amadan

carlos / platform Public

moves: S2 runs PrepareStrict's own preflight, held publication rechecks its authority under the frozen holds, failed strict cleanup is retried (Task 7b fix round 1)

- cgroup: strictRoot is PrepareStrict's host-independent runtime check
  (delegated root, agent placed in it or its supervise child, a cgroup2
  root); StrictPreflight exposes the same function. NextSpawnStrict uses it,
  so S2 refuses a box whose agent is misplaced, and also refuses while the
  host's last strict group awaits cleanup.
- release: MaterializeHeld is StageHeld then PublishHeld; PointLive runs
  inside a required guard. ServeHeldStageRequests asks its authority again
  around publication and drops a request refused there as stale.
- host: heldStageAuthority re-runs the full check after staging, then with
  the registry holds frozen re-checks the hold from the frozen rows and
  re-reads the claim (2 s bound, under the 5 s busy_timeout); publication
  runs last, inside the freeze.
- activator: a strict group whose drain or close fails keeps its handle,
  retried on a bounded schedule (a drain that succeeded is not repeated);
  the host's strict spawns are refused until a retry succeeds.
- copy: MV-N99 reworded; MV-N108 to MV-N110, pending.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude dfeee62736a9ea958586714572f54f2ef7115e1f parent f0e760a
10 files changed, +868 −157
  • docs/superpowers/plans/2026-10-06-moves-phase1-copy.md +4 −1
  • internal/activator/backend_exec.go +106 −25
  • internal/activator/backend_exec_test.go +192 −18
  • internal/cgroup/strict_linux.go +46 −12
  • internal/cgroup/strict_linux_test.go +57 −0
  • internal/cgroup/strict_other.go +1 −0
  • internal/host/heldstage.go +134 −58
  • internal/host/heldstage_test.go +165 −4
  • internal/release/materializeheld.go +78 −26
  • internal/release/materializeheld_test.go +85 −13
