amadan
Merge branch 'move-ready-delta-threshold' into main
6 files changed,
+1379
−111
internal/activator/backend_exec.go+21 −4internal/activator/backend_exec_test.go+79 −9internal/host/movespass.go+86 −7internal/host/movespass_test.go+456 −19internal/lease/mirror.go+355 −39internal/lease/mirror_test.go+382 −33
CI passed — run details
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index 615435540..114ee0987 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -284,7 +284,7 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, err |
| // and back must restore its own lineage, never regress to an old |
| // staged copy (the source's cut deleted the own prefix it left behind, |
| // so a non-empty own prefix is by construction the newer story). |
| - var promoted map[string]string |
| + var promoted map[string]lease.MirrorObject |
| if b.Mirror != nil { |
| var err error |
| if promoted, err = b.promoteStagedReplica(ctx, host); err != nil { |
| @@ -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 |
| @@ -1569,7 +1577,7 @@ func PurgeLocalDBArtifacts(dbPath string, full bool) { |
| // promoteStagedReplica copies moves/<host>/ over instances/<host>/ in this |
| // box's OWN bucket when — and only when — the ordinary prefix is empty. A |
| // no-op (nil error) when there is nothing staged. |
| -func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (map[string]string, error) { |
| +func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (map[string]lease.MirrorObject, error) { |
| // STAGING WINS when present (review round 1). A staged replica exists |
| // only while an inbound move is landing or has just landed: it was |
| // mirrored from the then-live data at cut, so it is by construction |
| @@ -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 |
| } |
| @@ -1600,7 +1617,7 @@ func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (ma |
| // third listing of a prefix this restore has already walked twice. |
| // Best-effort: a leftover staging copy is storage, not a correctness |
| // problem. |
| -func (b *ExecBackend) dropStagedReplica(ctx context.Context, staged map[string]string) { |
| +func (b *ExecBackend) dropStagedReplica(ctx context.Context, staged map[string]lease.MirrorObject) { |
| for k := range staged { |
| _ = b.Mirror.Delete(ctx, b.Cfg.Bucket, k) |
| } |
diff --git a/internal/activator/backend_exec_test.go b/internal/activator/backend_exec_test.go| index 84f291c12..8b4c41ceb 100644 |
| --- a/internal/activator/backend_exec_test.go |
| +++ b/internal/activator/backend_exec_test.go |
| @@ -13,6 +13,7 @@ import ( |
| _ "modernc.org/sqlite" |
| "github.com/carlosframework/platform/internal/cgroup" |
| + "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/registry" |
| ) |
| @@ -197,26 +198,78 @@ func TestRestoreStagingFallback(t *testing.T) { |
| f.failList = false |
| } |
| -// fakeMirrorStore is the test double for lease.MirrorStore, bucket→key→etag. |
| +// 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]string |
| + buckets map[string]map[string]lease.MirrorObject |
| failList bool |
| } |
| func newFakeMirrorStore() *fakeMirrorStore { |
| - return &fakeMirrorStore{buckets: map[string]map[string]string{}} |
| + return &fakeMirrorStore{buckets: map[string]map[string]lease.MirrorObject{}} |
| } |
| func (f *fakeMirrorStore) put(bucket, key, etag string) { |
| if f.buckets[bucket] == nil { |
| - f.buckets[bucket] = map[string]string{} |
| + f.buckets[bucket] = map[string]lease.MirrorObject{} |
| } |
| - f.buckets[bucket][key] = etag |
| + f.buckets[bucket][key] = lease.MirrorObject{ETag: etag} |
| } |
| -func (f *fakeMirrorStore) List(_ context.Context, bucket, prefix string) (map[string]string, error) { |
| +func (f *fakeMirrorStore) List(_ context.Context, bucket, prefix string) (map[string]lease.MirrorObject, error) { |
| if f.failList { |
| return nil, context.DeadlineExceeded |
| } |
| - out := map[string]string{} |
| + out := map[string]lease.MirrorObject{} |
| for k, v := range f.buckets[bucket] { |
| if strings.HasPrefix(k, prefix) { |
| out[k] = v |
| @@ -224,8 +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 { |
| - f.put(dstBucket, dstKey, f.buckets[srcBucket][srcKey]) |
| +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] = 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 6ec19b166..f64fe95ed 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -9,6 +9,7 @@ import ( |
| "log" |
| "os" |
| "path/filepath" |
| + "strconv" |
| "strings" |
| "sync" |
| "sync/atomic" |
| @@ -30,7 +31,14 @@ import ( |
| // same trigger as the reconcile pass (serial bumps land in seconds via |
| // 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. |
| +// every step derives from the record and the buckets. "Converged" is a |
| +// 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: |
| // |
| @@ -89,7 +97,10 @@ func (w *movesWorker) mirror(ctx context.Context, dstRegion string) (lease.Mirro |
| // the service that answers whether a child is down. Called once, before |
| // svc.Start. |
| func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service) { |
| - |
| + readyDelta, ok := moveReadyDeltaFromEnv(os.Getenv) |
| + if !ok { |
| + log.Printf("agent: %s=%q ignored: want a whole number of bytes, 0 or more — keeping the built-in default of %d", moveReadyDeltaEnv, os.Getenv(moveReadyDeltaEnv), defaultMoveReadyDelta) |
| + } |
| cfg := moveCfg{ |
| Region: strings.TrimSpace(os.Getenv("CARLOS_REGION")), |
| OwnBucket: hib.Bucket, |
| @@ -128,7 +139,8 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, |
| defer rep.Close(ctx) |
| return rep.SyncAndWait(ctx) |
| }, |
| - Now: time.Now, |
| + Now: time.Now, |
| + ReadyDelta: readyDelta, |
| } |
| armedMovesWorker.Store(&movesWorker{cfg: cfg, mirrors: map[string]lease.MirrorStore{}}) |
| } |
| @@ -184,6 +196,42 @@ 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 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 |
| +} |
| + |
| +// defaultMoveReadyDelta is the readiness bar when |
| +// CARLOS_MOVE_READY_DELTA_BYTES is unset or rejected. A busy instance's |
| +// per-round churn of new L0 files sits well under 16 MiB, so it reaches |
| +// ready; a round in which compaction rewrote a large level file (138 MB |
| +// for the 293 MB database in #175) does not, and waits for a quieter one. |
| +// Ready keeps mirroring every pass until the cut, so the bar roughly |
| +// bounds what the cut's final round, run while the instance is down, has |
| +// left to copy. |
| +const defaultMoveReadyDelta int64 = 16 << 20 |
| + |
| +// moveReadyDeltaEnv overrides defaultMoveReadyDelta with a whole number of |
| +// 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 |
| +// negative value keeps the default, so an operator's typo never stops the |
| +// agent; ok is false only when a value was SET and rejected, so the caller |
| +// can say so instead of leaving the operator to believe it took. |
| +func moveReadyDeltaFromEnv(getenv func(string) string) (delta int64, ok bool) { |
| + v := strings.TrimSpace(getenv(moveReadyDeltaEnv)) |
| + if v == "" { |
| + return defaultMoveReadyDelta, true |
| + } |
| + n, err := strconv.ParseInt(v, 10, 64) |
| + if err != nil || n < 0 { |
| + return defaultMoveReadyDelta, false |
| + } |
| + return n, true |
| } |
| // movesPass runs one round over every in-flight move this box owns. |
| @@ -315,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: |
| @@ -322,16 +374,22 @@ func runMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec prov |
| if err != nil { |
| return err |
| } |
| - changed, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix) |
| + round, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts) |
| if err != nil { |
| return err |
| } |
| - if rec.Move.State == provision.MovePreparing && changed == 0 { |
| + // Ready means the gap is small, not that it is closed (#175): an |
| + // awake instance writes new litestream files between any two |
| + // listings, so a round that copied NOTHING never comes. The cut |
| + // 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.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 — ready to cut", rec.Host, tgt.DataBucket, dstPrefix) |
| + 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 |
| @@ -349,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 |
| @@ -418,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. |
| @@ -437,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 9fca637a1..5f8ed6dd8 100644 |
| --- a/internal/host/movespass_test.go |
| +++ b/internal/host/movespass_test.go |
| @@ -3,52 +3,101 @@ package host |
| import ( |
| "context" |
| "errors" |
| + "fmt" |
| "os" |
| "path/filepath" |
| "strings" |
| "testing" |
| "time" |
| + "github.com/carlosframework/platform/internal/activator" |
| "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/provision" |
| "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| ) |
| -// fakeMoveMirror: bucket→key→etag 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]string |
| - 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]string{}} |
| + 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) put(bucket, key, etag string) { |
| +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]string{} |
| + f.buckets[bucket] = map[string]fakeMoveObject{} |
| } |
| - f.buckets[bucket][key] = etag |
| + f.buckets[bucket][key] = o |
| } |
| -func (f *fakeMoveMirror) List(_ context.Context, bucket, prefix string) (map[string]string, error) { |
| - out := map[string]string{} |
| +func (f *fakeMoveMirror) List(_ context.Context, bucket, prefix string) (map[string]lease.MirrorObject, error) { |
| + out := map[string]lease.MirrorObject{} |
| for k, v := range f.buckets[bucket] { |
| if strings.HasPrefix(k, prefix) { |
| - out[k] = v |
| + 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) |
| - f.put(dstBucket, dstKey, f.buckets[srcBucket][srcKey]) |
| - 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 |
| @@ -78,7 +127,8 @@ func newMoveHarness(t *testing.T) *moveHarness { |
| h.f.ops = append(h.f.ops, "checkpoint "+filepath.Base(dbPath)) |
| return nil |
| }, |
| - Now: time.Now, |
| + Now: time.Now, |
| + ReadyDelta: defaultMoveReadyDelta, // what armMovesPass sets with the env unset |
| } |
| // The target's edges file: uk publishes a data bucket. |
| if _, err := h.s.Put(context.Background(), "control/edges/uk-box.json", |
| @@ -127,12 +177,12 @@ func TestMovesPassPrepareAndPromote(t *testing.T) { |
| h := newMoveHarness(t) |
| host := "a.go.example.com" |
| h.record(t, host, provision.MoveRequest("uk", time.Now())) |
| - h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1") |
| + h.f.putSized("data-ie", "instances/"+host+"/generations/g/snapshot", "e1", 64<<20) |
| h.f.put("data-ie", "instances/"+host+"/generations/g/wal/000", "e2") |
| - // Round 1: copies happen; the SAME pass does not promote (the round |
| - // that copied cannot also claim caught-up — a write may have landed |
| - // mid-copy; only a round that finds NOTHING to do promotes). |
| + // Round 1: copies happen; the SAME pass does not promote, because the |
| + // round copied more than the readiness bar (the 64 MiB snapshot): the |
| + // cut's own round could have that much again to do. |
| if err := movesPass(context.Background(), h.s, h.worker()); err != nil { |
| t.Fatal(err) |
| } |
| @@ -143,7 +193,7 @@ func TestMovesPassPrepareAndPromote(t *testing.T) { |
| if len(staged) != 2 { |
| t.Fatalf("staging = %v", staged) |
| } |
| - // Round 2: nothing to do → ready. |
| + // Round 2: nothing to do → ready (zero bytes is under any bar). |
| if err := movesPass(context.Background(), h.s, h.worker()); err != nil { |
| t.Fatal(err) |
| } |
| @@ -476,6 +526,366 @@ func TestMovesPassCutStopsWhenTheFenceFails(t *testing.T) { |
| } |
| } |
| +// moveRound runs one pass and returns the record's move state after it. |
| +func (h *moveHarness) moveRound(t *testing.T, host string) string { |
| + t.Helper() |
| + if err := movesPass(context.Background(), h.s, h.worker()); err != nil { |
| + t.Fatal(err) |
| + } |
| + got := h.read(t, host) |
| + if got.Move == nil { |
| + t.Fatalf("move cleared unexpectedly: %+v", got) |
| + } |
| + return got.Move.State |
| +} |
| + |
| +// An AWAKE instance never stops writing: litestream lands a new L0 file |
| +// between every two listings, so no round ever copies nothing. Production |
| +// (#175, kpmg.bdn.oncarlos.com) sat in preparing for 80 minutes that way. |
| +// The bar is bytes, not zero: once the initial bulk has moved, a round |
| +// that copied only the small new files is ready. |
| +func TestMovesPassBusySourceReachesReady(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "a.go.example.com" |
| + src := "instances/" + host + "/generations/g/" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now())) |
| + h.f.putSized("data-ie", src+"snapshots/0", "s0", 293<<20) |
| + h.f.putSized("data-ie", "instances/"+host+"/lock.json", "renewal-0", 200) |
| + |
| + // Round 1 moves the bulk: well over the bar, still preparing. |
| + if st := h.moveRound(t, host); st != provision.MovePreparing { |
| + t.Fatalf("after the bulk round: %s, want preparing", st) |
| + } |
| + // Every later round finds a fresh L0 file and a lease renewal: never a |
| + // zero-change round, always a small one. |
| + st := "" |
| + for i := 1; i <= 3 && st != provision.MoveReady; i++ { |
| + h.f.putSized("data-ie", src+fmt.Sprint("ltx/0/", i), fmt.Sprint("l0-", i), 8<<10) |
| + h.f.putSized("data-ie", "instances/"+host+"/lock.json", fmt.Sprint("renewal-", i), 200) |
| + st = h.moveRound(t, host) |
| + } |
| + if st != provision.MoveReady { |
| + t.Fatalf("a busy source with small deltas never reached ready (state %s)", st) |
| + } |
| +} |
| + |
| +// A round that copied more than the bar does not promote, however many |
| +// rounds it takes: compaction rewriting a 138 MiB level file every round |
| +// (the #175 shape) keeps the move preparing. The bar is inclusive. |
| +func TestMovesPassLargeDeltaDoesNotPromote(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "a.go.example.com" |
| + src := "instances/" + host + "/generations/g/" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now())) |
| + h.f.putSized("data-ie", src+"snapshots/0", "s0", 293<<20) |
| + for i := 0; i < 4; i++ { |
| + // Compaction rewrites the level file under a new ETag each round. |
| + h.f.putSized("data-ie", src+"ltx/1/compacted", fmt.Sprint("c", i), 138<<20) |
| + if st := h.moveRound(t, host); st != provision.MovePreparing { |
| + t.Fatalf("round %d copied 138 MiB and promoted to %s", i, st) |
| + } |
| + } |
| + |
| + // The boundary, with a small bar: one byte over stays preparing, the |
| + // bar exactly promotes. |
| + h.cfg.ReadyDelta = 1000 |
| + h.f.putSized("data-ie", src+"ltx/0/a", "a", 1001) |
| + if st := h.moveRound(t, host); st != provision.MovePreparing { |
| + t.Fatalf("a 1001-byte round under a 1000-byte bar promoted to %s", st) |
| + } |
| + h.f.putSized("data-ie", src+"ltx/0/b", "b", 1000) |
| + if st := h.moveRound(t, host); st != provision.MoveReady { |
| + t.Fatalf("a 1000-byte round under a 1000-byte bar left the move %s", st) |
| + } |
| +} |
| + |
| +// 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" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now())) |
| + h.f.putSized("data-ie", "instances/"+host+"/generations/g/wal/000", "e1", 1) |
| + if st := h.moveRound(t, host); st != provision.MovePreparing { |
| + t.Fatalf("a 1-byte round under a zero bar promoted to %s", st) |
| + } |
| + 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 |
| +// small. The cut's final round still has to make staging exact: what was |
| +// written after the last ready round, what compaction rewrote, and what |
| +// retention pruned all reach the target, and the lease never does. |
| +func TestMovesPassCutAfterBusyReadyLandsByteExact(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "a.go.example.com" |
| + own := "instances/" + host + "/" |
| + gen := own + "generations/g/" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now())) |
| + h.f.putSized("data-ie", gen+"snapshots/0", "s0", 293<<20) |
| + h.f.putSized("data-ie", gen+"ltx/1/compacted", "c0", 138<<20) |
| + h.f.putSized("data-ie", gen+"ltx/0/0", "l0-0", 4<<10) |
| + h.f.putSized("data-ie", own+"lock.json", "renewal-0", 200) |
| + if st := h.moveRound(t, host); st != provision.MovePreparing { |
| + t.Fatalf("after the bulk round: %s", st) |
| + } |
| + h.f.putSized("data-ie", gen+"ltx/0/1", "l0-1", 4<<10) |
| + if st := h.moveRound(t, host); st != provision.MoveReady { |
| + t.Fatalf("a 4 KiB round left the move %s, want ready", st) |
| + } |
| + staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/") |
| + if _, ok := staged["moves/"+host+"/generations/g/ltx/0/1"]; !ok { |
| + t.Fatalf("the ready round's own delta was not staged: %v", staged) |
| + } |
| + |
| + // After ready and before the cut, the instance keeps writing: a new L0 |
| + // file, a compaction rewrite, retention pruning an old L0 file, and a |
| + // lease renewal. |
| + h.f.putSized("data-ie", gen+"ltx/0/2", "l0-2", 4<<10) |
| + h.f.putSized("data-ie", gen+"ltx/1/compacted", "c1", 139<<20) |
| + delete(h.f.buckets["data-ie"], gen+"ltx/0/0") |
| + h.f.putSized("data-ie", own+"lock.json", "renewal-1", 200) |
| + if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, provision.MoveMarkCutting()); err != nil { |
| + t.Fatal(err) |
| + } |
| + h.quiet = true |
| + 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) |
| +} |
| + |
| +// 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:]) |
| + } |
| + } |
| + // 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 := 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 |
| + want int64 |
| + ok bool |
| + }{ |
| + {"", defaultMoveReadyDelta, true}, // unset: the default, nothing to report |
| + {"0", 0, true}, // zero copied bytes: only a round that moved no data promotes |
| + {"1048576", 1 << 20, true}, |
| + {" 67108864 ", 64 << 20, true}, |
| + {"-1", defaultMoveReadyDelta, false}, |
| + {"16MiB", defaultMoveReadyDelta, false}, |
| + {"1.5", defaultMoveReadyDelta, false}, |
| + {"99999999999999999999", defaultMoveReadyDelta, false}, |
| + } { |
| + got, ok := moveReadyDeltaFromEnv(func(k string) string { |
| + if k != "CARLOS_MOVE_READY_DELTA_BYTES" { |
| + t.Fatalf("read %s, want CARLOS_MOVE_READY_DELTA_BYTES", k) |
| + } |
| + return tc.raw |
| + }) |
| + if got != tc.want || ok != tc.ok { |
| + t.Errorf("%q: got %d, %v; want %d, %v", tc.raw, got, ok, tc.want, tc.ok) |
| + } |
| + } |
| + if defaultMoveReadyDelta != 16<<20 { |
| + t.Fatalf("default bar = %d, want 16 MiB", defaultMoveReadyDelta) |
| + } |
| +} |
| + |
| +// armMovesPass is where the bar reaches the worker: the env override when |
| +// it parses, the default when it does not. |
| +func TestArmMovesPassReadsTheReadyDelta(t *testing.T) { |
| + prev := armedMovesWorker.Load() |
| + t.Cleanup(func() { armedMovesWorker.Store(prev) }) |
| + for _, tc := range []struct { |
| + raw string |
| + want int64 |
| + }{ |
| + {"", defaultMoveReadyDelta}, |
| + {"4096", 4096}, |
| + {"-5", defaultMoveReadyDelta}, |
| + {"lots", defaultMoveReadyDelta}, |
| + } { |
| + t.Setenv("CARLOS_MOVE_READY_DELTA_BYTES", tc.raw) |
| + armMovesPass(activator.Config{}, t.TempDir(), nil, nil) |
| + if got := armedMovesWorker.Load().cfg.ReadyDelta; got != tc.want { |
| + t.Errorf("CARLOS_MOVE_READY_DELTA_BYTES=%q armed a bar of %d, want %d", tc.raw, got, tc.want) |
| + } |
| + } |
| +} |
| + |
| func writeFileForTest(path, content string) error { return os.WriteFile(path, []byte(content), 0o600) } |
| func fileExistsForTest(path string) bool { _, err := os.Stat(path); return err == nil } |
| @@ -533,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 2061a4c4e..cb6ed72b0 100644 |
| --- a/internal/lease/mirror.go |
| +++ b/internal/lease/mirror.go |
| @@ -1,46 +1,148 @@ |
| 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 |
| // copies the instance's REPLICA prefix — the litestream generations the |
| // activator already maintains — never the database itself. MirrorPrefix is |
| // the one delta round; the move worker calls it repeatedly until a round |
| -// reports nothing changed, and once more after the child is down. |
| +// copies little enough that the cut's own round is cheap, and once more |
| +// after the child is down. |
| // |
| // The interface exists for the tests (an in-memory fake) and because the |
| // destination bucket lives in ANOTHER region: the implementation holds one |
| // client per side. |
| -// MirrorStore is the three S3 actions a delta round needs. List answers |
| -// key→ETag under a prefix (keys RELATIVE to the bucket root, not the |
| -// prefix); Copy is a server-side CopyObject; Delete removes one key. |
| +// 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, 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]string, error) |
| - Copy(ctx context.Context, srcBucket, srcKey, dstBucket, dstKey string) error |
| + List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, 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 { |
| + ETag string |
| + Size int64 |
| +} |
| + |
| +// MirrorRound is what one delta round did. |
| +type MirrorRound struct { |
| + // 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. |
| + 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 how many objects changed; 0 means the sides were |
| -// identical when listed — the move worker's "caught up". |
| +// 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) (int, 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 |
| @@ -54,37 +156,68 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB |
| } |
| src, err := m.List(ctx, srcBucket, srcPrefix) |
| if err != nil { |
| - return 0, fmt.Errorf("mirror: list source %s/%s: %w", srcBucket, srcPrefix, err) |
| + return MirrorRound{}, fmt.Errorf("mirror: list source %s/%s: %w", srcBucket, srcPrefix, err) |
| } |
| dst, err := m.List(ctx, dstBucket, dstPrefix) |
| if err != nil { |
| - return 0, fmt.Errorf("mirror: list staging %s/%s: %w", dstBucket, dstPrefix, err) |
| + return MirrorRound{}, fmt.Errorf("mirror: list staging %s/%s: %w", dstBucket, dstPrefix, err) |
| } |
| + 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 { |
| rel = append(rel, k[len(srcPrefix):]) |
| } |
| sort.Strings(rel) |
| - changed := 0 |
| + var round MirrorRound |
| for _, r := range rel { |
| // The lease's lock.json is COORDINATION state, not data (live |
| // rehearsal catch, 2026-08-29): the renewer rewrites it on every |
| - // TTL renewal, so mirroring it makes "changed==0" unreachable for |
| - // any awake instance — and promoting it would hand the target a |
| + // TTL renewal, so mirroring it would put a change in every round |
| + // for any awake instance — and promoting it would hand the target a |
| // stale foreign lease. Skipped on both sides: never copied, and |
| // the destination's own copy (if any) is never deleted. |
| if r == leaseLockName { |
| continue |
| } |
| sk, dk := srcPrefix+r, dstPrefix+r |
| - if et, ok := dst[dk]; ok && et == src[sk] { |
| + 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 { |
| - return changed, fmt.Errorf("mirror: copy %s: %w", sk, err) |
| + 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 += 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) |
| } |
| - changed++ |
| } |
| for dk := range dst { |
| r := dk[len(dstPrefix):] |
| @@ -93,12 +226,124 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB |
| } |
| if _, ok := src[srcPrefix+r]; !ok { |
| if err := m.Delete(ctx, dstBucket, dk); err != nil { |
| - return changed, fmt.Errorf("mirror: delete %s: %w", dk, err) |
| + book.save(ctx) |
| + return round, fmt.Errorf("mirror: delete %s: %w", dk, err) |
| } |
| - changed++ |
| + book.drop(r) |
| + round.Changed++ |
| } |
| } |
| - return changed, nil |
| + 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 |
| @@ -152,8 +397,8 @@ func NewS3Mirror(ctx context.Context, srcRegion, dstRegion string) (MirrorStore, |
| return &s3Mirror{src: src, dst: dst, byBucket: map[string]*awss3.Client{}}, nil |
| } |
| -func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix string) (map[string]string, error) { |
| - out := map[string]string{} |
| +func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix string) (map[string]MirrorObject, error) { |
| + out := map[string]MirrorObject{} |
| pager := awss3.NewListObjectsV2Paginator(c, &awss3.ListObjectsV2Input{Bucket: &bucket, Prefix: &prefix}) |
| for pager.HasMorePages() { |
| resp, err := pager.NextPage(ctx) |
| @@ -164,11 +409,14 @@ func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix str |
| if o.Key == nil { |
| continue |
| } |
| - et := "" |
| + var obj MirrorObject |
| if o.ETag != nil { |
| - et = *o.ETag |
| + obj.ETag = normalizeETag(*o.ETag) |
| } |
| - out[*o.Key] = et |
| + if o.Size != nil { |
| + obj.Size = *o.Size |
| + } |
| + out[*o.Key] = obj |
| } |
| } |
| return out, nil |
| @@ -178,7 +426,7 @@ func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix str |
| // never says which region a bucket lives in, so the FIRST touch of a bucket |
| // probes source-then-destination (a wrong-region List is an error, so the |
| // probe is unambiguous) and the winner is remembered. |
| -func (m *s3Mirror) List(ctx context.Context, bucket, prefix string) (map[string]string, error) { |
| +func (m *s3Mirror) List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error) { |
| m.mu.Lock() |
| c := m.byBucket[bucket] |
| m.mu.Unlock() |
| @@ -199,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 124b3b896..35cff26b0 100644 |
| --- a/internal/lease/mirror_test.go |
| +++ b/internal/lease/mirror_test.go |
| @@ -2,45 +2,85 @@ package lease |
| import ( |
| "context" |
| + "encoding/json" |
| "fmt" |
| "strings" |
| "testing" |
| ) |
| -// fakeMirror is an in-memory MirrorStore: buckets of key→etag, plus an op |
| -// log so ordering tests can assert the sequence. |
| +// fakeMirror is an in-memory MirrorStore 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]string |
| + 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]string{}} } |
| +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) put(bucket, key, etag string) { |
| +func (f *fakeMirror) store(bucket, key string, o fakeObject) { |
| if f.buckets[bucket] == nil { |
| - f.buckets[bucket] = map[string]string{} |
| + f.buckets[bucket] = map[string]fakeObject{} |
| } |
| - f.buckets[bucket][key] = etag |
| + f.buckets[bucket][key] = o |
| } |
| -func (f *fakeMirror) List(_ context.Context, bucket, prefix string) (map[string]string, error) { |
| - out := map[string]string{} |
| +func (f *fakeMirror) List(_ context.Context, bucket, prefix string) (map[string]MirrorObject, error) { |
| + out := map[string]MirrorObject{} |
| for k, v := range f.buckets[bucket] { |
| if strings.HasPrefix(k, prefix) { |
| - out[k] = v |
| + 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) |
| - f.put(dstBucket, dstKey, f.buckets[srcBucket][srcKey]) |
| - 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 { |
| @@ -49,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() |
| @@ -56,21 +117,21 @@ func TestMirrorPrefix(t *testing.T) { |
| f.put("src", "instances/h/generations/a/wal/000", "e2") |
| // Initial round: full copy. |
| - n, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| - if err != nil || n != 2 { |
| - t.Fatalf("initial round = %d, %v; want 2 copies", n, err) |
| + r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}) |
| + if err != nil || r.Changed != 2 { |
| + t.Fatalf("initial round = %+v, %v; want 2 copies", r, err) |
| } |
| // Identical: zero. |
| - if n, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || n != 0 { |
| - t.Fatalf("caught-up round = %d, %v; want 0", n, err) |
| + if r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", 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") |
| - n, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| - if err != nil || n != 3 { |
| - t.Fatalf("delta round = %d, %v; want 2 copies + 1 delete", n, err) |
| + r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}) |
| + if err != nil || r.Changed != 3 { |
| + t.Fatalf("delta round = %+v, %v; want 2 copies + 1 delete", r, err) |
| } |
| got, _ := f.List(ctx, "dst", "moves/h/") |
| want := map[string]string{ |
| @@ -81,14 +142,14 @@ func TestMirrorPrefix(t *testing.T) { |
| t.Fatalf("staging = %v, want %v", got, want) |
| } |
| for k, v := range want { |
| - if got[k] != v { |
| - t.Fatalf("staging[%s] = %q, want %q", k, got[k], v) |
| + if got[k].ETag != v { |
| + t.Fatalf("staging[%s] = %q, want %q", k, got[k].ETag, v) |
| } |
| } |
| // A copy failure surfaces with the partial count. |
| 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") |
| } |
| } |
| @@ -101,9 +162,9 @@ func TestMirrorPrefixNormalizesBoundaries(t *testing.T) { |
| f := newFakeMirror() |
| f.put("src", "instances/h/generations/a/snapshot", "e1") |
| f.put("src", "instances/h2/generations/a/snapshot", "eX") // the sibling |
| - n, err := MirrorPrefix(ctx, f, "src", "instances/h", "dst", "moves/h") |
| - if err != nil || n != 1 { |
| - t.Fatalf("round = %d, %v; want exactly the one h object", n, err) |
| + r, err := MirrorPrefix(ctx, f, "src", "instances/h", "dst", "moves/h", Receipts{}) |
| + if err != nil || r.Changed != 1 { |
| + t.Fatalf("round = %+v, %v; want exactly the one h object", r, err) |
| } |
| got, _ := f.List(ctx, "dst", "") |
| if _, ok := got["moves/h/generations/a/snapshot"]; !ok || len(got) != 1 { |
| @@ -119,15 +180,303 @@ func TestMirrorPrefixSkipsTheLease(t *testing.T) { |
| f.put("src", "instances/h/lock.json", "renewal-1") |
| f.put("src", "instances/h/generations/a/snapshot", "e1") |
| f.put("dst", "moves/h/lock.json", "old-foreign-lease") |
| - if n, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || n != 1 { |
| - t.Fatalf("round = %d, %v; want only the snapshot to move", n, err) |
| + if r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", 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 n, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || n != 0 { |
| - t.Fatalf("churn round = %d, %v; want caught-up despite the renewal", n, err) |
| + if r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", 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/") |
| - if got["moves/h/lock.json"] != "old-foreign-lease" || len(got) != 2 { |
| + if got["moves/h/lock.json"].ETag != "old-foreign-lease" || len(got) != 2 { |
| t.Fatalf("staging = %v; the dst lease must be untouched", got) |
| } |
| } |
| + |
| +// A round reports the bytes it copied: the listed source size of every |
| +// object it copied, and nothing for an object it skipped as unchanged, a |
| +// delete, or the lease. The move worker's readiness bar is this number. |
| +func TestMirrorPrefixCountsBytesCopied(t *testing.T) { |
| + ctx := context.Background() |
| + f := newFakeMirror() |
| + f.putSized("src", "instances/h/generations/a/snapshot", "e1", 1000) |
| + f.putSized("src", "instances/h/generations/a/wal/000", "e2", 30) |
| + f.putSized("src", "instances/h/lock.json", "renewal-1", 500) |
| + |
| + r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}) |
| + if err != nil || r.Changed != 2 || r.BytesCopied != 1030 { |
| + t.Fatalf("initial round = %+v, %v; want 2 changed, 1030 bytes", r, err) |
| + } |
| + |
| + // Unchanged objects are skipped and cost nothing; the lease's churn |
| + // costs nothing; a rewritten object costs its NEW size; a new one its |
| + // size; a pruned one is a change but no bytes. |
| + f.putSized("src", "instances/h/lock.json", "renewal-2", 500) |
| + f.putSized("src", "instances/h/generations/a/wal/000", "e2b", 70) |
| + f.putSized("src", "instances/h/generations/a/wal/001", "e3", 5) |
| + f.putSized("dst", "moves/h/generations/old/snapshot", "eOld", 9999) |
| + r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", 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/", Receipts{}) |
| + if err != nil || r.Changed != 0 || r.BytesCopied != 0 { |
| + t.Fatalf("caught-up round = %+v, %v; want nothing", r, err) |
| + } |
| + |
| + // A failed round still reports what it copied before the failure. |
| + f.putSized("src", "instances/h/generations/a/wal/002", "e4", 11) |
| + f.putSized("src", "instances/h/generations/a/wal/003", "e5", 13) |
| + f.failOn = "instances/h/generations/a/wal/003" |
| + r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", 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) |
| + } |
| +} |