casino/pkg/ledger/batch.go

package ledger

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

// Batcher raises write throughput by grouping transactions.
//
// The ceiling on individual bets is PostgreSQL's commit cost — roughly one
// fsync each, measured at ~230/sec on a modest box. Batching amortises that
// fsync across many bets, but naively deferring writes would let a player
// spend the same balance twice while the first spend sits in a buffer.
//
// So a bet takes two steps:
//
//  1. Reserve, synchronously and in memory. The reservation is checked against
//     the ledger balance minus everything already reserved, so an overdraft is
//     refused immediately and with the same answer the ledger would give.
//  2. Flush, in the background. Reservations are written as one transaction.
//
// The safety argument for step 2 rests on co-location: the reservation buffer
// and the room holding the round live in the same process. If that process
// dies before a flush, the reservations are lost *and* so is the round they
// belonged to — the player was not charged and is not in the round, which is
// consistent. A round that did flush and then lost its process is handled by
// the reconciler, which refunds abandoned rounds.
//
// A round must therefore never settle before its bets have flushed.
// Room.settle enforces that by calling Flush first.
type Batcher struct {
	ledger *Ledger

	// maxDelay and maxBatch are read by the flush loop while callers may be
	// tuning them, so they are atomic rather than plain fields. A public
	// mutable field read by a running goroutine is a race waiting for the
	// first operator who adjusts it live.
	maxDelay atomic.Int64 // nanoseconds
	maxBatch atomic.Int64

	mu       sync.Mutex
	pending  []pendingTx
	reserved map[int64]int64 // account -> millisatoshis reserved but unwritten
	waiters  []chan error

	flushing sync.Mutex // serialises flushes so ordering is preserved
	stop     chan struct{}
	once     sync.Once

	// negativeOK caches whether an account may go negative. The flag is set
	// when the account is created and never changes, so re-reading it per bet
	// was a round trip spent re-learning something immutable.
	negMu      sync.RWMutex
	negativeOK map[int64]bool
}

type pendingTx struct {
	kind     string
	roundID  *int64
	postings []Posting
}

var ErrBatcherClosed = errors.New("ledger: batcher is closed")

func NewBatcher(l *Ledger) *Batcher {
	b := &Batcher{
		ledger:     l,
		reserved:   make(map[int64]int64),
		negativeOK: make(map[int64]bool),
		stop:       make(chan struct{}),
	}
	b.SetMaxDelay(200 * time.Millisecond)
	b.SetMaxBatch(256)
	return b
}

// SetMaxDelay sets how long a reservation may wait before being written. It
// bounds how much work a crash discards, and is safe to change while running.
func (b *Batcher) SetMaxDelay(d time.Duration) {
	if d < time.Millisecond {
		d = time.Millisecond
	}
	b.maxDelay.Store(int64(d))
}

// MaxDelay reports the current flush interval.
func (b *Batcher) MaxDelay() time.Duration {
	return time.Duration(b.maxDelay.Load())
}

// SetMaxBatch sets how many transactions may queue before an early flush, so a
// burst does not build an unboundedly large database transaction.
func (b *Batcher) SetMaxBatch(n int) {
	if n < 1 {
		n = 1
	}
	b.maxBatch.Store(int64(n))
}

// Run flushes on a timer until the context ends.
func (b *Batcher) Run(ctx context.Context) {
	interval := b.MaxDelay()
	t := time.NewTicker(interval)
	defer t.Stop()
	for {
		// Pick up a changed interval without restarting the loop.
		if d := b.MaxDelay(); d != interval {
			interval = d
			t.Reset(interval)
		}
		select {
		case <-ctx.Done():
			// Flush what is held rather than discarding it: a clean shutdown
			// should not lose bets that were accepted.
			_ = b.Flush(context.WithoutCancel(ctx))
			return
		case <-b.stop:
			_ = b.Flush(context.WithoutCancel(ctx))
			return
		case <-t.C:
			if err := b.Flush(ctx); err != nil {
				fmt.Printf("ledger: batch flush failed: %v\n", err)
			}
		}
	}
}

// Close stops the batcher after a final flush.
func (b *Batcher) Close() {
	b.once.Do(func() { close(b.stop) })
}

