casino/pkg/ledger/batch_test.go

package ledger_test

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/drjones/quantum-arcade/pkg/ledger"
)

// Batching is only worth having if it cannot lose or create money. These pin
// that down before any throughput claim is made.

func newBatcher(t *testing.T) (*ledger.Batcher, *ledger.Ledger, context.Context) {
	t.Helper()
	l := ledger.New(testPool(t))
	b := ledger.NewBatcher(l)
	b.SetMaxDelay(50 * time.Millisecond)

	ctx, cancel := context.WithCancel(context.Background())
	go b.Run(ctx)
	t.Cleanup(func() {
		b.Close()
		cancel()
	})
	return b, l, context.Background()
}

func TestBatchedPostIsDurable(t *testing.T) {
	b, l, ctx := newBatcher(t)
	from, _ := l.EnsurePlayer(ctx, uniqueKey(t, "from"))
	to, _ := l.EnsurePlayer(ctx, uniqueKey(t, "to"))
	if _, err := l.Deposit(ctx, from, 100_000); err != nil {
		t.Fatal(err)
	}

	if err := b.Post(ctx, "transfer", nil, []ledger.Posting{
		{AccountID: from, AmountMsat: -10_000},
		{AccountID: to, AmountMsat: 10_000},
	}); err != nil {
		t.Fatal(err)
	}

	// Post returns only once written, so the ledger must already show it.
	if bal, _ := l.Balance(ctx, to); bal != 10_000 {
		t.Fatalf("recipient balance = %d after Post returned, want 10000", bal)
	}
}

// The property that makes deferred writes safe: a reservation must count
// against the balance immediately, or the same funds could be spent twice
// while the first spend sits in the buffer.
func TestReservationsPreventDoubleSpend(t *testing.T) {
	b, l, ctx := newBatcher(t)
	from, _ := l.EnsurePlayer(ctx, uniqueKey(t, "from"))
	to, _ := l.EnsurePlayer(ctx, uniqueKey(t, "to"))
	if _, err := l.Deposit(ctx, from, 10_000); err != nil {
		t.Fatal(err)
	}

	const workers = 12
	var wg sync.WaitGroup
	var ok atomic.Int64
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			// Each tries to spend the entire balance.
			if err := b.Post(ctx, "transfer", nil, []ledger.Posting{
				{AccountID: from, AmountMsat: -10_000},
				{AccountID: to, AmountMsat: 10_000},
			}); err == nil {
				ok.Add(1)
			}
		}()
	}
	wg.Wait()

	if ok.Load() != 1 {
		t.Fatalf("%d of %d concurrent spends of the same balance succeeded, want 1",
			ok.Load(), workers)
	}
	if bal, _ := l.Balance(ctx, from); bal != 0 {
		t.Fatalf("source balance = %d, want 0", bal)
	}
	if bal, _ := l.Balance(ctx, to); bal != 10_000 {
		t.Fatalf("recipient balance = %d, want exactly one transfer of 10000", bal)
	}
}

func TestAvailableBalanceReflectsReservations(t *testing.T) {
	b, l, ctx := newBatcher(t)
	b.SetMaxDelay(5 * time.Second) // hold the flush so the reservation is visible

	from, _ := l.EnsurePlayer(ctx, uniqueKey(t, "from"))
	to, _ := l.EnsurePlayer(ctx, uniqueKey(t, "to"))
	if _, err := l.Deposit(ctx, from, 50_000); err != nil {
		t.Fatal(err)
	}

	go func() {
		_ = b.Post(ctx, "transfer", nil, []ledger.Posting{
			{AccountID: from, AmountMsat: -20_000},
			{AccountID: to, AmountMsat: 20_000},
		})
	}()

	// Wait for the reservation to be taken but not yet written.
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		avail, err := b.AvailableBalance(ctx, from)
		if err != nil {
			t.Fatal(err)
		}
		if avail == 30_000 {
			return // reserved amount is subtracted, as it must be
		}
		time.Sleep(10 * time.Millisecond)
	}
	avail, _ := b.AvailableBalance(ctx, from)
	t.Fatalf("available balance = %d while 20000 is reserved, want 30000", avail)
}

func TestOverdraftRefusedBeforeReserving(t *testing.T) {
	b, l, ctx := newBatcher(t)
	from, _ := l.EnsurePlayer(ctx, uniqueKey(t, "from"))
	to, _ := l.EnsurePlayer(ctx, uniqueKey(t, "to"))
	if _, err := l.Deposit(ctx, from, 1_000); err != nil {
		t.Fatal(err)
	}

	err := b.Post(ctx, "transfer", nil, []ledger.Posting{
		{AccountID: from, AmountMsat: -5_000},
		{AccountID: to, AmountMsat: 5_000},
	})
	if !errors.Is(err, ledger.ErrInsufficientFunds) {
		t.Fatalf("got %v, want ErrInsufficientFunds", err)
	}
	// And a subsequent affordable spend must still work, proving the refused
	// attempt left no reservation behind.
	if err := b.Post(ctx, "transfer", nil, []ledger.Posting{
		{AccountID: from, AmountMsat: -1_000},
		{AccountID: to, AmountMsat: 1_000},
	}); err != nil {
		t.Fatalf("an affordable spend after a refused one failed: %v", err)
	}
}

