amadan

carlos / platform Public

schedule: serialize Converger.Pass — two agent loops share one converger

fastConvergeLoop (seed + serial leg) and adoptLoop (hourly floor) call Pass on
one shared Converger from two goroutines, and they overlap. Its four etag maps
had no mutex: a concurrent map read and write is a runtime throw that would
take the agent and the activator down, and two interleaved passes could both
see the run etag move and fire the same manual run twice.

The lock lives inside the Converger, not at the agent's call sites, so a future
caller cannot forget it.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev be49f6d6af03ce9ca75062186130d16433384877 parent 4e2e7ef
3 files changed, +153 −1
  • internal/host/agent.go +4 −1
  • internal/schedule/converge.go +34 −0
  • internal/schedule/converge_test.go +115 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go
index f3d00da79..a39f1cd88 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -317,7 +317,10 @@ func RunAgent(args []string) {
// hourly leg in adoptLoop, and the runner started here. One converger is a
// correctness requirement, not tidiness — its etag memory is what makes a
// declaration change cheap and a run signal exactly-once, and a second
- // instance elsewhere would replay or swallow one.
+ // instance elsewhere would replay or swallow one. The three legs run in
+ // two different goroutines and do overlap (an hourly pass mid-flight when
+ // a console bump fires the fast one); Converger.Pass serializes itself on
+ // its own mutex, so no call site here needs one.
//
// A box with no deploy store simply has no scheduled work: schedules live
// in the bucket, so there is nothing to converge and nowhere to report.
diff --git a/internal/schedule/converge.go b/internal/schedule/converge.go
index 803c9102b..8dd839dcd 100644
--- a/internal/schedule/converge.go
+++ b/internal/schedule/converge.go
@@ -7,6 +7,7 @@ import (
"fmt"
"log"
"os"
+ "sync"
"time"
"github.com/carlosframework/platform/internal/registry"
@@ -22,6 +23,11 @@ import (
// exactly-once: an unchanged declaration is one GET and no writes; a run
// signal fires when its etag MOVES, and the first sighting only seeds
// (a signal older than this process is history — restart-gen's rule).
+//
+// One instance, but THREE callers, and they are separate goroutines: the
+// agent's fast converge loop runs Pass at arming and on every serial move,
+// and its adopt loop runs Pass hourly. Pass is therefore serialized on mu —
+// see the field.
type Converger struct {
Reg *registry.Registry
Store release.Store
@@ -36,6 +42,27 @@ type Converger struct {
// reportedUnsupported tracks which unit-backed hosts have already had
// their one report — Report fires for them once, not every pass.
reportedUnsupported map[string]bool
+
+ // mu serializes Pass, and is held for the WHOLE of it. Two reasons, and
+ // the first one alone is fatal:
+ //
+ // 1. The four maps above are plain maps written on every pass. The agent
+ // calls Pass from two goroutines (fastConvergeLoop's seed and serial
+ // leg; adoptLoop's hourly floor), and they genuinely overlap — an
+ // hourly pass mid-flight when a console bump fires the fast one. A
+ // concurrent map read and write is a runtime THROW, not a data race
+ // you find later: it would kill the agent, and the activator and every
+ // exec child with it.
+ // 2. The etag memory IS the exactly-once mechanism for a manual run. Two
+ // interleaved passes can both read runEtag[k] before either writes it,
+ // both see movement, and both fire the same `carlos schedule run` —
+ // the one promise this struct's comment makes, broken.
+ //
+ // The lock lives HERE rather than at the agent's call sites (the shape
+ // configSyncMu/reconcileMu use in host/agent.go) because a Converger is a
+ // stateful object with an exported method: a future caller cannot forget a
+ // lock it never has to take.
+ mu sync.Mutex
}
const onceRetention = 7 * 24 * time.Hour
@@ -51,7 +78,14 @@ func NewConverger(reg *registry.Registry, store release.Store, box string, repor
// Pass is die-free per app: one app's unreadable declaration is logged and
// its rows left exactly as they were (a baseline with holes manufactures
// deletions), and every other app still converges.
+//
+// Serialized end to end (see mu): concurrent callers queue rather than
+// interleave. A pass is store-bound, so a queued caller can wait on a slow
+// bucket — that is the intended cost, and cheaper than the alternative for
+// both reasons mu documents.
func (c *Converger) Pass(ctx context.Context) error {
+ c.mu.Lock()
+ defer c.mu.Unlock()
routes, err := c.Reg.List()
if err != nil {
return err
diff --git a/internal/schedule/converge_test.go b/internal/schedule/converge_test.go
index 321d047f7..177c24b3d 100644
--- a/internal/schedule/converge_test.go
+++ b/internal/schedule/converge_test.go
@@ -3,6 +3,7 @@ package schedule
import (
"context"
"path/filepath"
+ "sync"
"testing"
"time"
@@ -233,3 +234,117 @@ func TestConvergeDropsScheduleRowsOnUnitFlip(t *testing.T) {
t.Fatalf("no unsupported report for the flipped host: %+v", reports)
}
}
+
+// gateStore blocks every Get until open is closed, recording the keys it was
+// asked for. It is how the two tests below observe whether a second Pass got
+// past the lock while the first one was parked mid-pass in the store.
+type gateStore struct {
+ release.Store
+ entered chan string
+ open chan struct{}
+}
+
+func (g *gateStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ g.entered <- key
+ <-g.open
+ return g.Store.Get(ctx, key)
+}
+
+// TestConvergePassHoldsTheLockForTheWholePass pins the serialization the agent
+// depends on: fastConvergeLoop (seed + serial leg) and adoptLoop (hourly floor)
+// call Pass on ONE shared Converger from two goroutines, and they genuinely
+// overlap. Without the lock the etag maps take a concurrent read and write —
+// a runtime throw that kills the agent and the activator with it — and two
+// interleaved passes can fire the same manual run twice.
+//
+// The store parks the first pass inside its very first Get, still holding the
+// lock; a second Pass must not reach the store at all until it is released.
+func TestConvergePassHoldsTheLockForTheWholePass(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
+ gate := &gateStore{Store: st, entered: make(chan string, 64), open: make(chan struct{})}
+ c := NewConverger(reg, gate, "box1", nil)
+ c.Now = func() time.Time { return now }
+
+ first := make(chan error, 1)
+ go func() { first <- c.Pass(ctx) }()
+ <-gate.entered // pass 1 is inside a Get, holding the lock
+
+ second := make(chan error, 1)
+ go func() { second <- c.Pass(ctx) }()
+ select {
+ case k := <-gate.entered:
+ // Only pass 2 can be here: pass 1 is parked inside its Get and cannot
+ // issue another until open closes.
+ t.Fatalf("a second Pass reached the store (%s) while the first held the lock", k)
+ case <-time.After(100 * time.Millisecond):
+ // Blocked, as it must be. A slow machine can only weaken this check,
+ // never fail it spuriously.
+ }
+ close(gate.open)
+ if err := <-first; err != nil {
+ t.Fatalf("first pass: %v", err)
+ }
+ if err := <-second; err != nil {
+ t.Fatalf("second pass: %v", err)
+ }
+ if len(gate.entered) == 0 {
+ t.Fatal("the gate store was never consulted — this test pinned nothing")
+ }
+}
+
+// TestConvergePassIsSafeUnderConcurrentCallers is the -race half of the same
+// finding, and the exactly-once half of it: four passes racing on one
+// Converger must leave the manual run fired ONCE, with the etag recorded so a
+// later pass does not replay it. The Report closure appends to a plain slice
+// on purpose — Pass calls it under the lock, so a caller's ordinary,
+// non-thread-safe callback is safe, and this would flag under -race if that
+// ever stopped being true.
+func TestConvergePassIsSafeUnderConcurrentCallers(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
+ var reports []Status
+ c := NewConverger(reg, st, "box1", func(_ registry.Route, s Status) { reports = append(reports, s) })
+ c.Now = func() time.Time { return now }
+ if err := c.Pass(ctx); err != nil { // seed: a run signal older than the converger is history
+ t.Fatal(err)
+ }
+ TouchRun(ctx, st, "bab", "app", "sync", now)
+
+ var wg sync.WaitGroup
+ errs := make(chan error, 8)
+ for i := 0; i < 4; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ if err := c.Pass(ctx); err != nil {
+ errs <- err
+ }
+ }()
+ }
+ wg.Wait()
+ close(errs)
+ for err := range errs {
+ t.Fatalf("concurrent Pass: %v", err)
+ }
+ for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
+ if tm, _ := reg.GetTimer(h, "sync"); tm.NextAt != now.Unix() {
+ t.Fatalf("%s: run signal not fired: %+v", h, tm)
+ }
+ }
+ // Exactly-once: the etag was recorded, so the next pass is a no-op.
+ reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 777)
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
+ t.Fatal("the run signal replayed after the concurrent passes")
+ }
+ if len(reports) == 0 {
+ t.Fatal("no reports at all — the fixture converged nothing")
+ }
+}