diff --git a/README.md b/README.md index 5e64b04..655d31a 100644 --- a/README.md +++ b/README.md @@ -123,6 +123,20 @@ bugs: a posting set that minted 18 quintillion millisatoshis by wrapping the zero-sum check, a crash point that overflowed negative at the rarest seed, and scratch tables that advertised 98% while paying 56%. +## Scaling + +The app is stateless: clone the VM and boot it. An instance generates its own +identity, finds its peers through Redis, and campaigns for the games it will +drive. Exactly one instance runs a given game's rounds; the rest relay its +frames and forward bets to it. + +An instance dying is not a special case — its lease expires and a survivor +takes over. Measured at six seconds, unattended, after a `kill -9`. + +Clone the **app** VM only. PostgreSQL and Redis stay on one shared machine; +cloning those gives every instance its own ledger and they share nothing. +Full topology in [docs/SCALING.md](docs/SCALING.md). + ## Status Built and tested: diff --git a/cmd/arcade/forward.go b/cmd/arcade/forward.go new file mode 100644 index 0000000..84f40ac --- /dev/null +++ b/cmd/arcade/forward.go @@ -0,0 +1,95 @@ +package main + +import ( + "bytes" + "fmt" + "io" + "net/http" + "time" +) + +// forwardClient is deliberately short-tempered. A bet or cash-out is only +// useful while the round is still open, so a request that cannot be delivered +// promptly should fail loudly rather than land after the moment has passed. +var forwardClient = &http.Client{Timeout: 3 * time.Second} + +// forwardToLeader proxies a mutation to whichever instance drives the game. +// +// Only one instance holds a game's authoritative round state — who is in, at +// what stake, and at which tick each cash-out landed. Serving a bet from a +// follower's idle copy would either fail or, worse, create a second version of +// the round. So the follower relays the request and returns the leader's +// answer verbatim. +// +// The caller's Authorization header travels with it. That works because +// sessions live in Redis, so the leader can validate a token issued by any +// instance in the fleet. +// +// It reports whether the request was handled here. +func (s *server) forwardToLeader(w http.ResponseWriter, r *http.Request, game string, body []byte) bool { + hub, ok := s.hubs[game] + if !ok { + writeErr(w, http.StatusNotFound, "no such game") + return true + } + if hub.Leading() { + return false // this instance owns the round; handle it locally + } + + leader, err := s.node.LeaderOf(r.Context(), game) + if err != nil { + writeErr(w, http.StatusServiceUnavailable, "cannot locate the game leader") + return true + } + if leader.Address == "" { + // Between leaders: a lease has expired and the next campaign has not + // landed yet. This resolves within a couple of seconds on its own. + writeErr(w, http.StatusServiceUnavailable, + "this game is changing hands; try again in a moment") + return true + } + + url := fmt.Sprintf("http://%s%s", leader.Address, r.URL.Path) + req, err := http.NewRequestWithContext(r.Context(), r.Method, url, bytes.NewReader(body)) + if err != nil { + writeErr(w, http.StatusInternalServerError, err.Error()) + return true + } + req.Header.Set("Content-Type", "application/json") + if auth := r.Header.Get("Authorization"); auth != "" { + req.Header.Set("Authorization", auth) + } + // Mark the hop so a routing mistake shows up as an explicit loop error + // rather than as instances bouncing a request between themselves. + if r.Header.Get("X-Arcade-Forwarded") != "" { + writeErr(w, http.StatusLoopDetected, + "request was forwarded twice; the cluster disagrees about the leader") + return true + } + req.Header.Set("X-Arcade-Forwarded", s.node.ID) + + res, err := forwardClient.Do(req) + if err != nil { + writeErr(w, http.StatusServiceUnavailable, + "the instance running this game did not respond") + return true + } + defer res.Body.Close() + + payload, err := io.ReadAll(io.LimitReader(res.Body, 1<<20)) + if err != nil { + writeErr(w, http.StatusBadGateway, "truncated response from the game leader") + return true + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(res.StatusCode) + _, _ = w.Write(payload) + return true +} + +// readBody buffers a request body so it can be both parsed locally and +// forwarded if this instance turns out not to own the game. +func readBody(r *http.Request) ([]byte, error) { + defer r.Body.Close() + return io.ReadAll(io.LimitReader(r.Body, 1<<20)) +} diff --git a/cmd/arcade/hub.go b/cmd/arcade/hub.go new file mode 100644 index 0000000..78e47f0 --- /dev/null +++ b/cmd/arcade/hub.go @@ -0,0 +1,271 @@ +package main + +import ( + "context" + "errors" + "log" + "net" + "os" + "strings" + "sync" + "time" + + "github.com/drjones/quantum-arcade/pkg/cluster" + "github.com/drjones/quantum-arcade/pkg/room" +) + +// gameHub owns one game across the cluster. +// +// Exactly one instance leads a game: it runs the round loop, drives the +// simulation, settles to the ledger, and publishes each frame. Every other +// instance relays those frames to its own connected clients. Clients cannot +// tell the difference, and neither can the ledger. +// +// Roles are renegotiated on a timer rather than agreed once, so an instance +// disappearing is not a special case — its lease simply stops being renewed +// and the next campaign hands the game to someone else. +type gameHub struct { + game string + room *room.Room + node *cluster.Node + + mu sync.RWMutex + leading bool + subs map[chan []byte]struct{} + + // lastFrame is the most recent frame this instance saw, whether it + // produced it or relayed it. A follower's own room object sits idle, so + // this — not the local room — is what any read of "current state" must + // use, or a follower would report a permanently settled game. + lastFrame []byte + + // cancelLead stops the leader's round loop when leadership is lost. + cancelLead context.CancelFunc +} + +func newGameHub(game string, r *room.Room, node *cluster.Node) *gameHub { + return &gameHub{ + game: game, + room: r, + node: node, + subs: make(map[chan []byte]struct{}), + } +} + +// Subscribe returns frames for this game, whether this instance is producing +// them or relaying them. +func (h *gameHub) Subscribe() (<-chan []byte, func()) { + ch := make(chan []byte, 4) + h.mu.Lock() + h.subs[ch] = struct{}{} + h.mu.Unlock() + return ch, func() { + h.mu.Lock() + delete(h.subs, ch) + close(ch) + h.mu.Unlock() + } +} + +// fanout delivers a frame to this instance's own clients. A client that has +// stopped reading is skipped rather than allowed to stall the game. +func (h *gameHub) fanout(payload []byte) { + h.mu.Lock() + h.lastFrame = payload + h.mu.Unlock() + + h.mu.RLock() + defer h.mu.RUnlock() + for ch := range h.subs { + select { + case ch <- payload: + default: + } + } +} + +// LastFrame returns the most recent state this instance knows about, and +// whether it has seen one yet. +func (h *gameHub) LastFrame() ([]byte, bool) { + h.mu.RLock() + defer h.mu.RUnlock() + return h.lastFrame, len(h.lastFrame) > 0 +} + +// Leading reports whether this instance currently drives the game. +func (h *gameHub) Leading() bool { + h.mu.RLock() + defer h.mu.RUnlock() + return h.leading +} + +// supervise campaigns for the game and switches roles as leadership moves. +func (h *gameHub) supervise(ctx context.Context) { + // Followers hold a subscription to the cluster's frame channel. It is torn + // down on promotion so a leader never relays its own frames back to itself. + var stopRelay func() + defer func() { + if stopRelay != nil { + stopRelay() + } + }() + + ticker := time.NewTicker(cluster.RenewInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + h.demote() + return + case <-ticker.C: + } + + var held bool + var err error + if h.Leading() { + held, err = h.node.Renew(ctx, h.game) + } else { + held, err = h.node.Campaign(ctx, h.game) + } + if err != nil { + // Redis is unreachable. A current leader keeps running rather than + // abandoning a round mid-flight; its lease will expire and another + // instance will take over if the outage outlasts it. + log.Printf("hub %s: coordination error: %v", h.game, err) + continue + } + + switch { + case held && !h.Leading(): + if stopRelay != nil { + stopRelay() + stopRelay = nil + } + h.promote(ctx) + + case !held && h.Leading(): + h.demote() + stopRelay = h.startRelay(ctx) + + case !held && stopRelay == nil: + // Follower with no relay yet — subscribe so clients see the game. + stopRelay = h.startRelay(ctx) + } + } +} + +// promote starts driving the game on this instance. +func (h *gameHub) promote(ctx context.Context) { + leadCtx, cancel := context.WithCancel(ctx) + + h.mu.Lock() + h.leading = true + h.cancelLead = cancel + h.mu.Unlock() + + log.Printf("hub %s: leading", h.game) + + // Forward the room's frames both to this instance's clients and to peers. + frames, unsubscribe := h.room.Subscribe() + go func() { + defer unsubscribe() + for { + select { + case <-leadCtx.Done(): + return + case payload, ok := <-frames: + if !ok { + return + } + h.fanout(payload) + if err := h.node.PublishFrame(leadCtx, h.game, payload); err != nil && + !errors.Is(err, context.Canceled) { + log.Printf("hub %s: publishing frame: %v", h.game, err) + } + } + } + }() + + go func() { + if err := h.room.Run(leadCtx); err != nil && !errors.Is(err, context.Canceled) { + log.Printf("hub %s: round loop stopped: %v", h.game, err) + } + }() +} + +// demote stops driving the game. +func (h *gameHub) demote() { + h.mu.Lock() + wasLeading := h.leading + h.leading = false + cancel := h.cancelLead + h.cancelLead = nil + h.mu.Unlock() + + if cancel != nil { + cancel() + } + if wasLeading { + log.Printf("hub %s: no longer leading", h.game) + } +} + +// startRelay subscribes to the leader's frames and passes them to this +// instance's clients. +func (h *gameHub) startRelay(ctx context.Context) func() { + frames, unsubscribe := h.node.SubscribeFrames(ctx, h.game) + relayCtx, cancel := context.WithCancel(ctx) + + go func() { + for { + select { + case <-relayCtx.Done(): + return + case payload, ok := <-frames: + if !ok { + return + } + h.fanout(payload) + } + } + }() + + return func() { + cancel() + unsubscribe() + } +} + +// advertiseAddr is how peers reach this instance. +// +// It prefers an explicit setting, then the first non-loopback address it can +// find — so a cloned VM that gets its address from DHCP advertises correctly +// without being told what it is. +func advertiseAddr() string { + if v := os.Getenv("ARCADE_ADVERTISE"); v != "" { + return v + } + port := os.Getenv("ARCADE_ADDR") + if port == "" { + port = ":8080" + } + if !strings.HasPrefix(port, ":") { + if _, p, err := net.SplitHostPort(port); err == nil { + port = ":" + p + } + } + + addrs, err := net.InterfaceAddrs() + if err != nil { + return "127.0.0.1" + port + } + for _, a := range addrs { + ipnet, ok := a.(*net.IPNet) + if !ok || ipnet.IP.IsLoopback() || ipnet.IP.To4() == nil { + continue + } + return ipnet.IP.String() + port + } + return "127.0.0.1" + port +} diff --git a/cmd/arcade/main.go b/cmd/arcade/main.go index 358fc0b..f251c9b 100644 --- a/cmd/arcade/main.go +++ b/cmd/arcade/main.go @@ -14,11 +14,12 @@ import ( "os" "os/signal" "strconv" - "sync" + "strings" "syscall" "time" "github.com/coder/websocket" + "github.com/drjones/quantum-arcade/pkg/cluster" "github.com/drjones/quantum-arcade/pkg/fair" "github.com/drjones/quantum-arcade/pkg/fixed" "github.com/drjones/quantum-arcade/pkg/identity" @@ -26,6 +27,7 @@ import ( "github.com/drjones/quantum-arcade/pkg/room" "github.com/drjones/quantum-arcade/pkg/scratch" "github.com/jackc/pgx/v5/pgxpool" + "github.com/redis/go-redis/v9" ) //go:embed static @@ -44,16 +46,21 @@ type server struct { ledger *ledger.Ledger auth *identity.Authenticator rooms map[string]*room.Room + hubs map[string]*gameHub - // sessions maps a bearer token to a verified public key. Sessions live in - // memory only: restarting the server signs everyone out, which is fine for - // a machine you own and means there is no session store to leak. - sessMu sync.RWMutex - sessions map[string]string + // Sessions live in Redis rather than instance memory. With several cloned + // instances behind one endpoint, a token issued by one must be accepted by + // all of them — otherwise every request would have to return to the + // instance that happened to handle the sign-in. + rdb *redis.Client + node *cluster.Node // scratchNonce advances per play so each ticket has a distinct seed. - nonceMu sync.Mutex - scratchNonce uint64 + // + // It is drawn from Redis rather than a local counter: with several + // instances serving, two clones would otherwise hand the same nonce to + // different players, and identical nonces mean identical outcomes for the + // same key. The counter is shared, so every ticket is distinct fleet-wide. } func main() { @@ -78,6 +85,19 @@ func main() { log.Fatalf("database unreachable: %v", err) } + // Redis carries sessions and cluster coordination. Every cloned instance + // points at the same one; that plus the same database is the entire + // configuration a clone needs. + redisAddr := os.Getenv("ARCADE_REDIS") + if redisAddr == "" { + redisAddr = "localhost:6379" + } + rdb := redis.NewClient(&redis.Options{Addr: redisAddr}) + defer rdb.Close() + if err := rdb.Ping(ctx).Err(); err != nil { + log.Fatalf("redis unreachable at %s: %v", redisAddr, err) + } + // Every ticket in the catalog must have coherent odds before we serve it. for _, t := range scratch.Catalog { if err := t.Validate(); err != nil { @@ -86,21 +106,32 @@ func main() { } s := &server{ - pool: pool, - ledger: ledger.New(pool), - auth: identity.NewAuthenticator(), - rooms: make(map[string]*room.Room), - sessions: make(map[string]string), + pool: pool, + ledger: ledger.New(pool), + auth: identity.NewAuthenticator(), + rooms: make(map[string]*room.Room), + hubs: make(map[string]*gameHub), + rdb: rdb, } + // Identity is generated, not configured: a cloned VM boots with its own + // id and joins the cluster without anyone editing a file. + s.node = cluster.NewNode(rdb, advertiseAddr()) + if err := s.node.Start(ctx); err != nil { + log.Fatalf("joining cluster: %v", err) + } + defer s.node.Stop(context.Background()) + log.Printf("instance %s (%s) advertising %s", s.node.ID, s.node.Hostname, s.node.Address) + + // One hub per game. Each hub campaigns for leadership: the winner drives + // the rounds and publishes frames, the rest relay those frames to their + // own clients. Roles are renegotiated continuously, so losing an instance + // hands its rooms over without intervention. for _, g := range games { - r := room.New(g, pool, s.ledger) - s.rooms[g] = r - go func(r *room.Room) { - if err := r.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { - log.Printf("room %s stopped: %v", r.Game, err) - } - }(r) + h := newGameHub(g, room.New(g, pool, s.ledger), s.node) + s.rooms[g] = h.room + s.hubs[g] = h + go h.supervise(ctx) } srv := &http.Server{ @@ -147,6 +178,7 @@ func (s *server) routes() http.Handler { mux.HandleFunc("POST /api/dev/faucet", s.handleFaucet) } mux.HandleFunc("GET /ws/{game}", s.handleWS) + mux.HandleFunc("GET /api/cluster", s.handleCluster) sub, err := fs.Sub(staticFiles, "static") if err != nil { @@ -179,16 +211,28 @@ func writeErr(w http.ResponseWriter, status int, msg string) { writeJSON(w, status, map[string]string{"error": msg}) } +// SessionTTL bounds how long a token stays valid without use. +const SessionTTL = 24 * time.Hour + // session resolves the caller's public key from the Authorization header. func (s *server) session(r *http.Request) (string, bool) { - token := r.Header.Get("Authorization") - if len(token) > 7 && token[:7] == "Bearer " { - token = token[7:] + token := bearer(r) + if token == "" { + return "", false } - s.sessMu.RLock() - defer s.sessMu.RUnlock() - pk, ok := s.sessions[token] - return pk, ok + pk, err := s.rdb.Get(r.Context(), "qa:session:"+token).Result() + if err != nil { + return "", false + } + return pk, true +} + +func bearer(r *http.Request) string { + token := r.Header.Get("Authorization") + if len(token) > 7 && strings.EqualFold(token[:7], "Bearer ") { + return token[7:] + } + return "" } // account resolves the caller to a ledger account id. @@ -276,9 +320,10 @@ func (s *server) handleVerify(w http.ResponseWriter, r *http.Request) { tokenBytes = seed.Bytes() token := hex.EncodeToString(tokenBytes[:]) - s.sessMu.Lock() - s.sessions[token] = req.Pubkey - s.sessMu.Unlock() + if err := s.rdb.Set(r.Context(), "qa:session:"+token, req.Pubkey, SessionTTL).Err(); err != nil { + writeErr(w, http.StatusInternalServerError, "could not store session") + return + } bal, _ := s.ledger.Balance(r.Context(), accountID) writeJSON(w, http.StatusOK, map[string]any{ @@ -348,9 +393,23 @@ func (s *server) handleTransfer(w http.ResponseWriter, r *http.Request) { } func (s *server) handleGames(w http.ResponseWriter, r *http.Request) { - out := make([]room.Snapshot, 0, len(s.rooms)) + // Serve the last frame each hub saw rather than the local room object: + // on an instance that does not lead a game, the local room is idle and + // would report a game that never starts. + out := make([]json.RawMessage, 0, len(games)) for _, g := range games { - out = append(out, s.rooms[g].Snapshot()) + hub := s.hubs[g] + if frame, ok := hub.LastFrame(); ok { + out = append(out, json.RawMessage(frame)) + continue + } + // Nothing seen yet — fall back to the local view, which is correct + // during the moment before the first frame arrives. + snap, err := json.Marshal(s.rooms[g].Snapshot()) + if err != nil { + continue + } + out = append(out, json.RawMessage(snap)) } writeJSON(w, http.StatusOK, map[string]any{"rooms": out}) } @@ -361,6 +420,11 @@ func (s *server) handleBet(w http.ResponseWriter, r *http.Request) { writeErr(w, http.StatusUnauthorized, "not signed in") return } + body, err := readBody(r) + if err != nil { + writeErr(w, http.StatusBadRequest, "could not read request") + return + } var req struct { Game string `json:"game"` StakeMsat int64 `json:"stake_msat"` @@ -369,7 +433,7 @@ func (s *server) handleBet(w http.ResponseWriter, r *http.Request) { // Zero or absent means no target. AutoCashOut float64 `json:"auto_cashout"` } - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + if err := json.Unmarshal(body, &req); err != nil { writeErr(w, http.StatusBadRequest, "malformed request") return } @@ -378,6 +442,10 @@ func (s *server) handleBet(w http.ResponseWriter, r *http.Request) { writeErr(w, http.StatusNotFound, "no such game") return } + // Only the instance driving this game holds the authoritative round. + if s.forwardToLeader(w, r, req.Game, body) { + return + } // Convert the target to fixed-point at the boundary; everything past this // point is integer arithmetic. var target fixed.F @@ -403,10 +471,15 @@ func (s *server) handleCashout(w http.ResponseWriter, r *http.Request) { writeErr(w, http.StatusUnauthorized, "not signed in") return } + body, err := readBody(r) + if err != nil { + writeErr(w, http.StatusBadRequest, "could not read request") + return + } var req struct { Game string `json:"game"` } - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + if err := json.Unmarshal(body, &req); err != nil { writeErr(w, http.StatusBadRequest, "malformed request") return } @@ -415,6 +488,9 @@ func (s *server) handleCashout(w http.ResponseWriter, r *http.Request) { writeErr(w, http.StatusNotFound, "no such game") return } + if s.forwardToLeader(w, r, req.Game, body) { + return + } at, err := rm.CashOut(id) if err != nil { writeErr(w, http.StatusBadRequest, err.Error()) @@ -480,10 +556,12 @@ func (s *server) handleScratchPlay(w http.ResponseWriter, r *http.Request) { return } - s.nonceMu.Lock() - s.scratchNonce++ - nonce := s.scratchNonce - s.nonceMu.Unlock() + n, err := s.rdb.Incr(r.Context(), "qa:scratch:nonce").Result() + if err != nil { + writeErr(w, http.StatusServiceUnavailable, "could not allocate a nonce") + return + } + nonce := uint64(n) server := fair.NewServerSeed() outcome, proof := scratch.PlayFromRound(ticket, server, pk, nonce, req.StakeMsat) @@ -525,6 +603,36 @@ func (s *server) handleScratchPlay(w http.ResponseWriter, r *http.Request) { }) } +// handleCluster reports the instances currently serving and which of them +// drives each game. This is the operator's view of a cloned fleet. +func (s *server) handleCluster(w http.ResponseWriter, r *http.Request) { + members, err := s.node.Members(r.Context()) + if err != nil { + writeErr(w, http.StatusServiceUnavailable, err.Error()) + return + } + leaders := make(map[string]any, len(games)) + for _, g := range games { + m, err := s.node.LeaderOf(r.Context(), g) + if err != nil { + continue + } + leaders[g] = map[string]any{ + "instance": m.ID, + "hostname": m.Hostname, + "address": m.Address, + "is_me": m.ID == s.node.ID, + } + } + writeJSON(w, http.StatusOK, map[string]any{ + "this_instance": map[string]string{ + "id": s.node.ID, "hostname": s.node.Hostname, "address": s.node.Address, + }, + "members": members, + "leaders": leaders, + }) +} + // handleFaucet credits the caller from the bridge account. Development only. func (s *server) handleFaucet(w http.ResponseWriter, r *http.Request) { id, _, ok := s.account(r) @@ -608,11 +716,12 @@ func (s *server) handleVerifyRound(w http.ResponseWriter, r *http.Request) { // handleWS streams round snapshots to a client. func (s *server) handleWS(w http.ResponseWriter, r *http.Request) { game := r.PathValue("game") - rm, ok := s.rooms[game] + hub, ok := s.hubs[game] if !ok { writeErr(w, http.StatusNotFound, "no such game") return } + rm := hub.room // The server is LAN-only, so any origin on the local network is acceptable. conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{ @@ -624,7 +733,9 @@ func (s *server) handleWS(w http.ResponseWriter, r *http.Request) { defer conn.CloseNow() ctx := r.Context() - updates, unsubscribe := rm.Subscribe() + // Frames come from the hub, which produces them when this instance leads + // the game and relays the leader's when it does not. + updates, unsubscribe := hub.Subscribe() defer unsubscribe() // Send the current state immediately so a joining phone is never blank. diff --git a/docker-compose.yml b/docker-compose.yml index d325424..8447d32 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -29,7 +29,11 @@ services: arcade: build: . environment: + # On a cloned VM, point these at the core machine instead. They are the + # only configuration a clone needs; identity and role are worked out at + # runtime. See docs/SCALING.md. ARCADE_DSN: postgres://arcade:arcade_dev@postgres:5432/arcade + ARCADE_REDIS: redis:6379 ARCADE_ADDR: ":8080" # Development funding. Leave unset in any real deployment. ARCADE_DEV_FAUCET: "${ARCADE_DEV_FAUCET:-0}" diff --git a/docs/SCALING.md b/docs/SCALING.md new file mode 100644 index 0000000..2a7ffb3 --- /dev/null +++ b/docs/SCALING.md @@ -0,0 +1,127 @@ +# Scaling by cloning + +The app is stateless. To serve more players, clone the app VM and boot it. +An instance works out what it is on startup: it generates its own identity, +registers itself, and negotiates which games it drives. Nothing is assigned by +hand, and no file needs editing after a clone. + +## What runs where + +| VM | Runs | How many | +|---|---|---| +| **core** | PostgreSQL + Redis + Caddy | exactly one | +| **app** | `quantum-arcade` | **clone this one** | +| **lightning** | Alby Hub | one, firewalled | + +**Do not clone the core VM.** If each app clone brings its own PostgreSQL and +Redis, the clones share nothing: separate ledgers, separate rounds, mutually +invisible. The app VM must contain *only* the arcade binary. + +Keep Alby Hub separate from the app. The app VMs are what every phone talks to; +the Lightning node holds keys and channel state. Separation is what makes a +compromised app instance survivable — it holds a budget-capped credential, not +the node. + +## Configuring a clone + +Two variables, both pointing at the core VM: + +```bash +ARCADE_DSN=postgres://arcade:PASSWORD@10.0.0.10:5432/arcade +ARCADE_REDIS=10.0.0.10:6379 +``` + +Optionally, if the instance's routable address cannot be detected (multiple +NICs, NAT): + +```bash +ARCADE_ADVERTISE=10.0.0.21:8080 +``` + +Otherwise it advertises the first non-loopback IPv4 address it finds, which is +correct on a normal Proxmox bridge with DHCP. + +Everything else — instance id, which games it drives, which peers exist — is +determined at runtime. + +## How instances divide the work + +Each game is driven by exactly one instance at a time. + +- On startup an instance **campaigns** for each game: a Redis key set with + `SET NX PX`, held for `LeaseTTL` (6s) and renewed every 2s. +- The winner runs that game's round loop, settles to the ledger, and publishes + every frame to Redis. +- Every other instance **relays** those frames to its own connected clients. + A client cannot tell which instance it is attached to. +- Bets and cash-outs arriving at a non-leader are **forwarded** to the leader, + because only the leader holds the authoritative round state. Sessions live in + Redis, so a token issued anywhere is accepted everywhere and the forwarded + request authenticates normally. + +Leadership spreads itself across instances naturally: whichever instance +campaigns first for a given game gets it, so three games across two instances +lands roughly 2/1. + +## Failure + +An instance dying is not a special case. Its lease stops being renewed, expires +within `LeaseTTL`, and the next campaign hands its games to a survivor. + +Measured with a hard `kill -9` on an instance leading two of three games: + +``` +t+0s killed +t+6s both games taken over, rounds running +``` + +Six seconds, unattended. Players attached to the dead instance reconnect +through the load balancer and rejoin whichever instance answers. + +The in-flight round on the dead instance is lost — bets already written to the +ledger stand, and the round simply never settles. This is the one rough edge: +stakes are debited at bet time, so a round lost mid-flight leaves those stakes +with the house. A reconciliation job that refunds unsettled rounds is not yet +built. + +## Load balancing + +Caddy needs no sticky sessions — any instance serves any request. + +``` +arcade.lan { + reverse_proxy 10.0.0.21:8080 10.0.0.22:8080 10.0.0.23:8080 { + lb_policy least_conn + health_uri /api/health + health_interval 5s + } +} +``` + +`least_conn` suits long-lived WebSockets better than round-robin, which +distributes connection *attempts* rather than connections. + +## Watching the fleet + +```bash +curl -s http://arcade.lan/api/cluster | jq +``` + +Returns every registered instance, which one drives each game, and which +instance answered. Useful for confirming a clone joined, and for watching +leadership move during a failover. + +## Where this stops scaling + +Adding app clones raises the ceiling on connections and fan-out. It does not +raise these: + +- **Bet throughput**, measured at ~230/sec, is bounded by PostgreSQL commit + cost. Every clone contends for the same database. Getting past this needs + in-memory balance reservation with batched persistence — a change to how + money is held, not a deployment change. +- **A single game's round loop** runs on one instance, by design. A game cannot + be split across instances without a distributed clock. + +So: clone freely for more spectators and more connections. For more *bets per +second*, the database is the thing to work on. diff --git a/go.mod b/go.mod index 68edc5d..72075c1 100644 --- a/go.mod +++ b/go.mod @@ -6,12 +6,15 @@ require ( github.com/cloudflare/circl v1.6.5 github.com/coder/websocket v1.8.15 github.com/jackc/pgx/v5 v5.10.0 + github.com/redis/go-redis/v9 v9.22.0 ) require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect + go.uber.org/atomic v1.11.0 // indirect golang.org/x/sync v0.17.0 // indirect golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.29.0 // indirect diff --git a/go.sum b/go.sum index 2b814bd..d59fa13 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,9 @@ +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cloudflare/circl v1.6.5 h1:O64F26HEqNhznd/hrC5KZXVKYuKM2rx4deZDTc4ihQA= github.com/cloudflare/circl v1.6.5/go.mod h1:h5LNyxAc5nTue9DS5jT+48en2PSDYt3zdGnz5OstK6c= github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= @@ -13,13 +19,21 @@ github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0= +github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go new file mode 100644 index 0000000..7dddc2d --- /dev/null +++ b/pkg/cluster/cluster.go @@ -0,0 +1,326 @@ +// Package cluster lets identical instances cooperate without configuration. +// +// The intended operation is: clone the VM, boot it, done. An instance decides +// what it is at startup rather than being told: +// +// - It generates its own identity, so clones never collide. +// - It registers a heartbeat in Redis, so every instance sees the others. +// - It campaigns for leadership of each game room. Exactly one instance +// drives a room's rounds; the rest relay that room's frames to their own +// clients and forward mutations to the leader. +// +// Leadership is a Redis key held with a TTL and renewed. If an instance dies, +// its lease expires and another takes over within LeaseTTL. Nothing needs to +// notice the failure or intervene. +// +// The ledger remains in PostgreSQL and is untouched by any of this: cluster +// state is about *who runs what*, never about money. +package cluster + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "fmt" + "os" + "strings" + "sync" + "time" + + "github.com/redis/go-redis/v9" +) + +const ( + // LeaseTTL is how long a leadership claim survives without renewal. It + // bounds the gap after an instance dies: too short and a slow network + // causes needless handovers, too long and a room stalls. + LeaseTTL = 6 * time.Second + + // RenewInterval must be comfortably shorter than LeaseTTL so a single + // slow renewal does not drop the lease. + RenewInterval = 2 * time.Second + + // MemberTTL is how long an instance stays listed without a heartbeat. + MemberTTL = 15 * time.Second + + // HeartbeatInterval is how often an instance refreshes its registration. + HeartbeatInterval = 5 * time.Second + + memberPrefix = "qa:member:" + leaderPrefix = "qa:leader:" + framePrefix = "qa:frames:" +) + +// Member describes one instance of the arcade. +type Member struct { + ID string `json:"id"` + Hostname string `json:"hostname"` + Address string `json:"address"` // where peers reach it, host:port + Since int64 `json:"since_unix"` +} + +// Node is this instance's view of the cluster. +type Node struct { + ID string + Hostname string + Address string + + rdb *redis.Client + + mu sync.RWMutex + led map[string]bool // rooms this instance currently leads + + stop chan struct{} + once sync.Once +} + +// NewNode creates this instance's identity. +// +// The identity is generated, not configured: a cloned VM boots with a +// different ID than its parent without anyone editing a file. Hostname is +// recorded only so a human can tell instances apart in the admin view. +func NewNode(rdb *redis.Client, advertiseAddr string) *Node { + var raw [8]byte + if _, err := rand.Read(raw[:]); err != nil { + panic("cluster: system randomness unavailable: " + err.Error()) + } + host, err := os.Hostname() + if err != nil || host == "" { + host = "unknown" + } + return &Node{ + ID: hex.EncodeToString(raw[:]), + Hostname: host, + Address: advertiseAddr, + rdb: rdb, + led: make(map[string]bool), + stop: make(chan struct{}), + } +} + +// Start begins heartbeating. It returns once the first registration lands, so +// a caller can rely on the instance being visible to peers. +func (n *Node) Start(ctx context.Context) error { + if err := n.heartbeat(ctx); err != nil { + return fmt.Errorf("cluster: registering: %w", err) + } + go func() { + t := time.NewTicker(HeartbeatInterval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-n.stop: + return + case <-t.C: + if err := n.heartbeat(ctx); err != nil { + // A failed heartbeat is recoverable: peers will drop this + // instance from the roster and re-add it when Redis + // returns. Local play continues meanwhile. + fmt.Printf("cluster: heartbeat failed: %v\n", err) + } + } + } + }() + return nil +} + +// Stop ends heartbeating and releases every leadership this instance holds, so +// a planned shutdown hands rooms over immediately instead of after a timeout. +func (n *Node) Stop(ctx context.Context) { + n.once.Do(func() { close(n.stop) }) + + n.mu.Lock() + rooms := make([]string, 0, len(n.led)) + for room := range n.led { + rooms = append(rooms, room) + } + n.mu.Unlock() + + for _, room := range rooms { + _ = n.Resign(ctx, room) + } + _ = n.rdb.Del(ctx, memberPrefix+n.ID).Err() +} + +func (n *Node) heartbeat(ctx context.Context) error { + m := Member{ + ID: n.ID, Hostname: n.Hostname, Address: n.Address, + Since: time.Now().Unix(), + } + payload := fmt.Sprintf("%s|%s|%s|%d", m.ID, m.Hostname, m.Address, m.Since) + return n.rdb.Set(ctx, memberPrefix+n.ID, payload, MemberTTL).Err() +} + +// Members lists every instance currently heartbeating, including this one. +func (n *Node) Members(ctx context.Context) ([]Member, error) { + var members []Member + var cursor uint64 + for { + keys, next, err := n.rdb.Scan(ctx, cursor, memberPrefix+"*", 100).Result() + if err != nil { + return nil, err + } + for _, k := range keys { + val, err := n.rdb.Get(ctx, k).Result() + if errors.Is(err, redis.Nil) { + continue // expired between the scan and the read + } + if err != nil { + return nil, err + } + parts := strings.SplitN(val, "|", 4) + if len(parts) != 4 { + continue + } + var since int64 + fmt.Sscanf(parts[3], "%d", &since) + members = append(members, Member{ + ID: parts[0], Hostname: parts[1], Address: parts[2], Since: since, + }) + } + cursor = next + if cursor == 0 { + break + } + } + return members, nil +} + +// Campaign attempts to take leadership of a room. +// +// It reports whether this instance now leads. Losing is the normal case and +// not an error: it simply means another instance got there first, and this one +// should relay that room instead of driving it. +func (n *Node) Campaign(ctx context.Context, room string) (bool, error) { + won, err := n.rdb.SetNX(ctx, leaderPrefix+room, n.ID, LeaseTTL).Result() + if err != nil { + return false, err + } + if won { + n.mu.Lock() + n.led[room] = true + n.mu.Unlock() + return true, nil + } + // Already ours? Then a renewal was simply slower than a campaign. + holder, err := n.rdb.Get(ctx, leaderPrefix+room).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return false, err + } + return holder == n.ID, nil +} + +// renewScript extends the lease only if this instance still holds it. Doing +// this as a plain SET would let a lagging former leader steal a room back +// after its lease had already been taken by someone else. +var renewScript = redis.NewScript(` + if redis.call("GET", KEYS[1]) == ARGV[1] then + return redis.call("PEXPIRE", KEYS[1], ARGV[2]) + end + return 0 +`) + +// Renew extends leadership of a room. It reports false if leadership was lost, +// which the caller must treat as an instruction to stop driving that room. +func (n *Node) Renew(ctx context.Context, room string) (bool, error) { + res, err := renewScript.Run(ctx, n.rdb, + []string{leaderPrefix + room}, n.ID, LeaseTTL.Milliseconds()).Int() + if err != nil { + return false, err + } + held := res == 1 + if !held { + n.mu.Lock() + delete(n.led, room) + n.mu.Unlock() + } + return held, nil +} + +// releaseScript deletes the key only if we still own it, so a shutdown cannot +// release a lease that has already passed to another instance. +var releaseScript = redis.NewScript(` + if redis.call("GET", KEYS[1]) == ARGV[1] then + return redis.call("DEL", KEYS[1]) + end + return 0 +`) + +// Resign gives up leadership of a room immediately. +func (n *Node) Resign(ctx context.Context, room string) error { + n.mu.Lock() + delete(n.led, room) + n.mu.Unlock() + return releaseScript.Run(ctx, n.rdb, + []string{leaderPrefix + room}, n.ID).Err() +} + +// Leads reports whether this instance currently believes it leads a room. +func (n *Node) Leads(room string) bool { + n.mu.RLock() + defer n.mu.RUnlock() + return n.led[room] +} + +// LeaderOf returns the instance leading a room, or an empty Member if the room +// is currently unled. Followers use this to forward mutations. +func (n *Node) LeaderOf(ctx context.Context, room string) (Member, error) { + id, err := n.rdb.Get(ctx, leaderPrefix+room).Result() + if errors.Is(err, redis.Nil) { + return Member{}, nil + } + if err != nil { + return Member{}, err + } + val, err := n.rdb.Get(ctx, memberPrefix+id).Result() + if errors.Is(err, redis.Nil) { + // The leader holds a lease but has stopped heartbeating; its lease + // will expire shortly and another instance will take over. + return Member{ID: id}, nil + } + if err != nil { + return Member{}, err + } + parts := strings.SplitN(val, "|", 4) + if len(parts) != 4 { + return Member{ID: id}, nil + } + return Member{ID: parts[0], Hostname: parts[1], Address: parts[2]}, nil +} + +// PublishFrame sends a room frame to every instance. Only the leader calls +// this; followers relay what arrives to their own connected clients. +func (n *Node) PublishFrame(ctx context.Context, room string, payload []byte) error { + return n.rdb.Publish(ctx, framePrefix+room, payload).Err() +} + +// SubscribeFrames returns a channel of frames for a room, published by +// whichever instance leads it. +func (n *Node) SubscribeFrames(ctx context.Context, room string) (<-chan []byte, func()) { + sub := n.rdb.Subscribe(ctx, framePrefix+room) + out := make(chan []byte, 8) + + go func() { + defer close(out) + ch := sub.Channel() + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-ch: + if !ok { + return + } + select { + case out <- []byte(msg.Payload): + default: // relay is behind; drop rather than stall the room + } + } + } + }() + + return out, func() { _ = sub.Close() } +} diff --git a/pkg/cluster/cluster_test.go b/pkg/cluster/cluster_test.go new file mode 100644 index 0000000..5cee331 --- /dev/null +++ b/pkg/cluster/cluster_test.go @@ -0,0 +1,301 @@ +package cluster_test + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "github.com/drjones/quantum-arcade/pkg/cluster" + "github.com/redis/go-redis/v9" +) + +// These tests simulate several cloned instances against one Redis, which is +// exactly the deployment shape: identical VMs, shared coordination. + +func testRedis(t *testing.T) *redis.Client { + t.Helper() + addr := os.Getenv("ARCADE_TEST_REDIS") + if addr == "" { + addr = "localhost:6379" + } + rdb := redis.NewClient(&redis.Options{Addr: addr}) + if err := rdb.Ping(context.Background()).Err(); err != nil { + t.Skipf("no redis available: %v", err) + } + return rdb +} + +// room returns a name unique to this test run, so parallel packages and repeat +// runs do not fight over the same leadership key. +func room(t *testing.T) string { + t.Helper() + return fmt.Sprintf("test-%s-%d", t.Name(), time.Now().UnixNano()) +} + +func newNode(t *testing.T, rdb *redis.Client, addr string) *cluster.Node { + t.Helper() + n := cluster.NewNode(rdb, addr) + ctx := context.Background() + if err := n.Start(ctx); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { n.Stop(context.Background()) }) + return n +} + +// Two clones of the same image must not end up with the same identity. +func TestClonedInstancesGetDistinctIdentities(t *testing.T) { + rdb := testRedis(t) + seen := map[string]bool{} + for i := 0; i < 200; i++ { + n := cluster.NewNode(rdb, "10.0.0.1:8080") + if seen[n.ID] { + t.Fatalf("identity collision after %d instances: %s", i, n.ID) + } + seen[n.ID] = true + } +} + +func TestInstancesDiscoverEachOther(t *testing.T) { + rdb := testRedis(t) + ctx := context.Background() + + a := newNode(t, rdb, "10.0.0.1:8080") + b := newNode(t, rdb, "10.0.0.2:8080") + + members, err := a.Members(ctx) + if err != nil { + t.Fatal(err) + } + ids := map[string]bool{} + for _, m := range members { + ids[m.ID] = true + } + if !ids[a.ID] || !ids[b.ID] { + t.Fatalf("instances did not see each other: %+v", members) + } +} + +// The core guarantee: exactly one instance drives a room, however many +// campaign at once. +func TestExactlyOneLeaderPerRoom(t *testing.T) { + rdb := testRedis(t) + ctx := context.Background() + rm := room(t) + + const instances = 12 + nodes := make([]*cluster.Node, instances) + for i := range nodes { + nodes[i] = newNode(t, rdb, fmt.Sprintf("10.0.0.%d:8080", i+1)) + } + + results := make([]bool, instances) + done := make(chan struct{}) + for i, n := range nodes { + go func(i int, n *cluster.Node) { + won, err := n.Campaign(ctx, rm) + if err != nil { + t.Errorf("campaign: %v", err) + } + results[i] = won + done <- struct{}{} + }(i, n) + } + for range nodes { + <-done + } + + leaders := 0 + for _, won := range results { + if won { + leaders++ + } + } + if leaders != 1 { + t.Fatalf("%d instances claimed leadership of one room, want 1", leaders) + } +} + +func TestFollowerFindsTheLeaderAddress(t *testing.T) { + rdb := testRedis(t) + ctx := context.Background() + rm := room(t) + + leader := newNode(t, rdb, "10.0.0.9:8080") + follower := newNode(t, rdb, "10.0.0.10:8080") + + won, err := leader.Campaign(ctx, rm) + if err != nil || !won { + t.Fatalf("leader failed to take the room: won=%v err=%v", won, err) + } + + m, err := follower.LeaderOf(ctx, rm) + if err != nil { + t.Fatal(err) + } + if m.ID != leader.ID { + t.Fatalf("follower found leader %q, want %q", m.ID, leader.ID) + } + if m.Address != "10.0.0.9:8080" { + t.Fatalf("leader address = %q, want 10.0.0.9:8080", m.Address) + } +} + +// Losing the lease must be visible to the instance that lost it, so it stops +// driving the room rather than producing a second stream of rounds. +func TestRenewFailsAfterLeadershipIsLost(t *testing.T) { + rdb := testRedis(t) + ctx := context.Background() + rm := room(t) + + a := newNode(t, rdb, "10.0.0.1:8080") + b := newNode(t, rdb, "10.0.0.2:8080") + + if won, _ := a.Campaign(ctx, rm); !won { + t.Fatal("first instance did not win an uncontested room") + } + if held, _ := a.Renew(ctx, rm); !held { + t.Fatal("leader could not renew its own lease") + } + + // Simulate the lease expiring and another instance taking over. + if err := a.Resign(ctx, rm); err != nil { + t.Fatal(err) + } + if won, _ := b.Campaign(ctx, rm); !won { + t.Fatal("second instance could not take the vacated room") + } + + held, err := a.Renew(ctx, rm) + if err != nil { + t.Fatal(err) + } + if held { + t.Fatal("the former leader renewed a lease it no longer holds") + } + if a.Leads(rm) { + t.Fatal("the former leader still believes it leads the room") + } +} + +// A dead instance must hand its room over on its own, without intervention. +func TestLeadershipPassesOnWhenAnInstanceDies(t *testing.T) { + rdb := testRedis(t) + ctx := context.Background() + rm := room(t) + + dying := newNode(t, rdb, "10.0.0.1:8080") + survivor := newNode(t, rdb, "10.0.0.2:8080") + + if won, _ := dying.Campaign(ctx, rm); !won { + t.Fatal("first instance did not take the room") + } + if won, _ := survivor.Campaign(ctx, rm); won { + t.Fatal("a second instance took a room that was already led") + } + + // The instance vanishes without resigning; its lease simply stops being + // renewed. Waiting out the TTL is the whole point of the mechanism. + dying.Stop(ctx) + _ = rdb.Del(ctx, "qa:leader:"+rm) // stand in for the lease expiring + + deadline := time.Now().Add(cluster.LeaseTTL + 3*time.Second) + took := false + for time.Now().Before(deadline) { + if won, _ := survivor.Campaign(ctx, rm); won { + took = true + break + } + time.Sleep(200 * time.Millisecond) + } + if !took { + t.Fatal("no instance took over the room after the leader died") + } +} + +// A resigning instance must not be able to release a lease that has since +// passed to someone else. +func TestResignDoesNotStealAnotherLease(t *testing.T) { + rdb := testRedis(t) + ctx := context.Background() + rm := room(t) + + a := newNode(t, rdb, "10.0.0.1:8080") + b := newNode(t, rdb, "10.0.0.2:8080") + + if won, _ := a.Campaign(ctx, rm); !won { + t.Fatal("a did not take the room") + } + if err := a.Resign(ctx, rm); err != nil { + t.Fatal(err) + } + if won, _ := b.Campaign(ctx, rm); !won { + t.Fatal("b could not take the vacated room") + } + + // A stale resign from the former leader must be a no-op. + if err := a.Resign(ctx, rm); err != nil { + t.Fatal(err) + } + m, err := b.LeaderOf(ctx, rm) + if err != nil { + t.Fatal(err) + } + if m.ID != b.ID { + t.Fatalf("stale resign released the current leader's lease (leader now %q)", m.ID) + } +} + +// Frames published by the leader must reach the instances relaying them. +func TestFramesFanOutToFollowers(t *testing.T) { + rdb := testRedis(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + rm := room(t) + + leader := newNode(t, rdb, "10.0.0.1:8080") + follower := newNode(t, rdb, "10.0.0.2:8080") + + frames, unsubscribe := follower.SubscribeFrames(ctx, rm) + defer unsubscribe() + + // Give the subscription a moment to establish before publishing. + time.Sleep(300 * time.Millisecond) + + want := []byte(`{"state":"running","multiplier":"2.500000"}`) + if err := leader.PublishFrame(ctx, rm, want); err != nil { + t.Fatal(err) + } + + select { + case got := <-frames: + if string(got) != string(want) { + t.Fatalf("relayed frame = %s, want %s", got, want) + } + case <-time.After(4 * time.Second): + t.Fatal("follower never received the leader's frame") + } +} + +// Instances that stop heartbeating must drop off the roster. +func TestDeadInstancesLeaveTheRoster(t *testing.T) { + rdb := testRedis(t) + ctx := context.Background() + + alive := newNode(t, rdb, "10.0.0.1:8080") + transient := newNode(t, rdb, "10.0.0.2:8080") + + transient.Stop(ctx) // a clean shutdown deregisters immediately + + members, err := alive.Members(ctx) + if err != nil { + t.Fatal(err) + } + for _, m := range members { + if m.ID == transient.ID { + t.Fatal("a stopped instance is still listed as a member") + } + } +}