path: root/internal/game/owner.go

package game

import (
	"log"
	"runtime"
	"runtime/debug"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
)

// owner is the single execution authority for live game state. Every
// mutation of players, sessions, the Hub membership, runtime maps, and all
// tick callbacks runs on exactly one goroutine (the loop below). Network and
// admin goroutines submit tasks; they never touch live objects directly.
//
// Rules:
//   - Game.run blocks the caller until the closure has executed on the
//     owner goroutine. Connection goroutines use it per input line, which
//     gives natural per-client backpressure without ever blocking another
//     client: the owner stays free because network I/O happens outside
//     (Session writes go through outbound queues, saves serialize copies).
//   - Code already executing on the owner must NOT call Game.run (it would
//     deadlock waiting for itself). Owner-executed code calls functions
//     directly.
//   - Task ordering is FIFO: submitted input lines, disconnect callbacks,
//     and admin mutations execute in submission order. Tick callbacks
//     serialize against these tasks the same way (the engine's ticker
//     goroutine submits each callback), preserving the documented tick
//     order relative to subscriber registration IDs.
type owner struct {
	mu     sync.Mutex
	cond   *sync.Cond
	closed bool
	tasks  chan func()
	done   chan struct{}
	// goid is the goroutine ID of the owner loop, recorded once the loop
	// starts. Re-entrancy checks (Game.executingOnOwner) compare it to the
	// CALLER's goroutine ID so only code actually running on the owner
	// detects itself — an "owner is busy" flag would wrongly let outside
	// callers bypass the queue while a task is in flight.
	goid atomic.Uint64
	// panics counts the task panics the loop has contained (see runTask).
	// A nonzero value means a command, tick callback, or disconnect handler
	// panicked and was logged instead of killing the process.
	panics atomic.Uint64
}

const ownerTaskQueue = 4096

func newOwner() *owner {
	o := &owner{
		tasks: make(chan func(), ownerTaskQueue),
		done:  make(chan struct{}),
	}
	o.cond = sync.NewCond(&o.mu)
	go o.loop()
	return o
}

func (o *owner) loop() {
	o.goid.Store(currentGoroutineID())
	defer close(o.done)
	for task := range o.tasks {
		o.runTask(task)
		// A queue slot freed up; wake one blocked submitter waiting for
		// room (see enqueue). The signal MUST be sent while holding o.mu:
		// a submitter can otherwise observe a full queue, watch this loop
		// free a slot and signal, and only then reach cond.Wait() — the
		// signal is lost and the submitter sleeps with capacity available.
		o.mu.Lock()
		o.cond.Signal()
		o.mu.Unlock()
	}
}

// runTask executes one queued task behind a panic barrier. A panic in a
// command, tick callback, or disconnect handler must not kill the owner
// goroutine: that goroutine is the only execution authority for live game
// state, so its death takes down every session and all unsaved state with
// it. The panic is logged with a stack trace and counted (see panics) so a
// storm is visible, and the loop keeps draining the queue.
func (o *owner) runTask(task func()) {
	defer func() {
		if r := recover(); r != nil {
			o.panics.Add(1)
			log.Printf("owner: recovered panic in task: %v\n%s", r, debug.Stack())
		}
	}()
	task()
}

// submit enqueues without blocking. It returns false when the owner has
// stopped or the queue is full; callers that need guaranteed delivery use
// enqueue instead.
func (o *owner) submit(fn func()) bool {
	o.mu.Lock()
	defer o.mu.Unlock()
	if o.closed {
		return false
	}
	select {
	case o.tasks <- fn:
		return true
	default:
		return false
	}
}

// enqueue submits with backpressure: it blocks while the queue is full and
// returns false only when the owner has stopped. A full queue must NEVER
// bypass the owner: running a task on the caller's goroutine while the
// owner keeps executing other tasks breaks the single-owner invariant that
// serializes all live state mutation.
func (o *owner) enqueue(fn func()) bool {
	o.mu.Lock()
	defer o.mu.Unlock()
	for !o.closed {
		select {
		case o.tasks <- fn:
			return true
		default:
			o.cond.Wait()
		}
	}
	return false
}

func (o *owner) stop() {
	o.mu.Lock()
	if !o.closed {
		o.closed = true
		close(o.tasks)
		o.cond.Broadcast()
	}
	o.mu.Unlock()
	<-o.done
}

