| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "log" |
| 11 | "os" |
| 12 | "path/filepath" |
| 13 | "strconv" |
| 14 | "strings" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/carlosframework/platform/internal/appconfig" |
| 18 | "github.com/carlosframework/platform/internal/hostbox" |
| 19 | "github.com/carlosframework/platform/internal/hostclaim" |
| 20 | "github.com/carlosframework/platform/internal/hostguard" |
| 21 | "github.com/carlosframework/platform/internal/lease" |
| 22 | "github.com/carlosframework/platform/internal/moves" |
| 23 | "github.com/carlosframework/platform/internal/moves/archive" |
| 24 | "github.com/carlosframework/platform/internal/park" |
| 25 | "github.com/carlosframework/platform/internal/provision" |
| 26 | "github.com/carlosframework/platform/internal/release" |
| 27 | "github.com/superfly/ltx" |
| 28 | ) |
| 29 | |
| 30 | // The rollback on the target (moves spec §7.8; Plan 4 Task 4), behind |
| 31 | // CARLOS_MOVES_PHASE1: the decision that ends a landing, then the target |
| 32 | // box's half of the rollback: its release, its return lane and its cleaning. |
| 33 | // |
| 34 | // The decision (rollBackLanding). A landing the target can no longer finish |
| 35 | // is rolled back by one record CAS, MoveRollBack, which moves the region back |
| 36 | // to From; the claim follows (RollBack, or RollBackFlipped over a claim the |
| 37 | // flip's half never reached, J16), journalling the target installation's |
| 38 | // stamp F (J15). It runs at the top of the landing's pass, before its arming |
| 39 | // gate (J10), and is the one place a landing's recorded mark (every |
| 40 | // markRollback site, Plans 2 and 3) becomes a rollback. The deadline that |
| 41 | // triggers it is landingDeadlineOf, the earlier of the cut's and the |
| 42 | // landing's, at every landing phase, probing and probed included (J11). The |
| 43 | // commit races it on the record: a commit that won makes MoveRollBack refuse |
| 44 | // with ErrMoveConflict, and the decision stops there with no trace, leaving |
| 45 | // the record to the commit and the open (Plan 3 Task 7). |
| 46 | // |
| 47 | // The target's half (rollbackTarget). Once the record is rolling_back its |
| 48 | // region is the source's, so the target reaches it through the post-flip |
| 49 | // participant rule (movesParticipantRecord, J10). One pass drives it as far |
| 50 | // as it can, each phase from durable state, under the record's |
| 51 | // target_phase (§7.8 target table; Plan 2's phase-name convention, G8: a |
| 52 | // record at X means X's work may have begun): |
| 53 | // |
| 54 | // - releasing: the row's hold handed from the move to the rollback |
| 55 | // (handHold, J3), the wake fence, the timer drain and the stop evidence |
| 56 | // (stopevidence.Reconstruct); a target not yet proven stopped retries, |
| 57 | // and the release deadline is the console's to page on. Then, if the |
| 58 | // landing reached probing and the database is here, a final checkpoint |
| 59 | // under the writer's lease (finalCheckpoint, writerLease), N' from the |
| 60 | // local position, and the digests: the database, the files plain digest |
| 61 | // and the timers. The release is clean when the landing never reached |
| 62 | // probing or all three equal what was installed (installed_db_sha256, |
| 63 | // source.files.plain, source.timers_sha256), and the staged seal covers |
| 64 | // the source (src_covered). stopped-target.json is written create-only, |
| 65 | // the record goes to released with Dirty and Evidence, and the claim |
| 66 | // follows (TargetReleased: execution to none, the target region's |
| 67 | // location park, its park-local only when the inventory lists F, the |
| 68 | // target region's predecessors covered); |
| 69 | // - returning (dirty only): the return lane, moves/<host>/<rb id>/ in the |
| 70 | // source's bucket, written as a staging lane under the staging-work lease |
| 71 | // for (source region, host), every object through the guarded staging |
| 72 | // writer, one presigned request per call. A target that served and |
| 73 | // changed sends a snapshot [1, N'] of its stopped database |
| 74 | // (moves.EncodeFull, held in memory: Phase 1's size limit), its files |
| 75 | // archive (bound to the move id, which the source's installer checks) |
| 76 | // and its timers; one whose state is still the staged seal's (dirty only |
| 77 | // for want of src_covered) sends the staged seal's versions as given. |
| 78 | // The return seal (moves.ReturnSeal, bound = the lease's SafeUntil after |
| 79 | // the last write) and the return baseline go into the claim's journal in |
| 80 | // one write, then the record says returned; |
| 81 | // - cleaning, only once the source has recovered (RecoveredAt, Review |
| 82 | // Focus 4): the target region's location park and the move's lane park |
| 83 | // (queued here) run through the park engine until their archives are |
| 84 | // complete (D6: this region is no longer authoritative, so the engine |
| 85 | // selects everything); the registered installation is retired through |
| 86 | // park.Local, which completes its own park-local (J15), and everything |
| 87 | // else at the flat paths is parked by the moves-owned rename (H5(b)) |
| 88 | // inside the host guard; the row and its timers rows go in one registry |
| 89 | // transaction under the host's hold lock (J12); the record says cleaned |
| 90 | // with target_parked_at, and the op's journal is parked last. |
| 91 | // |
| 92 | // Arming (ruling J18; Plan 5 Task 9a). Like the landing, the rollback's |
| 93 | // target half needs the claim's park leases, view and authorizer, which |
| 94 | // armMovesParks sets only once the wake-source gate opens |
| 95 | // (movesWakeSourceRegistered, Plan 5 Task 9b), so rollbackTargetArmed |
| 96 | // answers false in production and the pass logs a standing notice. The decision is |
| 97 | // not gated (J10). The move's event log (moves.LogEvent, Plan 6) is the |
| 98 | // LogEvent hook: nil in production, where the event is a log line. The |
| 99 | // stuck mark pages an operator (pageStuck, the Page hook); the release |
| 100 | // deadline is the console's. |
| 101 | |
| 102 | // rollbackStepsPerPass bounds the phases one pass runs. |
| 103 | const rollbackStepsPerPass = 8 |
| 104 | |
| 105 | // The target's rollback steps in its op journal (the landing's journal of |
| 106 | // the move id, moves.Journal). Each step is skipped on a replay once done. |
| 107 | const ( |
| 108 | rbStepTargetStamp = "rb-target-stamp" // F, noted from the claim before anything is retired |
| 109 | rbStepLaneQueued = "rb-lane-queued" // the move's lane park is queued in this region |
| 110 | rbStepRemote = "rb-target-park-remote" // both parks' archives are complete, or acknowledged |
| 111 | rbStepInstallation = "rb-target-park-installation" // retired, completed or unregistered |
| 112 | rbStepLocal = "rb-target-park-local" // parkLocalArtifacts' step: the retirement id |
| 113 | rbStepRowDeleted = "rb-target-row-deleted" // "pending" before the delete, "done" after it |
| 114 | rbStepCleaned = "rb-target-cleaned" // "done": the record says cleaned, only the journal's park is left |
| 115 | ) |
| 116 | |
| 117 | // stoppedTargetRole is the target's StoppedReport role. |
| 118 | const stoppedTargetRole = "target" |
| 119 | |
| 120 | // now is the worker's clock. |
| 121 | func (w *movesWorker) now() time.Time { |
| 122 | if n := w.cfg.Now; n != nil { |
| 123 | return n() |
| 124 | } |
| 125 | return time.Now() |
| 126 | } |
| 127 | |
| 128 | // stuckPage is the stuck mark's page (pageStuck, movespaging.go): the subject (MV-N313) and the body |
| 129 | // (MV-N314). |
| 130 | func stuckPage(host, id, box, role, reason string) (subject, body string) { |
| 131 | subject = fmt.Sprintf("carlos move stuck: %s", host) |
| 132 | body = fmt.Sprintf("The move %s of %s stopped for an operator on %s, its %s: %s. Run carlos move resolve --host %s --retry once the cause is fixed.", |
| 133 | id, host, box, role, reason, host) |
| 134 | return subject, body |
| 135 | } |
| 136 | |
| 137 | // rollBackLanding is the rollback decision on the target (§7.8; J10, J11): |
| 138 | // a landing marked for a rollback, or past landingDeadlineOf at any phase, |
| 139 | // is rolled back by MoveRollBack, and the claim follows. done is true when |
| 140 | // the record was this decision's to take (written, or refused because the |
| 141 | // record moved on); the landing's pass then ends. |
| 142 | func (w *movesWorker) rollBackLanding(ctx context.Context, s release.Store, rec provision.Record) (done bool, err error) { |
| 143 | m := rec.Move |
| 144 | if m == nil || m.Legacy() || m.State != provision.MoveLanding { |
| 145 | return false, nil |
| 146 | } |
| 147 | reason := m.Reason |
| 148 | if reason == "" { |
| 149 | d, which := landingDeadlineOf(m) |
| 150 | if d == nil || !w.now().After(*d) { |
| 151 | return false, nil |
| 152 | } |
| 153 | log.Printf("moves: %s: landing %s is past its %s (%s)", rec.Host, m.ID, which, d.UTC().Format(time.RFC3339)) |
| 154 | reason = provision.RollbackLandingDeadline |
| 155 | } |
| 156 | rbID, err := moves.NewRollbackID(rand.Reader) |
| 157 | if err != nil { |
| 158 | return true, err |
| 159 | } |
| 160 | g := provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation} |
| 161 | var pre provision.Record |
| 162 | fresh, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, provision.MoveRollBack(g, rbID, reason, w.now()))) |
| 163 | if errors.Is(err, provision.ErrMoveConflict) { |
| 164 | // A commit that won (the open is the commit's path), or another |
| 165 | // writer's rollback (the participant pass follows it). |
| 166 | log.Printf("moves: %s: the rollback of landing %s is refused (%v); the record moved on", rec.Host, m.ID, err) |
| 167 | return true, nil |
| 168 | } |
| 169 | if err != nil { |
| 170 | return true, fmt.Errorf("roll the landing back: %w", err) |
| 171 | } |
| 172 | log.Printf("moves: %s: landing %s rolls back as %s: %s", rec.Host, m.ID, rbID, fresh.Move.Rollback.Reason) |
| 173 | w.logTransition(ctx, s, pre, *fresh) |
| 174 | if _, err := w.rollbackClaim(ctx, s, *fresh); err != nil { |
| 175 | log.Printf("moves: %s: the claim half of rollback %s waits for the next pass: %v", rec.Host, rbID, err) |
| 176 | } |
| 177 | nudgeSerial(ctx, s) |
| 178 | return true, nil |
| 179 | } |
| 180 | |
| 181 | // rollbackClaim is the claim half of the record's rollback: the op of this |
| 182 | // move, for the record's incarnation, goes to restoring/releasing under the |
| 183 | // rollback id (RollBackFlipped writes the flip first over a claim still at |
| 184 | // moving/sealed). A claim already restoring for this rollback is done. |
| 185 | func (w *movesWorker) rollbackClaim(ctx context.Context, s release.Store, rec provision.Record) (hostclaim.Observed, error) { |
| 186 | m := rec.Move |
| 187 | obs, err := hostclaim.Read(ctx, s, rec.Host) |
| 188 | if err != nil { |
| 189 | return obs, fmt.Errorf("read the host claim: %w", err) |
| 190 | } |
| 191 | if obs.Absent { |
| 192 | return obs, fmt.Errorf("move %s has no host claim", m.ID) |
| 193 | } |
| 194 | if err := hostclaim.RequireOp(m.ID)(obs.Claim); err != nil { |
| 195 | return obs, fmt.Errorf("move %s: %w", m.ID, err) |
| 196 | } |
| 197 | h := obs.Claim.Holder |
| 198 | if h == nil || h.Incarnation != m.Incarnation { |
| 199 | return obs, fmt.Errorf("move %s: the host claim names another incarnation", m.ID) |
| 200 | } |
| 201 | rb := m.Rollback.ID |
| 202 | if op := obs.Claim.Op; op.Kind == hostclaim.KindRestoring && op.Intent["rollback"] == rb { |
| 203 | return obs, nil |
| 204 | } |
| 205 | if obs, err = hostclaim.Update(ctx, s, obs, hostclaim.RollBackFlipped(m.ID, rb, w.cfg.Box, h.Stamp, h.Generation)); err != nil { |
| 206 | return obs, fmt.Errorf("the claim half of the rollback: %w", err) |
| 207 | } |
| 208 | log.Printf("moves: %s: the claim follows rollback %s", rec.Host, rb) |
| 209 | return obs, nil |
| 210 | } |
| 211 | |
| 212 | // rollbackTargetArmed reports whether this box can run the target's half of |
| 213 | // a rollback: every hook it calls is wired, the claim's three park hooks and |
| 214 | // its local authorizer (M9: the installation's park.Local) included. In |
| 215 | // production armMovesParks sets them only once the wake-source gate opens |
| 216 | // (movesWakeSourceRegistered, Plan 5 Task 9b), so this answers false there |
| 217 | // (J18). |
| 218 | func rollbackTargetArmed(cfg moveCfg) bool { |
| 219 | return cfg.Box != "" && cfg.DataDir != "" && cfg.Guard.DataDir != "" && cfg.Local.DataDir != "" && parkHooksArmed(cfg) && cfg.ParkLocalAuthorize != nil && |
| 220 | cfg.HandHold != nil && cfg.Fence != nil && cfg.TimersDrained != nil && cfg.Evidence != nil && cfg.FinalCheckpoint != nil && |
| 221 | cfg.LocalPosition != nil && cfg.Timers != nil && cfg.FilesDir != nil && cfg.Route != nil && cfg.Versioned != nil && |
| 222 | cfg.StagingLease != nil && cfg.Staging != nil && cfg.DeleteHeld != nil && cfg.SecretsPub != nil |
| 223 | } |
| 224 | |
| 225 | // rollbackRun is one pass's state for one rolling_back record on its target. |
| 226 | type rollbackRun struct { |
| 227 | w *movesWorker |
| 228 | s release.Store |
| 229 | rec provision.Record |
| 230 | g provision.MoveGuard |
| 231 | rb string |
| 232 | } |
| 233 | |
| 234 | // rollbackTarget is the participant step for a rolling_back record on its |
| 235 | // target (movesPass, J10). |
| 236 | func (w *movesWorker) rollbackTarget(ctx context.Context, s release.Store, rec provision.Record) error { |
| 237 | cfg := w.cfg |
| 238 | m := rec.Move |
| 239 | if m == nil || m.Legacy() || m.State != provision.MoveRollingBack || m.Rollback == nil || m.To != cfg.Region || m.From == cfg.Region { |
| 240 | return nil |
| 241 | } |
| 242 | // The landing is abandoned: a location lease its build still holds on |
| 243 | // this worker (kept across passes from building through published) is |
| 244 | // released on entry, whoever decided the rollback, so the cleaning's |
| 245 | // park engine can take it (the final review's finding 2). Idempotent: |
| 246 | // nothing held is nothing to release. |
| 247 | releaseOwnLease(w, rec.Host) |
| 248 | if !rollbackTargetArmed(cfg) { |
| 249 | w.notice(rec.Host+"#rollback", "moves: %s: the target's half of rollback %s waits: this box's rollback 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) |
| 250 | return nil |
| 251 | } |
| 252 | r := &rollbackRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, rb: m.Rollback.ID} |
| 253 | for i := 0; i < rollbackStepsPerPass; i++ { |
| 254 | advanced, err := r.step(ctx) |
| 255 | if err != nil || !advanced { |
| 256 | return err |
| 257 | } |
| 258 | fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| 259 | if err != nil { |
| 260 | return fmt.Errorf("re-read the record: %w", err) |
| 261 | } |
| 262 | if fm := fresh.Move; fm == nil || fm.ID != r.g.ID || fm.Incarnation != r.g.Incarnation || fm.State != provision.MoveRollingBack || fm.Rollback == nil || fm.Rollback.ID != r.rb { |
| 263 | return nil |
| 264 | } |
| 265 | r.rec = *fresh |
| 266 | } |
| 267 | return nil |
| 268 | } |
| 269 | |
| 270 | func (r *rollbackRun) cfg() moveCfg { return r.w.cfg } |
| 271 | func (r *rollbackRun) now() time.Time { return r.w.now() } |
| 272 | func (r *rollbackRun) rollback() *provision.MoveRollback { return r.rec.Move.Rollback } |
| 273 | func (r *rollbackRun) dbPath() string { return provision.DefaultDBPath(r.cfg().DataDir, r.rec.Host) } |
| 274 | |
| 275 | // fault is the test seam after a side effect. |
| 276 | func (r *rollbackRun) fault(point string) error { |
| 277 | if f := r.cfg().rbFault; f != nil { |
| 278 | return f(point) |
| 279 | } |
| 280 | return nil |
| 281 | } |
| 282 | |
| 283 | // step runs the target's current rollback phase once. advanced is true when |
| 284 | // it wrote the record. |
| 285 | func (r *rollbackRun) step(ctx context.Context) (bool, error) { |
| 286 | switch r.rollback().TargetPhase { |
| 287 | case provision.RbTargetReleasing: |
| 288 | return r.releasing(ctx) |
| 289 | case provision.RbTargetReleased: |
| 290 | return r.released(ctx) |
| 291 | case provision.RbTargetReturning: |
| 292 | return r.returning(ctx) |
| 293 | case provision.RbTargetReturned: |
| 294 | return r.toCleaning(ctx, provision.RbTargetReturned) |
| 295 | case provision.RbTargetCleaning: |
| 296 | return r.cleaning(ctx) |
| 297 | case provision.RbTargetCleaned: |
| 298 | return false, r.cleaned() |
| 299 | } |
| 300 | return false, nil // lost: the declared loss's (Task 10) |
| 301 | } |
| 302 | |
| 303 | // phase writes one target phase step, logs it (movesevents.go names it from |
| 304 | // the record; the last argument, the step's old event name, is unused) and |
| 305 | // passes the step's fault point. A conflict is someone else's decision. |
| 306 | func (r *rollbackRun) phase(ctx context.Context, from, to string, stamp func(*provision.MoveRollback), _ string) (bool, error) { |
| 307 | host := r.rec.Host |
| 308 | var pre provision.Record |
| 309 | after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveRollbackTargetPhase(r.g, from, to, stamp))) |
| 310 | if err != nil { |
| 311 | if errors.Is(err, provision.ErrMoveConflict) { |
| 312 | log.Printf("moves: %s: the target's rollback %s to %s refused (%v); the record moved on", host, from, to, err) |
| 313 | return false, nil |
| 314 | } |
| 315 | return false, fmt.Errorf("the target's rollback %s to %s: %w", from, to, err) |
| 316 | } |
| 317 | log.Printf("moves: %s: the target of rollback %s is %s", host, r.rb, to) |
| 318 | r.w.logTransition(ctx, r.s, pre, *after) |
| 319 | if err := r.fault("after-record-" + to); err != nil { |
| 320 | return true, err |
| 321 | } |
| 322 | return true, nil |
| 323 | } |
| 324 | |
| 325 | // claim reads the host claim and requires this rollback's restoring op, |
| 326 | // with this box the move's target, for the record's incarnation. |
| 327 | func (r *rollbackRun) claim(ctx context.Context) (hostclaim.Observed, error) { |
| 328 | obs, err := hostclaim.Read(ctx, r.s, r.rec.Host) |
| 329 | if err != nil { |
| 330 | return obs, fmt.Errorf("read the host claim: %w", err) |
| 331 | } |
| 332 | if obs.Absent { |
| 333 | return obs, fmt.Errorf("move %s has no host claim", r.g.ID) |
| 334 | } |
| 335 | if err := r.ours(obs.Claim); err != nil { |
| 336 | return obs, fmt.Errorf("move %s: %w", r.g.ID, err) |
| 337 | } |
| 338 | return obs, nil |
| 339 | } |
| 340 | |
| 341 | // ours is the claim check of every target phase: this rollback's restoring |
| 342 | // op, this box the move's target, the record's incarnation. |
| 343 | func (r *rollbackRun) ours(c *hostclaim.Claim) error { |
| 344 | box := r.cfg().Box |
| 345 | if c == nil || c.Op.ID != r.g.ID || c.Op.Kind != hostclaim.KindRestoring || c.Op.Intent["rollback"] != r.rb { |
| 346 | return fmt.Errorf("%w: the claim is not rollback %s", hostclaim.ErrWrongState, r.rb) |
| 347 | } |
| 348 | if c.Holder == nil || c.Holder.Incarnation != r.g.Incarnation { |
| 349 | return fmt.Errorf("%w: the claim names another incarnation", hostclaim.ErrWrongState) |
| 350 | } |
| 351 | if len(c.Op.Participants) != 2 || c.Op.Participants[1] != box || c.Op.Intent["target"] != box { |
| 352 | return fmt.Errorf("%w: this box is not the move's target in the claim", hostclaim.ErrWrongState) |
| 353 | } |
| 354 | return nil |
| 355 | } |
| 356 | |
| 357 | // stillReleasing is the final checkpoint's authority: every replica request |
| 358 | // re-reads the claim, which must still name this rollback releasing with this |
| 359 | // box executing. |
| 360 | func (r *rollbackRun) stillReleasing(ctx context.Context) error { |
| 361 | obs, err := r.claim(ctx) |
| 362 | if err != nil { |
| 363 | return err |
| 364 | } |
| 365 | if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleasing || op.Exec != r.cfg().Box { |
| 366 | return fmt.Errorf("move %s: this box no longer executes the host", r.g.ID) |
| 367 | } |
| 368 | return nil |
| 369 | } |
| 370 | |
| 371 | // stillReleasingRecord is the hand-over's record check (J3, fix round 1, |
| 372 | // ruling 1), run by handHold under the host's hold lock: the record, read |
| 373 | // fresh, must still be this move and incarnation, rolling back under this |
| 374 | // rollback, with the target releasing. It reads the record only: the lock is |
| 375 | // the wrapper's, and nothing here takes it again. |
| 376 | func (r *rollbackRun) stillReleasingRecord(ctx context.Context) error { |
| 377 | fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host) |
| 378 | if err != nil { |
| 379 | return fmt.Errorf("re-read the record: %w", err) |
| 380 | } |
| 381 | if m := fresh.Move; m == nil || m.ID != r.g.ID || m.Incarnation != r.g.Incarnation || m.State != provision.MoveRollingBack || |
| 382 | m.Rollback == nil || m.Rollback.ID != r.rb || m.Rollback.TargetPhase != provision.RbTargetReleasing { |
| 383 | return fmt.Errorf("move %s: the record is no longer rollback %s releasing its target", r.g.ID, r.rb) |
| 384 | } |
| 385 | return nil |
| 386 | } |
| 387 | |
| 388 | // reachedProbing reports whether the landing reached probing: a writer may |
| 389 | // have run on this box. |
| 390 | func (r *rollbackRun) reachedProbing() bool { |
| 391 | order := provision.LandingOrder() |
| 392 | return landingIndex(order, r.rec.Move.LandingPhase) >= landingIndex(order, provision.LandProbing) |
| 393 | } |
| 394 | |
| 395 | // unchanged reports whether the stopped target's digests are what the |
| 396 | // landing installed: the database, the files and the timers. |
| 397 | func (r *rollbackRun) unchanged(report moves.StoppedReport) bool { |
| 398 | m := r.rec.Move |
| 399 | return m.Landing != nil && m.Landing.InstalledDBSHA256 != "" && report.DBSHA256 == m.Landing.InstalledDBSHA256 && |
| 400 | m.Source != nil && m.Source.Files != nil && report.FilesPlain == m.Source.Files.Plain && report.TimersSHA256 == m.Source.TimersSHA256 |
| 401 | } |
| 402 | |
| 403 | // releasing (§7.8 target releasing and released): the hold, the fence, the |
| 404 | // drain and the stop evidence; the final checkpoint and the digests; the |
| 405 | // stopped report; the record's released with Dirty and Evidence; the claim's |
| 406 | // release. A replay runs it all again: the hand-over, the fence and the drain |
| 407 | // are idempotent, the evidence reads durable state, the checkpoint is |
| 408 | // idempotent on a stopped database, and the report is create-only and read |
| 409 | // back, so the decision is the published report's. |
| 410 | func (r *rollbackRun) releasing(ctx context.Context) (bool, error) { |
| 411 | cfg := r.cfg() |
| 412 | host, id := r.rec.Host, r.g.ID |
| 413 | if _, err := r.w.rollbackClaim(ctx, r.s, r.rec); err != nil { |
| 414 | return false, err |
| 415 | } |
| 416 | obs, err := r.claim(ctx) |
| 417 | if err != nil { |
| 418 | return false, err |
| 419 | } |
| 420 | if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleasing || op.Exec != cfg.Box { |
| 421 | return false, fmt.Errorf("move %s: the claim is restoring/%q with executor %q, not releasing on this box", id, op.Phase, op.Exec) |
| 422 | } |
| 423 | holder := *obs.Claim.Holder |
| 424 | if _, err := writerLease(cfg.OwnPrefix, host, holder); err != nil { |
| 425 | return false, fmt.Errorf("move %s: %w", id, err) |
| 426 | } |
| 427 | seal, err := journalledSeal(obs.Claim) |
| 428 | if err != nil { |
| 429 | return false, err |
| 430 | } |
| 431 | handed, err := cfg.HandHold(host, "releasing", r.rb, id, func() error { return r.stillReleasingRecord(ctx) }) |
| 432 | if err != nil { |
| 433 | return false, fmt.Errorf("hand the row's hold to the rollback: %w", err) |
| 434 | } |
| 435 | if !handed { |
| 436 | return false, fmt.Errorf("move %s: this box's row is held by neither the move nor rollback %s", id, r.rb) |
| 437 | } |
| 438 | gen := moveFenceGeneration(r.rec.Move) |
| 439 | if err := cfg.Fence(ctx, host, gen); err != nil { |
| 440 | return false, fmt.Errorf("fence against wakes: %w", err) |
| 441 | } |
| 442 | r.w.fenced.Store(host, gen) |
| 443 | if err := cfg.TimersDrained(ctx, host); err != nil { |
| 444 | return false, fmt.Errorf("drain the timers: %w", err) |
| 445 | } |
| 446 | dbPath := r.dbPath() |
| 447 | ev, err := cfg.Evidence(ctx, host, holder.Location, []string{dbPath}) |
| 448 | if err != nil { |
| 449 | return false, fmt.Errorf("stop evidence: %w", err) |
| 450 | } |
| 451 | if !ev.TreeStopped() { |
| 452 | r.w.notice(host+"#rb-evidence", "moves: %s: the target is not proven stopped yet (%s); the release retries until its deadline", host, failedSteps(ev)) |
| 453 | return false, nil |
| 454 | } |
| 455 | reached := r.reachedProbing() |
| 456 | fromDB, err := regularFile(dbPath) |
| 457 | if err != nil { |
| 458 | return false, err |
| 459 | } |
| 460 | served := reached && fromDB |
| 461 | if served { |
| 462 | if err := cfg.FinalCheckpoint(ctx, host, dbPath, holder, r.stillReleasing); err != nil { |
| 463 | return false, fmt.Errorf("final checkpoint: %w", err) |
| 464 | } |
| 465 | } |
| 466 | if err := r.fault("after-checkpoint"); err != nil { |
| 467 | return false, err |
| 468 | } |
| 469 | n := seal.N |
| 470 | if served { |
| 471 | if n, err = cfg.LocalPosition(dbPath); err != nil { |
| 472 | return false, fmt.Errorf("read the stopped database's position: %w", err) |
| 473 | } |
| 474 | } |
| 475 | var dbSHA string |
| 476 | if fromDB { |
| 477 | if dbSHA, err = fileSHA256(dbPath); err != nil { |
| 478 | return false, fmt.Errorf("digest the stopped database: %w", err) |
| 479 | } |
| 480 | } |
| 481 | dir, envPath := cfg.FilesDir(host) |
| 482 | plain, _, _, err := archivePlainDigest(dir, envPath) |
| 483 | if err != nil { |
| 484 | if reached && archiveRefused(err) { |
| 485 | // Files the archive can never carry: no retry by itself can |
| 486 | // help, so the release stops for an operator (Plan 5 Task 9a), |
| 487 | // keeping its phase and everything the target holds. |
| 488 | return r.stuck(ctx, &returnRefusal{reason: provision.StuckReleaseTargetFiles, |
| 489 | err: fmt.Errorf("move %s: the target's stopped files cannot be archived: %w", id, err)}) |
| 490 | } |
| 491 | if reached { |
| 492 | return false, fmt.Errorf("digest the stopped files: %w", err) |
| 493 | } |
| 494 | // No writer ran here: the files are not compared, and may not |
| 495 | // have been extracted yet. |
| 496 | plain = "" |
| 497 | } |
| 498 | rows, err := cfg.Timers(host) |
| 499 | if err != nil { |
| 500 | return false, fmt.Errorf("read the timers: %w", err) |
| 501 | } |
| 502 | report := ev |
| 503 | report.V, report.Role, report.Box, report.Incarnation, report.ClaimGen = 1, stoppedTargetRole, cfg.Box, holder.Incarnation, obs.Claim.Gen |
| 504 | report.Host, report.OpID = host, id |
| 505 | report.Steps = map[string]moves.Check{} |
| 506 | for k, v := range ev.Steps { |
| 507 | report.Steps[k] = v |
| 508 | } |
| 509 | report.Steps[moves.StepReplica] = moves.Check{OK: true} |
| 510 | report.FinalTXID, report.DBSHA256, report.FilesPlain, report.TimersSHA256 = n, dbSHA, plain, moves.TimersDigest(rows) |
| 511 | // The report's time is after the checkpoint, the target's last write to |
| 512 | // its location: it is the horizon of that location's park. |
| 513 | report.CheckedAt = r.now().UTC() |
| 514 | if !report.Complete() { |
| 515 | return false, fmt.Errorf("the target's stopped report is not complete (%s)", failedSteps(report)) |
| 516 | } |
| 517 | key := moves.StoppedKey(moves.ReportPrefix(r.rec.Account, r.rec.App, host, id), stoppedTargetRole) |
| 518 | if report, err = publishStoppedReport(ctx, r.s, key, report); err != nil { |
| 519 | return false, err |
| 520 | } |
| 521 | if err := r.fault("after-report"); err != nil { |
| 522 | return false, err |
| 523 | } |
| 524 | clean := seal.SrcCovered && (!reached || r.unchanged(report)) |
| 525 | log.Printf("moves: %s: the target of rollback %s stopped at TXID %d (clean %t, src_covered %t)", host, r.rb, report.FinalTXID, clean, seal.SrcCovered) |
| 526 | ok, err := r.phase(ctx, provision.RbTargetReleasing, provision.RbTargetReleased, func(rb *provision.MoveRollback) { |
| 527 | rb.Dirty, rb.Evidence = !clean, key |
| 528 | }, "target-released") |
| 529 | if !ok || err != nil { |
| 530 | return ok, err |
| 531 | } |
| 532 | if err := r.claimReleased(ctx, key); err != nil { |
| 533 | log.Printf("moves: %s: the claim half of the target's release waits for the next pass: %v", host, err) |
| 534 | } |
| 535 | return true, nil |
| 536 | } |
| 537 | |
| 538 | // claimReleased is the claim half of released (TargetReleased): execution to |
| 539 | // none citing the target's evidence, the target region's location park whose |
| 540 | // horizon is the report's time, the target's park-local when its inventory |
| 541 | // lists F, and the region's predecessors covered. A claim already released |
| 542 | // with this evidence is done. |
| 543 | func (r *rollbackRun) claimReleased(ctx context.Context, evidence string) error { |
| 544 | cfg := r.cfg() |
| 545 | obs, err := r.claim(ctx) |
| 546 | if err != nil { |
| 547 | return err |
| 548 | } |
| 549 | op := obs.Claim.Op |
| 550 | if op.Phase != hostclaim.PhaseReleasing { |
| 551 | if op.Journal[hostclaim.JournalEvidenceTarget] == evidence { |
| 552 | return nil |
| 553 | } |
| 554 | return fmt.Errorf("move %s: the claim is restoring/%q and cites %q, not %s", r.g.ID, op.Phase, op.Journal[hostclaim.JournalEvidenceTarget], evidence) |
| 555 | } |
| 556 | report, err := readStoppedReport(ctx, r.s, evidence) |
| 557 | if err != nil { |
| 558 | return err |
| 559 | } |
| 560 | holder := obs.Claim.Holder |
| 561 | scope := moves.TargetLocationScope(*holder, report.CheckedAt) |
| 562 | if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.TargetReleased(r.g.ID, cfg.Box, cfg.parkRegion(), evidence, scope, obligationID)); err != nil { |
| 563 | return fmt.Errorf("the claim half of the release: %w", err) |
| 564 | } |
| 565 | log.Printf("moves: %s: the target gave execution up for rollback %s, citing %s", r.rec.Host, r.rb, evidence) |
| 566 | return nil |
| 567 | } |
| 568 | |
| 569 | // released: the claim half, if a crash left it, then returning (dirty) or, |
| 570 | // once the source has recovered, cleaning (clean). |
| 571 | func (r *rollbackRun) released(ctx context.Context) (bool, error) { |
| 572 | if err := r.claimReleased(ctx, r.rollback().Evidence); err != nil { |
| 573 | return false, err |
| 574 | } |
| 575 | if r.rollback().Dirty { |
| 576 | return r.phase(ctx, provision.RbTargetReleased, provision.RbTargetReturning, nil, "") |
| 577 | } |
| 578 | return r.toCleaning(ctx, provision.RbTargetReleased) |
| 579 | } |
| 580 | |
| 581 | // toCleaning waits for the source's recovery (Review Focus 4): until then the |
| 582 | // target keeps everything it holds. |
| 583 | func (r *rollbackRun) toCleaning(ctx context.Context, from string) (bool, error) { |
| 584 | if r.rollback().RecoveredAt == nil { |
| 585 | r.w.notice(r.rec.Host+"#rb-clean", "moves: %s: the target keeps what it holds for rollback %s until the source has recovered", r.rec.Host, r.rb) |
| 586 | return false, nil |
| 587 | } |
| 588 | return r.phase(ctx, from, provision.RbTargetCleaning, nil, "") |
| 589 | } |
| 590 | |
| 591 | // returnedLane is what a return lane holds and restores to. |
| 592 | type returnedLane struct { |
| 593 | n uint64 |
| 594 | replica, files, timers []moves.SealedObject |
| 595 | // state is the returned build state (settings spec rev 11 §4.3a), nil |
| 596 | // when there is none. |
| 597 | state []moves.SealedObject |
| 598 | baseline provision.MoveSource |
| 599 | } |
| 600 | |
| 601 | // returning (§7.8 target returning): the return lane under the source's |
| 602 | // staging-work lease, the return seal and baseline in the claim's journal, |
| 603 | // then the record's returned. A replay that finds the seal journalled only |
| 604 | // writes the record; one that does not writes the lane again (new versions; |
| 605 | // the seal names the ones it wrote, and the lane's park removes the rest). |
| 606 | func (r *rollbackRun) returning(ctx context.Context) (bool, error) { |
| 607 | cfg := r.cfg() |
| 608 | host, id, m := r.rec.Host, r.g.ID, r.rec.Move |
| 609 | rb := r.rollback() |
| 610 | obs, err := r.claim(ctx) |
| 611 | if err != nil { |
| 612 | return false, err |
| 613 | } |
| 614 | if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleased || op.Exec != "" || op.Journal[hostclaim.JournalEvidenceTarget] != rb.Evidence { |
| 615 | return false, fmt.Errorf("move %s: the claim is restoring/%q with executor %q, not released by this target", id, op.Phase, op.Exec) |
| 616 | } |
| 617 | if _, ok := obs.Claim.Op.Journal[moves.JournalReturnSeal]; ok { |
| 618 | return r.phase(ctx, provision.RbTargetReturning, provision.RbTargetReturned, nil, "target-returned") |
| 619 | } |
| 620 | report, err := readStoppedReport(ctx, r.s, rb.Evidence) |
| 621 | if err != nil { |
| 622 | return false, err |
| 623 | } |
| 624 | staged, err := journalledSeal(obs.Claim) |
| 625 | if err != nil { |
| 626 | return false, err |
| 627 | } |
| 628 | fleet, err := loadMoveFleet(ctx, r.s) |
| 629 | if err != nil { |
| 630 | return false, err |
| 631 | } |
| 632 | src, ok := fleet.targets[m.From] |
| 633 | if !ok { |
| 634 | return false, fmt.Errorf("source region %s publishes no data bucket. waiting", m.From) |
| 635 | } |
| 636 | held, release, err := cfg.StagingLease(ctx, m.From, host) |
| 637 | if err != nil { |
| 638 | var he *lease.HeldError |
| 639 | if errors.As(err, &he) { |
| 640 | r.w.notice(host+"#rb-lease", "moves: %s: another worker holds the staging lease; the return of %s waits", host, r.rb) |
| 641 | return false, nil |
| 642 | } |
| 643 | return false, fmt.Errorf("staging lease: %w", err) |
| 644 | } |
| 645 | defer func() { |
| 646 | rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 647 | defer cancel() |
| 648 | if err := release(rctx); err != nil { |
| 649 | log.Printf("moves: %s: release the staging lease: %v", host, err) |
| 650 | } |
| 651 | }() |
| 652 | stg, err := r.w.staging(ctx, src.DataBucketRegion) |
| 653 | if err != nil { |
| 654 | return false, err |
| 655 | } |
| 656 | own, err := r.w.mirror(ctx, cfg.OwnBucketRegion) |
| 657 | if err != nil { |
| 658 | return false, err |
| 659 | } |
| 660 | gm := &guardedMirror{w: stg, g: moves.Guarded{Lease: held, Now: r.now}, dst: src.DataBucket, own: own} |
| 661 | if err := r.laneMoveJSON(ctx, gm); err != nil { |
| 662 | return false, err |
| 663 | } |
| 664 | var lane returnedLane |
| 665 | if !r.reachedProbing() || r.unchanged(report) { |
| 666 | lane, err = r.copyAsGiven(ctx, gm, staged) |
| 667 | } else { |
| 668 | lane, err = r.snapshotStopped(ctx, gm, report) |
| 669 | } |
| 670 | if err == nil { |
| 671 | lane.state, err = r.returnState(ctx, gm, obs.Claim.Holder, staged) |
| 672 | } |
| 673 | var refused *returnRefusal |
| 674 | if errors.As(err, &refused) { |
| 675 | return r.stuck(ctx, refused) |
| 676 | } |
| 677 | if err != nil { |
| 678 | return false, err |
| 679 | } |
| 680 | // The bound is read after the last lane write: every request the guard |
| 681 | // let a writer sign expires before it. |
| 682 | bound, err := held.SafeUntil() |
| 683 | if err != nil { |
| 684 | return false, fmt.Errorf("the staging lease's bound: %w", err) |
| 685 | } |
| 686 | seal := moves.ReturnSeal(lane.n, lane.replica, lane.files, lane.timers, lane.state, r.now(), bound) |
| 687 | if err := seal.Validate(); err != nil { |
| 688 | return false, err |
| 689 | } |
| 690 | if !seal.ReachesN() { |
| 691 | return false, fmt.Errorf("the return lane does not end at TXID %d", seal.N) |
| 692 | } |
| 693 | base, err := moves.EncodeReturnBaseline(lane.baseline) |
| 694 | if err != nil { |
| 695 | return false, err |
| 696 | } |
| 697 | if obs, err = r.claim(ctx); err != nil { |
| 698 | return false, err |
| 699 | } |
| 700 | if _, err := hostclaim.Update(ctx, r.s, obs, journalReturn(id, r.rb, rb.Evidence, seal.Encode(), base)); err != nil { |
| 701 | return false, fmt.Errorf("journal the return seal: %w", err) |
| 702 | } |
| 703 | log.Printf("moves: %s: the return lane of %s is sealed: %d replica objects at TXID %d", host, r.rb, len(seal.Replica), seal.N) |
| 704 | if err := r.fault("after-return-seal"); err != nil { |
| 705 | return false, err |
| 706 | } |
| 707 | return r.phase(ctx, provision.RbTargetReturning, provision.RbTargetReturned, nil, "target-returned") |
| 708 | } |
| 709 | |
| 710 | // journalReturn is the return seal's claim write: the seal and the baseline, |
| 711 | // in one write, on this rollback's op released by this target. |
| 712 | func journalReturn(opID, rbID, evidence, seal, baseline string) func(*hostclaim.Claim) error { |
| 713 | return func(c *hostclaim.Claim) error { |
| 714 | if err := hostclaim.RequireOp(opID)(c); err != nil { |
| 715 | return err |
| 716 | } |
| 717 | op := c.Op |
| 718 | if op.Kind != hostclaim.KindRestoring || op.Intent["rollback"] != rbID || op.Phase != hostclaim.PhaseReleased || op.Exec != "" || |
| 719 | op.Journal[hostclaim.JournalEvidenceTarget] != evidence { |
| 720 | return fmt.Errorf("%w: the return seal needs this rollback released by its target", hostclaim.ErrWrongState) |
| 721 | } |
| 722 | return hostclaim.SetJournal(opID, map[string]string{moves.JournalReturnSeal: seal, moves.JournalReturnBaseline: baseline})(c) |
| 723 | } |
| 724 | } |
| 725 | |
| 726 | // laneMoveJSON writes the lane's move.json create-only (§7.3), naming the |
| 727 | // rollback; one already there must name it and this incarnation. |
| 728 | func (r *rollbackRun) laneMoveJSON(ctx context.Context, gm *guardedMirror) error { |
| 729 | host, m := r.rec.Host, r.rec.Move |
| 730 | body, err := json.Marshal(moves.MoveJSON{MoveID: r.rb, Incarnation: m.Incarnation, Host: host, From: m.To, To: m.From}) |
| 731 | if err != nil { |
| 732 | return err |
| 733 | } |
| 734 | key := moves.MoveJSONKey(host, r.rb) |
| 735 | if _, err := gm.PutIfAbsentVersioned(ctx, gm.dst, key, body); errors.Is(err, lease.ErrExists) { |
| 736 | prev, err := gm.Get(ctx, gm.dst, key) |
| 737 | if err != nil { |
| 738 | return fmt.Errorf("read the lane's move.json: %w", err) |
| 739 | } |
| 740 | var mj moves.MoveJSON |
| 741 | if err := json.Unmarshal(prev, &mj); err != nil || mj.MoveID != r.rb || mj.Incarnation != m.Incarnation { |
| 742 | return fmt.Errorf("the lane's move.json names %q/%q, not rollback %s", mj.MoveID, mj.Incarnation, r.rb) |
| 743 | } |
| 744 | } else if err != nil { |
| 745 | return fmt.Errorf("write the lane's move.json: %w", err) |
| 746 | } |
| 747 | return nil |
| 748 | } |
| 749 | |
| 750 | // copyAsGiven is the return of a target whose state is the staged seal's |
| 751 | // (dirty only for want of src_covered): every sealed version is read from |
| 752 | // this box's bucket and written into the lane under the same key relative to |
| 753 | // the lane, through the guarded staging writer (J22: PutReaderVersionedAt |
| 754 | // under Guarded.Before). The baseline is the source's own, cited under the |
| 755 | // target's evidence. |
| 756 | func (r *rollbackRun) copyAsGiven(ctx context.Context, gm *guardedMirror, staged moves.Seal) (returnedLane, error) { |
| 757 | cfg := r.cfg() |
| 758 | host, id, m := r.rec.Host, r.g.ID, r.rec.Move |
| 759 | if m.Source == nil || m.Source.Files == nil || staged.N != m.Source.FinalTXID { |
| 760 | return returnedLane{}, fmt.Errorf("move %s: the record's source baselines are not the staged seal's", id) |
| 761 | } |
| 762 | vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion) |
| 763 | if err != nil { |
| 764 | return returnedLane{}, err |
| 765 | } |
| 766 | tmp := filepath.Join(moves.JournalDir(cfg.DataDir, host, id), "return") |
| 767 | if err := os.MkdirAll(tmp, 0o700); err != nil { |
| 768 | return returnedLane{}, err |
| 769 | } |
| 770 | defer os.RemoveAll(tmp) |
| 771 | from, lane := moves.StagingPrefix(host, id), moves.ReturnLane(host, r.rb) |
| 772 | copyClass := func(objs []moves.SealedObject) ([]moves.SealedObject, error) { |
| 773 | var out []moves.SealedObject |
| 774 | for _, o := range objs { |
| 775 | rel, ok := strings.CutPrefix(o.Key, from) |
| 776 | if !ok || rel == "" { |
| 777 | return nil, fmt.Errorf("%w: %s is outside the move's lane", moves.ErrBadSeal, o.Key) |
| 778 | } |
| 779 | path := filepath.Join(tmp, "object") |
| 780 | if err := downloadExact(ctx, vs, cfg.OwnBucket, o.Key, o.VersionID, o.Size, path); err != nil { |
| 781 | return nil, fmt.Errorf("read %s: %w", o.Key, err) |
| 782 | } |
| 783 | sum, err := fileSHA256(path) |
| 784 | if err != nil { |
| 785 | return nil, err |
| 786 | } |
| 787 | ver, err := gm.PutFileVersioned(ctx, gm.dst, lane+rel, path, o.Size, digestHex(sum)) |
| 788 | os.Remove(path) |
| 789 | if err != nil { |
| 790 | return nil, fmt.Errorf("return %s: %w", o.Key, err) |
| 791 | } |
| 792 | if !lease.UsableVersionID(ver) { |
| 793 | return nil, fmt.Errorf("the returned %s has no usable version id %q", rel, ver) |
| 794 | } |
| 795 | out = append(out, moves.SealedObject{Key: lane + rel, VersionID: ver, Size: o.Size}) |
| 796 | } |
| 797 | return out, nil |
| 798 | } |
| 799 | var out returnedLane |
| 800 | if out.replica, err = copyClass(staged.Replica); err != nil { |
| 801 | return out, err |
| 802 | } |
| 803 | if out.files, err = copyClass(staged.Files); err != nil { |
| 804 | return out, err |
| 805 | } |
| 806 | if out.timers, err = copyClass(staged.Timers); err != nil { |
| 807 | return out, err |
| 808 | } |
| 809 | files := *m.Source.Files |
| 810 | out.n, out.baseline = staged.N, *m.Source |
| 811 | out.baseline.Files, out.baseline.Evidence = &files, r.rollback().Evidence |
| 812 | return out, nil |
| 813 | } |
| 814 | |
| 815 | // returnState is a dirty return's build state (settings spec rev 11 |
| 816 | // §4.3a), through the return lane's state/ key and sealed by the version |
| 817 | // written there. Once the target's install has built (its landing journal |
| 818 | // records the install), or the landing reached probing (reachedProbing, |
| 819 | // the predicate the return's lane choice uses: a writer may have run, and |
| 820 | // a journal that lost the install step is rebuilt by installManifest, |
| 821 | // which never places the build state again), the state beside the |
| 822 | // target's lease at the holder's location is the instance's: the moved |
| 823 | // state the install placed, with whatever the probe and the target's |
| 824 | // wakes recorded over it. Before both the target's location holds no |
| 825 | // state of this move that differs from the seal's (a return before the |
| 826 | // building step, or before any writer ran over the placed copy), so the |
| 827 | // forward lane's sealed state goes back as given: read at the move's own |
| 828 | // lane key by its exact version and its sealed size, as copyAsGiven reads |
| 829 | // the other classes. Neither holds one: none goes back. |
| 830 | func (r *rollbackRun) returnState(ctx context.Context, gm *guardedMirror, h *hostclaim.Holder, staged moves.Seal) ([]moves.SealedObject, error) { |
| 831 | cfg := r.cfg() |
| 832 | host := r.rec.Host |
| 833 | j, err := r.journal() |
| 834 | if err != nil { |
| 835 | return nil, err |
| 836 | } |
| 837 | if !j.Done(landStepInstall) && !r.reachedProbing() { |
| 838 | if len(staged.State) == 0 { |
| 839 | // A lane object an earlier attempt left is not sealed, so the |
| 840 | // source's install never places it. |
| 841 | return nil, nil |
| 842 | } |
| 843 | o := staged.State[0] |
| 844 | if want := moves.BuildStateKey(host, r.g.ID); o.Key != want { |
| 845 | return nil, fmt.Errorf("%w: the sealed build state %s is not the move's lane key %s", moves.ErrBadSeal, o.Key, want) |
| 846 | } |
| 847 | vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion) |
| 848 | if err != nil { |
| 849 | return nil, err |
| 850 | } |
| 851 | body, err := readExact(ctx, vs, cfg.OwnBucket, o.Key, o.VersionID, o.Size) |
| 852 | if err != nil { |
| 853 | return nil, fmt.Errorf("read the sealed build state %s: %w", o.Key, err) |
| 854 | } |
| 855 | key := moves.BuildStateKey(host, r.rb) |
| 856 | ver, err := gm.PutVersioned(ctx, gm.dst, key, body) |
| 857 | if err != nil { |
| 858 | return nil, fmt.Errorf("return the build state: %w", err) |
| 859 | } |
| 860 | if !lease.UsableVersionID(ver) { |
| 861 | return nil, fmt.Errorf("the returned build state has no usable version id %q", ver) |
| 862 | } |
| 863 | return []moves.SealedObject{{Key: key, VersionID: ver, Size: int64(len(body))}}, nil |
| 864 | } |
| 865 | if h == nil { |
| 866 | return nil, fmt.Errorf("move %s: the claim has no holder to return the build state from", r.g.ID) |
| 867 | } |
| 868 | // This box's request record goes back too (ruling R-SPEC-43A-2): a |
| 869 | // request the target closed as failed stays closed on the source. |
| 870 | if cfg.CarryRequest != nil { |
| 871 | if err := cfg.CarryRequest(ctx, host, cfg.OwnBucket, strings.TrimSuffix(h.Location, "/")); err != nil { |
| 872 | return nil, fmt.Errorf("carry the request record into the returned build state: %w", err) |
| 873 | } |
| 874 | } |
| 875 | _, o, err := mirrorBuildState(ctx, gm, cfg.OwnBucket, h.Location, gm.dst, host, r.rb) |
| 876 | if err != nil { |
| 877 | return nil, fmt.Errorf("return the build state: %w", err) |
| 878 | } |
| 879 | if o == nil { |
| 880 | return nil, nil |
| 881 | } |
| 882 | if !lease.UsableVersionID(o.VersionID) { |
| 883 | return nil, fmt.Errorf("the returned build state has no usable version id %q", o.VersionID) |
| 884 | } |
| 885 | return []moves.SealedObject{*o}, nil |
| 886 | } |
| 887 | |
| 888 | // readExact reads one exact version of size bytes into memory, as |
| 889 | // downloadExact does into a file: a version of any other size refuses. |
| 890 | func readExact(ctx context.Context, vs lease.VersionedStore, bucket, key, version string, size int64) ([]byte, error) { |
| 891 | rc, err := vs.OpenVersion(ctx, bucket, key, version) |
| 892 | if err != nil { |
| 893 | return nil, err |
| 894 | } |
| 895 | defer rc.Close() |
| 896 | body, err := io.ReadAll(io.LimitReader(rc, size+1)) |
| 897 | if err != nil { |
| 898 | return nil, err |
| 899 | } |
| 900 | if int64(len(body)) != size { |
| 901 | return nil, fmt.Errorf("the version is %d bytes, sealed as %d", len(body), size) |
| 902 | } |
| 903 | return body, nil |
| 904 | } |
| 905 | |
| 906 | // downloadExact reads one exact version of size bytes into path. |
| 907 | func downloadExact(ctx context.Context, vs lease.VersionedStore, bucket, key, version string, size int64, path string) error { |
| 908 | rc, err := vs.OpenVersion(ctx, bucket, key, version) |
| 909 | if err != nil { |
| 910 | return err |
| 911 | } |
| 912 | defer rc.Close() |
| 913 | f, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600) |
| 914 | if err != nil { |
| 915 | return err |
| 916 | } |
| 917 | n, err := io.Copy(f, io.LimitReader(rc, size+1)) |
| 918 | if cerr := f.Close(); err == nil { |
| 919 | err = cerr |
| 920 | } |
| 921 | if err != nil { |
| 922 | return err |
| 923 | } |
| 924 | if n != size { |
| 925 | return fmt.Errorf("the version is %d bytes, sealed as %d", n, size) |
| 926 | } |
| 927 | return nil |
| 928 | } |
| 929 | |
| 930 | // returnRefusal is a dirty return that cannot be built from what the target |
| 931 | // holds (fix round 1, ruling 2): its database data is missing, it changed |
| 932 | // after it stopped, or its files cannot be archived. No retry by itself can |
| 933 | // help, so the record stops for an operator with reason. Storage, network |
| 934 | // and local read errors are not refusals: they stay ordinary errors, and the |
| 935 | // next pass retries. |
| 936 | type returnRefusal struct { |
| 937 | reason string |
| 938 | err error |
| 939 | } |
| 940 | |
| 941 | func (e *returnRefusal) Error() string { return e.err.Error() } |
| 942 | func (e *returnRefusal) Unwrap() error { return e.err } |
| 943 | |
| 944 | func refuseReturn(reason string, err error) error { return &returnRefusal{reason: reason, err: err} } |
| 945 | |
| 946 | // archiveRefused reports whether err is one the archive gives for files it |
| 947 | // can never carry (a special file, over the cap, too many entries, a path |
| 948 | // too long), as opposed to a change under the walk or the disk's error, |
| 949 | // which a retry may get past. |
| 950 | func archiveRefused(err error) bool { |
| 951 | return errors.Is(err, archive.ErrSpecialFile) || errors.Is(err, archive.ErrOverCap) || |
| 952 | errors.Is(err, archive.ErrTooManyEntries) || errors.Is(err, archive.ErrBadPath) |
| 953 | } |
| 954 | |
| 955 | // databaseGone reports whether err says the database's data is not there to |
| 956 | // read (absent, short, or not a database), as opposed to a read that failed: |
| 957 | // an *os.PathError other than "does not exist" is the disk's, and transient. |
| 958 | func databaseGone(err error) bool { |
| 959 | if errors.Is(err, os.ErrNotExist) { |
| 960 | return true |
| 961 | } |
| 962 | var pe *os.PathError |
| 963 | return !errors.As(err, &pe) |
| 964 | } |
| 965 | |
| 966 | // stuck stops the record for an operator (MoveMarkStuck) at the target's |
| 967 | // releasing or returning: the phase, the row and everything the target |
| 968 | // holds are kept, and move --retry runs that phase again. |
| 969 | func (r *rollbackRun) stuck(ctx context.Context, refused *returnRefusal) (bool, error) { |
| 970 | host := r.rec.Host |
| 971 | log.Printf("moves: %s: the target's half of %s cannot go on (%v); it stops for an operator: %s", host, r.rb, refused.err, refused.reason) |
| 972 | var pre provision.Record |
| 973 | after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveMarkStuck(r.g, refused.reason))) |
| 974 | if err != nil { |
| 975 | if errors.Is(err, provision.ErrMoveConflict) { |
| 976 | log.Printf("moves: %s: the stuck mark of %s refused (%v); the record moved on", host, r.rb, err) |
| 977 | return false, nil |
| 978 | } |
| 979 | return false, fmt.Errorf("mark the return stuck: %w", err) |
| 980 | } |
| 981 | r.w.logTransition(ctx, r.s, pre, *after) |
| 982 | r.w.pageStuck(ctx, r.rec, "target", refused.reason) |
| 983 | nudgeSerial(ctx, r.s) |
| 984 | return true, nil |
| 985 | } |
| 986 | |
| 987 | // snapshotStopped is the return of a target that served and changed: one |
| 988 | // snapshot [1, N'] of the stopped database (moves.EncodeFull, in memory, the |
| 989 | // Phase 1 size limit), the target's files archive (manifest bound to the move |
| 990 | // id) and its timers. Each must still be what the stopped report recorded: |
| 991 | // data that is gone (StuckReturnTargetDatabase), a state that differs from |
| 992 | // the report (StuckReturnTargetChanged) and files that cannot be archived |
| 993 | // (StuckReturnTargetFiles) are refusals; anything else is retried. |
| 994 | func (r *rollbackRun) snapshotStopped(ctx context.Context, gm *guardedMirror, report moves.StoppedReport) (returnedLane, error) { |
| 995 | cfg := r.cfg() |
| 996 | host, id := r.rec.Host, r.g.ID |
| 997 | dbPath := r.dbPath() |
| 998 | sha, err := fileSHA256(dbPath) |
| 999 | if err != nil { |
| 1000 | err = fmt.Errorf("the target's stopped database: %w", err) |
| 1001 | if errors.Is(err, os.ErrNotExist) { |
| 1002 | err = refuseReturn(provision.StuckReturnTargetDatabase, err) |
| 1003 | } |
| 1004 | return returnedLane{}, err |
| 1005 | } |
| 1006 | if sha != report.DBSHA256 { |
| 1007 | return returnedLane{}, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the target's database changed since its release", id)) |
| 1008 | } |
| 1009 | n, err := cfg.LocalPosition(dbPath) |
| 1010 | if err != nil { |
| 1011 | return returnedLane{}, fmt.Errorf("read the stopped database's position: %w", err) |
| 1012 | } |
| 1013 | if n == 0 { |
| 1014 | return returnedLane{}, refuseReturn(provision.StuckReturnTargetDatabase, fmt.Errorf("move %s: the stopped database has no position", id)) |
| 1015 | } |
| 1016 | if n != report.FinalTXID { |
| 1017 | return returnedLane{}, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the stopped database is at TXID %d, the report at %d", id, n, report.FinalTXID)) |
| 1018 | } |
| 1019 | pageSize, pageCount, err := moves.ReadDBHeader(dbPath) |
| 1020 | if err != nil { |
| 1021 | if databaseGone(err) { |
| 1022 | err = refuseReturn(provision.StuckReturnTargetDatabase, err) |
| 1023 | } |
| 1024 | return returnedLane{}, err |
| 1025 | } |
| 1026 | f, err := os.Open(dbPath) |
| 1027 | if err != nil { |
| 1028 | if errors.Is(err, os.ErrNotExist) { |
| 1029 | err = refuseReturn(provision.StuckReturnTargetDatabase, err) |
| 1030 | } |
| 1031 | return returnedLane{}, err |
| 1032 | } |
| 1033 | data, _, err := moves.EncodeFull(f, pageSize, pageCount, 1, ltx.TXID(n), report.CheckedAt.UnixMilli(), 0) |
| 1034 | f.Close() |
| 1035 | if err != nil { |
| 1036 | err = fmt.Errorf("snapshot the stopped database: %w", err) |
| 1037 | if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { |
| 1038 | // The header names pages the file does not hold. |
| 1039 | err = refuseReturn(provision.StuckReturnTargetDatabase, err) |
| 1040 | } |
| 1041 | return returnedLane{}, err |
| 1042 | } |
| 1043 | var out returnedLane |
| 1044 | key := moves.SnapshotKey(moves.ReplicaPrefix(host, r.rb), n) |
| 1045 | ver, err := gm.PutVersioned(ctx, gm.dst, key, data) |
| 1046 | if err != nil { |
| 1047 | return out, fmt.Errorf("return the snapshot: %w", err) |
| 1048 | } |
| 1049 | if !lease.UsableVersionID(ver) { |
| 1050 | return out, fmt.Errorf("the returned snapshot has no usable version id %q", ver) |
| 1051 | } |
| 1052 | out.replica = []moves.SealedObject{{Key: key, VersionID: ver, Size: int64(len(data))}} |
| 1053 | a, why, retry, err := r.w.buildArchive(r.rec) |
| 1054 | switch { |
| 1055 | case err != nil: |
| 1056 | return out, fmt.Errorf("archive the target's files: %w", err) |
| 1057 | case why != "": |
| 1058 | return out, refuseReturn(provision.StuckReturnTargetFiles, fmt.Errorf("move %s: the target's files cannot be archived (%s)", id, why)) |
| 1059 | case retry: |
| 1060 | return out, errors.New("a stopped file changed while it was archived") |
| 1061 | case a.Plain != report.FilesPlain: |
| 1062 | return out, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the target's files are %s, the report %s", id, a.Plain, report.FilesPlain)) |
| 1063 | } |
| 1064 | ref, err := a.stage(ctx, gm, gm.dst, host, r.rb) |
| 1065 | if err != nil { |
| 1066 | return out, err |
| 1067 | } |
| 1068 | mst, err := os.Stat(a.manifestPath()) |
| 1069 | if err != nil { |
| 1070 | return out, err |
| 1071 | } |
| 1072 | if !lease.UsableVersionID(ref.SealedVersion) || !lease.UsableVersionID(ref.ManifestVersion) { |
| 1073 | return out, fmt.Errorf("the returned archive has no usable version ids (%q, %q)", ref.SealedVersion, ref.ManifestVersion) |
| 1074 | } |
| 1075 | out.files = []moves.SealedObject{{Key: ref.SealedObjectKey, VersionID: ref.SealedVersion, Size: a.SealedSize}, |
| 1076 | {Key: ref.ManifestKey, VersionID: ref.ManifestVersion, Size: mst.Size()}} |
| 1077 | rows, err := cfg.Timers(host) |
| 1078 | if err != nil { |
| 1079 | return out, fmt.Errorf("read the timers: %w", err) |
| 1080 | } |
| 1081 | if moves.TimersDigest(rows) != report.TimersSHA256 { |
| 1082 | return out, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the target's timers changed since its release", id)) |
| 1083 | } |
| 1084 | body, err := moves.TimersSnapshot{V: 1, Host: host, Rows: rows, TakenAt: report.CheckedAt}.Encode() |
| 1085 | if err != nil { |
| 1086 | return out, err |
| 1087 | } |
| 1088 | tkey := moves.TimersPrefix(host, r.rb) + "timers.json" |
| 1089 | tver, err := gm.PutVersioned(ctx, gm.dst, tkey, body) |
| 1090 | if err != nil { |
| 1091 | return out, fmt.Errorf("return the timers: %w", err) |
| 1092 | } |
| 1093 | if !lease.UsableVersionID(tver) { |
| 1094 | return out, fmt.Errorf("the returned timers have no usable version id %q", tver) |
| 1095 | } |
| 1096 | out.timers = []moves.SealedObject{{Key: tkey, VersionID: tver, Size: int64(len(body))}} |
| 1097 | rt, err := cfg.Route(host) |
| 1098 | if err != nil { |
| 1099 | return out, fmt.Errorf("read the registry row: %w", err) |
| 1100 | } |
| 1101 | if rt == nil { |
| 1102 | return out, fmt.Errorf("%s has no registry row on this box", host) |
| 1103 | } |
| 1104 | out.n = n |
| 1105 | out.baseline = provision.MoveSource{Replication: rt.Replication, NextDue: rt.NextDue, FinalTXID: n, DBSHA256: sha, |
| 1106 | Files: &provision.MoveFilesDigest{Plain: a.Plain, Sealed: a.Sealed, Bytes: a.Bytes, Entries: a.Entries}, |
| 1107 | TimersSHA256: report.TimersSHA256, Evidence: r.rollback().Evidence} |
| 1108 | return out, nil |
| 1109 | } |
| 1110 | |
| 1111 | // journal is the target's op journal of the move (the landing's). |
| 1112 | func (r *rollbackRun) journal() (*moves.Journal, error) { |
| 1113 | return moves.ReadJournal(r.cfg().DataDir, r.rec.Host, r.g.ID) |
| 1114 | } |
| 1115 | |
| 1116 | // cleaning (§7.8 target cleaning, INV-10), each step journalled and skipped |
| 1117 | // on a replay once done: the remote parks; the registered installation; the |
| 1118 | // moves-owned rename of everything else at the flat paths; the row; then the |
| 1119 | // record's cleaned with target_parked_at. The source has recovered (the |
| 1120 | // record says so), so nothing here is still needed by the move. |
| 1121 | func (r *rollbackRun) cleaning(ctx context.Context) (bool, error) { |
| 1122 | cfg := r.cfg() |
| 1123 | host, id := r.rec.Host, r.g.ID |
| 1124 | j, err := r.journal() |
| 1125 | if err != nil { |
| 1126 | return false, err |
| 1127 | } |
| 1128 | mark := func(step, value string) error { |
| 1129 | j.Mark(step, value) |
| 1130 | if err := j.Commit(cfg.DataDir); err != nil { |
| 1131 | return err |
| 1132 | } |
| 1133 | return r.fault("after-" + step) |
| 1134 | } |
| 1135 | obs, err := hostclaim.Read(ctx, r.s, host) |
| 1136 | if err != nil { |
| 1137 | return false, fmt.Errorf("read the host claim: %w", err) |
| 1138 | } |
| 1139 | if obs.Absent || obs.Claim.Holder == nil || obs.Claim.Holder.Incarnation != r.g.Incarnation { |
| 1140 | return false, fmt.Errorf("move %s: the host claim names no holder of this incarnation", id) |
| 1141 | } |
| 1142 | ours := r.ours(obs.Claim) == nil |
| 1143 | if ours && !j.Done(rbStepTargetStamp) { |
| 1144 | if err := mark(rbStepTargetStamp, obs.Claim.Op.Journal[hostclaim.JournalTargetStamp]); err != nil { |
| 1145 | return false, err |
| 1146 | } |
| 1147 | } |
| 1148 | if !j.Done(rbStepRemote) { |
| 1149 | // An op that has moved on had its chores acknowledged with it. |
| 1150 | if ours { |
| 1151 | done, err := r.remote(ctx, obs, j, mark) |
| 1152 | if err != nil || !done { |
| 1153 | return false, err |
| 1154 | } |
| 1155 | } |
| 1156 | if err := mark(rbStepRemote, ""); err != nil { |
| 1157 | return false, err |
| 1158 | } |
| 1159 | } |
| 1160 | if !j.Done(rbStepInstallation) { |
| 1161 | how, err := r.installation(ctx, ours) |
| 1162 | if err != nil { |
| 1163 | return false, err |
| 1164 | } |
| 1165 | if err := mark(rbStepInstallation, how); err != nil { |
| 1166 | return false, err |
| 1167 | } |
| 1168 | } |
| 1169 | if err := r.parkLeftovers(ctx, j, mark); err != nil { |
| 1170 | return false, fmt.Errorf("park the target's local artifacts: %w", err) |
| 1171 | } |
| 1172 | if err := r.deleteRow(ctx, j, mark); err != nil { |
| 1173 | return false, err |
| 1174 | } |
| 1175 | ok, err := r.phase(ctx, provision.RbTargetCleaning, provision.RbTargetCleaned, func(rb *provision.MoveRollback) { |
| 1176 | t := r.now().UTC() |
| 1177 | rb.TargetParkedAt = &t |
| 1178 | }, "target-cleaned") |
| 1179 | if !ok || err != nil { |
| 1180 | return ok, err |
| 1181 | } |
| 1182 | // The journal is parked in this same call: once target_parked_at is |
| 1183 | // stamped the source may close the record before this worker reads it |
| 1184 | // again, and then only the local sweep finds the journal. |
| 1185 | if err := mark(rbStepCleaned, "done"); err != nil { |
| 1186 | return true, err |
| 1187 | } |
| 1188 | return true, parkRollbackJournal(cfg.DataDir, host, id, j) |
| 1189 | } |
| 1190 | |
| 1191 | // remote parks the target region's location and the move's lane in it: the |
| 1192 | // lane's park is queued once (its horizon the staged seal's bound), then |
| 1193 | // every park-replica of this op in this region runs through the park engine |
| 1194 | // until its archive is complete (park.Engine.Parked). The acknowledgement, |
| 1195 | // after the horizon, is the engine's or the regional runner's (C4), never a |
| 1196 | // reason to wait here; a chore no longer in the claim was acknowledged. |
| 1197 | func (r *rollbackRun) remote(ctx context.Context, obs hostclaim.Observed, j *moves.Journal, mark func(step, value string) error) (bool, error) { |
| 1198 | cfg := r.cfg() |
| 1199 | host, id := r.rec.Host, r.g.ID |
| 1200 | if !j.Done(rbStepLaneQueued) { |
| 1201 | staged, err := journalledSeal(obs.Claim) |
| 1202 | if err != nil { |
| 1203 | return false, err |
| 1204 | } |
| 1205 | scope := moves.StagingScope(host, id, obs.Claim.Holder.Incarnation, staged.Bound()) |
| 1206 | if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.QueueStagingPark(id, cfg.parkRegion(), scope, obligationID)); err != nil { |
| 1207 | return false, fmt.Errorf("queue the move's lane park: %w", err) |
| 1208 | } |
| 1209 | if err := mark(rbStepLaneQueued, ""); err != nil { |
| 1210 | return false, err |
| 1211 | } |
| 1212 | } |
| 1213 | obs, err := r.claim(ctx) |
| 1214 | if err != nil { |
| 1215 | return false, err |
| 1216 | } |
| 1217 | vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion) |
| 1218 | if err != nil { |
| 1219 | return false, err |
| 1220 | } |
| 1221 | e := r.engine(vs) |
| 1222 | waiting := 0 |
| 1223 | for _, o := range obs.Claim.Obligations { |
| 1224 | if o.Op != id || o.Kind != "park-replica" || !park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) || o.Park == nil { |
| 1225 | continue |
| 1226 | } |
| 1227 | work := park.Work{Host: host, ID: o.ID, Op: id} |
| 1228 | if err := e.Run(ctx, work); err != nil && !errors.Is(err, park.ErrPending) && !errors.Is(err, park.ErrChanged) { |
| 1229 | return false, fmt.Errorf("park %s: %w", o.Park.Location, err) |
| 1230 | } |
| 1231 | parked, err := e.Parked(ctx, work) |
| 1232 | if errors.Is(err, park.ErrChanged) { |
| 1233 | continue |
| 1234 | } |
| 1235 | if err != nil { |
| 1236 | return false, fmt.Errorf("the park of %s: %w", o.Park.Location, err) |
| 1237 | } |
| 1238 | if !parked { |
| 1239 | waiting++ |
| 1240 | } |
| 1241 | } |
| 1242 | if waiting > 0 { |
| 1243 | r.w.notice(host+"#rb-remote", "moves: %s: the target's cleaning of %s waits for %d park archive(s)", host, r.rb, waiting) |
| 1244 | return false, nil |
| 1245 | } |
| 1246 | log.Printf("moves: %s: the target region's copies of %s are parked", host, id) |
| 1247 | return true, nil |
| 1248 | } |
| 1249 | |
| 1250 | // engine is the park engine over this box's own bucket with the claim's |
| 1251 | // leases and view, and the claim's authorizer behind this rollback's own |
| 1252 | // check (rollbackTargetArmed requires all three; nothing stands in). |
| 1253 | func (r *rollbackRun) engine(vs lease.VersionedStore) park.Engine { |
| 1254 | cfg := r.cfg() |
| 1255 | own := func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error { |
| 1256 | if obs.Absent { |
| 1257 | return park.ErrUnsafe |
| 1258 | } |
| 1259 | return r.authorizeChore(obs.Claim, o) |
| 1260 | } |
| 1261 | return park.Engine{HomeRegion: cfg.ParkHomeRegion, Store: vs, Claims: r.s, Bucket: cfg.OwnBucket, Region: cfg.parkRegion(), Leases: cfg.ParkLeases, |
| 1262 | Now: r.now, NewID: obligationID, View: cfg.ParkView, Authorize: movesArmAuthorize(cfg.ParkAuthorize, own), After: cfg.parkFault} |
| 1263 | } |
| 1264 | |
| 1265 | // authorizeChore is this rollback's own operation check (M17): a chore is |
| 1266 | // run only when it is this op's, the claim is still this rollback's with this |
| 1267 | // box no longer executing, and the chore is this region's or this box's. The |
| 1268 | // engine runs it before the claim's authorizer, and park.Local before the |
| 1269 | // claim's local authorizer (installation, movesLocalAuthorize). |
| 1270 | func (r *rollbackRun) authorizeChore(c *hostclaim.Claim, o hostclaim.Obligation) error { |
| 1271 | cfg := r.cfg() |
| 1272 | if r.ours(c) != nil || o.Op != r.g.ID || c.Op.Exec == cfg.Box { |
| 1273 | return fmt.Errorf("%w: not this rollback's chore", park.ErrUnsafe) |
| 1274 | } |
| 1275 | switch { |
| 1276 | case o.Kind == "park-replica" && park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region): |
| 1277 | case o.Kind == "park-local" && o.Box == cfg.Box: |
| 1278 | default: |
| 1279 | return fmt.Errorf("%w: not this target's chore", park.ErrUnsafe) |
| 1280 | } |
| 1281 | return nil |
| 1282 | } |
| 1283 | |
| 1284 | // targetStamp is F: the claim's journalled target stamp, as noted in this |
| 1285 | // box's journal when cleaning began. |
| 1286 | func targetStamp(j *moves.Journal) uint64 { |
| 1287 | f, err := strconv.ParseUint(j.Steps[rbStepTargetStamp], 10, 64) |
| 1288 | if err != nil { |
| 1289 | return 0 |
| 1290 | } |
| 1291 | return f |
| 1292 | } |
| 1293 | |
| 1294 | // guard is the host guard over this box's data directory, reading the claim |
| 1295 | // from the pass's store. |
| 1296 | func (r *rollbackRun) guard() hostguard.Guard { |
| 1297 | g := r.cfg().Guard |
| 1298 | if g.Claims == nil { |
| 1299 | g.Claims = hostguard.StoreReader{Store: r.s} |
| 1300 | } |
| 1301 | return g |
| 1302 | } |
| 1303 | |
| 1304 | // installation is the registered half of the local retirement (J15): with the |
| 1305 | // target's park-local and inventory paths of F, the installation is retired |
| 1306 | // through park.Local under the host guard, which completes the chore itself; |
| 1307 | // a park-local with nothing of F listed is completed; neither is an |
| 1308 | // unregistered target, whose artifacts are all the moves-owned rename's. |
| 1309 | func (r *rollbackRun) installation(ctx context.Context, ours bool) (string, error) { |
| 1310 | cfg := r.cfg() |
| 1311 | host, id := r.rec.Host, r.g.ID |
| 1312 | if !ours { |
| 1313 | return "moved-on", nil |
| 1314 | } |
| 1315 | obs, err := r.claim(ctx) |
| 1316 | if err != nil { |
| 1317 | return "", err |
| 1318 | } |
| 1319 | c := obs.Claim |
| 1320 | f, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalTargetStamp], 10, 64) |
| 1321 | if err != nil || f == 0 { |
| 1322 | return "", fmt.Errorf("move %s: the claim journals no target stamp", id) |
| 1323 | } |
| 1324 | var chore *hostclaim.Obligation |
| 1325 | for i := range c.Obligations { |
| 1326 | if o := &c.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box { |
| 1327 | chore = o |
| 1328 | } |
| 1329 | } |
| 1330 | registered := hostclaim.InventoryHasStamp(c, cfg.Box, f) |
| 1331 | switch { |
| 1332 | case chore != nil && registered: |
| 1333 | captured, err := hostguard.Capture(cfg.DataDir, host) |
| 1334 | if err != nil { |
| 1335 | return "", fmt.Errorf("the target's installation record: %w", err) |
| 1336 | } |
| 1337 | if captured.Incarnation != c.Holder.Incarnation || (captured.Stamp != 0 && captured.Stamp != f) { |
| 1338 | return "", fmt.Errorf("move %s: the target's installation record is at %s/%d, the claim's is %s/%d", id, captured.Incarnation, captured.Stamp, c.Holder.Incarnation, f) |
| 1339 | } |
| 1340 | local := cfg.Local |
| 1341 | local.Guard, local.Claims = r.guard(), r.s |
| 1342 | if local.Box == "" { |
| 1343 | local.Box = cfg.Box |
| 1344 | } |
| 1345 | if local.Authorize == nil { |
| 1346 | // park.Local's authorizer: the claim's MoveParks local authorizer |
| 1347 | // (ParkLocalAuthorize) behind this rollback's own check. The |
| 1348 | // claim's engine authorizer refuses a park-local (M9), so it is |
| 1349 | // never asked here. Without it the chore refuses |
| 1350 | // (rollbackTargetArmed requires it). |
| 1351 | local.Authorize = movesLocalAuthorize(cfg.ParkLocalAuthorize, func(_ context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error { |
| 1352 | return r.authorizeChore(c, o) |
| 1353 | }) |
| 1354 | } |
| 1355 | if err := r.fault("before-rb-local-retire"); err != nil { |
| 1356 | return "", err |
| 1357 | } |
| 1358 | if err := local.RetireInstallation(ctx, host, hostguard.Captured{Incarnation: captured.Incarnation, Stamp: f}, chore.ID, id); err != nil { |
| 1359 | return "", fmt.Errorf("retire the target's installation: %w", err) |
| 1360 | } |
| 1361 | if err := r.fault("after-rb-local-retire"); err != nil { |
| 1362 | return "", err |
| 1363 | } |
| 1364 | log.Printf("moves: %s: the target's registered installation (stamp %d) is retired", host, f) |
| 1365 | return "retired", nil |
| 1366 | case chore != nil: |
| 1367 | if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.CompleteLocalPark(id, cfg.Box, f)); err != nil { |
| 1368 | return "", fmt.Errorf("complete the target's local park: %w", err) |
| 1369 | } |
| 1370 | return "completed", nil |
| 1371 | case registered: |
| 1372 | return "", fmt.Errorf("move %s: this box's inventory lists the target's paths, and the release queued no park-local for them", id) |
| 1373 | } |
| 1374 | return "unregistered", nil |
| 1375 | } |
| 1376 | |
| 1377 | // parkLeftovers is the moves-owned rename (H5(b), parkLocalArtifacts) of |
| 1378 | // the artifacts this op created that the target still has at the flat paths, |
| 1379 | // its plan and its renames inside the host guard, which retires the |
| 1380 | // installation record at F if it is still active (an unregistered target), |
| 1381 | // with the claim read fresh there: this box no longer executes and no longer |
| 1382 | // takes part in the claim's op, and no inventory entry claims a path the |
| 1383 | // rename would move. |
| 1384 | // |
| 1385 | // The candidates are the installation's paths, the rendered config and the |
| 1386 | // sidecar's env (sourceArtifactPaths), but a candidate is moved only on a |
| 1387 | // record that this op created it (rollbackOwnership; the final review's |
| 1388 | // finding 1). Anything else, a files directory or env the landing refused as |
| 1389 | // occupied (RollbackFilesOccupied) included, is logged and left where it is. |
| 1390 | func (r *rollbackRun) parkLeftovers(ctx context.Context, j *moves.Journal, mark func(step, value string) error) error { |
| 1391 | if j.Done(rbStepLocal) { |
| 1392 | return nil |
| 1393 | } |
| 1394 | cfg := r.cfg() |
| 1395 | host, id := r.rec.Host, r.g.ID |
| 1396 | paths := sourceArtifactPaths(host) |
| 1397 | var owns func(path, identity string) (bool, error) |
| 1398 | if _, planned := j.Steps[rbStepLocal+"-plan"]; !planned { |
| 1399 | // A journalled plan already recorded what is the op's; the evidence |
| 1400 | // is read (the config ack is a store read) only before the plan. |
| 1401 | ev, err := r.ownership(ctx, j) |
| 1402 | if err != nil { |
| 1403 | return err |
| 1404 | } |
| 1405 | owns = ev.owns |
| 1406 | } |
| 1407 | var incarnation string |
| 1408 | check := func(c *hostclaim.Claim) error { |
| 1409 | if c.Op.Exec == cfg.Box { |
| 1410 | return fmt.Errorf("%w: this box executes the host again", hostclaim.ErrWrongState) |
| 1411 | } |
| 1412 | if c.Op.ID == id { |
| 1413 | if err := r.ours(c); err != nil { |
| 1414 | return err |
| 1415 | } |
| 1416 | } else if containsString(c.Op.Participants, cfg.Box) { |
| 1417 | return fmt.Errorf("%w: this box takes part in the claim's op %s", hostclaim.ErrWrongState, c.Op.ID) |
| 1418 | } |
| 1419 | for _, p := range paths { |
| 1420 | if _, ok := c.Inventory[cfg.Box].Paths[p]; ok { |
| 1421 | return fmt.Errorf("%w: the inventory claims %s", hostclaim.ErrWrongState, p) |
| 1422 | } |
| 1423 | } |
| 1424 | incarnation = c.Holder.Incarnation |
| 1425 | return nil |
| 1426 | } |
| 1427 | work := func() error { |
| 1428 | if incarnation != r.g.Incarnation { |
| 1429 | return fmt.Errorf("move %s: the claim names another incarnation", id) |
| 1430 | } |
| 1431 | if _, rendered := j.Steps[landStepPreRender]; rendered { |
| 1432 | if err := removeRenderTemp(cfg.DataDir, host); err != nil { |
| 1433 | return err |
| 1434 | } |
| 1435 | } |
| 1436 | _, err := parkLocalArtifacts(ctx, j, mark, localArtifactPark{DataDir: cfg.DataDir, Incarnation: incarnation, Host: host, |
| 1437 | Op: id, Step: rbStepLocal, Paths: paths, NewID: obligationID, Fault: r.fault, Owns: owns}) |
| 1438 | return err |
| 1439 | } |
| 1440 | g := r.guard() |
| 1441 | captured, err := hostguard.Capture(cfg.DataDir, host) |
| 1442 | switch { |
| 1443 | case errors.Is(err, hostguard.ErrNoRecord): |
| 1444 | return g.WithNoInstallation(ctx, host, r.g.Incarnation, check, work) |
| 1445 | case err != nil: |
| 1446 | return fmt.Errorf("the target's installation record: %w", err) |
| 1447 | } |
| 1448 | f := captured.Stamp |
| 1449 | if f == 0 { |
| 1450 | f = targetStamp(j) |
| 1451 | } |
| 1452 | if f == 0 { |
| 1453 | return fmt.Errorf("move %s: the target's retired installation stamp is not known", id) |
| 1454 | } |
| 1455 | return g.RetireInstallationWith(ctx, host, hostguard.Captured{Incarnation: r.g.Incarnation, Stamp: f}, check, work) |
| 1456 | } |
| 1457 | |
| 1458 | // removeRenderTemp removes the held renderer's temp sibling of |
| 1459 | // <host>.config.env (Task 9a fix round 5). The flat renderer writes the |
| 1460 | // rendered bytes to <path>.tmp and renames them over the path (appconfig |
| 1461 | // renderExecConfigBefore), so a crash between the two leaves the temp |
| 1462 | // holding the rendered secrets: never live config, only an unfinished write |
| 1463 | // of this op's, whose render the journal records (pre_render). It is |
| 1464 | // removed, never parked. Only a regular file is removed (Lstat'ed, never |
| 1465 | // followed); anything else at the name is logged and left. A moved row is |
| 1466 | // always flat (the landing refuses a nested row at checking), so the nested |
| 1467 | // renderer's temp (instancefs.WriteAtomic) never holds a held render. |
| 1468 | func removeRenderTemp(dataDir, host string) error { |
| 1469 | tmp := filepath.Join(dataDir, host+".config.env.tmp") |
| 1470 | fi, err := os.Lstat(tmp) |
| 1471 | switch { |
| 1472 | case errors.Is(err, os.ErrNotExist): |
| 1473 | return nil |
| 1474 | case err != nil: |
| 1475 | return err |
| 1476 | case !fi.Mode().IsRegular(): |
| 1477 | log.Printf("moves: %s: %s is not a regular file (%s); it is not the renderer's temp, and is left", host, tmp, fi.Mode()) |
| 1478 | return nil |
| 1479 | } |
| 1480 | if err := os.Remove(tmp); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 1481 | return fmt.Errorf("remove the held render's temp: %w", err) |
| 1482 | } |
| 1483 | log.Printf("moves: %s: the held render's unfinished write %s is removed", host, tmp) |
| 1484 | return syncPath(dataDir) |
| 1485 | } |
| 1486 | |
| 1487 | // rollbackOwnership is the record that this op created a path the target's |
| 1488 | // cleaning may park (the final review's finding 1). The registered |
| 1489 | // installation, the paths the landing listed under F, is park.Local's and is |
| 1490 | // retired before the moves-owned rename runs; what is left at the flat paths |
| 1491 | // is parked only on one of these: |
| 1492 | // |
| 1493 | // - the files directory and <host>.env: the landing's placement receipt |
| 1494 | // (files_placed), or before it the staged extraction's identities |
| 1495 | // (staged), which a rename keeps: the live path must have that identity; |
| 1496 | // - the database, its marker, WAL and shared memory: the landing installed |
| 1497 | // its database, as the record says (landing.installed_db_sha256, stamped |
| 1498 | // by the CAS that ends configured); the install refused a WAL or shared |
| 1499 | // memory beside an absent database, so what sits beside it since is the |
| 1500 | // op's instance's. Before that stamp, the database and its marker are |
| 1501 | // the op's only when they are the build itself: the database's sha256 |
| 1502 | // is the journalled install manifest's, and the marker names the op's |
| 1503 | // generation (a crash between the install and its stamp; carry (b)); |
| 1504 | // - the quarantine and the litestream shadow, which the install does not |
| 1505 | // refuse: the database installed as above, and the install's journal |
| 1506 | // records the path absent just before it put the database in place |
| 1507 | // (pre_install, notePreInstall). Present then, or no record, is never |
| 1508 | // the op's (the final re-review's residual); |
| 1509 | // - the rendered config, <host>.config.env: the file's sha256 is one this |
| 1510 | // op rendered, as the held render's ack on this box records it (the ack |
| 1511 | // names this op), or as the landing's journal records it: pre_render's |
| 1512 | // intended hash, journalled before the render wrote, or rendered, |
| 1513 | // journalled as soon as the render answered (carry (c); fix round 1, |
| 1514 | // I1). Any other content is not the op's; |
| 1515 | // - the sidecar's env: never, since a moved host has no sidecar. |
| 1516 | type rollbackOwnership struct { |
| 1517 | host, dataDir string |
| 1518 | placed stagedFiles // the landing's receipt, or its staged identities |
| 1519 | installed bool |
| 1520 | builtSHA256 string // the journalled install manifest's database, "sha256:<hex>", or "" |
| 1521 | generation string // the generation the install names (the op id) |
| 1522 | absentBefore map[string]bool // the siblings pre_install recorded absent |
| 1523 | configSHA256 []string // what this op rendered, "sha256:<hex>": the ack's and the journal's |
| 1524 | } |
| 1525 | |
| 1526 | // ownership reads the evidence: the landing's journal, the record and the |
| 1527 | // held render's config ack on this box. |
| 1528 | func (r *rollbackRun) ownership(ctx context.Context, j *moves.Journal) (rollbackOwnership, error) { |
| 1529 | cfg := r.cfg() |
| 1530 | host, id := r.rec.Host, r.g.ID |
| 1531 | ev := rollbackOwnership{host: host, dataDir: cfg.DataDir} |
| 1532 | for _, step := range []string{landStepFilesPlaced, landStepStaged} { |
| 1533 | if _, ok := j.Steps[step]; !ok { |
| 1534 | continue |
| 1535 | } |
| 1536 | placed, err := journalledStaged(j, step) |
| 1537 | if err != nil { |
| 1538 | return ev, err |
| 1539 | } |
| 1540 | ev.placed = placed |
| 1541 | break |
| 1542 | } |
| 1543 | if l := r.rec.Move.Landing; l != nil && l.InstalledDBSHA256 != "" { |
| 1544 | ev.installed = true |
| 1545 | } |
| 1546 | ev.generation = id |
| 1547 | if raw, ok := j.Steps[landStepInstall]; ok { |
| 1548 | var m moves.InstallManifest |
| 1549 | if err := json.Unmarshal([]byte(raw), &m); err == nil && m.Op == id && m.Host == host && m.DBSHA256 != "" { |
| 1550 | ev.builtSHA256 = "sha256:" + m.DBSHA256 |
| 1551 | } |
| 1552 | } |
| 1553 | if rendered := j.Steps[landStepRendered]; strings.HasPrefix(rendered, "sha256:") { |
| 1554 | ev.configSHA256 = append(ev.configSHA256, rendered) |
| 1555 | } |
| 1556 | if raw, ok := j.Steps[landStepPreRender]; ok { |
| 1557 | // The hash the render journalled before it wrote (fix round 1, I1): |
| 1558 | // a crash after the write and before either the ack or the |
| 1559 | // rendered record still leaves the file the op's by content. |
| 1560 | var pre preRender |
| 1561 | if err := json.Unmarshal([]byte(raw), &pre); err != nil { |
| 1562 | log.Printf("moves: %s: the journal's %s step is unreadable (%v); it vouches for no rendered config", host, landStepPreRender, err) |
| 1563 | } else { |
| 1564 | for _, h := range pre.Intended { |
| 1565 | if strings.HasPrefix(h, "sha256:") { |
| 1566 | ev.configSHA256 = append(ev.configSHA256, h) |
| 1567 | } |
| 1568 | } |
| 1569 | } |
| 1570 | } |
| 1571 | if raw, ok := j.Steps[landStepPreInstall]; ok { |
| 1572 | var pre preInstall |
| 1573 | if err := json.Unmarshal([]byte(raw), &pre); err != nil { |
| 1574 | log.Printf("moves: %s: the journal's %s step is unreadable (%v); the database's siblings are not counted as %s's", host, landStepPreInstall, err, id) |
| 1575 | } else { |
| 1576 | ev.absentBefore = map[string]bool{} |
| 1577 | for _, p := range pre.Absent { |
| 1578 | ev.absentBefore[p] = true |
| 1579 | } |
| 1580 | } |
| 1581 | } |
| 1582 | body, _, err := r.s.Get(ctx, appconfig.ConfigAckKey(cfg.Box, host)) |
| 1583 | switch { |
| 1584 | case errors.Is(err, os.ErrNotExist): |
| 1585 | case err != nil: |
| 1586 | return ev, fmt.Errorf("read the held render's config ack: %w", err) |
| 1587 | default: |
| 1588 | var a appconfig.ConfigAck |
| 1589 | if err := json.Unmarshal(body, &a); err != nil { |
| 1590 | log.Printf("moves: %s: the config ack is unreadable (%v); the rendered config is not counted as %s's", host, err, id) |
| 1591 | } else if a.MoveOp == id && a.Box == cfg.Box && a.Host == host && a.FileSHA256 != "" { |
| 1592 | ev.configSHA256 = append(ev.configSHA256, "sha256:"+a.FileSHA256) |
| 1593 | } |
| 1594 | } |
| 1595 | return ev, nil |
| 1596 | } |
| 1597 | |
| 1598 | // owns answers localArtifactPark.Owns for the target's leftovers. |
| 1599 | func (ev rollbackOwnership) owns(path, identity string) (bool, error) { |
| 1600 | db := ev.host + ".db" |
| 1601 | switch path { |
| 1602 | case ev.host: |
| 1603 | return ev.placed.Files != "" && identity == ev.placed.Files, nil |
| 1604 | case ev.host + ".env": |
| 1605 | return ev.placed.Env != "" && identity == ev.placed.Env, nil |
| 1606 | case ev.host + ".config.env": |
| 1607 | if len(ev.configSHA256) == 0 { |
| 1608 | return false, nil |
| 1609 | } |
| 1610 | full := filepath.Join(ev.dataDir, path) |
| 1611 | if fi, err := os.Lstat(full); err != nil || !fi.Mode().IsRegular() { |
| 1612 | return false, nil // a link or anything else is not the render's file |
| 1613 | } |
| 1614 | got, err := fileSHA256(full) |
| 1615 | if err != nil { |
| 1616 | return false, err |
| 1617 | } |
| 1618 | return containsString(ev.configSHA256, got), nil |
| 1619 | case db, db + ".gen": |
| 1620 | if ev.installed { |
| 1621 | return true, nil |
| 1622 | } |
| 1623 | return ev.isBuild(path) |
| 1624 | case db + "-wal", db + "-shm": |
| 1625 | return ev.installed, nil |
| 1626 | case db + ".corrupt", ".activator-" + db + "-litestream": |
| 1627 | return ev.installed && ev.absentBefore[path], nil |
| 1628 | } |
| 1629 | return false, nil |
| 1630 | } |
| 1631 | |
| 1632 | // isBuild reports whether path, the database or its marker at the flat |
| 1633 | // paths, is this op's build before the record stamped it installed: the |
| 1634 | // database's sha256 is the journalled build's, and the marker names the |
| 1635 | // op's generation. A link or anything else is not. |
| 1636 | func (ev rollbackOwnership) isBuild(path string) (bool, error) { |
| 1637 | if ev.builtSHA256 == "" { |
| 1638 | return false, nil |
| 1639 | } |
| 1640 | full := filepath.Join(ev.dataDir, path) |
| 1641 | if fi, err := os.Lstat(full); err != nil || !fi.Mode().IsRegular() { |
| 1642 | return false, nil |
| 1643 | } |
| 1644 | if strings.HasSuffix(path, ".gen") { |
| 1645 | gen, err := hostbox.ReadGenerationMarker(full) |
| 1646 | if err != nil { |
| 1647 | return false, nil |
| 1648 | } |
| 1649 | return gen == ev.generation, nil |
| 1650 | } |
| 1651 | got, err := fileSHA256(full) |
| 1652 | if err != nil { |
| 1653 | return false, err |
| 1654 | } |
| 1655 | return got == ev.builtSHA256, nil |
| 1656 | } |
| 1657 | |
| 1658 | // deleteRow deletes the row and its timers rows in one registry transaction |
| 1659 | // (registry.DeleteIfHoldWithTimers, J12), only while the rollback holds the |
| 1660 | // row, under the host's hold lock with the record re-read there; the journal |
| 1661 | // says "pending" before and "done" after, and a replay that finds "pending" |
| 1662 | // and no row of this rollback's treats the delete as done. |
| 1663 | func (r *rollbackRun) deleteRow(ctx context.Context, j *moves.Journal, mark func(step, value string) error) error { |
| 1664 | if j.Steps[rbStepRowDeleted] == "done" { |
| 1665 | return nil |
| 1666 | } |
| 1667 | cfg := r.cfg() |
| 1668 | host, id := r.rec.Host, r.g.ID |
| 1669 | var err error |
| 1670 | provision.WithHostHoldLock(host, func() { |
| 1671 | fresh, e := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, host) |
| 1672 | if e != nil { |
| 1673 | err = fmt.Errorf("re-read the record: %w", e) |
| 1674 | return |
| 1675 | } |
| 1676 | if m := fresh.Move; m == nil || m.ID != id || m.Incarnation != r.g.Incarnation || m.State != provision.MoveRollingBack || |
| 1677 | m.Rollback == nil || m.Rollback.ID != r.rb || m.Rollback.TargetPhase != provision.RbTargetCleaning { |
| 1678 | err = fmt.Errorf("move %s: the record is no longer this rollback's cleaning", id) |
| 1679 | return |
| 1680 | } |
| 1681 | if j.Steps[rbStepRowDeleted] != "pending" { |
| 1682 | j.Mark(rbStepRowDeleted, "pending") |
| 1683 | if err = j.Commit(cfg.DataDir); err != nil { |
| 1684 | return |
| 1685 | } |
| 1686 | if err = r.fault("after-rb-row-pending"); err != nil { |
| 1687 | return |
| 1688 | } |
| 1689 | } |
| 1690 | deleted, e := cfg.DeleteHeld(host, r.rb) |
| 1691 | if e != nil { |
| 1692 | err = fmt.Errorf("delete the target's row: %w", e) |
| 1693 | return |
| 1694 | } |
| 1695 | if deleted { |
| 1696 | log.Printf("moves: %s: the target's row and its timers are deleted", host) |
| 1697 | } |
| 1698 | if err = r.fault("after-rb-row-delete"); err != nil { |
| 1699 | return |
| 1700 | } |
| 1701 | err = mark(rbStepRowDeleted, "done") |
| 1702 | }) |
| 1703 | return err |
| 1704 | } |
| 1705 | |
| 1706 | // cleaned parks the op's journal into the local retirement once the record |
| 1707 | // says cleaned, if a crash or a quick close left it. |
| 1708 | func (r *rollbackRun) cleaned() error { |
| 1709 | cfg := r.cfg() |
| 1710 | host, id := r.rec.Host, r.g.ID |
| 1711 | if !moveJournalPresent(cfg.DataDir, host, id) { |
| 1712 | return nil |
| 1713 | } |
| 1714 | j, err := r.journal() |
| 1715 | if err != nil { |
| 1716 | return err |
| 1717 | } |
| 1718 | if j.Steps[rbStepCleaned] != "done" { |
| 1719 | j.Mark(rbStepCleaned, "done") |
| 1720 | if err := j.Commit(cfg.DataDir); err != nil { |
| 1721 | return err |
| 1722 | } |
| 1723 | } |
| 1724 | return parkRollbackJournal(cfg.DataDir, host, id, j) |
| 1725 | } |
| 1726 | |
| 1727 | // parkRollbackJournal parks the target's op journal (its steps, the landing's |
| 1728 | // build and extract scratch, and the built files archive of a dirty return) |
| 1729 | // into the retirement its own moves-owned rename recorded. |
| 1730 | func parkRollbackJournal(dataDir, host, op string, j *moves.Journal) error { |
| 1731 | retirement := j.Steps[rbStepLocal] |
| 1732 | if retirement == "" { |
| 1733 | return fmt.Errorf("move %s: the journal records no local retirement for the target's park", op) |
| 1734 | } |
| 1735 | plan, err := journalledLocalPlan(j, rbStepLocal) |
| 1736 | if err != nil { |
| 1737 | return err |
| 1738 | } |
| 1739 | if err := parkOpJournal(dataDir, plan.Incarnation, host, op, retirement); err != nil { |
| 1740 | return fmt.Errorf("park the op journal: %w", err) |
| 1741 | } |
| 1742 | log.Printf("moves: %s: the target's half of the rollback of %s is complete", host, op) |
| 1743 | return nil |
| 1744 | } |
| 1745 | |
| 1746 | // sweepRollbackJournal is the local journal sweep's rollback half |
| 1747 | // (sweepFinalJournals, fix round 1, ruling 3): a target's journal whose |
| 1748 | // cleaning reached its moves-owned rename (rbStepLocal) is parked once its |
| 1749 | // completion is shown, independent of any record that still routes here: |
| 1750 | // - the journal records cleaned; |
| 1751 | // - a listed record carries the op and its rollback shows target_parked_at |
| 1752 | // (the cleaned CAS landed); |
| 1753 | // - no listed record carries the op: the only close of a rolling_back move, |
| 1754 | // provision.MoveRolledBackClose, refuses unless target_parked_at is |
| 1755 | // stamped, and a journal at rbStepLocal is past every declared-loss |
| 1756 | // window, so the cleaned CAS landed before the close. |
| 1757 | // |
| 1758 | // A listed record that carries the op without target_parked_at is the |
| 1759 | // cleaning's own pass, and a host the listing does not name is left. A journal |
| 1760 | // that never reached rbStepLocal is never touched. |
| 1761 | func (w *movesWorker) sweepRollbackJournal(dataDir, host, op string, j *moves.Journal, recs []provision.Record) error { |
| 1762 | if _, ok := j.Steps[rbStepLocal]; !ok { |
| 1763 | return nil |
| 1764 | } |
| 1765 | if j.Steps[rbStepCleaned] != "done" { |
| 1766 | if len(recs) == 0 { |
| 1767 | w.notice(host+"#rb-sweep-"+op, "moves: %s: the target's journal of %s reached its park, and no listed record names the host; it is left", host, op) |
| 1768 | return nil |
| 1769 | } |
| 1770 | for _, rec := range recs { |
| 1771 | if m := rec.Move; m != nil && m.ID == op && (m.Rollback == nil || m.Rollback.TargetParkedAt == nil) { |
| 1772 | return nil |
| 1773 | } |
| 1774 | } |
| 1775 | j.Mark(rbStepCleaned, "done") |
| 1776 | if err := j.Commit(dataDir); err != nil { |
| 1777 | return err |
| 1778 | } |
| 1779 | } |
| 1780 | return parkRollbackJournal(dataDir, host, op, j) |
| 1781 | } |
| 1782 | |
amadan