| 1 | package hostrelocate |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "slices" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/hostclaim" |
| 12 | "github.com/carlosframework/platform/internal/release" |
| 13 | ) |
| 14 | |
| 15 | // Effects are a participant box's local effects, each under the host |
| 16 | // guard and bound to the claim observation it is given. Plan 6b's box |
| 17 | // assembly implements them (ClaimAssembly); this package only orders them |
| 18 | // and records each result in the claim. |
| 19 | type Effects interface { |
| 20 | // Acknowledge writes this box's pending controller state, refuses while |
| 21 | // root has an update begun (§5.10), takes the execution lease |
| 22 | // (hostexec.Open under the agent's lifetime, then Ready) and returns |
| 23 | // the acknowledgement to journal. After an agent restart it opens a |
| 24 | // fresh controller for the same participation; the journaled |
| 25 | // acknowledgement (the first lease, D6) is left as it is. |
| 26 | Acknowledge(context.Context, hostclaim.Observed) (hostclaim.ExecutionAcknowledgement, error) |
| 27 | // Ready reports whether this box's execution controller for the |
| 28 | // operation is open and active now: ErrNoController when none is |
| 29 | // retained (the agent restarted, or a controller that lost its lease |
| 30 | // was joined and its scoped cleanup repeated), and the join's or the |
| 31 | // cleanup's error while a lost one cannot be retired yet. |
| 32 | // Every effect after the acknowledgement runs only after Ready. |
| 33 | Ready(context.Context, hostclaim.Observed) error |
| 34 | // Release runs §5.8's release sequence for this box's current writer |
| 35 | // session: queue main's relocation-release obligation, stop every |
| 36 | // writer (main's ClaimAssembly.WritersStopped, with root's own |
| 37 | // observation), flush the database to its final TXID, publish every |
| 38 | // pending file and wake-state mutation, close the replicator and |
| 39 | // publisher, complete full stop evidence |
| 40 | // (ClaimStopCoordinator.FenceAndStop), and discharge the obligation with |
| 41 | // root's receipt. It returns the journal entry; it is idempotent for |
| 42 | // one session. |
| 43 | Release(context.Context, hostclaim.Observed) (hostclaim.Release, error) |
| 44 | // Build builds the prepared generation P from the journaled pin with |
| 45 | // moves' shared install routine (L16), under the replica lease. |
| 46 | Build(context.Context, hostclaim.Observed) error |
| 47 | // InstallLocal makes this box's local install from P's files manifest |
| 48 | // and returns the receipt (§5.8 "The local install"). |
| 49 | InstallLocal(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, error) |
| 50 | // EnsureLocal is called before P's first admission, each time: it |
| 51 | // verifies that the local install the journaled receipt names is |
| 52 | // complete on this box and, when it is not (a lost lease's scoped |
| 53 | // cleanup retires an unopened P's local install), restores it under |
| 54 | // install_stamp from P's files manifest. It is never called once P is |
| 55 | // opened: an opened P is never reinstalled (rev 15, round 14). A |
| 56 | // repair may write the claim: the typed inventory, and the receipt |
| 57 | // itself when the repaired journal's digest differs (a configuration |
| 58 | // changed since it was journaled; final review A3), replaced in one |
| 59 | // CAS before any admission. The caller re-reads the same installation |
| 60 | // after it. |
| 61 | EnsureLocal(context.Context, hostclaim.Observed) error |
| 62 | // VerifyReceipt is called before every admission after P is opened: |
| 63 | // it reports whether this box's durable local install receipt is |
| 64 | // still the one the claim journaled (box, enrollment, install_stamp, |
| 65 | // journal digest). It checks the receipt and its binding only, never |
| 66 | // the files, env or timers, which the admitted writers have changed |
| 67 | // legitimately since. A missing or different receipt is false: the |
| 68 | // local state cannot be rebuilt from P, so the operation is stuck |
| 69 | // (§5.8, B2-5; rev 16, round 15). |
| 70 | VerifyReceipt(context.Context, hostclaim.Observed) (bool, error) |
| 71 | // Probe makes one cold probe admission (AdmitWriter, PredecessorCertified |
| 72 | // as the caller names it, coverage "" as claim Plan 11's relocating |
| 73 | // admission requires), runs the probe and waits for it to complete before |
| 74 | // returning (R1: one attempt, its exact health). passed is the probe's |
| 75 | // result; false is a failed attempt, never a probe still running. The |
| 76 | // caller counts an attempt in the claim before each call, so a Probe |
| 77 | // that returned early would spend the attempts without probing. |
| 78 | Probe(ctx context.Context, obs hostclaim.Observed, certified bool) (session uint64, passed bool, err error) |
| 79 | // PassedProbe reports the probe this box already made for the claim's |
| 80 | // admitted session when its own durable attempt record holds it as |
| 81 | // healthy (stage B final review A2): after Probe returned passed, the |
| 82 | // claim read or the CAS that records the result may have failed, |
| 83 | // leaving the live, passed session with no journaled probe. It is |
| 84 | // asked once P is opened and no probe is journaled, before Preflight |
| 85 | // or any attempt, so the result is recovered rather than a new attempt |
| 86 | // launched (none can be: ruling 58, until claim ask C11). ok is false |
| 87 | // when the record holds no healthy start for the holder's session; a |
| 88 | // record that cannot be read is the error, never "no probe". |
| 89 | PassedProbe(ctx context.Context, obs hostclaim.Observed) (session uint64, ok bool, err error) |
| 90 | // Preflight is called in probing before an attempt is counted (region |
| 91 | // HA ruling 48): root's isolated launch must speak the protocol and |
| 92 | // install an attempt reader for relocating, or nothing is counted. |
| 93 | Preflight(context.Context, hostclaim.Observed) error |
| 94 | // EndProbe stops this box's passed probe session while the op is still |
| 95 | // relocating/probing and returns root's receipt for it (ruling 49): |
| 96 | // the activator's local fence and stop, then the claim's writer-session |
| 97 | // stop through root. Idempotent for one session. |
| 98 | EndProbe(context.Context, hostclaim.Observed) (hostclaim.ProbeStop, error) |
| 99 | // Retained lists the participations whose execution controllers this |
| 100 | // box still holds; Finish closes one and releases its lifetime lock. |
| 101 | // A controller is finished once the claim no longer runs that |
| 102 | // operation with this box as a participant (Sweep), so the next |
| 103 | // relocation of the host can open its own (hostexec keeps the lock |
| 104 | // until Close). Retained also lists every participation journaled on |
| 105 | // disk whose durable record a restart still has to clear; an entry it |
| 106 | // cannot read is the error, returned beside every one it can. |
| 107 | Retained() ([]Participation, error) |
| 108 | Finish(ctx context.Context, host, op string) error |
| 109 | } |
| 110 | |
| 111 | // Participation names one retained execution controller. |
| 112 | type Participation struct { |
| 113 | Host, Op string |
| 114 | } |
| 115 | |
| 116 | // Participant is one box's half of relocation. Step makes at most one |
| 117 | // participant transition for one host, chosen from the claim alone. |
| 118 | type Participant struct { |
| 119 | Claims release.Store |
| 120 | Box string |
| 121 | Enrollment string |
| 122 | Effects Effects |
| 123 | Now func() time.Time |
| 124 | } |
| 125 | |
| 126 | // Step acts for host if this box has work in its relocation, and reports |
| 127 | // ErrNotYet when it has none. |
| 128 | func (p Participant) Step(ctx context.Context, host string) (hostclaim.Observed, error) { |
| 129 | if p.Claims == nil || p.Box == "" || p.Enrollment == "" || p.Effects == nil || p.Now == nil { |
| 130 | return hostclaim.Observed{}, ErrUnsupported |
| 131 | } |
| 132 | obs, err := hostclaim.Read(ctx, p.Claims, host) |
| 133 | if err != nil || obs.Absent { |
| 134 | return obs, err |
| 135 | } |
| 136 | c := obs.Claim |
| 137 | if c.Op.Kind != hostclaim.KindRelocating || c.Op.Journal["stuck"] != "" || !slices.Contains(c.Op.Participants, p.Box) || slices.Contains(c.Lost, p.Box) { |
| 138 | return obs, ErrNotYet |
| 139 | } |
| 140 | id := c.Op.ID |
| 141 | if c.Op.Journal["exec_ack."+p.Box] == "" { |
| 142 | ack, err := p.Effects.Acknowledge(ctx, obs) |
| 143 | if err != nil { |
| 144 | return obs, err |
| 145 | } |
| 146 | return p.update(ctx, obs, hostclaim.AcknowledgeExecution(p.Box, ack)) |
| 147 | } |
| 148 | // The controller is checked before every effect, never inferred from |
| 149 | // the historical acknowledgement: a restarted agent reopens one (the |
| 150 | // claim keeps the first lease's record) and acts on the next pass. |
| 151 | if err := p.Effects.Ready(ctx, obs); errors.Is(err, ErrNoController) { |
| 152 | _, err := p.Effects.Acknowledge(ctx, obs) |
| 153 | return obs, errors.Join(ErrNotYet, err) |
| 154 | } else if err != nil { |
| 155 | return obs, err |
| 156 | } |
| 157 | executor := c.Op.Exec == p.Box |
| 158 | switch { |
| 159 | case executor && (c.Op.Phase == hostclaim.PhaseReleasing || c.Op.Phase == hostclaim.PhaseProbeFailed) && c.Op.Journal["release"] == "": |
| 160 | r, err := p.Effects.Release(ctx, obs) |
| 161 | if err != nil { |
| 162 | return obs, err |
| 163 | } |
| 164 | // The release wrote the claim (main's relocation-release obligation, |
| 165 | // queued before root's stop and discharged by its receipt): journal |
| 166 | // it on a fresh read. |
| 167 | fresh, err := hostclaim.Read(ctx, p.Claims, host) |
| 168 | if err != nil || fresh.Absent { |
| 169 | return obs, errors.Join(ErrNotYet, err) |
| 170 | } |
| 171 | return p.update(ctx, fresh, hostclaim.JournalRelease(id, r)) |
| 172 | case executor && c.Op.Phase == hostclaim.PhaseInstalling: |
| 173 | return obs, errors.Join(ErrNotYet, p.Effects.Build(ctx, obs)) |
| 174 | case executor && c.Op.Phase == hostclaim.PhaseInstalled: |
| 175 | if c.Op.Journal["local_install"] != "" { |
| 176 | // The receipt is a past fact; the files, env and timers it |
| 177 | // names are checked (and restored) before probing. |
| 178 | if err := p.Effects.EnsureLocal(ctx, obs); err != nil { |
| 179 | return obs, err |
| 180 | } |
| 181 | // A repairing EnsureLocal may have written the claim (the |
| 182 | // typed inventory): the transition is made on a fresh read of |
| 183 | // the same installation (rev 3, Astra round 2 finding 2). |
| 184 | fresh, err := p.sameInstallation(ctx, obs, hostclaim.PhaseInstalled) |
| 185 | if err != nil { |
| 186 | return obs, err |
| 187 | } |
| 188 | // The target enters probing itself (§5.8 phase table). |
| 189 | return p.update(ctx, fresh, hostclaim.ToProbing(id, p.Now())) |
| 190 | } |
| 191 | l, err := p.Effects.InstallLocal(ctx, obs) |
| 192 | if err != nil { |
| 193 | return obs, err |
| 194 | } |
| 195 | if l.Box != p.Box || l.Enrollment != p.Enrollment { |
| 196 | return obs, ErrRefused |
| 197 | } |
| 198 | // The install recorded its typed inventory in the claim (park's |
| 199 | // RecordInventory): record the receipt on a fresh read of the |
| 200 | // same installation, never on the stale observation. |
| 201 | fresh, err := p.sameInstallation(ctx, obs, hostclaim.PhaseInstalled) |
| 202 | if err != nil { |
| 203 | return obs, err |
| 204 | } |
| 205 | return p.update(ctx, fresh, hostclaim.RecordLocalInstall(id, l)) |
| 206 | case executor && c.Op.Phase == hostclaim.PhaseRelocatingProbing && c.Op.Journal["probe"] == "": |
| 207 | if !hostclaim.PreparedOpened(c) { |
| 208 | // Before P's first admission the local install must be whole. |
| 209 | if err := p.Effects.EnsureLocal(ctx, obs); err != nil { |
| 210 | return obs, err |
| 211 | } |
| 212 | } else { |
| 213 | // After it, the install is never remade; its receipt must |
| 214 | // still be this box's, or the operation is stuck. |
| 215 | ok, err := p.Effects.VerifyReceipt(ctx, obs) |
| 216 | if err != nil { |
| 217 | return obs, err |
| 218 | } |
| 219 | if !ok { |
| 220 | return p.update(ctx, obs, hostclaim.MarkStuck(id, "the local install receipt is missing after P was admitted (§5.8, B2-5)")) |
| 221 | } |
| 222 | // A probe this box already passed whose record never reached |
| 223 | // the claim (the read or the CAS after Probe failed) is |
| 224 | // recovered from the box's durable attempt record, keyed by |
| 225 | // the admitted session, before Preflight can turn the live, |
| 226 | // passed session into probe-failed (final review A2). |
| 227 | session, passed, err := p.Effects.PassedProbe(ctx, obs) |
| 228 | if err != nil { |
| 229 | return obs, err |
| 230 | } |
| 231 | if passed { |
| 232 | return p.recordProbe(ctx, obs, session) |
| 233 | } |
| 234 | } |
| 235 | // EnsureLocal may have written the claim: count on a fresh read |
| 236 | // of the same installation. |
| 237 | if !hostclaim.PreparedOpened(c) { |
| 238 | fresh, err := p.sameInstallation(ctx, obs, hostclaim.PhaseRelocatingProbing) |
| 239 | if err != nil { |
| 240 | return obs, err |
| 241 | } |
| 242 | obs, c = fresh, fresh.Claim |
| 243 | } |
| 244 | // The release the admission would certify is read before any |
| 245 | // attempt is counted (review I3, M1): a journaled release that is |
| 246 | // unreadable, or names another writer than the predecessor, is a |
| 247 | // claim defect no attempt can mend, so the operation is stuck for |
| 248 | // an operator and no attempt is spent on it. |
| 249 | certified, err := CertifiedFor(c) |
| 250 | if err != nil { |
| 251 | return p.update(ctx, obs, hostclaim.MarkStuck(id, "the journaled release cannot certify P's opening (§5.8, C4): "+err.Error())) |
| 252 | } |
| 253 | // Root must be able to authorize a relocating launch before an |
| 254 | // attempt is spent (ruling 48): a refusal counts nothing. A later |
| 255 | // attempt root cannot launch (ruling 58, until claim ask C11) is |
| 256 | // probe-failed now, not three uncountable refusals. |
| 257 | if err := p.Effects.Preflight(ctx, obs); errors.Is(err, ErrLaterAttemptUnlaunchable) { |
| 258 | return p.update(ctx, obs, hostclaim.ToProbeFailed(id, p.Now())) |
| 259 | } else if err != nil { |
| 260 | return obs, errors.Join(ErrNotYet, err) |
| 261 | } |
| 262 | // The attempt is counted in the claim before it is made (§5.8: |
| 263 | // at most three); the fourth is probe-failed, by this CAS. |
| 264 | counted, err := hostclaim.Update(ctx, p.Claims, obs, hostclaim.BeginProbeAttempt(id, p.Box)) |
| 265 | if errors.Is(err, hostclaim.ErrProbeAttemptsExhausted) { |
| 266 | return p.update(ctx, obs, hostclaim.ToProbeFailed(id, p.Now())) |
| 267 | } |
| 268 | if err != nil { |
| 269 | return obs, err |
| 270 | } |
| 271 | obs, c = counted, counted.Claim |
| 272 | session, passed, err := p.Effects.Probe(ctx, obs, certified) |
| 273 | if err != nil || !passed { |
| 274 | return obs, errors.Join(ErrNotYet, err) |
| 275 | } |
| 276 | return p.recordProbe(ctx, obs, session) |
| 277 | case executor && c.Op.Phase == hostclaim.PhaseRelocatingProbing && c.Op.Journal["probe_stop"] == "": |
| 278 | // Ruling 49: the passed probe's session is stopped, with root's |
| 279 | // receipt, before the serving CAS ends it. |
| 280 | s, err := p.Effects.EndProbe(ctx, obs) |
| 281 | if err != nil { |
| 282 | return obs, err |
| 283 | } |
| 284 | fresh, err := hostclaim.Read(ctx, p.Claims, host) |
| 285 | if err != nil || fresh.Absent { |
| 286 | return obs, errors.Join(ErrNotYet, err) |
| 287 | } |
| 288 | return p.update(ctx, fresh, hostclaim.RecordProbeStop(id, s)) |
| 289 | } |
| 290 | return obs, ErrNotYet |
| 291 | } |
| 292 | |
| 293 | // recordProbe journals the passed probe of session on a fresh read: the |
| 294 | // probe's admission wrote the claim, and RecordProbe binds the result to |
| 295 | // the admitted session. At is the time of this record; for a probe |
| 296 | // recovered from the box's attempt record (PassedProbe) that is later than |
| 297 | // the pass, never earlier, so the worker's release deadline (ruling 67) |
| 298 | // is counted from no sooner than the probe passed. |
| 299 | func (p Participant) recordProbe(ctx context.Context, obs hostclaim.Observed, session uint64) (hostclaim.Observed, error) { |
| 300 | fresh, err := hostclaim.Read(ctx, p.Claims, obs.Host) |
| 301 | if err != nil || fresh.Absent { |
| 302 | return obs, errors.Join(ErrNotYet, err) |
| 303 | } |
| 304 | return p.update(ctx, fresh, hostclaim.RecordProbe(obs.Claim.Op.ID, hostclaim.Probe{Box: p.Box, Session: session, At: p.Now()})) |
| 305 | } |
| 306 | |
| 307 | // ErrLaterAttemptUnlaunchable is Preflight's answer for a cold attempt |
| 308 | // after P's opening while root pins every relocating launch to |
| 309 | // install_stamp (claimLaunchAuthority, claim ask C11; ruling 58). |
| 310 | var ErrLaterAttemptUnlaunchable = errors.New("hostrelocate: root cannot launch a relocation attempt after P's opening") |
| 311 | |
| 312 | // sameInstallation re-reads the claim after an effect that may have written |
| 313 | // it and requires the same operation, executor, phase, prepared P and |
| 314 | // install stamp, unstuck: only the effect's own bookkeeping may differ. |
| 315 | func (p Participant) sameInstallation(ctx context.Context, obs hostclaim.Observed, phase string) (hostclaim.Observed, error) { |
| 316 | fresh, err := hostclaim.Read(ctx, p.Claims, obs.Host) |
| 317 | if err != nil { |
| 318 | return obs, err |
| 319 | } |
| 320 | a, b := obs.Claim, fresh.Claim |
| 321 | if fresh.Absent || b == nil || b.Holder == nil || b.Op.Kind != hostclaim.KindRelocating || b.Op.ID != a.Op.ID || b.Op.Exec != p.Box || b.Op.Phase != phase || |
| 322 | b.Op.Journal["prepared"] != a.Op.Journal["prepared"] || b.Op.Journal["install_stamp"] != a.Op.Journal["install_stamp"] || b.Op.Journal["stuck"] != "" || |
| 323 | b.Holder.Incarnation != a.Holder.Incarnation || b.Holder.Session != a.Holder.Session || b.Holder.Stamp != a.Holder.Stamp { |
| 324 | return obs, errors.Join(ErrNotYet, fmt.Errorf("%w: the installation changed under the effect", ErrRefused)) |
| 325 | } |
| 326 | return fresh, nil |
| 327 | } |
| 328 | |
| 329 | // Sweep finishes every retained controller whose participation is over: |
| 330 | // the claim no longer runs that operation, or this box is no longer a |
| 331 | // participant of it. It runs once per pass, for hosts the pass itself |
| 332 | // skips (a completed relocation is a serving claim). |
| 333 | func (p Participant) Sweep(ctx context.Context) error { |
| 334 | if p.Claims == nil || p.Effects == nil { |
| 335 | return ErrUnsupported |
| 336 | } |
| 337 | retained, err := p.Effects.Retained() |
| 338 | errs := []error{err} // reported, while every readable participation is still finished |
| 339 | for _, r := range retained { |
| 340 | obs, err := hostclaim.Read(ctx, p.Claims, r.Host) |
| 341 | if err != nil { |
| 342 | errs = append(errs, err) |
| 343 | continue |
| 344 | } |
| 345 | 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) { |
| 346 | continue |
| 347 | } |
| 348 | if err := p.Effects.Finish(ctx, r.Host, r.Op); err != nil { |
| 349 | errs = append(errs, err) |
| 350 | } |
| 351 | } |
| 352 | return errors.Join(errs...) |
| 353 | } |
| 354 | |
| 355 | // ErrRelease is a journaled release that is present but unreadable, or |
| 356 | // that names another writer than the holder's predecessor. Either is a |
| 357 | // claim defect, never "no release": an uncertified cold opening declares |
| 358 | // the predecessor's writes lost, which only a failover's evidence (the |
| 359 | // source lost and fenced, with no release at all) may do. |
| 360 | var ErrRelease = errors.New("hostrelocate: the journaled release is unreadable or does not name the predecessor") |
| 361 | |
| 362 | // CertifiedFor is whether the probe admission certifies its predecessor |
| 363 | // (§5.8, L9): only the admission that opens P, when this operation |
| 364 | // journaled a release whose session is the predecessor writer's. Claim |
| 365 | // Plan 11 gives a relocation no coverage value of its own, so the release |
| 366 | // is the generic PredecessorCertified, which AdmitWriter accepts on P's |
| 367 | // opening only with that release and a pin reaching it |
| 368 | // (relocationProbeOpening). No release at all (a failover, or a host that |
| 369 | // never had a writer) is an uncertified cold admission; after P's opening |
| 370 | // every later attempt is one too. A release that is present but |
| 371 | // unreadable, or that names another writer than the predecessor, is |
| 372 | // ErrRelease: absence is never evidence, and an unreadable or mismatched |
| 373 | // entry is not absence (fix round 1, I3). |
| 374 | func CertifiedFor(c *hostclaim.Claim) (bool, error) { |
| 375 | raw, ok := c.Op.Journal["release"] |
| 376 | if !ok { |
| 377 | return false, nil |
| 378 | } |
| 379 | var r hostclaim.Release |
| 380 | if json.Unmarshal([]byte(raw), &r) != nil || r.ReleasedBy == "" { |
| 381 | return false, fmt.Errorf("%w: unreadable", ErrRelease) |
| 382 | } |
| 383 | if hostclaim.PreparedOpened(c) { |
| 384 | return false, nil |
| 385 | } |
| 386 | if c.Holder == nil { |
| 387 | return false, fmt.Errorf("%w: no holder", ErrRelease) |
| 388 | } |
| 389 | pred := hostclaim.AdmissionRecord{Session: c.Holder.Session, Box: c.Holder.WriterBox, Generation: c.Holder.Generation} |
| 390 | if n := len(c.Holder.Admissions); n > 0 { |
| 391 | pred = c.Holder.Admissions[n-1] |
| 392 | } |
| 393 | if pred.Box != r.ReleasedBy || pred.Session != r.Session || pred.Generation != r.Generation { |
| 394 | return false, fmt.Errorf("%w: released by %s (session %d, generation %q), the predecessor is %s (session %d, generation %q)", ErrRelease, r.ReleasedBy, r.Session, r.Generation, pred.Box, pred.Session, pred.Generation) |
| 395 | } |
| 396 | return true, nil |
| 397 | } |
| 398 | |
| 399 | func (p Participant) update(ctx context.Context, obs hostclaim.Observed, tr func(*hostclaim.Claim) error) (hostclaim.Observed, error) { |
| 400 | next, err := hostclaim.Update(ctx, p.Claims, obs, tr) |
| 401 | if errors.Is(err, hostclaim.ErrAlreadyRecorded) || errors.Is(err, hostclaim.ErrAckAlreadyApplied) { |
| 402 | return obs, nil |
| 403 | } |
| 404 | return next, err |
| 405 | } |
| 406 | |
amadan