// AvailableBalance is what an account can actually spend: its ledger balance
// less anything reserved but not yet written.
func (b *Batcher) AvailableBalance(ctx context.Context, accountID int64) (int64, error) {
	settled, err := b.ledger.Balance(ctx, accountID)
	if err != nil {
		return 0, err
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	return settled + b.reserved[accountID], nil
}

// mayGoNegative reports whether an account is permitted a negative balance,
// caching the answer. The flag is immutable once an account exists.
func (b *Batcher) mayGoNegative(ctx context.Context, accountID int64) (bool, error) {
	b.negMu.RLock()
	v, ok := b.negativeOK[accountID]
	b.negMu.RUnlock()
	if ok {
		return v, nil
	}

	var allow bool
	if err := b.ledger.pool.QueryRow(ctx,
		`SELECT allow_negative FROM accounts WHERE id = $1`, accountID).Scan(&allow); err != nil {
		return false, fmt.Errorf("checking account %d: %w", accountID, err)
	}

	b.negMu.Lock()
	b.negativeOK[accountID] = allow
	b.negMu.Unlock()
	return allow, nil
}

// Post reserves a transaction and returns once it is durably written.
//
// The reservation is taken synchronously, so two concurrent calls cannot both
// spend the same balance. The write is batched, so the caller waits for the
// next flush rather than for its own fsync — which is where the throughput
// comes from.
func (b *Batcher) Post(ctx context.Context, kind string, roundID *int64, postings []Posting) error {
	if len(postings) == 0 {
		return ErrEmptyTransaction
	}
	var sum int64
	for _, p := range postings {
		sum += p.AmountMsat
	}
	if sum != 0 {
		return fmt.Errorf("%w: sum is %d", ErrUnbalanced, sum)
	}

	// Check every debit against the balance that will actually be available,
	// which is the settled balance plus reservations already taken.
	for _, p := range postings {
		if p.AmountMsat >= 0 {
			continue
		}
		available, err := b.AvailableBalance(ctx, p.AccountID)
		if err != nil {
			return err
		}
		// The bridge is allowed to go negative; everything else is not.
		allowNegative, err := b.mayGoNegative(ctx, p.AccountID)
		if err != nil {
			return err
		}
		if !allowNegative && available+p.AmountMsat < 0 {
			return fmt.Errorf("%w: account %d has %d available, needs %d",
				ErrInsufficientFunds, p.AccountID, available, -p.AmountMsat)
		}
	}

	done := make(chan error, 1)

	b.mu.Lock()
	select {
	case <-b.stop:
		b.mu.Unlock()
		return ErrBatcherClosed
	default:
	}
	for _, p := range postings {
		b.reserved[p.AccountID] += p.AmountMsat
	}
	b.pending = append(b.pending, pendingTx{kind: kind, roundID: roundID, postings: postings})
	b.waiters = append(b.waiters, done)
	full := int64(len(b.pending)) >= b.maxBatch.Load()
	b.mu.Unlock()

	if full {
		go func() {
			if err := b.Flush(context.WithoutCancel(ctx)); err != nil {
				fmt.Printf("ledger: batch flush failed: %v\n", err)
			}
		}()
	}

	select {
	case err := <-done:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

// Flush writes every pending transaction.
//
// Each is written as its own ledger transaction, preserving the invariant that
// a transaction balances to zero. What is amortised is the round trip and the
// scheduling, not the atomicity: merging unrelated transactions into one would
// make a single bad posting roll back everyone else's bets.
func (b *Batcher) Flush(ctx context.Context) error {
	b.flushing.Lock()
	defer b.flushing.Unlock()

	b.mu.Lock()
	if len(b.pending) == 0 {
		b.mu.Unlock()
		return nil
	}
	batch := b.pending
	waiters := b.waiters
	b.pending = nil
	b.waiters = nil
	b.mu.Unlock()

	release := func(tx pendingTx) {
		// Release the reservation whether or not the write succeeded: it has
		// either become a real posting or it never will, and in both cases
		// holding it would understate what the account can spend.
		b.mu.Lock()
		for _, p := range tx.postings {
			b.reserved[p.AccountID] -= p.AmountMsat
			if b.reserved[p.AccountID] == 0 {
				delete(b.reserved, p.AccountID)
			}
		}
		b.mu.Unlock()
	}

	groups := make([]Group, len(batch))
	for i, tx := range batch {
		groups[i] = Group{Kind: tx.kind, RoundID: tx.roundID, Postings: tx.postings}
	}

	// The fast path: one database transaction for the whole batch, so the
	// commit cost is paid once instead of once per bet.
	if _, err := b.ledger.PostMany(ctx, groups); err == nil {
		for i, tx := range batch {
			release(tx)
			waiters[i] <- nil
		}
		return nil
	}

	// Something in the batch was rejected. Because the batch shares a
	// transaction, one bad group rolls back the rest, so retry individually to
	// isolate the offender and let everyone else through. This is rare:
	// balances are checked before a group is ever queued.
	var firstErr error
	for i, tx := range batch {
		_, err := b.ledger.Post(ctx, tx.kind, tx.roundID, tx.postings)
		release(tx)
		waiters[i] <- err
		if err != nil && firstErr == nil {
			firstErr = err
		}
	}
	return firstErr
}

// Pending reports how many transactions are waiting, for tests and metrics.
func (b *Batcher) Pending() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return len(b.pending)
}