| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "reflect" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/hostclaim" |
| 13 | "github.com/carlosframework/platform/internal/moves" |
| 14 | "github.com/carlosframework/platform/internal/park" |
| 15 | "github.com/carlosframework/platform/internal/provision" |
| 16 | "github.com/carlosframework/platform/internal/registry" |
| 17 | "github.com/carlosframework/platform/internal/release" |
| 18 | ) |
| 19 | |
| 20 | // The revert on the source (moves spec §7.4 reverting; Plan 4 Task 6), |
| 21 | // behind CARLOS_MOVES_PHASE1: the pre-flip recovery of a cut abandoned before |
| 22 | // the flip. The record decided it (MoveRevert); the source takes the host |
| 23 | // back with its own evidence and runs a recovery journalled in the claim's |
| 24 | // rev_phase. revertSource is phase1MoveRound's MoveReverting arm (ruling J7), |
| 25 | // and movesPass reaches a reverting record through the participant rule |
| 26 | // (movesParticipantRecord, J10) as Tasks 4 and 5 do. |
| 27 | // |
| 28 | // Entry. A revert from stopping (or from stopped before execution was given |
| 29 | // up) finds the claim still moving with this box executing: the stop |
| 30 | // evidence was never obtained, so something may still run. The source holds |
| 31 | // the row moving and fences it again (an agent restart dropped the fence), |
| 32 | // drains the timers and asks for the evidence (§5.3), and starts nothing |
| 33 | // until it holds. Then the cut's own giveUpExecution publishes the complete |
| 34 | // stopped-source.json and gives execution up citing it (it records no |
| 35 | // baselines on a reverting record, J8), and Plan 2's revertClaim, unchanged, |
| 36 | // writes the claim half with its journalled revert_bound: reverting under the |
| 37 | // source, rev_phase evidence, the lane's park queued in the target region. |
| 38 | // No second staging scope is ever computed (J7). A revert decided after |
| 39 | // execution was given up starts at revertClaim. |
| 40 | // |
| 41 | // The recovery, from durable state, each step under the claim's rev_phase: |
| 42 | // |
| 43 | // - evidence: the evidence holds (Revert cites it). The row's hold goes |
| 44 | // from moving to restoring under the move id, a conditional status change |
| 45 | // under the host's hold lock with the record read fresh (RestatusHold; |
| 46 | // a revert never hands to a rollback id). Then rev_phase steps to verify |
| 47 | // when the record keeps the stopped baselines (Source), else straight to |
| 48 | // probing (J8), op.phase following from probing on (J1). |
| 49 | // - verify: the files plain digest and the timers digest equal what the |
| 50 | // source recorded at stopped (Source.Files.Plain, Source.TimersSHA256). |
| 51 | // A mismatch stops for an operator with the baseline-mismatch sentence |
| 52 | // (MoveMarkStuck); a retry returns to verify and checks again (INV-4). |
| 53 | // Passed, rev_phase and op.phase step to probing before the first attempt. |
| 54 | // - probing: the recovery probe only (recoveryProbe, Task 9's; warm or cold |
| 55 | // into the source's own generation, holder.generation, which a revert |
| 56 | // never changed). It records the inputs it renders before it renders them |
| 57 | // (MoveRecordRevertInputs). Never verify again: the probe's process may |
| 58 | // have written, and its writes are the authoritative state (INV-2). The |
| 59 | // probe has the recovery probe's nine attempts (§7.7), and the first |
| 60 | // attempt of a window is journalled: nine failed attempts |
| 61 | // (StuckRecoveryProbeAttempts) or fifteen minutes of failed probes |
| 62 | // (StuckRevertProbeBudget) stop for an operator. The window's end is |
| 63 | // journalled before the stuck mark, so a crash or a refused write between |
| 64 | // them still ends in the mark, never in a new window; only a retry starts |
| 65 | // one. Answered, the claim steps to probed. |
| 66 | // - probed: reopened_at on the record (MoveRevertReopened, J26), then the |
| 67 | // conditional open on hold_op = move id under the host's hold lock with |
| 68 | // the record read fresh (ReleaseHeld, a plain registry call, J12), then |
| 69 | // the claim steps to opened. |
| 70 | // - opened: the claim waits for the lane's park the revert queued in the |
| 71 | // target region (§7.11), which the target region's box runs |
| 72 | // (laneParkTarget) and the park engine acknowledges after its horizon. |
| 73 | // Then MoveRevertClose (LastMove rolled_back, offline from cut_at to |
| 74 | // the reopen), and then ReleaseMove to serving under this box: record |
| 75 | // first, ruling M4; a crash between is finished by the console's |
| 76 | // recovery pass from LastMove. The source's journal of the op is parked |
| 77 | // in the same pass; a crash before that leaves it to the local journal |
| 78 | // sweep (sweepRevertJournal). |
| 79 | // |
| 80 | // A claim this move already released (RequireReleasedMove: a binary that |
| 81 | // closed claim first crashed before the record's close) only closes. |
| 82 | // |
| 83 | // The target region's half (laneParkTarget). A revert never flipped, so its |
| 84 | // target box has neither the row nor a journal of the move; the participant |
| 85 | // rule answers it as the target on the state alone, and its one piece of work |
| 86 | // is the lane's park-replica the claim owes its region: the park engine over |
| 87 | // its own bucket archives moves/<host>/<op>/ under |
| 88 | // parked/<inc>/<host>/<ret>/moves/ and acknowledges once the horizon has |
| 89 | // passed (laneParkTarget dispatches an abort's the same way, Plan 5 Task 1). |
| 90 | // |
| 91 | // Every record write goes through a provision transition (the record_seq |
| 92 | // rule); every claim write through a hostclaim transition. |
| 93 | // |
| 94 | // Arming (rulings J18, G17). Like the landing and the rollback, the revert is |
| 95 | // unarmed in production until Plan 5: the recovery probe is nil there until |
| 96 | // Task 9 (claim Plan 10's recovery admission), so revertSourceArmed answers |
| 97 | // false and the source only writes the claim half, as Plan 2's arm did; and |
| 98 | // the target's lane park needs the claim's park leases, view and authorizer, |
| 99 | // which armMovesParks sets only once the wake-source gate opens |
| 100 | // (movesWakeSourceRegistered, Plan 5 Task 9b; revertLaneArmed). LogEvent is nil in production, |
| 101 | // where an event is a log line. |
| 102 | |
| 103 | // revertStepsPerPass bounds the steps one pass runs: evidence to the close is |
| 104 | // seven. |
| 105 | const revertStepsPerPass = 10 |
| 106 | |
| 107 | // revertProbeBudget is how long a revert's probes may fail before it stops |
| 108 | // for an operator (§7.4: "Fifteen minutes of failed probes: stuck"). |
| 109 | const revertProbeBudget = 15 * time.Minute |
| 110 | |
| 111 | // The revert's steps in the source's op journal (the cut's, under the move |
| 112 | // id). |
| 113 | const ( |
| 114 | // rvStepSource marks the journal as a revert's once the claim's recovery |
| 115 | // begins (revertJournalMark), so the sweep can tell it from a cut's that |
| 116 | // is still running, and knows the incarnation to park it under. |
| 117 | rvStepSource = "rv-source" |
| 118 | // rvStepProbeSince is the first probe attempt of this window of probing |
| 119 | // (RFC 3339), the start of the fifteen minutes. Cleared, with |
| 120 | // rvStepProbeExhausted, only once a retry has returned the record from |
| 121 | // stuck, so a retry starts a new window and nothing else does. |
| 122 | rvStepProbeSince = "rv-probe-since" |
| 123 | // rvStepProbeExhausted is the window's end, journalled before the stuck |
| 124 | // mark (probeExhaustion): the probe ran out of its nine attempts, or of |
| 125 | // its fifteen minutes. A replay finds it and marks the move stuck without |
| 126 | // probing again (Task 6 fix round 1). |
| 127 | rvStepProbeExhausted = "rv-probe-exhausted" |
| 128 | // rvStepClosed is "done" once the record closed. |
| 129 | rvStepClosed = "rv-source-closed" |
| 130 | // rvStepJournalPark is parkLocalArtifacts' step: the retirement the |
| 131 | // journal is parked into (no artifacts of its own). |
| 132 | rvStepJournalPark = "rv-source-journal-park" |
| 133 | ) |
| 134 | |
| 135 | // probeExhaustion is rvStepProbeExhausted's value: the recorded reason the |
| 136 | // window ended with, and the record's updated_at when it ended. SetRecordMove |
| 137 | // stamps updated_at on every transition, and once the window has ended the |
| 138 | // only transitions a reverting record takes are the stuck mark and then the |
| 139 | // operator's retry (this box records no probe inputs and no reopen after it, |
| 140 | // the door's abort refuses a reverting record, and MoveRevertClose needs the |
| 141 | // reopen). So a record still reverting at that updated_at has not been marked |
| 142 | // stuck, and one reverting at a later one has been retried. |
| 143 | type probeExhaustion struct { |
| 144 | Reason string `json:"reason"` |
| 145 | UpdatedAt time.Time `json:"updated_at"` |
| 146 | } |
| 147 | |
| 148 | // revertJournalMark is rvStepSource's value. |
| 149 | type revertJournalMark struct { |
| 150 | Move string `json:"move"` |
| 151 | Incarnation string `json:"incarnation"` |
| 152 | } |
| 153 | |
| 154 | // revertSourceArmed reports whether this box can run the revert's recovery: |
| 155 | // every hook it calls is wired, the recovery probe included. In production |
| 156 | // the probe is not (Task 9, J18), so this answers false there. |
| 157 | func revertSourceArmed(cfg moveCfg) bool { |
| 158 | return cfg.Box != "" && cfg.DataDir != "" && cfg.RestatusHold != nil && cfg.ReleaseHeld != nil && |
| 159 | cfg.Timers != nil && cfg.FilesDir != nil && cfg.recoveryProbe != nil |
| 160 | } |
| 161 | |
| 162 | // revertLaneArmed reports whether this box can run the target region's lane |
| 163 | // park: an engine region (parkRegion, the home's "" included as |
| 164 | // park.HomeRegion; a code that is no placement is not one), the claim's |
| 165 | // park leases, view and authorizer, and this box's own bucket. In |
| 166 | // production armMovesParks sets the three only once the wake-source gate |
| 167 | // opens (movesWakeSourceRegistered, Plan 5 Task 9b), so this |
| 168 | // answers false there until it lands. |
| 169 | func revertLaneArmed(cfg moveCfg) bool { |
| 170 | return cfg.Box != "" && cfg.parkRegion() != "" && cfg.OwnBucket != "" && parkHooksArmed(cfg) && cfg.Versioned != nil |
| 171 | } |
| 172 | |
| 173 | // restatusHold is moveCfg.RestatusHold in production: an op's own hold |
| 174 | // changes status (a revert's moving to restoring under the move id) by |
| 175 | // registry.SetHoldIf expecting the op itself, under the host's hold lock, |
| 176 | // after the caller's validation; written (holdStageRequester) runs when the |
| 177 | // change landed, so root re-judges what it staged. A row already held status |
| 178 | // by op is a replay and answers true; a row held by another op, unheld or |
| 179 | // absent answers false and is not touched. |
| 180 | func restatusHold(reg *registry.Registry, written func(host string)) func(host, status, op string, validate func() error) (bool, error) { |
| 181 | return func(host, status, op string, validate func() error) (bool, error) { |
| 182 | var held bool |
| 183 | var err error |
| 184 | provision.WithHostHoldLock(host, func() { |
| 185 | var rt *registry.Route |
| 186 | if rt, err = reg.Get(host); err != nil || rt == nil { |
| 187 | return |
| 188 | } |
| 189 | if validate != nil { |
| 190 | if err = validate(); err != nil { |
| 191 | return |
| 192 | } |
| 193 | } |
| 194 | if op == "" || rt.HoldOp != op { |
| 195 | return |
| 196 | } |
| 197 | if rt.Status == status { |
| 198 | held = true |
| 199 | return |
| 200 | } |
| 201 | if held, err = reg.SetHoldIf(host, status, op, op); err != nil || !held { |
| 202 | return |
| 203 | } |
| 204 | if written != nil { |
| 205 | written(host) |
| 206 | } |
| 207 | }) |
| 208 | return held, err |
| 209 | } |
| 210 | } |
| 211 | |
| 212 | // revertRun is one pass's state for one reverting record on its source. |
| 213 | type revertRun struct { |
| 214 | w *movesWorker |
| 215 | s release.Store |
| 216 | rec provision.Record |
| 217 | g provision.MoveGuard |
| 218 | fleet *moveFleet // nil when the pass could not read it: the entry waits |
| 219 | } |
| 220 | |
| 221 | // revertSource is phase1MoveRound's MoveReverting arm (J7) and the source's |
| 222 | // participant step for a reverting record (movesPass). |
| 223 | func (w *movesWorker) revertSource(ctx context.Context, s release.Store, rec provision.Record, fleet *moveFleet) error { |
| 224 | cfg := w.cfg |
| 225 | m := rec.Move |
| 226 | if m == nil || m.Legacy() || m.State != provision.MoveReverting || m.From != cfg.Region { |
| 227 | return nil |
| 228 | } |
| 229 | if !revertSourceArmed(cfg) { |
| 230 | // The record decided; the claim follows it (§7.5), as Plan 2's arm |
| 231 | // did. The recovery waits for Task 9's probe. |
| 232 | if err := w.revertClaim(ctx, s, rec); err != nil { |
| 233 | return err |
| 234 | } |
| 235 | w.notice(rec.Host+"#revert", "moves: %s: the revert of %s waits: this box's revert is not armed (the recovery probe is Task 9's, on claim Plan 10's recovery admission)", rec.Host, m.ID) |
| 236 | return nil |
| 237 | } |
| 238 | r := &revertRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, fleet: fleet} |
| 239 | for i := 0; i < revertStepsPerPass; i++ { |
| 240 | advanced, err := r.step(ctx) |
| 241 | if err != nil || !advanced { |
| 242 | return err |
| 243 | } |
| 244 | fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| 245 | if err != nil { |
| 246 | return fmt.Errorf("re-read the record: %w", err) |
| 247 | } |
| 248 | if !r.same(fresh) { |
| 249 | return nil |
| 250 | } |
| 251 | r.rec = *fresh |
| 252 | } |
| 253 | return nil |
| 254 | } |
| 255 | |
| 256 | func (r *revertRun) cfg() moveCfg { return r.w.cfg } |
| 257 | func (r *revertRun) now() time.Time { return r.w.now() } |
| 258 | |
| 259 | // same reports whether fresh is still this move, reverting. |
| 260 | func (r *revertRun) same(fresh *provision.Record) bool { |
| 261 | m := fresh.Move |
| 262 | return m != nil && m.ID == r.g.ID && m.Incarnation == r.g.Incarnation && m.State == provision.MoveReverting |
| 263 | } |
| 264 | |
| 265 | // fault is the test seam after a side effect. |
| 266 | func (r *revertRun) fault(point string) error { |
| 267 | if f := r.cfg().rvFault; f != nil { |
| 268 | return f(point) |
| 269 | } |
| 270 | return nil |
| 271 | } |
| 272 | |
| 273 | // stillReverting is the record check run under the host's hold lock before |
| 274 | // a row write: the record, read fresh, must still be this move reverting. |
| 275 | func (r *revertRun) stillReverting(ctx context.Context) error { |
| 276 | fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host) |
| 277 | if err != nil { |
| 278 | return fmt.Errorf("re-read the record: %w", err) |
| 279 | } |
| 280 | if !r.same(fresh) { |
| 281 | return fmt.Errorf("move %s: the record is no longer reverting", r.g.ID) |
| 282 | } |
| 283 | return nil |
| 284 | } |
| 285 | |
| 286 | // update writes one claim transition over obs. |
| 287 | func (r *revertRun) update(ctx context.Context, obs hostclaim.Observed, what string, step func(*hostclaim.Claim) error) error { |
| 288 | after, err := hostclaim.Update(ctx, r.s, obs, step) |
| 289 | if err != nil { |
| 290 | return fmt.Errorf("%s: %w", what, err) |
| 291 | } |
| 292 | r.w.logRevPhase(ctx, r.s, r.rec, obs, after) |
| 293 | return nil |
| 294 | } |
| 295 | |
| 296 | // record writes one record transition. A conflict is someone else's |
| 297 | // decision: logged, and the pass ends without error. |
| 298 | func (r *revertRun) record(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) { |
| 299 | var pre provision.Record |
| 300 | after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host, observedMove(&pre, mut)) |
| 301 | if err != nil { |
| 302 | if errors.Is(err, provision.ErrMoveConflict) { |
| 303 | log.Printf("moves: %s: %s of %s refused (%v); the record moved on", r.rec.Host, what, r.g.ID, err) |
| 304 | return false, nil |
| 305 | } |
| 306 | return false, fmt.Errorf("%s: %w", what, err) |
| 307 | } |
| 308 | r.w.logTransition(ctx, r.s, pre, *after) |
| 309 | return true, nil |
| 310 | } |
| 311 | |
| 312 | // step runs the revert's current step once, from the claim. advanced is true |
| 313 | // when it wrote the claim or the record. |
| 314 | func (r *revertRun) step(ctx context.Context) (bool, error) { |
| 315 | cfg := r.cfg() |
| 316 | host, id := r.rec.Host, r.g.ID |
| 317 | obs, err := hostclaim.Read(ctx, r.s, host) |
| 318 | if err != nil { |
| 319 | return false, fmt.Errorf("read the host claim: %w", err) |
| 320 | } |
| 321 | if obs.Absent { |
| 322 | return false, fmt.Errorf("move %s has no host claim", id) |
| 323 | } |
| 324 | c := obs.Claim |
| 325 | if hostclaim.RequireReleasedMove(id)(c) == nil { |
| 326 | // A binary that closed claim first crashed between the claim's |
| 327 | // release and the record's close: only the record's half is left. |
| 328 | return r.close(ctx, false) |
| 329 | } |
| 330 | if c.Op.ID != id { |
| 331 | return false, fmt.Errorf("move %s: the claim names op %q", id, c.Op.ID) |
| 332 | } |
| 333 | if c.Holder == nil || c.Holder.Incarnation != r.g.Incarnation { |
| 334 | return false, fmt.Errorf("move %s: the host claim names another incarnation", id) |
| 335 | } |
| 336 | if len(c.Op.Participants) == 0 || c.Op.Participants[0] != cfg.Box { |
| 337 | return false, fmt.Errorf("move %s: this box is not the move's source in the claim", id) |
| 338 | } |
| 339 | switch c.Op.Kind { |
| 340 | case hostclaim.KindMoving: |
| 341 | if c.Op.Exec == cfg.Box { |
| 342 | return r.evidence(ctx, obs) |
| 343 | } |
| 344 | // Execution given up with the source's evidence: the claim half. |
| 345 | if err := r.w.revertClaim(ctx, r.s, r.rec); err != nil { |
| 346 | return false, err |
| 347 | } |
| 348 | fresh, err := hostclaim.Read(ctx, r.s, host) |
| 349 | if err != nil { |
| 350 | return false, fmt.Errorf("read the host claim: %w", err) |
| 351 | } |
| 352 | if fresh.Absent || fresh.Claim.Op.Kind != hostclaim.KindReverting { |
| 353 | r.w.notice(host+"#revert", "moves: %s: the revert of %s waits: the claim is moving with executor %q", host, id, c.Op.Exec) |
| 354 | return false, nil |
| 355 | } |
| 356 | return true, nil |
| 357 | case hostclaim.KindReverting: |
| 358 | default: |
| 359 | return false, fmt.Errorf("move %s: the claim is %s, not reverting", id, c.Op.Kind) |
| 360 | } |
| 361 | if c.Op.Exec != cfg.Box { |
| 362 | return false, fmt.Errorf("move %s: the claim's executor is %q, not this box", id, c.Op.Exec) |
| 363 | } |
| 364 | switch phase := c.Op.Journal[hostclaim.JournalRevPhase]; phase { |
| 365 | case hostclaim.RevPhaseEvidence: |
| 366 | return r.begin(ctx, obs) |
| 367 | case hostclaim.RevPhaseVerify: |
| 368 | return r.verify(ctx, obs) |
| 369 | case hostclaim.PhaseProbing: |
| 370 | return r.probing(ctx, obs) |
| 371 | case hostclaim.PhaseProbed: |
| 372 | return r.probed(ctx, obs) |
| 373 | case hostclaim.PhaseOpened: |
| 374 | return r.opened(ctx, obs) |
| 375 | default: |
| 376 | return false, fmt.Errorf("move %s: the claim's rev_phase is %q", id, phase) |
| 377 | } |
| 378 | } |
| 379 | |
| 380 | // evidence is the entry from stopping (or from stopped before execution was |
| 381 | // given up): the hold, the fence and the drain again, the stop evidence until |
| 382 | // it holds, then the cut's giveUpExecution and Plan 2's revertClaim. |
| 383 | func (r *revertRun) evidence(ctx context.Context, obs hostclaim.Observed) (bool, error) { |
| 384 | cfg := r.cfg() |
| 385 | host, id := r.rec.Host, r.g.ID |
| 386 | if err := hostclaim.RequireMoveOp(id, hostclaim.KindMoving, hostclaim.PhaseMoving)(obs.Claim); err != nil { |
| 387 | return false, fmt.Errorf("move %s: %w", id, err) |
| 388 | } |
| 389 | if r.fleet == nil { |
| 390 | return false, fmt.Errorf("move %s: the fleet's data buckets could not be read; the revert's evidence waits", id) |
| 391 | } |
| 392 | c := &cutRun{w: r.w, s: r.s, fleet: *r.fleet, rec: r.rec, g: r.g} |
| 393 | defer c.releaseLease() |
| 394 | holder := *obs.Claim.Holder |
| 395 | if err := c.holderSupported(holder); err != nil { |
| 396 | return false, fmt.Errorf("move %s: %w", id, err) |
| 397 | } |
| 398 | if cfg.HoldMoving == nil || cfg.Fence == nil || cfg.TimersDrained == nil || cfg.Evidence == nil { |
| 399 | return false, errors.New("this box cannot stop an instance for a revert: a hook is not wired") |
| 400 | } |
| 401 | if err := cfg.HoldMoving(host, id); err != nil { |
| 402 | return false, fmt.Errorf("hold the row: %w", err) |
| 403 | } |
| 404 | gen := moveFenceGeneration(r.rec.Move) |
| 405 | if err := cfg.Fence(ctx, host, gen); err != nil { |
| 406 | return false, fmt.Errorf("fence against wakes: %w", err) |
| 407 | } |
| 408 | r.w.fenced.Store(host, gen) |
| 409 | if err := cfg.TimersDrained(ctx, host); err != nil { |
| 410 | return false, fmt.Errorf("drain the timers: %w", err) |
| 411 | } |
| 412 | j, err := c.journal() |
| 413 | if err != nil { |
| 414 | return false, err |
| 415 | } |
| 416 | var ev moves.StoppedReport |
| 417 | if raw, ok := j.Steps[cutStepEvidence]; !ok || json.Unmarshal([]byte(raw), &ev) != nil || !ev.TreeStopped() { |
| 418 | rep, err := cfg.Evidence(ctx, host, holder.Location, []string{c.dbPath()}) |
| 419 | if err != nil { |
| 420 | return false, fmt.Errorf("stop evidence: %w", err) |
| 421 | } |
| 422 | if !rep.TreeStopped() { |
| 423 | r.w.notice(host+"#rv-evidence", "moves: %s: the revert of %s starts nothing until the instance is proven stopped (%s)", host, id, failedSteps(rep)) |
| 424 | return false, nil |
| 425 | } |
| 426 | body, err := json.Marshal(rep) |
| 427 | if err != nil { |
| 428 | return false, err |
| 429 | } |
| 430 | j.Mark(cutStepEvidence, string(body)) |
| 431 | if err := j.Commit(cfg.DataDir); err != nil { |
| 432 | return false, err |
| 433 | } |
| 434 | if err := r.fault("after-revert-evidence"); err != nil { |
| 435 | return false, err |
| 436 | } |
| 437 | } |
| 438 | stoppedKey := moves.StoppedKey(moves.ReportPrefix(r.rec.Account, r.rec.App, host, id), "source") |
| 439 | report, err := c.giveUpExecution(ctx, obs, stoppedKey) |
| 440 | if err != nil { |
| 441 | return false, err |
| 442 | } |
| 443 | if report.V == 0 { |
| 444 | return false, nil // a file changed under the digest: the next pass |
| 445 | } |
| 446 | if err := r.w.revertClaim(ctx, r.s, r.rec); err != nil { |
| 447 | return false, err |
| 448 | } |
| 449 | if err := r.fault("after-revert-claim"); err != nil { |
| 450 | return false, err |
| 451 | } |
| 452 | return true, nil |
| 453 | } |
| 454 | |
| 455 | // begin is rev_phase evidence on a reverting claim: the journal marked as |
| 456 | // this revert's, the row's hold restoring under the move id, then verify or |
| 457 | // probing. |
| 458 | func (r *revertRun) begin(ctx context.Context, obs hostclaim.Observed) (bool, error) { |
| 459 | cfg := r.cfg() |
| 460 | host, id := r.rec.Host, r.g.ID |
| 461 | if obs.Claim.Op.Journal[hostclaim.JournalEvidenceSource] == "" { |
| 462 | return false, fmt.Errorf("move %s: the revert cites no stop evidence", id) |
| 463 | } |
| 464 | if err := r.markJournal(); err != nil { |
| 465 | return false, err |
| 466 | } |
| 467 | held, err := cfg.RestatusHold(host, "restoring", id, func() error { return r.stillReverting(ctx) }) |
| 468 | if err != nil { |
| 469 | return false, fmt.Errorf("hold the row restoring: %w", err) |
| 470 | } |
| 471 | if !held { |
| 472 | return false, fmt.Errorf("move %s: this box's row is not held by the move", id) |
| 473 | } |
| 474 | if err := r.fault("after-restatus"); err != nil { |
| 475 | return false, err |
| 476 | } |
| 477 | baseline := r.rec.Move.Source != nil |
| 478 | to := hostclaim.PhaseProbing |
| 479 | if baseline { |
| 480 | to = hostclaim.RevPhaseVerify |
| 481 | } |
| 482 | if err := r.update(ctx, obs, "the revert's "+to, hostclaim.SetRevPhase(id, hostclaim.RevPhaseEvidence, to, baseline)); err != nil { |
| 483 | return false, err |
| 484 | } |
| 485 | log.Printf("moves: %s: the revert of %s is at %s", host, id, to) |
| 486 | if err := r.fault("after-rev-" + to); err != nil { |
| 487 | return false, err |
| 488 | } |
| 489 | return true, nil |
| 490 | } |
| 491 | |
| 492 | // verify (§7.4 verify, INV-4): the files and the timers against the digests |
| 493 | // recorded at stopped. A mismatch stops for an operator; a retry runs this |
| 494 | // again. |
| 495 | func (r *revertRun) verify(ctx context.Context, obs hostclaim.Observed) (bool, error) { |
| 496 | cfg := r.cfg() |
| 497 | host, id, m := r.rec.Host, r.g.ID, r.rec.Move |
| 498 | if err := r.fault("verify"); err != nil { |
| 499 | return false, err |
| 500 | } |
| 501 | if obs.Claim.Op.Phase != hostclaim.PhaseMoving && obs.Claim.Op.Phase != hostclaim.PhaseSealed { |
| 502 | return false, fmt.Errorf("move %s: op.phase %q at rev_phase verify", id, obs.Claim.Op.Phase) |
| 503 | } |
| 504 | if m.Source == nil || m.Source.Files == nil || m.Source.TimersSHA256 == "" { |
| 505 | return false, fmt.Errorf("move %s: the record has no stopped baselines to verify against", id) |
| 506 | } |
| 507 | dir, envPath := cfg.FilesDir(host) |
| 508 | plain, _, _, err := archivePlainDigest(dir, envPath) |
| 509 | why, retry, ferr := archiveFailure(host, err) |
| 510 | switch { |
| 511 | case ferr != nil: |
| 512 | return false, fmt.Errorf("digest the source's files: %w", ferr) |
| 513 | case retry: |
| 514 | r.w.notice(host+"#rv-verify", "moves: %s: a file changed while the source's files were digested; the verification runs again", host) |
| 515 | return false, nil |
| 516 | case why != "": |
| 517 | return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the source's files cannot be digested: %s", why)) |
| 518 | case plain != m.Source.Files.Plain: |
| 519 | return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the files are %s, recorded %s", plain, m.Source.Files.Plain)) |
| 520 | } |
| 521 | rows, err := cfg.Timers(host) |
| 522 | if err != nil { |
| 523 | return false, fmt.Errorf("read the timers: %w", err) |
| 524 | } |
| 525 | if got := moves.TimersDigest(rows); got != m.Source.TimersSHA256 { |
| 526 | return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the timers are %s, recorded %s", got, m.Source.TimersSHA256)) |
| 527 | } |
| 528 | log.Printf("moves: %s: the source still matches what it recorded when it stopped", host) |
| 529 | if err := r.update(ctx, obs, "the revert's probing", hostclaim.SetRevPhase(id, hostclaim.RevPhaseVerify, hostclaim.PhaseProbing, true)); err != nil { |
| 530 | return false, err |
| 531 | } |
| 532 | if err := r.fault("after-rev-probing"); err != nil { |
| 533 | return false, err |
| 534 | } |
| 535 | return true, nil |
| 536 | } |
| 537 | |
| 538 | // probing (§7.4 probing): the probe only, bounded twice (§7.7): its nine |
| 539 | // attempts and fifteen minutes from its first attempt. Nothing earlier runs |
| 540 | // again from here. The window's end is journalled before the stuck mark, so |
| 541 | // a crash or a refused record write between the two marks the move stuck on |
| 542 | // the next pass without probing again; a new window starts only once a retry |
| 543 | // has returned the record from stuck. |
| 544 | func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool, error) { |
| 545 | cfg := r.cfg() |
| 546 | host, id := r.rec.Host, r.g.ID |
| 547 | if obs.Claim.Op.Phase != hostclaim.PhaseProbing { |
| 548 | return false, fmt.Errorf("move %s: op.phase %q at rev_phase probing", id, obs.Claim.Op.Phase) |
| 549 | } |
| 550 | j, err := moves.ReadJournal(cfg.DataDir, host, id) |
| 551 | if err != nil { |
| 552 | return false, err |
| 553 | } |
| 554 | if raw, ok := j.Steps[rvStepProbeExhausted]; ok { |
| 555 | var ex probeExhaustion |
| 556 | if err := json.Unmarshal([]byte(raw), &ex); err != nil { |
| 557 | return false, fmt.Errorf("move %s: the journalled end of the probe's window does not read: %w", id, err) |
| 558 | } |
| 559 | if r.rec.Move.UpdatedAt.Equal(ex.UpdatedAt) { |
| 560 | // The window ended and the stuck mark never landed. |
| 561 | return r.stuck(ctx, ex.Reason, errors.New("the probe's window ended before this pass")) |
| 562 | } |
| 563 | // Marked stuck and retried since: a new window. |
| 564 | j.Unmark(rvStepProbeExhausted) |
| 565 | j.Unmark(rvStepProbeSince) |
| 566 | if err := j.Commit(cfg.DataDir); err != nil { |
| 567 | return false, err |
| 568 | } |
| 569 | log.Printf("moves: %s: the revert of %s was retried; its probe has a new window", host, id) |
| 570 | } |
| 571 | since, perr := time.Parse(time.RFC3339Nano, j.Steps[rvStepProbeSince]) |
| 572 | if perr != nil { |
| 573 | since = r.now().UTC() |
| 574 | j.Mark(rvStepProbeSince, since.Format(time.RFC3339Nano)) |
| 575 | if err := j.Commit(cfg.DataDir); err != nil { |
| 576 | return false, err |
| 577 | } |
| 578 | } |
| 579 | // A revert has no rollback: it probes the source's own generation, warm |
| 580 | // or cold as for serving (§7.14 reverting row), with no recovery |
| 581 | // coverage, and the recovery probe's nine attempts. The probe records |
| 582 | // what it renders before it renders it. |
| 583 | req := recoveryProbeRequest{Clean: true, SourceGeneration: obs.Claim.Holder.Generation, Attempts: recoveryProbeAttempts, |
| 584 | Record: func(ctx context.Context, in provision.MoveInputs) error { |
| 585 | _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRecordRevertInputs(r.g, in)) |
| 586 | return err |
| 587 | }} |
| 588 | out, err := cfg.recoveryProbe(ctx, r.s, r.rec, req) |
| 589 | var missing *ClaimDependencyError |
| 590 | if errors.As(err, &missing) { |
| 591 | return false, nil |
| 592 | } |
| 593 | if err != nil { |
| 594 | log.Printf("moves: %s: the revert's probe of %s: %v", host, id, err) |
| 595 | out = recoveryProbeWaiting |
| 596 | } |
| 597 | if out == recoveryProbeAnswered { |
| 598 | // The probe's wake wrote the claim (its admission, then its |
| 599 | // restore's re-tag), so the step's own read is stale: read it |
| 600 | // again, as the rollback's probed does. The fresh claim is taken |
| 601 | // only when it is the step's read plus those writes and nothing |
| 602 | // else (revertProbeWrites, census-ruling-home-part2's condition); |
| 603 | // any other change is refused as the stale CAS was. |
| 604 | fresh, err := hostclaim.Read(ctx, r.s, host) |
| 605 | if err != nil { |
| 606 | return false, fmt.Errorf("re-read the host claim: %w", err) |
| 607 | } |
| 608 | if err := revertProbeWrites(cfg.Box, cfg.DataDir, obs, fresh); err != nil { |
| 609 | return false, fmt.Errorf("the revert's probed: %w", err) |
| 610 | } |
| 611 | obs = fresh |
| 612 | if err := r.update(ctx, obs, "the revert's probed", hostclaim.SetRevPhase(id, hostclaim.PhaseProbing, hostclaim.PhaseProbed, r.rec.Move.Source != nil)); err != nil { |
| 613 | return false, err |
| 614 | } |
| 615 | log.Printf("moves: %s: the source answered its probe; the revert of %s is probed", host, id) |
| 616 | if err := r.fault("after-rev-probed"); err != nil { |
| 617 | return false, err |
| 618 | } |
| 619 | return true, nil |
| 620 | } |
| 621 | var reason string |
| 622 | var why error |
| 623 | elapsed := r.now().Sub(since) |
| 624 | switch { |
| 625 | case out == recoveryProbeExhausted: |
| 626 | reason, why = provision.StuckRecoveryProbeAttempts, fmt.Errorf("%d attempts", recoveryProbeAttempts) |
| 627 | case elapsed >= revertProbeBudget: |
| 628 | reason, why = provision.StuckRevertProbeBudget, fmt.Errorf("no answer for %s", elapsed.Round(time.Second)) |
| 629 | default: |
| 630 | r.w.notice(host+"#rv-probe", "moves: %s: the revert's probe of %s has not answered yet", host, id) |
| 631 | return false, nil |
| 632 | } |
| 633 | // The window's end first, against the record as it stands now (the |
| 634 | // probe may have recorded its inputs this pass), then the stuck mark. |
| 635 | fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, host) |
| 636 | if err != nil { |
| 637 | return false, fmt.Errorf("re-read the record: %w", err) |
| 638 | } |
| 639 | if !r.same(fresh) { |
| 640 | return false, nil |
| 641 | } |
| 642 | r.rec = *fresh |
| 643 | body, err := json.Marshal(probeExhaustion{Reason: reason, UpdatedAt: fresh.Move.UpdatedAt}) |
| 644 | if err != nil { |
| 645 | return false, err |
| 646 | } |
| 647 | // The probe wrote its attempt count into the same journal. |
| 648 | j, err = moves.ReadJournal(cfg.DataDir, host, id) |
| 649 | if err != nil { |
| 650 | return false, err |
| 651 | } |
| 652 | j.Mark(rvStepProbeExhausted, string(body)) |
| 653 | if err := j.Commit(cfg.DataDir); err != nil { |
| 654 | return false, err |
| 655 | } |
| 656 | if err := r.fault("after-probe-exhausted"); err != nil { |
| 657 | return false, err |
| 658 | } |
| 659 | return r.stuck(ctx, reason, why) |
| 660 | } |
| 661 | |
| 662 | // revertProbeWrites is census-ruling-home-part2's condition on the revert's |
| 663 | // re-read after its probe answers: fresh must be earlier, the step's own |
| 664 | // read, with only the probe wake's own claim writes over it. Those are the |
| 665 | // claim's own transitions, in order: hostclaim.AdmitWriter's cold admission |
| 666 | // of a new generation for box, this op's source writer (with the origin its |
| 667 | // CAS records), then, when the admitted stamp is new, the restore's re-tag |
| 668 | // of box's database paths, checked by FS-R11's claimRestoreObservation.retag. |
| 669 | // Each is replayed over earlier and the result must equal fresh exactly, as |
| 670 | // FS-R11's advance compares, so the op, its phase and journal, the holder's |
| 671 | // identity, every chore, fence, quiet key and every other box's inventory |
| 672 | // stand as the step read them, and the admission must be the first write |
| 673 | // after that read (its session is the claim gen it was made at). A claim no |
| 674 | // write touched passes. The probe's attempt reservation is not a claim |
| 675 | // write here: a reverting op's attempts are journalled on the box |
| 676 | // (movesProbeAttempts), and hostclaim.ReserveWakeRetryAttempt refuses a |
| 677 | // claim that is not serving. Anything else is hostclaim.ErrConflict. |
| 678 | func revertProbeWrites(box, dataDir string, earlier, fresh hostclaim.Observed) error { |
| 679 | if sameClaimObservation(earlier, fresh) { |
| 680 | return nil |
| 681 | } |
| 682 | conflict := func(why string) error { |
| 683 | return fmt.Errorf("%w: the host claim changed under the probe other than by its own writes: %s", hostclaim.ErrConflict, why) |
| 684 | } |
| 685 | if earlier.Absent || fresh.Absent || earlier.Host != fresh.Host || earlier.ETag == "" || earlier.Claim == nil || fresh.Claim == nil || |
| 686 | earlier.Claim.Holder == nil || fresh.Claim.Holder == nil { |
| 687 | return conflict("it is not the claim the step read") |
| 688 | } |
| 689 | e, f := earlier.Claim, fresh.Claim |
| 690 | n := len(f.Holder.Admissions) |
| 691 | if n == 0 { |
| 692 | return conflict("no admission") |
| 693 | } |
| 694 | last := f.Holder.Admissions[n-1] |
| 695 | if last.LossID != "" { |
| 696 | // A loss-bound admission (as a fence-bound one, which AdmitWriter |
| 697 | // refuses below) is verified only inside hostclaim, which does not |
| 698 | // expose that; until a hostclaim-owned function re-verifies it |
| 699 | // (amadan #300), the step stalls here as it did before the re-read. |
| 700 | return conflict("a loss-bound admission, which only hostclaim can verify (amadan #300)") |
| 701 | } |
| 702 | if last.Box != box || last.Mode != "cold" || last.Generation == "" || last.Generation == e.Holder.Generation { |
| 703 | return conflict("the last admission is not a new generation for this box's writer") |
| 704 | } |
| 705 | in := hostclaim.Admission{ |
| 706 | ObservedOpID: e.Op.ID, ObservedPhase: e.Op.Phase, ObservedSession: e.Holder.Session, ObservedGeneration: e.Holder.Generation, |
| 707 | Box: box, Mode: "cold", NewGeneration: last.Generation, PredecessorCertified: last.WritesLost == nil, KeepStamp: f.Holder.Stamp != last.Session, |
| 708 | Coverage: last.Coverage, Region: last.Region, PredecessorRegion: last.PredecessorRegion, AdditionalWritesLost: last.AdditionalWritesLost, |
| 709 | } |
| 710 | if last.WritesLost != nil { |
| 711 | in.RemoteMaxTXID = last.WritesLost.RemoteMaxTXID |
| 712 | } |
| 713 | admitted, err := revertClaimCopy(e) |
| 714 | if err != nil { |
| 715 | return err |
| 716 | } |
| 717 | if err := hostclaim.AdmitWriter(in)(admitted); err != nil { |
| 718 | if errors.Is(err, hostclaim.ErrFence) { |
| 719 | return conflict("a fence-bound admission, which only hostclaim can verify (amadan #300)") |
| 720 | } |
| 721 | return conflict(err.Error()) |
| 722 | } |
| 723 | admitted.Gen = e.Gen + 1 |
| 724 | // The admission's CAS moves the origin head by exactly one node, or |
| 725 | // leaves it when the chain's last node is already this origin. |
| 726 | if eh, fh := e.Origins, f.Origins; (eh == nil) != (fh == nil) || eh != nil && *eh != *fh { |
| 727 | var count uint64 |
| 728 | if eh != nil { |
| 729 | count = eh.Count |
| 730 | } |
| 731 | if fh == nil || fh.Count != count+1 { |
| 732 | return conflict("the origin head moved by more than the admission") |
| 733 | } |
| 734 | head := *fh |
| 735 | admitted.Origins = &head |
| 736 | } |
| 737 | // The copy drops the origin AdmitWriter staged for Update's CAS. |
| 738 | if admitted, err = revertClaimCopy(admitted); err != nil { |
| 739 | return err |
| 740 | } |
| 741 | switch f.Gen { |
| 742 | case admitted.Gen: |
| 743 | if !reflect.DeepEqual(admitted, f) { |
| 744 | return conflict("the admission") |
| 745 | } |
| 746 | case admitted.Gen + 1: |
| 747 | db := provision.DefaultDBPath(dataDir, earlier.Host) |
| 748 | if e.Inventory[box].Layout == "nested" { |
| 749 | db = provision.NestedDBPath(dataDir, earlier.Host) |
| 750 | } |
| 751 | mid := hostclaim.Observed{Host: earlier.Host, ETag: earlier.ETag + "+admitted", Claim: admitted} |
| 752 | ob := &claimRestoreObservation{observed: mid, box: box, dataDir: dataDir, db: db} |
| 753 | if err := ob.retag(mid, fresh); err != nil { |
| 754 | return conflict("the restore's re-tag") |
| 755 | } |
| 756 | default: |
| 757 | return conflict(fmt.Sprintf("gen %d to %d", e.Gen, f.Gen)) |
| 758 | } |
| 759 | return nil |
| 760 | } |
| 761 | |
| 762 | // revertClaimCopy is a deep copy of c through its JSON form, as the claim |
| 763 | // store decodes it. |
| 764 | func revertClaimCopy(c *hostclaim.Claim) (*hostclaim.Claim, error) { |
| 765 | body, err := json.Marshal(c) |
| 766 | if err != nil { |
| 767 | return nil, err |
| 768 | } |
| 769 | var out hostclaim.Claim |
| 770 | if err := json.Unmarshal(body, &out); err != nil { |
| 771 | return nil, err |
| 772 | } |
| 773 | return &out, nil |
| 774 | } |
| 775 | |
| 776 | // probed (§7.4 probed): reopened_at on the record (J26), the conditional open |
| 777 | // on hold_op = move id under the host's hold lock (J12), the claim opened. |
| 778 | func (r *revertRun) probed(ctx context.Context, obs hostclaim.Observed) (bool, error) { |
| 779 | cfg := r.cfg() |
| 780 | host, id := r.rec.Host, r.g.ID |
| 781 | if r.rec.Move.ReopenedAt == nil { |
| 782 | ok, err := r.record(ctx, "the reopen", provision.MoveRevertReopened(r.g, r.now())) |
| 783 | if !ok { |
| 784 | return false, err |
| 785 | } |
| 786 | if err := r.fault("after-reopened"); err != nil { |
| 787 | return false, err |
| 788 | } |
| 789 | } |
| 790 | var err error |
| 791 | provision.WithHostHoldLock(host, func() { |
| 792 | if err = r.stillReverting(ctx); err != nil { |
| 793 | return |
| 794 | } |
| 795 | var released bool |
| 796 | if released, err = cfg.ReleaseHeld(host, id); err != nil || released { |
| 797 | return |
| 798 | } |
| 799 | var rt *registry.Route |
| 800 | if cfg.Route == nil { |
| 801 | err = errors.New("this box cannot read its registry row") |
| 802 | return |
| 803 | } |
| 804 | if rt, err = cfg.Route(host); err != nil { |
| 805 | return |
| 806 | } |
| 807 | switch { |
| 808 | case rt == nil: |
| 809 | err = fmt.Errorf("move %s: this box has no row to open", id) |
| 810 | case rt.HoldOp != "": |
| 811 | err = fmt.Errorf("move %s: the row is held %s by %q, not by the move", id, rt.Status, rt.HoldOp) |
| 812 | } |
| 813 | }) |
| 814 | if err != nil { |
| 815 | return false, fmt.Errorf("open the row: %w", err) |
| 816 | } |
| 817 | if err := r.fault("after-row-opened"); err != nil { |
| 818 | return false, err |
| 819 | } |
| 820 | if err := r.update(ctx, obs, "the revert's opened", hostclaim.SetRevPhase(id, hostclaim.PhaseProbed, hostclaim.PhaseOpened, r.rec.Move.Source != nil)); err != nil { |
| 821 | return false, err |
| 822 | } |
| 823 | log.Printf("moves: %s: the source serves again; the revert of %s waits for its lane's park", host, id) |
| 824 | nudgeSerial(ctx, r.s) |
| 825 | if err := r.fault("after-rev-opened"); err != nil { |
| 826 | return false, err |
| 827 | } |
| 828 | return true, nil |
| 829 | } |
| 830 | |
| 831 | // opened (§7.4 opened, §7.11): once no chore of the op is left in the claim |
| 832 | // (the lane's park acknowledged), the record's close, then the claim's |
| 833 | // release to serving under this box (record first, ruling M4). |
| 834 | func (r *revertRun) opened(ctx context.Context, obs hostclaim.Observed) (bool, error) { |
| 835 | host, id := r.rec.Host, r.g.ID |
| 836 | if r.rec.Move.ReopenedAt == nil { |
| 837 | return false, fmt.Errorf("move %s: the claim is opened and the record has no reopen", id) |
| 838 | } |
| 839 | // Nothing owed under the op, checked before the record's close (M4), as |
| 840 | // the console's grant checks it. |
| 841 | if owed := owedUnder(obs.Claim, id); owed > 0 { |
| 842 | r.w.notice(host+"#rv-opened", "moves: %s: the revert of %s waits for %d chore(s) of the move, its lane's park, to be acknowledged", host, id, owed) |
| 843 | return false, nil |
| 844 | } |
| 845 | return r.close(ctx, true) |
| 846 | } |
| 847 | |
| 848 | // close is the record's close (MoveRevertClose), then, when releaseClaim is set, |
| 849 | // the claim's release to serving under this box, then the source's journal |
| 850 | // of the op parked. A crash between the record's close and the release |
| 851 | // leaves the claim reverting/opened with nothing owed; the console's |
| 852 | // recovery pass releases it from LastMove. releaseClaim is false only for a |
| 853 | // claim a claim-first binary already released. |
| 854 | func (r *revertRun) close(ctx context.Context, releaseClaim bool) (bool, error) { |
| 855 | cfg := r.cfg() |
| 856 | host, id := r.rec.Host, r.g.ID |
| 857 | ok, err := r.record(ctx, "the revert's close", provision.MoveRevertClose(r.g, r.now())) |
| 858 | if !ok { |
| 859 | return false, err |
| 860 | } |
| 861 | log.Printf("moves: %s: move %s is reverted and closed", host, id) |
| 862 | r.w.pageRolledBack(ctx, r.rec) |
| 863 | nudgeSerial(ctx, r.s) |
| 864 | if releaseClaim { |
| 865 | if err := r.fault("after-close-record"); err != nil { |
| 866 | return true, err |
| 867 | } |
| 868 | if err := releaseAfterClose(ctx, r.s, host, id, hostclaim.ReleaseMove(id, cfg.Box)); err != nil { |
| 869 | return true, err |
| 870 | } |
| 871 | log.Printf("moves: %s: the claim is serving again after the revert of %s", host, id) |
| 872 | if err := r.fault("after-release-move"); err != nil { |
| 873 | return true, err |
| 874 | } |
| 875 | } |
| 876 | if err := r.fault("after-close"); err != nil { |
| 877 | return true, err |
| 878 | } |
| 879 | if err := parkRevertJournal(cfg.DataDir, host, id, true); err != nil { |
| 880 | return true, fmt.Errorf("park the source's journal of %s: %w", id, err) |
| 881 | } |
| 882 | return true, nil |
| 883 | } |
| 884 | |
| 885 | // stuck stops the revert for an operator (MoveMarkStuck) with a recorded |
| 886 | // reason; the row stays held and move --retry resumes the rev_phase. |
| 887 | func (r *revertRun) stuck(ctx context.Context, reason string, why error) (bool, error) { |
| 888 | host := r.rec.Host |
| 889 | log.Printf("moves: %s: the revert of %s stops for an operator (%v): %s", host, r.g.ID, why, reason) |
| 890 | ok, err := r.record(ctx, "the stuck mark", provision.MoveMarkStuck(r.g, reason)) |
| 891 | if !ok { |
| 892 | return false, err |
| 893 | } |
| 894 | r.w.pageStuck(ctx, r.rec, "source", reason) |
| 895 | nudgeSerial(ctx, r.s) |
| 896 | return true, nil |
| 897 | } |
| 898 | |
| 899 | // markJournal marks the source's journal of the op as this revert's. A mark |
| 900 | // already there is kept. |
| 901 | func (r *revertRun) markJournal() error { |
| 902 | dd, host := r.cfg().DataDir, r.rec.Host |
| 903 | j, err := moves.ReadJournal(dd, host, r.g.ID) |
| 904 | if err != nil { |
| 905 | return err |
| 906 | } |
| 907 | if j.Done(rvStepSource) { |
| 908 | return nil |
| 909 | } |
| 910 | body, err := json.Marshal(revertJournalMark{Move: r.g.ID, Incarnation: r.g.Incarnation}) |
| 911 | if err != nil { |
| 912 | return err |
| 913 | } |
| 914 | j.Mark(rvStepSource, string(body)) |
| 915 | return j.Commit(dd) |
| 916 | } |
| 917 | |
| 918 | // parkRevertJournal parks the source's journal of a closed revert: closed |
| 919 | // (true when the caller has seen the close) is journalled first, then a |
| 920 | // moves-owned retirement with no artifacts is made for it (its receipt names |
| 921 | // the op) and the journal, with the cut's archive inside it, is renamed into |
| 922 | // it. A journal that is not there was parked already; one with no revert mark |
| 923 | // is not this revert's and is left. |
| 924 | func parkRevertJournal(dataDir, host, op string, closed bool) error { |
| 925 | if !moveJournalPresent(dataDir, host, op) { |
| 926 | return nil |
| 927 | } |
| 928 | j, err := moves.ReadJournal(dataDir, host, op) |
| 929 | if err != nil { |
| 930 | return err |
| 931 | } |
| 932 | raw, ok := j.Steps[rvStepSource] |
| 933 | if !ok { |
| 934 | return nil |
| 935 | } |
| 936 | var mark revertJournalMark |
| 937 | if err := json.Unmarshal([]byte(raw), &mark); err != nil || !journalComponent(mark.Incarnation) { |
| 938 | return fmt.Errorf("the journal of %s has an unreadable revert mark", op) |
| 939 | } |
| 940 | write := func(step, value string) error { |
| 941 | j.Mark(step, value) |
| 942 | return j.Commit(dataDir) |
| 943 | } |
| 944 | if closed && j.Steps[rvStepClosed] != "done" { |
| 945 | if err := write(rvStepClosed, "done"); err != nil { |
| 946 | return err |
| 947 | } |
| 948 | } |
| 949 | retirement, err := parkLocalArtifacts(context.Background(), j, write, localArtifactPark{DataDir: dataDir, Incarnation: mark.Incarnation, |
| 950 | Host: host, Op: op, Step: rvStepJournalPark, NewID: obligationID}) |
| 951 | if err != nil { |
| 952 | return err |
| 953 | } |
| 954 | if err := parkOpJournal(dataDir, mark.Incarnation, host, op, retirement); err != nil { |
| 955 | return err |
| 956 | } |
| 957 | log.Printf("moves: %s: the source's journal of the revert %s is parked", host, op) |
| 958 | return nil |
| 959 | } |
| 960 | |
| 961 | // sweepRevertJournal is the local journal sweep's revert half |
| 962 | // (sweepFinalJournal): a journal the revert marked (rvStepSource) is parked |
| 963 | // once the close is shown by durable evidence: |
| 964 | // - the journal records the close (rvStepClosed "done"); |
| 965 | // - no listed record carries the move: a reverting or stuck move is |
| 966 | // cleared only by provision.MoveRevertClose (MoveRetry returns a stuck |
| 967 | // one to reverting; no abort or completion takes either state), and the |
| 968 | // mark is written while the record is reverting, so the close landed. |
| 969 | // |
| 970 | // A host the listing does not name, or a record that still carries the move, |
| 971 | // is left. |
| 972 | func (w *movesWorker) sweepRevertJournal(dataDir, host, op string, j *moves.Journal, recs []provision.Record) error { |
| 973 | var mark revertJournalMark |
| 974 | if err := json.Unmarshal([]byte(j.Steps[rvStepSource]), &mark); err != nil || mark.Move == "" { |
| 975 | return fmt.Errorf("the journal of %s has an unreadable revert mark", op) |
| 976 | } |
| 977 | if j.Steps[rvStepClosed] != "done" { |
| 978 | if len(recs) == 0 { |
| 979 | w.notice(host+"#rv-sweep-"+op, "moves: %s: the source's journal of %s is a revert's, and no listed record names the host; it is left", host, op) |
| 980 | return nil |
| 981 | } |
| 982 | for _, rec := range recs { |
| 983 | if m := rec.Move; m != nil && m.ID == mark.Move { |
| 984 | return nil |
| 985 | } |
| 986 | } |
| 987 | } |
| 988 | return parkRevertJournal(dataDir, host, op, true) |
| 989 | } |
| 990 | |
| 991 | // laneParkTarget is the target region's half of a move outcome that leaves |
| 992 | // the move's own staging lane behind in its bucket (movesPass through |
| 993 | // participantRound): the lane's park-replica the claim owes this region, run |
| 994 | // by the park engine over this box's own bucket until it is acknowledged |
| 995 | // after its horizon. It serves a revert (the claim reverting, any phase; the |
| 996 | // park hostclaim.Revert queued), an abort (moving/aborting; the park |
| 997 | // hostclaim.Abort queued, Plan 5 Task 1) and a committed landing |
| 998 | // (landing/committed; the park hostclaim.Commit queued, Plan 5 Task 3, §7.9 |
| 999 | // step 4, called from the committed arm). No phases means any phase of |
| 1000 | // kind. The view and the authorizer are the claim's (MoveParks), the |
| 1001 | // authorizer behind this lane's own check: only this move's lane park under |
| 1002 | // kind and its phases, with this box not executing the host unless the |
| 1003 | // landing is committed (the target serves it then). |
| 1004 | func (w *movesWorker) laneParkTarget(ctx context.Context, s release.Store, rec provision.Record, kind string, phases ...string) error { |
| 1005 | cfg := w.cfg |
| 1006 | m := rec.Move |
| 1007 | if m == nil || m.Legacy() || m.To != cfg.Region || m.From == cfg.Region { |
| 1008 | return nil |
| 1009 | } |
| 1010 | notice := rec.Host + "#lane-" + kind |
| 1011 | if kind == hostclaim.KindReverting { |
| 1012 | notice = rec.Host + "#rv-lane" |
| 1013 | } |
| 1014 | if !revertLaneArmed(cfg) { |
| 1015 | w.notice(notice, "moves: %s: the park of move %s's lane waits: this box's park engine is not armed (the claim's park hooks are set from MoveParks only once this box's moves wake source is registered, Plan 5 Task 9b)", rec.Host, m.ID) |
| 1016 | return nil |
| 1017 | } |
| 1018 | host, id := rec.Host, m.ID |
| 1019 | inOp := func(op hostclaim.Op) bool { |
| 1020 | return op.ID == id && op.Kind == kind && (len(phases) == 0 || containsString(phases, op.Phase)) |
| 1021 | } |
| 1022 | obs, err := hostclaim.Read(ctx, s, host) |
| 1023 | if err != nil { |
| 1024 | return fmt.Errorf("read the host claim: %w", err) |
| 1025 | } |
| 1026 | if obs.Absent || !inOp(obs.Claim.Op) { |
| 1027 | return nil // the claim half has not followed yet, or it is over |
| 1028 | } |
| 1029 | if h := obs.Claim.Holder; h == nil || h.Incarnation != m.Incarnation { |
| 1030 | return fmt.Errorf("move %s: the host claim names another incarnation", id) |
| 1031 | } |
| 1032 | lane := moves.ScopeLocation(host, id) |
| 1033 | var chores []hostclaim.Obligation |
| 1034 | for _, o := range obs.Claim.Obligations { |
| 1035 | if o.Op == id && o.Kind == "park-replica" && park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) && o.Park != nil && o.Park.Location == lane { |
| 1036 | chores = append(chores, o) |
| 1037 | } |
| 1038 | } |
| 1039 | if len(chores) == 0 { |
| 1040 | return nil |
| 1041 | } |
| 1042 | vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion) |
| 1043 | if err != nil { |
| 1044 | return err |
| 1045 | } |
| 1046 | committed := kind == hostclaim.KindLanding |
| 1047 | // The claim's view (movesParkView) names the target's region |
| 1048 | // authoritative after the commit, with its marker, and the source's |
| 1049 | // before it; the claim's authorizer runs behind this lane's own check. |
| 1050 | own := func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error { |
| 1051 | c := obs.Claim |
| 1052 | if obs.Absent || c == nil || !inOp(c.Op) || (!committed && c.Op.Exec == cfg.Box) || |
| 1053 | o.Op != id || o.Kind != "park-replica" || !park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) || o.Park == nil || o.Park.Location != lane { |
| 1054 | return fmt.Errorf("%w: not this move's lane park", park.ErrUnsafe) |
| 1055 | } |
| 1056 | return nil |
| 1057 | } |
| 1058 | e := park.Engine{HomeRegion: cfg.ParkHomeRegion, Store: vs, Claims: s, Bucket: cfg.OwnBucket, Region: cfg.parkRegion(), Leases: cfg.ParkLeases, |
| 1059 | Now: w.now, NewID: obligationID, View: cfg.ParkView, Authorize: movesArmAuthorize(cfg.ParkAuthorize, own), After: cfg.parkFault} |
| 1060 | for _, o := range chores { |
| 1061 | err := e.Run(ctx, park.Work{Host: host, ID: o.ID, Op: id}) |
| 1062 | switch { |
| 1063 | case errors.Is(err, park.ErrPending): |
| 1064 | w.notice(notice, "moves: %s: move %s's lane is archived; its park is acknowledged after its horizon", host, id) |
| 1065 | case errors.Is(err, park.ErrChanged): |
| 1066 | case err != nil: |
| 1067 | return fmt.Errorf("park move %s's lane: %w", id, err) |
| 1068 | default: |
| 1069 | log.Printf("moves: %s: move %s's lane in %s is parked", host, id, cfg.Region) |
| 1070 | } |
| 1071 | } |
| 1072 | return nil |
| 1073 | } |
| 1074 | |
amadan