Final fix 1: T7 re-verifies the live link under the publication's lock
heldReleaseReady passed whenever live/<host> named the pinned version, so a read between a publication's swap and its rollback, or a link whose bytes or proof policy changed since, passed T7. T7 now passes only on release.VerifyHeldLive: the held-stage authority holds at check time, D3 admits the release under the policy in force (the publication's own heldProofAdmit), and under the host's held-stage lock the link names the version and bind accepts the bytes it names (the publication's own bindRecording, recording nothing). It runs before any stage and again after a publication. The lock is a per-host flock, <DataDir>/.moves/held-<host>.lock, opened through movesDir (O_NOFOLLOW, regular file, read-only so root and the agent share it). heldStageAuthority's guard runs the publication (link read, bind, swap, re-hash, rollback) under it, inside the registry freeze, so it is a leaf after apply flock > floor.lock > registry; its holders take no other lock and make no network call. floor.lock's Lock is now lockNamed over the same code. Regressions: a T7 reading started inside the swap window waits out the rollback and does not pass (release, with a seam after the swap; host, with the real flock); an existing link with replaced bytes, a rotated prover key or a broken policy fails T7.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
13 files changed,
+600
−32
approved_moves_copy_test.go+1 −0internal/host/heldstage.go+11 −3internal/host/heldstage_test.go+7 −7internal/host/movesfloor.go+3 −1internal/host/movesfloor_linux.go+14 −9internal/host/movesfloor_other.go+3 −0internal/host/movesheldlock.go+67 −0internal/host/movesheldlock_test.go+82 −0internal/host/movesreport_test.go+2 −2internal/host/movesstage.go+33 −9internal/host/movesstage_proof_test.go+107 −0internal/release/heldverify_test.go+203 −0internal/release/materializeheld.go+67 −1
CI passed — run details
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go| index 249626389..54633a34a 100644 |
| --- a/approved_moves_copy_test.go |
| +++ b/approved_moves_copy_test.go |
| @@ -549,6 +549,7 @@ var movesScanFiles = []string{ |
| "internal/host/movesfloor_other.go", |
| "internal/host/movesguard.go", |
| "internal/host/movesheldenv.go", |
| + "internal/host/movesheldlock.go", |
| "internal/host/movesmeasure.go", |
| "internal/host/movesmeasure_linux.go", |
| "internal/host/movesmeasure_other.go", |
diff --git a/internal/host/heldstage.go b/internal/host/heldstage.go| index 2db105344..c7173cd8d 100644 |
| --- a/internal/host/heldstage.go |
| +++ b/internal/host/heldstage.go |
| @@ -55,7 +55,11 @@ const heldStageFrozenReadTimeout = 2 * time.Second |
| // 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 { |
| +// |
| +// Inside the freeze, then also holds the host's held-stage lock under |
| +// dataDir (withHeldStageLock), so T7's reading of the live link waits out a |
| +// publication and its rollback. |
| +func heldStageAuthority(reg *registry.Registry, store release.Store, box, dataDir 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 { |
| @@ -86,7 +90,11 @@ func heldStageAuthority(reg *registry.Registry, store release.Store, box string) |
| if err := heldStageClaim(obs, req); err != nil { |
| return err |
| } |
| - return then() |
| + // The publication (link read, bind, swap, re-hash, rollback) |
| + // holds the host's held-stage lock, which T7's reading takes |
| + // too (final review finding 1). Taken last, inside the freeze: |
| + // it is a leaf of the lock order. |
| + return withHeldStageLock(dataDir, req.Host, then) |
| }) |
| } |
| } |
| @@ -188,7 +196,7 @@ func heldStageClaim(obs hostclaim.Observed, req release.HeldStageRequest) error |
| // bake directory is the healthy history that pass reads. |
| func heldStagePass(ctx context.Context, dataDir, optDir string, reg *registry.Registry, store release.Store, box string) { |
| proof := release.NewProofGate(release.ProofPolicyPath, dataDir) |
| - if err := release.ServeHeldStageRequests(ctx, store, optDir, dataDir, proof, heldStageAuthority(reg, store, box)); err != nil { |
| + if err := release.ServeHeldStageRequests(ctx, store, optDir, dataDir, proof, heldStageAuthority(reg, store, box, dataDir)); err != nil { |
| log.Printf("adopt: held stage requests: %v", err) |
| } |
| retireEndedHeldRefusals(ctx, store, dataDir, box) |
diff --git a/internal/host/heldstage_test.go b/internal/host/heldstage_test.go| index e8a6b5eff..88aa68140 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, nil); !errors.Is(err, release.ErrStaleHeldStage) { |
| + if err := heldStageAuthority(f.reg, cs, hsBox, t.TempDir())(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(), nil); err != nil { |
| + if err := heldStageAuthority(f.reg, f.s, hsBox, t.TempDir())(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(), nil); err != nil { |
| + if err := heldStageAuthority(reg, s, hsBox, t.TempDir())(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, nil); !errors.Is(err, release.ErrStaleHeldStage) { |
| + if err := heldStageAuthority(f.reg, f.s, tc.box, t.TempDir())(req, nil); !errors.Is(err, release.ErrStaleHeldStage) { |
| t.Fatalf("authorize: %v, want ErrStaleHeldStage", err) |
| } |
| }) |
| @@ -350,7 +350,7 @@ func TestHeldPublicationIsGuardedByTheAuthority(t *testing.T) { |
| } |
| } |
| opt := t.TempDir() |
| - auth := heldStageAuthority(reg, s, hsBox) |
| + auth := heldStageAuthority(reg, s, hsBox, t.TempDir()) |
| req := hsRequest() |
| if err := auth(req, nil); err != nil { |
| t.Fatalf("authority before staging: %v", err) |
| @@ -418,7 +418,7 @@ func TestHeldPublicationRechecksInsideTheFreeze(t *testing.T) { |
| 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 }) |
| + err := heldStageAuthority(reg, s, hsBox, t.TempDir())(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) |
| } |
| @@ -434,7 +434,7 @@ func TestHeldPublicationRechecksInsideTheFreeze(t *testing.T) { |
| t.Fatal(err) |
| } |
| released := make(chan struct{}) |
| - err := heldStageAuthority(reg, s, hsBox)(hsRequest(), func() error { |
| + err := heldStageAuthority(reg, s, hsBox, t.TempDir())(hsRequest(), func() error { |
| go func() { _, _ = reg.ReleaseHold(hsHost, hsOp); close(released) }() |
| select { |
| case <-released: |
diff --git a/internal/host/movesfloor.go b/internal/host/movesfloor.go| index 1752139b6..8203cabbe 100644 |
| --- a/internal/host/movesfloor.go |
| +++ b/internal/host/movesfloor.go |
| @@ -125,7 +125,9 @@ var movesDirName = filepath.Dir(moves.FloorFile) |
| // Lock order, everywhere both are taken: the apply flock |
| // (system/router/lock, held by systemApplier.Run for the whole run), then |
| // the carlos.prev lock (root only), then this one, then the registry's |
| -// write lock (WithHoldsFrozen). The agent takes only this one and holds |
| +// write lock (WithHoldsFrozen), then a host's held-stage lock |
| +// (withHeldStageLock), which is a leaf: its holder takes no other lock. The |
| +// agent takes only this one and holds |
| // nothing else while it does; the adopt pass (rootJudgePrev) takes the |
| // carlos.prev lock and this one and never the apply flock. Nothing that |
| // holds this lock or the registry's may then take the apply flock: the |
diff --git a/internal/host/movesfloor_linux.go b/internal/host/movesfloor_linux.go| index a8916e823..8561740f0 100644 |
| --- a/internal/host/movesfloor_linux.go |
| +++ b/internal/host/movesfloor_linux.go |
| @@ -86,21 +86,26 @@ func (d *movesDir) Close() { |
| syscall.Close(d.fd) |
| } |
| -// Lock takes floor.lock, waiting up to floorLockWait. Opened read-only (a |
| +// Lock takes floor.lock, waiting up to floorLockWait. |
| +func (d *movesDir) Lock() error { return d.lockNamed(floorLockFile, floorLockWait, errFloorLockBusy) } |
| + |
| +// lockNamed takes an exclusive flock on the file name in the directory, |
| +// waiting up to wait and then failing with busy. Opened read-only (a |
| // read-only fd can be flocked), so the agent can lock a file root created |
| // and root a file the agent created; O_NOFOLLOW and O_NONBLOCK so a symlink |
| -// or a FIFO planted in its place cannot redirect or park root. |
| -func (d *movesDir) Lock() error { |
| - fd, err := syscall.Openat(d.fd, floorLockFile, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NOFOLLOW|syscall.O_NONBLOCK|syscall.O_CLOEXEC, 0o644) |
| +// or a FIFO planted in its place cannot redirect or park root. The fd is |
| +// kept until Close, which releases the lock. One lock per movesDir. |
| +func (d *movesDir) lockNamed(name string, wait time.Duration, busy error) error { |
| + fd, err := syscall.Openat(d.fd, name, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NOFOLLOW|syscall.O_NONBLOCK|syscall.O_CLOEXEC, 0o644) |
| if err != nil { |
| - return fmt.Errorf("open %s: %w", filepath.Join(d.path, floorLockFile), err) |
| + return fmt.Errorf("open %s: %w", filepath.Join(d.path, name), err) |
| } |
| var st syscall.Stat_t |
| if err := syscall.Fstat(fd, &st); err != nil || st.Mode&syscall.S_IFMT != syscall.S_IFREG { |
| syscall.Close(fd) |
| - return fmt.Errorf("%s is not a regular file", filepath.Join(d.path, floorLockFile)) |
| + return fmt.Errorf("%s is not a regular file", filepath.Join(d.path, name)) |
| } |
| - deadline := time.Now().Add(floorLockWait) |
| + deadline := time.Now().Add(wait) |
| for { |
| err := syscall.Flock(fd, syscall.LOCK_EX|syscall.LOCK_NB) |
| if err == nil { |
| @@ -109,11 +114,11 @@ func (d *movesDir) Lock() error { |
| } |
| if !errors.Is(err, syscall.EWOULDBLOCK) { |
| syscall.Close(fd) |
| - return fmt.Errorf("lock %s: %w", filepath.Join(d.path, floorLockFile), err) |
| + return fmt.Errorf("lock %s: %w", filepath.Join(d.path, name), err) |
| } |
| if time.Now().After(deadline) { |
| syscall.Close(fd) |
| - return fmt.Errorf("lock %s: %w", filepath.Join(d.path, floorLockFile), errFloorLockBusy) |
| + return fmt.Errorf("lock %s: %w", filepath.Join(d.path, name), busy) |
| } |
| time.Sleep(floorLockPoll) |
| } |
diff --git a/internal/host/movesfloor_other.go b/internal/host/movesfloor_other.go| index 1937948ec..c21c86f92 100644 |
| --- a/internal/host/movesfloor_other.go |
| +++ b/internal/host/movesfloor_other.go |
| @@ -5,6 +5,7 @@ package host |
| import ( |
| "errors" |
| "os" |
| + "time" |
| ) |
| // The floor files are written by root's apply and the agent, both of which |
| @@ -19,6 +20,8 @@ func (d *movesDir) Close() {} |
| func (d *movesDir) Lock() error { return errSystemUnsupported } |
| +func (d *movesDir) lockNamed(string, time.Duration, error) error { return errSystemUnsupported } |
| + |
| func (d *movesDir) WriteDurable(name string, data []byte, mode os.FileMode) error { |
| return errSystemUnsupported |
| } |
diff --git a/internal/host/movesheldlock.go b/internal/host/movesheldlock.go| new file mode 100644 |
| index 000000000..4e12216dd |
| --- /dev/null |
| +++ b/internal/host/movesheldlock.go |
| @@ -0,0 +1,67 @@ |
| +package host |
| + |
| +import ( |
| + "errors" |
| + "fmt" |
| + "os" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/controlname" |
| +) |
| + |
| +// The held-stage lock (final review finding 1): one flock per host, |
| +// <DataDir>/.moves/held-<host>.lock, taken through movesDir (O_NOFOLLOW |
| +// below a data dir root trusts, a regular file only), so root and the |
| +// sandboxed agent lock the same file whichever created it. |
| +// |
| +// A held publication (release.publishHeld, run through heldStageAuthority's |
| +// guard, by root's relay or by the agent on a writable box) holds it across |
| +// reading the live link, binding, the swap, the re-hash and any rollback. |
| +// T7's reading (release.VerifyHeldLive, from heldReleaseReady) takes it to |
| +// read the link and hash the bytes it names. So T7 never sees a link that a |
| +// failing publication is about to take back, and never hashes bytes the |
| +// publication has not finished vouching for. |
| +// |
| +// Lock order: the apply flock > floor.lock > the registry's write lock > |
| +// this one. The publication takes it inside WithHoldsFrozen; T7 takes it |
| +// holding nothing. Its holder takes no other lock and makes no network |
| +// call (D3's admit, which may read the bucket, runs before it on both |
| +// sides), so it never inverts the order and never holds the registry's |
| +// freeze for long. |
| + |
| +// heldLockWait bounds the wait for a held-stage lock. The publication waits |
| +// for it inside the registry's freeze, after a claim read bounded by |
| +// heldStageFrozenReadTimeout, so the two stay well under the registry's |
| +// 5 s busy_timeout. Its holders keep it for a hash or two of one binary. |
| +// A variable so a test can make contention time out quickly. |
| +var heldLockWait = time.Second |
| + |
| +// errHeldLockBusy is the held-stage lock still held after heldLockWait: |
| +// a publication, or a reading, is under way. Never a reason to go ahead |
| +// without it. |
| +var errHeldLockBusy = errors.New("another process holds this host's held-stage lock") |
| + |
| +// heldStageLockFile is host's lock file in .moves. |
| +func heldStageLockFile(host string) string { return "held-" + host + ".lock" } |
| + |
| +// withHeldStageLock runs fn holding host's held-stage lock, making .moves |
| +// when it is missing. An error opening or locking is returned without |
| +// running fn. Off Linux it always refuses (openMovesDir does), which fails |
| +// T7 and the publication closed; production boxes are Linux. |
| +func withHeldStageLock(dataDir, host string, fn func() error) error { |
| + if !controlname.ValidHostShape(host) { |
| + return fmt.Errorf("held-stage lock: malformed host %q", host) |
| + } |
| + d, err := openMovesDir(dataDir, false) |
| + if errors.Is(err, os.ErrNotExist) { |
| + d, err = openMovesDir(dataDir, true) |
| + } |
| + if err != nil { |
| + return err |
| + } |
| + defer d.Close() |
| + if err := d.lockNamed(heldStageLockFile(host), heldLockWait, errHeldLockBusy); err != nil { |
| + return err |
| + } |
| + return fn() |
| +} |
diff --git a/internal/host/movesheldlock_test.go b/internal/host/movesheldlock_test.go| new file mode 100644 |
| index 000000000..72c0405f7 |
| --- /dev/null |
| +++ b/internal/host/movesheldlock_test.go |
| @@ -0,0 +1,82 @@ |
| +//go:build linux |
| + |
| +package host |
| + |
| +import ( |
| + "errors" |
| + "os" |
| + "path/filepath" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// Final review finding 1: the held-stage lock is a per-host flock that a |
| +// second holder (another open, as another process makes) waits for and |
| +// then gives up on; other hosts do not contend; a planted symlink is |
| +// refused. |
| +func TestHeldStageLockSerialisesItsHolders(t *testing.T) { |
| + prev := heldLockWait |
| + t.Cleanup(func() { heldLockWait = prev }) |
| + heldLockWait = 50 * time.Millisecond |
| + data := t.TempDir() |
| + in, out, done := make(chan struct{}), make(chan struct{}), make(chan error, 1) |
| + go func() { |
| + done <- withHeldStageLock(data, hsHost, func() error { close(in); <-out; return nil }) |
| + }() |
| + <-in |
| + ran := false |
| + if err := withHeldStageLock(data, hsHost, func() error { ran = true; return nil }); !errors.Is(err, errHeldLockBusy) || ran { |
| + t.Fatalf("a second holder: %v (ran %v), want errHeldLockBusy", err, ran) |
| + } |
| + if err := withHeldStageLock(data, "other.example.com", func() error { ran = true; return nil }); err != nil || !ran { |
| + t.Fatalf("another host's lock: %v (ran %v)", err, ran) |
| + } |
| + close(out) |
| + if err := <-done; err != nil { |
| + t.Fatal(err) |
| + } |
| + ran = false |
| + if err := withHeldStageLock(data, hsHost, func() error { ran = true; return nil }); err != nil || !ran { |
| + t.Fatalf("after release: %v (ran %v)", err, ran) |
| + } |
| + if err := withHeldStageLock(data, "../x", func() error { return nil }); err == nil { |
| + t.Fatal("a malformed host was locked") |
| + } |
| + if err := os.Symlink(filepath.Join(t.TempDir(), "elsewhere"), filepath.Join(data, ".moves", heldStageLockFile("planted.example.com"))); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := withHeldStageLock(data, "planted.example.com", func() error { return nil }); err == nil { |
| + t.Fatal("a planted symlink was followed") |
| + } |
| +} |
| + |
| +// The publication runs under the lock: heldStageAuthority's then (the |
| +// publish closure: link read, bind, swap, re-hash, rollback) holds host's |
| +// held-stage lock, inside the registry's freeze. The pre-stage check |
| +// (then nil) takes no lock. |
| +func TestHeldPublicationHoldsTheHeldStageLock(t *testing.T) { |
| + prev := heldLockWait |
| + t.Cleanup(func() { heldLockWait = prev }) |
| + heldLockWait = 50 * time.Millisecond |
| + s := release.NewDirStore(t.TempDir()) |
| + hsClaim(t, s, movingOp()) |
| + hsReport(t, s, hsOp, "v7") |
| + data := t.TempDir() |
| + auth := heldStageAuthority(testRegistry(t), s, hsBox, data) |
| + if err := auth(hsRequest(), nil); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := os.Stat(filepath.Join(data, ".moves", heldStageLockFile(hsHost))); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("the pre-stage check took the lock: %v", err) |
| + } |
| + published := false |
| + err := auth(hsRequest(), func() error { |
| + published = true |
| + return withHeldStageLock(data, hsHost, func() error { return errors.New("the lock was free during the publication") }) |
| + }) |
| + if !published || !errors.Is(err, errHeldLockBusy) { |
| + t.Fatalf("publication: %v (ran %v), want the lock held around it", err, published) |
| + } |
| +} |
diff --git a/internal/host/movesreport_test.go b/internal/host/movesreport_test.go| index cd710b240..49f0109c5 100644 |
| --- a/internal/host/movesreport_test.go |
| +++ b/internal/host/movesreport_test.go |
| @@ -293,7 +293,7 @@ func TestPreparingRelaysHeldStageOnAHardenedBox(t *testing.T) { |
| // The root pass, with the relay's own authority, materializes it. |
| heldMaterialize = prev |
| - if err := release.ServeHeldStageRequests(ctx, h.s, targetOpt, targetData, offProofGate(t), heldStageAuthority(reg, h.s, "uk-box")); err != nil { |
| + if err := release.ServeHeldStageRequests(ctx, h.s, targetOpt, targetData, offProofGate(t), heldStageAuthority(reg, h.s, "uk-box", targetData)); err != nil { |
| t.Fatal(err) |
| } |
| if v, err := release.LiveVersion(targetOpt, p1Host); err != nil || v != "v7" { |
| @@ -313,7 +313,7 @@ func TestPreparingRelaysHeldStageOnAHardenedBox(t *testing.T) { |
| if err := writeHeldStageRequest(targetData, bad); err != nil { |
| t.Fatal(err) |
| } |
| - if err := release.ServeHeldStageRequests(ctx, h.s, opt, targetData, offProofGate(t), heldStageAuthority(reg, h.s, "uk-box")); err != nil { |
| + if err := release.ServeHeldStageRequests(ctx, h.s, opt, targetData, offProofGate(t), heldStageAuthority(reg, h.s, "uk-box", targetData)); err != nil { |
| t.Fatal(err) |
| } |
| if v, _ := release.LiveVersion(opt, p1Host); v != "" { |
diff --git a/internal/host/movesstage.go b/internal/host/movesstage.go| index c666f13b4..cd053b20d 100644 |
| --- a/internal/host/movesstage.go |
| +++ b/internal/host/movesstage.go |
| @@ -26,7 +26,16 @@ var heldMaterialize = release.MaterializeHeld |
| // the row is absent or held by the op, the source reports this release). |
| // On a hardened box the stage is blocked, so a held-stage request is left |
| // for the root stage pass (release.ServeHeldStageRequests) and its marker |
| -// touched; T7 passes once LiveVersion equals the pinned version. |
| +// touched. |
| +// |
| +// T7 passes only on release.VerifyHeldLive (final review finding 1): at |
| +// check time the authority holds, D3 admits the release under the policy in |
| +// force, and, under the host's held-stage lock (withHeldStageLock, the lock |
| +// every publication holds across its swap, re-hash and rollback), the link |
| +// names the version and D3's bind accepts the bytes it names. A link alone |
| +// is never evidence: the publication may be about to roll it back, and the |
| +// bytes or the policy may have changed since it was written. An existing |
| +// link that fails is t7NotProven and is left alone. |
| // |
| // proof is D3's gate (controller ruling D3-held): the held stage admits the |
| // pinned release as adoption would, before staging and so before any |
| @@ -37,16 +46,32 @@ var heldMaterialize = release.MaterializeHeld |
| // never visits. |
| func heldReleaseReady(reg *registry.Registry, optDir, dataDir, box string, proof *release.ProofGate, requestStage func(host, ver string, err error)) func(context.Context, release.Store, provision.Record, string) moves.Check { |
| return func(ctx context.Context, s release.Store, rec provision.Record, version string) moves.Check { |
| - if live, err := release.LiveVersion(optDir, rec.Host); err == nil && live == version { |
| - return moves.Check{OK: true} |
| - } |
| req := release.HeldStageRequest{Account: rec.Account, App: rec.App, Host: rec.Host, Version: version, Op: rec.Move.ID, Box: box} |
| - auth := heldStageAuthority(reg, s, box) |
| + auth := heldStageAuthority(reg, s, box, dataDir) |
| if err := auth(req, nil); err != nil { |
| logMoveStage(rec.Host, version, err) |
| return moves.Check{Reason: t7Failed} |
| } |
| rt := registry.Route{Host: rec.Host, Kind: "instance", App: rec.App, Account: rec.Account, HoldOp: rec.Move.ID} |
| + lock := func(read func() error) error { return withHeldStageLock(dataDir, rec.Host, read) } |
| + verified := func() (moves.Check, bool) { |
| + err := release.VerifyHeldLive(ctx, s, optDir, rt, version, proof, lock) |
| + switch { |
| + case err == nil: |
| + return moves.Check{OK: true}, true |
| + case errors.Is(err, release.ErrHeldNotProven): |
| + logMoveStage(rec.Host, version, err) |
| + return moves.Check{Reason: t7NotProven}, true |
| + case errors.Is(err, release.ErrHeldNotLive): |
| + return moves.Check{Reason: t7Failed}, false |
| + default: |
| + logMoveStage(rec.Host, version, err) |
| + return moves.Check{Reason: t7Failed}, true |
| + } |
| + } |
| + if c, done := verified(); done { |
| + return c |
| + } |
| _, err := heldMaterialize(ctx, s, optDir, rt, version, proof, func(publish func() error) error { return auth(req, publish) }) |
| switch { |
| case errors.Is(err, release.ErrHeldNotProven): |
| @@ -65,10 +90,9 @@ func heldReleaseReady(reg *registry.Registry, optDir, dataDir, box string, proof |
| logMoveStage(rec.Host, version, err) |
| return moves.Check{Reason: t7Failed} |
| } |
| - if live, err := release.LiveVersion(optDir, rec.Host); err != nil || live != version { |
| - return moves.Check{Reason: t7Failed} |
| - } |
| - return moves.Check{OK: true} |
| + // Published by this call: T7 passes on the same reading as ever. |
| + c, _ := verified() |
| + return c |
| } |
| } |
diff --git a/internal/host/movesstage_proof_test.go b/internal/host/movesstage_proof_test.go| index df364096e..ec532547b 100644 |
| --- a/internal/host/movesstage_proof_test.go |
| +++ b/internal/host/movesstage_proof_test.go |
| @@ -17,6 +17,7 @@ import ( |
| "github.com/carlosframework/platform/internal/hostclaim" |
| "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| ) |
| @@ -248,3 +249,109 @@ type failingClaimStore struct{ release.Store } |
| func (failingClaimStore) Get(context.Context, string) ([]byte, string, error) { |
| return nil, "", errors.New("bucket unreachable") |
| } |
| + |
| +// Final review finding 1, through heldReleaseReady: an existing link is |
| +// re-checked every time. Bytes replaced after the publication, or a proof |
| +// policy that no longer accepts them, fail T7 with the not-proven sentence |
| +// and nothing is restaged or republished. |
| +func TestT7RechecksAnExistingLink(t *testing.T) { |
| + ctx := context.Background() |
| + h := newPhase1Harness(t) |
| + rec := h.start(t, "manual") |
| + shipPinnedV7(t, h.s) |
| + h.pass(t) |
| + box := newHeldProofBox(t) |
| + box.require(t, "app") |
| + box.prove(t, h.s, "app", []byte("pinned")) |
| + reg, data, opt := testRegistry(t), t.TempDir(), t.TempDir() |
| + t7 := func() moves.Check { |
| + return heldReleaseReady(reg, opt, data, "uk-box", box.gate(), nil)(ctx, h.s, rec, "v7") |
| + } |
| + if c := t7(); !c.OK { |
| + t.Fatalf("T7 = %+v", c) |
| + } |
| + if c := t7(); !c.OK { |
| + t.Fatalf("T7 on the published link = %+v", c) |
| + } |
| + path := release.VersionPath(opt, "bab", "v7", "app") |
| + if err := os.WriteFile(path, []byte("replaced after publication"), 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + if c := t7(); c.OK || c.Reason != t7NotProven { |
| + t.Fatalf("T7 with replaced bytes = %+v, want %q", c, t7NotProven) |
| + } |
| + if got, _ := os.ReadFile(path); string(got) != "replaced after publication" { |
| + t.Fatalf("T7 restaged the release: %q", got) |
| + } |
| + if err := os.WriteFile(path, []byte("pinned"), 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + if c := t7(); !c.OK { |
| + t.Fatalf("T7 with the bytes restored = %+v", c) |
| + } |
| + // The policy rotates the prover's key: the proof no longer verifies. |
| + other := newHeldProofBox(t) |
| + other.policy = box.policy |
| + other.require(t, "app") |
| + if c := t7(); c.OK || c.Reason != t7NotProven { |
| + t.Fatalf("T7 under a rotated key = %+v, want %q", c, t7NotProven) |
| + } |
| +} |
| + |
| +// Final review finding 1, the review's interleaving with the real lock: a |
| +// publication holds host's held-stage lock while live/<host> names v7, and |
| +// then rolls the link back. T7, read inside that window, waits for the |
| +// lock, finds the link back on v0 and does not pass (here the stage |
| +// relays, so it is waiting for the root stage pass). |
| +func TestT7WaitsOutAPublicationsRollback(t *testing.T) { |
| + ctx := context.Background() |
| + h := newPhase1Harness(t) |
| + rec := h.start(t, "manual") |
| + shipPinnedV7(t, h.s) |
| + h.pass(t) |
| + box := newHeldProofBox(t) |
| + box.require(t, "app") |
| + box.prove(t, h.s, "app", []byte("pinned")) |
| + reg, data, opt := testRegistry(t), t.TempDir(), t.TempDir() |
| + for v, body := range map[string]string{"v0": "earlier", "v7": "pinned"} { |
| + if err := release.StageArtifact(opt, "bab", v, "app", []byte(body)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := release.PointLive(opt, p1Host, "bab", "v0", "app"); err != nil { |
| + t.Fatal(err) |
| + } |
| + prev := heldMaterialize |
| + t.Cleanup(func() { heldMaterialize = prev }) |
| + heldMaterialize = func(context.Context, release.Store, string, registry.Route, string, *release.ProofGate, func(func() error) error) (*release.Manifest, error) { |
| + return nil, fmt.Errorf("%w: read-only", release.ErrStageBlocked) |
| + } |
| + swapped, rolledBack := make(chan struct{}), make(chan time.Time, 1) |
| + go func() { |
| + err := withHeldStageLock(data, p1Host, func() error { |
| + if err := release.PointLive(opt, p1Host, "bab", "v7", "app"); err != nil { |
| + return err |
| + } |
| + close(swapped) |
| + time.Sleep(300 * time.Millisecond) |
| + err := release.PointLive(opt, p1Host, "bab", "v0", "app") |
| + rolledBack <- time.Now() |
| + return err |
| + }) |
| + if err != nil { |
| + t.Error(err) |
| + } |
| + }() |
| + <-swapped |
| + c := heldReleaseReady(reg, opt, data, "uk-box", box.gate(), stageRequester(data))(ctx, h.s, rec, "v7") |
| + answered := time.Now() |
| + if c.OK { |
| + t.Fatalf("T7 passed on a link the publication rolled back: %+v", c) |
| + } |
| + if at := <-rolledBack; answered.Before(at) { |
| + t.Fatalf("T7 answered %v before the rollback at %v: it did not wait for the lock", answered, at) |
| + } |
| + if c.Reason != t7Waiting { |
| + t.Fatalf("T7 = %+v, want %q", c, t7Waiting) |
| + } |
| +} |
diff --git a/internal/release/heldverify_test.go b/internal/release/heldverify_test.go| new file mode 100644 |
| index 000000000..5d076faf9 |
| --- /dev/null |
| +++ b/internal/release/heldverify_test.go |
| @@ -0,0 +1,203 @@ |
| +package release |
| + |
| +import ( |
| + "errors" |
| + "fmt" |
| + "os" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// Final review finding 1. T7 used to pass whenever live/<host> named the |
| +// pinned version: no authority, no proof policy, no look at the bytes. |
| +// VerifyHeldLive is its reading now: the link names the version, D3 admits |
| +// the release under the policy in force, and bind accepts the bytes the |
| +// link names, read under the lock the publication holds across its swap, |
| +// re-hash and rollback. |
| + |
| +// passLock is a lock that is not one: the bytes check alone. |
| +func passLock(read func() error) error { return read() } |
| + |
| +// chanLock is one lock shared by a publication's guard and a T7 reading, |
| +// as the per-host flock is in production. waiting is signalled each time a |
| +// reading starts to wait for it. |
| +type chanLock struct { |
| + mu chan struct{} |
| + waiting chan struct{} |
| +} |
| + |
| +func newChanLock() *chanLock { |
| + return &chanLock{mu: make(chan struct{}, 1), waiting: make(chan struct{}, 8)} |
| +} |
| + |
| +func (l *chanLock) guard(publish func() error) error { |
| + l.mu <- struct{}{} |
| + defer func() { <-l.mu }() |
| + return publish() |
| +} |
| + |
| +func (l *chanLock) read(fn func() error) error { |
| + l.waiting <- struct{}{} |
| + l.mu <- struct{}{} |
| + defer func() { <-l.mu }() |
| + return fn() |
| +} |
| + |
| +// The review's regression: T7 is read between the swap and a failed |
| +// post-swap verification. The publication holds the lock from before the |
| +// swap until after its rollback, so the reading waits it out and then sees |
| +// the link back on v0: not published, never a pass. |
| +func TestVerifyHeldLiveWaitsOutAFailedPublication(t *testing.T) { |
| + e := newHeldProofEnv(t) |
| + e.require(t, "console") |
| + e.prove(t, e.pinnedSHA) |
| + lock := newChanLock() |
| + agent := NewProofGate(e.policy, e.bake) // the agent's own gate |
| + prev := heldAfterSwap |
| + t.Cleanup(func() { heldAfterSwap = prev }) |
| + res := make(chan error, 1) |
| + var during error = errors.New("the seam never ran") |
| + heldAfterSwap = func() { |
| + // The link names v1 now. Its bytes are replaced, so the |
| + // publication's second bind will refuse and roll back. |
| + if err := StageArtifact(e.opt, "bab", "v1", "console", []byte("replaced after the swap")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if v := e.live(); v != "v1" { |
| + t.Fatalf("live = %q inside the swap window, want v1", v) |
| + } |
| + go func() { res <- VerifyHeldLive(e.ctx, e.store, e.opt, heldRoute(), "v1", agent, lock.read) }() |
| + select { |
| + case <-lock.waiting: |
| + case err := <-res: |
| + during = fmt.Errorf("T7 answered %v without waiting for the lock", err) |
| + return |
| + case <-time.After(5 * time.Second): |
| + during = errors.New("T7 never reached the lock") |
| + return |
| + } |
| + select { |
| + case err := <-res: |
| + during = fmt.Errorf("T7 answered %v while the publication held the lock", err) |
| + case <-time.After(200 * time.Millisecond): |
| + during = nil |
| + } |
| + } |
| + _, err := MaterializeHeld(e.ctx, e.store, e.opt, heldRoute(), "v1", e.gate, lock.guard) |
| + if !errors.Is(err, ErrHeldNotProven) || e.live() != "v0" { |
| + t.Fatalf("the publication: %v, live %q; want ErrHeldNotProven and v0", err, e.live()) |
| + } |
| + if during != nil { |
| + t.Fatal(during) |
| + } |
| + select { |
| + case got := <-res: |
| + if !errors.Is(got, ErrHeldNotLive) { |
| + t.Fatalf("T7 after the rollback = %v, want ErrHeldNotLive", got) |
| + } |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("T7 never answered") |
| + } |
| +} |
| + |
| +// Even with no lock, the reading never passes on the swapped link: it |
| +// hashes the bytes the link names, as the publication's second bind does. |
| +func TestVerifyHeldLiveRefusesTheSwappedBytes(t *testing.T) { |
| + e := newHeldProofEnv(t) |
| + e.require(t, "console") |
| + e.prove(t, e.pinnedSHA) |
| + agent := NewProofGate(e.policy, e.bake) |
| + prev := heldAfterSwap |
| + t.Cleanup(func() { heldAfterSwap = prev }) |
| + var during error = errors.New("the seam never ran") |
| + heldAfterSwap = func() { |
| + if err := StageArtifact(e.opt, "bab", "v1", "console", []byte("replaced after the swap")); err != nil { |
| + t.Fatal(err) |
| + } |
| + during = VerifyHeldLive(e.ctx, e.store, e.opt, heldRoute(), "v1", agent, passLock) |
| + } |
| + if _, err := MaterializeHeld(e.ctx, e.store, e.opt, heldRoute(), "v1", e.gate, openGuard); !errors.Is(err, ErrHeldNotProven) { |
| + t.Fatalf("the publication: %v", err) |
| + } |
| + if !errors.Is(during, ErrHeldNotProven) { |
| + t.Fatalf("T7 inside the swap window = %v, want ErrHeldNotProven", during) |
| + } |
| +} |
| + |
| +// An existing link is no evidence by itself: T7 re-checks the policy and |
| +// the bytes every time. Replaced bytes, a rotated prover key and a policy |
| +// that does not load each fail it; a link on another version is not live; |
| +// an app the policy does not list passes as adoption would. |
| +func TestVerifyHeldLiveRechecksAnExistingLink(t *testing.T) { |
| + e := newHeldProofEnv(t) |
| + e.require(t, "console") |
| + e.prove(t, e.pinnedSHA) |
| + if _, err := MaterializeHeld(e.ctx, e.store, e.opt, heldRoute(), "v1", e.gate, openGuard); err != nil || e.live() != "v1" { |
| + t.Fatalf("publish v1: %v, live %q", err, e.live()) |
| + } |
| + verify := func() error { |
| + return VerifyHeldLive(e.ctx, e.store, e.opt, heldRoute(), "v1", NewProofGate(e.policy, e.bake), passLock) |
| + } |
| + if err := verify(); err != nil { |
| + t.Fatalf("published and proven: %v", err) |
| + } |
| + |
| + // The artifact is replaced after publication. |
| + if err := StageArtifact(e.opt, "bab", "v1", "console", []byte("replaced later")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := verify(); !errors.Is(err, ErrHeldNotProven) { |
| + t.Fatalf("replaced bytes: %v, want ErrHeldNotProven", err) |
| + } |
| + if r, _ := ReadRefusal(HeldRefusalBase(e.bake), heldTestHost); r != nil { |
| + t.Fatalf("a reading recorded a refusal: %+v", r) |
| + } |
| + if err := StageArtifact(e.opt, "bab", "v1", "console", []byte("pinned build")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := verify(); err != nil { |
| + t.Fatalf("bytes restored: %v", err) |
| + } |
| + |
| + // The proof policy changes: another prover key. |
| + _, _, otherPub := proverKey(t) |
| + e.writePolicy(t, fmt.Sprintf(`{"schema":1,"require":[{"app":"console","prover":"dev","key":%q}]}`, otherPub)) |
| + if err := verify(); !errors.Is(err, ErrHeldNotProven) { |
| + t.Fatalf("rotated key: %v, want ErrHeldNotProven", err) |
| + } |
| + // A policy that does not load. |
| + e.writePolicy(t, "{not json") |
| + if err := verify(); !errors.Is(err, ErrHeldNotProven) { |
| + t.Fatalf("broken policy: %v, want ErrHeldNotProven", err) |
| + } |
| + // Uncovered: D3 has nothing to say about these bytes, as for adoption. |
| + e.require(t, "other") |
| + if err := StageArtifact(e.opt, "bab", "v1", "console", []byte("replaced later")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := verify(); err != nil { |
| + t.Fatalf("uncovered: %v", err) |
| + } |
| + |
| + // Another version live is not published. |
| + if err := PointLive(e.opt, heldTestHost, "bab", "v0", "console"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := verify(); !errors.Is(err, ErrHeldNotLive) { |
| + t.Fatalf("on v0: %v, want ErrHeldNotLive", err) |
| + } |
| + if err := os.Remove(LiveLink(e.opt, heldTestHost)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := verify(); !errors.Is(err, ErrHeldNotLive) { |
| + t.Fatalf("no link: %v, want ErrHeldNotLive", err) |
| + } |
| + |
| + // No lock, no gate: refused. |
| + if err := VerifyHeldLive(e.ctx, e.store, e.opt, heldRoute(), "v1", e.gate, nil); err == nil { |
| + t.Fatal("a reading without the lock passed") |
| + } |
| + if err := VerifyHeldLive(e.ctx, e.store, e.opt, heldRoute(), "v1", nil, passLock); err == nil { |
| + t.Fatal("a reading without a gate passed") |
| + } |
| +} |
diff --git a/internal/release/materializeheld.go b/internal/release/materializeheld.go| index 2c24a2ed9..c36d618a0 100644 |
| --- a/internal/release/materializeheld.go |
| +++ b/internal/release/materializeheld.go |
| @@ -55,6 +55,71 @@ var materializeHeldStage = ensureStaged |
| // swap. A test seam: the window a replaced staged file lands in. |
| var heldBeforeSwap = func() {} |
| +// heldAfterSwap runs inside the publication between the link swap and the |
| +// second bind. A test seam: the window in which live/<host> names a version |
| +// whose bytes are not yet vouched for, which the rollback may undo. |
| +var heldAfterSwap = func() {} |
| + |
| +// ErrHeldNotLive is VerifyHeldLive finding live/<host> on another version, |
| +// or on none: the held release is not published on this box. |
| +var ErrHeldNotLive = errors.New("release: live/<host> does not name the held release") |
| + |
| +var errHeldNoLock = errors.New("release: a held reading needs the publication's lock") |
| + |
| +// VerifyHeldLive is T7's reading of a published held release (final review |
| +// finding 1). A link that names the version is no evidence by itself: the |
| +// publication's rollback can undo it, and the bytes it names or the box's |
| +// proof policy can change after it was written. So the reading passes only |
| +// when, now: |
| +// |
| +// - D3 admits the release under the policy in force (heldProofAdmit, the |
| +// publication's own admit, after BeginPass re-reads the policy), and |
| +// - under lock, live/<host> names version and bind accepts the bytes it |
| +// names (ProofGate.bindRecording, the publication's own check; no |
| +// refusal is recorded, since a reading changes nothing). |
| +// |
| +// lock must be the lock the publication's guard holds across the swap, the |
| +// re-hash and the rollback (internal/host withHeldStageLock), so a reading |
| +// never sees a link a failing publication is about to take back. A nil |
| +// lock or gate is refused. A link on another version, or none, is |
| +// ErrHeldNotLive; a refusal is ErrHeldNotProven. |
| +func VerifyHeldLive(ctx context.Context, s Store, optDir string, rt registry.Route, version string, proof *ProofGate, lock func(read func() error) error) error { |
| + if err := heldStageable(rt, version); err != nil { |
| + return err |
| + } |
| + if proof == nil { |
| + return errHeldNoProofGate |
| + } |
| + if lock == nil { |
| + return errHeldNoLock |
| + } |
| + // A peek, so a release not yet published is not admitted twice (the |
| + // stage that follows admits it). Only the reading under lock passes. |
| + if live, err := LiveVersion(optDir, rt.Host); err != nil { |
| + return err |
| + } else if live != version { |
| + return ErrHeldNotLive |
| + } |
| + proof.BeginPass() |
| + ptr, v, err := heldProofAdmit(ctx, s, optDir, rt, version, proof) |
| + if err != nil { |
| + return err |
| + } |
| + return lock(func() error { |
| + live, err := LiveVersion(optDir, rt.Host) |
| + if err != nil { |
| + return err |
| + } |
| + if live != version { |
| + return ErrHeldNotLive |
| + } |
| + if !proof.bindRecording(optDir, rt, ptr, v, live, proof.now(), func(Refusal) error { return nil }) { |
| + return fmt.Errorf("%w: %s at %s: the bytes live/%s names are not the proven ones", ErrHeldNotProven, rt.Host, version, rt.Host) |
| + } |
| + return nil |
| + }) |
| +} |
| + |
| // 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 |
| @@ -176,6 +241,7 @@ func publishHeld(optDir string, rt registry.Route, version string, proof *ProofG |
| } |
| return err |
| } |
| + heldAfterSwap() |
| if !proof.bindRecording(optDir, rt, ptr, v, cur, proof.now(), record) { |
| if rerr := restoreLive(optDir, rt, cur); rerr != nil { |
| return errors.Join(notProven, fmt.Errorf("putting live/%s back on %q: %w", rt.Host, cur, rerr)) |
| @@ -341,7 +407,7 @@ type HeldStageRequest struct { |
| } |
| // HeldStaged is what the root pass leaves at HeldStagedDir(dataDir)/<host>.json |
| -// after serving a request. It is a hint only: T7 trusts LiveVersion. |
| +// after serving a request. It is a hint only: T7 trusts VerifyHeldLive. |
| type HeldStaged struct { |
| Version string `json:"version"` |
| Op string `json:"op"` |