amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package lease
2
3import (
4 "context"
5 "encoding/json"
6 "fmt"
7 "strings"
8 "sync"
9 "testing"
10)
11
12// fakeMirror is an in-memory MirrorStore that keeps S3's ETag rules, plus
13// an op log so ordering tests can assert the sequence. Every object carries
14// a body (its content's identity). A single-part object's ETag IS its body
15// (S3's MD5), a multipart one's is body+"-27", and a copy always lands
16// single-part: the destination's ETag is the body, whatever the source's
17// was. The fakes before #175 kept the source ETag on copy, and hid the bug
18// that made every multipart object look changed on every round.
19//
20// A round runs its copies side by side, so every method holds mu, and the
21// hooks (beforeCopy, onCopy) run without it: a hook may call the fake.
22// workers is the round's concurrency (MirrorConcurrency); zero is the
23// production default.
24type fakeMirror struct {
25 mu sync.Mutex
26 buckets map[string]map[string]fakeObject
27 ops []string
28 failOn string
29 // beforeCopy runs at the top of Copy, before the precondition check:
30 // the window between a round's listing and its copy.
31 beforeCopy func(srcKey string)
32 // onCopy runs after the precondition check and before the object
33 // lands; ctx is the copy's own. An error it returns is the copy's.
34 onCopy func(ctx context.Context, srcKey string) error
35 // onGet runs at the top of Get, without the lock.
36 onGet func(key string)
37 failPut bool
38 failGet bool
39 puts int
40 gets int
41 workers int
42 // inFlight and peak count the copies running at once.
43 inFlight, peak int
44}
45
46type fakeObject struct {
47 MirrorObject
48 body string
49}
50
51func newFakeMirror() *fakeMirror { return &fakeMirror{buckets: map[string]map[string]fakeObject{}} }
52
53func (f *fakeMirror) MirrorWorkers() int {
54 if f.workers == 0 {
55 return DefaultMirrorWorkers
56 }
57 return f.workers
58}
59
60func (f *fakeMirror) put(bucket, key, etag string) { f.putSized(bucket, key, etag, 0) }
61
62// putSized stores a single-part object: its ETag is its body.
63func (f *fakeMirror) putSized(bucket, key, etag string, size int64) {
64 f.store(bucket, key, fakeObject{MirrorObject{ETag: etag, Size: size}, etag})
65}
66
67// putMultipart stores a multipart upload of body (litestream's uploader
68// for anything over one 5 MiB part).
69func (f *fakeMirror) putMultipart(bucket, key, body string, size int64) {
70 f.store(bucket, key, fakeObject{MirrorObject{ETag: body + "-27", Size: size}, body})
71}
72
73func (f *fakeMirror) store(bucket, key string, o fakeObject) {
74 f.mu.Lock()
75 defer f.mu.Unlock()
76 f.storeLocked(bucket, key, o)
77}
78
79func (f *fakeMirror) storeLocked(bucket, key string, o fakeObject) {
80 if f.buckets[bucket] == nil {
81 f.buckets[bucket] = map[string]fakeObject{}
82 }
83 f.buckets[bucket][key] = o
84}
85
86// remove deletes one object, as a test's own hand does.
87func (f *fakeMirror) remove(bucket, key string) {
88 f.mu.Lock()
89 defer f.mu.Unlock()
90 delete(f.buckets[bucket], key)
91}
92
93// object is one stored object, as a test reads it.
94func (f *fakeMirror) object(bucket, key string) (fakeObject, bool) {
95 f.mu.Lock()
96 defer f.mu.Unlock()
97 o, ok := f.buckets[bucket][key]
98 return o, ok
99}
100
101func (f *fakeMirror) List(_ context.Context, bucket, prefix string) (map[string]MirrorObject, error) {
102 f.mu.Lock()
103 defer f.mu.Unlock()
104 out := map[string]MirrorObject{}
105 for k, v := range f.buckets[bucket] {
106 if strings.HasPrefix(k, prefix) {
107 out[k] = v.MirrorObject
108 }
109 }
110 return out, nil
111}
112
113func (f *fakeMirror) Copy(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) {
114 f.mu.Lock()
115 failOn, before, on := f.failOn, f.beforeCopy, f.onCopy
116 f.inFlight++
117 if f.inFlight > f.peak {
118 f.peak = f.inFlight
119 }
120 f.mu.Unlock()
121 defer func() {
122 f.mu.Lock()
123 f.inFlight--
124 f.mu.Unlock()
125 }()
126 if failOn == srcKey {
127 return "", fmt.Errorf("injected copy failure on %s", srcKey)
128 }
129 if before != nil {
130 before(srcKey)
131 }
132 f.mu.Lock()
133 o, ok := f.buckets[srcBucket][srcKey]
134 f.mu.Unlock()
135 if !ok {
136 return "", fmt.Errorf("no such key %s", srcKey)
137 }
138 if srcETag != "" && o.ETag != srcETag {
139 return "", ErrCopySourceChanged
140 }
141 if on != nil {
142 if err := on(ctx, srcKey); err != nil {
143 return "", err
144 }
145 }
146 f.mu.Lock()
147 defer f.mu.Unlock()
148 f.ops = append(f.ops, "copy "+srcKey+" -> "+dstBucket+"/"+dstKey)
149 f.storeLocked(dstBucket, dstKey, fakeObject{MirrorObject{ETag: o.body, Size: o.Size}, o.body})
150 return o.body, nil
151}
152
153func (f *fakeMirror) Delete(_ context.Context, bucket, key string) error {
154 f.mu.Lock()
155 defer f.mu.Unlock()
156 f.ops = append(f.ops, "delete "+bucket+"/"+key)
157 delete(f.buckets[bucket], key)
158 return nil
159}
160
161func (f *fakeMirror) Get(_ context.Context, bucket, key string) ([]byte, error) {
162 f.mu.Lock()
163 on := f.onGet
164 f.mu.Unlock()
165 if on != nil {
166 on(key)
167 }
168 f.mu.Lock()
169 defer f.mu.Unlock()
170 f.gets++
171 if f.failGet {
172 return nil, fmt.Errorf("injected get failure on %s", key)
173 }
174 o, ok := f.buckets[bucket][key]
175 if !ok {
176 return nil, ErrMirrorNotFound
177 }
178 return []byte(o.body), nil
179}
180
181func (f *fakeMirror) Put(_ context.Context, bucket, key string, body []byte) error {
182 f.mu.Lock()
183 defer f.mu.Unlock()
184 if f.failPut {
185 return fmt.Errorf("injected put failure on %s", key)
186 }
187 f.puts++
188 f.ops = append(f.ops, "put "+bucket+"/"+key)
189 f.storeLocked(bucket, key, fakeObject{MirrorObject{ETag: fmt.Sprint("put-", f.puts), Size: int64(len(body))}, string(body)})
190 return nil
191}
192
193func TestMirrorPrefix(t *testing.T) {
194 ctx := context.Background()
195 f := newFakeMirror()
196 f.put("src", "instances/h/generations/a/snapshot", "e1")
197 f.put("src", "instances/h/generations/a/wal/000", "e2")
198
199 // Initial round: full copy.
200 r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
201 if err != nil || r.Changed != 2 {
202 t.Fatalf("initial round = %+v, %v; want 2 copies", r, err)
203 }
204 // Identical: zero.
205 if r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); err != nil || r.Changed != 0 {
206 t.Fatalf("caught-up round = %+v, %v; want 0", r, err)
207 }
208 // Delta: one changed etag, one new, one pruned at source.
209 f.put("src", "instances/h/generations/a/wal/000", "e2b")
210 f.put("src", "instances/h/generations/a/wal/001", "e3")
211 delete(f.buckets["src"], "instances/h/generations/a/snapshot")
212 r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
213 if err != nil || r.Changed != 3 {
214 t.Fatalf("delta round = %+v, %v; want 2 copies + 1 delete", r, err)
215 }
216 got, _ := f.List(ctx, "dst", "moves/h/")
217 want := map[string]string{
218 "moves/h/generations/a/wal/000": "e2b",
219 "moves/h/generations/a/wal/001": "e3",
220 }
221 if len(got) != len(want) {
222 t.Fatalf("staging = %v, want %v", got, want)
223 }
224 for k, v := range want {
225 if got[k].ETag != v {
226 t.Fatalf("staging[%s] = %q, want %q", k, got[k].ETag, v)
227 }
228 }
229 // A copy failure surfaces with the partial count.
230 f.put("src", "instances/h/generations/a/wal/002", "e4")
231 f.failOn = "instances/h/generations/a/wal/002"
232 if _, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); err == nil {
233 t.Fatal("injected failure did not surface")
234 }
235}
236
237// Prefix boundaries: a slashless prefix (activator.HibernatePrefix's shape)
238// is normalized to a directory boundary — no fused keys, and a sibling host
239// whose name extends this one's is never swept in.
240func TestMirrorPrefixNormalizesBoundaries(t *testing.T) {
241 ctx := context.Background()
242 f := newFakeMirror()
243 f.put("src", "instances/h/generations/a/snapshot", "e1")
244 f.put("src", "instances/h2/generations/a/snapshot", "eX") // the sibling
245 r, err := MirrorPrefix(ctx, f, "src", "instances/h", "dst", "moves/h", Receipts{})
246 if err != nil || r.Changed != 1 {
247 t.Fatalf("round = %+v, %v; want exactly the one h object", r, err)
248 }
249 got, _ := f.List(ctx, "dst", "")
250 if _, ok := got["moves/h/generations/a/snapshot"]; !ok || len(got) != 1 {
251 t.Fatalf("staging keys = %v, want the one properly-joined key", got)
252 }
253}
254
255// lock.json is coordination state: never copied, never deleted, and its
256// churn (the renewer rewrites it every TTL) must not keep changed above 0.
257func TestMirrorPrefixSkipsTheLease(t *testing.T) {
258 ctx := context.Background()
259 f := newFakeMirror()
260 f.put("src", "instances/h/lock.json", "renewal-1")
261 f.put("src", "instances/h/generations/a/snapshot", "e1")
262 f.put("dst", "moves/h/lock.json", "old-foreign-lease")
263 if r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); err != nil || r.Changed != 1 {
264 t.Fatalf("round = %+v, %v; want only the snapshot to move", r, err)
265 }
266 f.put("src", "instances/h/lock.json", "renewal-2") // the churn
267 if r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); err != nil || r.Changed != 0 {
268 t.Fatalf("churn round = %+v, %v; want caught-up despite the renewal", r, err)
269 }
270 got, _ := f.List(ctx, "dst", "moves/h/")
271 if got["moves/h/lock.json"].ETag != "old-foreign-lease" || len(got) != 2 {
272 t.Fatalf("staging = %v; the dst lease must be untouched", got)
273 }
274}
275
276// A round reports the bytes it copied: the listed source size of every
277// object it copied, and nothing for an object it skipped as unchanged, a
278// delete, or the lease. The move worker's readiness bar is this number.
279func TestMirrorPrefixCountsBytesCopied(t *testing.T) {
280 ctx := context.Background()
281 f := newFakeMirror()
282 f.putSized("src", "instances/h/generations/a/snapshot", "e1", 1000)
283 f.putSized("src", "instances/h/generations/a/wal/000", "e2", 30)
284 f.putSized("src", "instances/h/lock.json", "renewal-1", 500)
285
286 r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
287 if err != nil || r.Changed != 2 || r.BytesCopied != 1030 {
288 t.Fatalf("initial round = %+v, %v; want 2 changed, 1030 bytes", r, err)
289 }
290
291 // Unchanged objects are skipped and cost nothing; the lease's churn
292 // costs nothing; a rewritten object costs its NEW size; a new one its
293 // size; a pruned one is a change but no bytes.
294 f.putSized("src", "instances/h/lock.json", "renewal-2", 500)
295 f.putSized("src", "instances/h/generations/a/wal/000", "e2b", 70)
296 f.putSized("src", "instances/h/generations/a/wal/001", "e3", 5)
297 f.putSized("dst", "moves/h/generations/old/snapshot", "eOld", 9999)
298 r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
299 if err != nil || r.Changed != 3 || r.BytesCopied != 75 {
300 t.Fatalf("delta round = %+v, %v; want 3 changed (2 copies + 1 delete), 75 bytes", r, err)
301 }
302
303 r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
304 if err != nil || r.Changed != 0 || r.BytesCopied != 0 {
305 t.Fatalf("caught-up round = %+v, %v; want nothing", r, err)
306 }
307
308 // A failed round still reports what it copied before the failure.
309 f.putSized("src", "instances/h/generations/a/wal/002", "e4", 11)
310 f.putSized("src", "instances/h/generations/a/wal/003", "e5", 13)
311 f.failOn = "instances/h/generations/a/wal/003"
312 r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
313 if err == nil || r.Changed != 1 || r.BytesCopied != 11 {
314 t.Fatalf("failed round = %+v, %v; want an error with the partial 1 change, 11 bytes", r, err)
315 }
316}
317
318var testReceipts = Receipts{Bucket: "src", Key: "move-receipts/h.json"}
319
320func mirrorRound(t *testing.T, f *fakeMirror, rc Receipts) MirrorRound {
321 t.Helper()
322 r, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", rc)
323 if err != nil {
324 t.Fatalf("round: %v", err)
325 }
326 return r
327}
328
329func readReceipts(t *testing.T, f *fakeMirror) map[string]copyReceipt {
330 t.Helper()
331 body, err := f.Get(context.Background(), testReceipts.Bucket, testReceipts.Key)
332 if err != nil {
333 t.Fatalf("receipts: %v", err)
334 }
335 var b receiptBook
336 if err := json.Unmarshal(body, &b); err != nil {
337 t.Fatalf("receipts: %v", err)
338 }
339 return b.Objects
340}
341
342// #175 as production saw it: a multipart source's copy lands with a
343// different ETag, so ETag equality alone recopies every multipart object
344// on every round. Receipts let an unchanged one be copied once.
345func TestMirrorPrefixReceiptsSkipAnUnchangedMultipartObject(t *testing.T) {
346 const big = 138 << 20
347 fill := func(f *fakeMirror) {
348 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "level1", big)
349 f.putSized("src", "instances/h/ltx/0/b.ltx", "l0", 4096)
350 }
351
352 // Without receipts: the bug, kept visible.
353 f := newFakeMirror()
354 fill(f)
355 mirrorRound(t, f, Receipts{})
356 if r := mirrorRound(t, f, Receipts{}); r.Changed != 1 || r.BytesCopied != big {
357 t.Fatalf("without receipts the unchanged multipart object should recopy: %+v", r)
358 }
359
360 f = newFakeMirror()
361 fill(f)
362 if r := mirrorRound(t, f, testReceipts); r.Changed != 2 || r.BytesCopied != big+4096 {
363 t.Fatalf("first round = %+v, want both copied", r)
364 }
365 if got := readReceipts(t, f); len(got) != 1 || got["ltx/1/a.ltx"] != (copyReceipt{SrcETag: "level1-27", SrcSize: big, DstETag: "level1"}) {
366 t.Fatalf("receipts = %+v, want exactly the multipart object's", got)
367 }
368 puts := f.puts
369 for i := 0; i < 3; i++ {
370 // MirrorRound gained a map (Versions), so it is no longer comparable
371 // with ==; this is the same "nothing happened" assertion, field by field.
372 if r := mirrorRound(t, f, testReceipts); r.Changed != 0 || r.BytesCopied != 0 || r.Deferred != 0 || r.BytesDeferred != 0 || r.Versions != nil {
373 t.Fatalf("unchanged round %d = %+v, want nothing copied", i, r)
374 }
375 }
376 if f.puts != puts {
377 t.Fatalf("receipts rewritten on rounds that changed nothing (%d puts, want %d)", f.puts, puts)
378 }
379 // Receipts are never data: nothing beside the two objects is staged.
380 if staged, _ := f.List(context.Background(), "dst", ""); len(staged) != 2 {
381 t.Fatalf("staging = %v, want only the two data objects", staged)
382 }
383}
384
385// Single-part sources keep the plain ETag skip: no receipts are written,
386// so the existing behaviour and request count do not change.
387func TestMirrorPrefixSinglePartWritesNoReceipts(t *testing.T) {
388 f := newFakeMirror()
389 f.putSized("src", "instances/h/ltx/0/a.ltx", "a", 100)
390 mirrorRound(t, f, testReceipts)
391 f.putSized("src", "instances/h/ltx/0/b.ltx", "b", 100)
392 mirrorRound(t, f, testReceipts)
393 if r := mirrorRound(t, f, testReceipts); r.Changed != 0 {
394 t.Fatalf("caught-up round = %+v", r)
395 }
396 if f.puts != 0 {
397 t.Fatalf("%d receipt writes for single-part objects, want none", f.puts)
398 }
399}
400
401func TestMirrorPrefixReceiptRecopiesWhenTheSourceChanges(t *testing.T) {
402 f := newFakeMirror()
403 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
404 mirrorRound(t, f, testReceipts)
405 // Compaction rewrote it under the same name.
406 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v2", 101<<20)
407 if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 101<<20 {
408 t.Fatalf("rewritten source = %+v, want it recopied", r)
409 }
410 if got := f.buckets["dst"]["moves/h/ltx/1/a.ltx"].body; got != "v2" {
411 t.Fatalf("staged body = %q, want v2", got)
412 }
413 // The same size with different content is still a change: never
414 // skip on size alone.
415 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v3", 101<<20)
416 if r := mirrorRound(t, f, testReceipts); r.Changed != 1 {
417 t.Fatalf("same-size rewrite = %+v, want it recopied", r)
418 }
419 if r := mirrorRound(t, f, testReceipts); r.Changed != 0 {
420 t.Fatalf("after the recopy = %+v, want the new receipt to hold", r)
421 }
422}
423
424func TestMirrorPrefixReceiptRecopiesWhenTheDestinationChanges(t *testing.T) {
425 for _, tc := range []struct {
426 name string
427 touch func(f *fakeMirror)
428 }{
429 {"overwritten", func(f *fakeMirror) { f.putSized("dst", "moves/h/ltx/1/a.ltx", "intruder", 100<<20) }},
430 {"deleted", func(f *fakeMirror) { delete(f.buckets["dst"], "moves/h/ltx/1/a.ltx") }},
431 } {
432 t.Run(tc.name, func(t *testing.T) {
433 f := newFakeMirror()
434 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
435 mirrorRound(t, f, testReceipts)
436 tc.touch(f)
437 if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 100<<20 {
438 t.Fatalf("destination %s behind the receipt = %+v, want a recopy", tc.name, r)
439 }
440 if got := f.buckets["dst"]["moves/h/ltx/1/a.ltx"].body; got != "v1" {
441 t.Fatalf("staged body = %q, want v1", got)
442 }
443 })
444 }
445}
446
447// A crash between the copy and the receipt write costs one extra copy on
448// the next round, never a wrong skip.
449func TestMirrorPrefixLostReceiptWriteRecopiesOnce(t *testing.T) {
450 f := newFakeMirror()
451 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
452 f.failPut = true
453 r, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", testReceipts)
454 if err == nil || r.BytesCopied != 100<<20 {
455 t.Fatalf("round with a failed receipt write = %+v, %v; want the copy reported and the error surfaced", r, err)
456 }
457 f.failPut = false
458 if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 100<<20 {
459 t.Fatalf("next round = %+v, want exactly one recopy", r)
460 }
461 if r := mirrorRound(t, f, testReceipts); r.Changed != 0 {
462 t.Fatalf("round after = %+v, want the receipt to hold", r)
463 }
464}
465
466// A source overwritten between the listing and the copy fails the
467// conditional copy. The round records no receipt for it and carries on;
468// the next round copies the new version.
469func TestMirrorPrefixSourceChangedMidRoundIsDeferred(t *testing.T) {
470 f := newFakeMirror()
471 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
472 f.putSized("src", "instances/h/ltx/0/b.ltx", "b", 10)
473 var once sync.Once
474 f.beforeCopy = func(k string) {
475 if k == "instances/h/ltx/1/a.ltx" {
476 once.Do(func() { f.putMultipart("src", k, "v2", 120<<20) })
477 }
478 }
479 r := mirrorRound(t, f, testReceipts)
480 if r.Changed != 2 || r.Deferred != 1 || r.BytesDeferred != 100<<20 || r.BytesCopied != 10 {
481 t.Fatalf("round = %+v, want b copied and a deferred at its listed size", r)
482 }
483 if r.DeltaBytes() != 100<<20+10 {
484 t.Fatalf("DeltaBytes = %d", r.DeltaBytes())
485 }
486 if _, ok := f.buckets["dst"]["moves/h/ltx/1/a.ltx"]; ok {
487 t.Fatal("a deferred copy landed anyway")
488 }
489 if _, err := f.Get(context.Background(), testReceipts.Bucket, testReceipts.Key); err == nil {
490 if _, ok := readReceipts(t, f)["ltx/1/a.ltx"]; ok {
491 t.Fatal("a receipt was recorded for a copy that failed its precondition")
492 }
493 }
494 if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 120<<20 || r.Deferred != 0 {
495 t.Fatalf("next round = %+v, want the new version copied", r)
496 }
497 if got := f.buckets["dst"]["moves/h/ltx/1/a.ltx"].body; got != "v2" {
498 t.Fatalf("staged body = %q, want v2", got)
499 }
500}
501
502// Receipts name the pair they were written for. Another pair's receipts,
503// even with matching keys and ETags, are not evidence about this one.
504func TestMirrorPrefixIgnoresAnotherPairsReceipts(t *testing.T) {
505 ctx := context.Background()
506 f := newFakeMirror()
507 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
508 mirrorRound(t, f, testReceipts)
509 // Another destination that happens to hold a lookalike object.
510 f.putSized("dst", "other/h/ltx/1/a.ltx", "v1", 100<<20)
511 r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "other/h/", testReceipts)
512 if err != nil || r.Changed != 1 {
513 t.Fatalf("round toward another destination = %+v, %v; want a copy", r, err)
514 }
515}
516
517// Unreadable receipts are no receipts: everything recopies once, and the
518// book is rewritten.
519func TestMirrorPrefixCorruptReceiptsRecopy(t *testing.T) {
520 f := newFakeMirror()
521 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
522 mirrorRound(t, f, testReceipts)
523 f.store(testReceipts.Bucket, testReceipts.Key, fakeObject{body: "{not json"})
524 if r := mirrorRound(t, f, testReceipts); r.Changed != 1 {
525 t.Fatalf("round over corrupt receipts = %+v, want a recopy", r)
526 }
527 if r := mirrorRound(t, f, testReceipts); r.Changed != 0 {
528 t.Fatalf("round after = %+v, want the rewritten receipts to hold", r)
529 }
530}
531
532// A source object retention pruned takes its receipt with it.
533func TestMirrorPrefixPrunesReceiptsWithTheirObjects(t *testing.T) {
534 f := newFakeMirror()
535 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
536 f.putMultipart("src", "instances/h/ltx/1/b.ltx", "w1", 100<<20)
537 mirrorRound(t, f, testReceipts)
538 delete(f.buckets["src"], "instances/h/ltx/1/a.ltx")
539 if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 0 {
540 t.Fatalf("prune round = %+v, want one delete", r)
541 }
542 if got := readReceipts(t, f); len(got) != 1 || got["ltx/1/b.ltx"].SrcETag != "w1-27" {
543 t.Fatalf("receipts = %+v, want only b's", got)
544 }
545}
546
547// Receipts that cannot be READ (a failure that is not "missing") fail the
548// round: guessing "no receipts" would quietly bring back the recopy loop,
549// and saving over a book this round never read would destroy it.
550func TestMirrorPrefixReceiptReadFailureFailsTheRound(t *testing.T) {
551 f := newFakeMirror()
552 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
553 mirrorRound(t, f, testReceipts)
554 before := readReceipts(t, f)
555 f.putMultipart("src", "instances/h/ltx/1/b.ltx", "w1", 100<<20)
556 f.failGet = true
557 if _, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", testReceipts); err == nil {
558 t.Fatal("an unreadable receipts object did not fail the round")
559 }
560 f.failGet = false
561 if got := readReceipts(t, f); len(got) != len(before) || got["ltx/1/a.ltx"] != before["ltx/1/a.ltx"] {
562 t.Fatalf("receipts after a failed read = %+v, want them untouched (%+v)", got, before)
563 }
564}
565
566// fakeVersionedMirror is fakeMirror plus version ids on copies and puts,
567// the S3 behaviour MirrorPrefix records for the seal (moves spec §7.3).
568type fakeVersionedMirror struct {
569 *fakeMirror
570 n int
571}
572
573func (f *fakeVersionedMirror) CopyVersioned(ctx context.Context, sb, sk, se, db, dk string) (string, string, error) {
574 etag, err := f.fakeMirror.Copy(ctx, sb, sk, se, db, dk)
575 if err != nil {
576 return "", "", err
577 }
578 return etag, f.next(), nil
579}
580
581func (f *fakeVersionedMirror) PutVersioned(ctx context.Context, b, k string, body []byte) (string, error) {
582 if err := f.fakeMirror.Put(ctx, b, k, body); err != nil {
583 return "", err
584 }
585 return f.next(), nil
586}
587
588// next is the next version id; copies run side by side, so under mu.
589func (f *fakeVersionedMirror) next() string {
590 f.mu.Lock()
591 defer f.mu.Unlock()
592 f.n++
593 return fmt.Sprintf("v%d", f.n)
594}
595
596func TestMirrorPrefixRecordsDestinationVersions(t *testing.T) {
597 f := &fakeVersionedMirror{fakeMirror: newFakeMirror()}
598 f.put("src", "instances/h/0000/a.ltx", "e1")
599 f.put("src", "instances/h/0009/s.ltx", "e2")
600 rc := Receipts{Bucket: "src", Key: "move-receipts/h/mv.json"}
601 r1, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
602 if err != nil {
603 t.Fatal(err)
604 }
605 if r1.Versions["0000/a.ltx"] == "" || r1.Versions["0009/s.ltx"] == "" {
606 t.Fatalf("versions after round 1: %v", r1.Versions)
607 }
608 // Round 2 copies nothing; the versions come from the receipts.
609 r2, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
610 if err != nil {
611 t.Fatal(err)
612 }
613 if r2.BytesCopied != 0 || r2.Versions["0000/a.ltx"] != r1.Versions["0000/a.ltx"] || len(r2.Versions) != 2 {
614 t.Fatalf("round 2: %+v", r2)
615 }
616 // A crash between the copy and the receipt write: the next round sees
617 // the destination's ETag equal to the source's but no receipt (or a
618 // receipt for an older version of the key). It recopies rather than
619 // sealing a version it cannot vouch for.
620 f.putSized("src", "instances/h/0000/a.ltx", "e1-new", 64)
621 f.n++ // a copy whose receipt was never persisted
622 f.store("dst", "moves/h/mv/replica/0000/a.ltx", fakeObject{MirrorObject{ETag: "e1-new", Size: 64}, "e1-new"})
623 r4, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
624 if err != nil || r4.Versions["0000/a.ltx"] == r1.Versions["0000/a.ltx"] || r4.Changed == 0 || r4.BytesCopied != 64 {
625 t.Fatalf("stale receipt sealed: %+v %v", r4, err)
626 }
627 // The plain store's second identical round copies nothing (the
628 // existing TestMirrorPrefix contract, kept): see the next block.
629 // A plain MirrorStore records no versions and keeps today's rule: an
630 // equal ETag skips the copy with no receipt needed.
631 g := newFakeMirror()
632 g.putSized("src", "instances/h/0000/a.ltx", "e1", 64)
633 r3, err := MirrorPrefix(context.Background(), g, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
634 if err != nil || r3.Versions != nil || r3.BytesCopied != 64 {
635 t.Fatalf("plain store round 1: %+v %v", r3, err)
636 }
637 r3b, err := MirrorPrefix(context.Background(), g, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
638 if err != nil || r3b.Changed != 0 || r3b.BytesCopied != 0 {
639 t.Fatalf("plain store recopied an unchanged object: %+v %v", r3b, err)
640 }
641}
642
643// A multipart source's copy lands with a destination ETag that differs from
644// the source's, so the receipt-match branch (not the equal-ETag one) must
645// also vouch for the version.
646func TestVersionedMirrorRecordsMultipartVersions(t *testing.T) {
647 ctx := context.Background()
648 f := &fakeVersionedMirror{fakeMirror: newFakeMirror()}
649 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
650 f.putMultipart("src", "instances/h/ltx/1/b.ltx", "w1", 50<<20)
651 rc := testReceipts
652 round := func() MirrorRound {
653 t.Helper()
654 r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", rc)
655 if err != nil {
656 t.Fatal(err)
657 }
658 return r
659 }
660 r1 := round()
661 if r1.Changed != 2 || r1.Versions["ltx/1/a.ltx"] == "" || r1.Versions["ltx/1/b.ltx"] == "" || len(r1.Versions) != 2 {
662 t.Fatalf("round 1: %+v", r1)
663 }
664 if d := f.buckets["dst"]["moves/h/ltx/1/a.ltx"]; d.ETag == "v1-27" {
665 t.Fatalf("fake did not model the multipart ETag change: %v", d.ETag)
666 }
667 r2 := round()
668 if r2.Changed != 0 || r2.BytesCopied != 0 || len(r2.Versions) != 2 ||
669 r2.Versions["ltx/1/a.ltx"] != r1.Versions["ltx/1/a.ltx"] || r2.Versions["ltx/1/b.ltx"] != r1.Versions["ltx/1/b.ltx"] {
670 t.Fatalf("round 2: %+v (round 1 versions %v)", r2, r1.Versions)
671 }
672 // A receipt hand-written without dst_version_id (the plain path, or
673 // before this change) is recopied exactly once, then carries one.
674 book := readReceipts(t, f.fakeMirror)
675 rcpt := book["ltx/1/a.ltx"]
676 if rcpt.DstVersionID == "" {
677 t.Fatalf("receipt carries no version: %+v", rcpt)
678 }
679 rcpt.DstVersionID = ""
680 book["ltx/1/a.ltx"] = rcpt
681 body, err := json.Marshal(receiptBook{V: 1, SrcBucket: "src", SrcPrefix: "instances/h/", DstBucket: "dst", DstPrefix: "moves/h/", Objects: book})
682 if err != nil {
683 t.Fatal(err)
684 }
685 if err := f.fakeMirror.Put(ctx, rc.Bucket, rc.Key, body); err != nil {
686 t.Fatal(err)
687 }
688 r3 := round()
689 if r3.Changed != 1 || r3.BytesCopied != 100<<20 || r3.Versions["ltx/1/a.ltx"] == "" || r3.Versions["ltx/1/a.ltx"] == r1.Versions["ltx/1/a.ltx"] {
690 t.Fatalf("round 3 (versionless receipt): %+v", r3)
691 }
692 r4 := round()
693 if r4.Changed != 0 || r4.Versions["ltx/1/a.ltx"] != r3.Versions["ltx/1/a.ltx"] || len(r4.Versions) != 2 {
694 t.Fatalf("round 4: %+v", r4)
695 }
696}
697
698// Final review finding 4: a receipt that names the version "null" (a copy
699// into a versioning-suspended bucket, recorded before the staging mirror
700// refused it) vouches for nothing, since a later write overwrites that
701// version. The round recopies the object rather than skip it or report the
702// null version, and a copier that answers "null" has no version recorded.
703func TestVersionedMirrorRecopiesANullVersionReceipt(t *testing.T) {
704 ctx := context.Background()
705 f := &fakeVersionedMirror{fakeMirror: newFakeMirror()}
706 f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
707 rc := testReceipts
708 round := func() MirrorRound {
709 t.Helper()
710 r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", rc)
711 if err != nil {
712 t.Fatal(err)
713 }
714 return r
715 }
716 r1 := round()
717 book := readReceipts(t, f.fakeMirror)
718 rcpt := book["ltx/1/a.ltx"]
719 rcpt.DstVersionID = "null"
720 book["ltx/1/a.ltx"] = rcpt
721 body, err := json.Marshal(receiptBook{V: 1, SrcBucket: "src", SrcPrefix: "instances/h/", DstBucket: "dst", DstPrefix: "moves/h/", Objects: book})
722 if err != nil {
723 t.Fatal(err)
724 }
725 if err := f.fakeMirror.Put(ctx, rc.Bucket, rc.Key, body); err != nil {
726 t.Fatal(err)
727 }
728 r2 := round()
729 if r2.Changed != 1 || r2.BytesCopied != 100<<20 || r2.Versions["ltx/1/a.ltx"] == "null" || r2.Versions["ltx/1/a.ltx"] == r1.Versions["ltx/1/a.ltx"] {
730 t.Fatalf("round after a null-version receipt: %+v (round 1 %v)", r2, r1.Versions)
731 }
732 n := &nullVersionMirror{fakeMirror: newFakeMirror()}
733 n.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
734 r, err := MirrorPrefix(ctx, n, "src", "instances/h/", "dst", "moves/h/", rc)
735 if err != nil {
736 t.Fatal(err)
737 }
738 if v, ok := r.Versions["ltx/1/a.ltx"]; ok {
739 t.Fatalf("a null version was recorded: %q", v)
740 }
741 if got := readReceipts(t, n.fakeMirror)["ltx/1/a.ltx"]; got.DstVersionID != "" {
742 t.Fatalf("a receipt carries the null version: %+v", got)
743 }
744}
745
746// nullVersionMirror is a copier into a versioning-suspended bucket.
747type nullVersionMirror struct{ *fakeMirror }
748
749func (f *nullVersionMirror) CopyVersioned(ctx context.Context, sb, sk, se, db, dk string) (string, string, error) {
750 etag, err := f.fakeMirror.Copy(ctx, sb, sk, se, db, dk)
751 return etag, "null", err
752}
753
754// Version collection is opt-in: the production s3Mirror must never enter
755// the receipt-required mode, since an unversioned destination can never
756// supply a version receipt and every unchanged object would recopy forever
757// (round-4 finding 7).
758func TestPlainS3MirrorRecordsNoVersions(t *testing.T) {
759 if _, ok := interface{}(&s3Mirror{}).(VersionedCopier); ok {
760 t.Fatal("s3Mirror implements VersionedCopier")
761 }
762 if _, ok := interface{}(&s3Mirror{}).(VersionedPutter); ok {
763 t.Fatal("s3Mirror implements VersionedPutter")
764 }
765}
766
767// MirrorPrefixKeep mirrors only the source keys keep admits: a refused key
768// is never copied, and a destination copy of it is deleted as if the source
769// no longer had it. A nil keep is MirrorPrefix.
770func TestMirrorPrefixKeepMirrorsOnlyKeptKeys(t *testing.T) {
771 ctx := context.Background()
772 f := newFakeMirror()
773 f.put("src", "loc/0000/a.ltx", "a")
774 f.put("src", "loc/0001/b.ltx", "b")
775 f.put("src", "loc/g-old/0000/c.ltx", "c")
776 f.put("src", "loc/generation.json", "g")
777 f.put("dst", "lane/g-old/0000/c.ltx", "c") // a stale copy of a refused key
778 keep := func(rel string) bool { return strings.Count(rel, "/") == 1 && !strings.HasPrefix(rel, "g-") }
779 round, err := MirrorPrefixKeep(ctx, f, "src", "loc", "dst", "lane", Receipts{}, keep)
780 if err != nil {
781 t.Fatal(err)
782 }
783 got, _ := f.List(ctx, "dst", "lane/")
784 if len(got) != 2 || got["lane/0000/a.ltx"].ETag != "a" || got["lane/0001/b.ltx"].ETag != "b" {
785 t.Fatalf("staged %v, want only the two kept level files", got)
786 }
787 if round.Changed != 3 {
788 t.Fatalf("round %+v: want two copies and one delete", round)
789 }
790 // nil keep: MirrorPrefix exactly, every key.
791 if _, err := MirrorPrefixKeep(ctx, f, "src", "loc", "dst", "all", Receipts{}, nil); err != nil {
792 t.Fatal(err)
793 }
794 if all, _ := f.List(ctx, "dst", "all/"); len(all) != 4 {
795 t.Fatalf("nil keep staged %d keys, want 4", len(all))
796 }
797}
798