amadan
host: the completion pass reads its registry rows and the hosts its own op journals name, so a completion whose row is gone is still said; it reads each claim once and asks which box it is only for a pending completion; an ErrChanged after its serving-here check is said once; alerts are one per host, step and cause and forget what clears; the startup full build runs in the first bounded pass and the launch reads nothing the loop owns; the sweep reads claims before its exclusive lock and checks again under it; edgesfile: no edges file at all is not complete evidence while a box reports fresh
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
12 files changed,
+867
−203
internal/console/regioncandidates_test.go+7 −4internal/console/regionfill_test.go+17 −0internal/edgesfile/snapshot.go+6 −5internal/edgesfile/snapshot_test.go+61 −0internal/host/agent.go+2 −3internal/host/relocatepass.go+219 −118internal/host/relocatepass_absence_test.go+11 −7internal/host/relocatepass_carries_test.go+11 −6internal/host/relocatepass_round1_test.go+290 −0internal/host/relocationfinish_unix_test.go+30 −0internal/host/relocationinstall_lock_linux_test.go+47 −0internal/host/relocationsweep.go+166 −60
diff --git a/internal/console/regioncandidates_test.go b/internal/console/regioncandidates_test.go| index d2f36a16f..6c598227a 100644 |
| --- a/internal/console/regioncandidates_test.go |
| +++ b/internal/console/regioncandidates_test.go |
| @@ -409,7 +409,11 @@ func TestStaleBoxesStillCountForTheKeylessProof(t *testing.T) { |
| } |
| } |
| -// F1: snapshots that cannot be loaded make no box fresh, said once. |
| +// F1: snapshots that cannot be loaded make no box fresh, and the failure |
| +// is never said twice. The once-per-process throttle lives in edgesfile |
| +// with the read (stage B Task 10), where TestReadLiveSnapshotSaysALoad |
| +// FailureOnce resets it; here an earlier test may already have tripped |
| +// it, so the count is at most one (Task 10 review M10). |
| func TestLiveEdgesSnapshotLoadFailureIsNotFreshAndLoggedOnce(t *testing.T) { |
| ctx := context.Background() |
| s := release.NewDirStore(t.TempDir()) |
| @@ -420,7 +424,6 @@ func TestLiveEdgesSnapshotLoadFailureIsNotFreshAndLoggedOnce(t *testing.T) { |
| var buf bytes.Buffer |
| restore := captureLogTo(&buf) |
| defer restore() |
| - livenessSaid.Delete("load") |
| for i := 0; i < 2; i++ { |
| snap, err := readLiveEdgesSnapshot(ctx, failingListPrefix{Store: s, prefix: "status/boxes/"}, time.Now()) |
| if err != nil { |
| @@ -430,8 +433,8 @@ func TestLiveEdgesSnapshotLoadFailureIsNotFreshAndLoggedOnce(t *testing.T) { |
| t.Fatalf("round %d: snapshots unloadable, uk = %+v, want ineligible", i, c) |
| } |
| } |
| - if n := strings.Count(buf.String(), "box snapshots"); n != 1 { |
| - t.Fatalf("load failure logged %d times, want once: %s", n, buf.String()) |
| + if n := strings.Count(buf.String(), "box snapshots"); n > 1 { |
| + t.Fatalf("load failure logged %d times, want at most once: %s", n, buf.String()) |
| } |
| } |
diff --git a/internal/console/regionfill_test.go b/internal/console/regionfill_test.go| index 100ae7778..ac3f62c69 100644 |
| --- a/internal/console/regionfill_test.go |
| +++ b/internal/console/regionfill_test.go |
| @@ -114,6 +114,10 @@ func TestFillStoreErrorIsC20(t *testing.T) { |
| } |
| // codex R2-2: dormancy keeps explicit requests, never app-derived ones. |
| +// Stage B Task 10 review M5: with no edges file at all but boxes still |
| +// reporting fresh, the evidence is incomplete (ruling (A)), so the |
| +// app-derived request is "could not confirm" (C28), not "absent"; once no |
| +// box reports either, it is C16 absent. |
| func TestAppDerivedNeedsPositiveEligibility(t *testing.T) { |
| fx, token, u := fillFixture(t, "uk") |
| for _, k := range []string{"control/edges/uk-box.json", "control/edges/ie-box.json"} { |
| @@ -123,6 +127,19 @@ func TestAppDerivedNeedsPositiveEligibility(t *testing.T) { |
| } |
| mustCreate(t, u, token, `{"host":"x.go.tito.io","region":"uk"}`) |
| status, _, out := create(t, u, token, `{"host":"y.go.tito.io"}`) |
| + if status != http.StatusServiceUnavailable || !strings.Contains(out, msgAppRegionUnconfirmed("uk")) { |
| + t.Fatalf("app-derived uk with fresh boxes and no edges files = %d %s, want C28 unconfirmed", status, out) |
| + } |
| + keys, err := fx.rel.List(t.Context(), "status/boxes/") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, k := range keys { |
| + if err := fx.rel.Delete(t.Context(), k); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + status, _, out = create(t, u, token, `{"host":"z.go.tito.io"}`) |
| if status != http.StatusBadRequest || !strings.Contains(out, reasonText("absent")) { |
| t.Fatalf("app-derived uk under dormancy = %d %s, want C16 absent", status, out) |
| } |
diff --git a/internal/edgesfile/snapshot.go b/internal/edgesfile/snapshot.go| index cffde957d..701d47bef 100644 |
| --- a/internal/edgesfile/snapshot.go |
| +++ b/internal/edgesfile/snapshot.go |
| @@ -106,10 +106,11 @@ func ReadLiveSnapshot(ctx context.Context, s release.Store, now time.Time) (Snap |
| if err != nil { |
| return snap, err |
| } |
| - if len(snap.Files) == 0 { |
| - snap.BoxesComplete = true // nothing to be incomplete about |
| - return snap, nil |
| - } |
| + // No edges file at all is not complete evidence by itself (Task 10 |
| + // review M5): the join still runs, so a fresh status snapshot with no |
| + // edges file (ruling (A) below) makes the evidence incomplete, as it |
| + // does beside any other file. With no file and no fresh snapshot the |
| + // evidence is complete and empty. |
| // Completeness is carried separately from staleness: LoadBoxesComplete |
| // returns what LoadBoxes would, plus whether every listed snapshot was |
| // actually read. A skipped read leaves that box's liveness UNKNOWN, |
| @@ -167,7 +168,7 @@ func ReadLiveSnapshot(ctx context.Context, s release.Store, now time.Time) (Snap |
| } |
| } |
| } |
| - if joined == 0 { |
| + if joined == 0 && len(snap.Files) > 0 { |
| if _, said := livenessSaid.LoadOrStore("none", true); !said { |
| log.Printf("edges: %d edges file(s) under %s but no fresh box snapshot matched any of them, so automatic region placement sees the home alone (are edges and status in different buckets? see CARLOS_STATUS_BUCKET/CARLOS_STATUS_PREFIX)", len(snap.Files), Prefix) |
| } |
diff --git a/internal/edgesfile/snapshot_test.go b/internal/edgesfile/snapshot_test.go| index 2f60ed8ba..4f142a5f5 100644 |
| --- a/internal/edgesfile/snapshot_test.go |
| +++ b/internal/edgesfile/snapshot_test.go |
| @@ -1,7 +1,12 @@ |
| package edgesfile |
| import ( |
| + "bytes" |
| "context" |
| + "errors" |
| + "log" |
| + "os" |
| + "strings" |
| "testing" |
| "time" |
| @@ -73,3 +78,59 @@ func TestLiveSnapshotJoinsEdgesFilesWithLiveness(t *testing.T) { |
| t.Fatalf("an unreadable file: %+v %v", plain, err) |
| } |
| } |
| + |
| +// Task 10 review M5: no edges file at all is not complete evidence by |
| +// itself. A fresh box with no edges file makes the evidence incomplete |
| +// (ruling (A)), exactly as it does beside other files; with no file and |
| +// no fresh box the evidence is complete and empty. |
| +func TestLiveSnapshotWithNoEdgesFilesStillJoinsLiveness(t *testing.T) { |
| + ctx := context.Background() |
| + now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) |
| + s := release.NewDirStore(t.TempDir()) |
| + live, err := ReadLiveSnapshot(ctx, s, now) |
| + if err != nil || !live.EvidenceComplete() || len(live.Files) != 0 { |
| + t.Fatalf("an empty deployment: %+v %v", live, err) |
| + } |
| + if err := status.WriteBox(ctx, s, status.Box{Label: "box-a", GeneratedAt: now.Add(-time.Second)}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if live, err = ReadLiveSnapshot(ctx, s, now); err != nil || live.EvidenceComplete() { |
| + t.Fatalf("a fresh box with no edges file left the evidence complete: %+v %v", live, err) |
| + } |
| + if choice, err := LiveSnapshot(ctx, s, now); err != nil || choice.Complete { |
| + t.Fatalf("boxchoice complete over a fresh box with no edges file: %+v %v", choice, err) |
| + } |
| +} |
| + |
| +// The load failure is said once per process: the throttle the console |
| +// and the pass share lives here (Task 10 review M10). |
| +func TestReadLiveSnapshotSaysALoadFailureOnce(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + if _, err := s.Put(ctx, Key("box-a"), []byte(`{"v":1,"box":"box-a","region":"eu-west-1","hosts":{}}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + var buf bytes.Buffer |
| + log.SetOutput(&buf) |
| + t.Cleanup(func() { log.SetOutput(os.Stderr) }) |
| + livenessSaid.Delete("load") |
| + for range 2 { |
| + live, err := ReadLiveSnapshot(ctx, failingStatusList{s}, time.Now()) |
| + if err != nil || live.EvidenceComplete() { |
| + t.Fatalf("snapshots unloadable: %+v %v", live, err) |
| + } |
| + } |
| + if n := strings.Count(buf.String(), "could not load box snapshots"); n != 1 { |
| + t.Fatalf("said %d times, want once:\n%s", n, buf.String()) |
| + } |
| +} |
| + |
| +// failingStatusList fails the status half of the join and serves the rest. |
| +type failingStatusList struct{ release.Store } |
| + |
| +func (f failingStatusList) List(ctx context.Context, prefix string) ([]string, error) { |
| + if strings.HasPrefix(prefix, "status/") { |
| + return nil, errors.New("status bucket unreachable") |
| + } |
| + return f.Store.List(ctx, prefix) |
| +} |
diff --git a/internal/host/agent.go b/internal/host/agent.go| index f8eddedc6..7234b24ed 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -605,9 +605,8 @@ func runAgent(args []string, buildClaims ClaimFactory) { |
| } |
| }() |
| - if relocation := startRelocationPass(agentCtx, claimCtx, claims, reg, svc); relocation != nil { |
| - log.Printf("agent: relocation pass started (completions only: %v)", relocation.Worker == nil) |
| - } |
| + // The relocation pass (stage B Task 10): it says itself what it started. |
| + startRelocationPass(agentCtx, claimCtx, claims, reg, svc) |
| go adoptLoop(reg, svc, sd, claims.Writes) |
| go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, claims.Providers.Provision) |
| if logSink != nil { |
diff --git a/internal/host/relocatepass.go b/internal/host/relocatepass.go| index eda18ecd5..708b7da15 100644 |
| --- a/internal/host/relocatepass.go |
| +++ b/internal/host/relocatepass.go |
| @@ -46,24 +46,32 @@ type RelocationRuntime struct { |
| Convert func(context.Context, release.Store, hostclaim.Observed) error |
| Finish func(ctx context.Context, host, op string) error |
| // Hosts, when set, is the hosts a pass reads instead of every claim |
| - // in the bucket: the completion-only runtime's (ruling 82) are this |
| - // box's own registry rows, since a completion this box can finish |
| - // always has its row here. |
| + // in the bucket. The completion-only runtime's (ruling 82) are this |
| + // box's registry rows and every host its own op journals name |
| + // (relocate/ops/*/{probe-fence,attempts}.json), so a completion whose |
| + // row is gone is still visited and its refusal said (Task 10 review |
| + // M1). An error beside a non-nil list is said, and the listed hosts |
| + // are still read. |
| Hosts func() ([]string, error) |
| - // Self, when Box is empty, answers this box once per pass: the |
| - // completion-only runtime has no roster read of its own, so it asks |
| - // the bound assembly's verified stop identity. A pass that cannot |
| - // answer it finishes nothing (carry 8: a completion is finished only |
| - // on the box it serves on). |
| + // Self, when Box is empty, answers this box: the completion-only |
| + // runtime has no roster read of its own, so it asks the bound |
| + // assembly's verified stop identity. It is asked at most once a pass, |
| + // and only once a serving claim with an undischarged completion is |
| + // found (review M2), so a box with nothing to finish makes no identity |
| + // read. A pass that cannot answer it finishes nothing (carry 8: a |
| + // completion is finished only on the box it serves on). |
| Self func(context.Context) (string, error) |
| - DataDir string // the managed data root: relocate/ journals (N9) |
| - alerted sync.Map // host\x00cause: one alert per host and cause (rulings 35, 36) |
| + DataDir string // the managed data root: relocate/ journals (N9) |
| + alerts relocationAlerts |
| // upgrade, when set, builds the full pass this completion-only runtime |
| - // stands in for: region HA is on, but the build failed (a roster, an |
| - // enrollment or a key that could not be read). Each pass tries it |
| - // again until it builds, or answers that the owners are not bound. A |
| + // stands in for while region HA is on. The first pass, bounded like |
| + // every pass, tries it (review M9); a pass whose build fails tries it |
| + // again, until it builds or answers that the owners are not bound. A |
| // read that failed once is never taken as "no relocation on this box". |
| upgrade func(context.Context) (*RelocationRuntime, error) |
| + // config is the assembly config the full pass was built with |
| + // (NewRelocationRuntime); zero for the completion-only runtime. |
| + config RelocationAssemblyConfig |
| } |
| // relocationStepper is hostrelocate.Worker's and Participant's Step. |
| @@ -319,7 +327,7 @@ func NewRelocationRuntime(ctx context.Context, getenv func(string) string, store |
| Self: termfence.WrittenBy{Box: in.Enrollment.Box, Enrollment: in.Enrollment.Enrollment}, |
| Region: stores.Region, Controller: newClaimID(), NewID: newClaimID, Now: time.Now} |
| participant := hostrelocate.Participant{Claims: store, Box: in.Enrollment.Box, Enrollment: in.Enrollment.Enrollment, Effects: effects, Now: time.Now} |
| - return &RelocationRuntime{Worker: worker, Participant: participant, Box: in.Enrollment.Box, Convert: convertFencedLanding(worker, store), Finish: effects.Finish, DataDir: cfg.DataDir}, nil |
| + return &RelocationRuntime{Worker: worker, Participant: participant, Box: in.Enrollment.Box, Convert: convertFencedLanding(worker, store), Finish: effects.Finish, DataDir: cfg.DataDir, config: cfg}, nil |
| } |
| // newCompletionRuntime is ruling 82: the smallest runtime that finishes a |
| @@ -335,16 +343,28 @@ func newCompletionRuntime(local *productionClaimLocal, reg *registry.Registry, a |
| return nil, hostwrite.ErrUnavailable |
| } |
| cfg := RelocationAssemblyConfig{Registry: reg, Activator: act, DataDir: a.Guard.DataDir, Discharge: local.DischargeCompleted, Now: time.Now} |
| + dataDir := a.Guard.DataDir |
| hosts := func() ([]string, error) { |
| rows, err := reg.List() |
| if err != nil { |
| return nil, err |
| } |
| + seen := make(map[string]bool, len(rows)) |
| out := make([]string, 0, len(rows)) |
| for _, row := range rows { |
| - out = append(out, row.Host) |
| + if !seen[row.Host] { |
| + seen[row.Host] = true |
| + out = append(out, row.Host) |
| + } |
| + } |
| + named, err := relocationJournalHosts(dataDir) |
| + for _, host := range named { |
| + if !seen[host] { |
| + seen[host] = true |
| + out = append(out, host) |
| + } |
| } |
| - return out, nil |
| + return out, err |
| } |
| // Carry 8: the completion-only runtime has no roster read, so this box |
| // is the bound assembly's own verified stop identity, asked each pass. |
| @@ -389,82 +409,87 @@ func (r *RelocationRuntime) Run(ctx context.Context, store release.Store, serial |
| // in one pass (ruling 49): the participant (its probe stop), the worker |
| // (the serving CAS), then the completion this box has not yet finished, |
| // found from the claim's record (ruling 80). A landing whose target is |
| -// fenced is converted when its move committed (J3). |
| +// fenced is converted when its move committed (J3). A claim is read once |
| +// a pass, and once more only after a step that may have written it. |
| func (r *RelocationRuntime) once(ctx context.Context, store release.Store) { |
| r.tryUpgrade(ctx) |
| hosts, err := r.hosts(ctx, store) |
| - if err != nil { |
| - log.Printf("relocate: list hosts: %v", err) |
| + r.result("", "list hosts", err) |
| + if err != nil && hosts == nil { |
| return |
| } |
| - box := r.Box |
| - if box == "" && r.Self != nil { |
| - // No answer is no box: this pass finishes nothing (carry 8). |
| - if box, err = r.Self(ctx); err != nil { |
| - box = "" |
| - r.alert("", "this box", err) |
| - } |
| - } |
| + var self passSelf |
| + read := make(map[string]bool, len(hosts)) |
| for _, host := range hosts { |
| + read[host] = true |
| obs, err := hostclaim.Read(ctx, store, host) |
| - if err != nil { |
| - // A claim that cannot be read is not "nothing to do": said once, |
| - // and read again next pass. |
| - r.alert(host, "read", err) |
| - continue |
| - } |
| - if obs.Absent { |
| + // A claim that cannot be read is not "nothing to do": said once, |
| + // and read again next pass. |
| + if r.result(host, "read", err); err != nil || obs.Absent { |
| continue |
| } |
| + stepped := false |
| switch obs.Claim.Op.Kind { |
| case hostclaim.KindLanding: |
| if r.Convert != nil { |
| - if err := r.Convert(ctx, store, obs); err != nil { |
| - r.alert(host, "convert", err) |
| - } |
| + stepped = true |
| + r.result(host, "convert", r.Convert(ctx, store, obs)) |
| } |
| case hostclaim.KindRelocating: |
| if r.Participant != nil { |
| - if _, err := r.Participant.Step(ctx, host); err != nil && !errors.Is(err, hostrelocate.ErrNotYet) { |
| - r.alert(host, r.Box, err) |
| - } |
| + stepped = true |
| + _, err := r.Participant.Step(ctx, host) |
| + r.result(host, "participant", notYetIsNoError(err)) |
| } |
| if r.Worker != nil { |
| - if _, err := r.Worker.Step(ctx, host); err != nil && !errors.Is(err, hostrelocate.ErrNotYet) { |
| - r.alert(host, "worker", err) |
| - } |
| + stepped = true |
| + _, err := r.Worker.Step(ctx, host) |
| + r.result(host, "worker", notYetIsNoError(err)) |
| } |
| } |
| - r.finishCompletion(ctx, store, host, box) |
| + if stepped { |
| + // A step may have just made the serving CAS (ruling 49). |
| + obs, err = hostclaim.Read(ctx, store, host) |
| + if r.result(host, "read", err); err != nil { |
| + continue |
| + } |
| + } |
| + r.finishCompletion(ctx, host, obs, &self) |
| } |
| + r.alerts.keep(read) |
| // Ruling 61: journals of operations that are over go, once per pass, |
| // after the pass's own steps and never beside a journal write (carry 3: |
| // sweepRelocationJournals holds relocationJournals exclusively). |
| if r.DataDir != "" { |
| - if err := sweepRelocationJournals(ctx, r.DataDir, store); err != nil { |
| - log.Printf("relocate: sweep journals: %v", err) |
| - } |
| + r.result("", "sweep journals", sweepRelocationJournals(ctx, r.DataDir, store)) |
| } |
| } |
| +func notYetIsNoError(err error) error { |
| + if errors.Is(err, hostrelocate.ErrNotYet) { |
| + return nil |
| + } |
| + return err |
| +} |
| + |
| // tryUpgrade replaces a completion-only stand-in with the full pass once |
| // it builds (upgrade). A build that answers no relocation owners ends the |
| // tries: completions are then all this box runs, as NewRelocationRuntime |
| -// would have answered at startup. |
| +// answers when the owners are not bound. |
| func (r *RelocationRuntime) tryUpgrade(ctx context.Context) { |
| if r.upgrade == nil { |
| return |
| } |
| full, err := r.upgrade(ctx) |
| - if err != nil { |
| - r.alert("", "region HA: relocation pass not built, finishing completions only", err) |
| + if r.result("", "region HA: relocation pass not built, finishing completions only", err); err != nil { |
| return |
| } |
| r.upgrade = nil |
| if full == nil || full.Worker == nil { |
| + log.Printf("agent: region HA: the relocation owners are not bound, so this box finishes completions only") |
| return |
| } |
| - r.Worker, r.Participant, r.Box, r.Convert, r.Finish, r.Hosts, r.Self, r.DataDir = full.Worker, full.Participant, full.Box, full.Convert, full.Finish, full.Hosts, full.Self, full.DataDir |
| + r.Worker, r.Participant, r.Box, r.Convert, r.Finish, r.Hosts, r.Self, r.DataDir, r.config = full.Worker, full.Participant, full.Box, full.Convert, full.Finish, full.Hosts, full.Self, full.DataDir, full.config |
| log.Printf("agent: region HA: relocation pass started") |
| } |
| @@ -487,51 +512,68 @@ func (r *RelocationRuntime) hosts(ctx context.Context, store release.Store) ([]s |
| return out, nil |
| } |
| +// passSelf is this box as one pass knows it: asked at most once. |
| +type passSelf struct { |
| + asked bool |
| + box string |
| +} |
| + |
| +// thisBox is Box, or Self's answer, asked once a pass and only by a |
| +// pending completion (review M2). No answer is no box. |
| +func (r *RelocationRuntime) thisBox(ctx context.Context, s *passSelf) string { |
| + if r.Box != "" { |
| + return r.Box |
| + } |
| + if !s.asked { |
| + s.asked = true |
| + err := errors.New("no identity to answer which box this is") |
| + if r.Self != nil { |
| + s.box, err = r.Self(ctx) |
| + } |
| + if r.result("", "this box", err); err != nil { |
| + s.box = "" |
| + } |
| + } |
| + return s.box |
| +} |
| + |
| // finishCompletion is ruling 80: a claim serving on this box whose |
| // completed relocation names this box and is not yet marked discharged is |
| // handed to Finish (the hold release, relocationReleaseHold) on every |
| -// pass, read afresh after this pass's steps so a serving CAS made just now |
| +// pass, from the pass's latest read of it, so a serving CAS made this pass |
| // is finished in the same pass. The work is found from the claim's record, |
| // never from the row: a row already released, a lift that never ran or a |
| // mark whose CAS failed is retried until the mark is written. |
| // |
| -// A claim that does not serve on this box is skipped before Finish is |
| -// called (carry 8). A record that cannot be read is alerted and retried, |
| -// never read as "no completion" (carry 1), and so is every refusal of |
| -// Finish, a probe-fence owner it cannot find among them (carry 6): the |
| -// record stays unmarked. ErrChanged from Finish is the claim moving on |
| -// under the pass: nothing to do now, and never "finished". |
| -func (r *RelocationRuntime) finishCompletion(ctx context.Context, store release.Store, host, box string) { |
| - if r.Finish == nil || box == "" { |
| - return |
| - } |
| - obs, err := hostclaim.Read(ctx, store, host) |
| - if err != nil { |
| - r.alert(host, "finish", err) |
| +// A claim with no undischarged completion asks nothing more, not even |
| +// which box this is (review M2). A claim that does not serve on this box is |
| +// skipped before Finish is called (carry 8). A record that cannot be read |
| +// on a claim serving here is said and retried, never read as "no |
| +// completion" (carry 1), and so is every refusal of Finish, a probe-fence |
| +// owner it cannot find (carry 6) and an ErrChanged after the serving-here |
| +// check (review M4) among them: the record stays unmarked. |
| +func (r *RelocationRuntime) finishCompletion(ctx context.Context, host string, obs hostclaim.Observed, self *passSelf) { |
| + if r.Finish == nil || obs.Absent || obs.Claim == nil || obs.Claim.Op.Kind != hostclaim.KindServing { |
| return |
| } |
| - if obs.Absent || obs.Claim == nil { |
| + c := obs.Claim |
| + done, ok, err := hostclaim.ReadCompletedRelocation(c) |
| + if err == nil && (!ok || done.Discharged) { |
| + r.alerts.clear(host, "finish") |
| return |
| } |
| - c := obs.Claim |
| - if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != box { |
| + box := r.thisBox(ctx, self) |
| + if box == "" || c.Op.Exec != box { |
| return |
| } |
| - done, ok, err := hostclaim.ReadCompletedRelocation(c) |
| if err != nil { |
| r.alert(host, "finish", err) |
| return |
| } |
| - if !ok || done.Discharged || done.Stop.Box != box { |
| + if done.Stop.Box != box { |
| return |
| } |
| - err = r.Finish(ctx, host, done.Op) |
| - if errors.Is(err, hostbox.ErrChanged) { |
| - return |
| - } |
| - if err != nil { |
| - r.alert(host, "finish", err) |
| - } |
| + r.result(host, "finish", r.Finish(ctx, host, done.Op)) |
| } |
| // convertFencedLanding is J3 through the worker's own authorization |
| @@ -551,16 +593,71 @@ func convertFencedLanding(w hostrelocate.Worker, store release.Store) func(conte |
| } |
| } |
| -func (r *RelocationRuntime) alert(host, who string, err error) { |
| - if _, said := r.alerted.LoadOrStore(host+"\x00"+err.Error(), struct{}{}); !said { |
| - if host == "" { |
| - log.Printf("relocate: %s: %v", who, err) |
| - return |
| +func (r *RelocationRuntime) alert(host, step string, err error) { r.alerts.say(host, step, err) } |
| + |
| +// result says err for host's step, or, with none, forgets the step's last |
| +// cause, so the cause is said again if it returns. |
| +func (r *RelocationRuntime) result(host, step string, err error) { |
| + if err != nil { |
| + r.alerts.say(host, step, err) |
| + return |
| + } |
| + r.alerts.clear(host, step) |
| +} |
| + |
| +// relocationAlerts is rulings 35 and 36's one alert per host, step and |
| +// cause (review M8): a cause is said when it is first seen for that host |
| +// and step, and again only after that step has succeeded since, or after |
| +// another cause. It holds one entry per host and step failing now; each |
| +// pass forgets the hosts it no longer reads, so it is bounded by them. |
| +type relocationAlerts struct { |
| + mu sync.Mutex |
| + last map[[2]string]string // {host, step} -> the cause last said |
| +} |
| + |
| +func (a *relocationAlerts) say(host, step string, err error) { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + if a.last == nil { |
| + a.last = map[[2]string]string{} |
| + } |
| + k := [2]string{host, step} |
| + if cause, said := a.last[k]; said && cause == err.Error() { |
| + return |
| + } |
| + a.last[k] = err.Error() |
| + if host == "" { |
| + log.Printf("relocate: %s: %v", step, err) |
| + return |
| + } |
| + log.Printf("relocate %s: %s: %v", host, step, err) |
| +} |
| + |
| +func (a *relocationAlerts) clear(host, step string) { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + delete(a.last, [2]string{host, step}) |
| +} |
| + |
| +// keep forgets every host's causes but those of hosts, and the box-wide |
| +// ones (host ""). |
| +func (a *relocationAlerts) keep(hosts map[string]bool) { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + for k := range a.last { |
| + if k[0] != "" && !hosts[k[0]] { |
| + delete(a.last, k) |
| } |
| - log.Printf("relocate %s: %s: %v", host, who, err) |
| } |
| } |
| +// size is the number of causes held (tests). |
| +func (a *relocationAlerts) size() int { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + return len(a.last) |
| +} |
| + |
| // registerRelocationSource registers region HA's owner source with the |
| // roster directory the claim's own owners read (claimproduction.go). |
| func registerRelocationSource(claimCtx context.Context, dataDir string) error { |
| @@ -582,28 +679,30 @@ func registerRelocationSource(claimCtx context.Context, dataDir string) error { |
| // startRelocationPass builds the relocation runtime from the claim entry's |
| // assembly and starts its loop. With claim mode on it always starts one: |
| -// the full pass with region HA on and the owners bound, otherwise the |
| -// completion-only pass (ruling 82). Nil with claim mode off, with no |
| -// activator, or when not even the completion pass can be built (logged). |
| -func startRelocationPass(agentCtx, claimCtx context.Context, claims *ClaimWiring, reg *registry.Registry, svc *activator.Service) *RelocationRuntime { |
| +// the completion-only pass (ruling 82), which with region HA on builds |
| +// the full pass on its first, bounded, pass (review M9) and becomes it |
| +// once the owners are bound. Nothing with claim mode off; with it on, |
| +// every way of not starting the pass is said. |
| +func startRelocationPass(agentCtx, claimCtx context.Context, claims *ClaimWiring, reg *registry.Registry, svc *activator.Service) { |
| getenv := claimEntryEnv(claimCtx) |
| if !hostwrite.Enabled(getenv) { |
| - return nil |
| + return |
| } |
| entry, _ := claimCtx.Value(claimEntryKey{}).(claimEntry) |
| // Claim mode is on, so a completion recorded on this box must be |
| - // finished here (ruling 82): every way of not starting the pass is said. |
| + // finished here (ruling 82). |
| if claims == nil || svc == nil || entry.Assembly == nil { |
| log.Printf("agent: relocation completion pass not started: claim mode is on but the claim wiring, the activator or the claim assembly is missing") |
| - return nil |
| + return |
| } |
| local, ok := entry.Assembly.Local.(*productionClaimLocal) |
| if !ok { |
| - log.Printf("agent: region HA: relocation pass not started: the assembly has no production participation owner") |
| - return nil |
| + log.Printf("agent: relocation completion pass not started: the assembly has no production participation owner") |
| + return |
| } |
| var full func(context.Context) (*RelocationRuntime, error) |
| - if regionHAEnabled(getenv) { |
| + ha := regionHAEnabled(getenv) |
| + if ha { |
| full = func(ctx context.Context) (*RelocationRuntime, error) { |
| in, err := productionRelocationRuntimeInputs(os.Getenv, release.OptDir()) |
| if err != nil { |
| @@ -612,37 +711,39 @@ func startRelocationPass(agentCtx, claimCtx context.Context, claims *ClaimWiring |
| return NewRelocationRuntime(ctx, getenv, entry.Store, local, claims.Providers, reg, svc, in) |
| } |
| } |
| - runtime, err := relocationPassFor(agentCtx, local, reg, svc, full) |
| + runtime, err := relocationPassFor(local, reg, svc, full) |
| if err != nil { |
| log.Printf("agent: relocation completion pass not started: %v", err) |
| - return nil |
| - } |
| - go runtime.Run(agentCtx, entry.Store, nil) |
| - return runtime |
| -} |
| - |
| -// relocationPassFor is the runtime the agent runs in claim mode: the full |
| -// pass when full (region HA on) builds it, otherwise the completion-only |
| -// one (ruling 82). A full build that fails is not "no relocation on this |
| -// box": the completion-only runtime stands in and tries the build again |
| -// every pass (upgrade) until it builds or answers that the owners are not |
| -// bound. |
| -func relocationPassFor(ctx context.Context, local *productionClaimLocal, reg *registry.Registry, act RelocationActivator, full func(context.Context) (*RelocationRuntime, error)) (*RelocationRuntime, error) { |
| - var failed error |
| - if full != nil { |
| - r, err := full(ctx) |
| - if err == nil && r != nil { |
| - return r, nil |
| - } |
| - failed = err |
| + return |
| } |
| + launchRelocationPass(agentCtx, runtime, entry.Store, ha) |
| +} |
| + |
| +// relocationPassFor is the runtime the agent runs in claim mode: the |
| +// completion-only one (ruling 82), standing in for the full pass while |
| +// full (region HA on) builds it. Nothing is built here: the build reads |
| +// the roster and the data bucket, so it runs inside the first pass, under |
| +// the pass's bound, never on the agent's startup path (review M9). A build |
| +// that fails is not "no relocation on this box": each pass tries it again. |
| +func relocationPassFor(local *productionClaimLocal, reg *registry.Registry, act RelocationActivator, full func(context.Context) (*RelocationRuntime, error)) (*RelocationRuntime, error) { |
| r, err := newCompletionRuntime(local, reg, act) |
| if err != nil { |
| return nil, err |
| } |
| - if failed != nil { |
| - r.alert("", "region HA: relocation pass not built, finishing completions only", failed) |
| - r.upgrade = full |
| - } |
| + r.upgrade = full |
| return r, nil |
| } |
| + |
| +// launchRelocationPass says the pass started and runs it. Everything it |
| +// says is decided before the loop starts: the loop owns the runtime's |
| +// fields from then on, and a pass may replace them (tryUpgrade; review |
| +// M3). The channel closes when the loop has returned. |
| +func launchRelocationPass(ctx context.Context, r *RelocationRuntime, store release.Store, regionHA bool) <-chan struct{} { |
| + log.Printf("agent: relocation pass started (completions only: %v)", !regionHA) |
| + done := make(chan struct{}) |
| + go func() { |
| + defer close(done) |
| + r.Run(ctx, store, nil) |
| + }() |
| + return done |
| +} |
diff --git a/internal/host/relocatepass_absence_test.go b/internal/host/relocatepass_absence_test.go| index b94ebc7e6..868c79efb 100644 |
| --- a/internal/host/relocatepass_absence_test.go |
| +++ b/internal/host/relocatepass_absence_test.go |
| @@ -82,10 +82,15 @@ func TestRelocationPassBuildsTheFullPassOnceItCan(t *testing.T) { |
| return &RelocationRuntime{Worker: stepRecorder{name: "worker", order: &order}, Participant: stepRecorder{name: "participant", order: &order}, Box: "dst", |
| Finish: func(context.Context, string, string) error { return nil }}, nil |
| } |
| - r, err := relocationPassFor(ctx, local, reg, &activator.Service{}, full) |
| + r, err := relocationPassFor(local, reg, &activator.Service{}, full) |
| if err != nil || r == nil || r.Worker != nil || r.Finish == nil || r.Hosts == nil { |
| t.Fatalf("not the completion-only stand-in: %+v %v", r, err) |
| } |
| + // Review M9: nothing is built on the startup path. |
| + if builds != 0 { |
| + t.Fatalf("built %d times before the first pass", builds) |
| + } |
| + r.once(ctx, store) // the first build fails |
| r.once(ctx, store) // the second build fails too |
| if len(order) != 0 || builds != 2 { |
| t.Fatalf("stepped %v after %d builds", order, builds) |
| @@ -104,7 +109,7 @@ func TestRelocationPassBuildsTheFullPassOnceItCan(t *testing.T) { |
| // Owners not bound: the build answers the completion-only runtime, and |
| // the tries end. |
| builds = 0 |
| - r, err = relocationPassFor(ctx, local, reg, &activator.Service{}, func(context.Context) (*RelocationRuntime, error) { |
| + r, err = relocationPassFor(local, reg, &activator.Service{}, func(context.Context) (*RelocationRuntime, error) { |
| builds++ |
| if builds == 1 { |
| return nil, unreadable |
| @@ -126,15 +131,14 @@ func TestRelocationPassBuildsTheFullPassOnceItCan(t *testing.T) { |
| func TestStartRelocationPassSaysWhyItDidNotStart(t *testing.T) { |
| buf := captureRacyLog(t) |
| claimCtx := context.WithValue(t.Context(), claimEntryKey{}, claimEntry{Enabled: true, FlagRead: true}) |
| - if r := startRelocationPass(t.Context(), claimCtx, &ClaimWiring{}, nil, &activator.Service{}); r != nil { |
| - t.Fatalf("started over no assembly: %+v", r) |
| - } |
| + startRelocationPass(t.Context(), claimCtx, &ClaimWiring{}, nil, &activator.Service{}) |
| if !strings.Contains(buf.String(), "relocation completion pass not started") { |
| t.Fatalf("not said:\n%s", buf.String()) |
| } |
| off := context.WithValue(t.Context(), claimEntryKey{}, claimEntry{FlagRead: true}) |
| before := buf.String() |
| - if r := startRelocationPass(t.Context(), off, &ClaimWiring{}, nil, &activator.Service{}); r != nil || buf.String() != before { |
| - t.Fatalf("claim mode off: %+v %q", r, strings.TrimPrefix(buf.String(), before)) |
| + startRelocationPass(t.Context(), off, &ClaimWiring{}, nil, &activator.Service{}) |
| + if buf.String() != before { |
| + t.Fatalf("claim mode off: %q", strings.TrimPrefix(buf.String(), before)) |
| } |
| } |
diff --git a/internal/host/relocatepass_carries_test.go b/internal/host/relocatepass_carries_test.go| index 962afc87d..a92c519a7 100644 |
| --- a/internal/host/relocatepass_carries_test.go |
| +++ b/internal/host/relocatepass_carries_test.go |
| @@ -83,11 +83,12 @@ func TestCompletionPassFinishesOnlyWhatServesOnThisBox(t *testing.T) { |
| } |
| } |
| -// Carry 8: ErrChanged from the hold release is the claim moving on under |
| -// the pass: nothing to do, no alert, and asked again next pass, never |
| -// taken as finished. Every other refusal is alerted once and retried |
| -// (carry 6: a probe-fence owner Finish cannot find is ErrRefused). |
| -func TestCompletionPassRetriesARefusedFinishAndIgnoresAChangedClaim(t *testing.T) { |
| +// Carry 8 and review M4: ErrChanged from the hold release, after the |
| +// pass's own serving-here check passed, is never taken as finished: it is |
| +// said once (a completion stuck that way keeps the barrier closed) and |
| +// asked again next pass. Every other refusal is alerted once and retried |
| +// too (carry 6: a probe-fence owner Finish cannot find is ErrRefused). |
| +func TestCompletionPassRetriesARefusedFinishAndAChangedClaim(t *testing.T) { |
| ctx := t.Context() |
| store := release.NewDirStore(t.TempDir()) |
| passClaim(t, store, "a.example", hostclaim.Op{}) |
| @@ -97,7 +98,7 @@ func TestCompletionPassRetriesARefusedFinishAndIgnoresAChangedClaim(t *testing.T |
| r := &RelocationRuntime{Finish: f.finish, Box: "dst"} |
| r.once(ctx, store) |
| r.once(ctx, store) |
| - if len(f.calls) != 2 || strings.Contains(buf.String(), "finish") { |
| + if len(f.calls) != 2 || strings.Count(buf.String(), "relocate a.example: finish: "+hostbox.ErrChanged.Error()) != 1 { |
| t.Fatalf("a changed claim: calls %v, log:\n%s", f.calls, buf.String()) |
| } |
| f.calls, f.err = nil, errors.Join(hostrelocate.ErrRefused, errors.New("a.example has no probe-fence owner")) |
| @@ -298,6 +299,10 @@ func TestRelocationRuntimeBuildsTheFullPassWithTheOwnersBound(t *testing.T) { |
| if !ok || r.Participant == nil || r.Convert == nil || r.Finish == nil || r.Box != "wrk" || r.DataDir != data || r.Hosts != nil { |
| t.Fatalf("not the full pass: %+v", r) |
| } |
| + // Review M7: the held-stage evidence reaches the full pass's config. |
| + if !bytes.Equal(r.config.ReleasePub, pub) || r.config.Ratchet == nil || r.config.Ratchet != pointerRatchet(false, in.OptDir, data, getenv) { |
| + t.Fatalf("the full pass's held-stage evidence: pub %v ratchet %p", r.config.ReleasePub, r.config.Ratchet) |
| + } |
| if worker.Self.Box != "wrk" || worker.Self.Enrollment != w.enrollment["wrk"] || worker.Region != "ie" { |
| t.Fatalf("the worker is not this box in its roster region: %+v %q", worker.Self, worker.Region) |
| } |
diff --git a/internal/host/relocatepass_round1_test.go b/internal/host/relocatepass_round1_test.go| new file mode 100644 |
| index 000000000..2d76244c6 |
| --- /dev/null |
| +++ b/internal/host/relocatepass_round1_test.go |
| @@ -0,0 +1,290 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "sync" |
| + "testing" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/hostbox" |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/hostguard" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// countingClaims counts the claim reads made through it, and runs onGet, |
| +// when set, before each. |
| +type countingClaims struct { |
| + release.Store |
| + mu sync.Mutex |
| + claims int |
| + onGet func(key string) |
| +} |
| + |
| +func (s *countingClaims) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if strings.HasPrefix(key, "control/hosts/") { |
| + s.mu.Lock() |
| + s.claims++ |
| + s.mu.Unlock() |
| + } |
| + if s.onGet != nil { |
| + s.onGet(key) |
| + } |
| + return s.Store.Get(ctx, key) |
| +} |
| + |
| +func (s *countingClaims) reads() int { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + n := s.claims |
| + s.claims = 0 |
| + return n |
| +} |
| + |
| +func passRegistry(t *testing.T, hosts ...string) *registry.Registry { |
| + t.Helper() |
| + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { reg.Close() }) |
| + for _, host := range hosts { |
| + if err := reg.Upsert(registry.Route{Host: host, Status: "ready", Socket: "/run/" + host + ".sock"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + return reg |
| +} |
| + |
| +// Review M2: the completion-only pass of a box with nothing to finish reads |
| +// each of its claims once and asks nothing more: no identity read, however |
| +// many rows. A pass that finds pending completions asks which box this is |
| +// once, however many it finds. |
| +func TestCompletionPassAsksNothingMoreWithNothingPending(t *testing.T) { |
| + ctx := t.Context() |
| + store := &countingClaims{Store: release.NewDirStore(t.TempDir())} |
| + for _, host := range []string{"a.example", "b.example", "c.example"} { |
| + passClaim(t, store.Store, host, hostclaim.Op{}) |
| + } |
| + completeOn(t, store.Store, "b.example", "rel-b", "dst", true) // finished already |
| + reg := passRegistry(t, "a.example", "b.example", "c.example") |
| + identities := 0 |
| + local := &productionClaimLocal{assembly: &ClaimAssembly{Claims: store, Guard: hostguard.Guard{DataDir: t.TempDir()}, stopLocal: func(context.Context) (claimStopLocal, error) { |
| + identities++ |
| + return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil |
| + }}} |
| + r, err := newCompletionRuntime(local, reg, &activator.Service{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + store.reads() |
| + r.once(ctx, store) |
| + if n := store.reads(); identities != 0 || n != 3 { |
| + t.Fatalf("nothing pending: %d identity reads, %d claim reads (want 0 and 3)", identities, n) |
| + } |
| + for _, host := range []string{"d.example", "e.example"} { |
| + passClaim(t, store.Store, host, hostclaim.Op{}) |
| + completeOn(t, store.Store, host, "rel-"+host[:1], "dst", false) |
| + if err := reg.Upsert(registry.Route{Host: host, Status: "ready", Socket: "/run/" + host + ".sock"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + f := &finishRecorder{} |
| + r.Finish = f.finish |
| + r.once(ctx, store) |
| + if n := store.reads(); identities != 1 || len(f.calls) != 2 || n != 5 { |
| + t.Fatalf("two pending: %d identity reads (want 1), finished %v, %d claim reads (want 5)", identities, f.calls, n) |
| + } |
| +} |
| + |
| +// Review M1: the completion-only pass reads the hosts its own op journals |
| +// name beside its registry rows, so a host with a journal and no row is |
| +// read; a journal it cannot read is said, and the rest are still read. |
| +func TestCompletionPassReadsTheHostsItsJournalsName(t *testing.T) { |
| + ctx := t.Context() |
| + store := release.NewDirStore(t.TempDir()) |
| + passClaim(t, store, "row.example", hostclaim.Op{}) |
| + passClaim(t, store, "rowless.example", hostclaim.Op{}) |
| + completeOn(t, store, "rowless.example", "rel-1", "dst", false) |
| + data := t.TempDir() |
| + marker, err := relocationProbeFencePath(data, "rel-1") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := relocationProbeFenceOwner(marker, "rowless.example", 1); err != nil { |
| + t.Fatal(err) |
| + } |
| + bad := filepath.Join(data, "relocate", "ops", "rel-2", "attempts.json") |
| + if err := os.MkdirAll(filepath.Dir(bad), 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(bad, []byte("not a journal"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg := passRegistry(t, "row.example") |
| + local := &productionClaimLocal{assembly: &ClaimAssembly{Claims: store, Guard: hostguard.Guard{DataDir: data}, stopLocal: func(context.Context) (claimStopLocal, error) { |
| + return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil |
| + }}} |
| + r, err := newCompletionRuntime(local, reg, &activator.Service{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + hosts, err := r.Hosts() |
| + if len(hosts) != 2 || hosts[0] != "row.example" || hosts[1] != "rowless.example" || err == nil { |
| + t.Fatalf("hosts %v, err %v: want the row, the journal's host and the unreadable journal", hosts, err) |
| + } |
| + buf := captureRacyLog(t) |
| + f := &finishRecorder{} |
| + r.Finish = f.finish |
| + r.once(ctx, store) |
| + if len(f.calls) != 1 || f.calls[0] != "rowless.example:rel-1" || !strings.Contains(buf.String(), "relocate: list hosts:") { |
| + t.Fatalf("finished %v, log:\n%s", f.calls, buf.String()) |
| + } |
| +} |
| + |
| +// Review M8: an alert is one per host, step and cause. Two steps failing |
| +// with one cause are both said; a cause that clears and returns is said |
| +// again; a host the pass no longer reads is forgotten. |
| +func TestRelocatePassAlertsPerStepAndForgetsClearedCauses(t *testing.T) { |
| + ctx := t.Context() |
| + store := release.NewDirStore(t.TempDir()) |
| + passClaim(t, store, "a.example", relocatingOp("rel-a")) |
| + same := errors.New("the roster could not be read") |
| + var order []string |
| + participant := &flakyStepper{name: "participant", order: &order, err: same} |
| + r := &RelocationRuntime{Worker: stepRecorder{name: "worker", order: &order, err: same}, Participant: participant, Box: "dst"} |
| + listed := []string{"a.example"} |
| + r.Hosts = func() ([]string, error) { return listed, nil } |
| + buf := captureRacyLog(t) |
| + r.once(ctx, store) |
| + r.once(ctx, store) |
| + if strings.Count(buf.String(), "relocate a.example: participant: "+same.Error()) != 1 || strings.Count(buf.String(), "relocate a.example: worker: "+same.Error()) != 1 { |
| + t.Fatalf("not one alert per step:\n%s", buf.String()) |
| + } |
| + participant.err = nil // the cause clears |
| + r.once(ctx, store) |
| + participant.err = same // and returns |
| + r.once(ctx, store) |
| + if n := strings.Count(buf.String(), "relocate a.example: participant: "+same.Error()); n != 2 { |
| + t.Fatalf("a returning cause said %d times, want twice:\n%s", n, buf.String()) |
| + } |
| + if r.alerts.size() != 2 { |
| + t.Fatalf("held %d causes, want 2", r.alerts.size()) |
| + } |
| + listed = nil |
| + r.once(ctx, store) |
| + if r.alerts.size() != 0 { |
| + t.Fatalf("held %d causes for hosts no longer read", r.alerts.size()) |
| + } |
| +} |
| + |
| +// flakyStepper is a stepRecorder whose error can change between passes. |
| +type flakyStepper struct { |
| + name string |
| + order *[]string |
| + err error |
| +} |
| + |
| +func (s *flakyStepper) Step(_ context.Context, host string) (hostclaim.Observed, error) { |
| + *s.order = append(*s.order, s.name+":"+host) |
| + return hostclaim.Observed{}, s.err |
| +} |
| + |
| +// Review M6: the sweep reads every claim before it takes the journals' |
| +// lock exclusively, so a wake's journal write never waits on a bucket |
| +// read; and what it decided unlocked it removes only as it finds it under |
| +// the lock. A directory that held no journal when it was read, and holds |
| +// one by the time the lock is held, is kept. |
| +func TestSweepReadsClaimsUnlockedAndChecksAgainUnderTheLock(t *testing.T) { |
| + ctx := t.Context() |
| + data := t.TempDir() |
| + ops := filepath.Join(data, "relocate", "ops") |
| + for _, dir := range []string{"aaa", "bbb"} { |
| + if err := os.MkdirAll(filepath.Join(ops, dir), 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + // aaa: a write's temporary only. bbb: b.example's journal, its claim gone. |
| + if err := os.WriteFile(filepath.Join(ops, "aaa", ".attempts.json.1"), []byte("{"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(filepath.Join(ops, "bbb", "install.json"), []byte(`{"v":1,"host":"b.example"}`), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + locked := false |
| + store := &countingClaims{Store: release.NewDirStore(t.TempDir())} |
| + store.onGet = func(key string) { |
| + if relocationJournals.TryRLock() { |
| + relocationJournals.RUnlock() |
| + } else { |
| + locked = true |
| + } |
| + // A write lands in aaa after the sweep read it as empty. |
| + if key == "control/hosts/b.example.json" { |
| + if err := os.WriteFile(filepath.Join(ops, "aaa", "attempts.json"), []byte(`{"v":1,"host":"a.example","op":"aaa"}`), 0o600); err != nil { |
| + t.Error(err) |
| + } |
| + } |
| + } |
| + if err := sweepRelocationJournals(ctx, data, store); err != nil { |
| + t.Fatal(err) |
| + } |
| + if locked { |
| + t.Fatal("a claim was read under the journals' exclusive lock") |
| + } |
| + if store.reads() == 0 { |
| + t.Fatal("the sweep read no claim") |
| + } |
| + if _, err := os.Stat(filepath.Join(ops, "bbb")); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("the ended op's journals survived: %v", err) |
| + } |
| + if _, err := os.Stat(filepath.Join(ops, "aaa", "attempts.json")); err != nil { |
| + t.Fatalf("a journal committed after the decision was swept: %v", err) |
| + } |
| +} |
| + |
| +// closingStepper closes stepped on its first Step. |
| +type closingStepper struct { |
| + once *sync.Once |
| + stepped chan struct{} |
| +} |
| + |
| +func (s closingStepper) Step(context.Context, string) (hostclaim.Observed, error) { |
| + s.once.Do(func() { close(s.stepped) }) |
| + return hostclaim.Observed{}, nil |
| +} |
| + |
| +// Review M3 (run with -race): the pass's loop owns the runtime's fields |
| +// once it starts, and its first pass replaces them with the full pass's |
| +// (tryUpgrade). The launch says what it started from what it decided |
| +// before the loop, never by reading the runtime after it. |
| +func TestLaunchRelocationPassReadsNothingTheLoopOwns(t *testing.T) { |
| + store := release.NewDirStore(t.TempDir()) |
| + passClaim(t, store, "a.example", relocatingOp("rel-a")) |
| + reg := passRegistry(t) |
| + local := &productionClaimLocal{assembly: &ClaimAssembly{Claims: store, Guard: hostguard.Guard{DataDir: t.TempDir()}, stopLocal: func(context.Context) (claimStopLocal, error) { |
| + return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil |
| + }}} |
| + stepped := make(chan struct{}) |
| + step := closingStepper{once: &sync.Once{}, stepped: stepped} |
| + r, err := relocationPassFor(local, reg, &activator.Service{}, func(context.Context) (*RelocationRuntime, error) { |
| + return &RelocationRuntime{Worker: step, Participant: step, Box: "dst", Finish: func(context.Context, string, string) error { return nil }}, nil |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + buf := captureRacyLog(t) |
| + ctx, cancel := context.WithCancel(t.Context()) |
| + done := launchRelocationPass(ctx, r, store, true) |
| + <-stepped |
| + cancel() |
| + <-done |
| + if !strings.Contains(buf.String(), "relocation pass started (completions only: false)") || !strings.Contains(buf.String(), "agent: region HA: relocation pass started") { |
| + t.Fatalf("not said:\n%s", buf.String()) |
| + } |
| +} |
diff --git a/internal/host/relocationfinish_unix_test.go b/internal/host/relocationfinish_unix_test.go| index fbb7345dc..93b550ab2 100644 |
| --- a/internal/host/relocationfinish_unix_test.go |
| +++ b/internal/host/relocationfinish_unix_test.go |
| @@ -278,3 +278,33 @@ func TestRelocatePassRetriesAMissingProbeFenceOwner(t *testing.T) { |
| w.pass() |
| w.done(1) |
| } |
| + |
| +// Review M1: a completion whose row is gone is still visited by the |
| +// completion-only pass (its probe-fence journal names the host), and the |
| +// hold release's refusal is said once; nothing is released or marked. |
| +func TestCompletionPassSaysACompletionWhoseRowIsGone(t *testing.T) { |
| + w := newFinishWorld(t) |
| + if err := w.reg.Delete(finishHost); err != nil { |
| + t.Fatal(err) |
| + } |
| + claimOnly := func(k string) string { |
| + if k == "CARLOS_HOST_CLAIM" { |
| + return "1" |
| + } |
| + return "" |
| + } |
| + r, err := NewRelocationRuntime(t.Context(), claimOnly, w.store, w.productionLocal(), ClaimProviders{}, w.reg, w.act, relocationRuntimeInputs{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + w.runtime = r |
| + buf := captureRacyLog(t) |
| + w.pass() |
| + w.pass() |
| + if n := strings.Count(buf.String(), "has no row on this box"); n != 1 { |
| + t.Fatalf("said %d times, want once:\n%s", n, buf.String()) |
| + } |
| + if ok, _ := w.marked(); ok || !w.act.svc.Fenced(finishHost) { |
| + t.Fatalf("finished over a missing row: marked=%v fenced=%v", ok, w.act.svc.Fenced(finishHost)) |
| + } |
| +} |
diff --git a/internal/host/relocationinstall_lock_linux_test.go b/internal/host/relocationinstall_lock_linux_test.go| new file mode 100644 |
| index 000000000..32ae3c095 |
| --- /dev/null |
| +++ b/internal/host/relocationinstall_lock_linux_test.go |
| @@ -0,0 +1,47 @@ |
| +//go:build linux |
| + |
| +package host |
| + |
| +import ( |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| +) |
| + |
| +// Carry 3, the install journal (Task 10 review M7): the local install |
| +// writes install.json under relocationJournals shared, so it waits while a |
| +// sweep holds the lock, and completes once the sweep lets go. |
| +func TestRelocationInstallJournalWaitsForTheSweep(t *testing.T) { |
| + f := newInstallFixture(t, nil) |
| + relocationJournals.Lock() |
| + locked := true |
| + defer func() { |
| + if locked { |
| + relocationJournals.Unlock() |
| + } |
| + }() |
| + type result struct { |
| + l hostclaim.LocalInstall |
| + err error |
| + } |
| + done := make(chan result, 1) |
| + go func() { |
| + l, err := f.install() |
| + done <- result{l, err} |
| + }() |
| + select { |
| + case r := <-done: |
| + t.Fatalf("installed beside the sweep: %+v %v", r.l, r.err) |
| + case <-time.After(500 * time.Millisecond): |
| + } |
| + relocationJournals.Unlock() |
| + locked = false |
| + if r := <-done; r.err != nil { |
| + t.Fatal(r.err) |
| + } |
| + path, _ := relocationInstallPath(f.r.DataDir, "rel-1") |
| + if _, err := relocationJournalHost(path); err != nil { |
| + t.Fatalf("the install journal: %v", err) |
| + } |
| +} |
diff --git a/internal/host/relocationsweep.go b/internal/host/relocationsweep.go| index b4ece1790..4bc760743 100644 |
| --- a/internal/host/relocationsweep.go |
| +++ b/internal/host/relocationsweep.go |
| @@ -48,14 +48,22 @@ func sayRelocationStray(path string) { |
| // release/, a build directory left empty. An entry the sweep does not know |
| // is left in place and logged once. |
| // |
| -// It holds relocationJournals exclusively for the whole sweep, so it never |
| -// runs beside a relocation journal write in this process (carry 3): the |
| -// pass's own steps write in turn, but the wake's probe attempts write |
| -// attempts.json whenever a probe runs. An ops directory with no committed |
| -// journal, and a temporary, are only ever a finished write's leftovers then. |
| +// It runs in two halves (Task 10 review M6). The first reads the |
| +// directories and the claims with no lock held, and decides. The second |
| +// holds relocationJournals exclusively (carry 3), so it never runs beside |
| +// a relocation journal write in this process (the pass's own steps write |
| +// in turn, the wake's probe attempts write attempts.json whenever a probe |
| +// runs), and removes only what it finds, read again under the lock, as it |
| +// was decided: the same host's journals for an op that is over, or still |
| +// no committed journal at all. A claim's "the op is over" never reverts |
| +// (op ids are never reused), so a decision read before the lock stands; |
| +// the local files are what a write can change, and they are read again. |
| +// |
| +// The build's scratch (relocationbuild.go, MkdirAll then BuildGeneration) |
| +// is written outside the lock, by Participant.Step on the pass's own |
| +// goroutine, so it is serialized with the sweep by program order. A build |
| +// moved off that goroutine would need the lock too. |
| func sweepRelocationJournals(ctx context.Context, dataDir string, claims release.Store) error { |
| - relocationJournals.Lock() |
| - defer relocationJournals.Unlock() |
| root := filepath.Join(dataDir, "relocate") |
| var errs []error |
| gone := func(host, op string) (bool, error) { |
| @@ -80,26 +88,14 @@ func sweepRelocationJournals(ctx context.Context, dataDir string, claims release |
| } |
| return obs.Claim.Op.ID != op, nil |
| } |
| - hostOf := func(path string) (string, error) { |
| - raw, err := os.ReadFile(path) |
| - if err != nil { |
| - return "", err |
| - } |
| - var probe struct { |
| - Host string `json:"host"` |
| - Request struct{ Host string } |
| - } |
| - if err := json.Unmarshal(raw, &probe); err != nil { |
| - return "", fmt.Errorf("%s: %w", path, err) |
| - } |
| - if probe.Host != "" { |
| - return probe.Host, nil |
| - } |
| - if probe.Request.Host != "" { |
| - return probe.Request.Host, nil |
| - } |
| - return "", fmt.Errorf("%s names no host", path) |
| + // A removal decided unlocked: check reads the entry again under the |
| + // lock and answers whether it is still as decided. |
| + type removal struct { |
| + path string |
| + all bool // RemoveAll, not Remove |
| + check func() bool |
| } |
| + var removals []removal |
| // ops/<op>/{install,attempts,probe-fence}.json |
| ops, err := os.ReadDir(filepath.Join(root, "ops")) |
| if err != nil && !errors.Is(err, os.ErrNotExist) { |
| @@ -111,32 +107,27 @@ func sweepRelocationJournals(ctx context.Context, dataDir string, claims release |
| sayRelocationStray(dir) |
| continue |
| } |
| - host, found, readErr := "", false, error(nil) |
| - for _, name := range relocationOpJournals { |
| - h, err := hostOf(filepath.Join(dir, name)) |
| - if errors.Is(err, os.ErrNotExist) { |
| - continue |
| - } |
| - found = true |
| - if err == nil { |
| - host = h |
| - break |
| - } |
| - readErr = errors.Join(readErr, err) |
| - } |
| + host, found, readErr := opJournalHost(dir) |
| switch { |
| case host != "": |
| if done, err := gone(host, e.Name()); err != nil { |
| errs = append(errs, err) |
| } else if done { |
| - errs = append(errs, os.RemoveAll(dir)) |
| + removals = append(removals, removal{path: dir, all: true, check: func() bool { |
| + h, _, err := opJournalHost(dir) |
| + return err == nil && (h == host || h == "") |
| + }}) |
| } |
| case found: |
| errs = append(errs, readErr) // unreadable: kept and reported |
| default: |
| // No host-bearing journal was ever committed here: what is |
| - // left is a write's temporary, which no op can need. |
| - errs = append(errs, os.RemoveAll(dir)) |
| + // left is a write's temporary, which no op can need, unless a |
| + // write committed one since (read again under the lock). |
| + removals = append(removals, removal{path: dir, all: true, check: func() bool { |
| + _, found, _ := opJournalHost(dir) |
| + return !found |
| + }}) |
| } |
| } |
| // release/<op>.json, and an atomic write's leftover temporaries. |
| @@ -151,7 +142,7 @@ func sweepRelocationJournals(ctx context.Context, dataDir string, claims release |
| case e.IsDir(): |
| sayRelocationStray(path) |
| case isJSON: |
| - host, err := hostOf(path) |
| + host, err := relocationJournalHost(path) |
| if err != nil { |
| errs = append(errs, err) |
| continue |
| @@ -159,10 +150,15 @@ func sweepRelocationJournals(ctx context.Context, dataDir string, claims release |
| if done, err := gone(host, op); err != nil { |
| errs = append(errs, err) |
| } else if done { |
| - errs = append(errs, os.Remove(path)) |
| + removals = append(removals, removal{path: path, check: func() bool { |
| + h, err := relocationJournalHost(path) |
| + return err == nil && h == host |
| + }}) |
| } |
| case relocationWriteLeftover(e.Name()): |
| - errs = append(errs, os.Remove(path)) |
| + // Under the lock no write is in flight, so a temporary still |
| + // there is a leftover, never a write's. |
| + removals = append(removals, removal{path: path, check: func() bool { return true }}) |
| default: |
| sayRelocationStray(path) |
| } |
| @@ -179,34 +175,144 @@ func sweepRelocationJournals(ctx context.Context, dataDir string, claims release |
| sayRelocationStray(dir) |
| continue |
| } |
| - files, err := os.ReadDir(dir) |
| - if err != nil { |
| - errs = append(errs, err) |
| - continue |
| - } |
| - host := "" |
| - for _, f := range files { |
| - if h, ok := buildScratchHost(f.Name()); ok { |
| - host = h |
| - break |
| - } |
| - } |
| + host, empty, err := buildScratchDirHost(dir) |
| switch { |
| + case err != nil: |
| + errs = append(errs, err) |
| case host != "": |
| if done, err := gone(host, e.Name()); err != nil { |
| errs = append(errs, err) |
| } else if done { |
| - errs = append(errs, os.RemoveAll(dir)) |
| + removals = append(removals, removal{path: dir, all: true, check: func() bool { |
| + h, _, err := buildScratchDirHost(dir) |
| + return err == nil && (h == host || h == "") |
| + }}) |
| } |
| - case len(files) == 0: |
| - errs = append(errs, os.Remove(dir)) |
| + case empty: |
| + removals = append(removals, removal{path: dir, check: func() bool { |
| + _, empty, err := buildScratchDirHost(dir) |
| + return err == nil && empty |
| + }}) |
| default: |
| sayRelocationStray(dir) |
| } |
| } |
| + if len(removals) == 0 { |
| + return errors.Join(errs...) |
| + } |
| + relocationJournals.Lock() |
| + defer relocationJournals.Unlock() |
| + for _, r := range removals { |
| + if _, err := os.Lstat(r.path); errors.Is(err, os.ErrNotExist) { |
| + continue |
| + } |
| + if !r.check() { |
| + continue // a write landed since the decision: the next pass decides again |
| + } |
| + if r.all { |
| + errs = append(errs, os.RemoveAll(r.path)) |
| + } else { |
| + errs = append(errs, os.Remove(r.path)) |
| + } |
| + } |
| return errors.Join(errs...) |
| } |
| +// relocationJournalHost is the host a relocation journal names inside: |
| +// "host", or a stop request's "Request.Host" (release positions). |
| +func relocationJournalHost(path string) (string, error) { |
| + raw, err := os.ReadFile(path) |
| + if err != nil { |
| + return "", err |
| + } |
| + var probe struct { |
| + Host string `json:"host"` |
| + Request struct{ Host string } |
| + } |
| + if err := json.Unmarshal(raw, &probe); err != nil { |
| + return "", fmt.Errorf("%s: %w", path, err) |
| + } |
| + if probe.Host != "" { |
| + return probe.Host, nil |
| + } |
| + if probe.Request.Host != "" { |
| + return probe.Request.Host, nil |
| + } |
| + return "", fmt.Errorf("%s names no host", path) |
| +} |
| + |
| +// opJournalHost is the host an ops/<op> directory's journals name: the |
| +// first readable one, found when any journal is present at all, and the |
| +// read errors of the unreadable ones when none is readable. |
| +func opJournalHost(dir string) (host string, found bool, err error) { |
| + for _, name := range relocationOpJournals { |
| + h, rerr := relocationJournalHost(filepath.Join(dir, name)) |
| + if errors.Is(rerr, os.ErrNotExist) { |
| + continue |
| + } |
| + found = true |
| + if rerr == nil { |
| + return h, true, nil |
| + } |
| + err = errors.Join(err, rerr) |
| + } |
| + return "", found, err |
| +} |
| + |
| +// buildScratchDirHost is the host a build/<op> directory's scratch names, |
| +// and whether the directory is empty. |
| +func buildScratchDirHost(dir string) (string, bool, error) { |
| + files, err := os.ReadDir(dir) |
| + if err != nil { |
| + return "", false, err |
| + } |
| + for _, f := range files { |
| + if h, ok := buildScratchHost(f.Name()); ok { |
| + return h, false, nil |
| + } |
| + } |
| + return "", len(files) == 0, nil |
| +} |
| + |
| +// relocationJournalHosts is every host this box's op journals name |
| +// (ops/*/{probe-fence,attempts}.json): the completion-only pass's positive |
| +// local evidence of a relocation this box took part in, written before any |
| +// completion exists and kept until its mark (ruling 76; Task 10 review |
| +// M1). A journal that cannot be read is returned as an error beside the |
| +// hosts that could. |
| +func relocationJournalHosts(dataDir string) ([]string, error) { |
| + ops, err := os.ReadDir(filepath.Join(dataDir, "relocate", "ops")) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return nil, nil |
| + } |
| + if err != nil { |
| + return nil, err |
| + } |
| + seen := map[string]bool{} |
| + var out []string |
| + var errs []error |
| + for _, e := range ops { |
| + if !e.IsDir() { |
| + continue |
| + } |
| + for _, name := range []string{"probe-fence.json", "attempts.json"} { |
| + host, err := relocationJournalHost(filepath.Join(dataDir, "relocate", "ops", e.Name(), name)) |
| + if errors.Is(err, os.ErrNotExist) { |
| + continue |
| + } |
| + if err != nil { |
| + errs = append(errs, err) |
| + continue |
| + } |
| + if !seen[host] { |
| + seen[host] = true |
| + out = append(out, host) |
| + } |
| + } |
| + } |
| + return out, errors.Join(errs...) |
| +} |
| + |
| // relocationJournals serializes the journal sweep against every relocation |
| // journal write in this process (carry 3). A writer holds it shared from |
| // creating the journal's directory until its atomic rename has landed; the |