amadan

carlos / platform Public

schedule: control-socket verbs for once-timers, bearer-gated

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 5d447e98e8a51dd4c753538ffb68be7ae5fa4559 parent a8636a0
2 files changed, +274 −0
  • internal/schedule/control.go +147 −0
  • internal/schedule/control_test.go +127 −0
diff --git a/internal/schedule/control.go b/internal/schedule/control.go
new file mode 100644
index 000000000..693417597
--- /dev/null
+++ b/internal/schedule/control.go
@@ -0,0 +1,147 @@
+package schedule
+
+import (
+ "crypto/sha256"
+ "crypto/subtle"
+ "encoding/json"
+ "errors"
+ "net/http"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// TimerStore is the registry subset the control channel needs.
+type TimerStore interface {
+ UpsertTimer(registry.Timer) error
+ DeleteTimer(host, name string) error
+ GetTimer(host, name string) (*registry.Timer, error)
+ TimersFor(host string) ([]registry.Timer, error)
+ CountTimers(host, kind string) (int, error)
+}
+
+// Mount adds the once-timer verbs to an instance's control socket (spec
+// §7.2). host is fixed here and never read from a request — the
+// confused-deputy guarantee edge.ListenPurge gives — and every verb needs
+// the host's own CARLOS_ADMIN_TOKEN as a bearer, which is what closes the
+// same-uid-sibling gap purge.go documents: a sibling can compute this
+// socket's path, but not this instance's token.
+func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (string, error), now func() time.Time) {
+ authed := func(fn http.HandlerFunc) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ want, err := token()
+ if err != nil || want == "" {
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ got := sha256.Sum256([]byte(r.Header.Get("Authorization")))
+ exp := sha256.Sum256([]byte("Bearer " + want))
+ if subtle.ConstantTimeCompare(got[:], exp[:]) != 1 {
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ fn(w, r)
+ }
+ }
+
+ mux.HandleFunc("PUT /schedule/{name}", authed(func(w http.ResponseWriter, r *http.Request) {
+ name := r.PathValue("name")
+ if !ValidName(name) {
+ http.Error(w, "name: want [a-z0-9][a-z0-9-]{0,31}", http.StatusBadRequest)
+ return
+ }
+ var body struct {
+ At string `json:"at"`
+ Path string `json:"path"`
+ }
+ if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&body); err != nil {
+ http.Error(w, "bad request: "+err.Error(), http.StatusBadRequest)
+ return
+ }
+ atT, err := time.Parse(time.RFC3339, body.At)
+ if err != nil {
+ http.Error(w, "at: want RFC3339, e.g. 2026-09-01T08:00:00Z", http.StatusBadRequest)
+ return
+ }
+ if atT.After(now().Add(MaxOnceAhead)) {
+ http.Error(w, "at: more than 400 days ahead", http.StatusBadRequest)
+ return
+ }
+ if err := ValidPath(body.Path); err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ existing, err := store.GetTimer(host, name)
+ switch {
+ case err == nil && existing.Kind != KindOnce:
+ http.Error(w, "name is a declared schedule", http.StatusConflict)
+ return
+ case err != nil && !errors.Is(err, os.ErrNotExist):
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ case err != nil:
+ n, cerr := store.CountTimers(host, KindOnce)
+ if cerr != nil {
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ }
+ if n >= MaxOncePerHost {
+ http.Error(w, "too many one-shot timers (limit 1000) — delete finished ones or use a schedule", http.StatusTooManyRequests)
+ return
+ }
+ }
+ sp := Spec{Kind: KindOnce, At: atT.UTC()}
+ t := registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: sp.String(), Path: body.Path, NextAt: atT.Unix(), CreatedAt: now().Unix()}
+ if err := store.UpsertTimer(t); err != nil {
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ }
+ row, err := store.GetTimer(host, name)
+ if err != nil {
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ }
+ writeTimer(w, *row)
+ }))
+
+ mux.HandleFunc("DELETE /schedule/{name}", authed(func(w http.ResponseWriter, r *http.Request) {
+ name := r.PathValue("name")
+ if !ValidName(name) {
+ http.Error(w, "bad name", http.StatusBadRequest)
+ return
+ }
+ if row, err := store.GetTimer(host, name); err == nil && row.Kind != KindOnce {
+ http.Error(w, "name is a declared schedule — remove it with carlos schedule rm", http.StatusConflict)
+ return
+ }
+ if err := store.DeleteTimer(host, name); err != nil {
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+ }))
+
+ mux.HandleFunc("GET /schedule", authed(func(w http.ResponseWriter, r *http.Request) {
+ rows, err := store.TimersFor(host)
+ if err != nil {
+ http.Error(w, "registry error", http.StatusInternalServerError)
+ return
+ }
+ out := struct {
+ Timers []TimerStatus `json:"timers"`
+ }{Timers: []TimerStatus{}}
+ for _, row := range rows {
+ out.Timers = append(out.Timers, toStatus(row))
+ }
+ w.Header().Set("Content-Type", "application/json")
+ json.NewEncoder(w).Encode(out)
+ }))
+}
+
+// writeTimer writes one registry row as its wire TimerStatus (toStatus is
+// converge.go's — one row-to-status mapping for the whole package).
+func writeTimer(w http.ResponseWriter, r registry.Timer) {
+ w.Header().Set("Content-Type", "application/json")
+ json.NewEncoder(w).Encode(toStatus(r))
+}
diff --git a/internal/schedule/control_test.go b/internal/schedule/control_test.go
new file mode 100644
index 000000000..3349da0d0
--- /dev/null
+++ b/internal/schedule/control_test.go
@@ -0,0 +1,127 @@
+package schedule
+
+import (
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+func controlFixture(t *testing.T) (*registry.Registry, *http.ServeMux) {
+ t.Helper()
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ mux := http.NewServeMux()
+ Mount(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil }, func() time.Time { return at("2026-08-23T13:07:00Z") })
+ return reg, mux
+}
+
+func do(mux *http.ServeMux, method, path, auth, body string) *httptest.ResponseRecorder {
+ req := httptest.NewRequest(method, path, strings.NewReader(body))
+ if auth != "" {
+ req.Header.Set("Authorization", auth)
+ }
+ rec := httptest.NewRecorder()
+ mux.ServeHTTP(rec, req)
+ return rec
+}
+
+func TestControlRequiresTheHostsToken(t *testing.T) {
+ _, mux := controlFixture(t)
+ for _, auth := range []string{"", "Bearer wrong", "Basic tok"} {
+ if rec := do(mux, "PUT", "/schedule/r", auth, `{"at":"2026-09-01T08:00:00Z","path":"/r"}`); rec.Code != http.StatusUnauthorized {
+ t.Errorf("auth %q → %d, want 401", auth, rec.Code)
+ }
+ if rec := do(mux, "GET", "/schedule", auth, ""); rec.Code != http.StatusUnauthorized {
+ t.Errorf("GET auth %q → %d", auth, rec.Code)
+ }
+ }
+}
+
+func TestControlPutGetDelete(t *testing.T) {
+ reg, mux := controlFixture(t)
+ rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/jobs/remind"}`)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("put = %d %s", rec.Code, rec.Body)
+ }
+ var ts TimerStatus
+ json.Unmarshal(rec.Body.Bytes(), &ts)
+ if ts.Kind != KindOnce || ts.NextAt != at("2026-09-01T08:00:00Z").Unix() || ts.Path != "/jobs/remind" || ts.Spec != "at "+itoa(ts.NextAt) {
+ t.Fatalf("put body = %+v", ts)
+ }
+ tm, _ := reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if tm == nil || tm.Kind != KindOnce {
+ t.Fatalf("row = %+v", tm)
+ }
+ // Upsert by name moves the time.
+ do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-02T08:00:00Z","path":"/jobs/remind"}`)
+ tm, _ = reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if tm.NextAt != at("2026-09-02T08:00:00Z").Unix() {
+ t.Fatalf("upsert did not move: %+v", tm)
+ }
+ rec = do(mux, "GET", "/schedule", "Bearer tok", "")
+ var list struct{ Timers []TimerStatus }
+ json.Unmarshal(rec.Body.Bytes(), &list)
+ if len(list.Timers) != 1 || list.Timers[0].Name != "remind" {
+ t.Fatalf("list = %+v", list)
+ }
+ if rec := do(mux, "DELETE", "/schedule/remind", "Bearer tok", ""); rec.Code != http.StatusNoContent {
+ t.Fatalf("delete = %d", rec.Code)
+ }
+ if rec := do(mux, "DELETE", "/schedule/remind", "Bearer tok", ""); rec.Code != http.StatusNoContent {
+ t.Fatalf("second delete = %d, want 204 (idempotent)", rec.Code)
+ }
+}
+
+func TestControlRefusals(t *testing.T) {
+ reg, mux := controlFixture(t)
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: 1})
+ cases := []struct {
+ name, body string
+ want int
+ }{
+ {"sync", `{"at":"2026-09-01T08:00:00Z","path":"/p"}`, http.StatusConflict},
+ {"Bad", `{"at":"2026-09-01T08:00:00Z","path":"/p"}`, http.StatusBadRequest},
+ {"ok", `{"at":"2026-09-01T08:00:00Z","path":"p"}`, http.StatusBadRequest},
+ {"ok", `{"at":"soon","path":"/p"}`, http.StatusBadRequest},
+ {"ok", `{"at":"2028-09-01T08:00:00Z","path":"/p"}`, http.StatusBadRequest}, // > 400 days ahead
+ {"ok", `not json`, http.StatusBadRequest},
+ }
+ for _, c := range cases {
+ if rec := do(mux, "PUT", "/schedule/"+c.name, "Bearer tok", c.body); rec.Code != c.want {
+ t.Errorf("PUT %s %s → %d, want %d (%s)", c.name, c.body, rec.Code, c.want, rec.Body)
+ }
+ }
+ // A past `at` is accepted and fires on the next tick.
+ if rec := do(mux, "PUT", "/schedule/past", "Bearer tok", `{"at":"2020-01-01T00:00:00Z","path":"/p"}`); rec.Code != http.StatusOK {
+ t.Fatalf("past at → %d", rec.Code)
+ }
+ if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "past"); tm.NextAt != at("2020-01-01T00:00:00Z").Unix() {
+ t.Fatalf("past row = %+v", tm)
+ }
+}
+
+func TestControlCapsOncePerHost(t *testing.T) {
+ reg, mux := controlFixture(t)
+ for i := 0; i < MaxOncePerHost; i++ {
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "n" + itoa(int64(i)), Kind: KindOnce, Spec: "at 1", Path: "/p", NextAt: 1})
+ }
+ if rec := do(mux, "PUT", "/schedule/over", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/p"}`); rec.Code != http.StatusTooManyRequests {
+ t.Fatalf("over cap → %d", rec.Code)
+ }
+ // Replacing an existing name is not a new row and must still work.
+ if rec := do(mux, "PUT", "/schedule/n1", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/p"}`); rec.Code != http.StatusOK {
+ t.Fatalf("replace at cap → %d", rec.Code)
+ }
+}
+
+func itoa(n int64) string { return strconv.FormatInt(n, 10) }