amadan
lease: run a mirror round's copies and deletes side by side
MirrorPrefixKeep copied one object at a time. On flagship S3 the offline move's seal round over a 305-object, 1.58 GB replica took six minutes while the instance was offline. The round now decides every skip first, in key order (the two listings and the receipts, exactly as before), then runs the copies on up to DefaultMirrorWorkers (8) goroutines, and only after every copy has landed runs the deletes the same way. Every invariant holds: the round and the receipt book are updated under one mutex; the first failure stops dispatch, cancels the copies in flight, and the best-effort save keeps a receipt for every copy that landed; a receipt is still written only after its copy returned; a VersionedCopier round still records versions only from copies and receipts that carry one (round-4 finding 7 unchanged for plain stores). A store may set its concurrency through MirrorConcurrency (the tests do). Every caller (offline seal, legacy and Phase 1 rounds, the cut, the rollback, the activator's staged promotion) gets the speedup. The test fakes that rounds run over (lease, host, activator) are made safe for concurrent use; the new tests run under -race.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
7 files changed,
+683
−44
internal/activator/backend_exec_test.go+18 −1internal/host/movescut_test.go+2 −0internal/host/movespass_test.go+37 −9internal/host/movesphase1_harness_test.go+6 −1internal/lease/mirror.go+145 −19internal/lease/mirror_parallel_test.go+383 −0internal/lease/mirror_test.go+92 −14
diff --git a/internal/activator/backend_exec_test.go b/internal/activator/backend_exec_test.go| index c3c5b9b5c..60eec7b27 100644 |
| --- a/internal/activator/backend_exec_test.go |
| +++ b/internal/activator/backend_exec_test.go |
| @@ -288,7 +288,9 @@ func TestRestoreIgnoresMoveReceipts(t *testing.T) { |
| } |
| // fakeMirrorStore is the test double for lease.MirrorStore, bucket→key→object. |
| +// A mirror round copies and deletes side by side, so every method holds mu. |
| type fakeMirrorStore struct { |
| + mu sync.Mutex |
| buckets map[string]map[string]lease.MirrorObject |
| failList bool |
| } |
| @@ -297,12 +299,19 @@ func newFakeMirrorStore() *fakeMirrorStore { |
| return &fakeMirrorStore{buckets: map[string]map[string]lease.MirrorObject{}} |
| } |
| func (f *fakeMirrorStore) put(bucket, key, etag string) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + f.putLocked(bucket, key, etag) |
| +} |
| +func (f *fakeMirrorStore) putLocked(bucket, key, etag string) { |
| if f.buckets[bucket] == nil { |
| f.buckets[bucket] = map[string]lease.MirrorObject{} |
| } |
| f.buckets[bucket][key] = lease.MirrorObject{ETag: etag} |
| } |
| func (f *fakeMirrorStore) List(_ context.Context, bucket, prefix string) (map[string]lease.MirrorObject, error) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| if f.failList { |
| return nil, context.DeadlineExceeded |
| } |
| @@ -315,6 +324,8 @@ func (f *fakeMirrorStore) List(_ context.Context, bucket, prefix string) (map[st |
| return out, nil |
| } |
| func (f *fakeMirrorStore) Copy(_ context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| o, ok := f.buckets[srcBucket][srcKey] |
| if !ok || (srcETag != "" && o.ETag != srcETag) { |
| return "", lease.ErrCopySourceChanged |
| @@ -326,16 +337,22 @@ func (f *fakeMirrorStore) Copy(_ context.Context, srcBucket, srcKey, srcETag, ds |
| return o.ETag, nil |
| } |
| func (f *fakeMirrorStore) Get(_ context.Context, bucket, key string) ([]byte, error) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| if _, ok := f.buckets[bucket][key]; !ok { |
| return nil, lease.ErrMirrorNotFound |
| } |
| return []byte("{}"), nil |
| } |
| func (f *fakeMirrorStore) Put(_ context.Context, bucket, key string, _ []byte) error { |
| - f.put(bucket, key, "receipts") |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + f.putLocked(bucket, key, "receipts") |
| return nil |
| } |
| func (f *fakeMirrorStore) Delete(_ context.Context, bucket, key string) error { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| delete(f.buckets[bucket], key) |
| return nil |
| } |
diff --git a/internal/host/movescut_test.go b/internal/host/movescut_test.go| index 0f3947853..f687f9906 100644 |
| --- a/internal/host/movescut_test.go |
| +++ b/internal/host/movescut_test.go |
| @@ -578,10 +578,12 @@ func TestCutRevertsAfterThreeDeferredRounds(t *testing.T) { |
| if strings.HasSuffix(srcKey, lastTXKey) { |
| // The source changes under the copy: the conditional copy is |
| // refused and the object deferred. |
| + h.f.bmu.Lock() |
| n++ |
| o := h.f.buckets["data-ie"][srcKey] |
| o.ETag = "changed-" + string(rune('a'+n)) |
| h.f.buckets["data-ie"][srcKey] = o |
| + h.f.bmu.Unlock() |
| } |
| } |
| h.cut(t) |
diff --git a/internal/host/movespass_test.go b/internal/host/movespass_test.go| index e93de9a1a..1edc0acd6 100644 |
| --- a/internal/host/movespass_test.go |
| +++ b/internal/host/movespass_test.go |
| @@ -25,7 +25,14 @@ import ( |
| // a single-part object's ETag is its body, a multipart one's is |
| // body+"-27", and a copy lands single-part with the body as its ETag. An |
| // op log records ordering. |
| +// |
| +// A mirror round runs its copies and deletes side by side |
| +// (lease.DefaultMirrorWorkers), so buckets and ops are guarded by bmu |
| +// whenever a round can be running; a test reads them directly between |
| +// rounds. beforeCopy runs without the lock, so it may call the fake. Lock |
| +// order: mu, then bmu. |
| type fakeMoveMirror struct { |
| + bmu sync.Mutex |
| buckets map[string]map[string]fakeMoveObject |
| ops []string |
| beforeCopy func(srcKey string) |
| @@ -62,12 +69,20 @@ func (f *fakeMoveMirror) putMultipart(bucket, key, body string, size int64) { |
| f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: body + "-27", Size: size}, body}) |
| } |
| func (f *fakeMoveMirror) store(bucket, key string, o fakeMoveObject) { |
| + f.bmu.Lock() |
| + defer f.bmu.Unlock() |
| + f.storeLocked(bucket, key, o) |
| +} |
| + |
| +func (f *fakeMoveMirror) storeLocked(bucket, key string, o fakeMoveObject) { |
| if f.buckets[bucket] == nil { |
| f.buckets[bucket] = map[string]fakeMoveObject{} |
| } |
| f.buckets[bucket][key] = o |
| } |
| func (f *fakeMoveMirror) List(_ context.Context, bucket, prefix string) (map[string]lease.MirrorObject, error) { |
| + f.bmu.Lock() |
| + defer f.bmu.Unlock() |
| out := map[string]lease.MirrorObject{} |
| for k, v := range f.buckets[bucket] { |
| if strings.HasPrefix(k, prefix) { |
| @@ -77,18 +92,25 @@ func (f *fakeMoveMirror) List(_ context.Context, bucket, prefix string) (map[str |
| return out, nil |
| } |
| func (f *fakeMoveMirror) Copy(_ context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) { |
| - if f.beforeCopy != nil { |
| - f.beforeCopy(srcKey) |
| - } |
| + f.bmu.Lock() |
| + before := f.beforeCopy |
| + f.bmu.Unlock() |
| + if before != nil { |
| + before(srcKey) |
| + } |
| + f.bmu.Lock() |
| + defer f.bmu.Unlock() |
| o, ok := f.buckets[srcBucket][srcKey] |
| if !ok || (srcETag != "" && o.ETag != srcETag) { |
| return "", lease.ErrCopySourceChanged |
| } |
| f.ops = append(f.ops, "copy "+dstBucket+"/"+dstKey) |
| - f.store(dstBucket, dstKey, fakeMoveObject{lease.MirrorObject{ETag: o.body, Size: o.Size}, o.body}) |
| + f.storeLocked(dstBucket, dstKey, fakeMoveObject{lease.MirrorObject{ETag: o.body, Size: o.Size}, o.body}) |
| return o.body, nil |
| } |
| func (f *fakeMoveMirror) Delete(_ context.Context, bucket, key string) error { |
| + f.bmu.Lock() |
| + defer f.bmu.Unlock() |
| f.ops = append(f.ops, "delete "+bucket+"/"+key) |
| delete(f.buckets[bucket], key) |
| return nil |
| @@ -99,6 +121,8 @@ func (f *fakeMoveMirror) Get(_ context.Context, bucket, key string) ([]byte, err |
| return nil, err |
| } |
| } |
| + f.bmu.Lock() |
| + defer f.bmu.Unlock() |
| o, ok := f.buckets[bucket][key] |
| if !ok { |
| return nil, lease.ErrMirrorNotFound |
| @@ -106,14 +130,18 @@ func (f *fakeMoveMirror) Get(_ context.Context, bucket, key string) ([]byte, err |
| return []byte(o.body), nil |
| } |
| func (f *fakeMoveMirror) Put(_ context.Context, bucket, key string, body []byte) error { |
| + f.bmu.Lock() |
| + defer f.bmu.Unlock() |
| f.ops = append(f.ops, "put "+bucket+"/"+key) |
| - f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "receipts", Size: int64(len(body))}, string(body)}) |
| + f.storeLocked(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "receipts", Size: int64(len(body))}, string(body)}) |
| return nil |
| } |
| // bodies is a prefix's content by relative key: what a byte-exact |
| // comparison compares (ETags differ across a multipart copy by design). |
| func (f *fakeMoveMirror) bodies(bucket, prefix string) map[string]string { |
| + f.bmu.Lock() |
| + defer f.bmu.Unlock() |
| out := map[string]string{} |
| for k, v := range f.buckets[bucket] { |
| if strings.HasPrefix(k, prefix) { |
| @@ -804,10 +832,10 @@ func TestMovesPassDeferredBytesCountAgainstTheBar(t *testing.T) { |
| h.moveRound(t, host) |
| big := "instances/" + host + "/ltx/1/0.ltx" |
| h.f.putMultipart("data-ie", big, "level-0b", 138<<20) |
| + var once sync.Once |
| h.f.beforeCopy = func(k string) { |
| if k == big { |
| - h.f.putMultipart("data-ie", big, "level-0c", 138<<20) |
| - h.f.beforeCopy = nil |
| + once.Do(func() { h.f.putMultipart("data-ie", big, "level-0c", 138<<20) }) |
| } |
| } |
| if st := h.moveRound(t, host); st != provision.MovePreparing { |
| @@ -864,10 +892,10 @@ func TestMovesPassCutStopsOnADeferredCopy(t *testing.T) { |
| t.Fatal(err) |
| } |
| big := "instances/" + host + "/ltx/1/3.ltx" |
| + var once sync.Once |
| h.f.beforeCopy = func(k string) { |
| if k == big { |
| - h.f.putMultipart("data-ie", big, "level-3b", 138<<20) |
| - h.f.beforeCopy = nil |
| + once.Do(func() { h.f.putMultipart("data-ie", big, "level-3b", 138<<20) }) |
| } |
| } |
| if err := movesPass(context.Background(), h.s, h.worker()); err != nil { |
diff --git a/internal/host/movesphase1_harness_test.go b/internal/host/movesphase1_harness_test.go| index 6e91d0118..b3b0f814f 100644 |
| --- a/internal/host/movesphase1_harness_test.go |
| +++ b/internal/host/movesphase1_harness_test.go |
| @@ -76,7 +76,10 @@ func (f *fakeMoveMirror) putAt(op string, signAt time.Time, bucket, key string, |
| if err := f.logAt(op, signAt, bucket, key); err != nil { |
| return "", err |
| } |
| - if _, ok := f.buckets[bucket][key]; ok && createOnly { |
| + f.bmu.Lock() |
| + _, exists := f.buckets[bucket][key] |
| + f.bmu.Unlock() |
| + if exists && createOnly { |
| return "", lease.ErrExists |
| } |
| ver := f.nextVersion() |
| @@ -99,7 +102,9 @@ func (f *fakeMoveMirror) setVersion(bucket, key, ver string) { |
| if f.history == nil { |
| f.history = map[string]fakeMoveObject{} |
| } |
| + f.bmu.Lock() |
| f.history[bucket+"/"+key+"@"+ver] = f.buckets[bucket][key] |
| + f.bmu.Unlock() |
| } |
| } |
diff --git a/internal/lease/mirror.go b/internal/lease/mirror.go| index 2aeb7bcb6..451bea05f 100644 |
| --- a/internal/lease/mirror.go |
| +++ b/internal/lease/mirror.go |
| @@ -223,6 +223,10 @@ func MirrorPrefixKeep(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, |
| if versioned { |
| round.Versions = map[string]string{} |
| } |
| + // The plan: which objects this round copies. Deciding a skip reads only |
| + // the two listings and the receipts, so it is done here, in key order, |
| + // before any copy starts; the copies then run side by side. |
| + var jobs []string |
| 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 |
| @@ -233,9 +237,8 @@ func MirrorPrefixKeep(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, |
| if r == leaseLockName { |
| continue |
| } |
| - sk, dk := srcPrefix+r, dstPrefix+r |
| - s := src[sk] |
| - d, staged := dst[dk] |
| + s := src[srcPrefix+r] |
| + d, staged := dst[dstPrefix+r] |
| if staged && d.ETag == s.ETag && !versioned { |
| continue |
| } |
| @@ -257,23 +260,38 @@ func MirrorPrefixKeep(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, |
| continue |
| } |
| } |
| + jobs = append(jobs, r) |
| + } |
| + // The copies, up to mirrorWorkers(m) at once (#175 offline-move |
| + // speedup: one server-side copy at a time took six minutes for a 1.6 GB |
| + // replica while the instance was offline). Every update to round and |
| + // to the book is made under mu, in whatever order the copies finish; |
| + // neither depends on that order. The first failure stops the round: no |
| + // further copy starts, the ones in flight are cancelled, and every copy |
| + // that did land keeps its receipt (the best-effort save below). |
| + var mu sync.Mutex |
| + err = forEachParallel(ctx, mirrorWorkers(m), len(jobs), func(cctx context.Context, i int) error { |
| + r := jobs[i] |
| + sk, dk := srcPrefix+r, dstPrefix+r |
| + s := src[sk] |
| var etag, ver string |
| var err error |
| if versioned { |
| - etag, ver, err = vc.CopyVersioned(ctx, srcBucket, sk, s.ETag, dstBucket, dk) |
| + etag, ver, err = vc.CopyVersioned(cctx, srcBucket, sk, s.ETag, dstBucket, dk) |
| } else { |
| - etag, err = m.Copy(ctx, srcBucket, sk, s.ETag, dstBucket, dk) |
| + etag, err = m.Copy(cctx, srcBucket, sk, s.ETag, dstBucket, dk) |
| } |
| + mu.Lock() |
| + defer mu.Unlock() |
| if errors.Is(err, ErrCopySourceChanged) { |
| book.drop(r) |
| round.Changed++ |
| round.Deferred++ |
| round.BytesDeferred += s.Size |
| - continue |
| + return nil |
| } |
| if err != nil { |
| - book.save(ctx) // best effort: what already landed keeps its receipts |
| - return round, fmt.Errorf("mirror: copy %s: %w", sk, err) |
| + return fmt.Errorf("mirror: copy %s: %w", sk, err) |
| } |
| if !UsableVersionID(ver) { |
| ver = "" // never recorded or sealed by |
| @@ -290,27 +308,43 @@ func MirrorPrefixKeep(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, |
| } |
| // A failed read here loses this receipt (one recopy next |
| // round), never a wrong skip. |
| - if err := book.set(ctx, r, copyReceipt{SrcETag: s.ETag, SrcSize: s.Size, DstETag: dstETag, DstVersionID: ver}); err != nil { |
| - return round, err |
| - } |
| - } else { |
| - book.drop(r) |
| + return book.set(ctx, r, copyReceipt{SrcETag: s.ETag, SrcSize: s.Size, DstETag: dstETag, DstVersionID: ver}) |
| } |
| + book.drop(r) |
| + return nil |
| + }) |
| + if err != nil { |
| + book.save(ctx) // best effort: what already landed keeps its receipts |
| + return round, err |
| } |
| + // The deletes, only once every copy has landed (as before), and side |
| + // by side among themselves: their order never mattered. |
| + var gone []string |
| for dk := range dst { |
| r := dk[len(dstPrefix):] |
| if r == leaseLockName { |
| continue |
| } |
| if _, ok := src[srcPrefix+r]; !ok { |
| - if err := m.Delete(ctx, dstBucket, dk); err != nil { |
| - book.save(ctx) |
| - return round, fmt.Errorf("mirror: delete %s: %w", dk, err) |
| - } |
| - book.drop(r) |
| - round.Changed++ |
| + gone = append(gone, r) |
| } |
| } |
| + sort.Strings(gone) |
| + err = forEachParallel(ctx, mirrorWorkers(m), len(gone), func(cctx context.Context, i int) error { |
| + r := gone[i] |
| + if err := m.Delete(cctx, dstBucket, dstPrefix+r); err != nil { |
| + return fmt.Errorf("mirror: delete %s: %w", dstPrefix+r, err) |
| + } |
| + mu.Lock() |
| + defer mu.Unlock() |
| + book.drop(r) |
| + round.Changed++ |
| + return nil |
| + }) |
| + if err != nil { |
| + book.save(ctx) |
| + return round, err |
| + } |
| book.prune(func(r string) bool { _, ok := src[srcPrefix+r]; return ok }) |
| if err := book.save(ctx); err != nil { |
| return round, err |
| @@ -318,6 +352,98 @@ func MirrorPrefixKeep(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, |
| return round, nil |
| } |
| +// DefaultMirrorWorkers is how many copies (and, after them, deletes) one |
| +// delta round runs at once. Each is one server-side request, so the round's |
| +// wall time is the slowest request of each batch rather than their sum. |
| +const DefaultMirrorWorkers = 8 |
| + |
| +// MirrorConcurrency is the optional half of a MirrorStore that sets how many |
| +// requests a round runs at once. A store that does not implement it gets |
| +// DefaultMirrorWorkers; an answer below one is one (a strictly serial |
| +// round, as every round was before). Tests use it; production stores do not. |
| +type MirrorConcurrency interface { |
| + MirrorWorkers() int |
| +} |
| + |
| +func mirrorWorkers(m MirrorStore) int { |
| + if c, ok := m.(MirrorConcurrency); ok { |
| + if n := c.MirrorWorkers(); n >= 1 { |
| + return n |
| + } |
| + return 1 |
| + } |
| + return DefaultMirrorWorkers |
| +} |
| + |
| +// forEachParallel calls do for every index below n, on up to workers |
| +// goroutines, starting them in index order. The first error do returns |
| +// stops it: no further index starts, the context handed to the calls still |
| +// in flight is cancelled, and once they have returned that first error is |
| +// the answer (any error a cancelled call returns after it is dropped). A ctx |
| +// that ends before every index has started is its own error. Calls that |
| +// were in flight when it stopped have returned by the time it does, so the |
| +// caller's state is final. |
| +func forEachParallel(ctx context.Context, workers, n int, do func(ctx context.Context, i int) error) error { |
| + if n == 0 { |
| + return nil |
| + } |
| + if workers < 1 { |
| + workers = 1 |
| + } |
| + if workers > n { |
| + workers = n |
| + } |
| + cctx, cancel := context.WithCancel(ctx) |
| + defer cancel() |
| + var ( |
| + mu sync.Mutex |
| + first error |
| + wg sync.WaitGroup |
| + ) |
| + next := make(chan int) |
| + for w := 0; w < workers; w++ { |
| + wg.Add(1) |
| + go func() { |
| + defer wg.Done() |
| + for i := range next { |
| + if err := do(cctx, i); err != nil { |
| + mu.Lock() |
| + if first == nil { |
| + first = err |
| + cancel() |
| + } |
| + mu.Unlock() |
| + } |
| + } |
| + }() |
| + } |
| + started := 0 |
| +feed: |
| + for ; started < n; started++ { |
| + // Checked first: a select with both cases ready picks at random, |
| + // and a stopped round must not start another copy. |
| + if cctx.Err() != nil { |
| + break |
| + } |
| + select { |
| + case next <- started: |
| + case <-cctx.Done(): |
| + break feed |
| + } |
| + } |
| + close(next) |
| + wg.Wait() |
| + if first != nil { |
| + return first |
| + } |
| + if started < n { |
| + if err := ctx.Err(); err != nil { |
| + return fmt.Errorf("mirror: %w", err) |
| + } |
| + } |
| + return nil |
| +} |
| + |
| // receiptLedger is one round's view of the receipts object. It reads the |
| // object lazily, on the first object whose ETags differ, so a round over |
| // single-part objects only costs no request it did not cost before. |
diff --git a/internal/lease/mirror_parallel_test.go b/internal/lease/mirror_parallel_test.go| new file mode 100644 |
| index 000000000..5cd834ec7 |
| --- /dev/null |
| +++ b/internal/lease/mirror_parallel_test.go |
| @@ -0,0 +1,383 @@ |
| +package lease |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "strings" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// The parallel round (offline-move speedup): copies and deletes run up to |
| +// mirrorWorkers(m) at once. These tests hold the round to what the serial |
| +// one promised: the same end state, receipts for exactly what landed, the |
| +// first failure stopping the round, and no skip a receipt cannot prove. |
| +// Run them under -race: the round's own state is shared by its workers. |
| + |
| +// fillReplica stores n objects under instances/h/: every third one |
| +// multipart (the copy lands with a new ETag, so it needs a receipt), the |
| +// rest single-part. Sizes differ so byte totals catch a double count. |
| +func fillReplica(f *fakeMirror, n int, gen string) (keys []string, bytes int64) { |
| + for i := 0; i < n; i++ { |
| + key := fmt.Sprintf("instances/h/ltx/0/%s-%04d.ltx", gen, i) |
| + size := int64(1000 + i) |
| + if i%3 == 0 { |
| + f.putMultipart("src", key, fmt.Sprintf("%s-body-%d", gen, i), size) |
| + } else { |
| + f.putSized("src", key, fmt.Sprintf("%s-etag-%d", gen, i), size) |
| + } |
| + keys = append(keys, key) |
| + bytes += size |
| + } |
| + return keys, bytes |
| +} |
| + |
| +// stagedBodies is moves/h/ by key relative to it, body for body. |
| +func stagedBodies(f *fakeMirror) map[string]string { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + out := map[string]string{} |
| + for k, v := range f.buckets["dst"] { |
| + if strings.HasPrefix(k, "moves/h/") { |
| + out[strings.TrimPrefix(k, "moves/h/")] = v.body |
| + } |
| + } |
| + return out |
| +} |
| + |
| +// sourceBodies is instances/h/ the same way, lock.json aside. |
| +func sourceBodies(f *fakeMirror) map[string]string { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + out := map[string]string{} |
| + for k, v := range f.buckets["src"] { |
| + if rel, ok := strings.CutPrefix(k, "instances/h/"); ok && rel != "lock.json" { |
| + out[rel] = v.body |
| + } |
| + } |
| + return out |
| +} |
| + |
| +func sameBodies(t *testing.T, got, want map[string]string) { |
| + t.Helper() |
| + if len(got) != len(want) { |
| + t.Fatalf("staged %d objects, source has %d", len(got), len(want)) |
| + } |
| + for k, v := range want { |
| + if got[k] != v { |
| + t.Fatalf("staged %s = %q, source %q", k, got[k], v) |
| + } |
| + } |
| +} |
| + |
| +func copiesIn(ops []string) int { |
| + n := 0 |
| + for _, op := range ops { |
| + if strings.HasPrefix(op, "copy ") { |
| + n++ |
| + } |
| + } |
| + return n |
| +} |
| + |
| +// A parallel round ends where a serial one does: every object staged body |
| +// for body, the same counts and bytes, receipts for exactly the multipart |
| +// objects, and a caught-up round after it. It really runs side by side, |
| +// and never more than the store allows. |
| +func TestMirrorPrefixParallelMatchesSerial(t *testing.T) { |
| + const n = 120 |
| + var rounds [2]MirrorRound |
| + var books [2]map[string]copyReceipt |
| + for i, workers := range []int{1, 6} { |
| + f := newFakeMirror() |
| + f.workers = workers |
| + _, bytes := fillReplica(f, n, "a") |
| + f.put("src", "instances/h/lock.json", "renewal") |
| + // Slow copies, so the workers overlap. |
| + f.onCopy = func(context.Context, string) error { time.Sleep(time.Millisecond); return nil } |
| + r := mirrorRound(t, f, testReceipts) |
| + if r.Changed != n || r.BytesCopied != bytes || r.Deferred != 0 { |
| + t.Fatalf("workers=%d: round = %+v, want %d copies of %d bytes", workers, r, n, bytes) |
| + } |
| + sameBodies(t, stagedBodies(f), sourceBodies(f)) |
| + if f.peak > workers || (workers > 1 && f.peak < 2) { |
| + t.Fatalf("workers=%d: peak concurrency %d", workers, f.peak) |
| + } |
| + book := readReceipts(t, f) |
| + if len(book) != n/3 { |
| + t.Fatalf("workers=%d: %d receipts, want one per multipart object (%d)", workers, len(book), n/3) |
| + } |
| + if again := mirrorRound(t, f, testReceipts); again.Changed != 0 { |
| + t.Fatalf("workers=%d: caught-up round = %+v", workers, again) |
| + } |
| + rounds[i], books[i] = r, book |
| + } |
| + if rounds[0].Changed != rounds[1].Changed || rounds[0].BytesCopied != rounds[1].BytesCopied { |
| + t.Fatalf("serial %+v and parallel %+v rounds differ", rounds[0], rounds[1]) |
| + } |
| + for k, v := range books[0] { |
| + if books[1][k] != v { |
| + t.Fatalf("receipt %s: serial %+v, parallel %+v", k, v, books[1][k]) |
| + } |
| + } |
| +} |
| + |
| +// The versioned mode under workers: every copied object's version lands in |
| +// Versions, and the receipts carry them, so the next round names them all |
| +// without copying. |
| +func TestMirrorPrefixParallelVersionedRound(t *testing.T) { |
| + f := &fakeVersionedMirror{fakeMirror: newFakeMirror()} |
| + f.workers = 8 |
| + keys, _ := fillReplica(f.fakeMirror, 64, "a") |
| + r1, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", testReceipts) |
| + if err != nil || len(r1.Versions) != len(keys) { |
| + t.Fatalf("round 1 = %d versions, %v; want %d", len(r1.Versions), err, len(keys)) |
| + } |
| + seen := map[string]bool{} |
| + for _, v := range r1.Versions { |
| + if seen[v] { |
| + t.Fatalf("version %s handed to two copies", v) |
| + } |
| + seen[v] = true |
| + } |
| + r2, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", testReceipts) |
| + if err != nil || r2.Changed != 0 || len(r2.Versions) != len(keys) { |
| + t.Fatalf("round 2 = %+v, %v; want nothing copied and every version named", r2, err) |
| + } |
| + for k, v := range r1.Versions { |
| + if r2.Versions[k] != v { |
| + t.Fatalf("%s: round 2 names %q, round 1 copied %q", k, r2.Versions[k], v) |
| + } |
| + } |
| +} |
| + |
| +// The first failure stops the round: it is the error, copies still in |
| +// flight see their context cancelled, nothing new starts, and every copy |
| +// that landed keeps its receipt, so the next round copies only what did |
| +// not land. |
| +func TestMirrorPrefixParallelFailureMidRound(t *testing.T) { |
| + f := newFakeMirror() |
| + f.workers = 4 |
| + // All multipart: a landed copy is skipped next round only by its receipt. |
| + for i := 0; i < 40; i++ { |
| + f.putMultipart("src", fmt.Sprintf("instances/h/ltx/1/%04d.ltx", i), fmt.Sprintf("b%d", i), 100) |
| + } |
| + bad := "instances/h/ltx/1/0010.ltx" |
| + var cancelled atomic.Int32 |
| + later := make(chan struct{}, 64) // a copy after the bad one has started |
| + f.onCopy = func(ctx context.Context, key string) error { |
| + switch { |
| + case key == bad: |
| + // Fail only once a later copy is in flight, so the cancel |
| + // has something to reach. |
| + select { |
| + case <-later: |
| + case <-time.After(5 * time.Second): |
| + } |
| + return errors.New("injected copy failure") |
| + case key > bad: |
| + later <- struct{}{} |
| + select { |
| + case <-ctx.Done(): |
| + cancelled.Add(1) |
| + return ctx.Err() |
| + case <-time.After(5 * time.Second): |
| + return errors.New("the round did not cancel a copy in flight") |
| + } |
| + } |
| + return nil |
| + } |
| + r, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", testReceipts) |
| + if err == nil || !strings.Contains(err.Error(), bad) || errors.Is(err, context.Canceled) { |
| + t.Fatalf("round error = %v, want the injected failure on %s", err, bad) |
| + } |
| + landed := stagedBodies(f) |
| + if r.Changed != len(landed) || r.BytesCopied != int64(len(landed))*100 { |
| + t.Fatalf("round = %+v, but %d objects landed", r, len(landed)) |
| + } |
| + if len(landed) == 0 || len(landed) > 10 { |
| + t.Fatalf("%d objects landed, want those before the failure only", len(landed)) |
| + } |
| + if cancelled.Load() == 0 { |
| + t.Fatal("no copy in flight was cancelled") |
| + } |
| + // Nothing past the workers in flight at the failure ever started. |
| + if started := copiesIn(f.ops) + int(cancelled.Load()) + 1; started > 10+1+f.workers { |
| + t.Fatalf("%d copies started, want the round to stop dispatching at the failure", started) |
| + } |
| + book := readReceipts(t, f) |
| + for rel := range landed { |
| + if _, ok := book[rel]; !ok { |
| + t.Fatalf("%s landed without a receipt", rel) |
| + } |
| + } |
| + if len(book) != len(landed) { |
| + t.Fatalf("%d receipts for %d landed copies", len(book), len(landed)) |
| + } |
| + f.onCopy = nil |
| + before := len(f.ops) |
| + next := mirrorRound(t, f, testReceipts) |
| + if next.Changed != 40-len(landed) { |
| + t.Fatalf("next round = %+v, want exactly the %d that did not land", next, 40-len(landed)) |
| + } |
| + for _, op := range f.ops[before:] { |
| + for rel := range landed { |
| + if strings.HasPrefix(op, "copy instances/h/"+rel+" ") { |
| + t.Fatalf("a landed copy was recopied: %s", op) |
| + } |
| + } |
| + } |
| + sameBodies(t, stagedBodies(f), sourceBodies(f)) |
| +} |
| + |
| +// A caller's context that ends mid-round is the round's error, never a |
| +// silent partial success. |
| +func TestMirrorPrefixParallelContextEnds(t *testing.T) { |
| + f := newFakeMirror() |
| + f.workers = 2 |
| + fillReplica(f, 30, "a") |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + var n atomic.Int32 |
| + f.onCopy = func(context.Context, string) error { |
| + if n.Add(1) == 3 { |
| + cancel() |
| + } |
| + return nil |
| + } |
| + if _, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", testReceipts); !errors.Is(err, context.Canceled) { |
| + t.Fatalf("round error = %v, want the context's", err) |
| + } |
| + if len(stagedBodies(f)) == 30 { |
| + t.Fatal("every copy ran after the context ended") |
| + } |
| +} |
| + |
| +// The pre-copy (offline move stage): rounds while the instance serves, then |
| +// the seal's round. The seal finds what stage copied and copies only what is |
| +// new since; its receipts prove every multipart skip. |
| +func TestMirrorPrefixReceiptsSkipAfterAPreCopy(t *testing.T) { |
| + f := newFakeMirror() |
| + _, staged := fillReplica(f, 90, "a") |
| + if r := mirrorRound(t, f, testReceipts); r.BytesCopied != staged { |
| + t.Fatalf("stage round = %+v, want %d bytes", r, staged) |
| + } |
| + // The instance kept writing: new LTX files. |
| + _, fresh := fillReplica(f, 6, "b") |
| + before := len(f.ops) |
| + seal := mirrorRound(t, f, testReceipts) |
| + if seal.BytesCopied != fresh || seal.Changed != 6 || seal.Deferred != 0 { |
| + t.Fatalf("seal round = %+v, want only the 6 new objects (%d bytes)", seal, fresh) |
| + } |
| + for _, op := range f.ops[before:] { |
| + if strings.HasPrefix(op, "copy ") && !strings.Contains(op, "/b-") { |
| + t.Fatalf("the seal recopied a pre-copied object: %s", op) |
| + } |
| + } |
| + sameBodies(t, stagedBodies(f), sourceBodies(f)) |
| +} |
| + |
| +// Between stage and seal, litestream's retention and compaction delete |
| +// replica files and write new ones. The seal's round deletes what the source |
| +// no longer has and drops those receipts, so the staged listing is the |
| +// source's again (the seal's sameListing). |
| +func TestMirrorPrefixSourceDeletionsBetweenStageAndSeal(t *testing.T) { |
| + f := newFakeMirror() |
| + keys, _ := fillReplica(f, 30, "a") |
| + mirrorRound(t, f, testReceipts) |
| + // Compaction: twelve level-0 files folded into one level-1 file. |
| + for _, k := range keys[:12] { |
| + f.remove("src", k) |
| + } |
| + f.putMultipart("src", "instances/h/ltx/1/compacted.ltx", "compacted", 12000) |
| + r := mirrorRound(t, f, testReceipts) |
| + if r.Changed != 13 || r.BytesCopied != 12000 { |
| + t.Fatalf("seal round = %+v, want 12 deletes and the one copy", r) |
| + } |
| + sameBodies(t, stagedBodies(f), sourceBodies(f)) |
| + for rel := range readReceipts(t, f) { |
| + if _, ok := sourceBodies(f)[rel]; !ok { |
| + t.Fatalf("a receipt outlived its source object: %s", rel) |
| + } |
| + } |
| + if again := mirrorRound(t, f, testReceipts); again.Changed != 0 { |
| + t.Fatalf("round after = %+v", again) |
| + } |
| +} |
| + |
| +// A staged prefix left by an earlier move is never trusted as such. A stale |
| +// object skips only when its ETag equals the source's (the same bytes) or a |
| +// receipt for this exact pair still matches both sides; anything else is |
| +// copied again, and every key the source lacks is deleted. |
| +func TestMirrorPrefixStaleStagedPrefixIsNotTrusted(t *testing.T) { |
| + f := newFakeMirror() |
| + f.putMultipart("src", "instances/h/ltx/1/a.ltx", "new-a", 500) |
| + f.putMultipart("src", "instances/h/ltx/1/b.ltx", "new-b", 500) |
| + f.putSized("src", "instances/h/ltx/0/c.ltx", "new-c", 50) |
| + f.putSized("src", "instances/h/ltx/0/d.ltx", "same-d", 50) |
| + // The stale prefix: a under the old move's copy of other bytes, b with |
| + // the same size as the source and a stale receipt, c changed, d byte |
| + // identical, and an extra the source never had. |
| + f.putSized("dst", "moves/h/ltx/1/a.ltx", "old-a", 500) |
| + f.putSized("dst", "moves/h/ltx/1/b.ltx", "old-b", 500) |
| + f.putSized("dst", "moves/h/ltx/0/c.ltx", "old-c", 50) |
| + f.putSized("dst", "moves/h/ltx/0/d.ltx", "same-d", 50) |
| + f.putSized("dst", "moves/h/ltx/0/extra.ltx", "old-x", 50) |
| + // An old book for this pair whose receipt for b names the source as it |
| + // is now but a destination ETag the staged object does not carry. |
| + stale := (`{"v":1,"src_bucket":"src","src_prefix":"instances/h/","dst_bucket":"dst","dst_prefix":"moves/h/",` + |
| + `"objects":{"ltx/1/b.ltx":{"src_etag":"new-b-27","src_size":500,"dst_etag":"new-b"},"ltx/0/extra.ltx":{"src_etag":"x","src_size":50,"dst_etag":"old-x"}}}`) |
| + f.store("src", testReceipts.Key, fakeObject{MirrorObject{ETag: "book", Size: int64(len(stale))}, stale}) |
| + |
| + r := mirrorRound(t, f, testReceipts) |
| + if r.Changed != 4 || r.BytesCopied != 1050 { |
| + t.Fatalf("round = %+v, want a, b and c copied and the extra deleted", r) |
| + } |
| + sameBodies(t, stagedBodies(f), sourceBodies(f)) |
| + book := readReceipts(t, f) |
| + if _, ok := book["ltx/0/extra.ltx"]; ok { |
| + t.Fatal("the stale receipt of a deleted key survived") |
| + } |
| + if got := book["ltx/1/b.ltx"]; got.DstETag != "new-b" || got.SrcETag != "new-b-27" { |
| + t.Fatalf("b's receipt = %+v", got) |
| + } |
| + |
| + // A book written for another pair (another move's target) is no |
| + // evidence at all, even when its receipt would match. |
| + g := newFakeMirror() |
| + g.putMultipart("src", "instances/h/ltx/1/a.ltx", "a", 500) |
| + g.putSized("dst", "moves/h/ltx/1/a.ltx", "a", 500) |
| + other := `{"v":1,"src_bucket":"src","src_prefix":"instances/h/","dst_bucket":"other-dst","dst_prefix":"moves/h/",` + |
| + `"objects":{"ltx/1/a.ltx":{"src_etag":"a-27","src_size":500,"dst_etag":"a"}}}` |
| + g.store("src", testReceipts.Key, fakeObject{MirrorObject{ETag: "book", Size: int64(len(other))}, other}) |
| + if r := mirrorRound(t, g, testReceipts); r.Changed != 1 { |
| + t.Fatalf("round over another pair's book = %+v, want the object copied again", r) |
| + } |
| +} |
| + |
| +// forEachParallel on its own: an error stops dispatch, and an already-ended |
| +// context starts nothing. |
| +func TestForEachParallel(t *testing.T) { |
| + var ran atomic.Int32 |
| + err := forEachParallel(context.Background(), 1, 10, func(_ context.Context, i int) error { |
| + ran.Add(1) |
| + if i == 3 { |
| + return errors.New("boom") |
| + } |
| + return nil |
| + }) |
| + if err == nil || err.Error() != "boom" || ran.Load() > 5 { |
| + t.Fatalf("err %v after %d calls, want boom after at most 5", err, ran.Load()) |
| + } |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + cancel() |
| + ran.Store(0) |
| + err = forEachParallel(ctx, 4, 10, func(context.Context, int) error { ran.Add(1); return nil }) |
| + if !errors.Is(err, context.Canceled) || ran.Load() > 4 { |
| + t.Fatalf("err %v after %d calls on an ended context", err, ran.Load()) |
| + } |
| + if err := forEachParallel(context.Background(), 8, 0, nil); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
diff --git a/internal/lease/mirror_test.go b/internal/lease/mirror_test.go| index 2e9bae104..169449135 100644 |
| --- a/internal/lease/mirror_test.go |
| +++ b/internal/lease/mirror_test.go |
| @@ -5,6 +5,7 @@ import ( |
| "encoding/json" |
| "fmt" |
| "strings" |
| + "sync" |
| "testing" |
| ) |
| @@ -15,16 +16,29 @@ import ( |
| // single-part: the destination's ETag is the body, whatever the source's |
| // was. The fakes before #175 kept the source ETag on copy, and hid the bug |
| // that made every multipart object look changed on every round. |
| +// |
| +// A round runs its copies side by side, so every method holds mu, and the |
| +// hooks (beforeCopy, onCopy) run without it: a hook may call the fake. |
| +// workers is the round's concurrency (MirrorConcurrency); zero is the |
| +// production default. |
| type fakeMirror struct { |
| + mu sync.Mutex |
| buckets map[string]map[string]fakeObject |
| ops []string |
| failOn string |
| // beforeCopy runs at the top of Copy, before the precondition check: |
| // the window between a round's listing and its copy. |
| beforeCopy func(srcKey string) |
| - failPut bool |
| - failGet bool |
| - puts int |
| + // onCopy runs after the precondition check and before the object |
| + // lands; ctx is the copy's own. An error it returns is the copy's. |
| + onCopy func(ctx context.Context, srcKey string) error |
| + failPut bool |
| + failGet bool |
| + puts int |
| + gets int |
| + workers int |
| + // inFlight and peak count the copies running at once. |
| + inFlight, peak int |
| } |
| type fakeObject struct { |
| @@ -34,6 +48,13 @@ type fakeObject struct { |
| func newFakeMirror() *fakeMirror { return &fakeMirror{buckets: map[string]map[string]fakeObject{}} } |
| +func (f *fakeMirror) MirrorWorkers() int { |
| + if f.workers == 0 { |
| + return DefaultMirrorWorkers |
| + } |
| + return f.workers |
| +} |
| + |
| func (f *fakeMirror) put(bucket, key, etag string) { f.putSized(bucket, key, etag, 0) } |
| // putSized stores a single-part object: its ETag is its body. |
| @@ -48,13 +69,36 @@ func (f *fakeMirror) putMultipart(bucket, key, body string, size int64) { |
| } |
| func (f *fakeMirror) store(bucket, key string, o fakeObject) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + f.storeLocked(bucket, key, o) |
| +} |
| + |
| +func (f *fakeMirror) storeLocked(bucket, key string, o fakeObject) { |
| if f.buckets[bucket] == nil { |
| f.buckets[bucket] = map[string]fakeObject{} |
| } |
| f.buckets[bucket][key] = o |
| } |
| +// remove deletes one object, as a test's own hand does. |
| +func (f *fakeMirror) remove(bucket, key string) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + delete(f.buckets[bucket], key) |
| +} |
| + |
| +// object is one stored object, as a test reads it. |
| +func (f *fakeMirror) object(bucket, key string) (fakeObject, bool) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + o, ok := f.buckets[bucket][key] |
| + return o, ok |
| +} |
| + |
| func (f *fakeMirror) List(_ context.Context, bucket, prefix string) (map[string]MirrorObject, error) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| out := map[string]MirrorObject{} |
| for k, v := range f.buckets[bucket] { |
| if strings.HasPrefix(k, prefix) { |
| @@ -64,32 +108,58 @@ func (f *fakeMirror) List(_ context.Context, bucket, prefix string) (map[string] |
| return out, nil |
| } |
| -func (f *fakeMirror) Copy(_ context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) { |
| - if f.failOn == srcKey { |
| +func (f *fakeMirror) Copy(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) { |
| + f.mu.Lock() |
| + failOn, before, on := f.failOn, f.beforeCopy, f.onCopy |
| + f.inFlight++ |
| + if f.inFlight > f.peak { |
| + f.peak = f.inFlight |
| + } |
| + f.mu.Unlock() |
| + defer func() { |
| + f.mu.Lock() |
| + f.inFlight-- |
| + f.mu.Unlock() |
| + }() |
| + if failOn == srcKey { |
| return "", fmt.Errorf("injected copy failure on %s", srcKey) |
| } |
| - if f.beforeCopy != nil { |
| - f.beforeCopy(srcKey) |
| + if before != nil { |
| + before(srcKey) |
| } |
| + f.mu.Lock() |
| o, ok := f.buckets[srcBucket][srcKey] |
| + f.mu.Unlock() |
| if !ok { |
| return "", fmt.Errorf("no such key %s", srcKey) |
| } |
| if srcETag != "" && o.ETag != srcETag { |
| return "", ErrCopySourceChanged |
| } |
| + if on != nil { |
| + if err := on(ctx, srcKey); err != nil { |
| + return "", err |
| + } |
| + } |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| f.ops = append(f.ops, "copy "+srcKey+" -> "+dstBucket+"/"+dstKey) |
| - f.store(dstBucket, dstKey, fakeObject{MirrorObject{ETag: o.body, Size: o.Size}, o.body}) |
| + f.storeLocked(dstBucket, dstKey, fakeObject{MirrorObject{ETag: o.body, Size: o.Size}, o.body}) |
| return o.body, nil |
| } |
| func (f *fakeMirror) Delete(_ context.Context, bucket, key string) error { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| f.ops = append(f.ops, "delete "+bucket+"/"+key) |
| delete(f.buckets[bucket], key) |
| return nil |
| } |
| func (f *fakeMirror) Get(_ context.Context, bucket, key string) ([]byte, error) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + f.gets++ |
| if f.failGet { |
| return nil, fmt.Errorf("injected get failure on %s", key) |
| } |
| @@ -101,12 +171,14 @@ func (f *fakeMirror) Get(_ context.Context, bucket, key string) ([]byte, error) |
| } |
| func (f *fakeMirror) Put(_ context.Context, bucket, key string, body []byte) error { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| if f.failPut { |
| return fmt.Errorf("injected put failure on %s", key) |
| } |
| f.puts++ |
| f.ops = append(f.ops, "put "+bucket+"/"+key) |
| - f.store(bucket, key, fakeObject{MirrorObject{ETag: fmt.Sprint("put-", f.puts), Size: int64(len(body))}, string(body)}) |
| + f.storeLocked(bucket, key, fakeObject{MirrorObject{ETag: fmt.Sprint("put-", f.puts), Size: int64(len(body))}, string(body)}) |
| return nil |
| } |
| @@ -390,10 +462,10 @@ func TestMirrorPrefixSourceChangedMidRoundIsDeferred(t *testing.T) { |
| f := newFakeMirror() |
| f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20) |
| f.putSized("src", "instances/h/ltx/0/b.ltx", "b", 10) |
| + var once sync.Once |
| f.beforeCopy = func(k string) { |
| if k == "instances/h/ltx/1/a.ltx" { |
| - f.putMultipart("src", k, "v2", 120<<20) |
| - f.beforeCopy = nil |
| + once.Do(func() { f.putMultipart("src", k, "v2", 120<<20) }) |
| } |
| } |
| r := mirrorRound(t, f, testReceipts) |
| @@ -495,16 +567,22 @@ func (f *fakeVersionedMirror) CopyVersioned(ctx context.Context, sb, sk, se, db, |
| if err != nil { |
| return "", "", err |
| } |
| - f.n++ |
| - return etag, fmt.Sprintf("v%d", f.n), nil |
| + return etag, f.next(), nil |
| } |
| func (f *fakeVersionedMirror) PutVersioned(ctx context.Context, b, k string, body []byte) (string, error) { |
| if err := f.fakeMirror.Put(ctx, b, k, body); err != nil { |
| return "", err |
| } |
| + return f.next(), nil |
| +} |
| + |
| +// next is the next version id; copies run side by side, so under mu. |
| +func (f *fakeVersionedMirror) next() string { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| f.n++ |
| - return fmt.Sprintf("v%d", f.n), nil |
| + return fmt.Sprintf("v%d", f.n) |
| } |
| func TestMirrorPrefixRecordsDestinationVersions(t *testing.T) { |