| new file mode 100644 |
| index 000000000..a374a4b2f |
| --- /dev/null |
| +++ b/internal/host/movesrollback.go |
| @@ -0,0 +1,1277 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "crypto/rand" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "log" |
| + "os" |
| + "path/filepath" |
| + "strconv" |
| + "strings" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/hostguard" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/park" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/superfly/ltx" |
| +) |
| + |
| +// The rollback on the target (moves spec §7.8; Plan 4 Task 4), behind |
| +// CARLOS_MOVES_PHASE1: the decision that ends a landing, then the target |
| +// box's half of the rollback: its release, its return lane and its cleaning. |
| +// |
| +// The decision (rollBackLanding). A landing the target can no longer finish |
| +// is rolled back by one record CAS, MoveRollBack, which moves the region back |
| +// to From; the claim follows (RollBack, or RollBackFlipped over a claim the |
| +// flip's half never reached, J16), journalling the target installation's |
| +// stamp F (J15). It runs at the top of the landing's pass, before its arming |
| +// gate (J10), and is the one place a landing's recorded mark (every |
| +// markRollback site, Plans 2 and 3) becomes a rollback. The deadline that |
| +// triggers it is landingDeadlineOf, the earlier of the cut's and the |
| +// landing's, at every landing phase, probing and probed included (J11). The |
| +// commit races it on the record: a commit that won makes MoveRollBack refuse |
| +// with ErrMoveConflict, and the decision stops there with no trace, leaving |
| +// the record to the commit and the open (Plan 3 Task 7). |
| +// |
| +// The target's half (rollbackTarget). Once the record is rolling_back its |
| +// region is the source's, so the target reaches it through the post-flip |
| +// participant rule (movesParticipantRecord, J10). One pass drives it as far |
| +// as it can, each phase from durable state, under the record's |
| +// target_phase (§7.8 target table; Plan 2's phase-name convention, G8: a |
| +// record at X means X's work may have begun): |
| +// |
| +// - releasing: the row's hold handed from the move to the rollback |
| +// (handHold, J3), the wake fence, the timer drain and the stop evidence |
| +// (stopevidence.Reconstruct); a target not yet proven stopped retries, |
| +// and the release deadline is the console's to page on. Then, if the |
| +// landing reached probing and the database is here, a final checkpoint |
| +// under the writer's lease (finalCheckpoint, writerLease), N' from the |
| +// local position, and the digests: the database, the files plain digest |
| +// and the timers. The release is clean when the landing never reached |
| +// probing or all three equal what was installed (installed_db_sha256, |
| +// source.files.plain, source.timers_sha256), and the staged seal covers |
| +// the source (src_covered). stopped-target.json is written create-only, |
| +// the record goes to released with Dirty and Evidence, and the claim |
| +// follows (TargetReleased: execution to none, the target region's |
| +// location park, its park-local only when the inventory lists F, the |
| +// target region's predecessors covered); |
| +// - returning (dirty only): the return lane, moves/<host>/<rb id>/ in the |
| +// source's bucket, written as a staging lane under the staging-work lease |
| +// for (source region, host), every object through the guarded staging |
| +// writer, one presigned request per call. A target that served and |
| +// changed sends a snapshot [1, N'] of its stopped database |
| +// (moves.EncodeFull, held in memory: Phase 1's size limit), its files |
| +// archive (bound to the move id, which the source's installer checks) |
| +// and its timers; one whose state is still the staged seal's (dirty only |
| +// for want of src_covered) sends the staged seal's versions as given. |
| +// The return seal (moves.ReturnSeal, bound = the lease's SafeUntil after |
| +// the last write) and the return baseline go into the claim's journal in |
| +// one write, then the record says returned; |
| +// - cleaning, only once the source has recovered (RecoveredAt, Review |
| +// Focus 4): the target region's location park and the move's lane park |
| +// (queued here) run through the park engine until their archives are |
| +// complete (D6: this region is no longer authoritative, so the engine |
| +// selects everything); the registered installation is retired through |
| +// park.Local, which completes its own park-local (J15), and everything |
| +// else at the flat paths is parked by the moves-owned rename (H5(b)) |
| +// inside the host guard; the row and its timers rows go in one registry |
| +// transaction under the host's hold lock (J12); the record says cleaned |
| +// with target_parked_at, and the op's journal is parked last. |
| +// |
| +// Arming (ruling J18). Like the landing, the rollback's target half is |
| +// unarmed in production until Plan 5: it needs the park engine's leases |
| +// (ParkLeases), which production does not wire, so rollbackTargetArmed |
| +// answers false there and the pass logs a standing notice. The decision is |
| +// not gated (J10). The move's event log (moves.LogEvent, Plan 6) is the |
| +// LogEvent hook: nil in production, where the event is a log line. Nothing |
| +// here pages: the release deadline is the console's. |
| + |
| +// rollbackStepsPerPass bounds the phases one pass runs. |
| +const rollbackStepsPerPass = 8 |
| + |
| +// The target's rollback steps in its op journal (the landing's journal of |
| +// the move id, moves.Journal). Each step is skipped on a replay once done. |
| +const ( |
| + rbStepTargetStamp = "rb-target-stamp" // F, noted from the claim before anything is retired |
| + rbStepLaneQueued = "rb-lane-queued" // the move's lane park is queued in this region |
| + rbStepRemote = "rb-target-park-remote" // both parks' archives are complete, or acknowledged |
| + rbStepInstallation = "rb-target-park-installation" // retired, completed or unregistered |
| + rbStepLocal = "rb-target-park-local" // parkLocalArtifacts' step: the retirement id |
| + rbStepRowDeleted = "rb-target-row-deleted" // "pending" before the delete, "done" after it |
| +) |
| + |
| +// stoppedTargetRole is the target's StoppedReport role. |
| +const stoppedTargetRole = "target" |
| + |
| +// now is the worker's clock. |
| +func (w *movesWorker) now() time.Time { |
| + if n := w.cfg.Now; n != nil { |
| + return n() |
| + } |
| + return time.Now() |
| +} |
| + |
| +// logEvent records a move event: the LogEvent hook when wired (Plan 6's |
| +// moves.LogEvent), else a log line. |
| +func (w *movesWorker) logEvent(ctx context.Context, rec provision.Record, event string) { |
| + if f := w.cfg.LogEvent; f != nil { |
| + f(ctx, rec, event) |
| + return |
| + } |
| + id := "" |
| + if rec.Move != nil { |
| + id = rec.Move.ID |
| + } |
| + log.Printf("moves: %s: %s %s (the move log is Plan 6's)", rec.Host, id, event) |
| +} |
| + |
| +// rollBackLanding is the rollback decision on the target (§7.8; J10, J11): |
| +// a landing marked for a rollback, or past landingDeadlineOf at any phase, |
| +// is rolled back by MoveRollBack, and the claim follows. done is true when |
| +// the record was this decision's to take (written, or refused because the |
| +// record moved on); the landing's pass then ends. |
| +func (w *movesWorker) rollBackLanding(ctx context.Context, s release.Store, rec provision.Record) (done bool, err error) { |
| + m := rec.Move |
| + if m == nil || m.Legacy() || m.State != provision.MoveLanding { |
| + return false, nil |
| + } |
| + reason := m.Reason |
| + if reason == "" { |
| + d, which := landingDeadlineOf(m) |
| + if d == nil || !w.now().After(*d) { |
| + return false, nil |
| + } |
| + log.Printf("moves: %s: landing %s is past its %s (%s)", rec.Host, m.ID, which, d.UTC().Format(time.RFC3339)) |
| + reason = provision.RollbackLandingDeadline |
| + } |
| + rbID, err := moves.NewRollbackID(rand.Reader) |
| + if err != nil { |
| + return true, err |
| + } |
| + g := provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation} |
| + fresh, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveRollBack(g, rbID, reason, w.now())) |
| + if errors.Is(err, provision.ErrMoveConflict) { |
| + // A commit that won (the open is the commit's path), or another |
| + // writer's rollback (the participant pass follows it). |
| + log.Printf("moves: %s: the rollback of landing %s is refused (%v); the record moved on", rec.Host, m.ID, err) |
| + return true, nil |
| + } |
| + if err != nil { |
| + return true, fmt.Errorf("roll the landing back: %w", err) |
| + } |
| + log.Printf("moves: %s: landing %s rolls back as %s: %s", rec.Host, m.ID, rbID, fresh.Move.Rollback.Reason) |
| + w.logEvent(ctx, *fresh, "rollback") |
| + if _, err := w.rollbackClaim(ctx, s, *fresh); err != nil { |
| + log.Printf("moves: %s: the claim half of rollback %s waits for the next pass: %v", rec.Host, rbID, err) |
| + } |
| + nudgeSerial(ctx, s) |
| + return true, nil |
| +} |
| + |
| +// rollbackClaim is the claim half of the record's rollback: the op of this |
| +// move, for the record's incarnation, goes to restoring/releasing under the |
| +// rollback id (RollBackFlipped writes the flip first over a claim still at |
| +// moving/sealed). A claim already restoring for this rollback is done. |
| +func (w *movesWorker) rollbackClaim(ctx context.Context, s release.Store, rec provision.Record) (hostclaim.Observed, error) { |
| + m := rec.Move |
| + obs, err := hostclaim.Read(ctx, s, rec.Host) |
| + if err != nil { |
| + return obs, fmt.Errorf("read the host claim: %w", err) |
| + } |
| + if obs.Absent { |
| + return obs, fmt.Errorf("move %s has no host claim", m.ID) |
| + } |
| + if err := hostclaim.RequireOp(m.ID)(obs.Claim); err != nil { |
| + return obs, fmt.Errorf("move %s: %w", m.ID, err) |
| + } |
| + h := obs.Claim.Holder |
| + if h == nil || h.Incarnation != m.Incarnation { |
| + return obs, fmt.Errorf("move %s: the host claim names another incarnation", m.ID) |
| + } |
| + rb := m.Rollback.ID |
| + if op := obs.Claim.Op; op.Kind == hostclaim.KindRestoring && op.Intent["rollback"] == rb { |
| + return obs, nil |
| + } |
| + if obs, err = hostclaim.Update(ctx, s, obs, hostclaim.RollBackFlipped(m.ID, rb, w.cfg.Box, h.Stamp, h.Generation)); err != nil { |
| + return obs, fmt.Errorf("the claim half of the rollback: %w", err) |
| + } |
| + log.Printf("moves: %s: the claim follows rollback %s", rec.Host, rb) |
| + return obs, nil |
| +} |
| + |
| +// rollbackTargetArmed reports whether this box can run the target's half of |
| +// a rollback: every hook it calls is wired, the park engine's leases |
| +// included. In production ParkLeases is not (ruling G17), so this answers |
| +// false there (J18). |
| +func rollbackTargetArmed(cfg moveCfg) bool { |
| + return cfg.Box != "" && cfg.DataDir != "" && cfg.Guard.DataDir != "" && cfg.Local.DataDir != "" && cfg.ParkLeases != nil && |
| + cfg.HandHold != nil && cfg.Fence != nil && cfg.TimersDrained != nil && cfg.Evidence != nil && cfg.FinalCheckpoint != nil && |
| + cfg.LocalPosition != nil && cfg.Timers != nil && cfg.FilesDir != nil && cfg.Route != nil && cfg.Versioned != nil && |
| + cfg.StagingLease != nil && cfg.Staging != nil && cfg.DeleteHeld != nil && cfg.SecretsPub != nil |
| +} |
| + |
| +// rollbackRun is one pass's state for one rolling_back record on its target. |
| +type rollbackRun struct { |
| + w *movesWorker |
| + s release.Store |
| + rec provision.Record |
| + g provision.MoveGuard |
| + rb string |
| +} |
| + |
| +// rollbackTarget is the participant step for a rolling_back record on its |
| +// target (movesPass, J10). |
| +func (w *movesWorker) rollbackTarget(ctx context.Context, s release.Store, rec provision.Record) error { |
| + cfg := w.cfg |
| + m := rec.Move |
| + if m == nil || m.Legacy() || m.State != provision.MoveRollingBack || m.Rollback == nil || m.To != cfg.Region || m.From == cfg.Region { |
| + return nil |
| + } |
| + if !rollbackTargetArmed(cfg) { |
| + w.notice(rec.Host+"#rollback", "moves: %s: the target's half of rollback %s waits: this box's rollback is not armed (Plan 5's park view and the park engine's leases are needed)", rec.Host, m.Rollback.ID) |
| + return nil |
| + } |
| + r := &rollbackRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, rb: m.Rollback.ID} |
| + for i := 0; i < rollbackStepsPerPass; i++ { |
| + advanced, err := r.step(ctx) |
| + if err != nil || !advanced { |
| + return err |
| + } |
| + fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| + if err != nil { |
| + return fmt.Errorf("re-read the record: %w", err) |
| + } |
| + 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 { |
| + return nil |
| + } |
| + r.rec = *fresh |
| + } |
| + return nil |
| +} |
| + |
| +func (r *rollbackRun) cfg() moveCfg { return r.w.cfg } |
| +func (r *rollbackRun) now() time.Time { return r.w.now() } |
| +func (r *rollbackRun) rollback() *provision.MoveRollback { return r.rec.Move.Rollback } |
| +func (r *rollbackRun) dbPath() string { return provision.DefaultDBPath(r.cfg().DataDir, r.rec.Host) } |
| + |
| +// fault is the test seam after a side effect. |
| +func (r *rollbackRun) fault(point string) error { |
| + if f := r.cfg().rbFault; f != nil { |
| + return f(point) |
| + } |
| + return nil |
| +} |
| + |
| +// step runs the target's current rollback phase once. advanced is true when |
| +// it wrote the record. |
| +func (r *rollbackRun) step(ctx context.Context) (bool, error) { |
| + switch r.rollback().TargetPhase { |
| + case provision.RbTargetReleasing: |
| + return r.releasing(ctx) |
| + case provision.RbTargetReleased: |
| + return r.released(ctx) |
| + case provision.RbTargetReturning: |
| + return r.returning(ctx) |
| + case provision.RbTargetReturned: |
| + return r.toCleaning(ctx, provision.RbTargetReturned) |
| + case provision.RbTargetCleaning: |
| + return r.cleaning(ctx) |
| + case provision.RbTargetCleaned: |
| + return false, r.cleaned() |
| + } |
| + return false, nil // lost: the declared loss's (Task 10) |
| +} |
| + |
| +// phase writes one target phase step, then logs the event, if any, and |
| +// passes the step's fault point. A conflict is someone else's decision. |
| +func (r *rollbackRun) phase(ctx context.Context, from, to string, stamp func(*provision.MoveRollback), event string) (bool, error) { |
| + host := r.rec.Host |
| + if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRollbackTargetPhase(r.g, from, to, stamp)); err != nil { |
| + if errors.Is(err, provision.ErrMoveConflict) { |
| + log.Printf("moves: %s: the target's rollback %s to %s refused (%v); the record moved on", host, from, to, err) |
| + return false, nil |
| + } |
| + return false, fmt.Errorf("the target's rollback %s to %s: %w", from, to, err) |
| + } |
| + log.Printf("moves: %s: the target of rollback %s is %s", host, r.rb, to) |
| + if event != "" { |
| + r.w.logEvent(ctx, r.rec, event) |
| + } |
| + if err := r.fault("after-record-" + to); err != nil { |
| + return true, err |
| + } |
| + return true, nil |
| +} |
| + |
| +// claim reads the host claim and requires this rollback's restoring op, |
| +// with this box the move's target, for the record's incarnation. |
| +func (r *rollbackRun) claim(ctx context.Context) (hostclaim.Observed, error) { |
| + obs, err := hostclaim.Read(ctx, r.s, r.rec.Host) |
| + if err != nil { |
| + return obs, fmt.Errorf("read the host claim: %w", err) |
| + } |
| + if obs.Absent { |
| + return obs, fmt.Errorf("move %s has no host claim", r.g.ID) |
| + } |
| + if err := r.ours(obs.Claim); err != nil { |
| + return obs, fmt.Errorf("move %s: %w", r.g.ID, err) |
| + } |
| + return obs, nil |
| +} |
| + |
| +// ours is the claim check of every target phase: this rollback's restoring |
| +// op, this box the move's target, the record's incarnation. |
| +func (r *rollbackRun) ours(c *hostclaim.Claim) error { |
| + box := r.cfg().Box |
| + if c == nil || c.Op.ID != r.g.ID || c.Op.Kind != hostclaim.KindRestoring || c.Op.Intent["rollback"] != r.rb { |
| + return fmt.Errorf("%w: the claim is not rollback %s", hostclaim.ErrWrongState, r.rb) |
| + } |
| + if c.Holder == nil || c.Holder.Incarnation != r.g.Incarnation { |
| + return fmt.Errorf("%w: the claim names another incarnation", hostclaim.ErrWrongState) |
| + } |
| + if len(c.Op.Participants) != 2 || c.Op.Participants[1] != box || c.Op.Intent["target"] != box { |
| + return fmt.Errorf("%w: this box is not the move's target in the claim", hostclaim.ErrWrongState) |
| + } |
| + return nil |
| +} |
| + |
| +// stillReleasing is the final checkpoint's authority: every replica request |
| +// re-reads the claim, which must still name this rollback releasing with this |
| +// box executing. |
| +func (r *rollbackRun) stillReleasing(ctx context.Context) error { |
| + obs, err := r.claim(ctx) |
| + if err != nil { |
| + return err |
| + } |
| + if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleasing || op.Exec != r.cfg().Box { |
| + return fmt.Errorf("move %s: this box no longer executes the host", r.g.ID) |
| + } |
| + return nil |
| +} |
| + |
| +// reachedProbing reports whether the landing reached probing: a writer may |
| +// have run on this box. |
| +func (r *rollbackRun) reachedProbing() bool { |
| + order := provision.LandingOrder() |
| + return landingIndex(order, r.rec.Move.LandingPhase) >= landingIndex(order, provision.LandProbing) |
| +} |
| + |
| +// unchanged reports whether the stopped target's digests are what the |
| +// landing installed: the database, the files and the timers. |
| +func (r *rollbackRun) unchanged(report moves.StoppedReport) bool { |
| + m := r.rec.Move |
| + return m.Landing != nil && m.Landing.InstalledDBSHA256 != "" && report.DBSHA256 == m.Landing.InstalledDBSHA256 && |
| + m.Source != nil && m.Source.Files != nil && report.FilesPlain == m.Source.Files.Plain && report.TimersSHA256 == m.Source.TimersSHA256 |
| +} |
| + |
| +// releasing (§7.8 target releasing and released): the hold, the fence, the |
| +// drain and the stop evidence; the final checkpoint and the digests; the |
| +// stopped report; the record's released with Dirty and Evidence; the claim's |
| +// release. A replay runs it all again: the hand-over, the fence and the drain |
| +// are idempotent, the evidence reads durable state, the checkpoint is |
| +// idempotent on a stopped database, and the report is create-only and read |
| +// back, so the decision is the published report's. |
| +func (r *rollbackRun) releasing(ctx context.Context) (bool, error) { |
| + cfg := r.cfg() |
| + host, id := r.rec.Host, r.g.ID |
| + if _, err := r.w.rollbackClaim(ctx, r.s, r.rec); err != nil { |
| + return false, err |
| + } |
| + obs, err := r.claim(ctx) |
| + if err != nil { |
| + return false, err |
| + } |
| + if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleasing || op.Exec != cfg.Box { |
| + return false, fmt.Errorf("move %s: the claim is restoring/%q with executor %q, not releasing on this box", id, op.Phase, op.Exec) |
| + } |
| + holder := *obs.Claim.Holder |
| + if _, err := writerLease(cfg.OwnPrefix, host, holder); err != nil { |
| + return false, fmt.Errorf("move %s: %w", id, err) |
| + } |
| + seal, err := journalledSeal(obs.Claim) |
| + if err != nil { |
| + return false, err |
| + } |
| + handed, err := cfg.HandHold(host, "releasing", r.rb, id) |
| + if err != nil { |
| + return false, fmt.Errorf("hand the row's hold to the rollback: %w", err) |
| + } |
| + if !handed { |
| + return false, fmt.Errorf("move %s: this box's row is held by neither the move nor rollback %s", id, r.rb) |
| + } |
| + gen := moveFenceGeneration(r.rec.Move) |
| + if err := cfg.Fence(ctx, host, gen); err != nil { |
| + return false, fmt.Errorf("fence against wakes: %w", err) |
| + } |
| + r.w.fenced.Store(host, gen) |
| + if err := cfg.TimersDrained(ctx, host); err != nil { |
| + return false, fmt.Errorf("drain the timers: %w", err) |
| + } |
| + dbPath := r.dbPath() |
| + ev, err := cfg.Evidence(ctx, host, holder.Location, []string{dbPath}) |
| + if err != nil { |
| + return false, fmt.Errorf("stop evidence: %w", err) |
| + } |
| + if !ev.TreeStopped() { |
| + 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)) |
| + return false, nil |
| + } |
| + reached := r.reachedProbing() |
| + fromDB, err := regularFile(dbPath) |
| + if err != nil { |
| + return false, err |
| + } |
| + served := reached && fromDB |
| + if served { |
| + if err := cfg.FinalCheckpoint(ctx, host, dbPath, holder, r.stillReleasing); err != nil { |
| + return false, fmt.Errorf("final checkpoint: %w", err) |
| + } |
| + } |
| + if err := r.fault("after-checkpoint"); err != nil { |
| + return false, err |
| + } |
| + n := seal.N |
| + if served { |
| + if n, err = cfg.LocalPosition(dbPath); err != nil { |
| + return false, fmt.Errorf("read the stopped database's position: %w", err) |
| + } |
| + } |
| + var dbSHA string |
| + if fromDB { |
| + if dbSHA, err = fileSHA256(dbPath); err != nil { |
| + return false, fmt.Errorf("digest the stopped database: %w", err) |
| + } |
| + } |
| + dir, envPath := cfg.FilesDir(host) |
| + plain, _, _, err := archivePlainDigest(dir, envPath) |
| + if err != nil { |
| + if reached { |
| + return false, fmt.Errorf("digest the stopped files: %w", err) |
| + } |
| + // No writer ran here: the files are not compared, and may not |
| + // have been extracted yet. |
| + plain = "" |
| + } |
| + rows, err := cfg.Timers(host) |
| + if err != nil { |
| + return false, fmt.Errorf("read the timers: %w", err) |
| + } |
| + report := ev |
| + report.V, report.Role, report.Box, report.Incarnation, report.ClaimGen = 1, stoppedTargetRole, cfg.Box, holder.Incarnation, obs.Claim.Gen |
| + report.Host, report.OpID = host, id |
| + report.Steps = map[string]moves.Check{} |
| + for k, v := range ev.Steps { |
| + report.Steps[k] = v |
| + } |
| + report.Steps[moves.StepReplica] = moves.Check{OK: true} |
| + report.FinalTXID, report.DBSHA256, report.FilesPlain, report.TimersSHA256 = n, dbSHA, plain, moves.TimersDigest(rows) |
| + // The report's time is after the checkpoint, the target's last write to |
| + // its location: it is the horizon of that location's park. |
| + report.CheckedAt = r.now().UTC() |
| + if !report.Complete() { |
| + return false, fmt.Errorf("the target's stopped report is not complete (%s)", failedSteps(report)) |
| + } |
| + key := moves.StoppedKey(moves.ReportPrefix(r.rec.Account, r.rec.App, host, id), stoppedTargetRole) |
| + if report, err = publishStoppedReport(ctx, r.s, key, report); err != nil { |
| + return false, err |
| + } |
| + if err := r.fault("after-report"); err != nil { |
| + return false, err |
| + } |
| + clean := seal.SrcCovered && (!reached || r.unchanged(report)) |
| + 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) |
| + ok, err := r.phase(ctx, provision.RbTargetReleasing, provision.RbTargetReleased, func(rb *provision.MoveRollback) { |
| + rb.Dirty, rb.Evidence = !clean, key |
| + }, "target-released") |
| + if !ok || err != nil { |
| + return ok, err |
| + } |
| + if err := r.claimReleased(ctx, key); err != nil { |
| + log.Printf("moves: %s: the claim half of the target's release waits for the next pass: %v", host, err) |
| + } |
| + return true, nil |
| +} |
| + |
| +// claimReleased is the claim half of released (TargetReleased): execution to |
| +// none citing the target's evidence, the target region's location park whose |
| +// horizon is the report's time, the target's park-local when its inventory |
| +// lists F, and the region's predecessors covered. A claim already released |
| +// with this evidence is done. |
| +func (r *rollbackRun) claimReleased(ctx context.Context, evidence string) error { |
| + cfg := r.cfg() |
| + obs, err := r.claim(ctx) |
| + if err != nil { |
| + return err |
| + } |
| + op := obs.Claim.Op |
| + if op.Phase != hostclaim.PhaseReleasing { |
| + if op.Journal[hostclaim.JournalEvidenceTarget] == evidence { |
| + return nil |
| + } |
| + 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) |
| + } |
| + report, err := readStoppedReport(ctx, r.s, evidence) |
| + if err != nil { |
| + return err |
| + } |
| + holder := obs.Claim.Holder |
| + scope := hostclaim.ParkScope{Location: strings.TrimSuffix(holder.Location, "/"), Incarnation: holder.Incarnation, |
| + EvidenceAt: report.CheckedAt.UTC(), RequestLimit: cutRequestLimit} |
| + if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.TargetReleased(r.g.ID, cfg.Box, cfg.Region, evidence, scope, obligationID)); err != nil { |
| + return fmt.Errorf("the claim half of the release: %w", err) |
| + } |
| + log.Printf("moves: %s: the target gave execution up for rollback %s, citing %s", r.rec.Host, r.rb, evidence) |
| + return nil |
| +} |
| + |
| +// released: the claim half, if a crash left it, then returning (dirty) or, |
| +// once the source has recovered, cleaning (clean). |
| +func (r *rollbackRun) released(ctx context.Context) (bool, error) { |
| + if err := r.claimReleased(ctx, r.rollback().Evidence); err != nil { |
| + return false, err |
| + } |
| + if r.rollback().Dirty { |
| + return r.phase(ctx, provision.RbTargetReleased, provision.RbTargetReturning, nil, "") |
| + } |
| + return r.toCleaning(ctx, provision.RbTargetReleased) |
| +} |
| + |
| +// toCleaning waits for the source's recovery (Review Focus 4): until then the |
| +// target keeps everything it holds. |
| +func (r *rollbackRun) toCleaning(ctx context.Context, from string) (bool, error) { |
| + if r.rollback().RecoveredAt == nil { |
| + 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) |
| + return false, nil |
| + } |
| + return r.phase(ctx, from, provision.RbTargetCleaning, nil, "") |
| +} |
| + |
| +// returnedLane is what a return lane holds and restores to. |
| +type returnedLane struct { |
| + n uint64 |
| + replica, files, timers []moves.SealedObject |
| + baseline provision.MoveSource |
| +} |
| + |
| +// returning (§7.8 target returning): the return lane under the source's |
| +// staging-work lease, the return seal and baseline in the claim's journal, |
| +// then the record's returned. A replay that finds the seal journalled only |
| +// writes the record; one that does not writes the lane again (new versions; |
| +// the seal names the ones it wrote, and the lane's park removes the rest). |
| +func (r *rollbackRun) returning(ctx context.Context) (bool, error) { |
| + cfg := r.cfg() |
| + host, id, m := r.rec.Host, r.g.ID, r.rec.Move |
| + rb := r.rollback() |
| + obs, err := r.claim(ctx) |
| + if err != nil { |
| + return false, err |
| + } |
| + if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleased || op.Exec != "" || op.Journal[hostclaim.JournalEvidenceTarget] != rb.Evidence { |
| + return false, fmt.Errorf("move %s: the claim is restoring/%q with executor %q, not released by this target", id, op.Phase, op.Exec) |
| + } |
| + if _, ok := obs.Claim.Op.Journal[moves.JournalReturnSeal]; ok { |
| + return r.phase(ctx, provision.RbTargetReturning, provision.RbTargetReturned, nil, "target-returned") |
| + } |
| + report, err := readStoppedReport(ctx, r.s, rb.Evidence) |
| + if err != nil { |
| + return false, err |
| + } |
| + staged, err := journalledSeal(obs.Claim) |
| + if err != nil { |
| + return false, err |
| + } |
| + fleet, err := loadMoveFleet(ctx, r.s) |
| + if err != nil { |
| + return false, err |
| + } |
| + src, ok := fleet.targets[m.From] |
| + if !ok { |
| + return false, fmt.Errorf("source region %s publishes no data bucket. waiting", m.From) |
| + } |
| + held, release, err := cfg.StagingLease(ctx, m.From, host) |
| + if err != nil { |
| + var he *lease.HeldError |
| + if errors.As(err, &he) { |
| + r.w.notice(host+"#rb-lease", "moves: %s: another worker holds the staging lease; the return of %s waits", host, r.rb) |
| + return false, nil |
| + } |
| + return false, fmt.Errorf("staging lease: %w", err) |
| + } |
| + defer func() { |
| + rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| + defer cancel() |
| + if err := release(rctx); err != nil { |
| + log.Printf("moves: %s: release the staging lease: %v", host, err) |
| + } |
| + }() |
| + stg, err := r.w.staging(ctx, src.DataBucketRegion) |
| + if err != nil { |
| + return false, err |
| + } |
| + own, err := r.w.mirror(ctx, cfg.OwnBucketRegion) |
| + if err != nil { |
| + return false, err |
| + } |
| + gm := &guardedMirror{w: stg, g: moves.Guarded{Lease: held, Now: r.now}, dst: src.DataBucket, own: own} |
| + if err := r.laneMoveJSON(ctx, gm); err != nil { |
| + return false, err |
| + } |
| + var lane returnedLane |
| + if !r.reachedProbing() || r.unchanged(report) { |
| + lane, err = r.copyAsGiven(ctx, gm, staged) |
| + } else { |
| + lane, err = r.snapshotStopped(ctx, gm, report) |
| + } |
| + if err != nil { |
| + return false, err |
| + } |
| + // The bound is read after the last lane write: every request the guard |
| + // let a writer sign expires before it. |
| + bound, err := held.SafeUntil() |
| + if err != nil { |
| + return false, fmt.Errorf("the staging lease's bound: %w", err) |
| + } |
| + seal := moves.ReturnSeal(lane.n, lane.replica, lane.files, lane.timers, r.now(), bound) |
| + if err := seal.Validate(); err != nil { |
| + return false, err |
| + } |
| + if !seal.ReachesN() { |
| + return false, fmt.Errorf("the return lane does not end at TXID %d", seal.N) |
| + } |
| + base, err := moves.EncodeReturnBaseline(lane.baseline) |
| + if err != nil { |
| + return false, err |
| + } |
| + if obs, err = r.claim(ctx); err != nil { |
| + return false, err |
| + } |
| + if _, err := hostclaim.Update(ctx, r.s, obs, journalReturn(id, r.rb, rb.Evidence, seal.Encode(), base)); err != nil { |
| + return false, fmt.Errorf("journal the return seal: %w", err) |
| + } |
| + 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) |
| + if err := r.fault("after-return-seal"); err != nil { |
| + return false, err |
| + } |
| + return r.phase(ctx, provision.RbTargetReturning, provision.RbTargetReturned, nil, "target-returned") |
| +} |
| + |
| +// journalReturn is the return seal's claim write: the seal and the baseline, |
| +// in one write, on this rollback's op released by this target. |
| +func journalReturn(opID, rbID, evidence, seal, baseline string) func(*hostclaim.Claim) error { |
| + return func(c *hostclaim.Claim) error { |
| + if err := hostclaim.RequireOp(opID)(c); err != nil { |
| + return err |
| + } |
| + op := c.Op |
| + if op.Kind != hostclaim.KindRestoring || op.Intent["rollback"] != rbID || op.Phase != hostclaim.PhaseReleased || op.Exec != "" || |
| + op.Journal[hostclaim.JournalEvidenceTarget] != evidence { |
| + return fmt.Errorf("%w: the return seal needs this rollback released by its target", hostclaim.ErrWrongState) |
| + } |
| + return hostclaim.SetJournal(opID, map[string]string{moves.JournalReturnSeal: seal, moves.JournalReturnBaseline: baseline})(c) |
| + } |
| +} |
| + |
| +// laneMoveJSON writes the lane's move.json create-only (§7.3), naming the |
| +// rollback; one already there must name it and this incarnation. |
| +func (r *rollbackRun) laneMoveJSON(ctx context.Context, gm *guardedMirror) error { |
| + host, m := r.rec.Host, r.rec.Move |
| + body, err := json.Marshal(moves.MoveJSON{MoveID: r.rb, Incarnation: m.Incarnation, Host: host, From: m.To, To: m.From}) |
| + if err != nil { |
| + return err |
| + } |
| + key := moves.MoveJSONKey(host, r.rb) |
| + if _, err := gm.PutIfAbsentVersioned(ctx, gm.dst, key, body); errors.Is(err, lease.ErrExists) { |
| + prev, err := gm.Get(ctx, gm.dst, key) |
| + if err != nil { |
| + return fmt.Errorf("read the lane's move.json: %w", err) |
| + } |
| + var mj moves.MoveJSON |
| + if err := json.Unmarshal(prev, &mj); err != nil || mj.MoveID != r.rb || mj.Incarnation != m.Incarnation { |
| + return fmt.Errorf("the lane's move.json names %q/%q, not rollback %s", mj.MoveID, mj.Incarnation, r.rb) |
| + } |
| + } else if err != nil { |
| + return fmt.Errorf("write the lane's move.json: %w", err) |
| + } |
| + return nil |
| +} |
| + |
| +// copyAsGiven is the return of a target whose state is the staged seal's |
| +// (dirty only for want of src_covered): every sealed version is read from |
| +// this box's bucket and written into the lane under the same key relative to |
| +// the lane, through the guarded staging writer (J22: PutReaderVersionedAt |
| +// under Guarded.Before). The baseline is the source's own, cited under the |
| +// target's evidence. |
| +func (r *rollbackRun) copyAsGiven(ctx context.Context, gm *guardedMirror, staged moves.Seal) (returnedLane, error) { |
| + cfg := r.cfg() |
| + host, id, m := r.rec.Host, r.g.ID, r.rec.Move |
| + if m.Source == nil || m.Source.Files == nil || staged.N != m.Source.FinalTXID { |
| + return returnedLane{}, fmt.Errorf("move %s: the record's source baselines are not the staged seal's", id) |
| + } |
| + vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion) |
| + if err != nil { |
| + return returnedLane{}, err |
| + } |
| + tmp := filepath.Join(moves.JournalDir(cfg.DataDir, host, id), "return") |
| + if err := os.MkdirAll(tmp, 0o700); err != nil { |
| + return returnedLane{}, err |
| + } |
| + defer os.RemoveAll(tmp) |
| + from, lane := moves.StagingPrefix(host, id), moves.ReturnLane(host, r.rb) |
| + copyClass := func(objs []moves.SealedObject) ([]moves.SealedObject, error) { |
| + var out []moves.SealedObject |
| + for _, o := range objs { |
| + rel, ok := strings.CutPrefix(o.Key, from) |
| + if !ok || rel == "" { |
| + return nil, fmt.Errorf("%w: %s is outside the move's lane", moves.ErrBadSeal, o.Key) |
| + } |
| + path := filepath.Join(tmp, "object") |
| + if err := downloadExact(ctx, vs, cfg.OwnBucket, o.Key, o.VersionID, o.Size, path); err != nil { |
| + return nil, fmt.Errorf("read %s: %w", o.Key, err) |
| + } |
| + sum, err := fileSHA256(path) |
| + if err != nil { |
| + return nil, err |
| + } |
| + ver, err := gm.PutFileVersioned(ctx, gm.dst, lane+rel, path, o.Size, digestHex(sum)) |
| + os.Remove(path) |
| + if err != nil { |
| + return nil, fmt.Errorf("return %s: %w", o.Key, err) |
| + } |
| + if !lease.UsableVersionID(ver) { |
| + return nil, fmt.Errorf("the returned %s has no usable version id %q", rel, ver) |
| + } |
| + out = append(out, moves.SealedObject{Key: lane + rel, VersionID: ver, Size: o.Size}) |
| + } |
| + return out, nil |
| + } |
| + var out returnedLane |
| + if out.replica, err = copyClass(staged.Replica); err != nil { |
| + return out, err |
| + } |
| + if out.files, err = copyClass(staged.Files); err != nil { |
| + return out, err |
| + } |
| + if out.timers, err = copyClass(staged.Timers); err != nil { |
| + return out, err |
| + } |
| + files := *m.Source.Files |
| + out.n, out.baseline = staged.N, *m.Source |
| + out.baseline.Files, out.baseline.Evidence = &files, r.rollback().Evidence |
| + return out, nil |
| +} |
| + |
| +// downloadExact reads one exact version of size bytes into path. |
| +func downloadExact(ctx context.Context, vs lease.VersionedStore, bucket, key, version string, size int64, path string) error { |
| + rc, err := vs.OpenVersion(ctx, bucket, key, version) |
| + if err != nil { |
| + return err |
| + } |
| + defer rc.Close() |
| + f, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600) |
| + if err != nil { |
| + return err |
| + } |
| + n, err := io.Copy(f, io.LimitReader(rc, size+1)) |
| + if cerr := f.Close(); err == nil { |
| + err = cerr |
| + } |
| + if err != nil { |
| + return err |
| + } |
| + if n != size { |
| + return fmt.Errorf("the version is %d bytes, sealed as %d", n, size) |
| + } |
| + return nil |
| +} |
| + |
| +// snapshotStopped is the return of a target that served and changed: one |
| +// snapshot [1, N'] of the stopped database (moves.EncodeFull, in memory, the |
| +// Phase 1 size limit), the target's files archive (manifest bound to the move |
| +// id) and its timers. Each must still be what the stopped report recorded. |
| +func (r *rollbackRun) snapshotStopped(ctx context.Context, gm *guardedMirror, report moves.StoppedReport) (returnedLane, error) { |
| + cfg := r.cfg() |
| + host, id := r.rec.Host, r.g.ID |
| + dbPath := r.dbPath() |
| + sha, err := fileSHA256(dbPath) |
| + if err != nil { |
| + return returnedLane{}, fmt.Errorf("the target's stopped database: %w", err) |
| + } |
| + if sha != report.DBSHA256 { |
| + return returnedLane{}, fmt.Errorf("move %s: the target's database changed since its release", id) |
| + } |
| + n, err := cfg.LocalPosition(dbPath) |
| + if err != nil { |
| + return returnedLane{}, fmt.Errorf("read the stopped database's position: %w", err) |
| + } |
| + if n != report.FinalTXID || n == 0 { |
| + return returnedLane{}, fmt.Errorf("move %s: the stopped database is at TXID %d, the report at %d", id, n, report.FinalTXID) |
| + } |
| + pageSize, pageCount, err := moves.ReadDBHeader(dbPath) |
| + if err != nil { |
| + return returnedLane{}, err |
| + } |
| + f, err := os.Open(dbPath) |
| + if err != nil { |
| + return returnedLane{}, err |
| + } |
| + data, _, err := moves.EncodeFull(f, pageSize, pageCount, 1, ltx.TXID(n), report.CheckedAt.UnixMilli(), 0) |
| + f.Close() |
| + if err != nil { |
| + return returnedLane{}, fmt.Errorf("snapshot the stopped database: %w", err) |
| + } |
| + var out returnedLane |
| + key := moves.SnapshotKey(moves.ReplicaPrefix(host, r.rb), n) |
| + ver, err := gm.PutVersioned(ctx, gm.dst, key, data) |
| + if err != nil { |
| + return out, fmt.Errorf("return the snapshot: %w", err) |
| + } |
| + if !lease.UsableVersionID(ver) { |
| + return out, fmt.Errorf("the returned snapshot has no usable version id %q", ver) |
| + } |
| + out.replica = []moves.SealedObject{{Key: key, VersionID: ver, Size: int64(len(data))}} |
| + a, why, retry, err := r.w.buildArchive(r.rec) |
| + switch { |
| + case err != nil: |
| + return out, fmt.Errorf("archive the target's files: %w", err) |
| + case why != "": |
| + return out, fmt.Errorf("move %s: the target's files cannot be archived (%s)", id, why) |
| + case retry: |
| + return out, errors.New("a stopped file changed while it was archived") |
| + case a.Plain != report.FilesPlain: |
| + return out, fmt.Errorf("move %s: the target's files are %s, the report %s", id, a.Plain, report.FilesPlain) |
| + } |
| + ref, err := a.stage(ctx, gm, gm.dst, host, r.rb) |
| + if err != nil { |
| + return out, err |
| + } |
| + mst, err := os.Stat(a.manifestPath()) |
| + if err != nil { |
| + return out, err |
| + } |
| + if !lease.UsableVersionID(ref.SealedVersion) || !lease.UsableVersionID(ref.ManifestVersion) { |
| + return out, fmt.Errorf("the returned archive has no usable version ids (%q, %q)", ref.SealedVersion, ref.ManifestVersion) |
| + } |
| + out.files = []moves.SealedObject{{Key: ref.SealedObjectKey, VersionID: ref.SealedVersion, Size: a.SealedSize}, |
| + {Key: ref.ManifestKey, VersionID: ref.ManifestVersion, Size: mst.Size()}} |
| + rows, err := cfg.Timers(host) |
| + if err != nil { |
| + return out, fmt.Errorf("read the timers: %w", err) |
| + } |
| + if moves.TimersDigest(rows) != report.TimersSHA256 { |
| + return out, fmt.Errorf("move %s: the target's timers changed since its release", id) |
| + } |
| + body, err := moves.TimersSnapshot{V: 1, Host: host, Rows: rows, TakenAt: report.CheckedAt}.Encode() |
| + if err != nil { |
| + return out, err |
| + } |
| + tkey := moves.TimersPrefix(host, r.rb) + "timers.json" |
| + tver, err := gm.PutVersioned(ctx, gm.dst, tkey, body) |
| + if err != nil { |
| + return out, fmt.Errorf("return the timers: %w", err) |
| + } |
| + if !lease.UsableVersionID(tver) { |
| + return out, fmt.Errorf("the returned timers have no usable version id %q", tver) |
| + } |
| + out.timers = []moves.SealedObject{{Key: tkey, VersionID: tver, Size: int64(len(body))}} |
| + rt, err := cfg.Route(host) |
| + if err != nil { |
| + return out, fmt.Errorf("read the registry row: %w", err) |
| + } |
| + if rt == nil { |
| + return out, fmt.Errorf("%s has no registry row on this box", host) |
| + } |
| + out.n = n |
| + out.baseline = provision.MoveSource{Replication: rt.Replication, NextDue: rt.NextDue, FinalTXID: n, DBSHA256: sha, |
| + Files: &provision.MoveFilesDigest{Plain: a.Plain, Sealed: a.Sealed, Bytes: a.Bytes, Entries: a.Entries}, |
| + TimersSHA256: report.TimersSHA256, Evidence: r.rollback().Evidence} |
| + return out, nil |
| +} |
| + |
| +// journal is the target's op journal of the move (the landing's). |
| +func (r *rollbackRun) journal() (*moves.Journal, error) { |
| + return moves.ReadJournal(r.cfg().DataDir, r.rec.Host, r.g.ID) |
| +} |
| + |
| +// cleaning (§7.8 target cleaning, INV-10), each step journalled and skipped |
| +// on a replay once done: the remote parks; the registered installation; the |
| +// moves-owned rename of everything else at the flat paths; the row; then the |
| +// record's cleaned with target_parked_at. The source has recovered (the |
| +// record says so), so nothing here is still needed by the move. |
| +func (r *rollbackRun) cleaning(ctx context.Context) (bool, error) { |
| + cfg := r.cfg() |
| + host, id := r.rec.Host, r.g.ID |
| + j, err := r.journal() |
| + if err != nil { |
| + return false, err |
| + } |
| + mark := func(step, value string) error { |
| + j.Mark(step, value) |
| + if err := j.Write(cfg.DataDir); err != nil { |
| + return err |
| + } |
| + return r.fault("after-" + step) |
| + } |
| + obs, err := hostclaim.Read(ctx, r.s, host) |
| + if err != nil { |
| + return false, fmt.Errorf("read the host claim: %w", err) |
| + } |
| + if obs.Absent || obs.Claim.Holder == nil || obs.Claim.Holder.Incarnation != r.g.Incarnation { |
| + return false, fmt.Errorf("move %s: the host claim names no holder of this incarnation", id) |
| + } |
| + ours := r.ours(obs.Claim) == nil |
| + if ours && !j.Done(rbStepTargetStamp) { |
| + if err := mark(rbStepTargetStamp, obs.Claim.Op.Journal[hostclaim.JournalTargetStamp]); err != nil { |
| + return false, err |
| + } |
| + } |
| + if !j.Done(rbStepRemote) { |
| + // An op that has moved on had its chores acknowledged with it. |
| + if ours { |
| + done, err := r.remote(ctx, obs, j, mark) |
| + if err != nil || !done { |
| + return false, err |
| + } |
| + } |
| + if err := mark(rbStepRemote, ""); err != nil { |
| + return false, err |
| + } |
| + } |
| + if !j.Done(rbStepInstallation) { |
| + how, err := r.installation(ctx, ours) |
| + if err != nil { |
| + return false, err |
| + } |
| + if err := mark(rbStepInstallation, how); err != nil { |
| + return false, err |
| + } |
| + } |
| + if err := r.parkLeftovers(ctx, j, mark); err != nil { |
| + return false, fmt.Errorf("park the target's local artifacts: %w", err) |
| + } |
| + if err := r.deleteRow(ctx, j, mark); err != nil { |
| + return false, err |
| + } |
| + return r.phase(ctx, provision.RbTargetCleaning, provision.RbTargetCleaned, func(rb *provision.MoveRollback) { |
| + t := r.now().UTC() |
| + rb.TargetParkedAt = &t |
| + }, "target-cleaned") |
| +} |
| + |
| +// remote parks the target region's location and the move's lane in it: the |
| +// lane's park is queued once (its horizon the staged seal's bound), then |
| +// every park-replica of this op in this region runs through the park engine |
| +// until its archive is complete (park.Engine.Parked). The acknowledgement, |
| +// after the horizon, is the engine's or the regional runner's (C4), never a |
| +// reason to wait here; a chore no longer in the claim was acknowledged. |
| +func (r *rollbackRun) remote(ctx context.Context, obs hostclaim.Observed, j *moves.Journal, mark func(step, value string) error) (bool, error) { |
| + cfg := r.cfg() |
| + host, id := r.rec.Host, r.g.ID |
| + if !j.Done(rbStepLaneQueued) { |
| + staged, err := journalledSeal(obs.Claim) |
| + if err != nil { |
| + return false, err |
| + } |
| + scope := hostclaim.ParkScope{Location: moves.ScopeLocation(host, id), Incarnation: obs.Claim.Holder.Incarnation, |
| + EvidenceAt: staged.Bound().UTC(), RequestLimit: cutRequestLimit} |
| + if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.QueueStagingPark(id, cfg.Region, scope, obligationID)); err != nil { |
| + return false, fmt.Errorf("queue the move's lane park: %w", err) |
| + } |
| + if err := mark(rbStepLaneQueued, ""); err != nil { |
| + return false, err |
| + } |
| + } |
| + obs, err := r.claim(ctx) |
| + if err != nil { |
| + return false, err |
| + } |
| + vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion) |
| + if err != nil { |
| + return false, err |
| + } |
| + e := r.engine(vs) |
| + waiting := 0 |
| + for _, o := range obs.Claim.Obligations { |
| + if o.Op != id || o.Kind != "park-replica" || o.Region != cfg.Region || o.Park == nil { |
| + continue |
| + } |
| + work := park.Work{Host: host, ID: o.ID, Op: id} |
| + if err := e.Run(ctx, work); err != nil && !errors.Is(err, park.ErrPending) && !errors.Is(err, park.ErrChanged) { |
| + return false, fmt.Errorf("park %s: %w", o.Park.Location, err) |
| + } |
| + parked, err := e.Parked(ctx, work) |
| + if errors.Is(err, park.ErrChanged) { |
| + continue |
| + } |
| + if err != nil { |
| + return false, fmt.Errorf("the park of %s: %w", o.Park.Location, err) |
| + } |
| + if !parked { |
| + waiting++ |
| + } |
| + } |
| + if waiting > 0 { |
| + r.w.notice(host+"#rb-remote", "moves: %s: the target's cleaning of %s waits for %d park archive(s)", host, r.rb, waiting) |
| + return false, nil |
| + } |
| + log.Printf("moves: %s: the target region's copies of %s are parked", host, id) |
| + return true, nil |
| +} |
| + |
| +// engine is the park engine over this box's own bucket: the wired leases, and |
| +// the view and authorizer Plan 5 supplies, or the stand-ins (the view with |
| +// the source's region authoritative, which it is again after the rollback). |
| +func (r *rollbackRun) engine(vs lease.VersionedStore) park.Engine { |
| + cfg := r.cfg() |
| + view, auth := cfg.ParkView, cfg.ParkAuthorize |
| + if view == nil { |
| + view = standInView(cfg.Region, roleTarget, r.rec.Move) |
| + } |
| + if auth == nil { |
| + auth = func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error { |
| + if obs.Absent { |
| + return park.ErrUnsafe |
| + } |
| + return r.authorizeChore(obs.Claim, o) |
| + } |
| + } |
| + return park.Engine{Store: vs, Claims: r.s, Bucket: cfg.OwnBucket, Region: cfg.Region, Leases: cfg.ParkLeases, |
| + Now: r.now, NewID: obligationID, View: view, Authorize: auth, After: cfg.parkFault} |
| +} |
| + |
| +// authorizeChore is the stand-in rule until Plan 5 (ruling G17): a chore is |
| +// run only when it is this op's, the claim is still this rollback's with this |
| +// box no longer executing, and the chore is this region's or this box's. |
| +func (r *rollbackRun) authorizeChore(c *hostclaim.Claim, o hostclaim.Obligation) error { |
| + cfg := r.cfg() |
| + if r.ours(c) != nil || o.Op != r.g.ID || c.Op.Exec == cfg.Box { |
| + return fmt.Errorf("%w: not this rollback's chore", park.ErrUnsafe) |
| + } |
| + switch { |
| + case o.Kind == "park-replica" && o.Region == cfg.Region: |
| + case o.Kind == "park-local" && o.Box == cfg.Box: |
| + default: |
| + return fmt.Errorf("%w: not this target's chore", park.ErrUnsafe) |
| + } |
| + return nil |
| +} |
| + |
| +// targetStamp is F: the claim's journalled target stamp, as noted in this |
| +// box's journal when cleaning began. |
| +func targetStamp(j *moves.Journal) uint64 { |
| + f, err := strconv.ParseUint(j.Steps[rbStepTargetStamp], 10, 64) |
| + if err != nil { |
| + return 0 |
| + } |
| + return f |
| +} |
| + |
| +// guard is the host guard over this box's data directory, reading the claim |
| +// from the pass's store. |
| +func (r *rollbackRun) guard() hostguard.Guard { |
| + g := r.cfg().Guard |
| + if g.Claims == nil { |
| + g.Claims = hostguard.StoreReader{Store: r.s} |
| + } |
| + return g |
| +} |
| + |
| +// installation is the registered half of the local retirement (J15): with the |
| +// target's park-local and inventory paths of F, the installation is retired |
| +// through park.Local under the host guard, which completes the chore itself; |
| +// a park-local with nothing of F listed is completed; neither is an |
| +// unregistered target, whose artifacts are all the moves-owned rename's. |
| +func (r *rollbackRun) installation(ctx context.Context, ours bool) (string, error) { |
| + cfg := r.cfg() |
| + host, id := r.rec.Host, r.g.ID |
| + if !ours { |
| + return "moved-on", nil |
| + } |
| + obs, err := r.claim(ctx) |
| + if err != nil { |
| + return "", err |
| + } |
| + c := obs.Claim |
| + f, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalTargetStamp], 10, 64) |
| + if err != nil || f == 0 { |
| + return "", fmt.Errorf("move %s: the claim journals no target stamp", id) |
| + } |
| + var chore *hostclaim.Obligation |
| + for i := range c.Obligations { |
| + if o := &c.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box { |
| + chore = o |
| + } |
| + } |
| + registered := hostclaim.InventoryHasStamp(c, cfg.Box, f) |
| + switch { |
| + case chore != nil && registered: |
| + captured, err := hostguard.Capture(cfg.DataDir, host) |
| + if err != nil { |
| + return "", fmt.Errorf("the target's installation record: %w", err) |
| + } |
| + if captured.Incarnation != c.Holder.Incarnation || (captured.Stamp != 0 && captured.Stamp != f) { |
| + 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) |
| + } |
| + local := cfg.Local |
| + local.Guard, local.Claims = r.guard(), r.s |
| + if local.Box == "" { |
| + local.Box = cfg.Box |
| + } |
| + if local.Authorize == nil { |
| + local.Authorize = func(ctx context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error { |
| + if auth := cfg.ParkAuthorize; auth != nil { |
| + return auth(ctx, hostclaim.Observed{Host: host, Claim: c}, o) |
| + } |
| + return r.authorizeChore(c, o) |
| + } |
| + } |
| + if err := local.RetireInstallation(ctx, host, hostguard.Captured{Incarnation: captured.Incarnation, Stamp: f}, chore.ID, id); err != nil { |
| + return "", fmt.Errorf("retire the target's installation: %w", err) |
| + } |
| + log.Printf("moves: %s: the target's registered installation (stamp %d) is retired", host, f) |
| + return "retired", nil |
| + case chore != nil: |
| + if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.CompleteLocalPark(id, cfg.Box, f)); err != nil { |
| + return "", fmt.Errorf("complete the target's local park: %w", err) |
| + } |
| + return "completed", nil |
| + case registered: |
| + return "", fmt.Errorf("move %s: this box's inventory lists the target's paths, and the release queued no park-local for them", id) |
| + } |
| + return "unregistered", nil |
| +} |
| + |
| +// parkLeftovers is the moves-owned rename (H5(b), parkLocalArtifacts) of |
| +// every artifact the target still has at the flat paths (the installation's, |
| +// the rendered config and the sidecar's env), its plan and its renames inside |
| +// the host guard, which retires the installation record at F if it is still |
| +// active (an unregistered target), with the claim read fresh there: this box |
| +// no longer executes and no longer takes part in the claim's op, and no |
| +// inventory entry claims a path the rename would move. |
| +func (r *rollbackRun) parkLeftovers(ctx context.Context, j *moves.Journal, mark func(step, value string) error) error { |
| + if j.Done(rbStepLocal) { |
| + return nil |
| + } |
| + cfg := r.cfg() |
| + host, id := r.rec.Host, r.g.ID |
| + paths := sourceArtifactPaths(host) |
| + var incarnation string |
| + check := func(c *hostclaim.Claim) error { |
| + if c.Op.Exec == cfg.Box { |
| + return fmt.Errorf("%w: this box executes the host again", hostclaim.ErrWrongState) |
| + } |
| + if c.Op.ID == id { |
| + if err := r.ours(c); err != nil { |
| + return err |
| + } |
| + } else if containsString(c.Op.Participants, cfg.Box) { |
| + return fmt.Errorf("%w: this box takes part in the claim's op %s", hostclaim.ErrWrongState, c.Op.ID) |
| + } |
| + for _, p := range paths { |
| + if _, ok := c.Inventory[cfg.Box].Paths[p]; ok { |
| + return fmt.Errorf("%w: the inventory claims %s", hostclaim.ErrWrongState, p) |
| + } |
| + } |
| + incarnation = c.Holder.Incarnation |
| + return nil |
| + } |
| + work := func() error { |
| + if incarnation != r.g.Incarnation { |
| + return fmt.Errorf("move %s: the claim names another incarnation", id) |
| + } |
| + _, err := parkLocalArtifacts(ctx, j, mark, localArtifactPark{DataDir: cfg.DataDir, Incarnation: incarnation, Host: host, |
| + Op: id, Step: rbStepLocal, Paths: paths, NewID: obligationID, Fault: r.fault}) |
| + return err |
| + } |
| + g := r.guard() |
| + captured, err := hostguard.Capture(cfg.DataDir, host) |
| + switch { |
| + case errors.Is(err, hostguard.ErrNoRecord): |
| + return g.WithNoInstallation(ctx, host, r.g.Incarnation, check, work) |
| + case err != nil: |
| + return fmt.Errorf("the target's installation record: %w", err) |
| + } |
| + f := captured.Stamp |
| + if f == 0 { |
| + f = targetStamp(j) |
| + } |
| + if f == 0 { |
| + return fmt.Errorf("move %s: the target's retired installation stamp is not known", id) |
| + } |
| + return g.RetireInstallationWith(ctx, host, hostguard.Captured{Incarnation: r.g.Incarnation, Stamp: f}, check, work) |
| +} |
| + |
| +// deleteRow deletes the row and its timers rows in one registry transaction |
| +// (registry.DeleteIfHoldWithTimers, J12), only while the rollback holds the |
| +// row, under the host's hold lock with the record re-read there; the journal |
| +// says "pending" before and "done" after, and a replay that finds "pending" |
| +// and no row of this rollback's treats the delete as done. |
| +func (r *rollbackRun) deleteRow(ctx context.Context, j *moves.Journal, mark func(step, value string) error) error { |
| + if j.Steps[rbStepRowDeleted] == "done" { |
| + return nil |
| + } |
| + cfg := r.cfg() |
| + host, id := r.rec.Host, r.g.ID |
| + var err error |
| + provision.WithHostHoldLock(host, func() { |
| + fresh, e := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, host) |
| + if e != nil { |
| + err = fmt.Errorf("re-read the record: %w", e) |
| + return |
| + } |
| + if m := fresh.Move; m == nil || m.ID != id || m.Incarnation != r.g.Incarnation || m.State != provision.MoveRollingBack || |
| + m.Rollback == nil || m.Rollback.ID != r.rb || m.Rollback.TargetPhase != provision.RbTargetCleaning { |
| + err = fmt.Errorf("move %s: the record is no longer this rollback's cleaning", id) |
| + return |
| + } |
| + if j.Steps[rbStepRowDeleted] != "pending" { |
| + if err = mark(rbStepRowDeleted, "pending"); err != nil { |
| + return |
| + } |
| + } |
| + deleted, e := cfg.DeleteHeld(host, r.rb) |
| + if e != nil { |
| + err = fmt.Errorf("delete the target's row: %w", e) |
| + return |
| + } |
| + if deleted { |
| + log.Printf("moves: %s: the target's row and its timers are deleted", host) |
| + } |
| + err = mark(rbStepRowDeleted, "done") |
| + }) |
| + return err |
| +} |
| + |
| +// cleaned parks the op's journal into the local retirement, after the record |
| +// says cleaned: until then it is what routes a replay back here. |
| +func (r *rollbackRun) cleaned() error { |
| + cfg := r.cfg() |
| + host, id := r.rec.Host, r.g.ID |
| + if !moveJournalPresent(cfg.DataDir, host, id) { |
| + return nil |
| + } |
| + j, err := r.journal() |
| + if err != nil { |
| + return err |
| + } |
| + retirement := j.Steps[rbStepLocal] |
| + if retirement == "" { |
| + return fmt.Errorf("move %s: the journal records no local retirement for the target's park", id) |
| + } |
| + plan, err := journalledLocalPlan(j, rbStepLocal) |
| + if err != nil { |
| + return err |
| + } |
| + if err := parkOpJournal(cfg.DataDir, plan.Incarnation, host, id, retirement); err != nil { |
| + return fmt.Errorf("park the op journal: %w", err) |
| + } |
| + log.Printf("moves: %s: the target's half of rollback %s is complete", host, r.rb) |
| + return nil |
| +} |
amadan