amadan
moves: copy receipts, so an unchanged multipart object is not recopied (#175)
S3 CopyObject always writes a single-part object, so a multipart source
(litestream uploads anything over one 5 MiB part that way) lands in
staging with a different ETag. MirrorPrefix compared ETags only, so it
recopied every multipart object on every round. In production every large
object under kpmg.bdn.oncarlos.com's replica is multipart ("...-27",
138 MB each), so each round re-staged ~1 GB and the move could never
reach ready, busy or idle. The 16 MiB bar alone could not fix that
(Astra's P2 on fe3d31b8).
A copy receipt records what one copy did: the source ETag and size it
copied, and the ETag the destination landed with (now returned by
MirrorStore.Copy). An object is skipped when the two ETags already match
(single-part, unchanged behaviour and request count), or when the source
still lists the receipt's ETag and size AND staging still lists the
receipt's destination ETag. Never on size alone. Copies are conditional
on the listed source ETag (CopySourceIfMatch): a source overwritten
between list and copy copies nothing, records no receipt, and is
deferred to the next round, with its bytes counted against the bar.
Receipts are written once, at the end of the round, after the copies
they describe, so a crash in between costs one recopy and can never
cause a wrong skip. A receipts book names its src/dst pair and is ignored
for any other. A receipts read that fails (not "missing") fails the
round rather than silently recopying everything or overwriting the book.
Where they live: the source box's grant on the target's moves/* is
push-only (put, delete, prefix-scoped list; no get), so receipts beside
the staged copy could be written but never read back. They live in the
source's OWN data bucket at move-receipts/<host>.json
(activator.MoveReceiptsKey), outside moves/ and the hibernate prefix, so
no restore, promotion or staging cleanup ever reads them. Cut and abort
delete them best-effort after the record write; a delete failure (some
deployments deny DeleteObject on the data bucket) only logs.
The cut refuses to purge and flip when its final round deferred anything,
and retries next pass, so it still lands byte-exact. The activator's
landing promotion takes no receipts (staged objects are single-part
copies) and fails the wake on a deferred object.
Also Astra's P3: a zero bar means zero bytes moved, so a deletion-only
round promotes; comment and test renamed to say so.
The fakes now follow S3's ETag rules (a copy's ETag is the content's,
whatever the source's was) and compare content, not metadata, for
byte-exact checks; the old fakes kept the source ETag and hid this bug.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
6 files changed,
+1029
−93
internal/activator/backend_exec.go+18 −1internal/activator/backend_exec_test.go+68 −2internal/host/movespass.go+42 −13internal/host/movespass_test.go+254 −25internal/lease/mirror.go+320 −29internal/lease/mirror_test.go+327 −23
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index 613d8e761..114ee0987 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -1548,6 +1548,14 @@ func (p *childProc) Stop(ctx context.Context) error { |
| // package spell it identically. |
| func MovesPrefix(host string) string { return "moves/" + host + "/" } |
| +// MoveReceiptsKey is where a residence move's mirror keeps its copy |
| +// receipts (#175, lease.Receipts) in the SOURCE box's own data bucket. The |
| +// source cannot read the target's moves/ prefix (its grant there is |
| +// push-only), so the receipts cannot live beside the staged copy. This key |
| +// is under neither MovesPrefix nor the hibernate prefix, so no restore, |
| +// promotion or staging cleanup ever sees it. |
| +func MoveReceiptsKey(host string) string { return "move-receipts/" + host + ".json" } |
| + |
| // PurgeLocalDBArtifacts removes the local litestream footprint next to a |
| // DB — WAL, SHM and both meta dirs — and, when full is true, the DB file |
| // itself. Restore's pre-restore cleanup uses the partial form (the DB's |
| @@ -1590,9 +1598,18 @@ func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (ma |
| return nil, nil |
| } |
| log.Printf("activator: %s: promoting %d staged object(s) from %s (residence move landing)", host, len(staged), MovesPrefix(host)) |
| - if _, err := lease.MirrorPrefix(ctx, b.Mirror, b.Cfg.Bucket, MovesPrefix(host), b.Cfg.Bucket, b.Cfg.hibernatePrefix(host)); err != nil { |
| + // No receipts: a staged object is the source box's single-part copy, |
| + // and an in-bucket copy of it keeps its ETag, so plain ETag equality |
| + // is exact here. A deferred object (staging overwritten mid-promote, |
| + // which nothing should do once the move has cut) fails the wake |
| + // rather than restoring from a promotion that is not byte-final. |
| + round, err := lease.MirrorPrefix(ctx, b.Mirror, b.Cfg.Bucket, MovesPrefix(host), b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), lease.Receipts{}) |
| + if err != nil { |
| return nil, err |
| } |
| + if round.Deferred > 0 { |
| + return nil, fmt.Errorf("promote staged replica for %s: %d staged object(s) changed during the copy; retrying on the next wake", host, round.Deferred) |
| + } |
| return staged, nil |
| } |
diff --git a/internal/activator/backend_exec_test.go b/internal/activator/backend_exec_test.go| index 329da4288..8b4c41ceb 100644 |
| --- a/internal/activator/backend_exec_test.go |
| +++ b/internal/activator/backend_exec_test.go |
| @@ -198,6 +198,58 @@ func TestRestoreStagingFallback(t *testing.T) { |
| f.failList = false |
| } |
| +// A move's copy receipts live in the SOURCE box's own bucket at |
| +// MoveReceiptsKey, and this box may be that source for an earlier move of |
| +// the same host (a move back). Landing must never take them for data: |
| +// they are not under MovesPrefix, so they are never promoted, and not |
| +// under the hibernate prefix, so they are never restored from or swept. |
| +func TestRestoreIgnoresMoveReceipts(t *testing.T) { |
| + ctx := context.Background() |
| + b := testExecBackend(t) |
| + f := newFakeMirrorStore() |
| + b.Mirror = f |
| + host := "moved.example.com" |
| + rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "i.sock")} |
| + |
| + receipts := MoveReceiptsKey(host) |
| + if strings.HasPrefix(receipts, MovesPrefix(host)) || strings.HasPrefix(receipts, b.Cfg.hibernatePrefix(host)+"/") || |
| + strings.HasPrefix(receipts, b.Cfg.Prefix+"/") { |
| + t.Fatalf("receipts key %q sits under a data prefix", receipts) |
| + } |
| + |
| + var restoredFrom []string |
| + oldRestore := restoreReplica |
| + restoreReplica = func(_ context.Context, dbPath, bucket, prefix, region string) (bool, error) { |
| + objs, _ := f.List(context.Background(), bucket, prefix) |
| + if len(objs) == 0 { |
| + return false, nil |
| + } |
| + restoredFrom = append(restoredFrom, bucket+"/"+prefix) |
| + return true, writeMinimalSQLite(dbPath) |
| + } |
| + defer func() { restoreReplica = oldRestore }() |
| + |
| + // Receipts alone: a fresh boot, nothing promoted, receipts untouched. |
| + f.put(b.Cfg.Bucket, receipts, "r1") |
| + fresh, err := b.Restore(ctx, rt) |
| + if err != nil || !fresh || len(restoredFrom) != 0 { |
| + t.Fatalf("receipts alone: fresh=%v err=%v restored=%v, want a fresh boot", fresh, err, restoredFrom) |
| + } |
| + // Beside a staged copy: only the staged data is promoted. |
| + f.put(b.Cfg.Bucket, "moves/"+host+"/ltx/0/a.ltx", "e1") |
| + _ = os.Remove(filepath.Join(b.DataDir, host+".db")) |
| + if _, err := b.Restore(ctx, rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + own, _ := f.List(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host)) |
| + if len(own) != 1 { |
| + t.Fatalf("own prefix after promotion = %v, want only the staged object", own) |
| + } |
| + if _, ok := f.buckets[b.Cfg.Bucket][receipts]; !ok { |
| + t.Fatal("landing deleted the receipts of a move it did not make") |
| + } |
| +} |
| + |
| // fakeMirrorStore is the test double for lease.MirrorStore, bucket→key→object. |
| type fakeMirrorStore struct { |
| buckets map[string]map[string]lease.MirrorObject |
| @@ -225,11 +277,25 @@ 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 { |
| +func (f *fakeMirrorStore) Copy(_ context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) { |
| + o, ok := f.buckets[srcBucket][srcKey] |
| + if !ok || (srcETag != "" && o.ETag != srcETag) { |
| + return "", lease.ErrCopySourceChanged |
| + } |
| if f.buckets[dstBucket] == nil { |
| f.buckets[dstBucket] = map[string]lease.MirrorObject{} |
| } |
| - f.buckets[dstBucket][dstKey] = f.buckets[srcBucket][srcKey] |
| + f.buckets[dstBucket][dstKey] = o |
| + return o.ETag, nil |
| +} |
| +func (f *fakeMirrorStore) Get(_ context.Context, bucket, key string) ([]byte, error) { |
| + 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") |
| 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 e43a2d48b..f64fe95ed 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -32,10 +32,13 @@ import ( |
| // carlos-adopt.path), and prepare's convergence is the round-over-round |
| // shrinking of the delta, not a resident goroutine — restart-safe because |
| // every step derives from the record and the buckets. "Converged" is a |
| -// round that copied at most moveCfg.ReadyDelta bytes (16 MiB by default), |
| -// not a round that copied nothing: an awake instance never produces the |
| -// latter (#175). Ready therefore means "small gap", and the cut's final |
| -// round below is what makes staging exact. |
| +// round that moved (copied or deferred) at most moveCfg.ReadyDelta bytes, |
| +// 16 MiB by default, not a round that copied nothing: an awake instance |
| +// never produces the latter (#175). Ready therefore means "small gap", and |
| +// the cut's final round below is what makes staging exact. "Unchanged" is |
| +// exact: the copy receipts (lease.Receipts, kept in this box's own bucket |
| +// at activator.MoveReceiptsKey) are what let a multipart source, whose |
| +// copy never shares its ETag, count as unchanged at all. |
| // |
| // The cut ordering is the never-two-writers argument, mechanically: |
| // |
| @@ -193,10 +196,11 @@ type moveCfg struct { |
| // replica really carries the tail. Nil skips (tests without S3). |
| Checkpoint func(ctx context.Context, dbPath string) error |
| Now func() time.Time |
| - // ReadyDelta is the readiness bar (#175): a preparing round that copied |
| - // at most this many bytes promotes the move to ready. Zero means only a |
| - // round that copied nothing at all qualifies, which an awake instance |
| - // never produces. armMovesPass sets it from moveReadyDeltaFromEnv. |
| + // ReadyDelta is the readiness bar (#175): a preparing round whose |
| + // lease.MirrorRound.DeltaBytes (bytes copied plus bytes deferred) is at |
| + // most this promotes the move to ready. Zero means zero bytes moved: a |
| + // round that only deleted still qualifies. armMovesPass sets it from |
| + // moveReadyDeltaFromEnv. |
| ReadyDelta int64 |
| } |
| @@ -211,7 +215,7 @@ type moveCfg struct { |
| const defaultMoveReadyDelta int64 = 16 << 20 |
| // moveReadyDeltaEnv overrides defaultMoveReadyDelta with a whole number of |
| -// bytes. Zero is accepted and restores the old "nothing changed" rule. |
| +// bytes. Zero is accepted and means a round must move zero bytes. |
| const moveReadyDeltaEnv = "CARLOS_MOVE_READY_DELTA_BYTES" |
| // moveReadyDeltaFromEnv reads the readiness bar. An unset, unparsable or |
| @@ -359,6 +363,10 @@ func runMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec prov |
| mirror := func() (lease.MirrorStore, error) { return w.mirror(ctx, tgt.DataBucketRegion) } |
| srcPrefix := activator.HibernatePrefix(cfg.OwnPrefix, rec.Host) |
| dstPrefix := activator.MovesPrefix(rec.Host) |
| + // The copy receipts (#175) live in this box's OWN bucket: the staging |
| + // grant on the target is push-only, so receipts there could be |
| + // written but never read back. |
| + receipts := lease.Receipts{Bucket: cfg.OwnBucket, Key: activator.MoveReceiptsKey(rec.Host)} |
| switch rec.Move.State { |
| case provision.MovePreparing, provision.MoveReady: |
| @@ -366,7 +374,7 @@ func runMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec prov |
| if err != nil { |
| return err |
| } |
| - round, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix) |
| + round, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts) |
| if err != nil { |
| return err |
| } |
| @@ -376,12 +384,12 @@ func runMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec prov |
| // closes the gap itself (quiesce, fence, final checkpoint, a final |
| // round), so this bar only bounds how much that last round, and |
| // the downtime around it, has left to copy. |
| - if rec.Move.State == provision.MovePreparing && round.BytesCopied <= cfg.ReadyDelta { |
| + if rec.Move.State == provision.MovePreparing && round.DeltaBytes() <= cfg.ReadyDelta { |
| 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 |
| } |
| nudgeSerial(ctx, s) |
| - log.Printf("moves: %s: staged replica caught up in %s/%s (last round copied %d bytes in %d changes, bar %d) — ready to cut", rec.Host, tgt.DataBucket, dstPrefix, round.BytesCopied, round.Changed, cfg.ReadyDelta) |
| + log.Printf("moves: %s: staged replica caught up in %s/%s (last round copied %d bytes, deferred %d, in %d changes; bar %d) — ready to cut", rec.Host, tgt.DataBucket, dstPrefix, round.BytesCopied, round.BytesDeferred, round.Changed, cfg.ReadyDelta) |
| } |
| return nil |
| @@ -399,6 +407,7 @@ func runMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec prov |
| if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveAbortComplete()); err != nil && !errors.Is(err, provision.ErrMoveConflict) { |
| return err |
| } |
| + clearMoveReceipts(ctx, m, receipts, rec.Host) |
| nudgeSerial(ctx, s) |
| log.Printf("moves: %s: abort complete — staged copy deleted, instance stays in %q", rec.Host, rec.Region) |
| return nil |
| @@ -468,15 +477,24 @@ func runMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec prov |
| if err != nil { |
| return err |
| } |
| - if _, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix); err != nil { |
| + round, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts) |
| + if err != nil { |
| return err |
| } |
| + // A deferred object is one the source changed under this round, so |
| + // staging is NOT byte-final. Nothing should write a quiesced, |
| + // fenced, checkpointed source, but the purge below is |
| + // irreversible, so it waits for a round that proves it. |
| + if round.Deferred > 0 { |
| + return fmt.Errorf("final mirror round: %d object(s) changed at the source during the copy; retrying the cut", round.Deferred) |
| + } |
| // Step 5: the warm-cache purge. |
| activator.PurgeLocalDBArtifacts(dbPath, true) |
| // Step 6: the last act — Region flips, Move clears, one write. |
| if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveComplete()); err != nil && !errors.Is(err, provision.ErrMoveConflict) { |
| return err |
| } |
| + clearMoveReceipts(ctx, m, receipts, rec.Host) |
| w.nudged.Delete(rec.Host) |
| // The flip's nudge is what turns the TARGET's mint from a |
| // 5-minute wait into a seconds-cadence converge. |
| @@ -487,6 +505,17 @@ func runMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec prov |
| return nil |
| } |
| +// clearMoveReceipts deletes a finished or abandoned move's copy receipts. |
| +// Best-effort, after the record write that ended the move: receipts are |
| +// self-checking (a stale one can only fail to match, never vouch for the |
| +// wrong object), and some deployments' store policy denies DeleteObject on |
| +// the data bucket, so a failure costs one small object, never a move. |
| +func clearMoveReceipts(ctx context.Context, m lease.MirrorStore, rc lease.Receipts, host string) { |
| + if err := m.Delete(ctx, rc.Bucket, rc.Key); err != nil { |
| + log.Printf("moves: %s: could not delete copy receipts %s/%s: %v (harmless; they no longer match anything)", host, rc.Bucket, rc.Key, err) |
| + } |
| +} |
| + |
| // nudgeSerial bumps control/serial after a box-side move write, so the |
| // NEXT actor — this box's own cut continuation, the target's mint, the |
| // tenant's poll — reacts on the fast-converge path instead of the 5-minute |
diff --git a/internal/host/movespass_test.go b/internal/host/movespass_test.go| index 7ee3701db..883797a3e 100644 |
| --- a/internal/host/movespass_test.go |
| +++ b/internal/host/movespass_test.go |
| @@ -17,42 +17,87 @@ import ( |
| "github.com/carlosframework/platform/internal/release" |
| ) |
| -// fakeMoveMirror: bucket→key→object plus an op log for ordering assertions. |
| +// fakeMoveMirror keeps S3's ETag rules (lease's fakeMirror explains them): |
| +// 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. |
| type fakeMoveMirror struct { |
| - buckets map[string]map[string]lease.MirrorObject |
| - ops []string |
| + buckets map[string]map[string]fakeMoveObject |
| + ops []string |
| + beforeCopy func(srcKey string) |
| +} |
| + |
| +type fakeMoveObject struct { |
| + lease.MirrorObject |
| + body string |
| } |
| func newFakeMoveMirror() *fakeMoveMirror { |
| - return &fakeMoveMirror{buckets: map[string]map[string]lease.MirrorObject{}} |
| + return &fakeMoveMirror{buckets: map[string]map[string]fakeMoveObject{}} |
| } |
| func (f *fakeMoveMirror) put(bucket, key, etag string) { f.putSized(bucket, key, etag, 0) } |
| func (f *fakeMoveMirror) putSized(bucket, key, etag string, size int64) { |
| + f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: etag, Size: size}, etag}) |
| +} |
| +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) { |
| if f.buckets[bucket] == nil { |
| - f.buckets[bucket] = map[string]lease.MirrorObject{} |
| + f.buckets[bucket] = map[string]fakeMoveObject{} |
| } |
| - f.buckets[bucket][key] = lease.MirrorObject{ETag: etag, Size: size} |
| + f.buckets[bucket][key] = o |
| } |
| 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 |
| + out[k] = v.MirrorObject |
| } |
| } |
| return out, nil |
| } |
| -func (f *fakeMoveMirror) Copy(_ context.Context, srcBucket, srcKey, dstBucket, dstKey string) error { |
| +func (f *fakeMoveMirror) Copy(_ context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) { |
| + if f.beforeCopy != nil { |
| + f.beforeCopy(srcKey) |
| + } |
| + o, ok := f.buckets[srcBucket][srcKey] |
| + if !ok || (srcETag != "" && o.ETag != srcETag) { |
| + return "", lease.ErrCopySourceChanged |
| + } |
| f.ops = append(f.ops, "copy "+dstBucket+"/"+dstKey) |
| - o := f.buckets[srcBucket][srcKey] |
| - f.putSized(dstBucket, dstKey, o.ETag, o.Size) |
| - return nil |
| + f.store(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.ops = append(f.ops, "delete "+bucket+"/"+key) |
| delete(f.buckets[bucket], key) |
| return nil |
| } |
| +func (f *fakeMoveMirror) Get(_ context.Context, bucket, key string) ([]byte, error) { |
| + o, ok := f.buckets[bucket][key] |
| + if !ok { |
| + return nil, lease.ErrMirrorNotFound |
| + } |
| + return []byte(o.body), nil |
| +} |
| +func (f *fakeMoveMirror) Put(_ context.Context, bucket, key string, body []byte) error { |
| + f.ops = append(f.ops, "put "+bucket+"/"+key) |
| + f.store(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 { |
| + out := map[string]string{} |
| + for k, v := range f.buckets[bucket] { |
| + if strings.HasPrefix(k, prefix) { |
| + out[strings.TrimPrefix(k, prefix)] = v.body |
| + } |
| + } |
| + return out |
| +} |
| type moveHarness struct { |
| s release.Store |
| @@ -554,8 +599,10 @@ func TestMovesPassLargeDeltaDoesNotPromote(t *testing.T) { |
| } |
| } |
| -// A zero bar is the old rule: only a round that copied nothing promotes. |
| -func TestMovesPassZeroBarNeedsAQuietRound(t *testing.T) { |
| +// A zero bar means zero bytes moved: a round that copied (or deferred) |
| +// even one byte does not promote. A deletion-only round moves no bytes, so |
| +// it does, which the old Changed == 0 rule did not allow. |
| +func TestMovesPassZeroBarNeedsZeroCopiedBytes(t *testing.T) { |
| h := newMoveHarness(t) |
| h.cfg.ReadyDelta = 0 |
| host := "a.go.example.com" |
| @@ -567,6 +614,17 @@ func TestMovesPassZeroBarNeedsAQuietRound(t *testing.T) { |
| if st := h.moveRound(t, host); st != provision.MoveReady { |
| t.Fatalf("a quiet round under a zero bar left the move %s", st) |
| } |
| + |
| + // Deletion-only: retention pruned a staged object, nothing to copy. |
| + other := "b.go.example.com" |
| + h.record(t, other, provision.MoveRequest("uk", time.Now())) |
| + h.f.putSized("data-ie", "instances/"+other+"/generations/g/wal/000", "e1", 1) |
| + h.moveRound(t, other) |
| + delete(h.f.buckets["data-ie"], "instances/"+other+"/generations/g/wal/000") |
| + h.f.putSized("data-ie", "instances/"+other+"/generations/g/wal/001", "e2", 0) |
| + if st := h.moveRound(t, other); st != provision.MoveReady { |
| + t.Fatalf("a zero-byte round (one delete, one empty copy) under a zero bar left the move %s", st) |
| + } |
| } |
| // Ready no longer means "staging matched the source". It means the gap was |
| @@ -613,25 +671,169 @@ func TestMovesPassCutAfterBusyReadyLandsByteExact(t *testing.T) { |
| t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move) |
| } |
| - src, _ := h.f.List(context.Background(), "data-ie", own) |
| - staged, _ = h.f.List(context.Background(), "data-uk", "moves/"+host+"/") |
| - want := map[string]lease.MirrorObject{} |
| - for k, v := range src { |
| - if strings.HasSuffix(k, "/lock.json") { |
| - continue |
| + assertStagedExact(t, h, host) |
| +} |
| + |
| +// assertStagedExact compares CONTENT, source against staging, lease |
| +// excluded: a multipart source and its copy carry different ETags by |
| +// design, so metadata equality would prove nothing. |
| +func assertStagedExact(t *testing.T, h *moveHarness, host string) { |
| + t.Helper() |
| + want := h.f.bodies("data-ie", "instances/"+host+"/") |
| + delete(want, "lock.json") |
| + got := h.f.bodies("data-uk", "moves/"+host+"/") |
| + if len(got) != len(want) { |
| + t.Fatalf("staged %d objects, want %d:\nstaged %v\nwant %v", len(got), len(want), got, want) |
| + } |
| + for k, v := range want { |
| + if got[k] != v { |
| + t.Fatalf("staged %s = %q, want %q (the cut's final round must make staging exact)", k, got[k], v) |
| + } |
| + } |
| +} |
| + |
| +// kpmgShape stocks a source like kpmg.bdn.oncarlos.com's (#175): a |
| +// multipart snapshot and six multipart 138 MB level files, all of which |
| +// land in staging with a DIFFERENT ETag than the source lists. |
| +func kpmgShape(h *moveHarness, host string) { |
| + gen := "instances/" + host + "/ltx/" |
| + h.f.putMultipart("data-ie", gen+"9/snapshot.ltx", "snap", 293<<20) |
| + for i := 0; i < 6; i++ { |
| + h.f.putMultipart("data-ie", fmt.Sprint(gen, "1/", i, ".ltx"), fmt.Sprint("level-", i), 138<<20) |
| + } |
| + h.f.putSized("data-ie", "instances/"+host+"/lock.json", "renewal-0", 200) |
| +} |
| + |
| +// The production failure (#175): multipart sources never compare equal to |
| +// their copies, so before receipts every round recopied ~1 GB and the |
| +// move could never reach ready, busy or not. With receipts, a busy |
| +// instance whose rounds add only small L0 files reaches ready on the |
| +// first round after the bulk. |
| +func TestMovesPassBusyMultipartInstanceReachesReady(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "kpmg.bdn.example.com" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now())) |
| + kpmgShape(h, host) |
| + if st := h.moveRound(t, host); st != provision.MovePreparing { |
| + t.Fatalf("after the bulk round: %s, want preparing", st) |
| + } |
| + h.f.putSized("data-ie", "instances/"+host+"/ltx/0/1.ltx", "l0-1", 8<<10) |
| + h.f.putSized("data-ie", "instances/"+host+"/lock.json", "renewal-1", 200) |
| + before := len(h.f.ops) |
| + if st := h.moveRound(t, host); st != provision.MoveReady { |
| + t.Fatalf("a busy multipart instance with an 8 KiB delta left the move %s", st) |
| + } |
| + for _, op := range h.f.ops[before:] { |
| + if strings.Contains(op, "/ltx/1/") || strings.Contains(op, "/ltx/9/") { |
| + t.Fatalf("an unchanged multipart object was recopied: %s (ops %v)", op, h.f.ops[before:]) |
| } |
| - want["moves/"+host+"/"+strings.TrimPrefix(k, own)] = v |
| } |
| - if len(staged) != len(want) { |
| - t.Fatalf("staged %d objects, want %d:\nstaged %v\nwant %v", len(staged), len(want), staged, want) |
| + // The receipts are in the SOURCE's bucket, never in staging. |
| + if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; !ok { |
| + t.Fatal("no receipts in the source's bucket") |
| } |
| - for k, v := range want { |
| - if staged[k] != v { |
| - t.Fatalf("staged[%s] = %+v, want %+v (the cut's final round must make staging exact)", k, staged[k], v) |
| + for k := range h.f.buckets["data-uk"] { |
| + if strings.Contains(k, "receipt") { |
| + t.Fatalf("receipts staged as data: %s", k) |
| } |
| } |
| } |
| +// A source object overwritten between the listing and the copy defers |
| +// that object. Its listed bytes still count against the bar, so a round |
| +// that deferred a large object does not promote. |
| +func TestMovesPassDeferredBytesCountAgainstTheBar(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "a.go.example.com" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now())) |
| + kpmgShape(h, host) |
| + h.moveRound(t, host) |
| + big := "instances/" + host + "/ltx/1/0.ltx" |
| + h.f.putMultipart("data-ie", big, "level-0b", 138<<20) |
| + h.f.beforeCopy = func(k string) { |
| + if k == big { |
| + h.f.putMultipart("data-ie", big, "level-0c", 138<<20) |
| + h.f.beforeCopy = nil |
| + } |
| + } |
| + if st := h.moveRound(t, host); st != provision.MovePreparing { |
| + t.Fatalf("a round that deferred 138 MiB promoted to %s", st) |
| + } |
| + if st := h.moveRound(t, host); st != provision.MovePreparing { |
| + t.Fatalf("the round that copied the deferred 138 MiB promoted to %s", st) |
| + } |
| + if st := h.moveRound(t, host); st != provision.MoveReady { |
| + t.Fatalf("the quiet round after left the move %s", st) |
| + } |
| +} |
| + |
| +// The cut's final round with multipart sources lands byte-exact, and the |
| +// move's receipts are cleared once the record has flipped. |
| +func TestMovesPassCutWithMultipartSourcesLandsExact(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "kpmg.bdn.example.com" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now())) |
| + kpmgShape(h, host) |
| + h.moveRound(t, host) |
| + if st := h.moveRound(t, host); st != provision.MoveReady { |
| + t.Fatalf("state %s, want ready", st) |
| + } |
| + // Compaction after ready: one level file rewritten, one pruned. |
| + h.f.putMultipart("data-ie", "instances/"+host+"/ltx/1/2.ltx", "level-2b", 140<<20) |
| + delete(h.f.buckets["data-ie"], "instances/"+host+"/ltx/1/5.ltx") |
| + if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, provision.MoveMarkCutting()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := movesPass(context.Background(), h.s, h.worker()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := h.read(t, host); got.Region != "uk" || got.Move != nil { |
| + t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move) |
| + } |
| + assertStagedExact(t, h, host) |
| + if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; ok { |
| + t.Fatal("the move's receipts outlived the cut") |
| + } |
| +} |
| + |
| +// A source that changes under the cut's final round (it must not, the |
| +// child is down, but nothing else proves it) leaves staging short of |
| +// exact. The cut stops before the purge and the flip, and the next pass |
| +// finishes it exactly. |
| +func TestMovesPassCutStopsOnADeferredCopy(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "a.go.example.com" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting()) |
| + kpmgShape(h, host) |
| + db := filepath.Join(h.cfg.DataDir, host+".db") |
| + if err := writeFileForTest(db, "warm"); err != nil { |
| + t.Fatal(err) |
| + } |
| + big := "instances/" + host + "/ltx/1/3.ltx" |
| + h.f.beforeCopy = func(k string) { |
| + if k == big { |
| + h.f.putMultipart("data-ie", big, "level-3b", 138<<20) |
| + h.f.beforeCopy = nil |
| + } |
| + } |
| + if err := movesPass(context.Background(), h.s, h.worker()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := h.read(t, host); got.Region != "ie" || got.Move == nil || got.Move.State != provision.MoveCutting { |
| + t.Fatalf("the cut completed over a deferred copy: region=%q move=%+v", got.Region, got.Move) |
| + } |
| + if !fileExistsForTest(db) { |
| + t.Fatal("the local DB was purged before staging was exact") |
| + } |
| + if err := movesPass(context.Background(), h.s, h.worker()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := h.read(t, host); got.Region != "uk" || got.Move != nil { |
| + t.Fatalf("the retried cut: region=%q move=%+v", got.Region, got.Move) |
| + } |
| + assertStagedExact(t, h, host) |
| +} |
| + |
| func TestMoveReadyDeltaFromEnv(t *testing.T) { |
| for _, tc := range []struct { |
| raw string |
| @@ -741,3 +943,30 @@ func TestMovesPassAbortPhase(t *testing.T) { |
| t.Fatalf("after abort phase: %+v region=%q", got.Move, got.Region) |
| } |
| } |
| + |
| +// Abort clears the move's receipts along with the staged copy. |
| +func TestMovesPassAbortClearsReceipts(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "a.go.example.com" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now())) |
| + kpmgShape(h, host) |
| + h.moveRound(t, host) |
| + if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; !ok { |
| + t.Fatal("no receipts after the bulk round") |
| + } |
| + if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, provision.MoveAbort()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := movesPass(context.Background(), h.s, h.worker()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := h.read(t, host); got.Move != nil { |
| + t.Fatalf("abort did not complete: %+v", got.Move) |
| + } |
| + if staged := h.f.bodies("data-uk", "moves/"+host+"/"); len(staged) != 0 { |
| + t.Fatalf("staging survived the abort: %v", staged) |
| + } |
| + if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; ok { |
| + t.Fatal("the receipts survived the abort") |
| + } |
| +} |
diff --git a/internal/lease/mirror.go b/internal/lease/mirror.go| index f5afb8b7b..cb6ed72b0 100644 |
| --- a/internal/lease/mirror.go |
| +++ b/internal/lease/mirror.go |
| @@ -1,14 +1,20 @@ |
| package lease |
| import ( |
| + "bytes" |
| "context" |
| + "encoding/json" |
| + "errors" |
| "fmt" |
| + "io" |
| "net/url" |
| "sort" |
| "strings" |
| "sync" |
| awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| + s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" |
| + "github.com/aws/smithy-go" |
| ) |
| // The staging mirror (issue #331, instance-move spec): a residence move |
| @@ -22,15 +28,31 @@ import ( |
| // 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 |
| +// MirrorStore is the S3 actions a delta round needs. List answers |
| // key→object under a prefix (keys RELATIVE to the bucket root, not the |
| -// prefix); Copy is a server-side CopyObject; Delete removes one key. |
| +// prefix). Copy is a server-side CopyObject, conditional on srcETag when it |
| +// is set: a source that no longer carries that ETag answers |
| +// ErrCopySourceChanged and copies nothing. It returns the ETag the |
| +// destination object landed with. Delete removes one key. Get and Put read |
| +// and write one small whole object (the copy receipts); Get answers |
| +// ErrMirrorNotFound for a missing key. |
| type MirrorStore interface { |
| List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error) |
| - Copy(ctx context.Context, srcBucket, srcKey, dstBucket, dstKey string) error |
| + Copy(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (dstETag string, err error) |
| Delete(ctx context.Context, bucket, key string) error |
| + Get(ctx context.Context, bucket, key string) ([]byte, error) |
| + Put(ctx context.Context, bucket, key string, body []byte) error |
| } |
| +// ErrCopySourceChanged is a conditional Copy whose source was overwritten |
| +// after the round listed it. The round defers that object to the next |
| +// round rather than failing: a source that keeps changing is a busy |
| +// source, not a broken one. |
| +var ErrCopySourceChanged = errors.New("mirror: copy source changed since it was listed") |
| + |
| +// ErrMirrorNotFound is Get's answer for a key that does not exist. |
| +var ErrMirrorNotFound = errors.New("mirror: no such object") |
| + |
| // MirrorObject is one listed object: the ETag the sameness test compares, |
| // and the size a round's byte accounting adds up. |
| type MirrorObject struct { |
| @@ -40,28 +62,87 @@ type MirrorObject struct { |
| // 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 counts objects copied, deferred 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 sums the listed SOURCE size of every object copied. |
| BytesCopied int64 |
| + // Deferred counts objects whose conditional copy found the source |
| + // overwritten since the listing (ErrCopySourceChanged); BytesDeferred |
| + // sums their listed sizes. They are still to copy, so a round that |
| + // has to be byte-final (the cut's) must not count as done while |
| + // Deferred > 0. |
| + Deferred int |
| + BytesDeferred int64 |
| +} |
| + |
| +// DeltaBytes is how much the round found to move: copied plus deferred. |
| +// The move worker's readiness bar measures this. 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 move bound the work left for the cut's round. |
| +func (r MirrorRound) DeltaBytes() int64 { return r.BytesCopied + r.BytesDeferred } |
| + |
| +// Receipts names the object where a mirror keeps its copy receipts. The |
| +// zero value keeps none, and every object whose ETags differ is copied. |
| +// |
| +// Why receipts exist (#175): S3's CopyObject always writes a single-part |
| +// object, so copying a MULTIPART source (litestream uploads anything over |
| +// one 5 MiB part that way) lands a destination whose ETag differs from the |
| +// source's. ETag equality then reports every multipart object as changed |
| +// on every round, forever: production recopied six 138 MB level files a |
| +// round. A receipt records what one copy did: the source ETag and size it |
| +// copied, and the ETag the destination landed with. An object is skipped |
| +// when the source still lists that ETag and size AND the destination still |
| +// lists that destination ETag, and copied otherwise. |
| +// |
| +// The receipts object must live where the mirroring box can READ it. A |
| +// move's destination is push-only to the source box (it may list, put and |
| +// delete under moves/*, never get), so the move worker keeps them in the |
| +// SOURCE's own bucket, outside every prefix a restore or a promotion |
| +// reads. They are never data. |
| +type Receipts struct { |
| + Bucket, Key string |
| +} |
| + |
| +// receiptBook is the receipts object's shape. It names the pair it |
| +// describes, and a book written for another pair is not evidence about |
| +// this one. |
| +type receiptBook struct { |
| + V int `json:"v"` |
| + SrcBucket string `json:"src_bucket"` |
| + SrcPrefix string `json:"src_prefix"` |
| + DstBucket string `json:"dst_bucket"` |
| + DstPrefix string `json:"dst_prefix"` |
| + Objects map[string]copyReceipt `json:"objects"` // relative key → receipt |
| +} |
| + |
| +// copyReceipt is one completed copy whose destination ETag differs from |
| +// its source's. Only such copies get one: where the ETags already match, |
| +// the plain equality test needs no help. |
| +type copyReceipt struct { |
| + SrcETag string `json:"src_etag"` |
| + SrcSize int64 `json:"src_size"` |
| + DstETag string `json:"dst_etag"` |
| } |
| // 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 what the round did; on error, what it did before the |
| -// failure. |
| +// srcBucket/srcPrefix: copy keys that are new or changed, 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 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) (MirrorRound, error) { |
| +// "Unchanged" is exact, never a guess from size: either the two listed |
| +// ETags are equal (a single-part source, whose copy keeps its ETag), or a |
| +// receipt (see Receipts) proves the destination is this mirror's copy of |
| +// exactly the source version now listed. |
| +// |
| +// Crash safety: a receipt is written only AFTER the copy it describes |
| +// returned, carrying the ETag that copy returned, once per round at the |
| +// end. A crash before that write loses receipts, and the next round copies |
| +// 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) { |
| // 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 |
| @@ -81,6 +162,7 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB |
| if err != nil { |
| return MirrorRound{}, fmt.Errorf("mirror: list staging %s/%s: %w", dstBucket, dstPrefix, err) |
| } |
| + book := receiptLedger{m: m, rc: rc, want: receiptBook{V: 1, SrcBucket: srcBucket, SrcPrefix: srcPrefix, DstBucket: dstBucket, DstPrefix: dstPrefix}} |
| // Deterministic order: a failed round resumes the same way everywhere. |
| rel := make([]string, 0, len(src)) |
| for k := range src { |
| @@ -99,14 +181,43 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB |
| continue |
| } |
| sk, dk := srcPrefix+r, dstPrefix+r |
| - if d, ok := dst[dk]; ok && d.ETag == src[sk].ETag { |
| + s := src[sk] |
| + d, staged := dst[dk] |
| + if staged && d.ETag == s.ETag { |
| continue |
| } |
| - if err := m.Copy(ctx, srcBucket, sk, dstBucket, dk); err != nil { |
| + if staged { |
| + rcpt, ok, err := book.get(ctx, r) |
| + if err != nil { |
| + return round, err |
| + } |
| + if ok && rcpt.SrcETag == s.ETag && rcpt.SrcSize == s.Size && rcpt.DstETag == d.ETag && d.Size == s.Size { |
| + continue |
| + } |
| + } |
| + etag, err := m.Copy(ctx, srcBucket, sk, s.ETag, dstBucket, dk) |
| + if errors.Is(err, ErrCopySourceChanged) { |
| + book.drop(r) |
| + round.Changed++ |
| + round.Deferred++ |
| + round.BytesDeferred += s.Size |
| + continue |
| + } |
| + if err != nil { |
| + book.save(ctx) // best effort: what already landed keeps its receipts |
| return round, fmt.Errorf("mirror: copy %s: %w", sk, err) |
| } |
| round.Changed++ |
| - round.BytesCopied += src[sk].Size |
| + round.BytesCopied += s.Size |
| + if etag != "" && etag != s.ETag { |
| + // 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: etag}); err != nil { |
| + return round, err |
| + } |
| + } else { |
| + book.drop(r) |
| + } |
| } |
| for dk := range dst { |
| r := dk[len(dstPrefix):] |
| @@ -115,14 +226,126 @@ 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 { |
| + book.save(ctx) |
| return round, fmt.Errorf("mirror: delete %s: %w", dk, err) |
| } |
| + book.drop(r) |
| round.Changed++ |
| } |
| } |
| + book.prune(func(r string) bool { _, ok := src[srcPrefix+r]; return ok }) |
| + if err := book.save(ctx); err != nil { |
| + return round, err |
| + } |
| return round, 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. |
| +type receiptLedger struct { |
| + m MirrorStore |
| + rc Receipts |
| + want receiptBook // the pair this round mirrors; Objects unused |
| + loaded bool |
| + book map[string]copyReceipt |
| + dirty bool |
| +} |
| + |
| +func (l *receiptLedger) load(ctx context.Context) error { |
| + if l.loaded || l.rc.Key == "" { |
| + return nil |
| + } |
| + body, err := l.m.Get(ctx, l.rc.Bucket, l.rc.Key) |
| + if err != nil && !errors.Is(err, ErrMirrorNotFound) { |
| + // Not loaded: a failed read must never be saved over as if the |
| + // book were empty. |
| + return fmt.Errorf("mirror: read receipts %s/%s: %w", l.rc.Bucket, l.rc.Key, err) |
| + } |
| + l.loaded = true |
| + l.book = map[string]copyReceipt{} |
| + if err != nil { |
| + return nil // no receipts yet |
| + } |
| + var b receiptBook |
| + if json.Unmarshal(body, &b) != nil || b.V != 1 || b.SrcBucket != l.want.SrcBucket || b.SrcPrefix != l.want.SrcPrefix || |
| + b.DstBucket != l.want.DstBucket || b.DstPrefix != l.want.DstPrefix { |
| + // Unreadable, or written for another pair: no evidence at all. |
| + // Rewritten at the end of the round. |
| + l.dirty = true |
| + return nil |
| + } |
| + for k, v := range b.Objects { |
| + l.book[k] = v |
| + } |
| + return nil |
| +} |
| + |
| +func (l *receiptLedger) get(ctx context.Context, rel string) (copyReceipt, bool, error) { |
| + if err := l.load(ctx); err != nil { |
| + return copyReceipt{}, false, err |
| + } |
| + r, ok := l.book[rel] |
| + return r, ok, nil |
| +} |
| + |
| +// set records a completed copy. It reads the book first if this round |
| +// has not (a copy to a key the destination lacked needs no lookup), so the |
| +// write at the end of the round keeps every other receipt. |
| +func (l *receiptLedger) set(ctx context.Context, rel string, r copyReceipt) error { |
| + if l.rc.Key == "" { |
| + return nil |
| + } |
| + if err := l.load(ctx); err != nil { |
| + return err |
| + } |
| + if old, ok := l.book[rel]; !ok || old != r { |
| + l.book[rel] = r |
| + l.dirty = true |
| + } |
| + return nil |
| +} |
| + |
| +// drop forgets rel's receipt. Only a book already read can hold one. |
| +func (l *receiptLedger) drop(rel string) { |
| + if !l.loaded { |
| + return |
| + } |
| + if _, ok := l.book[rel]; ok { |
| + delete(l.book, rel) |
| + l.dirty = true |
| + } |
| +} |
| + |
| +func (l *receiptLedger) prune(keep func(rel string) bool) { |
| + if !l.loaded { |
| + return |
| + } |
| + for rel := range l.book { |
| + if !keep(rel) { |
| + delete(l.book, rel) |
| + l.dirty = true |
| + } |
| + } |
| +} |
| + |
| +func (l *receiptLedger) save(ctx context.Context) error { |
| + if !l.dirty || l.rc.Key == "" { |
| + return nil |
| + } |
| + b := l.want |
| + b.Objects = l.book |
| + body, err := json.Marshal(b) |
| + if err != nil { |
| + return err |
| + } |
| + if err := l.m.Put(ctx, l.rc.Bucket, l.rc.Key, body); err != nil { |
| + return fmt.Errorf("mirror: write receipts %s/%s: %w", l.rc.Bucket, l.rc.Key, err) |
| + } |
| + l.dirty = false |
| + return nil |
| +} |
| + |
| // leaseLockName is the coordination file this package's own leaser keeps |
| // beside every replica. MirrorPrefix's exclusion of it is lease POLICY, not |
| // generic mirroring — which is why the mirror lives in this package. |
| @@ -188,7 +411,7 @@ func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix str |
| } |
| var obj MirrorObject |
| if o.ETag != nil { |
| - obj.ETag = *o.ETag |
| + obj.ETag = normalizeETag(*o.ETag) |
| } |
| if o.Size != nil { |
| obj.Size = *o.Size |
| @@ -224,15 +447,83 @@ func (m *s3Mirror) List(ctx context.Context, bucket, prefix string) (map[string] |
| return out, nil |
| } |
| -func (m *s3Mirror) Copy(ctx context.Context, srcBucket, srcKey, dstBucket, dstKey string) error { |
| +// normalizeETag strips the quotes S3 wraps an ETag in, so a listing and a |
| +// CopyObject result compare equal when they name the same object: a |
| +// receipt whose quoting differed from the next listing's would never match, |
| +// and the recopy-forever bug would come back silently. |
| +func normalizeETag(et string) string { return strings.Trim(et, `"`) } |
| + |
| +// clientFor is the client List bound bucket to, or fallback if this pair |
| +// has never listed it. |
| +func (m *s3Mirror) clientFor(bucket string, fallback *awss3.Client) *awss3.Client { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + if c := m.byBucket[bucket]; c != nil { |
| + return c |
| + } |
| + return fallback |
| +} |
| + |
| +func (m *s3Mirror) Copy(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) { |
| source := url.PathEscape(srcBucket + "/" + srcKey) |
| - _, err := m.dst.CopyObject(ctx, &awss3.CopyObjectInput{ |
| - Bucket: &dstBucket, Key: &dstKey, CopySource: &source, |
| - }) |
| - return err |
| + in := &awss3.CopyObjectInput{Bucket: &dstBucket, Key: &dstKey, CopySource: &source} |
| + if srcETag != "" { |
| + quoted := `"` + normalizeETag(srcETag) + `"` |
| + in.CopySourceIfMatch = "ed |
| + } |
| + out, err := m.dst.CopyObject(ctx, in) |
| + if err != nil { |
| + var ae smithy.APIError |
| + if srcETag != "" && errors.As(err, &ae) && ae.ErrorCode() == "PreconditionFailed" { |
| + return "", ErrCopySourceChanged |
| + } |
| + return "", err |
| + } |
| + if out.CopyObjectResult == nil || out.CopyObjectResult.ETag == nil { |
| + return "", nil // no receipt possible; the next round recopies |
| + } |
| + return normalizeETag(*out.CopyObjectResult.ETag), nil |
| } |
| +// Delete goes to the client List bound the bucket to: the receipts object |
| +// lives in the SOURCE's bucket, which the destination's client cannot |
| +// reach. A bucket this pair never listed (an abort after an agent restart |
| +// lists only the target) tries the destination client, as before, then |
| +// the source's; if both fail, the destination's error is the one reported. |
| func (m *s3Mirror) Delete(ctx context.Context, bucket, key string) error { |
| - _, err := m.dst.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: &bucket, Key: &key}) |
| + in := &awss3.DeleteObjectInput{Bucket: &bucket, Key: &key} |
| + if c := m.clientFor(bucket, nil); c != nil { |
| + _, err := c.DeleteObject(ctx, in) |
| + return err |
| + } |
| + _, err := m.dst.DeleteObject(ctx, in) |
| + if err != nil && m.src != m.dst { |
| + if _, err2 := m.src.DeleteObject(ctx, in); err2 == nil { |
| + return nil |
| + } |
| + } |
| + return err |
| +} |
| + |
| +// Get and Put serve the receipts object, which lives in the source's own |
| +// bucket: the bound client, else the source's. |
| +func (m *s3Mirror) Get(ctx context.Context, bucket, key string) ([]byte, error) { |
| + out, err := m.clientFor(bucket, m.src).GetObject(ctx, &awss3.GetObjectInput{Bucket: &bucket, Key: &key}) |
| + if err != nil { |
| + var nsk *s3types.NoSuchKey |
| + if errors.As(err, &nsk) { |
| + return nil, ErrMirrorNotFound |
| + } |
| + return nil, err |
| + } |
| + defer out.Body.Close() |
| + return io.ReadAll(out.Body) |
| +} |
| + |
| +func (m *s3Mirror) Put(ctx context.Context, bucket, key string, body []byte) error { |
| + ct := "application/json" |
| + _, err := m.clientFor(bucket, m.src).PutObject(ctx, &awss3.PutObjectInput{ |
| + Bucket: &bucket, Key: &key, Body: bytes.NewReader(body), ContentType: &ct, |
| + }) |
| return err |
| } |
diff --git a/internal/lease/mirror_test.go b/internal/lease/mirror_test.go| index 0d706f9ab..35cff26b0 100644 |
| --- a/internal/lease/mirror_test.go |
| +++ b/internal/lease/mirror_test.go |
| @@ -2,48 +2,85 @@ package lease |
| import ( |
| "context" |
| + "encoding/json" |
| "fmt" |
| "strings" |
| "testing" |
| ) |
| -// fakeMirror is an in-memory MirrorStore: buckets of key→object, plus an |
| -// op log so ordering tests can assert the sequence. |
| +// fakeMirror is an in-memory MirrorStore that keeps S3's ETag rules, plus |
| +// an op log so ordering tests can assert the sequence. Every object carries |
| +// a body (its content's identity). A single-part object's ETag IS its body |
| +// (S3's MD5), a multipart one's is body+"-27", and a copy always lands |
| +// 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. |
| type fakeMirror struct { |
| - buckets map[string]map[string]MirrorObject |
| + 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 |
| } |
| -func newFakeMirror() *fakeMirror { return &fakeMirror{buckets: map[string]map[string]MirrorObject{}} } |
| +type fakeObject struct { |
| + MirrorObject |
| + body string |
| +} |
| + |
| +func newFakeMirror() *fakeMirror { return &fakeMirror{buckets: map[string]map[string]fakeObject{}} } |
| 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. |
| func (f *fakeMirror) putSized(bucket, key, etag string, size int64) { |
| + f.store(bucket, key, fakeObject{MirrorObject{ETag: etag, Size: size}, etag}) |
| +} |
| + |
| +// putMultipart stores a multipart upload of body (litestream's uploader |
| +// for anything over one 5 MiB part). |
| +func (f *fakeMirror) putMultipart(bucket, key, body string, size int64) { |
| + f.store(bucket, key, fakeObject{MirrorObject{ETag: body + "-27", Size: size}, body}) |
| +} |
| + |
| +func (f *fakeMirror) store(bucket, key string, o fakeObject) { |
| if f.buckets[bucket] == nil { |
| - f.buckets[bucket] = map[string]MirrorObject{} |
| + f.buckets[bucket] = map[string]fakeObject{} |
| } |
| - f.buckets[bucket][key] = MirrorObject{ETag: etag, Size: size} |
| + f.buckets[bucket][key] = o |
| } |
| 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 |
| + out[k] = v.MirrorObject |
| } |
| } |
| return out, nil |
| } |
| -func (f *fakeMirror) Copy(_ context.Context, srcBucket, srcKey, dstBucket, dstKey string) error { |
| +func (f *fakeMirror) Copy(_ context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) { |
| if f.failOn == srcKey { |
| - return fmt.Errorf("injected copy failure on %s", srcKey) |
| + return "", fmt.Errorf("injected copy failure on %s", srcKey) |
| + } |
| + if f.beforeCopy != nil { |
| + f.beforeCopy(srcKey) |
| + } |
| + o, ok := f.buckets[srcBucket][srcKey] |
| + if !ok { |
| + return "", fmt.Errorf("no such key %s", srcKey) |
| + } |
| + if srcETag != "" && o.ETag != srcETag { |
| + return "", ErrCopySourceChanged |
| } |
| f.ops = append(f.ops, "copy "+srcKey+" -> "+dstBucket+"/"+dstKey) |
| - o := f.buckets[srcBucket][srcKey] |
| - f.putSized(dstBucket, dstKey, o.ETag, o.Size) |
| - return nil |
| + f.store(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 { |
| @@ -52,6 +89,27 @@ func (f *fakeMirror) Delete(_ context.Context, bucket, key string) error { |
| return nil |
| } |
| +func (f *fakeMirror) Get(_ context.Context, bucket, key string) ([]byte, error) { |
| + if f.failGet { |
| + return nil, fmt.Errorf("injected get failure on %s", key) |
| + } |
| + o, ok := f.buckets[bucket][key] |
| + if !ok { |
| + return nil, ErrMirrorNotFound |
| + } |
| + return []byte(o.body), nil |
| +} |
| + |
| +func (f *fakeMirror) Put(_ context.Context, bucket, key string, body []byte) error { |
| + 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)}) |
| + return nil |
| +} |
| + |
| func TestMirrorPrefix(t *testing.T) { |
| ctx := context.Background() |
| f := newFakeMirror() |
| @@ -59,19 +117,19 @@ func TestMirrorPrefix(t *testing.T) { |
| f.put("src", "instances/h/generations/a/wal/000", "e2") |
| // Initial round: full copy. |
| - r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| + r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}) |
| if err != nil || r.Changed != 2 { |
| t.Fatalf("initial round = %+v, %v; want 2 copies", r, err) |
| } |
| // Identical: zero. |
| - if r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || r.Changed != 0 { |
| + if r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); 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") |
| - r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| + r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}) |
| if err != nil || r.Changed != 3 { |
| t.Fatalf("delta round = %+v, %v; want 2 copies + 1 delete", r, err) |
| } |
| @@ -91,7 +149,7 @@ func TestMirrorPrefix(t *testing.T) { |
| // A copy failure surfaces with the partial count. |
| f.put("src", "instances/h/generations/a/wal/002", "e4") |
| f.failOn = "instances/h/generations/a/wal/002" |
| - if _, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err == nil { |
| + if _, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); err == nil { |
| t.Fatal("injected failure did not surface") |
| } |
| } |
| @@ -104,7 +162,7 @@ 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 |
| - r, err := MirrorPrefix(ctx, f, "src", "instances/h", "dst", "moves/h") |
| + r, err := MirrorPrefix(ctx, f, "src", "instances/h", "dst", "moves/h", Receipts{}) |
| if err != nil || r.Changed != 1 { |
| t.Fatalf("round = %+v, %v; want exactly the one h object", r, err) |
| } |
| @@ -122,11 +180,11 @@ 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 r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || r.Changed != 1 { |
| + if r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); 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 r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || r.Changed != 0 { |
| + if r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); 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/") |
| @@ -145,7 +203,7 @@ func TestMirrorPrefixCountsBytesCopied(t *testing.T) { |
| 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/") |
| + r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}) |
| if err != nil || r.Changed != 2 || r.BytesCopied != 1030 { |
| t.Fatalf("initial round = %+v, %v; want 2 changed, 1030 bytes", r, err) |
| } |
| @@ -157,12 +215,12 @@ func TestMirrorPrefixCountsBytesCopied(t *testing.T) { |
| 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/") |
| + r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}) |
| 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/") |
| + r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}) |
| if err != nil || r.Changed != 0 || r.BytesCopied != 0 { |
| t.Fatalf("caught-up round = %+v, %v; want nothing", r, err) |
| } |
| @@ -171,8 +229,254 @@ func TestMirrorPrefixCountsBytesCopied(t *testing.T) { |
| 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/") |
| + r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}) |
| 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) |
| } |
| } |
| + |
| +var testReceipts = Receipts{Bucket: "src", Key: "move-receipts/h.json"} |
| + |
| +func mirrorRound(t *testing.T, f *fakeMirror, rc Receipts) MirrorRound { |
| + t.Helper() |
| + r, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", rc) |
| + if err != nil { |
| + t.Fatalf("round: %v", err) |
| + } |
| + return r |
| +} |
| + |
| +func readReceipts(t *testing.T, f *fakeMirror) map[string]copyReceipt { |
| + t.Helper() |
| + body, err := f.Get(context.Background(), testReceipts.Bucket, testReceipts.Key) |
| + if err != nil { |
| + t.Fatalf("receipts: %v", err) |
| + } |
| + var b receiptBook |
| + if err := json.Unmarshal(body, &b); err != nil { |
| + t.Fatalf("receipts: %v", err) |
| + } |
| + return b.Objects |
| +} |
| + |
| +// #175 as production saw it: a multipart source's copy lands with a |
| +// different ETag, so ETag equality alone recopies every multipart object |
| +// on every round. Receipts let an unchanged one be copied once. |
| +func TestMirrorPrefixReceiptsSkipAnUnchangedMultipartObject(t *testing.T) { |
| + const big = 138 << 20 |
| + fill := func(f *fakeMirror) { |
| + f.putMultipart("src", "instances/h/ltx/1/a.ltx", "level1", big) |
| + f.putSized("src", "instances/h/ltx/0/b.ltx", "l0", 4096) |
| + } |
| + |
| + // Without receipts: the bug, kept visible. |
| + f := newFakeMirror() |
| + fill(f) |
| + mirrorRound(t, f, Receipts{}) |
| + if r := mirrorRound(t, f, Receipts{}); r.Changed != 1 || r.BytesCopied != big { |
| + t.Fatalf("without receipts the unchanged multipart object should recopy: %+v", r) |
| + } |
| + |
| + f = newFakeMirror() |
| + fill(f) |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 2 || r.BytesCopied != big+4096 { |
| + t.Fatalf("first round = %+v, want both copied", r) |
| + } |
| + if got := readReceipts(t, f); len(got) != 1 || got["ltx/1/a.ltx"] != (copyReceipt{SrcETag: "level1-27", SrcSize: big, DstETag: "level1"}) { |
| + t.Fatalf("receipts = %+v, want exactly the multipart object's", got) |
| + } |
| + puts := f.puts |
| + for i := 0; i < 3; i++ { |
| + if r := mirrorRound(t, f, testReceipts); r != (MirrorRound{}) { |
| + t.Fatalf("unchanged round %d = %+v, want nothing copied", i, r) |
| + } |
| + } |
| + if f.puts != puts { |
| + t.Fatalf("receipts rewritten on rounds that changed nothing (%d puts, want %d)", f.puts, puts) |
| + } |
| + // Receipts are never data: nothing beside the two objects is staged. |
| + if staged, _ := f.List(context.Background(), "dst", ""); len(staged) != 2 { |
| + t.Fatalf("staging = %v, want only the two data objects", staged) |
| + } |
| +} |
| + |
| +// Single-part sources keep the plain ETag skip: no receipts are written, |
| +// so the existing behaviour and request count do not change. |
| +func TestMirrorPrefixSinglePartWritesNoReceipts(t *testing.T) { |
| + f := newFakeMirror() |
| + f.putSized("src", "instances/h/ltx/0/a.ltx", "a", 100) |
| + mirrorRound(t, f, testReceipts) |
| + f.putSized("src", "instances/h/ltx/0/b.ltx", "b", 100) |
| + mirrorRound(t, f, testReceipts) |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 0 { |
| + t.Fatalf("caught-up round = %+v", r) |
| + } |
| + if f.puts != 0 { |
| + t.Fatalf("%d receipt writes for single-part objects, want none", f.puts) |
| + } |
| +} |
| + |
| +func TestMirrorPrefixReceiptRecopiesWhenTheSourceChanges(t *testing.T) { |
| + f := newFakeMirror() |
| + f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20) |
| + mirrorRound(t, f, testReceipts) |
| + // Compaction rewrote it under the same name. |
| + f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v2", 101<<20) |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 101<<20 { |
| + t.Fatalf("rewritten source = %+v, want it recopied", r) |
| + } |
| + if got := f.buckets["dst"]["moves/h/ltx/1/a.ltx"].body; got != "v2" { |
| + t.Fatalf("staged body = %q, want v2", got) |
| + } |
| + // The same size with different content is still a change: never |
| + // skip on size alone. |
| + f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v3", 101<<20) |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 1 { |
| + t.Fatalf("same-size rewrite = %+v, want it recopied", r) |
| + } |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 0 { |
| + t.Fatalf("after the recopy = %+v, want the new receipt to hold", r) |
| + } |
| +} |
| + |
| +func TestMirrorPrefixReceiptRecopiesWhenTheDestinationChanges(t *testing.T) { |
| + for _, tc := range []struct { |
| + name string |
| + touch func(f *fakeMirror) |
| + }{ |
| + {"overwritten", func(f *fakeMirror) { f.putSized("dst", "moves/h/ltx/1/a.ltx", "intruder", 100<<20) }}, |
| + {"deleted", func(f *fakeMirror) { delete(f.buckets["dst"], "moves/h/ltx/1/a.ltx") }}, |
| + } { |
| + t.Run(tc.name, func(t *testing.T) { |
| + f := newFakeMirror() |
| + f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20) |
| + mirrorRound(t, f, testReceipts) |
| + tc.touch(f) |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 100<<20 { |
| + t.Fatalf("destination %s behind the receipt = %+v, want a recopy", tc.name, r) |
| + } |
| + if got := f.buckets["dst"]["moves/h/ltx/1/a.ltx"].body; got != "v1" { |
| + t.Fatalf("staged body = %q, want v1", got) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// A crash between the copy and the receipt write costs one extra copy on |
| +// the next round, never a wrong skip. |
| +func TestMirrorPrefixLostReceiptWriteRecopiesOnce(t *testing.T) { |
| + f := newFakeMirror() |
| + f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20) |
| + f.failPut = true |
| + r, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", testReceipts) |
| + if err == nil || r.BytesCopied != 100<<20 { |
| + t.Fatalf("round with a failed receipt write = %+v, %v; want the copy reported and the error surfaced", r, err) |
| + } |
| + f.failPut = false |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 100<<20 { |
| + t.Fatalf("next round = %+v, want exactly one recopy", r) |
| + } |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 0 { |
| + t.Fatalf("round after = %+v, want the receipt to hold", r) |
| + } |
| +} |
| + |
| +// A source overwritten between the listing and the copy fails the |
| +// conditional copy. The round records no receipt for it and carries on; |
| +// the next round copies the new version. |
| +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) |
| + f.beforeCopy = func(k string) { |
| + if k == "instances/h/ltx/1/a.ltx" { |
| + f.putMultipart("src", k, "v2", 120<<20) |
| + f.beforeCopy = nil |
| + } |
| + } |
| + r := mirrorRound(t, f, testReceipts) |
| + if r.Changed != 2 || r.Deferred != 1 || r.BytesDeferred != 100<<20 || r.BytesCopied != 10 { |
| + t.Fatalf("round = %+v, want b copied and a deferred at its listed size", r) |
| + } |
| + if r.DeltaBytes() != 100<<20+10 { |
| + t.Fatalf("DeltaBytes = %d", r.DeltaBytes()) |
| + } |
| + if _, ok := f.buckets["dst"]["moves/h/ltx/1/a.ltx"]; ok { |
| + t.Fatal("a deferred copy landed anyway") |
| + } |
| + if _, err := f.Get(context.Background(), testReceipts.Bucket, testReceipts.Key); err == nil { |
| + if _, ok := readReceipts(t, f)["ltx/1/a.ltx"]; ok { |
| + t.Fatal("a receipt was recorded for a copy that failed its precondition") |
| + } |
| + } |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 120<<20 || r.Deferred != 0 { |
| + t.Fatalf("next round = %+v, want the new version copied", r) |
| + } |
| + if got := f.buckets["dst"]["moves/h/ltx/1/a.ltx"].body; got != "v2" { |
| + t.Fatalf("staged body = %q, want v2", got) |
| + } |
| +} |
| + |
| +// Receipts name the pair they were written for. Another pair's receipts, |
| +// even with matching keys and ETags, are not evidence about this one. |
| +func TestMirrorPrefixIgnoresAnotherPairsReceipts(t *testing.T) { |
| + ctx := context.Background() |
| + f := newFakeMirror() |
| + f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20) |
| + mirrorRound(t, f, testReceipts) |
| + // Another destination that happens to hold a lookalike object. |
| + f.putSized("dst", "other/h/ltx/1/a.ltx", "v1", 100<<20) |
| + r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "other/h/", testReceipts) |
| + if err != nil || r.Changed != 1 { |
| + t.Fatalf("round toward another destination = %+v, %v; want a copy", r, err) |
| + } |
| +} |
| + |
| +// Unreadable receipts are no receipts: everything recopies once, and the |
| +// book is rewritten. |
| +func TestMirrorPrefixCorruptReceiptsRecopy(t *testing.T) { |
| + f := newFakeMirror() |
| + f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20) |
| + mirrorRound(t, f, testReceipts) |
| + f.store(testReceipts.Bucket, testReceipts.Key, fakeObject{body: "{not json"}) |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 1 { |
| + t.Fatalf("round over corrupt receipts = %+v, want a recopy", r) |
| + } |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 0 { |
| + t.Fatalf("round after = %+v, want the rewritten receipts to hold", r) |
| + } |
| +} |
| + |
| +// A source object retention pruned takes its receipt with it. |
| +func TestMirrorPrefixPrunesReceiptsWithTheirObjects(t *testing.T) { |
| + f := newFakeMirror() |
| + f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20) |
| + f.putMultipart("src", "instances/h/ltx/1/b.ltx", "w1", 100<<20) |
| + mirrorRound(t, f, testReceipts) |
| + delete(f.buckets["src"], "instances/h/ltx/1/a.ltx") |
| + if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 0 { |
| + t.Fatalf("prune round = %+v, want one delete", r) |
| + } |
| + if got := readReceipts(t, f); len(got) != 1 || got["ltx/1/b.ltx"].SrcETag != "w1-27" { |
| + t.Fatalf("receipts = %+v, want only b's", got) |
| + } |
| +} |
| + |
| +// Receipts that cannot be READ (a failure that is not "missing") fail the |
| +// round: guessing "no receipts" would quietly bring back the recopy loop, |
| +// and saving over a book this round never read would destroy it. |
| +func TestMirrorPrefixReceiptReadFailureFailsTheRound(t *testing.T) { |
| + f := newFakeMirror() |
| + f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20) |
| + mirrorRound(t, f, testReceipts) |
| + before := readReceipts(t, f) |
| + f.putMultipart("src", "instances/h/ltx/1/b.ltx", "w1", 100<<20) |
| + f.failGet = true |
| + if _, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", testReceipts); err == nil { |
| + t.Fatal("an unreadable receipts object did not fail the round") |
| + } |
| + f.failGet = false |
| + if got := readReceipts(t, f); len(got) != len(before) || got["ltx/1/a.ltx"] != before["ltx/1/a.ltx"] { |
| + t.Fatalf("receipts after a failed read = %+v, want them untouched (%+v)", got, before) |
| + } |
| +} |