amadan
host: a round promotes or requests a cut only when its own published source report passed (Task 14 fix round 3)
writeSourceReport returns the report it published; the round ends unless that report's S1 and S2 pass, and decideReady judges that same report rather than re-reading one. Retries and errors already ended the round earlier. An unknown archive error on a ready automatic move therefore publishes MV-N151 and requests no cut.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2 files changed,
+143
−17
internal/host/movesphase1.go+32 −17internal/host/movesphase1_test.go+111 −0
diff --git a/internal/host/movesphase1.go b/internal/host/movesphase1.go| index 591ccda12..1871b0bdd 100644 |
| --- a/internal/host/movesphase1.go |
| +++ b/internal/host/movesphase1.go |
| @@ -196,7 +196,8 @@ func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec p |
| // 15-minute one, cannot fence staging writers. Without the deployment |
| // bucket there is no shared place for it, so the round refuses. |
| if cfg.StagingLease == nil { |
| - return w.writeSourceReport(ctx, s, rec, &srcRound{s1: &moves.Check{Reason: s1NoDeployBucket}}) |
| + _, err := w.writeSourceReport(ctx, s, rec, &srcRound{s1: &moves.Check{Reason: s1NoDeployBucket}}) |
| + return err |
| } |
| tgt, ok := fleet.targets[rec.Move.To] |
| if !ok { |
| @@ -239,7 +240,8 @@ func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec p |
| if why, err := w.sourceRefusal(ctx, s, rec); err != nil { |
| return err |
| } else if why != "" { |
| - return w.writeSourceReport(ctx, s, rec, &srcRound{s1: &moves.Check{Reason: why}}) |
| + _, err := w.writeSourceReport(ctx, s, rec, &srcRound{s1: &moves.Check{Reason: why}}) |
| + return err |
| } |
| // K12 and T9: the target region and box are in touched before the |
| // first staging write. |
| @@ -345,10 +347,18 @@ func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec p |
| sr.s1 = &moves.Check{Reason: why} |
| } |
| sr.ledger = ledger |
| - if err := w.writeSourceReport(ctx, s, rec, sr); err != nil { |
| + published, err := w.writeSourceReport(ctx, s, rec, sr) |
| + if err != nil { |
| return err |
| } |
| - return w.decideReady(ctx, s, rec, g, sr) |
| + // The invariant: a round promotes or requests a cut only when its own |
| + // source report passed every check and was published. Any failing |
| + // check ends the round here; a retry or an error ended it above. |
| + if !sourcePassed(published) { |
| + w.notice(rec.Host+"#ready", "moves: %s: this round's source report does not pass (S1 %v, S2 %v); no promotion and no cut", rec.Host, published.S1.OK, published.S2.OK) |
| + return nil |
| + } |
| + return w.decideReady(ctx, s, rec, g, sr, published) |
| } |
| // refreshCrossRate measures the cross-region path when no recent round did: |
| @@ -591,11 +601,14 @@ func digestHex(d string) string { |
| return h |
| } |
| +// sourcePassed is every check a source report carries: S1 and S2. |
| +func sourcePassed(r *moves.SourceReport) bool { return r != nil && r.S1.OK && r.S2.OK } |
| + |
| // writeSourceReport publishes the source's readiness report (§7.3): the |
| // launch environment as HMACs, S1 (a refusal, or the instance's own answer |
| // to its readiness path), S2, the published archive, the live release and |
| // the cross-region rate. It is replaced every round. |
| -func (w *movesWorker) writeSourceReport(ctx context.Context, s release.Store, rec provision.Record, r *srcRound) error { |
| +func (w *movesWorker) writeSourceReport(ctx context.Context, s release.Store, rec provision.Record, r *srcRound) (*moves.SourceReport, error) { |
| cfg := w.cfg |
| now := cfg.Now().UTC() |
| sr := moves.SourceReport{V: 1, Box: cfg.Box, MoveID: rec.Move.ID, At: now, |
| @@ -627,16 +640,19 @@ func (w *movesWorker) writeSourceReport(ctx context.Context, s release.Store, re |
| if err != nil { |
| sr.S1 = moves.Check{Reason: s1NoSecretsKey} |
| } else if env, _, err := cfg.LaunchEnv(rec.Host); err != nil { |
| - return fmt.Errorf("compose the launch environment: %w", err) |
| + return nil, fmt.Errorf("compose the launch environment: %w", err) |
| } else { |
| sr.Env = moves.ReportEnv(key, env) |
| } |
| } |
| body, err := json.Marshal(sr) |
| if err != nil { |
| - return err |
| + return nil, err |
| + } |
| + if err := moves.PutReplace(ctx, s, moves.SourceReportKey(moves.ReportPrefix(rec.Account, rec.App, rec.Host, rec.Move.ID)), body); err != nil { |
| + return nil, err |
| } |
| - return moves.PutReplace(ctx, s, moves.SourceReportKey(moves.ReportPrefix(rec.Account, rec.App, rec.Host, rec.Move.ID)), body) |
| + return &sr, nil |
| } |
| // probeSource is S1: the instance's own answer to its readiness path, |
| @@ -656,11 +672,14 @@ func (w *movesWorker) probeSource(ctx context.Context, s release.Store, rec prov |
| return w.cfg.ReadyProbe(ctx, rec.Host, m.ReadyPathOrDefault()) |
| } |
| -// decideReady reads the target's report and applies §7.3's rule; on ready |
| -// from preparing it pins the inputs and promotes, and an automatic cut then |
| -// marks the cut requested (§7.4). Every CAS is guarded by the op identity |
| -// the round read: a replacement move is ErrMoveConflict and the round ends. |
| -func (w *movesWorker) decideReady(ctx context.Context, s release.Store, rec provision.Record, g provision.MoveGuard, r *srcRound) error { |
| +// decideReady reads the target's report and applies §7.3's rule to it and |
| +// to source, the report this round just published (and that passed); on |
| +// ready from preparing it pins the inputs and promotes, and an automatic |
| +// cut then marks the cut requested (§7.4). The caller reaches it only for a |
| +// round whose own report passed, so a failing round never requests a cut. |
| +// Every CAS is guarded by the op identity the round read: a replacement |
| +// move is ErrMoveConflict and the round ends. |
| +func (w *movesWorker) decideReady(ctx context.Context, s release.Store, rec provision.Record, g provision.MoveGuard, r *srcRound, source *moves.SourceReport) error { |
| cfg := w.cfg |
| state := rec.Move.State |
| if state == provision.MovePreparing { |
| @@ -668,10 +687,6 @@ func (w *movesWorker) decideReady(ctx context.Context, s release.Store, rec prov |
| if err != nil { |
| return err |
| } |
| - source, err := readSourceReport(ctx, s, rec) |
| - if err != nil { |
| - return err |
| - } |
| ok, why := moves.Ready(r.deltaBytes, cfg.ReadyDelta, target, source, cfg.Now()) |
| if !ok { |
| w.notice(rec.Host+"#ready", "moves: %s: not ready: %s", rec.Host, why) |
diff --git a/internal/host/movesphase1_test.go b/internal/host/movesphase1_test.go| index dd623d13b..d7a64e2ad 100644 |
| --- a/internal/host/movesphase1_test.go |
| +++ b/internal/host/movesphase1_test.go |
| @@ -782,3 +782,114 @@ func TestArchiveSentencesMatchTheArchiveRows(t *testing.T) { |
| t.Fatalf("err %v; s1LongPath %q", err, s1LongPath(pe.Path)) |
| } |
| } |
| + |
| +// readyAuto drives a move to a persisted ready (cut manual, so the |
| +// promotion does not cut in the same pass), then makes it automatic: the |
| +// state a crash between the promotion and the cut leaves behind. |
| +func readyAuto(t *testing.T) *phase1Harness { |
| + t.Helper() |
| + h := newPhase1Harness(t) |
| + h.start(t, "manual") |
| + h.pass(t) |
| + h.targetPass(t, h.targetCfg()) |
| + h.pass(t) |
| + if got := h.read(t, p1Host).Move.State; got != provision.MoveReady { |
| + t.Fatalf("the fixture did not reach ready: %s", got) |
| + } |
| + if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error { |
| + rec.Move.Cut = "auto" |
| + return nil |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + return h |
| +} |
| + |
| +// The re-review's regression: from a persisted ready, automatic move, a |
| +// round whose archive fails with an unrecognised error publishes MV-N151 |
| +// and requests no cut. Without the failure, the same round does. |
| +func TestAnUnknownArchiveErrorRequestsNoCut(t *testing.T) { |
| + h := readyAuto(t) |
| + prev := archivePlainDigest |
| + t.Cleanup(func() { archivePlainDigest = prev }) |
| + archivePlainDigest = func(string, string) (string, int64, int, error) { return "", 0, 0, errors.New("EIO") } |
| + h.pass(t) |
| + if got := h.read(t, p1Host).Move.State; got != provision.MoveReady { |
| + t.Fatalf("an unknown archive error requested the cut: %s", got) |
| + } |
| + if sr := h.sourceReport(t); sr.S1.OK || sr.S1.Reason != s1ArchiveFailed { |
| + t.Fatalf("S1 = %+v", sr.S1) |
| + } |
| + archivePlainDigest = prev |
| + h.pass(t) |
| + if got := h.read(t, p1Host).Move; got.State != provision.MoveCutting { |
| + t.Fatalf("a clean round did not request the cut: %+v", got) |
| + } |
| +} |
| + |
| +// The invariant: a round requests a cut only when this round's own source |
| +// report passed every check and was published. A failure injected into |
| +// each source check, retry or error of the round leaves a ready, |
| +// automatic move ready. |
| +func TestOnlyAPassingRoundRequestsTheCut(t *testing.T) { |
| + ctx := context.Background() |
| + for name, inject := range map[string]func(t *testing.T, h *phase1Harness){ |
| + "none (the control)": nil, |
| + "S1 the instance does not answer": func(t *testing.T, h *phase1Harness) { |
| + h.probe = moves.Check{Reason: s1NoAnswer(p1Host)} |
| + }, |
| + "S1 the manifest cannot be read": func(t *testing.T, h *phase1Harness) { |
| + _, etag, _ := h.s.Get(ctx, provision.ManifestKey("bab", "app")) |
| + if _, err := h.s.Put(ctx, provision.ManifestKey("bab", "app"), []byte("{"), etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + }, |
| + "S1 no probe on this box": func(t *testing.T, h *phase1Harness) { h.cfg.ReadyProbe = nil }, |
| + "S1 no secrets key": func(t *testing.T, h *phase1Harness) { h.cfg.SecretsKey = "" }, |
| + "S1 a refused column": func(t *testing.T, h *phase1Harness) { h.route.Canary = true }, |
| + "S1 a special file": func(t *testing.T, h *phase1Harness) { |
| + if err := os.Symlink("/etc/passwd", filepath.Join(h.cfg.DataDir, p1Host, "link")); err != nil { |
| + t.Fatal(err) |
| + } |
| + }, |
| + "S2 the next spawn is not strict": func(t *testing.T, h *phase1Harness) { |
| + h.strict = moves.Check{Reason: "S2: the next spawn of this host is not strict"} |
| + }, |
| + "S2 no strict hook": func(t *testing.T, h *phase1Harness) { h.cfg.StrictNext = nil }, |
| + "the archive changed (retry)": func(t *testing.T, h *phase1Harness) { |
| + prev := archivePlainDigest |
| + t.Cleanup(func() { archivePlainDigest = prev }) |
| + archivePlainDigest = func(string, string) (string, int64, int, error) { return "", 0, 0, archive.ErrChanged } |
| + }, |
| + "the launch environment cannot be composed": func(t *testing.T, h *phase1Harness) { |
| + h.cfg.LaunchEnv = func(string) (map[string]string, []string, error) { return nil, nil, errors.New("no row") } |
| + }, |
| + "a staging write fails": func(t *testing.T, h *phase1Harness) { |
| + h.f.putSized("data-ie", "instances/"+p1Host+"/"+h.inc+"/g-cur/0000/0000000000000003-0000000000000003.ltx", "new", 10) |
| + h.f.failAt = func(string, string) error { return errors.New("injected 503") } |
| + }, |
| + "the staging lease is too close to its end": func(t *testing.T, h *phase1Harness) { |
| + h.f.putSized("data-ie", "instances/"+p1Host+"/"+h.inc+"/g-cur/0000/0000000000000003-0000000000000003.ltx", "new", 10) |
| + h.lease.until = h.clock.now.Add(time.Minute) |
| + }, |
| + "no deployment bucket": func(t *testing.T, h *phase1Harness) { h.cfg.StagingLease = nil }, |
| + "the claim is not the move's": func(t *testing.T, h *phase1Harness) { |
| + rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.Phase = hostclaim.PhaseAborting }) |
| + }, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + h := readyAuto(t) |
| + if inject != nil { |
| + inject(t, h) |
| + } |
| + h.pass(t) |
| + got := h.read(t, p1Host).Move.State |
| + if inject == nil && got != provision.MoveCutting { |
| + t.Fatalf("the control round did not request the cut: %s", got) |
| + } |
| + if inject != nil && got != provision.MoveReady { |
| + t.Fatalf("a failing round requested the cut: %s", got) |
| + } |
| + }) |
| + } |
| +} |