amadan

carlos / platform Public

Region HA plan 1b-next rev 3: Astra round 2 resolved (7 partials closed, 7 new fixed, none declined)

Probe attempts counted in the claim before they are made; discharge
completes discovery like a retirement; the fenced former writer's last
session retained; the pinned manifest's bytes retained by the pinner so a
late source PUT moves nothing; exhaustion checked before a spare is
chosen; the descriptor scan split at the privilege boundary (own /proc
in the agent, the complete scan root's, R3); flushed positions matched
by installation and session so a restarted agent asks root under the
recorded identity; controllers finished when a participation ends with
a cleanup scoped to the relocation's own work; the completed hold
cleared on the serving executor; the prepared flag only for P's opening
and an empty retry handled; the Task 9 adapter test separated into
"nothing minted" and "an unexplained row kept". Eleven new guards
removed in turn; un-gated tasks re-extracted and dry-run on 3407d114.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 42b1baadb33f193a09c4f87d3ec11933ac46df79 parent 47401a3
1 file changed, +1068 −271
  • docs/superpowers/plans/2026-10-07-region-ha-1b-next-relocation.md +1068 −271
diff --git a/docs/superpowers/plans/2026-10-07-region-ha-1b-next-relocation.md b/docs/superpowers/plans/2026-10-07-region-ha-1b-next-relocation.md
index f4693f9f9..bc3c662d9 100644
--- a/docs/superpowers/plans/2026-10-07-region-ha-1b-next-relocation.md
+++ b/docs/superpowers/plans/2026-10-07-region-ha-1b-next-relocation.md
@@ -1,4 +1,4 @@
-# Region HA 1b-next: Relocation, Row Cases, Create Choice and Fences Implementation Plan (rev 2)
+# Region HA 1b-next: Relocation, Row Cases, Create Choice and Fences Implementation Plan (rev 3)
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
@@ -10,6 +10,29 @@
**Spec:** `docs/superpowers/specs/2026-10-04-region-ha-box-placement-design.md`, rev 5.4 with §24 (rulings H1 to H19 final). Read §0, §2, §5.2, §5.3, §5.8, §5.9, §5.10, §9 (L1 to L20, accepted) and §24 first. Then the claim spec rev 6.30 on `origin/design-replica-ownership` (`docs/superpowers/specs/2026-10-02-host-lifecycle-claim-design.md`): §1.1's activation gate, §4 ("Same-region relocation (L1)" and the relocation exception to CC11), §5.1 and P5-D2/P5-D6 (the create intent), §6.1 ("Release before full stop (L10)"), §6.4a item 2 (`released`), §7.3 D2 ("Automatic termination (L2, L4)", "The fence's journal reference") and D4 ("Automatic loss persistence"). Then moves rev 7.22 on `origin/design-moves-any-region` §7.6 ("One install routine, two sources", L16 and L17) and §7.10 (the pre-flip conversion). The contracts note `docs/superpowers/notes/2026-10-07-region-ha-contracts.md` gives `hostfiles` v2, `CopyManifest`, `Publish` and `boxchoice`. Plan 1b (`docs/superpowers/plans/2026-10-06-region-ha-1b-claim-readers.md`) lists in "Waits for the claim" what this plan now builds.
+## Rev 3: round 2 dispositions
+
+Astra's round 2 (`CODEX-PLAN-REGION-HA-1BN-R2.md`, against `ed992c46`) left seven round 1 findings partly resolved and added seven. All fourteen are accepted; none is declined. Every un-gated change was dry-run again on a clean 3407d114 (build, vet for darwin and windows, cross-builds, targeted tests) and each new guard removed in turn (11 removals, each failing its test; listed in the Self-review).
+
+| # | Disposition | What changed | Where |
+|---|---|---|---|
+| R1-2 | accepted | The build's authority is checked at every write, not once: `authorityStore` wraps the versioned store so each `PutIfAbsent` and `CopyVersion` re-reads the claim and the host-guard record (`hostguard.Capture` at `install_stamp`) first; the lease's loss cancels the rest. | Task 8b |
+| R1-4 | accepted | `AutomaticLoss` retains the holder's incarnation and stamp, the box's last writer session whether or not it is still the writer (its last admission record on the holder, `LastAdmission`), so a source fenced after the target's admission keeps its session and consequences. | Task 1 (`TestRelocationLossRetainsAFormerWritersSession`); R5 |
+| R1-5 | accepted | The pinner retains the exact manifest bytes in the control bucket (`RetainedManifestKey`, create-only) before the pin; `CopyPinnedManifest` writes P from that copy, validated by digest, sequence and origins (`hostfiles.ValidateManifest`), and never reads the source again. A late source PUT changes nothing. | Task 5 (`TestCopyPinnedManifestEnforcesThePin`), ruling 21 |
+| R1-9 | accepted | `ReleasePositions.Recorded` matches the installation and session, not the controller, and returns the request the flush was made under; after a restart `Release` asks root under that identity (so root finds the stop's durable receipt), offers no drain of its own, and pairs the replayed receipt with the recorded positions. A stop root never completed cannot be finished by a new process: the release deadline makes it stuck. | Task 7 (`TestRelocationReleaseReplaysOnlyRecordedPositions/after_a_restart…`, `TestReleasePositionsFileBindsTheExactStop`) |
+| R1-13 | accepted | A later probe attempt whose predecessor generation holds no LTX file (`genselect.ErrEmpty`) is an ordinary empty cold admission: `Empty` set, no baseline, 6b's own empty authority. | Task 8c |
+| R1-14 | accepted | See new finding 2. | Task 9 |
+| R1-16 | accepted | Exhaustion is checked before a replacement is looked for, in both paths: with the replacements used the operation is stuck now, never retried on "no eligible box". | Task 5 (`TestWorkerMarksExhaustionStuckBeforeLookingForASpare`) |
+| N1 | accepted | `Prepared` is set only by the two branches that open P; the retry branch resets `Prepared` and `Empty` and selects its baseline as an ordinary cold wake, so its marker follows the first durable root. | Task 8c |
+| N2 | accepted | The adapter test separates "no row to mint" (a box with no role and no row: `ErrChanged`, nothing written) from "an existing unexplained row to keep" (held, never served, never deleted), and asserts `hold_op`, not a status the implementation does not write. | Task 9 |
+| N3 | accepted | `ProcessDescriptors` reads only `/proc/<pid>` (its own process; another process's unreadable table does not matter); the complete scan is root's: `RelocationEffects.Observe` is a provider, production asks root over the instance-control socket (`CollectWritersStopped`, R3's server half), and an agent-side scan fails closed on the first unreadable table. | Task 4 (`TestProcessDescriptorsNamesOneProcessesHolders`), Task 7, Task 8a, R3 |
+| N4 | accepted | A `RowReady` row whose `hold_op` names an operation the claim no longer runs has its hold cleared under the activation's verified ownership (`SetClaimHold(owner, old, "")`); a hold of the running operation is never touched. | Task 9 (`TestReconcileAdapterClearsTheHoldAfterCompletion`), ruling 20 |
+| N5 | accepted | `Effects.Retained` and `Finish`; `Participant.Sweep` finishes every retained controller whose operation the claim no longer runs with this box as a participant, once per pass; the relocation controller's `Cleanup` hook is scoped to the relocation's own partial work and is a no-op once this box has admitted under the operation or serves the host, so finishing never stops the serving writer. | Task 6 (`TestAParticipantFinishesItsControllerOnceTheOperationIsOver`), Task 7 (`Finish` in `TestRelocationAcknowledgeRetainsTheControllerUnderTheAgentLifetime`), Task 8a, Task 10; ruling 20 |
+| N6 | accepted | `MaxProbeAttempts = 3`; `BeginProbeAttempt` counts the attempt in the claim before it is made (per box and operation, so a replacement starts at zero); the participant enters `probe-failed` on the fourth. Durable across restarts and lost responses. | Task 1 (`TestRelocationProbeAttemptsAreBounded`), Task 6 (`TestAParticipantGivesUpAfterThreeProbeAttempts`); ruling 19 |
+| N7 | accepted | A qualified discharge completes each `park-local` with `CompleteObligation`'s own semantics, so a discovery operation's final chore marks discovery complete exactly as a retirement would. | Task 1 (`TestRelocationDischargeCompletesDiscoveryLikeARetirement`) |
+
+Astra's readings of rulings 1 to 4, 6 to 8 and 10 to 17 stand; rulings 5, 9 and 18 are completed by R1-5, N6 and R1-4 above.
+
## Rev 2: round 1 dispositions
Astra's round 1 (`CODEX-PLAN-REGION-HA-1BN-R1.md`, against `eb5e9051`) returned sixteen findings, "not ready". Each was checked against the tree at 3407d114 and the cited plans before anything changed. Fifteen are accepted in full and one in part (8: the un-gated half is built, the production body is a request to the claim session). None is declined. Every un-gated change below was dry-run again: applied to a clean 3407d114, built, vetted for darwin and windows, cross-built, tested, and each new guard removed in turn (25 removals, each failing its test; listed in the Self-review).
@@ -43,9 +66,9 @@ Interfaces this plan needs that no merged or declared plan supplies. Each gated
|---|---|---|---|
| R1 | claim 6b | `ClaimWakeAssembly.Build`'s runtime refuses every probe and every non-`serving` claim with a `ClaimDependencyError` naming moves C1. Region HA supplies `host.RelocationWakeInputs` (Task 8c) for `relocating/probing` where `op.exec` is this box. Ask: dispatch by `Op.Kind` to that provider in `Prerequisites`, `Admit` and `Prepare`, allowing `p.Probe` there, and expose one probe attempt as `(*ClaimWakeAssembly).Probe(ctx, obs, StartPolicy) (session uint64, passed bool, err error)` returning the attempt's exact health. | Task 8c |
| R2 | claim 6b | A writers-only stop on the admitted attempt, `(*claimWakeAttempt).StopWriters(ctx) (ClaimWritersDrained, error)`: close admission, settle spawn reservations, cancel and join the app's role launch, drain the schedule runner's in-flight deliveries for the host, wait for the host's cgroup to empty; keep the replicator and the files publisher open; idempotent once stopped; report the three drained facts, the cgroup and the database paths. `ClaimProducers.DrainClaim` stays the full drain. | Task 8a (finding 8) |
-| R3 | claim 6b | Record the replicator's descriptors when it opens: `stopevidence.ProcessDescriptors(ctx, "/proc", os.Getpid(), dbPaths)` immediately before and after `held.Replicate`, the difference kept on the attempt as `replicatorFDs`, with `claimStopLocal.ProcessStart`. L10 needs positively identified descriptors; nothing else can supply them. | Task 8a |
+| R3 | claim 6b | Two halves of L10's observation, split at the privilege boundary (rev 3). Agent side: record the replicator's descriptors when it opens, `stopevidence.ProcessDescriptors(ctx, "/proc", os.Getpid(), dbPaths)` immediately before and after `held.Replicate` (it reads only `/proc/self`), the difference kept on the attempt as `replicatorFDs` with `claimStopLocal.ProcessStart`. Root side: an authenticated instance-control endpoint `POST /claim/writers-stopped` taking Task 8a's `WritersStoppedRequest` (host, drained facts, database paths, replicator pid, start and descriptors) from the carlos uid, running `stopevidence.WritersStopped` with root's cgroup manager and the complete `/proc` scan, and returning the evidence. The agent cannot read other processes' descriptor tables (`carlos-edge.service`: uid carlos, `CAP_NET_BIND_SERVICE` only); an agent-side scan fails closed. | Task 8a |
| R4 | claim Plan 5 | The positive no-process witness for an installation with `writer_session = 0` (6b's `collectCapturedStop` already names it). Until it lands a drain of a never-woken host is refused at `Start` (ruling 11); a failover relocates it. | Task 5's ruling 11 |
-| R5 | claim spec, console | Adopt `evidence_refs` as the claim field for D3's "enrollment-qualified fence references": this plan's `hostclaim.AutomaticLoss` shape, one per (operation, box, fence), with the loss id in `loss_refs`. The console's claim-recovery pass materializes each into `control/host-loss/<id>.json` (D4's automatic variant) and sets `reported`; until then the claim keeps it. | After this plan |
+| R5 | claim spec, console | Adopt `evidence_refs` as the claim field for D3's "enrollment-qualified fence references": this plan's `hostclaim.AutomaticLoss` shape, one per (operation, box, fence), with the loss id in `loss_refs`, carrying the fence reference and observation time, the holder's incarnation and stamp, the box's role, its last writer session and that session's admission record (coverage, declarations), the discharge verdict and timer knowledge (rev 3: complete retained consequences). The console's claim-recovery pass materializes each into `control/host-loss/<id>.json` (D4's automatic variant, `write_sources` from `last_admission`, timers `unknown` or `partial`) and sets `reported`; until then the claim keeps it. | After this plan |
| R6 | claim spec | What evidences an empty generation on a failover of a host whose writer session is above 0 and whose listing holds no LTX file. Ruling 15 refuses it (the operation waits for an operator); if the claim session wants automatic recovery there, name the evidence. | ruling 15 |
| R7 | moves Plan 2 | Task 8's `moves.InstallManifestKey`, `SnapshotKey`, `AnchorKey` as declared, and `BuildInputs.Source` taking the pinned generation's full keys as `genselect` lists them (level and TXID range parsed from the key's tail by `SealedClient`). | Task 8b |
@@ -70,6 +93,10 @@ Each was a conflict between the authorities or a gap in them. Each says what it
15. **An empty pin needs evidence a writer can give** (rev 2; claim §6.4a "Empty bootstrap": no LTX file exists until the first commit, so a listing that finds nothing is a definite observation but says nothing about who wrote). A pin is `Empty` only when the journaled release reports `final_txid` 0 (the stopped writer flushed and found nothing), or the holder never admitted a writer (session 0, failover). A failover of a host whose session is above 0 with an empty listing refuses and waits for an operator; `JournalPin` enforces it, the pinner never writes it. *Cost if wrong:* that one case recovers by hand until R6 names its evidence.
16. **A pin is bound to the claim generation it was selected against** (rev 2; B2-3). Its key is `control/relocations/<host>/<op>/pin-<gen>.json` and `JournalPin` requires `gen` to be the observed claim's. A pin whose journal CAS lost is abandoned where it lies (never referenced, cheap, reaped by the storage follow-up of ruling 5); the next pass selects afresh at the new generation. Every bound `JournalPin` checks (the release's positions, the failover window) is checked before the object is written, so a doomed pin is never published. *Cost if wrong:* abandoned objects under a prefix nothing reads.
17. **Reconcile has three authorities for a relocation row** (rev 2; §5.3, 6b's `ClaimOwnership.Valid` needs a stamp). The executor (the source until its fence closes, the target from `installing` on) activates its own installation stamp and mints or keeps its row through the guarded path. A kept row of a box that is no longer the executor (the source after its release, a replaced target) gets its hold through `registry.SetClaimHold` under the flock, conditional on the row's recorded ownership: a hold grants nothing, so it needs no installation authority. A target before `installing` has no installation and mints nothing; the edges file already withholds the host from a box that is not `op.exec` (§24). *Cost if wrong:* a request that reaches a pre-installation target meets the box's unknown-host answer instead of a hold, for the seconds between `starting` and `installing`.
+19. **Probe attempts are counted before they are made** (rev 3; §5.8 "limits": at most three, then `probe-failed`). `BeginProbeAttempt` is one CAS per attempt, keyed by box, so a crash, a lost response or a restart never buys a fourth and a replacement target starts at zero; the participant enters `probe-failed` when refused. *Cost if wrong:* an attempt lost to a crash before its admission still counts; the bound is the spec's.
+20. **A relocation's execution controller is finished when its participation ends** (rev 3). `hostexec` keeps the host's lifetime lock until `Close`, so the participant sweeps its retained controllers once per pass and closes every one whose operation the claim no longer runs with this box as a participant. The relocation controller's `Cleanup` hook is scoped: it retires only the relocation's own partial installation (an unopened P's local install) and does nothing once this box has admitted a writer under the operation or serves the host, so closing it after a successful relocation stops nothing. The serving writer belongs to the wake assembly's own controller from its admission on. A completed operation's `hold_op` on the serving executor's row is cleared by reconcile under the activation's verified ownership; a running operation's hold is never touched. *Cost if wrong:* a scope rule in one hook; the lock release is hostexec's.
+21. **The pinned files manifest is retained by the pinner** (rev 3; B2-4). Before the pin object exists, the exact manifest bytes are written create-only to `control/relocations/<host>/<op>/manifest-<sha256>.json`, and P's manifest is written from that copy, never from the source prefix again: a late revision of the source changes nothing an admitted pin reads. *Cost if wrong:* one small object per pin under the prefix ruling 5's storage follow-up covers.
+22. **The writers-stopped observation is root's** (rev 3; L10 "a complete privileged descriptor scan"). The agent, uid carlos with `CAP_NET_BIND_SERVICE` alone, can read only its own descriptor table; it records the replicator's descriptors from `/proc/self` at open and asks root, over the authenticated instance-control socket, to make the complete scan with its cgroup manager (R3). An agent-side scan fails closed on the first unreadable table. *Cost if wrong:* one more root endpoint, beside the stop collector's.
18. **D4's automatic loss is retained as `evidence_refs`** (rev 2; claim §7.3 D3 "enrollment-qualified fence references in `evidence_refs` and stable loss identities in `loss_refs`"). `hostclaim.AutomaticLoss` holds the fence reference, its observation time, the box, enrollment and role, the box's last writer session, the discharge verdict, timer knowledge `unknown` and whether the console has reported it; the id is deterministic over (operation, box, fence) so a replayed transition dedupes. It lives beside `loss_refs`, outside `op`, and so survives the serving CAS. The console's materialization is after this plan (R5). *Cost if wrong:* a field name; the retention is what D4 requires.
## Why this is 1b-next, and what it leaves out
@@ -91,14 +118,14 @@ Left out, with where it goes:
- **H19, verbatim:** "The failover fence rests on the `terminated` read alone; disk evidence only decides discharge of local chores." No volume record gates a fence, a relocation step or an admission.
- **Absence is never evidence (§5.9).** A fence that cannot be read or does not validate is an error, never "not fenced"; a box missing from the roster refuses.
- **Every phase change is one claim CAS conditional on `op.id`, the phase and the ETag (claim C2).** Two workers racing make the same CAS and one gets `hostclaim.ErrConflict`.
-- **Exact values.** Failover pin delay **30 seconds** after the fence's `observed_at` (`hostclaim.FailoverPinDelay`); probe deadline **10 minutes** (`hostrelocate.ProbeDeadline`); at most **2** target replacements per operation (`hostclaim.MaxTargetReplacements`); reconcile's kept-row sweep **10 minutes** (H9); release deadline **10 minutes** (`hostrelocate.ReleaseDeadline`, ruling 14); create intent keys **`selected_executor`** and **`executor_enrollment`** (claim P5-D6); journal fence key **`fence.<box>`** holding `{"enrollment","key","sha256"}` (claim §7.3); release obligation **`relocation-release`**; pin objects **`control/relocations/<host>/<op>/pin-<claim gen>.json`** (`hostclaim.RecoveryPinKey`, ruling 16); loss ids **`auto-<32 hex>`** over (op, box, fence); the env platform item **`host.env`** (`hostfiles.EnvItem`).
+- **Exact values.** Failover pin delay **30 seconds** after the fence's `observed_at` (`hostclaim.FailoverPinDelay`); probe deadline **10 minutes** (`hostrelocate.ProbeDeadline`); at most **2** target replacements per operation (`hostclaim.MaxTargetReplacements`); reconcile's kept-row sweep **10 minutes** (H9); release deadline **10 minutes** (`hostrelocate.ReleaseDeadline`, ruling 14); probe attempts **3** per target installation (`hostclaim.MaxProbeAttempts`, ruling 19); create intent keys **`selected_executor`** and **`executor_enrollment`** (claim P5-D6); journal fence key **`fence.<box>`** holding `{"enrollment","key","sha256"}` (claim §7.3); release obligation **`relocation-release`**; pin objects **`control/relocations/<host>/<op>/pin-<claim gen>.json`** (`hostclaim.RecoveryPinKey`, ruling 16); loss ids **`auto-<32 hex>`** over (op, box, fence); the retained manifest **`control/relocations/<host>/<op>/manifest-<sha256>.json`** (ruling 21); the env platform item **`host.env`** (`hostfiles.EnvItem`).
- **Gated tasks.** Tasks 8a, 8b, 8c, 9 and 10 start with a "Gated on" line. Their code is written against the interfaces claim Plan 6b rev 5 declares (`ClaimAssembly`, `ClaimWakeAssembly`, `ClaimWakeInputs`, `CollectClaimStop`, `localStopIdentity`, `hostbox.ReconcileAdapter`, `registry.SetClaimHold`), moves Phase 1 Plan 2 Task 8 declares (`moves.BuildGeneration`, `BuildInputs`, `SealedObject`, the key helpers) and moves Plan 1 Task 7b declares (`RenderHeld`, `MaterializeHeld`), plus the requests R1 to R3 and R7. They were not compiled. When 6b and moves land, re-read their merged code and bind the same contracts to the names they actually have; the un-gated tasks do not change.
- **No new module dependencies. Do not add a method to `release.Store`.** Linux-only syscalls in `*_linux.go` with a `!linux` stub; CI cross-compiles darwin, windows and linux/arm64.
- **Copy.** No user-facing copy. Errors and log lines are operator-facing and follow the surrounding style. If a task finds it needs console or CLI text, stop: it waits for Paul's copy review.
- **Go environment.** `GOFLAGS=-buildvcs=false`. Leave `GOCACHE` at its default; if the sandbox refuses it, `GOCACHE=$TMPDIR/gocache` for that command. Socket tests need `TMPDIR=/tmp/k` (`mkdir -p /tmp/k`). Run tests niced (`nice -n 19`), targeted with `-run`; never the full `-race` suite, and no simulation sweeps. Scratch goes under `$TMPDIR`.
- **Gate before each commit:** `go build ./... && go vet ./<pkgs>/ && gofmt -l <dirs>` (empty), the task's targeted tests, then `GOOS=darwin go build ./... && GOOS=windows go build ./...`. (`GOOS=windows go vet ./internal/...` fails on main's own test files, `release/versions_test.go`, `appconfig/syncexec_test.go` and `activator/activator_test.go`; vet the task's packages instead.)
- Commit messages: lowercase `area:` prefix and a sentence stating the behaviour, then a blank line and `Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>`. Work on a branch cut from `origin/main` (suggested `region-ha-1b-next`). Merging is Paul's call, through `amadan branch merge`, and only after Task 11.
-- **Dry run.** Tasks 1 to 7 were applied in order to `origin/main` at 3407d114, built, vetted, cross-built for darwin, windows and linux/arm64, and their tests run; nineteen guards were each removed in turn and their test watched to fail (rev 1), then rev 2's twenty-five (listed in the Self-review). Then the code was removed; only this document is committed.
+- **Dry run.** Tasks 1 to 7 were applied in order to `origin/main` at 3407d114, built, vetted, cross-built for darwin, windows and linux/arm64, and their tests run; nineteen guards were each removed in turn and their test watched to fail (rev 1), then rev 2's twenty-five and rev 3's eleven (listed in the Self-review). Then the code was removed; only this document is committed.
## Review Focus
@@ -142,8 +169,8 @@ Spec §5.8 and claim §4. Every transition is a `func(*Claim) error` for `hostcl
**Interfaces:**
- Consumes (main): `Update`, `RequireOp`, `VerifyFence`, `FenceRef`, `FenceAuthority`, `AcknowledgeExecution`, `AdmitWriter`, `validBoxReference`, `isHex64`, `KindRelocating`, `validateRelocating`.
-- Produces: constants `ReasonDrain|ReasonRebalance|ReasonFailover`, `PhaseStarting … PhaseProbeFailed`, `MaxTargetReplacements = 2`, `FailoverPinDelay = 30s`; errors `ErrRelocation`, `ErrReplacementsExceeded`, `ErrAlreadyRecorded`; types `RelocationStart`, `LocalDischarge`, `Release`, `ReleaseAuthority` (`VerifyRelease(ctx, Claim, Release) error`), `RecoveryPin` (with `Empty`), `PreparedAuthority` (with `FilesSeq`, `FilesManifestSHA256`), `LocalInstall`, `Probe`, `Target{Box, Enrollment}`, `AutomaticLoss` and `Claim.EvidenceRefs`; functions `RecoveryPinKey(host, op string, gen uint64) string`, `RelocationStartGuard(c, region, source, target string, sourceFenced bool) error`, `StartRelocation(ctx, s, observed, RelocationStart, FenceAuthority) (Observed, error)`, `FenceSource(ctx, s, observed, opID, FenceRef, *LocalDischarge, FenceAuthority) (Observed, error)`, `ToReleasing(opID, now time.Time)`, `JournalRelease(opID, Release)`, `PinAfterRelease(ctx, s, observed, opID, ReleaseAuthority, *Target) (Observed, error)`, `PinFailover(opID)`, `JournalPin(opID, RecoveryPin)`, `NameTarget(opID, Target)`, `ToInstalling(opID, Target, prepared string)`, `ToInstalled(opID, PreparedAuthority)`, `RecordLocalInstall(opID, LocalInstall)`, `ToProbing(opID, now time.Time)`, `RecordProbe(opID, Probe)`, `ToProbeFailed(opID, now time.Time)`, `CompleteRelocation(opID, newID func() string)`, `ReplaceTarget(ctx, s, observed, opID, FenceRef, *LocalDischarge, next Target, prepared string, FenceAuthority) (Observed, error)`, `MarkStuck(opID, reason)`.
-- Journal keys later tasks read: `exec_ack.<box>`, `fence.<box>`, `fence.<box>.observed_at`, `discharge.<box>`, `reason`, `releasing_since`, `release`, `release.phase`, `released_by_target`, `pin`, `prepared`, `install_stamp`, `installed_boxes`, `prepared_authority`, `local_install`, `probing_since`, `probe`, `probe_failed_since`, `replacements`, `excluded_targets`, `stuck`. Claim fields: `evidence_refs`, `loss_refs`.
+- Produces: constants `ReasonDrain|ReasonRebalance|ReasonFailover`, `PhaseStarting … PhaseProbeFailed`, `MaxTargetReplacements = 2`, `FailoverPinDelay = 30s`; errors `ErrRelocation`, `ErrReplacementsExceeded`, `ErrAlreadyRecorded`; types `RelocationStart`, `LocalDischarge`, `Release`, `ReleaseAuthority` (`VerifyRelease(ctx, Claim, Release) error`), `RecoveryPin` (with `Empty`), `PreparedAuthority` (with `FilesSeq`, `FilesManifestSHA256`), `LocalInstall`, `Probe`, `Target{Box, Enrollment}`, `AutomaticLoss` (with `Incarnation`, `Stamp`, `LastAdmission`) and `Claim.EvidenceRefs`; `MaxProbeAttempts = 3`, `ErrProbeAttemptsExhausted`; functions `RecoveryPinKey(host, op string, gen uint64) string`, `BeginProbeAttempt(opID, box string)`, `RelocationStartGuard(c, region, source, target string, sourceFenced bool) error`, `StartRelocation(ctx, s, observed, RelocationStart, FenceAuthority) (Observed, error)`, `FenceSource(ctx, s, observed, opID, FenceRef, *LocalDischarge, FenceAuthority) (Observed, error)`, `ToReleasing(opID, now time.Time)`, `JournalRelease(opID, Release)`, `PinAfterRelease(ctx, s, observed, opID, ReleaseAuthority, *Target) (Observed, error)`, `PinFailover(opID)`, `JournalPin(opID, RecoveryPin)`, `NameTarget(opID, Target)`, `ToInstalling(opID, Target, prepared string)`, `ToInstalled(opID, PreparedAuthority)`, `RecordLocalInstall(opID, LocalInstall)`, `ToProbing(opID, now time.Time)`, `RecordProbe(opID, Probe)`, `ToProbeFailed(opID, now time.Time)`, `CompleteRelocation(opID, newID func() string)`, `ReplaceTarget(ctx, s, observed, opID, FenceRef, *LocalDischarge, next Target, prepared string, FenceAuthority) (Observed, error)`, `MarkStuck(opID, reason)`.
+- Journal keys later tasks read: `exec_ack.<box>`, `fence.<box>`, `fence.<box>.observed_at`, `discharge.<box>`, `reason`, `releasing_since`, `release`, `release.phase`, `released_by_target`, `pin`, `prepared`, `install_stamp`, `installed_boxes`, `prepared_authority`, `local_install`, `probing_since`, `probe_attempts.<box>`, `probe`, `probe_failed_since`, `replacements`, `excluded_targets`, `stuck`. Claim fields: `evidence_refs`, `loss_refs`.
- [ ] **Step 1: Write the failing tests**
@@ -158,6 +185,7 @@ import (
"encoding/hex"
"encoding/json"
"errors"
+ "reflect"
"slices"
"strings"
"testing"
@@ -746,6 +774,77 @@ func TestRelocationEmptyPinNeedsItsEvidence(t *testing.T) {
})
}
+func TestRelocationProbeAttemptsAreBounded(t *testing.T) {
+ ctx := context.Background()
+ s, o := relAtPhase(t, PhaseProbing)
+ for i := 0; i < MaxProbeAttempts; i++ {
+ o = step(t, s, o, BeginProbeAttempt("rel-1", relTarget))
+ }
+ if journalUint(o.Claim.Op, "probe_attempts."+relTarget) != MaxProbeAttempts {
+ t.Fatalf("count: %v", o.Claim.Op.Journal)
+ }
+ if _, err := Update(ctx, s, o, BeginProbeAttempt("rel-1", relTarget)); !errors.Is(err, ErrProbeAttemptsExhausted) {
+ t.Fatalf("a fourth attempt: %v", err)
+ }
+ if _, err := Update(ctx, s, o, BeginProbeAttempt("rel-1", relSpare)); !errors.Is(err, ErrRelocation) {
+ t.Fatalf("a non-executor counted an attempt: %v", err)
+ }
+ // A replacement target has its own count.
+ o, err := ReplaceTarget(ctx, s, o, "rel-1", putFence(t, s, relTarget), nil, Target{relSpare, relEnrollment(relSpare)}, "", relAuthority{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if journalUint(o.Claim.Op, "probe_attempts."+relSpare) != 0 {
+ t.Fatal("the replacement inherited attempts")
+ }
+}
+
+func TestRelocationDischargeCompletesDiscoveryLikeARetirement(t *testing.T) {
+ ctx := context.Background()
+ s, o := relServing(t)
+ o = step(t, s, o, func(c *Claim) error {
+ c.Touched.DiscoveryOp = "disc"
+ c.Obligations = append(c.Obligations, Obligation{ID: "d1", Op: "disc", Kind: "park-local", Box: relSource, Retirement: "r", Owner: "box:" + relSource})
+ return nil
+ })
+ ref := putFence(t, s, relSource)
+ in := relStart(ReasonFailover, &ref)
+ in.Discharge = &LocalDischarge{Qualified: true}
+ o, err := StartRelocation(ctx, s, o, in, relAuthority{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(o.Claim.Obligations) != 0 || !o.Claim.Touched.Complete || o.Claim.Touched.DiscoveryOp != "" {
+ t.Fatalf("discharging the final discovery chore: %+v touched %+v", o.Claim.Obligations, o.Claim.Touched)
+ }
+}
+
+func TestRelocationLossRetainsAFormerWritersSession(t *testing.T) {
+ // The source is fenced after the target's probe admission made the
+ // target the writer: the loss still names the source's last session.
+ ctx := context.Background()
+ s, o := relAtPhase(t, "probing+P")
+ before := o.Claim.Holder
+ var sourceAdmission AdmissionRecord
+ for _, a := range before.Admissions {
+ if a.Box == relSource {
+ sourceAdmission = a
+ }
+ }
+ o, err := FenceSource(ctx, s, o, "rel-1", putFence(t, s, relSource), nil, relAuthority{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ l := o.Claim.EvidenceRefs[0]
+ if l.Box != relSource || l.Session != sourceAdmission.Session || l.Generation != sourceAdmission.Generation || l.LastAdmission == nil || !reflect.DeepEqual(*l.LastAdmission, sourceAdmission) ||
+ l.Incarnation != before.Incarnation || l.Stamp != before.Stamp {
+ t.Fatalf("retained: %+v, source admission %+v", l, sourceAdmission)
+ }
+ if err := o.Claim.Validate(); err != nil {
+ t.Fatal(err)
+ }
+}
+
func TestRelocationRetainsEveryFencedBoxAcrossCompletion(t *testing.T) {
// A drain whose target is terminated in installing: the replacement
// completes, and the dead target's loss is still on the claim.
@@ -882,6 +981,12 @@ const (
// replacements of the target in one operation it goes to stuck.
MaxTargetReplacements = 2
+ // MaxProbeAttempts is §5.8's probe bound: at most three admissions
+ // per target installation within the probe deadline; running out is
+ // probe-failed. Attempts are counted in the claim before they are
+ // made, so a crash or a lost response never buys a fourth.
+ MaxProbeAttempts = 3
+
// FailoverPinDelay is §5.8's failover pin rule: pin no earlier than
// the fence's observed_at plus 30 seconds, so requests S3 had already
// accepted from the terminated box have completed (H3's window).
@@ -889,9 +994,10 @@ const (
)
var (
- ErrRelocation = errors.New("hostclaim: relocation transition refused")
- ErrReplacementsExceeded = errors.New("hostclaim: relocation has used its target replacements")
- ErrAlreadyRecorded = errors.New("hostclaim: relocation journal entry already recorded")
+ ErrRelocation = errors.New("hostclaim: relocation transition refused")
+ ErrReplacementsExceeded = errors.New("hostclaim: relocation has used its target replacements")
+ ErrAlreadyRecorded = errors.New("hostclaim: relocation journal entry already recorded")
+ ErrProbeAttemptsExhausted = errors.New("hostclaim: the target has used its probe attempts")
)
// RelocationStart is the start CAS's intent (spec 1 §5.8, claim §4). Fence
@@ -994,19 +1100,27 @@ type PreparedAuthority struct {
// D4 host loss record. It survives CompleteRelocation, free and holder
// replacement because it lives beside loss_refs, not in op.journal.
type AutomaticLoss struct {
- ID string `json:"id"` // the loss identity, also in loss_refs
- Box string `json:"box"`
- Enrollment string `json:"enrollment"`
- Fence FenceRef `json:"fence"`
- ObservedAt string `json:"observed_at"`
- Op string `json:"op"`
- Kind string `json:"kind"`
- Phase string `json:"phase"`
- Role string `json:"role"` // "source" | "target"
- Generation string `json:"generation,omitempty"`
- Session uint64 `json:"session,omitempty"`
- Discharge LocalDischarge `json:"discharge"`
- Timers TimerKnowledge `json:"timers"`
+ ID string `json:"id"` // the loss identity, also in loss_refs
+ Box string `json:"box"`
+ Enrollment string `json:"enrollment"`
+ Fence FenceRef `json:"fence"`
+ ObservedAt string `json:"observed_at"`
+ Op string `json:"op"`
+ Kind string `json:"kind"`
+ Phase string `json:"phase"`
+ Role string `json:"role"` // "source" | "target"
+ // Incarnation and Stamp bind the loss to the holder it was taken
+ // against; Generation and Session are the box's last writer session,
+ // whether it is still the writer or was superseded, and LastAdmission
+ // is that session's admission record when the claim holds one (the
+ // consequences the console reports: its coverage and declarations).
+ Incarnation string `json:"incarnation"`
+ Stamp uint64 `json:"stamp"`
+ Generation string `json:"generation,omitempty"`
+ Session uint64 `json:"session,omitempty"`
+ LastAdmission *AdmissionRecord `json:"last_admission,omitempty"`
+ Discharge LocalDischarge `json:"discharge"`
+ Timers TimerKnowledge `json:"timers"`
// Reported is set by the console once the D4 record holds this loss
// and its alert work exists. Until then reporting is pending.
Reported bool `json:"reported,omitempty"`
@@ -1151,13 +1265,24 @@ func fenceBox(c *Claim, box string, ref FenceRef, observedAt string, d *LocalDis
return err
}
if d.Qualified {
- kept := c.Obligations[:0]
- for _, o := range c.Obligations {
- if o.Kind != "park-local" || o.Box != box {
- kept = append(kept, o)
+ // Each chore is completed with CompleteObligation's own semantics,
+ // so a discovery operation whose final chore this discharges is
+ // marked complete exactly as a retirement would (delete.go).
+ for {
+ var chore *Obligation
+ for i := range c.Obligations {
+ if c.Obligations[i].Kind == "park-local" && c.Obligations[i].Box == box {
+ chore = &c.Obligations[i]
+ break
+ }
+ }
+ if chore == nil {
+ break
+ }
+ if err := CompleteObligation(chore.Op, chore.ID)(c); err != nil {
+ return err
}
}
- c.Obligations = kept
}
return retainAutomaticLoss(c, box, ref, observedAt, *d)
}
@@ -1177,8 +1302,23 @@ func retainAutomaticLoss(c *Claim, box string, ref FenceRef, observedAt string,
}
loss := AutomaticLoss{ID: id, Box: box, Enrollment: ref.Enrollment, Fence: ref, ObservedAt: observedAt,
Op: c.Op.ID, Kind: c.Op.Kind, Phase: c.Op.Phase, Role: role, Discharge: d, Timers: TimersUnknown}
- if h := c.Holder; h != nil && h.WriterBox == box {
- loss.Generation, loss.Session = h.Generation, h.Session
+ if h := c.Holder; h != nil {
+ loss.Incarnation, loss.Stamp = h.Incarnation, h.Stamp
+ // The box's last writer session: the live one when it is still
+ // the writer, otherwise its last admission on this holder.
+ if h.WriterBox == box {
+ loss.Generation, loss.Session = h.Generation, h.Session
+ }
+ for i := len(h.Admissions) - 1; i >= 0; i-- {
+ if h.Admissions[i].Box == box {
+ last := h.Admissions[i]
+ loss.LastAdmission = &last
+ if loss.Session == 0 {
+ loss.Generation, loss.Session = last.Generation, last.Session
+ }
+ break
+ }
+ }
}
c.EvidenceRefs = append(c.EvidenceRefs, loss)
if !slices.Contains(c.LossRefs, id) {
@@ -1628,6 +1768,29 @@ func ToProbing(opID string, now time.Time) func(*Claim) error {
}
}
+// BeginProbeAttempt counts one probe attempt for the executor before it
+// is made (§5.8 "limits"): the fourth is refused with
+// ErrProbeAttemptsExhausted, and the participant then enters
+// probe-failed. The count is per box and per operation, so a replacement
+// target starts at zero and a fenced one is never counted again.
+func BeginProbeAttempt(opID, box string) func(*Claim) error {
+ return func(c *Claim) error {
+ if err := requireRelocation(c, opID, PhaseProbing); err != nil {
+ return err
+ }
+ if box == "" || box != c.Op.Exec {
+ return fmt.Errorf("%w: only the executor probes", ErrRelocation)
+ }
+ key := "probe_attempts." + box
+ n := journalUint(c.Op, key)
+ if n >= MaxProbeAttempts {
+ return ErrProbeAttemptsExhausted
+ }
+ c.Op.Journal[key] = strconv.FormatUint(n+1, 10)
+ return nil
+ }
+}
+
// RecordProbe journals the executor's passing probe of its admitted session.
func RecordProbe(opID string, p Probe) func(*Claim) error {
return func(c *Claim) error {
@@ -3112,7 +3275,7 @@ Claim §6.1, "Release before full stop": release step 1 needs the admission fenc
**Interfaces:**
- Consumes: `realFDOps`, `readProcDir`, `procGone`, `ErrIncomplete`, `ErrInvalidInput`, `ErrUnsupported`, `ErrNoCgroupEvidence`, `ErrPopulated`, `ErrMembers` (all on main).
-- Produces: `stopevidence.Descriptor{PID, FD int; Target string}`, `stopevidence.WritersInputs`, `stopevidence.WritersEvidence`, `stopevidence.WritersStopped(ctx, WritersInputs) (WritersEvidence, error)`, `stopevidence.ProcessDescriptors(ctx, procRoot string, pid int, dbs []string) ([]Descriptor, error)` (one process's descriptors on the database, taken before and after the replicator opens: R3's input), `ErrWorkOpen`, `ErrForeignHolder`.
+- Produces: `stopevidence.Descriptor{PID, FD int; Target string}`, `stopevidence.WritersInputs`, `stopevidence.WritersEvidence`, `stopevidence.WritersStopped(ctx, WritersInputs) (WritersEvidence, error)`, `stopevidence.ProcessDescriptors(ctx, procRoot string, pid int, dbs []string) ([]Descriptor, error)` (one process's descriptors on the database, reading only `/proc/<pid>`, taken by the agent for its own pid before and after the replicator opens: R3's agent half), `ErrWorkOpen`, `ErrForeignHolder`. `WritersStopped`'s complete scan of every process is root's (ruling 22).
- [ ] **Step 1: Write the failing tests**
@@ -3205,6 +3368,21 @@ func TestProcessDescriptorsNamesOneProcessesHolders(t *testing.T) {
if _, err := ProcessDescriptors(t.Context(), filepath.Join(root, "missing"), 20, []string{db}); !errors.Is(err, ErrIncomplete) {
t.Fatalf("an unreadable table: %v", err)
}
+ // Another process's table being unreadable (root-owned, as the agent
+ // sees it) does not matter: only /proc/<pid> is read.
+ if err := os.Chmod(filepath.Join(root, "30", "task"), 0); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = os.Chmod(filepath.Join(root, "30", "task"), 0o700) })
+ if again, err := ProcessDescriptors(t.Context(), root, 20, []string{db}); err != nil || len(again) != 2 {
+ t.Fatalf("own descriptors with another table unreadable: %+v %v", again, err)
+ }
+ if _, err := WritersStopped(t.Context(), writersInputs(root, db)); !errors.Is(err, ErrIncomplete) {
+ t.Fatalf("the complete scan with an unreadable table must fail closed: %v", err)
+ }
+ if err := os.Chmod(filepath.Join(root, "30", "task"), 0o700); err != nil {
+ t.Fatal(err)
+ }
// The difference around an open is the replicator's set, and
// WritersStopped accepts exactly it.
in := writersInputs(root, db)
@@ -3391,11 +3569,14 @@ func WritersStopped(ctx context.Context, in WritersInputs) (WritersEvidence, err
// ProcessDescriptors lists the descriptors of one process (its leader's
// table) that hold one of dbs' database, WAL or SHM files, as /proc shows
-// them. The agent takes it for its own pid immediately before and after
-// it opens a host's replicator: the difference is that replicator's
-// descriptor set, which WritersInputs.Replicator names at release (L10:
-// "descriptors positively identified as belonging to this host's admitted
-// replicator"). An unreadable table is ErrIncomplete, never an empty set.
+// them. It reads only /proc/<pid>, never another process's table, so the
+// agent (uid carlos, no CAP_SYS_PTRACE) can take it for its own pid
+// immediately before and after it opens a host's replicator: the
+// difference is that replicator's descriptor set, which
+// WritersInputs.Replicator names at release (L10: "descriptors positively
+// identified as belonging to this host's admitted replicator"). An
+// unreadable table is ErrIncomplete, never an empty set. The complete
+// scan of every process (WritersStopped) stays root's.
func ProcessDescriptors(ctx context.Context, procRoot string, pid int, dbs []string) ([]Descriptor, error) {
if ctx == nil || pid <= 0 {
return nil, ErrInvalidInput
@@ -3407,13 +3588,13 @@ func ProcessDescriptors(ctx context.Context, procRoot string, pid int, dbs []str
if procRoot == "" {
procRoot = "/proc"
}
- held, err := scanDescriptors(ctx, procRoot, paths)
+ held, err := scanProcess(ctx, procRoot, pid, paths)
if err != nil {
return nil, err
}
var own []Descriptor
for _, d := range held {
- if d.PID == pid && d.FD >= 0 {
+ if d.FD >= 0 {
own = append(own, d)
}
}
@@ -3465,15 +3646,10 @@ func scanDescriptors(ctx context.Context, root string, paths []string) ([]Descri
if !filepath.IsAbs(root) {
return nil, ErrInvalidInput
}
- wanted := make(map[string]bool, len(paths))
- for _, p := range paths {
- wanted[filepath.Clean(p)] = true
- }
entries, err := ops.readDir(root)
if err != nil {
return nil, fmt.Errorf("%w: %w", ErrIncomplete, err)
}
- seen := map[Descriptor]bool{}
var out []Descriptor
for _, entry := range entries {
if err := ctx.Err(); err != nil {
@@ -3483,54 +3659,11 @@ func scanDescriptors(ctx context.Context, root string, paths []string) ([]Descri
if err != nil || pid <= 0 {
continue
}
- pidDir := filepath.Join(root, entry.Name())
- tasks, err := readProcDir(filepath.Join(pidDir, "task"), pidDir, ops)
+ held, err := scanOne(ctx, root, pid, paths, ops)
if err != nil {
- return nil, fmt.Errorf("%w: pid %d: %w", ErrIncomplete, pid, err)
- }
- // The leader's table first: a thread's descriptor is the leader's
- // only when the leader shows the same number for the same file.
- sort.SliceStable(tasks, func(i, j int) bool { return tasks[i].Name() == entry.Name() && tasks[j].Name() != entry.Name() })
- leader := map[int]string{}
- for _, task := range tasks {
- tid, err := strconv.Atoi(task.Name())
- if err != nil || tid <= 0 {
- continue
- }
- taskDir := filepath.Join(pidDir, "task", task.Name())
- fdDir := filepath.Join(taskDir, "fd")
- fds, err := readProcDir(fdDir, taskDir, ops)
- if err != nil {
- return nil, fmt.Errorf("%w: pid %d tid %d: %w", ErrIncomplete, pid, tid, err)
- }
- for _, fd := range fds {
- n, err := strconv.Atoi(fd.Name())
- if err != nil {
- continue
- }
- target, err := ops.readlink(filepath.Join(fdDir, fd.Name()))
- if procGone(err) {
- continue
- }
- if err != nil {
- return nil, fmt.Errorf("%w: pid %d tid %d fd %d: %w", ErrIncomplete, pid, tid, n, err)
- }
- clean := filepath.Clean(strings.TrimSuffix(target, " (deleted)"))
- if !filepath.IsAbs(target) || !wanted[clean] {
- continue
- }
- d := Descriptor{PID: pid, FD: n, Target: clean}
- if tid == pid {
- leader[n] = clean
- } else if leader[n] != clean {
- d.FD = -1 // not provably the leader's shared table
- }
- if !seen[d] {
- seen[d] = true
- out = append(out, d)
- }
- }
+ return nil, err
}
+ out = append(out, held...)
}
sort.Slice(out, func(i, j int) bool {
if out[i].PID != out[j].PID {
@@ -3541,6 +3674,93 @@ func scanDescriptors(ctx context.Context, root string, paths []string) ([]Descri
return out, nil
}
+// scanProcess is scanDescriptors for one process: it reads only
+// /proc/<pid>, so it needs no privilege over any other process.
+func scanProcess(ctx context.Context, root string, pid int, paths []string) ([]Descriptor, error) {
+ if !filepath.IsAbs(root) {
+ return nil, ErrInvalidInput
+ }
+ ops := realFDOps()
+ if _, err := ops.stat(filepath.Join(root, strconv.Itoa(pid))); err != nil {
+ // A process asking about itself cannot have vanished: an absent
+ // or unreadable /proc/<pid> is an incomplete observation.
+ return nil, fmt.Errorf("%w: pid %d: %w", ErrIncomplete, pid, err)
+ }
+ out, err := scanOne(ctx, root, pid, paths, ops)
+ if err != nil {
+ return nil, err
+ }
+ sort.Slice(out, func(i, j int) bool { return out[i].FD < out[j].FD })
+ return out, nil
+}
+
+// scanOne lists pid's descriptors on paths, deduplicated by descriptor
+// number. A descriptor seen only in a task that does not share the
+// leader's table (an unshared thread) is reported with FD -1, so no
+// allow-list can exempt it. A process that vanished mid-scan is skipped;
+// an unreadable table is ErrIncomplete.
+func scanOne(ctx context.Context, root string, pid int, paths []string, ops fdOps) ([]Descriptor, error) {
+ wanted := make(map[string]bool, len(paths))
+ for _, p := range paths {
+ wanted[filepath.Clean(p)] = true
+ }
+ pidDir := filepath.Join(root, strconv.Itoa(pid))
+ tasks, err := readProcDir(filepath.Join(pidDir, "task"), pidDir, ops)
+ if err != nil {
+ return nil, fmt.Errorf("%w: pid %d: %w", ErrIncomplete, pid, err)
+ }
+ // The leader's table first: a thread's descriptor is the leader's
+ // only when the leader shows the same number for the same file.
+ leaderName := strconv.Itoa(pid)
+ sort.SliceStable(tasks, func(i, j int) bool { return tasks[i].Name() == leaderName && tasks[j].Name() != leaderName })
+ leader := map[int]string{}
+ seen := map[Descriptor]bool{}
+ var out []Descriptor
+ for _, task := range tasks {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ tid, err := strconv.Atoi(task.Name())
+ if err != nil || tid <= 0 {
+ continue
+ }
+ taskDir := filepath.Join(pidDir, "task", task.Name())
+ fdDir := filepath.Join(taskDir, "fd")
+ fds, err := readProcDir(fdDir, taskDir, ops)
+ if err != nil {
+ return nil, fmt.Errorf("%w: pid %d tid %d: %w", ErrIncomplete, pid, tid, err)
+ }
+ for _, fd := range fds {
+ n, err := strconv.Atoi(fd.Name())
+ if err != nil {
+ continue
+ }
+ target, err := ops.readlink(filepath.Join(fdDir, fd.Name()))
+ if procGone(err) {
+ continue
+ }
+ if err != nil {
+ return nil, fmt.Errorf("%w: pid %d tid %d fd %d: %w", ErrIncomplete, pid, tid, n, err)
+ }
+ clean := filepath.Clean(strings.TrimSuffix(target, " (deleted)"))
+ if !filepath.IsAbs(target) || !wanted[clean] {
+ continue
+ }
+ d := Descriptor{PID: pid, FD: n, Target: clean}
+ if tid == pid {
+ leader[n] = clean
+ } else if leader[n] != clean {
+ d.FD = -1 // not provably the leader's shared table
+ }
+ if !seen[d] {
+ seen[d] = true
+ out = append(out, d)
+ }
+ }
+ }
+ return out, nil
+}
+
// processStart is field 22 of /proc/<pid>/stat: the process start time in
// clock ticks since boot, which with the boot ID names one process.
func processStart(root string, pid int) (string, error) {
@@ -3574,6 +3794,10 @@ func scanDescriptors(context.Context, string, []string) ([]Descriptor, error) {
return nil, ErrUnsupported
}
+func scanProcess(context.Context, string, int, []string) ([]Descriptor, error) {
+ return nil, ErrUnsupported
+}
+
func processStart(string, int) (string, error) { return "", ErrUnsupported }
```
@@ -3595,7 +3819,7 @@ Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
### Task 5: The regional worker (`internal/hostrelocate`)
-Spec §5.8 "Who acts". `Worker.Start` checks that this box is an active, unfenced worker of the region; that the claim serves on the source and the host is an instance of a relocatable shape (`Relocatable`); that the target is in the complete eligible set (`Eligible`, §2); the source's fence (a failover needs one, a drain refuses one); that a drain's source has a writer session to release (ruling 11); `Directory.CheckBoxStart` for the starter, the target and a live source (active, unfenced, no executable update reserved: H14); the box-started census (L5, L20); and the app admission registration; then `StartRelocation`. `Worker.Step` makes at most one worker CAS from the claim alone: a fenced source converts the operation (`FenceSource`), a fenced target is replaced (`ReplaceTarget`, or `MarkStuck` once two are used), then the phase's own step: the barrier, the pin after a release or after the failover window (selected and journaled under the replica lease on `holder.location`, finding 2), naming and installing the target, recording `installed` once `Prepared` verifies P, serving after a passing probe, `probe-failed` past the probe deadline, `stuck` past the release deadline (ruling 14) or on a live target's third replacement. `StorePinner` selects the database plan with `genselect.Select` and the files pin (the newest manifest in the admitted chain, independently of the database baseline, B2-4, with the chain as its origins), checks every bound before it writes, and writes the pin object create-only at the key of the claim generation it selected against (ruling 16); a retry at the same generation reuses its own object; an empty pin needs its evidence (ruling 15). `ReadPin` is every reader's digest-checked read; `CopyPinnedManifest` writes P's files manifest from the pinned bytes and nothing else.
+Spec §5.8 "Who acts". `Worker.Start` checks that this box is an active, unfenced worker of the region; that the claim serves on the source and the host is an instance of a relocatable shape (`Relocatable`); that the target is in the complete eligible set (`Eligible`, §2); the source's fence (a failover needs one, a drain refuses one); that a drain's source has a writer session to release (ruling 11); `Directory.CheckBoxStart` for the starter, the target and a live source (active, unfenced, no executable update reserved: H14); the box-started census (L5, L20); and the app admission registration; then `StartRelocation`. `Worker.Step` makes at most one worker CAS from the claim alone: a fenced source converts the operation (`FenceSource`), a fenced target is replaced (`ReplaceTarget`, or `MarkStuck` once two are used), then the phase's own step: the barrier, the pin after a release or after the failover window (selected and journaled under the replica lease on `holder.location`, finding 2), naming and installing the target, recording `installed` once `Prepared` verifies P, serving after a passing probe, `probe-failed` past the probe deadline, `stuck` past the release deadline (ruling 14) or on a live target's third replacement. `StorePinner` selects the database plan with `genselect.Select` and the files pin (the newest manifest in the admitted chain, independently of the database baseline, B2-4, with the chain as its origins), checks every bound before it writes, and writes the pin object create-only at the key of the claim generation it selected against (ruling 16); a retry at the same generation reuses its own object; an empty pin needs its evidence (ruling 15). `ReadPin` is every reader's digest-checked read; the pinner retains the pinned manifest's exact bytes in the control bucket (ruling 21) and `CopyPinnedManifest` writes P's files manifest from that copy and nothing else. A target replacement is looked for only while replacements remain; exhaustion is stuck at once.
**Files:**
- Create: `internal/hostrelocate/worker.go`, `internal/hostrelocate/pin.go`
@@ -3603,7 +3827,7 @@ Spec §5.8 "Who acts". `Worker.Start` checks that this box is an active, unfence
**Interfaces:**
- Consumes: Task 1; `hostactors.Directory` (`Read`, `TerminationFence`, `CheckBoxStart`), `hostactors.ValidateTerminationFence`, `hostfence.FenceKey`, `hostfence.ClaimFenceAuthority` (tests), `genselect.Request`, `genselect.Plan`, `genselect.ErrEmpty`, `hostfiles.ManifestKey`, `hostfiles.Manifest`, `hostfiles.CopyManifest`, `release.CreateExact`.
-- Produces: `hostrelocate.Worker{Claims, Directory, Fences, Releases, Admission, Census, Choose, Eligible, Relocatable, Leases, Pins, Prepared, Discharge, Notify, Self, Region, Controller, NewID, Now}` with `Start(ctx, host, Request) (hostclaim.Observed, error)` and `Step(ctx, host) (hostclaim.Observed, error)`; `Request{Reason, Source, Target}`; interfaces `Admission`, `Pinner`, `Prepared`, `Lease`; `Owner`; `ProbeDeadline`, `ReleaseDeadline`; errors `ErrUnsupported`, `ErrNotWorker`, `ErrNotYet`, `ErrRefused`, `ErrNoController`; `StorePinner{Claims, Data, SelectDB, Bucket, Region}`, `PinObject{V, Host, Op, ClaimGen, DB, Empty, Files, PinnedAt}`, `FilesPin{Prefix, Seq, SHA256, Origins}`, `PinKey(host, op, gen)`, `ReadPin(ctx, claims, RecoveryPin) (PinObject, error)`, `CopyPinnedManifest(ctx, data, FilesPin, toPrefix) (hostfiles.Manifest, error)`.
+- Produces: `hostrelocate.Worker{Claims, Directory, Fences, Releases, Admission, Census, Choose, Eligible, Relocatable, Leases, Pins, Prepared, Discharge, Notify, Self, Region, Controller, NewID, Now}` with `Start(ctx, host, Request) (hostclaim.Observed, error)` and `Step(ctx, host) (hostclaim.Observed, error)`; `Request{Reason, Source, Target}`; interfaces `Admission`, `Pinner`, `Prepared`, `Lease`; `Owner`; `ProbeDeadline`, `ReleaseDeadline`; errors `ErrUnsupported`, `ErrNotWorker`, `ErrNotYet`, `ErrRefused`, `ErrNoController`; `StorePinner{Claims, Data, SelectDB, Bucket, Region}`, `PinObject{V, Host, Op, ClaimGen, DB, Empty, Files, PinnedAt}`, `FilesPin{Prefix, Seq, SHA256, Origins, Retained}`, `PinKey(host, op, gen)`, `RetainedManifestKey(host, op, sha)`, `ReadPin(ctx, claims, RecoveryPin) (PinObject, error)`, `CopyPinnedManifest(ctx, claims, data, FilesPin, toPrefix) (hostfiles.Manifest, error)`; `hostfiles.ValidateManifest(m, origins) error`.
- [ ] **Step 1: Write the failing tests**
@@ -4136,6 +4360,26 @@ func TestWorkerMarksALiveTargetsThirdReplacementStuck(t *testing.T) {
}
}
+func TestWorkerMarksExhaustionStuckBeforeLookingForASpare(t *testing.T) {
+ // Two replacements used; no box is eligible for a third. The
+ // operation is stuck at once, not retried forever on "no eligible box".
+ f := newFixture(t)
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ for _, dead := range []string{"dst", "spa"} {
+ f.terminate(dead)
+ f.worker(nil)
+ }
+ f.w.Choose = func(context.Context, hostclaim.Observed, []string) (hostclaim.Target, error) {
+ return hostclaim.Target{}, errors.New("no eligible box")
+ }
+ f.terminate("sp2")
+ if o := f.worker(nil); o.Claim.Op.Journal["stuck"] == "" {
+ t.Fatalf("exhaustion with no spare was not stuck: %+v", o.Claim.Op.Journal)
+ }
+}
+
func TestWorkerMarksAnOverdueReleaseStuck(t *testing.T) {
f := newFixture(t)
if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
@@ -4227,6 +4471,14 @@ func TestStorePinnerPinsTheNewestFilesManifestInTheChain(t *testing.T) {
if want := []string{pinLocation + "/files/g3/", pinLocation + "/files/g2/", pinLocation + "/files/g1/"}; !equalStrings(obj.Files.Origins, want) || obj.ClaimGen != 12 {
t.Fatalf("origins %v gen %d", obj.Files.Origins, obj.ClaimGen)
}
+ // The exact bytes are retained in the control bucket before the pin.
+ kept, _, kerr := claims.Get(ctx, obj.Files.Retained)
+ if kerr != nil || obj.Files.Retained != RetainedManifestKey(host, "rel-1", obj.Files.SHA256) {
+ t.Fatalf("retained copy: %s %v", obj.Files.Retained, kerr)
+ }
+ if sum := sha256.Sum256(kept); hex.EncodeToString(sum[:]) != obj.Files.SHA256 {
+ t.Fatal("the retained copy is not the pinned bytes")
+ }
// A retry at the same observation reuses the object unchanged.
again, err := p.Pin(ctx, pinClaim(), at.Add(time.Minute))
if err != nil || again != pin || selects != 1 {
@@ -4361,7 +4613,7 @@ func TestStorePinnerRefusesAnUnreadableManifest(t *testing.T) {
// pinnedManifest publishes gen's manifest whose one entry names an object
// under origin (an earlier generation's prefix), and returns its pin.
-func pinnedManifest(t *testing.T, data release.Store, gen, origin string, seq int64) (FilesPin, hostfiles.Manifest) {
+func pinnedManifest(t *testing.T, claims, data release.Store, gen, origin string, seq int64) (FilesPin, hostfiles.Manifest) {
t.Helper()
sum := sha256.Sum256([]byte("hello"))
sha := hex.EncodeToString(sum[:])
@@ -4372,69 +4624,77 @@ func pinnedManifest(t *testing.T, data release.Store, gen, origin string, seq in
m := hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: seq, Entries: map[string]hostfiles.Entry{"a.txt": {Object: object, SHA256: sha, Size: 5, Mode: 0o644, Seq: 1}}, Platform: map[string]hostfiles.Entry{}}
_, raw := putManifest(t, data, gen, m)
rawSum := sha256.Sum256(raw)
+ manifestSHA := hex.EncodeToString(rawSum[:])
+ retained := RetainedManifestKey(host, "rel-1", manifestSHA)
+ if _, err := claims.Put(context.Background(), retained, raw, ""); err != nil {
+ t.Fatal(err)
+ }
prefix := pinLocation + "/files/" + gen + "/"
- return FilesPin{Prefix: prefix, Seq: seq, SHA256: hex.EncodeToString(rawSum[:]), Origins: []string{prefix, origin}}, m
+ return FilesPin{Prefix: prefix, Seq: seq, SHA256: manifestSHA, Origins: []string{prefix, origin}, Retained: retained}, m
}
func TestCopyPinnedManifestEnforcesThePin(t *testing.T) {
ctx := context.Background()
t.Run("a second relocation keeps objects of an earlier generation", func(t *testing.T) {
- data := release.NewDirStore(t.TempDir())
- pin, want := pinnedManifest(t, data, "P1", pinLocation+"/files/g1/", 3)
- got, err := CopyPinnedManifest(ctx, data, pin, pinLocation+"/files/P2/")
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ pin, want := pinnedManifest(t, claims, data, "P1", pinLocation+"/files/g1/", 3)
+ got, err := CopyPinnedManifest(ctx, claims, data, pin, pinLocation+"/files/P2/")
if err != nil || got.Seq != want.Seq || got.Entries["a.txt"] != want.Entries["a.txt"] {
t.Fatalf("copy: %+v %v", got, err)
}
// A replay is a success; the copy's entries still name g1's object.
- if _, err := CopyPinnedManifest(ctx, data, pin, pinLocation+"/files/P2/"); err != nil {
+ if _, err := CopyPinnedManifest(ctx, claims, data, pin, pinLocation+"/files/P2/"); err != nil {
t.Fatalf("replay: %v", err)
}
// Without the earlier generation among the origins the copy refuses.
narrow := pin
narrow.Origins = []string{pin.Prefix}
- if _, err := CopyPinnedManifest(ctx, data, narrow, pinLocation+"/files/P3/"); !errors.Is(err, hostfiles.ErrOrigin) {
+ if _, err := CopyPinnedManifest(ctx, claims, data, narrow, pinLocation+"/files/P3/"); !errors.Is(err, hostfiles.ErrOrigin) {
t.Fatalf("an inherited object outside the origins: %v", err)
}
})
- t.Run("a source manifest that moved after the pin", func(t *testing.T) {
- data := release.NewDirStore(t.TempDir())
- pin, m := pinnedManifest(t, data, "P1", pinLocation+"/files/g1/", 3)
+ t.Run("a source manifest that moved after the pin is not read", func(t *testing.T) {
+ // A late PUT to the source changes nothing: P is written from the
+ // retained bytes, so an admitted pin stays usable (B2-4).
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ pin, m := pinnedManifest(t, claims, data, "P1", pinLocation+"/files/g1/", 3)
m.Seq = 4
raw, _ := json.Marshal(m)
_, etag, _ := data.Get(ctx, hostfiles.ManifestKey(pin.Prefix))
if _, err := data.Put(ctx, hostfiles.ManifestKey(pin.Prefix), raw, etag); err != nil {
t.Fatal(err)
}
- if _, err := CopyPinnedManifest(ctx, data, pin, pinLocation+"/files/P2/"); !errors.Is(err, hostfiles.ErrSourceMoved) {
- t.Fatalf("a later revision was copied: %v", err)
- }
- if _, _, err := data.Get(ctx, hostfiles.ManifestKey(pinLocation+"/files/P2/")); err == nil {
- t.Fatal("P's manifest was written from a revision the pin did not admit")
+ got, err := CopyPinnedManifest(ctx, claims, data, pin, pinLocation+"/files/P2/")
+ if err != nil || got.Seq != 3 {
+ t.Fatalf("the pinned revision: %+v %v", got, err)
}
})
- t.Run("different bytes at the pinned sequence", func(t *testing.T) {
- data := release.NewDirStore(t.TempDir())
- pin, m := pinnedManifest(t, data, "P1", pinLocation+"/files/g1/", 3)
+ t.Run("a retained copy that is not the pinned bytes", func(t *testing.T) {
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ pin, m := pinnedManifest(t, claims, data, "P1", pinLocation+"/files/g1/", 3)
e := m.Entries["a.txt"]
- e.Size = 6 // the same seq, other content
+ e.Size = 6
m.Entries["a.txt"] = e
raw, _ := json.Marshal(m)
- _, etag, _ := data.Get(ctx, hostfiles.ManifestKey(pin.Prefix))
- if _, err := data.Put(ctx, hostfiles.ManifestKey(pin.Prefix), raw, etag); err != nil {
+ tampered := release.NewDirStore(t.TempDir())
+ if _, err := tampered.Put(ctx, pin.Retained, raw, ""); err != nil {
t.Fatal(err)
}
- if _, err := CopyPinnedManifest(ctx, data, pin, pinLocation+"/files/P2/"); !errors.Is(err, hostfiles.ErrSourceMoved) {
+ if _, err := CopyPinnedManifest(ctx, tampered, data, pin, pinLocation+"/files/P2/"); !errors.Is(err, ErrRefused) {
t.Fatalf("bytes the pin did not vouch for were copied: %v", err)
}
+ if _, _, err := data.Get(ctx, hostfiles.ManifestKey(pinLocation+"/files/P2/")); err == nil {
+ t.Fatal("P's manifest was written from bytes the pin did not admit")
+ }
})
t.Run("a host that never published files", func(t *testing.T) {
- data := release.NewDirStore(t.TempDir())
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
pin := FilesPin{Prefix: pinLocation + "/files/g3/", Origins: []string{pinLocation + "/files/g3/"}}
- m, err := CopyPinnedManifest(ctx, data, pin, pinLocation+"/files/P1/")
+ m, err := CopyPinnedManifest(ctx, claims, data, pin, pinLocation+"/files/P1/")
if err != nil || m.Seq != 0 || len(m.Entries) != 0 {
t.Fatalf("empty manifest: %+v %v", m, err)
}
- if _, err := CopyPinnedManifest(ctx, data, pin, pinLocation+"/files/P1/"); err != nil {
+ if _, err := CopyPinnedManifest(ctx, claims, data, pin, pinLocation+"/files/P1/"); err != nil {
t.Fatalf("replay: %v", err)
}
})
@@ -4473,6 +4733,7 @@ import (
"fmt"
"os"
"slices"
+ "strconv"
"time"
"github.com/carlosframework/platform/internal/hostactors"
@@ -4799,9 +5060,12 @@ func (w Worker) Step(ctx context.Context, host string) (hostclaim.Observed, erro
if c.Op.Journal["release"] == "" {
return w.releaseOverdue(ctx, obs, "probe_failed_since")
}
+ if exhausted(c) {
+ return w.stuckOn(ctx, obs, hostclaim.ErrReplacementsExceeded)
+ }
next, err := w.next(ctx, obs)
if err != nil {
- return w.stuckOn(ctx, obs, err)
+ return obs, err // no eligible box yet: a later pass may find one
}
out, err := hostclaim.PinAfterRelease(ctx, w.Claims, obs, id, w.Releases, &next)
if err != nil {
@@ -4927,9 +5191,15 @@ func (w Worker) next(ctx context.Context, obs hostclaim.Observed) (hostclaim.Tar
}
func (w Worker) replace(ctx context.Context, obs hostclaim.Observed, ref hostclaim.FenceRef, d *hostclaim.LocalDischarge) (hostclaim.Observed, error) {
+ // Exhaustion is checked before a box is chosen (§5.8 "The bound"):
+ // with the replacements used, no spare is looked for and the
+ // operation is stuck now, not after an eligible box never appears.
+ if exhausted(obs.Claim) {
+ return w.stuckOn(ctx, obs, hostclaim.ErrReplacementsExceeded)
+ }
next, err := w.next(ctx, obs)
if err != nil {
- return w.stuckOn(ctx, obs, err)
+ return obs, err
}
out, err := hostclaim.ReplaceTarget(ctx, w.Claims, obs, obs.Claim.Op.ID, ref, d, next, w.NewID(), w.Fences)
if errors.Is(err, hostclaim.ErrReplacementsExceeded) {
@@ -4954,6 +5224,12 @@ func (w Worker) update(ctx context.Context, obs hostclaim.Observed, tr func(*hos
return hostclaim.Update(ctx, w.Claims, obs, tr)
}
+// exhausted reports whether the operation has used its target replacements.
+func exhausted(c *hostclaim.Claim) bool {
+ n, _ := strconv.ParseUint(c.Op.Journal["replacements"], 10, 64)
+ return n >= hostclaim.MaxTargetReplacements
+}
+
// acked is the D6 barrier as the worker reads it before a CAS that needs it.
func acked(c *hostclaim.Claim) bool {
for _, box := range c.Op.Participants {
@@ -4980,7 +5256,6 @@ import (
"errors"
"fmt"
"os"
- "reflect"
"time"
"github.com/carlosframework/platform/internal/genselect"
@@ -5016,6 +5291,18 @@ type FilesPin struct {
Seq int64 `json:"seq"`
SHA256 string `json:"sha256"`
Origins []string `json:"origins"`
+ // Retained is the pinner's immutable copy of the exact manifest bytes
+ // (RetainedManifestKey), written create-only before the pin object. A
+ // late revision of the source manifest changes nothing the build
+ // reads: P is written from this copy, never from the source again.
+ // Empty for a host that never published files.
+ Retained string `json:"retained,omitempty"`
+}
+
+// RetainedManifestKey is where the pinner keeps the pinned manifest's
+// bytes: control/relocations/<host>/<op>/manifest-<sha256>.json.
+func RetainedManifestKey(host, op, sha string) string {
+ return "control/relocations/" + host + "/" + op + "/manifest-" + sha + ".json"
}
// PinKey is hostclaim.RecoveryPinKey.
@@ -5077,12 +5364,22 @@ func (p StorePinner) Pin(ctx context.Context, obs hostclaim.Observed, now time.T
if released && obj.DB != nil && obj.DB.MaxTXID < rel.FinalTXID {
return hostclaim.RecoveryPin{}, fmt.Errorf("%w: the listing is at %d, the release at %d", ErrNotYet, obj.DB.MaxTXID, rel.FinalTXID)
}
- if obj.Files, err = p.filesPin(ctx, c.Holder); err != nil {
+ files, retained, err := p.filesPin(ctx, c.Holder)
+ if err != nil {
return hostclaim.RecoveryPin{}, err
}
+ obj.Files = files
if released && obj.Files.Seq < rel.ManifestSeq {
return hostclaim.RecoveryPin{}, fmt.Errorf("%w: the files manifest is at %d, the release at %d", ErrNotYet, obj.Files.Seq, rel.ManifestSeq)
}
+ if retained != nil {
+ // The exact bytes, kept where nothing rewrites them, before the
+ // pin that names them exists.
+ obj.Files.Retained = RetainedManifestKey(c.Host, c.Op.ID, files.SHA256)
+ if err := release.CreateExact(ctx, p.Claims, obj.Files.Retained, retained); err != nil && !errors.Is(err, release.ErrCASConflict) {
+ return hostclaim.RecoveryPin{}, err
+ }
+ }
raw, err := json.Marshal(obj)
if err != nil {
return hostclaim.RecoveryPin{}, err
@@ -5124,7 +5421,7 @@ func pinRef(key string, raw []byte, obj PinObject) hostclaim.RecoveryPin {
// the chain is a host that never published files, pinned at sequence 0.
// Origins is the whole chain, so a manifest built by earlier copies (whose
// entries name earlier generations' objects, L13) restores.
-func (p StorePinner) filesPin(ctx context.Context, h *hostclaim.Holder) (FilesPin, error) {
+func (p StorePinner) filesPin(ctx context.Context, h *hostclaim.Holder) (FilesPin, []byte, error) {
chain := []string{h.Generation}
for i := len(h.Admissions) - 1; i >= 0; i-- {
chain = append(chain, h.Admissions[i].Generation, h.Admissions[i].Predecessor)
@@ -5144,21 +5441,21 @@ func (p StorePinner) filesPin(ctx context.Context, h *hostclaim.Holder) (FilesPi
continue
}
if err != nil {
- return FilesPin{}, err
+ return FilesPin{}, nil, err
}
var m hostfiles.Manifest // decoded from the bytes the digest names
if err := json.Unmarshal(raw, &m); err != nil || m.V != hostfiles.ManifestVersion {
- return FilesPin{}, errors.Join(fmt.Errorf("%w: files manifest %s", ErrRefused, prefix), err)
+ return FilesPin{}, nil, errors.Join(fmt.Errorf("%w: files manifest %s", ErrRefused, prefix), err)
}
sum := sha256.Sum256(raw)
- return FilesPin{Prefix: prefix, Seq: m.Seq, SHA256: hex.EncodeToString(sum[:]), Origins: origins}, nil
+ return FilesPin{Prefix: prefix, Seq: m.Seq, SHA256: hex.EncodeToString(sum[:]), Origins: origins}, raw, nil
}
gen := h.Generation
if gen == "" {
gen = hostclaim.RootGeneration
}
prefix := h.Location + "/files/" + gen + "/"
- return FilesPin{Prefix: prefix, Origins: []string{prefix}}, nil
+ return FilesPin{Prefix: prefix, Origins: []string{prefix}}, nil, nil
}
// ReadPin reads and verifies the pin object a journaled RecoveryPin names:
@@ -5181,57 +5478,54 @@ func ReadPin(ctx context.Context, claims release.Store, ref hostclaim.RecoveryPi
// CopyPinnedManifest writes the pinned files manifest into toPrefix
// (§5.8 "write P's files manifest, create-only, as the pinned manifest's
-// entries"), enforcing the pin: the source bytes must hash to the pin's
-// digest and sit at the pinned sequence, the copy validates every entry
-// against the pinned origins (L13), and what CopyManifest wrote must equal
-// what the digest vouched for. A source that moved is ErrSourceMoved; a
-// source whose bytes are not the pinned bytes is refused before anything
-// is written. A retry finding the same manifest at toPrefix succeeds.
-func CopyPinnedManifest(ctx context.Context, data release.Store, pin FilesPin, toPrefix string) (hostfiles.Manifest, error) {
- if pin.Prefix == "" || toPrefix == "" || len(pin.Origins) == 0 {
+// entries"), from the pinner's retained copy of its exact bytes (claims:
+// the control bucket), never from the source prefix again: a late
+// revision of the source manifest cannot move the pin (B2-4). The bytes
+// must hash to the pin's digest and sit at the pinned sequence; every
+// entry is validated against the pinned origins (L13); the write is
+// create-only, and a retry finding the same bytes at toPrefix succeeds.
+// A host that never published files gets an empty manifest.
+func CopyPinnedManifest(ctx context.Context, claims, data release.Store, pin FilesPin, toPrefix string) (hostfiles.Manifest, error) {
+ if pin.Prefix == "" || toPrefix == "" || len(pin.Origins) == 0 || (pin.SHA256 == "") != (pin.Retained == "") {
return hostfiles.Manifest{}, ErrUnsupported
}
var pinned hostfiles.Manifest
- raw, _, err := data.Get(ctx, hostfiles.ManifestKey(pin.Prefix))
- switch {
- case errors.Is(err, os.ErrNotExist) && pin.SHA256 == "" && pin.Seq == 0:
+ if pin.Retained == "" {
// A host that never published files: P starts an empty manifest.
pinned = hostfiles.Manifest{V: hostfiles.ManifestVersion, Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}}
- body, err := json.Marshal(pinned)
+ } else {
+ raw, _, err := claims.Get(ctx, pin.Retained)
if err != nil {
return hostfiles.Manifest{}, err
}
- if err := release.CreateExact(ctx, data, hostfiles.ManifestKey(toPrefix), body); err != nil {
- if errors.Is(err, release.ErrCASConflict) {
- return hostfiles.Manifest{}, hostfiles.ErrManifestExists
- }
+ if sum := sha256.Sum256(raw); hex.EncodeToString(sum[:]) != pin.SHA256 {
+ return hostfiles.Manifest{}, fmt.Errorf("%w: retained manifest %s is not the pinned bytes", ErrRefused, pin.Retained)
+ }
+ if err := json.Unmarshal(raw, &pinned); err != nil || pinned.Seq != pin.Seq {
+ return hostfiles.Manifest{}, fmt.Errorf("%w: retained manifest %s", ErrRefused, pin.Retained)
+ }
+ if err := hostfiles.ValidateManifest(pinned, pin.Origins); err != nil {
return hostfiles.Manifest{}, err
}
- return pinned, nil
- case err != nil:
- return hostfiles.Manifest{}, err
}
- if sum := sha256.Sum256(raw); hex.EncodeToString(sum[:]) != pin.SHA256 {
- return hostfiles.Manifest{}, fmt.Errorf("%w: files manifest %s is not the pinned bytes", hostfiles.ErrSourceMoved, pin.Prefix)
- }
- if err := json.Unmarshal(raw, &pinned); err != nil || pinned.V != hostfiles.ManifestVersion || pinned.Seq != pin.Seq {
- return hostfiles.Manifest{}, fmt.Errorf("%w: files manifest %s", ErrRefused, pin.Prefix)
- }
- m, err := hostfiles.CopyManifest(ctx, data, pin.Prefix, toPrefix, pin.Seq, pin.Origins)
+ body, err := json.Marshal(pinned)
if err != nil {
return hostfiles.Manifest{}, err
}
- if m.Seq != pinned.Seq || !reflect.DeepEqual(m.Entries, pinned.Entries) || !reflect.DeepEqual(m.Platform, pinned.Platform) {
- return hostfiles.Manifest{}, fmt.Errorf("%w: the copy is not the pinned manifest", hostfiles.ErrSourceMoved)
+ if err := release.CreateExact(ctx, data, hostfiles.ManifestKey(toPrefix), body); err != nil {
+ if errors.Is(err, release.ErrCASConflict) {
+ return hostfiles.Manifest{}, hostfiles.ErrManifestExists
+ }
+ return hostfiles.Manifest{}, err
}
- return m, nil
+ return pinned, nil
}
```
- [ ] **Step 5: Run the tests**
Run: `nice -n 19 go test ./internal/hostrelocate/ -count=1 -v 2>&1 | grep -E '^(---|ok|FAIL)'`
-Expected: PASS for all twenty tests.
+Expected: PASS for all twenty-one tests.
- [ ] **Step 6: Commit**
@@ -5246,7 +5540,7 @@ Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
### Task 6: The participant's half, ordered (`internal/hostrelocate`)
-A participant box acknowledges first (D6, after §5.10's pending-state and update-begun check, inside `Effects.Acknowledge`); before every later effect it checks its execution controller is still open and active (`Effects.Ready`), and after an agent restart (`ErrNoController`) it reopens one and journals nothing, the first acknowledgement standing; then, as `op.exec`, it releases in `releasing` or `probe-failed`, builds P in `installing`, makes its local install and enters `probing` in `installed`, and probes in `probing`. Coverage for the probe admission is `released` only when this operation journaled a release whose session is the predecessor writer and P has not been opened; `AdmitWriter` checks the rest. A box on `lost`, or not a participant, does nothing. The two end-to-end tests drive the worker and both participants against one claim store: a drain completes with `released` and no declared loss; a failover completes and declares the source's session.
+A participant box acknowledges first (D6, after §5.10's pending-state and update-begun check, inside `Effects.Acknowledge`); before every later effect it checks its execution controller is still open and active (`Effects.Ready`), and after an agent restart (`ErrNoController`) it reopens one and journals nothing, the first acknowledgement standing; then, as `op.exec`, it releases in `releasing` or `probe-failed`, builds P in `installing`, makes its local install and enters `probing` in `installed`, and probes in `probing`, counting each attempt in the claim first and entering `probe-failed` on the fourth (ruling 19). `Sweep`, once per pass, finishes every retained controller whose operation the claim no longer runs with this box as a participant (ruling 20). Coverage for the probe admission is `released` only when this operation journaled a release whose session is the predecessor writer and P has not been opened; `AdmitWriter` checks the rest. A box on `lost`, or not a participant, does nothing. The two end-to-end tests drive the worker and both participants against one claim store: a drain completes with `released` and no declared loss; a failover completes and declares the source's session.
**Files:**
- Create: `internal/hostrelocate/participant.go`
@@ -5254,7 +5548,7 @@ A participant box acknowledges first (D6, after §5.10's pending-state and updat
**Interfaces:**
- Consumes: Tasks 1 and 5 (the worker fixture in `worker_test.go`).
-- Produces: `hostrelocate.Effects` (`Acknowledge`, `Ready`, `Release`, `Build`, `InstallLocal`, `Probe`), `hostrelocate.Participant{Claims, Box, Enrollment, Effects, Now}` with `Step(ctx, host) (hostclaim.Observed, error)`.
+- Produces: `hostrelocate.Effects` (`Acknowledge`, `Ready`, `Release`, `Build`, `InstallLocal`, `Probe`, `Retained`, `Finish`), `hostrelocate.Participation{Host, Op}`, `hostrelocate.Participant{Claims, Box, Enrollment, Effects, Now}` with `Step(ctx, host) (hostclaim.Observed, error)` and `Sweep(ctx) error`.
- [ ] **Step 1: Write the failing tests**
@@ -5285,13 +5579,28 @@ type fakeBox struct {
opened int
restarted bool
effects []string
+ op string
+ finished bool
+ // failProbes makes every probe admission fail readiness.
+ failProbes bool
}
func (b *fakeBox) Acknowledge(_ context.Context, o hostclaim.Observed) (hostclaim.ExecutionAcknowledgement, error) {
b.opened++
- b.restarted = false
+ b.restarted, b.finished, b.op = false, false, o.Claim.Op.ID
return hostclaim.ExecutionAcknowledgement{Op: o.Claim.Op.ID, Enrollment: enrollment(b.box), BootID: "boot", ControllerID: "ctl-" + b.box, LeaseETag: "lease", Expires: b.f.now.Add(15 * 60e9)}, nil
}
+func (b *fakeBox) Retained() []Participation {
+ if b.opened == 0 || b.finished {
+ return nil
+ }
+ return []Participation{{Host: host, Op: b.op}}
+}
+func (b *fakeBox) Finish(_ context.Context, h, op string) error {
+ b.finished = true
+ b.effects = append(b.effects, "finish "+op)
+ return nil
+}
func (b *fakeBox) Ready(context.Context, hostclaim.Observed) error {
b.effects = append(b.effects, "ready")
if b.restarted || b.opened == 0 {
@@ -5320,6 +5629,9 @@ func (b *fakeBox) Probe(ctx context.Context, o hostclaim.Observed, coverage stri
if err != nil {
return 0, false, err
}
+ if b.failProbes {
+ return next.Claim.Holder.Session, false, nil
+ }
return next.Claim.Holder.Session, true, nil
}
@@ -5433,11 +5745,78 @@ func TestAParticipantChecksItsControllerBeforeEveryEffectAndReopensAfterARestart
}
}
}
+
+func TestAParticipantGivesUpAfterThreeProbeAttempts(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", failProbes: true}
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
+ pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
+ for i := 0; i < 40; i++ {
+ o := f.claim()
+ if o.Claim.Op.Phase == hostclaim.PhaseProbeFailed {
+ if n := len(dst.coverages); n != hostclaim.MaxProbeAttempts {
+ t.Fatalf("%d probe admissions, want %d", n, hostclaim.MaxProbeAttempts)
+ }
+ if journal := o.Claim.Op.Journal; journal["probe_attempts.dst"] != "3" {
+ t.Fatalf("attempts journaled: %v", journal["probe_attempts.dst"])
+ }
+ return
+ }
+ f.w.Step(ctx, host)
+ ps.Step(ctx, host)
+ pd.Step(ctx, host)
+ }
+ t.Fatalf("never probe-failed: %+v", f.claim().Claim.Op)
+}
+
+func TestAParticipantFinishesItsControllerOnceTheOperationIsOver(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
+ pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
+ run(t, f, ps, pd)
+ if err := pd.Sweep(ctx); err != nil || !dst.finished {
+ t.Fatalf("the serving target kept its relocation controller: %v finished=%v", err, dst.finished)
+ }
+ if err := ps.Sweep(ctx); err != nil || !src.finished {
+ t.Fatalf("the former source kept its controller: %v", err)
+ }
+ // A second relocation opens a fresh controller.
+ spa := &fakeBox{f: f, box: "spa"}
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "dst", Target: "spa"}); err != nil {
+ t.Fatal(err)
+ }
+ before := dst.opened
+ run(t, f, pd, Participant{Claims: f.s, Box: "spa", Enrollment: enrollment("spa"), Effects: spa, Now: f.clock})
+ if dst.opened != before+1 {
+ t.Fatalf("the second relocation did not open its own controller: %d -> %d", before, dst.opened)
+ }
+ // Mid-operation, Sweep keeps the controller.
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "spa", Target: "sp2"}); err != nil {
+ t.Fatal(err)
+ }
+ pspa := Participant{Claims: f.s, Box: "spa", Enrollment: enrollment("spa"), Effects: spa, Now: f.clock}
+ spa.finished = false
+ if _, err := pspa.Step(ctx, host); err != nil {
+ t.Fatal(err)
+ }
+ if err := pspa.Sweep(ctx); err != nil || spa.finished {
+ t.Fatalf("a running operation's controller was finished: %v", err)
+ }
+}
```
- [ ] **Step 2: Run them to see them fail**
-Run: `nice -n 19 go test ./internal/hostrelocate/ -run 'Completes|LostParticipant|ChecksItsController' -count=1`
+Run: `nice -n 19 go test ./internal/hostrelocate/ -run 'Completes|LostParticipant|ChecksItsController|ThreeProbeAttempts|FinishesItsController' -count=1`
Expected: FAIL to compile: `undefined: Participant`.
- [ ] **Step 3: The participant**
@@ -5491,6 +5870,19 @@ type Effects interface {
// Probe makes one cold probe admission (AdmitWriter, with coverage the
// caller names) and runs the probe. passed is false while it runs.
Probe(ctx context.Context, obs hostclaim.Observed, coverage string) (session uint64, passed bool, err error)
+ // Retained lists the participations whose execution controllers this
+ // box still holds; Finish closes one and releases its lifetime lock.
+ // A controller is finished once the claim no longer runs that
+ // operation with this box as a participant (Sweep), so the next
+ // relocation of the host can open its own (hostexec keeps the lock
+ // until Close).
+ Retained() []Participation
+ Finish(ctx context.Context, host, op string) error
+}
+
+// Participation names one retained execution controller.
+type Participation struct {
+ Host, Op string
}
// Participant is one box's half of relocation. Step makes at most one
@@ -5558,6 +5950,16 @@ func (p Participant) Step(ctx context.Context, host string) (hostclaim.Observed,
}
return p.update(ctx, obs, hostclaim.RecordLocalInstall(id, l))
case executor && c.Op.Phase == hostclaim.PhaseProbing && c.Op.Journal["probe"] == "":
+ // The attempt is counted in the claim before it is made (§5.8:
+ // at most three); the fourth is probe-failed, by this CAS.
+ counted, err := hostclaim.Update(ctx, p.Claims, obs, hostclaim.BeginProbeAttempt(id, p.Box))
+ if errors.Is(err, hostclaim.ErrProbeAttemptsExhausted) {
+ return p.update(ctx, obs, hostclaim.ToProbeFailed(id, p.Now()))
+ }
+ if err != nil {
+ return obs, err
+ }
+ obs, c = counted, counted.Claim
session, passed, err := p.Effects.Probe(ctx, obs, coverageFor(c))
if err != nil || !passed {
return obs, errors.Join(ErrNotYet, err)
@@ -5573,6 +5975,31 @@ func (p Participant) Step(ctx context.Context, host string) (hostclaim.Observed,
return obs, ErrNotYet
}
+// Sweep finishes every retained controller whose participation is over:
+// the claim no longer runs that operation, or this box is no longer a
+// participant of it. It runs once per pass, for hosts the pass itself
+// skips (a completed relocation is a serving claim).
+func (p Participant) Sweep(ctx context.Context) error {
+ if p.Claims == nil || p.Effects == nil {
+ return ErrUnsupported
+ }
+ var errs []error
+ for _, r := range p.Effects.Retained() {
+ obs, err := hostclaim.Read(ctx, p.Claims, r.Host)
+ if err != nil {
+ errs = append(errs, err)
+ continue
+ }
+ if !obs.Absent && obs.Claim.Op.Kind == hostclaim.KindRelocating && obs.Claim.Op.ID == r.Op && slices.Contains(obs.Claim.Op.Participants, p.Box) && !slices.Contains(obs.Claim.Lost, p.Box) {
+ continue
+ }
+ if err := p.Effects.Finish(ctx, r.Host, r.Op); err != nil {
+ errs = append(errs, err)
+ }
+ }
+ return errors.Join(errs...)
+}
+
// coverageFor is the probe admission's coverage (§5.8, L9): `released`
// only for the admission that opens P, when this operation journaled a
// release whose box is the predecessor writer; "" otherwise (a failover,
@@ -5626,7 +6053,7 @@ Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
### Task 7: Release evidence on the box (`internal/host`)
-`StopReceiptReleases` is `hostclaim.ReleaseAuthority` over main's stop receipts: the receipt the journal names hashes to the journaled digest, is canonical, sits at its own key, and is the full stop (`kind: stopped`) of exactly that host, incarnation, operation, phase, box, enrollment and writer session, with obligation `relocation-release` (ruling 2). `RelocationEffects` is `hostrelocate.Effects` over providers: `Release` asks root for stop evidence and runs steps 1 to 3 inside the drain root calls back (stop the writers, observe them stopped, database flush, files flush), records the flushed positions durably against the exact stop, and only then lets the producers drain; a failed step stops everything after it; a receipt root replays is paired with the positions this exact stop recorded, or refused, never guessed at. `ReleaseFlushFiles` is step 3 in full: the files directory, the wake state and the env file in one manifest revision. Controllers are opened under the agent's lifetime and retained (`RelocationControllers`); `Ready` checks one before every effect.
+`StopReceiptReleases` is `hostclaim.ReleaseAuthority` over main's stop receipts: the receipt the journal names hashes to the journaled digest, is canonical, sits at its own key, and is the full stop (`kind: stopped`) of exactly that host, incarnation, operation, phase, box, enrollment and writer session, with obligation `relocation-release` (ruling 2). `RelocationEffects` is `hostrelocate.Effects` over providers: `Release` asks root for stop evidence and runs steps 1 to 3 inside the drain root calls back (stop the writers, observe them stopped, database flush, files flush), records the flushed positions durably against the exact stop, and only then lets the producers drain; a failed step stops everything after it; a receipt root replays is paired with the positions this exact stop recorded, or refused, never guessed at. The observation (`Observe`) is a provider because the complete descriptor scan is root's (ruling 22). After a restart, `Release` asks root under the identity the flush was recorded with, offers no drain of its own, and pairs root's durable receipt with the record. `ReleaseFlushFiles` is step 3 in full: the files directory, the wake state and the env file in one manifest revision. Controllers are opened under the agent's lifetime and retained (`RelocationControllers`); `Ready` checks one before every effect.
**Files:**
- Create: `internal/host/relocationreleases.go`, `internal/host/relocationeffects.go`, `internal/host/relocationflush.go`, `internal/hostfiles/platform.go`
@@ -5634,7 +6061,7 @@ Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
**Interfaces:**
- Consumes: Tasks 1, 4 and 6; main's `StopReceipt`, `StopRequest`, `StopResult`, `StopDrainer`, `DrainAcknowledgement`, `mustMarshalStopReceipt`, `ErrClaimStopUnavailable`, `ErrClaimStopReceipt`, `hostbox.Actor`, `hostexec.Binding`, `hostexec.State`, `hostfiles.Open`, `ExportWake`, `StageWake`, `StagePlatform`, `SyncDir`, `Commit`, `registry.Registry`.
-- Produces: `host.RelocationReleaseObligation`, `host.StopReceiptReleases{Evidence}`, `host.RelocationIdentity{Actor, BootID, ControllerID}`, `host.RelocationController` (hostexec's `Ready` and `Close`), `host.RelocationControllers` (`Close`), `host.ReleasePositions` (`Record`, `Recorded`), `host.ReleasePositionsFile{Dir}`, `host.ReleaseFlushFiles{Data, Registry, FilesDir, EnvPath, Skip}.Flush`, `*host.RelocationEffects{Lifetime, Controllers, Identity, Controller, Producers, Collect, StopWriters, Writers, FlushDB, FlushFiles, Positions, BuildP, Install, Admit}`; `hostfiles.EnvItem`.
+- Produces: `host.RelocationReleaseObligation`, `host.StopReceiptReleases{Evidence}`, `host.RelocationIdentity{Actor, BootID, ControllerID}`, `host.RelocationController` (hostexec's `Ready` and `Close`), `host.RelocationControllers` (`Close`), `host.ReleasePositions` (`Record`, `Recorded` returning the recorded request), `host.ReleasePositionsFile{Dir}`, `host.ReleaseFlushFiles{Data, Registry, FilesDir, EnvPath, Skip}.Flush`, `*host.RelocationEffects{Lifetime, Controllers, Identity, Controller, Producers, Collect, StopWriters, Writers, Observe, FlushDB, FlushFiles, Positions, BuildP, Install, Admit}` with `Retained` and `Finish`; `hostfiles.EnvItem`.
Note: `internal/host`'s tests declare a function named `hostexec`, so this file imports the package as `execctl`.
@@ -5775,11 +6202,11 @@ func (m *memPositions) Record(_ context.Context, r StopRequest, txid uint64, seq
m.req, m.txid, m.seq, m.ok = r, txid, seq, true
return nil
}
-func (m *memPositions) Recorded(_ context.Context, r StopRequest) (uint64, int64, bool, error) {
+func (m *memPositions) Recorded(_ context.Context, r StopRequest) (StopRequest, uint64, int64, bool, error) {
if !m.ok || !sameStop(m.req, r) {
- return 0, 0, false, nil
+ return StopRequest{}, 0, 0, false, nil
}
- return m.txid, m.seq, true, nil
+ return m.req, m.txid, m.seq, true, nil
}
func effectsFixture(t *testing.T) (*RelocationEffects, hostclaim.Observed, *[]string, *StopRequest) {
@@ -5816,6 +6243,7 @@ func effectsFixture(t *testing.T) (*RelocationEffects, hostclaim.Observed, *[]st
return 8, nil
},
Positions: &memPositions{calls: calls},
+ Observe: stopevidence.WritersStopped,
Collect: func(ctx context.Context, r StopRequest, d StopDrainer) (StopResult, error) {
*seen = r
if _, err := d.DrainClaim(ctx, r, "fence"); err != nil {
@@ -5916,6 +6344,35 @@ func TestRelocationReleaseReplaysOnlyRecordedPositions(t *testing.T) {
t.Fatalf("replay: %+v %v", r, err)
}
})
+ t.Run("after a restart the recorded identity asks root", func(t *testing.T) {
+ // The agent flushed and recorded under controller "ctl", root
+ // completed its receipt, the agent restarted. The new process asks
+ // root under the recorded boot and controller (so root finds that
+ // stop's durable receipt), offers no drain of its own, and pairs
+ // the replayed receipt with the recorded positions.
+ e, obs, _, _ := effectsFixture(t)
+ if _, err := e.Release(context.Background(), obs); err != nil {
+ t.Fatal(err)
+ }
+ e.Identity = func(context.Context) (RelocationIdentity, error) {
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 4}, BootID: "boot-2", ControllerID: "ctl-2"}, nil
+ }
+ var asked StopRequest
+ e.Collect = func(_ context.Context, r StopRequest, d StopDrainer) (StopResult, error) {
+ asked = r
+ if _, err := d.DrainClaim(context.Background(), r, "fence"); !errors.Is(err, ErrClaimStopIdentity) {
+ return StopResult{}, errors.New("a restarted agent offered a drain for another process's stop")
+ }
+ return StopResult{Key: "k", Digest: strings.Repeat("a", 64)}, nil
+ }
+ r, err := e.Release(context.Background(), obs)
+ if err != nil || r.FinalTXID != 41 || r.ManifestSeq != 8 {
+ t.Fatalf("after restart: %+v %v", r, err)
+ }
+ if asked.BootID != "boot" || asked.ControllerID != "ctl" {
+ t.Fatalf("root was asked under the new identity: %+v", asked)
+ }
+ })
t.Run("a record of another session", func(t *testing.T) {
e, obs, _, _ := effectsFixture(t)
if _, err := e.Release(context.Background(), obs); err != nil {
@@ -5934,7 +6391,7 @@ func TestReleasePositionsFileBindsTheExactStop(t *testing.T) {
f := ReleasePositionsFile{Dir: filepath.Join(t.TempDir(), "relocate")}
r := StopRequest{Host: "shop.example", Incarnation: "inc", InstallationStamp: 5, WriterSession: 6, OperationID: "rel-1", Phase: "releasing", ObligationID: RelocationReleaseObligation,
Box: "src", Enrollment: "enr", BootID: "boot", ControllerID: "ctl", ExpectedClaimETag: "e1", ExpectedDirectoryRevision: "4"}
- if _, _, ok, err := f.Recorded(ctx, r); ok || err != nil {
+ if _, _, _, ok, err := f.Recorded(ctx, r); ok || err != nil {
t.Fatalf("before any record: %v %v", ok, err)
}
if err := f.Record(ctx, r, 41, 8); err != nil {
@@ -5942,27 +6399,33 @@ func TestReleasePositionsFileBindsTheExactStop(t *testing.T) {
}
moved := r
moved.ExpectedClaimETag, moved.ExpectedDirectoryRevision = "e2", "5" // a target replacement moved the claim
- if txid, seq, ok, err := f.Recorded(ctx, moved); !ok || err != nil || txid != 41 || seq != 8 {
+ if rec, txid, seq, ok, err := f.Recorded(ctx, moved); !ok || err != nil || txid != 41 || seq != 8 || rec.ControllerID != "ctl" {
t.Fatalf("recorded: %d %d %v %v", txid, seq, ok, err)
}
+ // A restarted controller of the same installation finds the record,
+ // and gets the identity the flush was made under.
+ restarted := r
+ restarted.BootID, restarted.ControllerID = "boot-2", "ctl-2"
+ if rec, _, _, ok, err := f.Recorded(ctx, restarted); !ok || err != nil || rec.BootID != "boot" || rec.ControllerID != "ctl" {
+ t.Fatalf("after a restart: %+v %v %v", rec, ok, err)
+ }
for name, change := range map[string]func(*StopRequest){
- "session": func(r *StopRequest) { r.WriterSession = 7 },
- "stamp": func(r *StopRequest) { r.InstallationStamp = 6 },
- "operation": func(r *StopRequest) { r.OperationID = "rel-2" },
- "phase": func(r *StopRequest) { r.Phase = "probe-failed" },
- "controller": func(r *StopRequest) { r.ControllerID = "other" },
- "boot": func(r *StopRequest) { r.BootID = "other" },
+ "session": func(r *StopRequest) { r.WriterSession = 7 },
+ "stamp": func(r *StopRequest) { r.InstallationStamp = 6 },
+ "operation": func(r *StopRequest) { r.OperationID = "rel-2" },
+ "phase": func(r *StopRequest) { r.Phase = "probe-failed" },
+ "box": func(r *StopRequest) { r.Box = "dst" },
} {
other := r
change(&other)
- if _, _, ok, _ := f.Recorded(ctx, other); ok {
+ if _, _, _, ok, _ := f.Recorded(ctx, other); ok {
t.Fatalf("%s: another stop's positions were returned", name)
}
}
if err := os.WriteFile(filepath.Join(f.Dir, "shop.example.json"), []byte("{"), 0o600); err != nil {
t.Fatal(err)
}
- if _, _, _, err := f.Recorded(ctx, r); !errors.Is(err, ErrClaimStopReceipt) {
+ if _, _, _, _, err := f.Recorded(ctx, r); !errors.Is(err, ErrClaimStopReceipt) {
t.Fatalf("a damaged record: %v", err)
}
}
@@ -6012,8 +6475,23 @@ func TestRelocationAcknowledgeRetainsTheControllerUnderTheAgentLifetime(t *testi
if _, err := restarted.Acknowledge(context.Background(), obs); err != nil || len(opened) != 2 {
t.Fatalf("reopen: %v opened=%d", err, len(opened))
}
- if err := restarted.Controllers.Close(context.Background()); err != nil || !opened[1].closed {
- t.Fatalf("close: %v closed=%v", err, opened[1].closed)
+ // Finishing a participation closes its controller and forgets it; a
+ // later Acknowledge for the same host opens a new one.
+ if got := restarted.Retained(); len(got) != 1 || got[0] != (hostrelocate.Participation{Host: "shop.example", Op: "rel-1"}) {
+ t.Fatalf("retained: %+v", got)
+ }
+ if err := restarted.Finish(context.Background(), "shop.example", "rel-1"); err != nil || !opened[1].closed || len(restarted.Retained()) != 0 {
+ t.Fatalf("finish: %v closed=%v retained=%v", err, opened[1].closed, restarted.Retained())
+ }
+ if err := restarted.Ready(context.Background(), obs); !errors.Is(err, hostrelocate.ErrNoController) {
+ t.Fatalf("after finish: %v", err)
+ }
+ obs.Claim.Op.ID = "rel-2"
+ if _, err := restarted.Acknowledge(context.Background(), obs); err != nil || len(opened) != 3 {
+ t.Fatalf("the next relocation: %v opened=%d", err, len(opened))
+ }
+ if err := restarted.Controllers.Close(context.Background()); err != nil || !opened[2].closed {
+ t.Fatalf("close: %v closed=%v", err, opened[2].closed)
}
}
@@ -6159,12 +6637,34 @@ func (a StopReceiptReleases) VerifyRelease(ctx context.Context, c hostclaim.Clai
```go
package hostfiles
+import "fmt"
+
// EnvItem is the host's env file (<host>.env) among the manifest's
// platform items. A release flush stages it beside the wake state so a
// relocation's prepared manifest carries it (region HA spec 1 §5.6:
// always continuous, whatever the setting); the local install restores it
// with ReadPlatform.
const EnvItem = "host.env"
+
+// ValidateManifest checks every entry and platform item of m against the
+// admitted origins, as Restore and CopyManifest do before they touch
+// anything (contracts note §1). A caller holding a manifest's exact bytes
+// from somewhere other than its prefix (a relocation's retained copy)
+// validates them with it before writing them under a new generation.
+func ValidateManifest(m Manifest, origins []string) error {
+ if m.V != ManifestVersion {
+ return fmt.Errorf("hostfiles: manifest is version %d; this binary reads %d", m.V, ManifestVersion)
+ }
+ if len(origins) == 0 {
+ return ErrOrigin
+ }
+ for _, o := range origins {
+ if err := checkPrefix(o); err != nil {
+ return err
+ }
+ }
+ return checkManifest(m, origins)
+}
```
`internal/host/relocationflush.go`:
@@ -6314,32 +6814,32 @@ func (f ReleasePositionsFile) Record(_ context.Context, r StopRequest, txid uint
return dir.Sync()
}
+// sameStop is the installation and session, not the controller.
func sameStop(a, b StopRequest) bool {
return a.Host == b.Host && a.Incarnation == b.Incarnation && a.InstallationStamp == b.InstallationStamp && a.WriterSession == b.WriterSession &&
- a.OperationID == b.OperationID && a.Phase == b.Phase && a.ObligationID == b.ObligationID && a.Box == b.Box && a.Enrollment == b.Enrollment &&
- a.BootID == b.BootID && a.ControllerID == b.ControllerID
+ a.OperationID == b.OperationID && a.Phase == b.Phase && a.ObligationID == b.ObligationID && a.Box == b.Box && a.Enrollment == b.Enrollment
}
-func (f ReleasePositionsFile) Recorded(_ context.Context, r StopRequest) (uint64, int64, bool, error) {
+func (f ReleasePositionsFile) Recorded(_ context.Context, r StopRequest) (StopRequest, uint64, int64, bool, error) {
path, err := f.path(r.Host)
if err != nil {
- return 0, 0, false, err
+ return StopRequest{}, 0, 0, false, err
}
raw, err := os.ReadFile(path)
if errors.Is(err, os.ErrNotExist) {
- return 0, 0, false, nil
+ return StopRequest{}, 0, 0, false, nil
}
if err != nil {
- return 0, 0, false, err
+ return StopRequest{}, 0, 0, false, err
}
var p releasePosition
if err := json.Unmarshal(raw, &p); err != nil || p.V != 1 {
- return 0, 0, false, fmt.Errorf("%w: release positions for %s", ErrClaimStopReceipt, r.Host)
+ return StopRequest{}, 0, 0, false, fmt.Errorf("%w: release positions for %s", ErrClaimStopReceipt, r.Host)
}
if !sameStop(p.Request, r) {
- return 0, 0, false, nil
+ return StopRequest{}, 0, 0, false, nil
}
- return p.TXID, p.Seq, true, nil
+ return p.Request, p.TXID, p.Seq, true, nil
}
```
@@ -6351,6 +6851,8 @@ package host
import (
"context"
"fmt"
+ "sort"
+ "strings"
"sync"
"github.com/carlosframework/platform/internal/hostbox"
@@ -6406,6 +6908,40 @@ func (r *RelocationControllers) put(host, op string, c RelocationController) (Re
return c, true
}
+// Retained lists the participations whose controllers are open.
+func (r *RelocationControllers) Retained() []hostrelocate.Participation {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ var out []hostrelocate.Participation
+ for key := range r.open {
+ host, op, _ := strings.Cut(key, "\x00")
+ out = append(out, hostrelocate.Participation{Host: host, Op: op})
+ }
+ sort.Slice(out, func(i, j int) bool {
+ return out[i].Host < out[j].Host || (out[i].Host == out[j].Host && out[i].Op < out[j].Op)
+ })
+ return out
+}
+
+// Finish closes and forgets one participation's controller, releasing the
+// host's lifetime lock for the next operation. Closing runs hostexec's
+// Cleanup hook: the relocation's controller is opened with a cleanup
+// scoped to the relocation's own partial work (Task 8a), never the serving
+// writer's session, so finishing after a successful relocation stops
+// nothing that serves.
+func (r *RelocationControllers) Finish(ctx context.Context, host, op string) error {
+ r.mu.Lock()
+ c, ok := r.open[controllerKey(host, op)]
+ if ok {
+ delete(r.open, controllerKey(host, op))
+ }
+ r.mu.Unlock()
+ if !ok {
+ return nil
+ }
+ return c.Close(ctx)
+}
+
// Close closes every retained controller (agent shutdown).
func (r *RelocationControllers) Close(ctx context.Context) error {
r.mu.Lock()
@@ -6428,9 +6964,14 @@ func (r *RelocationControllers) Close(ctx context.Context) error {
// lost response or a crash before JournalRelease) is paired with them;
// without a record for that exact request the release is refused rather
// than guessed.
+// Recorded matches the stop's installation and session (host, incarnation,
+// stamp, session, operation, phase, obligation, box, enrollment), not its
+// controller identity: a restarted agent asks with a new boot and
+// controller id, and gets back the request the flush was made under, so
+// it can ask root for that stop's durable receipt.
type ReleasePositions interface {
Record(ctx context.Context, r StopRequest, txid uint64, seq int64) error
- Recorded(ctx context.Context, r StopRequest) (txid uint64, seq int64, ok bool, err error)
+ Recorded(ctx context.Context, r StopRequest) (recorded StopRequest, txid uint64, seq int64, ok bool, err error)
}
// RelocationEffects is hostrelocate.Effects over providers Plan 6b's box
@@ -6458,8 +6999,13 @@ type RelocationEffects struct {
// and the files publisher open. Writers reports what it left.
StopWriters func(context.Context, hostclaim.Observed) error
// Writers reports the drained facts and the replicator's descriptors
- // for release step 1's observation (stopevidence.WritersStopped).
+ // for release step 1's observation; Observe makes it. The complete
+ // descriptor scan reads every process's table, which the agent (uid
+ // carlos) cannot: production's Observe asks root over the
+ // instance-control socket (R3); an agent-side scan fails closed on
+ // the first unreadable table.
Writers func(context.Context, hostclaim.Observed) (stopevidence.WritersInputs, error)
+ Observe func(context.Context, stopevidence.WritersInputs) (stopevidence.WritersEvidence, error)
// FlushDB is release step 2: final checkpoint and SyncAndWait, then the
// stopped database's final TXID N, returned only once the newest
// level-0 file's MaxTXID is N. A sync error is returned, never swallowed.
@@ -6530,7 +7076,7 @@ func (e *RelocationEffects) Ready(ctx context.Context, obs hostclaim.Observed) e
// before the producers close, so a receipt root replays later still
// carries them.
func (e *RelocationEffects) Release(ctx context.Context, obs hostclaim.Observed) (hostclaim.Release, error) {
- if e.Identity == nil || e.Producers == nil || e.Collect == nil || e.StopWriters == nil || e.Writers == nil || e.FlushDB == nil || e.FlushFiles == nil || e.Positions == nil || obs.Claim == nil || obs.Claim.Holder == nil {
+ if e.Identity == nil || e.Producers == nil || e.Collect == nil || e.StopWriters == nil || e.Writers == nil || e.Observe == nil || e.FlushDB == nil || e.FlushFiles == nil || e.Positions == nil || obs.Claim == nil || obs.Claim.Holder == nil {
return hostclaim.Release{}, ErrClaimStopUnavailable
}
id, err := e.Identity(ctx)
@@ -6538,30 +7084,39 @@ func (e *RelocationEffects) Release(ctx context.Context, obs hostclaim.Observed)
return hostclaim.Release{}, err
}
c, h := obs.Claim, obs.Claim.Holder
- inner, err := e.Producers(ctx, obs)
- if err != nil {
- return hostclaim.Release{}, err
- }
- d := &releaseDrain{e: e, obs: obs, inner: inner}
req := StopRequest{Host: c.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: h.Session,
OperationID: c.Op.ID, Phase: c.Op.Phase, ObligationID: RelocationReleaseObligation, Box: id.Actor.Box, Enrollment: id.Actor.Enrollment,
BootID: id.BootID, ControllerID: id.ControllerID, ExpectedClaimETag: obs.ETag, ExpectedDirectoryRevision: fmt.Sprint(id.Actor.Revision)}
- result, err := e.Collect(ctx, req, d)
+ // A flush this box already made for this stop, by any controller of
+ // this installation: after a restart the request is made under the
+ // identity that flushed, so root finds that stop's durable receipt.
+ recorded, txid, seq, flushed, err := e.Positions.Recorded(ctx, req)
if err != nil {
return hostclaim.Release{}, err
}
- txid, seq := d.txid, d.seq
- if !d.done {
- // Root replayed a durable receipt: this call did not flush. The
- // positions count only if this exact stop recorded them durably
- // before its producers closed. Never guess.
- var ok bool
- if txid, seq, ok, err = e.Positions.Recorded(ctx, req); err != nil {
+ var d StopDrainer = recoveryReleaseDrain{}
+ drain := &releaseDrain{e: e, obs: obs}
+ if flushed && (recorded.BootID != req.BootID || recorded.ControllerID != req.ControllerID) {
+ // The flushing process is gone with its producers: only root's
+ // durable receipt for the recorded identity can finish this
+ // release. A drain is never offered for another process's stop.
+ req.BootID, req.ControllerID = recorded.BootID, recorded.ControllerID
+ } else {
+ if drain.inner, err = e.Producers(ctx, obs); err != nil {
return hostclaim.Release{}, err
}
- if !ok {
- return hostclaim.Release{}, fmt.Errorf("%w: a replayed receipt has no recorded flushed positions for this stop", ErrClaimStopReceipt)
- }
+ d = drain
+ }
+ result, err := e.Collect(ctx, req, d)
+ if err != nil {
+ return hostclaim.Release{}, err
+ }
+ if drain.done {
+ txid, seq = drain.txid, drain.seq
+ } else if !flushed {
+ // Root replayed a durable receipt and nothing recorded a flush for
+ // this stop: never guess.
+ return hostclaim.Release{}, fmt.Errorf("%w: a replayed receipt has no recorded flushed positions for this stop", ErrClaimStopReceipt)
}
return hostclaim.Release{ReleasedBy: id.Actor.Box, Enrollment: id.Actor.Enrollment, Generation: h.Generation, Session: h.Session,
FinalTXID: txid, ManifestSeq: seq, Receipt: result.Key, ReceiptSHA256: result.Digest}, nil
@@ -6579,6 +7134,14 @@ type releaseDrain struct {
done bool
}
+// recoveryReleaseDrain refuses: a restarted agent's producers are not the
+// stop's, and root must replay the durable receipt or nothing.
+type recoveryReleaseDrain struct{}
+
+func (recoveryReleaseDrain) DrainClaim(context.Context, StopRequest, string) (DrainAcknowledgement, error) {
+ return DrainAcknowledgement{}, fmt.Errorf("%w: the process that flushed this stop is gone; only its durable receipt can finish the release", ErrClaimStopIdentity)
+}
+
func (d *releaseDrain) DrainClaim(ctx context.Context, r StopRequest, fence string) (DrainAcknowledgement, error) {
if err := d.e.StopWriters(ctx, d.obs); err != nil {
return DrainAcknowledgement{}, err
@@ -6587,7 +7150,7 @@ func (d *releaseDrain) DrainClaim(ctx context.Context, r StopRequest, fence stri
if err != nil {
return DrainAcknowledgement{}, err
}
- if _, err := stopevidence.WritersStopped(ctx, in); err != nil {
+ if _, err := d.e.Observe(ctx, in); err != nil {
return DrainAcknowledgement{}, err
}
if d.txid, err = d.e.FlushDB(ctx, d.obs); err != nil {
@@ -6607,6 +7170,20 @@ func (d *releaseDrain) DrainClaim(ctx context.Context, r StopRequest, fence stri
return ack, nil
}
+func (e *RelocationEffects) Retained() []hostrelocate.Participation {
+ if e.Controllers == nil {
+ return nil
+ }
+ return e.Controllers.Retained()
+}
+
+func (e *RelocationEffects) Finish(ctx context.Context, host, op string) error {
+ if e.Controllers == nil {
+ return ErrClaimStopUnavailable
+ }
+ return e.Controllers.Finish(ctx, host, op)
+}
+
func (e *RelocationEffects) Build(ctx context.Context, obs hostclaim.Observed) error {
if e.BuildP == nil {
return ErrClaimStopUnavailable
@@ -6689,13 +7266,13 @@ func TestRelocationEffectsAreBoundToTheAssembly(t *testing.T) {
a := ClaimAssembly{Wake: &ClaimWakeAssembly{}}
e := a.RelocationEffects(RelocationAssemblyConfig{Lifetime: context.Background(), ReleaseDir: t.TempDir()})
if e.Lifetime == nil || e.Controllers == nil || e.Identity == nil || e.Controller == nil || e.Producers == nil || e.Collect == nil ||
- e.StopWriters == nil || e.Writers == nil || e.FlushDB == nil || e.FlushFiles == nil || e.Positions == nil || e.BuildP == nil || e.Install == nil || e.Admit == nil {
+ e.StopWriters == nil || e.Writers == nil || e.Observe == nil || e.FlushDB == nil || e.FlushFiles == nil || e.Positions == nil || e.BuildP == nil || e.Install == nil || e.Admit == nil {
t.Fatalf("an unbound provider: %+v", e)
}
}
```
-A third test, written once R2 lands: `TestRelocationStopWritersLeavesTheReplicatorOpen` starts a real attempt with 6b's test launcher (a process in the host's cgroup and a timer delivery in flight through `schedule`), calls `StopWriters`, and checks the cgroup is empty, the delivery drained, and `rep` still open (its `Lost()` not closed); then `WritersStopped` with the attempt's recorded descriptors passes. It replaces the pre-set booleans Task 7's tests use.
+A third test, written once R2 lands: `TestRelocationStopWritersLeavesTheReplicatorOpen` starts a real attempt with 6b's test launcher (a process in the host's cgroup and a timer delivery in flight through `schedule`), calls `StopWriters`, and checks the cgroup is empty, the delivery drained, and `rep` still open (its `Lost()` not closed); then root's `writers-stopped` endpoint (R3) with the attempt's recorded descriptors passes, run as the production uid against a root-owned process holding another file in `/proc`. It replaces the pre-set booleans Task 7's tests use. A fourth, `TestRelocationCleanupNeverStopsAnAdmittedWriter`, closes the relocation controller after the probe's admission and after serving and checks the wake assembly's session is untouched (its producers still registered, `rep` open), and that before any admission it retires the partial local install.
- [ ] **Step 2: Run them to see them fail**
@@ -6710,9 +7287,14 @@ Expected: FAIL to compile: `undefined: RelocationAssemblyConfig`.
package host
import (
+ "bytes"
"context"
+ "encoding/json"
"errors"
"fmt"
+ "io"
+ "net"
+ "net/http"
"os"
"path/filepath"
"time"
@@ -6765,6 +7347,8 @@ func (a ClaimAssembly) RelocationEffects(cfg RelocationAssemblyConfig) *Relocati
Controller: func(ctx context.Context, b execctl.Binding) (RelocationController, error) {
c := cfg.Exec
c.Binding = b
+ // Ruling 20: this controller's cleanup is the relocation's own.
+ c.Cleanup = func(ctx context.Context) error { return a.relocationCleanup(ctx, cfg, b.Host, b.Op) }
return cfg.Open(ctx, c)
},
Producers: func(ctx context.Context, o hostclaim.Observed) (StopDrainer, error) { return a.Wake.Producers(ctx, o) },
@@ -6773,6 +7357,9 @@ func (a ClaimAssembly) RelocationEffects(cfg RelocationAssemblyConfig) *Relocati
},
StopWriters: a.relocationStopWriters(),
Writers: a.relocationWriters(),
+ Observe: func(ctx context.Context, in stopevidence.WritersInputs) (stopevidence.WritersEvidence, error) {
+ return CollectWritersStopped(ctx, controlsock.Path, in) // root's complete scan (R3, ruling 22)
+ },
FlushDB: a.relocationFlushDB(cfg),
FlushFiles: ReleaseFlushFiles{Data: cfg.Data, Registry: cfg.Registry, FilesDir: cfg.FilesDir, EnvPath: cfg.EnvPath}.Flush,
Positions: ReleasePositionsFile{Dir: cfg.ReleaseDir},
@@ -6853,6 +7440,96 @@ func (a ClaimAssembly) relocationFlushDB(cfg RelocationAssemblyConfig) func(cont
}
}
+// relocationCleanup is the relocation controller's Cleanup hook (hostexec
+// runs it when the controller is lost or closed): it retires only this
+// operation's partial local installation, the unopened P's files, env and
+// wake import made under install_stamp, through the assembly's
+// retire-artifacts path; and it does nothing once this box has admitted a
+// writer under the operation or serves the host (the wake assembly's own
+// controller and producers own that session, ruling 20).
+func (a ClaimAssembly) relocationCleanup(ctx context.Context, cfg RelocationAssemblyConfig, host, op string) error {
+ obs, err := hostclaim.Read(ctx, a.Claims, host)
+ if err != nil {
+ return err
+ }
+ local, err := a.localStopIdentity(ctx)
+ if err != nil {
+ return err
+ }
+ c := obs.Claim
+ if obs.Absent || c.Holder == nil {
+ return nil
+ }
+ if c.Holder.WriterBox == local.Actor.Box || (c.Op.Kind == hostclaim.KindServing && c.Op.Exec == local.Actor.Box) {
+ return nil // a writer this box admitted: not the relocation's to stop
+ }
+ for _, adm := range c.Holder.Admissions {
+ if adm.Box == local.Actor.Box && c.Op.ID == op {
+ return nil
+ }
+ }
+ if c.Op.Kind == hostclaim.KindRelocating && c.Op.ID == op && c.Op.Exec == local.Actor.Box && c.Op.Journal["local_install"] != "" {
+ return a.Wake.RetireArtifacts(ctx, obs, c.Op.Journal["prepared"]) // 6b's retire-artifacts under the host guard
+ }
+ return nil
+}
+
+// CollectWritersStopped asks root for L10's observation over the
+// instance-control socket (R3's root half): root runs
+// stopevidence.WritersStopped with its cgroup manager and the complete
+// /proc scan for the request's host, and answers with the evidence. The
+// request carries everything but the cgroup, which root supplies. It
+// mirrors CollectClaimStop's transport.
+func CollectWritersStopped(ctx context.Context, path string, in stopevidence.WritersInputs) (stopevidence.WritersEvidence, error) {
+ req := WritersStoppedRequest{Host: in.Host, FenceClosed: in.FenceClosed, SpawnsSettled: in.SpawnsSettled, WorkDrained: in.WorkDrained,
+ DBPaths: in.DBPaths, ReplicatorPID: in.ReplicatorPID, ReplicatorStart: in.ReplicatorStart, Replicator: in.Replicator}
+ ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
+ defer cancel()
+ body, err := json.Marshal(req)
+ if err != nil {
+ return stopevidence.WritersEvidence{}, err
+ }
+ transport := &http.Transport{DisableKeepAlives: true, DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, "unix", path)
+ }}
+ defer transport.CloseIdleConnections()
+ request, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://control/claim/writers-stopped", bytes.NewReader(body))
+ if err != nil {
+ return stopevidence.WritersEvidence{}, err
+ }
+ response, err := (&http.Client{Transport: transport}).Do(request)
+ if err != nil {
+ return stopevidence.WritersEvidence{}, err
+ }
+ defer response.Body.Close()
+ if response.StatusCode != http.StatusOK {
+ return stopevidence.WritersEvidence{}, fmt.Errorf("%w: writers-stopped: %s", ErrClaimStopUnavailable, response.Status)
+ }
+ var ev stopevidence.WritersEvidence
+ dec := json.NewDecoder(io.LimitReader(response.Body, 1<<20))
+ dec.DisallowUnknownFields()
+ if err := dec.Decode(&ev); err != nil {
+ return stopevidence.WritersEvidence{}, err
+ }
+ if ev.Host != in.Host || ev.CheckedAt.IsZero() {
+ return stopevidence.WritersEvidence{}, fmt.Errorf("%w: writers-stopped evidence is not for %s", ErrClaimStopUnavailable, in.Host)
+ }
+ return ev, nil
+}
+
+// WritersStoppedRequest is the wire form of stopevidence.WritersInputs
+// without the cgroup, which root supplies from its own manager.
+type WritersStoppedRequest struct {
+ Host string `json:"host"`
+ FenceClosed bool `json:"fence_closed"`
+ SpawnsSettled bool `json:"spawns_settled"`
+ WorkDrained bool `json:"work_drained"`
+ DBPaths []string `json:"db_paths"`
+ ReplicatorPID int `json:"replicator_pid"`
+ ReplicatorStart string `json:"replicator_start"`
+ Replicator []stopevidence.Descriptor `json:"replicator"`
+}
+
// relocationJournalDir is where a box journals its own relocation work for
// one host and operation.
func relocationJournalDir(cfg RelocationAssemblyConfig, host, op string) string {
@@ -6935,15 +7612,21 @@ func TestPreparedRefusesAPartialBuild(t *testing.T) {
}
// A build whose phase authority moved under it stops before the next
-// durable step (the lease still held, the claim replaced the target).
+// durable write, wherever it is: between the two steps, and between the
+// builder's own snapshot and anchor PUTs (the lease still held, the claim
+// replaced the target, or the guard record retired the stamp).
func TestRelocationBuildStopsWhenAuthorityChanges(t *testing.T) {
- f := buildFixture(t, "")
- f.moveClaimAfterFilesManifest() // ReplaceTarget to another box between the two steps
- if err := f.build(context.Background(), f.obs); !errors.Is(err, hostrelocate.ErrRefused) {
- t.Fatalf("the build went on after its authority changed: %v", err)
- }
- if f.installManifestExists() {
- t.Fatal("the database build ran for a box that is no longer op.exec")
+ for _, tc := range []struct{ name, after string }{{"after the files manifest", "files"}, {"after the snapshot PUT", "snapshot"}, {"after the stamp is retired", "stamp"}} {
+ t.Run(tc.name, func(t *testing.T) {
+ f := buildFixture(t, "")
+ f.moveClaimAfter(tc.after) // ReplaceTarget to another box, or Guard.RetireInstallation, at that point
+ if err := f.build(context.Background(), f.obs); !errors.Is(err, hostrelocate.ErrRefused) {
+ t.Fatalf("the build went on after its authority changed: %v", err)
+ }
+ if f.installManifestExists() {
+ t.Fatal("the install manifest was written for a box that is no longer op.exec")
+ }
+ })
}
}
```
@@ -6975,6 +7658,7 @@ import (
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/hostfiles"
+ "github.com/carlosframework/platform/internal/hostguard"
"github.com/carlosframework/platform/internal/hostrelocate"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/moves"
@@ -7022,8 +7706,11 @@ func (a ClaimAssembly) relocationBuild(cfg RelocationAssemblyConfig) func(contex
case <-alive.Done():
}
}()
- // authority re-reads the claim: the same operation, phase, executor,
- // stamp and pin, or nothing more is written.
+ // authority re-reads the claim and the host-guard record: the same
+ // operation, phase, executor, stamp and pin, and this box's
+ // activated installation still at install_stamp, or nothing more
+ // is written. It runs before every durable step, including every
+ // write the builder makes (authorityStore below).
authority := func() error {
fresh, err := hostclaim.Read(alive, a.Claims, o.Host)
if err != nil {
@@ -7033,13 +7720,20 @@ func (a ClaimAssembly) relocationBuild(cfg RelocationAssemblyConfig) func(contex
fresh.Claim.Op.Journal["prepared"] != p || fresh.Claim.Op.Journal["install_stamp"] != stamp || fresh.Claim.Op.Journal["pin"] != c.Op.Journal["pin"] || fresh.Claim.Op.Journal["stuck"] != "" {
return fmt.Errorf("%w: the build's authority changed", hostrelocate.ErrRefused)
}
+ captured, err := hostguard.Capture(a.Guard.DataDir, o.Host)
+ if err != nil {
+ return err
+ }
+ if captured.Incarnation != c.Holder.Incarnation || strconv.FormatUint(captured.Stamp, 10) != stamp {
+ return fmt.Errorf("%w: this box's installation is not the one the installing CAS issued", hostrelocate.ErrRefused)
+ }
return alive.Err()
}
prefix := c.Holder.Location + "/" + p + "/"
if err := authority(); err != nil {
return err
}
- if _, err := hostrelocate.CopyPinnedManifest(alive, cfg.Data, pin.Files, c.Holder.Location+"/files/"+p+"/"); err != nil {
+ if _, err := hostrelocate.CopyPinnedManifest(alive, a.Claims, cfg.Data, pin.Files, c.Holder.Location+"/files/"+p+"/"); err != nil {
return err
}
if pin.DB == nil {
@@ -7056,12 +7750,39 @@ func (a ClaimAssembly) relocationBuild(cfg RelocationAssemblyConfig) func(contex
if err := os.MkdirAll(work, 0o700); err != nil {
return err
}
- _, err = moves.BuildGeneration(alive, moves.BuildInputs{Store: cfg.Versions, Bucket: cfg.Bucket, Source: src, N: pin.DB.MaxTXID,
+ _, err = moves.BuildGeneration(alive, moves.BuildInputs{Store: authorityStore{VersionedStore: cfg.Versions, check: authority}, Bucket: cfg.Bucket, Source: src, N: pin.DB.MaxTXID,
GenerationPrefix: prefix, Op: c.Op.ID, Host: o.Host, WorkDir: work, Now: cfg.Now, Verify: lease.CheckDB})
return err
}
}
+// authorityStore is the versioned store the builder writes through: every
+// write re-checks the build's authority first (R1-2), so a delayed build
+// whose phase, executor, stamp or lease moved writes nothing more. Reads
+// are the store's own.
+type authorityStore struct {
+ lease.VersionedStore
+ check func() error
+}
+
+func (s authorityStore) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
+ if err := s.check(); err != nil {
+ return "", err
+ }
+ return s.VersionedStore.PutIfAbsent(ctx, bucket, key, body)
+}
+
+func (s authorityStore) CopyVersion(ctx context.Context, sb, sk, sv, db, dk string, createOnly bool) (string, string, error) {
+ if err := s.check(); err != nil {
+ return "", "", err
+ }
+ return s.VersionedStore.CopyVersion(ctx, sb, sk, sv, db, dk, createOnly)
+}
+
+func (s authorityStore) DeleteVersion(context.Context, string, string, string) error {
+ return fmt.Errorf("%w: the build never deletes", hostrelocate.ErrRefused)
+}
+
// PreparedFromInstallManifest is hostrelocate.Prepared: P is built when a
// fresh read verifies its files manifest at the pinned sequence with the
// pinned entries and, unless the pin is empty, its completed install
@@ -7174,7 +7895,7 @@ func sameEntries(a, b hostfiles.Manifest) bool {
return string(ra) == string(rb)
}
-var _ = strconv.Itoa // the journal's install_stamp is compared as the string the CAS wrote
+var _ = os.ErrNotExist
```
`Source` carries the pinned generation's full keys as `genselect` lists them (`<location>/<generation>/<level>/<min>-<max>.ltx`); moves' `SealedClient` parses the level and TXID range from the key's tail (`ltx.ParseFilename`), so `SealedObject` needs nothing more than key, version and size (R7). Replay from `installing` after a crash: `CopyPinnedManifest` is a success on its own earlier copy; `BuildGeneration` reuses its intent, snapshot and anchor by construction (moves §7.6). An empty pin builds no snapshot or anchor.
@@ -7219,9 +7940,16 @@ package host
func TestRelocationInstallIsBoundToTheInstallStamp(t *testing.T) { /* Guard record at another stamp: ErrStampMismatch; P opened: hostrelocate.ErrRefused; happy path: the receipt's InstallStamp is the journaled one and JournalSHA256 is the journal file's */ }
// The probe's inputs: the first opening restores P from its prepared pin
-// with KeepStamp and the operation's coverage; a later attempt restores
-// the predecessor with a fresh generation, no KeepStamp, coverage "".
-func TestRelocationWakeInputsOpenPOnceThenRestoreThePredecessor(t *testing.T) { /* table over a probing claim before and after P's admission */ }
+// with KeepStamp, Prepared and the operation's coverage; a later attempt
+// restores the predecessor with a fresh generation, no KeepStamp, coverage
+// "", Prepared false whatever the ordinary provider answered; a later
+// attempt over an empty predecessor is Empty, not Prepared.
+func TestRelocationWakeInputsOpenPOnceThenRestoreThePredecessor(t *testing.T) { /* table over a probing claim before and after P's admission, with an Ordinary that returns Prepared true */ }
+
+// A second attempt that crashes before its first replication publishes no
+// marker: with Prepared false 6b's runtime publishes only after the first
+// durable root (N1).
+func TestASecondProbeAttemptPublishesNoMarkerBeforeItsRoot(t *testing.T) { /* 6b's wake runtime with these inputs; crash injected in Replicate before the root; no generation.json under the attempt's generation */ }
```
- [ ] **Step 2: Run them to see them fail**
@@ -7409,6 +8137,7 @@ package host
import (
"context"
"encoding/json"
+ "errors"
"fmt"
"time"
@@ -7465,11 +8194,15 @@ func (r RelocationWakeInputs) Select(ctx context.Context, rt registry.Route, o h
req := genselect.Request{Bucket: r.Bucket, Region: c.Op.Intent["region"], Host: o.Host, Holder: *c.Holder}
adm := hostclaim.Admission{ObservedOpID: c.Op.ID, ObservedPhase: hostclaim.PhaseProbing, ObservedSession: c.Holder.Session, ObservedGeneration: c.Holder.Generation,
Box: r.Box, Mode: "cold", Region: c.Op.Intent["region"], PredecessorRegion: c.Op.Intent["region"]}
+ // Nothing is inherited from the ordinary provider's answer: each branch
+ // states its prepared and empty flags (N1).
+ in.Prepared, in.Empty, in.Baseline = false, false, nil
switch {
case !opened && pin.DB == nil:
- // An evidenced empty bootstrap: a fresh, empty generation.
+ // An evidenced empty bootstrap: a fresh, empty generation, opened
+ // as P (its marker precedes replication, as a prepared one's).
adm.NewGeneration, adm.KeepStamp, adm.Coverage = prepared, true, coverageOf(c)
- in.Empty, in.Baseline = true, nil
+ in.Empty, in.Prepared = true, true
case !opened:
var im moves.InstallManifest
if err := readInstallManifest(ctx, r.Data, r.Bucket, c.Holder.Location+"/"+prepared+"/", &im); err != nil {
@@ -7480,17 +8213,25 @@ func (r RelocationWakeInputs) Select(ctx context.Context, rt registry.Route, o h
return ClaimLaunchInputs{}, err
}
adm.NewGeneration, adm.KeepStamp, adm.Coverage, adm.RemoteMaxTXID = prepared, true, coverageOf(c), pin.DB.MaxTXID
- in.Baseline = plan
+ in.Baseline, in.Prepared = plan, true
default:
- // A later attempt: cold from the predecessor, as any cold wake.
+ // A later attempt: cold from the predecessor, as any cold wake. Its
+ // marker follows the first durable root (Prepared false). A
+ // predecessor that committed nothing (no LTX file, claim §6.4a
+ // "Empty bootstrap") is an ordinary empty cold admission under 6b's
+ // own empty authority (R1-13).
+ adm.NewGeneration = r.NewID()
plan, err := genselect.Select(ctx, r.Data, req)
- if err != nil {
+ switch {
+ case errors.Is(err, genselect.ErrEmpty):
+ in.Empty = true
+ case err != nil:
return ClaimLaunchInputs{}, err
+ default:
+ adm.RemoteMaxTXID, in.Baseline = plan.MaxTXID, plan
}
- adm.NewGeneration, adm.RemoteMaxTXID = r.NewID(), plan.MaxTXID
- in.Baseline = plan
}
- in.Admission, in.Prepared = adm, true
+ in.Admission = adm
return in, nil
}
@@ -7516,11 +8257,15 @@ func (r RelocationWakeInputs) Materialize(ctx context.Context, o hostclaim.Obser
return r.Ordinary.Materialize(ctx, o, in, w) // RenderHeld and MaterializeHeld ran at the local install; the session publisher starts from P's manifest (moves §7.6)
}
func (r RelocationWakeInputs) EmptyAuthority(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
- var ref hostclaim.RecoveryPin
- if json.Unmarshal([]byte(o.Claim.Op.Journal["pin"]), &ref) != nil || !ref.Empty {
- return fmt.Errorf("%w: no evidenced empty pin", hostrelocate.ErrRefused)
+ if in.Prepared {
+ // The opening of P: only an evidenced empty pin.
+ var ref hostclaim.RecoveryPin
+ if json.Unmarshal([]byte(o.Claim.Op.Journal["pin"]), &ref) != nil || !ref.Empty {
+ return fmt.Errorf("%w: no evidenced empty pin", hostrelocate.ErrRefused)
+ }
+ return nil
}
- return nil
+ return r.Ordinary.EmptyAuthority(ctx, o, in) // a later attempt: the assembly's own rule for an empty predecessor
}
func (r RelocationWakeInputs) Publish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
return r.Ordinary.Publish(ctx, o, in) // generation.json, then PublishGeneration (claim §6.4a item 5)
@@ -7598,51 +8343,91 @@ import (
)
// Every phase, for the source and the target, through the real adapter:
-// what the registry holds afterwards is the §5.3 table, and a box with no
-// role never gets a row.
+// what the registry holds afterwards is the §5.3 table. Two things are
+// distinct: a box with no role and no row mints nothing (ErrChanged,
+// nothing written); a box with no role and an existing row keeps it, held,
+// never deleted (RowUnexplained). A hold is hold_op; nothing here writes a
+// status.
func TestReconcileAdapterInstallFollowsTheRowCases(t *testing.T) {
for _, tc := range []struct {
- phase, exec, box string
- fenceClosed bool
- wantRow bool
- wantHold string
- wantReady bool
+ name, phase, exec, box string
+ fenceClosed bool
+ existing bool // a row exists before the pass
+ wantRow bool // a row exists after it
+ wantHold string // its hold_op
+ wantErr error // nil, or ErrChanged for "nothing minted"
}{
- {hostclaim.PhaseStarting, "src", "src", false, true, "", true},
- {hostclaim.PhaseStarting, "src", "dst", false, false, "", false}, // no installation yet: nothing minted
- {hostclaim.PhaseReleasing, "src", "src", false, true, "", true},
- {hostclaim.PhaseReleasing, "src", "src", true, true, "rel-1", false}, // kept, held, under WithLocal
- {hostclaim.PhasePinning, "none", "src", true, true, "rel-1", false},
- {hostclaim.PhaseInstalling, "dst", "dst", false, true, "rel-1", false}, // minted held at install_stamp
- {hostclaim.PhaseInstalled, "dst", "dst", false, true, "rel-1", false},
- {hostclaim.PhaseProbing, "dst", "dst", false, true, "rel-1", false},
- {hostclaim.PhaseProbing, "dst", "src", true, true, "rel-1", false},
- {hostclaim.PhaseProbing, "dst", "zzz", false, false, "", false},
+ {"source starting", hostclaim.PhaseStarting, "src", "src", false, true, true, "", nil},
+ {"target before installing, no row", hostclaim.PhaseStarting, "src", "dst", false, false, false, "", ErrChanged},
+ {"source releasing, fence open", hostclaim.PhaseReleasing, "src", "src", false, true, true, "", nil},
+ {"source releasing, fence closed", hostclaim.PhaseReleasing, "src", "src", true, true, true, "rel-1", nil}, // kept, held, under WithLocal
+ {"source pinning", hostclaim.PhasePinning, "none", "src", true, true, true, "rel-1", nil},
+ {"target installing, first pass", hostclaim.PhaseInstalling, "dst", "dst", false, false, true, "rel-1", nil}, // minted held at install_stamp
+ {"target installed", hostclaim.PhaseInstalled, "dst", "dst", false, true, true, "rel-1", nil},
+ {"target probing", hostclaim.PhaseProbing, "dst", "dst", false, true, true, "rel-1", nil},
+ {"source probing", hostclaim.PhaseProbing, "dst", "src", true, true, true, "rel-1", nil},
+ {"no role, no row", hostclaim.PhaseProbing, "dst", "zzz", false, false, false, "", ErrChanged},
+ {"no role, an existing row is kept held", hostclaim.PhaseProbing, "dst", "zzz", false, true, true, "rel-0", ErrChanged}, // its earlier hold, untouched
} {
- t.Run(tc.phase+"/"+tc.box, func(t *testing.T) {
+ t.Run(tc.name, func(t *testing.T) {
f := adapterFixture(t, tc.box) // 6b's reconcile_adapter_layout_test fixture: registry, inventory, guard dir, claims store
f.claimAt(tc.phase, tc.exec)
- if tc.box == "src" || tc.phase != hostclaim.PhaseStarting {
- f.withExistingRow() // the source always has its row; a target has one from installing on (an earlier pass minted it)
+ if tc.existing {
+ f.withExistingRow("rel-0") // minted by an earlier pass, with an earlier hold where the case says
}
f.adapter.Reconciler.FenceClosed = func(string) bool { return tc.fenceClosed }
err := f.adapter.Install(context.Background(), f.record(), f.write)
+ if !errors.Is(err, tc.wantErr) {
+ t.Fatalf("err %v, want %v", err, tc.wantErr)
+ }
row, _ := f.registry.Get(f.host)
switch {
case !tc.wantRow && row != nil:
t.Fatalf("a row for a box with no role: %+v", row)
case tc.wantRow && row == nil:
- t.Fatalf("no row: %v", err)
- case tc.wantRow && (row.HoldOp != tc.wantHold || (row.Status == "ready") != tc.wantReady):
- t.Fatalf("row %+v, want hold %q ready %v", row, tc.wantHold, tc.wantReady)
- }
- if !tc.wantRow && !errors.Is(err, ErrChanged) {
- t.Fatalf("minted for no role: %v", err)
+ t.Fatal("the row was deleted")
+ case tc.wantRow && row.HoldOp != tc.wantHold:
+ t.Fatalf("hold %q, want %q", row.HoldOp, tc.wantHold)
}
})
}
}
+// The full sequence on the target: minted held in installing, kept held
+// through probing, and after the serving CAS the next reconcile clears
+// the completed operation's hold under the serving executor's verified
+// ownership (N4), so the row routes.
+func TestReconcileAdapterClearsTheHoldAfterCompletion(t *testing.T) {
+ f := adapterFixture(t, "dst")
+ for _, phase := range []string{hostclaim.PhaseInstalling, hostclaim.PhaseInstalled, hostclaim.PhaseProbing} {
+ f.claimAt(phase, "dst")
+ if err := f.adapter.Install(context.Background(), f.record(), f.write); err != nil {
+ t.Fatalf("%s: %v", phase, err)
+ }
+ if row, _ := f.registry.Get(f.host); row == nil || row.HoldOp != "rel-1" {
+ t.Fatalf("%s: row %+v", phase, row)
+ }
+ }
+ f.serving("dst") // CompleteRelocation
+ if err := f.adapter.Install(context.Background(), f.record(), f.write); err != nil {
+ t.Fatal(err)
+ }
+ if row, _ := f.registry.Get(f.host); row == nil || row.HoldOp != "" {
+ t.Fatalf("the serving target is still held: %+v", row)
+ }
+ // A hold of a running operation is never cleared by a ready decision
+ // for another: a second relocation holds the row again, and the
+ // source's own pass leaves it.
+ f.claimAt(hostclaim.PhaseStarting, "dst")
+ f.setOp("rel-2")
+ if err := f.adapter.Install(context.Background(), f.record(), f.write); err != nil {
+ t.Fatal(err)
+ }
+ if row, _ := f.registry.Get(f.host); row == nil || row.HoldOp != "" {
+ t.Fatalf("the source in starting is ready: %+v", row)
+ }
+}
+
// A held row's hold is never cleared or moved by another operation, and
// a kept row's hold write needs the row's recorded ownership.
func TestReconcileAdapterHoldIsConditional(t *testing.T) {
@@ -7662,7 +8447,7 @@ func TestReconcileAdapterHoldIsConditional(t *testing.T) {
- [ ] **Step 2: Run them to see them fail**
-Run: `nice -n 19 go test ./internal/hostbox/ -run 'ReconcileAdapterInstall|ReconcileAdapterHold' -count=1`
+Run: `nice -n 19 go test ./internal/hostbox/ -run 'ReconcileAdapterInstall|ReconcileAdapterHold|ReconcileAdapterClears' -count=1`
Expected: FAIL to compile: `f.adapter.Reconciler.FenceClosed undefined`.
- [ ] **Step 3: The decision in `Reconciler`**
@@ -7742,12 +8527,18 @@ In 6b's `ReconcileAdapter.Install`, after the row is written through `write(owne
```go
d := RelocationRow(observed.Claim, a.Box, true, a.Reconciler.fenceClosed(rec.Host))
- want := ""
- if d.Case == RowHeld {
- want = d.HoldOp
- }
- if row.HoldOp != want && (row.HoldOp == "" || row.HoldOp == want || d.Case == RowHeld) {
- if err := a.Registry.SetClaimHold(ctx, rec.Host, owner, row.HoldOp, want); err != nil {
+ switch {
+ case d.Case == RowHeld && row.HoldOp != d.HoldOp:
+ // Held for the running operation; an earlier operation's hold is
+ // replaced, never a different running one's (the claim runs one).
+ if err := a.Registry.SetClaimHold(ctx, rec.Host, owner, row.HoldOp, d.HoldOp); err != nil {
+ return err
+ }
+ case d.Case == RowReady && row.HoldOp != "" && row.HoldOp != observed.Claim.Op.ID:
+ // The operation that held this row is over and this box is the
+ // serving executor with the captured stamp (the activation above
+ // verified both): the row routes again (N4).
+ if err := a.Registry.SetClaimHold(ctx, rec.Host, owner, row.HoldOp, ""); err != nil {
return err
}
}
@@ -7776,7 +8567,7 @@ func (a ReconcileAdapter) Configure(cfg *provision.ReconcileConfig) {
}
```
-A row whose status is `ready` and whose decision is `RowHeld` is also set to the held status 6b's `SetClaimHold` implies (`hold_op` non-empty is "held" for the edge, moves §5.2); the status column is not written here.
+A hold is `hold_op` alone: non-empty is "held" for the edge (moves §5.2), and the status column is not written here.
In 6b's `provision/reconcile.go` claim branch, a record whose `RelocationRow` is `RowRetiring` or `RowUnexplained` is kept, held, and left for typed retirement (`ClaimRetire`); an unexplained row is logged once per pass (`provision: %s: claim gives this box no role — held, not served`). Kept rows are re-read on a 10-minute sweep (`claimSweepEvery = 10 * time.Minute`), and at once when the control serial changes (§5.3 "Prompt changes"); a read error keeps the row and mints none (C6).
@@ -7800,7 +8591,7 @@ Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
**Gated on:** claim Plan 6b merged (Tasks 8a to 8c's effects, Task 9's rows); claim Plan 5's app gate merged (the `Admission` provider, ruling 4); and claim Plan 5's create door merged (`ProductionAuthority` as the console's create `Authority`). Not compiled in the dry run.
-On a box with `CARLOS_HOST_CLAIM` and `CARLOS_REGION_HA` both on, one pass every 30 seconds, and at once on a control serial change: for every claim under `control/hosts/` whose operation is `relocating` in this box's region, run `Worker.Step` (any active worker may), and for every claim naming this box as a participant, `Participant.Step`. The pass never starts a relocation by itself (spec 3's placer does); `carlos ops host relocate` (later, with Paul's copy) and the spec 3 placer call `Worker.Start`. The participant's effects are one `*RelocationEffects` per agent, opened with the agent's lifetime, so a pass's context bounds the pass and nothing else. On the console, create's `Authority.SelectExecutor` is the `ExecutorChooser` over console's live edges read, and only when `Config.RegionHA` is on (Task 2); off, `Executors` is nil and the create is today's.
+On a box with `CARLOS_HOST_CLAIM` and `CARLOS_REGION_HA` both on, one pass every 30 seconds, and at once on a control serial change: for every claim under `control/hosts/` whose operation is `relocating` in this box's region, run `Worker.Step` (any active worker may), and for every claim naming this box as a participant, `Participant.Step`. The pass never starts a relocation by itself (spec 3's placer does); `carlos ops host relocate` (later, with Paul's copy) and the spec 3 placer call `Worker.Start`. The participant's effects are one `*RelocationEffects` per agent, opened with the agent's lifetime, so a pass's context bounds the pass and nothing else; each pass ends with `Participant.Sweep`, which finishes the controllers of participations the claim no longer runs (ruling 20). On the console, create's `Authority.SelectExecutor` is the `ExecutorChooser` over console's live edges read, and only when `Config.RegionHA` is on (Task 2); off, `Executors` is nil and the create is today's.
**Files:**
- Create: `internal/host/relocatepass.go`
@@ -7891,6 +8682,11 @@ func runRelocatePass(ctx context.Context, store release.Store, w hostrelocate.Wo
log.Printf("relocate %s: %s: %v", host, p.Box, err)
}
}
+ // Participations that ended this pass (a serving claim is skipped
+ // above) release their controllers and the host's lifetime lock.
+ if err := p.Sweep(pass); err != nil {
+ log.Printf("relocate: sweep controllers: %v", err)
+ }
cancel()
select {
case <-ctx.Done():
@@ -7986,7 +8782,7 @@ Expected: exit 0. Exit 2 is pending: wait and read again. Exit 1: read the log a
## What the dry run could not check
- Tasks 8a, 8b, 8c, 9 and 10 were not compiled: they bind to Plan 6b rev 5's declared names, moves Plan 2 Task 8's `BuildGeneration` and key helpers, moves Plan 1 Task 7b's `RenderHeld` and `MaterializeHeld`, claim Plan 5's app gate and create door, and requests R1 to R3 and R7, none on main at 3407d114.
-- `StopWriters` against a live attempt (R2), and the probe through 6b's runtime (R1): their un-gated halves are tested with recorders; the live tests are Tasks 8a and 8c's.
+- `StopWriters` against a live attempt (R2), root's writers-stopped endpoint (R3) and the probe through 6b's runtime (R1): their un-gated halves are tested with recorders and the fixture `/proc`; the live tests, under the production uid, are Tasks 8a and 8c's.
- `WritersStopped` against a live agent with an embedded Litestream replicator. Its tests use a fixture `/proc`; the descriptor identification at replicator open belongs to Task 8.
- `genselect.Select` inside `StorePinner` against a real versioned bucket. The pin tests stub `SelectDB`.
- Socket-backed tests in `internal/host` inside the sandbox (the existing list in plan 1b's Global Constraints).
@@ -7998,3 +8794,4 @@ Expected: exit 0. Exit 2 is pending: wait and read again. Exit 1: read the log a
- **Type consistency.** `hostclaim.Release`, `RecoveryPin`, `PreparedAuthority`, `LocalInstall`, `Probe`, `Target`, `AutomaticLoss` (Task 1) are used unchanged by Tasks 5 to 8c; `hostrelocate.Effects` (Task 6) is implemented by `*host.RelocationEffects` (Task 7, `var _ hostrelocate.Effects`); `hostclaim.ReleaseAuthority` by `host.StopReceiptReleases` (`var _`); `hostrelocate.Lease` by `*lease.Held` (Task 10); `hostops.ExecutorChooser` (Task 2) by Task 10's worker adapters and the console constructor; `hostbox.RelocationRow` (Task 3) by Task 9; `hostrelocate.PinObject`, `ReadPin` and `CopyPinnedManifest` (Task 5) by Tasks 8b and 8c; `ReleaseFlushFiles`, `ReleasePositionsFile`, `RelocationControllers` (Task 7) by Task 8a.
- **Review Focus.** Each of the five has its test in the owning task. The dry run removed these guards one at a time and watched a test fail: `released`'s predecessor box; P's single opening; the failover pin delay (claim and worker); the pin reaching the release; the replacement bound; the nested failover after P; the receipt bound to `op.exec`; create's no-substitution; the start guard's live-source chore and region `park-replica`; the row case's closed fence and `lost`; the chooser's fence exclusion; discharge only when qualified; the writers-stopped thread-table check; the release receipt's obligation; the refusal of a replayed receipt; and a failed files flush stopping the release. Nineteen in all.
- **Rev 2 guards**, each removed in turn and its test watched to fail: the pin key bound to the claim generation; `ToInstalled`'s files manifest; the empty pin's evidence (claim); `stuck` before the phase table; the retained automatic loss; the lease lost before the pin CAS; the release bounds before the write (two); the empty pin's evidence (pinner); the pinned manifest's digest; the copied entries' equality; the target's eligibility at start; the session-0 drain refusal; `stuckOn` from `probe-failed`; the release deadline; the participant's readiness gate; `StopWriters` first in the drain; the replayed receipt without a record; the positions' session binding; the controller under the agent lifetime; the wake state in the flush; `Permits`; today's create without a selection; the `reconcile` action's participant and lost checks; the console's chooser off with HA off. Twenty-five in all.
+- **Rev 3 guards**, each removed in turn and its test watched to fail: the probe attempt bound; discharge through `CompleteObligation`; the former writer's last admission; the per-process descriptor scan; exhaustion before a spare is chosen; the retained manifest's digest; the attempt counted before the probe; the sweep's running-operation check; the recorded identity after a restart; the positions' box binding; `Finish` forgetting the controller. Eleven in all.