amadan

carlos / platform Public

host: the abort retires its own state on every pass before the release (Task 1 fix round 1)

A missing or unreadable journal directory no longer stands for "the
cleanup ran". While the claim is moving/aborting, the source runs the
idempotent retirement (snapshots by version, receipts, ledger, journal)
on every pass before ReleaseAbort, and any error ends the pass with the
claim unreleased. moveJournalPresent is unchanged. The census digest of
movesWorker.abortSource follows the change. The report test's comment
now says the record stays aborting during the park wait, and asserts it.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 730f2e05e7d1c2d558f5ebea424b85c4264def0a parent 5e04e9c
4 files changed, +101 −7
  • internal/host/movesabort.go +10 −5
  • internal/host/movesabort_test.go +83 −0
  • internal/host/movesreport_test.go +7 −1
  • internal/provision/plan9_b8_census_test.go +1 −1
diff --git a/internal/host/movesabort.go b/internal/host/movesabort.go
index e8d86cb79..7f1ecd3e1 100644
--- a/internal/host/movesabort.go
+++ b/internal/host/movesabort.go
@@ -45,7 +45,10 @@ import (
// park grammar, so "the source parks its own snapshots" means it
// retires them from its own bucket), then the receipts and the ledger,
// and last the op's journal directory (the bound is in the claim by
-// then);
+// then). It runs on every pass while the claim is aborting, before the
+// release, and is a no-op once everything is gone: nothing infers from
+// a missing journal that it already ran, and any error (a listing, a
+// delete, the journal directory) ends the pass unreleased;
// 5. it waits until no obligation under the op remains, then releases with
// hostclaim.ReleaseAbort (K7: from moving/aborting only, so never before
// the lane park was queued), and last clears the record with
@@ -119,10 +122,12 @@ func (w *movesWorker) abortSource(ctx context.Context, s release.Store, rec prov
}
fallthrough
case mine && op.Phase == hostclaim.PhaseAborting:
- if moveJournalPresent(cfg.DataDir, rec.Host, id) {
- if err := w.retireOwnMoveState(ctx, rec); err != nil {
- return err
- }
+ // Every pass, before the release: the retirement is idempotent, and
+ // a release never follows from inferring that it already ran (fix
+ // round 1). Any error, the journal directory's included, ends the
+ // pass with the claim still aborting.
+ if err := w.retireOwnMoveState(ctx, rec); err != nil {
+ return err
}
if released, err := w.releaseAbort(ctx, s, rec, obs); err != nil || !released {
return err
diff --git a/internal/host/movesabort_test.go b/internal/host/movesabort_test.go
index d0d65877f..44b728d79 100644
--- a/internal/host/movesabort_test.go
+++ b/internal/host/movesabort_test.go
@@ -2,7 +2,9 @@ package host
import (
"context"
+ "errors"
"os"
+ "path/filepath"
"strings"
"testing"
"time"
@@ -485,3 +487,84 @@ func TestAbortCompletesOnlyWithItsReleaseEvidence(t *testing.T) {
t.Fatalf("a serving claim without evidence completed the record: %+v", got.Move)
}
}
+
+// Fix round 1 (task-1-review.md, Important): release never follows from an
+// inference that the source's cleanup finished. After the claim's CAS, the
+// source retires its own state on every pass before ReleaseAbort, and any
+// failure to do so blocks the release until it succeeds.
+func TestAbortReleaseWaitsForTheSourcesCleanup(t *testing.T) {
+ // crashAfterClaim stops the source right after the Abort CAS, then lets
+ // the target region park and acknowledge the lane, so only the source's
+ // own cleanup stands between the claim and its release.
+ crashAfterClaim := func(t *testing.T) *abortHarness {
+ t.Helper()
+ h := newAbortHarness(t)
+ h.abort(t)
+ h.abAt = "after-abort-claim"
+ if err := movesPass(context.Background(), h.s, h.w()); err != nil {
+ t.Fatal(err)
+ }
+ if !h.abDone {
+ t.Fatalf("never reached after-abort-claim (%v)", h.abHits)
+ }
+ o := h.laneChore(t)
+ if o == nil {
+ t.Fatal("no lane park queued")
+ }
+ h.tPass(t)
+ h.pastHorizon(t, o)
+ h.tPass(t)
+ if h.laneChore(t) != nil {
+ t.Fatal("the lane park was not acknowledged")
+ }
+ h.worker1 = nil
+ return h
+ }
+ t.Run("the journal is gone but the state is not", func(t *testing.T) {
+ h := crashAfterClaim(t)
+ if err := os.RemoveAll(moves.JournalDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
+ t.Fatal(err)
+ }
+ h.pass(t)
+ h.closed(t)
+ h.ownStateGone(t)
+ })
+ t.Run("the journal cannot be inspected", func(t *testing.T) {
+ if os.Geteuid() == 0 {
+ t.Skip("root reads and removes past any mode")
+ }
+ h := crashAfterClaim(t)
+ hostDir := filepath.Dir(moves.JournalDir(h.cfg.DataDir, p1Host, p1ID))
+ if err := os.Chmod(hostDir, 0); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.Chmod(hostDir, 0o755) })
+ for i := 0; i < 3; i++ {
+ movesPass(context.Background(), h.s, h.w())
+ h.waiting(t)
+ }
+ if err := os.Chmod(hostDir, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ h.pass(t)
+ h.closed(t)
+ h.ownStateGone(t)
+ })
+ t.Run("the snapshots cannot be listed", func(t *testing.T) {
+ h := crashAfterClaim(t)
+ h.vs.deny = func(op, bucket, key string) error {
+ if op == "list-versions" && bucket == "data-ie" {
+ return errors.New("injected 503")
+ }
+ return nil
+ }
+ for i := 0; i < 3; i++ {
+ movesPass(context.Background(), h.s, h.w())
+ h.waiting(t)
+ }
+ h.vs.deny = nil
+ h.pass(t)
+ h.closed(t)
+ h.ownStateGone(t)
+ })
+}
diff --git a/internal/host/movesreport_test.go b/internal/host/movesreport_test.go
index 49f0109c5..ed4541d3e 100644
--- a/internal/host/movesreport_test.go
+++ b/internal/host/movesreport_test.go
@@ -582,7 +582,13 @@ func TestArchiveVerdictsAreEvictedWhenTheMoveEnds(t *testing.T) {
if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveGuardedAbort(g)); err != nil {
t.Fatal(err)
}
- h.pass(t) // the abort completes; the record is at rest
+ // The source's pass begins the abort; the record stays aborting while
+ // the target region parks the lane (Plan 5 Task 1). The move is over for
+ // the target's caches already.
+ h.pass(t)
+ if m := h.read(t, p1Host).Move; m == nil || m.State != provision.MoveAborting {
+ t.Fatalf("the record during the park wait: %+v", m)
+ }
h.targetPass(t, h.targetCfg())
if len(h.worker2.archives) != 0 {
t.Fatalf("a verdict outlived its move: %v", h.worker2.archives)
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go
index fa42f8606..26d0fe6c8 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -627,7 +627,7 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/host/fleetbind.go", "WriteFleetBinding", "nonrecord", "sel:Put", "44ab984c2c9b30abcc938132118f61a6922ccc0cd40ed2b6cb61c7654aa0a1c3"},
{"internal/host/logship.go", "logshipFlush", "nonrecord", "sel:Put", "a2fa594c1206028af170a2d0cbedef55ba00a2f1c698a6853e1214cc3b172dd9"},
{"internal/host/metrics_loop.go", "putAlertState", "nonrecord", "sel:Put", "d5b6e8b34dc5a731d6bbafeac0b7d0c5c20706d6f413dff38088b3321b35a26c"},
- {"internal/host/movesabort.go", "movesWorker.abortSource", "record-writer", "key:SetRecordMove", "5ba846213f439e5e55acac31ce7b7fe4f74810ed46834cea158d9d32d8ad7dbe"},
+ {"internal/host/movesabort.go", "movesWorker.abortSource", "record-writer", "key:SetRecordMove", "70f2cd7ec7709045043081c078f1abe63ef822325eef7016ab9f595a16cabad5"},
{"internal/host/movesarchive.go", "cutRun.publishCutArchive", "nonrecord", "sel:CopyVersioned,sel:PutIfAbsent", "6c7fcfd2d4b2a0319ac15ecbfe003ed0292fe160457fd5704be508c2ab302a1b"},
{"internal/host/movescut.go", "cutRun.cutTransition", "record-writer", "key:SetRecordMove", "d0d67399aa784036de43b3c143e88c3d85f8b3d9e6345f84908d3b58ff56191e"},
{"internal/host/movescut.go", "publishStoppedReport", "nonrecord", "sel:Put", "c710682f31d99127ad76524add246fda9f640c63934940e45c79ea25ba36eb3c"},