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 }