| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "sort" |
| 12 | "strconv" |
| 13 | "strings" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/hostclaim" |
| 17 | "github.com/carlosframework/platform/internal/hostfiles" |
| 18 | "github.com/carlosframework/platform/internal/hostguard" |
| 19 | "github.com/carlosframework/platform/internal/hostwrite" |
| 20 | "github.com/carlosframework/platform/internal/lease" |
| 21 | "github.com/carlosframework/platform/internal/moves" |
| 22 | "github.com/carlosframework/platform/internal/moves/archive" |
| 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 | ) |
| 28 | |
| 29 | // The generation installer (moves spec §7.6, §7.8; Plan 4 Task 3), shared by |
| 30 | // the landing on the target and the dirty return on the source. It runs the |
| 31 | // six phases parking, building, published, local, configured and restored |
| 32 | // exactly as Plan 2 Task 10 wrote them for the landing; installSource says |
| 33 | // what differs between the two kinds of install (rulings J13, J14, J20, J24): |
| 34 | // |
| 35 | // - the claim check: a landing's op is landing at the phase, mirrored into |
| 36 | // the claim by every record step; a return's op is restoring/claimed with |
| 37 | // the source executing at its pre-move stamp S throughout, and the claim |
| 38 | // does not follow the source's install phases; |
| 39 | // - the record check: landing at the phase with no rollback reason, or |
| 40 | // rolling_back, dirty, with the source at the phase; |
| 41 | // - the refusal action: a landing marks its rollback (MoveMarkRollback); a |
| 42 | // return stops for an operator (MoveMarkStuck) with a recorded reason; |
| 43 | // - the phase advance: MoveLandingPhase or MoveRollbackSourcePhase, bound |
| 44 | // by the caller, so neither kind is told apart by a type switch; |
| 45 | // - the generation: the move id for a landing, the rollback id for a return |
| 46 | // (J14: a return is cold and never reuses S's generation name). It names |
| 47 | // the build, the install's own journal, <dbPath>.gen and .files-gen; |
| 48 | // - configured: a landing renders and requires its pinned inputs; a return |
| 49 | // renders the current ones and records them in Rollback.RecoveryInputs, |
| 50 | // never compared with move.inputs (J2-R2-5); |
| 51 | // - local: a landing parks an earlier stay's artifacts tagged S'; a return |
| 52 | // parks the source's own pre-move installation before anything is |
| 53 | // extracted (J13, fix round 1): RetireArtifacts for S and for every |
| 54 | // stamp the claim shows superseded on this box, a stuck mark for any |
| 55 | // other stamp, then the moves-owned rename (H5(b), parkLocalArtifacts) |
| 56 | // of every pre-move artifact still at the flat paths, inside the host |
| 57 | // guard. placeStaged and restore then find nothing there, and refuse |
| 58 | // anything that appears; |
| 59 | // - the authoritative region the claim's view names for its parks (J24, |
| 60 | // §5.1): the source's for a landing, the target's for a return, each the |
| 61 | // other side of the move before this side's probe is admitted. |
| 62 | // |
| 63 | // The phase-name convention is Plan 2's (G8): restored is written by the CAS |
| 64 | // that ends configured, after the database is installed and its markers |
| 65 | // written. Every record write goes through a provision transition (the |
| 66 | // record_seq rule). The installer is unarmed in production, as the landing |
| 67 | // is (until the wake-source gate opens, movesWakeSourceRegistered, Plan 5 |
| 68 | // Task 9b): only the landing's pass, |
| 69 | // the recovery and the tests call it. |
| 70 | |
| 71 | // installStepPreMove is the moves-owned rename of a return's pre-move |
| 72 | // artifacts (parkLocalArtifacts' step: the retirement id). |
| 73 | const installStepPreMove = "pre_move_parked" |
| 74 | |
| 75 | // installStepResidueKept is a dirty return's no-op parking (Task 9a fix |
| 76 | // round 3, ruling (a)): the sentence saying the target never admitted the |
| 77 | // move, so the pre-move generation stays the source's live state. |
| 78 | const installStepResidueKept = "residue_kept" |
| 79 | |
| 80 | // installStamp is what a phase step records besides the phase: at the CAS |
| 81 | // that ends configured, the installed database's digest (a landing records |
| 82 | // it in landing.installed_db_sha256) and the inputs a return rendered. |
| 83 | type installStamp struct { |
| 84 | InstalledDBSHA256 string |
| 85 | RecoveryInputs *provision.MoveInputs |
| 86 | } |
| 87 | |
| 88 | // installRefusal is why an install stops for good: installSource.Refuse maps |
| 89 | // it to its kind's recorded sentence. |
| 90 | type installRefusal int |
| 91 | |
| 92 | const ( |
| 93 | refuseFilesOccupied installRefusal = iota + 1 // local: the files paths hold what the install did not write |
| 94 | refuseInstallMismatch // configured: the build does not verify |
| 95 | refuseConfigChanged // configured: a landing's render is not its pin |
| 96 | refuseDeadline // configured: a landing ran past its deadline |
| 97 | refuseUnknownStamp // local: a return found paths under a stamp it cannot account for |
| 98 | refuseManifestOccupied // local: the prepared files manifest's key holds a manifest the install did not write |
| 99 | ) |
| 100 | |
| 101 | // installSource is one kind of install: what the six phases need from the |
| 102 | // caller (Plan 4 Task 3's interface, with the rulings' additions). |
| 103 | type installSource struct { |
| 104 | OpID, Host string |
| 105 | // Role is roleTarget for a landing, roleSource for a dirty return. |
| 106 | Role string |
| 107 | // Name is the install's word in the log: "landing" or "return". |
| 108 | Name string |
| 109 | // Lane is the staging or return lane prefix in this box's own bucket; |
| 110 | // every object the seal names must be under it. |
| 111 | Lane string |
| 112 | // Seal reads the sealed set from the claim the phase works from. |
| 113 | Seal func(*hostclaim.Claim) (moves.Seal, error) |
| 114 | // Baseline is what the sealed set restores to: the final TXID, the |
| 115 | // files and timers digests, and the carried columns. |
| 116 | Baseline *provision.MoveSource |
| 117 | // GenerationOp names the generation this install builds and the |
| 118 | // install's own journal; HoldOp is the op the row is held by. |
| 119 | GenerationOp, HoldOp string |
| 120 | // PinInputs are a landing's pinned inputs; nil for a return. |
| 121 | PinInputs *provision.MoveInputs |
| 122 | // Phases are the six phase names of this install, in order. |
| 123 | Phases [6]string |
| 124 | // Phase is the record transition one step makes, bound to the guard. |
| 125 | Phase func(from, to string, st installStamp) func(*provision.Record) error |
| 126 | // Mirror is the claim half of a record step; nil when the claim does |
| 127 | // not follow this install's phases. |
| 128 | Mirror func(ctx context.Context, from, to string) error |
| 129 | // Sync is each phase's first claim read (a landing walks a claim that |
| 130 | // is behind forward); ClaimCheck is what the phase and the guard require |
| 131 | // of the claim at phase. |
| 132 | Sync func(ctx context.Context) (hostclaim.Observed, error) |
| 133 | ClaimCheck func(phase string) func(*hostclaim.Claim) error |
| 134 | // RecordPhase answers the phase a record is at while it is still this |
| 135 | // install's, and an error otherwise. |
| 136 | RecordPhase func(*provision.Record) (string, error) |
| 137 | // Refuse writes the refusal: written is true when the record took it, |
| 138 | // sentence is the recorded reason. |
| 139 | Refuse func(ctx context.Context, why installRefusal) (written bool, sentence string, err error) |
| 140 | // PastDeadline reports a landing's deadline passed; nil for a return. |
| 141 | PastDeadline func() bool |
| 142 | } |
| 143 | |
| 144 | func (src installSource) validate() error { |
| 145 | if src.OpID == "" || src.Host == "" || (src.Role != roleTarget && src.Role != roleSource) || src.Lane == "" || src.Seal == nil || |
| 146 | src.GenerationOp == "" || src.HoldOp == "" || src.Phase == nil || src.Sync == nil || src.ClaimCheck == nil || |
| 147 | src.RecordPhase == nil || src.Refuse == nil { |
| 148 | return errors.New("moves: an install source names its op, role, lane, seal, generation and hooks") |
| 149 | } |
| 150 | for _, p := range src.Phases { |
| 151 | if p == "" { |
| 152 | return errors.New("moves: an install source names its six phases") |
| 153 | } |
| 154 | } |
| 155 | return nil |
| 156 | } |
| 157 | |
| 158 | // kind is the claim op kind this install runs under. |
| 159 | func (src installSource) kind() string { |
| 160 | if src.Role == roleSource { |
| 161 | return hostclaim.KindRestoring |
| 162 | } |
| 163 | return hostclaim.KindLanding |
| 164 | } |
| 165 | |
| 166 | // installRun is one call's state for one install. |
| 167 | type installRun struct { |
| 168 | w *movesWorker |
| 169 | s release.Store |
| 170 | rec provision.Record |
| 171 | g provision.MoveGuard |
| 172 | src installSource |
| 173 | guard hostguard.Guard |
| 174 | local park.Local |
| 175 | // refused is the sentence a refusal recorded during this call. |
| 176 | refused string |
| 177 | } |
| 178 | |
| 179 | // installGeneration runs the record's current install phase once (one phase |
| 180 | // per call, as the landing's pass and the cut step): parking, building, |
| 181 | // published, local or configured. advanced is true when it wrote the record |
| 182 | // (a phase step or the refusal); refused is the recorded sentence when the |
| 183 | // refusal was written. A record at restored is the probe's and is left as it |
| 184 | // is; one that is no longer this install's is an error. |
| 185 | func installGeneration(ctx context.Context, s release.Store, w *movesWorker, rec provision.Record, src installSource) (advanced bool, refused string, err error) { |
| 186 | if err := src.validate(); err != nil { |
| 187 | return false, "", err |
| 188 | } |
| 189 | phase, err := src.RecordPhase(&rec) |
| 190 | if err != nil { |
| 191 | return false, "", err |
| 192 | } |
| 193 | if rec.Move.ID != src.OpID || rec.Host != src.Host { |
| 194 | return false, "", fmt.Errorf("moves: the install of %s is not %s's", src.OpID, rec.Host) |
| 195 | } |
| 196 | ir := newInstallRun(s, w, rec, src) |
| 197 | switch phase { |
| 198 | case src.Phases[0]: |
| 199 | advanced, err = ir.parking(ctx) |
| 200 | case src.Phases[1]: |
| 201 | advanced, err = ir.building(ctx) |
| 202 | case src.Phases[2]: |
| 203 | advanced, err = ir.published(ctx) |
| 204 | case src.Phases[3]: |
| 205 | advanced, err = ir.local_(ctx) |
| 206 | case src.Phases[4]: |
| 207 | advanced, err = ir.configured(ctx) |
| 208 | case src.Phases[5]: |
| 209 | return false, "", nil |
| 210 | default: |
| 211 | return false, "", fmt.Errorf("moves: %s is not a phase of the %s's install", phase, src.Name) |
| 212 | } |
| 213 | return advanced, ir.refused, err |
| 214 | } |
| 215 | |
| 216 | func newInstallRun(s release.Store, w *movesWorker, rec provision.Record, src installSource) *installRun { |
| 217 | cfg := w.cfg |
| 218 | ir := &installRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: rec.Move.ID, Incarnation: rec.Move.Incarnation}, src: src} |
| 219 | ir.guard = cfg.Guard |
| 220 | if ir.guard.Claims == nil { |
| 221 | ir.guard.Claims = hostguard.StoreReader{Store: s} |
| 222 | } |
| 223 | ir.local = cfg.Local |
| 224 | ir.local.Guard, ir.local.Claims = ir.guard, s |
| 225 | if ir.local.Box == "" { |
| 226 | ir.local.Box = cfg.Box |
| 227 | } |
| 228 | if ir.local.Authorize == nil { |
| 229 | // park.Local's authorizer: the claim's MoveParks local authorizer |
| 230 | // (ParkLocalAuthorize) behind this install's own check; the claim's |
| 231 | // engine authorizer refuses a park-local (M9). Without it every local |
| 232 | // chore refuses (landingArmed requires it). |
| 233 | ir.local.Authorize = movesLocalAuthorize(cfg.ParkLocalAuthorize, ir.authorizeChore) |
| 234 | } |
| 235 | return ir |
| 236 | } |
| 237 | |
| 238 | func (ir *installRun) cfg() moveCfg { return ir.w.cfg } |
| 239 | |
| 240 | // installAuthority is the install's authority re-check for the prepared |
| 241 | // manifest's bucket writes (fix round 1, finding 1; region HA's |
| 242 | // authorityStore pattern): a fresh read of the claim, the install's own |
| 243 | // claim check at phase (the op at that phase with this box executing) and |
| 244 | // the holder still at the captured stamp. It runs immediately before every |
| 245 | // write attempt, retries included, so a delayed build whose executor, |
| 246 | // phase or stamp moved writes nothing more. |
| 247 | func (ir *installRun) installAuthority(ctx context.Context, captured hostguard.Captured, phase string) func() error { |
| 248 | check := ir.src.ClaimCheck(phase) |
| 249 | return func() error { |
| 250 | if err := ctx.Err(); err != nil { |
| 251 | return err |
| 252 | } |
| 253 | obs, err := hostclaim.Read(ctx, ir.s, ir.rec.Host) |
| 254 | if err != nil { |
| 255 | return err |
| 256 | } |
| 257 | if obs.Absent || obs.Claim == nil { |
| 258 | return fmt.Errorf("%w: %s has no claim", errLandingChanged, ir.rec.Host) |
| 259 | } |
| 260 | if err := check(obs.Claim); err != nil { |
| 261 | return err |
| 262 | } |
| 263 | if obs.Claim.Holder == nil || obs.Claim.Holder.Stamp != captured.Stamp { |
| 264 | return fmt.Errorf("%w: the claim's holder is not at the installation stamp %d", errLandingChanged, captured.Stamp) |
| 265 | } |
| 266 | return nil |
| 267 | } |
| 268 | } |
| 269 | |
| 270 | // installAuthorityStore is the versioned store the prepared build writes |
| 271 | // through: every write re-checks the install's authority first. Reads are |
| 272 | // the store's own; the build never deletes. |
| 273 | type installAuthorityStore struct { |
| 274 | lease.VersionedStore |
| 275 | check func() error |
| 276 | } |
| 277 | |
| 278 | func (s installAuthorityStore) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) { |
| 279 | if err := s.check(); err != nil { |
| 280 | return "", err |
| 281 | } |
| 282 | return s.VersionedStore.PutIfAbsent(ctx, bucket, key, body) |
| 283 | } |
| 284 | |
| 285 | func (s installAuthorityStore) CopyVersion(ctx context.Context, sb, sk, sv, db, dk string, createOnly bool) (string, string, error) { |
| 286 | if err := s.check(); err != nil { |
| 287 | return "", "", err |
| 288 | } |
| 289 | return s.VersionedStore.CopyVersion(ctx, sb, sk, sv, db, dk, createOnly) |
| 290 | } |
| 291 | |
| 292 | func (installAuthorityStore) DeleteVersion(context.Context, string, string, string) error { |
| 293 | return fmt.Errorf("%w: the prepared build never deletes", errLandingChanged) |
| 294 | } |
| 295 | |
| 296 | // preparedManifestJournalled reports whether the journal records the |
| 297 | // prepared files manifest as built: the step done with this install's |
| 298 | // binding (decodePreparedManifest: P's key, the op and the captured |
| 299 | // stamp), not the "skipped" value journals written before C8 existed, |
| 300 | // which a replay builds (ruling Q3 item 2; the step itself is create-only |
| 301 | // and replay-safe). Any other value is an error (fix round 1, finding 3): |
| 302 | // an empty or malformed step, or a binding for another key, op or stamp, |
| 303 | // is a damaged or foreign journal, and the install neither advances over |
| 304 | // it nor rebuilds over it. |
| 305 | func preparedManifestJournalled(j *moves.Journal, key, op string, stamp uint64) (bool, error) { |
| 306 | v, ok := j.Steps[landStepFilesManifest] |
| 307 | if !ok || v == landingFilesManifestSkipped { |
| 308 | return false, nil |
| 309 | } |
| 310 | if _, err := decodePreparedManifest(v, key, op, stamp); err != nil { |
| 311 | return false, err |
| 312 | } |
| 313 | return true, nil |
| 314 | } |
| 315 | |
| 316 | // fault is the test seam after a side effect (the landing's points). |
| 317 | func (ir *installRun) fault(point string) error { |
| 318 | if f := ir.w.cfg.landFault; f != nil { |
| 319 | return f(point) |
| 320 | } |
| 321 | return nil |
| 322 | } |
| 323 | |
| 324 | // sync is each phase's first step: the kind's claim read, then its check at |
| 325 | // phase (K5's revalidation). |
| 326 | func (ir *installRun) sync(ctx context.Context, phase string) (hostclaim.Observed, error) { |
| 327 | obs, err := ir.src.Sync(ctx) |
| 328 | if err != nil { |
| 329 | return obs, err |
| 330 | } |
| 331 | if err := ir.src.ClaimCheck(phase)(obs.Claim); err != nil { |
| 332 | return obs, fmt.Errorf("move %s: %w", ir.g.ID, err) |
| 333 | } |
| 334 | return obs, nil |
| 335 | } |
| 336 | |
| 337 | // seal reads the sealed set from the claim and requires every object it |
| 338 | // names in this install's lane. |
| 339 | func (ir *installRun) seal(c *hostclaim.Claim) (moves.Seal, error) { |
| 340 | seal, err := ir.src.Seal(c) |
| 341 | if err != nil { |
| 342 | return seal, err |
| 343 | } |
| 344 | for _, class := range [][]moves.SealedObject{seal.Replica, seal.Files, seal.Timers, seal.State} { |
| 345 | for _, o := range class { |
| 346 | if !strings.HasPrefix(o.Key, ir.src.Lane) { |
| 347 | return seal, fmt.Errorf("%w: %s is not in the %s's lane %s", moves.ErrBadSeal, o.Key, ir.src.Name, ir.src.Lane) |
| 348 | } |
| 349 | } |
| 350 | } |
| 351 | return seal, nil |
| 352 | } |
| 353 | |
| 354 | // transition writes one record transition. A conflict is someone else's |
| 355 | // decision (a rollback, a stuck mark, a console write): logged, and the call |
| 356 | // ends without error. |
| 357 | func (ir *installRun) transition(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) { |
| 358 | var pre provision.Record |
| 359 | after, err := provision.SetRecordMove(ctx, ir.s, ir.rec.Account, ir.rec.App, ir.rec.Host, observedMove(&pre, mut)) |
| 360 | if err != nil { |
| 361 | if errors.Is(err, provision.ErrMoveConflict) { |
| 362 | log.Printf("moves: %s: %s of %s refused (%v); the record moved on", ir.rec.Host, what, ir.g.ID, err) |
| 363 | return false, nil |
| 364 | } |
| 365 | return false, fmt.Errorf("%s: %w", what, err) |
| 366 | } |
| 367 | ir.w.logTransition(ctx, ir.s, pre, *after) |
| 368 | return true, nil |
| 369 | } |
| 370 | |
| 371 | // advance writes the record's step and then, for a landing, mirrors it into |
| 372 | // the claim. A mirror that fails is logged: the next pass's sync writes it. |
| 373 | func (ir *installRun) advance(ctx context.Context, from, to string, st installStamp) (bool, error) { |
| 374 | name := ir.src.Name |
| 375 | ok, err := ir.transition(ctx, "the "+name+" "+from+" to "+to, ir.src.Phase(from, to, st)) |
| 376 | if !ok { |
| 377 | return ok, err |
| 378 | } |
| 379 | log.Printf("moves: %s: %s %s is %s", ir.rec.Host, name, ir.g.ID, to) |
| 380 | if err := ir.fault("after-record-" + to); err != nil { |
| 381 | return true, err |
| 382 | } |
| 383 | if ir.src.Mirror != nil { |
| 384 | if err := ir.src.Mirror(ctx, from, to); err != nil { |
| 385 | log.Printf("moves: %s: the claim half of %s %s waits for the next pass: %v", ir.rec.Host, name, to, err) |
| 386 | } |
| 387 | } |
| 388 | return true, nil |
| 389 | } |
| 390 | |
| 391 | // refuse writes the kind's refusal; written is true when the record took it. |
| 392 | func (ir *installRun) refuse(ctx context.Context, why installRefusal) (bool, error) { |
| 393 | ok, sentence, err := ir.src.Refuse(ctx, why) |
| 394 | if ok { |
| 395 | ir.refused = sentence |
| 396 | } |
| 397 | return ok, err |
| 398 | } |
| 399 | |
| 400 | func (ir *installRun) pastDeadline() bool { return ir.src.PastDeadline != nil && ir.src.PastDeadline() } |
| 401 | |
| 402 | // route is the host's registry row on this box. |
| 403 | func (ir *installRun) route() (registry.Route, error) { |
| 404 | rt, err := ir.cfg().Route(ir.rec.Host) |
| 405 | if err != nil { |
| 406 | return registry.Route{}, fmt.Errorf("read the registry row: %w", err) |
| 407 | } |
| 408 | if rt == nil { |
| 409 | return registry.Route{}, fmt.Errorf("%s has no registry row on this box", ir.rec.Host) |
| 410 | } |
| 411 | return *rt, nil |
| 412 | } |
| 413 | |
| 414 | func (ir *installRun) dbPath() string { return provision.DefaultDBPath(ir.cfg().DataDir, ir.rec.Host) } |
| 415 | |
| 416 | // journal is the install's own op journal, named by its generation. |
| 417 | func (ir *installRun) journal() (*moves.Journal, error) { |
| 418 | return moves.ReadJournal(ir.cfg().DataDir, ir.rec.Host, ir.src.GenerationOp) |
| 419 | } |
| 420 | |
| 421 | func (ir *installRun) journalDir() string { |
| 422 | return moves.JournalDir(ir.cfg().DataDir, ir.rec.Host, ir.src.GenerationOp) |
| 423 | } |
| 424 | |
| 425 | // buildDir is BuildGeneration's scratch, <JournalDir>/build/ (ruling G34), |
| 426 | // on the data disk so the result renames into DBPath. |
| 427 | func (ir *installRun) buildDir() string { return filepath.Join(ir.journalDir(), "build") } |
| 428 | |
| 429 | // versioned is this box's own bucket by exact version. |
| 430 | func (ir *installRun) versioned(ctx context.Context) (lease.VersionedStore, error) { |
| 431 | cfg := ir.cfg() |
| 432 | if cfg.Versioned == nil { |
| 433 | return nil, errors.New("no versioned store on this box") |
| 434 | } |
| 435 | return cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion) |
| 436 | } |
| 437 | |
| 438 | // parking (§7.6, D6): whatever an earlier stay of this incarnation left at |
| 439 | // holder.location in this box's own bucket is parked before the build |
| 440 | // writes beside it, one park-generation chore per generation (the origin as |
| 441 | // @root), scoped to the location with the flip's time as the evidence |
| 442 | // horizon. On the target that is an earlier stay's residue; on the source |
| 443 | // of a dirty return it is the pre-move generations (§7.8, D6). Each chore is |
| 444 | // run once per pass through the park engine. The location's lease is the |
| 445 | // install's own, held from here (a landing's from checking) through |
| 446 | // published, and each chore runs under it through park.Engine.RunHeld over |
| 447 | // the claim's leases (ParkRunHeld, M5). ErrPending is the expected answer: the acknowledgement waits for |
| 448 | // the horizon, and is C4's runner's. The phase advances once every chore's |
| 449 | // manifest exists and its deletion set is gone (Engine.Parked). |
| 450 | // Per-generation chores are what let the acknowledgement finish after the |
| 451 | // probe makes the new generation live here. |
| 452 | // |
| 453 | // A replay finds parking and continues from the claim: the residue list is |
| 454 | // journalled so the chores are the same ones, the queue is idempotent per |
| 455 | // generation, and Engine.Run resumes a chore from its manifest. |
| 456 | func (ir *installRun) parking(ctx context.Context) (bool, error) { |
| 457 | cfg := ir.cfg() |
| 458 | host, id, gen := ir.rec.Host, ir.g.ID, ir.src.GenerationOp |
| 459 | phase, next := ir.src.Phases[0], ir.src.Phases[1] |
| 460 | obs, err := ir.sync(ctx, phase) |
| 461 | if err != nil { |
| 462 | return false, err |
| 463 | } |
| 464 | holder := *obs.Claim.Holder |
| 465 | location := strings.TrimSuffix(holder.Location, "/") |
| 466 | vs, err := ir.versioned(ctx) |
| 467 | if err != nil { |
| 468 | return false, err |
| 469 | } |
| 470 | j, err := ir.journal() |
| 471 | if err != nil { |
| 472 | return false, err |
| 473 | } |
| 474 | _, listed := j.Steps[landStepResidue] |
| 475 | if !listed && ir.src.Role == roleSource && !targetAdmitted(ir.rec.Move, obs.Claim) { |
| 476 | // A dirty return whose target never admitted this move (ruling |
| 477 | // (a), fix round 3): holder.generation is still the pre-move one, |
| 478 | // the source's live state, so there is no residue to park. This |
| 479 | // is decided from the claim's recorded admissions only (fix round |
| 480 | // 4: not the landing phase), never from a park view's |
| 481 | // refusal, which can be transient. |
| 482 | why := fmt.Sprintf("the target never admitted move %s: the claim holds no admission of its generation; the pre-move generation stays the source's live state", id) |
| 483 | j.Phase = phase |
| 484 | j.Mark(landStepResidue, "[]") |
| 485 | j.Mark(installStepResidueKept, why) |
| 486 | if err := j.Commit(cfg.DataDir); err != nil { |
| 487 | return false, err |
| 488 | } |
| 489 | log.Printf("moves: %s: the %s parks nothing: %s", host, ir.src.Name, why) |
| 490 | return ir.advance(ctx, phase, next, installStamp{}) |
| 491 | } |
| 492 | var gens []string |
| 493 | if raw, ok := j.Steps[landStepResidue]; ok { |
| 494 | if err := json.Unmarshal([]byte(raw), &gens); err != nil { |
| 495 | return false, fmt.Errorf("the journal's residue list: %w", err) |
| 496 | } |
| 497 | } else { |
| 498 | objs, err := vs.ListVersions(ctx, cfg.OwnBucket, location+"/") |
| 499 | if err != nil { |
| 500 | return false, fmt.Errorf("list %s: %w", location, err) |
| 501 | } |
| 502 | seen := map[string]bool{} |
| 503 | for _, o := range objs { |
| 504 | if o.Key == location+"/lock.json" { |
| 505 | continue |
| 506 | } |
| 507 | if movesHolderBuildState(location, o.Key) { |
| 508 | // The build state beside the lease is the marker's twin |
| 509 | // (ruling R-PARK): no generation's, never residue here. |
| 510 | // The build places the moved state over it. |
| 511 | continue |
| 512 | } |
| 513 | g, ok := landingGenerationOf(location, o.Key) |
| 514 | if !ok { |
| 515 | return false, fmt.Errorf("%s is not an object of a generation this %s can classify", o.Key, ir.src.Name) |
| 516 | } |
| 517 | if g == gen || g == landingMarkerGeneration { |
| 518 | // The install's own generation is never residue, and the |
| 519 | // marker is superseded by the probe, never parked. |
| 520 | continue |
| 521 | } |
| 522 | if g == "" { |
| 523 | g = hostclaim.RootGeneration |
| 524 | } |
| 525 | seen[g] = true |
| 526 | } |
| 527 | for g := range seen { |
| 528 | gens = append(gens, g) |
| 529 | } |
| 530 | sort.Strings(gens) |
| 531 | body, err := json.Marshal(gens) |
| 532 | if err != nil { |
| 533 | return false, err |
| 534 | } |
| 535 | j.Phase = phase |
| 536 | j.Mark(landStepResidue, string(body)) |
| 537 | if err := j.Commit(cfg.DataDir); err != nil { |
| 538 | return false, err |
| 539 | } |
| 540 | } |
| 541 | if len(gens) == 0 { |
| 542 | return ir.advance(ctx, phase, next, installStamp{}) |
| 543 | } |
| 544 | if ir.rec.Move.FlippedAt == nil { |
| 545 | return false, fmt.Errorf("move %s: the record has no flip time", id) |
| 546 | } |
| 547 | // The parks run under the build's own lease on the location, which the |
| 548 | // install holds from here (a landing from checking) through published: |
| 549 | // the residue scope's lease set is that lock (M6), so Engine.Run would |
| 550 | // wait on the install itself. park.Engine.RunHeld (ParkRunHeld) runs |
| 551 | // each chore under the held lease instead, borrowed through the claim's |
| 552 | // AggregateLeases (M5). |
| 553 | held, waiting, err := ir.lease(ctx, holder) |
| 554 | if waiting || err != nil { |
| 555 | return false, err |
| 556 | } |
| 557 | scope := moves.ResidueScope(holder, *ir.rec.Move.FlippedAt) |
| 558 | if obs, err = hostclaim.Update(ctx, ir.s, obs, hostclaim.QueueResidueParks(id, cfg.parkRegion(), scope, gens, obligationID)); err != nil { |
| 559 | return false, fmt.Errorf("queue the residue parks: %w", err) |
| 560 | } |
| 561 | if err := ir.fault("after-residue-queued"); err != nil { |
| 562 | return false, err |
| 563 | } |
| 564 | eng := ir.engine(vs) |
| 565 | pending := 0 |
| 566 | for _, o := range obs.Claim.Obligations { |
| 567 | if o.Op != id || o.Kind != "park-generation" || !park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) { |
| 568 | continue |
| 569 | } |
| 570 | w := park.Work{Host: host, ID: o.ID, Op: id} |
| 571 | if cfg.ParkRunHeld == nil { |
| 572 | return false, fmt.Errorf("move %s: %w", id, &ClaimDependencyError{Contracts: []string{"the claim's MoveParks RunHeld (armMovesParks)"}}) |
| 573 | } |
| 574 | if err := cfg.ParkRunHeld(ctx, eng, w, held); err != nil && !errors.Is(err, park.ErrPending) { |
| 575 | return false, fmt.Errorf("park generation %s: %w", o.Generation, err) |
| 576 | } |
| 577 | parked, err := eng.Parked(ctx, w) |
| 578 | if errors.Is(err, park.ErrChanged) { |
| 579 | continue // acknowledged already |
| 580 | } |
| 581 | if err != nil { |
| 582 | return false, fmt.Errorf("park generation %s: %w", o.Generation, err) |
| 583 | } |
| 584 | if !parked { |
| 585 | pending++ |
| 586 | } |
| 587 | } |
| 588 | if pending > 0 { |
| 589 | ir.w.notice(host+"#parking", "moves: %s: %d residue park(s) are not complete yet", host, pending) |
| 590 | return false, nil |
| 591 | } |
| 592 | return ir.advance(ctx, phase, next, installStamp{}) |
| 593 | } |
| 594 | |
| 595 | // targetAdmitted reports whether the claim carries evidence that the |
| 596 | // move's target admitted a writer: the holder's admissions include the |
| 597 | // move's prepared generation (an admission appends its record atomically, |
| 598 | // and ReturnToSource keeps them). The record's landing phase is not |
| 599 | // evidence: it enters probing before the probe runs (fix round 4), so a |
| 600 | // landing that stopped at probing before its admission never admitted. |
| 601 | func targetAdmitted(m *provision.MoveState, c *hostclaim.Claim) bool { |
| 602 | if m == nil || c == nil || c.Holder == nil { |
| 603 | return false |
| 604 | } |
| 605 | for _, a := range c.Holder.Admissions { |
| 606 | if a.Generation == m.ID { |
| 607 | return true |
| 608 | } |
| 609 | } |
| 610 | return false |
| 611 | } |
| 612 | |
| 613 | // building (§7.6, §5.1 step 2): the generation built from the seal's pinned |
| 614 | // replica versions under the location's lease (moves.BuildGeneration: the |
| 615 | // intent, the snapshot, the anchor, the completed install manifest, every |
| 616 | // one create-only), with the result left at <JournalDir>/build/<host>.db |
| 617 | // for the restore. The manifest goes into the journal. |
| 618 | // |
| 619 | // A replay finds building and builds again: BuildGeneration reuses what it |
| 620 | // finds under the generation's keys and refuses anything it did not make. |
| 621 | func (ir *installRun) building(ctx context.Context) (bool, error) { |
| 622 | cfg := ir.cfg() |
| 623 | host, id, gen := ir.rec.Host, ir.g.ID, ir.src.GenerationOp |
| 624 | phase, next := ir.src.Phases[1], ir.src.Phases[2] |
| 625 | obs, err := ir.sync(ctx, phase) |
| 626 | if err != nil { |
| 627 | return false, err |
| 628 | } |
| 629 | holder := *obs.Claim.Holder |
| 630 | held, waiting, err := ir.lease(ctx, holder) |
| 631 | if waiting || err != nil { |
| 632 | return false, err |
| 633 | } |
| 634 | seal, err := ir.seal(obs.Claim) |
| 635 | if err != nil { |
| 636 | return false, err |
| 637 | } |
| 638 | src := ir.src.Baseline |
| 639 | if src == nil || seal.N != src.FinalTXID { |
| 640 | return false, fmt.Errorf("move %s: the seal is at TXID %d and the record at %d", id, seal.N, srcTXID(src)) |
| 641 | } |
| 642 | vs, err := ir.versioned(ctx) |
| 643 | if err != nil { |
| 644 | return false, err |
| 645 | } |
| 646 | if err := held.Check(ctx); err != nil { |
| 647 | return false, fmt.Errorf("the location's lease: %w", err) |
| 648 | } |
| 649 | // Every generation request runs under live lease authority (fix round |
| 650 | // 1, item 3): the store checks the lease before each call, as the cut's |
| 651 | // guarded writes check theirs, the requests run under a context the |
| 652 | // lease's loss cancels (leaseContext, fix round 2), and the lease is |
| 653 | // checked again before the build's completion is recorded. |
| 654 | m, err := moves.BuildGeneration(leaseContext(ctx, held), moves.BuildInputs{Store: leasedStore{VersionedStore: vs, check: held.Check}, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N, |
| 655 | GenerationPrefix: moves.GenerationPrefix(holder.Location, gen), Op: gen, Host: host, WorkDir: ir.buildDir(), Now: cfg.Now}) |
| 656 | if err != nil { |
| 657 | return false, fmt.Errorf("build the generation: %w", err) |
| 658 | } |
| 659 | if err := ir.placeBuildState(leaseContext(ctx, held), leasedStore{VersionedStore: vs, check: held.Check}, seal, holder); err != nil { |
| 660 | return false, err |
| 661 | } |
| 662 | if err := held.Check(ctx); err != nil { |
| 663 | return false, fmt.Errorf("the location's lease after the build: %w", err) |
| 664 | } |
| 665 | body, err := json.Marshal(m) |
| 666 | if err != nil { |
| 667 | return false, err |
| 668 | } |
| 669 | j, err := ir.journal() |
| 670 | if err != nil { |
| 671 | return false, err |
| 672 | } |
| 673 | j.Phase = phase |
| 674 | j.Mark(landStepInstall, string(body)) |
| 675 | if err := j.Commit(cfg.DataDir); err != nil { |
| 676 | return false, err |
| 677 | } |
| 678 | if err := ir.fault("after-build"); err != nil { |
| 679 | return false, err |
| 680 | } |
| 681 | log.Printf("moves: %s: generation %s built at TXID %d", host, gen, m.N) |
| 682 | return ir.advance(ctx, phase, next, installStamp{}) |
| 683 | } |
| 684 | |
| 685 | // placeBuildState copies the sealed build state (the seal's State class, |
| 686 | // the lane's state/ key) beside the location's lease, |
| 687 | // holder.Location/build-state.json in this box's own bucket (settings spec |
| 688 | // rev 11 §4.3a, ruling R-LANDING). building calls it under the location's |
| 689 | // lease, so it lands before the probe and the first wake: their record |
| 690 | // merges onto the moved state, and a held instance never reads as new |
| 691 | // here. It copies exactly the sealed version (ruling R-SEAL-STATE): a |
| 692 | // request a stale round signed can land a newer version in the lane until |
| 693 | // the seal's bound, and that version is never placed. A seal with no state |
| 694 | // (the source had none, or a seal from before the class existed) places |
| 695 | // nothing, and removes every version of a build state an earlier stay left |
| 696 | // at the location (Task 12a): it is the marker's twin, so the landing's park |
| 697 | // passes it over, and otherwise that stay's state would read as this |
| 698 | // instance's. A replay copies, or removes, again. |
| 699 | func (ir *installRun) placeBuildState(ctx context.Context, vs lease.VersionedStore, seal moves.Seal, holder hostclaim.Holder) error { |
| 700 | bucket := ir.cfg().OwnBucket |
| 701 | dst := release.BuildStateKey(strings.TrimSuffix(holder.Location, "/")) |
| 702 | if len(seal.State) == 0 { |
| 703 | vers, err := vs.ListVersions(ctx, bucket, dst) |
| 704 | if err != nil { |
| 705 | return fmt.Errorf("list a leftover build state at %s: %w", dst, err) |
| 706 | } |
| 707 | for _, v := range vers { |
| 708 | if v.Key != dst { |
| 709 | continue |
| 710 | } |
| 711 | if err := vs.DeleteVersion(ctx, bucket, dst, v.VersionID); err != nil { |
| 712 | return fmt.Errorf("remove a leftover build state at %s (version %s): %w", dst, v.VersionID, err) |
| 713 | } |
| 714 | log.Printf("moves: %s: the %s carries no build state; removed version %s of an earlier stay's at %s", ir.rec.Host, ir.src.Name, v.VersionID, dst) |
| 715 | } |
| 716 | return nil |
| 717 | } |
| 718 | o := seal.State[0] |
| 719 | if o.Key != ir.src.Lane+moves.BuildStateSuffix { |
| 720 | return fmt.Errorf("%w: the sealed build state %s is not the %s's lane key", moves.ErrBadSeal, o.Key, ir.src.Name) |
| 721 | } |
| 722 | if _, _, err := vs.CopyVersion(ctx, bucket, o.Key, o.VersionID, bucket, dst, false); err != nil { |
| 723 | return fmt.Errorf("place the build state at %s: %w", dst, err) |
| 724 | } |
| 725 | log.Printf("moves: %s: the %s placed the moved build state at %s", ir.rec.Host, ir.src.Name, dst) |
| 726 | return nil |
| 727 | } |
| 728 | |
| 729 | // published (§7.6): the build is complete and recorded; generation.json is |
| 730 | // not written here (R10-3: the probe's cold admission publishes it). The |
| 731 | // journal records the phase, the record advances to local and the location's |
| 732 | // lease is released: the build was the only writer that needed it. |
| 733 | func (ir *installRun) published(ctx context.Context) (bool, error) { |
| 734 | phase, next := ir.src.Phases[2], ir.src.Phases[3] |
| 735 | if _, err := ir.sync(ctx, phase); err != nil { |
| 736 | return false, err |
| 737 | } |
| 738 | j, err := ir.journal() |
| 739 | if err != nil { |
| 740 | return false, err |
| 741 | } |
| 742 | if !j.Done(landStepInstall) { |
| 743 | return false, fmt.Errorf("move %s: published without an install manifest in the journal", ir.g.ID) |
| 744 | } |
| 745 | j.Phase = phase |
| 746 | j.Mark(landStepPublished, ir.cfg().Now().UTC().Format(time.RFC3339Nano)) |
| 747 | if err := j.Commit(ir.cfg().DataDir); err != nil { |
| 748 | return false, err |
| 749 | } |
| 750 | ok, err := ir.advance(ctx, phase, next, installStamp{}) |
| 751 | if ok { |
| 752 | ir.releaseLease() |
| 753 | } |
| 754 | return ok, err |
| 755 | } |
| 756 | |
| 757 | // local (§7.6), each sub-step journalled and skipped on a replay once done, |
| 758 | // in this order (round-2 finding 8): (1) the installation record must be at |
| 759 | // the holder's stamp; (2) what is there before is parked: on the target an |
| 760 | // earlier stay's artifacts tagged S' through park.Local (which refuses a |
| 761 | // path tagged with another stamp), or a park-local left with nothing to |
| 762 | // retire is completed; on the source of a dirty return its own pre-move |
| 763 | // installation (retirePreMove); (3) the sealed archive, read by exact |
| 764 | // version and verified, is extracted into this install's own staging, |
| 765 | // <JournalDir>/extract/, and then, inside the host guard, moved to the files |
| 766 | // directory and <host>.env when nothing this install did not write is there |
| 767 | // (installFiles: the install never deletes at the live paths, final review |
| 768 | // finding 1; a retry clears only its staging); (4) the new paths are |
| 769 | // registered under the holder's stamp; (5) the timers table is the sealed |
| 770 | // snapshot with its registrations, the carried columns are set and the |
| 771 | // restore-verification entry removed; (6) the prepared files manifest (C8, |
| 772 | // Task 6): the placed tree, <host>.env and the wake state as create-only |
| 773 | // objects and one create-only sequence-0 manifest under files/g-<gen>/ in |
| 774 | // the location, each write re-checking the install's authority, the |
| 775 | // binding journalled with the manifest's digest and exact version, and a |
| 776 | // manifest already there that is not this install's refused |
| 777 | // (RollbackManifestOccupied); (7) the files directory is synced |
| 778 | // and .files-gen written last, so its presence vouches for every step |
| 779 | // before it. |
| 780 | // |
| 781 | // A replay finds local and finishes what the journal does not record; |
| 782 | // nothing here is read by the probe before .files-gen exists. |
| 783 | func (ir *installRun) local_(ctx context.Context) (bool, error) { |
| 784 | cfg := ir.cfg() |
| 785 | host, id, gen := ir.rec.Host, ir.g.ID, ir.src.GenerationOp |
| 786 | phase, next := ir.src.Phases[3], ir.src.Phases[4] |
| 787 | obs, err := ir.sync(ctx, phase) |
| 788 | if err != nil { |
| 789 | return false, err |
| 790 | } |
| 791 | holder := *obs.Claim.Holder |
| 792 | src := ir.src.Baseline |
| 793 | if src == nil || src.Files == nil { |
| 794 | return false, fmt.Errorf("move %s: the record has no source baselines", id) |
| 795 | } |
| 796 | seal, err := ir.seal(obs.Claim) |
| 797 | if err != nil { |
| 798 | return false, err |
| 799 | } |
| 800 | j, err := ir.journal() |
| 801 | if err != nil { |
| 802 | return false, err |
| 803 | } |
| 804 | j.Phase = phase |
| 805 | mark := func(step, value string) error { |
| 806 | j.Mark(step, value) |
| 807 | if err := j.Commit(cfg.DataDir); err != nil { |
| 808 | return err |
| 809 | } |
| 810 | return ir.fault("after-" + step) |
| 811 | } |
| 812 | captured, err := ir.capture(holder) |
| 813 | if err != nil { |
| 814 | return false, err |
| 815 | } |
| 816 | if !j.Done(landStepRetired) { |
| 817 | if ir.src.Role == roleSource { |
| 818 | outcome, err := ir.retirePreMove(ctx, captured, j, mark) |
| 819 | switch { |
| 820 | case err != nil: |
| 821 | return false, err |
| 822 | case outcome == restoreMarked: |
| 823 | return true, nil // the record changed: the refusal |
| 824 | case outcome == restoreWaiting: |
| 825 | return false, nil |
| 826 | } |
| 827 | } else if raw, ok := j.Steps[landStepPrevious]; ok { |
| 828 | earlier, err := strconv.ParseUint(raw, 10, 64) |
| 829 | if err != nil || earlier == 0 { |
| 830 | return false, fmt.Errorf("the journal's previous stamp %q", raw) |
| 831 | } |
| 832 | var chore *hostclaim.Obligation |
| 833 | for i := range obs.Claim.Obligations { |
| 834 | if o := &obs.Claim.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box { |
| 835 | chore = o |
| 836 | } |
| 837 | } |
| 838 | switch { |
| 839 | case chore != nil && hostclaim.InventoryHasStamp(obs.Claim, cfg.Box, earlier): |
| 840 | if err := ir.local.RetireArtifacts(ctx, host, captured, chore.ID, id, earlier); err != nil { |
| 841 | return false, fmt.Errorf("retire the earlier stay's artifacts: %w", err) |
| 842 | } |
| 843 | log.Printf("moves: %s: the earlier stay's artifacts (stamp %d) are parked", host, earlier) |
| 844 | case chore != nil: |
| 845 | if _, err := hostclaim.Update(ctx, ir.s, obs, hostclaim.CompleteLocalPark(id, cfg.Box, earlier)); err != nil { |
| 846 | return false, fmt.Errorf("complete the local park: %w", err) |
| 847 | } |
| 848 | } |
| 849 | } |
| 850 | if err := mark(landStepRetired, ""); err != nil { |
| 851 | return false, err |
| 852 | } |
| 853 | } |
| 854 | dir, envPath := cfg.FilesDir(host) |
| 855 | if !j.Done(landStepExtracted) { |
| 856 | outcome, err := ir.installFiles(ctx, j, mark, captured, seal, *src.Files, dir, envPath) |
| 857 | switch { |
| 858 | case err != nil: |
| 859 | return false, err |
| 860 | case outcome == restoreMarked: |
| 861 | return true, nil // the record changed: the refusal |
| 862 | case outcome == restoreWaiting: |
| 863 | return false, nil |
| 864 | } |
| 865 | } |
| 866 | if !j.Done(landStepInventory) { |
| 867 | // Only what this install has placed (carry (b)): the files directory |
| 868 | // and <host>.env, which installFiles moved into place when nothing |
| 869 | // it did not write was there. The database and its marker are |
| 870 | // registered by configured, once the install proves the database at |
| 871 | // DBPath is this build (registerDatabase): registered here, before |
| 872 | // the install, a rollback's retirement of F would park whatever |
| 873 | // database stands at DBPath, a foreign one included. |
| 874 | paths := []string{host, host + ".env"} |
| 875 | if err := ir.local.RecordInventory(ctx, host, captured, "flat", paths, ir.src.ClaimCheck(phase)); err != nil { |
| 876 | return false, fmt.Errorf("record the inventory: %w", err) |
| 877 | } |
| 878 | if err := mark(landStepInventory, ""); err != nil { |
| 879 | return false, err |
| 880 | } |
| 881 | } |
| 882 | if !j.Done(landStepTimers) { |
| 883 | rows, err := ir.sealedTimers(ctx, seal, src.TimersSHA256) |
| 884 | if err != nil { |
| 885 | return false, err |
| 886 | } |
| 887 | if err := cfg.ReplaceTimers(ctx, host, rows); err != nil { |
| 888 | return false, fmt.Errorf("replace the timers: %w", err) |
| 889 | } |
| 890 | if err := mark(landStepTimers, strconv.Itoa(len(rows))); err != nil { |
| 891 | return false, err |
| 892 | } |
| 893 | } |
| 894 | if !j.Done(landStepColumns) { |
| 895 | if err := cfg.SetReplication(host, src.Replication); err != nil { |
| 896 | return false, fmt.Errorf("set replication: %w", err) |
| 897 | } |
| 898 | if err := cfg.SetNextDue(host, src.NextDue); err != nil { |
| 899 | return false, fmt.Errorf("set next_due: %w", err) |
| 900 | } |
| 901 | if err := mark(landStepColumns, ""); err != nil { |
| 902 | return false, err |
| 903 | } |
| 904 | } |
| 905 | if !j.Done(landStepVerify) { |
| 906 | if err := cfg.RemoveVerifyEntry(ctx, ir.s, host); err != nil { |
| 907 | return false, fmt.Errorf("remove the restore-verification entry: %w", err) |
| 908 | } |
| 909 | if err := mark(landStepVerify, ""); err != nil { |
| 910 | return false, err |
| 911 | } |
| 912 | } |
| 913 | built, err := preparedManifestJournalled(j, hostfiles.ManifestKey(movesFilesPrefix(holder.Location, gen)), id, captured.Stamp) |
| 914 | if err != nil { |
| 915 | return false, err |
| 916 | } |
| 917 | if !built { |
| 918 | // Spec rev 7.22 §7.6 (L17, R19-1; C8, Task 6): the prepared files |
| 919 | // manifest under files/g-<generation>/ in the location, bound to |
| 920 | // {location, g-<generation>, op, stamp}, where the generation is the |
| 921 | // installation's prepared destination generation (the move id, or |
| 922 | // the rollback id of a dirty return), never holder.generation. It |
| 923 | // carries the placed files, <host>.env and the wake state the |
| 924 | // install put in place above, as create-only objects and one |
| 925 | // create-only sequence-0 manifest through the box's own versioned |
| 926 | // store, after the claim sync and stamp capture above (op.exec's |
| 927 | // installation authority). The binding is journalled with the |
| 928 | // manifest's digest; a replay reuses the objects by digest and reads |
| 929 | // the manifest back by key. A journal that records the step as |
| 930 | // skipped (written before C8 existed) counts as not done (ruling Q3 |
| 931 | // item 2). Another manifest at the key is not this install's to |
| 932 | // replace: the refusal (RollbackManifestOccupied). |
| 933 | // The image is the install's own placement, by the install receipt |
| 934 | // (files_placed: the tree's identity, the env file's identity or |
| 935 | // none): a vanished tree or env file refuses, and the manifest |
| 936 | // carries an env item exactly when the archive carried one (fix |
| 937 | // round 1, finding 4). |
| 938 | placed, err := journalledStaged(j, landStepFilesPlaced) |
| 939 | if err != nil { |
| 940 | return false, err |
| 941 | } |
| 942 | if ok, err := sameIdentity(dir, placed.Files); err != nil || !ok { |
| 943 | return false, fmt.Errorf("%w: %s is not the files directory this %s placed (%v)", errLandingChanged, dir, ir.src.Name, err) |
| 944 | } |
| 945 | wantEnv := placed.Env != "" |
| 946 | if wantEnv { |
| 947 | if ok, err := sameIdentity(envPath, placed.Env); err != nil || !ok { |
| 948 | return false, fmt.Errorf("%w: %s is not the env file this %s placed (%v)", errLandingChanged, envPath, ir.src.Name, err) |
| 949 | } |
| 950 | } |
| 951 | rows, err := ir.sealedTimers(ctx, seal, src.TimersSHA256) |
| 952 | if err != nil { |
| 953 | return false, err |
| 954 | } |
| 955 | vs, err := ir.versioned(ctx) |
| 956 | if err != nil { |
| 957 | return false, err |
| 958 | } |
| 959 | // Every write re-checks the authority first (installAuthority): a |
| 960 | // stale build never wins the create-only manifest. |
| 961 | authority := installAuthorityStore{VersionedStore: vs, check: ir.installAuthority(ctx, captured, phase)} |
| 962 | pm, err := buildPreparedManifest(ctx, authority, cfg.OwnBucket, holder.Location, gen, id, captured.Stamp, dir, envPath, wantEnv, hostfiles.WakeStateOf(host, rows, src.NextDue, src.Replication)) |
| 963 | if errors.Is(err, hostfiles.ErrManifestExists) { |
| 964 | log.Printf("moves: %s: the prepared files manifest of %s is not this install's: %v", host, id, err) |
| 965 | return ir.refuse(ctx, refuseManifestOccupied) |
| 966 | } |
| 967 | if err != nil { |
| 968 | return false, fmt.Errorf("build the prepared files manifest: %w", err) |
| 969 | } |
| 970 | raw, err := json.Marshal(pm) |
| 971 | if err != nil { |
| 972 | return false, err |
| 973 | } |
| 974 | if err := mark(landStepFilesManifest, string(raw)); err != nil { |
| 975 | return false, err |
| 976 | } |
| 977 | log.Printf("moves: %s: the prepared files manifest %s is written (sha256 %s)", host, pm.Key, pm.SHA256) |
| 978 | } |
| 979 | if err := ir.fault("before-files-gen"); err != nil { |
| 980 | return false, err |
| 981 | } |
| 982 | marker, err := json.Marshal(filesGenMarker{Op: gen, Incarnation: holder.Incarnation}) |
| 983 | if err != nil { |
| 984 | return false, err |
| 985 | } |
| 986 | placed, err := journalledStaged(j, landStepFilesPlaced) |
| 987 | if err != nil { |
| 988 | return false, err |
| 989 | } |
| 990 | // The marker goes into the files directory this install placed, and |
| 991 | // only while the installation is still this install's (final review, |
| 992 | // finding 2): inside the guard, the directory is the receipt's. |
| 993 | if err := ir.guarded(ctx, captured, phase, func() error { |
| 994 | if ok, err := sameIdentity(dir, placed.Files); err != nil || !ok { |
| 995 | return fmt.Errorf("%w: %s is not the files directory this %s placed (%v)", errLandingChanged, dir, ir.src.Name, err) |
| 996 | } |
| 997 | if err := syncPath(dir); err != nil { |
| 998 | return err |
| 999 | } |
| 1000 | return writeFileAtomic(filepath.Join(dir, archive.MarkerName), marker, 0o644) |
| 1001 | }); err != nil { |
| 1002 | return false, fmt.Errorf("write %s: %w", archive.MarkerName, err) |
| 1003 | } |
| 1004 | if err := mark(landStepFilesGen, ""); err != nil { |
| 1005 | return false, err |
| 1006 | } |
| 1007 | log.Printf("moves: %s: files, environment, timers and columns installed; %s written", host, archive.MarkerName) |
| 1008 | return ir.advance(ctx, phase, next, installStamp{}) |
| 1009 | } |
| 1010 | |
| 1011 | // installArtifactPaths are an installation's artifacts at the flat paths |
| 1012 | // (G11), relative to the data directory: the database with its marker, |
| 1013 | // journal, quarantine and litestream shadow, the files directory and |
| 1014 | // <host>.env. A dirty return parks them before it installs; the source's |
| 1015 | // finalization parks them and the rendered config beside them. |
| 1016 | func installArtifactPaths(host string) []string { |
| 1017 | db := host + ".db" |
| 1018 | return []string{db, db + ".gen", db + "-wal", db + "-shm", db + ".corrupt", ".activator-" + db + "-litestream", host, host + ".env"} |
| 1019 | } |
| 1020 | |
| 1021 | // retirePreMove is a dirty return's local sub-step (2) (ruling J13; fix |
| 1022 | // round 1): the source's own pre-move installation, and anything an earlier |
| 1023 | // installation of this holder left on this box, leaves the flat paths before |
| 1024 | // anything of the return's is placed there, so placeStaged and restore find |
| 1025 | // them empty and never install over them, and the return's registration |
| 1026 | // never meets a path tagged with another stamp. |
| 1027 | // |
| 1028 | // First the claim's inventory for this box, read fresh each round, is |
| 1029 | // accounted for stamp by stamp, before anything moves: |
| 1030 | // - S, the source's installation stamp (the holder's after |
| 1031 | // ReturnToSource): retired through park.Local under the host guard |
| 1032 | // against the park-local the source's claimed queued |
| 1033 | // (RetireArtifacts completes it); paths of S listed with no park-local |
| 1034 | // owed is an error, since only the claimed CAS queues it; |
| 1035 | // - a stamp the claim shows superseded on this box (supersededOnBox: a |
| 1036 | // cold admission here issued it, and it is neither S nor the holder's |
| 1037 | // live session; spec §7.8, rev 7.11): retired the same way, against a |
| 1038 | // park-local queued here for it, so its inventory entries clear; |
| 1039 | // - any other stamp (live, unknown, another holder's): nothing is moved, |
| 1040 | // and the return stops for an operator (StuckReturnUnknownStamp); a |
| 1041 | // retry after the operator has accounted for it runs this again. |
| 1042 | // |
| 1043 | // A park-local owed with nothing listed (a replay after a retirement) is |
| 1044 | // completed. Then every pre-move artifact still at the flat paths that no |
| 1045 | // inventory entry claims is parked by the moves-owned journalled rename |
| 1046 | // (H5(b), parkLocalArtifacts), its plan and its renames inside the host |
| 1047 | // guard with the claim and the record read fresh there (fix round 1, ruling |
| 1048 | // 1), so a call that resumes after authority moved on renames nothing. |
| 1049 | func (ir *installRun) retirePreMove(ctx context.Context, captured hostguard.Captured, j *moves.Journal, mark func(step, value string) error) (restoreOutcome, error) { |
| 1050 | cfg := ir.cfg() |
| 1051 | host, id, phase := ir.rec.Host, ir.g.ID, ir.src.Phases[3] |
| 1052 | stamp := captured.Stamp |
| 1053 | for round := 0; ; round++ { |
| 1054 | if round > 64 { |
| 1055 | return restoreWaiting, fmt.Errorf("move %s: the source's inventory did not clear", id) |
| 1056 | } |
| 1057 | obs, err := ir.sync(ctx, phase) |
| 1058 | if err != nil { |
| 1059 | return restoreWaiting, err |
| 1060 | } |
| 1061 | c := obs.Claim |
| 1062 | var retire []uint64 |
| 1063 | for _, x := range inventoryStamps(c, cfg.Box) { |
| 1064 | switch { |
| 1065 | case x == stamp: |
| 1066 | retire = append([]uint64{x}, retire...) |
| 1067 | case supersededOnBox(c, cfg.Box, x): |
| 1068 | retire = append(retire, x) |
| 1069 | default: |
| 1070 | log.Printf("moves: %s: this box's inventory lists paths under stamp %d, which the return of %s cannot account for; nothing is moved", host, x, id) |
| 1071 | ok, err := ir.refuse(ctx, refuseUnknownStamp) |
| 1072 | if err != nil || !ok { |
| 1073 | return restoreWaiting, err |
| 1074 | } |
| 1075 | return restoreMarked, nil |
| 1076 | } |
| 1077 | } |
| 1078 | var chore *hostclaim.Obligation |
| 1079 | for i := range c.Obligations { |
| 1080 | if o := &c.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box { |
| 1081 | chore = o |
| 1082 | } |
| 1083 | } |
| 1084 | if len(retire) == 0 { |
| 1085 | if chore != nil { |
| 1086 | if _, err := hostclaim.Update(ctx, ir.s, obs, hostclaim.CompleteLocalPark(id, cfg.Box, stamp)); err != nil { |
| 1087 | return restoreWaiting, fmt.Errorf("complete the source's local park: %w", err) |
| 1088 | } |
| 1089 | } |
| 1090 | break |
| 1091 | } |
| 1092 | x := retire[0] |
| 1093 | if chore == nil { |
| 1094 | if x == stamp { |
| 1095 | return restoreWaiting, fmt.Errorf("move %s: this box's inventory lists the source's pre-move paths at stamp %d, and the claim owes no park-local for them", id, stamp) |
| 1096 | } |
| 1097 | if obs, err = hostclaim.Update(ctx, ir.s, obs, hostclaim.QueueLocalPark(id, cfg.Box, obligationID)); err != nil { |
| 1098 | return restoreWaiting, fmt.Errorf("queue the local park of stamp %d: %w", x, err) |
| 1099 | } |
| 1100 | for i := range obs.Claim.Obligations { |
| 1101 | if o := &obs.Claim.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box { |
| 1102 | chore = o |
| 1103 | } |
| 1104 | } |
| 1105 | if chore == nil { |
| 1106 | return restoreWaiting, fmt.Errorf("move %s: the local park of stamp %d was not queued", id, x) |
| 1107 | } |
| 1108 | } |
| 1109 | if err := ir.local.RetireArtifacts(ctx, host, captured, chore.ID, id, x); err != nil { |
| 1110 | return restoreWaiting, fmt.Errorf("retire the source's artifacts of stamp %d: %w", x, err) |
| 1111 | } |
| 1112 | log.Printf("moves: %s: the source's artifacts of stamp %d are retired", host, x) |
| 1113 | } |
| 1114 | if err := ir.fault("before-" + installStepPreMove); err != nil { |
| 1115 | return restoreWaiting, err |
| 1116 | } |
| 1117 | paths := installArtifactPaths(host) |
| 1118 | unclaimed := func(c *hostclaim.Claim) error { |
| 1119 | for _, p := range paths { |
| 1120 | if _, ok := c.Inventory[cfg.Box].Paths[p]; ok { |
| 1121 | return fmt.Errorf("%w: the inventory claims %s", errLandingChanged, p) |
| 1122 | } |
| 1123 | } |
| 1124 | return nil |
| 1125 | } |
| 1126 | err := ir.guardedWith(ctx, captured, phase, unclaimed, func() error { |
| 1127 | _, err := parkLocalArtifacts(ctx, j, mark, localArtifactPark{DataDir: cfg.DataDir, Incarnation: captured.Incarnation, Host: host, |
| 1128 | Op: ir.src.GenerationOp, Step: installStepPreMove, Paths: paths, NewID: obligationID, Fault: ir.fault}) |
| 1129 | return err |
| 1130 | }) |
| 1131 | if err != nil { |
| 1132 | return restoreWaiting, fmt.Errorf("park the source's pre-move artifacts: %w", err) |
| 1133 | } |
| 1134 | return restoreInstalled, nil |
| 1135 | } |
| 1136 | |
| 1137 | // inventoryStamps are the distinct stamps box's inventory tags paths with, |
| 1138 | // in order. |
| 1139 | func inventoryStamps(c *hostclaim.Claim, box string) []uint64 { |
| 1140 | seen := map[uint64]bool{} |
| 1141 | var out []uint64 |
| 1142 | for _, s := range c.Inventory[box].Paths { |
| 1143 | if !seen[s] { |
| 1144 | seen[s] = true |
| 1145 | out = append(out, s) |
| 1146 | } |
| 1147 | } |
| 1148 | sort.Slice(out, func(i, k int) bool { return out[i] < out[k] }) |
| 1149 | return out |
| 1150 | } |
| 1151 | |
| 1152 | // supersededOnBox reports whether the claim shows stamp as an installation |
| 1153 | // stamp this holder issued on box and has since moved past: a cold |
| 1154 | // admission's stamp is its record's session (hostclaim.AdmitWriter), so a |
| 1155 | // cold admission record on box with that session names it, and it is |
| 1156 | // neither the holder's stamp nor its live writer session. A stamp the claim |
| 1157 | // has no such record of is not this box's to retire. |
| 1158 | func supersededOnBox(c *hostclaim.Claim, box string, stamp uint64) bool { |
| 1159 | h := c.Holder |
| 1160 | if h == nil || stamp == 0 || stamp == h.Stamp || stamp == h.Session { |
| 1161 | return false |
| 1162 | } |
| 1163 | for _, a := range h.Admissions { |
| 1164 | if a.Mode == "cold" && a.Box == box && a.Session == stamp { |
| 1165 | return true |
| 1166 | } |
| 1167 | } |
| 1168 | return false |
| 1169 | } |
| 1170 | |
| 1171 | // releaseLive is the configured step's release (§9.2 as amended |
| 1172 | // 2026-10-12): ReleaseLive, else, with CARLOS_HOST_CLAIM off, ReleaseReady, |
| 1173 | // which publishes then. Under the claim T7's hook only stages, so it is |
| 1174 | // never the fallback: nil, and configured waits. |
| 1175 | func (c moveCfg) releaseLive() func(context.Context, release.Store, provision.Record, string) moves.Check { |
| 1176 | switch { |
| 1177 | case c.ReleaseLive != nil: |
| 1178 | return c.ReleaseLive |
| 1179 | case hostwrite.Enabled(os.Getenv): |
| 1180 | return nil |
| 1181 | } |
| 1182 | return c.ReleaseReady |
| 1183 | } |
| 1184 | |
| 1185 | // configured (§7.6, §9.1, §9.2): the held row rendered (RenderHeld, the |
| 1186 | // move-bound ack). A landing requires the render's bundle and overlay etags |
| 1187 | // to be the ones pinned at readiness (else the refusal, "configuration |
| 1188 | // changed during the move") and the pinned release live and proven for the |
| 1189 | // held row through releaseLive (ruling G4: "waiting for the root stage |
| 1190 | // pass" waits, anything else waits until the landing's deadline refuses). |
| 1191 | // Under the claim (Paul's T7 ruling, 2026-10-12) the render is selected as |
| 1192 | // held-config and the release is published here, under held-release: T7 |
| 1193 | // only staged it. A |
| 1194 | // return renders the current inputs instead, never compared with |
| 1195 | // move.inputs, gates on the release live on this box, and records what it |
| 1196 | // rendered as Rollback.RecoveryInputs in the CAS that ends the phase |
| 1197 | // (J2-R2-5). Then the install the brief lists under `restored`: the built |
| 1198 | // database's sha256 against the manifest, lease.CheckDB, the rename into |
| 1199 | // DBPath (G34), <dbPath>.gen through hostbox's marker writer naming the |
| 1200 | // generation (ruling G2), and the CAS to restored. |
| 1201 | // |
| 1202 | // A replay finds configured and does it all again: the render and the |
| 1203 | // release check are idempotent reads and writes of the same bytes; a |
| 1204 | // database already at DBPath with the manifest's sha256 is this install's |
| 1205 | // own, so only the marker is written again. |
| 1206 | func (ir *installRun) configured(ctx context.Context) (bool, error) { |
| 1207 | cfg := ir.cfg() |
| 1208 | host, id := ir.rec.Host, ir.g.ID |
| 1209 | phase, next := ir.src.Phases[4], ir.src.Phases[5] |
| 1210 | obs, err := ir.sync(ctx, phase) |
| 1211 | if err != nil { |
| 1212 | return false, err |
| 1213 | } |
| 1214 | if obs.Claim.Op.Kind == hostclaim.KindLanding && cfg.registerLanding != nil { |
| 1215 | if err := cfg.registerLanding(ctx, obs); err != nil { |
| 1216 | return false, err |
| 1217 | } |
| 1218 | } |
| 1219 | var pin, recovery *provision.MoveInputs |
| 1220 | if ir.src.Role == roleTarget { |
| 1221 | if pin = ir.src.PinInputs; pin == nil { |
| 1222 | return false, fmt.Errorf("move %s: the record has no pinned inputs", id) |
| 1223 | } |
| 1224 | } |
| 1225 | rt, err := ir.route() |
| 1226 | if err != nil { |
| 1227 | return false, err |
| 1228 | } |
| 1229 | // The render's ownership is journalled before its write (carry (c), |
| 1230 | // fix round 1, I1): the hook runs once the rendered bytes are known and |
| 1231 | // before they are placed, and journals their sha256 with the file's |
| 1232 | // state before the render (pre_render). The rendered sha256 is journalled |
| 1233 | // again as soon as the render answers, even when its config ack failed. |
| 1234 | // A rollback's cleaning parks <host>.config.env only when its content is |
| 1235 | // one this op rendered (rollbackOwnership). |
| 1236 | // |
| 1237 | // Under the claim the render is selected as held-config, and the pins |
| 1238 | // decide first (Task 7, Q2.2): a merge that is not them is the |
| 1239 | // configuration-changed refusal, with nothing selected or written. |
| 1240 | hr, err := cfg.RenderHeld(ctx, ir.s, rt, ir.src.HoldOp, pin, ir.notePreRender) |
| 1241 | if hr.SHA256 != "" { |
| 1242 | if jerr := ir.noteRendered(hr.SHA256); jerr != nil { |
| 1243 | return false, errors.Join(err, jerr) |
| 1244 | } |
| 1245 | } |
| 1246 | if pin != nil && errors.Is(err, errHeldRenderUnpinned) { |
| 1247 | log.Printf("moves: %s: the held render's inputs are not the pinned ones (%v, overlay %q): %v", host, pin.BundleETags, pin.OverlayETag, err) |
| 1248 | return ir.refuse(ctx, refuseConfigChanged) |
| 1249 | } |
| 1250 | if err != nil { |
| 1251 | return false, fmt.Errorf("render the held row: %w", err) |
| 1252 | } |
| 1253 | want := "" |
| 1254 | if pin != nil { |
| 1255 | if !sameBundle(hr.BundleETags, pin.BundleETags) || hr.OverlayETag != pin.OverlayETag { |
| 1256 | log.Printf("moves: %s: the render's etags (%v, overlay %q) are not the pinned ones (%v, overlay %q)", host, hr.BundleETags, hr.OverlayETag, pin.BundleETags, pin.OverlayETag) |
| 1257 | return ir.refuse(ctx, refuseConfigChanged) |
| 1258 | } |
| 1259 | want = pin.Release |
| 1260 | } else { |
| 1261 | live, err := release.LiveVersion(cfg.OptDir, host) |
| 1262 | if err != nil { |
| 1263 | return false, fmt.Errorf("read the live release: %w", err) |
| 1264 | } |
| 1265 | bundle := make(map[string]string, len(hr.BundleETags)) |
| 1266 | for k, v := range hr.BundleETags { |
| 1267 | bundle[k] = v |
| 1268 | } |
| 1269 | recovery = &provision.MoveInputs{BundleETags: bundle, OverlayETag: hr.OverlayETag, Release: live} |
| 1270 | want = live |
| 1271 | } |
| 1272 | live := cfg.releaseLive() |
| 1273 | if live == nil { |
| 1274 | ir.w.notice(host+"#release", "moves: %s: release %s cannot be published for the held row: no held-release publication is wired under CARLOS_HOST_CLAIM; retrying", host, want) |
| 1275 | return false, nil |
| 1276 | } |
| 1277 | if c := live(ctx, ir.s, ir.rec, want); !c.OK { |
| 1278 | ir.w.notice(host+"#release", "moves: %s: release %s is not live and proven for the held row yet (%s); retrying", host, want, c.Reason) |
| 1279 | return false, nil |
| 1280 | } |
| 1281 | if err := ir.fault("after-release-ready"); err != nil { |
| 1282 | return false, err |
| 1283 | } |
| 1284 | // The render, the release gate and the install can outlast a landing's |
| 1285 | // deadline (fix round 1, item 4): it is checked again before the install |
| 1286 | // and right before the transition that names it complete. |
| 1287 | if ir.pastDeadline() { |
| 1288 | return ir.refuse(ctx, refuseDeadline) |
| 1289 | } |
| 1290 | sha, outcome, err := ir.restore(ctx, obs) |
| 1291 | if err != nil { |
| 1292 | return false, err |
| 1293 | } |
| 1294 | switch outcome { |
| 1295 | case restoreWaiting: |
| 1296 | return false, nil |
| 1297 | case restoreMarked: |
| 1298 | return true, nil // the record changed: the refusal, not an install |
| 1299 | } |
| 1300 | if err := ir.registerDatabase(ctx, obs, phase); err != nil { |
| 1301 | return false, err |
| 1302 | } |
| 1303 | if ir.pastDeadline() { |
| 1304 | return ir.refuse(ctx, refuseDeadline) |
| 1305 | } |
| 1306 | return ir.advance(ctx, phase, next, installStamp{InstalledDBSHA256: sha, RecoveryInputs: recovery}) |
| 1307 | } |
| 1308 | |
| 1309 | // registerDatabase registers the installed database and its marker under F |
| 1310 | // (carry (b)), once restore has proved the database at DBPath is this |
| 1311 | // install's build: renamed into place inside the guard, or found there |
| 1312 | // with the build's sha256. Journalled, so a replay registers once. |
| 1313 | func (ir *installRun) registerDatabase(ctx context.Context, obs hostclaim.Observed, phase string) error { |
| 1314 | j, err := ir.journal() |
| 1315 | if err != nil { |
| 1316 | return err |
| 1317 | } |
| 1318 | if j.Done(landStepInventoryDB) { |
| 1319 | return nil |
| 1320 | } |
| 1321 | captured, err := ir.capture(*obs.Claim.Holder) |
| 1322 | if err != nil { |
| 1323 | return err |
| 1324 | } |
| 1325 | host := ir.rec.Host |
| 1326 | if err := ir.local.RecordInventory(ctx, host, captured, "flat", []string{host + ".db", host + ".db.gen"}, ir.src.ClaimCheck(phase)); err != nil { |
| 1327 | return fmt.Errorf("register the installed database: %w", err) |
| 1328 | } |
| 1329 | j.Phase = phase |
| 1330 | j.Mark(landStepInventoryDB, "") |
| 1331 | if err := j.Commit(ir.cfg().DataDir); err != nil { |
| 1332 | return err |
| 1333 | } |
| 1334 | return ir.fault("after-" + landStepInventoryDB) |
| 1335 | } |
| 1336 | |
| 1337 | // configEnvPath is the held render's file, <DataDir>/<host>.config.env. |
| 1338 | func (ir *installRun) configEnvPath() string { |
| 1339 | return filepath.Join(ir.cfg().DataDir, ir.rec.Host+".config.env") |
| 1340 | } |
| 1341 | |
| 1342 | // preRender is the landStepPreRender record: <host>.config.env's state |
| 1343 | // before the first held render ("absent", "sha256:<hex>" of a regular file, |
| 1344 | // or "other"), and the intended set: the sha256 of each content a render of |
| 1345 | // this op may have placed, journalled before its write. The set is pruned |
| 1346 | // to the file on disk (fix round 4): before each attempt it keeps the hash |
| 1347 | // of the content now on disk, only when it is already one of the set's, and |
| 1348 | // adds the new content's. So it never holds more than two hashes, and every |
| 1349 | // content this op placed that can still be on disk stays named. |
| 1350 | type preRender struct { |
| 1351 | Before string `json:"before"` |
| 1352 | Intended []string `json:"intended"` |
| 1353 | } |
| 1354 | |
| 1355 | // notePreRender is the held render's before-write hook. It hashes the file |
| 1356 | // now on disk (Lstat'ed, never followed) and journals the intended set as |
| 1357 | // {that hash, if the set already names it} plus {the new content's hash}, |
| 1358 | // before the write. A file whose content the set does not name (the earlier |
| 1359 | // state, or foreign content) contributes nothing, and the earlier-state |
| 1360 | // record is kept as first seen. The renderer replaces the file by an atomic |
| 1361 | // rename (appconfig's renderExecConfigBefore), so the file on disk is always |
| 1362 | // one whole content: a hash pruned here is never on disk again unless this |
| 1363 | // op renders it again, when it is journalled again first. |
| 1364 | func (ir *installRun) notePreRender(sha string) error { |
| 1365 | if sha == "" { |
| 1366 | return errors.New("moves: the held render names no content") |
| 1367 | } |
| 1368 | j, err := ir.journal() |
| 1369 | if err != nil { |
| 1370 | return err |
| 1371 | } |
| 1372 | path := ir.configEnvPath() |
| 1373 | disk := "" |
| 1374 | fi, err := os.Lstat(path) |
| 1375 | switch { |
| 1376 | case errors.Is(err, os.ErrNotExist): |
| 1377 | case err != nil: |
| 1378 | return err |
| 1379 | case fi.Mode().IsRegular(): |
| 1380 | if disk, err = fileSHA256(path); err != nil { |
| 1381 | return err |
| 1382 | } |
| 1383 | } |
| 1384 | var pre preRender |
| 1385 | if raw, ok := j.Steps[landStepPreRender]; ok { |
| 1386 | if err := json.Unmarshal([]byte(raw), &pre); err != nil { |
| 1387 | return fmt.Errorf("the journal's %s step: %w", landStepPreRender, err) |
| 1388 | } |
| 1389 | } else { |
| 1390 | pre.Before = "other" |
| 1391 | switch { |
| 1392 | case errors.Is(err, os.ErrNotExist) || fi == nil: |
| 1393 | pre.Before = "absent" |
| 1394 | case disk != "": |
| 1395 | pre.Before = disk |
| 1396 | } |
| 1397 | } |
| 1398 | want := "sha256:" + sha |
| 1399 | var next []string |
| 1400 | if disk != "" && containsString(pre.Intended, disk) { |
| 1401 | next = append(next, disk) |
| 1402 | } |
| 1403 | if want != disk || len(next) == 0 { |
| 1404 | next = append(next, want) |
| 1405 | } |
| 1406 | if strings.Join(next, " ") == strings.Join(pre.Intended, " ") { |
| 1407 | return nil |
| 1408 | } |
| 1409 | pre.Intended = next |
| 1410 | body, err := json.Marshal(pre) |
| 1411 | if err != nil { |
| 1412 | return err |
| 1413 | } |
| 1414 | j.Mark(landStepPreRender, string(body)) |
| 1415 | if err := j.Commit(ir.cfg().DataDir); err != nil { |
| 1416 | return err |
| 1417 | } |
| 1418 | return ir.fault("after-" + landStepPreRender) |
| 1419 | } |
| 1420 | |
| 1421 | // noteRendered journals the sha256 (hex) of the file the held render wrote, |
| 1422 | // as "sha256:<hex>": the latest render's, which is what the file holds. |
| 1423 | func (ir *installRun) noteRendered(sha string) error { |
| 1424 | j, err := ir.journal() |
| 1425 | if err != nil { |
| 1426 | return err |
| 1427 | } |
| 1428 | want := "sha256:" + sha |
| 1429 | if j.Steps[landStepRendered] == want { |
| 1430 | return nil |
| 1431 | } |
| 1432 | j.Mark(landStepRendered, want) |
| 1433 | if err := j.Commit(ir.cfg().DataDir); err != nil { |
| 1434 | return err |
| 1435 | } |
| 1436 | return ir.fault("after-" + landStepRendered) |
| 1437 | } |
| 1438 | |
| 1439 | // recoveryInstallSource binds the installer to a dirty return on the source |
| 1440 | // (§7.8 source table, the dirty branch; rulings J13, J14, J20): the record |
| 1441 | // rolling_back and dirty with the source at an install phase; the claim |
| 1442 | // restoring/claimed for this rollback with this box, the move's source, |
| 1443 | // executing at the flip's journalled source stamp S; the generation and the |
| 1444 | // hold named by the rollback id; the return lane moves/<host>/<rb id>/ in |
| 1445 | // this box's own bucket; the refusal MoveMarkStuck; the phase advance |
| 1446 | // MoveRollbackSourcePhase, stamping RecoveryInputs at the end of configured. |
| 1447 | // seal reads the return seal from the claim the phase works from, and |
| 1448 | // baseline is what it restores to (the return's evidence): both are the |
| 1449 | // caller's (Plan 4 Tasks 4 and 5). Production does not call it yet. |
| 1450 | func recoveryInstallSource(w *movesWorker, s release.Store, rec provision.Record, seal func(*hostclaim.Claim) (moves.Seal, error), baseline *provision.MoveSource) (installSource, error) { |
| 1451 | cfg := w.cfg |
| 1452 | m := rec.Move |
| 1453 | if m == nil || m.Legacy() || m.State != provision.MoveRollingBack || m.Rollback == nil || !m.Rollback.Dirty || !moves.ValidOpID(m.Rollback.ID) || |
| 1454 | !strings.HasPrefix(m.Rollback.ID, "rb_") { |
| 1455 | return installSource{}, errors.New("moves: only a dirty rollback's source installs a return") |
| 1456 | } |
| 1457 | if cfg.Box == "" || seal == nil || baseline == nil { |
| 1458 | return installSource{}, errors.New("moves: a return's install names this box, its seal and its baseline") |
| 1459 | } |
| 1460 | g := provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation} |
| 1461 | host, rb, box := rec.Host, m.Rollback.ID, cfg.Box |
| 1462 | claimCheck := func(string) func(*hostclaim.Claim) error { |
| 1463 | return func(c *hostclaim.Claim) error { |
| 1464 | if err := hostclaim.RequireMoveOp(g.ID, hostclaim.KindRestoring, hostclaim.PhaseClaimed)(c); err != nil { |
| 1465 | return err |
| 1466 | } |
| 1467 | if c.Op.Exec != box || len(c.Op.Participants) == 0 || c.Op.Participants[0] != box || c.Op.Intent["source"] != box { |
| 1468 | return fmt.Errorf("%w: the executor is %q, not this box as the move's source", hostclaim.ErrWrongState, c.Op.Exec) |
| 1469 | } |
| 1470 | if c.Op.Intent["rollback"] != rb { |
| 1471 | return fmt.Errorf("%w: the op is rollback %q, not %s", hostclaim.ErrWrongState, c.Op.Intent["rollback"], rb) |
| 1472 | } |
| 1473 | stamp, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64) |
| 1474 | if err != nil || stamp == 0 || c.Holder == nil || c.Holder.Stamp != stamp { |
| 1475 | return fmt.Errorf("%w: the holder is not at the source's pre-move stamp", hostclaim.ErrWrongState) |
| 1476 | } |
| 1477 | return nil |
| 1478 | } |
| 1479 | } |
| 1480 | return installSource{ |
| 1481 | OpID: g.ID, Host: host, Role: roleSource, Name: "return", |
| 1482 | Lane: moves.StagingPrefix(host, rb), Seal: seal, Baseline: baseline, |
| 1483 | GenerationOp: rb, HoldOp: rb, |
| 1484 | Phases: [6]string{provision.RbSourceParking, provision.RbSourceBuilding, provision.RbSourcePublished, |
| 1485 | provision.RbSourceLocal, provision.RbSourceConfigured, provision.RbSourceRestored}, |
| 1486 | Phase: func(from, to string, st installStamp) func(*provision.Record) error { |
| 1487 | var stamp func(*provision.MoveRollback) |
| 1488 | if st.RecoveryInputs != nil { |
| 1489 | in := st.RecoveryInputs |
| 1490 | stamp = func(r *provision.MoveRollback) { r.RecoveryInputs = in } |
| 1491 | } |
| 1492 | return provision.MoveRollbackSourcePhase(g, from, to, stamp) |
| 1493 | }, |
| 1494 | Sync: func(ctx context.Context) (hostclaim.Observed, error) { |
| 1495 | obs, err := hostclaim.Read(ctx, s, host) |
| 1496 | if err != nil { |
| 1497 | return obs, fmt.Errorf("read the host claim: %w", err) |
| 1498 | } |
| 1499 | if obs.Absent { |
| 1500 | return obs, fmt.Errorf("move %s has no host claim", g.ID) |
| 1501 | } |
| 1502 | if err := hostclaim.RequireOp(g.ID)(obs.Claim); err != nil { |
| 1503 | return obs, fmt.Errorf("move %s: %w", g.ID, err) |
| 1504 | } |
| 1505 | if h := obs.Claim.Holder; h == nil || h.Incarnation != g.Incarnation { |
| 1506 | return obs, fmt.Errorf("move %s: the host claim names another incarnation", g.ID) |
| 1507 | } |
| 1508 | return obs, nil |
| 1509 | }, |
| 1510 | ClaimCheck: claimCheck, |
| 1511 | RecordPhase: func(r *provision.Record) (string, error) { |
| 1512 | m := r.Move |
| 1513 | if m == nil || m.ID != g.ID || m.Incarnation != g.Incarnation || m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.ID != rb || !m.Rollback.Dirty { |
| 1514 | return "", fmt.Errorf("%w: the record is no longer the dirty return of %s", errLandingChanged, rb) |
| 1515 | } |
| 1516 | return m.Rollback.SourcePhase, nil |
| 1517 | }, |
| 1518 | Refuse: func(ctx context.Context, why installRefusal) (bool, string, error) { |
| 1519 | var reason string |
| 1520 | switch why { |
| 1521 | case refuseFilesOccupied: |
| 1522 | reason = provision.StuckReturnFilesOccupied |
| 1523 | case refuseManifestOccupied: |
| 1524 | reason = provision.StuckReturnManifestOccupied |
| 1525 | case refuseInstallMismatch: |
| 1526 | reason = provision.StuckReturnInstallMismatch |
| 1527 | case refuseUnknownStamp: |
| 1528 | reason = provision.StuckReturnUnknownStamp |
| 1529 | default: |
| 1530 | return false, "", fmt.Errorf("move %s: a return's install has no refusal %d", g.ID, why) |
| 1531 | } |
| 1532 | var pre provision.Record |
| 1533 | after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, host, observedMove(&pre, provision.MoveMarkStuck(g, reason))) |
| 1534 | if err != nil { |
| 1535 | if errors.Is(err, provision.ErrMoveConflict) { |
| 1536 | log.Printf("moves: %s: the stuck mark of %s refused (%v); the record moved on", host, rb, err) |
| 1537 | return false, "", nil |
| 1538 | } |
| 1539 | return false, "", fmt.Errorf("mark the return stuck: %w", err) |
| 1540 | } |
| 1541 | w.logTransition(ctx, s, pre, *after) |
| 1542 | log.Printf("moves: %s: the return %s stops for an operator: %s", host, rb, reason) |
| 1543 | nudgeSerial(ctx, s) |
| 1544 | return true, reason, nil |
| 1545 | }, |
| 1546 | }, nil |
| 1547 | } |
| 1548 | |
amadan