amadan

carlos / platform Public

host: committed recovery after a restarted probe; the owner's input policy (Plan 3 Task 7 fix round 1, finding 1; claim Plan 10 rev 5)

A fresh generation's fenced publication first proves the files on disk
are the landing's (the install receipt's identities), then refreshes
.files-gen to the published generation under the admitted writer's
fence; a mismatch fails the publication and refreshes nothing. Task 4's
start check is unchanged, and a committed wake after a restarted probe,
and repeated wakes, now pass it. The owner's InputPolicy, the entry point
the claim's dispatcher routes move kinds to, binds the committed wake's
spawn to Move.Inputs; the committed wake launches the pinned inputs.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude d56170d0701ef1cfee42b2386523ac59ad2fd64f parent 078a6c3
4 files changed, +338 −15
  • internal/host/movesprobeowners.go +123 −8
  • internal/host/movesprobeowners_committed_test.go +186 −2
  • internal/host/movesprobeowners_fix1_test.go +9 −5
  • internal/host/movesprobeowners_test.go +20 −0
diff --git a/internal/host/movesprobeowners.go b/internal/host/movesprobeowners.go
index a51391589..66afde813 100644
--- a/internal/host/movesprobeowners.go
+++ b/internal/host/movesprobeowners.go
@@ -725,12 +725,7 @@ func (m *movesProbeOwners) Materialize(ctx context.Context, o hostclaim.Observed
if envDigest(m.EnvKey, in.Env) != pins.EnvReport {
return probeInputsChanged()
}
- m.mu.Lock()
- if m.fences == nil {
- m.fences = map[string]hostwrite.Work{}
- }
- m.fences[claimPublishedKey(o.Host, h.Session)] = work
- m.mu.Unlock()
+ m.recordFence(o.Host, h.Session, work)
receipt, err := m.preparedMarker(ctx, o.Host, v, fence)
if err != nil {
return err
@@ -934,6 +929,18 @@ func (m *movesProbeOwners) PublishFenced(ctx context.Context, o hostclaim.Observ
if in.Prepared || h.Generation == v.op {
return ClaimPublicationReceipt{}, hostbox.ErrChanged
}
+ // The admitted writer's spawn selection, recorded by Materialize: the
+ // fence the local files marker is refreshed under (fix round 1).
+ m.mu.Lock()
+ work := m.fences[claimPublishedKey(o.Host, h.Session)]
+ m.mu.Unlock()
+ if work == nil {
+ return ClaimPublicationReceipt{}, fmt.Errorf("%w: no admitted writer's fence for session %d", hostbox.ErrChanged, h.Session)
+ }
+ // The files on disk must be the landing's before anything is published.
+ if err := work.With(ctx, in.Identity, func() error { return m.verifyFiles(o.Host, v) }); err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
if err := pub.Sync(ctx); err != nil {
return ClaimPublicationReceipt{}, err
}
@@ -958,6 +965,9 @@ func (m *movesProbeOwners) PublishFenced(ctx context.Context, o hostclaim.Observ
return ClaimPublicationReceipt{}, hostbox.ErrChanged
}
receipt := ClaimPublicationReceipt{MarkerVersion: current.ID, Root: root}
+ if err := m.refreshFilesGen(ctx, o.Host, v, work, in.Identity); err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
m.record(o.Host, h.Session, receipt)
return receipt, nil
}
@@ -980,10 +990,74 @@ func (m *movesProbeOwners) PublishFenced(ctx context.Context, o hostclaim.Observ
return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
}
receipt := ClaimPublicationReceipt{MarkerVersion: version, Root: root}
+ if err := m.refreshFilesGen(ctx, o.Host, v, work, in.Identity); err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
m.record(o.Host, h.Session, receipt)
return receipt, nil
}
+// recordFence keeps the admitted spawn selection Materialize ran under, by
+// host and session: the writer's fence for that session's marker writes and
+// its files marker refresh.
+func (m *movesProbeOwners) recordFence(host string, session uint64, work hostwrite.Work) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ if m.fences == nil {
+ m.fences = map[string]hostwrite.Work{}
+ }
+ m.fences[claimPublishedKey(host, session)] = work
+}
+
+// verifyFiles requires the files directory and <host>.env on disk to be the
+// ones the landing placed: the install receipt's identities
+// (landStepFilesPlaced). Anything else fails the publication.
+func (m *movesProbeOwners) verifyFiles(host string, v probeLanding) error {
+ j, err := moves.ReadJournal(m.DataDir, host, v.op)
+ if err != nil {
+ return err
+ }
+ placed, err := journalledStaged(j, landStepFilesPlaced)
+ if err != nil {
+ return fmt.Errorf("%w: %v", hostbox.ErrChanged, err)
+ }
+ dir, env := m.FilesDir(host)
+ if ok, err := sameIdentity(dir, placed.Files); err != nil || !ok {
+ return fmt.Errorf("%w: %s is not the files directory the landing placed (%v)", hostbox.ErrChanged, dir, err)
+ }
+ if placed.Env != "" {
+ if ok, err := sameIdentity(env, placed.Env); err != nil || !ok {
+ return fmt.Errorf("%w: %s is not the environment file the landing placed (%v)", hostbox.ErrChanged, env, err)
+ }
+ }
+ return nil
+}
+
+// refreshFilesGen points .files-gen at the published generation (fix round
+// 1, finding 1), inside the admitted writer's fence, after the files on disk
+// are proved the landing's again: the files are the same bytes the move
+// installed, and the generation that now owns them is the one just
+// published. Task 4's start check is unchanged: the next admission's cold
+// predecessor is this generation. <dbPath>.gen needs no refresh: every cold
+// restore rewrites it to the admitted generation (H11).
+func (m *movesProbeOwners) refreshFilesGen(ctx context.Context, host string, v probeLanding, work hostwrite.Work, id hostwrite.Identity) error {
+ h := v.holder
+ marker, err := json.Marshal(moves.FilesGen{Op: h.Generation, Incarnation: h.Incarnation})
+ if err != nil {
+ return err
+ }
+ dir, _ := m.FilesDir(host)
+ return work.With(ctx, id, func() error {
+ if err := m.verifyFiles(host, v); err != nil {
+ return err
+ }
+ if err := syncPath(dir); err != nil {
+ return err
+ }
+ return writeFileAtomic(filepath.Join(dir, archive.MarkerName), marker, 0o644)
+ })
+}
+
func (m *movesProbeOwners) record(host string, session uint64, receipt ClaimPublicationReceipt) {
m.mu.Lock()
defer m.mu.Unlock()
@@ -1110,10 +1184,16 @@ func (m *movesProbeOwners) selectCommitted(ctx context.Context, rt registry.Rout
if err != nil {
return ClaimLaunchInputs{}, err
}
- if l.Identity.Config == "" || l.Identity.Release == "" || !filepath.IsAbs(l.Executable) || !claimHexDigest(l.ExecutableSHA256) || l.Version == "" {
+ if !filepath.IsAbs(l.Executable) || !claimHexDigest(l.ExecutableSHA256) || l.Version == "" {
return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
}
- return ClaimLaunchInputs{Roles: []string{"main"}, Identity: l.Identity, Baseline: plan, Admission: m.committedAdmission(o, v, gen, plan),
+ // The committed wake runs the move's pinned inputs (InputPolicy): the
+ // live release must be the pinned one.
+ 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)
+ }
+ return ClaimLaunchInputs{Roles: []string{"main"}, Identity: launchIdentity(pins), 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
}
@@ -1137,9 +1217,13 @@ 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) {
+ return hostbox.ErrChanged
+ }
if err := writerFence(work, in.Identity)(ctx); err != nil {
return err
}
+ m.recordFence(o.Host, v.holder.Session, work)
rt, err := m.Route(o.Host)
if err != nil {
return err
@@ -1152,3 +1236,34 @@ func (m *movesProbeOwners) materializeCommitted(ctx context.Context, o hostclaim
}
return m.startCheck(ctx, o, in)
}
+
+// InputPolicy is the moves owner's input policy: the entry point claim Plan
+// 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
+// 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 {
+ if m == nil || m.Records == nil || ctx == nil {
+ return hostwrite.ErrUnavailable
+ }
+ if c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseLandingCommitted ||
+ s.Phase != hostclaim.PhaseLandingCommitted || s.Action != movesActionSpawn || s.Op != c.Op.ID || s.Host != c.Host {
+ return hostbox.ErrChanged
+ }
+ rec, err := m.Records(ctx, c.Holder.Account, c.Holder.App, c.Host)
+ if err != nil {
+ return err
+ }
+ if rec == nil || rec.Host != c.Host || rec.Move == nil || rec.Move.ID != c.Op.ID || rec.Move.Incarnation != c.Holder.Incarnation ||
+ rec.Move.State != provision.MoveCommitted || rec.Move.Inputs == nil || rec.Move.Inputs.Release == "" {
+ return hostbox.ErrChanged
+ }
+ if id != launchIdentity(rec.Move.Inputs) {
+ return fmt.Errorf("%w: the committed launch is not the move's pinned inputs", hostbox.ErrChanged)
+ }
+ return nil
+}
diff --git a/internal/host/movesprobeowners_committed_test.go b/internal/host/movesprobeowners_committed_test.go
index d9d1fbb2d..c605085e6 100644
--- a/internal/host/movesprobeowners_committed_test.go
+++ b/internal/host/movesprobeowners_committed_test.go
@@ -2,14 +2,22 @@ package host
import (
"context"
+ "encoding/json"
"errors"
+ "os"
+ "path/filepath"
"strings"
"testing"
"time"
"github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/genselect"
"github.com/carlosframework/platform/internal/hostbox"
"github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
"github.com/carlosframework/platform/internal/provision"
"github.com/superfly/ltx"
)
@@ -55,8 +63,10 @@ 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)
}
- if in.Identity != f.launch.Identity {
- t.Fatalf("the committed wake's identity %+v, want the launch's %+v", in.Identity, f.launch.Identity)
+ // 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))
}
for _, kv := range in.Env {
if strings.HasPrefix(kv, "CARLOS_MOVE_PROBE=") {
@@ -95,3 +105,177 @@ func TestMovesProbeOwnersCommittedWake(t *testing.T) {
t.Fatal("a committed wake over a landing record")
}
}
+
+// filesGen reads the local .files-gen.
+func (f *poFixture) filesGen() moves.FilesGen {
+ f.t.Helper()
+ raw, err := os.ReadFile(filepath.Join(f.files, archive.MarkerName))
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ var fg moves.FilesGen
+ if err := json.Unmarshal(raw, &fg); err != nil {
+ f.t.Fatal(err)
+ }
+ return fg
+}
+
+// restore is ColdRestore's local effect on a fresh admission: <dbPath>.gen
+// names the admitted generation.
+func (f *poFixture) restore(gen string) {
+ f.t.Helper()
+ if err := hostbox.WriteGenerationMarker(f.route.DBPath+".gen", gen); err != nil {
+ f.t.Fatal(err)
+ }
+}
+
+// publishFresh puts the fresh generation's first root and publishes it
+// fenced, as the claim's sequence does once the session's root is durable.
+func (f *poFixture) publishFresh(admitted hostclaim.Observed, in ClaimLaunchInputs) error {
+ f.t.Helper()
+ gen := admitted.Claim.Holder.Generation
+ f.bucket.Put("data", "instances/"+poHost+"/g-"+gen+"/0009/"+ltx.FormatFilename(1, 44), f.ltxHeader(1, 44))
+ pub := ClaimPublication{Sync: func(context.Context) error { return nil }, Check: func(context.Context) error { return nil }}
+ _, err := f.owners.PublishFenced(f.t.Context(), admitted, in, pub)
+ return err
+}
+
+// restartedProbePublished runs a dead first attempt and a restarted attempt
+// that passes and publishes its fresh generation.
+func restartedProbePublished(t *testing.T) (*poFixture, hostclaim.Observed, ClaimLaunchInputs) {
+ t.Helper()
+ f := newPOFixture(t)
+ ctx := t.Context()
+ first, in := f.selectAndAdmit()
+ if err := f.owners.Materialize(ctx, first, in, newFenceWork()); err != nil {
+ t.Fatal(err)
+ }
+ again, err := f.owners.Select(ctx, f.route, first, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ restarted := f.admit(first, again)
+ f.restore(restarted.Claim.Holder.Generation)
+ if err := f.owners.Materialize(ctx, restarted, again, newFenceWork()); err != nil {
+ t.Fatalf("the restarted probe's start: %v", err)
+ }
+ return f, restarted, again
+}
+
+// commit moves the fixture to the commit: the claim at landing/committed,
+// the record committed, the row open.
+func (f *poFixture) commit() hostclaim.Observed {
+ f.t.Helper()
+ c := *f.read().Claim
+ c.Op.Phase = hostclaim.PhaseLandingCommitted
+ f.putClaim(&c)
+ f.rec.Move.State, f.rec.Move.LandingPhase = provision.MoveCommitted, ""
+ f.route.Status, f.route.HoldOp = "ready", ""
+ return f.read()
+}
+
+// Task 7 fix round 1, finding 1: a restarted probe's successful publication
+// refreshes .files-gen to the published generation, under the writer's
+// fence, after the files on disk are proved the landing's. After the commit,
+// a process death and a wake, and then several more, each start passes
+// Task 4's unweakened start check.
+func TestMovesProbeOwnersRestartedProbeThenCommittedWakes(t *testing.T) {
+ f, restarted, again := restartedProbePublished(t)
+ ctx := t.Context()
+ if err := f.publishFresh(restarted, again); err != nil {
+ t.Fatal(err)
+ }
+ fresh := restarted.Claim.Holder.Generation
+ if fg := f.filesGen(); fg.Op != fresh || fg.Incarnation != f.inc {
+ t.Fatalf(".files-gen after the restarted probe's publication: %+v, want %s", fg, fresh)
+ }
+ o := f.commit()
+ for i := 0; i < 3; i++ {
+ in, err := f.owners.Select(ctx, f.route, o, poWake)
+ if err != nil {
+ t.Fatalf("wake %d: %v", i, err)
+ }
+ admitted := f.admit(o, in)
+ f.restore(admitted.Claim.Holder.Generation)
+ if err := f.owners.Materialize(ctx, admitted, in, newFenceWork()); err != nil {
+ t.Fatalf("wake %d's start after the probe's %s: %v", i, fresh, err)
+ }
+ if err := f.publishFresh(admitted, in); err != nil {
+ t.Fatalf("wake %d's publication: %v", i, err)
+ }
+ if fg := f.filesGen(); fg.Op != admitted.Claim.Holder.Generation {
+ t.Fatalf("wake %d's .files-gen: %+v", i, fg)
+ }
+ o = f.read()
+ }
+}
+
+// A files directory that is not the one the landing placed fails the
+// publication: no marker is written and .files-gen is not refreshed.
+func TestMovesProbeOwnersPublicationRefusesMismatchedFiles(t *testing.T) {
+ f, restarted, again := restartedProbePublished(t)
+ before := f.bucket.LatestID("data", f.markerKey())
+ // Another directory in its place: made while the placed one still
+ // exists, so it cannot reuse its inode.
+ 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.writeLocal(poOp, restarted.Claim.Holder.Generation, f.inc)
+ if err := f.publishFresh(restarted, again); err == nil {
+ t.Fatal("published over files the landing did not place")
+ }
+ if f.bucket.LatestID("data", f.markerKey()) != before {
+ t.Fatal("a marker was written")
+ }
+ if fg := f.filesGen(); fg.Op != poOp {
+ t.Fatalf(".files-gen was refreshed: %+v", fg)
+ }
+}
+
+// platform-1e (claim Plan 10 rev 5): the moves owner's input policy binds
+// the committed wake's spawn to Move.Inputs, as MovesClaimPolicy binds the
+// landing's rows: a committed cold admission with the pins' launch identity
+// is accepted, and refused once the pins differ.
+func TestMovesProbeOwnersInputPolicyBindsTheCommittedWake(t *testing.T) {
+ f, o := committedAfterProbe(t)
+ ctx := t.Context()
+ in, err := f.owners.Select(ctx, f.route, o, poWake)
+ if err != nil {
+ t.Fatal(err)
+ }
+ admitted := f.admit(o, in)
+ c := admitted.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}
+ policy := MovesClaimPolicyWithInputs(f.owners.Records, f.owners.InputPolicy)
+ if err := f.owners.InputPolicy(ctx, c, sel, in.Identity); err != nil {
+ t.Fatalf("matching pins: %v", err)
+ }
+ if err := policy(ctx, c, sel, in.Identity); err != nil {
+ 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" },
+ } {
+ pins := *f.rec.Move.Inputs
+ mutate()
+ if err := f.owners.InputPolicy(ctx, c, sel, in.Identity); err == nil {
+ t.Fatalf("%s: accepted", name)
+ }
+ if err := policy(ctx, c, sel, in.Identity); err == nil {
+ t.Fatalf("%s: accepted through the policy", name)
+ }
+ f.rec.Move.Inputs = &pins
+ }
+ if err := f.owners.InputPolicy(ctx, c, sel, f.launch.Identity); err == nil {
+ t.Fatal("the ordinary identity was accepted")
+ }
+ _ = genselect.Marker{}
+ _ = hostwrite.Identity{}
+}
diff --git a/internal/host/movesprobeowners_fix1_test.go b/internal/host/movesprobeowners_fix1_test.go
index e481df618..8fa715228 100644
--- a/internal/host/movesprobeowners_fix1_test.go
+++ b/internal/host/movesprobeowners_fix1_test.go
@@ -274,9 +274,9 @@ func TestMovesProbeOwnersSupersedeTheEarlierStaysMarker(t *testing.T) {
// first attempt published the move id; the restarted attempt fresh1
// published its own marker after its root and crashed. The next attempt's
// baseline is the certified remote from fresh1, whose chain reaches the move
-// id, and fresh1's marker is kept, not superseded by the prepared one. (That
-// attempt then stops at Task 4's files rule: .files-gen names the move id,
-// two cold admissions back.)
+// id, and fresh1's marker is kept, not superseded by the prepared one. Since
+// Task 7's fix round 1 fresh1's publication refreshed .files-gen to fresh1,
+// so that attempt passes Task 4's files rule too.
func TestMovesProbeOwnersRecogniseARestartAfterACrash(t *testing.T) {
f := newPOFixture(t)
ctx := t.Context()
@@ -290,6 +290,10 @@ func TestMovesProbeOwnersRecogniseARestartAfterACrash(t *testing.T) {
}
restarted := f.admit(first, again)
fresh := again.Admission.NewGeneration
+ f.restore(fresh)
+ if err := f.owners.Materialize(ctx, restarted, again, newFenceWork()); err != nil {
+ t.Fatal(err)
+ }
f.bucket.Put("data", "instances/"+poHost+"/g-"+fresh+"/0009/"+ltx.FormatFilename(1, 44), f.ltxHeader(1, 44))
pub := ClaimPublication{Sync: func(context.Context) error { return nil }, Check: func(context.Context) error { return nil }}
if _, err := f.owners.PublishFenced(ctx, restarted, again, pub); err != nil {
@@ -304,13 +308,13 @@ func TestMovesProbeOwnersRecogniseARestartAfterACrash(t *testing.T) {
t.Fatalf("the restart's baseline is not the certified remote of %s: %+v", fresh, next.Baseline)
}
admitted := f.admit(restarted, next)
- f.writeLocal(poOp, admitted.Claim.Holder.Generation, f.inc)
+ f.restore(admitted.Claim.Holder.Generation)
written := f.bucket.LatestID("data", f.markerKey())
err = f.owners.Materialize(ctx, admitted, next, newFenceWork())
if m, _ := f.currentMarker(); m.Op != fresh || f.bucket.LatestID("data", f.markerKey()) != written {
t.Fatalf("the restart's marker was superseded: %+v", m)
}
- if !errors.Is(err, moves.ErrStartFiles) {
+ if err != nil {
t.Fatalf("the second restart: %v", err)
}
}
diff --git a/internal/host/movesprobeowners_test.go b/internal/host/movesprobeowners_test.go
index 129c084f6..0cbc8bc8c 100644
--- a/internal/host/movesprobeowners_test.go
+++ b/internal/host/movesprobeowners_test.go
@@ -112,6 +112,22 @@ func newPOFixture(t *testing.T) *poFixture {
}
body, _ := json.Marshal(f.manifest)
j.Mark(landStepInstall, string(body))
+ // The install receipt the landing journals: the files directory and
+ // <host>.env, placed (moveslandinstall.go).
+ envPath := filepath.Join(f.dataDir, poHost+".env")
+ if err := os.WriteFile(envPath, []byte("CARLOS_ADMIN_TOKEN=t\n"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ filesID, err := pathIdentity(f.files)
+ if err != nil {
+ t.Fatal(err)
+ }
+ envID, err := pathIdentity(envPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ placed, _ := json.Marshal(stagedFiles{Files: filesID, Env: envID})
+ j.Mark(landStepFilesPlaced, string(placed))
if err := j.Write(f.dataDir); err != nil {
t.Fatal(err)
}
@@ -582,6 +598,10 @@ func TestMovesProbeOwnersPublishARestartedAttemptFenced(t *testing.T) {
t.Fatal(err)
}
admitted := f.admit(first, again)
+ f.restore(admitted.Claim.Holder.Generation)
+ if err := f.owners.Materialize(ctx, admitted, again, newFenceWork()); err != nil {
+ t.Fatal(err)
+ }
synced := 0
pub := ClaimPublication{Sync: func(context.Context) error { synced++; return nil }, Check: func(context.Context) error { return nil }}
if err := f.owners.Publish(ctx, admitted, again); !errors.Is(err, hostbox.ErrChanged) {