amadan
settings-4: Astra round 2 fixes: the planner's recovery records; root's publication is fenced against a cut
Astra round 2 finding 1 (ruling R-PLANNED-RECOVERY): the activator carries the planner's checked selection of the good build in place of another (activator.PlannedSelection) on the context of that exec's record only: a planned fallback after B failed, the good build booted untried, and a recovery boot (step 2b, spawnPlanned, a claimed start). The pre-exec revalidation (release.HeldExecStale) admits that build when the state read under the lease holds exactly the replaced build as the held target; the record is still written. The good build is not exempt in general. Astra round 2 finding 2 (ruling R-CARRY-FENCE): root's consumed publication and the cut's carry serialize on a per-host flock under the consumed directory (Linux; a !linux stub). Root takes it, re-reads the route and publishes only for a row no move holds (publishGate), else installs nothing for it this pass. The agent's carry requires the row's move hold first, then takes the fence (read-only fd) across the read and the write; no lock file means no publication in flight.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
16 files changed,
+1090
−16
internal/activator/beforeexec.go+9 −1internal/activator/claimwake.go+6 −0internal/activator/plannedselection.go+73 −0internal/activator/wakeplan.go+15 −7internal/host/buildstate.go+31 −1internal/host/buildstate_carry_test.go+5 −1internal/host/buildstate_fence_linux_test.go+103 −0internal/host/buildstate_planned_test.go+316 −0internal/release/adopt.go+11 −4internal/release/consumedfence.go+112 −0internal/release/consumedfence_linux.go+75 −0internal/release/consumedfence_linux_test.go+269 −0internal/release/consumedfence_other.go+14 −0internal/release/heldwake.go+17 −1internal/release/heldwake_test.go+33 −0internal/release/updatesim_test.go+1 −1
diff --git a/internal/activator/beforeexec.go b/internal/activator/beforeexec.go| index 1777dd292..5f6a6be18 100644 |
| --- a/internal/activator/beforeexec.go |
| +++ b/internal/activator/beforeexec.go |
| @@ -152,7 +152,15 @@ func (a *Service) settleWakeIdentity(ctx context.Context, ib identityBackend, e |
| return id, pend, errStagedBuildNotPublished |
| } |
| } |
| - if err := a.recordBuild(ctx, rt.Host, id.Version); err != nil { |
| + // A recovery plan's boot (the good build in place of the installed one |
| + // the failure book blocks) is recorded with the planner's selection |
| + // (ruling R-PLANNED-RECOVERY), so the agent's revalidation admits it |
| + // before root's consumed file names the installed build failed. |
| + rctx := ctx |
| + if plan.planned() { |
| + rctx = withRecoverySelection(ctx, plan, rt, id.Version) |
| + } |
| + if err := a.recordBuild(rctx, rt.Host, id.Version); err != nil { |
| return id, pend, err |
| } |
| return id, pend, nil |
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go| index 2ad6aa3ea..81b6616ee 100644 |
| --- a/internal/activator/claimwake.go |
| +++ b/internal/activator/claimwake.go |
| @@ -533,6 +533,12 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc |
| } |
| } |
| at := WithBuildStateLocation(ctx, BuildStateLocation{Bucket: r.Bucket, Prefix: admitted.Claim.Holder.Location}) |
| + // A planned recovery boot records with the planner's selection |
| + // (ruling R-PLANNED-RECOVERY), as doWake's step 2b does. |
| + a.mu.Lock() |
| + plan, route := e.plan, e.route |
| + a.mu.Unlock() |
| + at = withRecoverySelection(at, plan, route, s.Staged.Version) |
| if err := a.recordBuild(at, e.host, s.Staged.Version); err != nil { |
| unrecorded = true |
| return err |
diff --git a/internal/activator/plannedselection.go b/internal/activator/plannedselection.go| new file mode 100644 |
| index 000000000..1ff9b515a |
| --- /dev/null |
| +++ b/internal/activator/plannedselection.go |
| @@ -0,0 +1,73 @@ |
| +package activator |
| + |
| +import ( |
| + "context" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +// PlannedSelection is the planner's checked choice of a build in place of |
| +// another, for one exec (settings plan 4, ruling R-PLANNED-RECOVERY; zero- |
| +// downtime spec §8.2): Build is the good build the plan boots instead of |
| +// InPlaceOf. It is one of |
| +// |
| +// - a planned fallback: a first wake's B failed and was proven stopped, |
| +// and the good build starts in its place (spawnPlanned, |
| +// spawnPlannedWorker); |
| +// - the good build booted untried, when too little of B's health window |
| +// was left at its exec (spawnPlanned); |
| +// - a recovery boot: decideWake planned the good build in place of the |
| +// installed build, which the failure book blocks (settleWakeIdentity, |
| +// spawnPlanned, a claimed start). |
| +// |
| +// The activator puts it on the context of that exec's record only, after |
| +// the plan's selection gate (Check) passed for Build. The agent's pre-exec |
| +// revalidation (release.HeldExecStale) admits Build when the state read under |
| +// the lease holds InPlaceOf as the instance's target: the planner's reason |
| +// for leaving InPlaceOf is then the reason the held target is left, as |
| +// zero-downtime spec §6 and settings spec §6 require. The record is still |
| +// written. It never admits the good build in general: a context with no |
| +// selection, or one made in place of another build, is revalidated as |
| +// before. |
| +type PlannedSelection struct{ Build, InPlaceOf string } |
| + |
| +type plannedSelectionKey struct{} |
| + |
| +// WithPlannedSelection returns ctx carrying s. A selection with either part |
| +// empty reads back as none, so a caller can mask one it inherited. |
| +func WithPlannedSelection(ctx context.Context, s PlannedSelection) context.Context { |
| + return context.WithValue(ctx, plannedSelectionKey{}, s) |
| +} |
| + |
| +// PlannedSelectionFrom is the selection ctx carries, if a whole one. |
| +func PlannedSelectionFrom(ctx context.Context) (PlannedSelection, bool) { |
| + s, ok := ctx.Value(plannedSelectionKey{}).(PlannedSelection) |
| + if !ok || s.Build == "" || s.InPlaceOf == "" { |
| + return PlannedSelection{}, false |
| + } |
| + return s, true |
| +} |
| + |
| +// recoverySelection is the selection a recovery plan makes (decideWake's |
| +// recovery: the good build booted because the failure book blocks the |
| +// installed build, or the installed build is not staged correctly): plan |
| +// boots the route's good build, as no first wake, in place of the build root |
| +// installed (rt.Version, the planner's "installed"). A plan that boots the |
| +// installed build itself, the good build included, selects nothing in place |
| +// of anything, and its record is revalidated as any other. |
| +func recoverySelection(plan WakePlan, rt registry.Route) (PlannedSelection, bool) { |
| + boot := plan.Boot.Version |
| + if !plan.planned() || plan.FirstWake || boot == "" || boot != rt.GoodVersion || rt.Version == "" || rt.Version == boot { |
| + return PlannedSelection{}, false |
| + } |
| + return PlannedSelection{Build: boot, InPlaceOf: rt.Version}, true |
| +} |
| + |
| +// withRecoverySelection is ctx carrying plan's recovery selection when plan |
| +// is a recovery and build is its boot; ctx itself otherwise. |
| +func withRecoverySelection(ctx context.Context, plan WakePlan, rt registry.Route, build string) context.Context { |
| + if s, ok := recoverySelection(plan, rt); ok && s.Build == build { |
| + return WithPlannedSelection(ctx, s) |
| + } |
| + return ctx |
| +} |
diff --git a/internal/activator/wakeplan.go b/internal/activator/wakeplan.go| index fb035f276..a74ae5839 100644 |
| --- a/internal/activator/wakeplan.go |
| +++ b/internal/activator/wakeplan.go |
| @@ -357,7 +357,11 @@ func (a *Service) spawnPlanned(ctx context.Context, e *actEntry, rt registry.Rou |
| healthBy = e.wakeStart.Add(firstWakeBudget - firstWakeKillReserve) |
| gate = verdictFloorGate(gate, healthBy, &left) |
| } |
| - proc, ver, g, err := a.startPinnedAt(ctx, rt, genOf(boot), gate, healthBy) |
| + // Every exec here that runs the good build in another build's place is |
| + // recorded with the planner's selection (ruling R-PLANNED-RECOVERY): a |
| + // recovery boot (normally recorded at step 2b already), the good build |
| + // booted untried and the fallback. Each exec's own context only. |
| + proc, ver, g, err := a.startPinnedAt(withRecoverySelection(ctx, plan, rt, boot.Version), rt, genOf(boot), gate, healthBy) |
| w.bootRan = err == nil || startMayHaveRun(proc, err) |
| if proc == nil && errors.Is(err, errNoTimeForVerdict) { |
| // B never ran: too little of its window was left for a verdict at |
| @@ -369,7 +373,8 @@ func (a *Service) spawnPlanned(ctx context.Context, e *actEntry, rt registry.Rou |
| log.Printf("activator: %s: first wake of %s had %v of its health window left at the exec, under the %v floor; booting the good build %s without trying it", |
| rt.Host, boot.Version, left.Round(time.Millisecond), firstWakeMinVerdict, good.Version) |
| w.untried = true |
| - gproc, gver, gg, gerr := a.startPinnedAt(ctx, rt, genOf(good), a.planGate(ctx, e, lease, plan, good.Version), time.Time{}) |
| + gctx := WithPlannedSelection(ctx, PlannedSelection{Build: good.Version, InPlaceOf: boot.Version}) |
| + gproc, gver, gg, gerr := a.startPinnedAt(gctx, rt, genOf(good), a.planGate(ctx, e, lease, plan, good.Version), time.Time{}) |
| if gerr != nil { |
| return gproc, gver, gg, fmt.Errorf("the good build %s, booted without trying %s: %w", good.Version, boot.Version, gerr) |
| } |
| @@ -397,7 +402,8 @@ func (a *Service) spawnPlanned(ctx context.Context, e *actEntry, rt registry.Rou |
| return proc, ver, g, fmt.Errorf("%w; its end was not proven, so the good build was not started: %v", err, kerr) |
| } |
| good := *plan.Fallback |
| - fproc, fver, fg, ferr := a.startPinnedAt(ctx, rt, genOf(good), a.planGate(ctx, e, lease, plan, good.Version), time.Time{}) |
| + fctx := WithPlannedSelection(ctx, PlannedSelection{Build: good.Version, InPlaceOf: boot.Version}) |
| + fproc, fver, fg, ferr := a.startPinnedAt(fctx, rt, genOf(good), a.planGate(ctx, e, lease, plan, good.Version), time.Time{}) |
| if ferr != nil { |
| if fproc == nil && !genStartUnproven(ferr) { |
| // Nothing of the good build ran (its gate refused: a fence, a |
| @@ -457,8 +463,8 @@ func (a *Service) spawnPlannedWorker(ctx context.Context, e *actEntry, rt regist |
| if err != nil { |
| return nil, release.StagedIdentity{}, err |
| } |
| - start := func(id release.StagedIdentity) (Proc, release.StagedIdentity, error) { |
| - p, got, pend, err := a.startWorker(ctx, rt, env, &id, a.planGate(ctx, e, lease, plan, id.Version)) |
| + start := func(sctx context.Context, id release.StagedIdentity) (Proc, release.StagedIdentity, error) { |
| + p, got, pend, err := a.startWorker(sctx, rt, env, &id, a.planGate(ctx, e, lease, plan, id.Version)) |
| if err != nil { |
| return p, got, err |
| } |
| @@ -474,7 +480,7 @@ func (a *Service) spawnPlannedWorker(ctx context.Context, e *actEntry, rt regist |
| } |
| canFall := plan.FirstWake && plan.Fallback != nil && plan.Mode == ModeOverlap |
| boot := plan.Boot |
| - proc, id, err := start(boot) |
| + proc, id, err := start(withRecoverySelection(ctx, plan, rt, boot.Version), boot) |
| w.bootRan = err == nil || startMayHaveRun(proc, err) |
| if err == nil || proc == nil { |
| // Ready, or nothing of the build ran: a refusal at the gate, an |
| @@ -496,7 +502,9 @@ func (a *Service) spawnPlannedWorker(ctx context.Context, e *actEntry, rt regist |
| if kerr := a.killFailed(rt, Gen{Build: id.Version}, proc, true); kerr != nil { |
| return proc, id, fmt.Errorf("%w; its end was not proven, so the good build was not started: %v", err, kerr) |
| } |
| - fproc, fid, ferr := start(*plan.Fallback) |
| + // The fallback is recorded with the planner's selection (ruling |
| + // R-PLANNED-RECOVERY), as spawnPlanned's is. |
| + fproc, fid, ferr := start(WithPlannedSelection(ctx, PlannedSelection{Build: plan.Fallback.Version, InPlaceOf: boot.Version}), *plan.Fallback) |
| if ferr != nil { |
| if fproc == nil { |
| // Nothing of the good build ran: B's failure is still recorded, |
diff --git a/internal/host/buildstate.go b/internal/host/buildstate.go| index 81dfa2e41..1634b42e5 100644 |
| --- a/internal/host/buildstate.go |
| +++ b/internal/host/buildstate.go |
| @@ -504,6 +504,10 @@ func (b *buildStates) write(ctx context.Context, host string, merge func(cur *re |
| // cannot parse rather than overwrite it (ruling R-BS-CORRUPT). For a row |
| // settings apply to, a state that now names another held target than build |
| // (release.HeldExecStale) refuses it too: the wake decided on a stale copy. |
| +// The one exception is the zero-downtime planner's checked selection of |
| +// build in place of that target, which the activator carries on ctx for this |
| +// exec only (activator.PlannedSelection; ruling R-PLANNED-RECOVERY): it is |
| +// admitted, and still recorded. |
| func (b *buildStates) beforeExec(ctx context.Context, host, build string) error { |
| if build == "" { |
| return fmt.Errorf("no build to record for %s", host) |
| @@ -517,9 +521,13 @@ func (b *buildStates) beforeExec(ctx context.Context, host, build string) error |
| // refused here, before anything is written or run. |
| rt, _ := b.reg.Get(host) |
| held := rt != nil && settingsApply(*rt) |
| + inPlaceOf := "" |
| + if sel, ok := activator.PlannedSelectionFrom(ctx); ok && sel.Build == build { |
| + inPlaceOf = sel.InPlaceOf |
| + } |
| _, err = b.write(ctx, host, func(cur *release.BuildState) (release.BuildState, bool, error) { |
| if held { |
| - if err := release.HeldExecStale(release.DecideHeldWake(b.heldWakeInput(*rt, cur)), cur, build); err != nil { |
| + if err := release.HeldExecStale(release.DecideHeldWake(b.heldWakeInput(*rt, cur)), cur, build, inPlaceOf); err != nil { |
| return release.BuildState{}, false, err |
| } |
| } |
| @@ -538,7 +546,29 @@ func (b *buildStates) beforeExec(ctx context.Context, host, build string) error |
| // host's lock and against the etag it read (write), so it never overwrites a |
| // newer record. A consumed file that does not read refuses the carry, and so |
| // the cut: a request this box closed would reopen on the destination. |
| +// |
| +// The carry runs inside the carry fence (release.FenceCarry; ruling |
| +// R-CARRY-FENCE): root's publication for host and this carry serialize on |
| +// one lock, and root publishes only for a row no move holds. So the row's |
| +// move hold must already be effective here (every lane sets it before its |
| +// carry: the legacy cut's "moving", the Phase 1 cut's held "moving", the |
| +// return lane's "releasing"); a row in no hold refuses the carry, as root |
| +// could still publish after it. A row gone from the registry gets no |
| +// publication either. Then the fence, held across the read and the write: a |
| +// root record in flight lands first and is carried. |
| func (b *buildStates) carryRequest(ctx context.Context, host, bucket, prefix string) error { |
| + rt, err := b.reg.Get(host) |
| + if err != nil { |
| + return fmt.Errorf("read the route of %s: %w", host, err) |
| + } |
| + if rt != nil && !registry.IsHold(rt.Status) { |
| + return fmt.Errorf("%s is in no move hold (status %q), so root could still publish a request record after the carry", host, rt.Status) |
| + } |
| + unlock, err := release.FenceCarry(ctx, b.optDir, host) |
| + if err != nil { |
| + return fmt.Errorf("the carry fence for %s: %w", host, err) |
| + } |
| + defer unlock() |
| consumed, err := release.ReadConsumed(b.optDir, host) |
| if err != nil { |
| return fmt.Errorf("read root's consumed file for %s: %w", host, err) |
diff --git a/internal/host/buildstate_carry_test.go b/internal/host/buildstate_carry_test.go| index 282a7241f..f2b454d19 100644 |
| --- a/internal/host/buildstate_carry_test.go |
| +++ b/internal/host/buildstate_carry_test.go |
| @@ -165,11 +165,15 @@ func TestTheReturnLaneCarriesTheRequest(t *testing.T) { |
| // given (release.CarryRequest), last_executed untouched, and writes only |
| // when something changes. |
| func TestCarryRequestWritesTheConsumedRecordAtTheLocation(t *testing.T) { |
| - b, s, _ := testBuildStates(t) |
| + b, s, reg := testBuildStates(t) |
| ctx := context.Background() |
| if err := put(t, b, ctx, astraHost, release.BuildState{LastExecuted: "A"}); err != nil { |
| t.Fatal(err) |
| } |
| + // The cut holds the row before it carries (ruling R-CARRY-FENCE). |
| + if err := reg.SetStatus(astraHost, "moving"); err != nil { |
| + t.Fatal(err) |
| + } |
| key := release.BuildStateKey("inst/" + astraHost) |
| etag := mustEtag(t, s, key) |
| // No consumed file: nothing to carry, nothing written. |
diff --git a/internal/host/buildstate_fence_linux_test.go b/internal/host/buildstate_fence_linux_test.go| new file mode 100644 |
| index 000000000..1c51c2012 |
| --- /dev/null |
| +++ b/internal/host/buildstate_fence_linux_test.go |
| @@ -0,0 +1,103 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "os" |
| + "path/filepath" |
| + "testing" |
| + "time" |
| + |
| + "golang.org/x/sys/unix" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// Settings plan 4, Astra round 2 finding 2 (ruling R-CARRY-FENCE): the |
| +// agent's carry (buildStates.carryRequest) takes the carry fence only once |
| +// the row's move hold is effective, and holds it across the carry, so a root |
| +// publication in flight lands before the carry reads the file and is |
| +// carried. |
| + |
| +// rootHoldsFence takes root's side of host's fence as root's publication |
| +// does (the lock file created, flock exclusive) and returns its release. |
| +func rootHoldsFence(t *testing.T, optDir, host string) func() { |
| + t.Helper() |
| + if err := os.MkdirAll(filepath.Join(optDir, "consumed"), 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + f, err := os.OpenFile(release.ConsumedLockPath(optDir, host), os.O_RDWR|os.O_CREATE, 0o644) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := unix.Flock(int(f.Fd()), unix.LOCK_EX); err != nil { |
| + t.Fatal(err) |
| + } |
| + return func() { _ = unix.Flock(int(f.Fd()), unix.LOCK_UN); _ = f.Close() } |
| +} |
| + |
| +// Root holds the fence and is about to publish R1 completed, failed; the |
| +// cut, its row held moving, asks for the carry: it waits for root, then |
| +// carries the record root published. |
| +func TestTheCarryWaitsForRootsPublicationAndCarriesIt(t *testing.T) { |
| + b, s, reg := testBuildStates(t) |
| + ctx := context.Background() |
| + if err := put(t, b, ctx, astraHost, release.BuildState{LastExecuted: "A", Resolved: &release.ResolvedRequest{ID: 1, Target: "T"}}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.SetStatus(astraHost, "moving"); err != nil { |
| + t.Fatal(err) |
| + } |
| + unlock := rootHoldsFence(t, b.optDir, astraHost) |
| + done := make(chan error, 1) |
| + go func() { done <- b.carryRequest(ctx, astraHost, "hib", "inst/"+astraHost) }() |
| + select { |
| + case err := <-done: |
| + unlock() |
| + t.Fatalf("the carry ran while root held the fence: %v", err) |
| + case <-time.After(200 * time.Millisecond): |
| + } |
| + writeConsumed(t, b, release.Consumed{Resolved: &release.ResolvedRequest{ID: 1, Target: "T"}, Completed: 1, CompletedFailed: true}) |
| + unlock() |
| + select { |
| + case err := <-done: |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + case <-time.After(10 * time.Second): |
| + t.Fatal("the carry never took the fence") |
| + } |
| + st, _, err := getState(ctx, s, release.BuildStateKey("inst/"+astraHost), astraHost) |
| + if err != nil || st == nil || st.Completed != 1 || !st.CompletedFailed || st.LastExecuted != "A" { |
| + t.Fatalf("state %+v, %v; want R1 completed failed carried, A untouched", st, err) |
| + } |
| +} |
| + |
| +// A carry for a row the move does not hold is refused: root could still |
| +// publish after it. Nothing is written. |
| +func TestTheCarryRefusesARowNoMoveHolds(t *testing.T) { |
| + b, s, _ := testBuildStates(t) |
| + ctx := context.Background() |
| + writeConsumed(t, b, release.Consumed{Resolved: &release.ResolvedRequest{ID: 1, Target: "T"}, Completed: 1, CompletedFailed: true}) |
| + if err := b.carryRequest(ctx, astraHost, "hib", "inst/"+astraHost); err == nil { |
| + t.Fatal("a carry for a row no move holds went ahead") |
| + } |
| + if _, _, err := s.Get(ctx, release.BuildStateKey("inst/"+astraHost)); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("the refused carry wrote: %v", err) |
| + } |
| +} |
| + |
| +// No consumed directory: no record to carry and no lock to take; the agent |
| +// creates nothing there. |
| +func TestTheCarryWithNoConsumedDirectoryTakesNoLock(t *testing.T) { |
| + b, _, reg := testBuildStates(t) |
| + if err := reg.SetStatus(astraHost, "moving"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := b.carryRequest(context.Background(), astraHost, "hib", "inst/"+astraHost); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := os.Stat(filepath.Join(b.optDir, "consumed")); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("the carry created the consumed directory: %v", err) |
| + } |
| +} |
diff --git a/internal/host/buildstate_planned_test.go b/internal/host/buildstate_planned_test.go| new file mode 100644 |
| index 000000000..38eec7b52 |
| --- /dev/null |
| +++ b/internal/host/buildstate_planned_test.go |
| @@ -0,0 +1,316 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "errors" |
| + "net" |
| + "net/http" |
| + "net/http/httptest" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/cgroup" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// Settings plan 4, Astra round 2 finding 1 (ruling R-PLANNED-RECOVERY): the |
| +// pre-exec write's revalidation admits the planner's checked recovery or |
| +// fallback onto the good build, for that transition only, and still writes |
| +// the record. These run a real activator wake (ServeHTTP, the planner, the |
| +// held wake check and every exec through startPinnedAt and step 2b) with the |
| +// real build-state hook (buildStates.beforeExec) and heldTarget. |
| + |
| +// plannedProc is a fake process: Stop ends it. |
| +type plannedProc struct{ exited atomic.Bool } |
| + |
| +func (p *plannedProc) Stop(context.Context) error { p.exited.Store(true); return nil } |
| +func (p *plannedProc) Exited() bool { return p.exited.Load() } |
| + |
| +type plannedLease struct{} |
| + |
| +func (plannedLease) Replicate(context.Context, string, time.Duration) (activator.Rep, error) { |
| + return plannedRep{}, nil |
| +} |
| +func (plannedLease) Release(context.Context) error { return nil } |
| + |
| +type plannedRep struct{} |
| + |
| +func (plannedRep) Err() error { return nil } |
| +func (plannedRep) Close(context.Context) error { return nil } |
| + |
| +// plannedBackend is the smallest exec backend a planned wake runs on: a |
| +// lease and restore that always succeed, staged generations (a build in |
| +// fail exits during its health check, with its process; any other answers |
| +// HTTP 200 on a real socket), a generation kill, the live link's version, |
| +// and the executable identity step 2b records. |
| +type plannedBackend struct { |
| + socket string |
| + live string |
| + fail map[string]bool |
| + |
| + mu sync.Mutex |
| + started []string |
| + killed []uint64 |
| + procs map[uint64]*plannedProc |
| +} |
| + |
| +func (b *plannedBackend) Acquire(context.Context, registry.Route) (activator.Lease, error) { |
| + return plannedLease{}, nil |
| +} |
| +func (b *plannedBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil } |
| +func (b *plannedBackend) Seed(context.Context, registry.Route) error { return nil } |
| +func (b *plannedBackend) Start(context.Context, registry.Route) (activator.Proc, string, error) { |
| + return nil, "", errors.New("a planned wake never starts the live link") |
| +} |
| +func (b *plannedBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil } |
| +func (b *plannedBackend) HasSidecar(registry.Route) bool { return false } |
| +func (b *plannedBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) { |
| + return nil, errors.New("no sidecar") |
| +} |
| +func (b *plannedBackend) StartWorker(context.Context, registry.Route, map[string]string) (activator.Proc, string, error) { |
| + return nil, "", errors.New("no worker") |
| +} |
| + |
| +func (b *plannedBackend) StartStaged(_ context.Context, _ registry.Route, g activator.Gen) (activator.Gen, activator.Proc, error) { |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + b.started = append(b.started, g.Build) |
| + p := &plannedProc{} |
| + if b.procs == nil { |
| + b.procs = map[uint64]*plannedProc{} |
| + } |
| + b.procs[g.N] = p |
| + if b.fail[g.Build] { |
| + return g, p, errors.New("the process exited during its health check") |
| + } |
| + g.Socket = b.socket |
| + return g, p, nil |
| +} |
| +func (b *plannedBackend) GenTier(registry.Route, string) cgroup.Limit { return cgroup.TierDefault } |
| +func (b *plannedBackend) GenReady(registry.Route, activator.Gen, activator.Proc) {} |
| +func (b *plannedBackend) RaiseGenParent(context.Context, registry.Route, cgroup.Limit, cgroup.Limit) error { |
| + return nil |
| +} |
| +func (b *plannedBackend) RestoreGenParent(context.Context, registry.Route, cgroup.Limit) error { |
| + return nil |
| +} |
| +func (b *plannedBackend) GenSwitched(registry.Route, activator.Gen, activator.Proc) error { return nil } |
| +func (b *plannedBackend) KillGen(_ context.Context, _ registry.Route, n uint64) error { |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + b.killed = append(b.killed, n) |
| + if p := b.procs[n]; p != nil { |
| + p.exited.Store(true) |
| + } |
| + return nil |
| +} |
| +func (b *plannedBackend) LiveVersion(string) (string, error) { return b.live, nil } |
| +func (b *plannedBackend) LiveIdentity(context.Context, registry.Route) (release.StagedIdentity, error) { |
| + return release.StagedIdentity{}, errors.New("a planned wake never captures the live link") |
| +} |
| +func (b *plannedBackend) StartGenExe(context.Context, registry.Route, uint64, bool, release.StagedIdentity) (activator.Proc, activator.Gen, error) { |
| + return nil, activator.Gen{}, errors.New("a planned wake never starts the live link") |
| +} |
| +func (b *plannedBackend) SeedExe(context.Context, registry.Route, release.StagedIdentity) error { |
| + return nil |
| +} |
| + |
| +func (b *plannedBackend) builds() []string { |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + return append([]string(nil), b.started...) |
| +} |
| + |
| +// healthySocket serves HTTP 200 on a unix socket under a short directory |
| +// (a socket path is bounded). |
| +func healthySocket(t *testing.T) string { |
| + t.Helper() |
| + dir, err := os.MkdirTemp("", "pr") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { os.RemoveAll(dir) }) |
| + path := filepath.Join(dir, "s") |
| + l, err := net.Listen("unix", path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })} |
| + go srv.Serve(l) //nolint:errcheck |
| + t.Cleanup(func() { srv.Close() }) |
| + return path |
| +} |
| + |
| +// plannedStaged writes a staged build and returns its identity (the |
| +// activator verifies the bytes before every staged exec). |
| +func plannedStaged(t *testing.T, dir, ver string) release.StagedIdentity { |
| + t.Helper() |
| + body := []byte("build " + ver) |
| + path := filepath.Join(dir, ver) |
| + if err := os.WriteFile(path, body, 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + sum := sha256.Sum256(body) |
| + return release.StagedIdentity{Version: ver, StagedPath: path, Digest: hex.EncodeToString(sum[:])} |
| +} |
| + |
| +// plannedWake is a manual instance with good build A (GoodVersion), a |
| +// planner that answers plan, and the real build-state hook and held wake |
| +// check over b's store and registry. The activator leases at the hibernate |
| +// prefix of bucket "hib", which b reads there too. |
| +func plannedWake(t *testing.T, live string, fail map[string]bool, plan func(good, t release.StagedIdentity) activator.WakePlan) (*activator.Service, *plannedBackend, *buildStates, release.Store) { |
| + t.Helper() |
| + b, s, reg := testBuildStates(t) |
| + b.prefix = func(h string) string { return activator.HibernatePrefix("", h) } |
| + dir := t.TempDir() |
| + good, next := plannedStaged(t, dir, "A"), plannedStaged(t, dir, "T") |
| + if err := reg.SetGoodBuild(astraHost, registry.GoodBuild{}, registry.GoodBuild{Version: "A", Digest: good.Digest}); err != nil { |
| + t.Fatal(err) |
| + } |
| + // Root's install stamp names the live build, as root's repoint leaves it. |
| + if err := reg.SetVersions(astraHost, live, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable", InstallRequest: 1}) |
| + be := &plannedBackend{socket: healthySocket(t), live: live, fail: fail} |
| + svc := activator.New(activator.Config{Bucket: "hib", Idle: time.Hour}, be, reg) |
| + svc.Planner = func(context.Context, registry.Route) activator.WakePlan { return plan(good, next) } |
| + svc.OnGen = func(activator.GenEvent) {} |
| + svc.BeforeExec = b.beforeExec |
| + svc.HeldTarget = b.heldTarget |
| + t.Cleanup(func() { svc.Shutdown(context.Background()) }) |
| + return svc, be, b, s |
| +} |
| + |
| +// serve sends one request through the activator: it wakes the instance. |
| +func serve(t *testing.T, svc *activator.Service, reg *registry.Registry) *httptest.ResponseRecorder { |
| + t.Helper() |
| + rt, err := reg.Get(astraHost) |
| + if err != nil || rt == nil { |
| + t.Fatalf("route: %v", err) |
| + } |
| + rec := httptest.NewRecorder() |
| + req := httptest.NewRequest(http.MethodGet, "http://"+astraHost+"/", nil) |
| + proxy := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.Write([]byte("served")) }) //nolint:errcheck |
| + if !svc.ServeHTTP(rec, req, *rt, proxy) { |
| + t.Fatal("the activator did not take the request") |
| + } |
| + return rec |
| +} |
| + |
| +func durable(t *testing.T, b *buildStates, s release.Store) *release.BuildState { |
| + t.Helper() |
| + st, _, err := getState(context.Background(), s, release.BuildStateKey(astraHost), astraHost) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return st |
| +} |
| + |
| +// The overlap wake's fallback (Astra round 2 finding 1's scenario): a manual |
| +// instance with good build A and Update now resolved to T. Its first wake |
| +// records T (completing R1) and T fails its health check. The planner's |
| +// fallback A must record and run, though root has not reported T failed and |
| +// the durable state's last_executed is T. |
| +func TestTheOverlapFallbackAfterTFailsIsRecordedAndRuns(t *testing.T) { |
| + svc, be, b, s := plannedWake(t, "T", map[string]bool{"T": true}, func(good, next release.StagedIdentity) activator.WakePlan { |
| + return activator.WakePlan{Boot: next, Mode: activator.ModeOverlap, FirstWake: true, Fallback: &good} |
| + }) |
| + if err := put(t, b, context.Background(), astraHost, release.BuildState{LastExecuted: "A"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + setCopy(t, b.reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "A"}}) |
| + decided(t, b, release.Consumed{Resolved: &release.ResolvedRequest{ID: 1, Target: "T"}, Reason: string(release.ReasonRequest), Desired: "T", Detail: "T"}) |
| + |
| + rec := serve(t, svc, b.reg) |
| + if rec.Code != http.StatusOK || rec.Body.String() != "served" { |
| + t.Fatalf("the wake was not served by the fallback: %d %q; generations %v", rec.Code, rec.Body.String(), be.builds()) |
| + } |
| + if got := be.builds(); len(got) != 2 || got[0] != "T" || got[1] != "A" { |
| + t.Fatalf("generations %v, want T then the good build A", got) |
| + } |
| + st := durable(t, b, s) |
| + if st == nil || st.LastExecuted != "A" || st.Completed != 1 { |
| + t.Fatalf("durable state %+v; want A recorded before it ran, R1 completed by T's run", st) |
| + } |
| +} |
| + |
| +// A planned recovery boot onto the good build (decideWake's recovery: the |
| +// failure book blocks T) before root's consumed file names T failed: the |
| +// durable state's last_executed is T, and the planner chose A in T's place. |
| +// The record admits A for that transition and the wake runs it. |
| +func TestAPlannedRecoveryBeforeRootRecordsTheFailureRuns(t *testing.T) { |
| + svc, be, b, s := plannedWake(t, "T", nil, func(good, _ release.StagedIdentity) activator.WakePlan { |
| + return activator.WakePlan{Boot: good, Mode: activator.ModeOverlap} |
| + }) |
| + tState := release.BuildState{LastExecuted: "T", Resolved: &release.ResolvedRequest{ID: 1, Target: "T"}, Completed: 1} |
| + if err := put(t, b, context.Background(), astraHost, tState); err != nil { |
| + t.Fatal(err) |
| + } |
| + setCopy(t, b.reg, release.BuildStateCopy{Known: true, Replica: true, State: &tState}) |
| + decided(t, b, release.Consumed{Resolved: &release.ResolvedRequest{ID: 1, Target: "T"}, Reason: string(release.ReasonRequest), Desired: "T", Detail: "T"}) |
| + |
| + rec := serve(t, svc, b.reg) |
| + if rec.Code != http.StatusOK || rec.Body.String() != "served" { |
| + t.Fatalf("the recovery boot was not served: %d %q; generations %v", rec.Code, rec.Body.String(), be.builds()) |
| + } |
| + if got := be.builds(); len(got) != 1 || got[0] != "A" { |
| + t.Fatalf("generations %v, want the good build A alone", got) |
| + } |
| + if st := durable(t, b, s); st == nil || st.LastExecuted != "A" || st.Completed != 1 { |
| + t.Fatalf("durable state %+v; want A recorded, R1 still completed", st) |
| + } |
| +} |
| + |
| +// The negative (GoodVersion is not exempt in general; Astra round 1 finding |
| +// 1): the wake decided on a stale copy that names the good build A, which is |
| +// installed, so the planner plans A as an ordinary wake, not in place of |
| +// anything. The durable state says the data last ran T. With no planner |
| +// selection the record refuses A and nothing runs. The same hook, called |
| +// directly with no selection, or with a selection for another transition, |
| +// refuses A too. |
| +func TestTheGoodBuildWithoutThePlannersSelectionIsRefused(t *testing.T) { |
| + svc, be, b, s := plannedWake(t, "A", nil, func(good, _ release.StagedIdentity) activator.WakePlan { |
| + return activator.WakePlan{Boot: good, Mode: activator.ModeOverlap} |
| + }) |
| + if err := put(t, b, context.Background(), astraHost, release.BuildState{LastExecuted: "T"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + setCopy(t, b.reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "A"}}) |
| + decided(t, b, release.Consumed{Reason: string(release.ReasonManual), Desired: "A", Detail: "A"}) |
| + |
| + rec := serve(t, svc, b.reg) |
| + if rec.Code == http.StatusOK { |
| + t.Fatalf("the good build ran over T's data with no planner selection: %q; generations %v", rec.Body.String(), be.builds()) |
| + } |
| + if got := be.builds(); len(got) != 0 { |
| + t.Fatalf("generations %v, want none", got) |
| + } |
| + if st := durable(t, b, s); st == nil || st.LastExecuted != "T" { |
| + t.Fatalf("durable state %+v; want T untouched", st) |
| + } |
| + ctx := context.Background() |
| + if err := b.beforeExec(ctx, astraHost, "A"); err == nil || !strings.Contains(err.Error(), "T") { |
| + t.Fatalf("beforeExec(A) with no selection = %v; want a stale refusal naming T", err) |
| + } |
| + other := activator.WithPlannedSelection(ctx, activator.PlannedSelection{Build: "A", InPlaceOf: "B"}) |
| + if err := b.beforeExec(other, astraHost, "A"); err == nil { |
| + t.Fatal("beforeExec(A) with a selection made in place of another build was admitted") |
| + } |
| + wrong := activator.WithPlannedSelection(ctx, activator.PlannedSelection{Build: "C", InPlaceOf: "T"}) |
| + if err := b.beforeExec(wrong, astraHost, "A"); err == nil { |
| + t.Fatal("beforeExec(A) with a selection of another build was admitted") |
| + } |
| + if st := durable(t, b, s); st == nil || st.LastExecuted != "T" { |
| + t.Fatalf("durable state %+v; want T untouched", st) |
| + } |
| +} |
diff --git a/internal/release/adopt.go b/internal/release/adopt.go| index 09993da94..6c7d515a7 100644 |
| --- a/internal/release/adopt.go |
| +++ b/internal/release/adopt.go |
| @@ -114,6 +114,9 @@ type passConfig struct { |
| managed func(registry.Route) bool |
| failed FailedBuild |
| failedErr error |
| + // route reads one registry row now, for root's re-check under the carry |
| + // fence (publishConsumed, ruling R-CARRY-FENCE); PassFilter sets it. |
| + route func(host string) (*registry.Route, error) |
| // pub verifies promotion records (§4.1 evidence); PassFilter sets it. |
| pub ed25519.PublicKey |
| // tags is "<acct>/<app>" -> release version -> its newest semver tag, read |
| @@ -424,6 +427,7 @@ func PassFilter(ctx context.Context, reg *registry.Registry, s Store, pub ed2551 |
| opt(&cfg) |
| } |
| cfg.pub = pub |
| + cfg.route = reg.Get |
| // Settings spec §4.3: the failure book is read fresh for every pass, so a |
| // record the agent persisted since the last one counts now. |
| if cfg.settings && cfg.openFailed != nil { |
| @@ -1217,7 +1221,7 @@ func adoptHost(ctx context.Context, work hostwrite.Work, reg *registry.Registry, |
| var dec TargetDecision |
| if settings { |
| var stop bool |
| - ver, dec, stop = settingsTarget(optDir, bake, rt, ptr, cur, now, cfg, holds, ignore) |
| + ver, dec, stop = settingsTarget(ctx, optDir, bake, rt, ptr, cur, now, cfg, holds, ignore) |
| if stop { |
| return |
| } |
| @@ -2157,7 +2161,7 @@ func backfillManifestDigest(ctx context.Context, s Store, optDir, acct, app, ver |
| // the head, with bake, on the pointer's evidence, and no book, no Target and |
| // no consumed write. Only an unreadable failure book on a managed route |
| // changes that (the zero-downtime plans' refuseManaged posture). |
| -func settingsTarget(optDir string, bake *FirstSeen, rt registry.Route, ptr *ChannelPointer, cur string, now time.Time, |
| +func settingsTarget(ctx context.Context, optDir string, bake *FirstSeen, rt registry.Route, ptr *ChannelPointer, cur string, now time.Time, |
| cfg passConfig, holds map[string]time.Duration, ignore bool) (string, TargetDecision, bool) { |
| stop := func(dec TargetDecision) (string, TargetDecision, bool) { |
| cfg.proof.clear(rt.Host) |
| @@ -2190,7 +2194,7 @@ func settingsTarget(optDir string, bake *FirstSeen, rt registry.Route, ptr *Chan |
| if cfg.resolve { |
| fmt.Printf(" %s: update settings unreadable (%v): held, nothing installed\n", rt.Host, uerr) |
| in := TargetInput{Route: rt, Pointer: ptr, Consumed: consumed, Installed: cur} |
| - if err := WriteConsumed(optDir, NextConsumed(in, dec)); err != nil { |
| + if err := publishConsumed(ctx, optDir, cfg.route, NextConsumed(in, dec)); err != nil { |
| fmt.Printf(" %s: recording the update decision: %v\n", rt.Host, err) |
| } |
| } |
| @@ -2219,7 +2223,10 @@ func settingsTarget(optDir string, bake *FirstSeen, rt registry.Route, ptr *Chan |
| } |
| dec := Target(in) |
| if cfg.resolve { |
| - if err := WriteConsumed(optDir, NextConsumed(in, dec)); err != nil { |
| + // Under the carry fence (ruling R-CARRY-FENCE): a route a move has |
| + // held since this pass listed it gets no record, and nothing is |
| + // installed for it on this pass. |
| + if err := publishConsumed(ctx, optDir, cfg.route, NextConsumed(in, dec)); err != nil { |
| fmt.Printf(" %s: recording the update decision: %v; nothing installed\n", rt.Host, err) |
| return stop(dec) |
| } |
diff --git a/internal/release/consumedfence.go b/internal/release/consumedfence.go| new file mode 100644 |
| index 000000000..16068845b |
| --- /dev/null |
| +++ b/internal/release/consumedfence.go |
| @@ -0,0 +1,112 @@ |
| +package release |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "path/filepath" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/controlname" |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +// The carry fence (settings plan 4, ruling R-CARRY-FENCE; Astra round 2 |
| +// finding 2). Root publishes its request record for a host in |
| +// /opt/carlos/consumed/<host> (WriteConsumed); before a graceful move seals |
| +// the build state, the agent carries that record into it (CarryRequest). |
| +// Root's pass decides from a registry snapshot taken before the pass, so |
| +// without a fence a pass already running when a cut begins could publish a |
| +// resolution or a failed completion after the carry read the file: the cut |
| +// seals a state without it, and the destination resolves the request again. |
| +// |
| +// The two serialize on one cross-process lock per host, flock on |
| +// ConsumedLockPath. Root takes it, re-reads the route and publishes only |
| +// when the route is in no move hold (publishGate), then releases. The cut |
| +// takes it only once its move hold is effective (the row's status is a hold: |
| +// registry.IsHold), carries, then releases. So a root publication lands |
| +// before the carry and is carried, or sees the hold and is not made at all |
| +// (nothing is installed for that route on that pass). Neither side waits on |
| +// the other for more than one record write or one build-state write. |
| + |
| +// consumedFenceWait bounds a wait for the carry fence, on either side. A |
| +// var so tests can shrink it. |
| +var consumedFenceWait = 30 * time.Second |
| + |
| +// ErrMoveHeld is root's refusal to publish a request record for a route a |
| +// move holds (publishGate): nothing is written and nothing installed for it |
| +// on that pass. |
| +var ErrMoveHeld = errors.New("the route is held by a move, so root publishes no request record for it") |
| + |
| +// ConsumedLockPath is host's carry-fence lock file, beside its consumed file. |
| +// The leading dot is no host's shape, so it never reads as a consumed file. |
| +// The caller validates host (controlname.ValidHostShape). |
| +func ConsumedLockPath(optDir, host string) string { |
| + return filepath.Join(optDir, "consumed", "."+host+".lock") |
| +} |
| + |
| +// publishGate is root's re-check inside the carry fence: nil when root may |
| +// publish a request record for rt, the route as the registry holds it now. |
| +// A row a move holds (any of registry.HoldStatuses: the source moving, a |
| +// rollback's restoring or releasing, a landing) or a row gone from this box |
| +// is refused with ErrMoveHeld. The update simulation calls it too. |
| +func publishGate(rt *registry.Route) error { |
| + if rt == nil { |
| + return fmt.Errorf("%w: the row is gone from this box", ErrMoveHeld) |
| + } |
| + if registry.IsHold(rt.Status) { |
| + return fmt.Errorf("%w: its status is %q", ErrMoveHeld, rt.Status) |
| + } |
| + return nil |
| +} |
| + |
| +// beforeConsumedWrite, when set (tests only), runs inside root's fenced |
| +// publication: at "fence" before the lock is taken, and at "write" with the |
| +// lock held and the re-check passed, immediately before WriteConsumed. |
| +var beforeConsumedWrite func(host, at string) |
| + |
| +// publishConsumed is root's publication of c under the carry fence: the |
| +// lock, the route re-read and re-checked (publishGate), then WriteConsumed. |
| +// route reads the registry row now. A refusal (ErrMoveHeld), a fence that |
| +// could not be taken, or a row that does not read writes nothing. |
| +func publishConsumed(ctx context.Context, optDir string, route func(host string) (*registry.Route, error), c Consumed) error { |
| + if !controlname.ValidHostShape(c.Host) { |
| + return fmt.Errorf("consumed: invalid host %q", c.Host) |
| + } |
| + if hook := beforeConsumedWrite; hook != nil { |
| + hook(c.Host, "fence") |
| + } |
| + unlock, err := lockConsumedPublish(ctx, optDir, c.Host) |
| + if err != nil { |
| + return fmt.Errorf("the carry fence: %w", err) |
| + } |
| + defer unlock() |
| + if route == nil { |
| + return errors.New("no registry to re-check the route under the carry fence") |
| + } |
| + rt, err := route(c.Host) |
| + if err != nil { |
| + return fmt.Errorf("re-read the route under the carry fence: %w", err) |
| + } |
| + if err := publishGate(rt); err != nil { |
| + return err |
| + } |
| + if hook := beforeConsumedWrite; hook != nil { |
| + hook(c.Host, "write") |
| + } |
| + return WriteConsumed(optDir, c) |
| +} |
| + |
| +// FenceCarry is the cut's side of the carry fence for host, taken once the |
| +// move hold on its row is effective and held across the carry: unlock |
| +// releases it. A host whose lock file does not exist (root has never |
| +// published under the fence for it, or the consumed directory does not |
| +// exist) is carried without the lock: any later publication takes the lock |
| +// first and then sees the hold. A lock file that does not open, or a fence |
| +// not taken within its bound, is an error, and the cut does not carry. |
| +func FenceCarry(ctx context.Context, optDir, host string) (unlock func(), err error) { |
| + if !controlname.ValidHostShape(host) { |
| + return nil, fmt.Errorf("consumed: invalid host %q", host) |
| + } |
| + return lockConsumedCarry(ctx, optDir, host) |
| +} |
diff --git a/internal/release/consumedfence_linux.go b/internal/release/consumedfence_linux.go| new file mode 100644 |
| index 000000000..06daec246 |
| --- /dev/null |
| +++ b/internal/release/consumedfence_linux.go |
| @@ -0,0 +1,75 @@ |
| +package release |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "path/filepath" |
| + "time" |
| + |
| + "golang.org/x/sys/unix" |
| +) |
| + |
| +// fencePoll is how often a wait for the carry fence asks again. |
| +const fencePoll = 20 * time.Millisecond |
| + |
| +// lockConsumedPublish is root's side of the carry fence for host: the lock |
| +// file is created if it is missing (root owns the consumed directory), and |
| +// held exclusively until unlock. |
| +func lockConsumedPublish(ctx context.Context, optDir, host string) (func(), error) { |
| + if err := os.MkdirAll(filepath.Join(optDir, "consumed"), 0o755); err != nil { |
| + return nil, err |
| + } |
| + f, err := os.OpenFile(ConsumedLockPath(optDir, host), os.O_RDWR|os.O_CREATE, 0o644) |
| + if err != nil { |
| + return nil, err |
| + } |
| + return flockWait(ctx, f) |
| +} |
| + |
| +// lockConsumedCarry is the cut's side for host. The agent may only be able |
| +// to read the consumed directory; flock takes an exclusive lock on a |
| +// read-only descriptor all the same. A lock file that does not exist is no |
| +// lock to take (FenceCarry). |
| +func lockConsumedCarry(ctx context.Context, optDir, host string) (func(), error) { |
| + f, err := os.Open(ConsumedLockPath(optDir, host)) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return func() {}, nil |
| + } |
| + if err != nil { |
| + return nil, err |
| + } |
| + return flockWait(ctx, f) |
| +} |
| + |
| +// flockWait takes an exclusive flock on f, asking again every fencePoll |
| +// until ctx ends or consumedFenceWait passes. f is closed on failure, and by |
| +// the unlock it returns. |
| +func flockWait(ctx context.Context, f *os.File) (func(), error) { |
| + fd := int(f.Fd()) |
| + deadline := time.Now().Add(consumedFenceWait) |
| + for { |
| + err := unix.Flock(fd, unix.LOCK_EX|unix.LOCK_NB) |
| + if err == nil { |
| + return func() { |
| + _ = unix.Flock(fd, unix.LOCK_UN) |
| + _ = f.Close() |
| + }, nil |
| + } |
| + if !errors.Is(err, unix.EWOULDBLOCK) && !errors.Is(err, unix.EINTR) { |
| + _ = f.Close() |
| + return nil, fmt.Errorf("lock %s: %w", f.Name(), err) |
| + } |
| + if !time.Now().Before(deadline) { |
| + _ = f.Close() |
| + return nil, fmt.Errorf("lock %s: still held after %s", f.Name(), consumedFenceWait) |
| + } |
| + select { |
| + case <-ctx.Done(): |
| + _ = f.Close() |
| + return nil, fmt.Errorf("lock %s: %w", f.Name(), ctx.Err()) |
| + case <-time.After(fencePoll): |
| + } |
| + } |
| +} |
diff --git a/internal/release/consumedfence_linux_test.go b/internal/release/consumedfence_linux_test.go| new file mode 100644 |
| index 000000000..2a755f2c8 |
| --- /dev/null |
| +++ b/internal/release/consumedfence_linux_test.go |
| @@ -0,0 +1,269 @@ |
| +package release |
| + |
| +import ( |
| + "context" |
| + "crypto/ed25519" |
| + "errors" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +// Settings plan 4, Astra round 2 finding 2 (ruling R-CARRY-FENCE): a root |
| +// pass already running when a graceful cut begins publishes its request |
| +// record before the cut's carry, and is carried, or not at all. A late |
| +// record is never left behind on the source box. |
| +// |
| +// The root pass is the real one (Pass with the update settings, root's |
| +// resolve). It is paused by the test hook at one of the two points of its |
| +// fenced publication. The cut's half is the agent's carry as |
| +// buildStates.carryRequest runs it: the row held moving first, then the |
| +// fence (FenceCarry), the consumed file read and merged (CarryRequest), the |
| +// fence released. |
| + |
| +const fenceHost = "a.example.com" |
| + |
| +// fenceCutResult is what the cut sealed: the build state after its carry. |
| +type fenceCutResult struct { |
| + state BuildState |
| + err error |
| +} |
| + |
| +// fenceCut runs the cut's half: marked is closed once the row is held. |
| +func fenceCut(reg *registry.Registry, opt string, state BuildState, marked chan<- struct{}) fenceCutResult { |
| + if err := reg.SetStatus(fenceHost, "moving"); err != nil { |
| + return fenceCutResult{err: err} |
| + } |
| + close(marked) |
| + unlock, err := FenceCarry(context.Background(), opt, fenceHost) |
| + if err != nil { |
| + return fenceCutResult{err: err} |
| + } |
| + defer unlock() |
| + c, err := ReadConsumed(opt, fenceHost) |
| + if err != nil { |
| + return fenceCutResult{err: err} |
| + } |
| + return fenceCutResult{state: CarryRequest(&state, c)} |
| +} |
| + |
| +// fenceScenario is one root pass over one route: setup prepares the store, |
| +// registry and consumed file and returns the build state the cut would |
| +// seal and the pass options. |
| +type fenceScenario struct { |
| + name string |
| + setup func(t *testing.T, s Store, priv ed25519.PrivateKey, opt string, reg *registry.Registry) (BuildState, []PassOption) |
| + // published reports whether the consumed file holds the pass's record. |
| + published func(c *Consumed) bool |
| +} |
| + |
| +func fenceScenarios() []fenceScenario { |
| + ctx := context.Background() |
| + return []fenceScenario{ |
| + { |
| + // Update now, resolved by the pass to the head before staging. |
| + name: "a resolution", |
| + setup: func(t *testing.T, s Store, priv ed25519.PrivateKey, opt string, reg *registry.Registry) (BuildState, []PassOption) { |
| + shipRelease(t, s, "bab", "km", "v1", adoptPlatform, []byte("v1")) |
| + if err := Promote(ctx, s, priv, "bab", "km", "canary/x", "v1"); err != nil { |
| + t.Fatal(err) |
| + } |
| + st := BuildState{LastExecuted: "v0"} |
| + settingsRoute(t, reg, fenceHost, "km", "canary/x", UpdateSettings{Updates: UpdatesManual, Channel: "canary/x", InstallRequest: 1}, |
| + BuildStateCopy{Known: true, Replica: true, State: &st}) |
| + return st, nil |
| + }, |
| + published: func(c *Consumed) bool { return c != nil && c.Resolved != nil && c.Resolved.ID == 1 }, |
| + }, |
| + { |
| + // R1 resolved to v2, which the failure book holds: the pass |
| + // records R1 completed, failed. |
| + name: "a failed completion", |
| + setup: func(t *testing.T, s Store, priv ed25519.PrivateKey, opt string, reg *registry.Registry) (BuildState, []PassOption) { |
| + for _, v := range []string{"v1", "v2"} { |
| + shipRelease(t, s, "bab", "km", v, adoptPlatform, []byte(v)) |
| + if err := Promote(ctx, s, priv, "bab", "km", "canary/x", v); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + st := BuildState{LastExecuted: "v1", Resolved: &ResolvedRequest{ID: 1, Target: "v2"}} |
| + settingsRoute(t, reg, fenceHost, "km", "canary/x", UpdateSettings{Updates: UpdatesManual, Channel: "canary/x", InstallRequest: 1}, |
| + BuildStateCopy{Known: true, Replica: true, State: &st}) |
| + if err := WriteConsumed(opt, Consumed{Host: fenceHost, Resolved: &ResolvedRequest{ID: 1, Target: "v2"}}); err != nil { |
| + t.Fatal(err) |
| + } |
| + book := func() (FailedBuild, error) { |
| + return func(_, _, build, _ string, _ bool) bool { return build == "v2" }, nil |
| + } |
| + return st, []PassOption{WithUpdateSettings(book, func(registry.Route) bool { return false }, true)} |
| + }, |
| + published: func(c *Consumed) bool { return c != nil && c.Completed == 1 && c.CompletedFailed }, |
| + }, |
| + } |
| +} |
| + |
| +func withFenceHook(t *testing.T, hook func(host, at string)) { |
| + t.Helper() |
| + old := beforeConsumedWrite |
| + beforeConsumedWrite = hook |
| + t.Cleanup(func() { beforeConsumedWrite = old }) |
| +} |
| + |
| +// leftBehind reports whether the source's consumed file holds request state |
| +// the sealed build state lacks: carrying it again would change the state. |
| +func leftBehind(t *testing.T, opt string, sealed BuildState) bool { |
| + t.Helper() |
| + c, err := ReadConsumed(opt, fenceHost) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + again := CarryRequest(&sealed, c) |
| + return !SameRequestState(&sealed, &again) |
| +} |
| + |
| +// A cut that holds the row and carries while root's pass is paused before |
| +// its fence: root then takes the fence, sees the hold, publishes nothing and |
| +// installs nothing. The record is never published, so nothing is left behind. |
| +func TestARootPassPausedBeforeTheFenceDoesNotPublishOverACut(t *testing.T) { |
| + for _, sc := range fenceScenarios() { |
| + t.Run(sc.name, func(t *testing.T) { |
| + s := NewDirStore(t.TempDir()) |
| + priv, pub := testKeys(t) |
| + opt, reg := t.TempDir(), testRegistry(t) |
| + state, opts := sc.setup(t, s, priv, opt, reg) |
| + before, _ := ReadConsumed(opt, fenceHost) |
| + var cut fenceCutResult |
| + withFenceHook(t, func(host, at string) { |
| + if at == "fence" { |
| + cut = fenceCut(reg, opt, state, make(chan struct{})) |
| + } |
| + }) |
| + passWithSettings(t, reg, s, pub, opt, opts...) |
| + if cut.err != nil { |
| + t.Fatalf("the cut: %v", cut.err) |
| + } |
| + after, _ := ReadConsumed(opt, fenceHost) |
| + if sc.published(after) && !sc.published(before) { |
| + t.Fatalf("root published %+v after the cut held the row and carried", after) |
| + } |
| + if leftBehind(t, opt, cut.state) { |
| + t.Fatalf("a record was left behind: consumed %+v, sealed %+v", after, cut.state) |
| + } |
| + if v, _ := LiveVersion(opt, fenceHost); v != "" { |
| + t.Fatalf("live %q: root installed for a route the cut holds", v) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// A root pass paused immediately before WriteConsumed, its fence held and |
| +// its re-check passed, while a cut holds the row and asks for the fence: |
| +// the cut waits, root's record lands, and the cut carries it. |
| +func TestARootPassPausedBeforeItsWriteIsCarriedByTheCut(t *testing.T) { |
| + for _, sc := range fenceScenarios() { |
| + t.Run(sc.name, func(t *testing.T) { |
| + s := NewDirStore(t.TempDir()) |
| + priv, pub := testKeys(t) |
| + opt, reg := t.TempDir(), testRegistry(t) |
| + state, opts := sc.setup(t, s, priv, opt, reg) |
| + done := make(chan fenceCutResult, 1) |
| + early := false |
| + withFenceHook(t, func(host, at string) { |
| + if at != "write" { |
| + return |
| + } |
| + marked := make(chan struct{}) |
| + go func() { done <- fenceCut(reg, opt, state, marked) }() |
| + <-marked |
| + select { |
| + case <-done: |
| + early = true |
| + case <-time.After(200 * time.Millisecond): |
| + } |
| + }) |
| + passWithSettings(t, reg, s, pub, opt, opts...) |
| + if early { |
| + t.Fatal("the cut carried while root held the fence") |
| + } |
| + var cut fenceCutResult |
| + select { |
| + case cut = <-done: |
| + case <-time.After(10 * time.Second): |
| + t.Fatal("the cut never took the fence") |
| + } |
| + if cut.err != nil { |
| + t.Fatalf("the cut: %v", cut.err) |
| + } |
| + after, _ := ReadConsumed(opt, fenceHost) |
| + if !sc.published(after) { |
| + t.Fatalf("consumed %+v: root's record did not land", after) |
| + } |
| + if leftBehind(t, opt, cut.state) { |
| + t.Fatalf("a record was left behind: consumed %+v, sealed %+v", after, cut.state) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// With no move in progress the fenced pass is the pass it was: the hook runs |
| +// at both points and the record lands as without it. |
| +func TestARootPassWithNoMoveIsUnchangedByTheFence(t *testing.T) { |
| + for _, sc := range fenceScenarios() { |
| + t.Run(sc.name, func(t *testing.T) { |
| + var want *Consumed |
| + var wantLive string |
| + for _, hooked := range []bool{false, true} { |
| + s := NewDirStore(t.TempDir()) |
| + priv, pub := testKeys(t) |
| + opt, reg := t.TempDir(), testRegistry(t) |
| + _, opts := sc.setup(t, s, priv, opt, reg) |
| + var seen []string |
| + if hooked { |
| + withFenceHook(t, func(_, at string) { seen = append(seen, at) }) |
| + } |
| + passWithSettings(t, reg, s, pub, opt, opts...) |
| + got, err := ReadConsumed(opt, fenceHost) |
| + if err != nil || !sc.published(got) { |
| + t.Fatalf("hooked %v: consumed %+v, %v; want the pass's record", hooked, got, err) |
| + } |
| + live, _ := LiveVersion(opt, fenceHost) |
| + if !hooked { |
| + want, wantLive = got, live |
| + continue |
| + } |
| + if len(seen) != 2 || seen[0] != "fence" || seen[1] != "write" { |
| + t.Fatalf("hook points %v, want fence then write", seen) |
| + } |
| + g, w := *got, *want |
| + g.At, w.At = time.Time{}, time.Time{} |
| + if !sameConsumedForTest(g, w) || live != wantLive { |
| + t.Fatalf("fenced pass: consumed %+v live %q; unfenced: %+v live %q", g, live, w, wantLive) |
| + } |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func sameConsumedForTest(a, b Consumed) bool { |
| + ra, rb := a.Resolved, b.Resolved |
| + a.Resolved, b.Resolved = nil, nil |
| + return a == b && ((ra == nil && rb == nil) || (ra != nil && rb != nil && *ra == *rb)) |
| +} |
| + |
| +// publishGate: a held or missing row is refused; a ready one is not. |
| +func TestPublishGateRefusesAMoveHold(t *testing.T) { |
| + for _, st := range []string{"moving", "landing", "restoring", "releasing"} { |
| + if err := publishGate(®istry.Route{Host: fenceHost, Status: st}); !errors.Is(err, ErrMoveHeld) { |
| + t.Errorf("status %q: %v; want ErrMoveHeld", st, err) |
| + } |
| + } |
| + if err := publishGate(nil); !errors.Is(err, ErrMoveHeld) { |
| + t.Errorf("no row: %v; want ErrMoveHeld", err) |
| + } |
| + for _, st := range []string{"", "ready"} { |
| + if err := publishGate(®istry.Route{Host: fenceHost, Status: st}); err != nil { |
| + t.Errorf("status %q: %v; want nil", st, err) |
| + } |
| + } |
| +} |
diff --git a/internal/release/consumedfence_other.go b/internal/release/consumedfence_other.go| new file mode 100644 |
| index 000000000..07fcc453e |
| --- /dev/null |
| +++ b/internal/release/consumedfence_other.go |
| @@ -0,0 +1,14 @@ |
| +//go:build !linux |
| + |
| +package release |
| + |
| +import "context" |
| + |
| +// Boxes run Linux; other builds (a developer's darwin, the windows cross |
| +// compile) have no carry-fence lock. Root's re-check of the move hold |
| +// (publishGate) still runs; only the serialization with a cut is absent, and |
| +// no cut runs off Linux. |
| + |
| +func lockConsumedPublish(context.Context, string, string) (func(), error) { return func() {}, nil } |
| + |
| +func lockConsumedCarry(context.Context, string, string) (func(), error) { return func() {}, nil } |
diff --git a/internal/release/heldwake.go b/internal/release/heldwake.go| index 58dd25342..d11fb4ade 100644 |
| --- a/internal/release/heldwake.go |
| +++ b/internal/release/heldwake.go |
| @@ -132,12 +132,28 @@ func DecideHeldWake(in HeldWakeInput) HeldWakeDecision { |
| // root reports failed is left to the planner's recovery (ruling |
| // R-HELD-FAILED), as an unknown target is to the wake's own refusal. nil |
| // when the build may run. |
| -func HeldExecStale(d HeldWakeDecision, st *BuildState, build string) error { |
| +// |
| +// inPlaceOf is the build the zero-downtime planner chose build in place of, |
| +// for this exec only ("" when it chose nothing: ruling R-PLANNED-RECOVERY): |
| +// a planned fallback after B failed, the good build booted untried, or a |
| +// recovery boot onto the good build because the failure book blocks the |
| +// installed build. When the state read under the lease holds exactly that |
| +// build as the target, the planner's reason for leaving it is the reason the |
| +// held target is left (settings spec §6: the scheduler, swap, failure |
| +// records and fallback act on the settings' build), so build runs, before |
| +// root's consumed file names the target failed. A selection made in place of |
| +// any other build admits nothing: the good build is never exempt in general, |
| +// since it is often the stale build a wake decided on (Astra round 1 |
| +// finding 1). |
| +func HeldExecStale(d HeldWakeDecision, st *BuildState, build, inPlaceOf string) error { |
| if !d.Held || d.Unknown || d.Unread || d.Failed || d.Target == "" || d.Target == build { |
| return nil |
| } |
| if st != nil && st.LastExecuted == build { |
| return nil |
| } |
| + if inPlaceOf != "" && inPlaceOf == d.Target { |
| + return nil |
| + } |
| return fmt.Errorf("the build state names %s as this instance's build, not %s: the wake decided on a stale copy", d.Target, build) |
| } |
diff --git a/internal/release/heldwake_test.go b/internal/release/heldwake_test.go| new file mode 100644 |
| index 000000000..cd2a3181f |
| --- /dev/null |
| +++ b/internal/release/heldwake_test.go |
| @@ -0,0 +1,33 @@ |
| +package release |
| + |
| +import "testing" |
| + |
| +// Ruling R-PLANNED-RECOVERY (Astra round 2 finding 1): the pre-exec |
| +// revalidation admits the planner's selection of a build in place of exactly |
| +// the held target, and nothing wider. |
| +func TestHeldExecStaleAdmitsOnlyThePlannersTransition(t *testing.T) { |
| + heldT := HeldWakeDecision{Target: "T", Held: true} |
| + st := &BuildState{LastExecuted: "T"} |
| + cases := []struct { |
| + name string |
| + d HeldWakeDecision |
| + st *BuildState |
| + build string |
| + inPlaceOf string |
| + admitted bool |
| + }{ |
| + {"the target itself", heldT, st, "T", "", true}, |
| + {"the good build, no selection", heldT, st, "A", "", false}, |
| + {"the good build, chosen in place of T", heldT, st, "A", "T", true}, |
| + {"the good build, chosen in place of another build", heldT, st, "A", "B", false}, |
| + {"last_executed restarts", HeldWakeDecision{Target: "R", Held: true}, &BuildState{LastExecuted: "A"}, "A", "", true}, |
| + {"a pending request's T, the fallback in place of it", HeldWakeDecision{Target: "T", Held: true}, &BuildState{LastExecuted: "A0"}, "A", "T", true}, |
| + {"not held", HeldWakeDecision{}, st, "A", "", true}, |
| + } |
| + for _, c := range cases { |
| + err := HeldExecStale(c.d, c.st, c.build, c.inPlaceOf) |
| + if (err == nil) != c.admitted { |
| + t.Errorf("%s: HeldExecStale = %v; admitted want %v", c.name, err, c.admitted) |
| + } |
| + } |
| +} |
diff --git a/internal/release/updatesim_test.go b/internal/release/updatesim_test.go| index 9006b425e..8ff4ee8d3 100644 |
| --- a/internal/release/updatesim_test.go |
| +++ b/internal/release/updatesim_test.go |
| @@ -1150,7 +1150,7 @@ func (w *simWorld) wake(b *simBox) string { |
| // lease, then the merge. A refused record runs nothing. |
| fresh := DecideHeldWake(HeldWakeInput{Settings: col, Build: BuildStateCopy{Known: true, Replica: true, State: cloneSimState(w.state)}, |
| Consumed: cloneSimConsumed(b.consumed), Installed: b.live}) |
| - if err := HeldExecStale(fresh, w.state, run); err != nil { |
| + if err := HeldExecStale(fresh, w.state, run, ""); err != nil { // the model's wake has no planner |
| return desc + ": record refused: " + err.Error() |
| } |
| w.checkWakeRun(b, run) |