games: the felt other people can sit at, and the version that settles the race
Phase B foundation for the multiplayer casino: the shared-table storage layer, the SSE fan-out, and the lock that only ever pretends to be the authority. - game_tables/game_seats/game_chat, plus a nullable table_id on game_live_hands so occupancy stays one row per player — the same primary key that stops a second solo hand stops a second seat. No second uniqueness domain, no split brain, no cash-out-to-zero while sitting on a pot. - The money model the plan sketched turned out simpler than it drew: chips cross the border only at sit-down and get-up, so a hand settles by moving the pot *within* the state blob and credits nobody. That deletes the payout ledger the design called for — there is no money write to make idempotent, only a state write conditional on the version. A replayed settle affects zero rows. - CommitTable/SitDown/LeaveTable each one transaction with the state write in it; the version column is the concurrency authority and the striped in-memory lock is only an optimisation over it, because a mutex does not survive a redeploy. - The SSE hub is a dumb byte fan-out: non-blocking sends (a stalled phone must not hold the table lock and freeze the clock for the room) and never a DB touch after the first read (holding the one connection open bricks the app). - DueTables/PushDeadlines for the turn clock to come; Chat keeps the hand_no it was said during, because at a money table collusion looks like chat. Storage and hub tested, including the version race and the never-block publish. No handlers wired yet, so nothing a player can see has changed. Claude-Session: https://claude.ai/code/session_013M5nD7PgUboJXoDcYHzpuJ
This commit is contained in:
117
internal/web/games_hub.go
Normal file
117
internal/web/games_hub.go
Normal file
@@ -0,0 +1,117 @@
|
||||
package web
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
// The SSE hub: how a move one player makes reaches the phones of everyone else
|
||||
// at the felt.
|
||||
//
|
||||
// It is in-memory and it is intentionally dumb. It holds no game state and makes
|
||||
// no decisions — it is a fan-out of opaque byte frames, keyed by table id. The
|
||||
// authority is always the database; a frame is a nudge that says "the table at
|
||||
// this version changed, come and look", and a subscriber that misses one (a
|
||||
// dropped send, a reconnect) refetches the table, which is authoritative anyway.
|
||||
// So a lost frame is a cosmetic hiccup, never a wrong balance.
|
||||
//
|
||||
// Two rules hold it together, and both are load-bearing:
|
||||
//
|
||||
// 1. **Sends are non-blocking.** A subscriber's channel is buffered, and a send
|
||||
// that would block is dropped, not waited on. The publish happens under the
|
||||
// table lock (which is what orders frames correctly for free), so a blocking
|
||||
// send would hold that lock while one phone on a train stalls — and the turn
|
||||
// clock behind that lock stalls with it, for the whole casino. A dropped frame
|
||||
// costs that one subscriber a refetch; a held lock costs everyone the room.
|
||||
//
|
||||
// 2. **The publisher never touches the database.** The hub is reached only after
|
||||
// the DB work is done and the connection released. Holding a *sql.Rows or a tx
|
||||
// open for the life of a stream would hold the one connection in the pool
|
||||
// forever, and a single subscriber would brick the whole application.
|
||||
|
||||
// hubFrame is what goes down the wire: an opaque payload the browser knows how to
|
||||
// read (a JSON table view), tagged with the version it represents so a subscriber
|
||||
// can tell a frame it already has from one it missed.
|
||||
type hubFrame struct {
|
||||
Version int64
|
||||
Data []byte
|
||||
}
|
||||
|
||||
// tableSub is one open EventSource: a buffered channel and the id that lets the
|
||||
// subscriber unregister itself when the stream closes.
|
||||
type tableSub struct {
|
||||
id int64
|
||||
ch chan hubFrame
|
||||
}
|
||||
|
||||
// gamesHub fans table frames out to whoever is watching each table.
|
||||
type gamesHub struct {
|
||||
mu sync.Mutex
|
||||
tables map[string]map[int64]*tableSub
|
||||
nextID atomic.Int64
|
||||
}
|
||||
|
||||
func newGamesHub() *gamesHub {
|
||||
return &gamesHub{tables: make(map[string]map[int64]*tableSub)}
|
||||
}
|
||||
|
||||
// subChanBuffer is how many frames a slow subscriber can fall behind before the
|
||||
// hub starts dropping theirs. A few is plenty: a subscriber that far behind is
|
||||
// going to refetch the authoritative table anyway, so buffering more just delays
|
||||
// that with staler frames.
|
||||
const subChanBuffer = 8
|
||||
|
||||
// subscribe registers a new watcher of a table and returns its channel plus the
|
||||
// unsubscribe to defer. The channel is buffered so a publish never blocks on a
|
||||
// reader that is mid-write to its socket.
|
||||
func (h *gamesHub) subscribe(tableID string) (<-chan hubFrame, func()) {
|
||||
sub := &tableSub{id: h.nextID.Add(1), ch: make(chan hubFrame, subChanBuffer)}
|
||||
|
||||
h.mu.Lock()
|
||||
subs := h.tables[tableID]
|
||||
if subs == nil {
|
||||
subs = make(map[int64]*tableSub)
|
||||
h.tables[tableID] = subs
|
||||
}
|
||||
subs[sub.id] = sub
|
||||
h.mu.Unlock()
|
||||
|
||||
return sub.ch, func() { h.unsubscribe(tableID, sub.id) }
|
||||
}
|
||||
|
||||
func (h *gamesHub) unsubscribe(tableID string, id int64) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
subs := h.tables[tableID]
|
||||
if subs == nil {
|
||||
return
|
||||
}
|
||||
delete(subs, id)
|
||||
if len(subs) == 0 {
|
||||
delete(h.tables, tableID)
|
||||
}
|
||||
}
|
||||
|
||||
// publish pushes a frame to everyone watching a table, dropping it for any
|
||||
// subscriber whose buffer is full rather than waiting on them. See rule 1: this
|
||||
// is called under the table lock, so it must never block.
|
||||
func (h *gamesHub) publish(tableID string, f hubFrame) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
for _, sub := range h.tables[tableID] {
|
||||
select {
|
||||
case sub.ch <- f:
|
||||
default:
|
||||
// Full buffer: this subscriber is behind. Dropping is correct — they will
|
||||
// refetch the authoritative table when they next read a version gap.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// watchers reports how many streams are open on a table. Used by the caller that
|
||||
// decides whether a frame is worth rendering at all.
|
||||
func (h *gamesHub) watchers(tableID string) int {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return len(h.tables[tableID])
|
||||
}
|
||||
94
internal/web/games_hub_test.go
Normal file
94
internal/web/games_hub_test.go
Normal file
@@ -0,0 +1,94 @@
|
||||
package web
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestHub_DeliversToSubscribers(t *testing.T) {
|
||||
h := newGamesHub()
|
||||
ch, done := h.subscribe("t1")
|
||||
defer done()
|
||||
|
||||
h.publish("t1", hubFrame{Version: 3, Data: []byte("hi")})
|
||||
f := <-ch
|
||||
if f.Version != 3 || string(f.Data) != "hi" {
|
||||
t.Fatalf("got %+v", f)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_OnlyToTheRightTable(t *testing.T) {
|
||||
h := newGamesHub()
|
||||
ch1, d1 := h.subscribe("t1")
|
||||
defer d1()
|
||||
ch2, d2 := h.subscribe("t2")
|
||||
defer d2()
|
||||
|
||||
h.publish("t1", hubFrame{Version: 1})
|
||||
select {
|
||||
case <-ch2:
|
||||
t.Fatal("t2 should not have received t1's frame")
|
||||
default:
|
||||
}
|
||||
if f := <-ch1; f.Version != 1 {
|
||||
t.Fatalf("t1 got %+v", f)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHub_PublishNeverBlocks is the load-bearing property: a subscriber that
|
||||
// never reads must not be able to hold up a publish, because publish happens
|
||||
// under the table lock and a blocked publish stalls the turn clock for everyone.
|
||||
func TestHub_PublishNeverBlocks(t *testing.T) {
|
||||
h := newGamesHub()
|
||||
_, done := h.subscribe("t1") // never read from
|
||||
defer done()
|
||||
|
||||
// Far more than the buffer. If any of these blocked, the test would hang.
|
||||
blocked := make(chan struct{})
|
||||
go func() {
|
||||
for i := 0; i < subChanBuffer*10; i++ {
|
||||
h.publish("t1", hubFrame{Version: int64(i)})
|
||||
}
|
||||
close(blocked)
|
||||
}()
|
||||
<-blocked
|
||||
}
|
||||
|
||||
func TestHub_UnsubscribeStopsDelivery(t *testing.T) {
|
||||
h := newGamesHub()
|
||||
ch, done := h.subscribe("t1")
|
||||
done()
|
||||
|
||||
if h.watchers("t1") != 0 {
|
||||
t.Fatalf("watchers should be 0 after unsubscribe, got %d", h.watchers("t1"))
|
||||
}
|
||||
h.publish("t1", hubFrame{Version: 1})
|
||||
select {
|
||||
case _, ok := <-ch:
|
||||
if ok {
|
||||
t.Fatal("a frame arrived after unsubscribe")
|
||||
}
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_ConcurrentSubscribers(t *testing.T) {
|
||||
h := newGamesHub()
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 50; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
ch, done := h.subscribe("t1")
|
||||
defer done()
|
||||
<-ch
|
||||
}()
|
||||
}
|
||||
// Let them all register, then flood so every one of them reads at least one.
|
||||
for h.watchers("t1") < 50 {
|
||||
}
|
||||
for i := 0; i < subChanBuffer; i++ {
|
||||
h.publish("t1", hubFrame{Version: int64(i)})
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
58
internal/web/games_lock.go
Normal file
58
internal/web/games_lock.go
Normal file
@@ -0,0 +1,58 @@
|
||||
package web
|
||||
|
||||
import (
|
||||
"hash/fnv"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// The striped table lock, and why it is only ever an optimisation.
|
||||
//
|
||||
// The database's version column is the real concurrency authority: every write
|
||||
// to a table is conditional on the version the writer read, so two writers that
|
||||
// race produce one winner and one ErrStaleTable no matter what happens in
|
||||
// memory. This lock exists purely to make the loser lose *before* it does the
|
||||
// work, rather than after — it serialises the read-modify-write on a table so the
|
||||
// common case doesn't burn an engine step and a marshal only to be told it was
|
||||
// stale.
|
||||
//
|
||||
// It is a fixed array hashed on table id, never a map you can delete from, and
|
||||
// that is deliberate. A map of mutexes keyed by table id, cleaned up when a table
|
||||
// empties, will hand two goroutines two different mutex objects for the same
|
||||
// table across a delete-and-recreate — which is no lock at all. A fixed array has
|
||||
// no lifecycle: the same id always hashes to the same mutex, forever. The only
|
||||
// cost is that two unrelated tables can collide onto one stripe and briefly wait
|
||||
// on each other, which is harmless.
|
||||
//
|
||||
// A redeploy is the case that proves the version column has to be the authority:
|
||||
// during a drain two processes are running, each with its own array, so a table
|
||||
// is "locked" by two mutexes that know nothing about each other. The version
|
||||
// column is the only thing both processes share, and it is what keeps them
|
||||
// correct while the mutexes are useless.
|
||||
|
||||
// lockStripes is how many mutexes the array holds. A power of two so the mask is
|
||||
// clean; large enough that collisions between live tables are rare.
|
||||
const lockStripes = 256
|
||||
|
||||
type stripedLocks struct {
|
||||
m [lockStripes]sync.Mutex
|
||||
}
|
||||
|
||||
func newStripedLocks() *stripedLocks { return &stripedLocks{} }
|
||||
|
||||
// forTable returns the mutex a given table hashes onto. The same id always
|
||||
// returns the same mutex.
|
||||
func (s *stripedLocks) forTable(id string) *sync.Mutex {
|
||||
h := fnv.New32a()
|
||||
_, _ = h.Write([]byte(id))
|
||||
return &s.m[h.Sum32()&(lockStripes-1)]
|
||||
}
|
||||
|
||||
// withTable runs fn while holding the table's stripe. The lock is released when
|
||||
// fn returns — it never spans a network read or an SSE send, only the
|
||||
// read-modify-write against the database.
|
||||
func (s *stripedLocks) withTable(id string, fn func() error) error {
|
||||
mu := s.forTable(id)
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
return fn()
|
||||
}
|
||||
@@ -76,6 +76,13 @@ type Server struct {
|
||||
metricsMu sync.Mutex
|
||||
saltDay int64
|
||||
salt [16]byte
|
||||
|
||||
// The shared-table machinery. hub fans SSE frames out to the phones at a felt;
|
||||
// tableLocks is the striped optimisation over the DB's version column (see
|
||||
// games_table.go). Both are nil-safe to construct always: they cost nothing
|
||||
// until a table is opened.
|
||||
hub *gamesHub
|
||||
tableLocks *stripedLocks
|
||||
}
|
||||
|
||||
// New builds the server. Templates are parsed once at startup. Each page gets
|
||||
@@ -136,7 +143,7 @@ func New(cfg config.WebConfig, sources []config.SourceConfig, postingEnabled boo
|
||||
live = append(live, ch)
|
||||
}
|
||||
|
||||
s := &Server{cfg: cfg, sources: infos, postingEnabled: postingEnabled, tpls: tpls, adminSubs: adminSubs, adv: adv, advPost: advPost, channels: live}
|
||||
s := &Server{cfg: cfg, sources: infos, postingEnabled: postingEnabled, tpls: tpls, adminSubs: adminSubs, adv: adv, advPost: advPost, channels: live, hub: newGamesHub(), tableLocks: newStripedLocks()}
|
||||
|
||||
// Optional OIDC sign-in (Authentik). Discovery is a network call; if the
|
||||
// provider is unreachable at boot we log and serve anonymously rather than
|
||||
|
||||
Reference in New Issue
Block a user