amadan

carlos / platform Public

carlos run: the ad-hoc run signal and its converge leg

`carlos run --path /admin/backfill` needs a path to reach the box, and
schedule-run.json cannot carry one: its contract is that etag movement
alone is the signal, with the path looked up in schedules.json. An ad-hoc
run has no declaration to look anything up in, so run-request.json is a
separate object whose content is load-bearing.

On the box it becomes a KindOnce row due now, which is why this is small:
one-shot timers at an arbitrary path already exist and are already
delivered with serialization, retry and status reporting — the app can
already ask for one over its control socket. This opens the same
mechanism to the member who owns the app.

Two things that are not copies of the named signal:

TouchRunPath RETRIES a CAS conflict where TouchRun treats one as success.
For a named signal the conflicting writer carried the same request and
coalescing is right; here the winner carried a different path, so
swallowing it would report a member's run as requested when nothing will
run it.

The row lands after the declaration sweep and as KindOnce, because
DeleteTimersExcept is scoped to KindSchedule — a KindSchedule row would
be deleted on the pass that created it. The nonce names the row, so two
runs of the same path are two rows rather than one overwrite.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0141TPozecoDNeapPJ94REuH
Paul Campbell pushed by paul@keymail.dev cb97234cff2f6b284f187787b0a4009e9fe2ca25 parent 0d128da
4 files changed, +463 −3
  • internal/schedule/converge.go +59 −3
  • internal/schedule/runconverge_test.go +200 −0
  • internal/schedule/runpath.go +95 −0
  • internal/schedule/runpath_test.go +109 −0
