amadan

carlos / platform Public

moves: ReplicaSource, MirrorPrefixKeep, the archive's entries in the T11 budget; edge: ServeUncached (§7.3)

moves.ReplicaSource binds a move's source copy to the claim holder's
location and generation and admits only that generation's LTX level
files; lease.MirrorPrefixKeep mirrors the keys it admits and nothing
else. Ruling T7-budget: Estimate and BudgetCheck take the archive's
entry count and Rates a per-file fsync rate, Archive = max(2,
entries/fsync); the target report carries both, and Ready re-prices
them. edge.ServeUncached is ServeHTTP with the response cache bypassed
both ways, for the source's S1 probe.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b1c27d87a6baf3d6917f877c4342881e5ab7a6bd parent 5832a73
12 files changed, +429 −58
  • docs/superpowers/plans/2026-10-06-moves-phase1-copy.md +1 −0
  • internal/edge/door.go +10 −0
  • internal/edge/edge.go +15 −1
  • internal/edge/edge_test.go +72 −0
  • internal/lease/mirror.go +17 −0
  • internal/lease/mirror_test.go +32 −0
  • internal/moves/budget.go +28 −10
  • internal/moves/budget_test.go +58 −12
  • internal/moves/readiness.go +51 −21
  • internal/moves/readiness_test.go +46 −14
  • internal/moves/source.go +53 −0
  • internal/moves/source_test.go +46 −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 9c0d3c7b3..1196e17bf 100644
--- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
+++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
@@ -152,6 +152,7 @@ Columns: id, file, text, status.
| MV-N110 | internal/release/materializeheld.go | release: a held publication needs its authority | pending |
| MV-N111 | internal/activator/backend_exec.go | activator: a strict child of this host is still running in its group | pending |
| MV-N112 | internal/activator/backend_exec.go | activator: this host's last strict group is still being cleaned up | pending |
+| MV-N113 | internal/moves/readiness.go | the archive's file count could not be measured | pending |
## Plan 2
diff --git a/internal/edge/door.go b/internal/edge/door.go
index c9438f5bb..9645df948 100644
--- a/internal/edge/door.go
+++ b/internal/edge/door.go
@@ -59,6 +59,16 @@ func doorCallerFrom(ctx context.Context) (doorCaller, bool) {
return c, ok
}
+// uncachedProbeKey marks a request ServeUncached is answering: the response
+// cache is neither read nor written for it. A context value, like
+// doorCallerKey, so nothing outside this process can set it.
+type uncachedProbeKey struct{}
+
+func uncachedProbe(ctx context.Context) bool {
+ v, _ := ctx.Value(uncachedProbeKey{}).(bool)
+ return v
+}
+
// doorOwnsRoute is the whole ownership rule: the route's account and app are
// the caller's.
func doorOwnsRoute(c doorCaller, rt registry.Route) bool {
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index a4222f36b..12921e1ad 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -1341,6 +1341,15 @@ func firstSegmentClean(p string) string {
return firstSegment(path.Clean(p))
}
+// ServeUncached is ServeHTTP with the response cache bypassed in both
+// directions: no cached answer is served and nothing answered is stored,
+// while a hibernating route is still woken through the activator. It is
+// the source's S1 probe (moves spec §7.3, round-4 finding 11): a cached 200
+// must never certify an instance that cannot wake.
+func (e *Edge) ServeUncached(w http.ResponseWriter, r *http.Request) {
+ e.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), uncachedProbeKey{}, true)))
+}
+
func (e *Edge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
host := r.Host
if h, _, err := net.SplitHostPort(host); err == nil {
@@ -1648,8 +1657,13 @@ func (e *Edge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// instead of being answered from cache, which is the invariant this cache
// exists to hold: the status host caches for itself; its responses never
// enter the shared response cache.
+ //
+ // A request ServeUncached marks is never cacheable either: a move's
+ // readiness probe must hear the instance's own process (a hibernating
+ // one is woken below), and what it hears must not be served to anyone
+ // else from the cache.
cacheable := e.cache != nil && (r.Method == http.MethodGet || r.Method == http.MethodHead) &&
- up.route.Kind != "status"
+ up.route.Kind != "status" && !uncachedProbe(r.Context())
var key CacheKey
if cacheable {
key = CacheKey{Host: host, Method: r.Method, Path: r.URL.Path, RawQuery: r.URL.RawQuery}
diff --git a/internal/edge/edge_test.go b/internal/edge/edge_test.go
index 2f92f98dc..0ef9fc46f 100644
--- a/internal/edge/edge_test.go
+++ b/internal/edge/edge_test.go
@@ -2324,3 +2324,75 @@ func TestSidecarRouteIsNeverServable(t *testing.T) {
t.Fatalf("GET %s = %d, want 404 — a sidecar must be unreachable", host, code)
}
}
+
+// wakeStub is a hibernation activator that answers every request itself:
+// with the wake failure when fail is set, else with a cacheable 200.
+type wakeStub struct {
+ fail bool
+ wakes int
+}
+
+func (a *wakeStub) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
+ a.wakes++
+ if a.fail {
+ MarkAnswered(r)
+ http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
+ return true
+ }
+ w.Header().Set("Cache-Control", "public, max-age=300")
+ w.WriteHeader(http.StatusOK)
+ _, _ = w.Write([]byte("woken"))
+ return true
+}
+func (a *wakeStub) RetireExcept(map[string]bool) {}
+
+// TestServeUncachedBypassesTheCacheAndWakes: a move's S1 probe (moves spec
+// §7.3, round-4 finding 11) must hear from the instance's own process, never
+// from the response cache. With the cache primed with a 200 for the path and
+// a wake that fails, ServeHTTP answers the cached 200 while ServeUncached
+// wakes and answers the failure; and what ServeUncached answers is never
+// stored, where ServeHTTP would have stored it.
+func TestServeUncachedBypassesTheCacheAndWakes(t *testing.T) {
+ reg := openReg(t)
+ rt := registry.Route{Host: "sleepy.example.com", Kind: "instance", Socket: t.TempDir() + "/x.sock", Status: "ready", Hibernate: true, App: "web"}
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ act := &wakeStub{fail: true}
+ e, err := New(reg, act)
+ if err != nil {
+ t.Fatal(err)
+ }
+ cache := NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)
+ e.EnableCache(cache)
+ cache.Put(CacheKey{Host: rt.Host, Method: http.MethodGet, Path: "/ready"}, "",
+ &CachedResponse{StatusCode: http.StatusOK, Header: http.Header{}, Body: []byte("cached"), StoredAt: time.Now(), TTL: time.Hour})
+
+ rec := httptest.NewRecorder()
+ e.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "http://sleepy.example.com/ready", nil))
+ if rec.Code != http.StatusOK || rec.Body.String() != "cached" || act.wakes != 0 {
+ t.Fatalf("ServeHTTP: %d %q after %d wakes, want the cached 200 and no wake", rec.Code, rec.Body.String(), act.wakes)
+ }
+ rec = httptest.NewRecorder()
+ e.ServeUncached(rec, httptest.NewRequest(http.MethodGet, "http://sleepy.example.com/ready", nil))
+ if rec.Code != http.StatusServiceUnavailable || act.wakes != 1 {
+ t.Fatalf("ServeUncached: %d after %d wakes, want the wake's 503", rec.Code, act.wakes)
+ }
+
+ // A wake that answers a cacheable 200: ServeUncached stores nothing.
+ act.fail = false
+ rec = httptest.NewRecorder()
+ e.ServeUncached(rec, httptest.NewRequest(http.MethodGet, "http://sleepy.example.com/fresh", nil))
+ if rec.Code != http.StatusOK || rec.Body.String() != "woken" {
+ t.Fatalf("ServeUncached: %d %q", rec.Code, rec.Body.String())
+ }
+ probe := httptest.NewRequest(http.MethodGet, "http://sleepy.example.com/fresh", nil)
+ if _, hit := cache.GetFor(CacheKey{Host: rt.Host, Method: http.MethodGet, Path: "/fresh"}, probe); hit {
+ t.Fatal("ServeUncached stored its answer in the response cache")
+ }
+ // The control: ServeHTTP stores the same answer.
+ e.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "http://sleepy.example.com/fresh", nil))
+ if _, hit := cache.GetFor(CacheKey{Host: rt.Host, Method: http.MethodGet, Path: "/fresh"}, probe); !hit {
+ t.Fatal("the control did not store: the cacheable answer is not cacheable, so the test proves nothing")
+ }
+}
diff --git a/internal/lease/mirror.go b/internal/lease/mirror.go
index 9e0ab9e11..ffc11d3d7 100644
--- a/internal/lease/mirror.go
+++ b/internal/lease/mirror.go
@@ -173,6 +173,16 @@ type copyReceipt struct {
// those objects once more. It can never produce a wrong skip, because a
// skip needs the destination to still list the ETag the copy produced.
func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstBucket, dstPrefix string, rc Receipts) (MirrorRound, error) {
+ return MirrorPrefixKeep(ctx, m, srcBucket, srcPrefix, dstBucket, dstPrefix, rc, nil)
+}
+
+// MirrorPrefixKeep is MirrorPrefix over the source keys keep admits: keep
+// is asked each source key relative to srcPrefix, and a key it refuses is
+// treated as absent, so it is never copied and a destination copy of it is
+// deleted. A move's lane (moves.ReplicaSource) mirrors one generation's
+// LTX files and nothing else this way. A nil keep admits every key, which
+// is MirrorPrefix.
+func MirrorPrefixKeep(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstBucket, dstPrefix string, rc Receipts, keep func(rel string) bool) (MirrorRound, error) {
// Both prefixes are DIRECTORY boundaries, normalized here to end in
// "/": activator.HibernatePrefix carries none (path.Join), MovesPrefix
// carries one, and relative-key math over a slashless prefix would both
@@ -188,6 +198,13 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB
if err != nil {
return MirrorRound{}, fmt.Errorf("mirror: list source %s/%s: %w", srcBucket, srcPrefix, err)
}
+ if keep != nil {
+ for k := range src {
+ if !keep(k[len(srcPrefix):]) {
+ delete(src, k)
+ }
+ }
+ }
dst, err := m.List(ctx, dstBucket, dstPrefix)
if err != nil {
return MirrorRound{}, fmt.Errorf("mirror: list staging %s/%s: %w", dstBucket, dstPrefix, err)
diff --git a/internal/lease/mirror_test.go b/internal/lease/mirror_test.go
index 3abf06856..9b25912bc 100644
--- a/internal/lease/mirror_test.go
+++ b/internal/lease/mirror_test.go
@@ -621,3 +621,35 @@ func TestPlainS3MirrorRecordsNoVersions(t *testing.T) {
t.Fatal("s3Mirror implements VersionedPutter")
}
}
+
+// MirrorPrefixKeep mirrors only the source keys keep admits: a refused key
+// is never copied, and a destination copy of it is deleted as if the source
+// no longer had it. A nil keep is MirrorPrefix.
+func TestMirrorPrefixKeepMirrorsOnlyKeptKeys(t *testing.T) {
+ ctx := context.Background()
+ f := newFakeMirror()
+ f.put("src", "loc/0000/a.ltx", "a")
+ f.put("src", "loc/0001/b.ltx", "b")
+ f.put("src", "loc/g-old/0000/c.ltx", "c")
+ f.put("src", "loc/generation.json", "g")
+ f.put("dst", "lane/g-old/0000/c.ltx", "c") // a stale copy of a refused key
+ keep := func(rel string) bool { return strings.Count(rel, "/") == 1 && !strings.HasPrefix(rel, "g-") }
+ round, err := MirrorPrefixKeep(ctx, f, "src", "loc", "dst", "lane", Receipts{}, keep)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, _ := f.List(ctx, "dst", "lane/")
+ if len(got) != 2 || got["lane/0000/a.ltx"].ETag != "a" || got["lane/0001/b.ltx"].ETag != "b" {
+ t.Fatalf("staged %v, want only the two kept level files", got)
+ }
+ if round.Changed != 3 {
+ t.Fatalf("round %+v: want two copies and one delete", round)
+ }
+ // nil keep: MirrorPrefix exactly, every key.
+ if _, err := MirrorPrefixKeep(ctx, f, "src", "loc", "dst", "all", Receipts{}, nil); err != nil {
+ t.Fatal(err)
+ }
+ if all, _ := f.List(ctx, "dst", "all/"); len(all) != 4 {
+ t.Fatalf("nil keep staged %d keys, want 4", len(all))
+ }
+}
diff --git a/internal/moves/budget.go b/internal/moves/budget.go
index dbd8238c9..dcc31851e 100644
--- a/internal/moves/budget.go
+++ b/internal/moves/budget.go
@@ -21,6 +21,7 @@ const (
SettleSeconds = 20.0
AdmitSeconds = 2.0 // the admitting CAS, the purge and generation.json
MirrorBarMiB = 16.0 // §7.3: the final mirror round is at most this much
+ ArchiveSeconds = 2.0 // the archive phase's floor (ruling T7-budget)
)
// CutEstimate is a cut priced phase by phase.
@@ -35,16 +36,22 @@ type CutEstimate struct {
// path, regional S3 (reads and writes from a box in the bucket's region)
// and the box's disk. MirrorBar is the final mirror round's size bar in
// MiB; zero means MirrorBarMiB. It is the one knob Task 11 turns
-// (CARLOS_MOVE_READY_DELTA_BYTES, ruling F28(e)).
+// (CARLOS_MOVE_READY_DELTA_BYTES, ruling F28(e)). FsyncRate is the target
+// disk's per-file sync rate in files per second (ruling T7-budget): the
+// landing's extract syncs every file, so the archive's entry count prices
+// the Archive phase.
type Rates struct {
Cross, S3, Disk float64
MirrorBar float64
+ FsyncRate float64
}
// ExampleRates are the rates the spec's worked example uses (§7.3 T11):
-// 10 MiB/s across regions, 250 MiB/s to regional S3, 500 MiB/s on disk.
-// Tests and documentation only: T11 reads measured rates.
-var ExampleRates = Rates{Cross: 10, S3: 250, Disk: 500}
+// 10 MiB/s across regions, 250 MiB/s to regional S3, 500 MiB/s on disk,
+// and (not in the spec's example, which has no files) 1600 files a second
+// synced, about what Task 7 measured for its 16-worker pool. Tests and
+// documentation only: T11 reads measured rates.
+var ExampleRates = Rates{Cross: 10, S3: 250, Disk: 500, FsyncRate: 1600}
// Estimate prices a cut for a database of dbMiB at the given rates. The
// final mirror is within the bar across regions. On the target every
@@ -52,13 +59,20 @@ var ExampleRates = Rates{Cross: 10, S3: 250, Disk: 500}
// disk, bar the verification, which reads the disk alone. Each probe
// attempt restores the prepared generation again and pays the admission;
// a retried probe also uploads the whole database as a fresh generation.
+// The archive phase is at least ArchiveSeconds, and longer when the
+// archive's entries take longer than that to sync on the target at
+// r.FsyncRate (ruling T7-budget): max(2, entries/FsyncRate).
+//
// An input that is not a usable measurement (a rate that is not a positive
-// finite number, a negative, NaN or infinite database size or bar) prices
-// the cut at infinity, so it is refused: a negative rate must never price a
-// phase below zero and let a move in.
-func Estimate(dbMiB float64, r Rates) CutEstimate {
+// finite number, a negative, NaN or infinite database size or bar, a
+// negative entry count, or entries with no usable fsync rate) prices the cut
+// at infinity, so it is refused: a negative rate must never price a phase
+// below zero and let a move in. With no entries the fsync rate is not
+// needed, but a negative, NaN or infinite one is still refused.
+func Estimate(dbMiB float64, entries int, r Rates) CutEstimate {
if !usable(r.Cross) || !usable(r.S3) || !usable(r.Disk) ||
- !finite(dbMiB) || dbMiB < 0 || !finite(r.MirrorBar) || r.MirrorBar < 0 {
+ !finite(dbMiB) || dbMiB < 0 || !finite(r.MirrorBar) || r.MirrorBar < 0 ||
+ entries < 0 || !finite(r.FsyncRate) || r.FsyncRate < 0 || (entries > 0 && !usable(r.FsyncRate)) {
inf := math.Inf(1)
return CutEstimate{Stop: inf, Checkpoint: inf, Mirror: inf, Archive: inf, Render: inf, Open: inf}
}
@@ -66,8 +80,12 @@ func Estimate(dbMiB float64, r Rates) CutEstimate {
if bar == 0 {
bar = MirrorBarMiB
}
+ archive := ArchiveSeconds
+ if entries > 0 {
+ archive = math.Max(ArchiveSeconds, float64(entries)/r.FsyncRate)
+ }
return CutEstimate{
- Stop: 2, Checkpoint: 1, Archive: 2, Render: 1, Open: OpenBudget,
+ Stop: 2, Checkpoint: 1, Archive: archive, Render: 1, Open: OpenBudget,
Mirror: bar / r.Cross,
RestoreStaged: dbMiB/r.S3 + dbMiB/r.Disk,
SnapshotBuild: dbMiB/r.Disk + dbMiB/r.S3,
diff --git a/internal/moves/budget_test.go b/internal/moves/budget_test.go
index 25491628d..794aa5ee0 100644
--- a/internal/moves/budget_test.go
+++ b/internal/moves/budget_test.go
@@ -12,7 +12,7 @@ import (
func TestReadinessBudget(t *testing.T) {
// a tiny instance: every fixed cost at the floor of its measured rate
- tiny := Estimate(1, ExampleRates)
+ tiny := Estimate(1, 0, ExampleRates)
// rev 7.3 required three attempts always: 150 s of probing alone, which
// the default window never held
if old := tiny.Fixed() + ProbeAttemptsMax*(tiny.Probe+tiny.Settle); old <= CutWindowDefault {
@@ -37,7 +37,7 @@ func TestReadinessBudget(t *testing.T) {
// R5-5: the spec's example, a 2 GiB database at the example rates. Rev
// 7.5's full-snapshot transfer alone takes 205 s and could never fit;
// the delta within the bar and the target's build do, with one attempt
- big := Estimate(2048, ExampleRates)
+ big := Estimate(2048, 0, ExampleRates)
fullTransfer := 2048.0 / ExampleRates.Cross
if fullTransfer < CutWindowDefault {
t.Fatalf("the example no longer illustrates R5-5")
@@ -64,7 +64,7 @@ func TestReadinessBudget(t *testing.T) {
// the review's 2.5 GiB (R10-8): rev 7.10's arithmetic admitted it at
// 113.8 s; with the probe's own restore it is 129.2 s, past the window,
// and refused
- bigger := Estimate(2560, ExampleRates)
+ bigger := Estimate(2560, 0, ExampleRates)
old := bigger.Fixed() + ProbeSeconds + SettleSeconds
if math.Abs(old-113.8) > 0.1 {
t.Fatalf("rev 7.10's arithmetic for 2.5 GiB is %.1f s; the review says 113.8 s", old)
@@ -85,7 +85,7 @@ func TestReadinessBudget(t *testing.T) {
// admit one attempt (108.68 s with the probe); the fourth pass takes the
// same workload to 124.04 s, past the window
r := Rates{Cross: 10, S3: 200, Disk: 200}
- three := Estimate(3072, r)
+ three := Estimate(3072, 0, r)
three.RestoreStaged, three.SnapshotBuild, three.Anchor, three.Verify = 3072/200.0, 3072/200.0, 3072/200.0, 0 // rev 7.6's arithmetic
three.RestoreAttempt, three.Admit, three.FreshUpload = 0, 0, 0 // rev 7.6's arithmetic knew no per-attempt restore
if n := three.AttemptsAdmitted(CutWindowDefault); n != 1 {
@@ -98,13 +98,13 @@ func TestReadinessBudget(t *testing.T) {
}
// rev 7.7 verifies the built database instead of restoring it again:
// the same workload is refused honestly, by its own arithmetic
- if n := Estimate(3072, r).AttemptsAdmitted(CutWindowDefault); n != 0 {
- t.Fatalf("rev 7.7 admits %d attempts for the review's workload, want 0 (fixed %.1f s)", n, Estimate(3072, r).Fixed())
+ if n := Estimate(3072, 0, r).AttemptsAdmitted(CutWindowDefault); n != 0 {
+ t.Fatalf("rev 7.7 admits %d attempts for the review's workload, want 0 (fixed %.1f s)", n, Estimate(3072, 0, r).Fixed())
}
}
func TestBudgetTotalAndDeadline(t *testing.T) {
- e := Estimate(2048, ExampleRates)
+ e := Estimate(2048, 0, ExampleRates)
n := e.AttemptsAdmitted(CutWindowDefault)
if n != 1 || math.Abs(e.Total(n)-117.9) > 0.1 {
t.Fatalf("attempts %d total %.1f", n, e.Total(n))
@@ -120,18 +120,18 @@ func TestBudgetTotalAndDeadline(t *testing.T) {
// (CARLOS_MOVE_READY_DELTA_BYTES, Ruling F28(e)): zero means the default,
// and the bar scales only the mirror.
func TestEstimateMirrorBar(t *testing.T) {
- def := Estimate(2048, ExampleRates)
+ def := Estimate(2048, 0, ExampleRates)
if want := MirrorBarMiB / ExampleRates.Cross; def.Mirror != want {
t.Fatalf("default mirror %.2f s, want %.2f s", def.Mirror, want)
}
explicit := ExampleRates
explicit.MirrorBar = MirrorBarMiB
- if got := Estimate(2048, explicit); got != def {
+ if got := Estimate(2048, 0, explicit); got != def {
t.Fatalf("an explicit default bar changed the estimate: %+v vs %+v", got, def)
}
wide := ExampleRates
wide.MirrorBar = 64
- w := Estimate(2048, wide)
+ w := Estimate(2048, 0, wide)
if math.Abs(w.Mirror-6.4) > 1e-9 {
t.Fatalf("a 64 MiB bar at 10 MiB/s mirrors in %.2f s, want 6.4", w.Mirror)
}
@@ -180,7 +180,7 @@ func TestEstimateInvalidInputsRefuse(t *testing.T) {
"db and s3 negative": {-1, with(func(r *Rates) { r.S3 = -250 })},
}
for name, c := range cases {
- e := Estimate(c.db, c.r)
+ e := Estimate(c.db, 0, c.r)
if n := e.AttemptsAdmitted(CutWindowDefault); n != 0 {
t.Errorf("%s admits %d attempts, want 0", name, n)
}
@@ -192,7 +192,53 @@ func TestEstimateInvalidInputsRefuse(t *testing.T) {
}
}
// the boundary the refusals must not touch: an empty database is valid
- if n := Estimate(0, ExampleRates).AttemptsAdmitted(CutWindowDefault); n != 2 {
+ if n := Estimate(0, 0, ExampleRates).AttemptsAdmitted(CutWindowDefault); n != 2 {
t.Errorf("an empty database admits %d attempts, want 2", n)
}
}
+
+// Ruling T7-budget: the landing's fsync scales with the archive's entry
+// count, so the Archive term is max(2, entries/FsyncRate) seconds. Every
+// new input keeps the invalid-input refusal.
+func TestEstimateArchiveScalesWithEntries(t *testing.T) {
+ r := ExampleRates
+ if got := Estimate(2048, 0, r).Archive; got != 2 {
+ t.Fatalf("no entries: Archive %v, want 2", got)
+ }
+ if got := Estimate(2048, 100, r).Archive; got != 2 {
+ t.Fatalf("a few entries: Archive %v, want the 2 s floor", got)
+ }
+ want := 32000 / r.FsyncRate
+ if got := Estimate(2048, 32000, r).Archive; got != want || got <= 2 {
+ t.Fatalf("32000 entries: Archive %v, want %v", got, want)
+ }
+ if Estimate(2048, 32000, r).Fixed() <= Estimate(2048, 0, r).Fixed() {
+ t.Fatal("entries did not reach the fixed cost")
+ }
+ // No entries needs no fsync rate.
+ noRate := r
+ noRate.FsyncRate = 0
+ if got := Estimate(2048, 0, noRate).Archive; got != 2 {
+ t.Fatalf("no entries, no rate: Archive %v, want 2", got)
+ }
+ nan, inf := math.NaN(), math.Inf(1)
+ for name, c := range map[string]struct {
+ entries int
+ fsync float64
+ }{
+ "negative entries": {-1, r.FsyncRate},
+ "entries and no rate": {10, 0},
+ "negative rate": {10, -1600},
+ "negative rate, no files": {0, -1600},
+ "NaN rate": {10, nan},
+ "NaN rate, no files": {0, nan},
+ "Inf rate": {10, inf},
+ } {
+ rr := r
+ rr.FsyncRate = c.fsync
+ e := Estimate(2048, c.entries, rr)
+ if !math.IsInf(e.Fixed(), 1) || e.AttemptsAdmitted(1e9) != 0 {
+ t.Errorf("%s: fixed %v, %d attempts; want refused", name, e.Fixed(), e.AttemptsAdmitted(1e9))
+ }
+ }
+}
diff --git a/internal/moves/readiness.go b/internal/moves/readiness.go
index d3077d6dd..000ffdecd 100644
--- a/internal/moves/readiness.go
+++ b/internal/moves/readiness.go
@@ -72,15 +72,23 @@ type TargetReport struct {
DBMiB float64 `json:"db_mib"`
S3At time.Time `json:"s3_at"`
DiskAt time.Time `json:"disk_at"`
- Notes []string `json:"notes,omitempty"`
+ // Entries is the archive's entry count the Archive phase was priced
+ // for, and FsyncAt when the per-file sync rate (Rates.FsyncRate) was
+ // measured (ruling T7-budget).
+ Entries int `json:"entries"`
+ FsyncAt time.Time `json:"fsync_at"`
+ Notes []string `json:"notes,omitempty"`
}
-// RecordBudget copies a BudgetCheck result and the measurements and database
-// size it was priced from into the report, the fields Ready re-checks.
-func (r *TargetReport) RecordBudget(dbMiB float64, m Measurements, res BudgetResult) {
+// RecordBudget copies a BudgetCheck result and the measurements, database
+// size and entry count it was priced from into the report, the fields Ready
+// re-checks.
+func (r *TargetReport) RecordBudget(dbMiB float64, entries int, m Measurements, res BudgetResult) {
r.DBMiB = dbMiB
+ r.Entries = entries
r.S3At = m.S3.At
r.DiskAt = m.Disk.At
+ r.FsyncAt = m.Fsync.At
r.Rates = res.Rates
r.AttemptsAdmitted = res.AttemptsAdmitted
}
@@ -220,15 +228,16 @@ func Ready(roundDelta, bar int64, target *TargetReport, source *SourceReport, no
// refreshed on their own schedules) but never higher: a report cannot claim
// more attempts than its own evidence buys.
func budgetEvidence(target *TargetReport, source *SourceReport, bar int64, now time.Time) string {
- if !usable(target.Rates.Cross) || !usable(target.Rates.S3) || !usable(target.Rates.Disk) {
+ if !usable(target.Rates.Cross) || !usable(target.Rates.S3) || !usable(target.Rates.Disk) || !usable(target.Rates.FsyncRate) {
return "T11: the target's report carries a rate that is not a usable number"
}
m := Measurements{
Cross: source.CrossMeasurement(),
S3: MeasuredRate{MiBPerS: target.Rates.S3, At: target.S3At},
Disk: MeasuredRate{MiBPerS: target.Rates.Disk, At: target.DiskAt},
+ Fsync: FilesRate{PerS: target.Rates.FsyncRate, At: target.FsyncAt},
}
- res := BudgetCheck(target.DBMiB, m, bar, float64(target.CutBudgetS), now)
+ res := BudgetCheck(target.DBMiB, target.Entries, m, bar, float64(target.CutBudgetS), now)
if !res.Check.OK {
return "T11: " + res.Check.Reason
}
@@ -265,12 +274,21 @@ func RateOf(bytes int64, d time.Duration, at time.Time) MeasuredRate {
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.
+// FilesRate is a per-file sync measurement: files synced per second, with
+// when and over how many files it was taken (ruling T7-budget).
+type FilesRate struct {
+ PerS float64 `json:"per_s"`
+ At time.Time `json:"at"`
+ Files int `json:"files"`
+}
+
+// Measurements are the rates T11 prices a cut from. The target measures
+// regional S3, its disk and its per-file sync rate itself. 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
+ Fsync FilesRate
}
// BarMiB is a ready-delta bar in bytes as the MiB that Rates.MirrorBar
@@ -296,12 +314,21 @@ func (m Measurements) Rates(now time.Time, barBytes int64) (Rates, error) {
{"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 {
+ if !fresh(c.m.MiBPerS, c.m.At, now) {
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
+ if !fresh(m.Fsync.PerS, m.Fsync.At, now) {
+ return Rates{}, fmt.Errorf("no fresh rate measurement for %s", "the disk's per-file sync")
+ }
+ return Rates{Cross: m.Cross.MiBPerS, S3: m.S3.MiBPerS, Disk: m.Disk.MiBPerS, MirrorBar: BarMiB(barBytes), FsyncRate: m.Fsync.PerS}, nil
+}
+
+// fresh is a usable rate measured within MeasurementFreshness of now and no
+// more than MeasurementSkew ahead of it.
+func fresh(rate float64, at, now time.Time) bool {
+ age := now.Sub(at)
+ return !at.IsZero() && usable(rate) && age <= MeasurementFreshness && age >= -MeasurementSkew
}
// BudgetResult is T11 decided: the check, the attempts the window admits,
@@ -313,17 +340,20 @@ type BudgetResult struct {
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 {
+// BudgetCheck is T11 (§7.3): price a cut of a dbMiB database whose archive
+// holds entries files 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 negative entry count,
+// 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, entries int, 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 entries < 0 {
+ return fail("the archive's file count could not be measured")
+ }
if !finite(windowSeconds) || windowSeconds <= 0 {
return fail("the cut window is not a usable number of seconds")
}
@@ -331,7 +361,7 @@ func BudgetCheck(dbMiB float64, m Measurements, barBytes int64, windowSeconds fl
if err != nil {
return fail(err.Error())
}
- e := Estimate(dbMiB, rates)
+ e := Estimate(dbMiB, entries, rates)
n := e.AttemptsAdmitted(windowSeconds)
need := e.Total(1)
res := BudgetResult{AttemptsAdmitted: n, Rates: rates, Seconds: need}
diff --git a/internal/moves/readiness_test.go b/internal/moves/readiness_test.go
index 8736b2a7d..2a6ba1fcf 100644
--- a/internal/moves/readiness_test.go
+++ b/internal/moves/readiness_test.go
@@ -29,12 +29,13 @@ func passingTarget(at time.Time) *TargetReport {
Cross: MeasuredRate{MiBPerS: 10, At: at.Add(-time.Minute)},
S3: MeasuredRate{MiBPerS: 250, At: at.Add(-time.Minute)},
Disk: MeasuredRate{MiBPerS: 500, At: at.Add(-time.Minute)},
+ Fsync: FilesRate{PerS: 1600, At: at.Add(-time.Minute)},
}
- res := BudgetCheck(100, m, 16<<20, 120, at)
+ res := BudgetCheck(100, 0, m, 16<<20, 120, at)
if !res.Check.OK {
panic("the test fixture does not fit its own window: " + res.Check.Reason)
}
- r.RecordBudget(100, m, res)
+ r.RecordBudget(100, 0, m, res)
return r
}
@@ -168,12 +169,13 @@ func TestReadyRefusesAnEstimateMadeUnderASmallerMirrorBar(t *testing.T) {
Cross: MeasuredRate{MiBPerS: 10, At: now.Add(-time.Minute)},
S3: MeasuredRate{MiBPerS: 250, At: tr.S3At},
Disk: MeasuredRate{MiBPerS: 500, At: tr.DiskAt},
+ Fsync: FilesRate{PerS: 1600, At: tr.FsyncAt},
}
- res := BudgetCheck(100, m, barBytes, 120, now)
+ res := BudgetCheck(100, 0, m, barBytes, 120, now)
if !res.Check.OK {
t.Fatalf("%+v", res)
}
- tr.RecordBudget(100, m, res)
+ tr.RecordBudget(100, 0, m, res)
return tr
}
// Priced at the 16 MiB default, the operator's bar raised to 32 MiB:
@@ -222,6 +224,7 @@ func TestReadyValidatesTheEvidenceItself(t *testing.T) {
c{fmt.Sprintf("disk rate %v", v), func(tr *TargetReport, _ *SourceReport) { tr.Rates.Disk = v }},
c{fmt.Sprintf("cross rate in the source %v", v), func(_ *TargetReport, sr *SourceReport) { sr.CrossRate = v }},
c{fmt.Sprintf("cross rate in the target %v", v), func(tr *TargetReport, _ *SourceReport) { tr.Rates.Cross = v }},
+ c{fmt.Sprintf("fsync rate %v", v), func(tr *TargetReport, _ *SourceReport) { tr.Rates.FsyncRate = v }},
)
}
expired := now.Add(-MeasurementFreshness - time.Second)
@@ -232,6 +235,11 @@ func TestReadyValidatesTheEvidenceItself(t *testing.T) {
c{"disk time missing", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = time.Time{} }},
c{"disk expired", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = expired }},
c{"disk from the future", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = now.Add(time.Hour) }},
+ c{"fsync time missing", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = time.Time{} }},
+ c{"fsync expired", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = expired }},
+ c{"fsync from the future", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = now.Add(time.Hour) }},
+ c{"entries negative", func(tr *TargetReport, _ *SourceReport) { tr.Entries = -1 }},
+ c{"too many entries for the window", func(tr *TargetReport, _ *SourceReport) { tr.Entries = 1 << 30 }},
c{"cross time missing", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = time.Time{} }},
c{"cross expired", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = expired }},
c{"cross from the future", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = now.Add(time.Hour) }},
@@ -260,6 +268,7 @@ func TestReadyValidatesTheEvidenceItself(t *testing.T) {
}{
{"s3", func(tr *TargetReport, _ *SourceReport) { tr.S3At = expired }, "regional S3"},
{"disk", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = expired }, "the disk"},
+ {"fsync", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = expired }, "the disk's per-file sync"},
{"cross", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = expired }, "the cross-region copy path"},
} {
tr, sr := passingTarget(now), goodSource(now)
@@ -396,6 +405,7 @@ func goodMeasurements() 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},
+ Fsync: FilesRate{PerS: 1600, At: at, Files: 512},
}
}
@@ -404,7 +414,7 @@ func TestMeasurementsNeedFreshUsableRates(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- if r.Cross != 10 || r.S3 != 250 || r.Disk != 500 || r.MirrorBar != 16 {
+ if r.Cross != 10 || r.S3 != 250 || r.Disk != 500 || r.MirrorBar != 16 || r.FsyncRate != 1600 {
t.Fatalf("%+v", r)
}
@@ -421,6 +431,9 @@ func TestMeasurementsNeedFreshUsableRates(t *testing.T) {
{"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"},
+ {"no fsync", func(m *Measurements) { m.Fsync = FilesRate{} }, "the disk's per-file sync"},
+ {"old fsync", func(m *Measurements) { m.Fsync.At = t11Now.Add(-MeasurementFreshness - time.Second) }, "the disk's per-file sync"},
+ {"NaN fsync", func(m *Measurements) { m.Fsync.PerS = math.NaN() }, "the disk's per-file sync"},
}
for _, c := range cases {
m := goodMeasurements()
@@ -470,7 +483,7 @@ func TestSourceReportCarriesTheCrossRate(t *testing.T) {
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)
+ res := BudgetCheck(100, 0, goodMeasurements(), 16<<20, CutWindowDefault, t11Now)
if !res.Check.OK || res.AttemptsAdmitted < 1 {
t.Fatalf("%+v", res)
}
@@ -483,7 +496,7 @@ func TestT11BudgetCheck(t *testing.T) {
// 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)
+ res = BudgetCheck(100000, 0, goodMeasurements(), 16<<20, CutWindowDefault, t11Now)
if res.Check.OK || res.AttemptsAdmitted != 0 || !strings.Contains(res.Check.Reason, "120") {
t.Fatalf("%+v", res)
}
@@ -491,19 +504,19 @@ func TestT11BudgetCheck(t *testing.T) {
// A missing measurement fails closed, naming which.
m := goodMeasurements()
m.Disk = MeasuredRate{}
- res = BudgetCheck(100, m, 16<<20, CutWindowDefault, t11Now)
+ res = BudgetCheck(100, 0, 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 {
+ if r := BudgetCheck(db, 0, 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 {
+ if r := BudgetCheck(100, 0, goodMeasurements(), 16<<20, w, t11Now); r.Check.OK || r.AttemptsAdmitted != 0 {
t.Fatalf("window %v: %+v", w, r)
}
}
@@ -514,8 +527,8 @@ func TestT11BudgetCheck(t *testing.T) {
// 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)
+ def := BudgetCheck(100, 0, m, 16<<20, CutWindowDefault, t11Now)
+ big := BudgetCheck(100, 0, m, 4<<30, CutWindowDefault, t11Now)
if !def.Check.OK {
t.Fatalf("%+v", def)
}
@@ -529,7 +542,7 @@ func TestT11BudgetCheckPricesTheOperatorsBar(t *testing.T) {
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)
+ zero := BudgetCheck(100, 0, m, 0, CutWindowDefault, t11Now)
if zero.Seconds != def.Seconds {
t.Fatalf("%v vs %v", zero.Seconds, def.Seconds)
}
@@ -537,7 +550,7 @@ func TestT11BudgetCheckPricesTheOperatorsBar(t *testing.T) {
// report leaves it with zero attempts, and Ready refuses it.
src := goodSource(t11Now)
tr := passingTarget(t11Now)
- tr.RecordBudget(100, m, big)
+ tr.RecordBudget(100, 0, m, big)
if tr.AttemptsAdmitted != 0 {
t.Fatalf("%d", tr.AttemptsAdmitted)
}
@@ -767,3 +780,22 @@ func TestMeasurementWindowEdges(t *testing.T) {
}
}
}
+
+// Ruling T7-budget: the archive's entry count is priced (and carried in the
+// report, so Ready re-prices it); a count that is not usable fails closed.
+func TestT11BudgetCheckPricesEntries(t *testing.T) {
+ m := goodMeasurements()
+ none := BudgetCheck(100, 0, m, 16<<20, CutWindowDefault, t11Now)
+ many := BudgetCheck(100, 50000, m, 16<<20, CutWindowDefault, t11Now)
+ if !none.Check.OK || many.Seconds <= none.Seconds {
+ t.Fatalf("entries were not priced: %v vs %v", many.Seconds, none.Seconds)
+ }
+ 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" {
+ t.Fatalf("%+v", r)
+ }
+ tr := passingTarget(t11Now)
+ tr.RecordBudget(100, 50000, m, many)
+ if tr.Entries != 50000 || tr.Rates.FsyncRate != 1600 || !tr.FsyncAt.Equal(m.Fsync.At) {
+ t.Fatalf("RecordBudget dropped the entry evidence: %+v", tr)
+ }
+}
diff --git a/internal/moves/source.go b/internal/moves/source.go
new file mode 100644
index 000000000..aa8cfd81b
--- /dev/null
+++ b/internal/moves/source.go
@@ -0,0 +1,53 @@
+package moves
+
+import (
+ "fmt"
+ "strconv"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/superfly/ltx"
+)
+
+// ReplicaSource is where a move's source copy reads from (§7.3, round-3
+// finding 5): the claim holder's location and generation, never the
+// legacy HibernatePrefix guess. keep admits, relative to prefix, only the
+// generation's own LTX files, `<level>/<min>-<max>.ltx` with a level from
+// 0000 to 0009, so the staging lane holds the generation's files flat
+// under replica/<level>/ and nothing else: no other generation (g-*/), no
+// child incarnation, no generation.json, lock.json or closed/.
+//
+// The holder's location never carries a trailing slash (Claim.Validate
+// accepts only instances/<host> or instances/<host>/<incarnation>), so the
+// separator is added here, as genselect.requestLocation does (round-4
+// finding 4). A generation of "" or hostclaim.RootGeneration is the root
+// layout: the level directories sit directly under the location.
+//
+// The generation prefix rule has twins: hostbox.GenerationPrefix (which
+// refuses the root generation and a holder with no session, so a source
+// copy cannot use it) and genselect's unexported prefix and coordinates.
+// The three change together.
+func ReplicaSource(h hostclaim.Holder) (prefix string, keep func(rel string) bool) {
+ base := strings.TrimSuffix(h.Location, "/") + "/"
+ prefix = base
+ if g := h.Generation; g != "" && g != hostclaim.RootGeneration {
+ prefix = base + "g-" + g + "/"
+ }
+ return prefix, levelFile
+}
+
+// levelFile is ReplicaSource's keep: exactly two path parts, a level
+// directory of four lowercase hex digits from 0000 to 0009, and an LTX
+// file name in its canonical form with 0 < min <= max.
+func levelFile(rel string) bool {
+ parts := strings.Split(rel, "/")
+ if len(parts) != 2 || len(parts[0]) != 4 {
+ return false
+ }
+ level, err := strconv.ParseUint(parts[0], 16, 16)
+ if err != nil || level > 9 || fmt.Sprintf("%04x", level) != parts[0] {
+ return false
+ }
+ min, max, err := ltx.ParseFilename(parts[1])
+ return err == nil && min != 0 && min <= max && ltx.FormatFilename(min, max) == parts[1]
+}
diff --git a/internal/moves/source_test.go b/internal/moves/source_test.go
new file mode 100644
index 000000000..26f0a3fe4
--- /dev/null
+++ b/internal/moves/source_test.go
@@ -0,0 +1,46 @@
+package moves
+
+import (
+ "testing"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+)
+
+func TestReplicaSourcePrefixAndKeep(t *testing.T) {
+ inc := "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
+ for _, tc := range []struct {
+ h hostclaim.Holder
+ want string
+ }{
+ {hostclaim.Holder{Location: "instances/h/" + inc, Generation: "cur"}, "instances/h/" + inc + "/g-cur/"},
+ {hostclaim.Holder{Location: "instances/h/" + inc + "/", Generation: "cur"}, "instances/h/" + inc + "/g-cur/"},
+ {hostclaim.Holder{Location: "instances/h", Generation: hostclaim.RootGeneration}, "instances/h/"},
+ {hostclaim.Holder{Location: "instances/h"}, "instances/h/"},
+ } {
+ if got, _ := ReplicaSource(tc.h); got != tc.want {
+ t.Errorf("ReplicaSource(%+v) prefix = %q, want %q", tc.h, got, tc.want)
+ }
+ }
+ _, keep := ReplicaSource(hostclaim.Holder{Location: "instances/h"})
+ for rel, want := range map[string]bool{
+ "0000/0000000000000001-0000000000000004.ltx": true,
+ "0009/0000000000000001-00000000000000ff.ltx": true,
+ "0001/0000000000000005-0000000000000005.ltx": true,
+ "000a/0000000000000001-0000000000000004.ltx": false, // level 10
+ "0000/0000000000000000-0000000000000004.ltx": false, // min 0
+ "0000/0000000000000005-0000000000000004.ltx": false, // min > max
+ "0000/0000000000000001-0000000000000004.LTX": false,
+ "0000/1-4.ltx": false, // not canonical
+ "00/0000000000000001-0000000000000004.ltx": false,
+ "g-x/0000/0000000000000001-0000000000000004.ltx": false,
+ inc + "/0000/0000000000000001-0000000000000004.ltx": false,
+ "generation.json": false,
+ "lock.json": false,
+ "closed/0000/0000000000000001-0000000000000004.ltx": false,
+ "0000/0000000000000001-0000000000000004.ltx/x": false,
+ } {
+ if got := keep(rel); got != want {
+ t.Errorf("keep(%q) = %v, want %v", rel, got, want)
+ }
+ }
+}