| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "encoding/hex" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "io" |
| 11 | "log" |
| 12 | "os" |
| 13 | "path/filepath" |
| 14 | "sort" |
| 15 | "strings" |
| 16 | "sync" |
| 17 | "time" |
| 18 | |
| 19 | "github.com/superfly/ltx" |
| 20 | |
| 21 | "github.com/carlosframework/platform/internal/activator" |
| 22 | "github.com/carlosframework/platform/internal/alerting" |
| 23 | "github.com/carlosframework/platform/internal/appconfig" |
| 24 | "github.com/carlosframework/platform/internal/instancekey" |
| 25 | "github.com/carlosframework/platform/internal/layout" |
| 26 | "github.com/carlosframework/platform/internal/lease" |
| 27 | "github.com/carlosframework/platform/internal/moves" |
| 28 | "github.com/carlosframework/platform/internal/moves/archive" |
| 29 | "github.com/carlosframework/platform/internal/park" |
| 30 | "github.com/carlosframework/platform/internal/provision" |
| 31 | "github.com/carlosframework/platform/internal/registry" |
| 32 | "github.com/carlosframework/platform/internal/release" |
| 33 | ) |
| 34 | |
| 35 | // The sealer of an offline operator move (docs/superpowers/plans/2026-10-20- |
| 36 | // offline-operator-move.md, Task 4): the one source box that turns "every |
| 37 | // source box holds, fences and reports" into a stop at the data layer, and |
| 38 | // records what the target must land. |
| 39 | // |
| 40 | // The tombstone is the linearization point of the whole move. Every wake on |
| 41 | // every box of the source region takes the host's single-writer lease |
| 42 | // (instances/<host>/lock.json in the source's own bucket) before it can |
| 43 | // write anything, and from the tombstone on that lease is held by |
| 44 | // moved:<id>. The order is the plan's (Review Focus 2): |
| 45 | // |
| 46 | // 1. this box's own report for the pass says held, fenced, quiesced, no |
| 47 | // process and no layout journal, and no source box keeps a sidecar; |
| 48 | // 2. the local database is checkpointed unless it is behind the replica |
| 49 | // (cfg.Checkpoint takes the lease, syncs and releases it); |
| 50 | // 3. lease.Tombstone, which never replaces a live holder; |
| 51 | // 4. only then the final position is read from the replica, and every |
| 52 | // source box's local position must be at or below it; |
| 53 | // 5. CarryRequest, a final MirrorPrefix round with nothing deferred, and |
| 54 | // the staged listing; |
| 55 | // 6. to 9. the files archive, the timers, the env and its pin, the release, |
| 56 | // replication and next due; |
| 57 | // 10. seal.json, written once and never rewritten (putOfflineCreateOnly), |
| 58 | // then the record's CAS to offline-sealed. |
| 59 | // |
| 60 | // Every step is an ensure. Until seal.json exists the whole seal is taken |
| 61 | // again on each pass (a tombstone that is already this move's is a no-op, |
| 62 | // the mirror resumes from its receipts, the archive is rebuilt and its |
| 63 | // objects overwritten), so a crash anywhere before it converges. Once |
| 64 | // seal.json exists the sealed position is fixed, and from then on the |
| 65 | // tombstone is never taken again: each pass reads lease.LockState first, |
| 66 | // and a tombstone that is absent, expired or anyone else's is a lost stop, |
| 67 | // as is a source report written since the seal at a position past the |
| 68 | // sealed one. The pass reports failed and pages: before open the operator |
| 69 | // aborts, from open on the runbook fixes it forward. Only a tombstone held |
| 70 | // by this move is re-asserted (lease.Reassert, If-Match on the ETag it |
| 71 | // reads, and lease.ErrTombstoneLost, never a new tombstone, when it went |
| 72 | // in between). After it is lifted (finish, abort) it is never re-asserted. |
| 73 | |
| 74 | // The objects the sealer writes beside the sealed archive, under |
| 75 | // moves.OfflineArchivePrefix(host, id) in the target's data bucket. The |
| 76 | // landing (Task 5) reads them there. |
| 77 | const ( |
| 78 | // offlineArchiveSealed is the sealed files archive (archive.Seal of |
| 79 | // archive.Build's output). Its sha256 is the seal's FilesSealed. |
| 80 | offlineArchiveSealed = "files.sealed" |
| 81 | // offlineArchiveManifest is its archive.Manifest (JSON), with the |
| 82 | // sealed data key and every entry. |
| 83 | offlineArchiveManifest = "files.json" |
| 84 | // offlineTimersObject is the moves.TimersSnapshot of the row's active |
| 85 | // timers. moves.TimersDigest of its rows is the seal's TimersDigest. |
| 86 | offlineTimersObject = "timers.json" |
| 87 | // offlineEnvReportObject is the source's moves.EnvReport (names and |
| 88 | // HMACs, never a value), so the landing can name the variables that |
| 89 | // differ (moves.CompareEnv) when the env pin does not match. Its |
| 90 | // Digest is the seal's EnvDigest. |
| 91 | offlineEnvReportObject = "env-report.json" |
| 92 | ) |
| 93 | |
| 94 | // offlineSealFresh is how old this box's own source report may be when the |
| 95 | // sealer reads it: the source step writes it earlier in the same pass, so |
| 96 | // anything older means that step failed this pass. |
| 97 | const offlineSealFresh = 2 * time.Minute |
| 98 | |
| 99 | // offlinePageLostStop is the page rule of a lost stop: one page per move. |
| 100 | const offlinePageLostStop = "offline-lost-stop" |
| 101 | |
| 102 | // offlineSealFaults is the sealer's test seam: a worker's entry is called |
| 103 | // after each step with the step's name, and an error returned is a crash |
| 104 | // there. Keyed by the worker, so tests that run side by side never share |
| 105 | // one. Empty in production. |
| 106 | var offlineSealFaults sync.Map // *movesWorker → func(point string) error |
| 107 | |
| 108 | func (w *movesWorker) sealFault(point string) error { |
| 109 | if f, ok := offlineSealFaults.Load(w); ok { |
| 110 | return f.(func(string) error)(point) |
| 111 | } |
| 112 | return nil |
| 113 | } |
| 114 | |
| 115 | // offlineSealerStep is the sealer's part of an offline move. offlineRound |
| 116 | // calls it on the box whose label is rec.Move.Sealer, after this box's own |
| 117 | // source step has written its report for the pass. |
| 118 | func (w *movesWorker) offlineSealerStep(ctx context.Context, s release.Store, rec provision.Record) error { |
| 119 | m := rec.Move |
| 120 | if !m.Offline() || w.cfg.Box == "" || m.Sealer != w.cfg.Box { |
| 121 | return nil |
| 122 | } |
| 123 | switch m.State { |
| 124 | case provision.OfflineSealing: |
| 125 | return w.offlineSeal(ctx, s, rec) |
| 126 | case provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded, provision.OfflineOpen: |
| 127 | raw, err := w.offlineSealRecord(ctx, s, rec) |
| 128 | if err != nil { |
| 129 | return err |
| 130 | } |
| 131 | return w.offlineKeepStop(ctx, s, rec, raw) |
| 132 | case provision.OfflineFinishing: |
| 133 | return w.offlineFinishLift(ctx, s, rec) |
| 134 | case provision.OfflineReverting: |
| 135 | return w.offlineSealRevert(ctx, s, rec) |
| 136 | } |
| 137 | return nil |
| 138 | } |
| 139 | |
| 140 | // tombstoneAt is where the host's single-writer lease lives in this box's |
| 141 | // own bucket: the activator's lease location, as ExecBackend.Acquire takes |
| 142 | // it for every wake. |
| 143 | func (c moveCfg) tombstoneAt(host string) (bucket, location, region string, err error) { |
| 144 | if c.OwnBucket == "" { |
| 145 | return "", "", "", errors.New("offline move: this box has no hibernate bucket") |
| 146 | } |
| 147 | return c.OwnBucket, activator.HibernatePrefix(c.OwnPrefix, host), c.OwnBucketRegion, nil |
| 148 | } |
| 149 | |
| 150 | // tombState is the lease as lease.LockState read it: an absent lock has no |
| 151 | // owner and a zero expiry. |
| 152 | type tombState struct { |
| 153 | owner string |
| 154 | expires time.Time |
| 155 | } |
| 156 | |
| 157 | // mine reports a tombstone this move holds now. "Expired" is the lease's |
| 158 | // own reading (Acquire takes a lease the moment its expiry has passed), so |
| 159 | // it is judged against the wall clock the lease is written by. |
| 160 | func (t tombState) mine(id string) bool { |
| 161 | return t.owner == moves.TombstoneOwner(id) && time.Now().Before(t.expires) |
| 162 | } |
| 163 | |
| 164 | // free reports a lease nothing holds: absent, or past its expiry, whoever |
| 165 | // wrote it (a lifted tombstone included: lease.IsLiftedTombstone). |
| 166 | func (t tombState) free() bool { |
| 167 | return t.owner == "" || !time.Now().Before(t.expires) |
| 168 | } |
| 169 | |
| 170 | func (t tombState) String() string { return offlineLeaseText(t) } |
| 171 | |
| 172 | // offlineLeaseText is the lease as the lost stop's page and the sealer's |
| 173 | // report name it (copy rows MV-O72 to MV-O75). |
| 174 | func offlineLeaseText(t tombState) string { |
| 175 | switch { |
| 176 | case t.owner == "": |
| 177 | return "absent" |
| 178 | case lease.IsLiftedTombstone(t.owner, t.expires) && t.expires.Unix() == 0: |
| 179 | // LiftTombstone's expiry: the epoch. |
| 180 | return fmt.Sprintf("lifted (%s)", t.owner) |
| 181 | case t.free(): |
| 182 | return fmt.Sprintf("expired (%s at %s)", t.owner, t.expires.UTC().Format(time.RFC3339)) |
| 183 | } |
| 184 | return fmt.Sprintf("held by %s until %s", t.owner, t.expires.UTC().Format(time.RFC3339)) |
| 185 | } |
| 186 | |
| 187 | // offlineLock reads the host's lease. |
| 188 | func (w *movesWorker) offlineLock(ctx context.Context, host string) (tombState, error) { |
| 189 | bucket, loc, region, err := w.cfg.tombstoneAt(host) |
| 190 | if err != nil { |
| 191 | return tombState{}, err |
| 192 | } |
| 193 | _, owner, expires, err := lease.LockState(ctx, bucket, loc, region) |
| 194 | if err != nil { |
| 195 | return tombState{}, fmt.Errorf("read the lease of %s: %w", host, err) |
| 196 | } |
| 197 | return tombState{owner: owner, expires: expires}, nil |
| 198 | } |
| 199 | |
| 200 | // offlineTarget is the target's coordinates as its edges file publishes |
| 201 | // them: the box at move.to (the lexically first, as every move reads the |
| 202 | // fleet), matched exactly and then by placement under the home binding. |
| 203 | func (w *movesWorker) offlineTarget(ctx context.Context, s release.Store, m *provision.MoveState) (edgesFile, error) { |
| 204 | fleet, err := loadMoveFleet(ctx, s) |
| 205 | if err != nil { |
| 206 | return edgesFile{}, fmt.Errorf("read the fleet's edges files: %w", err) |
| 207 | } |
| 208 | if t, ok := fleet.targets[m.To]; ok { |
| 209 | return t, nil |
| 210 | } |
| 211 | codes := make([]string, 0, len(fleet.targets)) |
| 212 | for code := range fleet.targets { |
| 213 | codes = append(codes, code) |
| 214 | } |
| 215 | sort.Strings(codes) |
| 216 | for _, code := range codes { |
| 217 | if park.PlacementMatches(w.cfg.ParkHomeRegion, m.To, park.PlacementRegion(code)) { |
| 218 | return fleet.targets[code], nil |
| 219 | } |
| 220 | } |
| 221 | return edgesFile{}, fmt.Errorf("the target region %q publishes no data bucket yet", m.To) |
| 222 | } |
| 223 | |
| 224 | // offlineMirror is the mirror pair between this box's bucket and region. |
| 225 | func (w *movesWorker) offlineMirror(ctx context.Context, region string) (lease.MirrorStore, error) { |
| 226 | if w.cfg.Mirror == nil { |
| 227 | return nil, errors.New("offline move: no S3 mirror on this box") |
| 228 | } |
| 229 | return w.mirror(ctx, region) |
| 230 | } |
| 231 | |
| 232 | // replicaPosition is the newest TXID of the replica listed in objs under |
| 233 | // prefix (newestLTX), 0 when it holds no LTX file. |
| 234 | func replicaPosition(objs map[string]lease.MirrorObject, prefix string) uint64 { |
| 235 | newest := newestLTX(objs, prefix, nil) |
| 236 | if newest == "" { |
| 237 | return 0 |
| 238 | } |
| 239 | _, maxTX, err := ltx.ParseFilename(filepath.Base(newest)) |
| 240 | if err != nil { |
| 241 | return 0 |
| 242 | } |
| 243 | return uint64(maxTX) |
| 244 | } |
| 245 | |
| 246 | // sealRefusal is a seal item that failed for a reason a later pass will |
| 247 | // meet again: the pass reports it (phase failed) and the operator aborts. |
| 248 | // Reason is the item's reason as `carlos move offline status` shows it. |
| 249 | // Each refusal is written as a literal where it is decided, so the copy |
| 250 | // gate reads every sentence a Reason can carry (copy rows MV-O76 to |
| 251 | // MV-O84); the two that carry an error's own text (the archive's refusal |
| 252 | // and the secrets key's) are box diagnostics (movesDiagnosticWrites). |
| 253 | type sealRefusal struct { |
| 254 | item, Reason string |
| 255 | } |
| 256 | |
| 257 | func (r *sealRefusal) Error() string { return r.item + ": " + r.Reason } |
| 258 | |
| 259 | // offlineSeal is the sealer's step in offline-sealing. |
| 260 | func (w *movesWorker) offlineSeal(ctx context.Context, s release.Store, rec provision.Record) error { |
| 261 | m, host := rec.Move, rec.Host |
| 262 | sealKey := moves.OfflineSealKey(host, m.ID) |
| 263 | if sealKey == "" { |
| 264 | return fmt.Errorf("offline move %s: %s has no seal key", m.ID, host) |
| 265 | } |
| 266 | raw, _, err := s.Get(ctx, sealKey) |
| 267 | switch { |
| 268 | case err == nil: |
| 269 | // Sealed on an earlier pass, which crashed before the CAS: the |
| 270 | // position is fixed, so the stop is only kept, never taken again. |
| 271 | return w.offlineKeepStop(ctx, s, rec, raw) |
| 272 | case !errors.Is(err, os.ErrNotExist): |
| 273 | return fmt.Errorf("read %s: %w", sealKey, err) |
| 274 | } |
| 275 | items := map[string]moves.Check{} |
| 276 | seal, err := w.offlineTakeSeal(ctx, s, rec, items) |
| 277 | var ref *sealRefusal |
| 278 | if errors.As(err, &ref) { |
| 279 | items[ref.item] = moves.Check{OK: false, Reason: ref.Reason} |
| 280 | r := moves.OfflineReport{Op: m.ID, Host: host, Box: w.cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseFailed, |
| 281 | At: w.cfg.now().UTC(), Items: items} |
| 282 | return errors.Join(fmt.Errorf("offline move %s: %s cannot be sealed: %w", m.ID, host, err), writeOfflineReport(ctx, s, r)) |
| 283 | } |
| 284 | if err != nil { |
| 285 | return fmt.Errorf("offline move %s: sealing %s: %w", m.ID, host, err) |
| 286 | } |
| 287 | body, err := json.Marshal(seal) |
| 288 | if err != nil { |
| 289 | return err |
| 290 | } |
| 291 | stored, created, err := putOfflineCreateOnly(ctx, s, sealKey, body) |
| 292 | if err != nil { |
| 293 | return err |
| 294 | } |
| 295 | if created { |
| 296 | log.Printf("moves: %s: offline move %s: sealed at TXID %d (%d staged objects, files %s)", host, m.ID, seal.FinalTXID, len(seal.Staged), seal.FilesPlain) |
| 297 | } |
| 298 | if err := w.sealFault("seal"); err != nil { |
| 299 | return err |
| 300 | } |
| 301 | return w.offlineKeepStop(ctx, s, rec, stored) |
| 302 | } |
| 303 | |
| 304 | // offlineTakeSeal runs steps 1 to 9 and returns the seal to write. items |
| 305 | // collects each item's check as it is decided; a *sealRefusal is the item |
| 306 | // that failed. |
| 307 | func (w *movesWorker) offlineTakeSeal(ctx context.Context, s release.Store, rec provision.Record, items map[string]moves.Check) (moves.OfflineReport, error) { |
| 308 | cfg := w.cfg |
| 309 | m, host := rec.Move, rec.Host |
| 310 | var seal moves.OfflineReport |
| 311 | |
| 312 | // 1. The stop, as the source reports say it now: this box held, fenced, |
| 313 | // quiesced, with no process and no layout journal; every box the seal |
| 314 | // named held; no sidecar anywhere. |
| 315 | if len(m.Boxes) == 0 { |
| 316 | return seal, errors.New("the record names no source boxes") |
| 317 | } |
| 318 | reports, err := readOfflineReports(ctx, s, host, m.ID, moves.OfflineRoleSource) |
| 319 | if err != nil { |
| 320 | return seal, fmt.Errorf("read the source reports: %w", err) |
| 321 | } |
| 322 | now := cfg.now().UTC() |
| 323 | own, ok := reports[cfg.Box] |
| 324 | if !ok || own.Phase != moves.OfflinePhaseStopped || !own.Held || !own.Fenced || !own.Quiesced || own.Procs != 0 || own.LayoutJournal || now.Sub(own.At) > offlineSealFresh { |
| 325 | return seal, fmt.Errorf("this box has not stopped %s this pass: report found %v, held %v, fenced %v, quiesced %v, processes %d, layout journal %v, at %s", |
| 326 | host, ok, own.Held, own.Fenced, own.Quiesced, own.Procs, own.LayoutJournal, own.At.Format(time.RFC3339)) |
| 327 | } |
| 328 | boxes := append([]string(nil), m.Boxes...) |
| 329 | sort.Strings(boxes) |
| 330 | for _, b := range boxes { |
| 331 | r, ok := reports[b] |
| 332 | if !ok || !r.Held { |
| 333 | return seal, fmt.Errorf("waiting for %s to report %s held", b, host) |
| 334 | } |
| 335 | } |
| 336 | for _, b := range boxes { |
| 337 | if reports[b].Sidecar { |
| 338 | return seal, &sealRefusal{item: "sidecar", Reason: fmt.Sprintf("%s keeps a sidecar file for %s", b, host)} |
| 339 | } |
| 340 | } |
| 341 | items["sidecar"] = moves.Check{OK: true, Reason: "no source box keeps a sidecar file"} |
| 342 | |
| 343 | rt, err := w.offlineRow(host) |
| 344 | if err != nil { |
| 345 | return seal, err |
| 346 | } |
| 347 | if rt == nil { |
| 348 | return seal, fmt.Errorf("this box has no row for %s", host) |
| 349 | } |
| 350 | tgt, err := w.offlineTarget(ctx, s, m) |
| 351 | if err != nil { |
| 352 | return seal, err |
| 353 | } |
| 354 | // A staged move hands over in the destination its stage fixed (Astra |
| 355 | // round 2, finding 2). A target that publishes anything else now is |
| 356 | // not sealed into; the seal door refuses it first (MV-O192). |
| 357 | if m.StagedAt != nil && (tgt.DataBucket != m.StageBucket || tgt.DataBucketRegion != m.StageRegion) { |
| 358 | return seal, fmt.Errorf("the target now publishes %s (%s), not the staging bucket %s (%s) the stage fixed", tgt.DataBucket, tgt.DataBucketRegion, m.StageBucket, m.StageRegion) |
| 359 | } |
| 360 | mirror, err := w.offlineMirror(ctx, tgt.DataBucketRegion) |
| 361 | if err != nil { |
| 362 | return seal, err |
| 363 | } |
| 364 | bucket, loc, region, err := cfg.tombstoneAt(host) |
| 365 | if err != nil { |
| 366 | return seal, err |
| 367 | } |
| 368 | srcPrefix := loc + "/" |
| 369 | dstPrefix := activator.MovesPrefix(host) |
| 370 | dbPath := rt.DBPath |
| 371 | if dbPath == "" { |
| 372 | dbPath = cfg.flatDBPath(host, rt) |
| 373 | } |
| 374 | |
| 375 | // 2. The checkpoint, while no tombstone is down: the local database is |
| 376 | // synced under the lease, which the checkpoint releases again. That |
| 377 | // carries a tail the stop failed to checkpoint, whether litestream had |
| 378 | // already turned it into a local LTX file (the local position is past |
| 379 | // the replica's) or it is still in the WAL (the positions are equal), |
| 380 | // as the legacy cut re-checkpoints unconditionally. A local copy behind |
| 381 | // the replica is a stale warm cache: the replica is the instance, and |
| 382 | // syncing the stale copy into it is the one thing never done. A |
| 383 | // tombstone already this move's means an earlier pass checkpointed and |
| 384 | // took it, and nothing has woken since; the checkpoint could not take |
| 385 | // the lease anyway. |
| 386 | local, err := w.localState(host, rt) |
| 387 | if err != nil { |
| 388 | return seal, err |
| 389 | } |
| 390 | objs, err := mirror.List(ctx, bucket, srcPrefix) |
| 391 | if err != nil { |
| 392 | return seal, fmt.Errorf("list the replica: %w", err) |
| 393 | } |
| 394 | before, err := w.offlineLock(ctx, host) |
| 395 | if err != nil { |
| 396 | return seal, err |
| 397 | } |
| 398 | replica := replicaPosition(objs, srcPrefix) |
| 399 | switch { |
| 400 | case before.mine(m.ID): |
| 401 | case local.LocalTXID < replica: |
| 402 | log.Printf("moves: %s: offline move %s: the local copy (position %d) is behind the replica (%d); the replica is sealed as it is", host, m.ID, local.LocalTXID, replica) |
| 403 | case lexists(dbPath): |
| 404 | if cfg.Checkpoint == nil { |
| 405 | return seal, errors.New("no checkpoint on this box") |
| 406 | } |
| 407 | if err := cfg.Checkpoint(ctx, dbPath); err != nil { |
| 408 | return seal, fmt.Errorf("final checkpoint of %s: %w", dbPath, err) |
| 409 | } |
| 410 | log.Printf("moves: %s: offline move %s: checkpointed the local database (position %d, replica %d)", host, m.ID, local.LocalTXID, replica) |
| 411 | case local.LocalTXID > replica: |
| 412 | return seal, &sealRefusal{item: "db-tail", Reason: fmt.Sprintf("this box's local position %d is past the replica's %d and there is no database at %s to checkpoint", local.LocalTXID, replica, dbPath)} |
| 413 | } |
| 414 | if err := w.sealFault("checkpoint"); err != nil { |
| 415 | return seal, err |
| 416 | } |
| 417 | |
| 418 | // 3. The tombstone: from here no wake on any box can take the lease. |
| 419 | owner := moves.TombstoneOwner(m.ID) |
| 420 | if err := lease.Tombstone(ctx, bucket, loc, region, owner, moves.TombstoneTTL); err != nil { |
| 421 | return seal, fmt.Errorf("tombstone %s/%s/lock.json: %w", bucket, loc, err) |
| 422 | } |
| 423 | if err := w.sealFault("tombstone"); err != nil { |
| 424 | return seal, err |
| 425 | } |
| 426 | |
| 427 | // 4. The final position, read only now, and no source box past it. |
| 428 | objs, err = mirror.List(ctx, bucket, srcPrefix) |
| 429 | if err != nil { |
| 430 | return seal, fmt.Errorf("list the replica: %w", err) |
| 431 | } |
| 432 | final := replicaPosition(objs, srcPrefix) |
| 433 | if final == 0 { |
| 434 | return seal, &sealRefusal{item: "db", Reason: fmt.Sprintf("the replica at %s/%s holds no database", bucket, srcPrefix)} |
| 435 | } |
| 436 | mine, err := w.localState(host, rt) // after the checkpoint, not as reported before it |
| 437 | if err != nil { |
| 438 | return seal, err |
| 439 | } |
| 440 | for _, b := range boxes { |
| 441 | pos := reports[b].LocalTXID |
| 442 | if b == cfg.Box { |
| 443 | pos = mine.LocalTXID |
| 444 | } |
| 445 | if pos > final { |
| 446 | return seal, &sealRefusal{item: "db-tail", Reason: fmt.Sprintf("%s's local position %d is past the replica's final position %d", b, pos, final)} |
| 447 | } |
| 448 | } |
| 449 | items["db-tail"] = moves.Check{OK: true, Reason: fmt.Sprintf("every source box is at or below TXID %d", final)} |
| 450 | items["lease"] = moves.Check{OK: true, Reason: fmt.Sprintf("%s holds %s/lock.json", owner, loc)} |
| 451 | if err := w.sealFault("final"); err != nil { |
| 452 | return seal, err |
| 453 | } |
| 454 | |
| 455 | // 5. The request record into the build state, then the byte-final |
| 456 | // round into the target's staging lane and its listing. |
| 457 | if cfg.CarryRequest != nil { |
| 458 | if err := cfg.CarryRequest(ctx, host, bucket, loc); err != nil { |
| 459 | return seal, fmt.Errorf("carry the request record into the build state: %w", err) |
| 460 | } |
| 461 | items["build-state"] = moves.Check{OK: true, Reason: "the request record is in the build state the replica prefix carries"} |
| 462 | } else { |
| 463 | items["build-state"] = moves.Check{OK: true, Reason: "this box keeps no request records; the build state travels as the replica prefix holds it"} |
| 464 | } |
| 465 | receipts := lease.Receipts{Bucket: bucket, Key: activator.MoveReceiptsKey(host)} |
| 466 | round, err := lease.MirrorPrefix(ctx, mirror, bucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts) |
| 467 | if err != nil { |
| 468 | return seal, err |
| 469 | } |
| 470 | if round.Deferred > 0 { |
| 471 | return seal, fmt.Errorf("final mirror round: %d object(s) changed at the source during the copy", round.Deferred) |
| 472 | } |
| 473 | source, err := mirror.List(ctx, bucket, srcPrefix) |
| 474 | if err != nil { |
| 475 | return seal, fmt.Errorf("list the replica: %w", err) |
| 476 | } |
| 477 | listed, err := mirror.List(ctx, tgt.DataBucket, dstPrefix) |
| 478 | if err != nil { |
| 479 | return seal, fmt.Errorf("list the staged copy: %w", err) |
| 480 | } |
| 481 | staged := map[string]moves.StagedObject{} |
| 482 | for k, o := range listed { |
| 483 | staged[strings.TrimPrefix(k, dstPrefix)] = moves.StagedObject{Size: o.Size, ETag: o.ETag} |
| 484 | } |
| 485 | if err := sameListing(source, srcPrefix, staged); err != nil { |
| 486 | return seal, fmt.Errorf("the staged copy is not the replica: %w", err) |
| 487 | } |
| 488 | if replicaPosition(listed, dstPrefix) != final { |
| 489 | return seal, fmt.Errorf("the staged copy's newest TXID is %d, the replica's %d", replicaPosition(listed, dstPrefix), final) |
| 490 | } |
| 491 | items["db"] = moves.Check{OK: true, Reason: fmt.Sprintf("%d objects staged at %s/%s, newest TXID %d", len(staged), tgt.DataBucket, dstPrefix, final)} |
| 492 | if err := w.sealFault("mirror"); err != nil { |
| 493 | return seal, err |
| 494 | } |
| 495 | |
| 496 | // 6. The files and <host>.env, sealed, beside the staging lane. |
| 497 | prefix := moves.OfflineArchivePrefix(host, m.ID) |
| 498 | files, envPath, configPath, stray, err := w.offlineFilesOf(host, rt) |
| 499 | if err != nil { |
| 500 | return seal, fmt.Errorf("read what %s keeps: %w", host, err) |
| 501 | } |
| 502 | if len(stray) > 0 { |
| 503 | sort.Strings(stray) |
| 504 | // What the app keeps there is the app's data, in a place it reads |
| 505 | // from: moving it into the files directory changes that place, and |
| 506 | // removing it loses it. So the refusal offers neither (MV-O76). |
| 507 | return seal, &sealRefusal{item: "files", Reason: fmt.Sprintf("%s keeps %s beside its files directory, and this move can carry only the files directory. see the offline move runbook", host, strings.Join(stray, ", "))} |
| 508 | } |
| 509 | a, err := w.offlineArchive(ctx, rec, mirror, tgt.DataBucket, prefix, files, envPath) |
| 510 | if err != nil { |
| 511 | return seal, err |
| 512 | } |
| 513 | items["files"] = moves.Check{OK: true, Reason: fmt.Sprintf("%d files, %d bytes, sealed at %s%s", a.entries, a.bytes, prefix, offlineArchiveSealed)} |
| 514 | if err := w.sealFault("archive"); err != nil { |
| 515 | return seal, err |
| 516 | } |
| 517 | |
| 518 | // 7. The timers. |
| 519 | if cfg.Registry == nil { |
| 520 | return seal, errors.New("no registry on this box") |
| 521 | } |
| 522 | rows, err := cfg.Registry.TimersFor(host) |
| 523 | if err != nil { |
| 524 | return seal, fmt.Errorf("read the timers: %w", err) |
| 525 | } |
| 526 | active := moves.ActiveTimers(rows) |
| 527 | timersBody, err := moves.TimersSnapshot{V: 1, Host: host, Rows: active, TakenAt: now}.Encode() |
| 528 | if err != nil { |
| 529 | return seal, err |
| 530 | } |
| 531 | if err := mirror.Put(ctx, tgt.DataBucket, prefix+offlineTimersObject, timersBody); err != nil { |
| 532 | return seal, fmt.Errorf("put the timers: %w", err) |
| 533 | } |
| 534 | timersDigest := moves.TimersDigest(rows) |
| 535 | items["timers"] = moves.Check{OK: true, Reason: fmt.Sprintf("%d active timers at %s%s", len(active), prefix, offlineTimersObject)} |
| 536 | if err := w.sealFault("timers"); err != nil { |
| 537 | return seal, err |
| 538 | } |
| 539 | |
| 540 | // 8. The env file, its effective instance key, and the env pin. |
| 541 | envSHA, keySHA, err := offlineEnvDigests(envPath, configPath, a.envSHA) |
| 542 | if err != nil { |
| 543 | return seal, err |
| 544 | } |
| 545 | // The digest itself is the seal's EnvSHA256, not part of the reason. |
| 546 | items["env"] = moves.Check{OK: true, Reason: "absent on the source, so none is carried"} |
| 547 | if envSHA != "" { |
| 548 | items["env"] = moves.Check{OK: true, Reason: "carried in the archive"} |
| 549 | } |
| 550 | if cfg.LaunchLayers == nil { |
| 551 | return seal, &sealRefusal{item: "env-pin", Reason: "this box has no exec backend to compose the launch environment"} |
| 552 | } |
| 553 | if cfg.SecretsKey == "" { |
| 554 | return seal, &sealRefusal{item: "env-pin", Reason: "this box holds no CARLOS_SECRETS_KEY to key the env report"} |
| 555 | } |
| 556 | env, instance, _, err := cfg.LaunchLayers(*rt) |
| 557 | if err != nil { |
| 558 | return seal, fmt.Errorf("compose the launch environment: %w", err) |
| 559 | } |
| 560 | mac, err := moves.ReportKeyFromSecrets(cfg.SecretsKey) |
| 561 | if err != nil { |
| 562 | return seal, &sealRefusal{item: "env-pin", Reason: err.Error()} |
| 563 | } |
| 564 | envReport := moves.ReportLaunch(mac, env, instance) |
| 565 | envReportBody, err := json.Marshal(envReport) |
| 566 | if err != nil { |
| 567 | return seal, err |
| 568 | } |
| 569 | if err := mirror.Put(ctx, tgt.DataBucket, prefix+offlineEnvReportObject, envReportBody); err != nil { |
| 570 | return seal, fmt.Errorf("put the env report: %w", err) |
| 571 | } |
| 572 | items["env-pin"] = moves.Check{OK: true, Reason: "launch environment digest " + envReport.Digest} |
| 573 | if err := w.sealFault("env"); err != nil { |
| 574 | return seal, err |
| 575 | } |
| 576 | |
| 577 | // 9. The build, the replication rung and the next due time. |
| 578 | if cfg.LiveVersion == nil { |
| 579 | return seal, errors.New("no version store on this box") |
| 580 | } |
| 581 | version, err := cfg.LiveVersion(host) |
| 582 | if err != nil { |
| 583 | return seal, fmt.Errorf("read the live build: %w", err) |
| 584 | } |
| 585 | if version == "" { |
| 586 | return seal, &sealRefusal{item: "release", Reason: fmt.Sprintf("no build is live for %s on this box", host)} |
| 587 | } |
| 588 | items["release"] = moves.Check{OK: true, Reason: "live build " + version} |
| 589 | items["replication"] = moves.Check{OK: true, Reason: fmt.Sprintf("replication %q", rt.Replication)} |
| 590 | items["next-due"] = moves.Check{OK: true, Reason: fmt.Sprintf("next due %d", rt.NextDue)} |
| 591 | |
| 592 | // The tombstone must still be this move's when the seal is written: |
| 593 | // the position it records was read under it. One lost before the seal |
| 594 | // is taken again on the next pass, which seals afresh. |
| 595 | st, err := w.offlineLock(ctx, host) |
| 596 | if err != nil { |
| 597 | return seal, err |
| 598 | } |
| 599 | if !st.mine(m.ID) { |
| 600 | return seal, fmt.Errorf("the lease is %s, not this move's tombstone; the next pass takes it again", st) |
| 601 | } |
| 602 | seal = moves.OfflineReport{ |
| 603 | V: moves.OfflineReportVersion, Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseSealed, |
| 604 | At: now, Items: items, TombstoneOwner: owner, |
| 605 | FinalTXID: final, Staged: staged, |
| 606 | FilesPlain: a.plain, FilesSealed: a.sealed, FilesBytes: a.bytes, FilesEntries: a.entries, |
| 607 | EnvSHA256: envSHA, KeySHA256: keySHA, EnvDigest: envReport.Digest, |
| 608 | Release: version, TimersDigest: timersDigest, Replication: rt.Replication, NextDue: rt.NextDue, |
| 609 | } |
| 610 | if err := moves.ValidateOfflineSeal(seal, host, m.ID); err != nil { |
| 611 | return seal, err |
| 612 | } |
| 613 | return seal, nil |
| 614 | } |
| 615 | |
| 616 | // sameListing reports whether staged (keyed relative to the staging |
| 617 | // prefix) holds exactly the source's objects under srcPrefix, lock.json |
| 618 | // aside (the mirror never copies it), size for size. ETags may differ: a |
| 619 | // multipart source's copy gets a new one (lease.Receipts). |
| 620 | func sameListing(source map[string]lease.MirrorObject, srcPrefix string, staged map[string]moves.StagedObject) error { |
| 621 | want := map[string]int64{} |
| 622 | for k, o := range source { |
| 623 | rel := strings.TrimPrefix(k, srcPrefix) |
| 624 | if rel == "lock.json" { |
| 625 | continue |
| 626 | } |
| 627 | want[rel] = o.Size |
| 628 | } |
| 629 | if len(want) != len(staged) { |
| 630 | return fmt.Errorf("%d objects at the source, %d staged", len(want), len(staged)) |
| 631 | } |
| 632 | for rel, size := range want { |
| 633 | got, ok := staged[rel] |
| 634 | if !ok || got.Size != size { |
| 635 | return fmt.Errorf("%s: staged %v, source size %d", rel, got, size) |
| 636 | } |
| 637 | } |
| 638 | return nil |
| 639 | } |
| 640 | |
| 641 | // offlineFilesOf is what the archive carries for the row's layout, and |
| 642 | // what it cannot. |
| 643 | // |
| 644 | // The target lands the instance flat (the row is minted at flat paths), so |
| 645 | // the archive's files root is the app's files directory as the flat layout |
| 646 | // names it: <DataDir>/<host>, with <host>.env beside it. The layout pass |
| 647 | // maps the flat data directory onto the nested instance directory name for |
| 648 | // name (layout/extras.go: an app finds its files beside its database, and |
| 649 | // nested the database is <instance dir>/<host>.db), so a nested row's |
| 650 | // files directory is <instance dir>/<host>, read through the raw |
| 651 | // directory as every agent read of it is, and its env file is |
| 652 | // <instance dir>/<host>.env. Archiving the whole instance directory would |
| 653 | // land the app's files one level down on the target (<DataDir>/<host>/ |
| 654 | // <host>/...), where the app would not find them. |
| 655 | // |
| 656 | // stray is every entry the app keeps beside its files directory, which |
| 657 | // the flat files directory cannot carry: on a nested row every entry of |
| 658 | // the instance directory that is neither the platform's |
| 659 | // (layout.IsNestedPlatformName: the seed marker and the launch descriptors |
| 660 | // included) nor <host>; on a flat row every entry of |
| 661 | // the data directory the layout's naming rules give this host |
| 662 | // (layout.FlatLeftovers; <host>-blobs is the live example) that is |
| 663 | // neither. The seal refuses on any, rather than leave it behind. |
| 664 | // |
| 665 | // A host that never kept a files directory (an app that writes nothing of |
| 666 | // its own) has none at the files path: the archive then carries the env |
| 667 | // file alone, FilesPlain is the digest of that, and the landing extracts |
| 668 | // an empty files directory. |
| 669 | // |
| 670 | // configPath is the rendered configuration beside the env file, which the |
| 671 | // effective instance key may come from. |
| 672 | func (w *movesWorker) offlineFilesOf(host string, rt *registry.Route) (files, envPath, configPath string, stray []string, err error) { |
| 673 | dd := w.cfg.DataDir |
| 674 | platform := func(name string) bool { return layout.IsNestedPlatformName(host, name) } |
| 675 | if rt != nil && provision.NestedPaths(dd, *rt) { |
| 676 | raw := layout.RawInstanceDir(dd, host) |
| 677 | ents, err := os.ReadDir(raw) |
| 678 | if err != nil && !errors.Is(err, os.ErrNotExist) { |
| 679 | return "", "", "", nil, err |
| 680 | } |
| 681 | for _, e := range ents { |
| 682 | if !platform(e.Name()) && e.Name() != host { |
| 683 | stray = append(stray, filepath.Join(raw, e.Name())) |
| 684 | } |
| 685 | } |
| 686 | return filepath.Join(raw, host), filepath.Join(raw, host+".env"), filepath.Join(raw, host+".config.env"), stray, nil |
| 687 | } |
| 688 | var hosts []string |
| 689 | if w.cfg.Registry != nil { |
| 690 | rows, err := w.cfg.Registry.List() |
| 691 | if err != nil { |
| 692 | return "", "", "", nil, err |
| 693 | } |
| 694 | for _, r := range rows { |
| 695 | hosts = append(hosts, r.Host) |
| 696 | } |
| 697 | } |
| 698 | for _, p := range layout.FlatLeftovers(dd, host, hosts) { |
| 699 | if name := filepath.Base(p); !platform(name) && name != host { |
| 700 | stray = append(stray, p) |
| 701 | } |
| 702 | } |
| 703 | return filepath.Join(dd, host), filepath.Join(dd, host+".env"), filepath.Join(dd, host+".config.env"), stray, nil |
| 704 | } |
| 705 | |
| 706 | // sealedArchive is what the seal records of the archive. |
| 707 | type sealedArchive struct { |
| 708 | plain, sealed string |
| 709 | bytes int64 |
| 710 | entries int |
| 711 | // envSHA is the env entry's digest as the archive recorded it |
| 712 | // ("sha256:<hex>"), "" when the archive carries no env. |
| 713 | envSHA string |
| 714 | } |
| 715 | |
| 716 | // offlineArchive builds the archive of files and envPath into a scratch |
| 717 | // directory on the data disk, seals it under the deployment's config key |
| 718 | // bound to the record's account and app, and puts it and its manifest |
| 719 | // under prefix in the target bucket. A file the archive refuses is a |
| 720 | // refusal of the files item, never a partial archive; a file that changed |
| 721 | // while it was read is retried. The files directory holds nothing of the |
| 722 | // platform's but the files marker, which every build leaves out, so no |
| 723 | // filter is needed (archive.Build is BuildFiltered with none). |
| 724 | func (w *movesWorker) offlineArchive(ctx context.Context, rec provision.Record, mirror lease.MirrorStore, bucket, prefix, files, envPath string) (sealedArchive, error) { |
| 725 | cfg := w.cfg |
| 726 | var out sealedArchive |
| 727 | if cfg.ConfigPub == nil { |
| 728 | return out, &sealRefusal{item: "files", Reason: "this box holds no CARLOS_SECRETS_KEY to seal the archive under"} |
| 729 | } |
| 730 | dir := moves.OfflineParkDir(cfg.DataDir, rec.Host, rec.Move.ID, moves.OfflineRoleSealer) |
| 731 | if dir == "" { |
| 732 | return out, fmt.Errorf("no scratch directory for %s under %q", rec.Host, cfg.DataDir) |
| 733 | } |
| 734 | if err := os.MkdirAll(dir, 0o700); err != nil { |
| 735 | return out, err |
| 736 | } |
| 737 | tmp, err := os.MkdirTemp(dir, "archive-") |
| 738 | if err != nil { |
| 739 | return out, err |
| 740 | } |
| 741 | defer os.RemoveAll(tmp) |
| 742 | built, err := os.Create(filepath.Join(tmp, "archive.tar.zst")) |
| 743 | if err != nil { |
| 744 | return out, err |
| 745 | } |
| 746 | defer built.Close() |
| 747 | plain, entries, size, err := archive.Build(files, envPath, built) |
| 748 | switch { |
| 749 | case errors.Is(err, archive.ErrChanged): |
| 750 | return out, fmt.Errorf("archive the files: %w; built again next pass", err) |
| 751 | case err != nil: |
| 752 | return out, &sealRefusal{item: "files", Reason: err.Error()} |
| 753 | } |
| 754 | if _, err := built.Seek(0, io.SeekStart); err != nil { |
| 755 | return out, err |
| 756 | } |
| 757 | sealedFile, err := os.Create(filepath.Join(tmp, "archive.sealed")) |
| 758 | if err != nil { |
| 759 | return out, err |
| 760 | } |
| 761 | defer sealedFile.Close() |
| 762 | dataKey, sealedSHA, err := archive.Seal(built, cfg.ConfigPub, appconfig.Binding{Account: rec.Account, App: rec.App}, sealedFile) |
| 763 | if err != nil { |
| 764 | return out, fmt.Errorf("seal the archive: %w", err) |
| 765 | } |
| 766 | manifest, err := json.Marshal(archive.Manifest{V: archive.ManifestVersion, Op: rec.Move.ID, Plain: plain, Sealed: sealedSHA, Bytes: size, |
| 767 | Entries: entries, SealedDataKey: dataKey, Chunk: archive.ChunkSize}) |
| 768 | if err != nil { |
| 769 | return out, err |
| 770 | } |
| 771 | body, err := os.ReadFile(sealedFile.Name()) |
| 772 | if err != nil { |
| 773 | return out, err |
| 774 | } |
| 775 | if err := mirror.Put(ctx, bucket, prefix+offlineArchiveSealed, body); err != nil { |
| 776 | return out, fmt.Errorf("put the sealed archive: %w", err) |
| 777 | } |
| 778 | if err := mirror.Put(ctx, bucket, prefix+offlineArchiveManifest, manifest); err != nil { |
| 779 | return out, fmt.Errorf("put the archive manifest: %w", err) |
| 780 | } |
| 781 | out = sealedArchive{plain: plain, sealed: sealedSHA, bytes: size, entries: len(entries)} |
| 782 | for _, e := range entries { |
| 783 | if e.Path == archive.EnvEntry { |
| 784 | out.envSHA = e.SHA256 |
| 785 | } |
| 786 | } |
| 787 | return out, nil |
| 788 | } |
| 789 | |
| 790 | // readOwnFile reads a regular file without following a link, as the agent |
| 791 | // reads every file a nested tenant can write: exists false when there is |
| 792 | // none. |
| 793 | func readOwnFile(path string) (body []byte, exists bool, err error) { |
| 794 | info, err := os.Lstat(path) |
| 795 | if errors.Is(err, os.ErrNotExist) { |
| 796 | return nil, false, nil |
| 797 | } |
| 798 | if err != nil { |
| 799 | return nil, false, err |
| 800 | } |
| 801 | if !info.Mode().IsRegular() { |
| 802 | return nil, true, fmt.Errorf("%s is not a regular file", path) |
| 803 | } |
| 804 | f, err := os.Open(path) |
| 805 | if err != nil { |
| 806 | return nil, true, err |
| 807 | } |
| 808 | defer f.Close() |
| 809 | got, err := f.Stat() |
| 810 | if err != nil { |
| 811 | return nil, true, err |
| 812 | } |
| 813 | if !os.SameFile(info, got) { |
| 814 | return nil, true, fmt.Errorf("%s changed while it was read", path) |
| 815 | } |
| 816 | body, err = io.ReadAll(io.LimitReader(f, instancekey.EnvReadLimit+1)) |
| 817 | if err != nil { |
| 818 | return nil, true, err |
| 819 | } |
| 820 | if len(body) > instancekey.EnvReadLimit { |
| 821 | return nil, true, fmt.Errorf("%s is larger than an env file", path) |
| 822 | } |
| 823 | return body, true, nil |
| 824 | } |
| 825 | |
| 826 | // offlineEnvDigests is the env item's two values: the lowercase hex sha256 |
| 827 | // of <host>.env's bytes ("" when the source has none), and the lowercase |
| 828 | // hex sha256 of the instance key a launch hands the app |
| 829 | // (instancekey.Effective over the rendered configuration and the own file, |
| 830 | // instancekey.Digest; "" when neither supplies one). archived is the env |
| 831 | // entry's digest as the archive recorded it ("sha256:<hex>", or "" for no |
| 832 | // env entry); the file read here must be the one archived. |
| 833 | func offlineEnvDigests(envPath, configPath, archived string) (envSHA, keySHA string, err error) { |
| 834 | own, exists, err := readOwnFile(envPath) |
| 835 | if err != nil { |
| 836 | return "", "", fmt.Errorf("read the env file: %w", err) |
| 837 | } |
| 838 | if exists { |
| 839 | sum := sha256.Sum256(own) |
| 840 | envSHA = hex.EncodeToString(sum[:]) |
| 841 | } |
| 842 | if want := strings.TrimPrefix(archived, "sha256:"); want != envSHA { |
| 843 | return "", "", fmt.Errorf("the env file changed while it was archived (archived %q, now %q); sealed again next pass", want, envSHA) |
| 844 | } |
| 845 | config, _, err := readOwnFile(configPath) |
| 846 | if err != nil { |
| 847 | return "", "", fmt.Errorf("read the rendered configuration: %w", err) |
| 848 | } |
| 849 | if key, _ := instancekey.Effective(instancekey.ParseEnv(config), instancekey.ParseEnv(own)); key != "" { |
| 850 | keySHA = instancekey.Digest(key) |
| 851 | } |
| 852 | return envSHA, keySHA, nil |
| 853 | } |
| 854 | |
| 855 | // offlineSealRecord reads seal.json, nil when it is not there. |
| 856 | func (w *movesWorker) offlineSealRecord(ctx context.Context, s release.Store, rec provision.Record) ([]byte, error) { |
| 857 | key := moves.OfflineSealKey(rec.Host, rec.Move.ID) |
| 858 | raw, _, err := s.Get(ctx, key) |
| 859 | if errors.Is(err, os.ErrNotExist) { |
| 860 | return nil, nil |
| 861 | } |
| 862 | if err != nil { |
| 863 | return nil, fmt.Errorf("read %s: %w", key, err) |
| 864 | } |
| 865 | return raw, nil |
| 866 | } |
| 867 | |
| 868 | // decodeSeal parses and checks the seal record; nil raw is no seal. |
| 869 | func decodeSeal(raw []byte, host, id string) (*moves.OfflineReport, error) { |
| 870 | if raw == nil { |
| 871 | return nil, nil |
| 872 | } |
| 873 | var seal moves.OfflineReport |
| 874 | if err := json.Unmarshal(raw, &seal); err != nil { |
| 875 | return nil, fmt.Errorf("the seal record does not parse: %w", err) |
| 876 | } |
| 877 | if err := moves.ValidateOfflineSeal(seal, host, id); err != nil { |
| 878 | return nil, err |
| 879 | } |
| 880 | return &seal, nil |
| 881 | } |
| 882 | |
| 883 | // offlineKeepStop is every pass after the seal, until the tombstone is |
| 884 | // lifted: read the lease first; a tombstone that is not this move's any |
| 885 | // more is a lost stop, never taken again. Then the source reports: one |
| 886 | // written since the seal at a position past the sealed one is a lost stop |
| 887 | // too (a write the stop did not hold off, which the staged copy lacks). |
| 888 | // Only then is this move's tombstone re-asserted. In |
| 889 | // offline-sealing (a seal written by a pass that crashed before the CAS) |
| 890 | // it also makes the CAS. Then it rewrites the living sealer report: the |
| 891 | // seal's values and items, the tombstone's owner and expiry, and At. |
| 892 | func (w *movesWorker) offlineKeepStop(ctx context.Context, s release.Store, rec provision.Record, raw []byte) error { |
| 893 | cfg := w.cfg |
| 894 | m, host := rec.Move, rec.Host |
| 895 | seal, err := decodeSeal(raw, host, m.ID) |
| 896 | if err != nil { |
| 897 | return fmt.Errorf("offline move %s: %s: %w; it is never rewritten, so abort the move", m.ID, host, err) |
| 898 | } |
| 899 | if seal == nil && m.State == provision.OfflineSealing { |
| 900 | return fmt.Errorf("offline move %s: %s has no seal record", m.ID, host) |
| 901 | } |
| 902 | st, err := w.offlineLock(ctx, host) |
| 903 | if err != nil { |
| 904 | return err |
| 905 | } |
| 906 | if !st.mine(m.ID) { |
| 907 | return w.offlineLeaseLost(ctx, s, rec, seal, st) |
| 908 | } |
| 909 | if seal != nil { |
| 910 | if err := w.offlineCheckPositions(ctx, s, rec, seal); err != nil { |
| 911 | return err |
| 912 | } |
| 913 | } |
| 914 | bucket, loc, region, err := cfg.tombstoneAt(host) |
| 915 | if err != nil { |
| 916 | return err |
| 917 | } |
| 918 | if err := w.sealFault("reassert"); err != nil { |
| 919 | return err |
| 920 | } |
| 921 | if err := lease.Reassert(ctx, bucket, loc, region, moves.TombstoneOwner(m.ID), moves.TombstoneTTL); err != nil { |
| 922 | var held *lease.HeldError |
| 923 | if errors.As(err, &held) { |
| 924 | return w.offlineLeaseLost(ctx, s, rec, seal, tombState{owner: held.Owner, expires: held.ExpiresAt}) |
| 925 | } |
| 926 | if errors.Is(err, lease.ErrTombstoneLost) { |
| 927 | // Gone between the read above and Reassert's own: the same |
| 928 | // lost stop, and Reassert wrote nothing. Read it again for |
| 929 | // what it is now (absent, when even that read fails). |
| 930 | now, rerr := w.offlineLock(ctx, host) |
| 931 | if rerr != nil { |
| 932 | now = tombState{} |
| 933 | } |
| 934 | return w.offlineLeaseLost(ctx, s, rec, seal, now) |
| 935 | } |
| 936 | return fmt.Errorf("offline move %s: re-assert the tombstone of %s: %w", m.ID, host, err) |
| 937 | } |
| 938 | if st, err = w.offlineLock(ctx, host); err != nil { |
| 939 | return err |
| 940 | } |
| 941 | if !st.mine(m.ID) { |
| 942 | return w.offlineLeaseLost(ctx, s, rec, seal, st) |
| 943 | } |
| 944 | if m.State == provision.OfflineSealing { |
| 945 | src := provision.MoveSource{ |
| 946 | FinalTXID: seal.FinalTXID, Replication: seal.Replication, NextDue: seal.NextDue, |
| 947 | Files: &provision.MoveFilesDigest{Plain: seal.FilesPlain, Sealed: seal.FilesSealed, Bytes: seal.FilesBytes, Entries: seal.FilesEntries}, |
| 948 | TimersSHA256: seal.TimersDigest, Release: seal.Release, EnvDigest: seal.EnvDigest, |
| 949 | Evidence: moves.OfflineSealEvidence(host, m.ID, raw), |
| 950 | } |
| 951 | _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, host, provision.OfflineSealedBy(m.ID, cfg.Box, src)) |
| 952 | switch { |
| 953 | case errors.Is(err, provision.ErrMoveConflict): |
| 954 | log.Printf("moves: %s: offline move %s: the record moved on before the seal's CAS: %v", host, m.ID, err) |
| 955 | case err != nil: |
| 956 | return fmt.Errorf("offline move %s: record the seal of %s: %w", m.ID, host, err) |
| 957 | default: |
| 958 | nudgeSerial(ctx, s) |
| 959 | log.Printf("moves: %s: offline move %s: sealed", host, m.ID) |
| 960 | } |
| 961 | if err := w.sealFault("cas"); err != nil { |
| 962 | return err |
| 963 | } |
| 964 | } |
| 965 | living := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer} |
| 966 | if seal != nil { |
| 967 | living = *seal |
| 968 | } |
| 969 | living.Phase = moves.OfflinePhaseSealed |
| 970 | living.At = cfg.now().UTC() |
| 971 | living.TombstoneOwner = st.owner |
| 972 | living.TombstoneUntil = st.expires.UTC() |
| 973 | return writeOfflineReport(ctx, s, living) |
| 974 | } |
| 975 | |
| 976 | // offlineSealerRefresh is the sealer's half of the report refresh |
| 977 | // (offlineReassert), from offline-sealed to offline-landed: offlineKeepStop's |
| 978 | // checks without its write to the lease. It reads the seal and the lease, a |
| 979 | // tombstone that is not this move's live one is the lost stop it always is, |
| 980 | // it checks the source positions against the seal, and it rewrites the |
| 981 | // living report with the tombstone's owner and expiry exactly as read. It |
| 982 | // never re-asserts the tombstone (fix round 1, Astra's finding 2: a refresh |
| 983 | // every 15 s through offlineKeepStop pushed the expiry out seven days each |
| 984 | // time); the sealer's ordinary pass keeps doing that, once a pass. |
| 985 | func (w *movesWorker) offlineSealerRefresh(ctx context.Context, s release.Store, rec provision.Record) error { |
| 986 | cfg := w.cfg |
| 987 | m, host := rec.Move, rec.Host |
| 988 | if !m.Offline() || cfg.Box == "" || m.Sealer != cfg.Box { |
| 989 | return nil |
| 990 | } |
| 991 | switch m.State { |
| 992 | case provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded: |
| 993 | default: |
| 994 | return nil |
| 995 | } |
| 996 | raw, err := w.offlineSealRecord(ctx, s, rec) |
| 997 | if err != nil { |
| 998 | return err |
| 999 | } |
| 1000 | seal, err := decodeSeal(raw, host, m.ID) |
| 1001 | if err != nil { |
| 1002 | return fmt.Errorf("offline move %s: %s: %w; it is never rewritten, so abort the move", m.ID, host, err) |
| 1003 | } |
| 1004 | st, err := w.offlineLock(ctx, host) |
| 1005 | if err != nil { |
| 1006 | return err |
| 1007 | } |
| 1008 | if !st.mine(m.ID) { |
| 1009 | return w.offlineLeaseLost(ctx, s, rec, seal, st) |
| 1010 | } |
| 1011 | if seal != nil { |
| 1012 | if err := w.offlineCheckPositions(ctx, s, rec, seal); err != nil { |
| 1013 | return err |
| 1014 | } |
| 1015 | } |
| 1016 | living := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer} |
| 1017 | if seal != nil { |
| 1018 | living = *seal |
| 1019 | } |
| 1020 | living.Phase = moves.OfflinePhaseSealed |
| 1021 | living.At = cfg.now().UTC() |
| 1022 | living.TombstoneOwner = st.owner |
| 1023 | living.TombstoneUntil = st.expires.UTC() |
| 1024 | return writeOfflineReport(ctx, s, living) |
| 1025 | } |
| 1026 | |
| 1027 | // offlineCheckPositions is the backstop behind the holds and the fence: |
| 1028 | // every source report written since the seal (at or after its At) must be |
| 1029 | // at or below the sealed final position. One past it is a box that wrote |
| 1030 | // what the staged copy does not hold, so the sealed position is no longer |
| 1031 | // the instance's: a lost stop, in the db-tail item. |
| 1032 | func (w *movesWorker) offlineCheckPositions(ctx context.Context, s release.Store, rec provision.Record, seal *moves.OfflineReport) error { |
| 1033 | reports, err := readOfflineReports(ctx, s, rec.Host, rec.Move.ID, moves.OfflineRoleSource) |
| 1034 | if err != nil { |
| 1035 | return fmt.Errorf("read the source reports: %w", err) |
| 1036 | } |
| 1037 | boxes := make([]string, 0, len(reports)) |
| 1038 | for b := range reports { |
| 1039 | boxes = append(boxes, b) |
| 1040 | } |
| 1041 | sort.Strings(boxes) |
| 1042 | for _, b := range boxes { |
| 1043 | r := reports[b] |
| 1044 | if r.At.Before(seal.At) || r.LocalTXID <= seal.FinalTXID { |
| 1045 | continue |
| 1046 | } |
| 1047 | return w.offlineLostStop(ctx, s, rec, seal, "db-tail", offlinePastSealCause(b, r.LocalTXID, seal.FinalTXID)) |
| 1048 | } |
| 1049 | return nil |
| 1050 | } |
| 1051 | |
| 1052 | // offlineLeaseLost is the lost stop of a tombstone that is absent, |
| 1053 | // expired, lifted or someone else's, in the lease item. |
| 1054 | func (w *movesWorker) offlineLeaseLost(ctx context.Context, s release.Store, rec provision.Record, seal *moves.OfflineReport, st tombState) error { |
| 1055 | return w.offlineLostStop(ctx, s, rec, seal, "lease", offlineLeaseCause(st)) |
| 1056 | } |
| 1057 | |
| 1058 | // offlineLeaseCause and offlinePastSealCause are the two causes of a lost |
| 1059 | // stop as its page and the failed item say them (MV-O69, MV-O70). |
| 1060 | func offlineLeaseCause(st tombState) string { |
| 1061 | return fmt.Sprintf("the source's lease is %s, not this move's tombstone", offlineLeaseText(st)) |
| 1062 | } |
| 1063 | |
| 1064 | func offlinePastSealCause(box string, pos, final uint64) string { |
| 1065 | return fmt.Sprintf("%s reports TXID %d, past the sealed position %d", box, pos, final) |
| 1066 | } |
| 1067 | |
| 1068 | // offlineLostStop reports a stop that no longer holds after the seal: |
| 1069 | // cause says why, and item is the seal item it fails. The tombstone is |
| 1070 | // never taken again (a wake may have written the replica in between, so |
| 1071 | // the sealed position is no longer the source's), the living report says |
| 1072 | // failed with that item not OK and no tombstone owner (so open refuses), |
| 1073 | // and an operator is paged once per move. |
| 1074 | func (w *movesWorker) offlineLostStop(ctx context.Context, s release.Store, rec provision.Record, seal *moves.OfflineReport, item, cause string) error { |
| 1075 | cfg := w.cfg |
| 1076 | m, host := rec.Move, rec.Host |
| 1077 | r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer} |
| 1078 | if seal != nil { |
| 1079 | r = *seal |
| 1080 | } |
| 1081 | r.Phase = moves.OfflinePhaseFailed |
| 1082 | r.At = cfg.now().UTC() |
| 1083 | r.TombstoneOwner, r.TombstoneUntil = "", time.Time{} |
| 1084 | items := map[string]moves.Check{} |
| 1085 | for k, v := range r.Items { |
| 1086 | items[k] = v |
| 1087 | } |
| 1088 | items[item] = moves.Check{OK: false, Reason: "lost stop: " + cause} |
| 1089 | r.Items = items |
| 1090 | werr := writeOfflineReport(ctx, s, r) |
| 1091 | subject, body := offlineLostStopPage(host, m.ID, m.State, cause) |
| 1092 | w.firePage(ctx, host, m.ID, offlinePageLostStop, alerting.LevelIncident, subject, body) |
| 1093 | log.Printf("moves: %s: offline move %s: LOST STOP: %s", host, m.ID, cause) |
| 1094 | return errors.Join(fmt.Errorf("offline move %s: %s: lost stop: %s", m.ID, host, cause), werr) |
| 1095 | } |
| 1096 | |
| 1097 | // offlineLostStopPage is the lost stop's page (copy rows MV-O66 to |
| 1098 | // MV-O68; the item's reason is MV-O71): its subject, and a body for the move's state. Before open, |
| 1099 | // abort puts the instance back with nothing lost, so the body says to |
| 1100 | // abort. From open on, abort is refused (MV-O40) and the move is fixed |
| 1101 | // forward by hand, so the body points to the status and the runbook. |
| 1102 | func offlineLostStopPage(host, id, state, cause string) (subject, body string) { |
| 1103 | subject = fmt.Sprintf("offline move of %s lost its stop", host) |
| 1104 | if state == provision.OfflineOpen || state == provision.OfflineFinishing { |
| 1105 | body = fmt.Sprintf("The offline move %s of %s is %s, and its stop no longer holds: %s. It cannot be aborted once it is open. Read carlos move offline status --host %s and follow the offline move runbook.", id, host, state, cause, host) |
| 1106 | return subject, body |
| 1107 | } |
| 1108 | body = fmt.Sprintf("The offline move %s of %s is %s, and its stop no longer holds: %s. The move cannot go on. Read carlos move offline status --host %s and abort the move.", id, host, state, cause, host) |
| 1109 | return subject, body |
| 1110 | } |
| 1111 | |
| 1112 | // offlineFinishLift is the sealer's step in offline-finishing. Until every |
| 1113 | // box the seal named reports gone (it parked at open and reconcile has |
| 1114 | // removed its row since), the stop is kept as before. Then the tombstone |
| 1115 | // is lifted (only this move's, with the ETag read), the copy receipts are |
| 1116 | // cleared and the report says lifted. A lease that is already free (lifted |
| 1117 | // by an earlier pass) is left as it is; one someone else holds live is a |
| 1118 | // lost stop. |
| 1119 | func (w *movesWorker) offlineFinishLift(ctx context.Context, s release.Store, rec provision.Record) error { |
| 1120 | cfg := w.cfg |
| 1121 | m, host := rec.Move, rec.Host |
| 1122 | reports, err := readOfflineReports(ctx, s, host, m.ID, moves.OfflineRoleSource) |
| 1123 | if err != nil { |
| 1124 | return fmt.Errorf("read the source reports: %w", err) |
| 1125 | } |
| 1126 | gone := len(m.Boxes) > 0 |
| 1127 | for _, b := range m.Boxes { |
| 1128 | if r, ok := reports[b]; !ok || r.Phase != moves.OfflinePhaseGone { |
| 1129 | gone = false |
| 1130 | } |
| 1131 | } |
| 1132 | raw, err := w.offlineSealRecord(ctx, s, rec) |
| 1133 | if err != nil { |
| 1134 | return err |
| 1135 | } |
| 1136 | if !gone { |
| 1137 | return w.offlineKeepStop(ctx, s, rec, raw) |
| 1138 | } |
| 1139 | st, err := w.offlineLock(ctx, host) |
| 1140 | if err != nil { |
| 1141 | return err |
| 1142 | } |
| 1143 | switch { |
| 1144 | case st.mine(m.ID): |
| 1145 | bucket, loc, region, err := cfg.tombstoneAt(host) |
| 1146 | if err != nil { |
| 1147 | return err |
| 1148 | } |
| 1149 | if _, err := lease.LiftTombstone(ctx, bucket, loc, region, func(owner string) bool { return owner == moves.TombstoneOwner(m.ID) }); err != nil { |
| 1150 | return fmt.Errorf("offline move %s: lift the tombstone of %s: %w", m.ID, host, err) |
| 1151 | } |
| 1152 | log.Printf("moves: %s: offline move %s: tombstone lifted", host, m.ID) |
| 1153 | case st.free(): |
| 1154 | default: |
| 1155 | seal, _ := decodeSeal(raw, host, m.ID) |
| 1156 | return w.offlineLeaseLost(ctx, s, rec, seal, st) |
| 1157 | } |
| 1158 | if err := w.sealFault("lift"); err != nil { |
| 1159 | return err |
| 1160 | } |
| 1161 | own, err := w.offlineMirror(ctx, cfg.OwnBucketRegion) |
| 1162 | if err != nil { |
| 1163 | return err |
| 1164 | } |
| 1165 | clearOfflineReceipts(ctx, own, cfg.OwnBucket, host) |
| 1166 | r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseLifted, At: cfg.now().UTC()} |
| 1167 | return writeOfflineReport(ctx, s, r) |
| 1168 | } |
| 1169 | |
| 1170 | // clearOfflineReceipts is clearMoveReceipts as an ensure: receipts that are |
| 1171 | // already gone are not deleted again, so a pass that repeats finish or |
| 1172 | // revert writes nothing (on a versioned bucket every DELETE of an absent |
| 1173 | // key leaves one more delete marker). |
| 1174 | func clearOfflineReceipts(ctx context.Context, m lease.MirrorStore, bucket, host string) { |
| 1175 | rc := lease.Receipts{Bucket: bucket, Key: activator.MoveReceiptsKey(host)} |
| 1176 | if _, err := m.Get(ctx, rc.Bucket, rc.Key); errors.Is(err, lease.ErrMirrorNotFound) { |
| 1177 | return |
| 1178 | } |
| 1179 | clearMoveReceipts(ctx, m, rc, host) |
| 1180 | } |
| 1181 | |
| 1182 | // offlineSealRevert is the sealer's step in offline-reverting: lift the |
| 1183 | // tombstone if it is this move's (a lease anyone else holds, or none, is |
| 1184 | // left alone), delete the two prefixes this move wrote in the target's |
| 1185 | // bucket (the staged replica and the archive), clear the copy receipts, |
| 1186 | // lift this move's fence (the source step lifts it too) and report |
| 1187 | // reverted. Every part is an ensure. |
| 1188 | func (w *movesWorker) offlineSealRevert(ctx context.Context, s release.Store, rec provision.Record) error { |
| 1189 | cfg := w.cfg |
| 1190 | m, host := rec.Move, rec.Host |
| 1191 | bucket, loc, region, err := cfg.tombstoneAt(host) |
| 1192 | if err != nil { |
| 1193 | return err |
| 1194 | } |
| 1195 | lifted, err := lease.LiftTombstone(ctx, bucket, loc, region, func(owner string) bool { return owner == moves.TombstoneOwner(m.ID) }) |
| 1196 | var held *lease.HeldError |
| 1197 | switch { |
| 1198 | case err == nil: |
| 1199 | if lifted { |
| 1200 | log.Printf("moves: %s: offline move %s: tombstone lifted on abort", host, m.ID) |
| 1201 | } |
| 1202 | case errors.As(err, &held): |
| 1203 | w.notice(host+"#offline-revert-lease", "moves: %s: offline move %s: the lease is %s's, not this move's; left as it is", host, m.ID, held.Owner) |
| 1204 | default: |
| 1205 | return fmt.Errorf("offline move %s: lift the tombstone of %s: %w", m.ID, host, err) |
| 1206 | } |
| 1207 | if err := w.sealFault("revert-lift"); err != nil { |
| 1208 | return err |
| 1209 | } |
| 1210 | tgt, err := w.offlineTarget(ctx, s, m) |
| 1211 | if err != nil { |
| 1212 | return err |
| 1213 | } |
| 1214 | mirror, err := w.offlineMirror(ctx, tgt.DataBucketRegion) |
| 1215 | if err != nil { |
| 1216 | return err |
| 1217 | } |
| 1218 | for _, prefix := range []string{activator.MovesPrefix(host), moves.OfflineArchivePrefix(host, m.ID)} { |
| 1219 | if prefix == "" || prefix == "moves/" { |
| 1220 | return fmt.Errorf("offline move %s: no staging prefix for %s", m.ID, host) |
| 1221 | } |
| 1222 | if err := lease.DeletePrefix(ctx, mirror, tgt.DataBucket, prefix); err != nil { |
| 1223 | return fmt.Errorf("offline move %s: delete %s/%s: %w", m.ID, tgt.DataBucket, prefix, err) |
| 1224 | } |
| 1225 | } |
| 1226 | clearOfflineReceipts(ctx, mirror, bucket, host) |
| 1227 | w.offlineLiftFence(host, m) |
| 1228 | r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC()} |
| 1229 | return writeOfflineReport(ctx, s, r) |
| 1230 | } |
| 1231 | |
amadan