amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "crypto/rand"
6 "crypto/sha256"
7 "encoding/hex"
8 "encoding/json"
9 "errors"
10 "fmt"
11 "io"
12 "log"
13 "os"
14 "path/filepath"
15 "strconv"
16 "strings"
17 "time"
18
19 "github.com/carlosframework/platform/internal/activator"
20 "github.com/carlosframework/platform/internal/hostclaim"
21 "github.com/carlosframework/platform/internal/lease"
22 "github.com/carlosframework/platform/internal/moves"
23 "github.com/carlosframework/platform/internal/park"
24 "github.com/carlosframework/platform/internal/provision"
25 "github.com/carlosframework/platform/internal/registry"
26 "github.com/carlosframework/platform/internal/release"
27 "github.com/superfly/ltx"
28)
29
30// The cut on the source (moves spec §5.3, §7.4, §7.5), behind
31// CARLOS_MOVES_PHASE1. One pass drives one cutting record as far as it can:
32// each phase does its work and then advances cut_phase by one guarded CAS,
33// and the next phase runs in the same pass on the re-read record. Nothing
34// is carried in memory between passes: every phase resumes from the
35// record, the claim, the deployment bucket, this box's own bucket and the
36// op journal (moves.Journal, <DataDir>/.moves/<host>/<op>/).
37//
38// Phase names (ruling G8). A phase is written when the previous phase's
39// work is complete, before its own: `stopping` before the hold and the
40// stop, `stopped` once the tree is proven stopped, `staged` once the final
41// round is sealed, `archived` once the seal names the archive. So a record
42// at X means "X's work may have begun, and everything before X is done",
43// and a replay of X redoes X's work from durable state. Each phase's
44// comment says what a replay finds.
45//
46// The deadline (§7.4): before any work, a phase past cut_deadline stops
47// the cut. Before anything stopped (admitted) that is back to ready; from
48// stopping on it is a revert, whose claim half follows the record once the
49// source has given up execution (revertClaim).
50
51// cutLandingWindow is the landing deadline the flip writes (now + 15 min).
52const cutLandingWindow = 15 * time.Minute
53
54// cutRequestLimit is the staging park's request limit, the longest a
55// staging request runs (lease's stagingRequestTimeout).
56const cutRequestLimit = moves.ParkRequestLimit
57
58// cutMaxDeferred is how many final rounds may defer before the cut reverts
59// (§7.4 staged: three deferred rounds).
60const cutMaxDeferred = 3
61
62// cutStepsPerPass bounds the phases one pass runs.
63const cutStepsPerPass = 8
64
65// Op journal steps the cut writes (moves.Journal).
66const (
67 cutStepEvidence = "evidence" // §5.3 steps 1 to 4, the StoppedReport JSON
68 cutStepCoverage = "coverage" // cutCoverage: the stopped database's position against N
69 cutStepArchive = "archive" // the cut archive's identity, before any PUT
70 cutStepRevertBound = "revert_bound" // the revert's staging park evidence_at
71)
72
73// cutCoverage is the journalled coverage decision (ruling 3): whether the
74// stopped database's own position equalled N after the final checkpoint.
75type cutCoverage struct {
76 N uint64 `json:"n"`
77 Covered bool `json:"covered"`
78}
79
80// cutRun is one pass's state for one cutting record.
81type cutRun struct {
82 w *movesWorker
83 s release.Store
84 fleet moveFleet
85 rec provision.Record
86 g provision.MoveGuard
87 // The staging lease and the guarded mirror, taken once per pass by the
88 // first phase that writes staging (ruling G6) and released at its end.
89 gm *guardedMirror
90 release func(context.Context) error
91}
92
93// runCut is the MoveCutting arm of phase1MoveRound.
94func (w *movesWorker) runCut(ctx context.Context, s release.Store, rec provision.Record, fleet moveFleet) error {
95 c := &cutRun{w: w, s: s, fleet: fleet, rec: rec, g: provision.MoveGuard{ID: rec.Move.ID, Incarnation: rec.Move.Incarnation}}
96 defer c.releaseLease()
97 for i := 0; i < cutStepsPerPass; i++ {
98 advanced, err := c.step(ctx)
99 if err != nil || !advanced {
100 return err
101 }
102 fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
103 if err != nil {
104 return fmt.Errorf("re-read the record: %w", err)
105 }
106 if fresh.Move == nil || fresh.Move.ID != c.g.ID || fresh.Move.Incarnation != c.g.Incarnation || fresh.Move.State != provision.MoveCutting {
107 return nil
108 }
109 c.rec = *fresh
110 }
111 return nil
112}
113
114func (c *cutRun) cfg() moveCfg { return c.w.cfg }
115
116// fault is the test seam after a side effect.
117func (c *cutRun) fault(point string) error {
118 if f := c.w.cfg.cutFault; f != nil {
119 return f(point)
120 }
121 return nil
122}
123
124// step runs the record's current phase once. advanced is true when it
125// wrote the record (a phase, back to ready, a revert or the flip).
126func (c *cutRun) step(ctx context.Context) (bool, error) {
127 m := c.rec.Move
128 if m.State != provision.MoveCutting {
129 return false, nil
130 }
131 if m.CutDeadline != nil && c.cfg().Now().After(*m.CutDeadline) {
132 switch m.CutPhase {
133 case provision.CutAdmitted:
134 return c.backToReady(ctx, "the cut's deadline passed before anything stopped")
135 case provision.CutStopping:
136 return c.revert(ctx, provision.RevertStopEvidenceBudget)
137 case provision.CutStopped, provision.CutStaged, provision.CutArchived:
138 return c.revert(ctx, provision.RevertCutDeadline)
139 }
140 }
141 switch m.CutPhase {
142 case provision.CutRequested:
143 return c.requested(ctx)
144 case provision.CutAdmitted:
145 return c.admitted(ctx)
146 case provision.CutStopping:
147 return c.stopping(ctx)
148 case provision.CutStopped:
149 return c.stopped(ctx)
150 case provision.CutStaged:
151 return c.staged(ctx)
152 case provision.CutArchived:
153 return c.archived(ctx)
154 }
155 return false, nil
156}
157
158// cutTransition writes one guarded record transition. A conflict is
159// someone else's decision (an abort, a console write): logged, and the
160// pass ends without error.
161func (c *cutRun) cutTransition(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) {
162 var pre provision.Record
163 after, err := provision.SetRecordMove(ctx, c.s, c.rec.Account, c.rec.App, c.rec.Host, observedMove(&pre, mut))
164 if err != nil {
165 if errors.Is(err, provision.ErrMoveConflict) {
166 log.Printf("moves: %s: %s of %s refused (%v); the record moved on", c.rec.Host, what, c.g.ID, err)
167 return false, nil
168 }
169 return false, fmt.Errorf("%s: %w", what, err)
170 }
171 c.w.logTransition(ctx, c.s, pre, *after)
172 return true, nil
173}
174
175func (c *cutRun) advance(ctx context.Context, from, to string, stamp func(*provision.MoveState)) (bool, error) {
176 ok, err := c.cutTransition(ctx, "the cut "+from+" to "+to, provision.MoveCutPhase(c.g, from, to, stamp))
177 if ok {
178 log.Printf("moves: %s: cut %s is %s", c.rec.Host, c.g.ID, to)
179 }
180 return ok, err
181}
182
183// backToReady undoes an admission whose checks are not fresh or whose
184// inputs moved (§7.4 admitted): nothing has stopped. The pin is cleared and
185// the next passing readiness round pins again (G1).
186func (c *cutRun) backToReady(ctx context.Context, why string) (bool, error) {
187 ok, err := c.cutTransition(ctx, "back to ready", provision.MoveBackToReady(c.g))
188 if ok {
189 log.Printf("moves: %s: cut %s goes back to ready: %s", c.rec.Host, c.g.ID, why)
190 nudgeSerial(ctx, c.s)
191 }
192 return ok, err
193}
194
195// revert starts the pre-flip recovery (§7.4 reverting): the record first,
196// then the claim half once execution is none (revertClaim).
197func (c *cutRun) revert(ctx context.Context, reason string) (bool, error) {
198 ok, err := c.cutTransition(ctx, "revert", provision.MoveRevert(c.g, reason))
199 if !ok {
200 return ok, err
201 }
202 log.Printf("moves: %s: cut %s reverts: %s", c.rec.Host, c.g.ID, reason)
203 nudgeSerial(ctx, c.s)
204 if err := c.w.revertClaim(ctx, c.s, c.rec); err != nil {
205 // The record decided; the next pass's reverting arm writes the
206 // claim half (phase1MoveRound).
207 log.Printf("moves: %s: the claim half of the revert waits: %v", c.rec.Host, err)
208 }
209 return true, nil
210}
211
212// requested (§7.4): one source round under the staging lease, with a fresh
213// source report (G13), then Ready on that round and the target's report,
214// then the pinned inputs against the current ones; the admission stamps
215// cut_at, the budget and the deadline. Any check that fails goes back to
216// ready, so readiness reruns. A replay finds requested and runs the round
217// again: nothing here is written but the reports and the lane, which every
218// round rewrites.
219func (c *cutRun) requested(ctx context.Context) (bool, error) {
220 cfg := c.cfg()
221 if refused, wrote, err := c.refuseUnarmed(ctx); refused || err != nil {
222 return wrote, err
223 }
224 sr, published, err := c.w.sourceRound(ctx, c.s, c.rec, c.fleet)
225 if err != nil || published == nil {
226 return false, err
227 }
228 if !sourcePassed(published) {
229 return c.backToReady(ctx, "this round's source report does not pass")
230 }
231 target, err := readTargetReport(ctx, c.s, c.rec)
232 if err != nil {
233 return false, err
234 }
235 if ok, why := moves.Ready(sr.deltaBytes, cfg.ReadyDelta, target, published, cfg.Now()); !ok {
236 return c.backToReady(ctx, why)
237 }
238 if same, err := c.inputsHold(ctx); err != nil || !same {
239 if err != nil {
240 return false, err
241 }
242 return c.backToReady(ctx, "the inputs pinned at readiness have changed")
243 }
244 if refused, wrote, err := c.refuseHolder(ctx); refused || err != nil {
245 return wrote, err
246 }
247 now := cfg.Now()
248 budget := target.CutBudgetS
249 return c.cutTransition(ctx, "the admission", provision.MoveAdmit(c.g, now, budget, moves.CutDeadline(now, float64(budget)), target.AttemptsAdmitted))
250}
251
252// admitted (§7.4): the pinned inputs still match the current bundle and
253// overlay versions, the release and the launch report, and the claim
254// holder is one this box can stop under the writer's lease (ruling
255// T5-lease), or back to ready. A replay finds admitted and compares again;
256// nothing has stopped.
257func (c *cutRun) admitted(ctx context.Context) (bool, error) {
258 same, err := c.inputsHold(ctx)
259 if err != nil {
260 return false, err
261 }
262 if !same {
263 return c.backToReady(ctx, "the inputs pinned at readiness have changed")
264 }
265 if refused, wrote, err := c.refuseHolder(ctx); refused || err != nil {
266 return wrote, err
267 }
268 if refused, wrote, err := c.refuseUnarmed(ctx); refused || err != nil {
269 return wrote, err
270 }
271 return c.advance(ctx, provision.CutAdmitted, provision.CutStopping, nil)
272}
273
274// refuseUnarmed is the cut guard (Paul's ruling, 2026-10-11): a cut starts
275// only when the target's latest report states that its box can land the
276// move and roll it back (T13, moves.TargetReport.Armed) and this box can
277// recover the move if the cut fails (recoverySourceArmed). Otherwise the
278// cut goes back to ready with the reason recorded, before this box takes
279// the staging lease, holds or stops anything, or seals the claim: it runs
280// first at requested and last at admitted, the step before stopping. It
281// fails closed: no report, or an older box's report with no T13, is
282// unarmed. The refusal stands while the cause does: decideReady pins no
283// move again while this box cannot recover, and the target's failing T13
284// keeps Ready from pinning it (§7.3), so nothing loops and the operator
285// reads why in `carlos move status`.
286//
287// refused and wrote are apart (fix round 1): refused is the guard's
288// verdict, wrote whether the refusal's record write landed. A refused cut
289// ends the step either way. When another writer moved the record first
290// (ErrMoveConflict, which cutTransition logs and answers false) nothing
291// was written, the pass ends, and the next pass reads the record again;
292// the cut never carries on to the source's round or to stopping.
293func (c *cutRun) refuseUnarmed(ctx context.Context) (refused, wrote bool, err error) {
294 reason := ""
295 if !recoverySourceArmed(c.cfg()) {
296 reason = provision.RefusedSourceUnarmed
297 } else {
298 target, err := readTargetReport(ctx, c.s, c.rec)
299 if err != nil {
300 return false, false, err
301 }
302 if !target.Armed() {
303 reason = provision.RefusedTargetUnarmed
304 }
305 }
306 if reason == "" {
307 return false, false, nil
308 }
309 log.Printf("moves: %s: cut %s refused: %s", c.rec.Host, c.g.ID, reason)
310 if err := c.fault("refusal"); err != nil {
311 return true, false, err
312 }
313 wrote, err = c.cutTransition(ctx, "the refusal", provision.MoveBackToReadyRefused(c.g, reason))
314 if wrote {
315 nudgeSerial(ctx, c.s)
316 }
317 return true, wrote, err
318}
319
320// refuseHolder reads the claim fresh and, when its holder is not at the
321// activator's lease location (holderSupported), sends the cut back to
322// ready with RefusedReplicaLayout recorded (ruling T5-lease). The
323// refusal stands as long as the claim does: a passing round pins again,
324// the next cut is refused the same way, and the operator reads why.
325//
326// refused and wrote are apart, as in refuseUnarmed (fix round 1): a
327// refused cut ends the step whether or not the refusal's record write
328// landed. A write another writer beat (ErrMoveConflict) writes nothing,
329// the pass ends, and the next pass reads the record again.
330func (c *cutRun) refuseHolder(ctx context.Context) (refused, wrote bool, err error) {
331 obs, err := c.claim(ctx)
332 if err != nil {
333 return false, false, err
334 }
335 if err := c.holderSupported(*obs.Claim.Holder); err != nil {
336 log.Printf("moves: %s: cut %s refused: %v", c.rec.Host, c.g.ID, err)
337 if err := c.fault("holder-refusal"); err != nil {
338 return true, false, err
339 }
340 wrote, err := c.cutTransition(ctx, "the refusal", provision.MoveBackToReadyRefused(c.g, provision.RefusedReplicaLayout))
341 if wrote {
342 nudgeSerial(ctx, c.s)
343 }
344 return true, wrote, err
345 }
346 return false, false, nil
347}
348
349// inputsHold compares the record's pin with the current inputs.
350func (c *cutRun) inputsHold(ctx context.Context) (bool, error) {
351 cur, err := c.w.currentInputs(ctx, c.s, c.rec)
352 if err != nil {
353 return false, err
354 }
355 return sameInputs(c.rec.Move.Inputs, cur), nil
356}
357
358// currentInputs are the inputs the admission compares (§7.4 admitted): the
359// bundle and overlay versions, the live release and the launch report's
360// digest, as this box would publish them now. The digest is the
361// move-portable one (moves.PortableDigest), since the pin it is compared
362// with crosses to the target.
363func (w *movesWorker) currentInputs(ctx context.Context, s release.Store, rec provision.Record) (provision.MoveInputs, error) {
364 return w.inputsDigested(ctx, s, rec, moves.PortableDigest)
365}
366
367// recoveryInputs are the inputs a recovery probe records and then checks on
368// this same box (movesrecoveryprobe.go): currentInputs' fields, with the
369// digest of the whole composed environment (moves.LocalDigest), box-owned
370// variables included. The record and the check are made on one box just
371// before the wake, so the box's own variables are part of what runs and
372// must not change between them either (fix round 2).
373func (w *movesWorker) recoveryInputs(ctx context.Context, s release.Store, rec provision.Record) (provision.MoveInputs, error) {
374 return w.inputsDigested(ctx, s, rec, func(key []byte, env map[string]string, _ map[string]bool) string {
375 return moves.LocalDigest(key, env)
376 })
377}
378
379func (w *movesWorker) inputsDigested(ctx context.Context, s release.Store, rec provision.Record, digest func(key []byte, env map[string]string, instance map[string]bool) string) (provision.MoveInputs, error) {
380 cfg := w.cfg
381 if cfg.InputVersions == nil || cfg.LaunchEnv == nil {
382 return provision.MoveInputs{}, errors.New("no way to read the current inputs on this box")
383 }
384 bundle, overlay, err := cfg.InputVersions(ctx, s, rec)
385 if err != nil {
386 return provision.MoveInputs{}, fmt.Errorf("read the configuration versions: %w", err)
387 }
388 live, err := release.LiveVersion(cfg.OptDir, rec.Host)
389 if err != nil {
390 return provision.MoveInputs{}, fmt.Errorf("read the live release: %w", err)
391 }
392 key, err := moves.ReportKeyFromSecrets(cfg.SecretsKey)
393 if err != nil {
394 return provision.MoveInputs{}, err
395 }
396 env, instance, err := cfg.LaunchEnv(rec.Host)
397 if err != nil {
398 return provision.MoveInputs{}, fmt.Errorf("compose the launch environment: %w", err)
399 }
400 // The probe-only variables are dropped as the owners' checks drop them
401 // (launchVars): the probe sets its own, whatever the files say.
402 vars := make(map[string]string, len(env))
403 for k, v := range env {
404 if !probeOnlyVars[k] {
405 vars[k] = v
406 }
407 }
408 return provision.MoveInputs{BundleETags: bundle, OverlayETag: overlay, Release: live, EnvReport: digest(key, vars, instance)}, nil
409}
410
411// sameInputs is the admission's comparison: every bundle etag, the overlay
412// etag, the release and the launch report. The files digest is not one of
413// them: a file written before the stop is carried by the cut's own archive.
414func sameInputs(pin *provision.MoveInputs, cur provision.MoveInputs) bool {
415 if pin == nil || pin.OverlayETag != cur.OverlayETag || pin.Release != cur.Release || pin.EnvReport != cur.EnvReport {
416 return false
417 }
418 if len(pin.BundleETags) != len(cur.BundleETags) {
419 return false
420 }
421 for k, v := range pin.BundleETags {
422 if cv, ok := cur.BundleETags[k]; !ok || cv != v {
423 return false
424 }
425 }
426 return true
427}
428
429// claim reads the host claim and checks it is this move's moving op for
430// the record's incarnation.
431func (c *cutRun) claim(ctx context.Context) (hostclaim.Observed, error) {
432 obs, err := hostclaim.Read(ctx, c.s, c.rec.Host)
433 if err != nil {
434 return obs, fmt.Errorf("read the host claim: %w", err)
435 }
436 if obs.Absent {
437 return obs, fmt.Errorf("move %s has no host claim", c.g.ID)
438 }
439 if err := hostclaim.RequireOp(c.g.ID)(obs.Claim); err != nil {
440 return obs, fmt.Errorf("move %s: %w", c.g.ID, err)
441 }
442 if h := obs.Claim.Holder; h == nil || h.Incarnation != c.rec.Move.Incarnation {
443 return obs, fmt.Errorf("move %s: the host claim names another incarnation", c.g.ID)
444 }
445 return obs, nil
446}
447
448// dbPath is the row's database path, or the flat default.
449func (c *cutRun) dbPath() string {
450 cfg := c.cfg()
451 if cfg.Route != nil {
452 if rt, err := cfg.Route(c.rec.Host); err == nil && rt != nil && rt.DBPath != "" {
453 return rt.DBPath
454 }
455 }
456 return provision.DefaultDBPath(cfg.DataDir, c.rec.Host)
457}
458
459func (c *cutRun) journal() (*moves.Journal, error) {
460 return moves.ReadJournal(c.cfg().DataDir, c.rec.Host, c.g.ID)
461}
462
463// stopping (§5.3, §7.4): the row held moving under the move id, the wake
464// fence, the timer drain, then the evidence of steps 1 to 4. A tree not
465// proven stopped waits for the next pass, until the deadline reverts with
466// "stop evidence budget". Proven, the evidence goes into the op journal
467// (local: the report's replica step, N and digests come at stopped) and
468// the phase advances.
469//
470// A replay finds stopping and runs it all again from the box's state:
471// every step is idempotent (the hold under the same op, the fence at the
472// same generation, the drain, and Evidence, which reads only durable
473// state). No in-memory state is needed (Review Focus 1).
474func (c *cutRun) stopping(ctx context.Context) (bool, error) {
475 cfg := c.cfg()
476 host := c.rec.Host
477 obs, err := c.claim(ctx)
478 if err != nil {
479 return false, err
480 }
481 if err := hostclaim.RequireMoveOp(c.g.ID, hostclaim.KindMoving, hostclaim.PhaseMoving)(obs.Claim); err != nil {
482 return false, fmt.Errorf("move %s: %w", c.g.ID, err)
483 }
484 if cfg.Box == "" || obs.Claim.Op.Exec != cfg.Box {
485 return false, fmt.Errorf("move %s: the claim's executor is %q, not this box", c.g.ID, obs.Claim.Op.Exec)
486 }
487 // Admission refused any other holder; a claim that differs now is an
488 // error, never a stop (fail closed).
489 if err := c.holderSupported(*obs.Claim.Holder); err != nil {
490 return false, fmt.Errorf("move %s: %w", c.g.ID, err)
491 }
492 if cfg.HoldMoving == nil || cfg.Fence == nil || cfg.TimersDrained == nil || cfg.Evidence == nil {
493 return false, errors.New("this box cannot stop an instance for a move: a hook is not wired")
494 }
495 if err := cfg.HoldMoving(host, c.g.ID); err != nil {
496 return false, fmt.Errorf("hold the row: %w", err)
497 }
498 gen := moveFenceGeneration(c.rec.Move)
499 if err := cfg.Fence(ctx, host, gen); err != nil {
500 return false, fmt.Errorf("fence against wakes: %w", err)
501 }
502 c.w.fenced.Store(host, gen)
503 if err := cfg.TimersDrained(ctx, host); err != nil {
504 return false, fmt.Errorf("drain the timers: %w", err)
505 }
506 rep, err := cfg.Evidence(ctx, host, obs.Claim.Holder.Location, []string{c.dbPath()})
507 if err != nil {
508 return false, fmt.Errorf("stop evidence: %w", err)
509 }
510 if !rep.TreeStopped() {
511 c.w.notice(host+"#evidence", "moves: %s: the instance is not proven stopped yet (%s); retrying until the cut's deadline", host, failedSteps(rep))
512 if strayRefused(rep) {
513 c.w.pageStray(ctx, c.rec)
514 }
515 return false, nil
516 }
517 j, err := c.journal()
518 if err != nil {
519 return false, err
520 }
521 body, err := json.Marshal(rep)
522 if err != nil {
523 return false, err
524 }
525 j.Phase = provision.CutStopping
526 j.Mark(cutStepEvidence, string(body))
527 if err := j.Commit(cfg.DataDir); err != nil {
528 return false, err
529 }
530 if err := c.fault("after-evidence"); err != nil {
531 return false, err
532 }
533 return c.advance(ctx, provision.CutStopping, provision.CutStopped, nil)
534}
535
536// failedSteps names the evidence steps that did not pass, for the log.
537func failedSteps(r moves.StoppedReport) string {
538 var out []string
539 for _, id := range []string{moves.StepTree, moves.StepStray, moves.StepFDs, moves.StepLease} {
540 if c, ok := r.Steps[id]; !ok || !c.OK {
541 out = append(out, id)
542 }
543 }
544 return strings.Join(out, ", ")
545}
546
547// stopped (§5.3 step 5, §7.4): the stopped source's report, its execution
548// given up, then the final round and the seal.
549//
550// While the claim's executor is still this box: the final checkpoint under
551// the claim's lineage (G3), N, the database and timers digests, the files'
552// plain digest, the complete stopped-source.json (create-only, K13), the
553// record's source baselines, and then exec to none with the cut's session
554// in one claim write (K3). Once exec is none, a replay reads the published
555// report instead: nothing on the source's replica is written again.
556//
557// Then, under the staging lease: the timers snapshot into timers/ and the
558// final mirror round into replica/. A round that deferred or moved more
559// than the bar is counted in the ledger (three revert); a round that
560// cannot name a usable version for every current key runs again. The seal
561// goes into the claim with the op's phase sealed in one CAS, and only then
562// does the record say staged (G8).
563//
564// A replay finds stopped and repeats from the claim's state: the
565// checkpoint is idempotent on a stopped database, the report and the
566// snapshots are create-only and read back, every claim write is
567// conditional, and the final round rewrites staging, which the seal pins by
568// version.
569func (c *cutRun) stopped(ctx context.Context) (bool, error) {
570 cfg := c.cfg()
571 obs, err := c.claim(ctx)
572 if err != nil {
573 return false, err
574 }
575 op := obs.Claim.Op
576 if op.Kind != hostclaim.KindMoving || (op.Phase != hostclaim.PhaseMoving && op.Phase != hostclaim.PhaseSealed) {
577 return false, fmt.Errorf("move %s: the claim is %s/%q", c.g.ID, op.Kind, op.Phase)
578 }
579 holder := *obs.Claim.Holder
580 if err := c.holderSupported(holder); err != nil {
581 return false, fmt.Errorf("move %s: %w", c.g.ID, err)
582 }
583 stoppedKey := moves.StoppedKey(moves.ReportPrefix(c.rec.Account, c.rec.App, c.rec.Host, c.g.ID), "source")
584 var report moves.StoppedReport
585 switch {
586 case cfg.Box != "" && op.Exec == cfg.Box:
587 if report, err = c.giveUpExecution(ctx, obs, stoppedKey); err != nil {
588 return false, err
589 }
590 if report.V == 0 {
591 return false, nil // reverted, or waiting
592 }
593 case op.Exec == "" && op.Journal[hostclaim.JournalEvidenceSource] == stoppedKey:
594 if report, err = readStoppedReport(ctx, c.s, stoppedKey); err != nil {
595 return false, err
596 }
597 default:
598 return false, fmt.Errorf("move %s: the claim's executor is %q and cites %q", c.g.ID, op.Exec, op.Journal[hostclaim.JournalEvidenceSource])
599 }
600 return c.finalRound(ctx, holder, report, stoppedKey)
601}
602
603// giveUpExecution is stopped's first half, run while this box executes.
604// It returns a zero report when the phase ended without one (a revert, or
605// a file that changed).
606//
607// A revert's entry runs it too (revertSource, Plan 4 Task 6, ruling J7), on a
608// record already reverting: it then records no baselines (MoveRecordSource is
609// a cut's, at stopped; a revert from stopping has none, J8), and a files
610// directory that cannot be archived is no reason to revert again: the report
611// carries no files digest, and nothing compares one (§7.4 evidence).
612func (c *cutRun) giveUpExecution(ctx context.Context, obs hostclaim.Observed, stoppedKey string) (moves.StoppedReport, error) {
613 cfg := c.cfg()
614 host := c.rec.Host
615 holder := *obs.Claim.Holder
616 reverting := c.rec.Move.State == provision.MoveReverting
617 j, err := c.journal()
618 if err != nil {
619 return moves.StoppedReport{}, err
620 }
621 var ev moves.StoppedReport
622 if raw, ok := j.Steps[cutStepEvidence]; !ok || json.Unmarshal([]byte(raw), &ev) != nil || !ev.TreeStopped() {
623 // The journal lost the evidence (a rebuilt data directory): the row
624 // is still held and fenced, so ask the box again.
625 if cfg.Evidence == nil {
626 return moves.StoppedReport{}, errors.New("no stop evidence on this box")
627 }
628 if ev, err = cfg.Evidence(ctx, host, holder.Location, []string{c.dbPath()}); err != nil {
629 return moves.StoppedReport{}, fmt.Errorf("stop evidence: %w", err)
630 }
631 if !ev.TreeStopped() {
632 return moves.StoppedReport{}, fmt.Errorf("the instance is no longer proven stopped (%s)", failedSteps(ev))
633 }
634 }
635 own, vs, err := c.ownStores(ctx)
636 if err != nil {
637 return moves.StoppedReport{}, err
638 }
639 prefix, keep := moves.ReplicaSource(holder)
640 dbPath := c.dbPath()
641 fromDB, err := regularFile(dbPath)
642 if err != nil {
643 return moves.StoppedReport{}, err
644 }
645 if fromDB {
646 if cfg.FinalCheckpoint == nil {
647 return moves.StoppedReport{}, errors.New("no final checkpoint on this box")
648 }
649 check := func(ctx context.Context) error { return c.stillExecuting(ctx) }
650 if err := cfg.FinalCheckpoint(ctx, host, dbPath, holder, check); err != nil {
651 return moves.StoppedReport{}, fmt.Errorf("final checkpoint: %w", err)
652 }
653 }
654 if err := c.fault("after-checkpoint"); err != nil {
655 return moves.StoppedReport{}, err
656 }
657 n, err := c.finalTXID(ctx, own, prefix, keep)
658 if err != nil {
659 return moves.StoppedReport{}, err
660 }
661 var dbSHA string
662 if fromDB {
663 if dbSHA, err = fileSHA256(dbPath); err != nil {
664 return moves.StoppedReport{}, fmt.Errorf("digest the stopped database: %w", err)
665 }
666 // Coverage (ruling 3): the stopped database's own position against
667 // N, decided here, with the database present after its checkpoint,
668 // and journalled so the seal can still state it if the disk is lost
669 // before the final round. Without a database nothing is journalled
670 // and the seal says uncovered.
671 if cfg.LocalPosition == nil {
672 return moves.StoppedReport{}, errors.New("no local position on this box")
673 }
674 pos, err := cfg.LocalPosition(dbPath)
675 if err != nil {
676 return moves.StoppedReport{}, fmt.Errorf("read the stopped database's position: %w", err)
677 }
678 body, err := json.Marshal(cutCoverage{N: n, Covered: pos == n})
679 if err != nil {
680 return moves.StoppedReport{}, err
681 }
682 j.Mark(cutStepCoverage, string(body))
683 if err := j.Commit(cfg.DataDir); err != nil {
684 return moves.StoppedReport{}, err
685 }
686 }
687 timersSHA, err := c.snapshotTimers(ctx, own, vs)
688 if err != nil {
689 return moves.StoppedReport{}, err
690 }
691 dir, envPath := cfg.FilesDir(host)
692 plain, _, _, err := archivePlainDigest(dir, envPath)
693 if why, retry, ferr := archiveFailure(host, err); ferr != nil || why != "" || retry {
694 switch {
695 case why != "" && reverting:
696 log.Printf("moves: %s: the revert's stopped report carries no files digest: %s", host, why)
697 plain = ""
698 case why != "":
699 _, rerr := c.revert(ctx, provision.RevertArchiveRefused)
700 return moves.StoppedReport{}, rerr
701 default:
702 return moves.StoppedReport{}, ferr
703 }
704 }
705 var rt registry.Route
706 if cfg.Route != nil {
707 if r, err := cfg.Route(host); err != nil {
708 return moves.StoppedReport{}, fmt.Errorf("read the registry row: %w", err)
709 } else if r != nil {
710 rt = *r
711 }
712 }
713 report := ev
714 report.V, report.Role, report.Box, report.Incarnation, report.ClaimGen = 1, "source", cfg.Box, holder.Incarnation, obs.Claim.Gen
715 report.Host, report.OpID = host, c.g.ID
716 report.Steps = map[string]moves.Check{}
717 for k, v := range ev.Steps {
718 report.Steps[k] = v
719 }
720 report.Steps[moves.StepReplica] = moves.Check{OK: true}
721 report.FinalTXID, report.DBSHA256, report.FilesPlain, report.TimersSHA256 = n, dbSHA, plain, timersSHA
722 if !report.Complete() {
723 return moves.StoppedReport{}, fmt.Errorf("the stopped report is not complete (%s)", failedSteps(report))
724 }
725 // K13: create-only. A report already there is this cut's own replay
726 // when it names the same host, op, role and N; it is then the report.
727 if report, err = publishStoppedReport(ctx, c.s, stoppedKey, report); err != nil {
728 return moves.StoppedReport{}, err
729 }
730 if err := c.fault("after-report"); err != nil {
731 return moves.StoppedReport{}, err
732 }
733 if !reverting {
734 src := provision.MoveSource{Replication: rt.Replication, NextDue: rt.NextDue, FinalTXID: report.FinalTXID, DBSHA256: report.DBSHA256,
735 Files: &provision.MoveFilesDigest{Plain: report.FilesPlain}, TimersSHA256: report.TimersSHA256, Evidence: stoppedKey}
736 if ok, err := c.cutTransition(ctx, "the source baselines", provision.MoveRecordSource(c.g, src)); !ok {
737 if err == nil {
738 err = errors.New("the record moved on")
739 }
740 return moves.StoppedReport{}, err
741 }
742 if err := c.fault("after-record-source"); err != nil {
743 return moves.StoppedReport{}, err
744 }
745 }
746 session := strconv.FormatUint(holder.Session, 10)
747 if _, err := hostclaim.Update(ctx, c.s, obs, func(cl *hostclaim.Claim) error {
748 if err := hostclaim.ExecToNone(c.g.ID, cfg.Box, stoppedKey, hostclaim.JournalEvidenceSource)(cl); err != nil {
749 return err
750 }
751 return hostclaim.SetJournal(c.g.ID, map[string]string{hostclaim.JournalCutSession: session})(cl)
752 }); err != nil {
753 return moves.StoppedReport{}, fmt.Errorf("give up execution: %w", err)
754 }
755 log.Printf("moves: %s: stopped at TXID %d; execution given up, citing %s", host, report.FinalTXID, stoppedKey)
756 if err := c.fault("after-exec-none"); err != nil {
757 return moves.StoppedReport{}, err
758 }
759 return report, nil
760}
761
762// stillExecuting is the final checkpoint's authority: every replica
763// request it makes first re-reads the claim, which must still name this
764// move with this box executing.
765func (c *cutRun) stillExecuting(ctx context.Context) error {
766 obs, err := c.claim(ctx)
767 if err != nil {
768 return err
769 }
770 if obs.Claim.Op.Kind != hostclaim.KindMoving || obs.Claim.Op.Exec == "" || obs.Claim.Op.Exec != c.cfg().Box {
771 return fmt.Errorf("move %s: this box no longer executes the host", c.g.ID)
772 }
773 return nil
774}
775
776// publishStoppedReport writes the report create-only. On a conflict the
777// existing report is read: the same host, op, role and N is a replay and
778// the existing report is returned; anything else is an error.
779func publishStoppedReport(ctx context.Context, s release.Store, key string, r moves.StoppedReport) (moves.StoppedReport, error) {
780 body, err := json.Marshal(r)
781 if err != nil {
782 return r, err
783 }
784 if _, err := s.Put(ctx, key, body, ""); err == nil {
785 return r, nil
786 } else if !errors.Is(err, release.ErrCASConflict) {
787 return r, fmt.Errorf("write %s: %w", key, err)
788 }
789 prev, err := readStoppedReport(ctx, s, key)
790 if err != nil {
791 return r, err
792 }
793 if prev.Host != r.Host || prev.OpID != r.OpID || prev.Role != r.Role || prev.FinalTXID != r.FinalTXID {
794 return r, fmt.Errorf("%s already names %s %s %s at TXID %d, not this stop at %d", key, prev.Host, prev.OpID, prev.Role, prev.FinalTXID, r.FinalTXID)
795 }
796 return prev, nil
797}
798
799func readStoppedReport(ctx context.Context, s release.Store, key string) (moves.StoppedReport, error) {
800 body, _, err := s.Get(ctx, key)
801 if err != nil {
802 return moves.StoppedReport{}, fmt.Errorf("read %s: %w", key, err)
803 }
804 var r moves.StoppedReport
805 if err := json.Unmarshal(body, &r); err != nil || r.V != 1 || !r.Complete() {
806 return moves.StoppedReport{}, fmt.Errorf("%s is not a complete stopped report", key)
807 }
808 return r, nil
809}
810
811// ownStores are this box's own bucket as a mirror (reads and plain writes)
812// and by version (create-only writes).
813func (c *cutRun) ownStores(ctx context.Context) (lease.MirrorStore, lease.VersionedStore, error) {
814 cfg := c.cfg()
815 tgt, ok := c.fleet.targets[c.rec.Move.To]
816 if !ok {
817 return nil, nil, fmt.Errorf("target region %s publishes no data bucket yet. waiting", c.rec.Move.To)
818 }
819 own, err := c.w.mirror(ctx, tgt.DataBucketRegion)
820 if err != nil {
821 return nil, nil, err
822 }
823 if cfg.Versioned == nil {
824 return nil, nil, errors.New("no versioned store on this box")
825 }
826 vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
827 if err != nil {
828 return nil, nil, err
829 }
830 return own, vs, nil
831}
832
833// finalTXID is N (§5.1 step 1): the highest max TXID among the claim
834// generation's level files in this box's own replica, the set the lane
835// mirrors (G3). After the final checkpoint it is the newest level-0
836// file's; a compaction only ever rewrites below it.
837func (c *cutRun) finalTXID(ctx context.Context, own lease.MirrorStore, prefix string, keep func(string) bool) (uint64, error) {
838 objs, err := own.List(ctx, c.cfg().OwnBucket, prefix)
839 if err != nil {
840 return 0, fmt.Errorf("list the replica: %w", err)
841 }
842 newest := newestLTX(objs, prefix, keep)
843 if newest == "" {
844 return 0, errors.New("the replica holds no level file")
845 }
846 _, maxTX, err := ltx.ParseFilename(filepath.Base(newest))
847 if err != nil {
848 return 0, err
849 }
850 return uint64(maxTX), nil
851}
852
853// snapshotTimers writes the host's timer rows create-only to
854// move-snapshots/<host>/<op>/timers.json in this box's bucket and returns
855// their digest. A snapshot already there is this cut's own (a replay): it
856// is read back and its rows are the ones digested.
857func (c *cutRun) snapshotTimers(ctx context.Context, own lease.MirrorStore, vs lease.VersionedStore) (string, error) {
858 cfg := c.cfg()
859 if cfg.Timers == nil {
860 return "", errors.New("no timer rows on this box")
861 }
862 rows, err := cfg.Timers(c.rec.Host)
863 if err != nil {
864 return "", fmt.Errorf("read the timers: %w", err)
865 }
866 body, err := moves.TimersSnapshot{V: 1, Host: c.rec.Host, Rows: rows, TakenAt: cfg.Now().UTC()}.Encode()
867 if err != nil {
868 return "", err
869 }
870 key := moves.SnapshotsPrefix(c.rec.Host, c.g.ID) + "timers.json"
871 if _, err := vs.PutIfAbsent(ctx, cfg.OwnBucket, key, body); errors.Is(err, lease.ErrExists) {
872 prev, err := own.Get(ctx, cfg.OwnBucket, key)
873 if err != nil {
874 return "", fmt.Errorf("read %s: %w", key, err)
875 }
876 snap, err := moves.DecodeTimers(prev)
877 if err != nil {
878 return "", err
879 }
880 rows = snap.Rows
881 } else if err != nil {
882 return "", fmt.Errorf("snapshot the timers: %w", err)
883 }
884 return moves.TimersDigest(rows), nil
885}
886
887// lease takes the staging lease once per pass (G6) and revalidates the
888// claim under it (K5). waiting is true when another worker holds it.
889func (c *cutRun) lease(ctx context.Context) (waiting bool, err error) {
890 if c.gm != nil {
891 return false, nil
892 }
893 cfg := c.cfg()
894 if cfg.StagingLease == nil {
895 return false, errors.New("no deployment bucket for the staging lease")
896 }
897 tgt, ok := c.fleet.targets[c.rec.Move.To]
898 if !ok {
899 return false, fmt.Errorf("target region %s publishes no data bucket yet. waiting", c.rec.Move.To)
900 }
901 held, rel, err := cfg.StagingLease(ctx, c.rec.Move.To, c.rec.Host)
902 if err != nil {
903 var he *lease.HeldError
904 if errors.As(err, &he) {
905 log.Printf("moves: %s: another worker holds the staging lease; the cut waits", c.rec.Host)
906 return true, nil
907 }
908 return false, fmt.Errorf("staging lease: %w", err)
909 }
910 stg, err := c.w.staging(ctx, tgt.DataBucketRegion)
911 if err != nil {
912 rel(context.Background())
913 return false, err
914 }
915 own, err := c.w.mirror(ctx, tgt.DataBucketRegion)
916 if err != nil {
917 rel(context.Background())
918 return false, err
919 }
920 c.gm = &guardedMirror{w: stg, g: moves.Guarded{Lease: held, Now: cfg.Now}, dst: tgt.DataBucket, own: own}
921 c.release = rel
922 return false, nil
923}
924
925func (c *cutRun) releaseLease() {
926 if c.release == nil {
927 return
928 }
929 rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
930 defer cancel()
931 if err := c.release(rctx); err != nil {
932 log.Printf("moves: %s: release the staging lease: %v", c.rec.Host, err)
933 }
934 c.release, c.gm = nil, nil
935}
936
937// sealedClaim revalidates the claim under the lease for the phases after
938// execution was given up: this move's moving op with exec none and this
939// stop's evidence, at one of the phases given.
940func (c *cutRun) sealedClaim(ctx context.Context, stoppedKey string, phases ...string) (hostclaim.Observed, error) {
941 obs, err := c.claim(ctx)
942 if err != nil {
943 return obs, err
944 }
945 op := obs.Claim.Op
946 if op.Kind != hostclaim.KindMoving || !containsString(phases, op.Phase) || op.Exec != "" || op.Journal[hostclaim.JournalEvidenceSource] != stoppedKey {
947 return obs, fmt.Errorf("move %s: the claim is %s/%q exec %q, not given up for this cut", c.g.ID, op.Kind, op.Phase, op.Exec)
948 }
949 return obs, nil
950}
951
952// bound is the writers' horizon for the seal (G7): the latest moment a
953// request this pass signed can still land, and never earlier than the
954// bound already recorded.
955func (c *cutRun) bound(prev time.Time) time.Time {
956 b := prev
957 if last := c.gm.lastSigned(); !last.IsZero() {
958 if t := last.Add(moves.SignatureLife + moves.DefaultSkew); t.After(b) {
959 b = t
960 }
961 }
962 return b.UTC()
963}
964
965// finalRound is stopped's second half: the timers and the replica into
966// staging under the lease, then the seal.
967func (c *cutRun) finalRound(ctx context.Context, holder hostclaim.Holder, report moves.StoppedReport, stoppedKey string) (bool, error) {
968 cfg := c.cfg()
969 host, id := c.rec.Host, c.g.ID
970 if waiting, err := c.lease(ctx); waiting || err != nil {
971 return false, err
972 }
973 obs, err := c.sealedClaim(ctx, stoppedKey, hostclaim.PhaseMoving, hostclaim.PhaseSealed)
974 if err != nil {
975 return false, err
976 }
977 own, _, err := c.ownStores(ctx)
978 if err != nil {
979 return false, err
980 }
981 dst := c.gm.dst
982 timersKey := moves.SnapshotsPrefix(host, id) + "timers.json"
983 timersBody, err := own.Get(ctx, cfg.OwnBucket, timersKey)
984 if err != nil {
985 return false, fmt.Errorf("read %s: %w", timersKey, err)
986 }
987 stagedTimers := moves.TimersPrefix(host, id) + "timers.json"
988 tver, err := c.gm.PutVersioned(ctx, dst, stagedTimers, timersBody)
989 if err != nil {
990 return false, fmt.Errorf("stage the timers: %w", err)
991 }
992 if !lease.UsableVersionID(tver) {
993 return false, fmt.Errorf("the staged timers have no usable version id %q", tver)
994 }
995 prefix, keep := moves.ReplicaSource(holder)
996 dstPrefix := moves.ReplicaPrefix(host, id)
997 receipts := lease.Receipts{Bucket: cfg.OwnBucket, Key: moves.ReceiptsKey(host, id)}
998 var round lease.MirrorRound
999 for attempt := 0; ; attempt++ {
1000 round, err = lease.MirrorPrefixKeep(ctx, c.gm, cfg.OwnBucket, prefix, dst, dstPrefix, receipts, keep)
1001 if err == nil || attempt == 1 || errors.Is(err, moves.ErrLeaseUnsafe) || errors.Is(err, moves.ErrInvalidSkew) {
1002 break
1003 }
1004 log.Printf("moves: %s: final round failed, retrying once: %v", host, err)
1005 }
1006 if err != nil {
1007 return false, fmt.Errorf("final round: %w", err)
1008 }
1009 // The build state of the stopped instance, so the lane carries the
1010 // final state and not a round-old one (settings spec rev 11 §4.3a). A
1011 // state that changed under the copy is not final: the pass retries.
1012 // The seal names the version this round copied (ruling R-SEAL-STATE).
1013 // This box's request record goes into it first (ruling R-SPEC-43A-2):
1014 // execution is given up, so no wake writes it now.
1015 if cfg.CarryRequest != nil {
1016 if err := cfg.CarryRequest(ctx, host, cfg.OwnBucket, strings.TrimSuffix(holder.Location, "/")); err != nil {
1017 return false, fmt.Errorf("final round: carry the request record into the build state: %w", err)
1018 }
1019 }
1020 stRound, stateObj, err := mirrorBuildState(ctx, c.gm, cfg.OwnBucket, holder.Location, dst, host, id)
1021 if err != nil {
1022 return false, fmt.Errorf("final round: build state lane: %w", err)
1023 }
1024 if stRound.Deferred > 0 {
1025 return false, fmt.Errorf("final round: the build state changed during the copy; retrying the cut")
1026 }
1027 if err := c.fault("after-final-round"); err != nil {
1028 return false, err
1029 }
1030 ledgerKey := moves.LedgerKey(host, id)
1031 ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey)
1032 if err != nil {
1033 return false, err
1034 }
1035 // §7.4 staged: a round that deferred is not byte-final; nor is one that
1036 // moved more than the bar the cut was priced for (G27).
1037 if round.Deferred > 0 || round.DeltaBytes() > cfg.ReadyDelta {
1038 ledger.DeferredRounds++
1039 if err := writeMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey, ledger); err != nil {
1040 return false, err
1041 }
1042 log.Printf("moves: %s: final round %d deferred %d object(s) and moved %d bytes", host, ledger.DeferredRounds, round.Deferred, round.DeltaBytes())
1043 if ledger.DeferredRounds >= cutMaxDeferred {
1044 return c.revert(ctx, provision.RevertDeferredRounds)
1045 }
1046 nudgeSerial(ctx, c.s)
1047 return false, nil
1048 }
1049 // G27: every kept current source key by a usable version, else another
1050 // round.
1051 src, err := own.List(ctx, cfg.OwnBucket, prefix)
1052 if err != nil {
1053 return false, fmt.Errorf("list the replica: %w", err)
1054 }
1055 var replica []moves.SealedObject
1056 for k, o := range src {
1057 rel := strings.TrimPrefix(k, prefix)
1058 if !keep(rel) {
1059 continue
1060 }
1061 ver := round.Versions[rel]
1062 if !lease.UsableVersionID(ver) {
1063 c.w.notice(host+"#seal", "moves: %s: the final round has no usable version for %s; another round", host, rel)
1064 return false, nil
1065 }
1066 replica = append(replica, moves.SealedObject{Key: dstPrefix + rel, VersionID: ver, Size: o.Size})
1067 }
1068 var state []moves.SealedObject
1069 if stateObj != nil {
1070 if !lease.UsableVersionID(stateObj.VersionID) {
1071 c.w.notice(host+"#seal", "moves: %s: the final round has no usable version for the build state; another round", host)
1072 return false, nil
1073 }
1074 state = []moves.SealedObject{*stateObj}
1075 }
1076 // Coverage starts false (ruling 3): a database still here is asked for
1077 // its position against N; one that is gone is covered only if the
1078 // checkpoint journalled that it was, at this N (giveUpExecution).
1079 fromDB, err := regularFile(c.dbPath())
1080 if err != nil {
1081 return false, err
1082 }
1083 covered := false
1084 if fromDB {
1085 if cfg.LocalPosition == nil {
1086 return false, errors.New("no local position on this box")
1087 }
1088 pos, err := cfg.LocalPosition(c.dbPath())
1089 if err != nil {
1090 return false, fmt.Errorf("read the stopped database's position: %w", err)
1091 }
1092 covered = pos == report.FinalTXID
1093 } else if j, err := c.journal(); err != nil {
1094 return false, err
1095 } else if raw, ok := j.Steps[cutStepCoverage]; ok {
1096 var cov cutCoverage
1097 if json.Unmarshal([]byte(raw), &cov) == nil && cov.N == report.FinalTXID {
1098 covered = cov.Covered
1099 }
1100 }
1101 now := cfg.Now().UTC()
1102 seal := moves.Seal{V: moves.SealVersion, N: report.FinalTXID, Replica: replica,
1103 Timers: []moves.SealedObject{{Key: stagedTimers, VersionID: tver, Size: int64(len(timersBody))}},
1104 State: state,
1105 SrcCovered: covered, FromDB: fromDB, SealedAt: now}
1106 seal.LeaseUntil = c.bound(now)
1107 if err := seal.Validate(); err != nil {
1108 return false, err
1109 }
1110 if !seal.ReachesN() {
1111 return false, fmt.Errorf("the sealed replica does not end at TXID %d", seal.N)
1112 }
1113 if _, err := hostclaim.Update(ctx, c.s, obs, sealStopped(id, stoppedKey, seal)); err != nil {
1114 return false, fmt.Errorf("seal the cut: %w", err)
1115 }
1116 if err := c.fault("after-seal"); err != nil {
1117 return false, err
1118 }
1119 ledger.LeaseUntil = seal.LeaseUntil
1120 if err := writeMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey, ledger); err != nil {
1121 return false, err
1122 }
1123 log.Printf("moves: %s: sealed %d replica objects at TXID %d", host, len(replica), seal.N)
1124 return c.advance(ctx, provision.CutStopped, provision.CutStaged, nil)
1125}
1126
1127// sealStopped is the stopped seal's claim write: the op's phase "" to
1128// sealed (or already sealed, a replay) and the seal, in one CAS, under
1129// exec none and this stop's evidence.
1130func sealStopped(id, stoppedKey string, seal moves.Seal) func(*hostclaim.Claim) error {
1131 return func(c *hostclaim.Claim) error {
1132 if c.Op.Exec != "" || c.Op.Journal[hostclaim.JournalEvidenceSource] != stoppedKey {
1133 return fmt.Errorf("%w: the seal needs exec none and this stop's evidence", hostclaim.ErrWrongState)
1134 }
1135 if c.Op.Phase == hostclaim.PhaseMoving {
1136 if err := hostclaim.SetMovePhase(id, hostclaim.KindMoving, hostclaim.PhaseMoving, hostclaim.PhaseSealed)(c); err != nil {
1137 return err
1138 }
1139 } else if err := hostclaim.RequireMoveOp(id, hostclaim.KindMoving, hostclaim.PhaseSealed)(c); err != nil {
1140 return err
1141 }
1142 return hostclaim.SetJournal(id, map[string]string{hostclaim.JournalSeal: seal.Encode()})(c)
1143 }
1144}
1145
1146// journalledSeal reads the seal the claim carries.
1147func journalledSeal(c *hostclaim.Claim) (moves.Seal, error) {
1148 raw, ok := c.Op.Journal[hostclaim.JournalSeal]
1149 if !ok {
1150 return moves.Seal{}, fmt.Errorf("%w: the claim carries no seal", moves.ErrBadSeal)
1151 }
1152 return moves.DecodeSeal(raw)
1153}
1154
1155// staged (§7.4, §8.3, codex 19, G14): the final archive of the stopped
1156// files directory, its identity fixed before any PUT, written create-only
1157// to move-snapshots/ and staged into files/, then the seal with the
1158// archive's two versions, and only then the record's archived with the
1159// archive's digests. A replay finds staged and redoes the archive from its
1160// journalled identity (cutArchive), then the seal and the phase.
1161func (c *cutRun) staged(ctx context.Context) (bool, error) {
1162 a, ok, err := c.sealArchive(ctx)
1163 if err != nil || !ok {
1164 return false, err
1165 }
1166 return c.advance(ctx, provision.CutStaged, provision.CutArchived, func(m *provision.MoveState) {
1167 if m.Source == nil {
1168 m.Source = &provision.MoveSource{}
1169 }
1170 m.Source.Files = &provision.MoveFilesDigest{Plain: a.Plain, Sealed: a.Sealed, Bytes: a.Bytes, Entries: a.Entries}
1171 })
1172}
1173
1174// sealArchive is staged's work, shared with archived's replay: the archive
1175// published and the claim's seal naming it. ok is false when the phase
1176// ended without it (a revert, the lease held elsewhere).
1177func (c *cutRun) sealArchive(ctx context.Context) (builtArchive, bool, error) {
1178 cfg := c.cfg()
1179 src := c.rec.Move.Source
1180 if src == nil || src.Files == nil || src.Evidence == "" {
1181 return builtArchive{}, false, fmt.Errorf("move %s: the record has no source baselines", c.g.ID)
1182 }
1183 if waiting, err := c.lease(ctx); waiting || err != nil {
1184 return builtArchive{}, false, err
1185 }
1186 obs, err := c.sealedClaim(ctx, src.Evidence, hostclaim.PhaseSealed)
1187 if err != nil {
1188 return builtArchive{}, false, err
1189 }
1190 seal, err := journalledSeal(obs.Claim)
1191 if err != nil {
1192 return builtArchive{}, false, err
1193 }
1194 if seal.N != src.FinalTXID {
1195 return builtArchive{}, false, fmt.Errorf("move %s: the seal is at TXID %d and the record at %d", c.g.ID, seal.N, src.FinalTXID)
1196 }
1197 a, files, why, err := c.cutArchive(ctx, src.Files.Plain)
1198 if err != nil {
1199 return builtArchive{}, false, err
1200 }
1201 if why != "" {
1202 _, err := c.revert(ctx, provision.RevertArchiveRefused)
1203 return builtArchive{}, false, err
1204 }
1205 seal.Files = files
1206 seal.LeaseUntil = c.bound(seal.LeaseUntil)
1207 if err := seal.Validate(); err != nil {
1208 return builtArchive{}, false, err
1209 }
1210 if _, err := hostclaim.Update(ctx, c.s, obs, func(cl *hostclaim.Claim) error {
1211 if cl.Op.Exec != "" || cl.Op.Journal[hostclaim.JournalEvidenceSource] != src.Evidence {
1212 return fmt.Errorf("%w: the seal needs exec none and this stop's evidence", hostclaim.ErrWrongState)
1213 }
1214 if err := hostclaim.RequireMoveOp(c.g.ID, hostclaim.KindMoving, hostclaim.PhaseSealed)(cl); err != nil {
1215 return err
1216 }
1217 return hostclaim.SetJournal(c.g.ID, map[string]string{hostclaim.JournalSeal: seal.Encode()})(cl)
1218 }); err != nil {
1219 return builtArchive{}, false, fmt.Errorf("seal the archive: %w", err)
1220 }
1221 if err := c.fault("after-archive-seal"); err != nil {
1222 return builtArchive{}, false, err
1223 }
1224 own, _, err := c.ownStores(ctx)
1225 if err != nil {
1226 return builtArchive{}, false, err
1227 }
1228 // G7: the bound in the ledger as well as the seal. A ledger that cannot
1229 // be read or written stops the phase here; the replay seals again and
1230 // writes it (ruling 5).
1231 ledgerKey := moves.LedgerKey(c.rec.Host, c.g.ID)
1232 ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey)
1233 if err != nil {
1234 return builtArchive{}, false, err
1235 }
1236 ledger.LeaseUntil = seal.LeaseUntil
1237 if err := writeMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey, ledger); err != nil {
1238 return builtArchive{}, false, err
1239 }
1240 return a, true, nil
1241}
1242
1243// archived (§7.4, §7.5): the flip. Its precondition is re-read from the
1244// claim: the seal names the archive at the record's N, and the record's
1245// digests are the seal's archive; otherwise this is a replay of staged's
1246// tail, which is redone (never a flip on a seal without the archive). Then
1247// the record flips (one CAS: region, state landing, the landing deadline)
1248// and the claim follows: landing/checking with the target executing,
1249// citing the source's evidence (hostclaim.Flip). The claim half is tried
1250// once here; anyone who sees the record ahead writes it (the console's
1251// recovery pass, Plan 5).
1252func (c *cutRun) archived(ctx context.Context) (bool, error) {
1253 cfg := c.cfg()
1254 src := c.rec.Move.Source
1255 if src == nil || src.Files == nil {
1256 return false, fmt.Errorf("move %s: the record has no source baselines", c.g.ID)
1257 }
1258 obs, err := c.sealedClaim(ctx, src.Evidence, hostclaim.PhaseSealed)
1259 if err != nil {
1260 return false, err
1261 }
1262 seal, err := journalledSeal(obs.Claim)
1263 if err != nil {
1264 return false, err
1265 }
1266 if !seal.HasFiles() || seal.N != src.FinalTXID || !sealNamesArchive(seal, src.Files.Sealed) {
1267 log.Printf("moves: %s: archived without a seal naming the archive; redoing the archive", c.rec.Host)
1268 if _, ok, err := c.sealArchive(ctx); err != nil || !ok {
1269 return false, err
1270 }
1271 if obs, err = c.claim(ctx); err != nil {
1272 return false, err
1273 }
1274 if seal, err = journalledSeal(obs.Claim); err != nil {
1275 return false, err
1276 }
1277 if !seal.HasFiles() || seal.N != src.FinalTXID || !sealNamesArchive(seal, src.Files.Sealed) {
1278 return false, fmt.Errorf("move %s: the record's archive %q is not the one the seal names", c.g.ID, src.Files.Sealed)
1279 }
1280 }
1281 targetBox := obs.Claim.Op.Intent["target"]
1282 holder := *obs.Claim.Holder
1283 now := cfg.Now()
1284 ok, err := c.cutTransition(ctx, "the flip", provision.MoveFlip(c.g, now, now.Add(cutLandingWindow)))
1285 if !ok {
1286 return ok, err
1287 }
1288 log.Printf("moves: %s: flipped to %s", c.rec.Host, c.rec.Move.To)
1289 if err := c.fault("after-record-flip"); err != nil {
1290 return true, err
1291 }
1292 if _, err := hostclaim.Update(ctx, c.s, obs, hostclaim.Flip(c.g.ID, targetBox, holder.Stamp, holder.Generation)); err != nil {
1293 log.Printf("moves: %s: the claim half of the flip waits for the recovery pass: %v", c.rec.Host, err)
1294 }
1295 nudgeSerial(ctx, c.s)
1296 return true, nil
1297}
1298
1299// revertClaim is the claim half of a revert (§7.4 reverting, §7.5): once
1300// the source has given up execution, the claim goes to reverting under
1301// the source with the staging lane's park queued (hostclaim.Revert). While
1302// the source still executes (a revert from stopping, or from stopped
1303// before exec went to none) there is nothing to hand back, and the claim
1304// stays moving: the recovery (Plan 4) starts from there. A claim already
1305// reverting for this move is done.
1306//
1307// The park's evidence_at is the writers' horizon (G7): the later of the
1308// seal's bound and this decision's time plus a signature's life and the
1309// skew, journalled once so a replay names the same scope.
1310func (w *movesWorker) revertClaim(ctx context.Context, s release.Store, rec provision.Record) error {
1311 cfg := w.cfg
1312 id := rec.Move.ID
1313 obs, err := hostclaim.Read(ctx, s, rec.Host)
1314 if err != nil {
1315 return fmt.Errorf("read the host claim: %w", err)
1316 }
1317 if obs.Absent || obs.Claim.Op.ID != id {
1318 return nil
1319 }
1320 op := obs.Claim.Op
1321 if op.Kind == hostclaim.KindReverting || op.Kind != hostclaim.KindMoving || op.Exec != "" || op.Journal[hostclaim.JournalEvidenceSource] == "" {
1322 return nil
1323 }
1324 if obs.Claim.Holder == nil || cfg.Box == "" {
1325 return errors.New("the revert needs the holder and this box's label")
1326 }
1327 j, err := moves.ReadJournal(cfg.DataDir, rec.Host, id)
1328 if err != nil {
1329 return err
1330 }
1331 var bound time.Time
1332 if raw, ok := j.Steps[cutStepRevertBound]; ok {
1333 if bound, err = time.Parse(time.RFC3339Nano, raw); err != nil {
1334 bound = time.Time{}
1335 }
1336 }
1337 if bound.IsZero() {
1338 bound = cfg.Now().Add(moves.SignatureLife + moves.DefaultSkew).UTC()
1339 if seal, err := journalledSeal(obs.Claim); err == nil && seal.Bound().After(bound) {
1340 bound = seal.Bound().UTC()
1341 }
1342 j.Mark(cutStepRevertBound, bound.Format(time.RFC3339Nano))
1343 if err := j.Commit(cfg.DataDir); err != nil {
1344 return err
1345 }
1346 }
1347 scope := moves.StagingScope(rec.Host, id, obs.Claim.Holder.Incarnation, bound)
1348 if _, err := hostclaim.Update(ctx, s, obs, hostclaim.Revert(id, cfg.Box, park.PlacementRegion(rec.Move.To), &scope, obligationID)); err != nil {
1349 return fmt.Errorf("revert the claim: %w", err)
1350 }
1351 log.Printf("moves: %s: the claim reverts to this box; the staging lane in %s is queued for its park", rec.Host, rec.Move.To)
1352 return nil
1353}
1354
1355// obligationID mints a claim obligation or retirement id.
1356func obligationID() string {
1357 var b [16]byte
1358 if _, err := rand.Read(b[:]); err != nil {
1359 return ""
1360 }
1361 return hex.EncodeToString(b[:])
1362}
1363
1364// regularFile reports whether path is a regular file. Absence (ENOENT) is
1365// false with no error; any other answer from the filesystem, an error or
1366// something there that is not a regular file, is an error, never absence
1367// (ruling 3): the cut decides coverage on it, so it must not read a
1368// transient fault as "the database is gone".
1369func regularFile(path string) (bool, error) {
1370 st, err := os.Lstat(path)
1371 if errors.Is(err, os.ErrNotExist) {
1372 return false, nil
1373 }
1374 if err != nil {
1375 return false, fmt.Errorf("stat %s: %w", path, err)
1376 }
1377 if !st.Mode().IsRegular() {
1378 return false, fmt.Errorf("%s is not a regular file (%s)", path, st.Mode())
1379 }
1380 return true, nil
1381}
1382
1383// fileSHA256 is "sha256:<hex>" of the file at path.
1384func fileSHA256(path string) (string, error) {
1385 f, err := os.Open(path)
1386 if err != nil {
1387 return "", err
1388 }
1389 defer f.Close()
1390 h := sha256.New()
1391 if _, err := io.Copy(h, f); err != nil {
1392 return "", err
1393 }
1394 return "sha256:" + hex.EncodeToString(h.Sum(nil)), nil
1395}
1396
1397// holdMoving is moveCfg.HoldMoving in production: the row held "moving"
1398// under op, conditional on its current hold (SetHoldIf) and under the
1399// host's hold lock, which reconcile's hold decision and the landing's open
1400// share (Task 6). A row already held moving by op is left as it is; a row
1401// held by another op is refused. SQLITE_BUSY while root freezes holds is
1402// an error the next pass retries (G30). A write that landed touches the
1403// stage-requested marker.
1404func holdMoving(reg *registry.Registry, written func(host string)) func(host, op string) error {
1405 return func(host, op string) error {
1406 var err error
1407 provision.WithHostHoldLock(host, func() {
1408 var rt *registry.Route
1409 if rt, err = reg.Get(host); err != nil {
1410 return
1411 }
1412 if rt == nil {
1413 err = fmt.Errorf("%s has no registry row on this box", host)
1414 return
1415 }
1416 if rt.HoldOp == op && rt.Status == "moving" {
1417 return
1418 }
1419 if rt.HoldOp != "" && rt.HoldOp != op {
1420 err = fmt.Errorf("%s is held %s by %s", host, rt.Status, rt.HoldOp)
1421 return
1422 }
1423 var held bool
1424 if held, err = reg.SetHoldIf(host, "moving", op, rt.HoldOp); err != nil {
1425 return
1426 }
1427 if !held {
1428 err = fmt.Errorf("%s: the row's hold changed under the cut", host)
1429 return
1430 }
1431 if written != nil {
1432 written(host)
1433 }
1434 })
1435 return err
1436 }
1437}
1438
1439// localLTXPosition is moveCfg.LocalPosition in production: the stopped
1440// database's TXID, the newest level-0 file in the activator's litestream
1441// meta directory (lease.Held's SetMetaPath, litestream's DB.MaxLTX). No
1442// directory is position 0.
1443func localLTXPosition(dbPath string) (uint64, error) {
1444 dir := filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream", "ltx", "0")
1445 ents, err := os.ReadDir(dir)
1446 if errors.Is(err, os.ErrNotExist) {
1447 return 0, nil
1448 }
1449 if err != nil {
1450 return 0, err
1451 }
1452 var newest ltx.TXID
1453 for _, e := range ents {
1454 if _, maxTX, err := ltx.ParseFilename(e.Name()); err == nil && maxTX > newest {
1455 newest = maxTX
1456 }
1457 }
1458 return uint64(newest), nil
1459}
1460
1461// errHolderLocation: the claim holder's location is not the activator's
1462// own lease prefix, so the cut cannot stop its writer (ruling T5-lease).
1463var errHolderLocation = errors.New("the claim holder is not at the activator's lease location")
1464
1465// writerLease is the one lease object the host's writer holds: the
1466// activator's, at activator.HibernatePrefix(prefix, host), as
1467// ExecBackend.Acquire takes it. Phase 1 cuts a holder only when its
1468// location is exactly that (ruling T5-lease): the final checkpoint and the
1469// stop evidence then exclude the real writer, and a scoped location
1470// (instances/<host>/<incarnation>), whose lease would be a different
1471// object, is refused at admission. The claim's generation is still bound
1472// on that lease, so the replica lineage stays the holder's.
1473func writerLease(prefix, host string, h hostclaim.Holder) (string, error) {
1474 want := activator.HibernatePrefix(prefix, host)
1475 if strings.TrimSuffix(h.Location, "/") != want {
1476 return "", fmt.Errorf("%w: %s is at %q, not %q", errHolderLocation, host, h.Location, want)
1477 }
1478 return want, nil
1479}
1480
1481// holderSupported is writerLease for this cut's box.
1482func (c *cutRun) holderSupported(h hostclaim.Holder) error {
1483 _, err := writerLease(c.cfg().OwnPrefix, c.rec.Host, h)
1484 return err
1485}
1486
1487// finalCheckpoint is moveCfg.FinalCheckpoint in production (§5.3 step 5,
1488// G3, ruling T5-lease): the activator's own lease for host, refused for a
1489// holder anywhere else before any request; the replica client bound to the
1490// claim's lineage with check as the authority read before every replica
1491// request, a bound generation (BindGeneration) or the root layout
1492// (BindRoot) alike; one checkpoint; the lease released.
1493func finalCheckpoint(bucket, prefix, region string) func(ctx context.Context, host, dbPath string, h hostclaim.Holder, check func(context.Context) error) error {
1494 return func(ctx context.Context, host, dbPath string, h hostclaim.Holder, check func(context.Context) error) error {
1495 at, err := writerLease(prefix, host, h)
1496 if err != nil {
1497 return err
1498 }
1499 held, err := lease.Acquire(ctx, bucket, at, region)
1500 if err != nil {
1501 return err
1502 }
1503 defer func() {
1504 rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
1505 defer cancel()
1506 if err := held.Release(rctx); err != nil {
1507 log.Printf("moves: release the checkpoint lease at %s: %v", at, err)
1508 }
1509 }()
1510 if g := h.Generation; g != "" && g != hostclaim.RootGeneration {
1511 lineage, _ := moves.ReplicaSource(h)
1512 err = held.BindGeneration(lineage, check)
1513 } else {
1514 err = held.BindRoot(check)
1515 }
1516 if err != nil {
1517 return err
1518 }
1519 _, err = held.Checkpoint(ctx, dbPath)
1520 return err
1521 }
1522}
1523