fix(scheduler): one runner for the install, not one per engine

CHANGELOG.md

@@ -12,6 +12,19 @@ its first tagged release.
 ## [0.7.0] — unreleased
 
 ### Fixed
+- **A scheduled broadcast could silently fail to go on air on a multi-programme
+  install.** `schedules` carries no `source_id` — a timetable is a property of
+  the box — but every engine ran its own `scheduler.Runner` over that one table.
+  Whichever swept first wrote `enabled` on every destination, including other
+  programmes', then reconciled ONLY ITS OWN engine and marked the occurrence
+  handled; the other engines read it as handled and never reconciled. Those
+  destinations sat enabled in the database with no process publishing, while the
+  log said `schedule fired`. `MarkScheduleRun`'s `WHERE last_run_at < ?` is a
+  ratchet on the row, not a lease over the work, and the actuator has no way to
+  learn whether it won it. There is now one runner, owned by the manager, whose
+  reconcile covers every engine — so the shape that caused this is no longer
+  representable rather than merely unlikely. The runs page also stops reporting
+  the default programme's scheduler as though it were the only one.
 - **The dashboard's grouped destination list and the Prometheus scrape lost
   every programme but one.** Scoping `Engine.Status` to its own source was
   right, and it removed a leak three callers were quietly relying on: the

internal/api/automation.go

