amadan

carlos / platform Public

host: repair an interrupted files-gen refresh; the committed wake judged on its launch (Plan 3 Task 7 fix round 2)

1. After the admission and before the start check, inside the writer's
   fence: when the published marker names the admitted generation's cold
   predecessor G, the local .files-gen names an older generation of G's
   ancestry, and the files on disk are the landing's, .files-gen moves
   to G. Anything else is left to the unchanged start check.
2. The committed wake's identity is computed from the launch that will
   run (its environment's report under the probe's normalisation, and its
   release) and compared with the pins by Select, Materialize and the
   owner's InputPolicy; a changed launch environment is refused though the
   record's pins are unchanged.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b007f4e9c32373aea7cd037f251f685448152ea9 parent d56170d
2 files changed, +238 −16
  • internal/host/movesprobeowners.go +107 −10
  • internal/host/movesprobeowners_committed_test.go +131 −6
diff --git a/internal/host/movesprobeowners.go b/internal/host/movesprobeowners.go
index 66afde813..7718d9350 100644
--- a/internal/host/movesprobeowners.go
+++ b/internal/host/movesprobeowners.go
@@ -9,6 +9,7 @@ import (
"errors"
"fmt"
"io/fs"
+ "log"
"os"
"path/filepath"
"reflect"
@@ -733,6 +734,9 @@ func (m *movesProbeOwners) Materialize(ctx context.Context, o hostclaim.Observed
if in.Prepared {
m.record(o.Host, h.Session, receipt)
}
+ if err := m.repairFilesGen(ctx, o, v, work, in.Identity); err != nil {
+ return err
+ }
return m.startCheck(ctx, o, in)
}
@@ -1187,13 +1191,14 @@ func (m *movesProbeOwners) selectCommitted(ctx context.Context, rt registry.Rout
if !filepath.IsAbs(l.Executable) || !claimHexDigest(l.ExecutableSHA256) || l.Version == "" {
return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
}
- // The committed wake runs the move's pinned inputs (InputPolicy): the
- // live release must be the pinned one.
+ // The committed wake runs the move's pinned inputs (InputPolicy), judged
+ // on the launch itself: its environment's report and its release.
pins := v.rec.Move.Inputs
- if l.Version != pins.Release {
- return ClaimLaunchInputs{}, fmt.Errorf("%w: the live release is not the move's pinned one", hostbox.ErrChanged)
+ id := committedLaunchIdentity(m.EnvKey, l.Env, l.Version)
+ if pins.EnvReport == "" || id != pinnedLaunchIdentity(pins) {
+ return ClaimLaunchInputs{}, fmt.Errorf("%w: the launch is not the move's pinned inputs", hostbox.ErrChanged)
}
- return ClaimLaunchInputs{Roles: []string{"main"}, Identity: launchIdentity(pins), Baseline: plan, Admission: m.committedAdmission(o, v, gen, plan),
+ return ClaimLaunchInputs{Roles: []string{"main"}, Identity: id, Baseline: plan, Admission: m.committedAdmission(o, v, gen, plan),
Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: withoutProbeVars(l.Env),
DBPath: rt.DBPath, Version: l.Version, Limit: l.Limit}, nil
}
@@ -1217,7 +1222,9 @@ func (m *movesProbeOwners) committedEvidence(ctx context.Context, o hostclaim.Ob
}
func (m *movesProbeOwners) materializeCommitted(ctx context.Context, o hostclaim.Observed, v probeLanding, in ClaimLaunchInputs, work hostwrite.Work) error {
- if in.Identity != launchIdentity(v.rec.Move.Inputs) {
+ // The identity is the launch that will run (in.Env, in.Version), and it
+ // is the pins'.
+ if in.Identity != committedLaunchIdentity(m.EnvKey, in.Env, in.Version) || in.Identity != pinnedLaunchIdentity(v.rec.Move.Inputs) {
return hostbox.ErrChanged
}
if err := writerFence(work, in.Identity)(ctx); err != nil {
@@ -1234,6 +1241,9 @@ func (m *movesProbeOwners) materializeCommitted(ctx context.Context, o hostclaim
if err := committedRoute(*rt, o.Host, v.op); err != nil {
return err
}
+ if err := m.repairFilesGen(ctx, o, v, work, in.Identity); err != nil {
+ return err
+ }
return m.startCheck(ctx, o, in)
}
@@ -1241,9 +1251,10 @@ func (m *movesProbeOwners) materializeCommitted(ctx context.Context, o hostclaim
// 10 rev 5's ClaimOwnerDispatch routes a move kind's input check to (the
// shape of ClaimOwnerBindings.InputPolicy; a move kind never falls back to
// serving's inputs). It answers the committed wake's spawn (Plan 3 Task 7):
-// at landing/committed the launch identity must be the move's pins',
-// HeldConfigIdentity(Move.Inputs) and Move.Inputs.Release, over a committed
-// record of this move, as MovesClaimPolicy binds the landing's rows. The
+// at landing/committed the launch identity, computed by the owner from the
+// launch that will run (committedLaunchIdentity: its environment's report
+// digest and its release), must be the move's pins' (Move.Inputs.EnvReport
+// and Move.Inputs.Release), over a committed record of this move. The
// other move-kind inputs are MovesClaimPolicy's own rows; anything else is
// refused here.
func (m *movesProbeOwners) InputPolicy(ctx context.Context, c *hostclaim.Claim, s hostbox.Selection, id hostwrite.Identity) error {
@@ -1262,8 +1273,94 @@ func (m *movesProbeOwners) InputPolicy(ctx context.Context, c *hostclaim.Claim,
rec.Move.State != provision.MoveCommitted || rec.Move.Inputs == nil || rec.Move.Inputs.Release == "" {
return hostbox.ErrChanged
}
- if id != launchIdentity(rec.Move.Inputs) {
+ if rec.Move.Inputs.EnvReport == "" || id != pinnedLaunchIdentity(rec.Move.Inputs) {
return fmt.Errorf("%w: the committed launch is not the move's pinned inputs", hostbox.ErrChanged)
}
return nil
}
+
+// committedEnvPrefix marks a committed launch's config identity: the report
+// digest of its environment.
+const committedEnvPrefix = "launch-env:"
+
+// committedLaunchIdentity is the committed wake's identity, computed from
+// the launch that will run (fix round 2, item 2): its environment's report
+// digest under the probe's normalisation contract (envDigest: CARLOS_MOVE_PROBE
+// excluded) and its release. The input policy compares it with the pins.
+func committedLaunchIdentity(key []byte, env []string, version string) hostwrite.Identity {
+ return hostwrite.Identity{Config: committedEnvPrefix + envDigest(key, env), Release: version}
+}
+
+// pinnedLaunchIdentity is the identity a launch matching the move's pins
+// carries.
+func pinnedLaunchIdentity(in *provision.MoveInputs) hostwrite.Identity {
+ return hostwrite.Identity{Config: committedEnvPrefix + in.EnvReport, Release: in.Release}
+}
+
+// repairFilesGen repairs a publication interrupted between its marker write
+// and its .files-gen refresh (fix round 2, item 1). After the admission and
+// before the start check, inside the admitted writer's fence: when the
+// location's published marker names G, G is the admitted generation's cold
+// predecessor (Task 4's rule), the local .files-gen names an older generation
+// of G's own ancestry, and the files on disk are proved the landing's (the
+// check publication runs), .files-gen is moved to G. Anything else is left
+// alone, for the unchanged start check to decide.
+func (m *movesProbeOwners) repairFilesGen(ctx context.Context, o hostclaim.Observed, v probeLanding, work hostwrite.Work, id hostwrite.Identity) error {
+ h := o.Claim.Holder
+ marker, err := m.readMarker(ctx, h)
+ if err != nil || marker == nil {
+ return err
+ }
+ g := marker.Op
+ if g == "" || marker.Incarnation != h.Incarnation || marker.Host != o.Host {
+ return nil
+ }
+ if pred, ok := coldPredecessor(h, h.Generation); !ok || pred != g {
+ return nil
+ }
+ dir, _ := m.FilesDir(o.Host)
+ files, _, err := m.localMarkers(o.Host, filepath.Join(dir, ".none"))
+ if err != nil || files == nil || files.Op == g || files.Incarnation != h.Incarnation || !ancestorOf(h, g, files.Op) {
+ return nil
+ }
+ body, err := json.Marshal(moves.FilesGen{Op: g, Incarnation: h.Incarnation})
+ if err != nil {
+ return err
+ }
+ return work.With(ctx, id, func() error {
+ if m.verifyFiles(o.Host, v) != nil {
+ return nil // not the landing's files: the start check decides
+ }
+ if err := syncPath(dir); err != nil {
+ return err
+ }
+ log.Printf("moves: %s: .files-gen repaired from %s to the published %s", o.Host, files.Op, g)
+ return writeFileAtomic(filepath.Join(dir, archive.MarkerName), body, 0o644)
+ })
+}
+
+// coldPredecessor is the predecessor of the newest cold admission into g.
+func coldPredecessor(h *hostclaim.Holder, g string) (string, bool) {
+ for i := len(h.Admissions) - 1; i >= 0; i-- {
+ if a := h.Admissions[i]; a.Generation == g && a.Mode == "cold" {
+ return a.Predecessor, true
+ }
+ }
+ return "", false
+}
+
+// ancestorOf reports whether old is reached by walking g's cold
+// predecessors in the holder's admissions.
+func ancestorOf(h *hostclaim.Holder, g, old string) bool {
+ for steps := 0; steps <= len(h.Admissions); steps++ {
+ pred, ok := coldPredecessor(h, g)
+ if !ok || pred == "" {
+ return false
+ }
+ if pred == old {
+ return true
+ }
+ g = pred
+ }
+ return false
+}
diff --git a/internal/host/movesprobeowners_committed_test.go b/internal/host/movesprobeowners_committed_test.go
index c605085e6..1ad06cf46 100644
--- a/internal/host/movesprobeowners_committed_test.go
+++ b/internal/host/movesprobeowners_committed_test.go
@@ -63,10 +63,11 @@ func TestMovesProbeOwnersCommittedWake(t *testing.T) {
if in.Prepared || in.Baseline == nil || in.Baseline.Generation != poOp || in.Baseline.MaxTXID != 42 || a.RemoteMaxTXID != 42 {
t.Fatalf("the committed baseline: %+v", in.Baseline)
}
- // Bound to the move's pins (platform-1e, claim Plan 10 rev 5): the
- // committed wake launches the inputs the move pinned.
- if in.Identity != launchIdentity(f.rec.Move.Inputs) {
- t.Fatalf("the committed wake's identity %+v, want the pins' %+v", in.Identity, launchIdentity(f.rec.Move.Inputs))
+ // Computed from the actual launch (fix round 2): its environment's
+ // report under the probe's normalisation, and its release; here they
+ // are the pins'.
+ if want := committedPinsIdentity(f.rec.Move.Inputs); in.Identity != want {
+ t.Fatalf("the committed wake's identity %+v, want %+v", in.Identity, want)
}
for _, kv := range in.Env {
if strings.HasPrefix(kv, "CARLOS_MOVE_PROBE=") {
@@ -260,8 +261,8 @@ func TestMovesProbeOwnersInputPolicyBindsTheCommittedWake(t *testing.T) {
t.Fatalf("matching pins through the policy: %v", err)
}
for name, mutate := range map[string]func(){
- "another release": func() { f.rec.Move.Inputs.Release = "v9" },
- "another overlay": func() { f.rec.Move.Inputs.OverlayETag = "e-other" },
+ "another release": func() { f.rec.Move.Inputs.Release = "v9" },
+ "another env report": func() { f.rec.Move.Inputs.EnvReport = "sha256:other" },
} {
pins := *f.rec.Move.Inputs
mutate()
@@ -279,3 +280,127 @@ func TestMovesProbeOwnersInputPolicyBindsTheCommittedWake(t *testing.T) {
_ = genselect.Marker{}
_ = hostwrite.Identity{}
}
+
+// writeFilesGenOnly writes .files-gen alone.
+func (f *poFixture) writeFilesGenOnly(op string) {
+ f.t.Helper()
+ body, _ := json.Marshal(moves.FilesGen{Op: op, Incarnation: f.inc})
+ if err := os.WriteFile(filepath.Join(f.files, archive.MarkerName), body, 0o600); err != nil {
+ f.t.Fatal(err)
+ }
+}
+
+// committedWake admits, restores and materializes one committed wake.
+func (f *poFixture) committedWake(o hostclaim.Observed) (hostclaim.Observed, ClaimLaunchInputs, error) {
+ f.t.Helper()
+ in, err := f.owners.Select(f.t.Context(), f.route, o, poWake)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ admitted := f.admit(o, in)
+ f.restore(admitted.Claim.Holder.Generation)
+ return admitted, in, f.owners.Materialize(f.t.Context(), admitted, in, newFenceWork())
+}
+
+// Task 7 fix round 2, item 1: a publication interrupted between its marker
+// write and the .files-gen refresh (local g1, published g2) is repaired
+// before the next admission's (g3) start check: the published marker names
+// g3's cold predecessor, and the files on disk are proved the landing's, so
+// .files-gen moves to g2 and the start check, unweakened, passes.
+func TestMovesProbeOwnersRepairAnInterruptedFilesGenRefresh(t *testing.T) {
+ f, restarted, again := restartedProbePublished(t)
+ if err := f.publishFresh(restarted, again); err != nil {
+ t.Fatal(err)
+ }
+ g1 := restarted.Claim.Holder.Generation
+ o := f.commit()
+ w1, in1, err := f.committedWake(o)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := f.publishFresh(w1, in1); err != nil {
+ t.Fatal(err)
+ }
+ g2 := w1.Claim.Holder.Generation
+ // The crash: the marker names g2, .files-gen still g1.
+ f.writeFilesGenOnly(g1)
+ w2, in2, err := f.committedWake(f.read())
+ if err != nil {
+ t.Fatalf("the next admission after an interrupted refresh: %v", err)
+ }
+ if fg := f.filesGen(); fg.Op != g2 {
+ t.Fatalf(".files-gen was not repaired to %s: %+v", g2, fg)
+ }
+ if err := f.publishFresh(w2, in2); err != nil {
+ t.Fatal(err)
+ }
+ if fg := f.filesGen(); fg.Op != w2.Claim.Holder.Generation {
+ t.Fatalf("after the next publication: %+v", fg)
+ }
+}
+
+// Item 1: files that are not the landing's are never repaired over; the
+// start check refuses as before.
+func TestMovesProbeOwnersRepairRefusesMismatchedFiles(t *testing.T) {
+ f, restarted, again := restartedProbePublished(t)
+ if err := f.publishFresh(restarted, again); err != nil {
+ t.Fatal(err)
+ }
+ g1 := restarted.Claim.Holder.Generation
+ o := f.commit()
+ w1, in1, err := f.committedWake(o)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := f.publishFresh(w1, in1); err != nil {
+ t.Fatal(err)
+ }
+ old := f.files + ".old"
+ if err := os.Rename(f.files, old); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Mkdir(f.files, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(old) })
+ f.writeFilesGenOnly(g1)
+ if _, _, err := f.committedWake(f.read()); !errors.Is(err, moves.ErrStartFiles) {
+ t.Fatalf("a repair over files the landing did not place: %v", err)
+ }
+ if fg := f.filesGen(); fg.Op != g1 {
+ t.Fatalf(".files-gen was rewritten: %+v", fg)
+ }
+}
+
+// Item 2: the committed wake's input check judges the real launch. A changed
+// launch environment with the record's pins unchanged is refused, by Select
+// and by the input policy on the identity that launch would carry; matching
+// inputs are accepted.
+func TestMovesProbeOwnersCommittedWakeJudgesTheRealLaunch(t *testing.T) {
+ f, o := committedAfterProbe(t)
+ ctx := t.Context()
+ in, err := f.owners.Select(ctx, f.route, o, poWake)
+ if err != nil {
+ t.Fatalf("matching inputs: %v", err)
+ }
+ c := o.Claim
+ sel := hostbox.Selection{Host: poHost, Action: "spawn", Box: "uk-box", Region: "uk", Captured: hostguard.Captured{Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp},
+ Kind: c.Op.Kind, Op: c.Op.ID, Phase: c.Op.Phase, Generation: c.Holder.Generation, Session: c.Holder.Session}
+ if err := f.owners.InputPolicy(ctx, c, sel, in.Identity); err != nil {
+ t.Fatalf("matching inputs through the input policy: %v", err)
+ }
+ f.launch.Env = []string{"A=2"}
+ if _, err := f.owners.Select(ctx, f.route, o, poWake); err == nil {
+ t.Fatal("a changed launch environment was selected")
+ }
+ changed := committedLaunchIdentity(poEnvKey, f.launch.Env, f.launch.Version)
+ if err := f.owners.InputPolicy(ctx, c, sel, changed); err == nil {
+ t.Fatal("the changed launch's identity was accepted")
+ }
+}
+
+// committedPinsIdentity is the identity a committed launch matching the pins
+// carries.
+func committedPinsIdentity(p *provision.MoveInputs) hostwrite.Identity {
+ return hostwrite.Identity{Config: committedEnvPrefix + p.EnvReport, Release: p.Release}
+}