amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package moves
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "math"
9 "os"
10 "reflect"
11 "strings"
12 "sync"
13 "testing"
14 "time"
15
16 "github.com/carlosframework/platform/internal/provision"
17 "github.com/carlosframework/platform/internal/release"
18)
19
20// passingTarget is a passing report made at `at` with real evidence: a
21// 100 MiB database priced from rates measured a minute earlier, in a 120 s
22// window. goodSource is the matching source report.
23func passingTarget(at time.Time) *TargetReport {
24 r := &TargetReport{V: 1, At: at, Checks: map[string]Check{}, CutBudgetS: 120, Inputs: provision.MoveInputs{Release: "v7"}}
25 for _, id := range AllChecks {
26 r.Checks[id] = Check{OK: true}
27 }
28 m := Measurements{
29 Cross: MeasuredRate{MiBPerS: 10, At: at.Add(-time.Minute)},
30 S3: MeasuredRate{MiBPerS: 250, At: at.Add(-time.Minute)},
31 Disk: MeasuredRate{MiBPerS: 500, At: at.Add(-time.Minute)},
32 Fsync: FilesRate{PerS: 1600, At: at.Add(-time.Minute)},
33 }
34 res := BudgetCheck(100, 0, m, 16<<20, 120, at)
35 if !res.Check.OK {
36 panic("the test fixture does not fit its own window: " + res.Check.Reason)
37 }
38 r.RecordBudget(100, 0, m, res)
39 return r
40}
41
42func sourceWith(now time.Time, s1, s2 Check) *SourceReport {
43 sr := goodSource(now)
44 sr.S1, sr.S2 = s1, s2
45 return sr
46}
47
48func goodSource(now time.Time) *SourceReport {
49 return &SourceReport{V: 1, At: now, S1: Check{OK: true}, S2: Check{OK: true}, CrossRate: 10, CrossRateAt: now.Add(-time.Minute), Release: "v7"}
50}
51
52func TestReadyNeedsFreshReports(t *testing.T) {
53 now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
54 pass := passingTarget
55 src := goodSource(now)
56 if ok, why := Ready(1<<20, 16<<20, pass(now.Add(-time.Minute)), src, now); !ok {
57 t.Fatal(why)
58 }
59 // Review Focus 5: a passing report older than 10 minutes does not promote.
60 if ok, why := Ready(1<<20, 16<<20, pass(now.Add(-11*time.Minute)), src, now); ok || !strings.Contains(why, "stale") {
61 t.Fatalf("stale report promoted: %v %s", ok, why)
62 }
63 // The same bound on the source's report.
64 oldSrc := goodSource(now)
65 oldSrc.At = now.Add(-11 * time.Minute)
66 if ok, why := Ready(0, 16<<20, pass(now), oldSrc, now); ok || !strings.Contains(why, "stale") {
67 t.Fatalf("stale source report promoted: %v %s", ok, why)
68 }
69 // Over the bar, a failed check, a failed S1, a missing report, zero attempts.
70 if ok, _ := Ready(17<<20, 16<<20, pass(now), src, now); ok {
71 t.Fatal("over the bar")
72 }
73 bad := pass(now)
74 bad.Checks[CheckBudget] = Check{Reason: "needs 131 s, window 120 s"}
75 if ok, why := Ready(0, 16<<20, bad, src, now); ok || !strings.Contains(why, "T11") {
76 t.Fatalf("failed check: %v %s", ok, why)
77 }
78 if ok, _ := Ready(0, 16<<20, pass(now), sourceWith(now, Check{Reason: "503"}, Check{OK: true}), now); ok {
79 t.Fatal("failed S1")
80 }
81 if ok, why := Ready(0, 16<<20, pass(now), sourceWith(now, Check{OK: true}, Check{Reason: "strict spawn refused"}), now); ok || !strings.Contains(why, "S2") {
82 t.Fatalf("failed S2: %v %s", ok, why)
83 }
84 if ok, _ := Ready(0, 16<<20, nil, src, now); ok {
85 t.Fatal("no report")
86 }
87 if ok, _ := Ready(0, 16<<20, pass(now), nil, now); ok {
88 t.Fatal("no source report")
89 }
90 none := pass(now)
91 none.AttemptsAdmitted = 0
92 if ok, _ := Ready(0, 16<<20, none, src, now); ok {
93 t.Fatal("zero attempts admitted")
94 }
95}
96
97// Review Focus 5, second half: a stale report after ready demotes nothing.
98// Ready only ever gates promotion. The one transition it feeds is
99// MovePromoteReady, which is legal from preparing alone, so a move that is
100// already ready stays ready whatever the next reading of Ready says, and
101// reading Ready changes neither report.
102func TestStaleReportAfterReadyDemotesNothing(t *testing.T) {
103 now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
104 target := passingTarget(now)
105 src := goodSource(now)
106 rec := &provision.Record{Move: &provision.MoveState{State: provision.MovePreparing}}
107
108 if ok, why := Ready(0, 16<<20, target, src, now); !ok {
109 t.Fatal(why)
110 }
111 if err := provision.MovePromoteReady(now)(rec); err != nil {
112 t.Fatal(err)
113 }
114 if rec.Move.State != provision.MoveReady {
115 t.Fatalf("not promoted: %s", rec.Move.State)
116 }
117
118 // Both reports go stale. The reading is false, and nothing moves.
119 targetBefore, srcBefore := *target, *src
120 later := now.Add(ReportFreshness + time.Minute)
121 if ok, why := Ready(0, 16<<20, target, src, later); ok || !strings.Contains(why, "stale") {
122 t.Fatalf("stale reading: %v %s", ok, why)
123 }
124 if !reflect.DeepEqual(*target, targetBefore) || !reflect.DeepEqual(*src, srcBefore) {
125 t.Fatal("Ready changed a report")
126 }
127 // The host acts on a true reading only, so the stale reading does not
128 // reach the journal; and if it did, there is no way back to preparing.
129 if err := provision.MovePromoteReady(later)(rec); err == nil {
130 t.Fatal("a second promotion was accepted")
131 }
132 if rec.Move.State != provision.MoveReady {
133 t.Fatalf("a stale report demoted the move to %s", rec.Move.State)
134 }
135}
136
137func TestReadyRejectsMismatchedInputs(t *testing.T) {
138 now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
139 mk := func(targetFiles, sourceFiles, targetEnv, sourceEnv string) (*TargetReport, *SourceReport) {
140 tr := passingTarget(now)
141 tr.Inputs = provision.MoveInputs{FilesPlain: targetFiles, EnvReport: targetEnv, Release: "v7"}
142 sr := goodSource(now)
143 sr.FilesPlain, sr.Env = sourceFiles, EnvReport{V: 1, Digest: sourceEnv}
144 return tr, sr
145 }
146 tr, sr := mk("sha256:aaa", "sha256:aaa", "env-1", "env-1")
147 if ok, why := Ready(0, 16<<20, tr, sr, now); !ok {
148 t.Fatalf("equal inputs: %s", why)
149 }
150 tr, sr = mk("sha256:aaa", "sha256:bbb", "env-1", "env-1")
151 ok, why := Ready(0, 16<<20, tr, sr, now)
152 if ok || !strings.Contains(why, "sha256:aaa") || !strings.Contains(why, "sha256:bbb") {
153 t.Fatalf("files mismatch: %v %q", ok, why)
154 }
155 tr, sr = mk("sha256:aaa", "sha256:aaa", "env-1", "env-2")
156 ok, why = Ready(0, 16<<20, tr, sr, now)
157 if ok || !strings.Contains(why, "env-1") || !strings.Contains(why, "env-2") {
158 t.Fatalf("env mismatch: %v %q", ok, why)
159 }
160}
161
162func TestReadyRefusesAnEstimateMadeUnderASmallerMirrorBar(t *testing.T) {
163 now := t11Now
164 src := goodSource(now)
165 // made returns a report whose budget was priced for the given bar.
166 made := func(barBytes int64) *TargetReport {
167 tr := passingTarget(now)
168 m := Measurements{
169 Cross: MeasuredRate{MiBPerS: 10, At: now.Add(-time.Minute)},
170 S3: MeasuredRate{MiBPerS: 250, At: tr.S3At},
171 Disk: MeasuredRate{MiBPerS: 500, At: tr.DiskAt},
172 Fsync: FilesRate{PerS: 1600, At: tr.FsyncAt},
173 }
174 res := BudgetCheck(100, 0, m, barBytes, 120, now)
175 if !res.Check.OK {
176 t.Fatalf("%+v", res)
177 }
178 tr.RecordBudget(100, 0, m, res)
179 return tr
180 }
181 // Priced at the 16 MiB default, the operator's bar raised to 32 MiB:
182 // the estimate is too small for it.
183 if ok, why := Ready(0, 32<<20, made(16<<20), src, now); ok || !strings.Contains(why, "T11") || !strings.Contains(why, "priced") {
184 t.Fatalf("under-priced estimate promoted: %v %q", ok, why)
185 }
186 // Priced at 32 MiB: fine at 32, and fine at a lower bar.
187 if ok, why := Ready(0, 32<<20, made(32<<20), src, now); !ok {
188 t.Fatal(why)
189 }
190 if ok, why := Ready(0, 8<<20, made(32<<20), src, now); !ok {
191 t.Fatal(why)
192 }
193 // Priced at 8, bar 16: too small.
194 if ok, _ := Ready(0, 16<<20, made(8<<20), src, now); ok {
195 t.Fatal("a smaller priced bar promoted")
196 }
197 // A mirror bar that is not finite never promotes, nor a negative one.
198 for _, bar := range []float64{math.Inf(1), math.NaN(), -16} {
199 tr := made(16 << 20)
200 tr.Rates.MirrorBar = bar
201 if ok, _ := Ready(0, 16<<20, tr, src, now); ok {
202 t.Fatalf("mirror bar %v promoted", bar)
203 }
204 }
205}
206
207// The controller's fail-closed ruling: Ready re-checks the evidence behind
208// the budget at its own `now`, whatever the report's checks say.
209func TestReadyValidatesTheEvidenceItself(t *testing.T) {
210 now := t11Now
211 if ok, why := Ready(0, 16<<20, passingTarget(now), goodSource(now), now); !ok {
212 t.Fatal(why)
213 }
214 bad := []float64{0, -1, math.NaN(), math.Inf(1), math.Inf(-1)}
215 type c struct {
216 name string
217 mutate func(*TargetReport, *SourceReport)
218 }
219 var cases []c
220 for _, v := range bad {
221 v := v
222 cases = append(cases,
223 c{fmt.Sprintf("s3 rate %v", v), func(tr *TargetReport, _ *SourceReport) { tr.Rates.S3 = v }},
224 c{fmt.Sprintf("disk rate %v", v), func(tr *TargetReport, _ *SourceReport) { tr.Rates.Disk = v }},
225 c{fmt.Sprintf("cross rate in the source %v", v), func(_ *TargetReport, sr *SourceReport) { sr.CrossRate = v }},
226 c{fmt.Sprintf("cross rate in the target %v", v), func(tr *TargetReport, _ *SourceReport) { tr.Rates.Cross = v }},
227 c{fmt.Sprintf("fsync rate %v", v), func(tr *TargetReport, _ *SourceReport) { tr.Rates.FsyncRate = v }},
228 )
229 }
230 expired := now.Add(-MeasurementFreshness - time.Second)
231 cases = append(cases,
232 c{"s3 time missing", func(tr *TargetReport, _ *SourceReport) { tr.S3At = time.Time{} }},
233 c{"s3 expired", func(tr *TargetReport, _ *SourceReport) { tr.S3At = expired }},
234 c{"s3 from the future", func(tr *TargetReport, _ *SourceReport) { tr.S3At = now.Add(time.Hour) }},
235 c{"disk time missing", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = time.Time{} }},
236 c{"disk expired", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = expired }},
237 c{"disk from the future", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = now.Add(time.Hour) }},
238 c{"fsync time missing", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = time.Time{} }},
239 c{"fsync expired", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = expired }},
240 c{"fsync from the future", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = now.Add(time.Hour) }},
241 c{"entries negative", func(tr *TargetReport, _ *SourceReport) { tr.Entries = -1 }},
242 c{"too many entries for the window", func(tr *TargetReport, _ *SourceReport) { tr.Entries = 1 << 30 }},
243 c{"cross time missing", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = time.Time{} }},
244 c{"cross expired", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = expired }},
245 c{"cross from the future", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = now.Add(time.Hour) }},
246 c{"source has measured no cross rate", func(_ *TargetReport, sr *SourceReport) { sr.CrossRate, sr.CrossRateAt = 0, time.Time{} }},
247 c{"database size NaN", func(tr *TargetReport, _ *SourceReport) { tr.DBMiB = math.NaN() }},
248 c{"database size negative", func(tr *TargetReport, _ *SourceReport) { tr.DBMiB = -1 }},
249 c{"no window", func(tr *TargetReport, _ *SourceReport) { tr.CutBudgetS = 0 }},
250 c{"database too big for the window", func(tr *TargetReport, _ *SourceReport) { tr.DBMiB = 1e6 }},
251 c{"more attempts claimed than admitted", func(tr *TargetReport, _ *SourceReport) { tr.AttemptsAdmitted = ProbeAttemptsMax + 1 }},
252 c{"no attempts claimed", func(tr *TargetReport, _ *SourceReport) { tr.AttemptsAdmitted = 0 }},
253 )
254 for _, cs := range cases {
255 tr, sr := passingTarget(now), goodSource(now)
256 cs.mutate(tr, sr)
257 ok, why := Ready(0, 16<<20, tr, sr, now)
258 if ok || !strings.Contains(why, "T11") {
259 t.Errorf("%s: ready=%v %q", cs.name, ok, why)
260 }
261 }
262
263 // Which rate is named when a measurement is missing or expired.
264 for _, w := range []struct {
265 name string
266 mutate func(*TargetReport, *SourceReport)
267 which string
268 }{
269 {"s3", func(tr *TargetReport, _ *SourceReport) { tr.S3At = expired }, "regional S3"},
270 {"disk", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = expired }, "the disk"},
271 {"fsync", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = expired }, "the disk's per-file sync"},
272 {"cross", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = expired }, "the cross-region copy path"},
273 } {
274 tr, sr := passingTarget(now), goodSource(now)
275 w.mutate(tr, sr)
276 if _, why := Ready(0, 16<<20, tr, sr, now); !strings.Contains(why, "no fresh rate measurement for "+w.which) {
277 t.Errorf("%s: %q", w.name, why)
278 }
279 }
280
281 // A measurement that expires while its report is still fresh is
282 // refused: the target's report is 5 minutes old, its rates 58 minutes
283 // old when written, so an hour later than their time they are stale.
284 tr := passingTarget(now.Add(-5 * time.Minute))
285 tr.S3At, tr.DiskAt = now.Add(-58*time.Minute), now.Add(-58*time.Minute)
286 if ok, why := Ready(0, 16<<20, tr, goodSource(now), now); !ok {
287 t.Fatal(why)
288 }
289 if ok, why := Ready(0, 16<<20, tr, goodSource(now.Add(5*time.Minute)), now.Add(5*time.Minute)); ok || !strings.Contains(why, "T11") {
290 t.Fatalf("a measurement that expired under a fresh report: %v %q", ok, why)
291 }
292
293 // A stored count at or below the recomputed one is accepted.
294 tr = passingTarget(now)
295 tr.AttemptsAdmitted = 1
296 if ok, why := Ready(0, 16<<20, tr, goodSource(now), now); !ok {
297 t.Fatal(why)
298 }
299}
300
301func TestPassingNamesTheFirstFailureInIDOrder(t *testing.T) {
302 r := passingTarget(time.Now())
303 if ok, why := r.Passing(); !ok || why != "" {
304 t.Fatalf("%v %q", ok, why)
305 }
306 r.Checks[CheckClaim] = Check{Reason: "touched"}
307 r.Checks[CheckArchive] = Check{Reason: "no archive"}
308 if ok, why := r.Passing(); ok || !strings.HasPrefix(why, "T5: ") || !strings.Contains(why, "no archive") {
309 t.Fatalf("%v %q", ok, why)
310 }
311 delete(r.Checks, CheckArchive)
312 if ok, why := r.Passing(); ok || !strings.HasPrefix(why, "T5") || strings.Contains(why, "T9") {
313 t.Fatalf("missing check: %v %q", ok, why)
314 }
315 // A failed check with no reason still says something.
316 r.Checks[CheckArchive] = Check{}
317 if ok, why := r.Passing(); ok || strings.TrimSpace(strings.TrimPrefix(why, "T5:")) == "" {
318 t.Fatalf("empty reason: %v %q", ok, why)
319 }
320 // A report with no checks at all is not passing.
321 if ok, _ := (TargetReport{}).Passing(); ok {
322 t.Fatal("an empty report passed")
323 }
324}
325
326func TestAllChecksListsT1ToT13(t *testing.T) {
327 want := []string{"T1", "T2", "T3", "T4", "T5", "T6", "T7", "T8", "T9", "T10", "T11", "T12", "T13"}
328 if !reflect.DeepEqual(AllChecks, want) {
329 t.Fatalf("%v", AllChecks)
330 }
331 got := []string{CheckProtocol, CheckSecrets, CheckEnv, CheckStorage, CheckArchive, CheckDisk, CheckRelease, CheckOneOwner, CheckClaim, CheckLane, CheckBudget, CheckQuiet, CheckArmed}
332 if !reflect.DeepEqual(got, want) {
333 t.Fatalf("%v", got)
334 }
335}
336
337func TestReportsRoundTripWithFixedWireNames(t *testing.T) {
338 at := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
339 tr := TargetReport{V: 1, Box: "uk-box", Region: "uk", MoveID: "mv_x", At: at,
340 Checks: map[string]Check{CheckProtocol: {OK: true}},
341 Inputs: provision.MoveInputs{FilesPlain: "p", EnvReport: "e"},
342 AttemptsAdmitted: 2, CutBudgetS: 120, Rates: Rates{Cross: 10, S3: 250, Disk: 500, MirrorBar: 16},
343 Notes: []string{"the carried .env overrides X"}}
344 b, err := json.Marshal(tr)
345 if err != nil {
346 t.Fatal(err)
347 }
348 var m map[string]json.RawMessage
349 if err := json.Unmarshal(b, &m); err != nil {
350 t.Fatal(err)
351 }
352 for _, k := range []string{"v", "box", "region", "move_id", "at", "checks", "inputs", "attempts_admitted", "cut_budget_s", "rates", "notes"} {
353 if _, ok := m[k]; !ok {
354 t.Errorf("target report has no %q: %s", k, b)
355 }
356 }
357 var back TargetReport
358 if err := json.Unmarshal(b, &back); err != nil || !reflect.DeepEqual(back, tr) {
359 t.Fatalf("%v %+v", err, back)
360 }
361
362 sr := SourceReport{V: 1, Box: "src", MoveID: "mv_x", At: at, S1: Check{OK: true}, S2: Check{OK: true},
363 FilesPlain: "p", Release: "v1.2.3", CrossRate: 9.5, CrossRateAt: at,
364 Archive: &ArchiveRef{ManifestKey: "a/m", ManifestVersion: "v1", SealedObjectKey: "a/s", SealedVersion: "v2", Plain: "p"}}
365 b, err = json.Marshal(sr)
366 if err != nil {
367 t.Fatal(err)
368 }
369 m = nil
370 if err := json.Unmarshal(b, &m); err != nil {
371 t.Fatal(err)
372 }
373 for _, k := range []string{"v", "box", "move_id", "at", "env", "s1", "s2", "files_plain", "release", "archive", "cross_rate", "cross_rate_at"} {
374 if _, ok := m[k]; !ok {
375 t.Errorf("source report has no %q: %s", k, b)
376 }
377 }
378 var ref map[string]string
379 if err := json.Unmarshal(m["archive"], &ref); err != nil {
380 t.Fatal(err)
381 }
382 for _, k := range []string{"manifest_key", "manifest_version", "sealed_object_key", "sealed_version", "plain"} {
383 if _, ok := ref[k]; !ok {
384 t.Errorf("archive ref has no %q", k)
385 }
386 }
387 var sback SourceReport
388 if err := json.Unmarshal(b, &sback); err != nil || !reflect.DeepEqual(sback, sr) {
389 t.Fatalf("%v %+v", err, sback)
390 }
391 // No archive yet: the field is absent, not null.
392 b, _ = json.Marshal(SourceReport{V: 1})
393 if strings.Contains(string(b), `"archive"`) {
394 t.Fatalf("an unpublished archive is written: %s", b)
395 }
396}
397
398// ---- measured rates and T11 ----
399
400var t11Now = time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
401
402func goodMeasurements() Measurements {
403 at := t11Now.Add(-time.Minute)
404 return Measurements{
405 Cross: MeasuredRate{MiBPerS: 10, At: at, Bytes: 100 << 20},
406 S3: MeasuredRate{MiBPerS: 250, At: at, Bytes: 16 << 20},
407 Disk: MeasuredRate{MiBPerS: 500, At: at, Bytes: 64 << 20},
408 Fsync: FilesRate{PerS: 1600, At: at, Files: 512},
409 }
410}
411
412func TestMeasurementsNeedFreshUsableRates(t *testing.T) {
413 r, err := goodMeasurements().Rates(t11Now, 16<<20)
414 if err != nil {
415 t.Fatal(err)
416 }
417 if r.Cross != 10 || r.S3 != 250 || r.Disk != 500 || r.MirrorBar != 16 || r.FsyncRate != 1600 {
418 t.Fatalf("%+v", r)
419 }
420
421 cases := []struct {
422 name string
423 mutate func(m *Measurements)
424 which string
425 }{
426 {"missing cross", func(m *Measurements) { m.Cross = MeasuredRate{} }, "the cross-region copy path"},
427 {"old s3", func(m *Measurements) { m.S3.At = t11Now.Add(-MeasurementFreshness - time.Second) }, "regional S3"},
428 {"future disk", func(m *Measurements) { m.Disk.At = t11Now.Add(time.Hour) }, "the disk"},
429 {"zero rate", func(m *Measurements) { m.Disk.MiBPerS = 0 }, "the disk"},
430 {"negative rate", func(m *Measurements) { m.S3.MiBPerS = -5 }, "regional S3"},
431 {"NaN", func(m *Measurements) { m.Cross.MiBPerS = math.NaN() }, "the cross-region copy path"},
432 {"+Inf", func(m *Measurements) { m.S3.MiBPerS = math.Inf(1) }, "regional S3"},
433 {"no time", func(m *Measurements) { m.Disk.At = time.Time{} }, "the disk"},
434 {"no fsync", func(m *Measurements) { m.Fsync = FilesRate{} }, "the disk's per-file sync"},
435 {"old fsync", func(m *Measurements) { m.Fsync.At = t11Now.Add(-MeasurementFreshness - time.Second) }, "the disk's per-file sync"},
436 {"NaN fsync", func(m *Measurements) { m.Fsync.PerS = math.NaN() }, "the disk's per-file sync"},
437 }
438 for _, c := range cases {
439 m := goodMeasurements()
440 c.mutate(&m)
441 _, err := m.Rates(t11Now, 16<<20)
442 if err == nil || !strings.Contains(err.Error(), "no fresh rate measurement for "+c.which) {
443 t.Errorf("%s: %v", c.name, err)
444 }
445 }
446 // Exactly an hour old is still fresh.
447 m := goodMeasurements()
448 m.S3.At = t11Now.Add(-MeasurementFreshness)
449 if _, err := m.Rates(t11Now, 16<<20); err != nil {
450 t.Fatal(err)
451 }
452}
453
454func TestMeasuredRateFromBytesAndTime(t *testing.T) {
455 m := RateOf(100<<20, 10*time.Second, t11Now)
456 if m.MiBPerS != 10 || m.Bytes != 100<<20 || !m.At.Equal(t11Now) {
457 t.Fatalf("%+v", m)
458 }
459 // No elapsed time or no bytes is no measurement, not an infinite rate.
460 for _, m := range []MeasuredRate{RateOf(1<<20, 0, t11Now), RateOf(0, time.Second, t11Now), RateOf(1<<20, -time.Second, t11Now)} {
461 if m.MiBPerS != 0 {
462 t.Fatalf("%+v", m)
463 }
464 }
465}
466
467func TestSourceReportCarriesTheCrossRate(t *testing.T) {
468 at := t11Now.Add(-time.Minute)
469 if got := (&SourceReport{CrossRate: 9.5, CrossRateAt: at}).CrossMeasurement(); got.MiBPerS != 9.5 || !got.At.Equal(at) {
470 t.Fatalf("%+v", got)
471 }
472 // A source that has measured nothing yields the zero measurement, which
473 // Measurements.Rates refuses.
474 m := goodMeasurements()
475 m.Cross = (&SourceReport{}).CrossMeasurement()
476 if _, err := m.Rates(t11Now, 16<<20); err == nil || !strings.Contains(err.Error(), "cross-region") {
477 t.Fatal(err)
478 }
479 if got := (*SourceReport)(nil).CrossMeasurement(); got != (MeasuredRate{}) {
480 t.Fatalf("%+v", got)
481 }
482}
483
484func TestT11BudgetCheck(t *testing.T) {
485 // 100 MiB database at the spec's example rates fits the default window.
486 res := BudgetCheck(100, 0, goodMeasurements(), 16<<20, CutWindowDefault, t11Now)
487 if !res.Check.OK || res.AttemptsAdmitted < 1 {
488 t.Fatalf("%+v", res)
489 }
490 if res.Rates.Cross != 10 || res.Rates.MirrorBar != 16 {
491 t.Fatalf("%+v", res.Rates)
492 }
493 if res.Seconds <= 0 || res.Seconds > CutWindowDefault {
494 t.Fatalf("%v", res.Seconds)
495 }
496
497 // A database too big for the window admits no attempt: T11 fails and
498 // the reason says what was needed.
499 res = BudgetCheck(100000, 0, goodMeasurements(), 16<<20, CutWindowDefault, t11Now)
500 if res.Check.OK || res.AttemptsAdmitted != 0 || !strings.Contains(res.Check.Reason, "120") {
501 t.Fatalf("%+v", res)
502 }
503
504 // A missing measurement fails closed, naming which.
505 m := goodMeasurements()
506 m.Disk = MeasuredRate{}
507 res = BudgetCheck(100, 0, m, 16<<20, CutWindowDefault, t11Now)
508 if res.Check.OK || res.AttemptsAdmitted != 0 || !strings.Contains(res.Check.Reason, "no fresh rate measurement for the disk") {
509 t.Fatalf("%+v", res)
510 }
511
512 // A database size that is not a number, or a window that is not, fails.
513 for _, db := range []float64{math.NaN(), math.Inf(1), -1} {
514 if r := BudgetCheck(db, 0, goodMeasurements(), 16<<20, CutWindowDefault, t11Now); r.Check.OK || r.AttemptsAdmitted != 0 {
515 t.Fatalf("db %v: %+v", db, r)
516 }
517 }
518 for _, w := range []float64{math.NaN(), math.Inf(1), 0, -3} {
519 if r := BudgetCheck(100, 0, goodMeasurements(), 16<<20, w, t11Now); r.Check.OK || r.AttemptsAdmitted != 0 {
520 t.Fatalf("window %v: %+v", w, r)
521 }
522 }
523}
524
525// F28(e): an override of the ready-delta bar reaches the estimate. A larger
526// mirror bar costs more seconds, so a window that fits the default bar can
527// refuse a bar of 4 GiB.
528func TestT11BudgetCheckPricesTheOperatorsBar(t *testing.T) {
529 m := goodMeasurements()
530 def := BudgetCheck(100, 0, m, 16<<20, CutWindowDefault, t11Now)
531 big := BudgetCheck(100, 0, m, 4<<30, CutWindowDefault, t11Now)
532 if !def.Check.OK {
533 t.Fatalf("%+v", def)
534 }
535 if big.Rates.MirrorBar != 4096 {
536 t.Fatalf("the bar did not reach the rates: %+v", big.Rates)
537 }
538 if big.Seconds <= def.Seconds {
539 t.Fatalf("a 4 GiB bar priced at %v, the default at %v", big.Seconds, def.Seconds)
540 }
541 if big.Check.OK {
542 t.Fatalf("a 4 GiB mirror at 10 MiB/s fits 120 s: %+v", big)
543 }
544 // A bar of zero bytes is priced at the default, never below it.
545 zero := BudgetCheck(100, 0, m, 0, CutWindowDefault, t11Now)
546 if zero.Seconds != def.Seconds {
547 t.Fatalf("%v vs %v", zero.Seconds, def.Seconds)
548 }
549 // A rejected budget never promotes: recording the 4 GiB result in a
550 // report leaves it with zero attempts, and Ready refuses it.
551 src := goodSource(t11Now)
552 tr := passingTarget(t11Now)
553 tr.RecordBudget(100, 0, m, big)
554 if tr.AttemptsAdmitted != 0 {
555 t.Fatalf("%d", tr.AttemptsAdmitted)
556 }
557 if ok, _ := Ready(0, 4<<30, tr, src, t11Now); ok {
558 t.Fatal("a report whose budget admits no attempt promoted")
559 }
560}
561
562func TestProbeKeyIsOutsideTheStagingPrefix(t *testing.T) {
563 host, op := "a.go.example.com", "mv_abcdefghijklmnopqrstuvwxyz"
564 k := ProbeKey(host, op)
565 if k != "move-probes/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz/probe" {
566 t.Fatal(k)
567 }
568 if strings.HasPrefix(k, "moves/") || strings.HasPrefix(k, StagingPrefix(host, op)) {
569 t.Fatal("the probe object is under the staging prefix")
570 }
571}
572
573// ---- PutReplace ----
574
575// casStore is an in-memory release.Store with the CAS rules of the real
576// ones. beforePut runs once ahead of the first Put, so a test can lose a
577// race.
578type casStore struct {
579 mu sync.Mutex
580 objs map[string][]byte
581 etags map[string]int
582 puts int
583 beforePut func()
584 failGet error
585}
586
587func newCASStore() *casStore {
588 return &casStore{objs: map[string][]byte{}, etags: map[string]int{}}
589}
590
591func (s *casStore) etag(k string) string {
592 if _, ok := s.objs[k]; !ok {
593 return ""
594 }
595 return "e" + string(rune('0'+s.etags[k]))
596}
597
598func (s *casStore) Get(_ context.Context, key string) ([]byte, string, error) {
599 s.mu.Lock()
600 defer s.mu.Unlock()
601 if s.failGet != nil {
602 return nil, "", s.failGet
603 }
604 b, ok := s.objs[key]
605 if !ok {
606 return nil, "", fmt.Errorf("no such key %q: %w", key, os.ErrNotExist)
607 }
608 return append([]byte(nil), b...), s.etag(key), nil
609}
610
611func (s *casStore) Put(_ context.Context, key string, body []byte, ifMatch string) (string, error) {
612 if s.beforePut != nil {
613 f := s.beforePut
614 s.beforePut = nil
615 f()
616 }
617 s.mu.Lock()
618 defer s.mu.Unlock()
619 s.puts++
620 if s.etag(key) != ifMatch {
621 return "", release.ErrCASConflict
622 }
623 s.objs[key] = append([]byte(nil), body...)
624 s.etags[key]++
625 return s.etag(key), nil
626}
627
628func (s *casStore) List(context.Context, string) ([]string, error) { return nil, nil }
629func (s *casStore) ListDirs(context.Context, string) ([]string, error) { return nil, nil }
630func (s *casStore) Delete(context.Context, string) error { return nil }
631
632func TestPutReplaceCreatesThenReplaces(t *testing.T) {
633 ctx := context.Background()
634 s := newCASStore()
635 key := TargetReportKey(ReportPrefix("bab", "app", "h", "mv_x"))
636 // A create-only Put refuses the second write: the reason PutReplace exists.
637 if _, err := s.Put(ctx, "other", []byte("1"), ""); err != nil {
638 t.Fatal(err)
639 }
640 if _, err := s.Put(ctx, "other", []byte("2"), ""); !errors.Is(err, release.ErrCASConflict) {
641 t.Fatal(err)
642 }
643 for _, body := range []string{"one", "two", "three"} {
644 if err := PutReplace(ctx, s, key, []byte(body)); err != nil {
645 t.Fatalf("%s: %v", body, err)
646 }
647 if got, _, _ := s.Get(ctx, key); string(got) != body {
648 t.Fatalf("%q", got)
649 }
650 }
651}
652
653func TestPutReplaceRetriesOnceOnAConflict(t *testing.T) {
654 ctx := context.Background()
655 s := newCASStore()
656 if err := PutReplace(ctx, s, "k", []byte("old")); err != nil {
657 t.Fatal(err)
658 }
659 // Someone else replaces the object between our Get and our Put.
660 s.beforePut = func() {
661 s.objs["k"] = []byte("theirs")
662 s.etags["k"]++
663 }
664 if err := PutReplace(ctx, s, "k", []byte("ours")); err != nil {
665 t.Fatal(err)
666 }
667 if got, _, _ := s.Get(ctx, "k"); string(got) != "ours" {
668 t.Fatalf("%q", got)
669 }
670}
671
672func TestPutReplaceGivesUpAfterTheSecondConflict(t *testing.T) {
673 ctx := context.Background()
674 s := &alwaysConflict{casStore: newCASStore()}
675 err := PutReplace(ctx, s, "k", []byte("x"))
676 if !errors.Is(err, release.ErrCASConflict) {
677 t.Fatalf("%v", err)
678 }
679 if s.puts != 2 {
680 t.Fatalf("%d puts, want exactly one retry", s.puts)
681 }
682}
683
684type alwaysConflict struct{ *casStore }
685
686func (a *alwaysConflict) Put(_ context.Context, _ string, _ []byte, _ string) (string, error) {
687 a.puts++
688 return "", release.ErrCASConflict
689}
690
691func TestPutReplaceDoesNotWriteBlindOnAReadError(t *testing.T) {
692 s := newCASStore()
693 s.failGet = errors.New("503 slow down")
694 err := PutReplace(context.Background(), s, "k", []byte("x"))
695 if err == nil || !strings.Contains(err.Error(), "503") {
696 t.Fatalf("%v", err)
697 }
698 if s.puts != 0 {
699 t.Fatal("wrote after a failed read")
700 }
701}
702
703func TestPutReplaceWorksOnADirStore(t *testing.T) {
704 ctx := context.Background()
705 s := release.NewDirStore(t.TempDir())
706 for _, body := range []string{"one", "two"} {
707 if err := PutReplace(ctx, s, "moves/a/b/target.json", []byte(body)); err != nil {
708 t.Fatal(err)
709 }
710 }
711 if got, _, err := s.Get(ctx, "moves/a/b/target.json"); err != nil || string(got) != "two" {
712 t.Fatalf("%q %v", got, err)
713 }
714}
715
716// A report dated ahead of the reader's clock would stay fresh until the
717// clock caught up and then some. One minute ahead is clock skew between
718// boxes and is accepted; beyond that is refused.
719func TestReadyBoundsReportTimestampsInTheFuture(t *testing.T) {
720 now := t11Now
721 // A target report one minute ahead is fine (its own measurements are
722 // dated relative to it, so they sit in the past of now).
723 tr := passingTarget(now.Add(ReportSkew))
724 if ok, why := Ready(0, 16<<20, tr, goodSource(now), now); !ok {
725 t.Fatalf("a target report one minute ahead: %s", why)
726 }
727 tr = passingTarget(now.Add(ReportSkew + time.Nanosecond))
728 if ok, why := Ready(0, 16<<20, tr, goodSource(now), now); ok || !strings.Contains(why, "ahead") || !strings.Contains(why, "target") {
729 t.Fatalf("a target report beyond the skew: %v %q", ok, why)
730 }
731 // Far ahead, as a stuck or wrong clock would write.
732 if ok, _ := Ready(0, 16<<20, passingTarget(now.Add(24*time.Hour)), goodSource(now), now); ok {
733 t.Fatal("a target report a day ahead promoted")
734 }
735
736 sr := goodSource(now)
737 sr.At = now.Add(ReportSkew)
738 if ok, why := Ready(0, 16<<20, passingTarget(now), sr, now); !ok {
739 t.Fatalf("a source report one minute ahead: %s", why)
740 }
741 sr.At = now.Add(ReportSkew + time.Nanosecond)
742 if ok, why := Ready(0, 16<<20, passingTarget(now), sr, now); ok || !strings.Contains(why, "ahead") || !strings.Contains(why, "source") {
743 t.Fatalf("a source report beyond the skew: %v %q", ok, why)
744 }
745 sr.At = now.Add(24 * time.Hour)
746 if ok, _ := Ready(0, 16<<20, passingTarget(now), sr, now); ok {
747 t.Fatal("a source report a day ahead promoted")
748 }
749
750 // The old edge is unchanged: exactly ReportFreshness old is fresh.
751 if ok, why := Ready(0, 16<<20, passingTarget(now.Add(-ReportFreshness)), goodSource(now), now); !ok {
752 t.Fatal(why)
753 }
754 if ok, _ := Ready(0, 16<<20, passingTarget(now.Add(-ReportFreshness-time.Nanosecond)), goodSource(now), now); ok {
755 t.Fatal("a report a nanosecond past the freshness window promoted")
756 }
757}
758
759// The exact edges of a measurement's window: an hour old and a minute
760// ahead are accepted, a nanosecond beyond either is refused.
761func TestMeasurementWindowEdges(t *testing.T) {
762 at := func(d time.Duration) error {
763 m := goodMeasurements()
764 m.S3.At = t11Now.Add(d)
765 _, err := m.Rates(t11Now, 16<<20)
766 return err
767 }
768 for _, c := range []struct {
769 d time.Duration
770 ok bool
771 }{
772 {-MeasurementFreshness, true},
773 {-MeasurementFreshness - time.Nanosecond, false},
774 {MeasurementSkew, true},
775 {MeasurementSkew + time.Nanosecond, false},
776 {0, true},
777 } {
778 if err := at(c.d); (err == nil) != c.ok {
779 t.Errorf("age %v: err %v, want ok=%v", c.d, err, c.ok)
780 }
781 }
782}
783
784// Ruling T7-budget: the archive's entry count is priced (and carried in the
785// report, so Ready re-prices it); a count that is not usable fails closed.
786func TestT11BudgetCheckPricesEntries(t *testing.T) {
787 m := goodMeasurements()
788 none := BudgetCheck(100, 0, m, 16<<20, CutWindowDefault, t11Now)
789 many := BudgetCheck(100, 50000, m, 16<<20, CutWindowDefault, t11Now)
790 if !none.Check.OK || many.Seconds <= none.Seconds {
791 t.Fatalf("entries were not priced: %v vs %v", many.Seconds, none.Seconds)
792 }
793 if r := BudgetCheck(100, -1, m, 16<<20, CutWindowDefault, t11Now); r.Check.OK || r.AttemptsAdmitted != 0 || r.Check.Reason != "the archive's file count could not be measured" {
794 t.Fatalf("%+v", r)
795 }
796 tr := passingTarget(t11Now)
797 tr.RecordBudget(100, 50000, m, many)
798 if tr.Entries != 50000 || tr.Rates.FsyncRate != 1600 || !tr.FsyncAt.Equal(m.Fsync.At) {
799 t.Fatalf("RecordBudget dropped the entry evidence: %+v", tr)
800 }
801}
802
803// The release a promotion pins is one the target checked (T7): a target
804// report with no release, or one for another release than the source now
805// runs, never promotes (Task 14 review finding 3).
806func TestReadyNeedsTheReleaseTheTargetChecked(t *testing.T) {
807 now := t11Now
808 if ok, why := Ready(0, 16<<20, passingTarget(now), goodSource(now), now); !ok {
809 t.Fatal(why)
810 }
811 tr := passingTarget(now)
812 tr.Inputs.Release = ""
813 if ok, why := Ready(0, 16<<20, tr, goodSource(now), now); ok || why != "the target has not checked a release yet" {
814 t.Fatalf("no release checked: %v %q", ok, why)
815 }
816 src := goodSource(now)
817 src.Release = "v8"
818 if ok, why := Ready(0, 16<<20, passingTarget(now), src, now); ok || why != "the target checked release v7 and the source now runs v8" {
819 t.Fatalf("release drift: %v %q", ok, why)
820 }
821 src.Release = ""
822 tr = passingTarget(now)
823 tr.Inputs.Release = ""
824 if ok, _ := Ready(0, 16<<20, tr, src, now); ok {
825 t.Fatal("two empty releases promoted")
826 }
827}
828
829// Measurements.Rates names the stale rate as data (*NoFreshRateError), and
830// BudgetCheck builds its reason from that, never from an error's text.
831func TestNoFreshRateIsTyped(t *testing.T) {
832 m := goodMeasurements()
833 m.S3 = MeasuredRate{}
834 _, err := m.Rates(t11Now, 16<<20)
835 var nf *NoFreshRateError
836 if !errors.As(err, &nf) || nf.Which != "regional S3" {
837 t.Fatalf("%v", err)
838 }
839 if r := BudgetCheck(100, 0, m, 16<<20, CutWindowDefault, t11Now); r.Check.Reason != NoFreshRate("regional S3") {
840 t.Fatalf("%q", r.Check.Reason)
841 }
842}
843
844// Final review finding 4: an archive reference pins its objects only by
845// version ids that name one immutable version each: neither "" nor the
846// suspended bucket's "null".
847func TestArchiveRefPinnedNeedsImmutableVersions(t *testing.T) {
848 good := ArchiveRef{ManifestKey: "m", ManifestVersion: "mv1", SealedObjectKey: "s", SealedVersion: "sv1"}
849 if !good.Pinned() {
850 t.Fatal("a reference with real version ids is not pinned")
851 }
852 for _, c := range []struct{ manifest, sealed string }{{"null", "sv1"}, {"mv1", "null"}, {"", "sv1"}, {"mv1", ""}, {"null", "null"}} {
853 r := good
854 r.ManifestVersion, r.SealedVersion = c.manifest, c.sealed
855 if r.Pinned() {
856 t.Errorf("versions %q and %q read as pinned", c.manifest, c.sealed)
857 }
858 }
859}
860
861// Final review finding 2: the target's report priced the cut (T11) and
862// checked the disk (T6) for the database size it saw. When the source's
863// report now carries a larger one (its local file, or the logical size of
864// its replica's newest LTX), the target's evidence no longer covers it and
865// Ready refuses until the target reports again. The review's example: a
866// fresh report for 100 MiB, then the database grows to 2.5 GiB, which at
867// 10/250/500 MiB/s no longer fits the 120-second window.
868func TestReadyRefusesADatabaseGrownSinceTheTargetPriced(t *testing.T) {
869 now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
870 target := passingTarget(now.Add(-time.Minute))
871 m := Measurements{
872 Cross: MeasuredRate{MiBPerS: 10, At: now.Add(-time.Minute)},
873 S3: MeasuredRate{MiBPerS: 250, At: now.Add(-time.Minute)},
874 Disk: MeasuredRate{MiBPerS: 500, At: now.Add(-time.Minute)},
875 Fsync: FilesRate{PerS: 1600, At: now.Add(-time.Minute)},
876 }
877 if res := BudgetCheck(2560, 0, m, 16<<20, 120, now); res.Check.OK {
878 t.Fatalf("the example does not overrun its window: %+v", res)
879 }
880 const grown = 2560 << 20
881 for name, src := range map[string]func(*SourceReport){
882 "local file": func(s *SourceReport) { s.DBBytes = grown },
883 "replica": func(s *SourceReport) { s.ReplicaBytes = grown },
884 "one byte": func(s *SourceReport) { s.ReplicaBytes = 100<<20 + 1 },
885 } {
886 s := goodSource(now)
887 src(s)
888 ok, why := Ready(0, 16<<20, target, s, now)
889 if ok || !strings.HasPrefix(why, "T11: the database has grown to ") {
890 t.Errorf("%s: Ready = %v %q, want the grown-database refusal", name, ok, why)
891 }
892 }
893 want := "T11: the database has grown to 2560 MiB since the target priced 100 MiB. waiting for the target's next report"
894 s := goodSource(now)
895 s.DBBytes = grown
896 if _, why := Ready(0, 16<<20, target, s, now); why != want {
897 t.Errorf("reason %q, want %q", why, want)
898 }
899 // The size the target priced, or less, is covered.
900 for _, size := range []int64{0, 1, 100 << 20} {
901 s := goodSource(now)
902 s.DBBytes, s.ReplicaBytes = size, size
903 if ok, why := Ready(0, 16<<20, target, s, now); !ok {
904 t.Errorf("size %d: not ready: %s", size, why)
905 }
906 }
907}
908
909// The cut guard (Paul's ruling, 2026-10-11): T13 is the target's statement
910// that it can land the move and roll it back. Only a report that states it
911// passing is armed: a nil report, an older box's report that has no T13,
912// and a failing T13 all read unarmed, and an older report is not ready.
913func TestTargetReportArmedFailsClosed(t *testing.T) {
914 now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
915 var none *TargetReport
916 if none.Armed() {
917 t.Fatal("a missing report is armed")
918 }
919 r := passingTarget(now)
920 if !r.Armed() {
921 t.Fatal("a report with a passing T13 is not armed")
922 }
923 r.Checks[CheckArmed] = Check{Reason: "not armed"}
924 if r.Armed() {
925 t.Fatal("a failing T13 is armed")
926 }
927 old := passingTarget(now)
928 delete(old.Checks, CheckArmed)
929 if old.Armed() {
930 t.Fatal("an older box's report, with no T13, is armed")
931 }
932 if ok, why := Ready(0, 16<<20, old, goodSource(now), now); ok || why != "T13 was not reported" {
933 t.Fatalf("an older box's report: ready %v (%q)", ok, why)
934 }
935 // The wire form: an older box's JSON has no T13 and reads unarmed.
936 body, err := json.Marshal(old)
937 if err != nil {
938 t.Fatal(err)
939 }
940 var back TargetReport
941 if err := json.Unmarshal(body, &back); err != nil || back.Armed() {
942 t.Fatalf("round trip of an older report: armed %v, %v", back.Armed(), err)
943 }
944}
945