func TestUnbalancedBatchIsRefused(t *testing.T) {
	b, l, ctx := newBatcher(t)
	a, _ := l.EnsurePlayer(ctx, uniqueKey(t, "a"))
	c, _ := l.EnsurePlayer(ctx, uniqueKey(t, "c"))

	if err := b.Post(ctx, "bad", nil, []ledger.Posting{
		{AccountID: a, AmountMsat: -100},
		{AccountID: c, AmountMsat: 50},
	}); !errors.Is(err, ledger.ErrUnbalanced) {
		t.Fatalf("got %v, want ErrUnbalanced", err)
	}
}

// Value must be conserved across a large batched workload.
func TestBatchedWorkloadConservesValue(t *testing.T) {
	b, l, ctx := newBatcher(t)

	const players = 10
	ids := make([]int64, players)
	for i := range ids {
		id, _ := l.EnsurePlayer(ctx, uniqueKey(t, string(rune('a'+i))))
		if _, err := l.Deposit(ctx, id, 100_000); err != nil {
			t.Fatal(err)
		}
		ids[i] = id
	}

	sumOwn := func() int64 {
		var total int64
		for _, id := range ids {
			bal, err := l.Balance(ctx, id)
			if err != nil {
				t.Fatal(err)
			}
			total += bal
		}
		return total
	}
	before := sumOwn()

	var wg sync.WaitGroup
	for i := 0; i < players; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			for j := 0; j < 20; j++ {
				from := ids[i]
				to := ids[(i+1)%players]
				_ = b.Post(ctx, "transfer", nil, []ledger.Posting{
					{AccountID: from, AmountMsat: -500},
					{AccountID: to, AmountMsat: 500},
				})
			}
		}(i)
	}
	wg.Wait()
	if err := b.Flush(ctx); err != nil {
		t.Fatal(err)
	}

	if after := sumOwn(); after != before {
		t.Fatalf("batched workload changed total value: %d -> %d", before, after)
	}
	for _, id := range ids {
		if bal, _ := l.Balance(ctx, id); bal < 0 {
			t.Fatalf("account %d went negative under batching: %d", id, bal)
		}
	}
}

// Flush must drain everything, so a caller can guarantee durability before
// settling a round.
func TestFlushDrainsEverything(t *testing.T) {
	b, l, ctx := newBatcher(t)
	b.SetMaxDelay(time.Hour) // only an explicit Flush will write

	from, _ := l.EnsurePlayer(ctx, uniqueKey(t, "from"))
	to, _ := l.EnsurePlayer(ctx, uniqueKey(t, "to"))
	if _, err := l.Deposit(ctx, from, 100_000); err != nil {
		t.Fatal(err)
	}

	for i := 0; i < 5; i++ {
		go func() {
			_ = b.Post(ctx, "transfer", nil, []ledger.Posting{
				{AccountID: from, AmountMsat: -1_000},
				{AccountID: to, AmountMsat: 1_000},
			})
		}()
	}
	// Wait for them to be queued.
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) && b.Pending() < 5 {
		time.Sleep(10 * time.Millisecond)
	}

	if err := b.Flush(ctx); err != nil {
		t.Fatal(err)
	}
	if p := b.Pending(); p != 0 {
		t.Fatalf("%d transactions still pending after Flush", p)
	}
	if bal, _ := l.Balance(ctx, to); bal != 5_000 {
		t.Fatalf("recipient balance = %d after flush, want 5000", bal)
	}
}

// The throughput claim, measured rather than asserted.
func TestBatchedThroughputBeatsDirect(t *testing.T) {
	if testing.Short() {
		t.Skip("throughput measurement")
	}
	b, l, ctx := newBatcher(t)
	// A longer window collects larger batches, which is where the amortisation
	// comes from. 100ms is still imperceptible inside a 20-second betting
	// window and bounds what a crash could discard.
	b.SetMaxDelay(100 * time.Millisecond)

	house, _ := l.EnsurePlayer(ctx, uniqueKey(t, "house"))
	const players = 32
	ids := make([]int64, players)
	for i := range ids {
		id, _ := l.EnsurePlayer(ctx, uniqueKey(t, "p"+string(rune('a'+i%26))+string(rune('0'+i/26))))
		if _, err := l.Deposit(ctx, id, 10_000_000); err != nil {
			t.Fatal(err)
		}
		ids[i] = id
	}

	run := func(post func(from int64) error) float64 {
		var wg sync.WaitGroup
		var ok atomic.Int64
		start := time.Now()
		for _, id := range ids {
			wg.Add(1)
			go func(id int64) {
				defer wg.Done()
				for i := 0; i < 15; i++ {
					if err := post(id); err == nil {
						ok.Add(1)
					}
				}
			}(id)
		}
		wg.Wait()
		return float64(ok.Load()) / time.Since(start).Seconds()
	}

	direct := run(func(from int64) error {
		_, err := l.Post(ctx, "bet", nil, []ledger.Posting{
			{AccountID: from, AmountMsat: -100},
			{AccountID: house, AmountMsat: 100},
		})
		return err
	})

	batched := run(func(from int64) error {
		return b.Post(ctx, "bet", nil, []ledger.Posting{
			{AccountID: from, AmountMsat: -100},
			{AccountID: house, AmountMsat: 100},
		})
	})

	t.Logf("direct:  %.0f bets/sec", direct)
	t.Logf("batched: %.0f bets/sec  (%.1fx)", batched, batched/direct)
	t.Logf("  -> a 20s betting window absorbs about %.0f batched bets", batched*20)

	if batched <= direct {
		t.Fatalf("batching did not improve throughput: %.0f vs %.0f", batched, direct)
	}
}