amadan

carlos / platform Public

host: probe owners fix round 1 (Plan 3 Task 5, findings 1, 2, 3, 4, 8)

1. Select and AdmissionEvidence check the landing's deadline, the record
   attempt within what T11 admitted, Landing.InstalledDBSHA256, and the
   baseline's pin, before the admission CAS.
2. The prepared marker is written under the admitted writer's fence (the
   spawn selection Materialize ran under, checked before every request
   attempt) and carries install.json's digest, verified against the
   bucket; Group D's Publish replays it and refuses without the fence.
3. The env pin is moves.ReportEnv over the selected launch's own env,
   with CARLOS_MOVE_PROBE normalised away, checked in Select and again in
   Materialize over the env that will run.
4. The render's etags are compared with the pins before any held work is
   selected, so a changed config is ProbeInputsChanged through the real
   policy.
8. Ruled C1 order: admit, publish the verified prepared marker (an
   earlier stay's is superseded by If-Match), start check, spawn. A first
   admission is decided by the writer box, not the admission history; a
   trimmed history that cannot be proved is still refused.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 5e803ac7ec35924e5d47d1fb0a99130f66716500 parent b02fd3b
3 files changed, +617 −87
  • internal/host/movesprobeowners.go +295 −72
  • internal/host/movesprobeowners_fix1_test.go +292 −0
  • internal/host/movesprobeowners_test.go +30 −15
diff --git a/internal/host/movesprobeowners.go b/internal/host/movesprobeowners.go
index 5d15da4b4..e3d31deda 100644
--- a/internal/host/movesprobeowners.go
+++ b/internal/host/movesprobeowners.go
@@ -3,6 +3,8 @@ package host
import (
"bytes"
"context"
+ "crypto/sha256"
+ "encoding/hex"
"encoding/json"
"errors"
"fmt"
@@ -41,9 +43,12 @@ import (
// replication and the publication sequence (claimwakeassembly.go). This
// owner set supplies what moves knows:
//
-// - Select (read-only, before admission): the launch-input pin at spawn
-// (§7.7, round-2 finding 14): the live release and the launch
-// environment's report are the pinned ones, else ProbeInputsChanged
+// - Select (read-only, before admission): the landing is still
+// admissible (its deadline, the record attempt within what T11
+// admitted, the install the record vouches for); the launch-input pin
+// at spawn (§7.7, round-2 finding 14): the live release and the report
+// of the selected launch's own environment are the pinned ones, else
+// ProbeInputsChanged
// with the recorded sentence "configuration changed during the move",
// which the landing turns into its rollback (Task 6); then the
// admission per §7.14. The first probe is a cold admission into the
@@ -59,18 +64,25 @@ import (
// once it has (H14). The launch identity is the pinned config and
// release (H4), CARLOS_MOVE_PROBE=1 is in the child's environment (H18)
// and the readiness path is the app's moves.ready_path (H12).
-// - AdmissionEvidence: the same decisions, re-derived and compared,
-// against a baseline selected again, immediately before the CAS.
-// - Materialize, before every start, under the admitted spawn selection:
-// the held render and the held release, each under its own claim
-// selection, held-config and held-release, through the existing held
-// paths (appconfig.RenderHeldUnder, release.MaterializeHeldUnder) and
-// MovesClaimPolicy's probing rows (H13); the rendered etags must be the
-// pins (the authoritative check); then the §5.1 start check
-// (moves.StartCheck) against the claim after the admission's CAS.
-// - Publish: the prepared generation's marker, before replication: the
-// move id with the build's pin, superseding the current marker by its
-// exact ETag or created when there is none. PublishFenced: a restarted
+// - AdmissionEvidence: the same decisions and the same admissibility,
+// re-derived and compared, pin included, against a baseline selected
+// again, immediately before the CAS.
+// - Materialize, before every start, under the admitted spawn selection,
+// in the order ruled for C1 (fix round 1, finding 8): the held render
+// and the held release, each under its own claim selection, held-config
+// and held-release, through the existing held paths
+// (appconfig.RenderHeldUnder, release.MaterializeHeldUnder) and
+// MovesClaimPolicy's probing rows (H13), the render's etags compared
+// with the pins before anything is selected or written; then the
+// verified prepared marker, published under the admitted writer's
+// fence; then the §5.1 start check (moves.StartCheck, not weakened)
+// against the claim after the admission's CAS; only then the spawn.
+// - Publish: Group D's call before replication finds the prepared marker
+// Materialize published and records it; it writes only under the fence
+// Materialize recorded for the session. The marker names the move id
+// with the build's pin and install.json's digest, superseding the
+// current marker (an earlier stay's, which Plan 2 keeps) by its exact
+// ETag, or created when there is none. PublishFenced: a restarted
// attempt's fresh generation, after its own first root, as serving
// publishes it. PublicationEvidence: this process's receipt for the
// admitted session, read back.
@@ -93,9 +105,20 @@ type movesProbeOwners struct {
// Launch is the held row's launch: the live executable, its digest,
// arguments, environment and version (the claim's launch snapshot).
Launch func(context.Context, registry.Route) (ClaimServingLaunch, error)
- // EnvReport is moves.ReportEnv's digest of the held row's launch
- // environment, the value Inputs.EnvReport pinned at readiness.
- EnvReport func(context.Context, registry.Route) (string, error)
+ // EnvKey is the env report's HMAC key (moves.ReportKeyFromSecrets). The
+ // pin is checked against moves.ReportEnv(EnvKey, env) of the selected
+ // launch's own environment, so no report can be decoupled from what
+ // runs (fix round 1, finding 3). Normalisation contract: env is the
+ // launch's K=V list, last value wins (childEnv's dedupe), with the
+ // probe-only variables (probeOnlyVars: CARLOS_MOVE_PROBE) removed,
+ // since the source never ran with them and the probe adds
+ // CARLOS_MOVE_PROBE=1 after the pin. The production Launch must
+ // therefore be the env the readiness pin was reported over
+ // (ExecBackend.LaunchEnvironment's composition; claim Plan 10's launch
+ // accessor).
+ EnvKey []byte
+ // Admitted is the number of probe attempts T11 admitted (H10).
+ Admitted func(context.Context, *provision.Record) (int, error)
// Works selects a held action's claim work at the admitted capture
// (hostbox.WorkSource.SelectAt).
Works func(ctx context.Context, host, action string, captured hostguard.Captured) (hostwrite.Work, error)
@@ -106,6 +129,9 @@ type movesProbeOwners struct {
mu sync.Mutex
published map[string]ClaimPublicationReceipt
+ // fences are the admitted spawn selections Materialize ran under, by
+ // host and session: the live fence for the session's marker writes.
+ fences map[string]hostwrite.Work
}
var (
@@ -141,7 +167,7 @@ func (m *movesProbeOwners) MovesOwners(ops OperationAttempts) ClaimMovesOwners {
func (m *movesProbeOwners) valid() bool {
return m != nil && m.Claims != nil && m.Records != nil && m.ReadyPath != nil && m.Data != nil && m.Bucket != "" && m.Region != "" && m.Box != "" &&
- m.DataDir != "" && m.FilesDir != nil && m.Markers != nil && m.Route != nil && m.Launch != nil && m.EnvReport != nil && m.Works != nil &&
+ m.DataDir != "" && m.FilesDir != nil && m.Markers != nil && m.Route != nil && m.Launch != nil && len(m.EnvKey) != 0 && m.Admitted != nil && m.Works != nil &&
m.Render != nil && m.Release != nil && m.NewID != nil && m.Now != nil
}
@@ -215,14 +241,85 @@ func heldRoute(rt registry.Route, host, op string) error {
return nil
}
-// opAdmitted reports whether the holder already admitted a writer into the
-// move id: a restarted attempt then admits a fresh generation.
-func opAdmitted(h *hostclaim.Holder, op string) bool {
- if h.Generation == op {
- return true
+// firstAdmission reports whether the holder's writer is still the move's
+// source: the flip leaves the source's session in place, and every
+// admission on this box makes it the writer. A first admission is the
+// prepared one; any later one is a restarted attempt into a fresh
+// generation. It reads no admission history, so trimming cannot change it.
+func firstAdmission(h *hostclaim.Holder, box string) bool {
+ return h.WriterBox != box
+}
+
+// admissible is the landing's own word that an admission may happen now
+// (fix round 1, finding 1): before the landing's deadline (the earlier of
+// cut_deadline and landing_deadline, H8), at a record attempt within what
+// T11 admitted (H10), over the install the record vouches for.
+func (m *movesProbeOwners) admissible(ctx context.Context, v probeLanding) error {
+ mv := v.rec.Move
+ d, _ := landingDeadlineOf(mv)
+ if d == nil || !m.Now().Before(*d) {
+ return fmt.Errorf("%w: the landing is past its deadline", hostbox.ErrChanged)
+ }
+ admitted, err := m.Admitted(ctx, v.rec)
+ if err != nil {
+ return err
+ }
+ if mv.Landing == nil || mv.Landing.Attempts < 1 || mv.Landing.Attempts > moves.AttemptsFor(moves.ProbeLanding, admitted) {
+ return fmt.Errorf("%w: the record's probe attempt is not one T11 admitted", hostbox.ErrChanged)
+ }
+ if mv.Landing.InstalledDBSHA256 != "sha256:"+v.manifest.DBSHA256 {
+ return fmt.Errorf("%w: the record does not vouch for this install", hostbox.ErrChanged)
+ }
+ return nil
+}
+
+// probeOnlyVars are the variables the probe adds to the launch and the
+// readiness pin never saw.
+var probeOnlyVars = map[string]bool{"CARLOS_MOVE_PROBE": true}
+
+// envDigest is the env report digest of a launch environment, normalised
+// as EnvKey's comment states.
+func envDigest(key []byte, env []string) string {
+ vars := map[string]string{}
+ for _, kv := range env {
+ k, v, ok := strings.Cut(kv, "=")
+ if !ok || probeOnlyVars[k] {
+ continue
+ }
+ vars[k] = v
+ }
+ return moves.ReportEnv(key, vars).Digest
+}
+
+// manifestDigest is install.json's digest: the build writes the manifest
+// as json.Marshal of the journalled value (moves.BuildGeneration).
+func manifestDigest(m moves.InstallManifest) (string, error) {
+ raw, err := json.Marshal(m)
+ if err != nil {
+ return "", err
+ }
+ sum := sha256.Sum256(raw)
+ return hex.EncodeToString(sum[:]), nil
+}
+
+func manifestPin(m moves.InstallManifest) genselect.Pin {
+ return genselect.Pin{SnapshotKey: m.SnapshotKey, SnapshotVersion: m.SnapshotVersion, SnapshotTXID: m.N, SnapshotCRC64: m.SnapshotCRC64}
+}
+
+// landingMarker reports whether the location's current marker already
+// names this landing: the move id's verified prepared marker, or the fresh
+// generation of a later admission on this box under it. Anything else (an
+// earlier stay's, another incarnation's) is superseded by the prepared one.
+func (m *movesProbeOwners) landingMarker(marker *genselect.Marker, host string, v probeLanding, digest string) bool {
+ h := v.holder
+ if marker == nil || marker.Incarnation != h.Incarnation || marker.Host != host || marker.Box != m.Box {
+ return false
+ }
+ if marker.Op == v.op {
+ return marker.Pin == manifestPin(v.manifest) && marker.ManifestSHA256 == digest
}
for _, a := range h.Admissions {
- if a.Generation == op {
+ if a.Generation == marker.Op && a.Box == m.Box && a.Region == m.Region && a.Mode == "cold" {
return true
}
}
@@ -243,6 +340,7 @@ func (m *movesProbeOwners) request(host string, h *hostclaim.Holder) genselect.R
}
func preparedPin(v probeLanding) genselect.PreparedPin {
+ // The build's own pin, from the journalled manifest.
return genselect.PreparedPin{Generation: v.op, SnapshotKey: v.manifest.SnapshotKey, SnapshotVersion: v.manifest.SnapshotVersion, SnapshotTXID: v.manifest.N, SnapshotCRC64: v.manifest.SnapshotCRC64}
}
@@ -271,17 +369,22 @@ func (m *movesProbeOwners) currentMarker(ctx context.Context, h *hostclaim.Holde
// baseline is the admission's baseline (H14): the build's pin for the
// prepared generation; for a restarted attempt, the prepared pin while the
-// location has no marker (the move id published nothing beyond the build)
-// and the certified remote once it has one.
+// location's marker is not this landing's (the move id published nothing
+// beyond the build; an earlier stay's pin is never a baseline), and the
+// certified remote once it is.
func (m *movesProbeOwners) baseline(ctx context.Context, host string, v probeLanding, prepared bool) (*genselect.Plan, error) {
if prepared {
return genselect.SelectPrepared(ctx, m.Data, m.request(host, v.holder), preparedPin(v))
}
- current, err := m.currentMarker(ctx, v.holder)
+ digest, err := manifestDigest(v.manifest)
+ if err != nil {
+ return nil, err
+ }
+ marker, err := m.readMarker(ctx, v.holder)
if err != nil {
return nil, err
}
- if current == nil {
+ if !m.landingMarker(marker, host, v, digest) {
return genselect.SelectPrepared(ctx, m.Data, m.request(host, v.holder), preparedPin(v))
}
return genselect.Select(ctx, m.Data, m.request(host, v.holder))
@@ -350,8 +453,11 @@ func (m *movesProbeOwners) Select(ctx context.Context, rt registry.Route, o host
if err := heldRoute(rt, o.Host, v.op); err != nil {
return ClaimLaunchInputs{}, err
}
+ if err := m.admissible(ctx, v); err != nil {
+ return ClaimLaunchInputs{}, err
+ }
h := v.holder
- prepared := !opAdmitted(h, v.op)
+ prepared := firstAdmission(h, m.Box)
admission := hostclaim.Admission{ObservedOpID: v.op, ObservedPhase: o.Claim.Op.Phase, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: m.Box, Mode: "cold",
KeepStamp: h.Stamp > h.Session, Region: m.Region, PredecessorRegion: predecessorRegion(h, v.rec.Move.From)}
if prepared {
@@ -368,7 +474,7 @@ func (m *movesProbeOwners) Select(ctx context.Context, rt registry.Route, o host
}
} else {
admission.NewGeneration = m.NewID()
- if !claimFreshGeneration(h, admission.NewGeneration) {
+ if admission.NewGeneration == v.op || !claimFreshGeneration(h, admission.NewGeneration) {
return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
}
}
@@ -393,11 +499,7 @@ func (m *movesProbeOwners) Select(ctx context.Context, rt registry.Route, o host
if l.Version != pins.Release {
return ClaimLaunchInputs{}, probeInputsChanged()
}
- report, err := m.EnvReport(ctx, rt)
- if err != nil {
- return ClaimLaunchInputs{}, err
- }
- if report == "" || report != pins.EnvReport {
+ if pins.EnvReport == "" || envDigest(m.EnvKey, l.Env) != pins.EnvReport {
return ClaimLaunchInputs{}, probeInputsChanged()
}
return ClaimLaunchInputs{Roles: []string{"main"}, Identity: launchIdentity(pins), Baseline: plan, Admission: admission,
@@ -412,20 +514,25 @@ func (m *movesProbeOwners) AdmissionEvidence(ctx context.Context, o hostclaim.Ob
if err != nil {
return err
}
+ if err := m.admissible(ctx, v); err != nil {
+ return err
+ }
h := v.holder
if in.ObservedOpID != v.op || in.ObservedPhase != o.Claim.Op.Phase || in.ObservedSession != h.Session || in.ObservedGeneration != h.Generation || in.Box != m.Box || in.Region != m.Region ||
in.Mode != "cold" || in.PredecessorCertified || in.LossID != "" || len(in.AdditionalWritesLost) != 0 || in.KeepStamp != (h.Stamp > h.Session) ||
in.PredecessorRegion != predecessorRegion(h, v.rec.Move.From) || baseline == nil || in.RemoteMaxTXID != baseline.MaxTXID {
return hostbox.ErrChanged
}
- prepared := !opAdmitted(h, v.op)
+ prepared := firstAdmission(h, m.Box)
switch {
case prepared:
coverage := ""
if v.seal.FromDB {
coverage = "sealed"
}
- if in.NewGeneration != v.op || in.Coverage != coverage {
+ pin := manifestPin(v.manifest)
+ if in.NewGeneration != v.op || in.Coverage != coverage || baseline.Generation != v.op || baseline.Pin == nil || *baseline.Pin != pin ||
+ baseline.MaxTXID != v.manifest.N || v.manifest.N != v.seal.N {
return hostbox.ErrChanged
}
rt, err := m.route(o.Host, v.op)
@@ -448,7 +555,7 @@ func (m *movesProbeOwners) AdmissionEvidence(ctx context.Context, o hostclaim.Ob
if err != nil {
return err
}
- if fresh.Generation != baseline.Generation || fresh.MaxTXID != baseline.MaxTXID || !reflect.DeepEqual(fresh.Files, baseline.Files) {
+ if fresh.Generation != baseline.Generation || fresh.MaxTXID != baseline.MaxTXID || !reflect.DeepEqual(fresh.Files, baseline.Files) || !reflect.DeepEqual(fresh.Pin, baseline.Pin) {
return hostbox.ErrChanged
}
return nil
@@ -491,8 +598,12 @@ func (w heldIdentityWork) With(ctx context.Context, _ hostwrite.Identity, fn fun
return w.Work.With(ctx, w.ID, fn)
}
-// Materialize runs before every start of the admitted probe: the held
-// render and release under their own selections, then the start check.
+// Materialize runs before every start of the admitted probe, in the ruled
+// order: the held render (its etags compared with the pins before anything
+// is selected or written) and the held release under their own selections,
+// the launch environment's pin again over the exact env that will run, the
+// verified prepared marker under the admitted writer's fence, then the
+// start check. The spawn follows only if all of it passed.
func (m *movesProbeOwners) Materialize(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
v, err := m.landing(ctx, o)
if err != nil {
@@ -508,8 +619,9 @@ func (m *movesProbeOwners) Materialize(ctx context.Context, o hostclaim.Observed
if in.Identity != launchIdentity(pins) {
return hostbox.ErrChanged
}
+ fence := writerFence(work, in.Identity)
// The admitted spawn selection still holds under the launch identity.
- if err := work.With(ctx, in.Identity, func() error { return nil }); err != nil {
+ if err := fence(ctx); err != nil {
return err
}
rt, err := m.route(o.Host, v.op)
@@ -522,6 +634,12 @@ func (m *movesProbeOwners) Materialize(ctx context.Context, o hostclaim.Observed
h := v.holder
captured := hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}
hr, err := m.Render(ctx, rt, v.op, func(pre appconfig.HeldRender) (hostwrite.Work, error) {
+ // The pin decides before any selection or write (fix round 1,
+ // finding 4): a changed configuration is the landing's rollback
+ // sentence, never the policy's refusal of the write.
+ if !sameBundle(pre.BundleETags, pins.BundleETags) || pre.OverlayETag != pins.OverlayETag {
+ return nil, probeInputsChanged()
+ }
w, err := m.Works(ctx, o.Host, movesActionHeldConfig, captured)
if err != nil {
return nil, err
@@ -541,9 +659,123 @@ func (m *movesProbeOwners) Materialize(ctx context.Context, o hostclaim.Observed
if err := m.Release(ctx, rt, pins.Release, heldIdentityWork{Work: released, ID: hostwrite.Identity{Release: pins.Release}}); err != nil {
return err
}
+ 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()
+ receipt, err := m.preparedMarker(ctx, o.Host, v, fence)
+ if err != nil {
+ return err
+ }
+ if in.Prepared {
+ m.record(o.Host, h.Session, receipt)
+ }
return m.startCheck(ctx, o, in)
}
+// writerFence is the admitted writer's live fence over its spawn selection:
+// the attempt's lease, the session, stamp and generation, and the policy,
+// checked afresh on every call (claimLeaseWork over hostbox's selection).
+func writerFence(work hostwrite.Work, id hostwrite.Identity) func(context.Context) error {
+ return func(ctx context.Context) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ return work.With(ctx, id, func() error { return nil })
+ }
+}
+
+// preparedMarker establishes the verified prepared marker under fence: the
+// install.json the build wrote is the journalled manifest, and the location's
+// generation.json names this landing (left as it is) or is superseded by the
+// move id's marker with the build's pin and that manifest's digest, by
+// If-Match on the current version or If-None-Match when there is none. The
+// store runs fence before every request attempt. It returns the receipt for
+// the move id's marker, or the zero receipt when a later landing generation's
+// marker is current.
+func (m *movesProbeOwners) preparedMarker(ctx context.Context, host string, v probeLanding, fence func(context.Context) error) (ClaimPublicationReceipt, error) {
+ h := v.holder
+ digest, err := manifestDigest(v.manifest)
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
+ defer cancel()
+ if err := m.verifyInstall(bounded, h, v, digest); err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ pin := manifestPin(v.manifest)
+ root := genselect.File{Key: v.manifest.SnapshotKey, VersionID: v.manifest.SnapshotVersion, Level: 9, MinTXID: 1, MaxTXID: v.manifest.N}
+ current, err := m.currentMarker(bounded, h)
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ if current != nil {
+ marker, err := m.readMarker(bounded, h)
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ if m.landingMarker(marker, host, v, digest) {
+ if marker.Op != v.op {
+ return ClaimPublicationReceipt{}, nil
+ }
+ return ClaimPublicationReceipt{MarkerVersion: current.ID, Root: root}, nil
+ }
+ }
+ body, err := json.Marshal(genselect.Marker{Op: v.op, Incarnation: h.Incarnation, Host: host, Box: m.Box, InstalledAt: m.Now().UTC(), ManifestSHA256: digest, Pin: pin})
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ expected := ""
+ if current != nil {
+ expected = current.ETag
+ }
+ if err := fence(bounded); err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ version, err := m.Markers.PutMarker(bounded, m.Bucket, markerKey(h), body, expected, fence)
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ if version == "" {
+ return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
+ }
+ return ClaimPublicationReceipt{MarkerVersion: version, Root: root}, nil
+}
+
+// verifyInstall requires the generation's current install.json to be the
+// journalled manifest, byte for byte (its digest is the marker's).
+func (m *movesProbeOwners) verifyInstall(ctx context.Context, h *hostclaim.Holder, v probeLanding, digest string) error {
+ key := moves.InstallManifestKey(moves.GenerationPrefix(h.Location, v.op))
+ versions, err := m.Data.ListVersions(ctx, m.Bucket, key)
+ if err != nil {
+ return err
+ }
+ id := ""
+ for _, ver := range versions {
+ if ver.Key == key && ver.IsLatest && !ver.DeleteMarker {
+ id = ver.VersionID
+ }
+ }
+ if id == "" {
+ return fmt.Errorf("%w: the generation has no install.json", hostbox.ErrChanged)
+ }
+ raw, err := m.Data.GetVersion(ctx, m.Bucket, key, id)
+ if err != nil {
+ return err
+ }
+ sum := sha256.Sum256(raw)
+ if hex.EncodeToString(sum[:]) != digest {
+ return fmt.Errorf("%w: install.json is not the journalled manifest", hostbox.ErrChanged)
+ }
+ return nil
+}
+
// startCheck is §5.1's check against the claim after the admission's CAS,
// read fresh: generation.json from this region's location, .files-gen and
// <dbPath>.gen from this box.
@@ -588,9 +820,10 @@ func (m *movesProbeOwners) EmptyAuthority(context.Context, hostclaim.Observed, C
return hostbox.ErrUnsupported
}
-// Publish writes the prepared generation's marker before replication: the
-// move id with the build's pin. A marker this box already wrote for the
-// move id is a replay and records the receipt again.
+// Publish is Group D's call for a prepared admission, before replication.
+// Materialize has already published the verified prepared marker; this
+// finds it and records the receipt. It writes only under the fence
+// Materialize recorded for the admitted session, and refuses without one.
func (m *movesProbeOwners) Publish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
v, err := m.landing(ctx, o)
if err != nil {
@@ -603,36 +836,20 @@ func (m *movesProbeOwners) Publish(ctx context.Context, o hostclaim.Observed, in
if !in.Prepared || in.Admission.NewGeneration != v.op {
return hostbox.ErrChanged
}
- pin := genselect.Pin{SnapshotKey: v.manifest.SnapshotKey, SnapshotVersion: v.manifest.SnapshotVersion, SnapshotTXID: v.manifest.N, SnapshotCRC64: v.manifest.SnapshotCRC64}
- root := genselect.File{Key: v.manifest.SnapshotKey, VersionID: v.manifest.SnapshotVersion, Level: 9, MinTXID: 1, MaxTXID: v.manifest.N}
- bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
- defer cancel()
- current, err := m.currentMarker(bounded, h)
- if err != nil {
- return err
- }
- if current != nil {
- if marker, err := m.readMarker(bounded, h); err == nil && marker != nil && marker.Op == v.op && marker.Box == m.Box && marker.Incarnation == h.Incarnation && marker.Host == o.Host && marker.Pin == pin {
- m.record(o.Host, h.Session, ClaimPublicationReceipt{MarkerVersion: current.ID, Root: root})
- return nil
- }
- }
- body, err := json.Marshal(genselect.Marker{Op: v.op, Incarnation: h.Incarnation, Host: o.Host, Box: m.Box, InstalledAt: m.Now().UTC(), Pin: pin})
- if err != nil {
- return err
- }
- expected := ""
- if current != nil {
- expected = current.ETag
+ m.mu.Lock()
+ work := m.fences[claimPublishedKey(o.Host, h.Session)]
+ m.mu.Unlock()
+ if work == nil {
+ return fmt.Errorf("%w: no admitted writer's fence for session %d", hostbox.ErrChanged, h.Session)
}
- version, err := m.Markers.PutMarker(bounded, m.Bucket, markerKey(h), body, expected, func(ctx context.Context) error { return ctx.Err() })
+ receipt, err := m.preparedMarker(ctx, o.Host, v, writerFence(work, in.Identity))
if err != nil {
return err
}
- if version == "" {
- return hostwrite.ErrUnavailable
+ if receipt.MarkerVersion == "" {
+ return hostbox.ErrChanged
}
- m.record(o.Host, h.Session, ClaimPublicationReceipt{MarkerVersion: version, Root: root})
+ m.record(o.Host, h.Session, receipt)
return nil
}
@@ -740,8 +957,14 @@ func (m *movesProbeOwners) PublicationEvidence(ctx context.Context, o hostclaim.
if plan.Generation != h.Generation || plan.Marker == nil || plan.MarkerVersion != receipt.MarkerVersion || plan.Marker.Op != h.Generation || plan.Marker.Box != m.Box || plan.Marker.Incarnation != h.Incarnation || plan.Marker.Host != o.Host {
return hostbox.ErrChanged
}
- if h.Generation == v.op && (plan.Marker.SnapshotKey != v.manifest.SnapshotKey || plan.Marker.SnapshotVersion != v.manifest.SnapshotVersion || plan.Marker.SnapshotTXID != v.manifest.N) {
- return hostbox.ErrChanged
+ if h.Generation == v.op {
+ digest, err := manifestDigest(v.manifest)
+ if err != nil {
+ return err
+ }
+ if plan.Marker.Pin != manifestPin(v.manifest) || plan.Marker.ManifestSHA256 != digest {
+ return hostbox.ErrChanged
+ }
}
versions, err := m.Data.ListVersions(ctx, m.Bucket, receipt.Root.Key)
if err != nil {
diff --git a/internal/host/movesprobeowners_fix1_test.go b/internal/host/movesprobeowners_fix1_test.go
new file mode 100644
index 000000000..012cebb04
--- /dev/null
+++ b/internal/host/movesprobeowners_fix1_test.go
@@ -0,0 +1,292 @@
+package host
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "testing"
+ "time"
+
+ "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/provision"
+)
+
+// fenceWork is an admitted spawn selection whose fence can be closed: once
+// shut, every check refuses. It counts the checks it answered.
+type fenceWork struct {
+ shut *bool
+ checks *int
+}
+
+func newFenceWork() fenceWork { return fenceWork{shut: new(bool), checks: new(int)} }
+
+func (w fenceWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
+ *w.checks++
+ if *w.shut {
+ return hostbox.ErrChanged
+ }
+ return fn()
+}
+
+func (f *poFixture) markerKey() string { return "instances/" + poHost + "/generation.json" }
+
+func (f *poFixture) currentMarker() (genselect.Marker, bool) {
+ f.t.Helper()
+ body, ok := f.bucket.Current("data", f.markerKey())
+ var m genselect.Marker
+ if ok && json.Unmarshal(body, &m) != nil {
+ f.t.Fatalf("marker %s", body)
+ }
+ return m, ok
+}
+
+// selectAndAdmit runs Select and the admission CAS for the claim as it is.
+func (f *poFixture) selectAndAdmit() (hostclaim.Observed, ClaimLaunchInputs) {
+ f.t.Helper()
+ o := f.read()
+ in, err := f.owners.Select(f.t.Context(), f.route, o, poProbe)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ return f.admit(o, in), in
+}
+
+// Finding 1: AdmissionEvidence, immediately before the admission's CAS,
+// refuses a landing past its deadline, a record attempt outside what T11
+// admitted, an install the record does not vouch for, and a baseline whose
+// pin is not the build's.
+func TestMovesProbeOwnersAdmissionEvidenceChecksTheLanding(t *testing.T) {
+ past := time.Date(2026, 10, 9, 11, 59, 0, 0, time.UTC)
+ for name, tc := range map[string]struct {
+ rec func(*provision.Record)
+ plan func(*genselect.Plan)
+ }{
+ "past the cut deadline": {rec: func(r *provision.Record) { r.Move.CutDeadline = &past }},
+ "past the landing deadline": {rec: func(r *provision.Record) { r.Move.LandingDeadline = &past }},
+ "no deadline": {rec: func(r *provision.Record) { r.Move.CutDeadline = nil }},
+ "no attempt counted": {rec: func(r *provision.Record) { r.Move.Landing.Attempts = 0 }},
+ "beyond the admitted": {rec: func(r *provision.Record) { r.Move.Landing.Attempts = 4 }},
+ "another installed db": {rec: func(r *provision.Record) {
+ r.Move.Landing.InstalledDBSHA256 = "sha256:" + hex.EncodeToString(make([]byte, 32))
+ }},
+ "no installed db": {rec: func(r *provision.Record) { r.Move.Landing.InstalledDBSHA256 = "" }},
+ "another pin version": {plan: func(p *genselect.Plan) {
+ pin := *p.Pin
+ pin.SnapshotVersion = "other"
+ p.Pin = &pin
+ }},
+ "no pin": {plan: func(p *genselect.Plan) { p.Pin = nil }},
+ } {
+ t.Run(name, func(t *testing.T) {
+ f := newPOFixture(t)
+ ctx := t.Context()
+ o := f.read()
+ in, err := f.owners.Select(ctx, f.route, o, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := f.owners.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); err != nil {
+ t.Fatalf("the unchanged landing: %v", err)
+ }
+ plan := *in.Baseline
+ if tc.rec != nil {
+ tc.rec(f.rec)
+ }
+ if tc.plan != nil {
+ tc.plan(&plan)
+ }
+ if err := f.owners.AdmissionEvidence(ctx, o, in.Admission, &plan); err == nil {
+ t.Fatal("admitted")
+ }
+ })
+ }
+}
+
+// Finding 2: the prepared marker is written only under the admitted
+// writer's live fence, checked by the store before every request attempt,
+// and carries the install manifest's digest, verified against install.json.
+func TestMovesProbeOwnersPreparedMarkerIsFenced(t *testing.T) {
+ f := newPOFixture(t)
+ ctx := t.Context()
+ admitted, in := f.selectAndAdmit()
+ // Group D's prepared Publish with no fence held for the session.
+ if err := f.owners.Publish(ctx, admitted, in); err == nil {
+ t.Fatal("an unfenced prepared marker was published")
+ }
+ if _, ok := f.currentMarker(); ok {
+ t.Fatal("a marker was written without the fence")
+ }
+ fence := newFenceWork()
+ if err := f.owners.Materialize(ctx, admitted, in, fence); err != nil {
+ t.Fatal(err)
+ }
+ m, ok := f.currentMarker()
+ install, _ := f.bucket.Current("data", moves.InstallManifestKey(moves.GenerationPrefix("instances/"+poHost, poOp)))
+ sum := sha256.Sum256(install)
+ if !ok || m.Op != poOp || m.ManifestSHA256 != hex.EncodeToString(sum[:]) || m.SnapshotTXID != 42 {
+ t.Fatalf("marker %+v", m)
+ }
+ // Group D's call now replays the marker this session wrote.
+ written := f.bucket.LatestID("data", f.markerKey())
+ if err := f.owners.Publish(ctx, admitted, in); err != nil || f.bucket.LatestID("data", f.markerKey()) != written {
+ t.Fatalf("replay: %v", err)
+ }
+ if err := f.owners.PublicationEvidence(ctx, admitted); err != nil {
+ t.Fatal(err)
+ }
+ // Another writer replaced it and the session's fence has closed: nothing
+ // is written over it.
+ f.bucket.Put("data", f.markerKey(), mustJSON(t, genselect.Marker{Op: "other", Incarnation: f.inc, Host: poHost, Box: "uk-box", InstalledAt: time.Unix(3, 0).UTC()}))
+ other := f.bucket.LatestID("data", f.markerKey())
+ *fence.shut = true
+ if err := f.owners.Publish(ctx, admitted, in); err == nil || f.bucket.LatestID("data", f.markerKey()) != other {
+ t.Fatalf("a marker written after the fence closed: %v", err)
+ }
+}
+
+// Finding 2: an install.json that is not the journalled manifest is not
+// published, and the probe does not start.
+func TestMovesProbeOwnersPreparedMarkerNeedsTheInstallManifest(t *testing.T) {
+ f := newPOFixture(t)
+ other := f.manifest
+ other.BuiltAt = time.Unix(9, 0).UTC()
+ f.bucket.Put("data", moves.InstallManifestKey(moves.GenerationPrefix("instances/"+poHost, poOp)), mustJSON(t, other))
+ admitted, in := f.selectAndAdmit()
+ if err := f.owners.Materialize(t.Context(), admitted, in, newFenceWork()); err == nil {
+ t.Fatal("published over an install.json the journal does not name")
+ }
+ if _, ok := f.currentMarker(); ok {
+ t.Fatal("a marker was written")
+ }
+}
+
+// Finding 3: the env pin is the report of the launch Select selected: a
+// launch environment that changed is refused though the pin and every other
+// input are unchanged; the probe-only CARLOS_MOVE_PROBE is not part of it.
+func TestMovesProbeOwnersEnvPinIsTheSelectedLaunch(t *testing.T) {
+ f := newPOFixture(t)
+ if _, err := f.owners.Select(t.Context(), f.route, f.read(), poProbe); err != nil {
+ t.Fatalf("the pinned launch: %v", err)
+ }
+ f.launch.Env = []string{"A=2", "CARLOS_MOVE_PROBE=0"}
+ _, err := f.owners.Select(t.Context(), f.route, f.read(), poProbe)
+ var changed *ProbeInputsChanged
+ if !errors.As(err, &changed) || changed.Reason != provision.RollbackConfigChanged {
+ t.Fatalf("a changed launch environment: %v", err)
+ }
+ f.launch.Env = []string{"A=1"}
+ if _, err := f.owners.Select(t.Context(), f.route, f.read(), poProbe); err != nil {
+ t.Fatalf("without the probe-only variable: %v", err)
+ }
+}
+
+// policyWork is a held selection that asks the real MovesClaimPolicy, as
+// hostbox.WorkSource does, before it runs anything.
+type policyWork struct {
+ f *poFixture
+ action string
+}
+
+func (w policyWork) With(ctx context.Context, id hostwrite.Identity, fn func() error) error {
+ o := w.f.read()
+ c := o.Claim
+ s := hostbox.Selection{Host: poHost, Action: w.action, 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 := MovesClaimPolicy(func(context.Context, string, string, string) (*provision.Record, error) { return w.f.rec, nil })(ctx, c, s, id); err != nil {
+ return err
+ }
+ return fn()
+}
+
+// Finding 4: a render whose etags are not the pins is ProbeInputsChanged
+// (the landing's RollbackConfigChanged), decided before any held work is
+// selected or written, through the real policy, which would otherwise
+// refuse the write with its own error.
+func TestMovesProbeOwnersChangedConfigThroughThePolicy(t *testing.T) {
+ f := newPOFixture(t)
+ ctx := t.Context()
+ f.owners.Works = func(_ context.Context, _, action string, _ hostguard.Captured) (hostwrite.Work, error) {
+ f.order = append(f.order, "select:"+action)
+ return policyWork{f: f, action: action}, nil
+ }
+ admitted, in := f.selectAndAdmit()
+ if err := f.owners.Materialize(ctx, admitted, in, newFenceWork()); err != nil {
+ t.Fatalf("the pinned render through the policy: %v", err)
+ }
+ f.order = nil
+ f.renderOverlay = "e-changed"
+ err := f.owners.Materialize(ctx, admitted, in, newFenceWork())
+ var changed *ProbeInputsChanged
+ var refusal *MovesPolicyRefusal
+ if !errors.As(err, &changed) || changed.Reason != provision.RollbackConfigChanged || errors.As(err, &refusal) {
+ t.Fatalf("a changed render: %v", err)
+ }
+ for _, step := range f.order {
+ if step == "select:held-config" || step == "select:held-release" {
+ t.Fatalf("a held work was selected for a changed render: %v", f.order)
+ }
+ }
+}
+
+// Finding 8, ruled order: admit, publish the verified prepared marker under
+// the writer's fence (superseding the earlier stay's marker by If-Match),
+// then the start check, then the spawn.
+func TestMovesProbeOwnersSupersedeTheEarlierStaysMarker(t *testing.T) {
+ f := newPOFixture(t)
+ ctx := t.Context()
+ f.bucket.Put("data", f.markerKey(), mustJSON(t, genselect.Marker{Op: "earlier", Incarnation: f.inc, Host: poHost, Box: "uk-box", InstalledAt: time.Unix(1, 0).UTC(), Pin: genselect.Pin{SnapshotTXID: 3}}))
+ admitted, in := f.selectAndAdmit()
+ if err := f.owners.Materialize(ctx, admitted, in, newFenceWork()); err != nil {
+ t.Fatalf("the first probe over an earlier stay's marker: %v", err)
+ }
+ if m, _ := f.currentMarker(); m.Op != poOp {
+ t.Fatalf("the earlier stay's marker was not superseded: %+v", m)
+ }
+}
+
+// Finding 8: a history trimmed so that the admitted generation's ancestry
+// back to the published marker cannot be proved is refused by the start
+// check, after the publication and before any spawn. StartCheck is not
+// weakened.
+func TestMovesProbeOwnersTrimmedHistoryIsRefused(t *testing.T) {
+ f := newPOFixture(t)
+ ctx := t.Context()
+ first, in := f.selectAndAdmit()
+ if err := f.owners.Materialize(ctx, first, in, newFenceWork()); err != nil {
+ t.Fatal(err)
+ }
+ // Two restarted attempts: fresh1 from the move id, fresh2 from fresh1.
+ again, err := f.owners.Select(ctx, f.route, first, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ second := f.admit(first, again)
+ third, err := f.owners.Select(ctx, f.route, second, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ admitted := f.admit(second, third)
+ // Trim the history to the last admission only (fresh2 from fresh1).
+ c := *admitted.Claim
+ h := *c.Holder
+ h.Admissions = h.Admissions[len(h.Admissions)-1:]
+ c.Holder = &h
+ f.putClaim(&c)
+ f.writeLocal(poOp, h.Generation, f.inc)
+ trimmed := f.read()
+ before := f.bucket.LatestID("data", f.markerKey())
+ err = f.owners.Materialize(ctx, trimmed, third, newFenceWork())
+ if !errors.Is(err, moves.ErrStartMarker) {
+ t.Fatalf("a trimmed history: %v", err)
+ }
+ if f.bucket.LatestID("data", f.markerKey()) != before {
+ t.Fatal("the verified marker was replaced")
+ }
+}
diff --git a/internal/host/movesprobeowners_test.go b/internal/host/movesprobeowners_test.go
index 5b07d432d..0e187443c 100644
--- a/internal/host/movesprobeowners_test.go
+++ b/internal/host/movesprobeowners_test.go
@@ -58,7 +58,7 @@ type poFixture struct {
route registry.Route
manifest moves.InstallManifest
launch ClaimServingLaunch
- envReport string
+ admitted int
renderETags map[string]string
renderOverlay string
order []string
@@ -105,6 +105,7 @@ func newPOFixture(t *testing.T) *poFixture {
key := "instances/" + poHost + "/g-" + poOp + "/0009/" + ltx.FormatFilename(1, 42)
f.snapshotVersion = f.bucket.Put("data", key, f.ltxHeader(1, 42))
f.manifest = moves.InstallManifest{V: 1, Op: poOp, Host: poHost, N: 42, SnapshotKey: key, SnapshotVersion: f.snapshotVersion, SnapshotCRC64: 7, DBSHA256: strings.Repeat("d", 64)}
+ f.bucket.Put("data", moves.InstallManifestKey(moves.GenerationPrefix("instances/"+poHost, poOp)), mustJSON(t, f.manifest))
j, err := moves.ReadJournal(f.dataDir, poHost, poOp)
if err != nil {
t.Fatal(err)
@@ -118,10 +119,13 @@ func newPOFixture(t *testing.T) *poFixture {
f.writeLocal(poOp, poOp, inc)
pins := mpPins()
pins.EnvReport = "sha256:env"
+ deadline := time.Date(2026, 10, 9, 12, 10, 0, 0, time.UTC)
f.rec = &provision.Record{Host: poHost, Account: "bab", App: "jam", Region: "uk",
- Move: &provision.MoveState{ID: poOp, Incarnation: inc, From: "ie", To: "uk", State: provision.MoveLanding, LandingPhase: provision.LandProbing, Inputs: pins, Landing: &provision.MoveLandingJournal{Attempts: 1}}}
+ Move: &provision.MoveState{ID: poOp, Incarnation: inc, From: "ie", To: "uk", State: provision.MoveLanding, LandingPhase: provision.LandProbing, Inputs: pins, CutDeadline: &deadline,
+ Landing: &provision.MoveLandingJournal{Attempts: 1, InstalledDBSHA256: "sha256:" + f.manifest.DBSHA256}}}
f.launch = ClaimServingLaunch{Identity: hostwrite.Identity{Config: "sha256:snapshot", Release: "sha256:bytes"}, Executable: "/opt/carlos/live/" + poHost, ExecutableSHA256: strings.Repeat("b", 64), Args: []string{"--socket", f.route.Socket, "--db", f.route.DBPath}, Env: []string{"A=1", "CARLOS_MOVE_PROBE=0"}, Version: pins.Release}
- f.envReport = pins.EnvReport
+ pins.EnvReport = envDigest(poEnvKey, f.launch.Env)
+ f.admitted = 3
f.renderETags, f.renderOverlay = pins.BundleETags, pins.OverlayETag
f.owners = &movesProbeOwners{Claims: f.claims, Data: f.bucket, Bucket: "data", Region: "uk", Box: "uk-box", DataDir: f.dataDir,
Records: func(context.Context, string, string, string) (*provision.Record, error) { f.reads++; return f.rec, nil },
@@ -133,7 +137,8 @@ func newPOFixture(t *testing.T) *poFixture {
f.launches++
return f.launch, nil
},
- EnvReport: func(context.Context, registry.Route) (string, error) { return f.envReport, nil },
+ EnvKey: poEnvKey,
+ Admitted: func(context.Context, *provision.Record) (int, error) { return f.admitted, nil },
Works: func(_ context.Context, host, action string, captured hostguard.Captured) (hostwrite.Work, error) {
f.order = append(f.order, "select:"+action)
return poWork{action: action, f: f}, nil
@@ -212,6 +217,8 @@ func (f *poFixture) read() hostclaim.Observed {
return o
}
+var poEnvKey = []byte("0123456789abcdef0123456789abcdef")
+
var poProbe = activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}
// admit applies the selected admission with the claim's own CAS.
@@ -421,7 +428,7 @@ func TestMovesProbeOwnersValidateThePreparedInputs(t *testing.T) {
func TestMovesProbeOwnersPinTheLaunchInputs(t *testing.T) {
for name, mutate := range map[string]func(*poFixture){
"another release": func(f *poFixture) { f.launch.Version = "v9" },
- "another environment": func(f *poFixture) { f.envReport = "sha256:other" },
+ "another environment": func(f *poFixture) { f.rec.Move.Inputs.EnvReport = "sha256:other" },
} {
t.Run(name, func(t *testing.T) {
f := newPOFixture(t)
@@ -461,6 +468,9 @@ func TestFirstProbeFromTheSourceGenerationIsAdmitted(t *testing.T) {
"spawn:" + launch.Config + "|" + launch.Release,
"render", "select:held-config", "held-config:" + HeldConfigIdentity(pins.BundleETags, pins.OverlayETag).Config + "|",
"select:held-release", "release:" + pins.Release, "held-release:|" + pins.Release,
+ // The prepared marker under the writer's fence: before the request
+ // and in the store before its attempt.
+ "spawn:" + launch.Config + "|" + launch.Release, "spawn:" + launch.Config + "|" + launch.Release,
}
if !reflect.DeepEqual(f.order, want) {
t.Fatalf("order:\n got %q\nwant %q", f.order, want)
@@ -487,10 +497,6 @@ func TestFirstProbeFromTheSourceGenerationIsAdmitted(t *testing.T) {
t.Fatalf("a database the restore did not write: %v", err)
}
f.writeLocal(poOp, poOp, f.inc)
- f.bucket.Put("data", "instances/"+poHost+"/generation.json", mustJSON(t, genselect.Marker{Op: "elsewhere", Incarnation: f.inc, Host: poHost, Box: "ie-box", InstalledAt: time.Unix(1, 0).UTC()}))
- if err := f.owners.Materialize(ctx, admitted, in, spawn); !errors.Is(err, moves.ErrStartMarker) {
- t.Fatalf("a marker the claim does not expect: %v", err)
- }
// The spawn selection itself refused: nothing is rendered.
f.order = nil
if err := f.owners.Materialize(ctx, admitted, in, poWork{action: "spawn", f: f, shut: true}); !errors.Is(err, hostwrite.ErrUnavailable) || contains(f.order, "render") {
@@ -525,13 +531,18 @@ func TestMovesProbeOwnersPublishThePreparedMarker(t *testing.T) {
if err := f.owners.PublicationEvidence(ctx, admitted); !errors.Is(err, hostbox.ErrChanged) {
t.Fatalf("evidence before publication: %v", err)
}
+ // Materialize publishes under the writer's fence; Group D's Publish
+ // records it.
+ if err := f.owners.Materialize(ctx, admitted, in, poWork{action: "spawn", f: f}); err != nil {
+ t.Fatal(err)
+ }
if err := f.owners.Publish(ctx, admitted, in); err != nil {
t.Fatal(err)
}
body, ok := f.bucket.Current("data", key)
var m genselect.Marker
if !ok || json.Unmarshal(body, &m) != nil || m.Op != poOp || m.Box != "uk-box" || m.Incarnation != f.inc || m.Host != poHost ||
- m.Pin != (genselect.Pin{SnapshotKey: f.manifest.SnapshotKey, SnapshotVersion: f.manifest.SnapshotVersion, SnapshotTXID: 42, SnapshotCRC64: 7}) {
+ m.Pin != (genselect.Pin{SnapshotKey: f.manifest.SnapshotKey, SnapshotVersion: f.manifest.SnapshotVersion, SnapshotTXID: 42, SnapshotCRC64: 7}) || m.ManifestSHA256 == "" {
t.Fatalf("marker: %s", body)
}
if err := f.owners.PublicationEvidence(ctx, admitted); err != nil {
@@ -546,7 +557,7 @@ func TestMovesProbeOwnersPublishThePreparedMarker(t *testing.T) {
if err != nil || again.Baseline.Generation != poOp || again.Baseline.Marker == nil || again.Baseline.Marker.Op != poOp {
t.Fatalf("restarted baseline after publication: %+v %v", again.Baseline, err)
}
- f.bucket.Put("data", key, mustJSON(t, genselect.Marker{Op: poOp, Incarnation: f.inc, Host: poHost, Box: "ie-box", InstalledAt: time.Unix(2, 0).UTC(), Pin: m.Pin}))
+ f.bucket.Put("data", key, mustJSON(t, genselect.Marker{Op: poOp, Incarnation: f.inc, Host: poHost, Box: "ie-box", InstalledAt: time.Unix(2, 0).UTC(), Pin: m.Pin, ManifestSHA256: m.ManifestSHA256}))
if err := f.owners.PublicationEvidence(ctx, admitted); !errors.Is(err, hostbox.ErrChanged) {
t.Fatalf("evidence for a marker this session did not write: %v", err)
}
@@ -617,7 +628,8 @@ func TestMovesProbeOwnersThroughTheClaimRuntime(t *testing.T) {
bucket.AddBucket("data")
key := strings.TrimSuffix(h.Location, "/") + "/g-" + poOp + "/0009/" + ltx.FormatFilename(1, 42)
hdr, _ := (&ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, MinTXID: 1, MaxTXID: 42}).MarshalBinary()
- manifest := moves.InstallManifest{V: 1, Op: poOp, Host: recoveryHost, N: 42, SnapshotKey: key, SnapshotVersion: bucket.Put("data", key, hdr)}
+ manifest := moves.InstallManifest{V: 1, Op: poOp, Host: recoveryHost, N: 42, SnapshotKey: key, SnapshotVersion: bucket.Put("data", key, hdr), DBSHA256: strings.Repeat("d", 64)}
+ bucket.Put("data", moves.InstallManifestKey(moves.GenerationPrefix(h.Location, poOp)), mustJSON(t, manifest))
j, _ := moves.ReadJournal(journalDir, recoveryHost, poOp)
body, _ := json.Marshal(manifest)
j.Mark(landStepInstall, string(body))
@@ -635,9 +647,11 @@ func TestMovesProbeOwnersThroughTheClaimRuntime(t *testing.T) {
t.Fatal(err)
}
pins := mpPins()
- pins.EnvReport = "sha256:env"
+ pins.EnvReport = envDigest(poEnvKey, []string{"A=1"})
+ deadline := f.now.Add(time.Hour)
rec := &provision.Record{Host: recoveryHost, Account: "bab", App: "jam", Region: "ie",
- Move: &provision.MoveState{ID: poOp, Incarnation: h.Incarnation, From: "uk", To: "ie", State: provision.MoveLanding, LandingPhase: provision.LandProbing, Inputs: pins, Landing: &provision.MoveLandingJournal{Attempts: 1}}}
+ Move: &provision.MoveState{ID: poOp, Incarnation: h.Incarnation, From: "uk", To: "ie", State: provision.MoveLanding, LandingPhase: provision.LandProbing, Inputs: pins, CutDeadline: &deadline,
+ Landing: &provision.MoveLandingJournal{Attempts: 1, InstalledDBSHA256: "sha256:" + manifest.DBSHA256}}}
records := func(context.Context, string, string, string) (*provision.Record, error) { return rec, nil }
var order []string
launches := 0
@@ -648,7 +662,8 @@ func TestMovesProbeOwnersThroughTheClaimRuntime(t *testing.T) {
Launch: func(context.Context, registry.Route) (ClaimServingLaunch, error) {
return ClaimServingLaunch{Executable: "/usr/bin/true", ExecutableSHA256: strings.Repeat("a", 64), Env: []string{"A=1"}, Version: pins.Release}, nil
},
- EnvReport: func(context.Context, registry.Route) (string, error) { return pins.EnvReport, nil },
+ EnvKey: poEnvKey,
+ Admitted: func(context.Context, *provision.Record) (int, error) { return 3, nil },
Works: func(_ context.Context, _, action string, _ hostguard.Captured) (hostwrite.Work, error) {
order = append(order, "select:"+action)
return plan9D5Work{}, nil