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 "errors"
6 "fmt"
7 "log"
8 "os"
9 "path/filepath"
10 "strconv"
11 "strings"
12 "time"
13
14 "github.com/carlosframework/platform/internal/appconfig"
15 "github.com/carlosframework/platform/internal/hostclaim"
16 "github.com/carlosframework/platform/internal/hostguard"
17 "github.com/carlosframework/platform/internal/lease"
18 "github.com/carlosframework/platform/internal/moves"
19 "github.com/carlosframework/platform/internal/park"
20 "github.com/carlosframework/platform/internal/provision"
21 "github.com/carlosframework/platform/internal/release"
22)
23
24// The source's finalization after the commit (moves spec §7.9 step 3; Plan 3
25// Task 8 with rulings H5, H7, H19 and H24), behind CARLOS_MOVES_PHASE1. Once
26// the record is committed the target's region is authoritative, and the
27// source box has four things left to do for the move, in this order:
28//
29// 1. the remote half: the source region's location park (the park-replica
30// the commit queued for instances/<host>, D6, H24) is run through the
31// park engine over this box's own bucket until its archive is complete
32// (park.Engine.Parked). Its acknowledgement after the horizon is the
33// engine's or the regional runner's (C4), never a reason to wait here.
34// The horizon starts at the flip (H7), which is after the source's last
35// write to its location: a park whose EvidenceAt is earlier is refused;
36// 2. the local half (ruling H5(b), Paul): a source whose installation is
37// registered (the commit queued its park-local, because the inventory
38// lists paths of the journalled source stamp S) retires it through
39// park.Local.RetireInstallation under the host guard, which completes the
40// chore itself (H19). Everything else the source kept at the live flat
41// paths, registered or not, is parked by a moves-owned journalled rename
42// into the engine's layout, .parked/<incarnation>/<host>/<retirement>/
43// artifacts/ (parkLocalArtifacts): the retirement id and every artifact's
44// identity are journalled before the first rename, and nothing the plan
45// did not record is ever moved or removed. The plan and the renames run
46// inside the host guard with the claim and the record read fresh there
47// (parkLocal; Plan 5 Task 3, carried follow-up (d)), so a call that
48// resumes after authority moved on renames nothing;
49// 3. the row and its timers rows go in one registry transaction, only while
50// the move holds the row (registry.DeleteIfHoldWithTimers, H5(d)), under
51// provision.WithHostHoldLock with the record re-read there; the journal
52// says row-deleted "pending" before it and "done" after it, and a replay
53// that finds "pending" and no row of this op's treats the delete as done
54// (round-2 finding 12). The box's config ack is copied into the move's
55// report prefix as config-ack-source.json and deleted (§6.3);
56// 4. MoveSourceParked stamps source_parked_at, and the serial is nudged.
57// Last of all the op's own journal directory is renamed into the same
58// retirement, so a finished source keeps nothing under .moves/. Then the
59// source tries the close (closeMove, movesparks.go): it closes only when
60// the target has opened and both parks are acknowledged already, and
61// this is the last pass that routes the record here, so otherwise the
62// close is the target's.
63//
64// Routing (ruling H5(a)): after the flip the record's region is the target's
65// and after step 3 this box has no row, so the region-and-row rule that
66// admits every other Phase 1 round would never bring the replay of a crash
67// between steps 3 and 4 back here. movesParticipantRecord is the rule for the
68// states after the flip: this box's region is the move's From (or To), and it
69// still has the row or the op's journal. Plan 4's rollback visits reuse it.
70//
71// Arming. Every hook that exists on main is wired (the guard, park.Local, the
72// registry's delete, the versioned store). The claim's park leases, view and
73// authorizer come from its MoveParks (armMovesParks, Plan 5 Task 9a), and
74// park.Local's authorizer is its MoveParks local authorizer (M9);
75// armMovesParks sets them only once the wake-source gate opens
76// (movesWakeSourceRegistered, Plan 5 Task 9b), so sourceFinalArmed answers
77// false in production and the pass only logs until then. The tests arm it
78// with fakes.
79
80// The finalization's op journal steps (moves.Journal, the source's own
81// journal of this op, which the cut began).
82const (
83 finalStepRemote = "source-park-remote" // the location park's archive is complete, or acknowledged
84 finalStepInstallation = "source-park-installation" // retired, completed or unregistered
85 finalStepLocal = "source-park-local" // parkLocalArtifacts' step: the retirement id
86 finalStepRowDeleted = "row-deleted" // "pending" before the delete, "done" after it
87 finalStepConfigAck = "config-ack-source" // the config ack is copied and deleted
88 finalStepParked = "source-parked" // "pending" before the stamp's CAS, "done" after it
89 finalStepJournal = "source-park-journal" // where the journal is going, written just before it goes
90)
91
92// finalConfigAckName is the config ack's copy under the move's report prefix.
93const finalConfigAckName = "config-ack-source.json"
94
95// movesParticipantStates are the record states after the flip (and a
96// revert or an abort) in which a participant may have work for a record
97// whose region no longer names it, or whose row it has already deleted.
98var movesParticipantStates = map[string]bool{
99 provision.MoveCommitted: true,
100 provision.MoveRollingBack: true,
101 provision.MoveReverting: true,
102 provision.MoveAborting: true,
103}
104
105// movesParticipantRecord is the routing rule for the states after the flip
106// (ruling H5(a), shared with Plan 4's J10 and J13): it answers this box's role
107// in rec's move, roleSource when this box's region is the move's From,
108// roleTarget when it is the move's To, and roleNone otherwise. A role is
109// answered only while the box still has something of the move's: its row
110// (OwnsRoute) or the op's journal directory (moveJournalPresent). The caller
111// decides what a role does in a state; this only says the box is a
112// participant with work it can find.
113func movesParticipantRecord(cfg moveCfg, rec provision.Record) string {
114 m := rec.Move
115 if m == nil || m.Legacy() || !movesParticipantStates[m.State] || m.From == m.To {
116 return roleNone
117 }
118 role := roleNone
119 switch cfg.Region {
120 case m.From:
121 role = roleSource
122 case m.To:
123 role = roleTarget
124 default:
125 return roleNone
126 }
127 if cfg.OwnsRoute == nil || cfg.OwnsRoute(rec.Host) || moveJournalPresent(cfg.DataDir, rec.Host, m.ID) {
128 return role
129 }
130 // A revert or an abort never flipped: its target has neither the row
131 // nor a journal, and its one piece of work is the lane park the claim
132 // owes its region (laneParkTarget; Plan 4 Task 6, Plan 5 Task 1).
133 if role == roleTarget && (m.State == provision.MoveReverting || m.State == provision.MoveAborting) {
134 return role
135 }
136 return roleNone
137}
138
139// moveJournalPresent reports whether this box holds the op's journal
140// directory, <dataDir>/.moves/<host>/<op>/. A name that could leave its
141// directory is never present.
142func moveJournalPresent(dataDir, host, op string) bool {
143 if dataDir == "" || !journalComponent(host) || !journalComponent(op) {
144 return false
145 }
146 fi, err := os.Lstat(moves.JournalDir(dataDir, host, op))
147 return err == nil && fi.IsDir()
148}
149
150// participantRound is a post-flip participant's step for a record the
151// region rule did not admit (movesPass): the source's finalization of a
152// committed move, the target's half of a rollback (movesrollback.go, J10),
153// and the target region's lane park of a revert or an abort
154// (laneParkTarget). The rest is Plan 4's later tasks'.
155func (w *movesWorker) participantRound(ctx context.Context, s release.Store, rec provision.Record, role string) error {
156 switch {
157 case role == roleSource && rec.Move.State == provision.MoveCommitted:
158 return w.finalizeSource(ctx, s, rec)
159 case role == roleTarget && rec.Move.State == provision.MoveCommitted:
160 return w.committedTarget(ctx, s, rec)
161 case role == roleTarget && rec.Move.State == provision.MoveRollingBack:
162 return w.rollbackTarget(ctx, s, rec)
163 case role == roleTarget && rec.Move.State == provision.MoveReverting:
164 return w.laneParkTarget(ctx, s, rec, hostclaim.KindReverting)
165 case role == roleTarget && rec.Move.State == provision.MoveAborting:
166 return w.laneParkTarget(ctx, s, rec, hostclaim.KindMoving, hostclaim.PhaseAborting)
167 }
168 return nil
169}
170
171// sourceFinalArmed reports whether this box can run the finalization: every
172// hook it calls is wired, the claim's three park hooks and its MoveParks
173// local authorizer (M9) included. Nothing stands in for a missing owner:
174// the tests inject their own. In production armMovesParks sets them only
175// once the wake-source gate opens (movesWakeSourceRegistered, Plan 5 Task
176// 9b), so this answers false there until then.
177func sourceFinalArmed(cfg moveCfg) bool {
178 return cfg.Box != "" && cfg.DataDir != "" && cfg.Guard.DataDir != "" && cfg.Local.DataDir != "" &&
179 parkHooksArmed(cfg) && cfg.ParkLocalAuthorize != nil &&
180 cfg.Versioned != nil && cfg.Route != nil && cfg.DeleteHeld != nil
181}
182
183// sourceLocationParkScope is the source region's location park the commit
184// queues (Task 7 builds the commit; Task 8 reads it): holder.location, the
185// holder's incarnation, the cut's request limit, and a horizon that starts at
186// the flip (ruling H7). The flip comes after ExecToNone, after which the
187// source never writes its location again; the stop evidence's CheckedAt does
188// not, because the final checkpoint writes after it.
189func sourceLocationParkScope(rec provision.Record, holder hostclaim.Holder) (hostclaim.ParkScope, error) {
190 if rec.Move == nil || rec.Move.FlippedAt == nil {
191 return hostclaim.ParkScope{}, errors.New("the source's location park needs the flip's time")
192 }
193 return moves.SourceLocationScope(holder, *rec.Move.FlippedAt), nil
194}
195
196// errFinalChanged is the record or claim no longer being this move's commit
197// when a step re-reads it.
198var errFinalChanged = errors.New("the record is no longer this move's commit")
199
200// sourceFinal is one pass's state for one committed record on its source.
201type sourceFinal struct {
202 w *movesWorker
203 s release.Store
204 rec provision.Record
205 g provision.MoveGuard
206 j *moves.Journal
207}
208
209// finalizeSource is the source's §7.9 step 3 for a committed record. It runs
210// only when the record is committed and either this box's row is held by the
211// move or the journal has row-deleted (Review Focus 5, round-2 finding 12):
212// a landing record, a row held by another op, or no row and no journalled
213// delete leaves everything as it is.
214func (w *movesWorker) finalizeSource(ctx context.Context, s release.Store, rec provision.Record) error {
215 cfg := w.cfg
216 m := rec.Move
217 if m == nil || m.Legacy() || m.State != provision.MoveCommitted || m.From != cfg.Region || m.To == cfg.Region {
218 return nil
219 }
220 if !sourceFinalArmed(cfg) {
221 w.notice(rec.Host+"#final", "moves: %s: the source's finalization of %s waits: this box's finalization is not armed (the claim's park hooks are set from MoveParks only once this box's moves wake source is registered, Plan 5 Task 9b)", rec.Host, m.ID)
222 return nil
223 }
224 f := &sourceFinal{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}}
225 return f.run(ctx)
226}
227
228func (f *sourceFinal) cfg() moveCfg { return f.w.cfg }
229
230func (f *sourceFinal) now() time.Time {
231 if n := f.cfg().Now; n != nil {
232 return n()
233 }
234 return time.Now()
235}
236
237// fault is the test seam after a side effect.
238func (f *sourceFinal) fault(point string) error {
239 if fn := f.cfg().finalFault; fn != nil {
240 return fn(point)
241 }
242 return nil
243}
244
245// mark journals a step durably, then passes its fault point.
246func (f *sourceFinal) mark(step, value string) error {
247 f.j.Mark(step, value)
248 if err := f.j.Commit(f.cfg().DataDir); err != nil {
249 return err
250 }
251 return f.fault("after-" + step)
252}
253
254func (f *sourceFinal) run(ctx context.Context) error {
255 cfg := f.cfg()
256 host, id := f.rec.Host, f.g.ID
257 j, err := moves.ReadJournal(cfg.DataDir, host, id)
258 if err != nil {
259 return err
260 }
261 f.j = j
262 rt, err := cfg.Route(host)
263 if err != nil {
264 return fmt.Errorf("read this box's row: %w", err)
265 }
266 held := rt != nil && rt.HoldOp == id
267 if !held && !j.Done(finalStepRowDeleted) {
268 f.w.notice(host+"#final-row", "moves: %s: the source's finalization of %s waits: this box's row is not held by the move", host, id)
269 return nil
270 }
271 if held && rt.DBPath != "" && rt.DBPath != provision.DefaultDBPath(cfg.DataDir, host) {
272 return fmt.Errorf("move %s: the source's row is not at the flat path (%s)", id, rt.DBPath)
273 }
274 // source_parked_at is stamped only after every step above it, so once it
275 // is only the journal's own park is left, and the claim may already
276 // have moved on to the close.
277 if f.rec.Move.SourceParkedAt != nil {
278 if j.Steps[finalStepParked] != "done" {
279 if err := f.mark(finalStepParked, "done"); err != nil {
280 return err
281 }
282 }
283 return f.parkJournalAndClose(ctx)
284 }
285 obs, err := hostclaim.Read(ctx, f.s, host)
286 if err != nil {
287 return fmt.Errorf("read the host claim: %w", err)
288 }
289 if err := f.sourceOf(obs); err != nil {
290 f.w.notice(host+"#final-claim", "moves: %s: the source's finalization of %s waits: %v", host, id, err)
291 return nil
292 }
293 holder := *obs.Claim.Holder
294 if !j.Done(finalStepRemote) {
295 done, err := f.remote(ctx, obs)
296 if err != nil || !done {
297 return err
298 }
299 if err := f.mark(finalStepRemote, ""); err != nil {
300 return err
301 }
302 }
303 if !j.Done(finalStepInstallation) {
304 how, err := f.installation(ctx)
305 if err != nil {
306 return err
307 }
308 if err := f.mark(finalStepInstallation, how); err != nil {
309 return err
310 }
311 }
312 if err := f.parkLocal(ctx, holder, obs.Claim); err != nil {
313 return fmt.Errorf("park the source's local artifacts: %w", err)
314 }
315 if err := f.deleteRow(ctx); err != nil {
316 return err
317 }
318 if !j.Done(finalStepConfigAck) {
319 if err := f.configAck(ctx); err != nil {
320 return err
321 }
322 if err := f.mark(finalStepConfigAck, ""); err != nil {
323 return err
324 }
325 }
326 if j.Steps[finalStepParked] != "done" {
327 // "pending" is the stamp step reached: from here the journal is
328 // the local sweep's to finish if the close overtakes this box
329 // (sweepFinalJournals).
330 if j.Steps[finalStepParked] != "pending" {
331 j.Mark(finalStepParked, "pending")
332 if err := j.Commit(cfg.DataDir); err != nil {
333 return err
334 }
335 if err := f.fault("after-source-parked-pending"); err != nil {
336 return err
337 }
338 }
339 if err := f.stampParked(ctx); err != nil {
340 return err
341 }
342 if err := f.mark(finalStepParked, "done"); err != nil {
343 return err
344 }
345 }
346 return f.parkJournalAndClose(ctx)
347}
348
349// parkJournalAndClose is step 4's end: the journal's park, then the close
350// when the target has opened and nothing is owed (closeMove).
351func (f *sourceFinal) parkJournalAndClose(ctx context.Context) error {
352 if err := f.parkJournal(); err != nil {
353 return err
354 }
355 return f.w.closeMove(ctx, f.s, f.rec, f.fault)
356}
357
358// sourceOf is the claim this finalization runs under: this move's landing,
359// committed, this box its source with its evidence journalled, and the
360// record's incarnation.
361func (f *sourceFinal) sourceOf(obs hostclaim.Observed) error {
362 cfg := f.cfg()
363 if obs.Absent || obs.Claim == nil {
364 return fmt.Errorf("%w: no host claim", park.ErrUnsafe)
365 }
366 c := obs.Claim
367 if c.Op.ID != f.g.ID || c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted {
368 return fmt.Errorf("%w: the claim is %s/%q for %q, not this move's commit", park.ErrUnsafe, c.Op.Kind, c.Op.Phase, c.Op.ID)
369 }
370 if len(c.Op.Participants) != 2 || c.Op.Participants[0] != cfg.Box || c.Op.Intent["source"] != cfg.Box {
371 return fmt.Errorf("%w: this box is not the move's source in the claim", park.ErrUnsafe)
372 }
373 if c.Op.Journal[hostclaim.JournalEvidenceSource] == "" {
374 return fmt.Errorf("%w: the claim does not journal the source's evidence", park.ErrUnsafe)
375 }
376 if c.Holder == nil || c.Holder.Incarnation != f.g.Incarnation {
377 return fmt.Errorf("%w: the claim's holder is not this move's incarnation", park.ErrUnsafe)
378 }
379 return nil
380}
381
382// engine is the park engine over this box's own bucket, with the claim's
383// leases and view, and the claim's authorizer behind the finalization's own
384// check (sourceFinalArmed requires all three): the claim is still this
385// move's commit with this box its source, and the chore is this op's in
386// this region.
387func (f *sourceFinal) engine(vs lease.VersionedStore) park.Engine {
388 cfg := f.cfg()
389 own := func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
390 if err := f.sourceOf(obs); err != nil {
391 return err
392 }
393 if o.Op != f.g.ID || !park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) {
394 return fmt.Errorf("%w: not this finalization's chore", park.ErrUnsafe)
395 }
396 return nil
397 }
398 return park.Engine{HomeRegion: cfg.ParkHomeRegion, Store: vs, Claims: f.s, Bucket: cfg.OwnBucket, Region: cfg.parkRegion(), Leases: cfg.ParkLeases,
399 Now: f.now, NewID: obligationID, View: cfg.ParkView, Authorize: movesArmAuthorize(cfg.ParkAuthorize, own), After: cfg.parkFault}
400}
401
402// remote is step 1. done is true once the source region's location park has
403// its archive complete, or is no longer in the claim (acknowledged: the
404// commit queues it in its own write, so a committed claim without it has
405// had it acknowledged). The engine takes the location's lease itself.
406func (f *sourceFinal) remote(ctx context.Context, obs hostclaim.Observed) (bool, error) {
407 cfg := f.cfg()
408 host, id := f.rec.Host, f.g.ID
409 var chore *hostclaim.Obligation
410 for i := range obs.Claim.Obligations {
411 o := &obs.Claim.Obligations[i]
412 if o.Op == id && o.Kind == "park-replica" && park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) && o.Park != nil && !isStagingLocation(o.Park.Location) {
413 chore = o
414 }
415 }
416 if chore == nil {
417 return true, nil
418 }
419 if flipped := f.rec.Move.FlippedAt; flipped == nil || chore.Park.EvidenceAt.Before(*flipped) {
420 return false, fmt.Errorf("move %s: the source's location park starts its horizon at %s, before the flip (ruling H7)", id, chore.Park.EvidenceAt.Format(time.RFC3339))
421 }
422 vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
423 if err != nil {
424 return false, err
425 }
426 e := f.engine(vs)
427 work := park.Work{Host: host, ID: chore.ID, Op: id}
428 if err := e.Run(ctx, work); err != nil && !errors.Is(err, park.ErrPending) && !errors.Is(err, park.ErrChanged) {
429 return false, fmt.Errorf("park the source's location: %w", err)
430 }
431 parked, err := e.Parked(ctx, work)
432 if errors.Is(err, park.ErrChanged) {
433 return true, nil
434 }
435 if err != nil {
436 return false, fmt.Errorf("the source's location park: %w", err)
437 }
438 if !parked {
439 f.w.notice(host+"#final-remote", "moves: %s: the source's finalization of %s waits for its location park's archive", host, id)
440 return false, nil
441 }
442 log.Printf("moves: %s: the source's location %s is parked", host, chore.Park.Location)
443 return true, nil
444}
445
446func isStagingLocation(location string) bool { return strings.HasPrefix(location, "moves/") }
447
448// installation is step 2's registered half (ruling H5(b)). With the box's
449// park-local and inventory paths of the journalled source stamp S, the
450// installation is retired under the host guard, which completes the chore; a
451// park-local with no such paths left is completed; no park-local is an
452// unregistered source, whose artifacts are all parkLocalArtifacts'. It
453// answers which, for the journal.
454func (f *sourceFinal) installation(ctx context.Context) (string, error) {
455 cfg := f.cfg()
456 host, id := f.rec.Host, f.g.ID
457 obs, err := hostclaim.Read(ctx, f.s, host)
458 if err != nil {
459 return "", fmt.Errorf("read the host claim: %w", err)
460 }
461 if err := f.sourceOf(obs); err != nil {
462 return "", fmt.Errorf("%w: %v", errFinalChanged, err)
463 }
464 c := obs.Claim
465 stamp, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
466 if err != nil || stamp == 0 {
467 return "", fmt.Errorf("move %s: the claim journals no source stamp", id)
468 }
469 var chore *hostclaim.Obligation
470 for i := range c.Obligations {
471 if o := &c.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box {
472 chore = o
473 }
474 }
475 registered := hostclaim.InventoryHasStamp(c, cfg.Box, stamp)
476 switch {
477 case chore != nil && registered:
478 captured, err := hostguard.Capture(cfg.DataDir, host)
479 if err != nil {
480 return "", fmt.Errorf("the source's installation record: %w", err)
481 }
482 // A retry after the guard retired the record and before the claim
483 // completed the chore captures stamp 0; the stamp is the claim's S.
484 if captured.Incarnation != c.Holder.Incarnation || (captured.Stamp != 0 && captured.Stamp != stamp) {
485 return "", fmt.Errorf("move %s: the source's installation record is at %s/%d, the claim's source is %s/%d", id, captured.Incarnation, captured.Stamp, c.Holder.Incarnation, stamp)
486 }
487 guard := cfg.Guard
488 if guard.Claims == nil {
489 guard.Claims = hostguard.StoreReader{Store: f.s}
490 }
491 local := cfg.Local
492 local.Guard, local.Claims = guard, f.s
493 if local.Box == "" {
494 local.Box = cfg.Box
495 }
496 if local.Authorize == nil {
497 // park.Local asks the claim's MoveParks local authorizer (M9),
498 // behind the finalization's own check over the claim it read
499 // under the guard. The claim's engine authorizer refuses a
500 // park-local, and nothing stands in: sourceFinalArmed requires it.
501 local.Authorize = movesLocalAuthorize(cfg.ParkLocalAuthorize, func(_ context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
502 if err := f.sourceOf(hostclaim.Observed{Host: host, Claim: c}); err != nil {
503 return err
504 }
505 if o.Op != id || o.Kind != "park-local" || o.Box != cfg.Box {
506 return fmt.Errorf("%w: not this finalization's local chore", park.ErrUnsafe)
507 }
508 return nil
509 })
510 }
511 if err := local.RetireInstallation(ctx, host, hostguard.Captured{Incarnation: captured.Incarnation, Stamp: stamp}, chore.ID, id); err != nil {
512 return "", fmt.Errorf("retire the source's installation: %w", err)
513 }
514 log.Printf("moves: %s: the source's registered installation (stamp %d) is retired", host, stamp)
515 return "retired", nil
516 case chore != nil:
517 if _, err := hostclaim.Update(ctx, f.s, obs, hostclaim.CompleteLocalPark(id, cfg.Box, stamp)); err != nil {
518 return "", fmt.Errorf("complete the source's local park: %w", err)
519 }
520 return "completed", nil
521 case registered:
522 return "", fmt.Errorf("move %s: this box's inventory lists the source's paths, and the commit queued no park-local for them", id)
523 }
524 return "unregistered", nil
525}
526
527// parkLocal is step 2's moves-owned rename (parkLocalArtifacts) of what the
528// source kept at the flat paths, its plan and its renames inside the host
529// guard (Plan 5 Task 3, carried follow-up (d); the pattern of Plan 4's
530// installer, installRun.guardedWith, and its rollback's parkLeftovers). In
531// the guard, with the claim and the record read fresh:
532// - the claim is still this move's commit with this box its source
533// (sourceOf), and this box does not execute the host;
534// - no inventory entry of this box claims a path the rename would move;
535// - the record is still this move's commit.
536//
537// The guard is the installation record's state: none (an unregistered
538// source with no record, WithNoInstallation), or the source's stamp, active
539// or retired by step 2's registered half (RetireInstallationWith, which
540// retires an active one first, so the guard refuses this box any later
541// write at it). A retirement already journalled is done and takes no guard.
542func (f *sourceFinal) parkLocal(ctx context.Context, holder hostclaim.Holder, c *hostclaim.Claim) error {
543 j := f.j
544 if _, done := j.Steps[finalStepLocal]; done {
545 return nil
546 }
547 cfg := f.cfg()
548 host, id := f.rec.Host, f.g.ID
549 paths := sourceArtifactPaths(host)
550 check := func(fresh *hostclaim.Claim) error {
551 if err := f.sourceOf(hostclaim.Observed{Host: host, Claim: fresh}); err != nil {
552 return fmt.Errorf("%w: %v", errFinalChanged, err)
553 }
554 if fresh.Op.Exec == cfg.Box {
555 return fmt.Errorf("%w: this box executes the host again", errFinalChanged)
556 }
557 for _, p := range paths {
558 if _, ok := fresh.Inventory[cfg.Box].Paths[p]; ok {
559 return fmt.Errorf("%w: the inventory claims %s", errFinalChanged, p)
560 }
561 }
562 rec, err := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host)
563 if err != nil {
564 return fmt.Errorf("re-read the record: %w", err)
565 }
566 if rec == nil || rec.Move == nil || rec.Move.ID != id || rec.Move.Incarnation != f.g.Incarnation || rec.Move.State != provision.MoveCommitted {
567 return errFinalChanged
568 }
569 return nil
570 }
571 work := func() error {
572 _, err := parkLocalArtifacts(ctx, j, f.mark, localArtifactPark{DataDir: cfg.DataDir, Incarnation: holder.Incarnation,
573 Host: host, Op: id, Step: finalStepLocal, Paths: paths, NewID: obligationID, Fault: f.fault})
574 return err
575 }
576 guard := cfg.Guard
577 if guard.Claims == nil {
578 guard.Claims = hostguard.StoreReader{Store: f.s}
579 }
580 // The guard's record directory: a Phase 1 source without host claims
581 // may not have one yet (the landing makes the same two, ensureLocalDirs).
582 for _, d := range []string{"hostguard", ".parked"} {
583 if err := os.MkdirAll(filepath.Join(guard.DataDir, d), 0o770); err != nil {
584 return err
585 }
586 }
587 captured, err := guard.Capture(host)
588 switch {
589 case errors.Is(err, hostguard.ErrNoRecord):
590 return guard.WithNoInstallation(ctx, host, f.g.Incarnation, check, work)
591 case err != nil:
592 return fmt.Errorf("the source's installation record: %w", err)
593 }
594 stamp := captured.Stamp
595 if stamp == 0 {
596 if stamp, err = strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64); err != nil || stamp == 0 {
597 return fmt.Errorf("move %s: the claim journals no source stamp", id)
598 }
599 }
600 return guard.RetireInstallationWith(ctx, host, hostguard.Captured{Incarnation: f.g.Incarnation, Stamp: stamp}, check, work)
601}
602
603// sourceArtifactPaths are the source's artifacts at the flat paths (G11),
604// relative to the data directory: the installation's (installArtifactPaths:
605// the database with its marker, journal, quarantine and litestream shadow,
606// the files directory and <host>.env), then the rendered config and the
607// sidecar's env. The order is the journalled plan's and does not change.
608func sourceArtifactPaths(host string) []string {
609 return append(installArtifactPaths(host), host+".config.env", host+"-sidecar.env")
610}
611
612// deleteRow is step 3's row and timers delete, under the host's hold lock
613// with the record re-read there (the lock reconcile's hold decision holds).
614func (f *sourceFinal) deleteRow(ctx context.Context) error {
615 if f.j.Steps[finalStepRowDeleted] == "done" {
616 return nil
617 }
618 cfg := f.cfg()
619 host, id := f.rec.Host, f.g.ID
620 var err error
621 provision.WithHostHoldLock(host, func() {
622 fresh, e := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host)
623 if e != nil {
624 err = fmt.Errorf("re-read the record: %w", e)
625 return
626 }
627 if m := fresh.Move; m == nil || m.ID != id || m.Incarnation != f.g.Incarnation || m.State != provision.MoveCommitted {
628 err = errFinalChanged
629 return
630 }
631 if f.j.Steps[finalStepRowDeleted] != "pending" {
632 f.j.Mark(finalStepRowDeleted, "pending")
633 if err = f.j.Commit(cfg.DataDir); err != nil {
634 return
635 }
636 if err = f.fault("after-row-pending"); err != nil {
637 return
638 }
639 }
640 // false: no row is held by this op. With "pending" journalled that is
641 // this delete, already done; a row another op holds was never this
642 // move's, and is left as it is.
643 deleted, e := cfg.DeleteHeld(host, id)
644 if e != nil {
645 err = fmt.Errorf("delete the source's row: %w", e)
646 return
647 }
648 if deleted {
649 log.Printf("moves: %s: the source's row and its timers are deleted", host)
650 }
651 if err = f.fault("after-row-delete"); err != nil {
652 return
653 }
654 err = f.mark(finalStepRowDeleted, "done")
655 })
656 return err
657}
658
659// configAck copies this box's config ack for the host into the move's report
660// prefix and deletes it (§6.3). Absent is nothing to copy, or a replay after
661// the delete.
662func (f *sourceFinal) configAck(ctx context.Context) error {
663 cfg := f.cfg()
664 key := appconfig.ConfigAckKey(cfg.Box, f.rec.Host)
665 body, _, err := f.s.Get(ctx, key)
666 if errors.Is(err, os.ErrNotExist) {
667 return nil
668 }
669 if err != nil {
670 return fmt.Errorf("read the config ack: %w", err)
671 }
672 // The copy is create-only. A replay after the copy finds it: the same
673 // body is this copy; another (the ack was rewritten in between) is
674 // replaced under its etag, so the copy is the last ack the box wrote.
675 dst := moves.ReportPrefix(f.rec.Account, f.rec.App, f.rec.Host, f.g.ID) + finalConfigAckName
676 if _, err := f.s.Put(ctx, dst, body, ""); errors.Is(err, release.ErrCASConflict) {
677 had, etag, gerr := f.s.Get(ctx, dst)
678 if gerr != nil {
679 return fmt.Errorf("read the config ack's copy: %w", gerr)
680 }
681 if string(had) != string(body) {
682 if _, err := f.s.Put(ctx, dst, body, etag); err != nil {
683 return fmt.Errorf("copy the config ack: %w", err)
684 }
685 }
686 } else if err != nil {
687 return fmt.Errorf("copy the config ack: %w", err)
688 }
689 if err := f.fault("after-config-ack-copy"); err != nil {
690 return err
691 }
692 if err := f.s.Delete(ctx, key); err != nil {
693 return fmt.Errorf("delete the config ack: %w", err)
694 }
695 return nil
696}
697
698// stampParked is step 4: MoveSourceParked, then the nudge. A record that has
699// moved past this commit (closed, or another move) had it stamped, since the
700// close needs it; this commit without the stamp is an error.
701func (f *sourceFinal) stampParked(ctx context.Context) error {
702 host := f.rec.Host
703 var pre provision.Record
704 after, err := provision.SetRecordMove(ctx, f.s, f.rec.Account, f.rec.App, host, observedMove(&pre, provision.MoveSourceParked(f.g, f.now())))
705 if err == nil {
706 f.w.logTransition(ctx, f.s, pre, *after)
707 }
708 if errors.Is(err, provision.ErrMoveConflict) {
709 fresh, gerr := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host)
710 if gerr != nil {
711 return fmt.Errorf("re-read the record: %w", gerr)
712 }
713 if m := fresh.Move; m != nil && m.ID == f.g.ID && m.SourceParkedAt == nil {
714 return fmt.Errorf("stamp source_parked_at: %w", err)
715 }
716 err = nil
717 }
718 if err != nil {
719 return fmt.Errorf("stamp source_parked_at: %w", err)
720 }
721 if err := f.fault("after-source-parked-record"); err != nil {
722 return err
723 }
724 nudgeSerial(ctx, f.s)
725 log.Printf("moves: %s: the source has parked what it kept for %s", host, f.g.ID)
726 return nil
727}
728
729// parkJournal renames the op's journal directory into the local retirement,
730// after source_parked_at is stamped: until then it is what brings a replay
731// back here. A crash before the rename leaves the journal, and the next pass
732// (or the local sweep) renames it; after it nothing of the move is left
733// under .moves/.
734func (f *sourceFinal) parkJournal() error {
735 return parkFinalJournal(f.cfg().DataDir, f.rec.Host, f.g.ID, f.j, f.mark)
736}
737
738// parkFinalJournal parks a finalized op's journal into the retirement its
739// own local park recorded, journalling the destination just before it goes.
740func parkFinalJournal(dataDir, host, op string, j *moves.Journal, mark func(step, value string) error) error {
741 retirement := j.Steps[finalStepLocal]
742 if retirement == "" {
743 return fmt.Errorf("move %s: the journal records no local retirement for its own park", op)
744 }
745 plan, err := journalledLocalPlan(j, finalStepLocal)
746 if err != nil {
747 return err
748 }
749 dst := filepath.Join(parkedArtifactsDir(dataDir, plan.Incarnation, host, retirement), ".moves", host, op)
750 if j.Steps[finalStepJournal] != dst {
751 if err := mark(finalStepJournal, dst); err != nil {
752 return err
753 }
754 }
755 if err := parkOpJournal(dataDir, plan.Incarnation, host, op, retirement); err != nil {
756 return fmt.Errorf("park the op journal: %w", err)
757 }
758 log.Printf("moves: %s: the source's finalization of %s is complete", host, op)
759 return nil
760}
761
762// sweepFinalJournals is the local sweep of this box's op journals (fix round
763// 1, ruling 1; a rollback target's journals too, sweepRollbackJournal),
764// independent of any active move: a crash after the stamp and
765// before the journal's park, then the close, leaves a journal no record
766// routes back here (movesParticipantRecord answers nothing for a record with
767// no move). Each journal under <DataDir>/.moves/<host>/<op>/ that has reached
768// the stamp step is finished here:
769// - the stamp journalled "done": the journal is parked;
770// - "pending", and the listed record still carries this op: with
771// source_parked_at set it is journalled done and parked; without, it is
772// left to the finalization's own pass;
773// - "pending", and the listed record no longer carries this op (closed, or
774// another move since): the stamp landed, because the only transition
775// that clears a committed move is MoveClose, which refuses unless
776// source_parked_at is set (the aborts end at cutting/requested); so the
777// stamp is journalled done and the journal parked;
778// - a host the listing does not name (its record deleted, or unreadable)
779// is left as it is.
780//
781// A journal that has not reached the stamp step (no source-parked step: a
782// landing's, a cut's, a finalization still under way) is never touched, and
783// nothing is ever deleted: a journal is only renamed into its retirement.
784func (w *movesWorker) sweepFinalJournals(ctx context.Context, s release.Store, recs []provision.Record) {
785 dd := w.cfg.DataDir
786 if dd == "" {
787 return
788 }
789 hosts, err := os.ReadDir(filepath.Join(dd, ".moves"))
790 if err != nil {
791 if !errors.Is(err, os.ErrNotExist) {
792 w.notice("#final-sweep", "moves: the op journal sweep cannot list %s: %v", filepath.Join(dd, ".moves"), err)
793 }
794 return
795 }
796 byHost := map[string][]provision.Record{}
797 for _, rec := range recs {
798 byHost[rec.Host] = append(byHost[rec.Host], rec)
799 }
800 for _, hd := range hosts {
801 host := hd.Name()
802 if !hd.IsDir() || !journalComponent(host) {
803 continue
804 }
805 if w.hostBlocked(ctx, s, host) {
806 continue // the lost-box gate (movespass.go hostBlocked)
807 }
808 ops, err := os.ReadDir(filepath.Join(dd, ".moves", host))
809 if err != nil {
810 w.notice(host+"#final-sweep", "moves: %s: the op journal sweep cannot list its journals: %v", host, err)
811 continue
812 }
813 for _, od := range ops {
814 op := od.Name()
815 if !od.IsDir() || !journalComponent(op) {
816 continue
817 }
818 if err := w.sweepFinalJournal(dd, host, op, byHost[host]); err != nil {
819 w.logTrouble(host, fmt.Errorf("the op journal sweep, %s: %w", op, err))
820 }
821 }
822 }
823}
824
825// sweepFinalJournal is one journal of sweepFinalJournals.
826func (w *movesWorker) sweepFinalJournal(dataDir, host, op string, recs []provision.Record) error {
827 j, err := moves.ReadJournal(dataDir, host, op)
828 if err != nil {
829 return err
830 }
831 stamp, reached := j.Steps[finalStepParked]
832 if !reached {
833 // Not a finalization's: a rollback source's recovery journal
834 // (movesrecover.go, Task 5 fix round 1), or a rollback's target
835 // journal (movesrollback.go, Task 4 fix round 1).
836 if _, ok := j.Steps[rcStepSource]; ok {
837 return w.sweepRecoveryJournal(dataDir, host, op, j, recs)
838 }
839 if _, ok := j.Steps[rvStepSource]; ok {
840 return w.sweepRevertJournal(dataDir, host, op, j, recs) // movesrevert.go, Task 6
841 }
842 return w.sweepRollbackJournal(dataDir, host, op, j, recs)
843 }
844 if stamp != "done" {
845 if len(recs) == 0 {
846 w.notice(host+"#final-sweep-"+op, "moves: %s: the journal of %s reached its stamp, and no listed record names the host; it is left", host, op)
847 return nil
848 }
849 for _, rec := range recs {
850 if m := rec.Move; m != nil && m.ID == op && m.SourceParkedAt == nil {
851 return nil // the finalization's own pass stamps it
852 }
853 }
854 }
855 mark := func(step, value string) error {
856 j.Mark(step, value)
857 if err := j.Commit(dataDir); err != nil {
858 return err
859 }
860 if fn := w.cfg.finalFault; fn != nil {
861 return fn("after-" + step)
862 }
863 return nil
864 }
865 if stamp != "done" {
866 if err := mark(finalStepParked, "done"); err != nil {
867 return err
868 }
869 }
870 return parkFinalJournal(dataDir, host, op, j, mark)
871}
872