| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "os" |
| 9 | "sort" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/hostclaim" |
| 13 | "github.com/carlosframework/platform/internal/lease" |
| 14 | "github.com/carlosframework/platform/internal/moves" |
| 15 | "github.com/carlosframework/platform/internal/park" |
| 16 | "github.com/carlosframework/platform/internal/provision" |
| 17 | "github.com/carlosframework/platform/internal/release" |
| 18 | ) |
| 19 | |
| 20 | // The abort (moves spec §7.11, joint N13; Plan 5 Task 1 with rulings K5, K6, |
| 21 | // K7, K19 and K21), behind CARLOS_MOVES_PHASE1. It replaces Plan 1's interim |
| 22 | // abort, which released the claim at once and left the staging lane in the |
| 23 | // target bucket with no owner. |
| 24 | // |
| 25 | // Who retires what. The source cannot read the target's moves/* nor write |
| 26 | // its parked/ (M17), so it never touches the lane: the lane is parked by the |
| 27 | // target region's box. The source retires only its own state. |
| 28 | // |
| 29 | // The source, for a record aborting in its region (phase1MoveRound): |
| 30 | // |
| 31 | // 1. with the claim at this move's moving/"" and this box executing, it |
| 32 | // takes the staging lease, so no staging writer is mid-round and none |
| 33 | // can sign another request (K5); |
| 34 | // 2. it journals abort_bound once: the later of now + SignatureLife + |
| 35 | // DefaultSkew and any seal's bound. Every staging request is signed at a |
| 36 | // check that leaves its whole signature life before the lease's |
| 37 | // SafeUntil, and none is signed after this moment while the lease is |
| 38 | // held, so no request can land after the bound. A replay reads the |
| 39 | // journalled bound and never computes a later one (revertClaim's |
| 40 | // pattern); |
| 41 | // 3. it CASes hostclaim.Abort with moves.AbortScope(host, op, inc, bound): |
| 42 | // moving/aborting, the lane's park-replica queued for the target region, |
| 43 | // owned by it. Execution stays with the source, which serves throughout; |
| 44 | // 4. only then it retires its own state: the move's snapshots in its own |
| 45 | // bucket, every version through DeleteVersion (K19: they are outside the |
| 46 | // park grammar, so "the source parks its own snapshots" means it |
| 47 | // retires them from its own bucket), then the receipts and the ledger, |
| 48 | // and last the op's journal directory (the bound is in the claim by |
| 49 | // then). It runs on every pass while the claim is aborting, before the |
| 50 | // release, and is a no-op once everything is gone: nothing infers from |
| 51 | // a missing journal that it already ran, and any error (a listing, a |
| 52 | // delete, the journal directory) ends the pass unreleased; |
| 53 | // 5. it waits until no obligation under the op remains, then clears the |
| 54 | // record with provision.MoveGuardedAbortComplete, and last releases |
| 55 | // with hostclaim.ReleaseAbort (K7: from moving/aborting only, so never |
| 56 | // before the lane park was queued). Record first, then the claim's |
| 57 | // release (§7.5 and §19 K8 as Paul's ruling M4 amends them): a crash |
| 58 | // between leaves the record at rest and the claim moving/aborting with |
| 59 | // nothing owed, and the console's recovery pass releases from LastMove |
| 60 | // (Plan 5 Task 7). A round that finds the claim already serving and |
| 61 | // recording this move's release (a binary that closed claim first) |
| 62 | // completes the record; any other serving claim is left alone. |
| 63 | // |
| 64 | // The target region's box, for a record aborting whose To is its region: |
| 65 | // movesParticipantRecord answers it as the target on the state alone (the |
| 66 | // record never flipped, so the target has neither the row nor a journal) |
| 67 | // and laneParkTarget runs the lane's park-replica through the park engine |
| 68 | // over its own bucket until the engine acknowledges it after the horizon, |
| 69 | // abort_bound + SignatureAge + RequestLimit. A late request found by the |
| 70 | // engine's verification listing is parked as a further retirement event. |
| 71 | // |
| 72 | // A dead target region (K21). If no box of the target region runs the lane |
| 73 | // park, the obligation stays owed, the claim stays moving/aborting and |
| 74 | // nothing releases. That blocks a delete (and any other operation) of the |
| 75 | // host until the region returns. Nothing needs to be done by hand; the |
| 76 | // overdue page is Task 8's. |
| 77 | // |
| 78 | // Arming. The source's half needs only what a preparing round already has. |
| 79 | // The target's lane park is unarmed in production, as the revert's is, |
| 80 | // until Task 9 wires the park engine's leases (revertLaneArmed); the tests |
| 81 | // inject it. Plan 1's DeletePrefix path stays for legacy records only |
| 82 | // (legacyMoveRound); this path never deletes from the target bucket. |
| 83 | |
| 84 | // abortStepBound is the abort's journalled writers' horizon (K5). |
| 85 | const abortStepBound = "abort_bound" |
| 86 | |
| 87 | // abFault is the abort's test seam after a side effect. |
| 88 | func (w *movesWorker) abFault(point string) error { |
| 89 | if f := w.cfg.abFault; f != nil { |
| 90 | return f(point) |
| 91 | } |
| 92 | return nil |
| 93 | } |
| 94 | |
| 95 | // abortSource is phase1MoveRound's MoveAborting arm. |
| 96 | func (w *movesWorker) abortSource(ctx context.Context, s release.Store, rec provision.Record) error { |
| 97 | cfg := w.cfg |
| 98 | id := rec.Move.ID |
| 99 | if cfg.Box == "" { |
| 100 | return errors.New("the abort needs this box's label") |
| 101 | } |
| 102 | obs, err := hostclaim.Read(ctx, s, rec.Host) |
| 103 | if err != nil { |
| 104 | return fmt.Errorf("read the host claim: %w", err) |
| 105 | } |
| 106 | if obs.Absent { |
| 107 | w.notice(rec.Host+"#abort", "moves: %s: aborting move %s has no host claim; left for the console pass", rec.Host, id) |
| 108 | return nil |
| 109 | } |
| 110 | op := obs.Claim.Op |
| 111 | mine := op.Kind == hostclaim.KindMoving && op.ID == id |
| 112 | releaseAfter := false |
| 113 | switch { |
| 114 | case mine && op.Exec != cfg.Box: |
| 115 | w.notice(rec.Host+"#abort", "moves: %s: aborting move %s: the claim's executor is %q, not this box; left for the console pass", rec.Host, id, op.Exec) |
| 116 | return nil |
| 117 | case mine && op.Phase == hostclaim.PhaseMoving: |
| 118 | ok, err := w.abortClaim(ctx, s, rec) |
| 119 | if err != nil || !ok { |
| 120 | return err |
| 121 | } |
| 122 | if obs, err = hostclaim.Read(ctx, s, rec.Host); err != nil { |
| 123 | return fmt.Errorf("read the host claim: %w", err) |
| 124 | } |
| 125 | if obs.Absent || obs.Claim.Op.ID != id || obs.Claim.Op.Phase != hostclaim.PhaseAborting { |
| 126 | return nil // another writer moved it; the next pass reads it again |
| 127 | } |
| 128 | fallthrough |
| 129 | case mine && op.Phase == hostclaim.PhaseAborting: |
| 130 | // Every pass, before the close: the retirement is idempotent, and |
| 131 | // a close never follows from inferring that it already ran (fix |
| 132 | // round 1). Any error, the journal directory's included, ends the |
| 133 | // pass with the record and the claim still aborting. |
| 134 | if err := w.retireOwnMoveState(ctx, rec); err != nil { |
| 135 | return err |
| 136 | } |
| 137 | // Nothing owed under the op, checked before the record's close |
| 138 | // (ruling M4), as the console's grant checks it. |
| 139 | if w.abortOwes(rec, obs) { |
| 140 | return nil |
| 141 | } |
| 142 | releaseAfter = true |
| 143 | case hostclaim.RequireReleasedMove(id)(obs.Claim) == nil: |
| 144 | // A binary that closed claim first released before a crash: |
| 145 | // finish the record. The own state went before the release; |
| 146 | // retiring it again is a no-op. |
| 147 | if err := w.retireOwnMoveState(ctx, rec); err != nil { |
| 148 | return err |
| 149 | } |
| 150 | default: |
| 151 | w.notice(rec.Host+"#abort", "moves: %s: aborting move %s: the claim is %s/%q (%s) and records no release of this move; left for the console pass", rec.Host, id, op.Kind, op.Phase, op.ID) |
| 152 | return nil |
| 153 | } |
| 154 | g := provision.MoveGuard{ID: id, Incarnation: rec.Move.Incarnation} |
| 155 | var pre provision.Record |
| 156 | after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, provision.MoveGuardedAbortComplete(g, cfg.Now()))) |
| 157 | if err == nil { |
| 158 | w.logTransition(ctx, s, pre, *after) |
| 159 | } |
| 160 | if err != nil { |
| 161 | if errors.Is(err, provision.ErrMoveConflict) { |
| 162 | log.Printf("moves: %s: abort of %s: the record moved on (%v)", rec.Host, id, err) |
| 163 | return nil |
| 164 | } |
| 165 | return err |
| 166 | } |
| 167 | nudgeSerial(ctx, s) |
| 168 | log.Printf("moves: %s: move %s aborted; the instance stays in %q", rec.Host, id, rec.Region) |
| 169 | if !releaseAfter { |
| 170 | return nil |
| 171 | } |
| 172 | if err := w.abFault("after-abort-record"); err != nil { |
| 173 | return err |
| 174 | } |
| 175 | return w.releaseAbort(ctx, s, rec) |
| 176 | } |
| 177 | |
| 178 | // abortClaim is steps 1 to 3: the staging lease, the journalled bound and |
| 179 | // the Abort CAS. ok is false when the pass should wait (another worker holds |
| 180 | // the lease). Without a deployment bucket no round ever took the lease and |
| 181 | // none staged, so there is no lease to take and the bound alone holds. |
| 182 | func (w *movesWorker) abortClaim(ctx context.Context, s release.Store, rec provision.Record) (bool, error) { |
| 183 | cfg := w.cfg |
| 184 | host, id := rec.Host, rec.Move.ID |
| 185 | if cfg.StagingLease != nil { |
| 186 | _, rel, err := cfg.StagingLease(ctx, rec.Move.To, host) |
| 187 | if err != nil { |
| 188 | var he *lease.HeldError |
| 189 | if errors.As(err, &he) { |
| 190 | w.notice(host+"#abort", "moves: %s: aborting move %s waits: another worker holds the staging lease", host, id) |
| 191 | return false, nil |
| 192 | } |
| 193 | return false, fmt.Errorf("staging lease: %w", err) |
| 194 | } |
| 195 | defer func() { |
| 196 | rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 197 | defer cancel() |
| 198 | if err := rel(rctx); err != nil { |
| 199 | log.Printf("moves: %s: release the staging lease: %v", host, err) |
| 200 | } |
| 201 | }() |
| 202 | } |
| 203 | // Under the lease, the claim again: still this move's, before the seal. |
| 204 | obs, err := hostclaim.Read(ctx, s, host) |
| 205 | if err != nil { |
| 206 | return false, fmt.Errorf("read the host claim: %w", err) |
| 207 | } |
| 208 | if obs.Absent || obs.Claim.Op.Kind != hostclaim.KindMoving || obs.Claim.Op.ID != id || obs.Claim.Op.Phase != hostclaim.PhaseMoving || obs.Claim.Op.Exec != cfg.Box { |
| 209 | return true, nil |
| 210 | } |
| 211 | if obs.Claim.Holder == nil { |
| 212 | return false, errors.New("the abort needs the claim's holder") |
| 213 | } |
| 214 | j, err := moves.ReadJournal(cfg.DataDir, host, id) |
| 215 | if err != nil { |
| 216 | return false, err |
| 217 | } |
| 218 | var bound time.Time |
| 219 | if raw, ok := j.Steps[abortStepBound]; ok { |
| 220 | if bound, err = time.Parse(time.RFC3339Nano, raw); err != nil { |
| 221 | bound = time.Time{} |
| 222 | } |
| 223 | } |
| 224 | if bound.IsZero() { |
| 225 | bound = cfg.Now().Add(moves.SignatureLife + moves.DefaultSkew).UTC() |
| 226 | if seal, err := journalledSeal(obs.Claim); err == nil && seal.Bound().After(bound) { |
| 227 | bound = seal.Bound().UTC() |
| 228 | } |
| 229 | j.Mark(abortStepBound, bound.Format(time.RFC3339Nano)) |
| 230 | if err := j.Commit(cfg.DataDir); err != nil { |
| 231 | return false, err |
| 232 | } |
| 233 | if err := w.abFault("after-abort-bound"); err != nil { |
| 234 | return false, err |
| 235 | } |
| 236 | } |
| 237 | scope := moves.AbortScope(host, id, obs.Claim.Holder.Incarnation, bound) |
| 238 | if _, err := hostclaim.Update(ctx, s, obs, hostclaim.Abort(id, park.PlacementRegion(rec.Move.To), scope, obligationID)); err != nil { |
| 239 | return false, fmt.Errorf("abort the claim: %w", err) |
| 240 | } |
| 241 | log.Printf("moves: %s: the claim aborts move %s; the staging lane in %s is queued for its park there", host, id, rec.Move.To) |
| 242 | if err := w.abFault("after-abort-claim"); err != nil { |
| 243 | return false, err |
| 244 | } |
| 245 | return true, nil |
| 246 | } |
| 247 | |
| 248 | // abortOwes is step 5's wait: true, with a notice, while an obligation of |
| 249 | // the op remains (the lane park the target region has not acknowledged). |
| 250 | func (w *movesWorker) abortOwes(rec provision.Record, obs hostclaim.Observed) bool { |
| 251 | id := rec.Move.ID |
| 252 | for _, o := range obs.Claim.Obligations { |
| 253 | if o.Op == id { |
| 254 | w.notice(rec.Host+"#abort-wait", "moves: %s: aborting move %s waits for %s to park the staging lane (%s %s)", rec.Host, id, rec.Move.To, o.Kind, o.Region+o.Box) |
| 255 | return true |
| 256 | } |
| 257 | } |
| 258 | return false |
| 259 | } |
| 260 | |
| 261 | // releaseAbort is step 5's second half: the record already says aborted, so |
| 262 | // hostclaim.ReleaseAbort from a fresh read. A failure is left to the |
| 263 | // console's recovery pass, which releases from LastMove. |
| 264 | func (w *movesWorker) releaseAbort(ctx context.Context, s release.Store, rec provision.Record) error { |
| 265 | if err := releaseAfterClose(ctx, s, rec.Host, rec.Move.ID, hostclaim.ReleaseAbort(rec.Move.ID, w.cfg.Box)); err != nil { |
| 266 | return err |
| 267 | } |
| 268 | return w.abFault("after-release-abort") |
| 269 | } |
| 270 | |
| 271 | // retireOwnMoveState is step 4: the move's snapshots in this box's own |
| 272 | // bucket, every version through DeleteVersion (K19), then its receipts and |
| 273 | // ledger, and last its journal directory, which holds the bound. Idempotent. |
| 274 | func (w *movesWorker) retireOwnMoveState(ctx context.Context, rec provision.Record) error { |
| 275 | cfg := w.cfg |
| 276 | if cfg.Versioned == nil { |
| 277 | return errors.New("the abort needs this box's versioned store to retire its snapshots") |
| 278 | } |
| 279 | vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion) |
| 280 | if err != nil { |
| 281 | return err |
| 282 | } |
| 283 | prefix := moves.SnapshotsPrefix(rec.Host, rec.Move.ID) |
| 284 | vers, err := vs.ListVersions(ctx, cfg.OwnBucket, prefix) |
| 285 | if err != nil { |
| 286 | return fmt.Errorf("list the move's snapshots: %w", err) |
| 287 | } |
| 288 | for _, v := range vers { |
| 289 | if err := vs.DeleteVersion(ctx, cfg.OwnBucket, v.Key, v.VersionID); err != nil { |
| 290 | return fmt.Errorf("delete %s version %s: %w", v.Key, v.VersionID, err) |
| 291 | } |
| 292 | } |
| 293 | own, err := w.mirror(ctx, cfg.OwnBucketRegion) |
| 294 | if err != nil { |
| 295 | return err |
| 296 | } |
| 297 | keys := []string{moves.ReceiptsKey(rec.Host, rec.Move.ID), moves.BuildStateReceiptsKey(rec.Host, rec.Move.ID), moves.LedgerKey(rec.Host, rec.Move.ID)} |
| 298 | sort.Strings(keys) |
| 299 | for _, k := range keys { |
| 300 | if err := own.Delete(ctx, cfg.OwnBucket, k); err != nil && !errors.Is(err, lease.ErrMirrorNotFound) { |
| 301 | return fmt.Errorf("delete %s: %w", k, err) |
| 302 | } |
| 303 | } |
| 304 | if err := w.abFault("after-abort-own-state"); err != nil { |
| 305 | return err |
| 306 | } |
| 307 | // The move's local directory: its journal and the archive kept for it. |
| 308 | if err := os.RemoveAll(moves.JournalDir(cfg.DataDir, rec.Host, rec.Move.ID)); err != nil { |
| 309 | return fmt.Errorf("remove the move's local directory: %w", err) |
| 310 | } |
| 311 | return w.abFault("after-abort-journal") |
| 312 | } |
| 313 | |
amadan