amadan

carlos / platform Public

host, appconfig, release, schedule, registry: the startup selection carried through config, release and timer writers (claim Plan 6 Task 9)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 1ecd7703adb5a58b494a63a7351fdf5d7c842eaf parent 4b4692e
15 files changed, +1433 −119
  • internal/appconfig/claimsync.go +41 −0
  • internal/appconfig/claimsync_test.go +168 −0
  • internal/appconfig/sync.go +70 −14
  • internal/appconfig/syncexec.go +81 −18
  • internal/host/agent.go +96 −21
  • internal/host/claimentry_test.go +56 −0
  • internal/host/claimwiring.go +106 −0
  • internal/host/claimwiring_test.go +67 −0
  • internal/registry/claimtimers.go +81 −0
  • internal/release/adopt.go +50 −12
  • internal/release/claims.go +30 −0
  • internal/release/claims_test.go +124 −0
  • internal/schedule/claimintegration_test.go +224 −0
  • internal/schedule/converge.go +130 −22
  • internal/schedule/runner.go +109 −32
diff --git a/internal/appconfig/claimsync.go b/internal/appconfig/claimsync.go
new file mode 100644
index 000000000..cc5b8226a
--- /dev/null
+++ b/internal/appconfig/claimsync.go
@@ -0,0 +1,41 @@
+package appconfig
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "fmt"
+
+ "github.com/carlosframework/platform/internal/hostwrite"
+)
+
+var errHostWriteOptions = errors.New("appconfig: more than one hostwrite configuration")
+
+func hostWriteOption(options []hostwrite.Config) (hostwrite.Config, error) {
+ if len(options) > 1 {
+ return hostwrite.Config{}, errHostWriteOptions
+ }
+ if len(options) == 1 {
+ return options[0], nil
+ }
+ return hostwrite.Config{}, nil
+}
+
+func renderedIdentity(rendered []byte, defaultPresent, environmentPresent, instancePresent bool) hostwrite.Identity {
+ sum := sha256.Sum256(rendered)
+ return hostwrite.Identity{Config: fmt.Sprintf("sha256:%s;default=%t;environment=%t;instance=%t", hex.EncodeToString(sum[:]), defaultPresent, environmentPresent, instancePresent)}
+}
+
+func guarded(ctx context.Context, work hostwrite.Work, id hostwrite.Identity, fn func() error) error {
+ if ctx == nil || work == nil {
+ return hostwrite.ErrUnavailable
+ }
+ return work.With(ctx, id, fn)
+}
+
+type legacyHostWork struct{}
+
+func (legacyHostWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
+ return fn()
+}
diff --git a/internal/appconfig/claimsync_test.go b/internal/appconfig/claimsync_test.go
new file mode 100644
index 000000000..77da13bf9
--- /dev/null
+++ b/internal/appconfig/claimsync_test.go
@@ -0,0 +1,168 @@
+package appconfig
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "sync"
+ "sync/atomic"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type pausedConfigWork struct {
+ valid atomic.Bool
+ once sync.Once
+ entered chan struct{}
+ resume chan struct{}
+}
+
+func newPausedConfigWork() *pausedConfigWork {
+ w := &pausedConfigWork{entered: make(chan struct{}), resume: make(chan struct{})}
+ w.valid.Store(true)
+ return w
+}
+
+func (w *pausedConfigWork) With(_ context.Context, id hostwrite.Identity, fn func() error) error {
+ if id.Config == "" {
+ return errors.New("missing rendered identity")
+ }
+ w.once.Do(func() {
+ close(w.entered)
+ <-w.resume
+ })
+ if !w.valid.Load() {
+ return errors.New("selection changed")
+ }
+ return fn()
+}
+
+func TestSyncExecChangedSelectionBeforeMutationWritesNothing(t *testing.T) {
+ dir := t.TempDir()
+ host := "app.example"
+ rt := registry.Route{Host: host, Account: "bab", App: "app"}
+ w := newPausedConfigWork()
+ bounces := atomic.Int64{}
+ done := make(chan error, 1)
+ go func() {
+ done <- syncExecHostClaimed(t.Context(), w, rt, &Manifest{Vars: map[string]Var{"A": {Kind: "env", Value: "one"}}}, Key("bab", "app"), nil, dir, func(string) error {
+ bounces.Add(1)
+ return nil
+ }, "", "", nil)
+ }()
+ <-w.entered
+ w.valid.Store(false)
+ close(w.resume)
+ if err := <-done; err == nil {
+ t.Fatal("stale config write succeeded")
+ }
+ if _, err := os.Stat(filepath.Join(dir, host+configEnvSuffix)); !os.IsNotExist(err) {
+ t.Fatalf("config file exists after refusal: %v", err)
+ }
+ if bounces.Load() != 0 {
+ t.Fatal("stale config bounced process")
+ }
+}
+
+func TestSyncExecChangedSelectionBeforeRemovalPreservesBytes(t *testing.T) {
+ dir := t.TempDir()
+ host := "app.example"
+ path := filepath.Join(dir, host+configEnvSuffix)
+ if err := os.WriteFile(path, []byte("KEEP=1\n"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ w := newPausedConfigWork()
+ done := make(chan error, 1)
+ go func() {
+ done <- syncExecHostClaimed(t.Context(), w, registry.Route{Host: host, Account: "bab", App: "app"}, nil, Key("bab", "app"), nil, dir, func(string) error { return nil }, "", "", nil)
+ }()
+ <-w.entered
+ w.valid.Store(false)
+ close(w.resume)
+ if err := <-done; err == nil {
+ t.Fatal("stale removal succeeded")
+ }
+ if got, err := os.ReadFile(path); err != nil || string(got) != "KEEP=1\n" {
+ t.Fatalf("preserved config = %q, %v", got, err)
+ }
+}
+
+func TestSyncEnabledMissingProviderRefusesBeforeConfigReads(t *testing.T) {
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ if err := reg.Upsert(registry.Route{Host: "app.example", Kind: "instance", Account: "bab", App: "app", Backing: registry.BackingUnit, Status: "ready"}); err != nil {
+ t.Fatal(err)
+ }
+ err = Sync(t.Context(), reg, nil, nil, t.TempDir(), func(string) error { return nil }, nil, hostwrite.Config{Enabled: true})
+ if !errors.Is(err, hostwrite.ErrUnavailable) {
+ t.Fatalf("Sync missing provider = %v", err)
+ }
+}
+
+func TestSyncRejectsMultipleHostWriteConfigurations(t *testing.T) {
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ if err := Sync(context.Background(), reg, nil, nil, t.TempDir(), nil, nil, hostwrite.Config{}, hostwrite.Config{}); !errors.Is(err, errHostWriteOptions) {
+ t.Fatalf("multiple configs = %v", err)
+ }
+}
+
+func TestClaimDisabledConfigErrorContinuesHostsAndReaps(t *testing.T) {
+ for _, exec := range []bool{false, true} {
+ t.Run(map[bool]string{false: "unit", true: "exec"}[exec], func(t *testing.T) {
+ reg := testRegistry(t)
+ defer reg.Close()
+ store := release.NewDirStore(t.TempDir())
+ dir := t.TempDir()
+ if err := SetEnv(t.Context(), store, "bab", "app", map[string]string{"VALUE": "new"}); err != nil {
+ t.Fatal(err)
+ }
+ suffix := ".env"
+ backing := registry.BackingUnit
+ if exec {
+ suffix = configEnvSuffix
+ backing = ""
+ }
+ for _, host := range []string{"a.example", "b.example"} {
+ if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Account: "bab", App: "app", Backing: backing, Status: "ready", Socket: "/unused.sock"}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ // A directory at the temporary-file path produces a deterministic error even as root.
+ if err := os.Mkdir(filepath.Join(dir, "a.example"+suffix+".tmp"), 0700); err != nil {
+ t.Fatal(err)
+ }
+ stale := filepath.Join(dir, "gone.example"+suffix)
+ if err := os.WriteFile(stale, []byte("old"), 0600); err != nil {
+ t.Fatal(err)
+ }
+ calls := map[string]int{}
+ restart := func(host string) error { calls[host]++; return nil }
+ var err error
+ if exec {
+ err = SyncExec(t.Context(), reg, store, nil, dir, restart, "", nil)
+ } else {
+ err = Sync(t.Context(), reg, store, nil, dir, restart, nil)
+ }
+ if err != nil {
+ t.Fatalf("legacy pass aborted: %v", err)
+ }
+ if _, err := os.Stat(filepath.Join(dir, "b.example"+suffix)); err != nil || calls["b.example"] != 1 {
+ t.Fatalf("later host: %v calls=%v", err, calls)
+ }
+ if _, err := os.Stat(stale); !os.IsNotExist(err) {
+ t.Fatalf("reap skipped: %v", err)
+ }
+ })
+ }
+}
diff --git a/internal/appconfig/sync.go b/internal/appconfig/sync.go
index 93ed75684..e242de8c8 100644
--- a/internal/appconfig/sync.go
+++ b/internal/appconfig/sync.go
@@ -20,6 +20,7 @@ import (
"strings"
"time"
+ "github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
@@ -47,11 +48,26 @@ const DefaultEnvDir = "/etc/carlos/apps"
// route's config-adoption acknowledgement (spec §5, configack.go) — nil
// means no acks, the shape every test and the exec-only loops that never
// touch /etc/carlos/apps use.
-func Sync(ctx context.Context, reg *registry.Registry, s release.Store, priv *ecdh.PrivateKey, envDir string, restart func(host string) error, ack func(ConfigAck)) error {
+func Sync(ctx context.Context, reg *registry.Registry, s release.Store, priv *ecdh.PrivateKey, envDir string, restart func(host string) error, ack func(ConfigAck), options ...hostwrite.Config) error {
routes, err := reg.List()
if err != nil {
return err
}
+ hostWrites, err := hostWriteOption(options)
+ if err != nil {
+ return err
+ }
+ selected := map[string]hostwrite.Work{}
+ for _, rt := range routes {
+ if !syncable(rt) || rt.Backing != registry.BackingUnit {
+ continue
+ }
+ work, selectErr := hostWrites.Select(ctx, rt.Host, "config")
+ if selectErr != nil {
+ return fmt.Errorf("%s: select config work: %w", rt.Host, selectErr)
+ }
+ selected[rt.Host] = work
+ }
type groupKey struct{ acct, app, env string }
// A resolution is one of three states, and they are NOT interchangeable:
// a manifest to converge to; "this app was never configured" (nil, nil),
@@ -257,7 +273,9 @@ func Sync(ctx context.Context, reg *registry.Registry, s release.Store, priv *ec
fmt.Printf(" %s: exec-backed — config delivered by the agent's in-process loop to /data/<host>.config.env, not this pass\n", rt.Host)
continue
}
- syncHost(rt, m, defaultMissing, priv, envDir, restart, instETag, ack)
+ if err := syncHostClaimed(ctx, selected[rt.Host], rt, m, defaultMissing, priv, envDir, restart, instETag, ack); err != nil && hostWrites.Enabled {
+ return err
+ }
}
// An empty route list is never authority to delete anything:
// registry.Open silently creates an empty database when handed a wrong
@@ -267,10 +285,36 @@ func Sync(ctx context.Context, reg *registry.Registry, s release.Store, priv *ec
if len(routes) == 0 {
return nil
}
- reap(routes, envDir)
+ if !hostWrites.Enabled {
+ reap(routes, envDir)
+ } else {
+ reapSelectedEnvTemps(ctx, envDir, selected)
+ }
return nil
}
+func reapSelectedEnvTemps(ctx context.Context, envDir string, selected map[string]hostwrite.Work) {
+ ents, err := os.ReadDir(envDir)
+ if err != nil {
+ return
+ }
+ for _, entry := range ents {
+ if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".env.tmp") {
+ continue
+ }
+ host := strings.TrimSuffix(entry.Name(), ".env.tmp")
+ work := selected[host]
+ if work == nil {
+ continue
+ }
+ if err := guarded(ctx, work, renderedIdentity(nil, false, false, false), func() error {
+ return os.Remove(filepath.Join(envDir, entry.Name()))
+ }); err != nil {
+ fmt.Printf(" %s: remove stale env temp file: %v\n", entry.Name(), err)
+ }
+ }
+}
+
// reap deletes <envDir>/<host>.env for any host with no route row at all —
// a decommissioned route's env file holds decrypted secrets and nothing
// else on the box removes it. Keyed off the full route list, not the
@@ -439,19 +483,23 @@ func routeBinding(rt registry.Route) Binding {
// since nothing was delivered on that pass and the ack already on file (if
// any) from a previous pass still describes what the route is running.
func syncHost(rt registry.Route, m *Manifest, defaultMissing bool, priv *ecdh.PrivateKey, envDir string, restart func(host string) error, instETag string, ack func(ConfigAck)) {
+ _ = syncHostClaimed(context.Background(), legacyHostWork{}, rt, m, defaultMissing, priv, envDir, restart, instETag, ack)
+}
+
+func syncHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.Route, m *Manifest, defaultMissing bool, priv *ecdh.PrivateKey, envDir string, restart func(host string) error, instETag string, ack func(ConfigAck)) error {
// The host is joined into a root-written path; a registry row carrying
// a separator must never place a file outside envDir. Same trust level
// as release's live/<host> symlinks, but this one is one line to pin.
if rt.Host == "" || rt.Host == "." || rt.Host == ".." || rt.Host != filepath.Base(rt.Host) {
fmt.Printf(" %q: not a bare hostname, refusing to derive an env file path from it\n", rt.Host)
- return
+ return nil
}
path := filepath.Join(envDir, rt.Host+".env")
cur, err := os.ReadFile(path)
existed := err == nil
if err != nil && !os.IsNotExist(err) {
fmt.Printf(" %s: read %s: %v\n", rt.Host, path, err)
- return
+ return nil
}
// The same protection an UNBOUND route gets from Sync's "no manifest
// object at all ⇒ leave the file alone" skip, applied to the DEFAULT half
@@ -502,19 +550,19 @@ func syncHost(rt registry.Route, m *Manifest, defaultMissing bool, priv *ecdh.Pr
"refusing to replace %s, which this pass cannot parse as KEY=value and so cannot prove is safe to overwrite; "+
"import its config (`carlos env set --app %s KEY=value ...`) or remove the file\n",
rt.Host, rt.App, alone, path, rt.App)
- return
+ return nil
case len(lost) > 0:
fmt.Printf(" %s: %s has no default config bundle, so the merged view is %s ALONE — "+
"refusing to drop %d var(s) already in %s (%s); import them (`carlos env set --app %s KEY=value ...`) "+
"and this route converges on the next pass, or remove the file if it is dead\n",
rt.Host, rt.App, alone, len(lost), path, strings.Join(lost, ", "), rt.App)
- return
+ return nil
}
}
rendered, err := RenderEnv(m, priv, routeBinding(rt))
if err != nil {
fmt.Printf(" %s: render config: %v\n", rt.Host, err)
- return
+ return nil
}
// Spec §7.3: deliver CARLOS_PURGE=header to unit tenants the same way
// the exec backend already does for exec-backed ones (withoutVars /
@@ -537,17 +585,18 @@ func syncHost(rt registry.Route, m *Manifest, defaultMissing bool, priv *ecdh.Pr
// feature landing, not a side effect of it.
rendered = applyAgentVars(rendered)
if existed && bytes.Equal(cur, rendered) {
- return // up to date — stay quiet on the periodic tick
+ return nil // up to date — stay quiet on the periodic tick
}
+ id := renderedIdentity(rendered, !defaultMissing, rt.Environment != "", instETag != "")
tmp := path + ".tmp"
- if err := os.WriteFile(tmp, rendered, 0o600); err != nil {
+ if err := guarded(ctx, work, id, func() error { return os.WriteFile(tmp, rendered, 0o600) }); err != nil {
fmt.Printf(" %s: write %s: %v\n", rt.Host, path, err)
- return
+ return err
}
- if err := os.Rename(tmp, path); err != nil {
- os.Remove(tmp) // don't leave rendered secrets behind
+ if err := guarded(ctx, work, id, func() error { return os.Rename(tmp, path) }); err != nil {
+ _ = guarded(ctx, work, id, func() error { return os.Remove(tmp) })
fmt.Printf(" %s: rename %s: %v\n", rt.Host, path, err)
- return
+ return err
}
// The config-adoption ack (spec §5): built from what THIS pass actually
// rendered and did, not re-derived later from the file on disk — the
@@ -571,6 +620,9 @@ func syncHost(rt registry.Route, m *Manifest, defaultMissing bool, priv *ecdh.Pr
// unit's try-restart returns success with nothing actually restarted,
// and reporting that as an ordinary "restarted" would tell a member
// their sleeping route already runs a key it will only pick up on wake.
+ if err := guarded(ctx, work, id, func() error { return nil }); err != nil {
+ return err
+ }
switch err := restart(rt.Host); {
case err != nil:
fmt.Printf(" %s: config updated but restart failed: %v\n", rt.Host, err)
@@ -583,6 +635,10 @@ func syncHost(rt registry.Route, m *Manifest, defaultMissing bool, priv *ecdh.Pr
fmt.Printf(" %s: config updated, restarted\n", rt.Host)
}
if ack != nil {
+ if err := guarded(ctx, work, id, func() error { return nil }); err != nil {
+ return err
+ }
ack(a)
}
+ return nil
}
diff --git a/internal/appconfig/syncexec.go b/internal/appconfig/syncexec.go
index 694f88863..3b13de308 100644
--- a/internal/appconfig/syncexec.go
+++ b/internal/appconfig/syncexec.go
@@ -45,6 +45,7 @@ import (
"strings"
"time"
+ "github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/instancefs"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
@@ -92,11 +93,26 @@ func ConfigEnvPath(dataDir string, rt registry.Route) string {
// route whose file is REMOVED (manifest absent) gets no ack: there is
// nothing for a route to have adopted once its config is gone, and a
// removal is not one of spec §5's three delivery shapes.
-func SyncExec(ctx context.Context, reg *registry.Registry, s release.Store, priv *ecdh.PrivateKey, dataDir string, bounce func(host string) error, appsDomain string, ack func(ConfigAck)) error {
+func SyncExec(ctx context.Context, reg *registry.Registry, s release.Store, priv *ecdh.PrivateKey, dataDir string, bounce func(host string) error, appsDomain string, ack func(ConfigAck), options ...hostwrite.Config) error {
routes, err := reg.List()
if err != nil {
return err
}
+ hostWrites, err := hostWriteOption(options)
+ if err != nil {
+ return err
+ }
+ selected := map[string]hostwrite.Work{}
+ for _, rt := range routes {
+ if !execSyncable(rt) && !execSidecarCandidate(rt) && !execAddrCandidate(rt) {
+ continue
+ }
+ work, selectErr := hostWrites.Select(ctx, rt.Host, "config")
+ if selectErr != nil {
+ return fmt.Errorf("%s: select config work: %w", rt.Host, selectErr)
+ }
+ selected[rt.Host] = work
+ }
// hostSet names every route on this box, in or out of scope, so the
// namespace-collision guard below sees a route x.config even though
// x.config itself may never reach syncExecHost. Built once: the guard
@@ -569,7 +585,9 @@ func SyncExec(ctx context.Context, reg *registry.Registry, s release.Store, priv
if m != nil || inst != nil {
m = applyInstanceOverlay(k.app, m, inst)
}
- syncExecHost(rt, m, Key(k.acct, k.app), priv, dataDir, hostBounce, header, instETag, ack)
+ if err := syncExecHostClaimed(ctx, selected[rt.Host], rt, m, Key(k.acct, k.app), priv, dataDir, hostBounce, header, instETag, ack); err != nil && hostWrites.Enabled {
+ return err
+ }
}
// Stale-secrets notice (issue #18 design ruling, ratified 2026-08-03
// review): an addr-shaped route whose app is DEFINITIVELY not opted in
@@ -622,10 +640,36 @@ func SyncExec(ctx context.Context, reg *registry.Registry, s release.Store, priv
if len(routes) == 0 {
return nil
}
- reapConfigEnv(routes, dataDir)
+ if !hostWrites.Enabled {
+ reapConfigEnv(routes, dataDir)
+ } else {
+ reapSelectedConfigTemps(ctx, dataDir, selected)
+ }
return nil
}
+func reapSelectedConfigTemps(ctx context.Context, dataDir string, selected map[string]hostwrite.Work) {
+ ents, err := os.ReadDir(dataDir)
+ if err != nil {
+ return
+ }
+ for _, entry := range ents {
+ if entry.IsDir() || !strings.HasSuffix(entry.Name(), configEnvSuffix+".tmp") {
+ continue
+ }
+ host := strings.TrimSuffix(entry.Name(), configEnvSuffix+".tmp")
+ work := selected[host]
+ if work == nil {
+ continue
+ }
+ if err := guarded(ctx, work, renderedIdentity(nil, false, false, false), func() error {
+ return os.Remove(filepath.Join(dataDir, entry.Name()))
+ }); err != nil {
+ fmt.Printf(" %s: remove stale config temp file: %v\n", entry.Name(), err)
+ }
+ }
+}
+
// execRouteBase is the lifecycle-managed-instance shape every exec route
// kind shares: the unit path's syncable() exclusions, narrowed to
// non-unit backing (unit routes belong to the root adopt path). Neither
@@ -1089,21 +1133,32 @@ const execAddrConfigHeader = `# Written by carlos (SyncExec, issue #18). Carlos
// this platform never restarts at all, so that shape's cutover is a
// member acknowledgement, `carlos relay adopted`, instead).
func syncExecHost(rt registry.Route, m *Manifest, key string, priv *ecdh.PrivateKey, dataDir string, bounce func(host string) error, header, instETag string, ack func(ConfigAck)) {
+ _ = syncExecHostClaimed(context.Background(), legacyHostWork{}, rt, m, key, priv, dataDir, bounce, header, instETag, ack)
+}
+
+func syncExecHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.Route, m *Manifest, key string, priv *ecdh.PrivateKey, dataDir string, bounce func(host string) error, header, instETag string, ack func(ConfigAck)) error {
if rt.Host == "" || rt.Host == "." || rt.Host == ".." || rt.Host != filepath.Base(rt.Host) {
fmt.Printf(" %q: not a bare hostname, refusing to derive a config file path from it\n", rt.Host)
- return
+ return nil
}
path := ConfigEnvPath(dataDir, rt)
if m == nil {
- if err := os.Remove(path); err != nil {
+ id := renderedIdentity(nil, false, rt.Environment != "", instETag != "")
+ if err := guarded(ctx, work, id, func() error { return os.Remove(path) }); err != nil {
if !os.IsNotExist(err) {
fmt.Printf(" %s: remove config for %s (manifest absent): %v\n", rt.Host, key, err)
}
- return
+ if os.IsNotExist(err) {
+ return nil
+ }
+ return err
}
fmt.Printf(" %s: manifest %s absent, removed its config file\n", rt.Host, key)
+ if err := guarded(ctx, work, id, func() error { return nil }); err != nil {
+ return err
+ }
bounceLogged(rt.Host, bounce)
- return
+ return nil
}
// A nested row's current file lives on tenant-owned ground, so its read
// goes through instancefs the same way its write below does: a plain
@@ -1118,14 +1173,14 @@ func syncExecHost(rt registry.Route, m *Manifest, key string, priv *ecdh.Private
existed = err == nil
if err != nil && !os.IsNotExist(err) {
fmt.Printf(" %s: read %s: %v\n", rt.Host, path, err)
- return
+ return nil
}
} else {
cur, err = os.ReadFile(path)
existed = err == nil
if err != nil && !os.IsNotExist(err) {
fmt.Printf(" %s: read %s: %v\n", rt.Host, path, err)
- return
+ return nil
}
}
// Where sync.go has its defaultMissing/lostKeys guard, this path has
@@ -1151,14 +1206,15 @@ func syncExecHost(rt registry.Route, m *Manifest, key string, priv *ecdh.Private
rendered, err := RenderEnv(m, priv, routeBinding(rt))
if err != nil {
fmt.Printf(" %s: render config: %v\n", rt.Host, err)
- return
+ return nil
}
if header != "" {
rendered = append([]byte(header), rendered...)
}
if existed && bytes.Equal(cur, rendered) {
- return // up to date — stay quiet on the periodic tick
+ return nil // up to date — stay quiet on the periodic tick
}
+ id := renderedIdentity(rendered, true, rt.Environment != "", instETag != "")
// A nested row's config file lives inside its own instance directory,
// tenant-owned ground (isolated-exec design §3.2). So it is written
// through instancefs.WriteAtomic's O_EXCL temp name and openat2
@@ -1167,23 +1223,26 @@ func syncExecHost(rt registry.Route, m *Manifest, key string, priv *ecdh.Private
// directory is agent-owned throughout, so it keeps the plain
// temp-then-rename it always used.
if nestedRoute(dataDir, rt) {
- if err := instancefs.WriteAtomic(filepath.Dir(path), filepath.Base(path), rendered, 0o600); err != nil {
+ if err := guarded(ctx, work, id, func() error { return instancefs.WriteAtomic(filepath.Dir(path), filepath.Base(path), rendered, 0o600) }); err != nil {
fmt.Printf(" %s: write %s: %v\n", rt.Host, path, err)
- return
+ return err
}
} else {
tmp := path + ".tmp"
- if err := os.WriteFile(tmp, rendered, 0o600); err != nil {
+ if err := guarded(ctx, work, id, func() error { return os.WriteFile(tmp, rendered, 0o600) }); err != nil {
fmt.Printf(" %s: write %s: %v\n", rt.Host, path, err)
- return
+ return err
}
- if err := os.Rename(tmp, path); err != nil {
- os.Remove(tmp) // don't leave rendered secrets behind
+ if err := guarded(ctx, work, id, func() error { return os.Rename(tmp, path) }); err != nil {
+ _ = guarded(ctx, work, id, func() error { return os.Remove(tmp) })
fmt.Printf(" %s: rename %s: %v\n", rt.Host, path, err)
- return
+ return err
}
}
fmt.Printf(" %s: config updated\n", rt.Host)
+ if err := guarded(ctx, work, id, func() error { return nil }); err != nil {
+ return err
+ }
bounceErr := bounceLogged(rt.Host, bounce)
if ack != nil {
a := ConfigAck{Schema: ConfigAckSchema, Host: rt.Host, Shape: shapeOf(rt), At: time.Now(), InstanceConfigETag: instETag}
@@ -1206,8 +1265,12 @@ func syncExecHost(rt registry.Route, m *Manifest, key string, priv *ecdh.Private
a.Restarted = true
}
}
+ if err := guarded(ctx, work, id, func() error { return nil }); err != nil {
+ return err
+ }
ack(a)
}
+ return nil
}
// bounceLogged calls bounce — best-effort in the sense that a failure is
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 328c83bef..0d0c0f7d0 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -178,6 +178,7 @@ import (
"github.com/carlosframework/platform/internal/edge"
"github.com/carlosframework/platform/internal/edgesfile"
"github.com/carlosframework/platform/internal/fleetapi"
+ "github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/instancectl"
"github.com/carlosframework/platform/internal/instancefs"
"github.com/carlosframework/platform/internal/lease"
@@ -200,7 +201,22 @@ func dataDir() string {
// edge.Run's (-registry, -dev, -http, -https, ...): the args pass through to
// edge.RunWithActivator, which owns the serving loop and blocks until
// SIGTERM/SIGINT.
-func RunAgent(args []string) {
+func RunAgent(args []string) { runAgent(args, productionClaimFactory) }
+
+// productionClaimFactory is the fail-closed production seam until Plan 8
+// supplies verified roster, evidence, lease, and move providers. Startup does
+// not call it while CARLOS_HOST_CLAIM is disabled.
+func productionClaimFactory(context.Context) (ClaimProviders, error) {
+ return ClaimProviders{}, hostwrite.ErrUnavailable
+}
+
+func runAgent(args []string, buildClaims ClaimFactory) {
+ agentCtx, cancelAgent := context.WithCancel(context.Background())
+ defer cancelAgent()
+ claims, err := newClaimWiring(agentCtx, os.Getenv, buildClaims)
+ if err != nil {
+ log.Fatalf("agent: host claims: %v", err)
+ }
dd := dataDir()
regPath := peekRegistryPath(args, dd)
// A second handle on the same registry file the edge opens — the registry
@@ -249,6 +265,9 @@ func RunAgent(args []string) {
if err != nil {
log.Fatalf("agent: %v", err)
}
+ if err := configureWakeClaims(claims, cfg); err != nil {
+ log.Fatalf("agent: host claims: %v", err)
+ }
// Fleet box identity (customer-fleets plan, Task 6 fix round — controller
// ruling): resolved ONCE, here, before the activator is even constructed.
@@ -467,12 +486,27 @@ func RunAgent(args []string) {
}
schedStore, schedErr := openDeployStore(context.Background())
sd := armSchedule(reg, schedStore, schedErr, svc, sink, dd)
+ if err := configureScheduleClaims(claims, sd); err != nil {
+ log.Fatalf("agent: host claims: %v", err)
+ }
if sd != nil {
- sd.runner.Start(context.Background()) // runs until the process ends, like every loop below
+ sd.runner.Start(agentCtx)
+ }
+ serialClaims := make(chan struct{}, 1)
+ if err := claims.startRegional(agentCtx, serialClaims, func(err error) { log.Printf("agent: regional claims: %v", err) }); err != nil {
+ log.Fatalf("agent: regional claims: %v", err)
}
+ defer func() {
+ cancelAgent()
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ if err := claims.close(ctx); err != nil {
+ log.Printf("agent: regional claims shutdown: %v", err)
+ }
+ }()
- go adoptLoop(reg, svc, sd)
- go fastConvergeLoop(reg, svc, sd)
+ go adoptLoop(reg, svc, sd, claims.Writes)
+ go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims)
if logSink != nil {
go logshipLoop(logSink, status.Label(readMachineID()))
}
@@ -509,8 +543,8 @@ func RunAgent(args []string) {
// The sweep reads the registry only through this closure — the stamped
// bespoke units to sample beside the carlos- ones, re-read each tick.
go metricsLoop(cgm.Root(), func() []string { return stampedUnitNames(reg) })
- go configSyncLoop(reg, svc, dd)
- go configGenLoop(reg, svc, dd)
+ go configSyncLoop(reg, svc, dd, claims.Writes)
+ go configGenLoop(reg, svc, dd, claims.Writes)
go socketAdvisoryLoop(reg, svc)
// The same serving loop as `carlos edge`, with the activator injected.
@@ -601,6 +635,34 @@ func RunAgent(args []string) {
stopHousekeeper()
}
+func configureWakeClaims(w *ClaimWiring, cfg *activator.Config) error {
+ if w == nil {
+ return hostwrite.ErrUnavailable
+ }
+ if !w.Writes.Enabled {
+ return nil
+ }
+ if cfg == nil || w.Providers.Wake == nil || w.Writes.Selector == nil {
+ return hostwrite.ErrUnavailable
+ }
+ cfg.HostWrites, cfg.Claims = w.Writes, w.Providers.Wake
+ return nil
+}
+
+func configureScheduleClaims(w *ClaimWiring, sd *scheduleDeps) error {
+ if w == nil {
+ return hostwrite.ErrUnavailable
+ }
+ if !w.Writes.Enabled {
+ return nil
+ }
+ if sd == nil || sd.runner == nil || sd.conv == nil || w.Writes.Selector == nil {
+ return hostwrite.ErrUnavailable
+ }
+ sd.runner.HostWrites, sd.conv.HostWrites = w.Writes, w.Writes
+ return nil
+}
+
// --- Task 8: the agent socket's retry/backoff, and the worker boot pass ----
// workersToStart is the boot pass's selection: a sidecar that does not
@@ -804,7 +866,7 @@ func materializeHook(reg *registry.Registry, dd string) func(context.Context, st
// sd carries the scheduled-work converger; this loop is its HOURLY floor (the
// fast one rides the serial in fastConvergeLoop). sd.converge is a no-op when
// the box has no deploy store — the agent logged that once at arming.
-func adoptLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps) {
+func adoptLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps, writes hostwrite.Config) {
ctx := context.Background()
store, err := openDeployStore(ctx)
if err != nil {
@@ -868,7 +930,7 @@ func adoptLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps)
if err := release.Pass(ctx, reg, store, pub, optDir, restart,
release.BakeDir(dataDir()), release.OnStageBlocked(stageRequester(dataDir())),
release.RestartHibernating(restart),
- release.WithStartUnit(makeStartRoute(reg, systemctlStartUnit))); err != nil {
+ release.WithStartUnit(makeStartRoute(reg, systemctlStartUnit)), release.WithHostWrites(writes)); err != nil {
log.Printf("agent: adopt pass: %v", err)
}
// The schedule converge's hourly floor (schedule spec §6.2), AFTER the
@@ -890,7 +952,7 @@ func adoptLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps)
// the readers; the root adopt path keeps the unit-backed half
// (/etc/carlos/apps needs root). Inert until CARLOS_SECRETS_KEY is set,
// the same convention as every other optional loop.
-func configSyncLoop(reg *registry.Registry, svc *activator.Service, dd string) {
+func configSyncLoop(reg *registry.Registry, svc *activator.Service, dd string, writes hostwrite.Config) {
raw := os.Getenv("CARLOS_SECRETS_KEY")
if raw == "" {
log.Printf("agent: config sync loop disabled: CARLOS_SECRETS_KEY not set")
@@ -912,7 +974,7 @@ func configSyncLoop(reg *registry.Registry, svc *activator.Service, dd string) {
for {
time.Sleep(delay)
delay = time.Hour + rand.N(10*time.Minute)
- if err := syncExecSerialized(ctx, reg, store, priv, dd, svc); err != nil {
+ if err := syncExecSerialized(ctx, reg, store, priv, dd, svc, writes); err != nil {
log.Printf("agent: config sync: %v", err)
}
}
@@ -933,10 +995,10 @@ var configSyncMu sync.Mutex
// syncExecSerialized is the one way either config loop runs SyncExec — the
// mutex lives in a shared wrapper, not at the call sites, so a future third
// caller cannot forget it.
-func syncExecSerialized(ctx context.Context, reg *registry.Registry, store release.Store, priv *ecdh.PrivateKey, dd string, svc *activator.Service) error {
+func syncExecSerialized(ctx context.Context, reg *registry.Registry, store release.Store, priv *ecdh.PrivateKey, dd string, svc *activator.Service, writes ...hostwrite.Config) error {
configSyncMu.Lock()
defer configSyncMu.Unlock()
- return appconfig.SyncExec(ctx, reg, store, priv, dd, bounce(svc), os.Getenv("PLATFORM_APPS_DOMAIN"), configAckFunc(ctx, store, "agent"))
+ return appconfig.SyncExec(ctx, reg, store, priv, dd, bounce(svc), os.Getenv("PLATFORM_APPS_DOMAIN"), configAckFunc(ctx, store, "agent"), writes...)
}
// configAckFunc builds the callback config sync's two passes (the root
@@ -1328,7 +1390,7 @@ func (t *stageThrottle) request(host, ver string) {
//
// Thin and untested by convention, like adoptLoop/statusLoop — the tick's
// whole decision lives in genPollPass, which is tested.
-func configGenLoop(reg *registry.Registry, svc *activator.Service, dd string) {
+func configGenLoop(reg *registry.Registry, svc *activator.Service, dd string, writes hostwrite.Config) {
raw := os.Getenv("CARLOS_SECRETS_KEY")
if raw == "" {
log.Printf("agent: config generation loop disabled: CARLOS_SECRETS_KEY not set")
@@ -1383,7 +1445,7 @@ func configGenLoop(reg *registry.Registry, svc *activator.Service, dd string) {
if !execChanged {
continue
}
- if err := syncExecSerialized(ctx, reg, store, priv, dd, svc); err != nil {
+ if err := syncExecSerialized(ctx, reg, store, priv, dd, svc, writes); err != nil {
log.Printf("agent: config generation sync: %v", err)
}
}
@@ -1436,6 +1498,7 @@ type fastDeps struct {
step map[string]time.Duration
restartHold map[string]*restartFailure // hosts held out of the sweep after a failed restart; see noteRestartFailed
serialBroken bool // last ReadSerial failed and was logged; reset on success
+ serialNotify chan<- struct{}
}
// restartFailure is one host's cross-pass memo of a version its restart
@@ -1841,6 +1904,12 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
if d.scheduleLeg != nil {
d.scheduleLeg(ctx)
}
+ if d.serialNotify != nil {
+ select {
+ case d.serialNotify <- struct{}{}:
+ default:
+ }
+ }
// The platform self-update leg rides the move too, last: a
// `carlos system update` is a console action like any other, and
// this is what makes a promote noticed in seconds instead of at
@@ -1905,7 +1974,7 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
//
// Thin and untested by convention, like adoptLoop/statusLoop — the tick's
// whole decision lives in fastTick/fastMembers, which are tested.
-func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps) {
+func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps, writes hostwrite.Config, serialNotify chan<- struct{}) {
ctx := context.Background()
store, err := openDeployStore(ctx)
if err != nil {
@@ -1927,11 +1996,11 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
// observer (observeRestart): a failed restart holds its host out of the
// sweep instead of the sweep re-cycling the service every tick. The
// remaining fields are filled in right below.
- d := &fastDeps{reg: reg, store: store, pub: pub, optDir: optDir}
+ d := &fastDeps{reg: reg, store: store, pub: pub, optDir: optDir, serialNotify: serialNotify}
restart := d.observeRestart(makeRestartRoute(reg, systemctlTryRestart, systemctlTryRestartUnit, adoptBounceFn(svc)))
// Built once, used by both pass shapes below — see fastPassOpts for the
// DO-NOT list and why the stage requester is per-loop, not per-tick.
- opts := fastPassOpts(dd, stageRequester(dd), restart, makeStartRoute(reg, systemctlStartUnit))
+ opts := append(fastPassOpts(dd, stageRequester(dd), restart, makeStartRoute(reg, systemctlStartUnit)), release.WithHostWrites(writes))
// The restart check's memory (issue #113, narrowed requests added by
// discussion #16), seeded ONCE here — two reads per app at arming (the
// app-wide generation and the narrowed request log) and never again except
@@ -2007,7 +2076,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
}
mark("adopt")
if priv != nil {
- if err := syncExecSerialized(ctx, reg, store, priv, dd, svc); err != nil {
+ if err := syncExecSerialized(ctx, reg, store, priv, dd, svc, writes); err != nil {
log.Printf("agent: fast converge config sync: %v", err)
}
mark("config")
@@ -3453,7 +3522,13 @@ func openStatusStore(ctx context.Context, getenv func(string) string) (release.S
// capability converge, or nag about canaries. It deliberately needs NO release
// key and NO secrets key — restarting verifies no signature and renders no
// config — so a box missing either still honours it.
-func RunAdopt(args []string) {
+func RunAdopt(args []string) { runAdopt(args, productionClaimFactory) }
+
+func runAdopt(args []string, buildClaims ClaimFactory) {
+ claims, err := newClaimWiring(context.Background(), os.Getenv, buildClaims)
+ if err != nil {
+ log.Fatalf("adopt: host claims: %v", err)
+ }
fs := flag.NewFlagSet("adopt", flag.ExitOnError)
regPath := fs.String("registry", filepath.Join(dataDir(), "registry.db"), "registry database")
configOnly := fs.Bool("config-only", false, "sync unit-backed config and restart changed routes only: no release adoption, no canary nag")
@@ -3573,7 +3648,7 @@ func RunAdopt(args []string) {
secretsPriv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY")
if err != nil {
log.Printf("adopt: config sync disabled: %v", err)
- } else if err := appconfig.Sync(ctx, reg, store, secretsPriv, appconfig.DefaultEnvDir, restart, configAckFunc(ctx, store, "adopt")); err != nil {
+ } else if err := appconfig.Sync(ctx, reg, store, secretsPriv, appconfig.DefaultEnvDir, restart, configAckFunc(ctx, store, "adopt"), claims.Writes); err != nil {
log.Printf("adopt: config sync: %v", err)
}
}
@@ -3658,7 +3733,7 @@ func RunAdopt(args []string) {
// run). -stage-only keeps it for the same reason it keeps the converge:
// a relayed stage is exactly the request that performs such a restart.
passOpts := []release.PassOption{release.WithCapabilities(caps.Ops{}), release.BakeDir(dataDir()),
- release.WithStartUnit(makeStartRoute(reg, systemctlStartUnit))}
+ release.WithStartUnit(makeStartRoute(reg, systemctlStartUnit)), release.WithHostWrites(claims.Writes)}
if !*stageOnly {
passOpts = append(passOpts, release.NagExpiredCanaries(os.Stdout))
}
diff --git a/internal/host/claimentry_test.go b/internal/host/claimentry_test.go
new file mode 100644
index 000000000..618777f51
--- /dev/null
+++ b/internal/host/claimentry_test.go
@@ -0,0 +1,56 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/schedule"
+)
+
+type entrySelector struct{}
+
+func (entrySelector) Select(context.Context, string, string) (hostwrite.Work, error) { return nil, nil }
+
+func TestAgentClaimConsumersUseStartupWiring(t *testing.T) {
+ sel := entrySelector{}
+ w := &ClaimWiring{Writes: hostwrite.Config{Enabled: true, Selector: sel}, Providers: ClaimProviders{Wake: &activator.ClaimRuntime{}}}
+ cfg := &activator.Config{}
+ sd := &scheduleDeps{runner: &schedule.Runner{}, conv: &schedule.Converger{}}
+ if err := configureWakeClaims(w, cfg); err != nil {
+ t.Fatal(err)
+ }
+ if err := configureScheduleClaims(w, sd); err != nil {
+ t.Fatal(err)
+ }
+ if !cfg.HostWrites.Enabled || cfg.Claims != w.Providers.Wake || !sd.runner.HostWrites.Enabled || !sd.conv.HostWrites.Enabled {
+ t.Fatal("startup claim wiring was not propagated to every agent consumer")
+ }
+}
+
+func TestAgentClaimConsumersRefuseMissingEnabledDependencies(t *testing.T) {
+ w := &ClaimWiring{Writes: hostwrite.Config{Enabled: true, Selector: entrySelector{}}, Providers: ClaimProviders{Wake: &activator.ClaimRuntime{}}}
+ if err := configureWakeClaims(w, nil); !errors.Is(err, hostwrite.ErrUnavailable) {
+ t.Fatalf("wake error = %v", err)
+ }
+ if err := configureScheduleClaims(w, &scheduleDeps{runner: &schedule.Runner{}}); !errors.Is(err, hostwrite.ErrUnavailable) {
+ t.Fatalf("schedule error = %v", err)
+ }
+}
+
+func TestDisabledAgentClaimConsumersPreserveZeroConfig(t *testing.T) {
+ w := &ClaimWiring{}
+ cfg := &activator.Config{}
+ sd := &scheduleDeps{runner: &schedule.Runner{}, conv: &schedule.Converger{}}
+ if err := configureWakeClaims(w, cfg); err != nil {
+ t.Fatal(err)
+ }
+ if err := configureScheduleClaims(w, sd); err != nil {
+ t.Fatal(err)
+ }
+ if cfg.HostWrites.Enabled || sd.runner.HostWrites.Enabled || sd.conv.HostWrites.Enabled {
+ t.Fatal("disabled startup changed consumers")
+ }
+}
diff --git a/internal/host/claimwiring.go b/internal/host/claimwiring.go
new file mode 100644
index 000000000..d5c64595e
--- /dev/null
+++ b/internal/host/claimwiring.go
@@ -0,0 +1,106 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "sync"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/schedule"
+)
+
+// ClaimProviders contains deployment providers after identity, floor and
+// quarantine verification. Build must fail if any of those gates is unknown.
+type RegionalLoop interface {
+ Loop(context.Context, <-chan struct{}, func(error)) error
+}
+
+type ClaimProviders struct {
+ Selector hostwrite.Selector
+ Wake *activator.ClaimRuntime
+ Regional RegionalLoop
+ Stop *ClaimStopCoordinator
+}
+type ClaimFactory func(context.Context) (ClaimProviders, error)
+
+type ClaimWiring struct {
+ Writes hostwrite.Config
+ Providers ClaimProviders
+ cancel context.CancelFunc
+ joined chan struct{}
+ once sync.Once
+}
+
+func newClaimWiring(ctx context.Context, getenv func(string) string, build ClaimFactory) (*ClaimWiring, error) {
+ w := &ClaimWiring{}
+ cfg, err := hostbox.Startup(ctx, getenv, func(ctx context.Context) (hostwrite.Selector, error) {
+ if build == nil {
+ return nil, hostwrite.ErrUnavailable
+ }
+ providers, err := build(ctx)
+ if err != nil {
+ return nil, err
+ }
+ if providers.Selector == nil || providers.Wake == nil || providers.Regional == nil || providers.Stop == nil {
+ return nil, hostwrite.ErrUnavailable
+ }
+ w.Providers = providers
+ return providers.Selector, nil
+ })
+ w.Writes = cfg
+ return w, err
+}
+
+func (w *ClaimWiring) configure(cfg *activator.Config, runner *schedule.Runner, converger *schedule.Converger) error {
+ if w == nil || cfg == nil || runner == nil || converger == nil {
+ return hostwrite.ErrUnavailable
+ }
+ if !w.Writes.Enabled {
+ return nil
+ }
+ if w.Writes.Selector == nil || w.Providers.Wake == nil {
+ return hostwrite.ErrUnavailable
+ }
+ cfg.HostWrites = w.Writes
+ cfg.Claims = w.Providers.Wake
+ runner.HostWrites = w.Writes
+ converger.HostWrites = w.Writes
+ return nil
+}
+func (w *ClaimWiring) startRegional(ctx context.Context, serial <-chan struct{}, report func(error)) error {
+ if w == nil {
+ return hostwrite.ErrUnavailable
+ }
+ if !w.Writes.Enabled {
+ return nil
+ }
+ if w.Providers.Regional == nil || w.joined != nil {
+ return hostwrite.ErrUnavailable
+ }
+ workerCtx, cancel := context.WithCancel(ctx)
+ w.cancel = cancel
+ w.joined = make(chan struct{})
+ go func() { defer close(w.joined); w.Providers.Regional.Loop(workerCtx, serial, report) }()
+ return nil
+}
+func (w *ClaimWiring) close(ctx context.Context) error {
+ if w == nil {
+ return nil
+ }
+ w.once.Do(func() {
+ if w.cancel != nil {
+ w.cancel()
+ }
+ })
+ if w.joined == nil {
+ return nil
+ }
+ select {
+ case <-w.joined:
+ return nil
+ case <-ctx.Done():
+ return errors.Join(ctx.Err(), hostwrite.ErrUnavailable)
+ }
+}
diff --git a/internal/host/claimwiring_test.go b/internal/host/claimwiring_test.go
new file mode 100644
index 000000000..84da62c16
--- /dev/null
+++ b/internal/host/claimwiring_test.go
@@ -0,0 +1,67 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/schedule"
+ "testing"
+)
+
+func TestClaimWiringOffConstructsNothing(t *testing.T) {
+ reads := 0
+ w, err := newClaimWiring(context.Background(), func(string) string { reads++; return " " }, func(context.Context) (ClaimProviders, error) {
+ t.Fatal("provider construction flag-off")
+ return ClaimProviders{}, nil
+ })
+ if err != nil || reads != 1 {
+ t.Fatal(err, reads)
+ }
+ cfg := activator.Config{}
+ runner := schedule.Runner{}
+ converger := schedule.Converger{}
+ if err := w.configure(&cfg, &runner, &converger); err != nil {
+ t.Fatal(err)
+ }
+ if cfg.HostWrites.Enabled || runner.HostWrites.Enabled || converger.HostWrites.Enabled {
+ t.Fatal("flag off changed")
+ }
+ if err := w.startRegional(context.Background(), nil, nil); err != nil {
+ t.Fatal(err)
+ }
+ if w.joined != nil {
+ t.Fatal("worker flag off")
+ }
+}
+func TestClaimWiringOnMissingProviderRefuses(t *testing.T) {
+ w, err := newClaimWiring(context.Background(), func(string) string { return "0" }, nil)
+ if !w.Writes.Enabled || !errors.Is(err, hostwrite.ErrUnavailable) {
+ t.Fatal(w, err)
+ }
+ if err := w.configure(&activator.Config{}, &schedule.Runner{}, &schedule.Converger{}); err == nil {
+ t.Fatal("missing provider configured")
+ }
+}
+
+type testRegionalLoop struct{ started chan struct{} }
+
+func (r testRegionalLoop) Loop(ctx context.Context, _ <-chan struct{}, _ func(error)) error {
+ close(r.started)
+ <-ctx.Done()
+ return ctx.Err()
+}
+
+var _ RegionalLoop = testRegionalLoop{}
+
+func TestClaimWiringRegionalContractBeforeTask10(t *testing.T) {
+ loop := testRegionalLoop{started: make(chan struct{})}
+ w := &ClaimWiring{Writes: hostwrite.Config{Enabled: true}, Providers: ClaimProviders{Regional: loop}}
+ if err := w.startRegional(t.Context(), nil, nil); err != nil {
+ t.Fatal(err)
+ }
+ <-loop.started
+ if err := w.close(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+}
diff --git a/internal/registry/claimtimers.go b/internal/registry/claimtimers.go
new file mode 100644
index 000000000..c788fa3a0
--- /dev/null
+++ b/internal/registry/claimtimers.go
@@ -0,0 +1,81 @@
+package registry
+
+import (
+ "database/sql"
+ "errors"
+)
+
+var ErrTimerChanged = errors.New("registry: timer registration changed")
+
+// UpsertTimerRegistration compares the captured registration and returns the
+// resulting registration in the same transaction. Empty means absent at read.
+func (r *Registry) UpsertTimerRegistration(t Timer, expected string) (*Timer, error) {
+ tx, err := r.sql.Begin()
+ if err != nil {
+ return nil, err
+ }
+ defer tx.Rollback()
+ old, err := scanTimer(tx.QueryRow(`SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, t.Host, t.Name))
+ if errors.Is(err, sql.ErrNoRows) {
+ if expected != "" {
+ return nil, ErrTimerChanged
+ }
+ } else if err != nil {
+ return nil, err
+ } else {
+ if old.Kind != t.Kind {
+ return nil, ErrTimerKind
+ }
+ if expected == "" || old.Registration != expected {
+ return nil, ErrTimerChanged
+ }
+ }
+ const rearm = `(timers.spec != excluded.spec OR timers.path != excluded.path OR excluded.next_at != 0)`
+ row, err := scanTimer(tx.QueryRow(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,lower(hex(randomblob(16))))
+ON CONFLICT(host, name) DO UPDATE SET
+ spec=excluded.spec, path=excluded.path,
+ next_at=CASE WHEN `+rearm+` THEN excluded.next_at ELSE timers.next_at END,
+ due_at=CASE WHEN `+rearm+` THEN excluded.due_at ELSE timers.due_at END,
+ attempts=CASE WHEN `+rearm+` THEN 0 ELSE timers.attempts END,
+ registration=CASE WHEN `+rearm+` THEN excluded.registration ELSE timers.registration END
+WHERE timers.kind=excluded.kind AND timers.registration=?
+RETURNING `+timerCols,
+ t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt, expected))
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, ErrTimerChanged
+ }
+ if err != nil {
+ return nil, err
+ }
+ if err := tx.Commit(); err != nil {
+ return nil, err
+ }
+ return &row, nil
+}
+
+func (r *Registry) SetTimerStatusRegistration(host, name, registration, status string) error {
+ if registration == "" {
+ return ErrTimerChanged
+ }
+ res, err := r.sql.Exec(`UPDATE timers SET last_status=? WHERE host=? AND name=? AND registration=?`, status, host, name, registration)
+ return timerRegistrationResult(res, err)
+}
+
+func (r *Registry) FireTimerNowRegistration(host, name, registration string, now int64) error {
+ if registration == "" {
+ return ErrTimerChanged
+ }
+ res, err := r.sql.Exec(`UPDATE timers SET next_at=?, due_at=?, attempts=0, registration=lower(hex(randomblob(16))) WHERE host=? AND name=? AND registration=?`, now, now, host, name, registration)
+ return timerRegistrationResult(res, err)
+}
+
+func timerRegistrationResult(res sql.Result, err error) error {
+ if err != nil {
+ return err
+ }
+ n, err := res.RowsAffected()
+ if err == nil && n == 0 {
+ return ErrTimerChanged
+ }
+ return err
+}
diff --git a/internal/release/adopt.go b/internal/release/adopt.go
index ed2b3485b..d254cc43f 100644
--- a/internal/release/adopt.go
+++ b/internal/release/adopt.go
@@ -18,6 +18,7 @@ import (
"time"
"github.com/carlosframework/platform/internal/caps"
+ "github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/registry"
)
@@ -32,6 +33,7 @@ var adoptPlatform = runtime.GOOS + "-" + runtime.GOARCH
type PassOption func(*passConfig)
type passConfig struct {
+ hostWrites hostwrite.Config
// nag is where the canary-TTL nag (spec §5) is printed; nil (the
// default — every call site that does not pass NagExpiredCanaries)
// means the nag never fires at all, not even to a discarded writer.
@@ -312,6 +314,17 @@ func PassFilter(ctx context.Context, reg *registry.Registry, s Store, pub ed2551
adopts := func(rt registry.Route) bool {
return Adoptable(rt) && (extra == nil || extra(rt))
}
+ selected := make(map[string]hostwrite.Work, len(routes))
+ for _, rt := range routes {
+ if !adopts(rt) {
+ continue
+ }
+ work, selectErr := cfg.hostWrites.Select(ctx, rt.Host, "release")
+ if selectErr != nil {
+ return fmt.Errorf("%s: select release work: %w", rt.Host, selectErr)
+ }
+ selected[rt.Host] = work
+ }
// Resolve every distinct (Account, App, Channel) once. Verification
// failures kill the group (nil entry), not the pass.
@@ -774,7 +787,7 @@ func PassFilter(ctx context.Context, reg *registry.Registry, s Store, pub ed2551
continue
}
ptr := pointers[groupKey{RouteAccount(rt), rt.App, routeChannel(rt)}]
- adoptHost(ctx, reg, s, optDir, bake, poisoned, rt, ptr, restart, now, cfg, routeHolds(rt), bakeIgnore(rt))
+ adoptHost(ctx, selected[rt.Host], reg, s, optDir, bake, poisoned, rt, ptr, restart, now, cfg, routeHolds(rt), bakeIgnore(rt))
}
for _, rt := range hibernating {
ptr := pointers[groupKey{RouteAccount(rt), rt.App, routeChannel(rt)}]
@@ -789,7 +802,7 @@ func PassFilter(ctx context.Context, reg *registry.Registry, s Store, pub ed2551
// no-op on a stopped unit). The flag says a route MAY sleep, not that
// it is asleep now, and a warm exec child must not keep serving a
// binary its symlink no longer names.
- adoptHost(ctx, reg, s, optDir, bake, poisoned, rt, ptr, cfg.restartHibernating, now, cfg, routeHolds(rt), bakeIgnore(rt))
+ adoptHost(ctx, selected[rt.Host], reg, s, optDir, bake, poisoned, rt, ptr, cfg.restartHibernating, now, cfg, routeHolds(rt), bakeIgnore(rt))
}
return nil
}
@@ -898,7 +911,7 @@ func RouteAccount(rt registry.Route) string {
// means the supply-chain window does not CURRENTLY apply to this app on
// this box (not declared production, or declared off and past the ratchet);
// the caller computes it once per route from the flag it read this pass.
-func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir string, bake *FirstSeen, poisoned map[string]bool, rt registry.Route, ptr *ChannelPointer, restart func(host string) error, now time.Time, cfg passConfig, holds map[string]time.Duration, ignore bool) {
+func adoptHost(ctx context.Context, work hostwrite.Work, reg *registry.Registry, s Store, optDir string, bake *FirstSeen, poisoned map[string]bool, rt registry.Route, ptr *ChannelPointer, restart func(host string) error, now time.Time, cfg passConfig, holds map[string]time.Duration, ignore bool) {
if ptr == nil {
return
}
@@ -968,10 +981,19 @@ func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir stri
// privileged pass would finish; everything else (bucket, signature,
// manifest) fails the same way as root. See OnStageBlocked.
if cfg.stageBlocked != nil && stageBlocked(err) {
- cfg.stageBlocked(rt.Host, ver, err)
+ // No manifest identity exists on this refusal path, so there is no
+ // admitted host mutation to relay under lifecycle mode.
+ if !cfg.hostWrites.Enabled {
+ cfg.stageBlocked(rt.Host, ver, err)
+ }
}
return
}
+ id, err := releaseIdentity(m)
+ if err != nil {
+ fmt.Printf(" %s: release identity: %v\n", rt.Host, err)
+ return
+ }
// The capability converge for the NEW version, BEFORE PointLive (spec
// §4d). m is the manifest ensureStaged has in hand — never a re-read of
@@ -1000,7 +1022,11 @@ func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir stri
plan, err = capabilityPlan(rt, m, restart != nil)
var res caps.Result
if err == nil {
- res, err = caps.Converge(*cfg.capsOps, plan)
+ err = withReleaseWork(ctx, work, id, func() error {
+ var convergeErr error
+ res, convergeErr = caps.Converge(*cfg.capsOps, plan)
+ return convergeErr
+ })
prev = res.Prev
}
if err != nil {
@@ -1032,7 +1058,7 @@ func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir stri
// Prev is empty and would be read as "this route had no drop-ins"
// — that is a deletion, not a restore.
if res.Changed {
- if err := caps.Restore(*cfg.capsOps, prev); err != nil {
+ if err := withReleaseWork(ctx, work, id, func() error { return caps.Restore(*cfg.capsOps, prev) }); err != nil {
fmt.Printf(" %s: restoring the previous version's capability drop-ins: %v\n", rt.Host, err)
}
}
@@ -1042,7 +1068,7 @@ func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir stri
converged = true
}
- if err := PointLive(optDir, rt.Host, RouteAccount(rt), ver, rt.App); err != nil {
+ if err := withReleaseWork(ctx, work, id, func() error { return PointLive(optDir, rt.Host, RouteAccount(rt), ver, rt.App) }); err != nil {
// The symlink did not move, so the box is still serving cur — and the
// unit shape on disk is the new version's. Put the old one back, or a
// port the running binary never asked for stays publicly bound.
@@ -1051,7 +1077,7 @@ func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir stri
// names" means removing what this pass just wrote.
rolled := ""
if converged {
- if rerr := caps.Restore(*cfg.capsOps, prev); rerr != nil {
+ if rerr := withReleaseWork(ctx, work, id, func() error { return caps.Restore(*cfg.capsOps, prev) }); rerr != nil {
fmt.Printf(" %s: restoring the previous version's capability drop-ins: %v\n", rt.Host, rerr)
} else {
rolled = " — restored the capability drop-ins the symlink still names"
@@ -1062,7 +1088,11 @@ func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir stri
// swap its live symlink. Relay that write refusal too, or the fast
// loop never requests another pass once the artifact exists.
if cfg.stageBlocked != nil && stageBlocked(err) {
- cfg.stageBlocked(rt.Host, ver, err)
+ if guardErr := withReleaseWork(ctx, work, id, func() error { return nil }); guardErr != nil {
+ fmt.Printf(" %s: release authority changed before privileged staging relay: %v\n", rt.Host, guardErr)
+ } else {
+ cfg.stageBlocked(rt.Host, ver, err)
+ }
}
return
}
@@ -1084,6 +1114,10 @@ func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir stri
}
if restart != nil {
+ if err := withReleaseWork(ctx, work, id, func() error { return nil }); err != nil {
+ fmt.Printf(" %s: release authority changed before restart: %v\n", rt.Host, err)
+ return
+ }
// A converged DAEMON's restart is caps.Restart, not the caller's
// callback: this is the only call site that knows the plan, and it is
// the only restart whose error must be real. rootRestartRoute's
@@ -1117,7 +1151,7 @@ func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir stri
poisoned[ver] = true
rolledBack := false
if cur != "" {
- if perr := PointLive(optDir, rt.Host, RouteAccount(rt), cur, rt.App); perr != nil {
+ if perr := withReleaseWork(ctx, work, id, func() error { return PointLive(optDir, rt.Host, RouteAccount(rt), cur, rt.App) }); perr != nil {
fmt.Printf(" %s: could not roll the symlink back to %s: %v\n", rt.Host, cur, perr)
} else {
rolledBack = true
@@ -1142,7 +1176,7 @@ func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir stri
// Bytes are not a binding: Restore re-runs the socket
// sequence itself when the socket half moves, or the MX
// stays off :25 after a rollback.
- if err := caps.Restore(*cfg.capsOps, prev); err != nil {
+ if err := withReleaseWork(ctx, work, id, func() error { return caps.Restore(*cfg.capsOps, prev) }); err != nil {
fmt.Printf(" %s: restoring the previous version's capability drop-ins: %v\n", rt.Host, err)
rolled = " but could not restore the previous version's capability drop-ins"
} else {
@@ -1192,6 +1226,10 @@ func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir stri
// same way while the old process still serves — the two errors on
// the line tell them apart.
if cfg.startUnit != nil && rolledBack && !daemonRestart {
+ if guardErr := withReleaseWork(ctx, work, id, func() error { return nil }); guardErr != nil {
+ fmt.Printf(" %s: release authority changed before rollback start: %v\n", rt.Host, guardErr)
+ return
+ }
if serr := cfg.startUnit(rt.Host); serr != nil {
fmt.Printf(" %s: rolled back to %s but could not start it: %v — this host may be serving nothing\n",
rt.Host, cur, serr)
@@ -1200,7 +1238,7 @@ func adoptHost(ctx context.Context, reg *registry.Registry, s Store, optDir stri
return
}
}
- if err := reg.SetVersions(rt.Host, ver, cur); err != nil {
+ if err := withReleaseWork(ctx, work, id, func() error { return reg.SetVersions(rt.Host, ver, cur) }); err != nil {
fmt.Printf(" %s: repointed but recording the version failed: %v\n", rt.Host, err)
return
}
diff --git a/internal/release/claims.go b/internal/release/claims.go
new file mode 100644
index 000000000..a7fddfef0
--- /dev/null
+++ b/internal/release/claims.go
@@ -0,0 +1,30 @@
+package release
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+
+ "github.com/carlosframework/platform/internal/hostwrite"
+)
+
+func WithHostWrites(config hostwrite.Config) PassOption {
+ return func(c *passConfig) { c.hostWrites = config }
+}
+
+func releaseIdentity(m *Manifest) (hostwrite.Identity, error) {
+ raw, err := json.Marshal(m)
+ if err != nil {
+ return hostwrite.Identity{}, err
+ }
+ sum := sha256.Sum256(raw)
+ return hostwrite.Identity{Release: "sha256:" + hex.EncodeToString(sum[:])}, nil
+}
+
+func withReleaseWork(ctx context.Context, work hostwrite.Work, id hostwrite.Identity, fn func() error) error {
+ if ctx == nil || work == nil {
+ return hostwrite.ErrUnavailable
+ }
+ return work.With(ctx, id, fn)
+}
diff --git a/internal/release/claims_test.go b/internal/release/claims_test.go
new file mode 100644
index 000000000..2c7aac1a5
--- /dev/null
+++ b/internal/release/claims_test.go
@@ -0,0 +1,124 @@
+package release
+
+import (
+ "context"
+ "errors"
+ "sync"
+ "sync/atomic"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/hostwrite"
+)
+
+type pausedReleaseWork struct {
+ pauseAt int
+ calls atomic.Int64
+ valid atomic.Bool
+ once sync.Once
+ entered chan struct{}
+ resume chan struct{}
+}
+
+func newPausedReleaseWork(pauseAt int) *pausedReleaseWork {
+ w := &pausedReleaseWork{pauseAt: pauseAt, entered: make(chan struct{}), resume: make(chan struct{})}
+ w.valid.Store(true)
+ return w
+}
+
+func (w *pausedReleaseWork) With(_ context.Context, id hostwrite.Identity, fn func() error) error {
+ if id.Release == "" {
+ return errors.New("missing release identity")
+ }
+ call := int(w.calls.Add(1))
+ if call == w.pauseAt {
+ w.once.Do(func() { close(w.entered) })
+ <-w.resume
+ }
+ if !w.valid.Load() {
+ return errors.New("release selection changed")
+ }
+ return fn()
+}
+
+type oneReleaseSelector struct{ work hostwrite.Work }
+
+func (s oneReleaseSelector) Select(context.Context, string, string) (hostwrite.Work, error) {
+ return s.work, nil
+}
+
+func TestPassStaleSelectionBeforePointLiveMakesNoHostEffect(t *testing.T) {
+ s := NewDirStore(t.TempDir())
+ priv, pub := testKeys(t)
+ opt := t.TempDir()
+ reg := testRegistry(t)
+ t.Cleanup(func() { reg.Close() })
+ shipRelease(t, s, "bab", "myapp", "v1", adoptPlatform, []byte("build"))
+ if err := Promote(t.Context(), s, priv, "bab", "myapp", "canary/x", "v1"); err != nil {
+ t.Fatal(err)
+ }
+ addRoute(t, reg, "app.example", "myapp", "canary/x", false)
+ w := newPausedReleaseWork(1)
+ restarts := atomic.Int64{}
+ done := make(chan error, 1)
+ go func() {
+ done <- Pass(t.Context(), reg, s, pub, opt, func(string) error { restarts.Add(1); return nil }, WithHostWrites(hostwrite.Config{Enabled: true, Selector: oneReleaseSelector{work: w}}))
+ }()
+ <-w.entered
+ w.valid.Store(false)
+ close(w.resume)
+ if err := <-done; err != nil {
+ t.Fatal(err)
+ }
+ if live, err := LiveVersion(opt, "app.example"); err != nil || live != "" {
+ t.Fatalf("live = %q, %v", live, err)
+ }
+ rt, err := reg.Get("app.example")
+ if err != nil || rt.Version != "" || restarts.Load() != 0 {
+ t.Fatalf("route=%+v err=%v restarts=%d", rt, err, restarts.Load())
+ }
+}
+
+func TestPassRollbackReusesOriginalSelection(t *testing.T) {
+ s := NewDirStore(t.TempDir())
+ priv, pub := testKeys(t)
+ opt := t.TempDir()
+ reg := testRegistry(t)
+ t.Cleanup(func() { reg.Close() })
+ for _, version := range []string{"v1", "v2"} {
+ shipRelease(t, s, "bab", "myapp", version, adoptPlatform, []byte(version))
+ }
+ if err := StageArtifact(opt, "bab", "v1", "myapp", []byte("v1")); err != nil {
+ t.Fatal(err)
+ }
+ if err := PointLive(opt, "app.example", "bab", "v1", "myapp"); err != nil {
+ t.Fatal(err)
+ }
+ if err := Promote(t.Context(), s, priv, "bab", "myapp", "canary/x", "v2"); err != nil {
+ t.Fatal(err)
+ }
+ addRoute(t, reg, "app.example", "myapp", "canary/x", false)
+ w := newPausedReleaseWork(3)
+ done := make(chan error, 1)
+ go func() {
+ done <- Pass(t.Context(), reg, s, pub, opt, func(string) error { return errors.New("restart failed") }, WithHostWrites(hostwrite.Config{Enabled: true, Selector: oneReleaseSelector{work: w}}))
+ }()
+ <-w.entered
+ w.valid.Store(false)
+ close(w.resume)
+ if err := <-done; err != nil {
+ t.Fatal(err)
+ }
+ if live, err := LiveVersion(opt, "app.example"); err != nil || live != "v2" {
+ t.Fatalf("stale rollback changed live link: %q, %v", live, err)
+ }
+}
+
+func TestPassEnabledMissingHostWriteProviderRefusesBeforeChannelRead(t *testing.T) {
+ reg := testRegistry(t)
+ t.Cleanup(func() { reg.Close() })
+ addRoute(t, reg, "app.example", "myapp", "canary/x", false)
+ err := Pass(t.Context(), reg, nil, nil, t.TempDir(), nil, WithHostWrites(hostwrite.Config{Enabled: true}))
+ if !errors.Is(err, hostwrite.ErrUnavailable) {
+ t.Fatalf("Pass = %v", err)
+ }
+}
diff --git a/internal/schedule/claimintegration_test.go b/internal/schedule/claimintegration_test.go
new file mode 100644
index 000000000..c71e4db93
--- /dev/null
+++ b/internal/schedule/claimintegration_test.go
@@ -0,0 +1,224 @@
+package schedule
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "path/filepath"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "encoding/json"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "reflect"
+)
+
+type lifecycleWork struct{ valid *atomic.Bool }
+
+func (w lifecycleWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
+ if !w.valid.Load() {
+ return errors.New("selection changed")
+ }
+ return fn()
+}
+
+type lifecycleSelector struct{ valid *atomic.Bool }
+
+func (s lifecycleSelector) Select(context.Context, string, string) (hostwrite.Work, error) {
+ return lifecycleWork{valid: s.valid}, nil
+}
+
+type pausedDelivery struct {
+ started chan struct{}
+ resume chan struct{}
+}
+
+func (d pausedDelivery) Deliver(context.Context, registry.Route, *http.Request) (int, error) {
+ close(d.started)
+ <-d.resume
+ return 204, nil
+}
+
+func TestRunnerChangedSelectionAfterDeliveryWritesNoBookkeeping(t *testing.T) {
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ host := "app.bab.oncarlos.com"
+ if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
+ t.Fatal(err)
+ }
+ now := time.Unix(10_000, 0).UTC()
+ if err := reg.UpsertTimer(registry.Timer{Host: host, Name: "tick", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/tick", NextAt: now.Unix()}); err != nil {
+ t.Fatal(err)
+ }
+ valid := &atomic.Bool{}
+ valid.Store(true)
+ delivery := pausedDelivery{started: make(chan struct{}), resume: make(chan struct{})}
+ events := atomic.Int64{}
+ r := &Runner{
+ Reg: reg, Deliver: delivery, Now: func() time.Time { return now }, Timeout: time.Minute,
+ Token: func(string) (string, error) { return "token", nil },
+ Event: func(registry.Route, string, string) { events.Add(1) },
+ HostWrites: hostwrite.Config{Enabled: true, Selector: lifecycleSelector{valid: valid}},
+ }
+ r.Tick(t.Context())
+ <-delivery.started
+ valid.Store(false)
+ close(delivery.resume)
+ r.wait()
+ row, err := reg.GetTimer(host, "tick")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.LastAt != 0 || row.NextAt != now.Unix() || events.Load() != 0 {
+ t.Fatalf("bookkeeping changed after authority loss: row=%+v events=%d", row, events.Load())
+ }
+}
+
+func TestEnabledRunnerMissingLifecycleProviderRefusesTimer(t *testing.T) {
+ d := &fakeDeliverer{status: 204}
+ reg, r := runnerFixture(t, d)
+ now := time.Unix(20_000, 0).UTC()
+ r.Now = func() time.Time { return now }
+ r.HostWrites = hostwrite.Config{Enabled: true}
+ if err := reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "tick", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/tick", NextAt: now.Unix()}); err != nil {
+ t.Fatal(err)
+ }
+ r.Tick(t.Context())
+ r.wait()
+ if len(d.calls) != 0 {
+ t.Fatal("delivery ran without lifecycle provider")
+ }
+}
+
+type rotatingSelector struct {
+ current *atomic.Int64
+ calls int
+}
+type capturedEpoch struct {
+ current *atomic.Int64
+ selected int64
+}
+
+func (s *rotatingSelector) Select(context.Context, string, string) (hostwrite.Work, error) {
+ s.calls++
+ return capturedEpoch{s.current, s.current.Load()}, nil
+}
+func (w capturedEpoch) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
+ if w.current.Load() != w.selected {
+ return errors.New("installation changed")
+ }
+ return fn()
+}
+
+type changeOnRead struct {
+ release.Store
+ key string
+ current *atomic.Int64
+}
+
+func (s changeOnRead) Get(ctx context.Context, key string) ([]byte, string, error) {
+ b, e, err := s.Store.Get(ctx, key)
+ if key == s.key {
+ s.current.Add(1)
+ }
+ return b, e, err
+}
+
+func TestConvergerCapturesBeforeDeclarationAndRunReads(t *testing.T) {
+ for _, key := range []string{DeclarationKey("bab", "app"), RunKey("bab", "app")} {
+ t.Run(key, func(t *testing.T) {
+ reg, store := convergeFixture(t)
+ decl := []byte(`{"schedules":[{"name":"tick","every":"1h","path":"/tick"}]}`)
+ if _, err := store.Put(t.Context(), DeclarationKey("bab", "app"), decl, ""); err != nil {
+ t.Fatal(err)
+ }
+ epoch := &atomic.Int64{}
+ selector := &rotatingSelector{current: epoch}
+ c := NewConverger(reg, changeOnRead{store, key, epoch}, "box", nil)
+ c.HostWrites = hostwrite.Config{Enabled: true, Selector: selector}
+ if err := c.Pass(t.Context()); err == nil {
+ t.Fatal("old inputs acquired successor authority")
+ }
+ rows, err := reg.TimersFor("a.bab.oncarlos.com")
+ if err != nil || len(rows) != 0 {
+ t.Fatalf("stale rows=%v %v", rows, err)
+ }
+ if selector.calls != 4 {
+ t.Fatalf("selection refreshed: %d", selector.calls)
+ }
+ })
+ }
+}
+
+type replaceAtWork struct {
+ n, at int
+ replace func()
+}
+
+func (w *replaceAtWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
+ w.n++
+ if w.n == w.at {
+ w.replace()
+ }
+ return fn()
+}
+func TestConvergerRegistrationMutationsPreserveReplacement(t *testing.T) {
+ for _, kind := range []string{"delete", "unit-delete", "status", "run", "upsert"} {
+ t.Run(kind, func(t *testing.T) {
+ reg, _ := convergeFixture(t)
+ host := "a.bab.oncarlos.com"
+ original := registry.Timer{Host: host, Name: "tick", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/tick", NextAt: 100, DueAt: 100}
+ if err := reg.UpsertTimer(original); err != nil {
+ t.Fatal(err)
+ }
+ rt, _ := reg.Get(host)
+ var decl Declaration
+ if err := json.Unmarshal([]byte(`{"schedules":[{"name":"tick","every":"1h","path":"/tick"}]}`), &decl); err != nil {
+ t.Fatal(err)
+ }
+ fire := ""
+ at := 2
+ switch kind {
+ case "delete":
+ decl = Declaration{}
+ case "unit-delete":
+ rt.Backing = registry.BackingUnit
+ decl = Declaration{}
+ case "status":
+ decl = Declaration{}
+ if err := json.Unmarshal([]byte(`{"schedules":[{"name":"tick","cron":"0 0 31 2 *","path":"/tick"}]}`), &decl); err != nil {
+ t.Fatal(err)
+ }
+ at = 3
+ case "run":
+ fire = "tick"
+ at = 4
+ }
+ var replacement *registry.Timer
+ w := &replaceAtWork{at: at, replace: func() {
+ new := original
+ new.Path = "/successor"
+ new.NextAt = 999
+ new.DueAt = 999
+ if err := reg.UpsertTimer(new); err != nil {
+ t.Fatal(err)
+ }
+ replacement, _ = reg.GetTimer(host, "tick")
+ }}
+ ctx := context.WithValue(t.Context(), timerWorkKey{}, timerWork{ctx: t.Context(), work: w})
+ c := NewConverger(reg, nil, "box", nil)
+ c.HostWrites.Enabled = true
+ _, _ = c.convergeHost(ctx, *rt, decl, fire, time.Unix(1000, 0))
+ got, err := reg.GetTimer(host, "tick")
+ if replacement == nil || err != nil || !reflect.DeepEqual(got, replacement) {
+ t.Fatalf("%s replaced row: got=%+v want=%+v err=%v work=%d", kind, got, replacement, err, w.n)
+ }
+ })
+ }
+}
diff --git a/internal/schedule/converge.go b/internal/schedule/converge.go
index adbfca188..687eaf201 100644
--- a/internal/schedule/converge.go
+++ b/internal/schedule/converge.go
@@ -10,6 +10,7 @@ import (
"sync"
"time"
+ "github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
@@ -29,12 +30,13 @@ import (
// 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
- Now func() time.Time
- Box string // reported in Status.Box
- Report func(rt registry.Route, st Status) // called for every host whose rows changed or that is unsupported; nil = no reports
- Idle time.Duration // the activator's idle window; zero when no hibernation is armed on this box
+ Reg *registry.Registry
+ Store release.Store
+ Now func() time.Time
+ Box string // reported in Status.Box
+ Report func(rt registry.Route, st Status) // called for every host whose rows changed or that is unsupported; nil = no reports
+ Idle time.Duration // the activator's idle window; zero when no hibernation is armed on this box
+ HostWrites hostwrite.Config
declEtag map[[2]string]string // (acct, app) → last etag seen
declDoc map[[2]string]Declaration
@@ -111,15 +113,28 @@ func (c *Converger) Pass(ctx context.Context) error {
errs = append(errs, fmt.Errorf("%s/%s: %w", k[0], k[1], err))
}
}
- if n, err := c.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil {
- errs = append(errs, err)
- } else if n > 0 {
- log.Printf("schedule: pruned %d timer row(s)", n)
+ if !c.HostWrites.Enabled {
+ if n, err := c.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil {
+ errs = append(errs, err)
+ } else if n > 0 {
+ log.Printf("schedule: pruned %d timer row(s)", n)
+ }
}
return errors.Join(errs...)
}
func (c *Converger) convergeApp(ctx context.Context, acct, app string, rts []registry.Route, now time.Time) error {
+ // Capture once before either grouped remote input read. Reuse on every effect.
+ selected := map[string]hostwrite.Work{}
+ var errs []error
+ for _, rt := range rts {
+ work, err := c.HostWrites.Select(ctx, rt.Host, "timers")
+ if err != nil {
+ errs = append(errs, fmt.Errorf("%s: select lifecycle authority: %w", rt.Host, err))
+ continue
+ }
+ selected[rt.Host] = work
+ }
k := [2]string{acct, app}
// Reads the object directly rather than calling ReadDeclaration: the
// etag has to be compared against declEtag[k] BEFORE deciding whether
@@ -164,15 +179,24 @@ func (c *Converger) convergeApp(ctx context.Context, acct, app string, rts []reg
}
c.runEtag[k] = runEtag
- var errs []error
for _, rt := range rts {
- changed, err := c.convergeHost(rt, decl, fireName, now)
+ work, ok := selected[rt.Host]
+ if !ok {
+ continue
+ }
+ hctx := context.WithValue(ctx, timerWorkKey{}, timerWork{ctx: ctx, work: work})
+ changed, err := c.convergeHost(hctx, rt, decl, fireName, now)
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", rt.Host, err))
continue
}
if changed && c.Report != nil {
- st, err := StatusFor(c.Reg, rt, c.Box, now, c.Idle)
+ var st Status
+ err := withTimerWork(hctx, func() error {
+ var statusErr error
+ st, statusErr = StatusFor(c.Reg, rt, c.Box, now, c.Idle)
+ return statusErr
+ })
if err != nil {
errs = append(errs, err)
continue
@@ -185,13 +209,43 @@ 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(ctx context.Context, rt registry.Route, decl Declaration, fireName string, now time.Time) (bool, error) {
+ var before []registry.Timer
+ if c.HostWrites.Enabled {
+ err := withTimerWork(ctx, func() error {
+ var readErr error
+ before, readErr = c.Reg.TimersFor(rt.Host)
+ return readErr
+ })
+ if err != nil {
+ return false, err
+ }
+ cutoff := now.Add(-onceRetention).Unix()
+ kept := before[:0]
+ for _, row := range before {
+ if row.Kind == KindOnce && row.NextAt == 0 && row.LastAt < cutoff {
+ if err := withTimerWork(ctx, func() error {
+ return c.Reg.DeleteTimerRegistration(row.Host, row.Name, row.Registration)
+ }); err != nil {
+ return false, err
+ }
+ continue
+ }
+ kept = append(kept, row)
+ }
+ before = kept
+ }
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
// still carries from before the flip, so "no rows" stays true for
// every unit-backed host, not just ones that were always unit-backed.
- n, err := c.Reg.DeleteTimersExcept(rt.Host, KindSchedule, map[string]bool{})
+ var n int
+ err := withTimerWork(ctx, func() error {
+ var writeErr error
+ n, writeErr = c.deleteSchedules(rt.Host, before, map[string]bool{})
+ return writeErr
+ })
if err != nil {
return false, err
}
@@ -203,10 +257,14 @@ func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName s
}
return changed, nil
}
- before, err := c.Reg.TimersFor(rt.Host)
- if err != nil {
- return false, err
+ if !c.HostWrites.Enabled {
+ var err error
+ before, err = c.Reg.TimersFor(rt.Host)
+ if err != nil {
+ return false, err
+ }
}
+ registrations := map[string]string{}
keep := map[string]bool{}
changed := false
for _, sc := range decl.Schedules {
@@ -238,7 +296,20 @@ func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName s
}
// A new occurrence is due whenever next moves (DueAt rides NextAt —
// UpsertTimer leaves both alone when NextAt is 0 and nothing changed).
- err = c.Reg.UpsertTimer(registry.Timer{Host: rt.Host, Name: sc.Name, Kind: KindSchedule, Spec: sp.String(), Path: sc.Path, NextAt: next, DueAt: next, CreatedAt: now.Unix()})
+ var installed *registry.Timer
+ err = withTimerWork(ctx, func() error {
+ t := registry.Timer{Host: rt.Host, Name: sc.Name, Kind: KindSchedule, Spec: sp.String(), Path: sc.Path, NextAt: next, DueAt: next, CreatedAt: now.Unix()}
+ if !c.HostWrites.Enabled {
+ return c.Reg.UpsertTimer(t)
+ }
+ expected := ""
+ if prev != nil {
+ expected = prev.Registration
+ }
+ var err error
+ installed, err = c.Reg.UpsertTimerRegistration(t, expected)
+ return err
+ })
if errors.Is(err, registry.ErrTimerKind) {
// Nothing was written — a collision must not pin `changed`,
// or a permanently-colliding name reports every single pass.
@@ -248,9 +319,17 @@ func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName s
if err != nil {
return changed, err
}
+ if installed != nil {
+ registrations[sc.Name] = installed.Registration
+ }
if neverFires {
log.Printf("schedule: %s: schedule %q never fires (no occurrence within the search window) — parked", rt.Host, sc.Name)
- if err := c.Reg.SetTimerStatus(rt.Host, sc.Name, StatusNeverFires); err != nil {
+ if err := withTimerWork(ctx, func() error {
+ if c.HostWrites.Enabled {
+ return c.Reg.SetTimerStatusRegistration(rt.Host, sc.Name, installed.Registration, StatusNeverFires)
+ }
+ return c.Reg.SetTimerStatus(rt.Host, sc.Name, StatusNeverFires)
+ }); err != nil {
return changed, err
}
}
@@ -258,7 +337,12 @@ func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName s
changed = true
}
}
- n, err := c.Reg.DeleteTimersExcept(rt.Host, KindSchedule, keep)
+ var n int
+ err := withTimerWork(ctx, func() error {
+ var writeErr error
+ n, writeErr = c.deleteSchedules(rt.Host, before, keep)
+ return writeErr
+ })
if err != nil {
return changed, err
}
@@ -266,7 +350,12 @@ func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName s
changed = true
}
if fireName != "" && keep[fireName] {
- if err := c.Reg.FireTimerNow(rt.Host, fireName, now.Unix()); err != nil {
+ if err := withTimerWork(ctx, func() error {
+ if c.HostWrites.Enabled {
+ return c.Reg.FireTimerNowRegistration(rt.Host, fireName, registrations[fireName], now.Unix())
+ }
+ return c.Reg.FireTimerNow(rt.Host, fireName, now.Unix())
+ }); err != nil {
return changed, err
}
changed = true
@@ -387,3 +476,22 @@ func holdsAwake(t registry.Timer, idle time.Duration, now time.Time) bool {
}
return second.Sub(first) <= idle
}
+
+// deleteSchedules only removes rows observed by this pass. A same-name
+// replacement keeps its registration and survives the stale deletion.
+func (c *Converger) deleteSchedules(host string, before []registry.Timer, keep map[string]bool) (int, error) {
+ if !c.HostWrites.Enabled {
+ return c.Reg.DeleteTimersExcept(host, KindSchedule, keep)
+ }
+ n := 0
+ for _, row := range before {
+ if row.Kind != KindSchedule || keep[row.Name] {
+ continue
+ }
+ if err := c.Reg.DeleteTimerRegistration(host, row.Name, row.Registration); err != nil {
+ return n, err
+ }
+ n++
+ }
+ return n, nil
+}
diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go
index cfc6dfd46..eb964d164 100644
--- a/internal/schedule/runner.go
+++ b/internal/schedule/runner.go
@@ -10,6 +10,7 @@ import (
"sync"
"time"
+ "github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/registry"
)
@@ -50,16 +51,17 @@ type Deliverer interface {
// parallelism across hosts (spec §6.3). Every seam is a field so a tick is
// testable with no 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
- Idle time.Duration // the activator's idle window; zero when no hibernation is armed on this box
+ 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
+ Idle time.Duration // the activator's idle window; zero when no hibernation is armed on this box
+ HostWrites hostwrite.Config
// Per-host work is DETACHED from the tick: Tick launches one goroutine
// per host with due rows and returns without waiting for any of them.
@@ -116,7 +118,7 @@ func (r *Runner) timeout() time.Duration {
// goroutine is launched — it never waits for a delivery.
func (r *Runner) Tick(ctx context.Context) {
now := r.now()
- r.prune(now)
+ r.prune(ctx, now)
// Busy hosts are excluded IN THE QUERY, not after it. The batch is
// capped at dueBatch rows: a host parked in a long handler (or one
// holding more due rows than the batch) would otherwise fill the batch
@@ -140,7 +142,18 @@ func (r *Runner) Tick(ctx context.Context) {
byHost[t.Host] = append(byHost[t.Host], t)
}
for _, host := range order {
- rt, err := r.Reg.Get(host)
+ work, err := r.HostWrites.Select(ctx, host, "timers")
+ if err != nil {
+ log.Printf("schedule: %s: select lifecycle authority: %v", host, err)
+ continue
+ }
+ hctx := context.WithValue(ctx, timerWorkKey{}, timerWork{ctx: ctx, work: work})
+ var rt *registry.Route
+ err = withTimerWork(hctx, func() error {
+ var readErr error
+ rt, readErr = r.Reg.Get(host)
+ return readErr
+ })
if err != nil {
// A registry error is not "route gone": say so, or a broken
// registry reads as a fleet of retired routes.
@@ -165,7 +178,7 @@ func (r *Runner) Tick(ctx context.Context) {
return
}
r.inflight.Add(1)
- go func(host string, rt registry.Route, timers []registry.Timer) {
+ go func(ctx context.Context, host string, rt registry.Route, timers []registry.Timer) {
defer r.inflight.Done()
defer func() { <-r.slots() }()
defer r.release(host)
@@ -176,13 +189,19 @@ func (r *Runner) Tick(ctx context.Context) {
// A report that cannot be built is a silent hole in `carlos
// schedule ls` — the member's only view of this box. Say so:
// the fires themselves still landed on the rows.
- if st, err := StatusFor(r.Reg, rt, r.Box, r.now(), r.Idle); err != nil {
+ var st Status
+ err := withTimerWork(ctx, func() error {
+ var statusErr error
+ st, statusErr = StatusFor(r.Reg, rt, r.Box, r.now(), r.Idle)
+ return statusErr
+ })
+ if err != nil {
log.Printf("schedule: %s: build status report: %v", rt.Host, err)
} else {
r.Report(rt, st)
}
}
- }(host, *rt, byHost[host])
+ }(hctx, host, *rt, byHost[host])
}
}
@@ -213,7 +232,7 @@ func (r *Runner) busyHosts() []string {
// second goroutine to stop, and the first tick after start prunes once.
// Cheap either way (one DELETE over an indexed table), and idempotent with
// the converger's sweep on a box that has both.
-func (r *Runner) prune(now time.Time) {
+func (r *Runner) prune(ctx context.Context, now time.Time) {
r.mu.Lock()
if !r.lastPruned.IsZero() && now.Sub(r.lastPruned) < pruneEvery {
r.mu.Unlock()
@@ -229,8 +248,8 @@ func (r *Runner) prune(now time.Time) {
log.Printf("schedule: runner prune: list routes: %v — leaving the rows alone", err)
return
}
- alive := map[string]bool{}
sockets := map[string]bool{}
+ alive := map[string]bool{}
for _, rt := range routes {
alive[rt.Host] = true
if rt.Socket != "" {
@@ -245,7 +264,7 @@ func (r *Runner) prune(now time.Time) {
if sr, ok := r.Deliver.(SocketRetainer); ok {
sr.RetainSockets(sockets)
}
- if len(alive) == 0 {
+ if len(routes) == 0 {
// PruneTimers reads an empty alive set as "every host is gone" and
// deletes the whole table — correct for a genuinely empty registry
// (routes and timers live in the same file, so every row really is
@@ -257,11 +276,56 @@ func (r *Runner) prune(now time.Time) {
log.Printf("schedule: runner prune: the registry has no routes — declining to sweep (that would delete every timer row)")
return
}
- if n, err := r.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil {
- log.Printf("schedule: runner prune: %v", err)
- } else if n > 0 {
- log.Printf("schedule: runner prune: pruned %d timer row(s)", n)
+ if !r.HostWrites.Enabled {
+ if n, err := r.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil {
+ log.Printf("schedule: runner prune: %v", err)
+ } else if n > 0 {
+ log.Printf("schedule: runner prune: pruned %d timer row(s)", n)
+ }
+ return
+ }
+ cutoff := now.Add(-onceRetention).Unix()
+ for _, rt := range routes {
+ work, err := r.HostWrites.Select(ctx, rt.Host, "timers")
+ if err != nil {
+ log.Printf("schedule: runner prune %s: select lifecycle authority: %v", rt.Host, err)
+ continue
+ }
+ hctx := context.WithValue(ctx, timerWorkKey{}, timerWork{ctx: ctx, work: work})
+ var rows []registry.Timer
+ if err := withTimerWork(hctx, func() error {
+ var readErr error
+ rows, readErr = r.Reg.TimersFor(rt.Host)
+ return readErr
+ }); err != nil {
+ log.Printf("schedule: runner prune %s: %v", rt.Host, err)
+ continue
+ }
+ for _, row := range rows {
+ if row.Kind != KindOnce || row.NextAt != 0 || row.LastAt >= cutoff {
+ continue
+ }
+ if err := withTimerWork(hctx, func() error {
+ return r.Reg.DeleteTimerRegistration(row.Host, row.Name, row.Registration)
+ }); err != nil {
+ log.Printf("schedule: runner prune %s/%s: %v", row.Host, row.Name, err)
+ }
+ }
+ }
+}
+
+type timerWorkKey struct{}
+type timerWork struct {
+ ctx context.Context
+ work hostwrite.Work
+}
+
+func withTimerWork(ctx context.Context, fn func() error) error {
+ selected, ok := ctx.Value(timerWorkKey{}).(timerWork)
+ if !ok || selected.ctx == nil || selected.work == nil {
+ return hostwrite.ErrUnavailable
}
+ return selected.work.With(selected.ctx, hostwrite.Identity{}, fn)
}
// SocketRetainer is the optional half of Deliverer that keeps per-socket
@@ -308,36 +372,43 @@ func (r *Runner) slots() chan struct{} {
// sleeping. Nothing in the agent calls it.
func (r *Runner) wait() { r.inflight.Wait() }
-func (r *Runner) timerStillDue(t registry.Timer) bool {
- current, err := r.Reg.GetTimer(t.Host, t.Name)
+func (r *Runner) timerStillDue(ctx context.Context, t registry.Timer) bool {
+ var current *registry.Timer
+ err := withTimerWork(ctx, func() error {
+ var readErr error
+ current, readErr = r.Reg.GetTimer(t.Host, t.Name)
+ return readErr
+ })
return err == nil && current.Registration == t.Registration && current.NextAt == t.NextAt && current.NextAt > 0 && current.NextAt <= r.now().Unix()
}
func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer) {
- if !r.timerStillDue(t) {
+ if !r.timerStillDue(ctx, t) {
return
}
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.DeleteTimerRegistration(rt.Host, t.Name, t.Registration)
+ if err := withTimerWork(ctx, func() error { return r.Reg.DeleteTimerRegistration(rt.Host, t.Name, t.Registration) }); err != nil {
+ log.Printf("schedule: %s/%s: drop bad row: %v", rt.Host, t.Name, err)
+ }
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)
+ r.record(ctx, rt, t, sp, start, Outcome{Status: "no instance token: " + err.Error(), Retry: true}, 0)
return
}
req, err := BuildRequest(rt, t, token, r.Build)
if err != nil {
// Not retried: the path is the row's, and it will be exactly as bad
// next time. The row advances (or finishes) with the reason on it.
- r.record(rt, t, sp, start, Outcome{Status: "bad path: " + err.Error()}, 0)
+ r.record(ctx, rt, t, sp, start, Outcome{Status: "bad path: " + err.Error()}, 0)
return
}
- if !r.timerStillDue(t) {
+ if !r.timerStillDue(ctx, t) {
return
}
dctx, cancel := context.WithTimeout(ctx, r.timeout())
@@ -347,12 +418,12 @@ func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer)
if errors.Is(derr, context.DeadlineExceeded) {
derr = ErrTimeout
}
- r.record(rt, t, sp, start, classify(status, derr, r.timeout()), r.now().Sub(start))
+ r.record(ctx, 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) {
+func (r *Runner) record(ctx context.Context, 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
@@ -391,9 +462,15 @@ func (r *Runner) record(rt registry.Route, t registry.Timer, sp Spec, at time.Ti
} else if !isSuccess(o.Status) {
level = "warn"
}
- applied, err := r.Reg.MarkTimerFired(rt.Host, t.Name, t.Registration, at.Unix(), status, took.Milliseconds(), attempts, next, due)
+ var applied bool
+ err := withTimerWork(ctx, func() error {
+ var writeErr error
+ applied, writeErr = r.Reg.MarkTimerFired(rt.Host, t.Name, t.Registration, at.Unix(), status, took.Milliseconds(), attempts, next, due)
+ return writeErr
+ })
if err != nil {
log.Printf("schedule: %s/%s: record fire: %v", rt.Host, t.Name, err)
+ return
}
if r.Event != nil {
msg := fmt.Sprintf("schedule %s fired POST %s → %s in %s", t.Name, t.Path, status, took.Truncate(time.Millisecond))