| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "strconv" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/hostbox" |
| 13 | "github.com/carlosframework/platform/internal/hostclaim" |
| 14 | "github.com/carlosframework/platform/internal/hostguard" |
| 15 | "github.com/carlosframework/platform/internal/moves" |
| 16 | "github.com/carlosframework/platform/internal/provision" |
| 17 | "github.com/carlosframework/platform/internal/registry" |
| 18 | "github.com/carlosframework/platform/internal/release" |
| 19 | ) |
| 20 | |
| 21 | // The source's recovery after a rollback (moves spec §7.8 source table; Plan |
| 22 | // 4 Task 5), behind CARLOS_MOVES_PHASE1: the target has released, and the |
| 23 | // source box takes the host back, clean or dirty, through opened, and then |
| 24 | // the rolled_back finalization. |
| 25 | // |
| 26 | // Routing. MoveRollBack moved the record's region back to From, so the |
| 27 | // source reaches a rolling_back record through the post-flip participant |
| 28 | // rule (movesParticipantRecord, J10) in movesPass, whether or not it still |
| 29 | // has its row. One pass drives it as far as it can, each phase from durable |
| 30 | // state, under the record's source_phase. A record at X means X's work may |
| 31 | // have begun (Plan 2's convention, G8; provision's "each is written before |
| 32 | // its side effects"): |
| 33 | // |
| 34 | // - waiting: until the target's evidence, released clean or returned |
| 35 | // dirty on the record, and the claim restoring/released with nobody |
| 36 | // executing ("" after the target's evidence, "none" after a |
| 37 | // declaration). Then the record says claimed. |
| 38 | // - claimed: the claim's return (hostclaim.ReturnToSource: this box |
| 39 | // executing at the source's own stamp S; a clean return also names the |
| 40 | // source's generation, src_gen, again), then the guard's activation |
| 41 | // (the fresh read finds holder.stamp == S, matching the installation |
| 42 | // record, or re-initialises a lost one to S), then, for a dirty return |
| 43 | // whose inventory lists paths of S, the source's park-local (J13: the |
| 44 | // installer's local retires S against it). The claim's phase is |
| 45 | // claimed from here through the install (J1). |
| 46 | // - restoring: the row handed from the move's moving hold to the |
| 47 | // rollback's restoring hold (handHold, J3), with the record read fresh |
| 48 | // under the host's hold lock. |
| 49 | // - clean verified: the files plain digest and the timers digest equal |
| 50 | // the source's stopped baselines (Source.Files.Plain, |
| 51 | // Source.TimersSHA256), and the warm database's <dbPath>.gen names |
| 52 | // src_gen (plain text, H11). A mismatch stops for an operator with the |
| 53 | // baseline-mismatch sentence at verified (MoveMarkStuck, INV-4); a |
| 54 | // retry verifies again, never skips it. |
| 55 | // - dirty parking to configured: the shared installer (installGeneration, |
| 56 | // Task 3) bound by recoveryInstallSource to the return seal and the |
| 57 | // return baseline the target journalled in the claim (Task 4), from the |
| 58 | // return lane moves/<host>/<rb id>/ in this box's own bucket, under S; |
| 59 | // configured records the inputs it rendered as recovery_inputs, never |
| 60 | // compared with move.inputs. restored hands over to probing. |
| 61 | // - probing: the claim steps claimed to probing first (J1), then the |
| 62 | // recovery probe: recoveryProbe, Task 9's (BLOCKED-ON-CLAIM C1, rulings |
| 63 | // J2 and J18), a cold admission of kind recovery with nine attempts, |
| 64 | // into the return generation (dirty, coverage sealed) or a fresh one |
| 65 | // restored from src_gen (clean, coverage clean-return; J14: both |
| 66 | // returns are cold and never keep S). The probe records the inputs it |
| 67 | // renders before it renders them (MoveRecordRecoveryInputs, at |
| 68 | // probing: recoveryProbeAdapter, movesrecoveryprobe.go, claim Plan 10 |
| 69 | // J2; a clean return records only here). Once the |
| 70 | // record says probing nothing earlier runs again: a restarted probe |
| 71 | // restarts only the probe (INV-2, INV-7). Nine failed attempts stop |
| 72 | // for an operator at probing. |
| 73 | // - probed: the claim steps to probed and, for a dirty return, queues the |
| 74 | // return lane's park in this region in the same write |
| 75 | // (QueueReturnLanePark, its horizon the return seal's bound); then |
| 76 | // recovered_at, and the serial is nudged (the target's cleaning waits |
| 77 | // on it). |
| 78 | // - recovered: the open: the rollback's hold on the row ends |
| 79 | // (ReleaseHeld, a plain registry call under the host's hold lock with |
| 80 | // the record read fresh, J12), the claim steps to opened, and the record |
| 81 | // says opened with opened_at. |
| 82 | // - opened: the finalization. Once recovered_at, opened_at and |
| 83 | // target_parked_at are stamped and no obligation of the op remains in |
| 84 | // the claim, the terminal event (LogEvent, Plan 6's moves.LogEvent), |
| 85 | // the record's close (MoveRolledBackClose) and then the claim's |
| 86 | // release to serving (ReleaseMove): record first, ruling M4. Whichever |
| 87 | // actor stamps last tries; a crash before the record's close is the |
| 88 | // source's next pass, and one after it the console's recovery pass |
| 89 | // (Plan 5 Task 7), which releases from LastMove. |
| 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; Plan 5 Task 9a). Like the landing and the |
| 95 | // target's half of the rollback, the source's recovery needs the claim's |
| 96 | // park hooks, RunHeld (for the dirty return's residue parks) and the local |
| 97 | // authorizer, which armMovesParks sets only once the wake-source gate opens |
| 98 | // (movesWakeSourceRegistered, Plan 5 Task 9b), so |
| 99 | // recoverySourceArmed answers false in production and the pass logs a |
| 100 | // standing notice. The probe hook is |
| 101 | // the recovery probe adapter (armRecoveryProbe) with CARLOS_HOST_CLAIM on, |
| 102 | // bound to the activator's Group D entry. Without the claim switch the |
| 103 | // legacy nil hook still waits, recording nothing. LogEvent |
| 104 | // is nil in production, where the event is a log line. |
| 105 | |
| 106 | // recoverStepsPerPass bounds the phases one pass runs: the dirty path from |
| 107 | // waiting to the finalization is fifteen. |
| 108 | const recoverStepsPerPass = 16 |
| 109 | |
| 110 | // recoveryProbeAttempts is the recovery probe's budget (§7.8: three times |
| 111 | // the landing's three). |
| 112 | const recoveryProbeAttempts = 9 |
| 113 | |
| 114 | // The recovery probe's admission coverage (§7.14): a dirty return admits |
| 115 | // the return generation from its sealed lane, a clean one restores the |
| 116 | // source's own generation. |
| 117 | const ( |
| 118 | recoveryCoverageSealed = "sealed" |
| 119 | recoveryCoverageClean = "clean-return" |
| 120 | ) |
| 121 | |
| 122 | // recoveryProbeOutcome is what one call of the recovery probe found. |
| 123 | type recoveryProbeOutcome int |
| 124 | |
| 125 | const ( |
| 126 | // recoveryProbeWaiting: no answer yet; the next pass calls the probe |
| 127 | // again (and only the probe). |
| 128 | recoveryProbeWaiting recoveryProbeOutcome = iota |
| 129 | // recoveryProbeAnswered: the app answered; the recovery goes to probed. |
| 130 | recoveryProbeAnswered |
| 131 | // recoveryProbeExhausted: every attempt failed; the recovery stops for |
| 132 | // an operator (StuckRecoveryProbeAttempts). |
| 133 | recoveryProbeExhausted |
| 134 | ) |
| 135 | |
| 136 | // recoveryProbeRequest is what the source asks of the recovery probe (Task |
| 137 | // 9 builds it against claim Group D for recovery, J2). |
| 138 | type recoveryProbeRequest struct { |
| 139 | // RollbackID names the rollback, whose id names a dirty return's |
| 140 | // generation and journal. |
| 141 | RollbackID string |
| 142 | // Clean is the return's kind; Coverage is the admission's coverage, |
| 143 | // "sealed" (dirty), "clean-return", or "" for a declared return. |
| 144 | Clean bool |
| 145 | Coverage string |
| 146 | // Declared is a clean return after a declared-lost or fenced target |
| 147 | // (Q6): the record's rollback cites the loss the claim journals. It asks |
| 148 | // no coverage, for every session, and carries no loss id: the claim's |
| 149 | // writer derives it. |
| 150 | Declared bool |
| 151 | // Generation is the generation the dirty return installed (the |
| 152 | // rollback id); empty for a clean return, whose probe admits a fresh |
| 153 | // one restored from SourceGeneration (src_gen, J14). |
| 154 | Generation, SourceGeneration string |
| 155 | // Attempts is the budget, nine. |
| 156 | Attempts int |
| 157 | // Record records the inputs the probe is about to render, before it |
| 158 | // renders them (MoveRecordRecoveryInputs; Task 3b concern 3). |
| 159 | Record func(context.Context, provision.MoveInputs) error |
| 160 | } |
| 161 | |
| 162 | // recoverySourceArmed reports whether this box can run the source's |
| 163 | // recovery: every hook a dirty return's install and the open call is wired, |
| 164 | // the claim's park hooks, RunHeld and local authorizer included |
| 165 | // (landingArmed). In production armMovesParks sets those only once the |
| 166 | // wake-source gate opens (movesWakeSourceRegistered, Plan 5 Task 9b), so this |
| 167 | // answers false there until then (J18). |
| 168 | func recoverySourceArmed(cfg moveCfg) bool { |
| 169 | return landingArmed(cfg) && cfg.DataDir != "" && cfg.HandHold != nil && cfg.ReleaseHeld != nil && cfg.Timers != nil |
| 170 | } |
| 171 | |
| 172 | // releaseHeld is the recovery's open in production: registry.ReleaseHold |
| 173 | // under the op that holds the row, then written (holdStageRequester) when |
| 174 | // it landed, so root re-judges what it staged now the hold has ended. The |
| 175 | // caller holds provision.WithHostHoldLock. |
| 176 | func releaseHeld(reg *registry.Registry, written func(host string)) func(host, op string) (bool, error) { |
| 177 | return func(host, op string) (bool, error) { |
| 178 | released, err := reg.ReleaseHold(host, op) |
| 179 | if err == nil && released && written != nil { |
| 180 | written(host) |
| 181 | } |
| 182 | return released, err |
| 183 | } |
| 184 | } |
| 185 | |
| 186 | // recoverRun is one pass's state for one rolling_back record on its source. |
| 187 | type recoverRun struct { |
| 188 | w *movesWorker |
| 189 | s release.Store |
| 190 | rec provision.Record |
| 191 | g provision.MoveGuard |
| 192 | rb string |
| 193 | } |
| 194 | |
| 195 | // recoverSource is the participant step for a rolling_back record on its |
| 196 | // source (movesPass). |
| 197 | func (w *movesWorker) recoverSource(ctx context.Context, s release.Store, rec provision.Record) error { |
| 198 | cfg := w.cfg |
| 199 | m := rec.Move |
| 200 | if m == nil || m.Legacy() || m.State != provision.MoveRollingBack || m.Rollback == nil || m.From != cfg.Region || m.To == cfg.Region { |
| 201 | return nil |
| 202 | } |
| 203 | if !recoverySourceArmed(cfg) { |
| 204 | w.notice(rec.Host+"#recover", "moves: %s: the source's recovery of rollback %s waits: this box's recovery 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.Rollback.ID) |
| 205 | return nil |
| 206 | } |
| 207 | r := &recoverRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, rb: m.Rollback.ID} |
| 208 | defer r.settleLease() |
| 209 | for i := 0; i < recoverStepsPerPass; i++ { |
| 210 | advanced, err := r.step(ctx) |
| 211 | if err != nil || !advanced { |
| 212 | return err |
| 213 | } |
| 214 | fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| 215 | if err != nil { |
| 216 | return fmt.Errorf("re-read the record: %w", err) |
| 217 | } |
| 218 | if !r.same(fresh) { |
| 219 | return nil |
| 220 | } |
| 221 | r.rec = *fresh |
| 222 | } |
| 223 | return nil |
| 224 | } |
| 225 | |
| 226 | func (r *recoverRun) cfg() moveCfg { return r.w.cfg } |
| 227 | func (r *recoverRun) now() time.Time { return r.w.now() } |
| 228 | func (r *recoverRun) rollback() *provision.MoveRollback { return r.rec.Move.Rollback } |
| 229 | func (r *recoverRun) dbPath() string { return provision.DefaultDBPath(r.cfg().DataDir, r.rec.Host) } |
| 230 | |
| 231 | // same reports whether fresh is still this rollback, rolling back. |
| 232 | func (r *recoverRun) same(fresh *provision.Record) bool { |
| 233 | m := fresh.Move |
| 234 | return m != nil && m.ID == r.g.ID && m.Incarnation == r.g.Incarnation && m.State == provision.MoveRollingBack && m.Rollback != nil && m.Rollback.ID == r.rb |
| 235 | } |
| 236 | |
| 237 | // fault is the test seam after a side effect. |
| 238 | func (r *recoverRun) fault(point string) error { |
| 239 | if f := r.cfg().rcFault; f != nil { |
| 240 | return f(point) |
| 241 | } |
| 242 | return nil |
| 243 | } |
| 244 | |
| 245 | // settleLease releases the own-location lease a dirty return's build holds |
| 246 | // from building through published, once the record is past them or no |
| 247 | // longer this return: the lease is the build's alone. |
| 248 | func (r *recoverRun) settleLease() { |
| 249 | m := r.rec.Move |
| 250 | if m != nil && m.State == provision.MoveRollingBack && m.Rollback != nil && m.Rollback.Dirty && |
| 251 | (m.Rollback.SourcePhase == provision.RbSourceBuilding || m.Rollback.SourcePhase == provision.RbSourcePublished) { |
| 252 | return |
| 253 | } |
| 254 | releaseOwnLease(r.w, r.rec.Host) |
| 255 | } |
| 256 | |
| 257 | // step runs the source's current recovery phase once. advanced is true when |
| 258 | // it wrote the record. |
| 259 | func (r *recoverRun) step(ctx context.Context) (bool, error) { |
| 260 | rb := r.rollback() |
| 261 | switch rb.SourcePhase { |
| 262 | case provision.RbSourceWaiting: |
| 263 | return r.waiting(ctx) |
| 264 | case provision.RbSourceClaimed: |
| 265 | return r.claimed(ctx) |
| 266 | case provision.RbSourceRestoring: |
| 267 | return r.restoring(ctx) |
| 268 | case provision.RbSourceVerified: |
| 269 | return r.verified(ctx) |
| 270 | case provision.RbSourceParking, provision.RbSourceBuilding, provision.RbSourcePublished, provision.RbSourceLocal, provision.RbSourceConfigured: |
| 271 | return r.install(ctx) |
| 272 | case provision.RbSourceRestored: |
| 273 | return r.phase(ctx, provision.RbSourceRestored, provision.RbSourceProbing, nil, "") |
| 274 | case provision.RbSourceProbing: |
| 275 | return r.probing(ctx) |
| 276 | case provision.RbSourceProbed: |
| 277 | return r.probed(ctx) |
| 278 | case provision.RbSourceRecovered: |
| 279 | return r.open(ctx) |
| 280 | case provision.RbSourceOpened: |
| 281 | return r.finalize(ctx) |
| 282 | } |
| 283 | return false, nil |
| 284 | } |
| 285 | |
| 286 | // phase writes one source phase step, logs it (movesevents.go names it from |
| 287 | // the record; the last argument, the step's old event name, is unused) and |
| 288 | // passes the step's fault point. A conflict is someone else's decision. |
| 289 | func (r *recoverRun) phase(ctx context.Context, from, to string, stamp func(*provision.MoveRollback), _ string) (bool, error) { |
| 290 | host := r.rec.Host |
| 291 | var pre provision.Record |
| 292 | after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveRollbackSourcePhase(r.g, from, to, stamp))) |
| 293 | if err != nil { |
| 294 | if errors.Is(err, provision.ErrMoveConflict) { |
| 295 | log.Printf("moves: %s: the source's recovery %s to %s refused (%v); the record moved on", host, from, to, err) |
| 296 | return false, nil |
| 297 | } |
| 298 | return false, fmt.Errorf("the source's recovery %s to %s: %w", from, to, err) |
| 299 | } |
| 300 | log.Printf("moves: %s: the source of rollback %s is %s", host, r.rb, to) |
| 301 | r.w.logTransition(ctx, r.s, pre, *after) |
| 302 | if err := r.fault("after-record-" + to); err != nil { |
| 303 | return true, err |
| 304 | } |
| 305 | return true, nil |
| 306 | } |
| 307 | |
| 308 | // stuck stops the recovery for an operator (MoveMarkStuck) with a recorded |
| 309 | // reason; the row stays held and move --retry returns to the phase. |
| 310 | func (r *recoverRun) stuck(ctx context.Context, reason string, why error) (bool, error) { |
| 311 | host := r.rec.Host |
| 312 | log.Printf("moves: %s: the recovery of %s stops for an operator (%v): %s", host, r.rb, why, reason) |
| 313 | var pre provision.Record |
| 314 | after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveMarkStuck(r.g, reason))) |
| 315 | if err != nil { |
| 316 | if errors.Is(err, provision.ErrMoveConflict) { |
| 317 | log.Printf("moves: %s: the stuck mark of %s refused (%v); the record moved on", host, r.rb, err) |
| 318 | return false, nil |
| 319 | } |
| 320 | return false, fmt.Errorf("mark the recovery stuck: %w", err) |
| 321 | } |
| 322 | r.w.logTransition(ctx, r.s, pre, *after) |
| 323 | r.w.pageStuck(ctx, r.rec, "source", reason) |
| 324 | nudgeSerial(ctx, r.s) |
| 325 | return true, nil |
| 326 | } |
| 327 | |
| 328 | // claim reads the host claim and requires this rollback's restoring op, |
| 329 | // with this box the move's source, for the record's incarnation. |
| 330 | func (r *recoverRun) claim(ctx context.Context) (hostclaim.Observed, error) { |
| 331 | obs, err := hostclaim.Read(ctx, r.s, r.rec.Host) |
| 332 | if err != nil { |
| 333 | return obs, fmt.Errorf("read the host claim: %w", err) |
| 334 | } |
| 335 | if obs.Absent { |
| 336 | return obs, fmt.Errorf("move %s has no host claim", r.g.ID) |
| 337 | } |
| 338 | if err := r.ours(obs.Claim); err != nil { |
| 339 | return obs, fmt.Errorf("move %s: %w", r.g.ID, err) |
| 340 | } |
| 341 | return obs, nil |
| 342 | } |
| 343 | |
| 344 | // ours is the claim check of every source phase before the finalization: |
| 345 | // this rollback's restoring op, this box the move's source, the record's |
| 346 | // incarnation. |
| 347 | func (r *recoverRun) ours(c *hostclaim.Claim) error { |
| 348 | box := r.cfg().Box |
| 349 | if c == nil || c.Op.ID != r.g.ID || c.Op.Kind != hostclaim.KindRestoring || c.Op.Intent["rollback"] != r.rb { |
| 350 | return fmt.Errorf("%w: the claim is not rollback %s", hostclaim.ErrWrongState, r.rb) |
| 351 | } |
| 352 | if c.Holder == nil || c.Holder.Incarnation != r.g.Incarnation { |
| 353 | return fmt.Errorf("%w: the claim names another incarnation", hostclaim.ErrWrongState) |
| 354 | } |
| 355 | if len(c.Op.Participants) == 0 || c.Op.Participants[0] != box || c.Op.Intent["source"] != box { |
| 356 | return fmt.Errorf("%w: this box is not the move's source in the claim", hostclaim.ErrWrongState) |
| 357 | } |
| 358 | return nil |
| 359 | } |
| 360 | |
| 361 | // executing requires the claim at phase with this box executing. |
| 362 | func (r *recoverRun) executing(c *hostclaim.Claim, phase string) error { |
| 363 | if c.Op.Phase != phase || c.Op.Exec != r.cfg().Box { |
| 364 | return fmt.Errorf("move %s: the claim is restoring/%q with executor %q, not %s on this box", r.g.ID, c.Op.Phase, c.Op.Exec, phase) |
| 365 | } |
| 366 | return nil |
| 367 | } |
| 368 | |
| 369 | // sourceStamp is S, the source installation's stamp the flip journalled. |
| 370 | func sourceStamp(c *hostclaim.Claim) (uint64, error) { |
| 371 | s, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64) |
| 372 | if err != nil || s == 0 { |
| 373 | return 0, fmt.Errorf("%w: the flip's source stamp is not journalled", hostclaim.ErrWrongState) |
| 374 | } |
| 375 | return s, nil |
| 376 | } |
| 377 | |
| 378 | // stillAt is the record check run under the host's hold lock before a row |
| 379 | // write: the record, read fresh, must still be this rollback with the |
| 380 | // source at phase. It reads the record only. |
| 381 | func (r *recoverRun) stillAt(ctx context.Context, phase string) error { |
| 382 | fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host) |
| 383 | if err != nil { |
| 384 | return fmt.Errorf("re-read the record: %w", err) |
| 385 | } |
| 386 | if !r.same(fresh) || fresh.Move.Rollback.SourcePhase != phase { |
| 387 | return fmt.Errorf("move %s: the record is no longer rollback %s at %s", r.g.ID, r.rb, phase) |
| 388 | } |
| 389 | return nil |
| 390 | } |
| 391 | |
| 392 | // waiting (§7.8 source waiting): the target's evidence, released clean or |
| 393 | // returned dirty, and the claim released with nobody executing. |
| 394 | func (r *recoverRun) waiting(ctx context.Context) (bool, error) { |
| 395 | host, rb := r.rec.Host, r.rollback() |
| 396 | if !(rb.TargetPhase == provision.RbTargetReleased && !rb.Dirty) && !(rb.TargetPhase == provision.RbTargetReturned && rb.Dirty) { |
| 397 | r.w.notice(host+"#rc-wait", "moves: %s: the source waits for the target of rollback %s (%s, dirty %t)", host, r.rb, rb.TargetPhase, rb.Dirty) |
| 398 | return false, nil |
| 399 | } |
| 400 | obs, err := r.claim(ctx) |
| 401 | if err != nil { |
| 402 | return false, err |
| 403 | } |
| 404 | if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleased || (op.Exec != "" && op.Exec != "none") { |
| 405 | r.w.notice(host+"#rc-wait", "moves: %s: the source waits for the claim's release of rollback %s (restoring/%q, executor %q)", host, r.rb, op.Phase, op.Exec) |
| 406 | return false, nil |
| 407 | } |
| 408 | return r.phase(ctx, provision.RbSourceWaiting, provision.RbSourceClaimed, nil, "") |
| 409 | } |
| 410 | |
| 411 | // claimed (§7.8 source claimed): the claim's return to this box at S (and |
| 412 | // src_gen for a clean return), the guard's activation at S, and a dirty |
| 413 | // return's park-local for its registered pre-move installation (J13). |
| 414 | func (r *recoverRun) claimed(ctx context.Context) (bool, error) { |
| 415 | cfg := r.cfg() |
| 416 | host, id, rb := r.rec.Host, r.g.ID, r.rollback() |
| 417 | clean := !rb.Dirty |
| 418 | if err := r.markJournals(); err != nil { |
| 419 | return false, err |
| 420 | } |
| 421 | obs, err := r.claim(ctx) |
| 422 | if err != nil { |
| 423 | return false, err |
| 424 | } |
| 425 | if op := obs.Claim.Op; op.Phase != hostclaim.PhaseClaimed || op.Exec != cfg.Box { |
| 426 | if obs, err = hostclaim.Update(ctx, r.s, obs, hostclaim.ReturnToSource(id, cfg.Box, clean, hostclaim.ReturnFence{})); err != nil { |
| 427 | return false, fmt.Errorf("the claim's return to the source: %w", err) |
| 428 | } |
| 429 | log.Printf("moves: %s: the source executes the host again for rollback %s (clean %t)", host, r.rb, clean) |
| 430 | } |
| 431 | if err := r.fault("after-return-claimed"); err != nil { |
| 432 | return false, err |
| 433 | } |
| 434 | c := obs.Claim |
| 435 | s, err := sourceStamp(c) |
| 436 | if err != nil { |
| 437 | return false, err |
| 438 | } |
| 439 | if c.Holder.Stamp != s { |
| 440 | return false, fmt.Errorf("move %s: the holder is at stamp %d, not the source's %d", id, c.Holder.Stamp, s) |
| 441 | } |
| 442 | if clean && c.Holder.Generation != c.Op.Journal[hostclaim.JournalSrcGen] { |
| 443 | return false, fmt.Errorf("move %s: a clean return's holder names generation %q, not src_gen", id, c.Holder.Generation) |
| 444 | } |
| 445 | g := cfg.Guard |
| 446 | if g.Claims == nil { |
| 447 | g.Claims = hostguard.StoreReader{Store: r.s} |
| 448 | } |
| 449 | if err := g.Activate(ctx, host, cfg.Box, nil); err != nil { |
| 450 | return false, fmt.Errorf("activate the source's installation at stamp %d: %w", s, err) |
| 451 | } |
| 452 | if err := r.fault("after-activate"); err != nil { |
| 453 | return false, err |
| 454 | } |
| 455 | if rb.Dirty && hostclaim.InventoryHasStamp(c, cfg.Box, s) { |
| 456 | if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.QueueLocalPark(id, cfg.Box, obligationID)); err != nil { |
| 457 | return false, fmt.Errorf("queue the source's local park: %w", err) |
| 458 | } |
| 459 | } |
| 460 | return r.phase(ctx, provision.RbSourceClaimed, provision.RbSourceRestoring, nil, "") |
| 461 | } |
| 462 | |
| 463 | // restoring (§7.8 source restoring): the row handed from the move's hold to |
| 464 | // the rollback's, as restoring (J3), with the record checked under the |
| 465 | // host's hold lock; then the branch, clean or dirty. |
| 466 | func (r *recoverRun) restoring(ctx context.Context) (bool, error) { |
| 467 | cfg := r.cfg() |
| 468 | host, id := r.rec.Host, r.g.ID |
| 469 | handed, err := cfg.HandHold(host, "restoring", r.rb, id, func() error { return r.stillAt(ctx, provision.RbSourceRestoring) }) |
| 470 | if err != nil { |
| 471 | return false, fmt.Errorf("hand the row's hold to the rollback: %w", err) |
| 472 | } |
| 473 | if !handed { |
| 474 | return false, fmt.Errorf("move %s: this box's row is held by neither the move nor rollback %s", id, r.rb) |
| 475 | } |
| 476 | if err := r.fault("after-hand"); err != nil { |
| 477 | return false, err |
| 478 | } |
| 479 | next := provision.RbSourceVerified |
| 480 | if r.rollback().Dirty { |
| 481 | next = provision.RbSourceParking |
| 482 | } |
| 483 | return r.phase(ctx, provision.RbSourceRestoring, next, nil, "") |
| 484 | } |
| 485 | |
| 486 | // verified (§7.8 clean verified, INV-4): the source's files and timers hash |
| 487 | // to what it recorded when it stopped, and its warm database's marker names |
| 488 | // src_gen. A mismatch stops for an operator; a retry runs this again. |
| 489 | func (r *recoverRun) verified(ctx context.Context) (bool, error) { |
| 490 | cfg := r.cfg() |
| 491 | host, id, m := r.rec.Host, r.g.ID, r.rec.Move |
| 492 | if err := r.fault("verify"); err != nil { |
| 493 | return false, err |
| 494 | } |
| 495 | obs, err := r.claim(ctx) |
| 496 | if err != nil { |
| 497 | return false, err |
| 498 | } |
| 499 | if err := r.executing(obs.Claim, hostclaim.PhaseClaimed); err != nil { |
| 500 | return false, err |
| 501 | } |
| 502 | srcGen, ok := obs.Claim.Op.Journal[hostclaim.JournalSrcGen] |
| 503 | if !ok { |
| 504 | return false, fmt.Errorf("move %s: the flip's source generation is not journalled", id) |
| 505 | } |
| 506 | if m.Source == nil || m.Source.Files == nil || m.Source.Files.Plain == "" || m.Source.TimersSHA256 == "" { |
| 507 | return false, fmt.Errorf("move %s: the record has no stopped baselines to verify against", id) |
| 508 | } |
| 509 | dir, envPath := cfg.FilesDir(host) |
| 510 | plain, _, _, err := archivePlainDigest(dir, envPath) |
| 511 | why, retry, ferr := archiveFailure(host, err) |
| 512 | switch { |
| 513 | case ferr != nil: |
| 514 | return false, fmt.Errorf("digest the source's files: %w", ferr) |
| 515 | case retry: |
| 516 | r.w.notice(host+"#rc-verify", "moves: %s: a file changed while the source's files were digested; the verification runs again", host) |
| 517 | return false, nil |
| 518 | case why != "": |
| 519 | return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the source's files cannot be digested: %s", why)) |
| 520 | case plain != m.Source.Files.Plain: |
| 521 | return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the files are %s, recorded %s", plain, m.Source.Files.Plain)) |
| 522 | } |
| 523 | rows, err := cfg.Timers(host) |
| 524 | if err != nil { |
| 525 | return false, fmt.Errorf("read the timers: %w", err) |
| 526 | } |
| 527 | if got := moves.TimersDigest(rows); got != m.Source.TimersSHA256 { |
| 528 | return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the timers are %s, recorded %s", got, m.Source.TimersSHA256)) |
| 529 | } |
| 530 | if why, err := r.warmMismatch(srcGen); err != nil { |
| 531 | return false, err |
| 532 | } else if why != nil { |
| 533 | return r.stuck(ctx, provision.StuckBaselineMismatch, why) |
| 534 | } |
| 535 | log.Printf("moves: %s: the source still matches what it recorded when it stopped", host) |
| 536 | return r.phase(ctx, provision.RbSourceVerified, provision.RbSourceProbing, nil, "") |
| 537 | } |
| 538 | |
| 539 | // warmMismatch is the warm database's half of the clean verification (fix |
| 540 | // round 1, ruling 1). A marker that is there must name src_gen. A database |
| 541 | // with no marker is refused unless src_gen is the origin (""), whose |
| 542 | // database never had one. A database evicted with its marker passes: the |
| 543 | // probe restores the source cold from src_gen (J14), so nothing warm is |
| 544 | // read. why is the mismatch; err a read that failed. |
| 545 | func (r *recoverRun) warmMismatch(srcGen string) (why, err error) { |
| 546 | db := r.dbPath() |
| 547 | gen, gerr := hostbox.ReadGenerationMarker(db + ".gen") |
| 548 | if gerr != nil { |
| 549 | return fmt.Errorf("the warm database's marker: %w", gerr), nil |
| 550 | } |
| 551 | present, err := regularFile(db) |
| 552 | if err != nil { |
| 553 | return nil, err |
| 554 | } |
| 555 | switch { |
| 556 | case gen != "" && gen != srcGen: |
| 557 | return fmt.Errorf("the warm database's marker names %q, the pre-move generation is %q", gen, srcGen), nil |
| 558 | case gen == "" && present && srcGen != "": |
| 559 | return fmt.Errorf("the warm database has no marker, the pre-move generation is %q", srcGen), nil |
| 560 | case gen == "" && !present: |
| 561 | log.Printf("moves: %s: the source's warm database was evicted; the probe restores generation %q cold", r.rec.Host, srcGen) |
| 562 | } |
| 563 | return nil, nil |
| 564 | } |
| 565 | |
| 566 | // returnSeal reads the return seal the target journalled in the claim. |
| 567 | func returnSeal(c *hostclaim.Claim) (moves.Seal, error) { |
| 568 | raw, ok := c.Op.Journal[moves.JournalReturnSeal] |
| 569 | if !ok { |
| 570 | return moves.Seal{}, fmt.Errorf("%w: the claim carries no return seal", moves.ErrBadSeal) |
| 571 | } |
| 572 | return moves.DecodeSeal(raw) |
| 573 | } |
| 574 | |
| 575 | // install (§7.8 dirty parking to configured): one phase of the shared |
| 576 | // installer over the return lane, its seal and baseline the target's |
| 577 | // journalled return. A refusal it records stops the return for an operator. |
| 578 | func (r *recoverRun) install(ctx context.Context) (bool, error) { |
| 579 | host := r.rec.Host |
| 580 | obs, err := r.claim(ctx) |
| 581 | if err != nil { |
| 582 | return false, err |
| 583 | } |
| 584 | raw, ok := obs.Claim.Op.Journal[moves.JournalReturnBaseline] |
| 585 | if !ok { |
| 586 | return false, fmt.Errorf("move %s: the claim carries no return baseline", r.g.ID) |
| 587 | } |
| 588 | baseline, err := moves.DecodeReturnBaseline(raw) |
| 589 | if err != nil { |
| 590 | return false, err |
| 591 | } |
| 592 | src, err := recoveryInstallSource(r.w, r.s, r.rec, returnSeal, &baseline) |
| 593 | if err != nil { |
| 594 | return false, err |
| 595 | } |
| 596 | advanced, refused, err := installGeneration(ctx, r.s, r.w, r.rec, src) |
| 597 | if refused != "" { |
| 598 | log.Printf("moves: %s: the return of %s stopped for an operator: %s", host, r.rb, refused) |
| 599 | // The stuck mark is installGeneration's write, logged there. |
| 600 | r.w.pageStuck(ctx, r.rec, "source", refused) |
| 601 | } |
| 602 | return advanced, err |
| 603 | } |
| 604 | |
| 605 | // probing (§7.8 source probing): the claim steps to probing before the |
| 606 | // first attempt (J1), then the recovery probe. Nothing earlier is run again |
| 607 | // from here. |
| 608 | func (r *recoverRun) probing(ctx context.Context) (bool, error) { |
| 609 | cfg := r.cfg() |
| 610 | host, id, rb := r.rec.Host, r.g.ID, r.rollback() |
| 611 | obs, err := r.claim(ctx) |
| 612 | if err != nil { |
| 613 | return false, err |
| 614 | } |
| 615 | if obs.Claim.Op.Phase == hostclaim.PhaseClaimed { |
| 616 | if err := r.executing(obs.Claim, hostclaim.PhaseClaimed); err != nil { |
| 617 | return false, err |
| 618 | } |
| 619 | if obs, err = hostclaim.Update(ctx, r.s, obs, hostclaim.SetRestoringPhase(id, hostclaim.PhaseClaimed, hostclaim.PhaseProbing)); err != nil { |
| 620 | return false, fmt.Errorf("the claim's probing: %w", err) |
| 621 | } |
| 622 | if err := r.fault("after-claim-probing"); err != nil { |
| 623 | return false, err |
| 624 | } |
| 625 | } |
| 626 | if err := r.executing(obs.Claim, hostclaim.PhaseProbing); err != nil { |
| 627 | return false, err |
| 628 | } |
| 629 | if cfg.recoveryProbe == nil { |
| 630 | r.w.notice(host+"#rc-probe", "moves: %s: the recovery probe of rollback %s waits: it is Task 9's, on claim Plan 10's recovery admission", host, r.rb) |
| 631 | return false, nil |
| 632 | } |
| 633 | req := recoveryProbeRequest{RollbackID: r.rb, Clean: !rb.Dirty, Attempts: recoveryProbeAttempts, |
| 634 | SourceGeneration: obs.Claim.Op.Journal[hostclaim.JournalSrcGen], |
| 635 | Record: func(ctx context.Context, in provision.MoveInputs) error { |
| 636 | _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRecordRecoveryInputs(r.g, r.rb, in)) |
| 637 | return err |
| 638 | }} |
| 639 | // A declared or fenced return asks no coverage (Q6); while the record |
| 640 | // and the claim disagree on the target's loss the probe waits. |
| 641 | declared, err := recoveryDeclared(obs.Claim, rb.LossRef) |
| 642 | if err != nil { |
| 643 | r.w.notice(host+"#rc-probe", "moves: %s: the recovery probe of rollback %s waits: %v", host, r.rb, err) |
| 644 | return false, nil |
| 645 | } |
| 646 | switch { |
| 647 | case rb.Dirty: |
| 648 | req.Coverage, req.Generation = recoveryCoverageSealed, r.rb |
| 649 | case declared: |
| 650 | req.Declared = true |
| 651 | default: |
| 652 | req.Coverage = recoveryCoverageClean |
| 653 | } |
| 654 | out, err := cfg.recoveryProbe(ctx, r.s, r.rec, req) |
| 655 | var missing *ClaimDependencyError |
| 656 | if errors.As(err, &missing) { |
| 657 | // The adapter refused before recording or waking (Plan 10 J2): |
| 658 | // probing waits, as it did with no hook. |
| 659 | r.w.notice(host+"#rc-probe", "moves: %s: the recovery probe of rollback %s waits: %v", host, r.rb, err) |
| 660 | return false, nil |
| 661 | } |
| 662 | if err != nil { |
| 663 | return false, fmt.Errorf("the recovery probe: %w", err) |
| 664 | } |
| 665 | switch out { |
| 666 | case recoveryProbeAnswered: |
| 667 | return r.phase(ctx, provision.RbSourceProbing, provision.RbSourceProbed, nil, "") |
| 668 | case recoveryProbeExhausted: |
| 669 | return r.stuck(ctx, provision.StuckRecoveryProbeAttempts, fmt.Errorf("%d attempts", recoveryProbeAttempts)) |
| 670 | } |
| 671 | r.w.notice(host+"#rc-probe", "moves: %s: the recovery probe of rollback %s has not answered yet", host, r.rb) |
| 672 | return false, nil |
| 673 | } |
| 674 | |
| 675 | // probed (§7.8 source probed): the claim steps to probed and, for a dirty |
| 676 | // return, queues the return lane's park in the same write; then recovered |
| 677 | // with recovered_at, and the serial is nudged for the target's cleaning. |
| 678 | func (r *recoverRun) probed(ctx context.Context) (bool, error) { |
| 679 | cfg := r.cfg() |
| 680 | host, id, rb := r.rec.Host, r.g.ID, r.rollback() |
| 681 | obs, err := r.claim(ctx) |
| 682 | if err != nil { |
| 683 | return false, err |
| 684 | } |
| 685 | var lane func(*hostclaim.Claim) error |
| 686 | if rb.Dirty { |
| 687 | seal, err := returnSeal(obs.Claim) |
| 688 | if err != nil { |
| 689 | return false, err |
| 690 | } |
| 691 | scope := moves.ReturnLaneScope(host, r.rb, obs.Claim.Holder.Incarnation, seal.Bound()) |
| 692 | lane = hostclaim.QueueReturnLanePark(id, cfg.parkRegion(), scope, obligationID) |
| 693 | } |
| 694 | switch obs.Claim.Op.Phase { |
| 695 | case hostclaim.PhaseProbing: |
| 696 | if err := r.executing(obs.Claim, hostclaim.PhaseProbing); err != nil { |
| 697 | return false, err |
| 698 | } |
| 699 | step := hostclaim.SetRestoringPhase(id, hostclaim.PhaseProbing, hostclaim.PhaseProbed) |
| 700 | if _, err := hostclaim.Update(ctx, r.s, obs, func(c *hostclaim.Claim) error { |
| 701 | if err := step(c); err != nil { |
| 702 | return err |
| 703 | } |
| 704 | if lane != nil { |
| 705 | return lane(c) |
| 706 | } |
| 707 | return nil |
| 708 | }); err != nil { |
| 709 | return false, fmt.Errorf("the claim's probed: %w", err) |
| 710 | } |
| 711 | case hostclaim.PhaseProbed: |
| 712 | // A replay after the claim's write: the lane park is idempotent, |
| 713 | // and once the park engine completed it it is not queued again. |
| 714 | if err := r.executing(obs.Claim, hostclaim.PhaseProbed); err != nil { |
| 715 | return false, err |
| 716 | } |
| 717 | if lane != nil { |
| 718 | if _, err := hostclaim.Update(ctx, r.s, obs, lane); err != nil { |
| 719 | return false, fmt.Errorf("queue the return lane's park: %w", err) |
| 720 | } |
| 721 | } |
| 722 | default: |
| 723 | return false, r.executing(obs.Claim, hostclaim.PhaseProbing) |
| 724 | } |
| 725 | if err := r.fault("after-claim-probed"); err != nil { |
| 726 | return false, err |
| 727 | } |
| 728 | ok, err := r.phase(ctx, provision.RbSourceProbed, provision.RbSourceRecovered, func(rb *provision.MoveRollback) { |
| 729 | t := r.now().UTC() |
| 730 | rb.RecoveredAt = &t |
| 731 | }, "source-recovered") |
| 732 | if ok { |
| 733 | nudgeSerial(ctx, r.s) |
| 734 | } |
| 735 | return ok, err |
| 736 | } |
| 737 | |
| 738 | // open (§7.8 source opened): the rollback's hold on the row ends, under the |
| 739 | // host's hold lock with the record read fresh (J12); the claim steps to |
| 740 | // opened; the record says opened with opened_at. A replay that finds the |
| 741 | // row already open goes on. |
| 742 | func (r *recoverRun) open(ctx context.Context) (bool, error) { |
| 743 | cfg := r.cfg() |
| 744 | host, id := r.rec.Host, r.g.ID |
| 745 | var err error |
| 746 | provision.WithHostHoldLock(host, func() { |
| 747 | if err = r.stillAt(ctx, provision.RbSourceRecovered); err != nil { |
| 748 | return |
| 749 | } |
| 750 | var released bool |
| 751 | if released, err = cfg.ReleaseHeld(host, r.rb); err != nil || released { |
| 752 | return |
| 753 | } |
| 754 | var rt *registry.Route |
| 755 | if rt, err = cfg.Route(host); err != nil { |
| 756 | return |
| 757 | } |
| 758 | switch { |
| 759 | case rt == nil: |
| 760 | err = fmt.Errorf("move %s: this box has no row to open", id) |
| 761 | case rt.HoldOp != "": |
| 762 | err = fmt.Errorf("move %s: the row is held %s by %q, not by rollback %s", id, rt.Status, rt.HoldOp, r.rb) |
| 763 | } |
| 764 | }) |
| 765 | if err != nil { |
| 766 | return false, fmt.Errorf("open the row: %w", err) |
| 767 | } |
| 768 | if err := r.fault("after-row-opened"); err != nil { |
| 769 | return false, err |
| 770 | } |
| 771 | obs, err := r.claim(ctx) |
| 772 | if err != nil { |
| 773 | return false, err |
| 774 | } |
| 775 | if obs.Claim.Op.Phase != hostclaim.PhaseOpened { |
| 776 | if err := r.executing(obs.Claim, hostclaim.PhaseProbed); err != nil { |
| 777 | return false, err |
| 778 | } |
| 779 | if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.SetRestoringPhase(id, hostclaim.PhaseProbed, hostclaim.PhaseOpened)); err != nil { |
| 780 | return false, fmt.Errorf("the claim's opened: %w", err) |
| 781 | } |
| 782 | } |
| 783 | if err := r.fault("after-claim-opened"); err != nil { |
| 784 | return false, err |
| 785 | } |
| 786 | return r.phase(ctx, provision.RbSourceRecovered, provision.RbSourceOpened, func(rb *provision.MoveRollback) { |
| 787 | t := r.now().UTC() |
| 788 | rb.OpenedAt = &t |
| 789 | }, "source-opened") |
| 790 | } |
| 791 | |
| 792 | // finalize (§7.8 rolled_back): with the source opened and recovered and the |
| 793 | // target parked, and no obligation of the op left in the claim, the |
| 794 | // terminal event, the record's close, then the claim's release to serving: |
| 795 | // record first (§7.5 and §19 K8 as Paul's ruling M4 amends them). A crash |
| 796 | // between the halves leaves the record at rest and the claim |
| 797 | // restoring/opened with nothing owed; the console's recovery pass releases |
| 798 | // from LastMove. A claim already released by this move (a binary that |
| 799 | // closed claim first) only closes. |
| 800 | func (r *recoverRun) finalize(ctx context.Context) (bool, error) { |
| 801 | cfg := r.cfg() |
| 802 | host, id, rb := r.rec.Host, r.g.ID, r.rollback() |
| 803 | if rb.RecoveredAt == nil || rb.OpenedAt == nil || rb.TargetParkedAt == nil { |
| 804 | r.w.notice(host+"#rc-final", "moves: %s: rollback %s waits for the target's park before it closes", host, r.rb) |
| 805 | return false, nil |
| 806 | } |
| 807 | obs, err := hostclaim.Read(ctx, r.s, host) |
| 808 | if err != nil { |
| 809 | return false, fmt.Errorf("read the host claim: %w", err) |
| 810 | } |
| 811 | if obs.Absent { |
| 812 | return false, fmt.Errorf("move %s has no host claim", id) |
| 813 | } |
| 814 | released := hostclaim.RequireReleasedMove(id)(obs.Claim) == nil |
| 815 | if !released { |
| 816 | if err := r.ours(obs.Claim); err != nil { |
| 817 | return false, fmt.Errorf("move %s: %w", id, err) |
| 818 | } |
| 819 | if err := r.executing(obs.Claim, hostclaim.PhaseOpened); err != nil { |
| 820 | return false, err |
| 821 | } |
| 822 | // Nothing owed under the op, checked before the record's close |
| 823 | // (M4), as the console's grant checks it. |
| 824 | if owed := owedUnder(obs.Claim, id); owed > 0 { |
| 825 | r.w.notice(host+"#rc-final", "moves: %s: rollback %s waits for %d chore(s) of the move to be acknowledged before it closes", host, r.rb, owed) |
| 826 | return false, nil |
| 827 | } |
| 828 | } |
| 829 | var pre provision.Record |
| 830 | closed, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveRolledBackClose(r.g, r.now()))) |
| 831 | if err != nil { |
| 832 | if errors.Is(err, provision.ErrMoveConflict) { |
| 833 | log.Printf("moves: %s: the close of rollback %s refused (%v); the record moved on", host, r.rb, err) |
| 834 | return false, nil |
| 835 | } |
| 836 | return false, fmt.Errorf("close the rolled-back move: %w", err) |
| 837 | } |
| 838 | r.w.logTransition(ctx, r.s, pre, *closed) |
| 839 | log.Printf("moves: %s: move %s is rolled back and closed", host, id) |
| 840 | r.w.pageRolledBack(ctx, r.rec) |
| 841 | nudgeSerial(ctx, r.s) |
| 842 | if !released { |
| 843 | if err := r.fault("after-close-record"); err != nil { |
| 844 | return true, err |
| 845 | } |
| 846 | if err := releaseAfterClose(ctx, r.s, host, id, hostclaim.ReleaseMove(id, cfg.Box)); err != nil { |
| 847 | return true, err |
| 848 | } |
| 849 | log.Printf("moves: %s: the claim is serving again after rollback %s", host, r.rb) |
| 850 | if err := r.fault("after-release-move"); err != nil { |
| 851 | return true, err |
| 852 | } |
| 853 | } |
| 854 | if err := r.fault("after-close"); err != nil { |
| 855 | return true, err |
| 856 | } |
| 857 | // The journals go in the pass that closes; a crash before this leaves |
| 858 | // them to the local journal sweep (sweepRecoveryJournal). |
| 859 | for _, op := range []string{r.rb, id} { |
| 860 | if err := parkRecoveryJournal(cfg.DataDir, host, op, true); err != nil { |
| 861 | return true, fmt.Errorf("park the source's journal of %s: %w", op, err) |
| 862 | } |
| 863 | } |
| 864 | return true, nil |
| 865 | } |
| 866 | |
| 867 | // The source's recovery steps in its two op journals, the cut's under the |
| 868 | // move id and the recovery's under the rollback id (§7.8: "journaled ... in |
| 869 | // <DataDir>/.moves/<host>/<rollback id>/"; a dirty return's install keeps |
| 870 | // its build and extract scratch there too). Fix round 1, ruling 2. |
| 871 | const ( |
| 872 | // rcStepSource marks a journal as a source recovery's (written at |
| 873 | // claimed): sourceJournalMark, so the sweep can tell it from a cut, a |
| 874 | // landing or a target's, and knows the incarnation to park it under. |
| 875 | rcStepSource = "rb-source" |
| 876 | // rcStepClosed is "done" once the record closed. |
| 877 | rcStepClosed = "rb-source-closed" |
| 878 | // rcStepJournalPark is parkLocalArtifacts' step: the retirement the |
| 879 | // journal is parked into (no artifacts of its own). |
| 880 | rcStepJournalPark = "rb-source-journal-park" |
| 881 | ) |
| 882 | |
| 883 | // sourceJournalMark is rcStepSource's value. |
| 884 | type sourceJournalMark struct { |
| 885 | Move string `json:"move"` |
| 886 | Rollback string `json:"rollback"` |
| 887 | Incarnation string `json:"incarnation"` |
| 888 | } |
| 889 | |
| 890 | // markJournals marks the source's journals as this recovery's: the cut's, |
| 891 | // when this box holds it, and the recovery's own, made here for a clean |
| 892 | // return (the installer's for a dirty one). A mark already there is kept. |
| 893 | func (r *recoverRun) markJournals() error { |
| 894 | dd, host := r.cfg().DataDir, r.rec.Host |
| 895 | body, err := json.Marshal(sourceJournalMark{Move: r.g.ID, Rollback: r.rb, Incarnation: r.g.Incarnation}) |
| 896 | if err != nil { |
| 897 | return err |
| 898 | } |
| 899 | for _, op := range []string{r.g.ID, r.rb} { |
| 900 | if op == r.g.ID && !moveJournalPresent(dd, host, op) { |
| 901 | continue |
| 902 | } |
| 903 | j, err := moves.ReadJournal(dd, host, op) |
| 904 | if err != nil { |
| 905 | return err |
| 906 | } |
| 907 | if j.Done(rcStepSource) { |
| 908 | continue |
| 909 | } |
| 910 | j.Mark(rcStepSource, string(body)) |
| 911 | if err := j.Commit(dd); err != nil { |
| 912 | return err |
| 913 | } |
| 914 | } |
| 915 | return nil |
| 916 | } |
| 917 | |
| 918 | // parkRecoveryJournal parks one source journal of a closed rollback: 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 journal's op) and the journal is renamed into it. A journal that is |
| 922 | // not there was parked already; one with no recovery mark is not this |
| 923 | // recovery's and is left. |
| 924 | func parkRecoveryJournal(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 | var mark sourceJournalMark |
| 933 | raw, ok := j.Steps[rcStepSource] |
| 934 | if !ok { |
| 935 | return nil |
| 936 | } |
| 937 | if err := json.Unmarshal([]byte(raw), &mark); err != nil || !journalComponent(mark.Incarnation) { |
| 938 | return fmt.Errorf("the journal of %s has an unreadable recovery 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[rcStepClosed] != "done" { |
| 945 | if err := write(rcStepClosed, "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: rcStepJournalPark, 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 %s is parked", host, op) |
| 958 | return nil |
| 959 | } |
| 960 | |
| 961 | // sweepRecoveryJournal is the local journal sweep's source-recovery half |
| 962 | // (sweepFinalJournal, fix round 1, ruling 2): a journal the recovery marked |
| 963 | // (rcStepSource) is parked once the close is shown by durable evidence: |
| 964 | // - the journal records the close (rcStepClosed "done"); |
| 965 | // - no listed record carries the move: a rolling_back or stuck move is |
| 966 | // cleared only by provision.MoveRolledBackClose, which refuses unless |
| 967 | // the rollback's recovered_at, opened_at and target_parked_at are all |
| 968 | // stamped (the aborts clear only an aborting move, MoveComplete only a |
| 969 | // cutting one), and the mark is written while the record is rolling |
| 970 | // back, so the close landed; |
| 971 | // - a listed record carries the move with exactly those three stamps on |
| 972 | // this rollback: the close's preconditions hold, nothing of the |
| 973 | // recovery reads its journals again, and the close follows. |
| 974 | // |
| 975 | // A host the listing does not name, or a record that carries the move |
| 976 | // without the three stamps, is left. |
| 977 | func (w *movesWorker) sweepRecoveryJournal(dataDir, host, op string, j *moves.Journal, recs []provision.Record) error { |
| 978 | var mark sourceJournalMark |
| 979 | if err := json.Unmarshal([]byte(j.Steps[rcStepSource]), &mark); err != nil || mark.Move == "" || mark.Rollback == "" { |
| 980 | return fmt.Errorf("the journal of %s has an unreadable recovery mark", op) |
| 981 | } |
| 982 | closed := j.Steps[rcStepClosed] == "done" |
| 983 | if !closed { |
| 984 | if len(recs) == 0 { |
| 985 | w.notice(host+"#rc-sweep-"+op, "moves: %s: the source's journal of %s is a recovery's, and no listed record names the host; it is left", host, op) |
| 986 | return nil |
| 987 | } |
| 988 | for _, rec := range recs { |
| 989 | m := rec.Move |
| 990 | if m == nil || m.ID != mark.Move { |
| 991 | continue |
| 992 | } |
| 993 | rb := m.Rollback |
| 994 | if rb == nil || rb.ID != mark.Rollback || rb.RecoveredAt == nil || rb.OpenedAt == nil || rb.TargetParkedAt == nil { |
| 995 | return nil |
| 996 | } |
| 997 | return parkRecoveryJournal(dataDir, host, op, false) |
| 998 | } |
| 999 | } |
| 1000 | return parkRecoveryJournal(dataDir, host, op, true) |
| 1001 | } |
| 1002 | |
amadan