diff --git a/cmd/arcade/main.go b/cmd/arcade/main.go index 3621b31..358fc0b 100644 --- a/cmd/arcade/main.go +++ b/cmd/arcade/main.go @@ -19,7 +19,6 @@ import ( "time" "github.com/coder/websocket" - "github.com/coder/websocket/wsjson" "github.com/drjones/quantum-arcade/pkg/fair" "github.com/drjones/quantum-arcade/pkg/fixed" "github.com/drjones/quantum-arcade/pkg/identity" @@ -629,18 +628,26 @@ func (s *server) handleWS(w http.ResponseWriter, r *http.Request) { defer unsubscribe() // Send the current state immediately so a joining phone is never blank. - if err := wsjson.Write(ctx, conn, rm.Snapshot()); err != nil { + first, err := json.Marshal(rm.Snapshot()) + if err != nil { return } + if err := conn.Write(ctx, websocket.MessageText, first); err != nil { + return + } + + // Frames arrive pre-serialised: the room marshals once per broadcast and + // every connection writes the same bytes. Marshalling per connection was + // the dominant cost under load. for { select { case <-ctx.Done(): return - case snap, ok := <-updates: + case payload, ok := <-updates: if !ok { return } - if err := wsjson.Write(ctx, conn, snap); err != nil { + if err := conn.Write(ctx, websocket.MessageText, payload); err != nil { return } } diff --git a/cmd/arcade/static/app.js b/cmd/arcade/static/app.js index eb8bc7d..eff50e9 100644 --- a/cmd/arcade/static/app.js +++ b/cmd/arcade/static/app.js @@ -191,8 +191,44 @@ function connect(game) { socket.onclose = () => setTimeout(() => connect(currentGame), 1200); } +/* The server sends a frame a few times a second; the multiplier curve is + * deterministic, so between frames the client computes it locally from the + * round's start time. This is why the animation is smooth without the server + * pushing sixty frames a second to every phone. */ +const TICK_HZ = 60; +const ROUND_TICKS = 60 * TICK_HZ; + +function multiplierAtTick(tick) { + if (tick <= 0) return 1; + if (tick >= ROUND_TICKS) tick = ROUND_TICKS - 1; + const remaining = 1 - tick / ROUND_TICKS; + return 1 / (remaining * remaining); +} + +function localMultiplier() { + if (!snapshot || snapshot.state !== 'running' || !snapshot.started_unix_milli) { + return snapshot ? parseFloat(snapshot.multiplier) : 1; + } + const elapsedMs = Date.now() - snapshot.started_unix_milli; + return multiplierAtTick(Math.floor((elapsedMs / 1000) * TICK_HZ)); +} + +/* Runs every animation frame while a round is in flight, so the number and the + * 3D scene update at display rate rather than at network rate. */ +function interpolate() { + if (!snapshot || snapshot.state !== 'running') return; + const m = localMultiplier(); + $('multiplier').textContent = m.toFixed(2) + '×'; + if (scene) scene.setState(Math.log(Math.max(1, m)) / Math.log(25), false); + if (myBet === 'in') { + $('action').textContent = `Cash out ${sats(stake * m)}`; + } + requestAnimationFrame(interpolate); +} + function onSnapshot(s) { const roundChanged = !snapshot || snapshot.round_id !== s.round_id; + const wasRunning = snapshot && snapshot.state === 'running'; const wasSettled = snapshot && snapshot.state === 'settled'; snapshot = s; if (roundChanged) myBet = null; @@ -205,9 +241,11 @@ function onSnapshot(s) { $('commitment').textContent = s.commitment || '—'; $('revealed').textContent = s.server_seed || 'sealed until the round ends'; - const potMsat = (s.players || []).reduce((n, p) => n + p.stake_msat, 0); - $('potline').textContent = potMsat - ? `${(s.players || []).length} in · ${sats(potMsat)} sats at stake` + // Counts come from the server aggregate: the player list in each frame is + // only the leaderboard, capped so a large room stays cheap to broadcast. + $('potline').textContent = s.player_count + ? `${s.player_count} in · ${sats(s.pot_msat)} sats at stake` + + (s.cashed_out_count ? ` · ${s.cashed_out_count} out` : '') : 'no bets yet'; // Drive the 3D scene. Progress is log-scaled so the early climb is visible @@ -239,6 +277,7 @@ function onSnapshot(s) { case 'running': { $('state').textContent = 'in flight'; + if (!wasRunning) requestAnimationFrame(interpolate); if (myBet === 'in') { const payout = stake * current; action.textContent = `Cash out ${sats(payout)}`; diff --git a/pkg/ledger/ledger.go b/pkg/ledger/ledger.go index 21ffa7c..7dff132 100644 --- a/pkg/ledger/ledger.go +++ b/pkg/ledger/ledger.go @@ -15,6 +15,7 @@ import ( "fmt" "math/big" "sort" + "strings" "time" "github.com/jackc/pgx/v5" @@ -83,58 +84,126 @@ func (l *Ledger) Post(ctx context.Context, kind string, roundID *int64, postings return 0, err } - ordered := append([]Posting(nil), postings...) - sort.Slice(ordered, func(i, j int) bool { - return ordered[i].AccountID < ordered[j].AccountID - }) - - for _, p := range ordered { - // Lock the account row first, then read its latest balance. Taking the - // lock before the read is what serializes concurrent spends. - var allowNegative bool - if err := tx.QueryRow(ctx, - `SELECT allow_negative FROM accounts WHERE id = $1 FOR UPDATE`, - p.AccountID).Scan(&allowNegative); err != nil { - return 0, fmt.Errorf("locking account %d: %w", p.AccountID, err) + // Merge duplicate accounts before locking. A transaction that touched the + // same account twice would otherwise read a stale balance for the second + // leg and write a posting that contradicts the first. + merged := make(map[int64]int64, len(postings)) + order := make([]int64, 0, len(postings)) + for _, p := range postings { + if _, seen := merged[p.AccountID]; !seen { + order = append(order, p.AccountID) } + merged[p.AccountID] += p.AmountMsat + } + sort.Slice(order, func(i, j int) bool { return order[i] < order[j] }) - var before int64 - if err := tx.QueryRow(ctx, - `SELECT COALESCE( - (SELECT balance_after FROM postings - WHERE account_id = $1 ORDER BY id DESC LIMIT 1), 0)`, - p.AccountID).Scan(&before); err != nil { - return 0, fmt.Errorf("reading balance of account %d: %w", p.AccountID, err) + ids := make([]int64, 0, len(order)) + amounts := make([]int64, 0, len(order)) + for _, id := range order { + if merged[id] == 0 { + continue // legs cancelled out; nothing to record } - - // Detect wraparound before trusting the result: a credit that - // overflows would otherwise land as a negative balance, and a debit - // that underflows as a positive one. - after := before + p.AmountMsat - if (p.AmountMsat > 0 && after < before) || (p.AmountMsat < 0 && after > before) { - return 0, fmt.Errorf("ledger: amount %d overflows the balance of account %d (%d)", - p.AmountMsat, p.AccountID, before) - } - if after < 0 && !allowNegative { - return 0, fmt.Errorf("%w: account %d holds %d, needs %d", - ErrInsufficientFunds, p.AccountID, before, -p.AmountMsat) - } - - if _, err := tx.Exec(ctx, - `INSERT INTO postings - (transaction_id, account_id, amount_msat, balance_before, balance_after) - VALUES ($1, $2, $3, $4, $5)`, - txID, p.AccountID, p.AmountMsat, before, after); err != nil { + ids = append(ids, id) + amounts = append(amounts, merged[id]) + } + if len(ids) == 0 { + // Every leg cancelled. The transaction row stands as a record that + // something was attempted, but there is no balance change to write. + if err := tx.Commit(ctx); err != nil { return 0, err } + return txID, nil + } + + // Lock every account first, in ascending id order so concurrent + // transactions cannot deadlock against each other. + // + // This must be its own statement. A single statement — even one whose CTE + // does FOR UPDATE — evaluates against one snapshot taken before the locks + // are held, so the balance read would see pre-lock data and concurrent + // transactions would silently overwrite each other. Splitting it means the + // second statement takes a fresh snapshot, by which point we hold the + // locks and no other writer can commit against these accounts. + if _, err := tx.Exec(ctx, + `SELECT id FROM accounts WHERE id = ANY($1) ORDER BY id FOR UPDATE`, + ids); err != nil { + return 0, fmt.Errorf("locking accounts: %w", err) + } + + // Now write every posting in one statement, however many legs there are. + // Doing this per-posting cost three round trips each, which made + // settlement scale in network latency rather than in real work. + rows, err := tx.Query(ctx, ` + WITH input AS ( + SELECT unnest($2::bigint[]) AS account_id, + unnest($3::bigint[]) AS amount + ), + current AS ( + SELECT i.account_id, + i.amount, + COALESCE((SELECT p.balance_after + FROM postings p + WHERE p.account_id = i.account_id + ORDER BY p.id DESC + LIMIT 1), 0) AS balance_before + FROM input i + ) + INSERT INTO postings + (transaction_id, account_id, amount_msat, balance_before, balance_after) + SELECT $1, account_id, amount, balance_before, balance_before + amount + FROM current + RETURNING account_id, balance_after`, + txID, ids, amounts) + if err != nil { + // The balance floor is enforced by a database trigger, so an overdraft + // surfaces here. Translate it into the domain error callers expect. + if isBalanceFloorViolation(err) { + return 0, fmt.Errorf("%w: %v", ErrInsufficientFunds, err) + } + return 0, err + } + + written := 0 + for rows.Next() { + var acct, after int64 + if err := rows.Scan(&acct, &after); err != nil { + rows.Close() + return 0, err + } + written++ + } + rows.Close() + if err := rows.Err(); err != nil { + if isBalanceFloorViolation(err) { + return 0, fmt.Errorf("%w: %v", ErrInsufficientFunds, err) + } + return 0, err + } + if written != len(ids) { + return 0, fmt.Errorf("ledger: wrote %d postings for %d accounts; "+ + "an account id does not exist", written, len(ids)) } if err := tx.Commit(ctx); err != nil { + if isBalanceFloorViolation(err) { + return 0, fmt.Errorf("%w: %v", ErrInsufficientFunds, err) + } return 0, err } return txID, nil } +// isBalanceFloorViolation reports whether an error is the database refusing to +// let an account go negative. +func isBalanceFloorViolation(err error) bool { + if err == nil { + return false + } + msg := err.Error() + return strings.Contains(msg, "may not go negative") || + strings.Contains(msg, "balance_nonnegative") +} + // Transfer moves funds between two accounts. This is the peer-to-peer path. func (l *Ledger) Transfer(ctx context.Context, from, to int64, amountMsat int64) (int64, error) { if amountMsat <= 0 { diff --git a/pkg/ledger/load_test.go b/pkg/ledger/load_test.go new file mode 100644 index 0000000..7d3d426 --- /dev/null +++ b/pkg/ledger/load_test.go @@ -0,0 +1,227 @@ +package ledger_test + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/drjones/quantum-arcade/pkg/ledger" +) + +// These measure throughput on the paths that decide whether the platform +// survives a crowd. They print numbers rather than asserting thresholds, +// because the numbers depend on the host — but the shape of the result is what +// matters, and a regression shows up immediately. + +// Every bet debits the player and credits the house. All of them contend on +// the same house account row, which is the first thing to check: if that lock +// serialises the workload, no amount of hardware helps. +func TestThroughputContendedHouseAccount(t *testing.T) { + if testing.Short() { + t.Skip("load test") + } + l := ledger.New(testPool(t)) + ctx := context.Background() + + house, err := l.EnsurePlayer(ctx, uniqueKey(t, "house")) + if err != nil { + t.Fatal(err) + } + + const players = 64 + const perPlayer = 20 + + ids := make([]int64, players) + for i := range ids { + id, err := l.EnsurePlayer(ctx, uniqueKey(t, fmt.Sprintf("p%d", i))) + if err != nil { + t.Fatal(err) + } + if _, err := l.Deposit(ctx, id, 10_000_000); err != nil { + t.Fatal(err) + } + ids[i] = id + } + + var ok, failed atomic.Int64 + start := time.Now() + + var wg sync.WaitGroup + for _, id := range ids { + wg.Add(1) + go func(id int64) { + defer wg.Done() + for i := 0; i < perPlayer; i++ { + _, err := l.Post(ctx, "bet", nil, []ledger.Posting{ + {AccountID: id, AmountMsat: -1000}, + {AccountID: house, AmountMsat: 1000}, + }) + if err != nil { + failed.Add(1) + } else { + ok.Add(1) + } + } + }(id) + } + wg.Wait() + + elapsed := time.Since(start) + rate := float64(ok.Load()) / elapsed.Seconds() + t.Logf("contended (shared house row): %d bets in %v = %.0f bets/sec (%d failed)", + ok.Load(), elapsed.Round(time.Millisecond), rate, failed.Load()) + t.Logf(" -> a 20s betting window absorbs about %.0f bets", rate*20) +} + +// The same workload with the house side spread over several accounts, to +// isolate how much of the cost is lock contention rather than raw database +// throughput. +func TestThroughputShardedHouseAccount(t *testing.T) { + if testing.Short() { + t.Skip("load test") + } + l := ledger.New(testPool(t)) + ctx := context.Background() + + const shards = 16 + shardIDs := make([]int64, shards) + for i := range shardIDs { + id, err := l.EnsurePlayer(ctx, uniqueKey(t, fmt.Sprintf("houseshard%d", i))) + if err != nil { + t.Fatal(err) + } + shardIDs[i] = id + } + + const players = 64 + const perPlayer = 20 + + ids := make([]int64, players) + for i := range ids { + id, err := l.EnsurePlayer(ctx, uniqueKey(t, fmt.Sprintf("sp%d", i))) + if err != nil { + t.Fatal(err) + } + if _, err := l.Deposit(ctx, id, 10_000_000); err != nil { + t.Fatal(err) + } + ids[i] = id + } + + var ok, failed atomic.Int64 + start := time.Now() + + var wg sync.WaitGroup + for n, id := range ids { + wg.Add(1) + go func(n int, id int64) { + defer wg.Done() + for i := 0; i < perPlayer; i++ { + // Each player uses a fixed shard, the way a real sharded + // house account would be selected. + shard := shardIDs[n%shards] + _, err := l.Post(ctx, "bet", nil, []ledger.Posting{ + {AccountID: id, AmountMsat: -1000}, + {AccountID: shard, AmountMsat: 1000}, + }) + if err != nil { + failed.Add(1) + } else { + ok.Add(1) + } + } + }(n, id) + } + wg.Wait() + + elapsed := time.Since(start) + rate := float64(ok.Load()) / elapsed.Seconds() + t.Logf("sharded (%d house rows): %d bets in %v = %.0f bets/sec (%d failed)", + shards, ok.Load(), elapsed.Round(time.Millisecond), rate, failed.Load()) + t.Logf(" -> a 20s betting window absorbs about %.0f bets", rate*20) +} + +// Settlement writes every payout for a round. At scale this is one large +// transaction, so its cost per posting is what decides how long a crowd waits +// between rounds. +func TestThroughputBatchSettlement(t *testing.T) { + if testing.Short() { + t.Skip("load test") + } + l := ledger.New(testPool(t)) + ctx := context.Background() + + house, _ := l.EnsurePlayer(ctx, uniqueKey(t, "settlehouse")) + if _, err := l.Deposit(ctx, house, 1_000_000_000); err != nil { + t.Fatal(err) + } + + for _, size := range []int{10, 100, 500, 1000} { + winners := make([]int64, size) + for i := range winners { + id, err := l.EnsurePlayer(ctx, uniqueKey(t, fmt.Sprintf("w%d-%d", size, i))) + if err != nil { + t.Fatal(err) + } + winners[i] = id + } + + postings := make([]ledger.Posting, 0, size+1) + for _, w := range winners { + postings = append(postings, ledger.Posting{AccountID: w, AmountMsat: 1000}) + } + postings = append(postings, + ledger.Posting{AccountID: house, AmountMsat: -int64(size) * 1000}) + + start := time.Now() + if _, err := l.Post(ctx, "payout", nil, postings); err != nil { + t.Fatalf("settling %d winners: %v", size, err) + } + elapsed := time.Since(start) + t.Logf("settle %4d winners in one transaction: %8v (%.2fms per winner)", + size, elapsed.Round(time.Millisecond), + float64(elapsed.Microseconds())/1000/float64(size)) + } +} + +// Balance reads are the most frequent query in the system: every client polls +// after every round. +func TestThroughputBalanceReads(t *testing.T) { + if testing.Short() { + t.Skip("load test") + } + l := ledger.New(testPool(t)) + ctx := context.Background() + + id, _ := l.EnsurePlayer(ctx, uniqueKey(t, "reader")) + if _, err := l.Deposit(ctx, id, 1_000_000); err != nil { + t.Fatal(err) + } + + const readers = 32 + const each = 100 + var ok atomic.Int64 + start := time.Now() + + var wg sync.WaitGroup + for i := 0; i < readers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < each; j++ { + if _, err := l.Balance(ctx, id); err == nil { + ok.Add(1) + } + } + }() + } + wg.Wait() + + elapsed := time.Since(start) + t.Logf("balance reads: %d in %v = %.0f reads/sec", + ok.Load(), elapsed.Round(time.Millisecond), + float64(ok.Load())/elapsed.Seconds()) +} diff --git a/pkg/room/bench_test.go b/pkg/room/bench_test.go new file mode 100644 index 0000000..02aeab1 --- /dev/null +++ b/pkg/room/bench_test.go @@ -0,0 +1,69 @@ +package room + +import ( + "encoding/json" + "testing" +) + +// Broadcast cost decides whether a crowd can watch the same round. At 60Hz +// with N subscribers the server does N marshals per tick unless the payload is +// serialised once and shared. +func BenchmarkSnapshotMarshal(b *testing.B) { + r := New("rocket", nil, nil) + for i := 0; i < 200; i++ { + r.bets[int64(i)] = &Bet{ + AccountID: int64(i), + Pubkey: []byte("0123456789abcdef0123456789abcdef"), + Nickname: "player", + StakeMsat: 10000, + } + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + snap := r.Snapshot() + if _, err := json.Marshal(snap); err != nil { + b.Fatal(err) + } + } +} + +// Snapshot alone, without serialisation: this is the lock-held portion, which +// blocks every other operation on the room. +func BenchmarkSnapshotOnly(b *testing.B) { + r := New("rocket", nil, nil) + for i := 0; i < 200; i++ { + r.bets[int64(i)] = &Bet{ + AccountID: int64(i), + Pubkey: []byte("0123456789abcdef0123456789abcdef"), + Nickname: "player", + StakeMsat: 10000, + } + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + _ = r.Snapshot() + } +} + +// Fan-out to many subscriber channels. +func BenchmarkBroadcast1000Subscribers(b *testing.B) { + r := New("rocket", nil, nil) + for i := 0; i < 50; i++ { + r.bets[int64(i)] = &Bet{ + AccountID: int64(i), Pubkey: []byte("key"), + Nickname: "p", StakeMsat: 1000, + } + } + // Drain subscribers so the buffered channels do not simply fill. + for i := 0; i < 1000; i++ { + ch, _ := r.Subscribe() + go func(c <-chan []byte) { + for range c { + } + }(ch) + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + r.broadcast() + } +} diff --git a/pkg/room/room.go b/pkg/room/room.go index 7bdb64b..7135041 100644 --- a/pkg/room/room.go +++ b/pkg/room/room.go @@ -12,7 +12,9 @@ package room import ( "context" "encoding/hex" + "encoding/json" "fmt" + "sort" "sync" "time" @@ -61,23 +63,47 @@ type Bet struct { // Snapshot is what clients render. It carries the seed inputs so a client can // verify the round the moment it settles. type Snapshot struct { - RoundID int64 `json:"round_id"` - Game string `json:"game"` - State State `json:"state"` - Tick int `json:"tick"` - Multiplier string `json:"multiplier"` - Commitment string `json:"commitment"` - ServerSeed string `json:"server_seed,omitempty"` // only once settled - CrashPoint string `json:"crash_point,omitempty"` // only once settled + RoundID int64 `json:"round_id"` + Game string `json:"game"` + State State `json:"state"` + Tick int `json:"tick"` + Multiplier string `json:"multiplier"` + Commitment string `json:"commitment"` + ServerSeed string `json:"server_seed,omitempty"` // only once settled + CrashPoint string `json:"crash_point,omitempty"` // only once settled + // Players is capped at MaxListedPlayers. Sending every player to every + // subscriber is O(n^2) in bandwidth and makes a large room impossible: + // 50k players broadcast to 50k phones is gigabytes per second. The full + // list is available on request; the feed carries the leaderboard. Players []Player `json:"players"` + PlayerCount int `json:"player_count"` + PotMsat int64 `json:"pot_msat"` + CashedOut int `json:"cashed_out_count"` HousePotMsat int64 `json:"house_pot_msat"` NextPhaseIn float64 `json:"next_phase_in_seconds"` + + // StartedUnixMilli is when the running phase began. Because the multiplier + // curve is deterministic, a client can compute the current value locally + // from this instead of being told it sixty times a second. + StartedUnixMilli int64 `json:"started_unix_milli,omitempty"` } +// MaxListedPlayers bounds the per-frame player list. +const MaxListedPlayers = 24 + +// BroadcastHz is how often a running round pushes a frame. Clients compute the +// multiplier locally between frames, so this only has to be often enough to +// correct drift and deliver cash-out news. +const BroadcastHz = 5 + // Player is the public view of a participant. type Player struct { - Nickname string `json:"nickname"` - PubkeyHex string `json:"pubkey"` + Nickname string `json:"nickname"` + // Pubkey is omitted from the live feed: it is 64 hex characters, it is + // most of the frame, and nothing in the interface displays it. The full + // participant list, with keys, is served by the verification endpoint + // after settlement — which is where it actually matters. + PubkeyHex string `json:"pubkey,omitempty"` StakeMsat int64 `json:"stake_msat"` CashedOut string `json:"cashed_out,omitempty"` PayoutMsat int64 `json:"payout_msat"` @@ -104,8 +130,11 @@ type Room struct { order [][]byte // participant pubkeys in join order phaseEnds time.Time - subscribers map[chan Snapshot]struct{} + subscribers map[chan []byte]struct{} subMu sync.Mutex + + // runStarted is when the current running phase began. + runStarted time.Time } func New(game string, pool *pgxpool.Pool, l *ledger.Ledger) *Room { @@ -115,7 +144,7 @@ func New(game string, pool *pgxpool.Pool, l *ledger.Ledger) *Room { ledger: l, state: StateSettled, bets: make(map[int64]*Bet), - subscribers: make(map[chan Snapshot]struct{}), + subscribers: make(map[chan []byte]struct{}), phaseEnds: time.Now(), } } @@ -123,8 +152,8 @@ func New(game string, pool *pgxpool.Pool, l *ledger.Ledger) *Room { // Subscribe returns a channel of snapshots. The channel is buffered and drops // updates rather than blocking the round loop: a slow phone must never stall // the game for everyone else. -func (r *Room) Subscribe() (<-chan Snapshot, func()) { - ch := make(chan Snapshot, 8) +func (r *Room) Subscribe() (<-chan []byte, func()) { + ch := make(chan []byte, 4) r.subMu.Lock() r.subscribers[ch] = struct{}{} r.subMu.Unlock() @@ -137,14 +166,24 @@ func (r *Room) Subscribe() (<-chan Snapshot, func()) { } } +// broadcast serialises the snapshot once and hands the same bytes to every +// subscriber. +// +// Letting each connection marshal its own copy costs ~355us per subscriber per +// frame, which at any real crowd size exceeds the tick interval by orders of +// magnitude. One marshal per frame turns fan-out into a pointer copy. func (r *Room) broadcast() { - snap := r.Snapshot() + payload, err := json.Marshal(r.Snapshot()) + if err != nil { + fmt.Printf("room %s: marshalling snapshot: %v\n", r.Game, err) + return + } r.subMu.Lock() defer r.subMu.Unlock() for ch := range r.subscribers { select { - case ch <- snap: - default: // subscriber is behind; skip this frame + case ch <- payload: + default: // subscriber is behind; drop this frame rather than stall } } } @@ -208,7 +247,17 @@ func (r *Room) step(ctx context.Context) error { if crashed { return r.settle(ctx) } - r.broadcast() + // The multiplier is a pure function of the tick, and the client has + // the same curve. So the feed does not need to carry it sixty times a + // second: clients interpolate locally from StartedUnixMilli and the + // server sends a correcting frame a few times a second. + // + // At 60Hz this fan-out was the single largest cost in the system. At + // BroadcastHz it is a rounding error, and the animation is smoother + // because it is no longer gated on network jitter. + if r.tick%(sim.TickHz/BroadcastHz) == 0 { + r.broadcast() + } } return nil } @@ -256,6 +305,7 @@ func (r *Room) startRunning(ctx context.Context) error { r.crashPoint = sim.CrashPoint(roundSeed) r.state = StateRunning r.tick = 0 + r.runStarted = time.Now() roundID := r.roundID crash := r.crashPoint r.mu.Unlock() @@ -449,11 +499,27 @@ func (r *Room) Snapshot() Snapshot { r.mu.RLock() defer r.mu.RUnlock() - players := make([]Player, 0, len(r.bets)) + // Aggregate over every player, but only serialise the largest few. + var pot int64 + cashed := 0 + all := make([]*Bet, 0, len(r.bets)) for _, b := range r.bets { + pot += b.StakeMsat + if b.CashedOutAt != 0 { + cashed++ + } + all = append(all, b) + } + // Partial ordering is enough: the list is a leaderboard, not a ledger. + sort.Slice(all, func(i, j int) bool { return all[i].StakeMsat > all[j].StakeMsat }) + if len(all) > MaxListedPlayers { + all = all[:MaxListedPlayers] + } + + players := make([]Player, 0, len(all)) + for _, b := range all { p := Player{ Nickname: b.Nickname, - PubkeyHex: hex.EncodeToString(b.Pubkey), StakeMsat: b.StakeMsat, PayoutMsat: b.PayoutMsat, } @@ -472,8 +538,14 @@ func (r *Room) Snapshot() Snapshot { Multiplier: sim.MultiplierAt(r.tick).String(), Commitment: hex.EncodeToString(r.commitment[:]), Players: players, + PlayerCount: len(r.bets), + PotMsat: pot, + CashedOut: cashed, NextPhaseIn: time.Until(r.phaseEnds).Seconds(), } + if r.state == StateRunning { + s.StartedUnixMilli = r.runStarted.UnixMilli() + } // The seed is revealed only once the round is over. if r.state == StateSettled && r.crashPoint != 0 { s.ServerSeed = r.serverSeed.Hex() diff --git a/pkg/room/room_test.go b/pkg/room/room_test.go index eb6394d..6138667 100644 --- a/pkg/room/room_test.go +++ b/pkg/room/room_test.go @@ -2,6 +2,7 @@ package room import ( "context" + "encoding/json" "fmt" "math/rand" "os" @@ -489,7 +490,11 @@ func TestSubscriberReceivesUpdates(t *testing.T) { f.openBetting() select { - case snap := <-ch: + case payload := <-ch: + var snap Snapshot + if err := json.Unmarshal(payload, &snap); err != nil { + t.Fatal(err) + } if snap.State != StateBetting { t.Fatalf("received state %q, want betting_open", snap.State) }