// run executes fn on the owner goroutine and waits for it. When no owner
// exists (synthetic Games in unit tests) fn runs inline on the caller's
// goroutine; production Games always have one, started by New.
func (g *Game) run(fn func()) {
	// Already running on the owner goroutine itself (e.g. a tick callback
	// re-entering through a Hub removal callback): run inline, exactly like
	// owner code calling functions directly. Re-submitting would deadlock
	// waiting for the running task to finish — itself. Only true identity
	// passes: an outside caller while the owner is merely BUSY still
	// queues (and blocks), preserving the single-owner invariant.
	if g.executingOnOwner() {
		fn()
		return
	}
	o := g.owner
	if o == nil {
		fn()
		return
	}
	task := make(chan struct{})
	if !o.enqueue(func() {
		// Deferred so a panic contained by runTask still releases the
		// caller blocked on <-task below instead of hanging it forever.
		defer close(task)
		fn()
	}) {
		// The owner loop has stopped (graceful shutdown in progress).
		// Admissions are closed: DROP the task. Executing it on the
		// caller's goroutine would mutate live state concurrently with the
		// shutdown sequence and after the final-save barrier (network
		// transports are still draining until transportStop runs). See
		// TestShutdownRejectsSessionInput.
		return
	}
	select {
	case <-task:
	case <-o.done:
		// The owner goroutine exited without draining the queue (a task
		// called runtime.Goexit, e.g. a test's t.Fatal). Return instead of
		// blocking forever on a task that no longer has a consumer.
	}
}

// adminTaskPanic is the value RunAdminOnOwner re-raises on its caller's
// goroutine when the owner loop contained a panic in an editor task. It is a
// distinct value so a caller can tell a contained task panic from the
// shutdown refusal that a false result means.
const adminTaskPanic = "owner: admin task panicked; the owner loop contained it, see the log for the stack trace"

// RunAdminOnOwner synchronously executes one complete editor request. Unlike
// RunOnOwner it rejects work during shutdown and NEVER falls back to executing
// on the HTTP goroutine when the owner has stopped. Call only from outside the
// owner; runtime callbacks within fn must call owner-only methods directly.
//
// A panic inside fn is contained by the owner loop's panic barrier (runTask
// logs and counts it and the loop keeps draining), but it is NOT reported as a
// false result: false means "the owner refused the work", which the admin
// middleware renders as 503 "game is shutting down" - a wrong diagnosis for a
// task that panicked, and one that hides a real internal error. Instead the
// panic is re-raised on the CALLER's goroutine, so the admin HTTP handler's
// own panic recovery (admin.ownerMiddleware preserves net/http's) answers an
// internal error while the owner loop stays untouched.
func (g *Game) RunAdminOnOwner(fn func()) bool {
	if g.isShuttingDown() {
		return false
	}
	accepted := false
	task := func() {
		if !g.isShuttingDown() {
			fn()
			accepted = true
		}
	}
	if g.owner == nil {
		// Synthetic Games only; production always has an owner.
		task()
		return accepted
	}
	done := make(chan struct{})
	// finished is written on the owner goroutine before close(done), so the
	// caller reads it race-free after <-done. A deferred close(done) releases
	// the caller whether the task completed or panicked.
	finished := false
	if !g.owner.enqueue(func() {
		defer close(done)
		task()
		finished = true
	}) {
		return false
	}
	select {
	case <-done:
	case <-g.owner.done:
		// Owner goroutine exited; treat as a refused task rather than hanging.
		return false
	}
	if !finished {
		panic(adminTaskPanic)
	}
	return accepted
}

// currentGoroutineID returns the calling goroutine's ID parsed from its
// stack header ("goroutine 123 [running]:"), 0 when unparseable.
func currentGoroutineID() uint64 {
	var buf [64]byte
	n := runtime.Stack(buf[:], false)
	s := strings.TrimPrefix(string(buf[:n]), "goroutine ")
	end := strings.IndexByte(s, ' ')
	if end <= 0 {
		return 0
	}
	id, err := strconv.ParseUint(s[:end], 10, 64)
	if err != nil {
		return 0
	}
	return id
}

// executingOnOwner reports whether the CALLING goroutine is the owner loop
// goroutine. Owner-executed code must invoke callbacks directly instead of
// re-submitting through run(), which would deadlock waiting for itself.
func (g *Game) executingOnOwner() bool {
	o := g.owner
	if o == nil {
		return false
	}
	id := o.goid.Load()
	return id != 0 && id == currentGoroutineID()
}

// stopOwner drains and stops the owner loop. Called once during graceful
// shutdown after all entry points have stopped submitting.
func (g *Game) stopOwner() {
	g.ownerStop.Do(func() {
		if g.owner != nil {
			g.owner.stop()
		}
	})
}