diff --git a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
index ddeff3ade..38cb3aee4 100644
--- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
+++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
@@ -138,7 +138,7 @@ Columns: id, file, text, status.
| MV-N96 | internal/activator/backend_exec.go | activator: a host in a move spawns only under strict containment | pending |
| MV-N97 | internal/activator/backend_exec.go | S2: the next spawn of this host is not strict | pending |
| MV-N98 | internal/activator/backend_exec.go | S2: this box has no delegated cgroup subtree, so it cannot spawn strictly | pending |
-| MV-N99 | internal/activator/backend_exec.go | S2: this build cannot spawn strictly | pending |
+| MV-N99 | internal/activator/backend_exec.go | S2: this box cannot spawn strictly | pending |
| MV-N100 | internal/release/materializeheld.go | release: this process cannot write the version store, so the root stage pass stages this release | pending |
| MV-N101 | internal/release/materializeheld.go | release: held stage request no longer matches its move | pending |
| MV-N102 | internal/release/materializeheld.go | release: held rows are not staged while CARLOS_HOST_CLAIM is on | pending |
@@ -147,6 +147,9 @@ Columns: id, file, text, status.
| MV-N105 | internal/host/systemapply.go | moves protocol floor: another writer holds the floor lock on this box; this waits for the next run | pending |
| MV-N106 | internal/host/systemapply.go | <error>; the root passes paused for the restore could not be started again: <error>, and the retry could not be recorded: <error> | pending |
| MV-N107 | internal/host/systemapply.go | <error>; the journal stays: <error> | pending |
+| MV-N108 | internal/activator/backend_exec.go | S2: the agent is not placed in its delegated cgroup subtree, so it cannot spawn strictly | pending |
+| MV-N109 | internal/activator/backend_exec.go | S2: this host's last strict group has not been cleaned up yet | pending |
+| MV-N110 | internal/release/materializeheld.go | release: a held publication needs its authority | pending |
## Plan 2
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 972e63751..16b8fe122 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -14,7 +14,6 @@ import (
"os"
"os/exec"
"path/filepath"
- "runtime"
"sort"
"strings"
"sync"
@@ -174,6 +173,15 @@ type ExecBackend struct {
// and a nil Cgroups under Strict is ErrStrictRequired. Tests replace
// this seam, never the Cgroups field.
prepareStrict func(host string, limit cgroup.Limit) (strictProc, error)
+ // strictPreflight is NextSpawnStrict's seam for PrepareStrict's runtime
+ // preflight. Nil: Cgroups.StrictPreflight, the very function
+ // PrepareStrict runs, so S2 and the spawn cannot disagree.
+ strictPreflight func() error
+ // strictPending holds, per host, a strict group whose drain or close
+ // failed on exit, with its handle, until a retry cleans it up. While a
+ // host has one, its strict spawns are refused.
+ strictMu sync.Mutex
+ strictPending map[string]*pendingStrict
}
// ErrStrictRequired is a spawn refused because its host is in a move and
@@ -193,17 +201,16 @@ type strictProc interface {
var _ strictProc = (*cgroup.Prepared)(nil)
-// strictBuild reports whether this build can prepare a strict group at all:
-// cgroup.PrepareStrict is Linux-only and answers ErrStrictUnsupported
-// elsewhere. A variable so a test can take the other branch.
-var strictBuild = runtime.GOOS == "linux"
-
-// strictDrainTimeout bounds the KillAndEmpty a strict child's exit runs.
-// A group still populated after it keeps its reservation (Close refuses),
-// so the host's next strict spawn is refused rather than sharing a group
-// with an unaccounted process.
+// strictDrainTimeout bounds each KillAndEmpty a strict group's cleanup runs.
var strictDrainTimeout = 30 * time.Second
+// strictCleanupBackoff is the bounded retry schedule for a strict group
+// whose cleanup failed on exit: one more attempt after each wait. When it
+// runs out the handle is still kept, and the host's strict spawns stay
+// refused until the agent restarts: a group that never drained may still
+// hold a process nothing accounts for.
+var strictCleanupBackoff = []time.Duration{time.Second, 5 * time.Second, 15 * time.Second, 30 * time.Second, time.Minute, 2 * time.Minute, 5 * time.Minute, 10 * time.Minute}
+
// execLogLineMax caps one teed log line (spec §5): longer is truncated with
// a marker, never split into phantom lines.
const execLogLineMax = 8 << 10
@@ -258,16 +265,19 @@ func (b *ExecBackend) spawnInCgroup(rt registry.Route, build func() *exec.Cmd) (
cmd, err := b.Cgroups.Start(rt.Host, b.tier(rt), build)
return cmd, func() {}, err
}
+ if b.strictCleanupPending(rt.Host) {
+ return nil, nil, fmt.Errorf("%s: %w: its last strict group has not been cleaned up yet", rt.Host, ErrStrictRequired)
+ }
sp, err := b.prepareStrictProc(rt.Host, b.tier(rt))
if err != nil {
return nil, nil, fmt.Errorf("%s: %w: %v", rt.Host, ErrStrictRequired, err)
}
cmd := build()
if err := sp.Start(cmd); err != nil {
- settleStrict(rt.Host, sp)
+ b.settleStrict(rt.Host, sp)
return nil, nil, fmt.Errorf("%s: %w: %v", rt.Host, ErrStrictRequired, err)
}
- return cmd, func() { settleStrict(rt.Host, sp) }, nil
+ return cmd, func() { b.settleStrict(rt.Host, sp) }, nil
}
// prepareStrictProc is prepareStrict with its default applied.
@@ -285,32 +295,103 @@ func (b *ExecBackend) prepareStrictProc(host string, limit cgroup.Limit) (strict
return p, nil
}
-// settleStrict drains a strict child's group and closes the handle. A
-// failure is logged and leaves the handle's reservation in place, which
-// refuses the host's next strict spawn on this agent: the group may still
-// hold a process nothing accounts for.
-func settleStrict(host string, sp strictProc) {
- ctx, cancel := context.WithTimeout(context.Background(), strictDrainTimeout)
- defer cancel()
- if err := sp.KillAndEmpty(ctx); err != nil {
- log.Printf("activator: %s: strict group did not drain: %v", host, err)
+// pendingStrict is a strict group whose cleanup has not succeeded yet.
+// drained records a KillAndEmpty that already succeeded: the handle refuses
+// a second one, so a retry after a failed Close only closes.
+type pendingStrict struct {
+ sp strictProc
+ drained bool
+}
+
+// attempt is one cleanup attempt: drain (unless already drained), then
+// close.
+func (p *pendingStrict) attempt(host string) error {
+ if !p.drained {
+ ctx, cancel := context.WithTimeout(context.Background(), strictDrainTimeout)
+ err := p.sp.KillAndEmpty(ctx)
+ cancel()
+ if err != nil {
+ log.Printf("activator: %s: strict group did not drain: %v", host, err)
+ return err
+ }
+ p.drained = true
}
- if err := sp.Close(); err != nil {
+ if err := p.sp.Close(); err != nil {
log.Printf("activator: %s: strict group not closed: %v", host, err)
+ return err
+ }
+ return nil
+}
+
+// settleStrict drains a strict child's group and closes the handle. On a
+// failure the handle is kept as the host's pending cleanup, retried on
+// strictCleanupBackoff, and the host's strict spawns are refused until a
+// retry succeeds.
+func (b *ExecBackend) settleStrict(host string, sp strictProc) {
+ p := &pendingStrict{sp: sp}
+ if p.attempt(host) == nil {
+ return
+ }
+ b.strictMu.Lock()
+ if b.strictPending == nil {
+ b.strictPending = map[string]*pendingStrict{}
+ }
+ b.strictPending[host] = p
+ b.strictMu.Unlock()
+ go b.retryStrictCleanup(host, p, append([]time.Duration(nil), strictCleanupBackoff...))
+}
+
+// retryStrictCleanup retries p's cleanup once after each wait in schedule,
+// a copy taken when the cleanup was queued.
+func (b *ExecBackend) retryStrictCleanup(host string, p *pendingStrict, schedule []time.Duration) {
+ for i, wait := range schedule {
+ time.Sleep(wait)
+ if p.attempt(host) == nil {
+ b.strictMu.Lock()
+ if b.strictPending[host] == p {
+ delete(b.strictPending, host)
+ }
+ b.strictMu.Unlock()
+ log.Printf("activator: %s: strict group cleaned up on retry %d", host, i+1)
+ return
+ }
}
+ log.Printf("activator: %s: strict group still not cleaned up after %d retries; strict spawns of this host stay refused until the agent restarts", host, len(schedule))
+}
+
+// strictCleanupPending reports whether host has a strict group awaiting
+// cleanup.
+func (b *ExecBackend) strictCleanupPending(host string) bool {
+ b.strictMu.Lock()
+ defer b.strictMu.Unlock()
+ return b.strictPending[host] != nil
}
// NextSpawnStrict is the source's S2 input (moves design §7.3, §5.3): will
// host's next spawn be strict? OK only when Strict names the host, the box
-// has a delegated cgroup subtree, and this build can prepare a strict group.
+// has a delegated cgroup subtree, PrepareStrict's own runtime preflight
+// passes (the agent placed in that subtree or its supervise child, on a
+// real cgroup2 mount, on a build that supports it), and no earlier strict
+// group of the host awaits cleanup. The per-host group state PrepareStrict
+// also judges (a populated group) is the idle stop's to clear.
func (b *ExecBackend) NextSpawnStrict(host string) moves.Check {
switch {
case b.Strict == nil || !b.Strict(host):
return moves.Check{Reason: "S2: the next spawn of this host is not strict"}
case b.Cgroups.Root() == "":
return moves.Check{Reason: "S2: this box has no delegated cgroup subtree, so it cannot spawn strictly"}
- case !strictBuild:
- return moves.Check{Reason: "S2: this build cannot spawn strictly"}
+ case b.strictCleanupPending(host):
+ return moves.Check{Reason: "S2: this host's last strict group has not been cleaned up yet"}
+ }
+ preflight := b.strictPreflight
+ if preflight == nil {
+ preflight = b.Cgroups.StrictPreflight
+ }
+ if err := preflight(); err != nil {
+ if errors.Is(err, cgroup.ErrStrictRefused) {
+ return moves.Check{Reason: "S2: the agent is not placed in its delegated cgroup subtree, so it cannot spawn strictly"}
+ }
+ return moves.Check{Reason: "S2: this box cannot spawn strictly"}
}
return moves.Check{OK: true}
}
diff --git a/internal/activator/backend_exec_test.go b/internal/activator/backend_exec_test.go
index 5ffbacf2a..8f57b4525 100644
--- a/internal/activator/backend_exec_test.go
+++ b/internal/activator/backend_exec_test.go
@@ -7,6 +7,7 @@ import (
"os"
"os/exec"
"path/filepath"
+ "runtime"
"strings"
"sync"
"testing"
@@ -540,10 +541,10 @@ func (f *failingStrict) Start(*exec.Cmd) error {
return cgroup.ErrStrictRefused
}
-// TestNextSpawnStrict is S2's input: OK only when the host is in a move
-// (Strict), the box has a delegated cgroup subtree, and this build can
-// prepare a strict group.
-func TestNextSpawnStrict(t *testing.T) {
+// delegatedManager is a cgroup manager over a temp-dir fixture: delegation
+// "exists" (Root() is set), but the test process is not placed in it.
+func delegatedManager(t *testing.T) *cgroup.Manager {
+ t.Helper()
root := t.TempDir()
for _, f := range []string{"cgroup.procs", "cgroup.subtree_control"} {
if err := os.WriteFile(filepath.Join(root, f), nil, 0o644); err != nil {
@@ -554,26 +555,40 @@ func TestNextSpawnStrict(t *testing.T) {
if err != nil {
t.Fatal(err)
}
+ return m
+}
+
+// TestNextSpawnStrict is S2's input: OK only when the host is in a move
+// (Strict), the box has a delegated cgroup subtree, PrepareStrict's own
+// runtime preflight passes (cgroup.Manager.StrictPreflight, the same
+// function PrepareStrict runs, so S2 cannot approve a box PrepareStrict
+// refuses), and no earlier strict group of the host awaits cleanup.
+func TestNextSpawnStrict(t *testing.T) {
+ m := delegatedManager(t)
host := "s2.example.com"
yes := func(string) bool { return true }
+ ok := func() error { return nil }
for name, tc := range map[string]struct {
- strict func(string) bool
- cg *cgroup.Manager
- build bool
- ok bool
- inWords string
+ strict func(string) bool
+ cg *cgroup.Manager
+ preflight func() error
+ ok bool
+ inWords string
}{
- "in a move, delegated, supported": {yes, m, true, true, ""},
- "not in a move": {func(string) bool { return false }, m, true, false, "not strict"},
- "Strict unset": {nil, m, true, false, "not strict"},
- "no delegation": {yes, nil, true, false, "delegated"},
- "unsupported build": {yes, m, false, false, "build"},
+ "in a move, delegated, preflight passes": {yes, m, ok, true, ""},
+ "not in a move": {func(string) bool { return false }, m, ok, false, "not strict"},
+ "Strict unset": {nil, m, ok, false, "not strict"},
+ "no delegation": {yes, nil, ok, false, "delegated"},
+ // The real preflight, no seam: delegation exists, but this test
+ // process is not placed in the fixture's subtree.
+ "delegated, agent misplaced": {yes, m, nil, false, "not placed"},
+ "preflight says unsupported": {yes, m, func() error { return cgroup.ErrStrictUnsupported }, false, "cannot spawn strictly"},
} {
t.Run(name, func(t *testing.T) {
- orig := strictBuild
- strictBuild = tc.build
- t.Cleanup(func() { strictBuild = orig })
- b := ExecBackend{Strict: tc.strict, Cgroups: tc.cg}
+ if tc.preflight == nil && runtime.GOOS != "linux" {
+ t.Skip("the real preflight answers unsupported off Linux")
+ }
+ b := &ExecBackend{Strict: tc.strict, Cgroups: tc.cg, strictPreflight: tc.preflight}
c := b.NextSpawnStrict(host)
if c.OK != tc.ok || (!c.OK && !strings.Contains(c.Reason, tc.inWords)) || (c.OK && c.Reason != "") {
t.Fatalf("NextSpawnStrict = %+v, want ok=%v naming %q", c, tc.ok, tc.inWords)
@@ -581,3 +596,162 @@ func TestNextSpawnStrict(t *testing.T) {
})
}
}
+
+// flakyStrict is a recordingStrict whose KillAndEmpty fails its first
+// killFails calls and whose Close fails its first closeFails calls, the
+// way a drain that outlives its deadline or a group still settling does.
+type flakyStrict struct {
+ *recordingStrict
+ killFails, closeFails int
+}
+
+func (f *flakyStrict) KillAndEmpty(ctx context.Context) error {
+ f.mu.Lock()
+ fail := f.killFails > 0
+ if fail {
+ f.killFails--
+ }
+ f.mu.Unlock()
+ if fail {
+ f.note("kill-and-empty failed")
+ return cgroup.ErrStrictDeadline
+ }
+ return f.recordingStrict.KillAndEmpty(ctx)
+}
+
+func (f *flakyStrict) Close() error {
+ f.mu.Lock()
+ fail := f.closeFails > 0
+ if fail {
+ f.closeFails--
+ }
+ f.mu.Unlock()
+ if fail {
+ f.note("close failed")
+ return cgroup.ErrStrictRefused
+ }
+ return f.recordingStrict.Close()
+}
+
+// TestStrictCleanupIsRetriedAndRefusesUntilItSucceeds: a strict group whose
+// drain or close fails on exit keeps its handle; cleanup is retried on a
+// bounded schedule, and until it succeeds the host's strict spawns are
+// refused (and S2 says so) without preparing a new group. Once it
+// succeeds, strict spawns work again. A drain that succeeded is not
+// repeated when only the close failed.
+func TestStrictCleanupIsRetriedAndRefusesUntilItSucceeds(t *testing.T) {
+ old := spawnHealthTimeout
+ spawnHealthTimeout = 300 * time.Millisecond
+ t.Cleanup(func() { spawnHealthTimeout = old })
+ host := "retry.example.com"
+ for name, tc := range map[string]struct {
+ killFails, closeFails int
+ want string
+ }{
+ "the drain fails once": {1, 0, "start,kill-and-empty failed,kill-and-empty,close"},
+ "the close fails once": {0, 1, "start,kill-and-empty,close failed,close"},
+ } {
+ t.Run(name, func(t *testing.T) {
+ oldBackoff := strictCleanupBackoff
+ strictCleanupBackoff = []time.Duration{400 * time.Millisecond, 400 * time.Millisecond}
+ t.Cleanup(func() { strictCleanupBackoff = oldBackoff })
+ b := testExecBackend(t)
+ b.Cgroups = delegatedManager(t)
+ b.strictPreflight = func() error { return nil }
+ argvPath, _ := stageRecordingLive(t, b, host, "v1")
+ b.Strict = func(string) bool { return true }
+ var mu sync.Mutex
+ var handles []*flakyStrict
+ b.prepareStrict = func(string, cgroup.Limit) (strictProc, error) {
+ mu.Lock()
+ defer mu.Unlock()
+ h := &flakyStrict{recordingStrict: newRecordingStrict()}
+ if len(handles) == 0 {
+ h.killFails, h.closeFails = tc.killFails, tc.closeFails
+ }
+ handles = append(handles, h)
+ return h, nil
+ }
+ prepared := func() int { mu.Lock(); defer mu.Unlock(); return len(handles) }
+ startOnce := func() error {
+ rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "i.sock")}
+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ defer cancel()
+ proc, _, err := b.Start(ctx, rt)
+ if proc != nil {
+ _ = proc.Stop(context.Background())
+ }
+ return err
+ }
+
+ _ = startOnce()
+ waitForFile(t, argvPath, time.Second)
+ deadline := time.Now().Add(2 * time.Second)
+ for !b.strictCleanupPending(host) {
+ if time.Now().After(deadline) {
+ t.Fatalf("no cleanup pending after a failed drain: %v", handles[0].seen())
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ if c := b.NextSpawnStrict(host); c.OK || !strings.Contains(c.Reason, "cleaned up") {
+ t.Fatalf("S2 while cleanup is pending: %+v", c)
+ }
+ if err := startOnce(); !errors.Is(err, ErrStrictRequired) {
+ t.Fatalf("spawn while cleanup is pending: %v, want ErrStrictRequired", err)
+ }
+ if n := prepared(); n != 1 {
+ t.Fatalf("a new strict group was prepared while the old one awaited cleanup (%d)", n)
+ }
+
+ deadline = time.Now().Add(3 * time.Second)
+ for b.strictCleanupPending(host) {
+ if time.Now().After(deadline) {
+ t.Fatalf("cleanup never succeeded on retry: %v", handles[0].seen())
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ if got := strings.Join(handles[0].seen(), ","); got != tc.want {
+ t.Fatalf("first handle's calls = %s, want %s", got, tc.want)
+ }
+ if c := b.NextSpawnStrict(host); !c.OK {
+ t.Fatalf("S2 after cleanup: %+v", c)
+ }
+ if err := os.Remove(argvPath); err != nil {
+ t.Fatal(err)
+ }
+ _ = startOnce()
+ waitForFile(t, argvPath, time.Second)
+ if n := prepared(); n != 2 {
+ t.Fatalf("strict groups prepared = %d, want a second one after cleanup", n)
+ }
+ })
+ }
+}
+
+// TestStrictCleanupGivesUpButStaysRefused: the retry schedule is bounded.
+// When it runs out the handle is kept and the host's strict spawns stay
+// refused: a group that never drained may still hold a process.
+func TestStrictCleanupGivesUpButStaysRefused(t *testing.T) {
+ oldBackoff := strictCleanupBackoff
+ strictCleanupBackoff = []time.Duration{10 * time.Millisecond, 10 * time.Millisecond}
+ t.Cleanup(func() { strictCleanupBackoff = oldBackoff })
+ host := "stuck.example.com"
+ b := testExecBackend(t)
+ h := &flakyStrict{recordingStrict: newRecordingStrict(), killFails: 100}
+ b.settleStrict(host, h)
+ time.Sleep(200 * time.Millisecond)
+ if got := strings.Count(strings.Join(h.seen(), ","), "kill-and-empty failed"); got != 3 {
+ t.Fatalf("drain attempts = %d, want the first and two retries", got)
+ }
+ if !b.strictCleanupPending(host) {
+ t.Fatal("the handle was dropped after the schedule ran out")
+ }
+ b.Strict = func(string) bool { return true }
+ b.prepareStrict = func(string, cgroup.Limit) (strictProc, error) {
+ t.Fatal("a strict group was prepared for a host awaiting cleanup")
+ return nil, nil
+ }
+ if _, _, err := b.spawnInCgroup(registry.Route{Host: host}, func() *exec.Cmd { return exec.Command("/bin/true") }); !errors.Is(err, ErrStrictRequired) {
+ t.Fatalf("spawn: %v, want ErrStrictRequired", err)
+ }
+}
diff --git a/internal/cgroup/strict_linux.go b/internal/cgroup/strict_linux.go
index 076c8cbc9..37a057902 100644
--- a/internal/cgroup/strict_linux.go
+++ b/internal/cgroup/strict_linux.go
@@ -146,16 +146,19 @@ func (p *Prepared) closeFiles() {
}
}
-// PrepareStrict never uses accounting's fallback. Only the two existing tiers
-// are accepted; a zero or missing limit cannot create an uncapped command.
-func (m *Manager) PrepareStrict(host string, limit Limit) (out *Prepared, err error) {
+// selfCgroupFile is where the strict preflight reads this process's own
+// cgroup. A variable so a test can place the agent where it needs.
+var selfCgroupFile = "/proc/self/cgroup"
+
+// strictRoot is the host-independent half of PrepareStrict's runtime
+// checks, and the whole of StrictPreflight: a delegated root, this process
+// placed in it or in its supervise child, and the root a real cgroup2
+// directory. It returns the root opened without following a symlink.
+func (m *Manager) strictRoot() (*os.File, error) {
if m == nil || m.root == "" {
return nil, ErrStrictUnsupported
}
- if !controlname.ValidHostShape(host) || (limit != TierDefault && limit != TierChildren) {
- return nil, ErrStrictRefused
- }
- own, e := os.ReadFile("/proc/self/cgroup")
+ own, e := os.ReadFile(selfCgroupFile)
rel, ok := ParseSelfCgroup(string(own))
if e != nil || !ok {
return nil, ErrStrictUnsupported
@@ -165,10 +168,45 @@ func (m *Manager) PrepareStrict(host string, limit Limit) (out *Prepared, err er
if current != rootPath && current != filepath.Join(rootPath, "supervise") {
return nil, ErrStrictRefused
}
- root, e := os.OpenFile(filepath.Clean(m.root), os.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW, 0)
+ root, e := os.OpenFile(rootPath, os.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW, 0)
if e != nil {
return nil, ErrStrictUnsupported
}
+ var fs unix.Statfs_t
+ if e = unix.Fstatfs(int(root.Fd()), &fs); e != nil || uint64(fs.Type) != uint64(unix.CGROUP2_SUPER_MAGIC) {
+ root.Close()
+ return nil, ErrStrictUnsupported
+ }
+ return root, nil
+}
+
+// StrictPreflight reports whether PrepareStrict's runtime checks that do
+// not depend on the host pass now: the same function, not a copy, so a
+// readiness check (moves S2) cannot approve a box PrepareStrict refuses
+// for its placement or its mount. Nil means a strict spawn can be
+// attempted; the per-host state (a populated group, a held reservation)
+// is still PrepareStrict's to judge.
+func (m *Manager) StrictPreflight() error {
+ root, e := m.strictRoot()
+ if e != nil {
+ return e
+ }
+ return root.Close()
+}
+
+// PrepareStrict never uses accounting's fallback. Only the two existing tiers
+// are accepted; a zero or missing limit cannot create an uncapped command.
+func (m *Manager) PrepareStrict(host string, limit Limit) (out *Prepared, err error) {
+ if m == nil || m.root == "" {
+ return nil, ErrStrictUnsupported
+ }
+ if !controlname.ValidHostShape(host) || (limit != TierDefault && limit != TierChildren) {
+ return nil, ErrStrictRefused
+ }
+ root, e := m.strictRoot()
+ if e != nil {
+ return nil, e
+ }
p := &Prepared{strictPreparedState: &strictPreparedState{root: root, host: host, limits: map[string]*os.File{}, start: func(cmd *exec.Cmd) error { return cmd.Start() }}}
reserved := false
defer func() {
@@ -181,10 +219,6 @@ func (m *Manager) PrepareStrict(host string, limit Limit) (out *Prepared, err er
}
}
}()
- var fs unix.Statfs_t
- if e = unix.Fstatfs(int(root.Fd()), &fs); e != nil || uint64(fs.Type) != uint64(unix.CGROUP2_SUPER_MAGIC) {
- return nil, ErrStrictUnsupported
- }
if e = unix.Mkdirat(int(root.Fd()), ExecName(host), 0755); e != nil && !errors.Is(e, unix.EEXIST) {
return nil, ErrStrictRefused
}
diff --git a/internal/cgroup/strict_linux_test.go b/internal/cgroup/strict_linux_test.go
index d8d8f3908..486127ffb 100644
--- a/internal/cgroup/strict_linux_test.go
+++ b/internal/cgroup/strict_linux_test.go
@@ -12,6 +12,8 @@ import (
"syscall"
"testing"
"time"
+
+ "golang.org/x/sys/unix"
)
// Filesystem fixtures test handle state only. PrepareStrict must reject them;
@@ -269,3 +271,58 @@ func TestStrictConcurrentAndInvalidStartsConsumeAttempt(t *testing.T) {
}
}
}
+
+// fakeSelfCgroup points the strict preflight's /proc/self/cgroup read at a
+// file holding line, so a test can place the agent wherever it needs.
+func fakeSelfCgroup(t *testing.T, line string) {
+ t.Helper()
+ path := filepath.Join(t.TempDir(), "cgroup")
+ if e := os.WriteFile(path, []byte(line+"\n"), 0o644); e != nil {
+ t.Fatal(e)
+ }
+ orig := selfCgroupFile
+ selfCgroupFile = path
+ t.Cleanup(func() { selfCgroupFile = orig })
+}
+
+// TestStrictPreflightIsPrepareStrictsRuntimeCheck: StrictPreflight answers
+// what PrepareStrict's host-independent runtime checks answer, because it
+// is the same function. Delegation present but the agent placed outside
+// the delegated root or its supervise child is refused by both; placed in
+// either, the preflight passes on a real cgroup2 mount.
+func TestStrictPreflightIsPrepareStrictsRuntimeCheck(t *testing.T) {
+ var nilManager *Manager
+ if e := nilManager.StrictPreflight(); !errors.Is(e, ErrStrictUnsupported) {
+ t.Fatalf("nil manager: %v", e)
+ }
+ var fs unix.Statfs_t
+ if e := unix.Statfs("/sys/fs/cgroup", &fs); e != nil || uint64(fs.Type) != uint64(unix.CGROUP2_SUPER_MAGIC) {
+ t.Skip("no cgroup2 mount at /sys/fs/cgroup")
+ }
+ m := &Manager{root: "/sys/fs/cgroup/carlos.slice/carlos-edge.service"}
+ fakeSelfCgroup(t, "0::/user.slice/elsewhere.service")
+ if e := m.StrictPreflight(); !errors.Is(e, ErrStrictRefused) {
+ t.Fatalf("misplaced agent: preflight %v, want ErrStrictRefused", e)
+ }
+ if p, e := m.PrepareStrict("x.example", TierDefault); p != nil || !errors.Is(e, ErrStrictRefused) {
+ t.Fatalf("misplaced agent: PrepareStrict %v, want the preflight's ErrStrictRefused", e)
+ }
+ // The test's own cgroup stands in for the delegated root: a real
+ // cgroup2 directory the agent is placed in, or in its supervise child.
+ raw, e := os.ReadFile("/proc/self/cgroup")
+ rel, ok := ParseSelfCgroup(string(raw))
+ if e != nil || !ok {
+ t.Skip("no cgroup v2 placement for this test process")
+ }
+ own := &Manager{root: filepath.Join("/sys/fs/cgroup", rel)}
+ for _, line := range []string{"0::/" + rel, "0::/" + rel + "/supervise"} {
+ fakeSelfCgroup(t, line)
+ if e := own.StrictPreflight(); e != nil {
+ t.Fatalf("agent at %q: preflight %v, want nil", line, e)
+ }
+ }
+ fakeSelfCgroup(t, "0::/"+rel)
+ if e := (&Manager{root: t.TempDir()}).StrictPreflight(); !errors.Is(e, ErrStrictRefused) {
+ t.Fatalf("root outside the agent's placement: %v", e)
+ }
+}
diff --git a/internal/cgroup/strict_other.go b/internal/cgroup/strict_other.go
index b5c10bd57..34ddee7cb 100644
--- a/internal/cgroup/strict_other.go
+++ b/internal/cgroup/strict_other.go
@@ -12,6 +12,7 @@ type Prepared struct{}
func (m *Manager) PrepareStrict(host string, limit Limit) (*Prepared, error) {
return nil, ErrStrictUnsupported
}
+func (m *Manager) StrictPreflight() error { return ErrStrictUnsupported }
func (p *Prepared) Start(cmd *exec.Cmd) error { return ErrStrictUnsupported }
func (p *Prepared) Populated() (bool, error) { return false, ErrStrictUnsupported }
func (p *Prepared) KillAndEmpty(ctx context.Context) error { return ErrStrictUnsupported }
diff --git a/internal/host/heldstage.go b/internal/host/heldstage.go
index 4c9696a3b..23621d040 100644
--- a/internal/host/heldstage.go
+++ b/internal/host/heldstage.go
@@ -18,6 +18,11 @@ import (
// heldStageReadTimeout bounds the bucket reads one authorization makes.
const heldStageReadTimeout = 30 * time.Second
+// heldStageFrozenReadTimeout bounds the one claim read made while the
+// registry's write lock is held: well under the registry's 5 s busy_timeout,
+// so a SetHold or ReleaseHold waiting on the freeze is delayed, never failed.
+const heldStageFrozenReadTimeout = 2 * time.Second
+
// heldStageAuthority is the authorize closure the root adopt pass hands
// release.ServeHeldStageRequests (moves design §7.3, T7). A held-stage
// request is honoured only for the move this box is the target of, right
@@ -37,69 +42,140 @@ const heldStageReadTimeout = 30 * time.Second
//
// Any mismatch is release.ErrStaleHeldStage, so a request queued for an
// operation that was aborted or replaced never repoints live/<host>. A
-// read that fails is returned as itself: the request is dropped either way,
-// and the agent queues it again on its next round.
-func heldStageAuthority(reg *registry.Registry, store release.Store, box string) func(release.HeldStageRequest) error {
- return func(req release.HeldStageRequest) error {
- stale := func(format string, args ...any) error {
- return fmt.Errorf("%w: "+format, append([]any{release.ErrStaleHeldStage}, args...)...)
- }
- if !moves.ValidOpID(req.Op) {
- return stale("malformed op id")
- }
- if box == "" || req.Box != box {
- return stale("it names box %q and this is %q", req.Box, box)
- }
- rt, err := reg.Get(req.Host)
- if err != nil {
- return fmt.Errorf("registry row for %s: %w", req.Host, err)
+// read that fails is returned as itself.
+//
+// With a non-nil then (the publication), the whole check runs again, and
+// then, with the registry's holds frozen (WithHoldsFrozen, so no SetHold or
+// ReleaseHold can land in between), the hold is checked from the frozen
+// rows and the claim is re-read; then runs last, inside the freeze. Staging
+// takes minutes, so this is what keeps a move aborted or replaced meanwhile
+// from having its release published. The registry allows one connection,
+// so inside the freeze only the held rows it hands over are read: a row the
+// first check found held must still be held by the op there, and a row
+// that appeared since must be held by the op too. An unheld row appearing
+// where there was none would mean the move finished, which the claim
+// re-read refuses.
+func heldStageAuthority(reg *registry.Registry, store release.Store, box string) func(release.HeldStageRequest, func() error) error {
+ return func(req release.HeldStageRequest, then func() error) error {
+ rowHeld, err := heldStageCheck(reg, store, box, req)
+ if err != nil || then == nil {
+ return err
}
- if rt != nil {
- if rt.HoldOp != req.Op {
- return stale("the registry row is held by %q", rt.HoldOp)
+ heldStageBeforeFreeze()
+ return reg.WithHoldsFrozen(func(held []registry.Route) error {
+ var row *registry.Route
+ for i := range held {
+ if held[i].Host == req.Host {
+ row = &held[i]
+ }
}
- if release.RouteAccount(*rt) != req.Account || rt.App != req.App {
- return stale("the registry row is %s/%s", release.RouteAccount(*rt), rt.App)
+ switch {
+ case row != nil:
+ if err := heldStageRow(*row, req); err != nil {
+ return err
+ }
+ case rowHeld:
+ return staleHeldStage("the registry row is no longer held by %s", req.Op)
}
+ ctx, cancel := context.WithTimeout(context.Background(), heldStageFrozenReadTimeout)
+ defer cancel()
+ obs, err := hostclaim.Read(ctx, store, req.Host)
+ if err != nil {
+ return fmt.Errorf("host claim for %s: %w", req.Host, err)
+ }
+ if err := heldStageClaim(obs, req); err != nil {
+ return err
+ }
+ return then()
+ })
+ }
+}
+
+// heldStageBeforeFreeze runs between the full check and the freeze. It is
+// a test seam: the window a hold or claim change can land in.
+var heldStageBeforeFreeze = func() {}
+
+func staleHeldStage(format string, args ...any) error {
+ return fmt.Errorf("%w: "+format, append([]any{release.ErrStaleHeldStage}, args...)...)
+}
+
+// heldStageCheck is the authority's full check, outside any freeze. It
+// reports whether the host has a registry row (held by the op, or the check
+// would have refused).
+func heldStageCheck(reg *registry.Registry, store release.Store, box string, req release.HeldStageRequest) (bool, error) {
+ if !moves.ValidOpID(req.Op) {
+ return false, staleHeldStage("malformed op id")
+ }
+ if box == "" || req.Box != box {
+ return false, staleHeldStage("it names box %q and this is %q", req.Box, box)
+ }
+ rt, err := reg.Get(req.Host)
+ if err != nil {
+ return false, fmt.Errorf("registry row for %s: %w", req.Host, err)
+ }
+ if rt != nil {
+ if err := heldStageRow(*rt, req); err != nil {
+ return false, err
}
- ctx, cancel := context.WithTimeout(context.Background(), heldStageReadTimeout)
- defer cancel()
- obs, err := hostclaim.Read(ctx, store, req.Host)
- if err != nil {
- return fmt.Errorf("host claim for %s: %w", req.Host, err)
- }
- if obs.Absent || obs.Claim == nil {
- return stale("the host has no claim")
- }
- op := obs.Claim.Op
- if op.ID != req.Op {
- return stale("the claim names op %q", op.ID)
- }
- if op.Kind != hostclaim.KindMoving && op.Kind != hostclaim.KindLanding {
- return stale("the claim's op is %s", op.Kind)
- }
- if op.Intent["target"] != req.Box {
- return stale("the claim's target is %q", op.Intent["target"])
- }
- if h := obs.Claim.Holder; h == nil || h.Account != req.Account || h.App != req.App {
- return stale("the claim is not held by %s/%s", req.Account, req.App)
- }
- raw, _, err := store.Get(ctx, moves.SourceReportKey(moves.ReportPrefix(req.Account, req.App, req.Host, req.Op)))
- if errors.Is(err, os.ErrNotExist) {
- return stale("the source has not reported its release")
- }
- if err != nil {
- return fmt.Errorf("source report for %s: %w", req.Host, err)
- }
- var sr moves.SourceReport
- if err := json.Unmarshal(raw, &sr); err != nil {
- return fmt.Errorf("source report for %s: %w", req.Host, err)
- }
- if sr.Release == "" || sr.Release != req.Version {
- return stale("the source runs %q", sr.Release)
- }
- return nil
}
+ ctx, cancel := context.WithTimeout(context.Background(), heldStageReadTimeout)
+ defer cancel()
+ obs, err := hostclaim.Read(ctx, store, req.Host)
+ if err != nil {
+ return false, fmt.Errorf("host claim for %s: %w", req.Host, err)
+ }
+ if err := heldStageClaim(obs, req); err != nil {
+ return false, err
+ }
+ raw, _, err := store.Get(ctx, moves.SourceReportKey(moves.ReportPrefix(req.Account, req.App, req.Host, req.Op)))
+ if errors.Is(err, os.ErrNotExist) {
+ return false, staleHeldStage("the source has not reported its release")
+ }
+ if err != nil {
+ return false, fmt.Errorf("source report for %s: %w", req.Host, err)
+ }
+ var sr moves.SourceReport
+ if err := json.Unmarshal(raw, &sr); err != nil {
+ return false, fmt.Errorf("source report for %s: %w", req.Host, err)
+ }
+ if sr.Release == "" || sr.Release != req.Version {
+ return false, staleHeldStage("the source runs %q", sr.Release)
+ }
+ return rt != nil, nil
+}
+
+// heldStageRow admits a registry row: held by the request's op, of the
+// request's account and app.
+func heldStageRow(rt registry.Route, req release.HeldStageRequest) error {
+ if rt.HoldOp != req.Op {
+ return staleHeldStage("the registry row is held by %q", rt.HoldOp)
+ }
+ if release.RouteAccount(rt) != req.Account || rt.App != req.App {
+ return staleHeldStage("the registry row is %s/%s", release.RouteAccount(rt), rt.App)
+ }
+ return nil
+}
+
+// heldStageClaim admits a host claim: it names the op, is moving or
+// landing, targets the request's box, and is held by its account and app.
+func heldStageClaim(obs hostclaim.Observed, req release.HeldStageRequest) error {
+ if obs.Absent || obs.Claim == nil {
+ return staleHeldStage("the host has no claim")
+ }
+ op := obs.Claim.Op
+ if op.ID != req.Op {
+ return staleHeldStage("the claim names op %q", op.ID)
+ }
+ if op.Kind != hostclaim.KindMoving && op.Kind != hostclaim.KindLanding {
+ return staleHeldStage("the claim's op is %s", op.Kind)
+ }
+ if op.Intent["target"] != req.Box {
+ return staleHeldStage("the claim's target is %q", op.Intent["target"])
+ }
+ if h := obs.Claim.Holder; h == nil || h.Account != req.Account || h.App != req.App {
+ return staleHeldStage("the claim is not held by %s/%s", req.Account, req.App)
+ }
+ return nil
}
// heldStagePass serves the held-stage requests the sandboxed agent left
diff --git a/internal/host/heldstage_test.go b/internal/host/heldstage_test.go
index f60f3d0e2..a9e002782 100644
--- a/internal/host/heldstage_test.go
+++ b/internal/host/heldstage_test.go
@@ -100,7 +100,7 @@ func TestHeldStageAuthority(t *testing.T) {
cs := &hsCountingStore{Store: f.s}
req := hsRequest()
req.Op = "../" + hsOp
- if err := heldStageAuthority(f.reg, cs, hsBox)(req); !errors.Is(err, release.ErrStaleHeldStage) {
+ if err := heldStageAuthority(f.reg, cs, hsBox)(req, nil); !errors.Is(err, release.ErrStaleHeldStage) {
t.Fatalf("authorize: %v, want ErrStaleHeldStage", err)
}
if cs.n != 0 {
@@ -109,7 +109,7 @@ func TestHeldStageAuthority(t *testing.T) {
})
t.Run("absent row, moving claim, matching report", func(t *testing.T) {
f := good(t)
- if err := heldStageAuthority(f.reg, f.s, hsBox)(hsRequest()); err != nil {
+ if err := heldStageAuthority(f.reg, f.s, hsBox)(hsRequest(), nil); err != nil {
t.Fatalf("authorize: %v", err)
}
})
@@ -123,7 +123,7 @@ func TestHeldStageAuthority(t *testing.T) {
if err := reg.Upsert(registry.Route{Host: hsHost, Kind: "instance", App: "titogo", Account: "bab", Status: "landing", HoldOp: hsOp}); err != nil {
t.Fatal(err)
}
- if err := heldStageAuthority(reg, s, hsBox)(hsRequest()); err != nil {
+ if err := heldStageAuthority(reg, s, hsBox)(hsRequest(), nil); err != nil {
t.Fatalf("authorize: %v", err)
}
})
@@ -192,7 +192,7 @@ func TestHeldStageAuthority(t *testing.T) {
f := good(t)
req := hsRequest()
tc.setup(t, f, &req)
- if err := heldStageAuthority(f.reg, f.s, tc.box)(req); !errors.Is(err, release.ErrStaleHeldStage) {
+ if err := heldStageAuthority(f.reg, f.s, tc.box)(req, nil); !errors.Is(err, release.ErrStaleHeldStage) {
t.Fatalf("authorize: %v, want ErrStaleHeldStage", err)
}
})
@@ -292,3 +292,164 @@ func TestHeldStagePassServesAnAuthorizedRequest(t *testing.T) {
t.Fatalf("requests left behind: %s", strings.Join(names, ", "))
}
}
+
+// shipV7 ships bab/titogo v7 for this platform.
+func shipV7(t *testing.T, s release.Store) {
+ t.Helper()
+ plat := runtime.GOOS + "-" + runtime.GOARCH
+ m := release.Manifest{App: "titogo", Version: "v7", Kind: "binary", Artifacts: map[string]release.Artifact{plat: {}}}
+ if err := release.Ship(context.Background(), s, "bab", m, map[string][]byte{plat: []byte("pinned")}); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// TestHeldPublicationIsGuardedByTheAuthority: the authority that admitted a
+// request before staging is re-run around the publication. If the claim
+// moved to another op, or the registry hold was released, while the release
+// was staging, publication returns ErrStaleHeldStage and live/<host> is
+// never written. Unchanged, the publication happens inside the guard.
+func TestHeldPublicationIsGuardedByTheAuthority(t *testing.T) {
+ heldRow := registry.Route{Host: hsHost, Kind: "instance", App: "titogo", Account: "bab", Status: "landing", HoldOp: hsOp}
+ for name, tc := range map[string]struct {
+ row bool
+ during func(*testing.T, *registry.Registry, release.Store)
+ stale bool
+ }{
+ "unchanged, no row yet": {false, func(*testing.T, *registry.Registry, release.Store) {}, false},
+ "unchanged, held row": {true, func(*testing.T, *registry.Registry, release.Store) {}, false},
+ "the claim moves to another op": {false, func(t *testing.T, _ *registry.Registry, s release.Store) {
+ rewriteClaim(t, s, func(op *hostclaim.Op) { op.ID = hsOther })
+ }, true},
+ "the claim leaves the move": {true, func(t *testing.T, _ *registry.Registry, s release.Store) {
+ rewriteClaim(t, s, func(op *hostclaim.Op) { op.Kind = hostclaim.KindDeleting })
+ }, true},
+ "the hold is released": {true, func(t *testing.T, reg *registry.Registry, _ release.Store) {
+ if ok, err := reg.ReleaseHold(hsHost, hsOp); err != nil || !ok {
+ t.Fatalf("ReleaseHold: %v %v", ok, err)
+ }
+ }, true},
+ "the hold passes to another op": {true, func(t *testing.T, reg *registry.Registry, _ release.Store) {
+ if ok, err := reg.ReleaseHold(hsHost, hsOp); err != nil || !ok {
+ t.Fatalf("ReleaseHold: %v %v", ok, err)
+ }
+ if err := reg.SetHold(hsHost, "landing", hsOther); err != nil {
+ t.Fatal(err)
+ }
+ }, true},
+ } {
+ t.Run(name, func(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ hsClaim(t, s, movingOp())
+ hsReport(t, s, hsOp, "v7")
+ shipV7(t, s)
+ reg := testRegistry(t)
+ if tc.row {
+ if err := reg.Upsert(heldRow); err != nil {
+ t.Fatal(err)
+ }
+ }
+ opt := t.TempDir()
+ auth := heldStageAuthority(reg, s, hsBox)
+ req := hsRequest()
+ if err := auth(req, nil); err != nil {
+ t.Fatalf("authority before staging: %v", err)
+ }
+ rt := registry.Route{Host: req.Host, Account: req.Account, App: req.App}
+ if _, err := release.StageHeld(ctx, s, opt, rt, req.Version); err != nil {
+ t.Fatalf("StageHeld: %v", err)
+ }
+ tc.during(t, reg, s)
+ err := release.PublishHeld(opt, rt, req.Version, func(publish func() error) error { return auth(req, publish) })
+ got, _ := release.LiveVersion(opt, hsHost)
+ if tc.stale {
+ if !errors.Is(err, release.ErrStaleHeldStage) || got != "" {
+ t.Fatalf("PublishHeld: %v, live %q; want ErrStaleHeldStage and no live link", err, got)
+ }
+ return
+ }
+ if err != nil || got != "v7" {
+ t.Fatalf("PublishHeld: %v, live %q; want v7", err, got)
+ }
+ })
+ }
+}
+
+// TestHeldPublicationRechecksInsideTheFreeze: a change that lands after the
+// full check and before the freeze is caught by the re-check made with the
+// holds frozen, and the publication itself runs while they are frozen: a
+// ReleaseHold issued during it waits until it returns.
+func TestHeldPublicationRechecksInsideTheFreeze(t *testing.T) {
+ heldRow := registry.Route{Host: hsHost, Kind: "instance", App: "titogo", Account: "bab", Status: "landing", HoldOp: hsOp}
+ for name, tc := range map[string]struct {
+ row bool
+ window func(*testing.T, *registry.Registry, release.Store)
+ }{
+ "the hold is released": {true, func(t *testing.T, reg *registry.Registry, _ release.Store) {
+ if ok, err := reg.ReleaseHold(hsHost, hsOp); err != nil || !ok {
+ t.Fatalf("ReleaseHold: %v %v", ok, err)
+ }
+ }},
+ "a row appears held by another op": {false, func(t *testing.T, reg *registry.Registry, _ release.Store) {
+ r := heldRow
+ r.HoldOp = hsOther
+ if err := reg.Upsert(r); err != nil {
+ t.Fatal(err)
+ }
+ }},
+ "the claim moves to another op": {false, func(t *testing.T, _ *registry.Registry, s release.Store) {
+ rewriteClaim(t, s, func(op *hostclaim.Op) { op.ID = hsOther })
+ }},
+ "the claim retargets": {true, func(t *testing.T, _ *registry.Registry, s release.Store) {
+ rewriteClaim(t, s, func(op *hostclaim.Op) { op.Intent["target"] = "fr-box" })
+ }},
+ } {
+ t.Run(name, func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ hsClaim(t, s, movingOp())
+ hsReport(t, s, hsOp, "v7")
+ reg := testRegistry(t)
+ if tc.row {
+ if err := reg.Upsert(heldRow); err != nil {
+ t.Fatal(err)
+ }
+ }
+ orig := heldStageBeforeFreeze
+ t.Cleanup(func() { heldStageBeforeFreeze = orig })
+ heldStageBeforeFreeze = func() { tc.window(t, reg, s) }
+ published := false
+ err := heldStageAuthority(reg, s, hsBox)(hsRequest(), func() error { published = true; return nil })
+ if !errors.Is(err, release.ErrStaleHeldStage) || published {
+ t.Fatalf("authority: %v (published %v), want ErrStaleHeldStage before publishing", err, published)
+ }
+ })
+ }
+
+ t.Run("publication runs with the holds frozen", func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ hsClaim(t, s, movingOp())
+ hsReport(t, s, hsOp, "v7")
+ reg := testRegistry(t)
+ if err := reg.Upsert(heldRow); err != nil {
+ t.Fatal(err)
+ }
+ released := make(chan struct{})
+ err := heldStageAuthority(reg, s, hsBox)(hsRequest(), func() error {
+ go func() { _, _ = reg.ReleaseHold(hsHost, hsOp); close(released) }()
+ select {
+ case <-released:
+ return errors.New("a ReleaseHold landed during the publication")
+ case <-time.After(300 * time.Millisecond):
+ return nil
+ }
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ select {
+ case <-released:
+ case <-time.After(5 * time.Second):
+ t.Fatal("the ReleaseHold never ran after the freeze")
+ }
+ })
+}
diff --git a/internal/release/materializeheld.go b/internal/release/materializeheld.go
index 02b0f8ac7..7567b833a 100644
--- a/internal/release/materializeheld.go
+++ b/internal/release/materializeheld.go
@@ -39,27 +39,33 @@ var ErrHeldClaimWired = errors.New("release: held rows are not staged while CARL
// a variable so a test can make it fail the way a read-only mount does.
var materializeHeldStage = ensureStaged
-// MaterializeHeld stages version of rt's app and points live/<host> at it:
-// ensureStaged, then PointLive. A held row runs exactly the pinned release,
-// so there is no channel resolution, no bake-hold check and no restart.
-// When this process cannot write optDir it returns ErrStageBlocked wrapping
-// the cause and has written nothing to live/<host>.
-func MaterializeHeld(ctx context.Context, s Store, optDir string, rt registry.Route, version string) (*Manifest, error) {
- if hostwrite.Enabled(os.Getenv) {
- return nil, ErrHeldClaimWired
+// MaterializeHeld is StageHeld then PublishHeld: version of rt's app is
+// staged, and live/<host> is pointed at it inside guard. A held row runs
+// exactly the pinned release, so there is no channel resolution, no
+// bake-hold check and no restart. When this process cannot write optDir it
+// returns ErrStageBlocked wrapping the cause and has written nothing to
+// live/<host>.
+func MaterializeHeld(ctx context.Context, s Store, optDir string, rt registry.Route, version string, guard func(publish func() error) error) (*Manifest, error) {
+ m, err := StageHeld(ctx, s, optDir, rt, version)
+ if err != nil {
+ return nil, err
}
- if !controlname.ValidHostShape(rt.Host) || !ValidApp(rt.App) || !ValidVersion(version) {
- return nil, fmt.Errorf("release: held stage of %q/%q at %q: malformed", rt.Host, rt.App, version)
+ if err := PublishHeld(optDir, rt, version, guard); err != nil {
+ return nil, err
}
- acct := RouteAccount(rt)
- m, err := materializeHeldStage(ctx, s, optDir, acct, rt.App, version, rt.Host)
- if err != nil {
- if StageBlocked(err) {
- return nil, fmt.Errorf("%w: %w", ErrStageBlocked, err)
- }
+ return m, nil
+}
+
+// StageHeld is a held row's staging step, ensureStaged, with nothing
+// published: the version lands in the version store and live/<host> is
+// left alone. It refuses while CARLOS_HOST_CLAIM is on, and a refused write
+// is ErrStageBlocked wrapping the cause.
+func StageHeld(ctx context.Context, s Store, optDir string, rt registry.Route, version string) (*Manifest, error) {
+ if err := heldStageable(rt, version); err != nil {
return nil, err
}
- if err := PointLive(optDir, rt.Host, acct, version, rt.App); err != nil {
+ m, err := materializeHeldStage(ctx, s, optDir, RouteAccount(rt), rt.App, version, rt.Host)
+ if err != nil {
if StageBlocked(err) {
return nil, fmt.Errorf("%w: %w", ErrStageBlocked, err)
}
@@ -68,6 +74,42 @@ func MaterializeHeld(ctx context.Context, s Store, optDir string, rt registry.Ro
return m, nil
}
+// PublishHeld points live/<host> at a staged version, and does it inside
+// guard: guard re-checks whatever authorised the stage (for the root relay,
+// the registry hold frozen and the host claim re-read) and calls publish
+// only if that still holds, as its last step. Staging can take minutes, so
+// an authority checked only before it could be stale by now: a move aborted
+// or replaced meanwhile must never have live/<host> repointed for it. A nil
+// guard is refused: nothing is published on a request's word alone.
+func PublishHeld(optDir string, rt registry.Route, version string, guard func(publish func() error) error) error {
+ if err := heldStageable(rt, version); err != nil {
+ return err
+ }
+ if guard == nil {
+ return errors.New("release: a held publication needs its authority")
+ }
+ return guard(func() error {
+ if err := PointLive(optDir, rt.Host, RouteAccount(rt), version, rt.App); err != nil {
+ if StageBlocked(err) {
+ return fmt.Errorf("%w: %w", ErrStageBlocked, err)
+ }
+ return err
+ }
+ return nil
+ })
+}
+
+// heldStageable is the admission StageHeld and PublishHeld share.
+func heldStageable(rt registry.Route, version string) error {
+ if hostwrite.Enabled(os.Getenv) {
+ return ErrHeldClaimWired
+ }
+ if !controlname.ValidHostShape(rt.Host) || !ValidApp(rt.App) || !ValidVersion(version) {
+ return fmt.Errorf("release: held stage of %q/%q at %q: malformed", rt.Host, rt.App, version)
+ }
+ return nil
+}
+
// HeldStageRequest is what the sandboxed agent leaves at
// HeldStageDir(dataDir)/<Host>.json when MaterializeHeld returned
// ErrStageBlocked: stage Version of Account/App for Host under move Op,
@@ -104,16 +146,22 @@ const heldStageRequestMax = 4096
// `carlos adopt` after its release adoption. For each <name>.json in
// HeldStageDir it validates every field before any filesystem effect (Host
// equal to name and a valid host shape, valid account, app and version, a
-// non-empty op and box), asks authorize, then runs MaterializeHeld, writes
-// the HeldStaged ack and removes the request. A request that fails
-// validation or authorize is removed with a log line and never reaches a
-// path. A request whose materialization or ack fails is kept for the next
-// pass. With no request directory it reads nothing and creates nothing.
+// non-empty op and box), asks authorize, stages, asks authorize again with
+// the publication as its then, writes the HeldStaged ack and removes the
+// request. A request that fails validation, or that authorize refuses at
+// either point with ErrStaleHeldStage, is removed with a log line, and
+// live/<host> is never repointed for it. A request whose staging, ack or
+// publication fails any other way is kept for the next pass. With no
+// request directory it reads nothing and creates nothing.
+//
+// authorize(req, then) checks req and, if it passes, runs then (when
+// non-nil) under whatever guard keeps the check true until then returns.
+// Before staging it is called with a nil then.
//
// The directory is agent-owned and this runs as root, so every read, write
// and removal goes through instancefs: no symlink is followed at any
// component, and nothing but a regular file is read.
-func ServeHeldStageRequests(ctx context.Context, s Store, optDir, dataDir string, authorize func(HeldStageRequest) error) error {
+func ServeHeldStageRequests(ctx context.Context, s Store, optDir, dataDir string, authorize func(req HeldStageRequest, then func() error) error) error {
reqDir := HeldStageDir(dataDir)
ents, err := os.ReadDir(reqDir)
if errors.Is(err, os.ErrNotExist) {
@@ -136,7 +184,7 @@ func ServeHeldStageRequests(ctx context.Context, s Store, optDir, dataDir string
// serveHeldStageRequest serves one request file. It returns an error only
// when this build cannot do the work at all (instancefs unsupported).
-func serveHeldStageRequest(ctx context.Context, s Store, optDir, dataDir, reqDir, name string, authorize func(HeldStageRequest) error) error {
+func serveHeldStageRequest(ctx context.Context, s Store, optDir, dataDir, reqDir, name string, authorize func(HeldStageRequest, func() error) error) error {
drop := func(format string, args ...any) error {
fmt.Printf(" held stage request %q: "+format+"; removed\n", append([]any{name}, args...)...)
if err := instancefs.RemoveBeneath(reqDir, name); err != nil {
@@ -161,11 +209,15 @@ func serveHeldStageRequest(ctx context.Context, s Store, optDir, dataDir, reqDir
if why := heldStageRequestInvalid(req, strings.TrimSuffix(name, ".json")); why != "" {
return drop("refused: %s", why)
}
- if err := authorize(req); err != nil {
+ if err := authorize(req, nil); err != nil {
return drop("refused: %v", err)
}
rt := registry.Route{Host: req.Host, Account: req.Account, App: req.App}
- if _, err := MaterializeHeld(ctx, s, optDir, rt, req.Version); err != nil {
+ guard := func(publish func() error) error { return authorize(req, publish) }
+ if _, err := MaterializeHeld(ctx, s, optDir, rt, req.Version, guard); err != nil {
+ if errors.Is(err, ErrStaleHeldStage) {
+ return drop("refused at publication: %v", err)
+ }
fmt.Printf(" %s: held stage of %s for %s: %v; kept for the next pass\n", req.Host, req.Version, req.Op, err)
return nil
}
diff --git a/internal/release/materializeheld_test.go b/internal/release/materializeheld_test.go
index 3afbcc8ff..6510dc427 100644
--- a/internal/release/materializeheld_test.go
+++ b/internal/release/materializeheld_test.go
@@ -42,7 +42,7 @@ func heldRoute() registry.Route {
// pinned release, never the channel head, and LiveVersion reads it back.
func TestMaterializeHeldStagesThePinnedVersion(t *testing.T) {
ctx, s, opt := heldFixture(t)
- m, err := MaterializeHeld(ctx, s, opt, heldRoute(), "v1")
+ m, err := MaterializeHeld(ctx, s, opt, heldRoute(), "v1", openGuard)
if err != nil {
t.Fatalf("MaterializeHeld: %v", err)
}
@@ -70,7 +70,7 @@ func TestMaterializeHeldStageBlocked(t *testing.T) {
materializeHeldStage = func(context.Context, Store, string, string, string, string, string) (*Manifest, error) {
return nil, &os.PathError{Op: "mkdir", Path: filepath.Join(opt, "versions"), Err: errno}
}
- _, err := MaterializeHeld(ctx, s, opt, heldRoute(), "v1")
+ _, err := MaterializeHeld(ctx, s, opt, heldRoute(), "v1", openGuard)
if !errors.Is(err, ErrStageBlocked) || !errors.Is(err, errno) {
t.Fatalf("MaterializeHeld: %v, want ErrStageBlocked wrapping %v", err, errno)
}
@@ -91,7 +91,7 @@ func TestMaterializeHeldStageBlocked(t *testing.T) {
func TestMaterializeHeldRefusesWhileHostClaimIsWired(t *testing.T) {
ctx, s, opt := heldFixture(t)
t.Setenv("CARLOS_HOST_CLAIM", "1")
- if _, err := MaterializeHeld(ctx, s, opt, heldRoute(), "v1"); !errors.Is(err, ErrHeldClaimWired) {
+ if _, err := MaterializeHeld(ctx, s, opt, heldRoute(), "v1", openGuard); !errors.Is(err, ErrHeldClaimWired) {
t.Fatalf("MaterializeHeld: %v, want ErrHeldClaimWired", err)
}
if ents, _ := os.ReadDir(opt); len(ents) != 0 {
@@ -154,7 +154,7 @@ func TestServeHeldStageRequestsRefusesStaleAndTraversal(t *testing.T) {
rec := &heldRecordingStore{Store: s}
dd := t.TempDir()
called := false
- if err := ServeHeldStageRequests(ctx, rec, opt, dd, func(HeldStageRequest) error { called = true; return nil }); err != nil {
+ if err := ServeHeldStageRequests(ctx, rec, opt, dd, passAuth(func(HeldStageRequest) { called = true })); err != nil {
t.Fatalf("ServeHeldStageRequests: %v", err)
}
if called || len(rec.calls) != 0 {
@@ -174,11 +174,11 @@ func TestServeHeldStageRequestsRefusesStaleAndTraversal(t *testing.T) {
req := goodHeldRequest()
req.Op = "mv_bcdefghijklmnopqrstuvwxyza"
path := writeHeldRequest(t, dd, req.Host+".json", req)
- authorize := func(r HeldStageRequest) error {
+ authorize := func(r HeldStageRequest, then func() error) error {
if r.Op != heldTestOp {
return fmt.Errorf("%w: the claim names %s", ErrStaleHeldStage, heldTestOp)
}
- return nil
+ return runThen(then)
}
if err := ServeHeldStageRequests(ctx, s, opt, dd, authorize); err != nil {
t.Fatalf("ServeHeldStageRequests: %v", err)
@@ -209,7 +209,7 @@ func TestServeHeldStageRequestsRefusesStaleAndTraversal(t *testing.T) {
tc.mutate(&req)
path := writeHeldRequest(t, dd, tc.file, req)
called := false
- if err := ServeHeldStageRequests(ctx, rec, opt, dd, func(HeldStageRequest) error { called = true; return nil }); err != nil {
+ if err := ServeHeldStageRequests(ctx, rec, opt, dd, passAuth(func(HeldStageRequest) { called = true })); err != nil {
t.Fatalf("ServeHeldStageRequests: %v", err)
}
gone(t, path)
@@ -232,7 +232,7 @@ func TestServeHeldStageRequestsRefusesStaleAndTraversal(t *testing.T) {
if err := os.WriteFile(path, []byte("{not json"), 0o644); err != nil {
t.Fatal(err)
}
- if err := ServeHeldStageRequests(ctx, s, opt, dd, func(HeldStageRequest) error { return nil }); err != nil {
+ if err := ServeHeldStageRequests(ctx, s, opt, dd, passAuth(nil)); err != nil {
t.Fatal(err)
}
gone(t, path)
@@ -254,7 +254,7 @@ func TestServeHeldStageRequestsRefusesStaleAndTraversal(t *testing.T) {
t.Fatal(err)
}
called := false
- if err := ServeHeldStageRequests(ctx, s, opt, dd, func(HeldStageRequest) error { called = true; return nil }); err != nil {
+ if err := ServeHeldStageRequests(ctx, s, opt, dd, passAuth(func(HeldStageRequest) { called = true })); err != nil {
t.Fatal(err)
}
gone(t, link)
@@ -272,7 +272,7 @@ func TestServeHeldStageRequestsRefusesStaleAndTraversal(t *testing.T) {
req := goodHeldRequest()
req.Version = "v9" // never shipped
path := writeHeldRequest(t, dd, req.Host+".json", req)
- if err := ServeHeldStageRequests(ctx, s, opt, dd, func(HeldStageRequest) error { return nil }); err != nil {
+ if err := ServeHeldStageRequests(ctx, s, opt, dd, passAuth(nil)); err != nil {
t.Fatal(err)
}
if _, err := os.Stat(path); err != nil {
@@ -287,11 +287,11 @@ func TestServeHeldStageRequestsRefusesStaleAndTraversal(t *testing.T) {
req := goodHeldRequest()
path := writeHeldRequest(t, dd, req.Host+".json", req)
var seen []HeldStageRequest
- if err := ServeHeldStageRequests(ctx, s, opt, dd, func(r HeldStageRequest) error { seen = append(seen, r); return nil }); err != nil {
+ if err := ServeHeldStageRequests(ctx, s, opt, dd, passAuth(func(r HeldStageRequest) { seen = append(seen, r) })); err != nil {
t.Fatalf("ServeHeldStageRequests: %v", err)
}
- if len(seen) != 1 || seen[0] != req {
- t.Fatalf("authorize saw %+v, want exactly %+v", seen, req)
+ if len(seen) != 2 || seen[0] != req || seen[1] != req {
+ t.Fatalf("authorize saw %+v, want %+v before staging and again around publication", seen, req)
}
gone(t, path)
if got, err := LiveVersion(opt, heldTestHost); err != nil || got != "v1" {
@@ -307,3 +307,75 @@ func TestServeHeldStageRequestsRefusesStaleAndTraversal(t *testing.T) {
}
})
}
+
+// openGuard publishes unconditionally: the guard of a test whose authority
+// is not the point.
+func openGuard(publish func() error) error { return publish() }
+
+func runThen(then func() error) error {
+ if then == nil {
+ return nil
+ }
+ return then()
+}
+
+// passAuth is an authority that accepts everything, noting each call.
+func passAuth(note func(HeldStageRequest)) func(HeldStageRequest, func() error) error {
+ return func(r HeldStageRequest, then func() error) error {
+ if note != nil {
+ note(r)
+ }
+ return runThen(then)
+ }
+}
+
+// TestMaterializeHeldNeedsAGuard: publication without an authority would
+// repoint live/<host> on the word of a request alone, so a nil guard is
+// refused after staging and before anything is published.
+func TestMaterializeHeldNeedsAGuard(t *testing.T) {
+ ctx, s, opt := heldFixture(t)
+ if _, err := MaterializeHeld(ctx, s, opt, heldRoute(), "v1", nil); err == nil {
+ t.Fatal("MaterializeHeld published with no guard")
+ }
+ gone(t, LiveLink(opt, heldTestHost))
+}
+
+// TestHeldPublicationRechecksItsAuthority: the authority that admitted a
+// request before staging is asked again around publication, and a move
+// aborted or replaced while the release was staging leaves live/<host>
+// untouched, drops the request, and reports ErrStaleHeldStage.
+func TestHeldPublicationRechecksItsAuthority(t *testing.T) {
+ ctx, s, opt := heldFixture(t)
+ orig := materializeHeldStage
+ t.Cleanup(func() { materializeHeldStage = orig })
+ staged := false
+ materializeHeldStage = func(ctx context.Context, s Store, optDir, acct, app, ver, host string) (*Manifest, error) {
+ m, err := orig(ctx, s, optDir, acct, app, ver, host)
+ staged = true // the move is replaced while this runs
+ return m, err
+ }
+ published := false
+ authorize := func(_ HeldStageRequest, then func() error) error {
+ if staged {
+ return fmt.Errorf("%w: the claim names another op", ErrStaleHeldStage)
+ }
+ return runThen(then)
+ }
+ _, err := MaterializeHeld(ctx, s, opt, heldRoute(), "v1", func(publish func() error) error {
+ return authorize(goodHeldRequest(), func() error { published = true; return publish() })
+ })
+ if !errors.Is(err, ErrStaleHeldStage) || published {
+ t.Fatalf("MaterializeHeld: %v (published %v), want ErrStaleHeldStage and nothing published", err, published)
+ }
+ gone(t, LiveLink(opt, heldTestHost))
+
+ staged = false
+ dd := t.TempDir()
+ path := writeHeldRequest(t, dd, heldTestHost+".json", goodHeldRequest())
+ if err := ServeHeldStageRequests(ctx, s, opt, dd, authorize); err != nil {
+ t.Fatal(err)
+ }
+ gone(t, path)
+ gone(t, LiveLink(opt, heldTestHost))
+ gone(t, filepath.Join(HeldStagedDir(dd), heldTestHost+".json"))
+}