| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/sha256" |
| 7 | "encoding/hex" |
| 8 | "encoding/json" |
| 9 | "errors" |
| 10 | "fmt" |
| 11 | "io" |
| 12 | "log" |
| 13 | "os" |
| 14 | "path/filepath" |
| 15 | "reflect" |
| 16 | "sort" |
| 17 | "strings" |
| 18 | "time" |
| 19 | |
| 20 | "github.com/carlosframework/platform/internal/activator" |
| 21 | "github.com/carlosframework/platform/internal/appconfig" |
| 22 | "github.com/carlosframework/platform/internal/instancekey" |
| 23 | "github.com/carlosframework/platform/internal/lease" |
| 24 | "github.com/carlosframework/platform/internal/moves" |
| 25 | "github.com/carlosframework/platform/internal/moves/archive" |
| 26 | "github.com/carlosframework/platform/internal/provision" |
| 27 | "github.com/carlosframework/platform/internal/registry" |
| 28 | "github.com/carlosframework/platform/internal/release" |
| 29 | ) |
| 30 | |
| 31 | // The target box's part of an offline move |
| 32 | // (docs/superpowers/plans/2026-10-20-offline-operator-move.md, Task 5): the |
| 33 | // landing, open on the target, and the target's revert. offlineRound calls |
| 34 | // offlineTargetStep on a box whose placement equals rec.Move.To, in every |
| 35 | // offline state, after the source and sealer steps. |
| 36 | // |
| 37 | // The landing is a list of ensures (Fable's B2). Every item is judged from |
| 38 | // what is on disk, in the registry and in the bucket, compared with the |
| 39 | // immutable seal (seal.json, moves.OfflineSealKey), never with the sealer's |
| 40 | // living report: an item that matches is OK and left alone, an item that is |
| 41 | // absent is landed, and an item that is present and different is renamed into |
| 42 | // the move's target-before park directory and landed again. So a pass may run |
| 43 | // at any point, after any item, after the CAS or after an agent restart, and |
| 44 | // it neither parks a correct item nor fails on its own earlier extraction. |
| 45 | // |
| 46 | // Nothing here starts the instance or takes its lease. The row stays held |
| 47 | // "landing" until open; open re-verifies every item against the seal and only |
| 48 | // then releases the hold, and the build arrives the way every new instance's |
| 49 | // does, through ordinary adoption and root's existing stage pass. The first |
| 50 | // request promotes the staged replica (ExecBackend.restorePlain). |
| 51 | // |
| 52 | // An item's reason is what `carlos move offline status` shows beside it, so |
| 53 | // each is a recorded sentence (copy rows MV-O110 on). An error behind a reason |
| 54 | // is logged on the box, never put in the reason. |
| 55 | |
| 56 | // offlineTargetStep is the target step of one offline state. Every step is an |
| 57 | // ensure: an error is logged by the pass, throttled per host, and the next |
| 58 | // pass runs the step again. |
| 59 | func (w *movesWorker) offlineTargetStep(ctx context.Context, s release.Store, rec provision.Record) error { |
| 60 | if w.cfg.Box == "" { |
| 61 | return errors.New("offline move: this box has no label to report under") |
| 62 | } |
| 63 | switch rec.Move.State { |
| 64 | case provision.OfflineLanding, provision.OfflineLanded: |
| 65 | return w.offlineLand(ctx, s, rec) |
| 66 | case provision.OfflineOpen: |
| 67 | return w.offlineTargetOpen(ctx, s, rec) |
| 68 | case provision.OfflineReverting: |
| 69 | return w.offlineTargetRevert(ctx, s, rec) |
| 70 | } |
| 71 | return nil |
| 72 | } |
| 73 | |
| 74 | // offlineLand is the landing pass in offline-landing and offline-landed: |
| 75 | // ensure every item, rewrite the target report, and, once every item is OK |
| 76 | // in offline-landing, move the record to offline-landed. |
| 77 | func (w *movesWorker) offlineLand(ctx context.Context, s release.Store, rec provision.Record) error { |
| 78 | m := rec.Move |
| 79 | if rec.Region != m.To { |
| 80 | return nil // the flip has not reached this record yet |
| 81 | } |
| 82 | r, ok, err := w.landReport(ctx, s, rec) |
| 83 | if err != nil { |
| 84 | return err |
| 85 | } |
| 86 | if err := writeOfflineReport(ctx, s, r); err != nil { |
| 87 | return err |
| 88 | } |
| 89 | if err := w.landCrash("report"); err != nil { |
| 90 | return err |
| 91 | } |
| 92 | if !ok { |
| 93 | _, bad := r.ItemsOK(moves.OfflineOwnerLanding) |
| 94 | w.notice(rec.Host+"#offline-land", "moves: %s: offline move %s: landing, not yet OK: %s", rec.Host, m.ID, strings.Join(bad, ", ")) |
| 95 | return nil |
| 96 | } |
| 97 | if m.State != provision.OfflineLanding { |
| 98 | return nil |
| 99 | } |
| 100 | if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.OfflineLandedBy(m.ID, m.To)); err != nil { |
| 101 | if errors.Is(err, provision.ErrMoveConflict) { |
| 102 | log.Printf("moves: %s: offline move %s: not marked landed: %v", rec.Host, m.ID, err) |
| 103 | return nil |
| 104 | } |
| 105 | return err |
| 106 | } |
| 107 | log.Printf("moves: %s: offline move %s: landed; every item is OK", rec.Host, m.ID) |
| 108 | return w.landCrash("landed") |
| 109 | } |
| 110 | |
| 111 | // landCrash is the landing's test seam (moveCfg.landFault, the landing's |
| 112 | // crash points): an error stands for a crash at point, after the item of |
| 113 | // that name was ensured ("report" after the report is written, "landed" |
| 114 | // after the CAS, "released" after open's release). Nil in production. |
| 115 | func (w *movesWorker) landCrash(point string) error { |
| 116 | if w.cfg.landFault == nil { |
| 117 | return nil |
| 118 | } |
| 119 | return w.cfg.landFault("offline-" + point) |
| 120 | } |
| 121 | |
| 122 | // offlineTargetOpen is the target step at offline-open. While the row is |
| 123 | // still held by the move, every landing item is ensured and verified against |
| 124 | // the seal once more, and the verified report is written; only when every |
| 125 | // item is OK is the hold released (under reconcileMu and the host's hold |
| 126 | // lock, as the source's revert releases its own). The row then reads ready: |
| 127 | // ordinary adoption stages the channel's build, and the first request |
| 128 | // promotes moves/<host>/. Then the move's archive prefix is deleted from this |
| 129 | // box's bucket and the box reports opened. A pass after the release finds the |
| 130 | // row ready and does not verify again (the staging may already be promoted |
| 131 | // and gone); it goes on only from this box's own report with every item OK. |
| 132 | func (w *movesWorker) offlineTargetOpen(ctx context.Context, s release.Store, rec provision.Record) error { |
| 133 | cfg := w.cfg |
| 134 | m, host := rec.Move, rec.Host |
| 135 | rt, err := w.offlineRow(host) |
| 136 | if err != nil { |
| 137 | return err |
| 138 | } |
| 139 | prev := w.prevTargetReport(ctx, s, host, m.ID) |
| 140 | var items map[string]moves.Check |
| 141 | switch { |
| 142 | case rt == nil: |
| 143 | return fmt.Errorf("offline move %s: %s: no row on the target at open", m.ID, host) |
| 144 | case rt.HoldOp == m.ID && registry.IsHold(rt.Status): |
| 145 | r, ok, err := w.landReport(ctx, s, rec) |
| 146 | if err != nil { |
| 147 | return err |
| 148 | } |
| 149 | // The verified report goes first, so a crash after the release |
| 150 | // leaves the items this pass verified, not an older pass's. |
| 151 | if err := writeOfflineReport(ctx, s, r); err != nil { |
| 152 | return err |
| 153 | } |
| 154 | if !ok { |
| 155 | _, bad := r.ItemsOK(moves.OfflineOwnerLanding) |
| 156 | return fmt.Errorf("offline move %s: %s: not opening: %s not OK on this box", m.ID, host, strings.Join(bad, ", ")) |
| 157 | } |
| 158 | items = r.Items |
| 159 | if cfg.ReleaseHold == nil { |
| 160 | return errors.New("offline move: no registry to release the hold through") |
| 161 | } |
| 162 | var relErr error |
| 163 | reconcileMu.Lock() |
| 164 | provision.WithHostHoldLock(host, func() { |
| 165 | _, relErr = cfg.ReleaseHold(host, m.ID) |
| 166 | }) |
| 167 | reconcileMu.Unlock() |
| 168 | if relErr != nil { |
| 169 | return relErr |
| 170 | } |
| 171 | if rt, err = w.offlineRow(host); err != nil { |
| 172 | return err |
| 173 | } |
| 174 | if rt == nil || registry.IsHold(rt.Status) { |
| 175 | return fmt.Errorf("offline move %s: %s: the hold was not released", m.ID, host) |
| 176 | } |
| 177 | log.Printf("moves: %s: offline move %s: opened; the row is ready and adoption stages its build", host, m.ID) |
| 178 | if err := w.landCrash("released"); err != nil { |
| 179 | return err |
| 180 | } |
| 181 | case rt.HoldOp != "" || registry.IsHold(rt.Status): |
| 182 | return fmt.Errorf("offline move %s: %s is held %s by %q, not by this move", m.ID, host, rt.Status, rt.HoldOp) |
| 183 | default: |
| 184 | // The row is ready: this box released it on an earlier pass, which |
| 185 | // wrote the verified report first. A ready row this box never |
| 186 | // verified (reconcile mints a missing row ready once the record is |
| 187 | // open) is not vouched for. |
| 188 | if prev == nil { |
| 189 | return fmt.Errorf("offline move %s: %s is ready on this box, which has no landing report for the move", m.ID, host) |
| 190 | } |
| 191 | if ok, bad := prev.ItemsOK(moves.OfflineOwnerLanding); !ok { |
| 192 | return fmt.Errorf("offline move %s: %s is ready on this box, but its last report has %s not OK", m.ID, host, strings.Join(bad, ", ")) |
| 193 | } |
| 194 | items = prev.Items |
| 195 | } |
| 196 | if prev == nil || prev.Phase != moves.OfflinePhaseOpened { |
| 197 | st, err := w.landStore(ctx) |
| 198 | if err != nil { |
| 199 | return err |
| 200 | } |
| 201 | if err := lease.DeletePrefix(ctx, st, cfg.OwnBucket, moves.OfflineArchivePrefix(host, m.ID)); err != nil { |
| 202 | return fmt.Errorf("offline move %s: delete the archive: %w", m.ID, err) |
| 203 | } |
| 204 | } |
| 205 | return writeOfflineReport(ctx, s, moves.OfflineReport{ |
| 206 | Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: moves.OfflinePhaseOpened, At: cfg.now().UTC(), |
| 207 | Items: landingItemsOnly(items), RowPresent: rt != nil, Files: isDir(filepath.Join(cfg.DataDir, host)), Env: lexists(filepath.Join(cfg.DataDir, host+".env")), |
| 208 | }) |
| 209 | } |
| 210 | |
| 211 | // offlineTargetRevert is the target step at offline-reverting. When this box |
| 212 | // holds the row for the move, or has reported for the move at all, it first |
| 213 | // renames what the landing may have extracted and rendered (the files |
| 214 | // directory, <host>.env and <host>.config.env) into the move's target park |
| 215 | // directory, and only then deletes the held row (DeleteIfHold, which takes |
| 216 | // its timers with it) under reconcileMu and the hold lock, so a reconcile |
| 217 | // pass that listed the record at landing cannot mint it held again after |
| 218 | // the delete. Parking first means a crash anywhere leaves nothing of the |
| 219 | // move's in the data directory without a row to say why; a first pass that |
| 220 | // extracted and crashed before it could report is parked all the same, |
| 221 | // since its row is still held. Then it reports reverted. The source is |
| 222 | // untouched, so the revert loses nothing. A box that never had the row held |
| 223 | // for the move and never reported does nothing. |
| 224 | func (w *movesWorker) offlineTargetRevert(ctx context.Context, s release.Store, rec provision.Record) error { |
| 225 | cfg := w.cfg |
| 226 | m, host := rec.Move, rec.Host |
| 227 | rt, err := w.offlineRow(host) |
| 228 | if err != nil { |
| 229 | return err |
| 230 | } |
| 231 | prev := w.prevTargetReport(ctx, s, host, m.ID) |
| 232 | held := rt != nil && rt.HoldOp == m.ID && registry.IsHold(rt.Status) |
| 233 | if rt != nil && !held { |
| 234 | // Not this move's row (the move holds its row until open, and an |
| 235 | // open move is never aborted): leave it and what it serves. |
| 236 | return fmt.Errorf("offline move %s: %s is %q held by %q on this box, not by the move; the revert leaves it", m.ID, host, rt.Status, rt.HoldOp) |
| 237 | } |
| 238 | if !held && prev == nil { |
| 239 | return nil |
| 240 | } |
| 241 | parkDir := moves.OfflineParkDir(cfg.DataDir, host, m.ID, moves.OfflineRoleTarget) |
| 242 | if parkDir == "" { |
| 243 | return fmt.Errorf("offline move %s: no park directory for %s under %q", m.ID, host, cfg.DataDir) |
| 244 | } |
| 245 | for _, p := range []string{filepath.Join(cfg.DataDir, host), filepath.Join(cfg.DataDir, host+".env"), filepath.Join(cfg.DataDir, host+".config.env")} { |
| 246 | if err := parkInto(p, parkDir); err != nil { |
| 247 | return fmt.Errorf("offline move %s: park %s: %w", m.ID, p, err) |
| 248 | } |
| 249 | } |
| 250 | if held { |
| 251 | if cfg.DeleteIfHold == nil { |
| 252 | return errors.New("offline move: no registry to delete the held row through") |
| 253 | } |
| 254 | var delErr error |
| 255 | reconcileMu.Lock() |
| 256 | provision.WithHostHoldLock(host, func() { |
| 257 | _, delErr = cfg.DeleteIfHold(host, m.ID) |
| 258 | }) |
| 259 | reconcileMu.Unlock() |
| 260 | if delErr != nil { |
| 261 | return delErr |
| 262 | } |
| 263 | if rt, err = w.offlineRow(host); err != nil { |
| 264 | return err |
| 265 | } |
| 266 | if rt != nil && rt.HoldOp == m.ID { |
| 267 | return fmt.Errorf("offline move %s: %s is still held by the move", m.ID, host) |
| 268 | } |
| 269 | log.Printf("moves: %s: offline move %s: the target's held row is deleted", host, m.ID) |
| 270 | } |
| 271 | return writeOfflineReport(ctx, s, moves.OfflineReport{ |
| 272 | Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC(), |
| 273 | RowPresent: rt != nil, Files: isDir(filepath.Join(cfg.DataDir, host)), Env: lexists(filepath.Join(cfg.DataDir, host+".env")), |
| 274 | }) |
| 275 | } |
| 276 | |
| 277 | // parkInto renames p into dir under its own name (parkName numbers it when |
| 278 | // the name is taken). An absent p is nothing to do. |
| 279 | func parkInto(p, dir string) error { |
| 280 | if !lexists(p) { |
| 281 | return nil |
| 282 | } |
| 283 | if err := os.MkdirAll(dir, 0o700); err != nil { |
| 284 | return err |
| 285 | } |
| 286 | dst := parkName(dir, filepath.Base(p)) |
| 287 | if err := os.Rename(p, dst); err != nil { |
| 288 | return err |
| 289 | } |
| 290 | log.Printf("moves: parked %s at %s", p, dst) |
| 291 | return nil |
| 292 | } |
| 293 | |
| 294 | // landingItemsOnly keeps the items a target report may carry. |
| 295 | func landingItemsOnly(in map[string]moves.Check) map[string]moves.Check { |
| 296 | if len(in) == 0 { |
| 297 | return nil |
| 298 | } |
| 299 | out := map[string]moves.Check{} |
| 300 | for id, c := range in { |
| 301 | if owner, ok := moves.OfflineItemOwner(id); ok && owner == moves.OfflineOwnerLanding { |
| 302 | out[id] = c |
| 303 | } |
| 304 | } |
| 305 | return out |
| 306 | } |
| 307 | |
| 308 | // prevTargetReport is this box's last target report for the move, nil when |
| 309 | // there is none or it cannot be trusted. |
| 310 | func (w *movesWorker) prevTargetReport(ctx context.Context, s release.Store, host, op string) *moves.OfflineReport { |
| 311 | key := moves.OfflineReportKey(host, op, moves.OfflineRoleTarget, w.cfg.Box) |
| 312 | if key == "" { |
| 313 | return nil |
| 314 | } |
| 315 | body, _, err := s.Get(ctx, key) |
| 316 | if err != nil { |
| 317 | return nil |
| 318 | } |
| 319 | var r moves.OfflineReport |
| 320 | if json.Unmarshal(body, &r) != nil || moves.ValidateOfflineReport(r) != nil || r.Host != host || r.Op != op || r.Role != moves.OfflineRoleTarget { |
| 321 | return nil |
| 322 | } |
| 323 | return &r |
| 324 | } |
| 325 | |
| 326 | // landStore is this box's own data bucket (moveCfg.Mirror for its own |
| 327 | // region): the staged replica, the archive, the timers snapshot and the |
| 328 | // source's env report. |
| 329 | func (w *movesWorker) landStore(ctx context.Context) (lease.MirrorStore, error) { |
| 330 | if w.cfg.Mirror == nil || w.cfg.OwnBucket == "" { |
| 331 | return nil, errors.New("offline move: no data bucket on this box") |
| 332 | } |
| 333 | return w.mirror(ctx, w.cfg.OwnBucketRegion) |
| 334 | } |
| 335 | |
| 336 | // landRun is one landing pass's state. |
| 337 | type landRun struct { |
| 338 | w *movesWorker |
| 339 | s release.Store |
| 340 | rec provision.Record |
| 341 | m *provision.MoveState |
| 342 | host string |
| 343 | seal moves.OfflineReport |
| 344 | rt *registry.Route |
| 345 | prev *moves.OfflineReport |
| 346 | |
| 347 | items map[string]moves.Check |
| 348 | hard bool // an item failed in a way the next pass will not clear by itself |
| 349 | // staged is the staged replica's listing as verified this pass (the |
| 350 | // report's Staged): the db item restores once per listing. |
| 351 | staged map[string]moves.StagedObject |
| 352 | |
| 353 | st lease.MirrorStore |
| 354 | arch *offlineArchive |
| 355 | archErr error |
| 356 | archHard bool |
| 357 | } |
| 358 | |
| 359 | // offlineArchive is the move's archive in this box's bucket, read once per |
| 360 | // pass and only when an item needs it. |
| 361 | type offlineArchive struct { |
| 362 | manifest archive.Manifest |
| 363 | sealed []byte |
| 364 | } |
| 365 | |
| 366 | func (r *landRun) cfg() moveCfg { return r.w.cfg } |
| 367 | |
| 368 | func (r *landRun) ok(id string) { r.items[id] = moves.Check{OK: true} } |
| 369 | |
| 370 | // landWait records an item not yet OK that a later pass may make OK. reason is |
| 371 | // a recorded sentence. |
| 372 | func (r *landRun) landWait(id, reason string) { r.items[id] = moves.Check{Reason: reason} } |
| 373 | |
| 374 | // landWaitErr is landWait for an item an error stopped this pass: the error is the |
| 375 | // box's log line, never the reason. |
| 376 | func (r *landRun) landWaitErr(id, reason string, err error) { |
| 377 | r.w.notice(r.host+"#offline-land-"+id, "moves: %s: offline move %s: %s: %s: %v", r.host, r.m.ID, id, reason, err) |
| 378 | r.items[id] = moves.Check{Reason: reason} |
| 379 | } |
| 380 | |
| 381 | // landFail records a mismatch with the seal: the report's phase becomes failed. |
| 382 | func (r *landRun) landFail(id, reason string) { |
| 383 | r.items[id] = moves.Check{Reason: reason} |
| 384 | r.hard = true |
| 385 | } |
| 386 | |
| 387 | // landFailErr is landFail with the error behind it logged on the box. |
| 388 | func (r *landRun) landFailErr(id, reason string, err error) { |
| 389 | r.w.notice(r.host+"#offline-land-"+id, "moves: %s: offline move %s: %s: %s: %v", r.host, r.m.ID, id, reason, err) |
| 390 | r.landFail(id, reason) |
| 391 | } |
| 392 | |
| 393 | func (r *landRun) isOK(id string) bool { return r.items[id].OK } |
| 394 | |
| 395 | func (r *landRun) filesDir() string { return filepath.Join(r.cfg().DataDir, r.host) } |
| 396 | func (r *landRun) envPath() string { return filepath.Join(r.cfg().DataDir, r.host+".env") } |
| 397 | func (r *landRun) configPath() string { return filepath.Join(r.cfg().DataDir, r.host+".config.env") } |
| 398 | func (r *landRun) beforeDir() string { |
| 399 | return moves.OfflineParkDir(r.cfg().DataDir, r.host, r.m.ID, moves.OfflineParkTargetBefore) |
| 400 | } |
| 401 | |
| 402 | // landReport ensures every landing item and returns the target report with |
| 403 | // its phase (landed, or failed on a mismatch), and whether every item is OK. |
| 404 | // Items wait, never fail, for what a later pass can mend (an S3 error, a row |
| 405 | // not minted yet, a render that could not run); they fail on a mismatch with |
| 406 | // the seal. |
| 407 | func (w *movesWorker) landReport(ctx context.Context, s release.Store, rec provision.Record) (moves.OfflineReport, bool, error) { |
| 408 | cfg := w.cfg |
| 409 | r := &landRun{w: w, s: s, rec: rec, m: rec.Move, host: rec.Host, items: map[string]moves.Check{}} |
| 410 | r.prev = w.prevTargetReport(ctx, s, r.host, r.m.ID) |
| 411 | rt, err := w.offlineRow(r.host) |
| 412 | if err != nil { |
| 413 | return moves.OfflineReport{}, false, err |
| 414 | } |
| 415 | r.rt = rt |
| 416 | r.row() |
| 417 | if r.isOK("row") && r.readSeal(ctx) { |
| 418 | // In this order: files before config and the env check (the key is |
| 419 | // read from both files), config before the env pin. |
| 420 | for _, step := range []struct { |
| 421 | id string |
| 422 | run func() |
| 423 | }{ |
| 424 | {"warm-cache", r.warmCache}, |
| 425 | {"key-adoption", r.keyAdoption}, |
| 426 | {"lease", func() { r.lease(ctx) }}, |
| 427 | {"db", func() { r.db(ctx) }}, |
| 428 | {"files", func() { r.files(ctx) }}, |
| 429 | {"config", func() { r.config(ctx) }}, |
| 430 | {"env", r.env}, |
| 431 | {"env-pin", func() { r.envPin(ctx) }}, |
| 432 | {"timers", func() { r.timers(ctx) }}, |
| 433 | {"next-due", r.nextDue}, |
| 434 | {"replication", r.replication}, |
| 435 | {"release", func() { r.release(ctx) }}, |
| 436 | } { |
| 437 | step.run() |
| 438 | if err := w.landCrash(step.id); err != nil { |
| 439 | return moves.OfflineReport{}, false, err |
| 440 | } |
| 441 | } |
| 442 | } |
| 443 | for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) { |
| 444 | if _, done := r.items[it.ID]; done { |
| 445 | continue |
| 446 | } |
| 447 | if r.isOK("row") { |
| 448 | r.landWait(it.ID, "not checked: the seal could not be used") |
| 449 | } else { |
| 450 | r.landWait(it.ID, "not checked: the row is not held for this move") |
| 451 | } |
| 452 | } |
| 453 | phase := moves.OfflinePhaseLanded |
| 454 | if r.hard { |
| 455 | phase = moves.OfflinePhaseFailed |
| 456 | } |
| 457 | out := moves.OfflineReport{ |
| 458 | Op: r.m.ID, Host: r.host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: phase, At: cfg.now().UTC(), Items: r.items, |
| 459 | Held: r.rt != nil && r.rt.HoldOp == r.m.ID && registry.IsHold(r.rt.Status), RowPresent: r.rt != nil, |
| 460 | LocalDB: r.anyWarm(), Files: isDir(r.filesDir()), Env: lexists(r.envPath()), Staged: r.staged, |
| 461 | } |
| 462 | if r.rt != nil { |
| 463 | out.NextDue, out.Replication = r.rt.NextDue, r.rt.Replication |
| 464 | } |
| 465 | ok, _ := out.ItemsOK(moves.OfflineOwnerLanding) |
| 466 | return out, ok, nil |
| 467 | } |
| 468 | |
| 469 | // row: the reconciled row exists, held "landing" by this move, at flat paths. |
| 470 | func (r *landRun) row() { |
| 471 | rt := r.rt |
| 472 | switch { |
| 473 | case rt == nil: |
| 474 | r.landWait("row", "no row yet: reconcile mints it held") |
| 475 | case rt.HoldOp != "" && rt.HoldOp != r.m.ID: |
| 476 | r.landFail("row", "the row is held by another operation") |
| 477 | case rt.HoldOp != r.m.ID || rt.Status != "landing": |
| 478 | r.landWait("row", "the row is not held landing by this move yet") |
| 479 | case provision.NestedPaths(r.cfg().DataDir, *rt): |
| 480 | r.landFail("row", "the row's paths are nested, and an offline move lands flat rows only") |
| 481 | default: |
| 482 | r.ok("row") |
| 483 | } |
| 484 | } |
| 485 | |
| 486 | // readSeal reads seal.json and checks it against the record's evidence: |
| 487 | // the key and the sha256 of the exact bytes the sealer wrote, once. |
| 488 | func (r *landRun) readSeal(ctx context.Context) bool { |
| 489 | src := r.m.Source |
| 490 | if src == nil || src.Evidence == "" { |
| 491 | r.landFailAll("the record carries no seal evidence") |
| 492 | return false |
| 493 | } |
| 494 | key := moves.OfflineSealKey(r.host, r.m.ID) |
| 495 | raw, _, err := r.s.Get(ctx, key) |
| 496 | if errors.Is(err, os.ErrNotExist) { |
| 497 | r.landFailAll("the seal is missing") |
| 498 | return false |
| 499 | } |
| 500 | if err != nil { |
| 501 | r.w.notice(r.host+"#offline-land-seal", "moves: %s: offline move %s: read %s: %v", r.host, r.m.ID, key, err) |
| 502 | r.landWaitAll("the seal could not be read") |
| 503 | return false |
| 504 | } |
| 505 | if moves.OfflineSealEvidence(r.host, r.m.ID, raw) != src.Evidence { |
| 506 | r.landFailAll("the seal is not the one the record names") |
| 507 | return false |
| 508 | } |
| 509 | var seal moves.OfflineReport |
| 510 | if err := json.Unmarshal(raw, &seal); err != nil { |
| 511 | r.landFailAll("the seal does not decode") |
| 512 | return false |
| 513 | } |
| 514 | if err := moves.ValidateOfflineSeal(seal, r.host, r.m.ID); err != nil { |
| 515 | r.w.notice(r.host+"#offline-land-seal", "moves: %s: offline move %s: the seal: %v", r.host, r.m.ID, err) |
| 516 | r.landFailAll("the seal is not a valid seal of this move") |
| 517 | return false |
| 518 | } |
| 519 | r.seal = seal |
| 520 | return true |
| 521 | } |
| 522 | |
| 523 | // landFailAll fails every item but row for a seal that cannot be used. |
| 524 | func (r *landRun) landFailAll(reason string) { |
| 525 | for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) { |
| 526 | if it.ID != "row" { |
| 527 | r.landFail(it.ID, reason) |
| 528 | } |
| 529 | } |
| 530 | } |
| 531 | |
| 532 | func (r *landRun) landWaitAll(reason string) { |
| 533 | for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) { |
| 534 | if it.ID != "row" { |
| 535 | r.landWait(it.ID, reason) |
| 536 | } |
| 537 | } |
| 538 | } |
| 539 | |
| 540 | // warmPaths are the local database artifacts a box could trust as a warm |
| 541 | // cache for the host: the row's database and the default flat one, each |
| 542 | // with its WAL, SHM, both litestream meta directories and its seed marker. |
| 543 | func (r *landRun) warmPaths() []string { |
| 544 | dbs := []string{provision.DefaultDBPath(r.cfg().DataDir, r.host)} |
| 545 | if r.rt != nil && r.rt.DBPath != "" && r.rt.DBPath != dbs[0] { |
| 546 | dbs = append(dbs, r.rt.DBPath) |
| 547 | } |
| 548 | var out []string |
| 549 | for _, db := range dbs { |
| 550 | out = append(out, db, db+"-wal", db+"-shm", db+"-litestream", activatorMetaDir(db), db+activator.SeedMarkerSuffix) |
| 551 | } |
| 552 | return out |
| 553 | } |
| 554 | |
| 555 | func (r *landRun) anyWarm() bool { |
| 556 | for _, p := range r.warmPaths() { |
| 557 | if lexists(p) { |
| 558 | return true |
| 559 | } |
| 560 | } |
| 561 | return false |
| 562 | } |
| 563 | |
| 564 | // ensureAbsent renames every present path into target-before and reports |
| 565 | // the item from what is left. |
| 566 | func (r *landRun) ensureAbsent(id string, paths []string) { |
| 567 | dir := r.beforeDir() |
| 568 | for _, p := range paths { |
| 569 | if !lexists(p) { |
| 570 | continue |
| 571 | } |
| 572 | if err := parkInto(p, dir); err != nil { |
| 573 | r.landWaitErr(id, "a path the target held for the host could not be parked", err) |
| 574 | return |
| 575 | } |
| 576 | } |
| 577 | for _, p := range paths { |
| 578 | if lexists(p) { |
| 579 | r.landWait(id, "a path the target held for the host is still there") |
| 580 | return |
| 581 | } |
| 582 | } |
| 583 | r.ok(id) |
| 584 | } |
| 585 | |
| 586 | // warmCache: no local database or meta directory for the host. The target |
| 587 | // never woke it, so anything here predates the move and would win over the |
| 588 | // staged replica on the first wake. |
| 589 | func (r *landRun) warmCache() { r.ensureAbsent("warm-cache", r.warmPaths()) } |
| 590 | |
| 591 | // keyAdoption: no instance-key adoption record (written only for claim |
| 592 | // launches; never carried). |
| 593 | func (r *landRun) keyAdoption() { |
| 594 | r.ensureAbsent("key-adoption", []string{filepath.Join(r.cfg().DataDir, r.host+instancekey.AdoptionSuffix)}) |
| 595 | } |
| 596 | |
| 597 | // lease: the target bucket's lock for the host is absent, a lifted |
| 598 | // tombstone, or an ordinary lease that has expired. A tombstone of another |
| 599 | // move is lifted. Nothing here ever takes the lease. Expiry is read on the |
| 600 | // wall clock, as the lease package reads it. |
| 601 | func (r *landRun) lease(ctx context.Context) { |
| 602 | cfg := r.cfg() |
| 603 | loc := activator.HibernatePrefix(cfg.OwnPrefix, r.host) |
| 604 | _, owner, expires, err := lease.LockState(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion) |
| 605 | if err != nil { |
| 606 | r.landWaitErr("lease", "the target's lock could not be read", err) |
| 607 | return |
| 608 | } |
| 609 | switch { |
| 610 | case owner == "" && expires.IsZero(): |
| 611 | r.ok("lease") |
| 612 | return |
| 613 | case lease.IsLiftedTombstone(owner, expires): |
| 614 | r.ok("lease") |
| 615 | return |
| 616 | case owner == moves.TombstoneOwner(r.m.ID): |
| 617 | r.landFail("lease", "the target's lock is this move's own tombstone") |
| 618 | return |
| 619 | case !strings.HasPrefix(owner, lease.TombstonePrefix): |
| 620 | if time.Now().After(expires) { |
| 621 | r.ok("lease") |
| 622 | return |
| 623 | } |
| 624 | r.w.notice(r.host+"#offline-land-lease", "moves: %s: offline move %s: the target's lock is held by %s until %s", r.host, r.m.ID, owner, expires.UTC().Format(time.RFC3339)) |
| 625 | r.landWait("lease", "a writer holds the target's lock") |
| 626 | return |
| 627 | } |
| 628 | stale := owner |
| 629 | if _, err := lease.LiftTombstone(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion, func(o string) bool { return o == stale }); err != nil { |
| 630 | r.landWaitErr("lease", "a stale tombstone in the target's bucket could not be lifted", err) |
| 631 | return |
| 632 | } |
| 633 | _, owner, expires, err = lease.LockState(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion) |
| 634 | if err != nil { |
| 635 | r.landWaitErr("lease", "the target's lock could not be read", err) |
| 636 | return |
| 637 | } |
| 638 | if (owner == "" && expires.IsZero()) || lease.IsLiftedTombstone(owner, expires) { |
| 639 | log.Printf("moves: %s: offline move %s: lifted the stale tombstone %s in the target's bucket", r.host, r.m.ID, stale) |
| 640 | r.ok("lease") |
| 641 | return |
| 642 | } |
| 643 | r.landWait("lease", "the target's lock is still held after the lift") |
| 644 | } |
| 645 | |
| 646 | func (r *landRun) store(ctx context.Context) (lease.MirrorStore, error) { |
| 647 | if r.st != nil { |
| 648 | return r.st, nil |
| 649 | } |
| 650 | st, err := r.w.landStore(ctx) |
| 651 | if err != nil { |
| 652 | return nil, err |
| 653 | } |
| 654 | r.st = st |
| 655 | return st, nil |
| 656 | } |
| 657 | |
| 658 | // stagedDiff describes how got differs from want, "" when equal: a log line. |
| 659 | func stagedDiff(want, got map[string]moves.StagedObject) string { |
| 660 | var missing, extra, changed []string |
| 661 | for k, w := range want { |
| 662 | g, ok := got[k] |
| 663 | switch { |
| 664 | case !ok: |
| 665 | missing = append(missing, k) |
| 666 | case g.Size != w.Size || g.ETag != w.ETag: |
| 667 | changed = append(changed, k) |
| 668 | } |
| 669 | } |
| 670 | for k := range got { |
| 671 | if _, ok := want[k]; !ok { |
| 672 | extra = append(extra, k) |
| 673 | } |
| 674 | } |
| 675 | sort.Strings(missing) |
| 676 | sort.Strings(extra) |
| 677 | sort.Strings(changed) |
| 678 | if len(missing)+len(extra)+len(changed) == 0 { |
| 679 | return "" |
| 680 | } |
| 681 | return fmt.Sprintf("missing %v, extra %v, changed %v", missing, extra, changed) |
| 682 | } |
| 683 | |
| 684 | // db: the staged replica under moves/<host>/ is exactly what the seal |
| 685 | // staged (nothing consumes it before open), its newest TXID |
| 686 | // (replicaPosition, the sealer's own rule) is the seal's FinalTXID, and it |
| 687 | // restores clean (lease.VerifyReplica into a scratch directory under the |
| 688 | // move's verify park directory, removed after). The restore runs once per |
| 689 | // listing: the verified listing is the report's Staged, and a pass whose |
| 690 | // listing equals the last report's, with db OK there, does not restore |
| 691 | // again. The report is this move's, so the seal it was verified against is |
| 692 | // the same immutable one. |
| 693 | func (r *landRun) db(ctx context.Context) { |
| 694 | cfg := r.cfg() |
| 695 | st, err := r.store(ctx) |
| 696 | if err != nil { |
| 697 | r.landWaitErr("db", "the staged replica could not be listed", err) |
| 698 | return |
| 699 | } |
| 700 | prefix := activator.MovesPrefix(r.host) |
| 701 | objs, err := st.List(ctx, cfg.OwnBucket, prefix) |
| 702 | if err != nil { |
| 703 | r.landWaitErr("db", "the staged replica could not be listed", err) |
| 704 | return |
| 705 | } |
| 706 | got := make(map[string]moves.StagedObject, len(objs)) |
| 707 | for k, o := range objs { |
| 708 | got[strings.TrimPrefix(k, prefix)] = moves.StagedObject{Size: o.Size, ETag: o.ETag} |
| 709 | } |
| 710 | if len(r.seal.Staged) == 0 { |
| 711 | r.landFail("db", "the seal staged nothing") |
| 712 | return |
| 713 | } |
| 714 | if diff := stagedDiff(r.seal.Staged, got); diff != "" { |
| 715 | r.w.notice(r.host+"#offline-land-db", "moves: %s: offline move %s: the staged replica differs from the seal's: %s", r.host, r.m.ID, diff) |
| 716 | r.landFail("db", "the staged replica is not the one the seal staged") |
| 717 | return |
| 718 | } |
| 719 | if final := replicaPosition(objs, prefix); final != r.seal.FinalTXID { |
| 720 | r.w.notice(r.host+"#offline-land-db", "moves: %s: offline move %s: the staged replica ends at TXID %d, the seal at %d", r.host, r.m.ID, final, r.seal.FinalTXID) |
| 721 | r.landFail("db", "the staged replica does not end at the sealed position") |
| 722 | return |
| 723 | } |
| 724 | if p := r.prev; p != nil && p.Items["db"].OK && reflect.DeepEqual(p.Staged, got) { |
| 725 | r.staged = got |
| 726 | r.ok("db") |
| 727 | return |
| 728 | } |
| 729 | dir := moves.OfflineParkDir(cfg.DataDir, r.host, r.m.ID, moves.OfflineParkVerify) |
| 730 | if dir == "" { |
| 731 | r.landFail("db", "the staged replica does not restore clean") |
| 732 | return |
| 733 | } |
| 734 | if err := os.MkdirAll(dir, 0o700); err != nil { |
| 735 | r.landWaitErr("db", "the staged replica could not be verified this pass", err) |
| 736 | return |
| 737 | } |
| 738 | defer os.RemoveAll(dir) |
| 739 | if _, err := lease.VerifyReplica(ctx, cfg.OwnBucket, strings.TrimSuffix(prefix, "/"), cfg.OwnBucketRegion, dir); err != nil { |
| 740 | r.landFailErr("db", "the staged replica does not restore clean", err) |
| 741 | return |
| 742 | } |
| 743 | r.staged = got |
| 744 | r.ok("db") |
| 745 | } |
| 746 | |
| 747 | // readArchive fetches the move's archive and its manifest from this box's |
| 748 | // bucket (the sealer's offlineArchiveSealed and offlineArchiveManifest |
| 749 | // under the move's archive prefix), once per pass and only when an item |
| 750 | // needs it. hard is a manifest that is not the seal's. |
| 751 | func (r *landRun) readArchive(ctx context.Context) (*offlineArchive, bool, error) { |
| 752 | if r.arch != nil || r.archErr != nil { |
| 753 | return r.arch, r.archHard, r.archErr |
| 754 | } |
| 755 | r.arch, r.archHard, r.archErr = r.fetchArchive(ctx) |
| 756 | return r.arch, r.archHard, r.archErr |
| 757 | } |
| 758 | |
| 759 | func (r *landRun) fetchArchive(ctx context.Context) (*offlineArchive, bool, error) { |
| 760 | body, err := r.object(ctx, offlineArchiveManifest) |
| 761 | if err != nil { |
| 762 | return nil, false, err |
| 763 | } |
| 764 | out := &offlineArchive{} |
| 765 | if err := json.Unmarshal(body, &out.manifest); err != nil { |
| 766 | return nil, true, fmt.Errorf("the archive manifest: %w", err) |
| 767 | } |
| 768 | am := out.manifest |
| 769 | if am.Op != r.m.ID || am.Plain != r.seal.FilesPlain || (r.seal.FilesSealed != "" && am.Sealed != r.seal.FilesSealed) { |
| 770 | return nil, true, fmt.Errorf("the archive manifest names op %s, plain %s, sealed %s", am.Op, am.Plain, am.Sealed) |
| 771 | } |
| 772 | if out.sealed, err = r.object(ctx, offlineArchiveSealed); err != nil { |
| 773 | return nil, false, err |
| 774 | } |
| 775 | return out, false, nil |
| 776 | } |
| 777 | |
| 778 | // object reads one of the sealer's objects under the move's archive prefix |
| 779 | // in this box's bucket. |
| 780 | func (r *landRun) object(ctx context.Context, name string) ([]byte, error) { |
| 781 | st, err := r.store(ctx) |
| 782 | if err != nil { |
| 783 | return nil, err |
| 784 | } |
| 785 | return st.Get(ctx, r.cfg().OwnBucket, moves.OfflineArchivePrefix(r.host, r.m.ID)+name) |
| 786 | } |
| 787 | |
| 788 | // filesMatch reports whether the flat files directory and <host>.env are |
| 789 | // the seal's, by the archive's own plaintext digest over the pair. A seal |
| 790 | // of a host with no files directory carries no files/ entry at all and |
| 791 | // requires no file; extraction then leaves an empty directory, whose digest |
| 792 | // with the env file is the seal's. |
| 793 | func (r *landRun) filesMatch() bool { |
| 794 | if !isDir(r.filesDir()) { |
| 795 | return false |
| 796 | } |
| 797 | plain, _, _, err := archive.PlainDigest(r.filesDir(), r.envPath()) |
| 798 | return err == nil && plain == r.seal.FilesPlain |
| 799 | } |
| 800 | |
| 801 | // files: <DataDir>/<host>/ and <host>.env carry the seal's plaintext digest. |
| 802 | // OK: left alone. Absent: fetched, verified and extracted. Present and |
| 803 | // different: the files directory (when not empty) and an env file that is |
| 804 | // not the seal's are renamed into target-before first, then extracted. |
| 805 | func (r *landRun) files(ctx context.Context) { |
| 806 | if r.filesMatch() { |
| 807 | r.ok("files") |
| 808 | return |
| 809 | } |
| 810 | cfg := r.cfg() |
| 811 | if cfg.ConfigPriv == nil { |
| 812 | r.landWait("files", "this box holds no CARLOS_SECRETS_KEY to open the archive with") |
| 813 | return |
| 814 | } |
| 815 | a, hard, err := r.readArchive(ctx) |
| 816 | switch { |
| 817 | case err != nil && hard: |
| 818 | r.landFailErr("files", "the archive is not the one the seal names", err) |
| 819 | return |
| 820 | case err != nil: |
| 821 | r.landWaitErr("files", "the archive could not be read", err) |
| 822 | return |
| 823 | } |
| 824 | bind := appconfig.Binding{Account: r.rec.Account, App: r.rec.App} |
| 825 | if err := archive.Verify(bytes.NewReader(a.sealed), a.manifest, cfg.ConfigPriv, bind); err != nil { |
| 826 | r.landFailErr("files", "the archive does not verify", err) |
| 827 | return |
| 828 | } |
| 829 | if nonEmptyOrNotDir(r.filesDir()) { |
| 830 | if err := parkInto(r.filesDir(), r.beforeDir()); err != nil { |
| 831 | r.landWaitErr("files", "the files directory the target held could not be parked", err) |
| 832 | return |
| 833 | } |
| 834 | } |
| 835 | if lexists(r.envPath()) && !sameSHA(r.seal.EnvSHA256, hexFileSHA256(r.envPath())) { |
| 836 | if err := parkInto(r.envPath(), r.beforeDir()); err != nil { |
| 837 | r.landWaitErr("files", "the env file the target held could not be parked", err) |
| 838 | return |
| 839 | } |
| 840 | } |
| 841 | if err := extractSealed(a, cfg, bind, r.filesDir(), r.envPath()); err != nil { |
| 842 | r.landWaitErr("files", "the archive could not be extracted", err) |
| 843 | return |
| 844 | } |
| 845 | log.Printf("moves: %s: offline move %s: extracted the files and env", r.host, r.m.ID) |
| 846 | if !r.filesMatch() { |
| 847 | r.landFail("files", "after extraction the files are not the seal's") |
| 848 | return |
| 849 | } |
| 850 | r.ok("files") |
| 851 | } |
| 852 | |
| 853 | // extractSealed opens the sealed archive and extracts it, as the Phase 1 |
| 854 | // install does (openAndExtract): Extract admits nothing it did not read |
| 855 | // whole. |
| 856 | func extractSealed(a *offlineArchive, cfg moveCfg, bind appconfig.Binding, filesDir, envPath string) error { |
| 857 | pr, pw := io.Pipe() |
| 858 | done := make(chan error, 1) |
| 859 | go func() { |
| 860 | err := archive.Open(bytes.NewReader(a.sealed), a.manifest.SealedDataKey, cfg.ConfigPriv, bind, pw) |
| 861 | pw.CloseWithError(err) |
| 862 | done <- err |
| 863 | }() |
| 864 | xerr := archive.Extract(pr, filesDir, envPath) |
| 865 | pr.CloseWithError(errors.New("extract stopped reading")) |
| 866 | if oerr := <-done; oerr != nil && xerr == nil { |
| 867 | xerr = oerr |
| 868 | } |
| 869 | return xerr |
| 870 | } |
| 871 | |
| 872 | // nonEmptyOrNotDir reports a path Extract would refuse as its files |
| 873 | // directory: present and not an empty directory. |
| 874 | func nonEmptyOrNotDir(p string) bool { |
| 875 | st, err := os.Lstat(p) |
| 876 | if err != nil { |
| 877 | return false |
| 878 | } |
| 879 | if !st.IsDir() { |
| 880 | return true |
| 881 | } |
| 882 | ents, err := os.ReadDir(p) |
| 883 | return err != nil || len(ents) > 0 |
| 884 | } |
| 885 | |
| 886 | // hexFileSHA256 is the hex sha256 of a file's bytes, "" when it cannot be |
| 887 | // read. |
| 888 | func hexFileSHA256(p string) string { |
| 889 | b, err := os.ReadFile(p) |
| 890 | if err != nil { |
| 891 | return "" |
| 892 | } |
| 893 | sum := sha256.Sum256(b) |
| 894 | return hex.EncodeToString(sum[:]) |
| 895 | } |
| 896 | |
| 897 | // sameSHA reports whether a sealed sha256 names gotHex, the bare hex form a |
| 898 | // box computes. The seal may carry it bare or as "sha256:<hex>"; "" names |
| 899 | // "" only (nothing there). |
| 900 | func sameSHA(want, gotHex string) bool { |
| 901 | if gotHex == "" { |
| 902 | return want == "" |
| 903 | } |
| 904 | return want == gotHex || want == "sha256:"+gotHex |
| 905 | } |
| 906 | |
| 907 | // config: the held row's <host>.config.env rendered from the bucket bundle |
| 908 | // (RenderHeld, which admits the row held by this move and writes a |
| 909 | // move-bound ack). OK when the file holds the rendered bytes and this box's |
| 910 | // ack names this move and those bytes. Rendered every pass, so a bundle |
| 911 | // changed since the seal reaches the file and fails the env pin. |
| 912 | func (r *landRun) config(ctx context.Context) { |
| 913 | cfg := r.cfg() |
| 914 | if cfg.RenderHeld == nil { |
| 915 | r.landWait("config", "this box has no held render") |
| 916 | return |
| 917 | } |
| 918 | hr, err := cfg.RenderHeld(ctx, r.s, *r.rt, r.m.ID, nil, func(string) error { return nil }) |
| 919 | if err != nil { |
| 920 | r.landWaitErr("config", "the config could not be rendered", err) |
| 921 | return |
| 922 | } |
| 923 | if hr.Path != r.configPath() { |
| 924 | r.landFail("config", "the config was rendered at another path") |
| 925 | return |
| 926 | } |
| 927 | if hexFileSHA256(hr.Path) != hr.SHA256 { |
| 928 | r.landWait("config", "the config file does not hold the rendered bytes") |
| 929 | return |
| 930 | } |
| 931 | body, _, err := r.s.Get(ctx, appconfig.ConfigAckKey(cfg.Box, r.host)) |
| 932 | if err != nil { |
| 933 | r.landWaitErr("config", "the config ack could not be read", err) |
| 934 | return |
| 935 | } |
| 936 | var ack appconfig.ConfigAck |
| 937 | if err := json.Unmarshal(body, &ack); err != nil || ack.MoveOp != r.m.ID || ack.FileSHA256 != hr.SHA256 { |
| 938 | r.landWait("config", "the config ack does not name this move and the rendered file") |
| 939 | return |
| 940 | } |
| 941 | r.ok("config") |
| 942 | } |
| 943 | |
| 944 | // env: <host>.env is the source's byte for byte (the seal's sha256), and |
| 945 | // the instance key the app would be handed, from the env file and the |
| 946 | // rendered config as the activator merges them (instancekey.Effective), is |
| 947 | // the source's (the seal's key sha256). Both travel in the archive, so the |
| 948 | // files item lands them; this is the check. |
| 949 | func (r *landRun) env() { |
| 950 | if !r.isOK("files") { |
| 951 | r.landWait("env", "waits for the files item") |
| 952 | return |
| 953 | } |
| 954 | own, err := os.ReadFile(r.envPath()) |
| 955 | switch { |
| 956 | case errors.Is(err, os.ErrNotExist) && r.seal.EnvSHA256 == "": |
| 957 | case errors.Is(err, os.ErrNotExist): |
| 958 | r.landFail("env", "the env file is missing") |
| 959 | return |
| 960 | case err != nil: |
| 961 | r.landWaitErr("env", "the env file could not be read", err) |
| 962 | return |
| 963 | default: |
| 964 | if sum := sha256.Sum256(own); !sameSHA(r.seal.EnvSHA256, hex.EncodeToString(sum[:])) { |
| 965 | r.landFail("env", "the env file is not the source's") |
| 966 | return |
| 967 | } |
| 968 | } |
| 969 | if !r.isOK("config") { |
| 970 | r.landWait("env", "waits for the config item: the instance key is read from both files") |
| 971 | return |
| 972 | } |
| 973 | cfgEnv, err := os.ReadFile(r.configPath()) |
| 974 | if err != nil && !errors.Is(err, os.ErrNotExist) { |
| 975 | r.landWaitErr("env", "the config file could not be read", err) |
| 976 | return |
| 977 | } |
| 978 | key, _ := instancekey.Effective(instancekey.ParseEnv(cfgEnv), instancekey.ParseEnv(own)) |
| 979 | got := "" |
| 980 | if key != "" { |
| 981 | got = instancekey.Digest(key) |
| 982 | } |
| 983 | if !sameSHA(r.seal.KeySHA256, got) { |
| 984 | r.landFail("env", "the instance key is not the source's") |
| 985 | return |
| 986 | } |
| 987 | r.ok("env") |
| 988 | } |
| 989 | |
| 990 | // pinNames is the variable names a failed env pin reports. It is not a |
| 991 | // string, so a reason that prints it carries names, never values. |
| 992 | type pinNames []string |
| 993 | |
| 994 | func (n pinNames) String() string { return strings.Join(n, ", ") } |
| 995 | |
| 996 | // envPin: the composed launch environment of the held row |
| 997 | // (ExecBackend.LaunchLayers) has the source's portable digest. App settings |
| 998 | // must be equal; box settings are left out (moves.Pinned). On a mismatch the |
| 999 | // reason names the variables that differ (moves.CompareEnv against the |
| 1000 | // source's report the sealer wrote beside the archive), never their values. |
| 1001 | func (r *landRun) envPin(ctx context.Context) { |
| 1002 | cfg := r.cfg() |
| 1003 | if !r.isOK("files") || !r.isOK("config") { |
| 1004 | r.landWait("env-pin", "waits for the files and config items") |
| 1005 | return |
| 1006 | } |
| 1007 | if r.m.Source.EnvDigest != "" && r.m.Source.EnvDigest != r.seal.EnvDigest { |
| 1008 | r.landFail("env-pin", "the record and the seal disagree on this item") |
| 1009 | return |
| 1010 | } |
| 1011 | if cfg.LaunchLayers == nil { |
| 1012 | r.landWait("env-pin", "this box has no exec backend to compose the launch environment") |
| 1013 | return |
| 1014 | } |
| 1015 | key, err := moves.ReportKeyFromSecrets(cfg.SecretsKey) |
| 1016 | if err != nil { |
| 1017 | r.landWaitErr("env-pin", "this box holds no CARLOS_SECRETS_KEY to key the env report", err) |
| 1018 | return |
| 1019 | } |
| 1020 | env, inst, _, err := cfg.LaunchLayers(*r.rt) |
| 1021 | if err != nil { |
| 1022 | r.landWaitErr("env-pin", "the launch environment could not be composed", err) |
| 1023 | return |
| 1024 | } |
| 1025 | target := moves.ReportLaunch(key, env, inst) |
| 1026 | if target.Digest == r.seal.EnvDigest { |
| 1027 | r.ok("env-pin") |
| 1028 | return |
| 1029 | } |
| 1030 | names := r.envDiff(ctx, target) |
| 1031 | if len(names) == 0 { |
| 1032 | r.landFail("env-pin", "the launch environment differs from the source's") |
| 1033 | return |
| 1034 | } |
| 1035 | r.landFail("env-pin", fmt.Sprintf("the launch environment differs from the source's in %v", names)) |
| 1036 | } |
| 1037 | |
| 1038 | // envDiff is the names CompareEnv finds between the source's env report |
| 1039 | // (the sealer's offlineEnvReportObject, used only when its digest is the |
| 1040 | // seal's EnvDigest) and the target's, notes left out. None when the report |
| 1041 | // cannot be read or trusted. |
| 1042 | func (r *landRun) envDiff(ctx context.Context, target moves.EnvReport) pinNames { |
| 1043 | body, err := r.object(ctx, offlineEnvReportObject) |
| 1044 | if err != nil { |
| 1045 | r.w.notice(r.host+"#offline-land-envreport", "moves: %s: offline move %s: the source's env report: %v", r.host, r.m.ID, err) |
| 1046 | return nil |
| 1047 | } |
| 1048 | var src moves.EnvReport |
| 1049 | if json.Unmarshal(body, &src) != nil || !moves.CurrentEnvReport(src) || src.Digest != r.seal.EnvDigest { |
| 1050 | return nil |
| 1051 | } |
| 1052 | var out pinNames |
| 1053 | for _, d := range moves.CompareEnv(src, target, true, r.m.To == "") { |
| 1054 | if !d.Note() { |
| 1055 | out = append(out, d.Name+" ("+d.Kind+")") |
| 1056 | } |
| 1057 | } |
| 1058 | return out |
| 1059 | } |
| 1060 | |
| 1061 | // timers: the row's timer rows digest to the seal's. Otherwise they are |
| 1062 | // replaced from the snapshot beside the archive (its digest checked first) |
| 1063 | // and read back. |
| 1064 | func (r *landRun) timers(ctx context.Context) { |
| 1065 | reg := r.cfg().Registry |
| 1066 | if reg == nil { |
| 1067 | r.landWait("timers", "no registry on this box") |
| 1068 | return |
| 1069 | } |
| 1070 | if r.m.Source.TimersSHA256 != "" && r.m.Source.TimersSHA256 != r.seal.TimersDigest { |
| 1071 | r.landFail("timers", "the record and the seal disagree on this item") |
| 1072 | return |
| 1073 | } |
| 1074 | rows, err := reg.TimersFor(r.host) |
| 1075 | if err != nil { |
| 1076 | r.landWaitErr("timers", "the timers could not be read", err) |
| 1077 | return |
| 1078 | } |
| 1079 | if moves.TimersDigest(rows) == r.seal.TimersDigest { |
| 1080 | r.ok("timers") |
| 1081 | return |
| 1082 | } |
| 1083 | body, err := r.object(ctx, offlineTimersObject) |
| 1084 | if err != nil { |
| 1085 | r.landWaitErr("timers", "the timers snapshot could not be read", err) |
| 1086 | return |
| 1087 | } |
| 1088 | snap, err := moves.DecodeTimers(body) |
| 1089 | if err != nil || snap.Host != r.host || moves.TimersDigest(snap.Rows) != r.seal.TimersDigest { |
| 1090 | r.landFail("timers", "the timers snapshot is not the one the seal names") |
| 1091 | return |
| 1092 | } |
| 1093 | if err := reg.ReplaceTimers(r.host, snap.Rows); err != nil { |
| 1094 | r.landWaitErr("timers", "the timers could not be replaced", err) |
| 1095 | return |
| 1096 | } |
| 1097 | if rows, err = reg.TimersFor(r.host); err != nil { |
| 1098 | r.landWaitErr("timers", "the timers could not be read", err) |
| 1099 | return |
| 1100 | } |
| 1101 | if moves.TimersDigest(rows) != r.seal.TimersDigest { |
| 1102 | r.landFail("timers", "the timers read back differently from the seal's") |
| 1103 | return |
| 1104 | } |
| 1105 | r.ok("timers") |
| 1106 | } |
| 1107 | |
| 1108 | // reread refreshes the row after a column write. |
| 1109 | func (r *landRun) reread(id string) bool { |
| 1110 | rt, err := r.w.offlineRow(r.host) |
| 1111 | if err == nil && rt == nil { |
| 1112 | err = errors.New("the row is gone") |
| 1113 | } |
| 1114 | if err != nil { |
| 1115 | r.landWaitErr(id, "the row could not be read back", err) |
| 1116 | return false |
| 1117 | } |
| 1118 | r.rt = rt |
| 1119 | return true |
| 1120 | } |
| 1121 | |
| 1122 | // nextDue: the row's NextDue is the seal's; else set and read back. A write |
| 1123 | // error waits; only a read-back that differs fails. |
| 1124 | func (r *landRun) nextDue() { |
| 1125 | if r.rt.NextDue == r.seal.NextDue { |
| 1126 | r.ok("next-due") |
| 1127 | return |
| 1128 | } |
| 1129 | reg := r.cfg().Registry |
| 1130 | if reg == nil { |
| 1131 | r.landWait("next-due", "no registry on this box") |
| 1132 | return |
| 1133 | } |
| 1134 | if err := reg.SetNextDue(r.host, r.seal.NextDue); err != nil { |
| 1135 | r.landWaitErr("next-due", "next due could not be set", err) |
| 1136 | return |
| 1137 | } |
| 1138 | if !r.reread("next-due") { |
| 1139 | return |
| 1140 | } |
| 1141 | if r.rt.NextDue != r.seal.NextDue { |
| 1142 | r.landFail("next-due", "next due reads back differently from the seal's") |
| 1143 | return |
| 1144 | } |
| 1145 | r.ok("next-due") |
| 1146 | } |
| 1147 | |
| 1148 | // replication: the row's replication rung is the seal's; else set and read |
| 1149 | // back. A write error waits; only a read-back that differs fails. |
| 1150 | func (r *landRun) replication() { |
| 1151 | if r.rt.Replication == r.seal.Replication { |
| 1152 | r.ok("replication") |
| 1153 | return |
| 1154 | } |
| 1155 | reg := r.cfg().Registry |
| 1156 | if reg == nil { |
| 1157 | r.landWait("replication", "no registry on this box") |
| 1158 | return |
| 1159 | } |
| 1160 | if err := reg.SetReplication(r.host, r.seal.Replication); err != nil { |
| 1161 | r.landWaitErr("replication", "the replication rung could not be set", err) |
| 1162 | return |
| 1163 | } |
| 1164 | if !r.reread("replication") { |
| 1165 | return |
| 1166 | } |
| 1167 | if r.rt.Replication != r.seal.Replication { |
| 1168 | r.landFail("replication", "the replication rung reads back differently from the seal's") |
| 1169 | return |
| 1170 | } |
| 1171 | r.ok("replication") |
| 1172 | } |
| 1173 | |
| 1174 | // release: the app's channel resolves to the build the seal recorded, and |
| 1175 | // the target holds no live link to another build of the host |
| 1176 | // (liveNotSealed). A promote during the move fails it; the build itself |
| 1177 | // arrives after open through ordinary adoption. |
| 1178 | func (r *landRun) release(ctx context.Context) { |
| 1179 | cfg := r.cfg() |
| 1180 | if r.m.Source.Release != "" && r.m.Source.Release != r.seal.Release { |
| 1181 | r.landFail("release", "the record and the seal disagree on this item") |
| 1182 | return |
| 1183 | } |
| 1184 | if cfg.ChannelVersion == nil { |
| 1185 | r.landWait("release", "this box cannot resolve channels") |
| 1186 | return |
| 1187 | } |
| 1188 | v, err := cfg.ChannelVersion(ctx, r.s, *r.rt) |
| 1189 | if err != nil { |
| 1190 | r.landWaitErr("release", "the channel could not be resolved", err) |
| 1191 | return |
| 1192 | } |
| 1193 | if v != r.seal.Release { |
| 1194 | r.w.notice(r.host+"#offline-land-release", "moves: %s: offline move %s: the channel points at %q, the seal recorded %q", r.host, r.m.ID, v, r.seal.Release) |
| 1195 | r.landFail("release", "the channel points at another build than the sealed one") |
| 1196 | return |
| 1197 | } |
| 1198 | if err := r.liveNotSealed(); err != nil { |
| 1199 | r.landFailErr("release", "the target box still holds an older build of this instance. see the offline move runbook", err) |
| 1200 | return |
| 1201 | } |
| 1202 | r.ok("release") |
| 1203 | } |
| 1204 | |
| 1205 | // liveNotSealed is the release item's second half: nil when the target has |
| 1206 | // no live link for the host, or when the link names the sealed build. The |
| 1207 | // link (<opt>/live/<host>) is the one file ExecBackend.ensureLiveLink and |
| 1208 | // the activator exec. A box the instance once left keeps it, root-owned and |
| 1209 | // still naming the build it ran then (retirePaths leaves it); a wake before |
| 1210 | // ordinary adoption execs whatever it names, since ensureLiveLink accepts |
| 1211 | // any executable and default update settings gate no version. So on a |
| 1212 | // return move after an upgrade the old build could open the sealed |
| 1213 | // database. Any other link fails the item, and open, which verifies every |
| 1214 | // item again, never releases the hold over it. No link is fine: nothing can |
| 1215 | // run until adoption stages the sealed build. The link is only read here, |
| 1216 | // never changed. The error is the box's log line. |
| 1217 | // |
| 1218 | // The check reads the link itself and nothing beyond it (the census ruling |
| 1219 | // on this unit): Lstat, then Readlink, then a string comparison of the |
| 1220 | // target with the sealed build's path, exactly as PointLive writes it. It |
| 1221 | // never follows the link to stat or open what it names, so a target that |
| 1222 | // reaches the sealed build only through another link or a relative |
| 1223 | // spelling fails, and so does a live path that is not a symlink. |
| 1224 | func (r *landRun) liveNotSealed() error { |
| 1225 | optDir := r.cfg().OptDir |
| 1226 | if optDir == "" { |
| 1227 | optDir = release.OptDir() |
| 1228 | } |
| 1229 | link := release.LiveLink(optDir, r.host) |
| 1230 | fi, err := os.Lstat(link) |
| 1231 | if errors.Is(err, os.ErrNotExist) { |
| 1232 | return nil |
| 1233 | } else if err != nil { |
| 1234 | return err |
| 1235 | } |
| 1236 | if fi.Mode()&os.ModeSymlink == 0 { |
| 1237 | return fmt.Errorf("%s is not a symlink (%s)", link, fi.Mode().Type()) |
| 1238 | } |
| 1239 | target, err := os.Readlink(link) |
| 1240 | if err != nil { |
| 1241 | return err |
| 1242 | } |
| 1243 | want := release.VersionPath(optDir, release.RouteAccount(*r.rt), r.seal.Release, r.rt.App) |
| 1244 | if target != want { |
| 1245 | return fmt.Errorf("%s names %s, the seal recorded %q at %s", link, target, r.seal.Release, want) |
| 1246 | } |
| 1247 | return nil |
| 1248 | } |
| 1249 | |
amadan