amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "crypto/ecdh"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "github.com/carlosframework/platform/internal/park"
10 "io"
11 "io/fs"
12 "log"
13 "os"
14 "path/filepath"
15 "slices"
16 "strings"
17 "time"
18
19 "github.com/carlosframework/platform/internal/appconfig"
20 "github.com/carlosframework/platform/internal/hostclaim"
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/provision"
25 "github.com/carlosframework/platform/internal/release"
26 "github.com/superfly/ltx"
27)
28
29// The target's readiness report (moves spec §7.3, T1 to T12, and the cut
30// guard's T13), behind
31// CARLOS_MOVES_PHASE1. The named target box (the claim's op.intent.target)
32// writes it every pass for every move into its placement that is preparing
33// or ready; no other box writes one. Each check is its own small function
34// returning a moves.Check. The reasons are rows of the copy record, pending
35// review; they reach `carlos move status` through the report.
36const (
37 t1FloorUnreadable = "the moves protocol floor could not be read"
38 t1LocalUnreadable = "this box cannot read its own copy of the moves protocol floor"
39 t1NotArmed = "this box has not armed the moves protocol floor yet"
40 t2NoConsole = "the console has not published its secrets key fingerprint yet"
41 t2NoKey = "this box has no secrets key"
42 t2Differs = "this box's secrets key is not the console's"
43 t3NoSource = "the source has not published a readiness report yet" // MV-N62's text
44 t3NoCompose = "this box could not compose the launch environment of the held row"
45 // t3OldReport is MV-N513: the source's env report is version 1, whose
46 // pin cannot tell an instance's box variable from the box's
47 // (moves.EnvReportVersion).
48 t3OldReport = "the source reported its launch environment in an older format"
49 t5NoArchive = "the source has published no archive yet" // MV-N73
50 t5Unreadable = "this box could not read the archive the source published"
51 t5Refused = "the archive does not open with this box's key or does not match its manifest"
52 t5NotPinned = "the source staged its archive without a fixed version id"
53 t6Unmeasured = "this box could not measure its free disk space"
54 t7NoRelease = "the source has not reported its release yet" // MV-N76
55 t7Waiting = "waiting for the root stage pass" // MV-N75
56 t7Failed = "this box could not stage the source's release"
57 t7NotProven = "this box requires a proof for the source's release and has not accepted one"
58 t7NotStaged = "the release is not staged on this box yet" // MV-N455: T7 under the claim, the version store not writable (heldReleaseStaged)
59 t9NotThisMove = "the host claim is not held by this move"
60 t9Incarnation = "the host claim names another incarnation of this instance"
61 t9NoMoveJSON = "the staging lane has no move.json yet"
62 t9OtherMove = "the staging lane holds another move's move.json" // MV-N9, aligned
63 t10NoBucket = "the source region publishes no data bucket"
64 t10Failed = "this box cannot write the return lane in the source's bucket"
65 t11NoReplica = "no replica has been staged yet"
66 t11Unmeasured = "the database size could not be measured" // MV-N69
67 t13NotArmed = "this box cannot land the move or roll it back yet"
68)
69
70func t1Below(min int) string {
71 return fmt.Sprintf("this box speaks moves protocol %d, below the floor's %d", moves.ProtocolVersion, min)
72}
73func t3Differs(names []string) string {
74 return "the launch environment differs from the source's: " + strings.Join(names, ", ")
75}
76func t3Shadowed(name string) string {
77 return name + " in the carried .env overrides the rendered configuration"
78}
79func t4Failed(prefix string) string {
80 return "the store bound as " + prefix + " did not answer the probe from this box"
81}
82func t6NoRoom(needGiB float64) string {
83 return fmt.Sprintf("this box needs %.1f GiB free for the move without filling its disk past 85%%", needGiB)
84}
85func t8Count(n int, to string) string {
86 if n == 0 {
87 return "no box publishes " + to
88 }
89 return fmt.Sprintf("%d boxes publish %s, and a move needs exactly one", n, to)
90}
91func t9Untouched(what string) string { return "the host claim has not touched " + what + " yet" } // MV-N74
92func t12NotQuiet(region string, until time.Time) string {
93 return "the instance left " + region + " recently and can move back after " + until.UTC().Format(time.RFC3339)
94}
95
96// moveFleet is the fleet as the edges files publish it: per placement, the
97// box whose coordinates a move uses (the lexically first, as moveTargets
98// picks) and every box that publishes it (T8). The home publishes no
99// placement code, so it is kept under "", where a move out of the home
100// finds its source bucket (T10).
101type moveFleet struct {
102 targets map[string]edgesFile
103 boxes map[string][]string
104}
105
106func loadMoveFleet(ctx context.Context, s release.Store) (moveFleet, error) {
107 bodies, err := loadEdgesFileBodies(ctx, s)
108 if err != nil {
109 return moveFleet{}, err
110 }
111 f := moveFleet{targets: map[string]edgesFile{}, boxes: map[string][]string{}}
112 for _, body := range bodies {
113 var t edgesFile
114 if err := json.Unmarshal(body, &t); err != nil || t.V < 1 {
115 continue
116 }
117 if !containsString(f.boxes[t.Placement], t.Box) {
118 f.boxes[t.Placement] = append(f.boxes[t.Placement], t.Box)
119 }
120 if t.DataBucket == "" {
121 continue
122 }
123 if prev, dup := f.targets[t.Placement]; dup && prev.Box <= t.Box {
124 continue
125 }
126 f.targets[t.Placement] = t
127 }
128 return f, nil
129}
130
131// targetReportPass writes the target report for every move into this box's
132// placement whose claim names this box as the target.
133func (w *movesWorker) targetReportPass(ctx context.Context, s release.Store, recs []provision.Record, fleet moveFleet) {
134 live := map[string]bool{}
135 defer w.evictArchives(live)
136 defer w.retireHeldRefusals(live)
137 for _, rec := range recs {
138 m := rec.Move
139 if m == nil || m.Legacy() || m.To != w.cfg.Region {
140 continue
141 }
142 // A move into this box is live from preparing through the cut and
143 // the landing (ruling G26): the held refusal a landing's configured
144 // phase writes, and T5's verdict, stay until the move is over. The
145 // report itself is written only while preparing or ready. An
146 // aborting move is over for this box: it never flipped, so no
147 // landing wrote a refusal, and its record stays aborting only until
148 // the lane park is acknowledged (Plan 5 Task 1).
149 if m.State == provision.MoveAborting {
150 continue
151 }
152 live[m.ID] = true
153 if m.State != provision.MovePreparing && m.State != provision.MoveReady {
154 continue
155 }
156 if w.lost[rec.Host] {
157 continue // the lost-box gate (movespass.go lostHere)
158 }
159 if m.State == provision.MovePreparing && rec.Seed.MoveBlock() != "" {
160 continue // movesPass's seed rule (spec §5.8), which logged it
161 }
162 if err := w.targetReport(ctx, s, rec, fleet); err != nil {
163 w.logTrouble(rec.Host+" (target)", err)
164 continue
165 }
166 w.trouble.Delete(rec.Host + " (target)")
167 }
168}
169
170// evictArchives drops T5's verdicts (and the carried .env they hold) for
171// every move no longer preparing or ready into this box.
172func (w *movesWorker) evictArchives(live map[string]bool) {
173 w.mu.Lock()
174 defer w.mu.Unlock()
175 for id := range w.archives {
176 if !live[id] {
177 delete(w.archives, id)
178 }
179 }
180}
181
182// retireHeldRefusals retires the held-stage refusal (D3-held) of every
183// move no longer preparing or ready into this box, with the rest of its
184// staging state: live is the same set evictArchives keeps. A record root
185// wrote cannot be removed here (the agent does not own it), which is not
186// worth a line: the root stage pass retires those (retireEndedHeldRefusals).
187func (w *movesWorker) retireHeldRefusals(live map[string]bool) {
188 if w.cfg.DataDir == "" {
189 return
190 }
191 retired, err := release.RetireHeldRefusals(w.cfg.DataDir, func(r release.Refusal) bool { return live[r.MoveOp] })
192 for _, host := range retired {
193 log.Printf("moves: %s: retired the held refusal of a move that is over", host)
194 }
195 if err != nil && !errors.Is(err, fs.ErrPermission) {
196 w.logTrouble("held refusals", err)
197 }
198}
199
200// verifiedArchive is what T5 learned from one published archive, kept per
201// exact version so a pass does not open the same archive again.
202type verifiedArchive struct {
203 ref moves.ArchiveRef
204 check moves.Check
205 plain string
206 entries int
207 bytes int64
208 env []byte // the carried <host>.env, nil when the archive has none
209}
210
211// heldEnv is T3's composition: the environment, the names a carried .env
212// shadows, and the input versions rendered.
213type heldEnv struct {
214 env map[string]string
215 shadowed []string
216 inputs provision.MoveInputs
217}
218
219func (w *movesWorker) targetReport(ctx context.Context, s release.Store, rec provision.Record, fleet moveFleet) error {
220 cfg := w.cfg
221 id := rec.Move.ID
222 obs, err := hostclaim.Read(ctx, s, rec.Host)
223 if err != nil {
224 return fmt.Errorf("read the host claim: %w", err)
225 }
226 if obs.Absent || obs.Claim.Op.ID != id {
227 w.notice(rec.Host+"#target", "moves: %s: the host claim does not name move %s; no target report", rec.Host, id)
228 return nil
229 }
230 // T8 and T9 bind to the named target: only it writes the report.
231 if target := obs.Claim.Op.Intent["target"]; cfg.Box == "" || target != cfg.Box {
232 w.notice(rec.Host+"#target", "moves: %s: this box is not the move's target (%q); no target report", rec.Host, target)
233 return nil
234 }
235 source, err := readSourceReport(ctx, s, rec)
236 if err != nil {
237 return err
238 }
239 now := cfg.Now().UTC()
240 rep := moves.TargetReport{V: 1, Box: cfg.Box, Region: cfg.Region, MoveID: id, At: now,
241 Checks: map[string]moves.Check{}, CutBudgetS: int(moves.CutWindowDefault)}
242 rep.Checks[moves.CheckProtocol] = w.checkProtocol(ctx, s)
243 rep.Checks[moves.CheckSecrets] = w.checkSecrets(ctx, s)
244 art := w.checkArchive(ctx, rec, source)
245 rep.Checks[moves.CheckArchive] = art.check
246 held, t3, notes := w.checkEnv(ctx, s, rec, source, art)
247 rep.Checks[moves.CheckEnv] = t3
248 rep.Notes = notes
249 rep.Checks[moves.CheckStorage] = w.checkStorage(ctx, held, id)
250 // The database's logical size: the newest staged LTX header's, or the
251 // source's local file when that is larger (review finding 1).
252 staged, dbBytes, replicaErr := w.stagedReplica(ctx, rec)
253 if source != nil && source.DBBytes > dbBytes {
254 dbBytes = source.DBBytes
255 }
256 rep.Checks[moves.CheckDisk] = w.checkDisk(max(staged, dbBytes), replicaErr, art)
257 rep.Checks[moves.CheckRelease] = w.checkRelease(ctx, s, rec, source)
258 rep.Checks[moves.CheckOneOwner] = checkOneOwner(fleet, rec.Move.To, moveRegionName(obs.Claim, park.PlacementRegion(rec.Move.To)))
259 rep.Checks[moves.CheckClaim] = w.checkClaim(ctx, obs, rec)
260 rep.Checks[moves.CheckLane] = w.checkLane(ctx, rec, fleet)
261 rep.Checks[moves.CheckQuiet] = checkQuiet(obs, park.PlacementRegion(rec.Move.To), now)
262 rep.Checks[moves.CheckArmed] = checkArmed(cfg)
263
264 // T11 from measured rates: this box's own, and the source's cross rate.
265 m := w.measure(ctx, rec)
266 m.Cross = source.CrossMeasurement()
267 dbMiB := float64(dbBytes) / (1 << 20)
268 if replicaErr != nil {
269 rep.Checks[moves.CheckBudget] = moves.Check{Reason: replicaReason(replicaErr)}
270 } else {
271 res := moves.BudgetCheck(dbMiB, art.entries, m, cfg.ReadyDelta, moves.CutWindowDefault, now)
272 rep.Checks[moves.CheckBudget] = res.Check
273 rep.RecordBudget(dbMiB, art.entries, m, res)
274 }
275
276 rep.Inputs = held.inputs
277 if source != nil {
278 rep.Inputs.Release = source.Release
279 if t3.OK {
280 rep.Inputs.EnvReport = source.Env.Digest
281 }
282 }
283 if art.check.OK {
284 rep.Inputs.FilesPlain = art.plain
285 }
286 body, err := json.Marshal(rep)
287 if err != nil {
288 return err
289 }
290 return moves.PutReplace(ctx, s, moves.TargetReportKey(moves.ReportPrefix(rec.Account, rec.App, rec.Host, id)), body)
291}
292
293// checkArmed is T13, the cut guard's arming statement (Paul's ruling,
294// 2026-10-11): this box can land the move and roll it back, every hook of
295// both wired, judged as the report is written. The source refuses to start
296// a cut unless the target's latest report states it (cutRun.refuseUnarmed).
297// In production both stay unarmed until the wake-source gate opens
298// (movesWakeSourceRegistered, Plan 5 Task 9b). Its reason is
299// copy record MV-N440.
300func checkArmed(cfg moveCfg) moves.Check {
301 if landingArmed(cfg) && rollbackTargetArmed(cfg) {
302 return moves.Check{OK: true}
303 }
304 return moves.Check{Reason: t13NotArmed}
305}
306
307// checkProtocol is T1 (§11): this box speaks at least the published floor
308// and has armed it locally, so its executable selection enforces it. No
309// published floor is a pass: the floor is opt-in until the first
310// production move (§11).
311func (w *movesWorker) checkProtocol(ctx context.Context, s release.Store) moves.Check {
312 fl, err := moves.ReadFloor(ctx, s)
313 if err != nil {
314 log.Printf("moves: T1: %v", err)
315 return moves.Check{Reason: t1FloorUnreadable}
316 }
317 if fl == nil {
318 return moves.Check{OK: true}
319 }
320 if moves.ProtocolVersion < fl.Min {
321 return moves.Check{Reason: t1Below(fl.Min)}
322 }
323 local, err := readLocalFloor(w.cfg.DataDir)
324 if err != nil {
325 log.Printf("moves: T1: local floor: %v", err)
326 return moves.Check{Reason: t1LocalUnreadable}
327 }
328 if local == nil || local.MinRelease != fl.MinRelease || local.Min < fl.Min {
329 return moves.Check{Reason: t1NotArmed}
330 }
331 return moves.Check{OK: true}
332}
333
334// checkSecrets is T2 (§10): this box's secrets key fingerprint equals the
335// one the console publishes.
336func (w *movesWorker) checkSecrets(ctx context.Context, s release.Store) moves.Check {
337 if w.cfg.SecretsPub == nil {
338 return moves.Check{Reason: t2NoKey}
339 }
340 var rec moves.ConsoleRecord
341 if ok, err := readReport(ctx, s, moves.ConsoleKey, &rec); err != nil || !ok || rec.SecretsKeyFP == "" {
342 if err != nil {
343 log.Printf("moves: T2: %v", err)
344 }
345 return moves.Check{Reason: t2NoConsole}
346 }
347 if appconfig.KeyFingerprint(w.cfg.SecretsPub) != rec.SecretsKeyFP {
348 return moves.Check{Reason: t2Differs}
349 }
350 return moves.Check{OK: true}
351}
352
353// checkArchive is T5: the archive the source report names, read by exact
354// version from this box's bucket, opens with this box's key and matches its
355// manifest (archive.Verify: digests, every entry, the entry types, the
356// limits), and is this move's. Its carried <host>.env is kept for T3. A
357// verdict is kept per exact version; a read that fails is not.
358func (w *movesWorker) checkArchive(ctx context.Context, rec provision.Record, source *moves.SourceReport) verifiedArchive {
359 if source == nil || source.Archive == nil {
360 return verifiedArchive{check: moves.Check{Reason: t5NoArchive}}
361 }
362 ref := *source.Archive
363 // A reference by "null" (or no) version pins nothing, so it is refused
364 // before the cache: a verdict kept for it would vouch for whatever the
365 // key holds now (final review finding 4).
366 if !ref.Pinned() {
367 return verifiedArchive{ref: ref, check: moves.Check{Reason: t5NotPinned}}
368 }
369 w.mu.Lock()
370 cached, ok := w.archives[rec.Move.ID]
371 w.mu.Unlock()
372 if ok && cached.ref == ref {
373 return cached
374 }
375 v, keep := w.verifyArchive(ctx, rec, ref)
376 if keep {
377 w.mu.Lock()
378 if w.archives == nil {
379 w.archives = map[string]verifiedArchive{}
380 }
381 w.archives[rec.Move.ID] = v
382 w.mu.Unlock()
383 }
384 return v
385}
386
387func (w *movesWorker) verifyArchive(ctx context.Context, rec provision.Record, ref moves.ArchiveRef) (verifiedArchive, bool) {
388 cfg := w.cfg
389 out := verifiedArchive{ref: ref}
390 fail := func(reason string, keep bool, err error) (verifiedArchive, bool) {
391 log.Printf("moves: %s: T5: %v", rec.Host, err)
392 out.check = moves.Check{Reason: reason}
393 return out, keep
394 }
395 if cfg.SecretsPriv == nil {
396 out.check = moves.Check{Reason: t2NoKey}
397 return out, false
398 }
399 if cfg.Versioned == nil {
400 return fail(t5Unreadable, false, errors.New("no versioned store"))
401 }
402 vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
403 if err != nil {
404 return fail(t5Unreadable, false, err)
405 }
406 raw, err := vs.GetVersion(ctx, cfg.OwnBucket, ref.ManifestKey, ref.ManifestVersion)
407 if err != nil {
408 return fail(t5Unreadable, false, err)
409 }
410 var m archive.Manifest
411 if err := json.Unmarshal(raw, &m); err != nil {
412 return fail(t5Refused, true, err)
413 }
414 if m.Op != rec.Move.ID || m.Incarnation != rec.Move.Incarnation || m.Plain != ref.Plain {
415 return fail(t5Refused, true, fmt.Errorf("the manifest is for %s/%s/%s", m.Op, m.Incarnation, m.Plain))
416 }
417 tmp, err := os.MkdirTemp("", "carlos-move-t5-")
418 if err != nil {
419 return fail(t5Unreadable, false, err)
420 }
421 defer os.RemoveAll(tmp)
422 sealedPath := filepath.Join(tmp, "archive.sealed")
423 if err := downloadVersion(ctx, vs, cfg.OwnBucket, ref.SealedObjectKey, ref.SealedVersion, sealedPath); err != nil {
424 return fail(t5Unreadable, false, err)
425 }
426 bind := appconfig.Binding{Account: rec.Account, App: rec.App}
427 f, err := os.Open(sealedPath)
428 if err != nil {
429 return fail(t5Unreadable, false, err)
430 }
431 defer f.Close()
432 if err := archive.Verify(f, m, cfg.SecretsPriv, bind); err != nil {
433 return fail(t5Refused, true, err)
434 }
435 env, err := carriedEnv(f, m, cfg.SecretsPriv, bind, tmp, rec.Host)
436 if err != nil {
437 return fail(t5Refused, true, err)
438 }
439 out.check, out.plain, out.entries, out.bytes, out.env = moves.Check{OK: true}, m.Plain, len(m.Entries), m.Bytes, env
440 return out, true
441}
442
443func downloadVersion(ctx context.Context, vs lease.VersionedStore, bucket, key, version, path string) error {
444 rc, err := vs.OpenVersion(ctx, bucket, key, version)
445 if err != nil {
446 return err
447 }
448 defer rc.Close()
449 f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
450 if err != nil {
451 return err
452 }
453 if _, err := io.Copy(f, io.LimitReader(rc, archive.MaxDecoded+archive.MaxDecoded/16)); err != nil {
454 f.Close()
455 return err
456 }
457 return f.Close()
458}
459
460// carriedEnv extracts the verified archive into a scratch directory and
461// returns its <host>.env (nil when it carries none).
462func carriedEnv(f *os.File, m archive.Manifest, priv *ecdh.PrivateKey, bind appconfig.Binding, tmp, host string) ([]byte, error) {
463 if _, err := f.Seek(0, io.SeekStart); err != nil {
464 return nil, err
465 }
466 pr, pw := io.Pipe()
467 done := make(chan error, 1)
468 go func() {
469 err := archive.Open(f, m.SealedDataKey, priv, bind, pw)
470 pw.CloseWithError(err)
471 done <- err
472 }()
473 envPath := filepath.Join(tmp, host+".env")
474 xerr := archive.Extract(pr, filepath.Join(tmp, "files"), envPath)
475 pr.CloseWithError(errors.New("extract stopped reading"))
476 if oerr := <-done; oerr != nil && xerr == nil {
477 xerr = oerr
478 }
479 if xerr != nil {
480 return nil, xerr
481 }
482 env, err := os.ReadFile(envPath)
483 if errors.Is(err, os.ErrNotExist) {
484 return nil, nil
485 }
486 return env, err
487}
488
489// checkEnv is T3: the environment this box would launch the flat held row
490// with, compared by HMAC with the source's launch report. Shadowed names
491// and advisory differences are notes, never failures.
492func (w *movesWorker) checkEnv(ctx context.Context, s release.Store, rec provision.Record, source *moves.SourceReport, art verifiedArchive) (heldEnv, moves.Check, []string) {
493 cfg := w.cfg
494 if source == nil {
495 return heldEnv{}, moves.Check{Reason: t3NoSource}, nil
496 }
497 if !art.check.OK {
498 return heldEnv{}, moves.Check{Reason: t5NoArchive}, nil
499 }
500 key, err := moves.ReportKeyFromSecrets(cfg.SecretsKey)
501 if err != nil {
502 return heldEnv{}, moves.Check{Reason: t2NoKey}, nil
503 }
504 if cfg.ComposeHeldEnv == nil {
505 return heldEnv{}, moves.Check{Reason: t3NoCompose}, nil
506 }
507 env, instance, shadowed, in, err := cfg.ComposeHeldEnv(ctx, s, rec, art.env)
508 if err != nil {
509 log.Printf("moves: %s: T3: %v", rec.Host, err)
510 return heldEnv{}, moves.Check{Reason: t3NoCompose}, nil
511 }
512 held := heldEnv{env: env, shadowed: shadowed, inputs: in}
513 var notes, failed []string
514 for _, name := range shadowed {
515 notes = append(notes, t3Shadowed(name))
516 }
517 // A v1 report cannot say which box variables the instance supplied, so
518 // it is refused before any comparison (fail closed).
519 if !moves.CurrentEnvReport(source.Env) {
520 return held, moves.Check{Reason: t3OldReport}, notes
521 }
522 // A move into home names the home's code, "", on the record: T3's home
523 // rule follows the record, never this box's region, so a second
524 // placementless box stays T8's failure and is not T3's pass.
525 targetHome := rec.Move.To == ""
526 for _, d := range moves.CompareEnv(source.Env, moves.ReportLaunch(key, env, instance), true, targetHome) {
527 if d.Note() {
528 notes = append(notes, d.Name+" ("+d.Kind+")")
529 continue
530 }
531 failed = append(failed, d.Name+" ("+d.Kind+")")
532 }
533 if len(failed) > 0 {
534 return held, moves.Check{Reason: t3Differs(failed)}, notes
535 }
536 return held, moves.Check{OK: true}, notes
537}
538
539// checkStorage is T4: every store bound in the composed environment answers
540// a list, put, get and delete of .carlos-move-probe/<move id> with its own
541// credentials.
542func (w *movesWorker) checkStorage(ctx context.Context, held heldEnv, id string) moves.Check {
543 if held.env == nil {
544 return moves.Check{Reason: t3NoCompose}
545 }
546 if w.cfg.ProbeStore == nil {
547 return moves.Check{Reason: t4Failed("(no probe on this box)")}
548 }
549 if prefix, err := w.cfg.ProbeStore(ctx, held.env, id); err != nil {
550 log.Printf("moves: T4: %s: %v", prefix, err)
551 return moves.Check{Reason: t4Failed(prefix)}
552 }
553 return moves.Check{OK: true}
554}
555
556// errNoReplicaStaged is a lane with no LTX file in it yet.
557var errNoReplicaStaged = errors.New("no replica has been staged")
558
559func replicaReason(err error) string {
560 if errors.Is(err, errNoReplicaStaged) {
561 return t11NoReplica
562 }
563 return t11Unmeasured
564}
565
566// stagedReplica reads the replica staged in this box's bucket for the move:
567// the bytes staged, and the database's logical size, Commit pages of
568// PageSize from the header of the newest LTX file (the highest max TXID).
569// The staged bytes are no measure of the database: LTX compresses its
570// pages, so a compressible database stages far fewer bytes than it holds.
571// A newest file whose header cannot be read is an error, never a guess.
572func (w *movesWorker) stagedReplica(ctx context.Context, rec provision.Record) (staged, logical int64, err error) {
573 m, err := w.mirror(ctx, w.cfg.OwnBucketRegion)
574 if err != nil {
575 return 0, 0, err
576 }
577 prefix := moves.ReplicaPrefix(rec.Host, rec.Move.ID)
578 objs, err := m.List(ctx, w.cfg.OwnBucket, prefix)
579 if err != nil {
580 return 0, 0, err
581 }
582 for _, o := range objs {
583 staged += o.Size
584 }
585 newest := newestLTX(objs, prefix, nil)
586 if newest == "" {
587 return staged, 0, errNoReplicaStaged
588 }
589 hdr, err := w.ltxHeader(ctx, newest)
590 if err != nil {
591 return staged, 0, fmt.Errorf("the newest staged LTX file %s: %w", newest, err)
592 }
593 return staged, int64(hdr.Commit) * int64(hdr.PageSize), nil
594}
595
596// newestLTX is the key in objs of the newest LTX file under prefix, the one
597// with the highest max TXID (ties go to the greater key), or "" when there
598// is none. keep, when non-nil, must admit a key relative to prefix.
599func newestLTX(objs map[string]lease.MirrorObject, prefix string, keep func(string) bool) string {
600 var newest string
601 var newestTX ltx.TXID
602 for k := range objs {
603 rel := strings.TrimPrefix(k, prefix)
604 if keep != nil && !keep(rel) {
605 continue
606 }
607 _, name, ok := strings.Cut(rel, "/")
608 if !ok {
609 continue
610 }
611 _, maxTX, perr := ltx.ParseFilename(name)
612 if perr != nil {
613 continue
614 }
615 if newest == "" || maxTX > newestTX || (maxTX == newestTX && k > newest) {
616 newest, newestTX = k, maxTX
617 }
618 }
619 return newest
620}
621
622// replicaLogicalSize is the source's half of the size evidence (final
623// review finding 2): the database's logical size as its own replica states
624// it, from the header of the newest LTX file under the claim's generation
625// (srcPrefix, filtered by keep: the files the lane mirrors), read as the
626// target reads its staged copy. No LTX file yet is 0; a newest file whose
627// header cannot be read is an error, which ends the round before its
628// report, so nothing promotes on a size nobody measured.
629func (w *movesWorker) replicaLogicalSize(ctx context.Context, m lease.MirrorStore, srcPrefix string, keep func(string) bool) (int64, error) {
630 objs, err := m.List(ctx, w.cfg.OwnBucket, srcPrefix)
631 if err != nil {
632 return 0, fmt.Errorf("list the replica source: %w", err)
633 }
634 newest := newestLTX(objs, srcPrefix, keep)
635 if newest == "" {
636 return 0, nil
637 }
638 hdr, err := w.ltxHeader(ctx, newest)
639 if err != nil {
640 return 0, fmt.Errorf("the replica's newest LTX file %s: %w", newest, err)
641 }
642 return int64(hdr.Commit) * int64(hdr.PageSize), nil
643}
644
645// ltxHeader reads the header of the current version of key in this box's
646// bucket: its first ltx.HeaderSize bytes, by exact version.
647func (w *movesWorker) ltxHeader(ctx context.Context, key string) (ltx.Header, error) {
648 cfg := w.cfg
649 if cfg.Versioned == nil {
650 return ltx.Header{}, errors.New("no versioned store")
651 }
652 vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
653 if err != nil {
654 return ltx.Header{}, err
655 }
656 vers, err := vs.ListVersions(ctx, cfg.OwnBucket, key)
657 if err != nil {
658 return ltx.Header{}, err
659 }
660 var version string
661 for _, v := range vers {
662 if v.Key == key && v.IsLatest && !v.DeleteMarker {
663 version = v.VersionID
664 }
665 }
666 if version == "" {
667 return ltx.Header{}, errors.New("no current version")
668 }
669 var rc io.ReadCloser
670 if ro, ok := vs.(lease.VersionRangeOpener); ok {
671 rc, err = ro.OpenVersionRange(ctx, cfg.OwnBucket, key, version, 0, ltx.HeaderSize)
672 } else {
673 rc, err = vs.OpenVersion(ctx, cfg.OwnBucket, key, version)
674 }
675 if err != nil {
676 return ltx.Header{}, err
677 }
678 defer rc.Close()
679 buf := make([]byte, ltx.HeaderSize)
680 if _, err := io.ReadFull(rc, buf); err != nil {
681 return ltx.Header{}, err
682 }
683 var hdr ltx.Header
684 if err := hdr.UnmarshalBinary(buf); err != nil {
685 return ltx.Header{}, err
686 }
687 if !ltx.IsValidPageSize(hdr.PageSize) || hdr.Commit == 0 {
688 return ltx.Header{}, fmt.Errorf("header names %d pages of %d bytes", hdr.Commit, hdr.PageSize)
689 }
690 return hdr, nil
691}
692
693// checkDisk is T6: free space covers twice the replica, plus the archive,
694// plus 2 GiB, and the disk stays under 85% full after it.
695func (w *movesWorker) checkDisk(replicaBytes int64, replicaErr error, art verifiedArchive) moves.Check {
696 if replicaErr != nil {
697 log.Printf("moves: T6: %v", replicaErr)
698 return moves.Check{Reason: replicaReason(replicaErr)}
699 }
700 if w.cfg.DiskFree == nil {
701 return moves.Check{Reason: t6Unmeasured}
702 }
703 total, avail, err := w.cfg.DiskFree(w.cfg.DataDir)
704 if err != nil || total == 0 {
705 log.Printf("moves: T6: %v", err)
706 return moves.Check{Reason: t6Unmeasured}
707 }
708 need := uint64(2*replicaBytes+art.bytes) + 2<<30
709 used := total - avail
710 if avail < need || float64(used+need)/float64(total) >= 0.85 {
711 return moves.Check{Reason: t6NoRoom(float64(need) / (1 << 30))}
712 }
713 return moves.Check{OK: true}
714}
715
716// checkRelease is T7: the source's pinned release materializes for the held
717// row on this box. With the claim off that is staged and published
718// (heldReleaseReady); under the claim it is staged and verified, and the
719// landing's configured step publishes it (heldReleaseStaged; Paul's T7
720// ruling, 2026-10-12).
721func (w *movesWorker) checkRelease(ctx context.Context, s release.Store, rec provision.Record, source *moves.SourceReport) moves.Check {
722 if source == nil || source.Release == "" {
723 return moves.Check{Reason: t7NoRelease}
724 }
725 if w.cfg.ReleaseReady == nil {
726 return moves.Check{Reason: t7Failed}
727 }
728 return w.cfg.ReleaseReady(ctx, s, rec, source.Release)
729}
730
731// checkOneOwner is T8: exactly one hosting box publishes the target
732// placement.
733func checkOneOwner(fleet moveFleet, to string, names ...string) moves.Check {
734 name := moves.RegionText(to)
735 if len(names) == 1 && names[0] != "" {
736 name = names[0]
737 }
738 if n := len(fleet.boxes[to]); n != 1 {
739 return moves.Check{Reason: t8Count(n, name)}
740 }
741 return moves.Check{OK: true}
742}
743
744// checkClaim is T9: the claim is moving under this move, held by the
745// record's incarnation, has touched this region and box, and the staging
746// lane's move.json is this move's (the target owns that validation: the
747// source cannot read the lane, round-2 finding 7).
748func (w *movesWorker) checkClaim(ctx context.Context, obs hostclaim.Observed, rec provision.Record) moves.Check {
749 c := obs.Claim
750 if err := hostclaim.RequireMoveOp(rec.Move.ID, hostclaim.KindMoving, hostclaim.PhaseMoving)(c); err != nil {
751 return moves.Check{Reason: t9NotThisMove}
752 }
753 if c.Holder == nil || c.Holder.Incarnation != rec.Move.Incarnation {
754 return moves.Check{Reason: t9Incarnation}
755 }
756 if !slices.ContainsFunc(c.Touched.Regions, func(region string) bool {
757 return park.PlacementMatches(c.Op.Intent["home_region"], rec.Move.To, region)
758 }) {
759 return moves.Check{Reason: t9Untouched(moveRegionName(c, park.PlacementRegion(rec.Move.To)))}
760 }
761 if !containsString(c.Touched.Boxes, w.cfg.Box) {
762 return moves.Check{Reason: t9Untouched(w.cfg.Box)}
763 }
764 m, err := w.mirror(ctx, w.cfg.OwnBucketRegion)
765 if err != nil {
766 log.Printf("moves: %s: T9: %v", rec.Host, err)
767 return moves.Check{Reason: t9NoMoveJSON}
768 }
769 raw, err := m.Get(ctx, w.cfg.OwnBucket, moves.MoveJSONKey(rec.Host, rec.Move.ID))
770 if err != nil {
771 if !errors.Is(err, lease.ErrMirrorNotFound) {
772 log.Printf("moves: %s: T9: %v", rec.Host, err)
773 }
774 return moves.Check{Reason: t9NoMoveJSON}
775 }
776 var mj moves.MoveJSON
777 if err := json.Unmarshal(raw, &mj); err != nil || mj.MoveID != rec.Move.ID || mj.Incarnation != c.Holder.Incarnation ||
778 mj.Host != rec.Host || mj.From != rec.Move.From || mj.To != rec.Move.To {
779 return moves.Check{Reason: t9OtherMove}
780 }
781 return moves.Check{OK: true}
782}
783
784// checkLane is T10: this box can put and delete moves/_lane/<id> in the
785// source's bucket (M17's return lane).
786func (w *movesWorker) checkLane(ctx context.Context, rec provision.Record, fleet moveFleet) moves.Check {
787 src, ok := fleet.targets[rec.Move.From]
788 if !ok {
789 src, ok = fleet.targets[rec.Region]
790 }
791 if !ok || src.DataBucket == "" {
792 return moves.Check{Reason: t10NoBucket}
793 }
794 if w.cfg.LaneProbe == nil {
795 return moves.Check{Reason: t10Failed}
796 }
797 if err := w.cfg.LaneProbe(ctx, src.DataBucket, src.DataBucketRegion, rec.Move.ID); err != nil {
798 log.Printf("moves: %s: T10: %v", rec.Host, err)
799 return moves.Check{Reason: t10Failed}
800 }
801 return moves.Check{OK: true}
802}
803
804// checkQuiet is T12: the claim's quiet_after for the target region passed.
805func checkQuiet(obs hostclaim.Observed, region string, now time.Time) moves.Check {
806 if until := park.QuietUntil(obs.Claim.QuietAfter, obs.Claim.Op.Intent["home_region"], region); !until.IsZero() && now.Before(until) {
807 return moves.Check{Reason: t12NotQuiet(moveRegionName(obs.Claim, region), until)}
808 }
809 return moves.Check{OK: true}
810}
811
812// measureEvery is how long this box's own measurements are reused: half of
813// moves.MeasurementFreshness, so a report never carries one about to expire,
814// and a pass does not move 80 MiB through S3 and the disk every few seconds.
815const measureEvery = moves.MeasurementFreshness / 2
816
817// measure is this box's regional S3, disk and per-file sync rates, measured
818// again when older than measureEvery. A failed measurement is logged and
819// left out, which T11 reports as missing.
820func (w *movesWorker) measure(ctx context.Context, rec provision.Record) moves.Measurements {
821 cfg := w.cfg
822 now := cfg.Now()
823 w.mu.Lock()
824 m := w.measured
825 w.mu.Unlock()
826 if cfg.MeasureS3 != nil && (m.S3.At.IsZero() || now.Sub(m.S3.At) > measureEvery) {
827 if r, err := cfg.MeasureS3(ctx, moves.ProbeKey(rec.Host, rec.Move.ID)); err != nil {
828 log.Printf("moves: measure regional S3: %v", err)
829 } else {
830 m.S3 = r
831 }
832 }
833 if cfg.MeasureDisk != nil && (m.Disk.At.IsZero() || now.Sub(m.Disk.At) > measureEvery) {
834 if d, f, err := cfg.MeasureDisk(ctx); err != nil {
835 log.Printf("moves: measure the disk: %v", err)
836 } else {
837 m.Disk, m.Fsync = d, f
838 }
839 }
840 w.mu.Lock()
841 w.measured = m
842 w.mu.Unlock()
843 return m
844}
845
846func moveRegionName(c *hostclaim.Claim, region string) string {
847 if park.SameRegion(c.Op.Intent["home_region"], region, park.HomeRegion) {
848 if home := c.Op.Intent["home_region"]; home != "" {
849 return home
850 }
851 return moves.HomeRegionText
852 }
853 return region
854}
855