Test against over spending #2024
-
|
As a proof of concept. I would like to write some client code to test against double spending. Let me explain. I would like to have this sequence of 3 events happen at a massive scale with the last 2 competing with each other:
My worry is if event-2 lookup the balance, see 100 while event-3 happen to do the same thing at the same time and see 100 as well. I want to test that one of the transaction will actually fail and will not end up with a negative balance. In order to prevent that, the client will have to check if the balance will be > 0 after executing the withdraw and reject the transaction if otherwise. Can someone please help me model this simple problem in either: Go, Python or Node please? |
Beta Was this translation helpful? Give feedback.
Replies: 5 comments 8 replies
-
|
Hey, @pierrekttipay, thanks for asking and opening the discussion. |
Beta Was this translation helpful? Give feedback.
-
|
Hey @ostafen |
Beta Was this translation helpful? Give feedback.
-
|
The following snippets creates an The snippet then spawns If all works, we expect the balance to be decremented at most package main
import (
"context"
"errors"
"fmt"
"log"
"sync"
"github.com/codenotary/immudb/embedded/sql"
"github.com/codenotary/immudb/embedded/store"
)
func fatalOnErr(err error) {
if err != nil {
log.Fatal(err)
}
}
func main() {
st, err := store.Open("./data", store.DefaultOptions().WithMultiIndexing(true))
fatalOnErr(err)
defer st.Close()
e, err := sql.NewEngine(st, sql.DefaultOptions())
fatalOnErr(err)
ctx := context.Background()
_, _, err = e.Exec(ctx, nil, `
CREATE TABLE IF NOT EXISTS accounts(
id INTEGER,
balance INTEGER,
CHECK (balance >= 0),
PRIMARY KEY id
)
`, nil)
fatalOnErr(err)
_, _, err = e.Exec(ctx, nil, "UPSERT INTO accounts(id, balance) VALUES (1, 65)", nil)
fatalOnErr(err)
n := 10
var wg sync.WaitGroup
wg.Add(n)
for i := 0; i < n; i++ {
go func() {
defer wg.Done()
err := store.ErrTxReadConflict
for err == store.ErrTxReadConflict {
_, _, err = e.Exec(ctx, nil, "UPDATE accounts SET balance = balance - 10", nil)
}
if err != nil && !errors.Is(err, sql.ErrCheckConstraintViolation) {
fatalOnErr(err)
}
}()
}
wg.Wait()
reader, err := e.Query(ctx, nil, "SELECT * FROM accounts", nil)
fatalOnErr(err)
row, err := reader.Read(ctx)
fatalOnErr(err)
balance, _ := row.ValuesByPosition[1].RawValue().(int64)
if balance != 5 {
fatalOnErr(fmt.Errorf("unexpected balance: %d", balance))
}
fatalOnErr(reader.Close())
}@pierrekttipay: let me know if that is helpful and if you need further assistance. Thanks! |
Beta Was this translation helpful? Give feedback.
-
|
Thanks @ostafen , I will check the response in detail.
I think this would work well with interfacing with real 3rd party payment system. For instance:
|
Beta Was this translation helpful? Give feedback.
-
|
Hi @ostafen my code is: package main
import (
"context"
"fmt"
"log"
"sync"
immudb "github.com/codenotary/immudb/pkg/client"
)
func fatalOnErr(err error) {
if err != nil {
log.Fatal(err)
}
}
func main() {
opts := immudb.DefaultOptions().WithAddress("127.0.0.1").WithPort(3322)
e := immudb.NewClient().WithOptions(opts)
// connect with immudb server (user, password, database)
err := e.OpenSession(context.Background(), []byte("immudb"), []byte("immudb"), "defaultdb")
if err != nil {
log.Fatal(err)
}
// ensure connection is closed
defer e.CloseSession(context.Background())
ctx := context.Background()
if err != nil {
log.Fatalf("Load Table %s. Error while creating transaction", err)
}
_, err = e.SQLExec(ctx, `
CREATE TABLE IF NOT EXISTS accounts(
id INTEGER,
balance INTEGER,
CHECK (balance >= 0),
PRIMARY KEY id
)
`, nil)
fatalOnErr(err)
_, err = e.SQLExec(ctx, "UPSERT INTO accounts(id, balance) VALUES (1, 65)", nil)
fatalOnErr(err)
log.Printf("UPSERT $65")
n := 10
var wg sync.WaitGroup
wg.Add(n)
for i := 0; i < n; i++ {
go func() {
defer wg.Done()
tx, err := e.NewTx(ctx)
if err != nil {
fatalOnErr(err)
}
err = tx.SQLExec(ctx, "UPDATE accounts SET balance = balance - 10", nil)
if err != nil {
fatalOnErr(err)
}
log.Printf("UPDATE balance - $10")
_, err = tx.Commit(ctx)
if err != nil {
log.Printf("%v", err)
// fatalOnErr(err)
}
}()
}
wg.Wait()
res, err := e.SQLQuery(ctx, "SELECT * FROM accounts", map[string]interface{}{}, true)
fatalOnErr(err)
for _, row := range res.Rows {
log.Printf("Got row: %v\n", row)
vals := row.GetValues()
log.Printf("Got values: %v\n", vals)
balance := vals[1].GetN()
if balance != 5 {
fatalOnErr(fmt.Errorf("unexpected balance: %d", balance))
}
}
}my output varies. Sometimes balances is $45, sometimes $55. (starting with a balance of $65) If I try to extrapolate to a real scenario. I will have many customers trying to withdraw money at the same time. |
Beta Was this translation helpful? Give feedback.
-
|
Hey, @pierrekttipay, I think you can just retry the update in a loop (bounded by some maximum number of retries) and add a small random delay (or exponential backoff delay) when you get a read conflict. |
Beta Was this translation helpful? Give feedback.
-
|
I am struggling a little with my loop for i := 0; i < n; i++ {
go func() {
defer wg.Done()
_, err := updateAccount(ctx, e)
log.Printf("+%s", err)
for err != nil {
time.Sleep(1 * time.Nanosecond)
_, err = updateAccount(ctx, e)
log.Printf("-%v", err)
// fatalOnErr(err)
}
}()
}my conditional My updateAccount function is as such: func updateAccount(ctx context.Context, e immudb.ImmuClient) (string, error) {
tx, err := e.NewTx(ctx)
if err != nil {
fatalOnErr(err)
}
err = tx.SQLExec(ctx, "UPDATE accounts SET balance = balance - 10", nil)
if err != nil {
return "", err
}
log.Printf("UPDATE balance - $10")
_, err = tx.Commit(ctx)
if err != nil {
log.Printf("-%v", err)
// fatalOnErr(err)
}
return "", err
}I could only find the |
Beta Was this translation helpful? Give feedback.
-
|
For instance, doing this log.Printf("+%s", err)
for err != nil && !errors.Is(err, sql.ErrCheckConstraintViolation) {
time.Sleep(1 * time.Nanosecond)
_, err = updateAccount(ctx, e)
log.Printf("-%v", err)
// fatalOnErr(err)
}will not avoid the infinite loop here is a |
Beta Was this translation helpful? Give feedback.
-
|
Because the error returned from the remote server is wrapped in an immuError error. try: strings.Contains(err.Error(), sql.ErrCheckConstraintViolation.Error()) |
Beta Was this translation helpful? Give feedback.
-
|
Ok this worked. Thanks I would have thought there will be a more idiomatic way of doing it using |
Beta Was this translation helpful? Give feedback.
-
|
Hi again @ostafen Do you see any downside with that? package main
import (
"context"
"errors"
"fmt"
immudb "github.com/codenotary/immudb/pkg/client"
"log"
"sync"
"time"
)
var (
id int = 3
ErrNotEnoughBalance = errors.New("not enough balance")
)
func fatalOnErr(err error) {
if err != nil {
log.Fatal(err)
}
}
func updateAccount(ctx context.Context, e immudb.ImmuClient) error {
tx, err := e.NewTx(ctx)
if err != nil {
return err
}
res, err := tx.SQLQuery(ctx, fmt.Sprintf("SELECT balance FROM accounts2 WHERE id=%v", id), map[string]interface{}{})
fatalOnErr(err)
if res.Rows[0].GetValues()[0].GetN() > 10 {
err = tx.SQLExec(ctx, fmt.Sprintf("UPDATE accounts2 SET balance = balance - 10 WHERE id=%v", id), nil)
if err != nil {
return err
}
log.Printf("UPDATE balance - $10")
_, err = tx.Commit(ctx)
if err != nil {
log.Printf("%v", err)
}
} else {
err = tx.Rollback(ctx)
if err != nil {
log.Printf("%v", err)
}
return ErrNotEnoughBalance
}
return err
}
func main() {
opts := immudb.DefaultOptions().WithAddress("127.0.0.1").WithPort(3322)
e := immudb.NewClient().WithOptions(opts)
// connect with immudb server (user, password, database)
err := e.OpenSession(context.Background(), []byte("immudb"), []byte("immudb"), "defaultdb")
if err != nil {
log.Fatal(err)
}
// ensure connection is closed
defer e.CloseSession(context.Background())
ctx := context.Background()
if err != nil {
log.Fatalf("Load Table %s. Error while creating transaction", err)
}
_, err = e.SQLExec(ctx, `
CREATE TABLE IF NOT EXISTS accounts2(
id INTEGER,
balance INTEGER,
PRIMARY KEY id
)
`, nil)
fatalOnErr(err)
_, err = e.SQLExec(ctx, fmt.Sprintf("UPSERT INTO accounts2(id, balance) VALUES (%v, 65)", id), nil)
fatalOnErr(err)
log.Printf("UPSERT $65")
n := 10
var wg sync.WaitGroup
wg.Add(n)
for i := 0; i < n; i++ {
go func() {
defer wg.Done()
success := false
err := updateAccount(ctx, e)
for err != nil && !errors.Is(err, ErrNotEnoughBalance) {
time.Sleep(1 * time.Nanosecond)
err = updateAccount(ctx, e)
}
if err == nil {
success = true
}
log.Printf("STOP success:%v", success)
}()
}
wg.Wait()
log.Printf("ALL FINISHED")
res, err := e.SQLQuery(ctx, fmt.Sprintf("SELECT * FROM accounts2 WHERE id=%v", id), map[string]interface{}{}, true)
fatalOnErr(err)
for _, row := range res.Rows {
log.Printf("Got row: %v\n", row)
vals := row.GetValues()
log.Printf("Got values: %v\n", vals)
balance := vals[1].GetN()
if balance != 5 {
fatalOnErr(fmt.Errorf("unexpected balance: %d", balance))
}
}
} |
Beta Was this translation helpful? Give feedback.
-
|
It's totally fine, but I would use the check syntax unless you need to implement more complex checks |
Beta Was this translation helpful? Give feedback.
-
|
I marked as resolved, thank you very much. |
Beta Was this translation helpful? Give feedback.
The following snippets creates an
accountstable with two simple fields: id and balance. Then one account is inserted, with an initial balance of65. Note theCHECK balance >= 0part, which makes sure no update can result in a negative balance.The snippet then spawns
10concurrent goroutines, each trying to decrement the balance by10(they try as soon as a read conflict happens). If a unique constraint violation is detected the goroutine simply stops, as it would otherwise loop forever.If all works, we expect the balance to be decremented at most
6times, with a final balance of value of5.