amadan
appconfig, release, cgroup, hostwrite: callbacks keep captured authority, enabled config errors refuse, cleanup latches first (Plan 6 review findings 5-7)
Co-Authored-By: Codex <noreply@openai.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
11 files changed,
+534
−28
internal/appconfig/claimsync.go+45 −0internal/appconfig/claimsync_test.go+241 −2internal/appconfig/sync.go+18 −7internal/appconfig/syncexec.go+39 −9internal/cgroup/role_linux.go+1 −1internal/cgroup/role_linux_test.go+28 −0internal/cgroup/strict_linux.go+6 −5internal/hostwrite/callback.go+22 −0internal/release/adopt.go+20 −3internal/release/claims.go+7 −0internal/release/claims_test.go+107 −1
diff --git a/internal/appconfig/claimsync.go b/internal/appconfig/claimsync.go| index cc5b8226a..5e8bcf9af 100644 |
| --- a/internal/appconfig/claimsync.go |
| +++ b/internal/appconfig/claimsync.go |
| @@ -39,3 +39,48 @@ type legacyHostWork struct{} |
| func (legacyHostWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error { |
| return fn() |
| } |
| + |
| +// Callbacks keeps legacy handlers for flag-off passes. Enabled passes require |
| +// the selected handlers, which must retain the selection until their final |
| +// effect. They run outside the config writer's flock. |
| +type Callbacks struct { |
| + Restart func(string) error |
| + Ack func(ConfigAck) |
| + SelectedRestart hostwrite.Callback |
| + SelectedAck func(context.Context, hostwrite.Selection, ConfigAck) error |
| + enabled bool |
| +} |
| + |
| +func (c Callbacks) restart(ctx context.Context, selected hostwrite.Selection) error { |
| + if c.enabled { |
| + if c.SelectedRestart == nil { |
| + return hostwrite.ErrUnavailable |
| + } |
| + return c.SelectedRestart(ctx, selected) |
| + } |
| + if c.Restart == nil { |
| + return nil |
| + } |
| + return c.Restart(selected.Host) |
| +} |
| + |
| +func (c Callbacks) acknowledge(ctx context.Context, selected hostwrite.Selection, a ConfigAck) error { |
| + if c.enabled { |
| + if c.SelectedAck == nil { |
| + return hostwrite.ErrUnavailable |
| + } |
| + return c.SelectedAck(ctx, selected, a) |
| + } |
| + if c.Ack != nil { |
| + c.Ack(a) |
| + } |
| + return nil |
| +} |
| + |
| +func configCallbacks(work hostwrite.Work, restart func(string) error, ack func(ConfigAck), options []Callbacks) Callbacks { |
| + if len(options) != 0 { |
| + return options[0] |
| + } |
| + _, legacy := work.(legacyHostWork) |
| + return Callbacks{Restart: restart, Ack: ack, enabled: !legacy} |
| +} |
diff --git a/internal/appconfig/claimsync_test.go b/internal/appconfig/claimsync_test.go| index 77da13bf9..71004c3aa 100644 |
| --- a/internal/appconfig/claimsync_test.go |
| +++ b/internal/appconfig/claimsync_test.go |
| @@ -10,6 +10,7 @@ import ( |
| "testing" |
| "github.com/carlosframework/platform/internal/hostwrite" |
| + "github.com/carlosframework/platform/internal/provision" |
| "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| ) |
| @@ -149,10 +150,17 @@ func TestClaimDisabledConfigErrorContinuesHostsAndReaps(t *testing.T) { |
| calls := map[string]int{} |
| restart := func(host string) error { calls[host]++; return nil } |
| var err error |
| + callbacks := Callbacks{Restart: restart, |
| + SelectedRestart: func(context.Context, hostwrite.Selection) error { t.Fatal("disabled selected restart"); return nil }, |
| + SelectedAck: func(context.Context, hostwrite.Selection, ConfigAck) error { |
| + t.Fatal("disabled selected ack") |
| + return nil |
| + }, |
| + } |
| if exec { |
| - err = SyncExec(t.Context(), reg, store, nil, dir, restart, "", nil) |
| + err = SyncExecWithCallbacks(t.Context(), reg, store, nil, dir, "", callbacks) |
| } else { |
| - err = Sync(t.Context(), reg, store, nil, dir, restart, nil) |
| + err = SyncWithCallbacks(t.Context(), reg, store, nil, dir, callbacks) |
| } |
| if err != nil { |
| t.Fatalf("legacy pass aborted: %v", err) |
| @@ -166,3 +174,234 @@ func TestClaimDisabledConfigErrorContinuesHostsAndReaps(t *testing.T) { |
| }) |
| } |
| } |
| + |
| +type configTestSelector struct{} |
| + |
| +func (configTestSelector) Select(context.Context, string, string) (hostwrite.Work, error) { |
| + return legacyHostWork{}, nil |
| +} |
| + |
| +func TestClaimEnabledConfigReadAndRenderFailures(t *testing.T) { |
| + for _, shape := range []string{"unit", "flat-exec", "nested-exec"} { |
| + for _, failure := range []string{"read", "render"} { |
| + t.Run(shape+"/"+failure, func(t *testing.T) { |
| + reg := testRegistry(t) |
| + defer reg.Close() |
| + store := release.NewDirStore(t.TempDir()) |
| + dir := t.TempDir() |
| + rt := registry.Route{Host: "app.example", Kind: "instance", Account: "bab", App: "app", Status: "ready", Socket: "/unused.sock"} |
| + path := filepath.Join(dir, rt.Host+".env") |
| + if shape == "unit" { |
| + rt.Backing = registry.BackingUnit |
| + } else { |
| + if shape == "nested-exec" { |
| + rt.DBPath = provision.NestedDBPath(dir, rt.Host) |
| + rt.Socket = provision.NestedSocket(rt.Host) |
| + } |
| + path = ConfigEnvPath(dir, rt) |
| + } |
| + if err := reg.Upsert(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + vars := map[string]string{"VALUE": "new"} |
| + if err := SetEnv(t.Context(), store, "bab", "app", vars); err != nil { |
| + t.Fatal(err) |
| + } |
| + if failure == "read" { |
| + if err := os.MkdirAll(path, 0700); err != nil { |
| + t.Fatal(err) |
| + } |
| + } else { |
| + _, etag, err := store.Get(t.Context(), Key("bab", "app")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := store.Put(t.Context(), Key("bab", "app"), []byte(`{"schema":1,"vars":{"VALUE":{"kind":"secret"}}}`), etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + callbacks := 0 |
| + restart := func(string) error { callbacks++; return nil } |
| + ack := func(ConfigAck) { callbacks++ } |
| + writes := hostwrite.Config{Enabled: true, Selector: configTestSelector{}} |
| + var err error |
| + if shape == "unit" { |
| + err = Sync(t.Context(), reg, store, nil, dir, restart, ack, writes) |
| + } else { |
| + err = SyncExec(t.Context(), reg, store, nil, dir, restart, "", ack, writes) |
| + } |
| + if err == nil { |
| + t.Fatal("enabled pass reported success after config " + failure + " failure") |
| + } |
| + if callbacks != 0 { |
| + t.Fatalf("callbacks after failure: %d", callbacks) |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +func TestClaimConfigRefusesHostnameOnlyCallbacks(t *testing.T) { |
| + for _, shape := range []string{"unit", "exec", "exec-removal", "ack"} { |
| + t.Run(shape, func(t *testing.T) { |
| + reg := testRegistry(t) |
| + defer reg.Close() |
| + store := release.NewDirStore(t.TempDir()) |
| + dir := t.TempDir() |
| + rt := registry.Route{Host: "app.example", Kind: "instance", Account: "bab", App: "app", Status: "ready", Socket: "/unused.sock"} |
| + if shape == "unit" { |
| + rt.Backing = registry.BackingUnit |
| + } |
| + if shape == "ack" { |
| + rt.Socket = "" |
| + rt.Addr = "127.0.0.1:9999" |
| + } |
| + if err := reg.Upsert(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + if shape == "exec-removal" { |
| + if err := os.WriteFile(ConfigEnvPath(dir, rt), []byte("OLD=1\n"), 0600); err != nil { |
| + t.Fatal(err) |
| + } |
| + } else if err := SetEnv(t.Context(), store, "bab", "app", map[string]string{"VALUE": "new"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if shape == "ack" { |
| + if err := SetExecDelivery(t.Context(), store, "bab", "app", ExecDeliveryFile); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + calls := 0 |
| + restart := func(string) error { calls++; return nil } |
| + ack := func(ConfigAck) { calls++ } |
| + writes := hostwrite.Config{Enabled: true, Selector: configTestSelector{}} |
| + var err error |
| + if shape == "unit" { |
| + err = Sync(t.Context(), reg, store, nil, dir, restart, ack, writes) |
| + } else { |
| + err = SyncExec(t.Context(), reg, store, nil, dir, restart, "", ack, writes) |
| + } |
| + if calls != 0 { |
| + t.Fatalf("enabled callbacks lost selection: %d effects, err=%v", calls, err) |
| + } |
| + if !errors.Is(err, hostwrite.ErrUnavailable) { |
| + t.Fatalf("missing selected callback: %v", err) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +type callbackConfigWork struct { |
| + valid bool |
| + held bool |
| +} |
| + |
| +func (w *callbackConfigWork) With(_ context.Context, id hostwrite.Identity, fn func() error) error { |
| + if !w.valid { |
| + return errors.New("selection changed") |
| + } |
| + if w.held || id.Config == "" { |
| + return errors.New("invalid callback guard") |
| + } |
| + w.held = true |
| + defer func() { w.held = false }() |
| + return fn() |
| +} |
| + |
| +type selectedConfigTestSelector struct { |
| + work hostwrite.Work |
| + calls int |
| +} |
| + |
| +func (s *selectedConfigTestSelector) Select(context.Context, string, string) (hostwrite.Work, error) { |
| + s.calls++ |
| + return s.work, nil |
| +} |
| + |
| +func TestClaimConfigCallbacksRetainSelection(t *testing.T) { |
| + for _, shape := range []string{"unit", "exec", "exec-removal"} { |
| + for _, boundary := range []string{"restart", "ack", "current"} { |
| + if shape == "exec-removal" && boundary == "ack" { |
| + continue |
| + } |
| + t.Run(shape+"/"+boundary, func(t *testing.T) { |
| + reg := testRegistry(t) |
| + defer reg.Close() |
| + store := release.NewDirStore(t.TempDir()) |
| + dir := t.TempDir() |
| + rt := registry.Route{Host: "app.example", Kind: "instance", Account: "bab", App: "app", Status: "ready", Socket: "/unused.sock"} |
| + if shape == "unit" { |
| + rt.Backing = registry.BackingUnit |
| + } |
| + if err := reg.Upsert(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + if shape == "exec-removal" { |
| + if err := os.WriteFile(ConfigEnvPath(dir, rt), []byte("OLD=1\n"), 0600); err != nil { |
| + t.Fatal(err) |
| + } |
| + } else if err := SetEnv(t.Context(), store, "bab", "app", map[string]string{"VALUE": "new"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + w := &callbackConfigWork{valid: true} |
| + selector := &selectedConfigTestSelector{work: w} |
| + restarts, acks := 0, 0 |
| + var identity hostwrite.Identity |
| + check := func(s hostwrite.Selection) { |
| + if w.held { |
| + t.Fatal("callback/network wait under guard") |
| + } |
| + if s.Work != w || s.Host != rt.Host || s.Identity.Config == "" { |
| + t.Fatalf("lost selection: %+v", s) |
| + } |
| + if identity.Config != "" && s.Identity != identity { |
| + t.Fatal("identity changed between callbacks") |
| + } |
| + identity = s.Identity |
| + } |
| + callbacks := Callbacks{ |
| + Restart: func(string) error { t.Fatal("legacy restart"); return nil }, |
| + Ack: func(ConfigAck) { t.Fatal("legacy ack") }, |
| + SelectedRestart: func(ctx context.Context, s hostwrite.Selection) error { |
| + check(s) |
| + if boundary == "restart" { |
| + w.valid = false |
| + } |
| + return s.With(ctx, func() error { restarts++; return nil }) |
| + }, |
| + SelectedAck: func(ctx context.Context, s hostwrite.Selection, a ConfigAck) error { |
| + check(s) |
| + if a.Host != rt.Host { |
| + t.Fatal(a.Host) |
| + } |
| + if boundary == "ack" { |
| + w.valid = false |
| + } |
| + return s.With(ctx, func() error { acks++; return nil }) |
| + }, |
| + } |
| + writes := hostwrite.Config{Enabled: true, Selector: selector} |
| + var err error |
| + if shape == "unit" { |
| + err = SyncWithCallbacks(t.Context(), reg, store, nil, dir, callbacks, writes) |
| + } else { |
| + err = SyncExecWithCallbacks(t.Context(), reg, store, nil, dir, "", callbacks, writes) |
| + } |
| + wantRestarts, wantAcks := 1, 1 |
| + if shape == "exec-removal" { |
| + wantAcks = 0 |
| + } |
| + if boundary == "restart" { |
| + wantRestarts = 0 |
| + wantAcks = 0 |
| + } |
| + if boundary == "ack" { |
| + wantAcks = 0 |
| + } |
| + if (err == nil) != (boundary == "current") || restarts != wantRestarts || acks != wantAcks || selector.calls != 1 { |
| + t.Fatalf("err=%v restarts=%d acks=%d selections=%d", err, restarts, acks, selector.calls) |
| + } |
| + }) |
| + } |
| + } |
| +} |
diff --git a/internal/appconfig/sync.go b/internal/appconfig/sync.go| index e242de8c8..a6915f16f 100644 |
| --- a/internal/appconfig/sync.go |
| +++ b/internal/appconfig/sync.go |
| @@ -49,6 +49,11 @@ const DefaultEnvDir = "/etc/carlos/apps" |
| // 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), options ...hostwrite.Config) error { |
| + return SyncWithCallbacks(ctx, reg, s, priv, envDir, Callbacks{Restart: restart, Ack: ack}, options...) |
| +} |
| + |
| +func SyncWithCallbacks(ctx context.Context, reg *registry.Registry, s release.Store, priv *ecdh.PrivateKey, envDir string, callbacks Callbacks, options ...hostwrite.Config) error { |
| + restart, ack := callbacks.Restart, callbacks.Ack |
| routes, err := reg.List() |
| if err != nil { |
| return err |
| @@ -57,6 +62,7 @@ func Sync(ctx context.Context, reg *registry.Registry, s release.Store, priv *ec |
| if err != nil { |
| return err |
| } |
| + callbacks.enabled = hostWrites.Enabled |
| selected := map[string]hostwrite.Work{} |
| for _, rt := range routes { |
| if !syncable(rt) || rt.Backing != registry.BackingUnit { |
| @@ -273,7 +279,7 @@ 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 |
| } |
| - if err := syncHostClaimed(ctx, selected[rt.Host], rt, m, defaultMissing, priv, envDir, restart, instETag, ack); err != nil && hostWrites.Enabled { |
| + if err := syncHostClaimed(ctx, selected[rt.Host], rt, m, defaultMissing, priv, envDir, restart, instETag, ack, callbacks); err != nil && hostWrites.Enabled { |
| return err |
| } |
| } |
| @@ -486,7 +492,8 @@ func syncHost(rt registry.Route, m *Manifest, defaultMissing bool, priv *ecdh.Pr |
| _ = 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 { |
| +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), options ...Callbacks) error { |
| + callbacks := configCallbacks(work, restart, ack, options) |
| // 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. |
| @@ -499,7 +506,7 @@ func syncHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.Route |
| existed := err == nil |
| if err != nil && !os.IsNotExist(err) { |
| fmt.Printf(" %s: read %s: %v\n", rt.Host, path, err) |
| - return nil |
| + return err |
| } |
| // 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 |
| @@ -562,7 +569,7 @@ func syncHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.Route |
| rendered, err := RenderEnv(m, priv, routeBinding(rt)) |
| if err != nil { |
| fmt.Printf(" %s: render config: %v\n", rt.Host, err) |
| - return nil |
| + return err |
| } |
| // Spec §7.3: deliver CARLOS_PURGE=header to unit tenants the same way |
| // the exec backend already does for exec-backed ones (withoutVars / |
| @@ -623,7 +630,11 @@ func syncHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.Route |
| if err := guarded(ctx, work, id, func() error { return nil }); err != nil { |
| return err |
| } |
| - switch err := restart(rt.Host); { |
| + restartErr := callbacks.restart(ctx, hostwrite.Selection{Host: rt.Host, Work: work, Identity: id}) |
| + if callbacks.enabled && restartErr != nil { |
| + return restartErr |
| + } |
| + switch err := restartErr; { |
| case err != nil: |
| fmt.Printf(" %s: config updated but restart failed: %v\n", rt.Host, err) |
| a.RestartError = err.Error() |
| @@ -634,11 +645,11 @@ func syncHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.Route |
| a.Restarted = true |
| fmt.Printf(" %s: config updated, restarted\n", rt.Host) |
| } |
| - if ack != nil { |
| + if ack != nil || callbacks.SelectedAck != nil { |
| if err := guarded(ctx, work, id, func() error { return nil }); err != nil { |
| return err |
| } |
| - ack(a) |
| + return callbacks.acknowledge(ctx, hostwrite.Selection{Host: rt.Host, Work: work, Identity: id}, a) |
| } |
| return nil |
| } |
diff --git a/internal/appconfig/syncexec.go b/internal/appconfig/syncexec.go| index 3b13de308..decb89b91 100644 |
| --- a/internal/appconfig/syncexec.go |
| +++ b/internal/appconfig/syncexec.go |
| @@ -94,6 +94,14 @@ func ConfigEnvPath(dataDir string, rt registry.Route) string { |
| // 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), options ...hostwrite.Config) error { |
| + return SyncExecWithCallbacks(ctx, reg, s, priv, dataDir, appsDomain, Callbacks{Restart: bounce, Ack: ack}, options...) |
| +} |
| + |
| +func SyncExecWithCallbacks(ctx context.Context, reg *registry.Registry, s release.Store, priv *ecdh.PrivateKey, dataDir, appsDomain string, callbacks Callbacks, options ...hostwrite.Config) error { |
| + bounce, ack := callbacks.Restart, callbacks.Ack |
| + if len(options) == 1 && options[0].Enabled && bounce == nil && callbacks.SelectedRestart != nil { |
| + bounce = func(string) error { return hostwrite.ErrUnavailable } |
| + } |
| routes, err := reg.List() |
| if err != nil { |
| return err |
| @@ -102,6 +110,7 @@ func SyncExec(ctx context.Context, reg *registry.Registry, s release.Store, priv |
| if err != nil { |
| return err |
| } |
| + callbacks.enabled = hostWrites.Enabled |
| selected := map[string]hostwrite.Work{} |
| for _, rt := range routes { |
| if !execSyncable(rt) && !execSidecarCandidate(rt) && !execAddrCandidate(rt) { |
| @@ -585,7 +594,12 @@ func SyncExec(ctx context.Context, reg *registry.Registry, s release.Store, priv |
| if m != nil || inst != nil { |
| m = applyInstanceOverlay(k.app, m, inst) |
| } |
| - if err := syncExecHostClaimed(ctx, selected[rt.Host], rt, m, Key(k.acct, k.app), priv, dataDir, hostBounce, header, instETag, ack); err != nil && hostWrites.Enabled { |
| + hostCallbacks := callbacks |
| + if hostBounce == nil { |
| + hostCallbacks.Restart = nil |
| + hostCallbacks.SelectedRestart = nil |
| + } |
| + if err := syncExecHostClaimed(ctx, selected[rt.Host], rt, m, Key(k.acct, k.app), priv, dataDir, hostBounce, header, instETag, ack, hostCallbacks); err != nil && hostWrites.Enabled { |
| return err |
| } |
| } |
| @@ -1136,7 +1150,8 @@ func syncExecHost(rt registry.Route, m *Manifest, key string, priv *ecdh.Private |
| _ = 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 { |
| +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), options ...Callbacks) error { |
| + callbacks := configCallbacks(work, bounce, ack, options) |
| 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 nil |
| @@ -1157,7 +1172,12 @@ func syncExecHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.R |
| if err := guarded(ctx, work, id, func() error { return nil }); err != nil { |
| return err |
| } |
| - bounceLogged(rt.Host, bounce) |
| + if bounce != nil || callbacks.SelectedRestart != nil { |
| + if callbacks.enabled { |
| + return callbacks.restart(ctx, hostwrite.Selection{Host: rt.Host, Work: work, Identity: id}) |
| + } |
| + bounceLogged(rt.Host, bounce) |
| + } |
| return nil |
| } |
| // A nested row's current file lives on tenant-owned ground, so its read |
| @@ -1173,14 +1193,14 @@ func syncExecHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.R |
| existed = err == nil |
| if err != nil && !os.IsNotExist(err) { |
| fmt.Printf(" %s: read %s: %v\n", rt.Host, path, err) |
| - return nil |
| + return err |
| } |
| } 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 nil |
| + return err |
| } |
| } |
| // Where sync.go has its defaultMissing/lostKeys guard, this path has |
| @@ -1206,7 +1226,7 @@ func syncExecHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.R |
| rendered, err := RenderEnv(m, priv, routeBinding(rt)) |
| if err != nil { |
| fmt.Printf(" %s: render config: %v\n", rt.Host, err) |
| - return nil |
| + return err |
| } |
| if header != "" { |
| rendered = append([]byte(header), rendered...) |
| @@ -1243,8 +1263,18 @@ func syncExecHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.R |
| if err := guarded(ctx, work, id, func() error { return nil }); err != nil { |
| return err |
| } |
| - bounceErr := bounceLogged(rt.Host, bounce) |
| - if ack != nil { |
| + var bounceErr error |
| + if bounce != nil || callbacks.SelectedRestart != nil { |
| + if callbacks.enabled { |
| + bounceErr = callbacks.restart(ctx, hostwrite.Selection{Host: rt.Host, Work: work, Identity: id}) |
| + if bounceErr != nil { |
| + return bounceErr |
| + } |
| + } else { |
| + bounceErr = bounceLogged(rt.Host, bounce) |
| + } |
| + } |
| + if ack != nil || callbacks.SelectedAck != nil { |
| a := ConfigAck{Schema: ConfigAckSchema, Host: rt.Host, Shape: shapeOf(rt), At: time.Now(), InstanceConfigETag: instETag} |
| if suffixes, err := relaySuffixesFromRelayVar(m, priv, routeBinding(rt)); err != nil { |
| fmt.Printf(" %s: config updated, but CARLOS_RELAY could not be read for the config-ack: %v\n", rt.Host, err) |
| @@ -1268,7 +1298,7 @@ func syncExecHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.R |
| if err := guarded(ctx, work, id, func() error { return nil }); err != nil { |
| return err |
| } |
| - ack(a) |
| + return callbacks.acknowledge(ctx, hostwrite.Selection{Host: rt.Host, Work: work, Identity: id}, a) |
| } |
| return nil |
| } |
diff --git a/internal/cgroup/role_linux.go b/internal/cgroup/role_linux.go| index 708e5f931..d29cb80e0 100644 |
| --- a/internal/cgroup/role_linux.go |
| +++ b/internal/cgroup/role_linux.go |
| @@ -33,7 +33,7 @@ func (p *Prepared) PrepareRoleChild(role string) (out *Prepared, err error) { |
| } |
| p.mu.Lock() |
| defer p.mu.Unlock() |
| - if p.parent != nil || p.empty || p.draining { |
| + if p.parent != nil || p.empty || p.draining.Load() { |
| return nil, ErrStrictState |
| } |
| if err = p.identityOK(); err != nil { |
diff --git a/internal/cgroup/role_linux_test.go b/internal/cgroup/role_linux_test.go| index dfecd0aa5..6cb75d1bb 100644 |
| --- a/internal/cgroup/role_linux_test.go |
| +++ b/internal/cgroup/role_linux_test.go |
| @@ -141,3 +141,31 @@ func TestPrepareRoleChildRefusals(t *testing.T) { |
| } |
| }) |
| } |
| + |
| +func TestRoleChildRefusesAfterCleanupAcquisitionCancellation(t *testing.T) { |
| + p := strictFixture(t) |
| + makeRoleFixture(t, p, "seed", "populated 0\n") |
| + child, err := p.PrepareRoleChild("seed") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer child.Close() |
| + starts := 0 |
| + child.start = func(*exec.Cmd) error { starts++; return nil } |
| + p.mu.Lock() |
| + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) |
| + defer cancel() |
| + done := make(chan error, 1) |
| + go func() { done <- p.KillAndEmpty(ctx) }() |
| + err = <-done |
| + p.mu.Unlock() |
| + if !errors.Is(err, ErrStrictDeadline) { |
| + t.Fatalf("cleanup acquisition: %v", err) |
| + } |
| + if err := child.Start(exec.Command("unused")); !errors.Is(err, ErrStrictState) || starts != 0 { |
| + t.Fatalf("child reopened parent after cancelled cleanup: %v, starts=%d", err, starts) |
| + } |
| + if _, err := p.PrepareRoleChild("another"); !errors.Is(err, ErrStrictState) { |
| + t.Fatalf("new role after cancelled cleanup: %v", err) |
| + } |
| +} |
diff --git a/internal/cgroup/strict_linux.go b/internal/cgroup/strict_linux.go| index 3914e32e2..076c8cbc9 100644 |
| --- a/internal/cgroup/strict_linux.go |
| +++ b/internal/cgroup/strict_linux.go |
| @@ -12,6 +12,7 @@ import ( |
| "strconv" |
| "strings" |
| "sync" |
| + "sync/atomic" |
| "syscall" |
| "time" |
| @@ -39,7 +40,7 @@ type strictPreparedState struct { |
| entryParent *os.File |
| entryName string |
| parent *Prepared |
| - draining bool |
| + draining atomic.Bool |
| host string |
| attempted, empty, closed bool |
| start func(*exec.Cmd) error |
| @@ -245,7 +246,7 @@ func (p *Prepared) Start(cmd *exec.Cmd) error { |
| if p.parent != nil { |
| p.parent.mu.Lock() |
| defer p.parent.mu.Unlock() |
| - if p.parent.closed || p.parent.empty || p.parent.draining { |
| + if p.parent.closed || p.parent.empty || p.parent.draining.Load() { |
| return ErrStrictState |
| } |
| if err := p.parent.identityOK(); err != nil { |
| @@ -257,7 +258,7 @@ func (p *Prepared) Start(cmd *exec.Cmd) error { |
| } |
| p.mu.Lock() |
| defer p.mu.Unlock() |
| - if p.closed || p.attempted || p.empty || p.draining { |
| + if p.closed || p.attempted || p.empty || p.draining.Load() { |
| return ErrStrictState |
| } |
| p.attempted = true |
| @@ -306,6 +307,7 @@ func (p *Prepared) KillAndEmpty(ctx context.Context) error { |
| if p == nil || p.strictPreparedState == nil { |
| return ErrStrictState |
| } |
| + p.draining.Store(true) |
| if ctx == nil { |
| return ErrStrictDeadline |
| } |
| @@ -332,7 +334,6 @@ func (p *Prepared) KillAndEmpty(ctx context.Context) error { |
| if e := p.identityOK(); e != nil { |
| return e |
| } |
| - p.draining = true |
| p.attempted = true |
| if e := strictWrite(p.kill, "1"); e != nil { |
| return ErrStrictRefused |
| @@ -373,7 +374,7 @@ func (p *Prepared) Close() error { |
| } |
| p.mu.Lock() |
| defer p.mu.Unlock() |
| - p.draining = true |
| + p.draining.Store(true) |
| if p.closed { |
| return ErrStrictState |
| } |
diff --git a/internal/hostwrite/callback.go b/internal/hostwrite/callback.go| new file mode 100644 |
| index 000000000..cdfb4af1b |
| --- /dev/null |
| +++ b/internal/hostwrite/callback.go |
| @@ -0,0 +1,22 @@ |
| +package hostwrite |
| + |
| +import "context" |
| + |
| +// Selection carries the original work and rendered inputs across a callback. |
| +// The receiver uses With at its final local effect, never a new hostname lookup. |
| +// Network waits belong outside With; remote effects need a provider that binds |
| +// the request to this selection and refuses a successor at publication. |
| +type Selection struct { |
| + Host string |
| + Work Work |
| + Identity Identity |
| +} |
| + |
| +func (s Selection) With(ctx context.Context, effect func() error) error { |
| + if ctx == nil || s.Host == "" || s.Work == nil || effect == nil { |
| + return ErrUnavailable |
| + } |
| + return s.Work.With(ctx, s.Identity, effect) |
| +} |
| + |
| +type Callback func(context.Context, Selection) error |
diff --git a/internal/release/adopt.go b/internal/release/adopt.go| index d254cc43f..e1b3050b2 100644 |
| --- a/internal/release/adopt.go |
| +++ b/internal/release/adopt.go |
| @@ -33,7 +33,8 @@ var adoptPlatform = runtime.GOOS + "-" + runtime.GOARCH |
| type PassOption func(*passConfig) |
| type passConfig struct { |
| - hostWrites hostwrite.Config |
| + hostWrites hostwrite.Config |
| + selectedRestart, selectedStart hostwrite.Callback |
| // 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. |
| @@ -1129,7 +1130,13 @@ func adoptHost(ctx context.Context, work hostwrite.Work, reg *registry.Registry, |
| // on this path would panic the root adopt pass for the whole box. |
| var rerr error |
| daemonRestart := converged && plan.Set != nil && plan.Set.Daemon |
| - if daemonRestart { |
| + if cfg.hostWrites.Enabled { |
| + if cfg.selectedRestart == nil { |
| + rerr = hostwrite.ErrUnavailable |
| + } else { |
| + rerr = cfg.selectedRestart(ctx, hostwrite.Selection{Host: rt.Host, Work: work, Identity: id}) |
| + } |
| + } else if daemonRestart { |
| rerr = caps.Restart(*cfg.capsOps, plan) |
| } else { |
| rerr = restart(rt.Host) |
| @@ -1230,7 +1237,17 @@ func adoptHost(ctx context.Context, work hostwrite.Work, reg *registry.Registry, |
| fmt.Printf(" %s: release authority changed before rollback start: %v\n", rt.Host, guardErr) |
| return |
| } |
| - if serr := cfg.startUnit(rt.Host); serr != nil { |
| + var serr error |
| + if cfg.hostWrites.Enabled { |
| + if cfg.selectedStart == nil { |
| + serr = hostwrite.ErrUnavailable |
| + } else { |
| + serr = cfg.selectedStart(ctx, hostwrite.Selection{Host: rt.Host, Work: work, Identity: id}) |
| + } |
| + } else { |
| + serr = cfg.startUnit(rt.Host) |
| + } |
| + if 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) |
| } |
diff --git a/internal/release/claims.go b/internal/release/claims.go| index a7fddfef0..418817942 100644 |
| --- a/internal/release/claims.go |
| +++ b/internal/release/claims.go |
| @@ -28,3 +28,10 @@ func withReleaseWork(ctx context.Context, work hostwrite.Work, id hostwrite.Iden |
| } |
| return work.With(ctx, id, fn) |
| } |
| + |
| +// WithSelectedRestarts supplies receivers that retain the original selection |
| +// through restart and rollback start. Receivers validate at their final effect; |
| +// network waits must not hold the hostguard flock. |
| +func WithSelectedRestarts(restart, start hostwrite.Callback) PassOption { |
| + return func(c *passConfig) { c.selectedRestart = restart; c.selectedStart = start } |
| +} |
diff --git a/internal/release/claims_test.go b/internal/release/claims_test.go| index 2c7aac1a5..dc1b3cb41 100644 |
| --- a/internal/release/claims_test.go |
| +++ b/internal/release/claims_test.go |
| @@ -100,7 +100,7 @@ func TestPassRollbackReusesOriginalSelection(t *testing.T) { |
| 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}})) |
| + done <- Pass(t.Context(), reg, s, pub, opt, func(string) error { return errors.New("unexpected legacy restart") }, WithHostWrites(hostwrite.Config{Enabled: true, Selector: oneReleaseSelector{work: w}}), WithSelectedRestarts(func(context.Context, hostwrite.Selection) error { return errors.New("restart failed") }, nil)) |
| }() |
| <-w.entered |
| w.valid.Store(false) |
| @@ -122,3 +122,109 @@ func TestPassEnabledMissingHostWriteProviderRefusesBeforeChannelRead(t *testing. |
| t.Fatalf("Pass = %v", err) |
| } |
| } |
| + |
| +func TestPassClaimRefusesHostnameOnlyRestart(t *testing.T) { |
| + s := NewDirStore(t.TempDir()) |
| + priv, pub := testKeys(t) |
| + 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(0) |
| + restarts := 0 |
| + if err := Pass(t.Context(), reg, s, pub, t.TempDir(), func(string) error { restarts++; return nil }, WithHostWrites(hostwrite.Config{Enabled: true, Selector: oneReleaseSelector{w}})); err != nil { |
| + t.Fatal(err) |
| + } |
| + rt, err := reg.Get("app.example") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if restarts != 0 || rt.Version != "" { |
| + t.Fatalf("hostname-only restart accepted: restarts=%d version=%q", restarts, rt.Version) |
| + } |
| +} |
| + |
| +func TestPassClaimSelectedRestartBoundary(t *testing.T) { |
| + for _, stale := range []bool{false, true} { |
| + t.Run(map[bool]string{false: "current", true: "successor"}[stale], func(t *testing.T) { |
| + s := NewDirStore(t.TempDir()) |
| + priv, pub := testKeys(t) |
| + 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(0) |
| + effects := 0 |
| + restart := func(ctx context.Context, selected hostwrite.Selection) error { |
| + if selected.Work != w || selected.Host != "app.example" || selected.Identity.Release == "" { |
| + t.Fatalf("lost selection: %+v", selected) |
| + } |
| + if stale { |
| + w.valid.Store(false) |
| + } |
| + return selected.With(ctx, func() error { effects++; return nil }) |
| + } |
| + if err := Pass(t.Context(), reg, s, pub, t.TempDir(), func(string) error { t.Fatal("legacy callback"); return nil }, WithHostWrites(hostwrite.Config{Enabled: true, Selector: oneReleaseSelector{w}}), WithSelectedRestarts(restart, nil)); err != nil { |
| + t.Fatal(err) |
| + } |
| + rt, err := reg.Get("app.example") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if stale && (effects != 0 || rt.Version != "") || !stale && (effects != 1 || rt.Version != "v1") { |
| + t.Fatalf("effects=%d version=%s", effects, rt.Version) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestPassClaimRollbackStartRetainsSelection(t *testing.T) { |
| + s := NewDirStore(t.TempDir()) |
| + priv, pub := testKeys(t) |
| + reg := testRegistry(t) |
| + t.Cleanup(func() { reg.Close() }) |
| + opt := t.TempDir() |
| + for _, v := range []string{"v1", "v2"} { |
| + shipRelease(t, s, "bab", "myapp", v, adoptPlatform, []byte(v)) |
| + } |
| + 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(0) |
| + var first hostwrite.Selection |
| + starts, callbacks := 0, 0 |
| + restart := func(ctx context.Context, s hostwrite.Selection) error { |
| + first = s |
| + return s.With(ctx, func() error { return errors.New("restart failed") }) |
| + } |
| + start := func(ctx context.Context, s hostwrite.Selection) error { |
| + callbacks++ |
| + if s.Work != first.Work || s.Identity != first.Identity || s.Host != first.Host { |
| + t.Fatal("rollback recaptured authority") |
| + } |
| + w.valid.Store(false) |
| + return s.With(ctx, func() error { starts++; return nil }) |
| + } |
| + if err := Pass(t.Context(), reg, s, pub, opt, func(string) error { t.Fatal("legacy restart"); return nil }, WithHostWrites(hostwrite.Config{Enabled: true, Selector: oneReleaseSelector{w}}), WithSelectedRestarts(restart, start), WithStartUnit(func(string) error { t.Fatal("legacy start"); return nil })); err != nil { |
| + t.Fatal(err) |
| + } |
| + if callbacks != 1 || starts != 0 { |
| + t.Fatalf("callbacks=%d starts=%d", callbacks, starts) |
| + } |
| + if live, err := LiveVersion(opt, "app.example"); err != nil || live != "v1" { |
| + t.Fatalf("rollback=%s %v", live, err) |
| + } |
| +} |