| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/hostclaim" |
| 10 | "github.com/carlosframework/platform/internal/moves" |
| 11 | "github.com/carlosframework/platform/internal/park" |
| 12 | "github.com/carlosframework/platform/internal/provision" |
| 13 | "github.com/carlosframework/platform/internal/registry" |
| 14 | "github.com/carlosframework/platform/internal/release" |
| 15 | ) |
| 16 | |
| 17 | // The commit and the open on the target (moves spec §7.5, §7.9 steps 1 and |
| 18 | // 2; Plan 3 Task 7; preflight H6 to H9). |
| 19 | // |
| 20 | // probed: the commit is the record's CAS (provision.MoveCommit), which |
| 21 | // refuses a recorded rollback and a commit after the landing's deadline |
| 22 | // (H8); that refusal is the commit-deadline rollback mark, converted at the |
| 23 | // pass's one rollback site. Then the committed arm runs in the same pass. |
| 24 | // |
| 25 | // The committed arm (H9), routed for the target by the post-flip participant |
| 26 | // rule, runs every pass while the record is committed: |
| 27 | // - the claim half: a claim lagging at probing is stepped to probed first |
| 28 | // (platform-1e m-2), then hostclaim.Commit (probed only) under |
| 29 | // KeepWriterSession, so the admitted probe session is the serving one |
| 30 | // (H3); idempotent once committed. The source's location park is scoped |
| 31 | // from the flip (H7), the staging lane from the seal's bound; |
| 32 | // - the open, conditional: under the host's hold lock, with the record and |
| 33 | // the claim re-read there and both committed, the probe's live activator |
| 34 | // entry is marked open (MarkProbeOpen), and only then is the hold |
| 35 | // released (ReleaseHeld), so an edge reload in between keeps the entry |
| 36 | // (Task 5's open race). The entry from the probe is the serving process: |
| 37 | // it is never restarted; |
| 38 | // - MoveOpened, the first time only; |
| 39 | // - the staging park and the close (§7.9 steps 4 and 5, Plan 5 Task 3, |
| 40 | // movesparks.go): the lane's park through laneParkTarget at |
| 41 | // landing/committed, then closeMove. A claim that already records this |
| 42 | // move's release (a crash between the close's two halves) skips the |
| 43 | // rest and only clears the record. |
| 44 | // |
| 45 | // Unarmed in production with the rest of the landing: landingArmed fails |
| 46 | // there until the wake-source gate opens (movesWakeSourceRegistered, Plan 5 |
| 47 | // Task 9b), and the open needs |
| 48 | // ReleaseHeld. |
| 49 | |
| 50 | // probed commits the landing: the record's CAS, then the committed arm. |
| 51 | func (l *landingRun) probed(ctx context.Context) (bool, error) { |
| 52 | obs, err := l.syncClaimPhase(ctx) |
| 53 | if err != nil { |
| 54 | return false, err |
| 55 | } |
| 56 | if err := l.requirePhase(obs, provision.LandProbed); err != nil { |
| 57 | return false, err |
| 58 | } |
| 59 | if err := l.fault("before-commit"); err != nil { |
| 60 | return false, err |
| 61 | } |
| 62 | var pre provision.Record |
| 63 | fresh, err := provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, l.rec.Host, observedMove(&pre, provision.MoveCommit(l.g, l.cfg().Now()))) |
| 64 | if err == nil { |
| 65 | l.w.logTransition(ctx, l.s, pre, *fresh) |
| 66 | } |
| 67 | switch { |
| 68 | case errors.Is(err, provision.ErrCutDeadline): |
| 69 | log.Printf("moves: %s: the commit of landing %s came after its deadline", l.rec.Host, l.g.ID) |
| 70 | return l.markRollback(ctx, provision.RollbackCommitDeadline) |
| 71 | case errors.Is(err, provision.ErrMoveConflict): |
| 72 | // A replay finds committed; anything else is another writer's. |
| 73 | if fresh, err = provision.GetRecord(ctx, l.s, l.rec.Account, l.rec.App, l.rec.Host); err != nil { |
| 74 | return false, fmt.Errorf("re-read the record: %w", err) |
| 75 | } |
| 76 | if fresh == nil || fresh.Move == nil || fresh.Move.ID != l.g.ID || fresh.Move.State != provision.MoveCommitted { |
| 77 | log.Printf("moves: %s: the commit of landing %s is refused; the record moved on", l.rec.Host, l.g.ID) |
| 78 | return false, nil |
| 79 | } |
| 80 | case err != nil: |
| 81 | return false, fmt.Errorf("the commit: %w", err) |
| 82 | } |
| 83 | log.Printf("moves: %s: landing %s is committed", l.rec.Host, l.g.ID) |
| 84 | if err := l.fault("after-record-committed"); err != nil { |
| 85 | return true, err |
| 86 | } |
| 87 | return true, l.w.committedTarget(ctx, l.s, *fresh) |
| 88 | } |
| 89 | |
| 90 | // committedTarget is the committed arm on the target (see the file comment). |
| 91 | func (w *movesWorker) committedTarget(ctx context.Context, s release.Store, rec provision.Record) error { |
| 92 | cfg := w.cfg |
| 93 | m := rec.Move |
| 94 | if m == nil || m.Legacy() || m.State != provision.MoveCommitted || m.To != cfg.Region || m.From == cfg.Region { |
| 95 | return nil |
| 96 | } |
| 97 | if !landingArmed(cfg) { |
| 98 | w.notice(rec.Host+"#committed", "moves: %s: the open of %s waits: this box's landing is not armed", rec.Host, m.ID) |
| 99 | return nil |
| 100 | } |
| 101 | l := &landingRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, guard: cfg.Guard} |
| 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 && hostclaim.RequireReleasedMove(m.ID)(obs.Claim) == nil { |
| 107 | // A binary that closed claim first released the claim and stopped |
| 108 | // before the record's half: only that is left (closeMove). |
| 109 | return w.closeMove(ctx, s, rec, l.fault) |
| 110 | } |
| 111 | if err := l.commitClaim(ctx); err != nil { |
| 112 | return err |
| 113 | } |
| 114 | if err := l.fault("after-claim-committed"); err != nil { |
| 115 | return err |
| 116 | } |
| 117 | if cfg.ReleaseHeld == nil { |
| 118 | w.notice(rec.Host+"#open", "moves: %s: the open of %s waits: this box cannot release the hold", rec.Host, m.ID) |
| 119 | } else { |
| 120 | stamped, err := l.open(ctx) |
| 121 | if err != nil { |
| 122 | return err |
| 123 | } |
| 124 | if stamped { |
| 125 | log.Printf("moves: %s: landing %s is open", rec.Host, m.ID) |
| 126 | nudgeSerial(ctx, s) |
| 127 | } |
| 128 | } |
| 129 | // §7.9 step 4: the lane's park the commit queued (movesparks.go). |
| 130 | if err := w.laneParkTarget(ctx, s, rec, hostclaim.KindLanding, hostclaim.PhaseCommitted); err != nil { |
| 131 | return err |
| 132 | } |
| 133 | // §7.9 step 5: the close, once both stamps are in and nothing is owed. |
| 134 | return w.closeMove(ctx, s, rec, l.fault) |
| 135 | } |
| 136 | |
| 137 | // commitClaim is the claim half of the commit (§7.5): written by whoever |
| 138 | // sees the record ahead. A claim already committed is done. |
| 139 | func (l *landingRun) commitClaim(ctx context.Context) error { |
| 140 | cfg := l.cfg() |
| 141 | id := l.g.ID |
| 142 | obs, err := l.claim(ctx) |
| 143 | if err != nil { |
| 144 | return err |
| 145 | } |
| 146 | op := obs.Claim.Op |
| 147 | if op.Kind != hostclaim.KindLanding { |
| 148 | return fmt.Errorf("move %s: the claim is %s/%q, not landing", id, op.Kind, op.Phase) |
| 149 | } |
| 150 | if op.Phase == hostclaim.PhaseLandingCommitted { |
| 151 | return nil |
| 152 | } |
| 153 | if cfg.Box == "" || op.Exec != cfg.Box { |
| 154 | return fmt.Errorf("move %s: the claim's executor is %q, not this box", id, op.Exec) |
| 155 | } |
| 156 | if op.Phase == hostclaim.PhaseLandingProbing { |
| 157 | // m-2: the commit follows probed; a claim lagging behind the |
| 158 | // record's probed is stepped there on the same session. |
| 159 | if obs, err = hostclaim.Update(ctx, l.s, obs, hostclaim.KeepWriterSession(id, hostclaim.SetMovePhase(id, hostclaim.KindLanding, hostclaim.PhaseLandingProbing, hostclaim.PhaseLandingProbed))); err != nil { |
| 160 | return fmt.Errorf("move %s: the claim's probed before the commit: %w", id, err) |
| 161 | } |
| 162 | } |
| 163 | source, staging, err := l.commitScopes(obs.Claim) |
| 164 | if err != nil { |
| 165 | return err |
| 166 | } |
| 167 | if _, err := hostclaim.Update(ctx, l.s, obs, hostclaim.KeepWriterSession(id, hostclaim.Commit(id, park.PlacementRegion(l.rec.Move.From), obs.Claim.Op.Intent["source"], cfg.parkRegion(), source, staging, obligationID))); err != nil { |
| 168 | return fmt.Errorf("move %s: the claim's commit: %w", id, err) |
| 169 | } |
| 170 | log.Printf("moves: %s: the claim of landing %s is committed", l.rec.Host, id) |
| 171 | return nil |
| 172 | } |
| 173 | |
| 174 | // commitScopes are the parks the commit queues: moves.CommitScopes, the one |
| 175 | // builder the console's pass uses too (Plan 5 Task 3, K9), so the box and |
| 176 | // the console write the same obligations. The source region's location is |
| 177 | // scoped from the flip (H7), the staging lane to the seal's bound. |
| 178 | func (l *landingRun) commitScopes(c *hostclaim.Claim) (source, staging hostclaim.ParkScope, err error) { |
| 179 | if l.rec.Move == nil { |
| 180 | return source, staging, errors.New("the commit's parks need the record's move") |
| 181 | } |
| 182 | return moves.CommitScopes(c, l.rec.Move.FlippedAt) |
| 183 | } |
| 184 | |
| 185 | // open is the conditional open, proved (fix round 1, finding 3): under the |
| 186 | // host's hold lock, with the record and the claim re-read and both |
| 187 | // committed, the probe's entry is marked open, the hold released, and then |
| 188 | // opened_at stamped, in that order. A replay is only a row already ready and |
| 189 | // unheld; a release that changed nothing is re-read and accepted only when |
| 190 | // the row is ready and unheld; anything else refuses and stamps nothing. |
| 191 | // stamped reports that this call wrote opened_at. |
| 192 | func (l *landingRun) open(ctx context.Context) (stamped bool, err error) { |
| 193 | cfg := l.cfg() |
| 194 | host, id := l.rec.Host, l.g.ID |
| 195 | provision.WithHostHoldLock(host, func() { |
| 196 | var fresh *provision.Record |
| 197 | if fresh, err = provision.GetRecord(ctx, l.s, l.rec.Account, l.rec.App, host); err != nil { |
| 198 | return |
| 199 | } |
| 200 | if fresh == nil || fresh.Move == nil || fresh.Move.ID != id || fresh.Move.Incarnation != l.g.Incarnation || fresh.Move.State != provision.MoveCommitted { |
| 201 | err = fmt.Errorf("move %s: the record is no longer committed", id) |
| 202 | return |
| 203 | } |
| 204 | var obs hostclaim.Observed |
| 205 | if obs, err = l.claim(ctx); err != nil { |
| 206 | return |
| 207 | } |
| 208 | if op := obs.Claim.Op; op.Kind != hostclaim.KindLanding || op.Phase != hostclaim.PhaseLandingCommitted { |
| 209 | err = fmt.Errorf("move %s: the claim is %s/%q, not committed", id, op.Kind, op.Phase) |
| 210 | return |
| 211 | } |
| 212 | var rt *registry.Route |
| 213 | if rt, err = cfg.Route(host); err != nil { |
| 214 | return |
| 215 | } |
| 216 | switch { |
| 217 | case rt == nil: |
| 218 | err = fmt.Errorf("move %s: this box has no row to open", id) |
| 219 | return |
| 220 | case rt.HoldOp == "" && rt.Status == "ready": |
| 221 | // open already: a replay |
| 222 | case rt.HoldOp == "": |
| 223 | err = fmt.Errorf("move %s: the row is unheld but %q, not open", id, rt.Status) |
| 224 | return |
| 225 | case rt.HoldOp != id: |
| 226 | err = fmt.Errorf("move %s: the row is held %s by %q", id, rt.Status, rt.HoldOp) |
| 227 | return |
| 228 | default: |
| 229 | if cfg.MarkProbeOpen != nil && !cfg.MarkProbeOpen(host, id) { |
| 230 | log.Printf("moves: %s: no live probe entry for %s; the next request wakes it under the commit", host, id) |
| 231 | } |
| 232 | var released bool |
| 233 | if released, err = cfg.ReleaseHeld(host, id); err != nil { |
| 234 | return |
| 235 | } |
| 236 | if !released { |
| 237 | if rt, err = cfg.Route(host); err != nil { |
| 238 | return |
| 239 | } |
| 240 | if rt == nil || rt.Status != "ready" || rt.HoldOp != "" { |
| 241 | err = fmt.Errorf("move %s: the release matched nothing and the row is not open", id) |
| 242 | return |
| 243 | } |
| 244 | } |
| 245 | } |
| 246 | // The release is proved (the row is ready and unheld): the probe's |
| 247 | // entry is opened, ending its idle protection (final review, ruling |
| 248 | // 3). A failed release returned above and confirmed nothing. |
| 249 | if cfg.ConfirmProbeOpen != nil { |
| 250 | cfg.ConfirmProbeOpen(host, id) |
| 251 | } |
| 252 | if fresh.Move.OpenedAt == nil { |
| 253 | var pre provision.Record |
| 254 | var opened *provision.Record |
| 255 | if opened, err = provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, host, observedMove(&pre, provision.MoveOpened(l.g, cfg.Now()))); err != nil { |
| 256 | err = fmt.Errorf("the record's opened_at: %w", err) |
| 257 | return |
| 258 | } |
| 259 | l.w.logTransition(ctx, l.s, pre, *opened) |
| 260 | stamped = true |
| 261 | } |
| 262 | }) |
| 263 | if err != nil { |
| 264 | return false, fmt.Errorf("open the row: %w", err) |
| 265 | } |
| 266 | return stamped, nil |
| 267 | } |
| 268 | |
amadan