memby/server/internal/runtimestats/workers.go

// Package runtimestats answers the one question the console's Process card exists for: is
// the gateway healthy, and if something is abnormal, which area should be looked at?
//
// A bare goroutine count cannot answer it. Twenty-five means nothing on its own — an
// operator cannot tell from it what those goroutines are doing, which part of Memby they
// belong to, whether the number is normal, or whether it has been climbing all week. This
// package supplies the three things that make the number readable, and keeps them apart by
// what they cost:
//
//   - The registry here, which is free. A long-running Memby worker says its own name when
//     it starts, so "Library ingest" and "Emby health probe" are named rather than inferred
//     from a stack.
//   - The sampler in sample.go, which is a handful of counters on a slow tick. A single
//     instantaneous figure cannot show a leak; a trend can.
//   - The stack breakdown in goroutines.go, which is genuinely expensive and is therefore
//     collected only when an operator asks for it.
//
// Nothing here recovers a panic. A worker that dies must die exactly as it always did —
// this package reports, and reporting must never change what it is reporting on.
package runtimestats

import (
	"sort"
	"sync"
	"time"
)

// WorkerState is what became of a tracked worker. A worker that ends is not a fault —
// several of the gateway's background jobs are one-shot startup work — but "still running"
// and "finished" are different answers and the console must not merge them.
type WorkerState string

const (
	WorkerRunning  WorkerState = "running"
	WorkerFinished WorkerState = "finished"
)

// Worker is one named background goroutine.
type Worker struct {
	Name      string      `json:"name"`
	Component string      `json:"component"`
	State     WorkerState `json:"state"`
	Started   time.Time   `json:"started"`
	Stopped   time.Time   `json:"stopped,omitempty"`
	// Starts counts how many times this name has been launched. It is on the record
	// because a worker that is being restarted in a loop reads exactly like a healthy one
	// from a single snapshot, and does not from this number.
	Starts int `json:"starts"`
}

type workerEntry struct {
	component string
	running   int
	starts    int
	started   time.Time
	stopped   time.Time
}

var (
	workersMu sync.Mutex
	workers   = map[string]*workerEntry{}
)

// Go starts fn on its own goroutine and records it under a name an operator can read. The
// name is the worker's identity across restarts, so it must be stable and must not carry a
// count or an address in it.
func Go(name, component string, fn func()) {
	begin(name, component)
	go func() {
		defer end(name)
		fn()
	}()
}

// Track records a worker whose goroutine something else owns. The returned function marks
// it finished, and is safe to call more than once.
func Track(name, component string) (done func()) {
	begin(name, component)
	var once sync.Once
	return func() { once.Do(func() { end(name) }) }
}

func begin(name, component string) {
	now := time.Now()
	workersMu.Lock()
	defer workersMu.Unlock()
	entry := workers[name]
	if entry == nil {
		entry = &workerEntry{}
		workers[name] = entry
	}
	entry.component = component
	entry.running++
	entry.starts++
	entry.started = now
	entry.stopped = time.Time{}
}

func end(name string) {
	now := time.Now()
	workersMu.Lock()
	defer workersMu.Unlock()
	entry := workers[name]
	if entry == nil {
		return
	}
	if entry.running > 0 {
		entry.running--
	}
	if entry.running == 0 {
		entry.stopped = now
	}
}

// Workers lists what has been registered, running first and then by name, so the table
// does not reorder itself under the operator on every poll.
func Workers() []Worker {
	workersMu.Lock()
	list := make([]Worker, 0, len(workers))
	for name, entry := range workers {
		worker := Worker{
			Name:      name,
			Component: entry.component,
			State:     WorkerFinished,
			Started:   entry.started,
			Stopped:   entry.stopped,
			Starts:    entry.starts,
		}
		if entry.running > 0 {
			worker.State = WorkerRunning
			worker.Stopped = time.Time{}
		}
		list = append(list, worker)
	}
	workersMu.Unlock()
	sort.Slice(list, func(a, b int) bool {
		if (list[a].State == WorkerRunning) != (list[b].State == WorkerRunning) {
			return list[a].State == WorkerRunning
		}
		return list[a].Name < list[b].Name
	})
	return list
}

// resetWorkers exists for the tests; the registry is process-wide by design.
func resetWorkers() {
	workersMu.Lock()
	workers = map[string]*workerEntry{}
	workersMu.Unlock()
}