// 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() } }