casino/pkg/cluster/cluster.go

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