| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/hex" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "github.com/carlosframework/platform/internal/park" |
| 10 | "io" |
| 11 | "log" |
| 12 | "os" |
| 13 | "path/filepath" |
| 14 | "slices" |
| 15 | "strings" |
| 16 | "time" |
| 17 | |
| 18 | "github.com/carlosframework/platform/internal/appconfig" |
| 19 | "github.com/carlosframework/platform/internal/hostclaim" |
| 20 | "github.com/carlosframework/platform/internal/hostfiles" |
| 21 | "github.com/carlosframework/platform/internal/lease" |
| 22 | "github.com/carlosframework/platform/internal/moves" |
| 23 | "github.com/carlosframework/platform/internal/moves/archive" |
| 24 | "github.com/carlosframework/platform/internal/objstore" |
| 25 | "github.com/carlosframework/platform/internal/provision" |
| 26 | "github.com/carlosframework/platform/internal/registry" |
| 27 | "github.com/carlosframework/platform/internal/release" |
| 28 | ) |
| 29 | |
| 30 | // The Phase 1 source round (moves spec §7.3, §7.11, K5, K12), behind |
| 31 | // CARLOS_MOVES_PHASE1. It drives a move whose record carries an id: the |
| 32 | // staging lane under moves/<host>/<id>/ in the target bucket, written only |
| 33 | // through the guarded staging mirror while this box holds the staging-work |
| 34 | // lease; the per-operation receipts and ledger in this box's own bucket; the |
| 35 | // source's readiness report in the deployment bucket; and the promotion to |
| 36 | // ready (and, for an automatic cut, to cutting) by guarded CAS. The legacy |
| 37 | // worker (legacyMoveRound) is untouched and runs only with the switch off. |
| 38 | // |
| 39 | // The operator-visible sentences below are rows of the copy record |
| 40 | // (docs/superpowers/plans/2026-10-06-moves-phase1-copy.md), pending review; |
| 41 | // they reach `carlos move status` through the source report's S1. |
| 42 | const ( |
| 43 | s1NoDeployBucket = "this deployment has no deployment bucket for the staging lease" |
| 44 | s1Nested = "nested instances cannot move yet" |
| 45 | s1NoSecretsKey = "this box has no secrets key, so it cannot report the launch environment or seal the archive" |
| 46 | s1CannotProbe = "this box cannot probe the instance yet" |
| 47 | s1NoReadyPath = "the app's readiness path could not be read" |
| 48 | ) |
| 49 | |
| 50 | // archivePlainDigest and archiveBuild are the archive's digest and build, |
| 51 | // as variables so a test can make a file change under the walk. |
| 52 | var ( |
| 53 | archivePlainDigest = archive.PlainDigest |
| 54 | archiveBuild = archive.Build |
| 55 | ) |
| 56 | |
| 57 | func s1NotExec(host string) string { |
| 58 | return host + " is not an exec-backed instance, so it cannot move" |
| 59 | } |
| 60 | func s1NoHibernate(host string) string { return host + " does not hibernate, so it cannot move" } |
| 61 | func s1Sidecar(host string) string { |
| 62 | return host + " has a sidecar environment file, so it cannot move" |
| 63 | } |
| 64 | func s1Column(host, column string) string { |
| 65 | return host + " has " + column + " set in the registry, so it cannot move" |
| 66 | } |
| 67 | |
| 68 | // s1PrivateStorage is M-C17, the door's sentence repeated every round. |
| 69 | func s1PrivateStorage(host string) string { |
| 70 | return host + " uses private storage. Instances with private storage cannot move yet." |
| 71 | } |
| 72 | |
| 73 | // s1SpecialFile and s1OverCap are M-C14 and M-C15. |
| 74 | func s1SpecialFile(host, path string) string { |
| 75 | return host + " keeps a file this move cannot copy: " + path + ". Only regular files and folders move." |
| 76 | } |
| 77 | func s1OverCap(host string) string { |
| 78 | return host + " keeps more than 256 MiB of files beside its database. Keep large files in object storage." |
| 79 | } |
| 80 | |
| 81 | // s1FilesPastCap is MV-N501 (#236): the files manifest a recovery of the |
| 82 | // host would restore holds an entry path past the readers' cap, so no |
| 83 | // recovery after the cut could restore it (sourceFilesRefusal). |
| 84 | func s1FilesPastCap(host string) string { |
| 85 | return host + " has a files manifest with a path longer than 4095 bytes, so it cannot move. an operator must replace the manifest" |
| 86 | } |
| 87 | |
| 88 | // s1FilePastCap is MV-N502 (#236 fix round 1): the files directory itself |
| 89 | // holds a path longer than the files tree's cap (archive.ErrPastTreeCap, |
| 90 | // one byte under the archive format's MV-N81 bound). The archive could |
| 91 | // carry it, but the target's prepared files manifest could not, so the |
| 92 | // landing would fail after the cut; the source refuses here instead. |
| 93 | func s1FilePastCap(host string) string { |
| 94 | return host + " has a file whose path is longer than 4095 bytes, so it cannot move. shorten the path or remove the file" |
| 95 | } |
| 96 | |
| 97 | // The rows say 4095, so the sentences do; this fails to compile if |
| 98 | // hostfiles.MaxPathBytes is ever changed without the rows. |
| 99 | const _ = uint(hostfiles.MaxPathBytes-4095) + uint(4095-hostfiles.MaxPathBytes) |
| 100 | |
| 101 | // s1TooManyEntries and s1LongPath are MV-N79 and MV-N81, the archive's own |
| 102 | // rows, written here from data (the sentinel, the path the error carries) |
| 103 | // rather than taken from an error's text. s1ArchiveFailed is every archive |
| 104 | // error this round does not recognise; its text goes to the log only. |
| 105 | const ( |
| 106 | s1TooManyEntries = "archive: the files directory holds more than 131072 entries" |
| 107 | s1ArchiveFailed = "the files beside the database could not be archived" |
| 108 | ) |
| 109 | |
| 110 | func s1LongPath(path string) string { |
| 111 | return "archive: refused path: " + path + " is longer than 4096 bytes" |
| 112 | } |
| 113 | |
| 114 | // s1NoAnswer is M-C16. |
| 115 | func s1NoAnswer(host string) string { |
| 116 | return host + " does not answer its readiness path. Declare moves.ready_path for this app." |
| 117 | } |
| 118 | |
| 119 | // crossRefreshAfter is how old the source's cross-region rate may get before |
| 120 | // a round that copied nothing measures it again by recopying one staged |
| 121 | // object. Half of moves.MeasurementFreshness, so a quiet (hibernating) |
| 122 | // source keeps a fresh rate in its report while it waits for ready. |
| 123 | const crossRefreshAfter = moves.MeasurementFreshness / 2 |
| 124 | |
| 125 | // crossMinBytes is the least a round must copy for its rate to be recorded: |
| 126 | // a round of a few small files measures latency, not throughput. |
| 127 | const crossMinBytes = 1 << 20 |
| 128 | |
| 129 | // moveLedger is the per-operation ledger beside the copy receipts, at |
| 130 | // moves.LedgerKey in this box's own bucket (§6.3). It is what the rounds |
| 131 | // record about the lane: move.json's version (or "present" when a replay |
| 132 | // found it already there), the published archive by digest and by the |
| 133 | // version ids its PUTs returned, and the most recent copying round's |
| 134 | // cross-region rate. |
| 135 | type moveLedger struct { |
| 136 | V int `json:"v"` |
| 137 | MoveJSONVersion string `json:"move_json_version,omitempty"` |
| 138 | ArchivePlain string `json:"archive_plain,omitempty"` |
| 139 | ArchiveSealed string `json:"archive_sealed,omitempty"` |
| 140 | ArchiveVersions *moves.ArchiveRef `json:"archive_versions,omitempty"` |
| 141 | LastRoundBytes int64 `json:"last_round_bytes,omitempty"` |
| 142 | // LastRoundSeconds and LastRoundAt time the most recent round that |
| 143 | // copied at least crossMinBytes, or the recopy that refreshed it. |
| 144 | LastRoundSeconds float64 `json:"last_round_seconds,omitempty"` |
| 145 | LastRoundAt time.Time `json:"last_round_at,omitempty"` |
| 146 | // DeferredRounds counts the cut's final rounds that deferred an object |
| 147 | // or moved more than the bar (§7.4 staged; three revert). LeaseUntil is |
| 148 | // the seal's writers' bound (ruling G7), recorded beside it. |
| 149 | DeferredRounds int `json:"deferred_rounds,omitempty"` |
| 150 | LeaseUntil time.Time `json:"lease_until,omitempty"` |
| 151 | } |
| 152 | |
| 153 | // moveJSONPresent is the ledger's note that move.json already existed when |
| 154 | // this source wrote it: its own lost-response PUT, or a foreign object. The |
| 155 | // source cannot tell which (M17 forbids the read), so the target's T9 |
| 156 | // validates the object's fields. |
| 157 | const moveJSONPresent = "present" |
| 158 | |
| 159 | func (l moveLedger) crossRate() (float64, time.Time) { |
| 160 | if l.LastRoundBytes <= 0 || l.LastRoundSeconds <= 0 { |
| 161 | return 0, time.Time{} |
| 162 | } |
| 163 | return float64(l.LastRoundBytes) / (1 << 20) / l.LastRoundSeconds, l.LastRoundAt |
| 164 | } |
| 165 | |
| 166 | func readMoveLedger(ctx context.Context, m lease.MirrorStore, bucket, key string) (moveLedger, error) { |
| 167 | body, err := m.Get(ctx, bucket, key) |
| 168 | if errors.Is(err, lease.ErrMirrorNotFound) { |
| 169 | return moveLedger{V: 1}, nil |
| 170 | } |
| 171 | if err != nil { |
| 172 | return moveLedger{}, fmt.Errorf("read the move ledger %s/%s: %w", bucket, key, err) |
| 173 | } |
| 174 | var l moveLedger |
| 175 | if err := json.Unmarshal(body, &l); err != nil || l.V != 1 { |
| 176 | // An unreadable ledger is no evidence: start over. Every fact in it |
| 177 | // is recomputed (move.json answers present, the archive is rebuilt). |
| 178 | return moveLedger{V: 1}, nil |
| 179 | } |
| 180 | // Nor is a version id that pins nothing (final review finding 4): a |
| 181 | // "null" recorded before the staging mirror refused it. The archive is |
| 182 | // restaged and move.json asked for again (it answers present). |
| 183 | if l.ArchiveVersions != nil && !l.ArchiveVersions.Pinned() { |
| 184 | l.ArchivePlain, l.ArchiveSealed, l.ArchiveVersions = "", "", nil |
| 185 | } |
| 186 | if l.MoveJSONVersion != moveJSONPresent && !lease.UsableVersionID(l.MoveJSONVersion) { |
| 187 | l.MoveJSONVersion = "" |
| 188 | } |
| 189 | return l, nil |
| 190 | } |
| 191 | |
| 192 | func writeMoveLedger(ctx context.Context, m lease.MirrorStore, bucket, key string, l moveLedger) error { |
| 193 | body, err := json.Marshal(l) |
| 194 | if err != nil { |
| 195 | return err |
| 196 | } |
| 197 | if err := m.Put(ctx, bucket, key, body); err != nil { |
| 198 | return fmt.Errorf("write the move ledger %s/%s: %w", bucket, key, err) |
| 199 | } |
| 200 | return nil |
| 201 | } |
| 202 | |
| 203 | // phase1MoveRound is the Phase 1 worker's step for one id'd record this box |
| 204 | // is the source of. |
| 205 | func phase1MoveRound(ctx context.Context, s release.Store, w *movesWorker, rec provision.Record, fleet moveFleet) error { |
| 206 | // The transitions another writer made (the console's doors, a declared |
| 207 | // loss) are logged by the box that sees them (movesevents.go). |
| 208 | w.observeMove(ctx, s, rec) |
| 209 | switch rec.Move.State { |
| 210 | case provision.MovePreparing, provision.MoveReady: |
| 211 | return w.preparingRound(ctx, s, rec, fleet) |
| 212 | case provision.MoveAborting: |
| 213 | // The abort's source half (movesabort.go, Plan 5 Task 1): the lane |
| 214 | // is the target region's to park. |
| 215 | return w.abortSource(ctx, s, rec) |
| 216 | case provision.MoveCutting: |
| 217 | return w.runCut(ctx, s, rec, fleet) |
| 218 | case provision.MoveReverting: |
| 219 | // The record decided the revert; the claim follows it (§7.5), and |
| 220 | // the source recovers (movesrevert.go, Plan 4 Task 6, ruling J7). |
| 221 | // movesPass routes a reverting record by the participant rule |
| 222 | // before it gets here; this arm is the same step. |
| 223 | return w.revertSource(ctx, s, rec, &fleet) |
| 224 | case provision.MoveLanding: |
| 225 | // The target's (ruling G34): after the flip the record's region is |
| 226 | // this box's and reconcile has minted the held row, so OwnsRoute |
| 227 | // admits it here. checking to restored is Task 10's (movesland.go); |
| 228 | // the probe and the commit are Plan 3's. |
| 229 | return w.landingPass(ctx, s, rec) |
| 230 | case provision.MoveCommitted: |
| 231 | // The target's committed arm (Plan 3 Task 7, H9): the claim half |
| 232 | // and the conditional open, every pass while committed. |
| 233 | if movesParticipantRecord(w.cfg, rec) == roleTarget { |
| 234 | return w.committedTarget(ctx, s, rec) |
| 235 | } |
| 236 | } |
| 237 | return nil // later: Plans 4 and 5 |
| 238 | } |
| 239 | |
| 240 | // srcRound is what one preparing round learned for the source report. |
| 241 | type srcRound struct { |
| 242 | s1 *moves.Check // a refusal; nil means probe the instance |
| 243 | ledger moveLedger |
| 244 | deltaBytes int64 |
| 245 | // replicaBytes is the replica's logical size this round |
| 246 | // (replicaLogicalSize), carried as SourceReport.ReplicaBytes. |
| 247 | replicaBytes int64 |
| 248 | // holder is the claim holder this round read under the lease, for |
| 249 | // decideReady's layout rule (ruling T5-lease). |
| 250 | holder hostclaim.Holder |
| 251 | } |
| 252 | |
| 253 | // preparingRound is §7.3's source round (sourceRound), then Ready, the |
| 254 | // promotion (or the repin of a ready move whose pin a cut's admission |
| 255 | // cleared) and the automatic cut. |
| 256 | func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec provision.Record, fleet moveFleet) error { |
| 257 | sr, published, err := w.sourceRound(ctx, s, rec, fleet) |
| 258 | if err != nil || published == nil { |
| 259 | return err |
| 260 | } |
| 261 | // The invariant: a round promotes or requests a cut only when its own |
| 262 | // source report passed every check and was published. Any failing |
| 263 | // check ends the round here; a retry or an error ended it above. |
| 264 | if !sourcePassed(published) { |
| 265 | w.notice(rec.Host+"#ready", "moves: %s: this round's source report does not pass (S1 %v, S2 %v); no promotion and no cut", rec.Host, published.S1.OK, published.S2.OK) |
| 266 | return nil |
| 267 | } |
| 268 | return w.decideReady(ctx, s, rec, provision.MoveGuard{ID: rec.Move.ID, Incarnation: rec.Move.Incarnation}, sr, published) |
| 269 | } |
| 270 | |
| 271 | // sourceRound is §7.3's source round, in order: no deployment bucket |
| 272 | // refuses; the staging lease is taken and the claim revalidated under it; |
| 273 | // the box-step refusals and the one-box rule; the store binding; Touch; |
| 274 | // move.json; the replica lane; the archive; the source report. It returns |
| 275 | // the round and the report it published. A round that ended before any |
| 276 | // report (another worker holds the lease, a file changed under the |
| 277 | // archive) returns a nil report and no error: the next pass runs it again. |
| 278 | // The cut's `requested` runs it too (G13), and decides admission instead |
| 279 | // of readiness. |
| 280 | func (w *movesWorker) sourceRound(ctx context.Context, s release.Store, rec provision.Record, fleet moveFleet) (*srcRound, *moves.SourceReport, error) { |
| 281 | cfg := w.cfg |
| 282 | id := rec.Move.ID |
| 283 | report := func(sr *srcRound) (*srcRound, *moves.SourceReport, error) { |
| 284 | published, err := w.writeSourceReport(ctx, s, rec, sr) |
| 285 | return sr, published, err |
| 286 | } |
| 287 | // Round-5 finding 11: a lease in a per-participant bucket, or a |
| 288 | // 15-minute one, cannot fence staging writers. Without the deployment |
| 289 | // bucket there is no shared place for it, so the round refuses. |
| 290 | if cfg.StagingLease == nil { |
| 291 | return report(&srcRound{s1: &moves.Check{Reason: s1NoDeployBucket}}) |
| 292 | } |
| 293 | tgt, ok := fleet.targets[rec.Move.To] |
| 294 | if !ok { |
| 295 | return nil, nil, fmt.Errorf("target region %s publishes no data bucket yet. waiting", rec.Move.To) |
| 296 | } |
| 297 | held, releaseLease, err := cfg.StagingLease(ctx, rec.Move.To, rec.Host) |
| 298 | if err != nil { |
| 299 | var he *lease.HeldError |
| 300 | if errors.As(err, &he) { |
| 301 | log.Printf("moves: %s: another worker holds the staging lease; this round waits", rec.Host) |
| 302 | return nil, nil, nil |
| 303 | } |
| 304 | return nil, nil, fmt.Errorf("staging lease: %w", err) |
| 305 | } |
| 306 | defer func() { |
| 307 | rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 308 | defer cancel() |
| 309 | if err := releaseLease(rctx); err != nil { |
| 310 | log.Printf("moves: %s: release the staging lease: %v", rec.Host, err) |
| 311 | } |
| 312 | }() |
| 313 | // K5: revalidate the claim under the lease. |
| 314 | obs, err := hostclaim.Read(ctx, s, rec.Host) |
| 315 | if err != nil { |
| 316 | return nil, nil, fmt.Errorf("read the host claim: %w", err) |
| 317 | } |
| 318 | if obs.Absent { |
| 319 | return nil, nil, fmt.Errorf("move %s has no host claim", id) |
| 320 | } |
| 321 | if err := hostclaim.RequireMoveOp(id, hostclaim.KindMoving, hostclaim.PhaseMoving)(obs.Claim); err != nil { |
| 322 | return nil, nil, fmt.Errorf("move %s: %w", id, err) |
| 323 | } |
| 324 | if h := obs.Claim.Holder; h == nil || h.Incarnation != rec.Move.Incarnation { |
| 325 | return nil, nil, fmt.Errorf("move %s: the host claim names another incarnation", id) |
| 326 | } |
| 327 | if cfg.Box != "" && obs.Claim.Op.Exec != cfg.Box { |
| 328 | return nil, nil, fmt.Errorf("move %s: the claim's executor is %q, not this box", id, obs.Claim.Op.Exec) |
| 329 | } |
| 330 | // The box-step refusals (§7.1: every box step re-checks), and the |
| 331 | // one-box rule the door applied (D15, ruling G13): a region that has |
| 332 | // gained a second box since is refused here, every round. |
| 333 | if why, err := w.sourceRefusal(ctx, s, rec, obs.Claim.Holder); err != nil { |
| 334 | return nil, nil, err |
| 335 | } else if why != "" { |
| 336 | return report(&srcRound{s1: &moves.Check{Reason: why}}) |
| 337 | } |
| 338 | if why := oneBoxRefusal(fleet, rec); why != "" { |
| 339 | return report(&srcRound{s1: &moves.Check{Reason: why}}) |
| 340 | } |
| 341 | // K12 and T9: the target region and box are in touched before the |
| 342 | // first staging write. |
| 343 | targetBox := obs.Claim.Op.Intent["target"] |
| 344 | if !slices.ContainsFunc(obs.Claim.Touched.Regions, func(region string) bool { |
| 345 | return park.PlacementMatches(obs.Claim.Op.Intent["home_region"], rec.Move.To, region) |
| 346 | }) || (targetBox != "" && !containsString(obs.Claim.Touched.Boxes, targetBox)) { |
| 347 | var boxes []string |
| 348 | if targetBox != "" { |
| 349 | boxes = []string{targetBox} |
| 350 | } |
| 351 | if _, err := hostclaim.Update(ctx, s, obs, hostclaim.Touch(id, []string{park.PlacementRegion(rec.Move.To)}, boxes)); err != nil { |
| 352 | return nil, nil, fmt.Errorf("touch %s in the host claim: %w", rec.Move.To, err) |
| 353 | } |
| 354 | } |
| 355 | stg, err := w.staging(ctx, tgt.DataBucketRegion) |
| 356 | if err != nil { |
| 357 | return nil, nil, err |
| 358 | } |
| 359 | own, err := w.mirror(ctx, tgt.DataBucketRegion) |
| 360 | if err != nil { |
| 361 | return nil, nil, err |
| 362 | } |
| 363 | gm := &guardedMirror{w: stg, g: moves.Guarded{Lease: held, Now: cfg.Now}, dst: tgt.DataBucket, own: own} |
| 364 | ledgerKey := moves.LedgerKey(rec.Host, id) |
| 365 | ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey) |
| 366 | if err != nil { |
| 367 | return nil, nil, err |
| 368 | } |
| 369 | before := ledger |
| 370 | saveLedger := func() error { |
| 371 | if ledger == before { |
| 372 | return nil |
| 373 | } |
| 374 | if err := writeMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey, ledger); err != nil { |
| 375 | return err |
| 376 | } |
| 377 | before = ledger |
| 378 | return nil |
| 379 | } |
| 380 | |
| 381 | // move.json, create-only, before the first round. Once the ledger |
| 382 | // knows it, it is not written again. |
| 383 | if ledger.MoveJSONVersion == "" { |
| 384 | body, err := json.Marshal(moves.MoveJSON{MoveID: id, Incarnation: rec.Move.Incarnation, Host: rec.Host, From: rec.Move.From, To: rec.Move.To}) |
| 385 | if err != nil { |
| 386 | return nil, nil, err |
| 387 | } |
| 388 | ver, err := gm.PutIfAbsentVersioned(ctx, tgt.DataBucket, moves.MoveJSONKey(rec.Host, id), body) |
| 389 | switch { |
| 390 | case errors.Is(err, lease.ErrExists): |
| 391 | ledger.MoveJSONVersion = moveJSONPresent |
| 392 | case err != nil: |
| 393 | return nil, nil, fmt.Errorf("write move.json: %w", err) |
| 394 | default: |
| 395 | ledger.MoveJSONVersion = ver |
| 396 | } |
| 397 | if err := saveLedger(); err != nil { |
| 398 | return nil, nil, err |
| 399 | } |
| 400 | } |
| 401 | |
| 402 | // The replica lane: the claim's generation, flat under replica/. |
| 403 | srcPrefix, keep := moves.ReplicaSource(*obs.Claim.Holder) |
| 404 | dstPrefix := moves.ReplicaPrefix(rec.Host, id) |
| 405 | receipts := lease.Receipts{Bucket: cfg.OwnBucket, Key: moves.ReceiptsKey(rec.Host, id)} |
| 406 | var round lease.MirrorRound |
| 407 | start := cfg.Now() |
| 408 | for attempt := 0; ; attempt++ { |
| 409 | round, err = lease.MirrorPrefixKeep(ctx, gm, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts, keep) |
| 410 | // One retry, and only by calling again: a new check, a new |
| 411 | // signature. A guard refusal is never retried. |
| 412 | if err == nil || attempt == 1 || errors.Is(err, moves.ErrLeaseUnsafe) || errors.Is(err, moves.ErrInvalidSkew) { |
| 413 | break |
| 414 | } |
| 415 | log.Printf("moves: %s: replica round failed, retrying once: %v", rec.Host, err) |
| 416 | } |
| 417 | if err != nil { |
| 418 | return nil, nil, fmt.Errorf("replica lane: %w", err) |
| 419 | } |
| 420 | // The build state travels with the replica, in every round. |
| 421 | if _, _, err := mirrorBuildState(ctx, gm, cfg.OwnBucket, obs.Claim.Holder.Location, tgt.DataBucket, rec.Host, id); err != nil { |
| 422 | return nil, nil, fmt.Errorf("build state lane: %w", err) |
| 423 | } |
| 424 | if round.BytesCopied >= crossMinBytes { |
| 425 | ledger.LastRoundBytes, ledger.LastRoundSeconds, ledger.LastRoundAt = round.BytesCopied, cfg.Now().Sub(start).Seconds(), cfg.Now().UTC() |
| 426 | } else if _, at := ledger.crossRate(); at.IsZero() || cfg.Now().Sub(at) > crossRefreshAfter { |
| 427 | if err := w.refreshCrossRate(ctx, gm, srcPrefix, keep, tgt.DataBucket, dstPrefix, &ledger); err != nil { |
| 428 | return nil, nil, err |
| 429 | } |
| 430 | } |
| 431 | |
| 432 | // The database's size as this round's replica states it, for Ready to |
| 433 | // hold against the size the target's report priced (final review |
| 434 | // finding 2). An unreadable header ends the round: no report, no |
| 435 | // promotion. |
| 436 | replicaBytes, err := w.replicaLogicalSize(ctx, gm, srcPrefix, keep) |
| 437 | if err != nil { |
| 438 | return nil, nil, err |
| 439 | } |
| 440 | sr := &srcRound{deltaBytes: round.DeltaBytes(), replicaBytes: replicaBytes, holder: *obs.Claim.Holder} |
| 441 | // The files archive, rebuilt only when its plaintext digest changes. A |
| 442 | // file that changed under the walk ends the round here, before any |
| 443 | // report or readiness: the archive the target checked is no longer the |
| 444 | // files, and the next round builds the one that is (review finding 2). |
| 445 | why, retry, err := w.publishArchive(ctx, gm, rec, tgt.DataBucket, &ledger) |
| 446 | if err != nil { |
| 447 | return nil, nil, err |
| 448 | } |
| 449 | if err := saveLedger(); err != nil { |
| 450 | return nil, nil, err |
| 451 | } |
| 452 | if retry { |
| 453 | return nil, nil, nil |
| 454 | } |
| 455 | if why != "" { |
| 456 | sr.s1 = &moves.Check{Reason: why} |
| 457 | } |
| 458 | sr.ledger = ledger |
| 459 | return report(sr) |
| 460 | } |
| 461 | |
| 462 | // buildStateName is the build-state object's name beside a lease |
| 463 | // (release.BuildStateKey) and in a lane's state/ prefix. |
| 464 | const buildStateName = "build-state.json" |
| 465 | |
| 466 | // mirrorBuildState is the build-state lane (settings spec rev 11 §4.3a): |
| 467 | // the instance's build-state.json, beside the lease at location in |
| 468 | // srcBucket, mirrored to opID's lane state/ prefix in dstBucket, and |
| 469 | // nothing else. The destination must know which build last ran against |
| 470 | // this data. A source with no build state leaves none in the lane (the |
| 471 | // mirror deletes a stale copy) and answers a nil object. The preparing |
| 472 | // rounds, the cut's final round and a rollback's return lane all call it. |
| 473 | // Receipts live in srcBucket beside the replica lane's, under their own |
| 474 | // key. The object it answers is the lane copy by the version this round |
| 475 | // copied or vouched for (round.Versions, a versioned mirror only), for the |
| 476 | // seal: a later write by a stale round is never what a seal names (ruling |
| 477 | // R-SEAL-STATE). Its VersionID is "" on a mirror that records none. |
| 478 | func mirrorBuildState(ctx context.Context, m lease.MirrorStore, srcBucket, location, dstBucket, host, opID string) (lease.MirrorRound, *moves.SealedObject, error) { |
| 479 | src := strings.TrimSuffix(location, "/") + "/" |
| 480 | dst := moves.BuildStateKey(host, opID) |
| 481 | keep := func(rel string) bool { return rel == buildStateName } |
| 482 | round, err := lease.MirrorPrefixKeep(ctx, m, srcBucket, src, dstBucket, strings.TrimSuffix(dst, buildStateName), |
| 483 | lease.Receipts{Bucket: srcBucket, Key: moves.BuildStateReceiptsKey(host, opID)}, keep) |
| 484 | if err != nil { |
| 485 | return round, nil, err |
| 486 | } |
| 487 | listed, err := m.List(ctx, srcBucket, src+buildStateName) |
| 488 | if err != nil { |
| 489 | return round, nil, fmt.Errorf("list the build state: %w", err) |
| 490 | } |
| 491 | o, ok := listed[src+buildStateName] |
| 492 | if !ok { |
| 493 | return round, nil, nil |
| 494 | } |
| 495 | return round, &moves.SealedObject{Key: dst, VersionID: round.Versions[buildStateName], Size: o.Size}, nil |
| 496 | } |
| 497 | |
| 498 | // oneBoxRefusal is the one-box rule (D15), re-checked from the edges |
| 499 | // snapshot every round (ruling G13): either region publishing more than one |
| 500 | // box refuses with M-C21's sentence. It counts every box that publishes the |
| 501 | // placement, as T8 does for the target. |
| 502 | func oneBoxRefusal(fleet moveFleet, rec provision.Record) string { |
| 503 | for _, region := range []string{rec.Move.To, rec.Move.From} { |
| 504 | if len(fleet.boxes[region]) > 1 { |
| 505 | return s1ManyBoxes(region) |
| 506 | } |
| 507 | } |
| 508 | return "" |
| 509 | } |
| 510 | |
| 511 | // s1ManyBoxes is M-C21 as the source says it; the home region has no code |
| 512 | // of its own on a box. |
| 513 | func s1ManyBoxes(region string) string { |
| 514 | if region == "" { |
| 515 | region = "the home region" |
| 516 | } |
| 517 | return region + " has more than one box. Moves between regions with several boxes are not supported yet." |
| 518 | } |
| 519 | |
| 520 | // refreshCrossRate measures the cross-region path when no recent round did: |
| 521 | // it recopies the largest object of the lane (guarded, as every staging |
| 522 | // write) and times it. The recopy lands a new version of an object the lane |
| 523 | // already holds with the same bytes; the receipts keep naming the earlier |
| 524 | // version, which holds the same bytes too. |
| 525 | func (w *movesWorker) refreshCrossRate(ctx context.Context, gm *guardedMirror, srcPrefix string, keep func(string) bool, dstBucket, dstPrefix string, ledger *moveLedger) error { |
| 526 | src, err := gm.List(ctx, w.cfg.OwnBucket, srcPrefix) |
| 527 | if err != nil { |
| 528 | return fmt.Errorf("list the replica source: %w", err) |
| 529 | } |
| 530 | var key string |
| 531 | var obj lease.MirrorObject |
| 532 | for k, o := range src { |
| 533 | rel := strings.TrimPrefix(k, srcPrefix) |
| 534 | if keep(rel) && (o.Size > obj.Size || (o.Size == obj.Size && k < key) || key == "") { |
| 535 | key, obj = k, o |
| 536 | } |
| 537 | } |
| 538 | if key == "" || obj.Size <= 0 { |
| 539 | return nil // nothing to measure with yet |
| 540 | } |
| 541 | start := w.cfg.Now() |
| 542 | if _, _, err := gm.CopyVersioned(ctx, w.cfg.OwnBucket, key, obj.ETag, dstBucket, dstPrefix+strings.TrimPrefix(key, srcPrefix)); err != nil { |
| 543 | if errors.Is(err, lease.ErrCopySourceChanged) { |
| 544 | return nil // the next round copies it anyway, and times that |
| 545 | } |
| 546 | return fmt.Errorf("measure the cross-region copy: %w", err) |
| 547 | } |
| 548 | if d := w.cfg.Now().Sub(start); d > 0 { |
| 549 | ledger.LastRoundBytes, ledger.LastRoundSeconds, ledger.LastRoundAt = obj.Size, d.Seconds(), w.cfg.Now().UTC() |
| 550 | } |
| 551 | return nil |
| 552 | } |
| 553 | |
| 554 | // sourceRefusal is the box-step re-check (§6.2, §7.1, F14): the reason this |
| 555 | // source cannot move, or "". A registry or bucket read that fails is an |
| 556 | // error and ends the round. holder is the claim's, as the round read it |
| 557 | // under the staging lease. |
| 558 | func (w *movesWorker) sourceRefusal(ctx context.Context, s release.Store, rec provision.Record, holder *hostclaim.Holder) (string, error) { |
| 559 | cfg := w.cfg |
| 560 | if cfg.Route == nil { |
| 561 | return "", errors.New("no registry to re-check the source against") |
| 562 | } |
| 563 | rt, err := cfg.Route(rec.Host) |
| 564 | if err != nil { |
| 565 | return "", fmt.Errorf("read the registry row: %w", err) |
| 566 | } |
| 567 | if rt == nil { |
| 568 | return "", fmt.Errorf("%s has no registry row on this box", rec.Host) |
| 569 | } |
| 570 | if why := routeRefusal(*rt, cfg.DataDir); why != "" { |
| 571 | return why, nil |
| 572 | } |
| 573 | if fileExists(filepath.Join(cfg.DataDir, rec.Host+"-sidecar.env")) { |
| 574 | return s1Sidecar(rec.Host), nil |
| 575 | } |
| 576 | // The door's private-storage check (Task 4), repeated before the first |
| 577 | // staging write of every round. |
| 578 | if env := provision.InstanceStoreEnv(&rec); env == "" || appconfig.ValidEnvironmentName(env) { |
| 579 | store, _, err := objstore.GetIn(ctx, s, rec.Account, rec.App, env) |
| 580 | if err != nil { |
| 581 | return "", fmt.Errorf("read the instance's storage binding: %w", err) |
| 582 | } |
| 583 | if store != nil { |
| 584 | return s1PrivateStorage(rec.Host), nil |
| 585 | } |
| 586 | } |
| 587 | if cfg.SecretsPub == nil || cfg.SecretsKey == "" { |
| 588 | return s1NoSecretsKey, nil |
| 589 | } |
| 590 | return w.sourceFilesRefusal(ctx, rec, holder) |
| 591 | } |
| 592 | |
| 593 | // movesFilesPreflightTimeout bounds sourceFilesRefusal's reads together: |
| 594 | // creating the versioned reader, listing the manifest key's versions for |
| 595 | // each generation of the chain, and downloading the selected manifest. A |
| 596 | // files manifest is one JSON object, small next to what a round copies, so |
| 597 | // a healthy read takes well under a second; 30 seconds absorbs a slow |
| 598 | // bucket, is a quarter of the claim's own request limit |
| 599 | // (hostbox.RequestLimit), and is far inside the 45-minute staging lease the |
| 600 | // round holds (moves.StagingLeaseTTL). Past it the round ends with an |
| 601 | // error and no report, and the next pass tries again. |
| 602 | const movesFilesPreflightTimeout = 30 * time.Second |
| 603 | |
| 604 | // movesFilesPreflightBound is the bound in force: movesFilesPreflightTimeout, |
| 605 | // a variable only so a test can watch a stalled read end without waiting |
| 606 | // the full 30 seconds. |
| 607 | var movesFilesPreflightBound = movesFilesPreflightTimeout |
| 608 | |
| 609 | // sourceFilesRefusal is the cut's files preflight (#236; Paul's ruling of |
| 610 | // 2026-10-15): the files manifest a recovery of this host would restore |
| 611 | // must pass the readers' path cap, or the move never reaches its cut. A |
| 612 | // manifest written before the cap (a8cebad) can hold a path past it, and |
| 613 | // every reader refuses that manifest terminally: a recovery that selected |
| 614 | // it would fail its establishment and leave the source offline after the |
| 615 | // cut. It runs in every preparing round and in the cut's requested round |
| 616 | // (G13), before any staging write, stop or cut. |
| 617 | // |
| 618 | // The manifest is the one the establishment selects (movesFilesSelect): |
| 619 | // the newest manifest of the holder's admitted chain in this region. That |
| 620 | // is the one a revert restores (its holder never leaves the source), and |
| 621 | // the one a clean return's src_gen bridge restores, since the flip |
| 622 | // journals the holder's generation as src_gen and every later write of a |
| 623 | // manifest is held to the cap by Commit. A dirty return restores from a |
| 624 | // prepared manifest its install builds through the capped writers, so |
| 625 | // nothing published before the cap reaches it. The rule is the |
| 626 | // establishment's own (movesFilesRestorable). A host with no files |
| 627 | // manifest has nothing to check here. A read that fails ends the round. |
| 628 | // |
| 629 | // It is deny-only (census ruling moves-overcap-cut-236, platform-1e, |
| 630 | // 2026-10-15; binding): it can only refuse a move, never allow anything. |
| 631 | // Passing it grants nothing, and it never skips or short-circuits the |
| 632 | // establishment's own check of the same manifest, which reads through the |
| 633 | // claim's identity-bound store (hostbox.RequestStore) and applies |
| 634 | // movesFilesRestorable and ValidateManifest again. Its reads go through |
| 635 | // cfg.Versioned, the moves worker's own reader, and are bounded by |
| 636 | // movesFilesPreflightTimeout. A change that lets it authorize anything |
| 637 | // needs a new census ruling. |
| 638 | func (w *movesWorker) sourceFilesRefusal(ctx context.Context, rec provision.Record, holder *hostclaim.Holder) (string, error) { |
| 639 | cfg := w.cfg |
| 640 | if holder == nil || holder.Location == "" { |
| 641 | return "", nil // no location, no files manifest (a legacy holder) |
| 642 | } |
| 643 | if cfg.Versioned == nil { |
| 644 | return "", errors.New("no versioned reader to check the files manifest with") |
| 645 | } |
| 646 | // The round runs under the staging lease and the pass mutex every moves |
| 647 | // pass shares, and the instances loop calls it with no deadline: a |
| 648 | // stalled listing or download must end the round, not hold both. |
| 649 | ctx, cancel := context.WithTimeout(ctx, movesFilesPreflightBound) |
| 650 | defer cancel() |
| 651 | vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion) |
| 652 | if err != nil { |
| 653 | return "", fmt.Errorf("the data bucket, to check the files manifest: %w", err) |
| 654 | } |
| 655 | m, key, version, found, err := movesFilesSelect(ctx, vs, cfg.OwnBucket, holder.Location, movesFilesChain(holder, cfg.ParkHomeRegion, cfg.parkRegion())) |
| 656 | if err != nil { |
| 657 | return "", fmt.Errorf("read the files manifest: %w", err) |
| 658 | } |
| 659 | if !found { |
| 660 | return "", nil |
| 661 | } |
| 662 | if err := movesFilesRestorable(m); err != nil { |
| 663 | if !errors.Is(err, hostfiles.ErrTooDeep) { |
| 664 | return "", fmt.Errorf("the files manifest %s@%s: %w", key, version, err) |
| 665 | } |
| 666 | log.Printf("moves: %s: the files manifest %s@%s cannot be restored (%v); an operator must replace it", rec.Host, key, version, err) |
| 667 | return s1FilesPastCap(rec.Host), nil |
| 668 | } |
| 669 | return "", nil |
| 670 | } |
| 671 | |
| 672 | // routeRefusal is the registry half of the box-step re-check. Backing is |
| 673 | // refused whenever it is set: §6.2 names only Backing (unit), and the plan |
| 674 | // refuses any non-empty value, which is the stricter reading (F14). |
| 675 | func routeRefusal(rt registry.Route, dataDir string) string { |
| 676 | switch { |
| 677 | case rt.Kind != "instance" || rt.Platform: |
| 678 | return s1NotExec(rt.Host) |
| 679 | case !rt.Hibernate: |
| 680 | return s1NoHibernate(rt.Host) |
| 681 | case rt.DBPath != provision.DefaultDBPath(dataDir, rt.Host): |
| 682 | return s1Nested |
| 683 | } |
| 684 | for _, c := range []struct { |
| 685 | name string |
| 686 | set bool |
| 687 | }{ |
| 688 | {"caps", rt.Caps != ""}, {"unit", rt.Unit != ""}, {"backing", rt.Backing != ""}, |
| 689 | {"tunnel", rt.Tunnel != ""}, {"fleet_box", rt.FleetBox != ""}, {"pool", rt.Pool != ""}, |
| 690 | {"canary", rt.Canary}, {"addr", rt.Addr != ""}, {"addrs", rt.Addrs != ""}, {"health", rt.Health != ""}, |
| 691 | {"replica", rt.Replica}, {"replicas", rt.Replicas != ""}, |
| 692 | } { |
| 693 | if c.set { |
| 694 | return s1Column(rt.Host, c.name) |
| 695 | } |
| 696 | } |
| 697 | return "" |
| 698 | } |
| 699 | |
| 700 | func fileExists(path string) bool { |
| 701 | _, err := os.Lstat(path) |
| 702 | return err == nil |
| 703 | } |
| 704 | |
| 705 | func containsString(xs []string, x string) bool { |
| 706 | for _, v := range xs { |
| 707 | if v == x { |
| 708 | return true |
| 709 | } |
| 710 | } |
| 711 | return false |
| 712 | } |
| 713 | |
| 714 | // publishArchive builds, seals and stages the files archive when its |
| 715 | // plaintext digest differs from the one the ledger published, and records |
| 716 | // the new one. It returns an S1 reason for an archive this move can never |
| 717 | // carry, or retry for a file that changed while it was read: the round |
| 718 | // then ends, and the next one tries again. |
| 719 | func (w *movesWorker) publishArchive(ctx context.Context, gm *guardedMirror, rec provision.Record, dstBucket string, ledger *moveLedger) (why string, retry bool, err error) { |
| 720 | cfg := w.cfg |
| 721 | dir, envPath := cfg.FilesDir(rec.Host) |
| 722 | plain, _, _, err := archivePlainDigest(dir, envPath) |
| 723 | if why, retry, ferr := archiveFailure(rec.Host, err); ferr != nil || why != "" || retry { |
| 724 | return why, retry, ferr |
| 725 | } |
| 726 | if plain == ledger.ArchivePlain && ledger.ArchiveVersions != nil { |
| 727 | return "", false, nil |
| 728 | } |
| 729 | a, why, retry, err := w.buildArchive(rec) |
| 730 | if err != nil || why != "" || retry { |
| 731 | return why, retry, err |
| 732 | } |
| 733 | ref, err := a.stage(ctx, gm, dstBucket, rec.Host, rec.Move.ID) |
| 734 | if err != nil { |
| 735 | return "", false, err |
| 736 | } |
| 737 | ledger.ArchivePlain, ledger.ArchiveSealed, ledger.ArchiveVersions = a.Plain, a.Sealed, ref |
| 738 | log.Printf("moves: %s: staged the files archive %s (%d files, %d bytes)", rec.Host, a.Plain, a.Entries, a.Bytes) |
| 739 | return "", false, nil |
| 740 | } |
| 741 | |
| 742 | // builtArchive is one built and sealed archive on the data disk, in the |
| 743 | // move's archive directory (archiveDir): <Name>.sealed and <Name>.json. |
| 744 | // The preparing round keeps the newest one there, so the cut can write the |
| 745 | // very object the target verified into move-snapshots/ (ruling G14). |
| 746 | type builtArchive struct { |
| 747 | Name string `json:"name"` // files-<sealed hex> |
| 748 | Plain string `json:"plain"` // the plaintext digest |
| 749 | Sealed string `json:"sealed"` // the sealed object's sha256 |
| 750 | Bytes int64 `json:"bytes"` // plaintext bytes |
| 751 | Entries int `json:"entries"` |
| 752 | SealedSize int64 `json:"sealed_size"` |
| 753 | dir string |
| 754 | } |
| 755 | |
| 756 | func (a builtArchive) sealedPath() string { return filepath.Join(a.dir, a.Name+".sealed") } |
| 757 | func (a builtArchive) manifestPath() string { return filepath.Join(a.dir, a.Name+".json") } |
| 758 | |
| 759 | // archiveDir is where a move's built archive lives on the data disk, under |
| 760 | // its journal directory: never a shared tmpfs. |
| 761 | func archiveDir(dataDir, host, op string) string { |
| 762 | return filepath.Join(moves.JournalDir(dataDir, host, op), "archive") |
| 763 | } |
| 764 | |
| 765 | // archiveName is the object name for a sealed digest. |
| 766 | func archiveName(sealed string) string { return "files-" + digestHex(sealed) } |
| 767 | |
| 768 | // buildArchive builds and seals the files archive into the move's archive |
| 769 | // directory, streamed to disk so a 256 MiB archive is never held in |
| 770 | // memory, and removes any older one there. Sealing is randomized (§8.2): |
| 771 | // every build is a new object with a new name. It returns an S1 reason for |
| 772 | // an archive this move can never carry, or retry for a file that changed |
| 773 | // while it was read. |
| 774 | func (w *movesWorker) buildArchive(rec provision.Record) (a builtArchive, why string, retry bool, err error) { |
| 775 | cfg := w.cfg |
| 776 | dir, envPath := cfg.FilesDir(rec.Host) |
| 777 | adir := archiveDir(cfg.DataDir, rec.Host, rec.Move.ID) |
| 778 | if err := os.MkdirAll(adir, 0o700); err != nil { |
| 779 | return a, "", false, err |
| 780 | } |
| 781 | tmp, err := os.MkdirTemp(adir, "build-") |
| 782 | if err != nil { |
| 783 | return a, "", false, err |
| 784 | } |
| 785 | defer os.RemoveAll(tmp) |
| 786 | built, err := os.Create(filepath.Join(tmp, "archive.tar.zst")) |
| 787 | if err != nil { |
| 788 | return a, "", false, err |
| 789 | } |
| 790 | defer built.Close() |
| 791 | plain, entries, bytes, err := archiveBuild(dir, envPath, built) |
| 792 | if why, retry, ferr := archiveFailure(rec.Host, err); ferr != nil || why != "" || retry { |
| 793 | return a, why, retry, ferr |
| 794 | } |
| 795 | if _, err := built.Seek(0, io.SeekStart); err != nil { |
| 796 | return a, "", false, err |
| 797 | } |
| 798 | sealedTmp := filepath.Join(tmp, "archive.sealed") |
| 799 | sealed, err := os.Create(sealedTmp) |
| 800 | if err != nil { |
| 801 | return a, "", false, err |
| 802 | } |
| 803 | defer sealed.Close() |
| 804 | // Seal hashes what it writes: sealedSHA is the sealed file's sha256, |
| 805 | // the payload hash its upload is signed with. |
| 806 | dataKey, sealedSHA, err := archive.Seal(built, cfg.SecretsPub, appconfig.Binding{Account: rec.Account, App: rec.App}, sealed) |
| 807 | if err != nil { |
| 808 | return a, "", false, fmt.Errorf("seal the archive: %w", err) |
| 809 | } |
| 810 | if err := sealed.Sync(); err != nil { |
| 811 | return a, "", false, err |
| 812 | } |
| 813 | st, err := sealed.Stat() |
| 814 | if err != nil { |
| 815 | return a, "", false, err |
| 816 | } |
| 817 | m := archive.Manifest{V: archive.ManifestVersion, Op: rec.Move.ID, Incarnation: rec.Move.Incarnation, Plain: plain, |
| 818 | Sealed: sealedSHA, Bytes: bytes, Entries: entries, SealedDataKey: dataKey, Chunk: archive.ChunkSize} |
| 819 | manifest, err := json.Marshal(m) |
| 820 | if err != nil { |
| 821 | return a, "", false, err |
| 822 | } |
| 823 | a = builtArchive{Name: archiveName(sealedSHA), Plain: plain, Sealed: sealedSHA, Bytes: bytes, Entries: len(entries), SealedSize: st.Size(), dir: adir} |
| 824 | if err := writeFileSync(filepath.Join(tmp, "manifest.json"), manifest); err != nil { |
| 825 | return a, "", false, err |
| 826 | } |
| 827 | if err := os.Rename(sealedTmp, a.sealedPath()); err != nil { |
| 828 | return a, "", false, err |
| 829 | } |
| 830 | if err := os.Rename(filepath.Join(tmp, "manifest.json"), a.manifestPath()); err != nil { |
| 831 | return a, "", false, err |
| 832 | } |
| 833 | if err := syncDir(adir); err != nil { |
| 834 | return a, "", false, err |
| 835 | } |
| 836 | pruneArchives(adir, a.Name) |
| 837 | return a, "", false, nil |
| 838 | } |
| 839 | |
| 840 | // stage puts the archive's two objects into the staging lane's files/ |
| 841 | // through the guard (the sealed object streamed from disk) and returns |
| 842 | // them by the versions the PUTs answered. |
| 843 | func (a builtArchive) stage(ctx context.Context, gm *guardedMirror, dstBucket, host, op string) (*moves.ArchiveRef, error) { |
| 844 | manifest, err := os.ReadFile(a.manifestPath()) |
| 845 | if err != nil { |
| 846 | return nil, err |
| 847 | } |
| 848 | prefix := moves.FilesPrefix(host, op) |
| 849 | ref := &moves.ArchiveRef{ManifestKey: prefix + a.Name + ".json", SealedObjectKey: prefix + a.Name + ".sealed", Plain: a.Plain} |
| 850 | if ref.SealedVersion, err = gm.PutFileVersioned(ctx, dstBucket, ref.SealedObjectKey, a.sealedPath(), a.SealedSize, digestHex(a.Sealed)); err != nil { |
| 851 | return nil, fmt.Errorf("stage the sealed archive: %w", err) |
| 852 | } |
| 853 | if ref.ManifestVersion, err = gm.PutVersioned(ctx, dstBucket, ref.ManifestKey, manifest); err != nil { |
| 854 | return nil, fmt.Errorf("stage the archive manifest: %w", err) |
| 855 | } |
| 856 | return ref, nil |
| 857 | } |
| 858 | |
| 859 | // pruneArchives removes every archive in dir but keep. Best effort: a file |
| 860 | // left behind costs disk until the move's directory is removed. |
| 861 | func pruneArchives(dir, keep string) { |
| 862 | ents, err := os.ReadDir(dir) |
| 863 | if err != nil { |
| 864 | return |
| 865 | } |
| 866 | for _, e := range ents { |
| 867 | name := e.Name() |
| 868 | if strings.HasPrefix(name, "files-") && strings.TrimSuffix(strings.TrimSuffix(name, ".sealed"), ".json") != keep { |
| 869 | os.Remove(filepath.Join(dir, name)) |
| 870 | } |
| 871 | } |
| 872 | } |
| 873 | |
| 874 | // writeFileSync writes body to path and syncs it. |
| 875 | func writeFileSync(path string, body []byte) error { |
| 876 | f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) |
| 877 | if err != nil { |
| 878 | return err |
| 879 | } |
| 880 | if _, err := f.Write(body); err != nil { |
| 881 | f.Close() |
| 882 | return err |
| 883 | } |
| 884 | if err := f.Sync(); err != nil { |
| 885 | f.Close() |
| 886 | return err |
| 887 | } |
| 888 | return f.Close() |
| 889 | } |
| 890 | |
| 891 | // archiveFailure maps an archive error to an S1 reason (a standing refusal |
| 892 | // the operator must act on), a retry (a file changed under the walk), or |
| 893 | // nothing. The reason is always a recorded sentence chosen by errors.Is, |
| 894 | // with a path only from the *archive.PathError the error carries; the |
| 895 | // error's own text, which can name anything the walk met, is logged and |
| 896 | // never published (ruling T15-sinks). An error this does not recognise |
| 897 | // gets s1ArchiveFailed. |
| 898 | func archiveFailure(host string, err error) (why string, retry bool, ferr error) { |
| 899 | if err == nil { |
| 900 | return "", false, nil |
| 901 | } |
| 902 | if errors.Is(err, archive.ErrChanged) { |
| 903 | log.Printf("moves: %s: %v; the archive is rebuilt next round", host, err) |
| 904 | return "", true, nil |
| 905 | } |
| 906 | log.Printf("moves: %s: archive the files: %v", host, err) |
| 907 | var pe *archive.PathError |
| 908 | hasPath := errors.As(err, &pe) && pe.Path != "" |
| 909 | switch { |
| 910 | case errors.Is(err, archive.ErrSpecialFile) && hasPath: |
| 911 | return s1SpecialFile(host, pe.Path), false, nil // M-C14, hard links too |
| 912 | case errors.Is(err, archive.ErrOverCap): |
| 913 | return s1OverCap(host), false, nil // M-C15 |
| 914 | case errors.Is(err, archive.ErrPastTreeCap): |
| 915 | return s1FilePastCap(host), false, nil // MV-N502 |
| 916 | case errors.Is(err, archive.ErrTooManyEntries): |
| 917 | return s1TooManyEntries, false, nil // MV-N79 |
| 918 | case errors.Is(err, archive.ErrBadPath) && hasPath: |
| 919 | return s1LongPath(pe.Path), false, nil // MV-N81 |
| 920 | } |
| 921 | return s1ArchiveFailed, false, nil |
| 922 | } |
| 923 | |
| 924 | // digestHex is "sha256:<hex>" without its algorithm, for an object name. |
| 925 | func digestHex(d string) string { |
| 926 | _, h, ok := strings.Cut(d, ":") |
| 927 | if !ok { |
| 928 | return d |
| 929 | } |
| 930 | if _, err := hex.DecodeString(h); err != nil { |
| 931 | return d |
| 932 | } |
| 933 | return h |
| 934 | } |
| 935 | |
| 936 | // sourcePassed is every check a source report carries: S1 and S2. |
| 937 | func sourcePassed(r *moves.SourceReport) bool { return r != nil && r.S1.OK && r.S2.OK } |
| 938 | |
| 939 | // writeSourceReport publishes the source's readiness report (§7.3): the |
| 940 | // launch environment as HMACs, S1 (a refusal, or the instance's own answer |
| 941 | // to its readiness path), S2, the published archive, the live release and |
| 942 | // the cross-region rate. It is replaced every round. |
| 943 | func (w *movesWorker) writeSourceReport(ctx context.Context, s release.Store, rec provision.Record, r *srcRound) (*moves.SourceReport, error) { |
| 944 | cfg := w.cfg |
| 945 | now := cfg.Now().UTC() |
| 946 | sr := moves.SourceReport{V: 1, Box: cfg.Box, MoveID: rec.Move.ID, At: now, |
| 947 | FilesPlain: r.ledger.ArchivePlain, Archive: r.ledger.ArchiveVersions} |
| 948 | sr.CrossRate, sr.CrossRateAt = r.ledger.crossRate() |
| 949 | if r.s1 != nil { |
| 950 | sr.S1 = *r.s1 |
| 951 | } else { |
| 952 | sr.S1 = w.probeSource(ctx, s, rec) |
| 953 | } |
| 954 | sr.S2 = moves.Check{Reason: "the next spawn of this host is not strict"} |
| 955 | if cfg.StrictNext != nil { |
| 956 | sr.S2 = cfg.StrictNext(rec.Host) |
| 957 | // NextSpawnStrict prefixes its reasons with "S2: ", which Ready adds. |
| 958 | sr.S2.Reason = strings.TrimPrefix(sr.S2.Reason, "S2: ") |
| 959 | } |
| 960 | if st, err := os.Stat(provision.DefaultDBPath(cfg.DataDir, rec.Host)); err == nil && st.Mode().IsRegular() { |
| 961 | sr.DBBytes = st.Size() |
| 962 | } |
| 963 | sr.ReplicaBytes = r.replicaBytes |
| 964 | if cfg.OptDir != "" { |
| 965 | if v, err := release.LiveVersion(cfg.OptDir, rec.Host); err == nil { |
| 966 | sr.Release = v |
| 967 | } else { |
| 968 | log.Printf("moves: %s: live version: %v", rec.Host, err) |
| 969 | } |
| 970 | } |
| 971 | if r.s1 == nil && cfg.LaunchEnv != nil { |
| 972 | key, err := moves.ReportKeyFromSecrets(cfg.SecretsKey) |
| 973 | if err != nil { |
| 974 | sr.S1 = moves.Check{Reason: s1NoSecretsKey} |
| 975 | } else if env, instance, err := cfg.LaunchEnv(rec.Host); err != nil { |
| 976 | return nil, fmt.Errorf("compose the launch environment: %w", err) |
| 977 | } else { |
| 978 | sr.Env = moves.ReportLaunch(key, env, instance) |
| 979 | } |
| 980 | } |
| 981 | body, err := json.Marshal(sr) |
| 982 | if err != nil { |
| 983 | return nil, err |
| 984 | } |
| 985 | if err := moves.PutReplace(ctx, s, moves.SourceReportKey(moves.ReportPrefix(rec.Account, rec.App, rec.Host, rec.Move.ID)), body); err != nil { |
| 986 | return nil, err |
| 987 | } |
| 988 | return &sr, nil |
| 989 | } |
| 990 | |
| 991 | // probeSource is S1: the instance's own answer to its readiness path, |
| 992 | // woken if it sleeps, never from the response cache. |
| 993 | func (w *movesWorker) probeSource(ctx context.Context, s release.Store, rec provision.Record) moves.Check { |
| 994 | if w.cfg.ReadyProbe == nil { |
| 995 | return moves.Check{Reason: s1CannotProbe} |
| 996 | } |
| 997 | // "/" only when the manifest was read and declares no path: an app |
| 998 | // whose declared path fails must not pass on a healthy homepage while |
| 999 | // its manifest cannot be read (review finding 5). |
| 1000 | m, err := provision.GetManifest(ctx, s, rec.Account, rec.App) |
| 1001 | if err != nil { |
| 1002 | log.Printf("moves: %s: S1: instances manifest: %v", rec.Host, err) |
| 1003 | return moves.Check{Reason: s1NoReadyPath} |
| 1004 | } |
| 1005 | return w.cfg.ReadyProbe(ctx, rec.Host, m.ReadyPathOrDefault()) |
| 1006 | } |
| 1007 | |
| 1008 | // decideReady reads the target's report and applies §7.3's rule to it and |
| 1009 | // to source, the report this round just published (and that passed); on |
| 1010 | // ready from preparing it pins the inputs and promotes, and an automatic |
| 1011 | // cut then marks the cut requested (§7.4). The caller reaches it only for a |
| 1012 | // round whose own report passed, so a failing round never requests a cut. |
| 1013 | // Every CAS is guarded by the op identity the round read: a replacement |
| 1014 | // move is ErrMoveConflict and the round ends. |
| 1015 | func (w *movesWorker) decideReady(ctx context.Context, s release.Store, rec provision.Record, g provision.MoveGuard, r *srcRound, source *moves.SourceReport) error { |
| 1016 | cfg := w.cfg |
| 1017 | state := rec.Move.State |
| 1018 | // A ready move with no pin is one a cut's admission sent back (G1): the |
| 1019 | // same rule pins it again, from this passing round, before any cut. |
| 1020 | repin := state == provision.MoveReady && rec.Move.Inputs == nil |
| 1021 | // Ruling T5-lease (fix round 2): a holder whose location is not the |
| 1022 | // activator's lease prefix cannot be cut, so a ready move on one is |
| 1023 | // neither pinned again nor cut automatically. Nothing is written: the |
| 1024 | // door refuses such a move at request time and the cut's admission |
| 1025 | // refuses a replay, so a move that got here waits, with a notice, |
| 1026 | // until an operator aborts it. |
| 1027 | // (Its promotion to ready still happens, so the operator sees it there.) |
| 1028 | _, layoutErr := writerLease(cfg.OwnPrefix, rec.Host, r.holder) |
| 1029 | if layoutErr != nil && (repin || rec.Move.Cut == "auto") { |
| 1030 | w.notice(rec.Host+"#layout", "moves: %s: move %s cannot be cut (%v); no repin and no automatic cut", rec.Host, g.ID, layoutErr) |
| 1031 | if repin { |
| 1032 | return nil |
| 1033 | } |
| 1034 | } |
| 1035 | // The cut guard (Paul's ruling, 2026-10-11): a cut this box refused |
| 1036 | // because it cannot recover the move (cutRun.refuseUnarmed) is not |
| 1037 | // pinned again while that holds, so an automatic cut does not loop |
| 1038 | // through the refusal. The first cut is still requested, and its |
| 1039 | // refusal records the reason the operator reads. A target that cannot |
| 1040 | // land fails T13, so Ready below keeps it from pinning. |
| 1041 | if repin && !recoverySourceArmed(cfg) { |
| 1042 | w.notice(rec.Host+"#unarmed", "moves: %s: move %s is not pinned again: %s", rec.Host, g.ID, provision.RefusedSourceUnarmed) |
| 1043 | return nil |
| 1044 | } |
| 1045 | if state == provision.MovePreparing || repin { |
| 1046 | target, err := readTargetReport(ctx, s, rec) |
| 1047 | if err != nil { |
| 1048 | return err |
| 1049 | } |
| 1050 | ok, why := moves.Ready(r.deltaBytes, cfg.ReadyDelta, target, source, cfg.Now()) |
| 1051 | if !ok { |
| 1052 | w.notice(rec.Host+"#ready", "moves: %s: not ready: %s", rec.Host, why) |
| 1053 | return nil |
| 1054 | } |
| 1055 | // moves.Ready has required target.Inputs.Release to equal the |
| 1056 | // source's, so what is pinned is what T7 checked (review finding 3). |
| 1057 | in := target.Inputs |
| 1058 | mut := provision.MoveGuardedPromoteReady(g, cfg.Now(), in) |
| 1059 | if repin { |
| 1060 | mut = provision.MoveGuardedRepin(g, in) |
| 1061 | } |
| 1062 | var pre provision.Record |
| 1063 | after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, mut)) |
| 1064 | if err != nil { |
| 1065 | if errors.Is(err, provision.ErrMoveConflict) { |
| 1066 | log.Printf("moves: %s: the record no longer holds move %s as %s without a pin (%v); this round ends", rec.Host, g.ID, state, err) |
| 1067 | return nil |
| 1068 | } |
| 1069 | return err |
| 1070 | } |
| 1071 | w.logTransition(ctx, s, pre, *after) |
| 1072 | nudgeSerial(ctx, s) |
| 1073 | if repin { |
| 1074 | log.Printf("moves: %s: move %s pinned its inputs again", rec.Host, g.ID) |
| 1075 | } else { |
| 1076 | log.Printf("moves: %s: move %s is ready (last round moved %d bytes; bar %d)", rec.Host, g.ID, r.deltaBytes, cfg.ReadyDelta) |
| 1077 | } |
| 1078 | state = provision.MoveReady |
| 1079 | } |
| 1080 | if state == provision.MoveReady && rec.Move.Cut == "auto" && layoutErr == nil { |
| 1081 | var pre provision.Record |
| 1082 | after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, provision.MoveGuardedCut(g))) |
| 1083 | if err != nil { |
| 1084 | if errors.Is(err, provision.ErrMoveConflict) { |
| 1085 | log.Printf("moves: %s: the automatic cut of %s was refused (%v)", rec.Host, g.ID, err) |
| 1086 | return nil |
| 1087 | } |
| 1088 | return err |
| 1089 | } |
| 1090 | w.logTransition(ctx, s, pre, *after) |
| 1091 | nudgeSerial(ctx, s) |
| 1092 | log.Printf("moves: %s: move %s cut requested automatically", rec.Host, g.ID) |
| 1093 | } |
| 1094 | return nil |
| 1095 | } |
| 1096 | |
| 1097 | func readSourceReport(ctx context.Context, s release.Store, rec provision.Record) (*moves.SourceReport, error) { |
| 1098 | var r moves.SourceReport |
| 1099 | ok, err := readReport(ctx, s, moves.SourceReportKey(moves.ReportPrefix(rec.Account, rec.App, rec.Host, rec.Move.ID)), &r) |
| 1100 | if !ok { |
| 1101 | return nil, err |
| 1102 | } |
| 1103 | return &r, nil |
| 1104 | } |
| 1105 | |
| 1106 | func readTargetReport(ctx context.Context, s release.Store, rec provision.Record) (*moves.TargetReport, error) { |
| 1107 | var r moves.TargetReport |
| 1108 | ok, err := readReport(ctx, s, moves.TargetReportKey(moves.ReportPrefix(rec.Account, rec.App, rec.Host, rec.Move.ID)), &r) |
| 1109 | if !ok { |
| 1110 | return nil, err |
| 1111 | } |
| 1112 | return &r, nil |
| 1113 | } |
| 1114 | |
| 1115 | // readReport reads a report: (false, nil) when absent or unreadable as a |
| 1116 | // report (Ready then says no report), an error when the read failed. |
| 1117 | func readReport(ctx context.Context, s release.Store, key string, into any) (bool, error) { |
| 1118 | body, _, err := s.Get(ctx, key) |
| 1119 | if errors.Is(err, os.ErrNotExist) { |
| 1120 | return false, nil |
| 1121 | } |
| 1122 | if err != nil { |
| 1123 | return false, fmt.Errorf("read %s: %w", key, err) |
| 1124 | } |
| 1125 | if err := json.Unmarshal(body, into); err != nil { |
| 1126 | log.Printf("moves: %s does not parse: %v", key, err) |
| 1127 | return false, nil |
| 1128 | } |
| 1129 | return true, nil |
| 1130 | } |
| 1131 | |
amadan