amadan

carlos / platform Public

moves: readiness reports T1 to T12, S1, S2, the measured T11 budget and the promotion rule (§7.3)

TargetReport, SourceReport, ArchiveRef, Ready, Passing, PutReplace,
Measurements and BudgetCheck, plus ProbeKey outside the staging prefix.
Ready also refuses an estimate priced for a smaller mirror bar than the
operator's. Copy rows MV-N55 to MV-N76 recorded as pending.
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
claude 95e3d459cf16c6ec7011e4c958117a860d012e91 parent 6b3fd93
4 files changed, +915 −0
  • docs/superpowers/plans/2026-10-06-moves-phase1-copy.md +22 −0
  • internal/moves/readiness.go +316 −0
  • internal/moves/readiness_test.go +571 −0
  • internal/moves/staging.go +6 −0
diff --git a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
index 95f1ba358..3a0296fe3 100644
--- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
+++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
@@ -94,6 +94,28 @@ Columns: id, file, text, status.
| MV-N52 | internal/console/apiinstancemove.go | too many instance requests. wait a minute and try again | pending |
| MV-N53 | internal/console/apiinstancemove.go | this console has no deployment bucket, so it cannot answer this | pending |
| MV-N54 | internal/console/apiinstancemove.go | account "<segment>" is ambiguous. pass --account-id with one of: <sqid> ("<name>"), ... | pending |
+| MV-N55 | internal/moves/readiness.go | the last round moved <bytes> bytes and the bar is <bytes> | pending |
+| MV-N56 | internal/moves/readiness.go | the target has not published a readiness report yet | pending |
+| MV-N57 | internal/moves/readiness.go | the target's readiness report is stale | pending |
+| MV-N58 | internal/moves/readiness.go | <check> was not reported | pending |
+| MV-N59 | internal/moves/readiness.go | <check>: did not pass | pending |
+| MV-N60 | internal/moves/readiness.go | T11: the cut budget admits no probe attempt | pending |
+| MV-N61 | internal/moves/readiness.go | T11: the cut budget was priced for a final round of <MiB> MiB and the bar is <MiB> MiB | pending |
+| MV-N62 | internal/moves/readiness.go | the source has not published a readiness report yet | pending |
+| MV-N63 | internal/moves/readiness.go | the source's readiness report is stale | pending |
+| MV-N64 | internal/moves/readiness.go | S1: the source did not answer | pending |
+| MV-N65 | internal/moves/readiness.go | S2: the source could not confirm its spawn state | pending |
+| MV-N66 | internal/moves/readiness.go | the target checked files <digest> and the source now publishes <digest> | pending |
+| MV-N67 | internal/moves/readiness.go | the target checked launch environment <digest> and the source now reports <digest> | pending |
+| MV-N68 | internal/moves/readiness.go | no fresh rate measurement for <which> | pending |
+| MV-N69 | internal/moves/readiness.go | the database size could not be measured | pending |
+| MV-N70 | internal/moves/readiness.go | the cut window is not a usable number of seconds | pending |
+| MV-N71 | internal/moves/readiness.go | the cut needs about <seconds> seconds with one probe attempt and the window is <seconds> seconds | pending |
+| MV-N72 | internal/host/movesreport.go | nested instances cannot move yet | pending |
+| MV-N73 | internal/host/movesreport.go | the source has published no archive yet | pending |
+| MV-N74 | internal/host/movesreport.go | the host claim has not touched <region> yet | pending |
+| MV-N75 | internal/host/movesreport.go | waiting for the root stage pass | pending |
+| MV-N76 | internal/host/movesreport.go | the source has not reported its release yet | pending |
## Plan 2
diff --git a/internal/moves/readiness.go b/internal/moves/readiness.go
new file mode 100644
index 000000000..a1800d77e
--- /dev/null
+++ b/internal/moves/readiness.go
@@ -0,0 +1,316 @@
+package moves
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The readiness checks of §7.3. The target box writes T1 to T12 into its
+// report; the source box writes S1 and S2 into its own. Ready joins them.
+const (
+ CheckProtocol = "T1"
+ CheckSecrets = "T2"
+ CheckEnv = "T3"
+ CheckStorage = "T4"
+ CheckArchive = "T5"
+ CheckDisk = "T6"
+ CheckRelease = "T7"
+ CheckOneOwner = "T8"
+ CheckClaim = "T9"
+ CheckLane = "T10"
+ CheckBudget = "T11"
+ CheckQuiet = "T12"
+)
+
+// AllChecks lists the target's checks in id order. Passing walks it, so the
+// first failure it names is the lowest-numbered one.
+var AllChecks = []string{CheckProtocol, CheckSecrets, CheckEnv, CheckStorage, CheckArchive, CheckDisk, CheckRelease, CheckOneOwner, CheckClaim, CheckLane, CheckBudget, CheckQuiet}
+
+// ReportFreshness bounds how old a report may be and still decide ready
+// (§7.3); the cut re-checks it at `admitted` (§7.4).
+const ReportFreshness = 10 * time.Minute
+
+// MeasurementFreshness bounds how old a rate measurement may be and still
+// price a cut (§7.3 T11). MeasurementSkew is how far ahead of the reader's
+// clock a measurement made on another box may be dated.
+const (
+ MeasurementFreshness = time.Hour
+ MeasurementSkew = time.Minute
+)
+
+// TargetReport is what the target's box publishes at TargetReportKey, every
+// pass (§7.3). Inputs are what its checks were made against; Ready compares
+// them with what the source publishes now. Notes are advisories that do not
+// fail a check (the `shadowed` names of T3).
+type TargetReport struct {
+ V int `json:"v"`
+ Box string `json:"box"`
+ Region string `json:"region"`
+ MoveID string `json:"move_id"`
+ At time.Time `json:"at"`
+ Checks map[string]Check `json:"checks"`
+ Inputs provision.MoveInputs `json:"inputs"`
+ AttemptsAdmitted int `json:"attempts_admitted"`
+ CutBudgetS int `json:"cut_budget_s"`
+ Rates Rates `json:"rates"`
+ Notes []string `json:"notes,omitempty"`
+}
+
+// SourceReport is what the source's box publishes at SourceReportKey.
+// Release is the source's live version (release.LiveVersion), the pinned
+// release T7 stages. CrossRate is the cross-region copy rate in MiB/s from
+// the source's most recent copying round, made through the target bucket's
+// region (the path the cut uses), and CrossRateAt is when that round ended.
+type SourceReport struct {
+ V int `json:"v"`
+ Box string `json:"box"`
+ MoveID string `json:"move_id"`
+ At time.Time `json:"at"`
+ Env EnvReport `json:"env"`
+ S1 Check `json:"s1"`
+ S2 Check `json:"s2"`
+ FilesPlain string `json:"files_plain"`
+ Release string `json:"release"`
+ Archive *ArchiveRef `json:"archive,omitempty"`
+ CrossRate float64 `json:"cross_rate"`
+ CrossRateAt time.Time `json:"cross_rate_at"`
+}
+
+// ArchiveRef names the archive the source published, by exact version, so
+// T5 reads the very objects the source wrote. SealedObjectKey is the sealed
+// archive's object key. It is not archive.Manifest's SealedDataKey, which is
+// the key that seals the data inside.
+type ArchiveRef struct {
+ ManifestKey string `json:"manifest_key"`
+ ManifestVersion string `json:"manifest_version"`
+ SealedObjectKey string `json:"sealed_object_key"`
+ SealedVersion string `json:"sealed_version"`
+ Plain string `json:"plain"`
+}
+
+// CrossMeasurement is the source's cross-region rate as a measurement. A
+// nil report, or one that has measured nothing, gives the zero measurement,
+// which Measurements.Rates refuses.
+func (s *SourceReport) CrossMeasurement() MeasuredRate {
+ if s == nil {
+ return MeasuredRate{}
+ }
+ return MeasuredRate{MiBPerS: s.CrossRate, At: s.CrossRateAt}
+}
+
+// Passing reports whether every check passed, naming the first that did not
+// in id order. A check that is missing from the report fails.
+func (r TargetReport) Passing() (bool, string) {
+ for _, id := range AllChecks {
+ c, ok := r.Checks[id]
+ if !ok {
+ return false, id + " was not reported"
+ }
+ if !c.OK {
+ reason := c.Reason
+ if reason == "" {
+ reason = "did not pass"
+ }
+ return false, id + ": " + reason
+ }
+ }
+ return true, ""
+}
+
+// Ready is §7.3's promotion rule: a round at most the bar, a passing target
+// report and source readiness, each under ReportFreshness old, and at least
+// one probe attempt admitted by the budget (R3-7). It is a pure reading and
+// only ever gates promotion. A later stale report reads false, and nothing
+// here, or in MovePromoteReady, which is legal from preparing only, takes a
+// move back out of ready (Review Focus 5).
+//
+// The target's checks must also have been made against what the source
+// publishes now (round-5 finding 9): a fresh passing report for archive A
+// must not promote after the source has published archive B. And the budget
+// must have been priced for a final mirror round as large as the bar allows
+// (F28(e)), so raising CARLOS_MOVE_READY_DELTA_BYTES cannot promote on an
+// estimate made under a smaller one.
+func Ready(roundDelta, bar int64, target *TargetReport, source *SourceReport, now time.Time) (bool, string) {
+ if roundDelta > bar {
+ return false, fmt.Sprintf("the last round moved %d bytes and the bar is %d", roundDelta, bar)
+ }
+ if target == nil {
+ return false, "the target has not published a readiness report yet"
+ }
+ if now.Sub(target.At) > ReportFreshness {
+ return false, "the target's readiness report is stale"
+ }
+ if ok, why := target.Passing(); !ok {
+ return false, why
+ }
+ if target.AttemptsAdmitted < 1 {
+ return false, "T11: the cut budget admits no probe attempt"
+ }
+ priced := target.Rates.MirrorBar
+ if priced == 0 {
+ priced = MirrorBarMiB
+ }
+ if want := BarMiB(bar); !(priced >= want) { // written so NaN fails
+ return false, fmt.Sprintf("T11: the cut budget was priced for a final round of %g MiB and the bar is %g MiB", priced, want)
+ }
+ if source == nil {
+ return false, "the source has not published a readiness report yet"
+ }
+ if now.Sub(source.At) > ReportFreshness {
+ return false, "the source's readiness report is stale"
+ }
+ if !source.S1.OK {
+ return false, "S1: " + reasonOr(source.S1, "the source did not answer")
+ }
+ if !source.S2.OK {
+ return false, "S2: " + reasonOr(source.S2, "the source could not confirm its spawn state")
+ }
+ if target.Inputs.FilesPlain != source.FilesPlain {
+ return false, fmt.Sprintf("the target checked files %s and the source now publishes %s", target.Inputs.FilesPlain, source.FilesPlain)
+ }
+ if target.Inputs.EnvReport != source.Env.Digest {
+ return false, fmt.Sprintf("the target checked launch environment %s and the source now reports %s", target.Inputs.EnvReport, source.Env.Digest)
+ }
+ return true, ""
+}
+
+func reasonOr(c Check, fallback string) string {
+ if c.Reason == "" {
+ return fallback
+ }
+ return c.Reason
+}
+
+// ---- measured rates and T11 ----
+
+// MeasuredRate is one throughput measurement, in MiB/s, with when and over
+// how many bytes it was taken.
+type MeasuredRate struct {
+ MiBPerS float64 `json:"mib_per_s"`
+ At time.Time `json:"at"`
+ Bytes int64 `json:"bytes"`
+}
+
+// RateOf is bytes over d as a measurement made at `at`. No bytes or no
+// elapsed time is no measurement (a zero rate), never an infinite one.
+func RateOf(bytes int64, d time.Duration, at time.Time) MeasuredRate {
+ m := MeasuredRate{At: at, Bytes: bytes}
+ if bytes > 0 && d > 0 {
+ m.MiBPerS = float64(bytes) / (1 << 20) / d.Seconds()
+ }
+ return m
+}
+
+// Measurements are the three rates T11 prices a cut from. The target
+// measures regional S3 and its disk itself, once per report. Cross comes
+// from the source's report (SourceReport.CrossMeasurement), as only the
+// source's copying rounds cross the regions.
+type Measurements struct {
+ Cross, S3, Disk MeasuredRate
+}
+
+// BarMiB is a ready-delta bar in bytes as the MiB that Rates.MirrorBar
+// counts. A bar of zero or less is zero, which Estimate prices at the
+// default, never below it.
+func BarMiB(bytes int64) float64 {
+ if bytes <= 0 {
+ return 0
+ }
+ return float64(bytes) / (1 << 20)
+}
+
+// Rates turns the measurements into the rates Estimate takes, with the
+// operator's ready-delta bar (bytes) as the mirror bar. Every rate must be
+// a positive finite number taken in the last MeasurementFreshness; any other
+// is refused, naming which. The worked example's rates are never a fallback.
+func (m Measurements) Rates(now time.Time, barBytes int64) (Rates, error) {
+ for _, c := range []struct {
+ which string
+ m MeasuredRate
+ }{
+ {"the cross-region copy path", m.Cross},
+ {"regional S3", m.S3},
+ {"the disk", m.Disk},
+ } {
+ age := now.Sub(c.m.At)
+ if c.m.At.IsZero() || !usable(c.m.MiBPerS) || age > MeasurementFreshness || age < -MeasurementSkew {
+ return Rates{}, fmt.Errorf("no fresh rate measurement for %s", c.which)
+ }
+ }
+ return Rates{Cross: m.Cross.MiBPerS, S3: m.S3.MiBPerS, Disk: m.Disk.MiBPerS, MirrorBar: BarMiB(barBytes)}, nil
+}
+
+// BudgetResult is T11 decided: the check, the attempts the window admits,
+// the rates and the price, for the report to carry.
+type BudgetResult struct {
+ Check Check
+ AttemptsAdmitted int
+ Rates Rates
+ Seconds float64 // the cut priced with one probe attempt
+}
+
+// BudgetCheck is T11 (§7.3): price a cut of a dbMiB database at the
+// measured rates, with barBytes as the final mirror bar, and admit the probe
+// attempts that fit windowSeconds. It fails closed: a missing, stale or
+// unusable measurement, a database size or window that is not a usable
+// number, or a window that admits no attempt, fails the check with zero
+// attempts admitted.
+func BudgetCheck(dbMiB float64, m Measurements, barBytes int64, windowSeconds float64, now time.Time) BudgetResult {
+ fail := func(why string) BudgetResult { return BudgetResult{Check: Check{Reason: why}} }
+ if !finite(dbMiB) || dbMiB < 0 {
+ return fail("the database size could not be measured")
+ }
+ if !finite(windowSeconds) || windowSeconds <= 0 {
+ return fail("the cut window is not a usable number of seconds")
+ }
+ rates, err := m.Rates(now, barBytes)
+ if err != nil {
+ return fail(err.Error())
+ }
+ e := Estimate(dbMiB, rates)
+ n := e.AttemptsAdmitted(windowSeconds)
+ need := e.Total(1)
+ res := BudgetResult{AttemptsAdmitted: n, Rates: rates, Seconds: need}
+ if n < 1 {
+ res.Check.Reason = fmt.Sprintf("the cut needs about %.0f seconds with one probe attempt and the window is %.0f seconds", need, windowSeconds)
+ return res
+ }
+ res.Check.OK = true
+ return res
+}
+
+// ---- writing a recurring object ----
+
+// PutReplace writes key whether or not it exists. release.Store.Put with an
+// empty ifMatch is create-only, so a recurring object (a report, the
+// console record) is written by Get for the current etag (absent is "") and
+// then Put with it, retrying once when another writer got in between. A Get
+// that fails for any reason but absence is returned, never written over.
+// Every mutable object the series writes to the deployment bucket goes
+// through here.
+func PutReplace(ctx context.Context, s release.Store, key string, body []byte) error {
+ var err error
+ for attempt := 0; attempt < 2; attempt++ {
+ _, etag, gerr := s.Get(ctx, key)
+ switch {
+ case gerr == nil:
+ case errors.Is(gerr, os.ErrNotExist):
+ etag = ""
+ default:
+ return fmt.Errorf("read %s before replacing it: %w", key, gerr)
+ }
+ if _, err = s.Put(ctx, key, body, etag); err == nil {
+ return nil
+ }
+ if !errors.Is(err, release.ErrCASConflict) {
+ break
+ }
+ }
+ return fmt.Errorf("write %s: %w", key, err)
+}
diff --git a/internal/moves/readiness_test.go b/internal/moves/readiness_test.go
new file mode 100644
index 000000000..639479aba
--- /dev/null
+++ b/internal/moves/readiness_test.go
@@ -0,0 +1,571 @@
+package moves
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "math"
+ "os"
+ "reflect"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func passingTarget(at time.Time) *TargetReport {
+ r := &TargetReport{V: 1, At: at, Checks: map[string]Check{}, AttemptsAdmitted: 1}
+ for _, id := range AllChecks {
+ r.Checks[id] = Check{OK: true}
+ }
+ return r
+}
+
+func TestReadyNeedsFreshReports(t *testing.T) {
+ now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ pass := passingTarget
+ src := &SourceReport{V: 1, At: now, S1: Check{OK: true}, S2: Check{OK: true}}
+ if ok, why := Ready(1<<20, 16<<20, pass(now.Add(-time.Minute)), src, now); !ok {
+ t.Fatal(why)
+ }
+ // Review Focus 5: a passing report older than 10 minutes does not promote.
+ if ok, why := Ready(1<<20, 16<<20, pass(now.Add(-11*time.Minute)), src, now); ok || !strings.Contains(why, "stale") {
+ t.Fatalf("stale report promoted: %v %s", ok, why)
+ }
+ // The same bound on the source's report.
+ oldSrc := &SourceReport{V: 1, At: now.Add(-11 * time.Minute), S1: Check{OK: true}, S2: Check{OK: true}}
+ if ok, why := Ready(0, 16<<20, pass(now), oldSrc, now); ok || !strings.Contains(why, "stale") {
+ t.Fatalf("stale source report promoted: %v %s", ok, why)
+ }
+ // Over the bar, a failed check, a failed S1, a missing report, zero attempts.
+ if ok, _ := Ready(17<<20, 16<<20, pass(now), src, now); ok {
+ t.Fatal("over the bar")
+ }
+ bad := pass(now)
+ bad.Checks[CheckBudget] = Check{Reason: "needs 131 s, window 120 s"}
+ if ok, why := Ready(0, 16<<20, bad, src, now); ok || !strings.Contains(why, "T11") {
+ t.Fatalf("failed check: %v %s", ok, why)
+ }
+ if ok, _ := Ready(0, 16<<20, pass(now), &SourceReport{V: 1, At: now, S1: Check{Reason: "503"}, S2: Check{OK: true}}, now); ok {
+ t.Fatal("failed S1")
+ }
+ if ok, why := Ready(0, 16<<20, pass(now), &SourceReport{V: 1, At: now, S1: Check{OK: true}, S2: Check{Reason: "strict spawn refused"}}, now); ok || !strings.Contains(why, "S2") {
+ t.Fatalf("failed S2: %v %s", ok, why)
+ }
+ if ok, _ := Ready(0, 16<<20, nil, src, now); ok {
+ t.Fatal("no report")
+ }
+ if ok, _ := Ready(0, 16<<20, pass(now), nil, now); ok {
+ t.Fatal("no source report")
+ }
+ none := pass(now)
+ none.AttemptsAdmitted = 0
+ if ok, _ := Ready(0, 16<<20, none, src, now); ok {
+ t.Fatal("zero attempts admitted")
+ }
+}
+
+// Review Focus 5, second half: a stale report after ready demotes nothing.
+// Ready only ever gates promotion. The one transition it feeds is
+// MovePromoteReady, which is legal from preparing alone, so a move that is
+// already ready stays ready whatever the next reading of Ready says, and
+// reading Ready changes neither report.
+func TestStaleReportAfterReadyDemotesNothing(t *testing.T) {
+ now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ target := passingTarget(now)
+ src := &SourceReport{V: 1, At: now, S1: Check{OK: true}, S2: Check{OK: true}}
+ rec := &provision.Record{Move: &provision.MoveState{State: provision.MovePreparing}}
+
+ if ok, why := Ready(0, 16<<20, target, src, now); !ok {
+ t.Fatal(why)
+ }
+ if err := provision.MovePromoteReady(now)(rec); err != nil {
+ t.Fatal(err)
+ }
+ if rec.Move.State != provision.MoveReady {
+ t.Fatalf("not promoted: %s", rec.Move.State)
+ }
+
+ // Both reports go stale. The reading is false, and nothing moves.
+ targetBefore, srcBefore := *target, *src
+ later := now.Add(ReportFreshness + time.Minute)
+ if ok, why := Ready(0, 16<<20, target, src, later); ok || !strings.Contains(why, "stale") {
+ t.Fatalf("stale reading: %v %s", ok, why)
+ }
+ if !reflect.DeepEqual(*target, targetBefore) || !reflect.DeepEqual(*src, srcBefore) {
+ t.Fatal("Ready changed a report")
+ }
+ // The host acts on a true reading only, so the stale reading does not
+ // reach the journal; and if it did, there is no way back to preparing.
+ if err := provision.MovePromoteReady(later)(rec); err == nil {
+ t.Fatal("a second promotion was accepted")
+ }
+ if rec.Move.State != provision.MoveReady {
+ t.Fatalf("a stale report demoted the move to %s", rec.Move.State)
+ }
+}
+
+func TestReadyRejectsMismatchedInputs(t *testing.T) {
+ now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ mk := func(targetFiles, sourceFiles, targetEnv, sourceEnv string) (*TargetReport, *SourceReport) {
+ tr := passingTarget(now)
+ tr.Inputs = provision.MoveInputs{FilesPlain: targetFiles, EnvReport: targetEnv}
+ sr := &SourceReport{V: 1, At: now, S1: Check{OK: true}, S2: Check{OK: true}, FilesPlain: sourceFiles, Env: EnvReport{V: 1, Digest: sourceEnv}}
+ return tr, sr
+ }
+ tr, sr := mk("sha256:aaa", "sha256:aaa", "env-1", "env-1")
+ if ok, why := Ready(0, 16<<20, tr, sr, now); !ok {
+ t.Fatalf("equal inputs: %s", why)
+ }
+ tr, sr = mk("sha256:aaa", "sha256:bbb", "env-1", "env-1")
+ ok, why := Ready(0, 16<<20, tr, sr, now)
+ if ok || !strings.Contains(why, "sha256:aaa") || !strings.Contains(why, "sha256:bbb") {
+ t.Fatalf("files mismatch: %v %q", ok, why)
+ }
+ tr, sr = mk("sha256:aaa", "sha256:aaa", "env-1", "env-2")
+ ok, why = Ready(0, 16<<20, tr, sr, now)
+ if ok || !strings.Contains(why, "env-1") || !strings.Contains(why, "env-2") {
+ t.Fatalf("env mismatch: %v %q", ok, why)
+ }
+}
+
+func TestReadyRefusesAnEstimateMadeUnderASmallerMirrorBar(t *testing.T) {
+ now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ src := &SourceReport{V: 1, At: now, S1: Check{OK: true}, S2: Check{OK: true}}
+ tr := passingTarget(now)
+
+ // Priced at the default 16 MiB bar (MirrorBar zero), the operator's
+ // bar raised to 32 MiB: the estimate is too small for it.
+ if ok, why := Ready(0, 32<<20, tr, src, now); ok || !strings.Contains(why, "T11") {
+ t.Fatalf("under-priced estimate promoted: %v %q", ok, why)
+ }
+ // Priced at 32 MiB, bar 32 MiB: fine. Priced at 32, bar lowered: fine.
+ tr.Rates.MirrorBar = 32
+ if ok, why := Ready(0, 32<<20, tr, src, now); !ok {
+ t.Fatal(why)
+ }
+ if ok, why := Ready(0, 8<<20, tr, src, now); !ok {
+ t.Fatal(why)
+ }
+ // Priced at 8, bar 16: too small.
+ tr.Rates.MirrorBar = 8
+ if ok, _ := Ready(0, 16<<20, tr, src, now); ok {
+ t.Fatal("a smaller priced bar promoted")
+ }
+}
+
+func TestPassingNamesTheFirstFailureInIDOrder(t *testing.T) {
+ r := passingTarget(time.Now())
+ if ok, why := r.Passing(); !ok || why != "" {
+ t.Fatalf("%v %q", ok, why)
+ }
+ r.Checks[CheckClaim] = Check{Reason: "touched"}
+ r.Checks[CheckArchive] = Check{Reason: "no archive"}
+ if ok, why := r.Passing(); ok || !strings.HasPrefix(why, "T5: ") || !strings.Contains(why, "no archive") {
+ t.Fatalf("%v %q", ok, why)
+ }
+ delete(r.Checks, CheckArchive)
+ if ok, why := r.Passing(); ok || !strings.HasPrefix(why, "T5") || strings.Contains(why, "T9") {
+ t.Fatalf("missing check: %v %q", ok, why)
+ }
+ // A failed check with no reason still says something.
+ r.Checks[CheckArchive] = Check{}
+ if ok, why := r.Passing(); ok || strings.TrimSpace(strings.TrimPrefix(why, "T5:")) == "" {
+ t.Fatalf("empty reason: %v %q", ok, why)
+ }
+ // A report with no checks at all is not passing.
+ if ok, _ := (TargetReport{}).Passing(); ok {
+ t.Fatal("an empty report passed")
+ }
+}
+
+func TestAllChecksListsT1ToT12(t *testing.T) {
+ want := []string{"T1", "T2", "T3", "T4", "T5", "T6", "T7", "T8", "T9", "T10", "T11", "T12"}
+ if !reflect.DeepEqual(AllChecks, want) {
+ t.Fatalf("%v", AllChecks)
+ }
+ got := []string{CheckProtocol, CheckSecrets, CheckEnv, CheckStorage, CheckArchive, CheckDisk, CheckRelease, CheckOneOwner, CheckClaim, CheckLane, CheckBudget, CheckQuiet}
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("%v", got)
+ }
+}
+
+func TestReportsRoundTripWithFixedWireNames(t *testing.T) {
+ at := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ tr := TargetReport{V: 1, Box: "uk-box", Region: "uk", MoveID: "mv_x", At: at,
+ Checks: map[string]Check{CheckProtocol: {OK: true}},
+ Inputs: provision.MoveInputs{FilesPlain: "p", EnvReport: "e"},
+ AttemptsAdmitted: 2, CutBudgetS: 120, Rates: Rates{Cross: 10, S3: 250, Disk: 500, MirrorBar: 16},
+ Notes: []string{"the carried .env overrides X"}}
+ b, err := json.Marshal(tr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var m map[string]json.RawMessage
+ if err := json.Unmarshal(b, &m); err != nil {
+ t.Fatal(err)
+ }
+ for _, k := range []string{"v", "box", "region", "move_id", "at", "checks", "inputs", "attempts_admitted", "cut_budget_s", "rates", "notes"} {
+ if _, ok := m[k]; !ok {
+ t.Errorf("target report has no %q: %s", k, b)
+ }
+ }
+ var back TargetReport
+ if err := json.Unmarshal(b, &back); err != nil || !reflect.DeepEqual(back, tr) {
+ t.Fatalf("%v %+v", err, back)
+ }
+
+ sr := SourceReport{V: 1, Box: "src", MoveID: "mv_x", At: at, S1: Check{OK: true}, S2: Check{OK: true},
+ FilesPlain: "p", Release: "v1.2.3", CrossRate: 9.5, CrossRateAt: at,
+ Archive: &ArchiveRef{ManifestKey: "a/m", ManifestVersion: "v1", SealedObjectKey: "a/s", SealedVersion: "v2", Plain: "p"}}
+ b, err = json.Marshal(sr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ m = nil
+ if err := json.Unmarshal(b, &m); err != nil {
+ t.Fatal(err)
+ }
+ for _, k := range []string{"v", "box", "move_id", "at", "env", "s1", "s2", "files_plain", "release", "archive", "cross_rate", "cross_rate_at"} {
+ if _, ok := m[k]; !ok {
+ t.Errorf("source report has no %q: %s", k, b)
+ }
+ }
+ var ref map[string]string
+ if err := json.Unmarshal(m["archive"], &ref); err != nil {
+ t.Fatal(err)
+ }
+ for _, k := range []string{"manifest_key", "manifest_version", "sealed_object_key", "sealed_version", "plain"} {
+ if _, ok := ref[k]; !ok {
+ t.Errorf("archive ref has no %q", k)
+ }
+ }
+ var sback SourceReport
+ if err := json.Unmarshal(b, &sback); err != nil || !reflect.DeepEqual(sback, sr) {
+ t.Fatalf("%v %+v", err, sback)
+ }
+ // No archive yet: the field is absent, not null.
+ b, _ = json.Marshal(SourceReport{V: 1})
+ if strings.Contains(string(b), `"archive"`) {
+ t.Fatalf("an unpublished archive is written: %s", b)
+ }
+}
+
+// ---- measured rates and T11 ----
+
+var t11Now = time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+
+func goodMeasurements() Measurements {
+ at := t11Now.Add(-time.Minute)
+ return Measurements{
+ Cross: MeasuredRate{MiBPerS: 10, At: at, Bytes: 100 << 20},
+ S3: MeasuredRate{MiBPerS: 250, At: at, Bytes: 16 << 20},
+ Disk: MeasuredRate{MiBPerS: 500, At: at, Bytes: 64 << 20},
+ }
+}
+
+func TestMeasurementsNeedFreshUsableRates(t *testing.T) {
+ r, err := goodMeasurements().Rates(t11Now, 16<<20)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if r.Cross != 10 || r.S3 != 250 || r.Disk != 500 || r.MirrorBar != 16 {
+ t.Fatalf("%+v", r)
+ }
+
+ cases := []struct {
+ name string
+ mutate func(m *Measurements)
+ which string
+ }{
+ {"missing cross", func(m *Measurements) { m.Cross = MeasuredRate{} }, "the cross-region copy path"},
+ {"old s3", func(m *Measurements) { m.S3.At = t11Now.Add(-MeasurementFreshness - time.Second) }, "regional S3"},
+ {"future disk", func(m *Measurements) { m.Disk.At = t11Now.Add(time.Hour) }, "the disk"},
+ {"zero rate", func(m *Measurements) { m.Disk.MiBPerS = 0 }, "the disk"},
+ {"negative rate", func(m *Measurements) { m.S3.MiBPerS = -5 }, "regional S3"},
+ {"NaN", func(m *Measurements) { m.Cross.MiBPerS = math.NaN() }, "the cross-region copy path"},
+ {"+Inf", func(m *Measurements) { m.S3.MiBPerS = math.Inf(1) }, "regional S3"},
+ {"no time", func(m *Measurements) { m.Disk.At = time.Time{} }, "the disk"},
+ }
+ for _, c := range cases {
+ m := goodMeasurements()
+ c.mutate(&m)
+ _, err := m.Rates(t11Now, 16<<20)
+ if err == nil || !strings.Contains(err.Error(), "no fresh rate measurement for "+c.which) {
+ t.Errorf("%s: %v", c.name, err)
+ }
+ }
+ // Exactly an hour old is still fresh.
+ m := goodMeasurements()
+ m.S3.At = t11Now.Add(-MeasurementFreshness)
+ if _, err := m.Rates(t11Now, 16<<20); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestMeasuredRateFromBytesAndTime(t *testing.T) {
+ m := RateOf(100<<20, 10*time.Second, t11Now)
+ if m.MiBPerS != 10 || m.Bytes != 100<<20 || !m.At.Equal(t11Now) {
+ t.Fatalf("%+v", m)
+ }
+ // No elapsed time or no bytes is no measurement, not an infinite rate.
+ for _, m := range []MeasuredRate{RateOf(1<<20, 0, t11Now), RateOf(0, time.Second, t11Now), RateOf(1<<20, -time.Second, t11Now)} {
+ if m.MiBPerS != 0 {
+ t.Fatalf("%+v", m)
+ }
+ }
+}
+
+func TestSourceReportCarriesTheCrossRate(t *testing.T) {
+ at := t11Now.Add(-time.Minute)
+ if got := (&SourceReport{CrossRate: 9.5, CrossRateAt: at}).CrossMeasurement(); got.MiBPerS != 9.5 || !got.At.Equal(at) {
+ t.Fatalf("%+v", got)
+ }
+ // A source that has measured nothing yields the zero measurement, which
+ // Measurements.Rates refuses.
+ m := goodMeasurements()
+ m.Cross = (&SourceReport{}).CrossMeasurement()
+ if _, err := m.Rates(t11Now, 16<<20); err == nil || !strings.Contains(err.Error(), "cross-region") {
+ t.Fatal(err)
+ }
+ if got := (*SourceReport)(nil).CrossMeasurement(); got != (MeasuredRate{}) {
+ t.Fatalf("%+v", got)
+ }
+}
+
+func TestT11BudgetCheck(t *testing.T) {
+ // 100 MiB database at the spec's example rates fits the default window.
+ res := BudgetCheck(100, goodMeasurements(), 16<<20, CutWindowDefault, t11Now)
+ if !res.Check.OK || res.AttemptsAdmitted < 1 {
+ t.Fatalf("%+v", res)
+ }
+ if res.Rates.Cross != 10 || res.Rates.MirrorBar != 16 {
+ t.Fatalf("%+v", res.Rates)
+ }
+ if res.Seconds <= 0 || res.Seconds > CutWindowDefault {
+ t.Fatalf("%v", res.Seconds)
+ }
+
+ // A database too big for the window admits no attempt: T11 fails and
+ // the reason says what was needed.
+ res = BudgetCheck(100000, goodMeasurements(), 16<<20, CutWindowDefault, t11Now)
+ if res.Check.OK || res.AttemptsAdmitted != 0 || !strings.Contains(res.Check.Reason, "120") {
+ t.Fatalf("%+v", res)
+ }
+
+ // A missing measurement fails closed, naming which.
+ m := goodMeasurements()
+ m.Disk = MeasuredRate{}
+ res = BudgetCheck(100, m, 16<<20, CutWindowDefault, t11Now)
+ if res.Check.OK || res.AttemptsAdmitted != 0 || !strings.Contains(res.Check.Reason, "no fresh rate measurement for the disk") {
+ t.Fatalf("%+v", res)
+ }
+
+ // A database size that is not a number, or a window that is not, fails.
+ for _, db := range []float64{math.NaN(), math.Inf(1), -1} {
+ if r := BudgetCheck(db, goodMeasurements(), 16<<20, CutWindowDefault, t11Now); r.Check.OK || r.AttemptsAdmitted != 0 {
+ t.Fatalf("db %v: %+v", db, r)
+ }
+ }
+ for _, w := range []float64{math.NaN(), math.Inf(1), 0, -3} {
+ if r := BudgetCheck(100, goodMeasurements(), 16<<20, w, t11Now); r.Check.OK || r.AttemptsAdmitted != 0 {
+ t.Fatalf("window %v: %+v", w, r)
+ }
+ }
+}
+
+// F28(e): an override of the ready-delta bar reaches the estimate. A larger
+// mirror bar costs more seconds, so a window that fits the default bar can
+// refuse a bar of 4 GiB.
+func TestT11BudgetCheckPricesTheOperatorsBar(t *testing.T) {
+ m := goodMeasurements()
+ def := BudgetCheck(100, m, 16<<20, CutWindowDefault, t11Now)
+ big := BudgetCheck(100, m, 4<<30, CutWindowDefault, t11Now)
+ if !def.Check.OK {
+ t.Fatalf("%+v", def)
+ }
+ if big.Rates.MirrorBar != 4096 {
+ t.Fatalf("the bar did not reach the rates: %+v", big.Rates)
+ }
+ if big.Seconds <= def.Seconds {
+ t.Fatalf("a 4 GiB bar priced at %v, the default at %v", big.Seconds, def.Seconds)
+ }
+ if big.Check.OK {
+ t.Fatalf("a 4 GiB mirror at 10 MiB/s fits 120 s: %+v", big)
+ }
+ // A bar of zero bytes is priced at the default, never below it.
+ zero := BudgetCheck(100, m, 0, CutWindowDefault, t11Now)
+ if zero.Seconds != def.Seconds {
+ t.Fatalf("%v vs %v", zero.Seconds, def.Seconds)
+ }
+ // A report made by BudgetCheck is accepted by Ready at the same bar,
+ // and refused at a larger one.
+ src := &SourceReport{V: 1, At: t11Now, S1: Check{OK: true}, S2: Check{OK: true}}
+ tr := passingTarget(t11Now)
+ tr.Rates, tr.AttemptsAdmitted = big.Rates, 1
+ if ok, why := Ready(0, 4<<30, tr, src, t11Now); !ok {
+ t.Fatal(why)
+ }
+ tr.Rates = def.Rates
+ if ok, _ := Ready(0, 4<<30, tr, src, t11Now); ok {
+ t.Fatal("a report priced at 16 MiB promoted under a 4 GiB bar")
+ }
+}
+
+func TestProbeKeyIsOutsideTheStagingPrefix(t *testing.T) {
+ host, op := "a.go.example.com", "mv_abcdefghijklmnopqrstuvwxyz"
+ k := ProbeKey(host, op)
+ if k != "move-probes/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz/probe" {
+ t.Fatal(k)
+ }
+ if strings.HasPrefix(k, "moves/") || strings.HasPrefix(k, StagingPrefix(host, op)) {
+ t.Fatal("the probe object is under the staging prefix")
+ }
+}
+
+// ---- PutReplace ----
+
+// casStore is an in-memory release.Store with the CAS rules of the real
+// ones. beforePut runs once ahead of the first Put, so a test can lose a
+// race.
+type casStore struct {
+ mu sync.Mutex
+ objs map[string][]byte
+ etags map[string]int
+ puts int
+ beforePut func()
+ failGet error
+}
+
+func newCASStore() *casStore {
+ return &casStore{objs: map[string][]byte{}, etags: map[string]int{}}
+}
+
+func (s *casStore) etag(k string) string {
+ if _, ok := s.objs[k]; !ok {
+ return ""
+ }
+ return "e" + string(rune('0'+s.etags[k]))
+}
+
+func (s *casStore) Get(_ context.Context, key string) ([]byte, string, error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if s.failGet != nil {
+ return nil, "", s.failGet
+ }
+ b, ok := s.objs[key]
+ if !ok {
+ return nil, "", fmt.Errorf("no such key %q: %w", key, os.ErrNotExist)
+ }
+ return append([]byte(nil), b...), s.etag(key), nil
+}
+
+func (s *casStore) Put(_ context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if s.beforePut != nil {
+ f := s.beforePut
+ s.beforePut = nil
+ f()
+ }
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.puts++
+ if s.etag(key) != ifMatch {
+ return "", release.ErrCASConflict
+ }
+ s.objs[key] = append([]byte(nil), body...)
+ s.etags[key]++
+ return s.etag(key), nil
+}
+
+func (s *casStore) List(context.Context, string) ([]string, error) { return nil, nil }
+func (s *casStore) ListDirs(context.Context, string) ([]string, error) { return nil, nil }
+func (s *casStore) Delete(context.Context, string) error { return nil }
+
+func TestPutReplaceCreatesThenReplaces(t *testing.T) {
+ ctx := context.Background()
+ s := newCASStore()
+ key := TargetReportKey(ReportPrefix("bab", "app", "h", "mv_x"))
+ // A create-only Put refuses the second write: the reason PutReplace exists.
+ if _, err := s.Put(ctx, "other", []byte("1"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := s.Put(ctx, "other", []byte("2"), ""); !errors.Is(err, release.ErrCASConflict) {
+ t.Fatal(err)
+ }
+ for _, body := range []string{"one", "two", "three"} {
+ if err := PutReplace(ctx, s, key, []byte(body)); err != nil {
+ t.Fatalf("%s: %v", body, err)
+ }
+ if got, _, _ := s.Get(ctx, key); string(got) != body {
+ t.Fatalf("%q", got)
+ }
+ }
+}
+
+func TestPutReplaceRetriesOnceOnAConflict(t *testing.T) {
+ ctx := context.Background()
+ s := newCASStore()
+ if err := PutReplace(ctx, s, "k", []byte("old")); err != nil {
+ t.Fatal(err)
+ }
+ // Someone else replaces the object between our Get and our Put.
+ s.beforePut = func() {
+ s.objs["k"] = []byte("theirs")
+ s.etags["k"]++
+ }
+ if err := PutReplace(ctx, s, "k", []byte("ours")); err != nil {
+ t.Fatal(err)
+ }
+ if got, _, _ := s.Get(ctx, "k"); string(got) != "ours" {
+ t.Fatalf("%q", got)
+ }
+}
+
+func TestPutReplaceGivesUpAfterTheSecondConflict(t *testing.T) {
+ ctx := context.Background()
+ s := &alwaysConflict{casStore: newCASStore()}
+ err := PutReplace(ctx, s, "k", []byte("x"))
+ if !errors.Is(err, release.ErrCASConflict) {
+ t.Fatalf("%v", err)
+ }
+ if s.puts != 2 {
+ t.Fatalf("%d puts, want exactly one retry", s.puts)
+ }
+}
+
+type alwaysConflict struct{ *casStore }
+
+func (a *alwaysConflict) Put(_ context.Context, _ string, _ []byte, _ string) (string, error) {
+ a.puts++
+ return "", release.ErrCASConflict
+}
+
+func TestPutReplaceDoesNotWriteBlindOnAReadError(t *testing.T) {
+ s := newCASStore()
+ s.failGet = errors.New("503 slow down")
+ err := PutReplace(context.Background(), s, "k", []byte("x"))
+ if err == nil || !strings.Contains(err.Error(), "503") {
+ t.Fatalf("%v", err)
+ }
+ if s.puts != 0 {
+ t.Fatal("wrote after a failed read")
+ }
+}
+
+func TestPutReplaceWorksOnADirStore(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ for _, body := range []string{"one", "two"} {
+ if err := PutReplace(ctx, s, "moves/a/b/target.json", []byte(body)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if got, _, err := s.Get(ctx, "moves/a/b/target.json"); err != nil || string(got) != "two" {
+ t.Fatalf("%q %v", got, err)
+ }
+}
diff --git a/internal/moves/staging.go b/internal/moves/staging.go
index d2258b254..6adca63f3 100644
--- a/internal/moves/staging.go
+++ b/internal/moves/staging.go
@@ -47,3 +47,9 @@ func StoppedKey(prefix, role string) string { return prefix + "stopped-" + role
func LogKey(prefix string, seq int, event string) string {
return fmt.Sprintf("%slog/%07d-%s.json", prefix, seq, event)
}
+
+// ProbeKey is where a target box measures regional S3 by putting and
+// getting a probe object (§7.3 T11). It is outside moves/ on purpose: the
+// target writes it with its own credentials and deletes it itself, and no
+// staging prefix, guarded list or mirror ever reaches it.
+func ProbeKey(host, opID string) string { return "move-probes/" + host + "/" + opID + "/probe" }