diff --git a/internal/schedule/converge.go b/internal/schedule/converge.go
index adbfca188..8138d39e6 100644
--- a/internal/schedule/converge.go
+++ b/internal/schedule/converge.go
@@ -39,6 +39,12 @@ type Converger struct {
declEtag map[[2]string]string // (acct, app) → last etag seen
declDoc map[[2]string]Declaration
runEtag map[[2]string]string
+ // runReqEtag is run-request.json's own memory — `carlos run`, the
+ // ad-hoc path (runpath.go). Separate from runEtag because the two
+ // signals are separate objects with separate movement: a member firing
+ // a declared schedule and a member firing an ad-hoc path in the same
+ // pass are two requests, and one shared etag would drop one of them.
+ runReqEtag map[[2]string]string
// reportedUnsupported tracks which unit-backed hosts have already had
// their one report — Report fires for them once, not every pass.
@@ -72,7 +78,8 @@ func NewConverger(reg *registry.Registry, store release.Store, box string, repor
return &Converger{
Reg: reg, Store: store, Now: time.Now, Box: box, Report: report,
declEtag: map[[2]string]string{}, declDoc: map[[2]string]Declaration{},
- runEtag: map[[2]string]string{}, reportedUnsupported: map[string]bool{},
+ runEtag: map[[2]string]string{}, runReqEtag: map[[2]string]string{},
+ reportedUnsupported: map[string]bool{},
}
}
@@ -164,9 +171,34 @@ func (c *Converger) convergeApp(ctx context.Context, acct, app string, rts []reg
}
c.runEtag[k] = runEtag
+ // Ad-hoc run request (`carlos run`): same seed-then-movement rule as the
+ // named signal, but the PATH travels in the object — there is no
+ // declaration to look it up in. See runpath.go.
+ adhoc := RunRequest{}
+ reqRaw, reqEtag, err := c.Store.Get(ctx, RunRequestKey(acct, app))
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ reqEtag = ""
+ case err != nil:
+ return fmt.Errorf("read run-request.json: %w", err)
+ }
+ if prev, seen := c.runReqEtag[k]; seen && reqEtag != prev && reqEtag != "" {
+ var rq RunRequest
+ if err := json.Unmarshal(reqRaw, &rq); err != nil {
+ log.Printf("schedule: %s/%s: unreadable run request — ignored: %v", acct, app, err)
+ } else if err := ValidPath(rq.Path); err != nil {
+ // Validated at the door too; a box that trusted this blindly
+ // would POST wherever a bucket writer said.
+ log.Printf("schedule: %s/%s: run requested for invalid path %q — ignored: %v", acct, app, rq.Path, err)
+ } else {
+ adhoc = rq
+ }
+ }
+ c.runReqEtag[k] = reqEtag
+
var errs []error
for _, rt := range rts {
- changed, err := c.convergeHost(rt, decl, fireName, now)
+ changed, err := c.convergeHost(rt, decl, fireName, adhoc, now)
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", rt.Host, err))
continue
@@ -185,7 +217,7 @@ func (c *Converger) convergeApp(ctx context.Context, acct, app string, rts []reg
// convergeHost returns whether anything about the host's rows changed (or
// the host is unsupported and should be reported so).
-func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName string, now time.Time) (bool, error) {
+func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName string, adhoc RunRequest, now time.Time) (bool, error) {
if rt.Backing == registry.BackingUnit {
// A host can flip exec -> unit mid-lifetime (the app-capabilities
// migration door, registry.SetBacking): drop any schedule rows it
@@ -265,6 +297,30 @@ func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName s
if n > 0 {
changed = true
}
+ // The ad-hoc run lands AFTER the sweep, and as KindOnce — which is what
+ // keeps it alive. DeleteTimersExcept is scoped to KindSchedule, so a row
+ // that is not in schedules.json (and never will be) is not swept; a
+ // KindSchedule row here would be deleted on this very pass.
+ //
+ // The nonce names the row, so two runs of the same path are two rows
+ // rather than one overwrite, and a name is recognisable in `schedule ls`
+ // as a run rather than a schedule.
+ if adhoc.Path != "" && adhoc.Nonce != "" {
+ name := "run-" + adhoc.Nonce[:12]
+ err := c.Reg.UpsertTimer(registry.Timer{
+ Host: rt.Host, Name: name, Kind: KindOnce,
+ Spec: Spec{Kind: KindOnce, At: now.UTC()}.String(),
+ Path: adhoc.Path,
+ NextAt: now.Unix(), DueAt: now.Unix(), CreatedAt: now.Unix(),
+ })
+ if errors.Is(err, registry.ErrTimerKind) {
+ log.Printf("schedule: %s: ad-hoc run %q collides with a declared schedule of that name — skipped", rt.Host, name)
+ } else if err != nil {
+ return changed, err
+ } else {
+ changed = true
+ }
+ }
if fireName != "" && keep[fireName] {
if err := c.Reg.FireTimerNow(rt.Host, fireName, now.Unix()); err != nil {
return changed, err
diff --git a/internal/schedule/runconverge_test.go b/internal/schedule/runconverge_test.go
new file mode 100644
index 000000000..40693746f
--- /dev/null
+++ b/internal/schedule/runconverge_test.go
@@ -0,0 +1,200 @@
+package schedule
+
+import (
+ "context"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// An ad-hoc run lands as a KindOnce row due now on every EXEC instance of
+// the app — the same fan-out `carlos schedule run` gets, reusing the
+// runner's delivery, serialization and retry rather than growing a second
+// delivery leg.
+func TestConvergeRunRequestMaterialisesAOnceTimerPerExecInstance(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+
+ c := NewConverger(reg, st, "box1", nil)
+ c.Now = func() time.Time { return now }
+ c.Pass(ctx) // seed: nothing to replay
+
+ if err := TouchRunPath(ctx, st, "bab", "app", "/admin/backfill", now); err != nil {
+ t.Fatal(err)
+ }
+ c.Pass(ctx)
+
+ for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
+ rows, err := reg.TimersFor(h)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var found *registry.Timer
+ for i := range rows {
+ if rows[i].Kind == KindOnce {
+ found = &rows[i]
+ }
+ }
+ if found == nil {
+ t.Fatalf("%s got no once-timer for the ad-hoc run: %+v", h, rows)
+ }
+ if found.Path != "/admin/backfill" {
+ t.Errorf("%s path = %q, want /admin/backfill", h, found.Path)
+ }
+ if found.DueAt != now.Unix() {
+ t.Errorf("%s due at %d, want now (%d)", h, found.DueAt, now.Unix())
+ }
+ }
+
+ // A unit-backed instance has no exec child to POST to.
+ rows, _ := reg.TimersFor("u.bab.oncarlos.com")
+ for _, r := range rows {
+ if r.Kind == KindOnce {
+ t.Fatalf("unit-backed host got an ad-hoc run row: %+v", r)
+ }
+ }
+ // Another app's instance must never see this app's run.
+ rows, _ = reg.TimersFor("other.bab.oncarlos.com")
+ if len(rows) != 0 {
+ t.Fatalf("another app's host got rows: %+v", rows)
+ }
+}
+
+// The etag rule the named signal already follows: a request that predates
+// this converger is history, and an unchanged etag is not a new request.
+// Without this a box restart would replay every past run.
+func TestConvergeRunRequestDoesNotReplay(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+
+ // Requested BEFORE this converger ever looked: seeding must not fire it.
+ if err := TouchRunPath(ctx, st, "bab", "app", "/admin/backfill", now); err != nil {
+ t.Fatal(err)
+ }
+ c := NewConverger(reg, st, "box1", nil)
+ c.Now = func() time.Time { return now }
+ c.Pass(ctx)
+ if onceRows(t, reg, "a.bab.oncarlos.com") != 0 {
+ t.Fatal("a pre-existing run request was replayed at seed")
+ }
+
+ // A real request fires exactly once, and the unchanged etag does not
+ // fire it again.
+ if err := TouchRunPath(ctx, st, "bab", "app", "/admin/backfill", now); err != nil {
+ t.Fatal(err)
+ }
+ c.Pass(ctx)
+ if n := onceRows(t, reg, "a.bab.oncarlos.com"); n != 1 {
+ t.Fatalf("after one request: %d once-rows, want 1", n)
+ }
+ c.Pass(ctx)
+ if n := onceRows(t, reg, "a.bab.oncarlos.com"); n != 1 {
+ t.Fatalf("unchanged etag replayed the run: %d once-rows, want 1", n)
+ }
+}
+
+// Two different ad-hoc runs are two rows, not one overwritten row: firing
+// a backfill and then a reindex must run both.
+func TestConvergeRunRequestKeepsDistinctRunsApart(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+
+ c := NewConverger(reg, st, "box1", nil)
+ c.Now = func() time.Time { return now }
+ c.Pass(ctx)
+
+ if err := TouchRunPath(ctx, st, "bab", "app", "/admin/backfill", now); err != nil {
+ t.Fatal(err)
+ }
+ c.Pass(ctx)
+ if err := TouchRunPath(ctx, st, "bab", "app", "/admin/reindex", now); err != nil {
+ t.Fatal(err)
+ }
+ c.Pass(ctx)
+
+ rows, _ := reg.TimersFor("a.bab.oncarlos.com")
+ paths := map[string]bool{}
+ for _, r := range rows {
+ if r.Kind == KindOnce {
+ paths[r.Path] = true
+ }
+ }
+ if !paths["/admin/backfill"] || !paths["/admin/reindex"] {
+ t.Fatalf("both runs should be pending, got %v", paths)
+ }
+}
+
+// The declaration sweep deletes undeclared KindSchedule rows. An ad-hoc run
+// is not in schedules.json and never will be, so a sweep that reached it
+// would delete the run between converge and delivery.
+func TestConvergeDeclarationSweepLeavesAdHocRunsAlone(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+
+ c := NewConverger(reg, st, "box1", nil)
+ c.Now = func() time.Time { return now }
+ c.Pass(ctx)
+ if err := TouchRunPath(ctx, st, "bab", "app", "/admin/backfill", now); err != nil {
+ t.Fatal(err)
+ }
+ c.Pass(ctx)
+
+ // A declaration arrives and is converged: its sweep must not touch the
+ // pending ad-hoc row.
+ if _, err := UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now); err != nil {
+ t.Fatal(err)
+ }
+ c.Pass(ctx)
+
+ if n := onceRows(t, reg, "a.bab.oncarlos.com"); n != 1 {
+ t.Fatalf("the declaration sweep ate the ad-hoc run: %d once-rows, want 1", n)
+ }
+}
+
+// A run at a path the app does not serve is still a run: the platform never
+// interprets the path, and a 404 is the app's answer, visible in the logs.
+// What must NOT happen is the request being silently dropped on the box.
+func TestConvergeRunRequestNamesTheRowAfterTheRequest(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+
+ c := NewConverger(reg, st, "box1", nil)
+ c.Now = func() time.Time { return now }
+ c.Pass(ctx)
+ if err := TouchRunPath(ctx, st, "bab", "app", "/admin/backfill", now); err != nil {
+ t.Fatal(err)
+ }
+ c.Pass(ctx)
+
+ rows, _ := reg.TimersFor("a.bab.oncarlos.com")
+ for _, r := range rows {
+ if r.Kind != KindOnce {
+ continue
+ }
+ if !strings.HasPrefix(r.Name, "run-") {
+ t.Fatalf("ad-hoc row name %q should be recognisable as a run", r.Name)
+ }
+ }
+}
+
+func onceRows(t *testing.T, reg *registry.Registry, host string) int {
+ t.Helper()
+ rows, err := reg.TimersFor(host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ n := 0
+ for _, r := range rows {
+ if r.Kind == KindOnce {
+ n++
+ }
+ }
+ return n
+}
diff --git a/internal/schedule/runpath.go b/internal/schedule/runpath.go
new file mode 100644
index 000000000..25d97b45b
--- /dev/null
+++ b/internal/schedule/runpath.go
@@ -0,0 +1,95 @@
+// This file is apps/<acct>/<app>/run-request.json — the ad-hoc run signal
+// behind `carlos run`, and the one place in this package where the object's
+// CONTENT is load-bearing rather than its etag movement alone.
+//
+// schedule-run.json (declaration.go) names a schedule, and the box looks the
+// path up in schedules.json. An ad-hoc run has no declaration to look
+// anything up in, so the path travels in the object. That is the whole
+// difference between the two signals, and it is why this is a separate key
+// rather than another field on RunSignal: a box that coalesced the two would
+// have to decide which of a name and a path won.
+
+package schedule
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func RunRequestKey(acct, app string) string {
+ return release.AppPrefix(acct, app) + "run-request.json"
+}
+
+// RunRequest is run-request.json: one ad-hoc POST a member asked for, at a
+// path the app serves. Nonce is what makes two runs of the SAME path two
+// distinct signals — without it the bytes would be identical and the etag
+// would not move, which is a run every box would ignore.
+type RunRequest struct {
+ Schema int `json:"schema"`
+ Path string `json:"path"`
+ RequestedAt string `json:"requested_at"`
+ Nonce string `json:"nonce"`
+}
+
+// runPathAttempts bounds the CAS retry. Two members firing runs at the same
+// instant is already unusual; three losing the race in a row is not a case
+// worth spinning on, and the refusal a member reads ("try again") is honest
+// where a silent success would not be.
+const runPathAttempts = 3
+
+// TouchRunPath records an ad-hoc run of path for acct/app.
+//
+// A CAS conflict is RETRIED here, where TouchRun treats one as success. The
+// difference is what the conflicting writer carried: for a named signal it
+// is the same request twice and coalescing is right, but the winner of a
+// race here holds a DIFFERENT path, and swallowing that would report a
+// member's run as requested when nothing will ever run it.
+func TouchRunPath(ctx context.Context, s release.Store, acct, app, path string, now time.Time) error {
+ if err := checkIDs(acct, app); err != nil {
+ return err
+ }
+ if err := ValidPath(path); err != nil {
+ return fmt.Errorf("schedule: %w", err)
+ }
+ key := RunRequestKey(acct, app)
+ for attempt := 0; ; attempt++ {
+ _, etag, err := s.Get(ctx, key)
+ if errors.Is(err, os.ErrNotExist) {
+ etag = ""
+ } else if err != nil {
+ return err
+ }
+ nonce := make([]byte, 16)
+ if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
+ return err
+ }
+ raw, err := json.Marshal(RunRequest{
+ Schema: 1,
+ Path: path,
+ RequestedAt: now.UTC().Format(time.RFC3339),
+ Nonce: hex.EncodeToString(nonce),
+ })
+ if err != nil {
+ return err
+ }
+ _, err = s.Put(ctx, key, raw, etag)
+ if err == nil {
+ return nil
+ }
+ if !errors.Is(err, release.ErrCASConflict) {
+ return err
+ }
+ if attempt+1 >= runPathAttempts {
+ return fmt.Errorf("schedule: run request for %s/%s lost %d races with another writer; nothing was requested — try again", acct, app, runPathAttempts)
+ }
+ }
+}
diff --git a/internal/schedule/runpath_test.go b/internal/schedule/runpath_test.go
new file mode 100644
index 000000000..a4fa498fc
--- /dev/null
+++ b/internal/schedule/runpath_test.go
@@ -0,0 +1,109 @@
+package schedule
+
+import (
+ "context"
+ "encoding/json"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The ad-hoc run request carries its PATH, unlike schedule-run.json whose
+// etag movement alone is the signal: there is no declaration to look the
+// path up in, so the object is the only place it can live.
+func TestTouchRunPathCarriesThePath(t *testing.T) {
+ ctx := context.Background()
+ st := release.NewDirStore(t.TempDir())
+ if err := TouchRunPath(ctx, st, "bab", "app", "/admin/backfill", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ raw, etag, err := st.Get(ctx, RunRequestKey("bab", "app"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if etag == "" {
+ t.Fatal("no etag on the run request")
+ }
+ var got RunRequest
+ if err := json.Unmarshal(raw, &got); err != nil {
+ t.Fatal(err)
+ }
+ if got.Path != "/admin/backfill" {
+ t.Fatalf("Path = %q, want /admin/backfill", got.Path)
+ }
+}
+
+// Two runs of the same path must be two distinct signals — a member firing
+// the same backfill twice is asking for it twice, and an unmoved etag is a
+// run every box would ignore.
+func TestTouchRunPathMovesTheEtagForTheSamePath(t *testing.T) {
+ ctx := context.Background()
+ st := release.NewDirStore(t.TempDir())
+ if err := TouchRunPath(ctx, st, "bab", "app", "/admin/backfill", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ _, e1, _ := st.Get(ctx, RunRequestKey("bab", "app"))
+ if err := TouchRunPath(ctx, st, "bab", "app", "/admin/backfill", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ _, e2, _ := st.Get(ctx, RunRequestKey("bab", "app"))
+ if e1 == e2 {
+ t.Fatalf("etag did not move for a repeated run: %q", e1)
+ }
+}
+
+// The path is validated before anything is written, so a member typo costs
+// a round trip and leaves no object for a box to act on.
+func TestTouchRunPathRefusesABadPath(t *testing.T) {
+ ctx := context.Background()
+ st := release.NewDirStore(t.TempDir())
+ if err := TouchRunPath(ctx, st, "bab", "app", "../etc/passwd", time.Now()); err == nil {
+ t.Fatal("TouchRunPath(../etc/passwd) = nil, want a refusal")
+ }
+ if _, _, err := st.Get(ctx, RunRequestKey("bab", "app")); err == nil {
+ t.Fatal("a refused path still wrote the object")
+ }
+}
+
+// conflictOnce wraps a store and fails the first Put on key with a CAS
+// conflict — a second writer that landed between our Get and our Put.
+type conflictOnce struct {
+ release.Store
+ key string
+ done bool
+}
+
+func (c *conflictOnce) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if key == c.key && !c.done {
+ c.done = true
+ return "", release.ErrCASConflict
+ }
+ return c.Store.Put(ctx, key, body, ifMatch)
+}
+
+// A CAS conflict is SUCCESS for TouchRun — the named signal already fired,
+// and a second bump would be the same request twice. It is a LOST RUN here:
+// the winner carried a DIFFERENT path, so swallowing the conflict would
+// tell a member their backfill was requested when nothing will run it.
+// Retry, and leave our own path in the object.
+func TestTouchRunPathRetriesPastAConflictingWriter(t *testing.T) {
+ ctx := context.Background()
+ base := release.NewDirStore(t.TempDir())
+ st := &conflictOnce{Store: base, key: RunRequestKey("bab", "app")}
+
+ if err := TouchRunPath(ctx, st, "bab", "app", "/admin/backfill", time.Now()); err != nil {
+ t.Fatalf("TouchRunPath past one conflict = %v, want nil", err)
+ }
+ raw, _, err := base.Get(ctx, RunRequestKey("bab", "app"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var got RunRequest
+ if err := json.Unmarshal(raw, &got); err != nil {
+ t.Fatal(err)
+ }
+ if got.Path != "/admin/backfill" {
+ t.Fatalf("Path = %q after a conflict, want our own path — the run was lost", got.Path)
+ }
+}