amadan

carlos / platform Public

schedule: runner — due ticks, outcome policy, backoff, events

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 2f23ed872ed73c8c0d81bde70b17b38baa5680b9 parent 5628dfb
3 files changed, +493 −0
  • internal/schedule/envfile.go +32 −0
  • internal/schedule/runner.go +253 −0
  • internal/schedule/runner_test.go +208 −0
diff --git a/internal/schedule/envfile.go b/internal/schedule/envfile.go
new file mode 100644
index 000000000..f8ee0c5ae
--- /dev/null
+++ b/internal/schedule/envfile.go
@@ -0,0 +1,32 @@
+package schedule
+
+import (
+ "bufio"
+ "os"
+ "strings"
+)
+
+// readEnvFile parses a KEY=value EnvironmentFile (blank lines and #comments
+// ignored). Copied from internal/activator/backend_exec.go's function of the
+// same name — importing activator here would be a cycle (activator is a
+// consumer of this package's runner), so the ~20-line parser is duplicated
+// rather than shared. Keep the two in sync if either changes.
+func readEnvFile(path string) (map[string]string, error) {
+ f, err := os.Open(path)
+ if err != nil {
+ return nil, err
+ }
+ defer f.Close()
+ m := map[string]string{}
+ sc := bufio.NewScanner(f)
+ for sc.Scan() {
+ line := strings.TrimSpace(sc.Text())
+ if line == "" || strings.HasPrefix(line, "#") {
+ continue
+ }
+ if k, v, ok := strings.Cut(line, "="); ok {
+ m[strings.TrimSpace(k)] = strings.TrimSpace(v)
+ }
+ }
+ return m, sc.Err()
+}
diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go
new file mode 100644
index 000000000..3c53f7d76
--- /dev/null
+++ b/internal/schedule/runner.go
@@ -0,0 +1,253 @@
+package schedule
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "net/http"
+ "path/filepath"
+ "strconv"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+const (
+ DeliveryTimeout = 30 * time.Minute
+ TickEvery = 15 * time.Second
+ MaxAttempts = 5
+ MaxParallelHosts = 4
+ dueBatch = 64
+ backoffBase = time.Minute
+ backoffCap = 15 * time.Minute
+)
+
+// ErrTimeout is what a Deliverer returns when the handler ran past the
+// delivery timeout. The policy treats it as delivered (spec §6.3): the job
+// may have finished, or still be running — a retry would double it.
+var ErrTimeout = errors.New("schedule: delivery timed out")
+
+// Deliverer sends one tick and reports the handler's status, or a
+// transport-level error (wake failed, connect refused, ErrTimeout).
+type Deliverer interface {
+ Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error)
+}
+
+// Runner fires due timers: at-least-once, serialized per host, bounded
+// parallelism across hosts (spec §6.3). Every seam is a field so a tick is
+// testable with no goroutines, sockets or clocks.
+type Runner struct {
+ Reg *registry.Registry
+ Deliver Deliverer
+ Now func() time.Time
+ Timeout time.Duration
+ Token func(host string) (string, error)
+ Event func(rt registry.Route, level, msg string)
+ Report func(rt registry.Route, st Status)
+ Box string
+ Build string
+}
+
+// Start runs Tick every TickEvery until ctx ends.
+func (r *Runner) Start(ctx context.Context) {
+ go func() {
+ t := time.NewTicker(TickEvery)
+ defer t.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-t.C:
+ r.Tick(ctx)
+ }
+ }
+ }()
+}
+
+func (r *Runner) now() time.Time {
+ if r.Now != nil {
+ return r.Now()
+ }
+ return time.Now()
+}
+
+func (r *Runner) timeout() time.Duration {
+ if r.Timeout > 0 {
+ return r.Timeout
+ }
+ return DeliveryTimeout
+}
+
+// Tick is one pass: everything due, grouped by host, hosts in parallel up to
+// MaxParallelHosts, timers within a host one after another.
+func (r *Runner) Tick(ctx context.Context) {
+ due, err := r.Reg.DueTimers(r.now().Unix(), dueBatch)
+ if err != nil {
+ log.Printf("schedule: due query: %v", err)
+ return
+ }
+ if len(due) == 0 {
+ return
+ }
+ byHost := map[string][]registry.Timer{}
+ var order []string
+ for _, t := range due {
+ if _, seen := byHost[t.Host]; !seen {
+ order = append(order, t.Host)
+ }
+ byHost[t.Host] = append(byHost[t.Host], t)
+ }
+ sem := make(chan struct{}, MaxParallelHosts)
+ var wg sync.WaitGroup
+ for _, host := range order {
+ rt, err := r.Reg.Get(host)
+ if err != nil || rt == nil {
+ // Converge prunes these; until it does, they are not ours to fire.
+ continue
+ }
+ wg.Add(1)
+ sem <- struct{}{}
+ go func(rt registry.Route, timers []registry.Timer) {
+ defer wg.Done()
+ defer func() { <-sem }()
+ for _, t := range timers {
+ r.fire(ctx, rt, t)
+ }
+ if r.Report != nil {
+ if st, err := StatusFor(r.Reg, rt, r.Box, r.now()); err == nil {
+ r.Report(rt, st)
+ }
+ }
+ }(*rt, byHost[host])
+ }
+ wg.Wait()
+}
+
+func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer) {
+ sp, err := ParseSpec(t.Spec)
+ if err != nil {
+ log.Printf("schedule: %s/%s: bad spec %q: %v — dropping the row", rt.Host, t.Name, t.Spec, err)
+ r.Reg.DeleteTimer(rt.Host, t.Name)
+ return
+ }
+ start := r.now()
+ token, err := r.Token(rt.Host)
+ if err != nil {
+ r.record(rt, t, sp, start, Outcome{Status: "no instance token: " + err.Error(), Retry: true}, 0)
+ return
+ }
+ dctx, cancel := context.WithTimeout(ctx, r.timeout())
+ req := BuildRequest(rt, t, token, r.Build).WithContext(dctx)
+ status, derr := r.Deliver.Deliver(dctx, rt, req)
+ cancel()
+ if errors.Is(derr, context.DeadlineExceeded) {
+ derr = ErrTimeout
+ }
+ r.record(rt, t, sp, start, classify(status, derr, r.timeout()), r.now().Sub(start))
+}
+
+// record applies the policy table to one outcome and writes the row, the
+// member-facing event, and nothing else.
+func (r *Runner) record(rt registry.Route, t registry.Timer, sp Spec, at time.Time, o Outcome, took time.Duration) {
+ attempts := 0
+ status := o.Status
+ natural := sp.Next(at) // zero for a finished once
+ next := natural.Unix()
+ if natural.IsZero() {
+ next = 0
+ }
+ level := "info"
+ if o.Retry {
+ attempts = t.Attempts + 1
+ if attempts >= MaxAttempts {
+ status = fmt.Sprintf("failed after %d attempts: %s", MaxAttempts, o.Status)
+ attempts = 0
+ level = "warn"
+ } else {
+ backoff := backoffBase << (attempts - 1)
+ if backoff > backoffCap {
+ backoff = backoffCap
+ }
+ retryAt := at.Add(backoff)
+ if !natural.IsZero() && retryAt.After(natural) {
+ retryAt = natural
+ }
+ next = retryAt.Unix()
+ level = "warn"
+ }
+ } else if !isSuccess(o.Status) {
+ level = "warn"
+ }
+ if err := r.Reg.MarkTimerFired(rt.Host, t.Name, at.Unix(), status, took.Milliseconds(), attempts, next); err != nil {
+ log.Printf("schedule: %s/%s: record fire: %v", rt.Host, t.Name, err)
+ }
+ if r.Event != nil {
+ msg := fmt.Sprintf("schedule %s fired POST %s → %s in %s", t.Name, t.Path, status, took.Truncate(time.Millisecond))
+ if attempts > 0 {
+ msg += fmt.Sprintf(" (attempt %d of %d; retrying at %s)", attempts, MaxAttempts, time.Unix(next, 0).UTC().Format(time.RFC3339))
+ }
+ r.Event(rt, level, msg)
+ }
+}
+
+func isSuccess(status string) bool { return len(status) == 3 && status[0] == '2' }
+
+// Outcome is one line of the policy table (spec §6.3).
+type Outcome struct {
+ Status string
+ Retry bool
+}
+
+func classify(status int, err error, timeout time.Duration) Outcome {
+ switch {
+ case errors.Is(err, ErrTimeout):
+ return Outcome{Status: "timeout after " + timeout.String()}
+ case err != nil:
+ return Outcome{Status: "503: " + err.Error(), Retry: true}
+ case status >= 200 && status < 300:
+ return Outcome{Status: strconv.Itoa(status)}
+ case status >= 500, status == 408, status == 429:
+ return Outcome{Status: strconv.Itoa(status), Retry: true}
+ default:
+ return Outcome{Status: strconv.Itoa(status) + " (app refused)"}
+ }
+}
+
+// Classify is the exported policy for the host package's tests.
+func Classify(status int, err error) Outcome { return classify(status, err, DeliveryTimeout) }
+
+// BuildRequest is the tick (spec §6.3's header table). Content-Length 0 and
+// no body: a tick carries no data, by design.
+func BuildRequest(rt registry.Route, t registry.Timer, token, build string) *http.Request {
+ req, err := http.NewRequest(http.MethodPost, "http://"+rt.Host+t.Path, http.NoBody)
+ if err != nil {
+ // Path was validated at declaration; a failure here is a bug, not
+ // a runtime condition. Fall back to the root so the app still hears.
+ req, _ = http.NewRequest(http.MethodPost, "http://"+rt.Host+"/", http.NoBody)
+ }
+ req.Host = rt.Host
+ req.ContentLength = 0
+ req.Header.Set("Authorization", "Bearer "+token)
+ req.Header.Set("X-Carlos-Schedule", t.Name)
+ req.Header.Set("X-Carlos-Schedule-Kind", t.Kind)
+ req.Header.Set("X-Carlos-Schedule-At", strconv.FormatInt(t.NextAt, 10))
+ req.Header.Set("User-Agent", "carlos-schedule/"+build)
+ return req
+}
+
+// TokenFromEnvFile is the production Token: CARLOS_ADMIN_TOKEN from
+// <dataDir>/<host>.env, the file the activator owns (spec §6.4).
+func TokenFromEnvFile(dataDir string) func(host string) (string, error) {
+ return func(host string) (string, error) {
+ env, err := readEnvFile(filepath.Join(dataDir, host+".env"))
+ if err != nil {
+ return "", err
+ }
+ if env["CARLOS_ADMIN_TOKEN"] == "" {
+ return "", errors.New("no CARLOS_ADMIN_TOKEN")
+ }
+ return env["CARLOS_ADMIN_TOKEN"], nil
+ }
+}
diff --git a/internal/schedule/runner_test.go b/internal/schedule/runner_test.go
new file mode 100644
index 000000000..073e0e4b4
--- /dev/null
+++ b/internal/schedule/runner_test.go
@@ -0,0 +1,208 @@
+package schedule
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "path/filepath"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+type fakeDeliverer struct {
+ mu sync.Mutex
+ status int
+ err error
+ calls []*http.Request
+ hosts []string
+}
+
+func (f *fakeDeliverer) Deliver(_ context.Context, rt registry.Route, req *http.Request) (int, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.calls = append(f.calls, req)
+ f.hosts = append(f.hosts, rt.Host)
+ return f.status, f.err
+}
+
+func runnerFixture(t *testing.T, d *fakeDeliverer) (*registry.Registry, *Runner) {
+ t.Helper()
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ if err := reg.Upsert(registry.Route{Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true}); err != nil {
+ t.Fatal(err)
+ }
+ r := &Runner{Reg: reg, Deliver: d, Timeout: time.Minute, Box: "box1", Build: "test",
+ Token: func(string) (string, error) { return "tok", nil }}
+ return reg, r
+}
+
+func TestClassify(t *testing.T) {
+ cases := []struct {
+ status int
+ err error
+ want Outcome
+ }{
+ {204, nil, Outcome{Status: "204", Retry: false}},
+ {200, nil, Outcome{Status: "200", Retry: false}},
+ {404, nil, Outcome{Status: "404 (app refused)", Retry: false}},
+ {403, nil, Outcome{Status: "403 (app refused)", Retry: false}},
+ {302, nil, Outcome{Status: "302 (app refused)", Retry: false}},
+ {500, nil, Outcome{Status: "500", Retry: true}},
+ {503, nil, Outcome{Status: "503", Retry: true}},
+ {429, nil, Outcome{Status: "429", Retry: true}},
+ {408, nil, Outcome{Status: "408", Retry: true}},
+ {0, errors.New("instance failed to wake"), Outcome{Status: "503: instance failed to wake", Retry: true}},
+ {0, ErrTimeout, Outcome{Status: "timeout after 1m0s", Retry: false}},
+ }
+ for _, c := range cases {
+ if got := classify(c.status, c.err, time.Minute); got != c.want {
+ t.Errorf("classify(%d, %v) = %+v, want %+v", c.status, c.err, got, c.want)
+ }
+ }
+}
+
+func TestBuildRequestCarriesTheContract(t *testing.T) {
+ rt := registry.Route{Host: "a.bab.oncarlos.com"}
+ tm := registry.Timer{Name: "sync", Kind: KindSchedule, Path: "/jobs/sync?x=1", NextAt: 1787004000}
+ req := BuildRequest(rt, tm, "tok", "abc")
+ if req.Method != http.MethodPost || req.Host != rt.Host || req.URL.Path != "/jobs/sync" || req.URL.RawQuery != "x=1" {
+ t.Fatalf("req = %s %s %s", req.Method, req.Host, req.URL)
+ }
+ h := req.Header
+ if h.Get("Authorization") != "Bearer tok" || h.Get("X-Carlos-Schedule") != "sync" || h.Get("X-Carlos-Schedule-Kind") != "schedule" ||
+ h.Get("X-Carlos-Schedule-At") != "1787004000" || h.Get("User-Agent") != "carlos-schedule/abc" || req.ContentLength != 0 {
+ t.Fatalf("headers = %v", h)
+ }
+}
+
+func TestTickFiresDueAdvancesScheduleAndFinishesOnce(t *testing.T) {
+ d := &fakeDeliverer{status: 204}
+ reg, r := runnerFixture(t, d)
+ now := at("2026-08-23T13:07:00Z")
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix() - 5})
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "later", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/l", NextAt: now.Unix() + 60})
+ var events []string
+ r.Event = func(_ registry.Route, level, msg string) { events = append(events, level+": "+msg) }
+ var reports int
+ r.Report = func(registry.Route, Status) { reports++ }
+ r.Tick(context.Background())
+ if len(d.calls) != 2 {
+ t.Fatalf("delivered %d, want 2 (not the future one)", len(d.calls))
+ }
+ sync, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
+ if sync.LastStatus != "204" || sync.LastAt != now.Unix() || sync.NextAt != at("2026-08-23T18:00:00Z").Unix() || sync.Attempts != 0 {
+ t.Fatalf("sync = %+v", sync)
+ }
+ once, _ := reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if once.NextAt != 0 || once.LastStatus != "204" {
+ t.Fatalf("once = %+v", once)
+ }
+ if len(events) != 2 || events[0][:5] != "info:" {
+ t.Fatalf("events = %v", events)
+ }
+ if reports != 1 { // one report per host per tick, not per timer
+ t.Fatalf("reports = %d", reports)
+ }
+}
+
+func TestTickRetriesWithBackoffThenGivesUp(t *testing.T) {
+ d := &fakeDeliverer{status: 503}
+ reg, r := runnerFixture(t, d)
+ now := at("2026-08-23T13:07:00Z")
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix()})
+ wantBackoff := []time.Duration{time.Minute, 2 * time.Minute, 4 * time.Minute, 8 * time.Minute}
+ for i, b := range wantBackoff {
+ r.Tick(context.Background())
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
+ if tm.Attempts != i+1 || tm.NextAt != now.Add(b).Unix() || tm.LastStatus != "503" {
+ t.Fatalf("attempt %d: %+v (want next=+%s)", i+1, tm, b)
+ }
+ now = now.Add(b)
+ }
+ r.Tick(context.Background()) // 5th failure: give up, advance to the natural next
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
+ if tm.Attempts != 0 || tm.LastStatus != "failed after 5 attempts: 503" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
+ t.Fatalf("after giving up: %+v", tm)
+ }
+}
+
+func TestBackoffNeverPassesTheNaturalNext(t *testing.T) {
+ d := &fakeDeliverer{status: 503}
+ reg, r := runnerFixture(t, d)
+ now := at("2026-08-23T13:59:30Z") // next natural 1m fire is 14:00:00, 30s away
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "m", Kind: KindSchedule, Spec: "every 1m0s", Path: "/m", NextAt: now.Unix()})
+ r.Tick(context.Background())
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "m")
+ if tm.NextAt != at("2026-08-23T14:00:00Z").Unix() {
+ t.Fatalf("next = %d, want the natural next, not now+1m", tm.NextAt)
+ }
+}
+
+func TestAppRefusalDoesNotRetry(t *testing.T) {
+ d := &fakeDeliverer{status: 404}
+ reg, r := runnerFixture(t, d)
+ now := at("2026-08-23T13:07:00Z")
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "o", Kind: KindOnce, Spec: "at 1", Path: "/o", NextAt: 1})
+ var levels []string
+ r.Event = func(_ registry.Route, level, _ string) { levels = append(levels, level) }
+ r.Tick(context.Background())
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "o")
+ if tm.NextAt != 0 || tm.LastStatus != "404 (app refused)" || tm.Attempts != 0 {
+ t.Fatalf("%+v", tm)
+ }
+ if len(levels) != 1 || levels[0] != "warn" {
+ t.Fatalf("levels = %v", levels)
+ }
+}
+
+func TestTimeoutCountsAsDelivered(t *testing.T) {
+ d := &fakeDeliverer{err: ErrTimeout}
+ reg, r := runnerFixture(t, d)
+ now := at("2026-08-23T13:07:00Z")
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: now.Unix()})
+ r.Tick(context.Background())
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s")
+ if tm.Attempts != 0 || tm.LastStatus != "timeout after 1m0s" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
+ t.Fatalf("%+v", tm)
+ }
+}
+
+func TestNoTokenIsRecordedNotDelivered(t *testing.T) {
+ d := &fakeDeliverer{status: 204}
+ reg, r := runnerFixture(t, d)
+ r.Token = func(string) (string, error) { return "", errors.New("no CARLOS_ADMIN_TOKEN") }
+ now := at("2026-08-23T13:07:00Z")
+ r.Now = func() time.Time { return now }
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: now.Unix()})
+ r.Tick(context.Background())
+ if len(d.calls) != 0 {
+ t.Fatal("delivered without a token")
+ }
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s")
+ if tm.Attempts != 1 || tm.LastStatus != "no instance token: no CARLOS_ADMIN_TOKEN" {
+ t.Fatalf("%+v", tm)
+ }
+}
+
+func TestTickSkipsRowsWhoseRouteIsGone(t *testing.T) {
+ d := &fakeDeliverer{status: 204}
+ reg, r := runnerFixture(t, d)
+ reg.UpsertTimer(registry.Timer{Host: "ghost.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: 1})
+ r.Tick(context.Background())
+ if len(d.calls) != 0 {
+ t.Fatal("delivered to a host with no route")
+ }
+}