lease: MirrorPrefix reports the bytes a round copied
MirrorStore.List now answers key -> MirrorObject{ETag, Size} (ListObjectsV2
already returns the size, so this costs no extra request), and MirrorPrefix
returns a MirrorRound{Changed, BytesCopied} instead of a bare count. On an
error the round still reports what it did before the failure, as the count
did.
No behaviour change: the move worker still promotes on Changed == 0. The
next commit uses BytesCopied as the readiness bar (#175).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
6 files changed,
+149
−73
internal/activator/backend_exec.go+3 −3internal/activator/backend_exec_test.go+12 −8internal/host/movespass.go+2 −2internal/host/movespass_test.go+11 −9internal/lease/mirror.go+48 −23internal/lease/mirror_test.go+73 −28
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index 615435540..613d8e761 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -284,7 +284,7 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, err |
| // and back must restore its own lineage, never regress to an old |
| // staged copy (the source's cut deleted the own prefix it left behind, |
| // so a non-empty own prefix is by construction the newer story). |
| - var promoted map[string]string |
| + var promoted map[string]lease.MirrorObject |
| if b.Mirror != nil { |
| var err error |
| if promoted, err = b.promoteStagedReplica(ctx, host); err != nil { |
| @@ -1569,7 +1569,7 @@ func PurgeLocalDBArtifacts(dbPath string, full bool) { |
| // promoteStagedReplica copies moves/<host>/ over instances/<host>/ in this |
| // box's OWN bucket when — and only when — the ordinary prefix is empty. A |
| // no-op (nil error) when there is nothing staged. |
| -func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (map[string]string, error) { |
| +func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (map[string]lease.MirrorObject, error) { |
| // STAGING WINS when present (review round 1). A staged replica exists |
| // only while an inbound move is landing or has just landed: it was |
| // mirrored from the then-live data at cut, so it is by construction |
| @@ -1600,7 +1600,7 @@ func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (ma |
| // third listing of a prefix this restore has already walked twice. |
| // Best-effort: a leftover staging copy is storage, not a correctness |
| // problem. |
| -func (b *ExecBackend) dropStagedReplica(ctx context.Context, staged map[string]string) { |
| +func (b *ExecBackend) dropStagedReplica(ctx context.Context, staged map[string]lease.MirrorObject) { |
| for k := range staged { |
| _ = b.Mirror.Delete(ctx, b.Cfg.Bucket, k) |
| } |
diff --git a/internal/activator/backend_exec_test.go b/internal/activator/backend_exec_test.go| index 84f291c12..329da4288 100644 |
| --- a/internal/activator/backend_exec_test.go |
| +++ b/internal/activator/backend_exec_test.go |
| @@ -13,6 +13,7 @@ import ( |
| _ "modernc.org/sqlite" |
| "github.com/carlosframework/platform/internal/cgroup" |
| + "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/registry" |
| ) |
| @@ -197,26 +198,26 @@ func TestRestoreStagingFallback(t *testing.T) { |
| f.failList = false |
| } |
| -// fakeMirrorStore is the test double for lease.MirrorStore, bucket→key→etag. |
| +// fakeMirrorStore is the test double for lease.MirrorStore, bucket→key→object. |
| type fakeMirrorStore struct { |
| - buckets map[string]map[string]string |
| + buckets map[string]map[string]lease.MirrorObject |
| failList bool |
| } |
| func newFakeMirrorStore() *fakeMirrorStore { |
| - return &fakeMirrorStore{buckets: map[string]map[string]string{}} |
| + return &fakeMirrorStore{buckets: map[string]map[string]lease.MirrorObject{}} |
| } |
| func (f *fakeMirrorStore) put(bucket, key, etag string) { |
| if f.buckets[bucket] == nil { |
| - f.buckets[bucket] = map[string]string{} |
| + f.buckets[bucket] = map[string]lease.MirrorObject{} |
| } |
| - f.buckets[bucket][key] = etag |
| + f.buckets[bucket][key] = lease.MirrorObject{ETag: etag} |
| } |
| -func (f *fakeMirrorStore) List(_ context.Context, bucket, prefix string) (map[string]string, error) { |
| +func (f *fakeMirrorStore) List(_ context.Context, bucket, prefix string) (map[string]lease.MirrorObject, error) { |
| if f.failList { |
| return nil, context.DeadlineExceeded |
| } |
| - out := map[string]string{} |
| + out := map[string]lease.MirrorObject{} |
| for k, v := range f.buckets[bucket] { |
| if strings.HasPrefix(k, prefix) { |
| out[k] = v |
| @@ -225,7 +226,10 @@ func (f *fakeMirrorStore) List(_ context.Context, bucket, prefix string) (map[st |
| return out, nil |
| } |
| func (f *fakeMirrorStore) Copy(_ context.Context, srcBucket, srcKey, dstBucket, dstKey string) error { |
| - f.put(dstBucket, dstKey, f.buckets[srcBucket][srcKey]) |
| + if f.buckets[dstBucket] == nil { |
| + f.buckets[dstBucket] = map[string]lease.MirrorObject{} |
| + } |
| + f.buckets[dstBucket][dstKey] = f.buckets[srcBucket][srcKey] |
| return nil |
| } |
| func (f *fakeMirrorStore) Delete(_ context.Context, bucket, key string) error { |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index 6ec19b166..6f9951983 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -322,11 +322,11 @@ func runMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec prov |
| if err != nil { |
| return err |
| } |
| - changed, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix) |
| + round, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix) |
| if err != nil { |
| return err |
| } |
| - if rec.Move.State == provision.MovePreparing && changed == 0 { |
| + if rec.Move.State == provision.MovePreparing && round.Changed == 0 { |
| if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MovePromoteReady(cfg.Now())); err != nil && !errors.Is(err, provision.ErrMoveConflict) { |
| return err |
| } |
diff --git a/internal/host/movespass_test.go b/internal/host/movespass_test.go| index 9fca637a1..d57ee7b1e 100644 |
| --- a/internal/host/movespass_test.go |
| +++ b/internal/host/movespass_test.go |
| @@ -15,23 +15,24 @@ import ( |
| "github.com/carlosframework/platform/internal/release" |
| ) |
| -// fakeMoveMirror: bucket→key→etag plus an op log for ordering assertions. |
| +// fakeMoveMirror: bucket→key→object plus an op log for ordering assertions. |
| type fakeMoveMirror struct { |
| - buckets map[string]map[string]string |
| + buckets map[string]map[string]lease.MirrorObject |
| ops []string |
| } |
| func newFakeMoveMirror() *fakeMoveMirror { |
| - return &fakeMoveMirror{buckets: map[string]map[string]string{}} |
| + return &fakeMoveMirror{buckets: map[string]map[string]lease.MirrorObject{}} |
| } |
| -func (f *fakeMoveMirror) put(bucket, key, etag string) { |
| +func (f *fakeMoveMirror) put(bucket, key, etag string) { f.putSized(bucket, key, etag, 0) } |
| +func (f *fakeMoveMirror) putSized(bucket, key, etag string, size int64) { |
| if f.buckets[bucket] == nil { |
| - f.buckets[bucket] = map[string]string{} |
| + f.buckets[bucket] = map[string]lease.MirrorObject{} |
| } |
| - f.buckets[bucket][key] = etag |
| + f.buckets[bucket][key] = lease.MirrorObject{ETag: etag, Size: size} |
| } |
| -func (f *fakeMoveMirror) List(_ context.Context, bucket, prefix string) (map[string]string, error) { |
| - out := map[string]string{} |
| +func (f *fakeMoveMirror) List(_ context.Context, bucket, prefix string) (map[string]lease.MirrorObject, error) { |
| + out := map[string]lease.MirrorObject{} |
| for k, v := range f.buckets[bucket] { |
| if strings.HasPrefix(k, prefix) { |
| out[k] = v |
| @@ -41,7 +42,8 @@ func (f *fakeMoveMirror) List(_ context.Context, bucket, prefix string) (map[str |
| } |
| func (f *fakeMoveMirror) Copy(_ context.Context, srcBucket, srcKey, dstBucket, dstKey string) error { |
| f.ops = append(f.ops, "copy "+dstBucket+"/"+dstKey) |
| - f.put(dstBucket, dstKey, f.buckets[srcBucket][srcKey]) |
| + o := f.buckets[srcBucket][srcKey] |
| + f.putSized(dstBucket, dstKey, o.ETag, o.Size) |
| return nil |
| } |
| func (f *fakeMoveMirror) Delete(_ context.Context, bucket, key string) error { |
diff --git a/internal/lease/mirror.go b/internal/lease/mirror.go| index 2061a4c4e..f5afb8b7b 100644 |
| --- a/internal/lease/mirror.go |
| +++ b/internal/lease/mirror.go |
| @@ -15,32 +15,53 @@ import ( |
| // copies the instance's REPLICA prefix — the litestream generations the |
| // activator already maintains — never the database itself. MirrorPrefix is |
| // the one delta round; the move worker calls it repeatedly until a round |
| -// reports nothing changed, and once more after the child is down. |
| +// copies little enough that the cut's own round is cheap, and once more |
| +// after the child is down. |
| // |
| // The interface exists for the tests (an in-memory fake) and because the |
| // destination bucket lives in ANOTHER region: the implementation holds one |
| // client per side. |
| // MirrorStore is the three S3 actions a delta round needs. List answers |
| -// key→ETag under a prefix (keys RELATIVE to the bucket root, not the |
| +// key→object under a prefix (keys RELATIVE to the bucket root, not the |
| // prefix); Copy is a server-side CopyObject; Delete removes one key. |
| type MirrorStore interface { |
| - List(ctx context.Context, bucket, prefix string) (map[string]string, error) |
| + List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error) |
| Copy(ctx context.Context, srcBucket, srcKey, dstBucket, dstKey string) error |
| Delete(ctx context.Context, bucket, key string) error |
| } |
| +// MirrorObject is one listed object: the ETag the sameness test compares, |
| +// and the size a round's byte accounting adds up. |
| +type MirrorObject struct { |
| + ETag string |
| + Size int64 |
| +} |
| + |
| +// MirrorRound is what one delta round did. |
| +type MirrorRound struct { |
| + // Changed counts objects copied or deleted; 0 means the sides were |
| + // identical when listed. |
| + Changed int |
| + // BytesCopied sums the listed SOURCE size of every object copied. It |
| + // is what the move worker's readiness bar measures: an awake instance |
| + // never reaches Changed == 0 (litestream writes new files and |
| + // compaction rewrites old ones between any two listings), but the |
| + // bytes a round had to copy bound the work left for the cut's round. |
| + BytesCopied int64 |
| +} |
| + |
| // MirrorPrefix makes dstBucket/dstPrefix an exact mirror of |
| // srcBucket/srcPrefix: copy keys that are new or whose ETag differs, delete |
| // keys the source no longer has (litestream retention must hold in staging |
| // too — a restore that replays pruned WAL segments is a restore nobody |
| -// tested). Returns how many objects changed; 0 means the sides were |
| -// identical when listed — the move worker's "caught up". |
| +// tested). Returns what the round did; on error, what it did before the |
| +// failure. |
| // |
| // ETag equality is a valid sameness test here even for multipart uploads: |
| // both sides of a server-side COPY carry the copy's own single-part ETag, |
| // and litestream's own objects are single PUTs. |
| -func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstBucket, dstPrefix string) (int, error) { |
| +func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstBucket, dstPrefix string) (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 |
| @@ -54,11 +75,11 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB |
| } |
| src, err := m.List(ctx, srcBucket, srcPrefix) |
| if err != nil { |
| - return 0, fmt.Errorf("mirror: list source %s/%s: %w", srcBucket, srcPrefix, err) |
| + return MirrorRound{}, fmt.Errorf("mirror: list source %s/%s: %w", srcBucket, srcPrefix, err) |
| } |
| dst, err := m.List(ctx, dstBucket, dstPrefix) |
| if err != nil { |
| - return 0, fmt.Errorf("mirror: list staging %s/%s: %w", dstBucket, dstPrefix, err) |
| + return MirrorRound{}, fmt.Errorf("mirror: list staging %s/%s: %w", dstBucket, dstPrefix, err) |
| } |
| // Deterministic order: a failed round resumes the same way everywhere. |
| rel := make([]string, 0, len(src)) |
| @@ -66,25 +87,26 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB |
| rel = append(rel, k[len(srcPrefix):]) |
| } |
| sort.Strings(rel) |
| - changed := 0 |
| + var round MirrorRound |
| for _, r := range rel { |
| // The lease's lock.json is COORDINATION state, not data (live |
| // rehearsal catch, 2026-08-29): the renewer rewrites it on every |
| - // TTL renewal, so mirroring it makes "changed==0" unreachable for |
| - // any awake instance — and promoting it would hand the target a |
| + // TTL renewal, so mirroring it would put a change in every round |
| + // for any awake instance — and promoting it would hand the target a |
| // stale foreign lease. Skipped on both sides: never copied, and |
| // the destination's own copy (if any) is never deleted. |
| if r == leaseLockName { |
| continue |
| } |
| sk, dk := srcPrefix+r, dstPrefix+r |
| - if et, ok := dst[dk]; ok && et == src[sk] { |
| + if d, ok := dst[dk]; ok && d.ETag == src[sk].ETag { |
| continue |
| } |
| if err := m.Copy(ctx, srcBucket, sk, dstBucket, dk); err != nil { |
| - return changed, fmt.Errorf("mirror: copy %s: %w", sk, err) |
| + return round, fmt.Errorf("mirror: copy %s: %w", sk, err) |
| } |
| - changed++ |
| + round.Changed++ |
| + round.BytesCopied += src[sk].Size |
| } |
| for dk := range dst { |
| r := dk[len(dstPrefix):] |
| @@ -93,12 +115,12 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB |
| } |
| if _, ok := src[srcPrefix+r]; !ok { |
| if err := m.Delete(ctx, dstBucket, dk); err != nil { |
| - return changed, fmt.Errorf("mirror: delete %s: %w", dk, err) |
| + return round, fmt.Errorf("mirror: delete %s: %w", dk, err) |
| } |
| - changed++ |
| + round.Changed++ |
| } |
| } |
| - return changed, nil |
| + return round, nil |
| } |
| // leaseLockName is the coordination file this package's own leaser keeps |
| @@ -152,8 +174,8 @@ func NewS3Mirror(ctx context.Context, srcRegion, dstRegion string) (MirrorStore, |
| return &s3Mirror{src: src, dst: dst, byBucket: map[string]*awss3.Client{}}, nil |
| } |
| -func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix string) (map[string]string, error) { |
| - out := map[string]string{} |
| +func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix string) (map[string]MirrorObject, error) { |
| + out := map[string]MirrorObject{} |
| pager := awss3.NewListObjectsV2Paginator(c, &awss3.ListObjectsV2Input{Bucket: &bucket, Prefix: &prefix}) |
| for pager.HasMorePages() { |
| resp, err := pager.NextPage(ctx) |
| @@ -164,11 +186,14 @@ func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix str |
| if o.Key == nil { |
| continue |
| } |
| - et := "" |
| + var obj MirrorObject |
| if o.ETag != nil { |
| - et = *o.ETag |
| + obj.ETag = *o.ETag |
| + } |
| + if o.Size != nil { |
| + obj.Size = *o.Size |
| } |
| - out[*o.Key] = et |
| + out[*o.Key] = obj |
| } |
| } |
| return out, nil |
| @@ -178,7 +203,7 @@ func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix str |
| // never says which region a bucket lives in, so the FIRST touch of a bucket |
| // probes source-then-destination (a wrong-region List is an error, so the |
| // probe is unambiguous) and the winner is remembered. |
| -func (m *s3Mirror) List(ctx context.Context, bucket, prefix string) (map[string]string, error) { |
| +func (m *s3Mirror) List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error) { |
| m.mu.Lock() |
| c := m.byBucket[bucket] |
| m.mu.Unlock() |
diff --git a/internal/lease/mirror_test.go b/internal/lease/mirror_test.go| index 124b3b896..0d706f9ab 100644 |
| --- a/internal/lease/mirror_test.go |
| +++ b/internal/lease/mirror_test.go |
| @@ -7,25 +7,27 @@ import ( |
| "testing" |
| ) |
| -// fakeMirror is an in-memory MirrorStore: buckets of key→etag, plus an op |
| -// log so ordering tests can assert the sequence. |
| +// fakeMirror is an in-memory MirrorStore: buckets of key→object, plus an |
| +// op log so ordering tests can assert the sequence. |
| type fakeMirror struct { |
| - buckets map[string]map[string]string |
| + buckets map[string]map[string]MirrorObject |
| ops []string |
| failOn string |
| } |
| -func newFakeMirror() *fakeMirror { return &fakeMirror{buckets: map[string]map[string]string{}} } |
| +func newFakeMirror() *fakeMirror { return &fakeMirror{buckets: map[string]map[string]MirrorObject{}} } |
| -func (f *fakeMirror) put(bucket, key, etag string) { |
| +func (f *fakeMirror) put(bucket, key, etag string) { f.putSized(bucket, key, etag, 0) } |
| + |
| +func (f *fakeMirror) putSized(bucket, key, etag string, size int64) { |
| if f.buckets[bucket] == nil { |
| - f.buckets[bucket] = map[string]string{} |
| + f.buckets[bucket] = map[string]MirrorObject{} |
| } |
| - f.buckets[bucket][key] = etag |
| + f.buckets[bucket][key] = MirrorObject{ETag: etag, Size: size} |
| } |
| -func (f *fakeMirror) List(_ context.Context, bucket, prefix string) (map[string]string, error) { |
| - out := map[string]string{} |
| +func (f *fakeMirror) List(_ context.Context, bucket, prefix string) (map[string]MirrorObject, error) { |
| + out := map[string]MirrorObject{} |
| for k, v := range f.buckets[bucket] { |
| if strings.HasPrefix(k, prefix) { |
| out[k] = v |
| @@ -39,7 +41,8 @@ func (f *fakeMirror) Copy(_ context.Context, srcBucket, srcKey, dstBucket, dstKe |
| return fmt.Errorf("injected copy failure on %s", srcKey) |
| } |
| f.ops = append(f.ops, "copy "+srcKey+" -> "+dstBucket+"/"+dstKey) |
| - f.put(dstBucket, dstKey, f.buckets[srcBucket][srcKey]) |
| + o := f.buckets[srcBucket][srcKey] |
| + f.putSized(dstBucket, dstKey, o.ETag, o.Size) |
| return nil |
| } |
| @@ -56,21 +59,21 @@ func TestMirrorPrefix(t *testing.T) { |
| f.put("src", "instances/h/generations/a/wal/000", "e2") |
| // Initial round: full copy. |
| - n, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| - if err != nil || n != 2 { |
| - t.Fatalf("initial round = %d, %v; want 2 copies", n, err) |
| + r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| + if err != nil || r.Changed != 2 { |
| + t.Fatalf("initial round = %+v, %v; want 2 copies", r, err) |
| } |
| // Identical: zero. |
| - if n, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || n != 0 { |
| - t.Fatalf("caught-up round = %d, %v; want 0", n, err) |
| + if r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || r.Changed != 0 { |
| + t.Fatalf("caught-up round = %+v, %v; want 0", r, err) |
| } |
| // Delta: one changed etag, one new, one pruned at source. |
| f.put("src", "instances/h/generations/a/wal/000", "e2b") |
| f.put("src", "instances/h/generations/a/wal/001", "e3") |
| delete(f.buckets["src"], "instances/h/generations/a/snapshot") |
| - n, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| - if err != nil || n != 3 { |
| - t.Fatalf("delta round = %d, %v; want 2 copies + 1 delete", n, err) |
| + r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| + if err != nil || r.Changed != 3 { |
| + t.Fatalf("delta round = %+v, %v; want 2 copies + 1 delete", r, err) |
| } |
| got, _ := f.List(ctx, "dst", "moves/h/") |
| want := map[string]string{ |
| @@ -81,8 +84,8 @@ func TestMirrorPrefix(t *testing.T) { |
| t.Fatalf("staging = %v, want %v", got, want) |
| } |
| for k, v := range want { |
| - if got[k] != v { |
| - t.Fatalf("staging[%s] = %q, want %q", k, got[k], v) |
| + if got[k].ETag != v { |
| + t.Fatalf("staging[%s] = %q, want %q", k, got[k].ETag, v) |
| } |
| } |
| // A copy failure surfaces with the partial count. |
| @@ -101,9 +104,9 @@ func TestMirrorPrefixNormalizesBoundaries(t *testing.T) { |
| f := newFakeMirror() |
| f.put("src", "instances/h/generations/a/snapshot", "e1") |
| f.put("src", "instances/h2/generations/a/snapshot", "eX") // the sibling |
| - n, err := MirrorPrefix(ctx, f, "src", "instances/h", "dst", "moves/h") |
| - if err != nil || n != 1 { |
| - t.Fatalf("round = %d, %v; want exactly the one h object", n, err) |
| + r, err := MirrorPrefix(ctx, f, "src", "instances/h", "dst", "moves/h") |
| + if err != nil || r.Changed != 1 { |
| + t.Fatalf("round = %+v, %v; want exactly the one h object", r, err) |
| } |
| got, _ := f.List(ctx, "dst", "") |
| if _, ok := got["moves/h/generations/a/snapshot"]; !ok || len(got) != 1 { |
| @@ -119,15 +122,57 @@ func TestMirrorPrefixSkipsTheLease(t *testing.T) { |
| f.put("src", "instances/h/lock.json", "renewal-1") |
| f.put("src", "instances/h/generations/a/snapshot", "e1") |
| f.put("dst", "moves/h/lock.json", "old-foreign-lease") |
| - if n, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || n != 1 { |
| - t.Fatalf("round = %d, %v; want only the snapshot to move", n, err) |
| + if r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || r.Changed != 1 { |
| + t.Fatalf("round = %+v, %v; want only the snapshot to move", r, err) |
| } |
| f.put("src", "instances/h/lock.json", "renewal-2") // the churn |
| - if n, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || n != 0 { |
| - t.Fatalf("churn round = %d, %v; want caught-up despite the renewal", n, err) |
| + if r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || r.Changed != 0 { |
| + t.Fatalf("churn round = %+v, %v; want caught-up despite the renewal", r, err) |
| } |
| got, _ := f.List(ctx, "dst", "moves/h/") |
| - if got["moves/h/lock.json"] != "old-foreign-lease" || len(got) != 2 { |
| + if got["moves/h/lock.json"].ETag != "old-foreign-lease" || len(got) != 2 { |
| t.Fatalf("staging = %v; the dst lease must be untouched", got) |
| } |
| } |
| + |
| +// A round reports the bytes it copied: the listed source size of every |
| +// object it copied, and nothing for an object it skipped as unchanged, a |
| +// delete, or the lease. The move worker's readiness bar is this number. |
| +func TestMirrorPrefixCountsBytesCopied(t *testing.T) { |
| + ctx := context.Background() |
| + f := newFakeMirror() |
| + f.putSized("src", "instances/h/generations/a/snapshot", "e1", 1000) |
| + f.putSized("src", "instances/h/generations/a/wal/000", "e2", 30) |
| + f.putSized("src", "instances/h/lock.json", "renewal-1", 500) |
| + |
| + r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| + if err != nil || r.Changed != 2 || r.BytesCopied != 1030 { |
| + t.Fatalf("initial round = %+v, %v; want 2 changed, 1030 bytes", r, err) |
| + } |
| + |
| + // Unchanged objects are skipped and cost nothing; the lease's churn |
| + // costs nothing; a rewritten object costs its NEW size; a new one its |
| + // size; a pruned one is a change but no bytes. |
| + f.putSized("src", "instances/h/lock.json", "renewal-2", 500) |
| + f.putSized("src", "instances/h/generations/a/wal/000", "e2b", 70) |
| + f.putSized("src", "instances/h/generations/a/wal/001", "e3", 5) |
| + f.putSized("dst", "moves/h/generations/old/snapshot", "eOld", 9999) |
| + r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| + if err != nil || r.Changed != 3 || r.BytesCopied != 75 { |
| + t.Fatalf("delta round = %+v, %v; want 3 changed (2 copies + 1 delete), 75 bytes", r, err) |
| + } |
| + |
| + r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| + if err != nil || r.Changed != 0 || r.BytesCopied != 0 { |
| + t.Fatalf("caught-up round = %+v, %v; want nothing", r, err) |
| + } |
| + |
| + // A failed round still reports what it copied before the failure. |
| + f.putSized("src", "instances/h/generations/a/wal/002", "e4", 11) |
| + f.putSized("src", "instances/h/generations/a/wal/003", "e5", 13) |
| + f.failOn = "instances/h/generations/a/wal/003" |
| + r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| + if err == nil || r.Changed != 1 || r.BytesCopied != 11 { |
| + t.Fatalf("failed round = %+v, %v; want an error with the partial 1 change, 11 bytes", r, err) |
| + } |
| +} |