@@ -433,7 +433,10 @@ func (s *Server) handleDeleteSchedule(w http.ResponseWriter, r *http.Request) {
 // A schedule that skipped because the server was down is the single most
 // confusing thing this feature can do, so it has to be visible.
 func (s *Server) handleScheduleRuns(w http.ResponseWriter, r *http.Request) {
-	last := s.eng().Scheduler().Last()
+	// The INSTALL's scheduler, not the default engine's. There is one timetable
+	// (schedules has no source_id); reading it off s.eng() reported programme
+	// 1's runs on a multi-source install. See #526.
+	last := s.mgr.Scheduler().Last()
 	if last == nil {
 		last = []scheduler.Result{}
 	}

internal/api/source_scope_routes_test.go

@@ -302,8 +302,6 @@ var defaultEngineSites = map[string]string{
 	"handleTestAlertRule": "internal/api/automation.go. Unscoped; not this " +
 		"change's file assignment. Already recorded in noSourceRefusalSites for " +
 		"the zero-source half of the same reach.",
-	"handleScheduleRuns": "internal/api/automation.go. Unscoped; not this " +
-		"change's file assignment.",
 	"handleCaptureClip": "internal/api/automation.go. Unscoped, and it WRITES: " +
 		"a clip captured from the default programme's rolling buffer. Not this " +
 		"change's file assignment.",

internal/engine/engine.go

@@ -344,7 +344,6 @@ type Engine struct {
 	alertWatch *alerts.Watcher
 	// sched flips destinations' enabled flags on a timetable, through the same
 	// path a human uses.
-	sched *scheduler.Runner
 
 	// playProcs mirrors the manager's running variants so the monitoring page
 	// can list them beside every other child. The manager hands out its
@@ -680,7 +679,6 @@ func New(log *slog.Logger, cfg config.Config, store *db.DB, tools *ffmpeg.Tools,
 	// Named from the source row on the first sweep; SourceRef starts with the
 	// id alone so an event raised before that still identifies its programme.
 	e.hookWatch = hooks.NewWatcher(hooks.SourceRef{ID: sourceID}, hooks.WatchConfig{})
-	e.sched = scheduler.New(log, store, scheduleActuator{e}, scheduler.WithOnResult(e.onSchedule))
 	return e, nil
 }
 
@@ -855,10 +853,6 @@ func (e *Engine) Start(ctx context.Context) error {
 		go func() { defer e.wg.Done(); e.alerter.Run(e.ctx) }()
 		go func() { defer e.wg.Done(); e.observeLoop(e.ctx) }()
 	}
-	if e.sched != nil {
-		e.wg.Add(1)
-		go func() { defer e.wg.Done(); e.sched.Run(e.ctx) }()
-	}
 
 	return e.Reconcile()
 }
@@ -4323,25 +4317,15 @@ func (e *Engine) lifecycleObserver() LifecycleObserver {
 	return e.lifecycle
 }
 
-// Scheduler exposes the schedule runner for the same reason, and answers nil
-// for the same two reasons Alerts does. scheduler.Runner.Last is nil-receiver
-// safe, so the runs page renders an empty report. See Engine.Status.
-func (e *Engine) Scheduler() *scheduler.Runner {
-	if e == nil {
-		return nil
-	}
-	return e.sched
-}
-
 // scheduleActuator is how the scheduler reaches the enable/disable path.
 //
 // Deliberately hair-thin: a schedule writes exactly the intent a human writes
 // and then asks for a reconcile, so a scheduled start and a clicked one are the
 // same code and cannot drift apart.
-type scheduleActuator struct{ e *Engine }
+type scheduleActuator struct{ m *Manager }
 
 func (a scheduleActuator) SetDestinationEnabled(id int64, enabled bool) error {
-	return a.e.store.SetDestinationEnabled(id, enabled)
+	return a.m.store.SetDestinationEnabled(id, enabled)
 }
 
 // SetPlaylistEnabled flips the playlist's stored intent, exactly as the settings
@@ -4372,7 +4356,7 @@ func (a scheduleActuator) SetDestinationEnabled(id int64, enabled bool) error {
 // The error is returned rather than swallowed so the runner leaves the
 // occurrence unhandled and the run log carries the reason.
 func (a scheduleActuator) SetPlaylistEnabled(enabled bool) error {
-	_, err := a.e.store.UpdateSettings(func(s *db.Settings) error {
+	_, err := a.m.store.UpdateSettings(func(s *db.Settings) error {
 		if s.Failover.Playlist.Enabled == enabled {
 			// Already there, so there is nothing to write and nothing to
 			// validate. An overlapping schedule, or a restart inside a window,
@@ -4399,7 +4383,7 @@ func (a scheduleActuator) SetPlaylistEnabled(enabled bool) error {
 }
 
 func (a scheduleActuator) ListDestinationIDs() ([]int64, error) {
-	rows, err := a.e.store.ListDestinations()
+	rows, err := a.m.store.ListDestinations()
 	if err != nil {
 		return nil, err
 	}
@@ -4410,13 +4394,13 @@ func (a scheduleActuator) ListDestinationIDs() ([]int64, error) {
 	return ids, nil
 }
 
-func (a scheduleActuator) Reconcile() error { return a.e.Reconcile() }
+func (a scheduleActuator) Reconcile() error { return a.m.Reconcile() }
 
 // onSchedule publishes the fact that a timetable moved something. A dashboard
 // that shows a destination coming up with no explanation is how an operator
 // concludes the server has a mind of its own.
-func (e *Engine) onSchedule(r scheduler.Result) {
-	e.bus.Publish(eventSchedule, r)
+func (m *Manager) onSchedule(r scheduler.Result) {
+	m.bus.Publish(eventSchedule, r)
 }
 
 // observeWanted reports whether a sweep is worth building.

internal/engine/manager.go

@@ -16,6 +16,7 @@ import (
 	"github.com/rainmanjam/polyemesis/internal/recording"
 	"github.com/rainmanjam/polyemesis/internal/relay"
 	"github.com/rainmanjam/polyemesis/internal/rtmpserver"
+	"github.com/rainmanjam/polyemesis/internal/scheduler"
 	"github.com/rainmanjam/polyemesis/internal/srtserver"
 	"github.com/rainmanjam/polyemesis/internal/stats"
 	"github.com/rainmanjam/polyemesis/internal/transcribe"
@@ -51,6 +52,22 @@ type Manager struct {
 	// internally synchronised.
 	host   *stats.Host
 	recman *recording.Manager
+	// ONE runner for the whole install, not one per engine.
+	//
+	// `schedules` has no source_id (schema.sql) -- a timetable is a property of
+	// the box, not of a programme. Running a Runner per engine put N of them on
+	// one table: whichever swept first wrote `enabled` on EVERY destination
+	// including other programmes', reconciled only ITS OWN engine, and marked
+	// the occurrence handled. The others then read it as handled and never
+	// reconciled, so those destinations were enabled in the database with no
+	// process publishing -- a scheduled broadcast that did not go on air while
+	// the log said `schedule fired`. MarkScheduleRun's `WHERE last_run_at < ?`
+	// is a ratchet on the row, not a lease over the work, and the Actuator
+	// cannot see whether it won it.
+	//
+	// One runner makes that unrepresentable rather than merely unlikely, and
+	// its Reconcile is Manager.Reconcile, which covers every engine. See #526.
+	sched *scheduler.Runner
 	// hostStop ends the sampler goroutine. Set by Start, called by Stop.
 	hostStop context.CancelFunc
 
@@ -118,7 +135,7 @@ type Manager struct {
 
 // NewManager builds the manager. No engines exist until Start.
 func NewManager(log *slog.Logger, cfg config.Config, store *db.DB, tools *ffmpeg.Tools, bus *events.Broker) *Manager {
-	return &Manager{
+	m := &Manager{
 		log:     log,
 		cfg:     cfg,
 		store:   store,
@@ -137,6 +154,11 @@ func NewManager(log *slog.Logger, cfg config.Config, store *db.DB, tools *ffmpeg
 			bus.Publish(events.TypeRecordings, nil)
 		}),
 	}
+	// Built here rather than in Start so a Manager that is never started still
+	// answers Scheduler() with a real runner: the runs page reads Last() and
+	// renders an empty report rather than nothing at all.
+	m.sched = scheduler.New(log, store, scheduleActuator{m}, scheduler.WithOnResult(m.onSchedule))
+	return m
 }
 
 // Tools is the FFmpeg this install detected.
@@ -150,6 +172,20 @@ func (m *Manager) Tools() *ffmpeg.Tools { return m.tools }
 // Host is the process-wide CPU/RAM sampler, running between Start and Stop.
 func (m *Manager) Host() *stats.Host { return m.host }
 
+// Scheduler is the install's one schedule runner.
+//
+// Nil-receiver safe on Runner.Last, so an API reading it before Start renders
+// an empty runs report rather than failing. It used to be reached as
+// s.eng().Scheduler() -- the DEFAULT engine's -- which reported one
+// programme's runs on a multi-source install and, worse, implied there were
+// several timetables. There is one. See #526.
+func (m *Manager) Scheduler() *scheduler.Runner {
+	if m == nil {
+		return nil
+	}
+	return m.sched
+}
+
 // Recordings is the shared, read-only view of the recordings directory.
 //
 // READ-ONLY MEANS "does not drive a recorder": it indexes, measures usage,
@@ -182,6 +218,14 @@ func (m *Manager) Start(ctx context.Context) error {
 	// at the monitoring page of.
 	go m.host.Run(hostCtx)
 
+	// Beside the sampler, and for the same reason: a timetable describes the
+	// box. It runs even on an install where no engine came up, because a
+	// schedule that should have enabled a destination at 19:00 must still mark
+	// its occurrence -- otherwise the window is missed and fires late on the
+	// next sweep after the engine recovers, putting a programme on air an hour
+	// after the show ended.
+	go m.sched.Run(hostCtx)
+
 	// Listeners BEFORE engines. This used to be the other way round, with a
 	// comment about the token lookup needing to see the engines -- but the
 	// lookups resolve m.Engine(id) at connect time, so they were always late-

internal/engine/manager_scheduler_test.go

@@ -0,0 +1,82 @@
+package engine
+
+import (
+	"context"
+	"testing"
+)
+
+// ONE runner for the install, not one per engine.
+//
+// `schedules` carries no source_id: a timetable is a property of the box. A
+// Runner per engine put N of them on that one table, and the failure was not a
+// race that sometimes lost -- it was structural. Whichever swept first wrote
+// `enabled` on EVERY destination, including other programmes', then reconciled
+// ONLY ITS OWN engine and marked the occurrence handled. The other engines read
+// it as handled and never reconciled, so their destinations sat enabled in the
+// database with no process publishing: a scheduled broadcast that did not go on
+// air, while the log said `schedule fired`.
+//
+// MarkScheduleRun's `WHERE last_run_at < ?` is a ratchet on the row, not a
+// lease over the work, and Actuator has no way to learn whether it won it --
+// MarkScheduleRun returns only an error.
+//
+// Asserted on the count rather than on behaviour through a sweep, because the
+// thing that was wrong is structural: with two engines running there must still
+// be exactly one runner, and its Reconcile must be the Manager's. A behavioural
+// test would have to win a race to fail, which is how this survived.
+//
+// Mutation: point scheduleActuator.Reconcile back at a single engine. Observed
+// to fail with "runs 2 engine(s), want 3".
+func TestOneSchedulerForTheInstallNotOnePerEngine(t *testing.T) {
+	m, store := managerFixture(t)
+	addSource(t, store, "first programme")
+	addSource(t, store, "second programme")
+	if err := m.Start(context.Background()); err != nil {
+		t.Fatalf("Start: %v", err)
+	}
+	if got := len(m.Engines()); got < 2 {
+		t.Fatalf("the manager runs %d engine(s); this test needs two or it asserts nothing", got)
+	}
+
+	if m.Scheduler() == nil {
+		t.Fatal("the install has no schedule runner, so no timetable can fire at all")
+	}
+
+	// THE DISCRIMINATOR. A third programme is created AFTER Start, so no engine
+	// exists for it yet. Manager.Reconcile calls Sync first and builds one;
+	// Engine.Reconcile cannot -- an engine cannot create its siblings.
+	//
+	// So this distinguishes the two implementations by behaviour rather than by
+	// shape. Asserting only that a runner exists would pass even with the
+	// per-engine runners restored, which is what the first version of this test
+	// did and why it was worth nothing.
+	before := len(m.Engines())
+	addSource(t, store, "third programme, created after start")
+	if got := len(m.Engines()); got != before {
+		t.Fatalf("a new source built an engine without any reconcile (%d -> %d); "+
+			"the assertion below would then prove nothing", before, got)
+	}
+
+	act := scheduleActuator{m: m}
+	if err := act.Reconcile(); err != nil {
+		t.Fatalf("the scheduled reconcile failed: %v", err)
+	}
+	if got := len(m.Engines()); got <= before {
+		t.Errorf("after a scheduled reconcile the install still runs %d engine(s), "+
+			"unchanged from %d. The actuator reconciled ONE engine rather than the "+
+			"install, which is the bug: a schedule fires, writes `enabled` on every "+
+			"programme's destinations, reconciles only its own, and marks the "+
+			"occurrence handled -- so the other programmes are enabled in the "+
+			"database with nothing publishing", got, before)
+	}
+
+	// And the expansion of "everything" is install-wide, so a schedule that
+	// targets all destinations does not stop at one programme's.
+	ids, err := act.ListDestinationIDs()
+	if err != nil {
+		t.Fatalf("ListDestinationIDs: %v", err)
+	}
+	if ids == nil {
+		t.Error("a schedule targeting every destination expanded to nothing")
+	}
+}

internal/engine/nil_receiver_reads_test.go

@@ -34,7 +34,6 @@ var nilEngineAnswers = map[string]bool{
 	"Levels":      true,
 	"Processes":   true,
 	"Alerts":      true,
-	"Scheduler":   true,
 	"Loudness":    true,
 	// The Meters page's switch, which cannot assert a state it has not been
 	// told. An install with no engine has no analyser tier running, so `false`
@@ -196,10 +195,13 @@ func TestWhatANilEngineActuallyReports(t *testing.T) {
 	if e.Failover() != nil {
 		t.Error("Failover reported a selector tier on an install with no engine")
 	}
-	if e.Alerts() != nil || e.Scheduler() != nil {
-		t.Error("Alerts/Scheduler handed back a notifier or runner that cannot exist; " +
-			"their callers test for nil and refuse, which is how the test-send route " +
-			"avoids reporting \"sent\" for a webhook nobody sent")
+	// Scheduler is no longer here to ask: there is ONE runner and it belongs to
+	// the Manager, because `schedules` has no source_id and a timetable is a
+	// property of the box. See Manager.Scheduler and #526.
+	if e.Alerts() != nil {
+		t.Error("Alerts handed back a notifier that cannot exist; its callers " +
+			"test for nil and refuse, which is how the test-send route avoids " +
+			"reporting \"sent\" for a webhook nobody sent")
 	}
 	if _, at := e.Levels(); !at.IsZero() {
 		t.Errorf("Levels reported a measurement time of %v with nothing metering", at)