| new file mode 100644 |
| index 000000000..b5d7364ca |
| --- /dev/null |
| +++ b/internal/lease/observed_test.go |
| @@ -0,0 +1,210 @@ |
| +package lease |
| + |
| +import ( |
| + "context" |
| + "reflect" |
| + "sort" |
| + "strings" |
| + "sync" |
| + "testing" |
| +) |
| + |
| +// eventLog collects an observer's events, in the order the round made them, |
| +// keyed by what they said. |
| +type eventLog struct { |
| + mu sync.Mutex |
| + evs []CopyEvent |
| +} |
| + |
| +func (l *eventLog) observe(e CopyEvent) { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + l.evs = append(l.evs, e) |
| +} |
| + |
| +// byRel is every non-final event by key. |
| +func (l *eventLog) byRel(t *testing.T) map[string]CopyEvent { |
| + t.Helper() |
| + out := map[string]CopyEvent{} |
| + for _, e := range l.evs { |
| + if e.Complete { |
| + continue |
| + } |
| + if _, dup := out[e.Rel]; dup { |
| + t.Fatalf("two events for %s: %+v", e.Rel, l.evs) |
| + } |
| + out[e.Rel] = e |
| + } |
| + return out |
| +} |
| + |
| +func (l *eventLog) completes() []CopyEvent { |
| + var out []CopyEvent |
| + for _, e := range l.evs { |
| + if e.Complete { |
| + out = append(out, e) |
| + } |
| + } |
| + return out |
| +} |
| + |
| +// mirrorState is everything a round leaves behind that a reader can see: |
| +// the destination's objects and the receipts object's bytes. |
| +func mirrorState(f *fakeMirror) map[string]string { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + out := map[string]string{} |
| + for b, objs := range f.buckets { |
| + for k, o := range objs { |
| + out[b+"/"+k] = o.ETag + "|" + o.body |
| + } |
| + } |
| + return out |
| +} |
| + |
| +func TestObservedMirrorEmitsEveryDecision(t *testing.T) { |
| + ctx := context.Background() |
| + rc := Receipts{Bucket: "src", Key: "receipts/h.json"} |
| + seed := func() *fakeMirror { |
| + f := newFakeMirror() |
| + f.workers = 1 |
| + f.putSized("src", "instances/h/same", "e-same", 4) // staged with the same ETag |
| + f.putMultipart("src", "instances/h/multi", "b-multi", 64) // copied once, then vouched by a receipt |
| + f.putSized("src", "instances/h/new", "e-new", 8) // to copy |
| + f.putSized("src", "instances/h/busy", "e-busy", 16) // changes under its copy: deferred |
| + f.putSized("src", "instances/h/lock.json", "lease", 1) // never decided |
| + f.putSized("dst", "moves/h/same", "e-same", 4) // already there |
| + f.putSized("dst", "moves/h/gone", "e-gone", 2) // the source no longer has it |
| + f.putSized("dst", "moves/h/lock.json", "dst-lease", 1) // never decided |
| + return f |
| + } |
| + // The multipart object's receipt comes from an earlier round of the |
| + // same pair, so this round skips it by receipt. |
| + prime := func(f *fakeMirror) { |
| + if _, err := MirrorPrefixKeep(ctx, f, "src", "instances/h/", "dst", "moves/h/", rc, func(rel string) bool { return rel == "multi" || rel == "same" }); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + busy := func(f *fakeMirror) { |
| + f.beforeCopy = func(k string) { |
| + if k == "instances/h/busy" { |
| + f.putSized("src", "instances/h/busy", "e-busy-2", 16) |
| + } |
| + } |
| + } |
| + |
| + // The observed round and the plain one leave the same destination and |
| + // the same receipts: the observer changes nothing. |
| + plain, observed := seed(), seed() |
| + prime(plain) |
| + prime(observed) |
| + plain.putSized("dst", "moves/h/gone", "e-gone", 2) |
| + observed.putSized("dst", "moves/h/gone", "e-gone", 2) |
| + busy(plain) |
| + busy(observed) |
| + rPlain, err := MirrorPrefixKeep(ctx, plain, "src", "instances/h/", "dst", "moves/h/", rc, nil) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var log eventLog |
| + rObs, err := MirrorPrefixKeepObserved(ctx, observed, "src", "instances/h/", "dst", "moves/h/", rc, nil, log.observe) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !reflect.DeepEqual(rPlain, rObs) { |
| + t.Fatalf("rounds differ: plain %+v, observed %+v", rPlain, rObs) |
| + } |
| + if a, b := mirrorState(plain), mirrorState(observed); !reflect.DeepEqual(a, b) { |
| + t.Fatalf("the observer changed what the round left:\nplain %v\nobserved %v", a, b) |
| + } |
| + |
| + evs := log.byRel(t) |
| + multi, _ := observed.object("dst", "moves/h/multi") |
| + newObj, _ := observed.object("dst", "moves/h/new") |
| + want := map[string]CopyEvent{ |
| + "same": {Rel: "same", SrcETag: "e-same", SrcSize: 4, DstETag: "e-same", Unchanged: true}, |
| + "multi": {Rel: "multi", SrcETag: "b-multi-27", SrcSize: 64, DstETag: multi.ETag, Unchanged: true}, |
| + "new": {Rel: "new", SrcETag: "e-new", SrcSize: 8, DstETag: newObj.ETag}, |
| + "busy": {Rel: "busy", SrcETag: "e-busy", SrcSize: 16, Deferred: true}, |
| + "gone": {Rel: "gone", Deleted: true}, |
| + } |
| + if !reflect.DeepEqual(evs, want) { |
| + t.Fatalf("events:\n got %+v\nwant %+v", evs, want) |
| + } |
| + if multi.ETag == "b-multi-27" { |
| + t.Fatal("the fixture's multipart copy kept the source ETag; the receipt case is not exercised") |
| + } |
| + done := log.completes() |
| + if len(done) != 1 || !log.evs[len(log.evs)-1].Complete { |
| + t.Fatalf("want exactly one Complete, and last: %+v", log.evs) |
| + } |
| + src, dst := append([]string(nil), done[0].SourceKeys...), append([]string(nil), done[0].DestKeys...) |
| + if !sort.StringsAreSorted(src) || !sort.StringsAreSorted(dst) { |
| + t.Fatalf("key sets are not sorted: %v %v", src, dst) |
| + } |
| + if !reflect.DeepEqual(src, []string{"busy", "multi", "new", "same"}) || !reflect.DeepEqual(dst, []string{"multi", "new", "same"}) { |
| + t.Fatalf("Complete key sets: source %v, destination %v", src, dst) |
| + } |
| + |
| + // A round that fails emits no Complete, and the events it made before |
| + // the failure stand. |
| + f := seed() |
| + f.putSized("src", "instances/h/zz-fails", "e-z", 1) |
| + f.failOn = "instances/h/zz-fails" |
| + var flog eventLog |
| + if _, err := MirrorPrefixKeepObserved(ctx, f, "src", "instances/h/", "dst", "moves/h/", rc, nil, flog.observe); err == nil || !strings.Contains(err.Error(), "zz-fails") { |
| + t.Fatalf("injected failure: %v", err) |
| + } |
| + if len(flog.completes()) != 0 { |
| + t.Fatalf("a failed round completed: %+v", flog.evs) |
| + } |
| + if e, ok := flog.byRel(t)["same"]; !ok || !e.Unchanged { |
| + t.Fatalf("a skip decided before the failure was not reported: %+v", flog.evs) |
| + } |
| + |
| + // A versioned store's copies and receipt skips carry the version. |
| + vf := &versionedFake{fakeMirror: seed()} |
| + vf.workers = 1 |
| + var vlog eventLog |
| + if _, err := MirrorPrefixKeepObserved(ctx, vf, "src", "instances/h/", "dst", "moves/h/", rc, nil, vlog.observe); err != nil { |
| + t.Fatal(err) |
| + } |
| + for rel, e := range vlog.byRel(t) { |
| + if (e.Deleted || e.Deferred) == (e.DstVersionID != "") { |
| + t.Errorf("%s: version %q on %+v", rel, e.DstVersionID, e) |
| + } |
| + } |
| + var again eventLog |
| + if _, err := MirrorPrefixKeepObserved(ctx, vf, "src", "instances/h/", "dst", "moves/h/", rc, nil, again.observe); err != nil { |
| + t.Fatal(err) |
| + } |
| + for rel, e := range again.byRel(t) { |
| + if e.Unchanged && e.DstVersionID == "" { |
| + t.Errorf("%s: a receipt-vouched skip without the receipt's version: %+v", rel, e) |
| + } |
| + } |
| + |
| + // MirrorPrefixKeep is the observed round with no observer. |
| + if _, err := MirrorPrefixKeepObserved(ctx, seed(), "src", "instances/h/", "dst", "moves/h/", rc, nil, nil); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// versionedFake is fakeMirror as a VersionedCopier, each copy a fresh |
| +// version. |
| +type versionedFake struct { |
| + *fakeMirror |
| + n int |
| +} |
| + |
| +func (v *versionedFake) CopyVersioned(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, string, error) { |
| + etag, err := v.Copy(ctx, srcBucket, srcKey, srcETag, dstBucket, dstKey) |
| + if err != nil { |
| + return "", "", err |
| + } |
| + v.mu.Lock() |
| + v.n++ |
| + n := v.n |
| + v.mu.Unlock() |
| + return etag, "ver-" + string(rune('a'+n)), nil |
| +} |
amadan