amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package genselect
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "io"
9 "reflect"
10 "strings"
11 "testing"
12 "time"
13
14 "github.com/carlosframework/platform/internal/hostclaim"
15 "github.com/carlosframework/platform/internal/lease"
16 "github.com/carlosframework/platform/internal/lease/leasetest"
17 "github.com/superfly/ltx"
18)
19
20type fixture struct {
21 t *testing.T
22 b *leasetest.VersionedBucket
23 req Request
24 marker Marker
25 markerVersion string
26 pin File
27 store Reader
28}
29
30func newFixture(t *testing.T) *fixture {
31 b := leasetest.NewVersionedBucket()
32 b.AddBucket("data")
33 b.PageSize = 1
34 f := &fixture{t: t, b: b, store: b, req: Request{Bucket: "data", Region: "ie", Host: "h",
35 Holder: hostclaim.Holder{Incarnation: "inc", Location: "instances/h/inc", Generation: "g0"}}}
36 f.pin = f.file("g0", 9, 1, 11)
37 f.file("g0", 0, 11, 11)
38 f.marker = Marker{Op: "g0", Incarnation: "inc", Host: "h", Box: "box", InstalledAt: time.Unix(1, 0).UTC(),
39 ManifestSHA256: strings.Repeat("a", 64), Pin: Pin{SnapshotKey: f.pin.Key, SnapshotVersion: f.pin.VersionID, SnapshotTXID: 11}}
40 f.publish()
41 return f
42}
43
44func header(t *testing.T, min, max uint64) []byte {
45 t.Helper()
46 h := ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096,
47 MinTXID: ltx.TXID(min), MaxTXID: ltx.TXID(max)}
48 b, err := h.MarshalBinary()
49 if err != nil {
50 t.Fatal(err)
51 }
52 return b
53}
54
55func (f *fixture) file(gen string, level int, min, max uint64) File {
56 f.t.Helper()
57 key := fmt.Sprintf("%s%04x/%016x-%016x.ltx", prefix(strings.TrimSuffix(f.req.Holder.Location, "/")+"/", origin(gen)), level, min, max)
58 b := header(f.t, min, max)
59 id := f.b.Put("data", key, b)
60 return File{Key: key, VersionID: id, Level: level, MinTXID: min, MaxTXID: max, Size: int64(len(b))}
61}
62
63func (f *fixture) publish() {
64 f.t.Helper()
65 b, err := json.Marshal(f.marker)
66 if err != nil {
67 f.t.Fatal(err)
68 }
69 f.markerVersion = f.b.Put("data", f.req.Holder.Location+"/generation.json", b)
70}
71
72func (f *fixture) chain(base string) {
73 f.req.Holder.Generation = "g2"
74 f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
75 {Generation: "g1", Predecessor: base, Region: "ie", PredecessorRegion: "ie", Mode: "cold"},
76 {Generation: "g2", Predecessor: "g1", Region: "ie", PredecessorRegion: "ie", Mode: "cold"},
77 }
78}
79
80func (f *fixture) origin() {
81 f.req.Holder.Generation = ""
82 if err := f.b.DeleteVersion(f.t.Context(), "data", f.req.Holder.Location+"/generation.json", f.markerVersion); err != nil {
83 f.t.Fatal(err)
84 }
85 f.markerVersion = ""
86}
87
88// These are pure-selection ports, not simulations of wake, park or replication.
89// Paths refer to platform-moves-design/internal/sim/hostlife at rev 7.13.
90func TestModelWitnesses(t *testing.T) {
91 tests := []struct {
92 name, source string
93 setup func(*fixture) []File
94 gen string
95 wantErr error
96 }{
97 {"wRestoreOverlap", "witness_test.go:1204", func(f *fixture) []File {
98 return []File{f.pin, f.file("g0", 1, 10, 13)}
99 }, "g0", nil},
100 {"wRestoreGap", "witness_test.go:1233", func(f *fixture) []File {
101 f.file("g0", 0, 13, 13)
102 return nil
103 }, "", ErrGap},
104 {"wRootAdvance", "witness_test.go:1295", func(f *fixture) []File {
105 f.file("g0", 0, 13, 13)
106 root := f.file("g0", 9, 1, 20)
107 f.b.BareDelete("data", f.pin.Key)
108 return []File{root, f.file("g0", 0, 21, 21)}
109 }, "g0", nil},
110 {"wOriginSwept/origin", "witness_test.go:2373", func(f *fixture) []File {
111 f.origin()
112 return []File{f.file("", 0, 1, 1), f.file("", 1, 2, 5)}
113 }, hostclaim.RootGeneration, nil},
114 {"wOriginSwept/unfinished_chain", "witness_test.go:2373; protocol_test.go:1188", func(f *fixture) []File {
115 f.origin()
116 f.chain(hostclaim.RootGeneration)
117 return []File{f.file("", 0, 1, 3)}
118 }, hostclaim.RootGeneration, nil},
119 {"wBaselineParked/all_empty", "witness_test.go:2168; protocol_test.go:1154", func(f *fixture) []File {
120 f.chain("g0")
121 return []File{f.pin}
122 }, "g0", nil},
123 {"wBaselineParked/newest_nonempty", "witness_test.go:2168; protocol_test.go:1172", func(f *fixture) []File {
124 f.chain("g0")
125 return []File{f.file("g1", 0, 1, 1)}
126 }, "g1", nil},
127 {"wBaselineParked/holder_nonempty", "protocol_test.go:1161", func(f *fixture) []File {
128 f.chain("g0")
129 f.file("g1", 0, 1, 9)
130 return []File{f.file("g2", 0, 1, 1)}
131 }, "g2", nil},
132 }
133 for _, tt := range tests {
134 t.Run(tt.name, func(t *testing.T) {
135 t.Log("port of", tt.source)
136 f := newFixture(t)
137 want := tt.setup(f)
138 before := f.b.Snapshot()
139 f.b.FailOn = func(op, key string) error {
140 if op != "list" && op != "get" {
141 t.Fatalf("mutation %s %s", op, key)
142 }
143 return nil
144 }
145 got, err := Select(t.Context(), f.store, f.req)
146 if !errors.Is(err, tt.wantErr) {
147 t.Fatalf("got %v, want %v", err, tt.wantErr)
148 }
149 if f.b.Snapshot() != before {
150 t.Fatal("selection mutated storage")
151 }
152 if tt.wantErr != nil {
153 if got != nil {
154 t.Fatal("partial plan")
155 }
156 return
157 }
158 if got.Generation != tt.gen || !reflect.DeepEqual(got.Files, want) {
159 t.Fatalf("got %+v, want %s %+v", got, tt.gen, want)
160 }
161 if got.MarkerVersion != f.markerVersion || got.Root == nil || *got.Root != want[0] || got.MaxTXID != want[len(want)-1].MaxTXID {
162 t.Fatalf("root, marker or final TXID: %+v", got)
163 }
164 })
165 }
166}
167
168type altered struct {
169 Reader
170 list func([]lease.ObjectVersion) ([]lease.ObjectVersion, error)
171 open func(context.Context, string, string, string) (io.ReadCloser, error)
172}
173
174func (s altered) ListVersions(ctx context.Context, bucket, base string) ([]lease.ObjectVersion, error) {
175 vs, err := s.Reader.ListVersions(ctx, bucket, base)
176 if err != nil {
177 return nil, err
178 }
179 if s.list != nil {
180 return s.list(vs)
181 }
182 return vs, nil
183}
184func (s altered) OpenVersion(ctx context.Context, bucket, key, id string) (io.ReadCloser, error) {
185 if s.open != nil {
186 return s.open(ctx, bucket, key, id)
187 }
188 return s.Reader.OpenVersion(ctx, bucket, key, id)
189}
190
191func TestR13_1UnfinishedRegionalWalk(t *testing.T) {
192 tests := []struct {
193 name string
194 setup func(*fixture) []File
195 gen string
196 want error
197 }{
198 {"interrupted_clean_return", func(f *fixture) []File {
199 return []File{f.pin}
200 }, "g0", nil},
201 {"another_interrupted_admission", func(f *fixture) []File {
202 f.req.Holder.Generation = "R2"
203 f.req.Holder.Admissions = append(f.req.Holder.Admissions,
204 hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold"})
205 return []File{f.pin}
206 }, "g0", nil},
207 {"covered_foreign_link_deeper_in_run", func(f *fixture) []File {
208 f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
209 {Generation: "T", Predecessor: "g0", Region: "us", PredecessorRegion: "ie", Mode: "cold"},
210 {Generation: "U", Predecessor: "T", Region: "uk", PredecessorRegion: "us", Mode: "cold", Covered: true},
211 {Generation: "R", Predecessor: "U", Region: "ie", PredecessorRegion: "uk", Mode: "cold"},
212 }
213 return []File{f.pin}
214 }, "g0", nil},
215 {"uncovered_foreign_link", func(f *fixture) []File {
216 f.req.Holder.Admissions[1].Covered = false
217 return []File{f.pin}
218 }, "g0", nil},
219 {"foreign_objects_never_selected", func(f *fixture) []File {
220 f.req.Holder.Admissions[1].Covered = false
221 f.file("T", 9, 1, 99)
222 return []File{f.pin}
223 }, "g0", nil},
224 {"holder_objects_with_covered_foreign_predecessor", func(f *fixture) []File {
225 return []File{f.file("R", 0, 1, 1)}
226 }, "R", nil},
227 {"newest_local_objects", func(f *fixture) []File {
228 f.req.Holder.Generation = "R2"
229 f.req.Holder.Admissions = append(f.req.Holder.Admissions,
230 hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold"})
231 return []File{f.file("R", 0, 1, 1)}
232 }, "R", nil},
233 {"unsafe_covered_local_baseline", func(f *fixture) []File {
234 f.req.Holder.Generation = "R2"
235 f.req.Holder.Admissions = append(f.req.Holder.Admissions,
236 hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold", Covered: true})
237 f.file("R", 0, 1, 1)
238 return nil
239 }, "", ErrCovered},
240 {"unsafe_covered_empty_local_baseline", func(f *fixture) []File {
241 f.req.Holder.Generation = "R2"
242 f.req.Holder.Admissions = append(f.req.Holder.Admissions,
243 hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold", Covered: true})
244 return nil
245 }, "", ErrCovered},
246 {"unsafe_covered_local_marker", func(f *fixture) []File {
247 f.req.Holder.Admissions[0].Covered = true
248 return nil
249 }, "", ErrCovered},
250 {"older_covered_local_not_needed", func(f *fixture) []File {
251 f.req.Holder.Admissions[0].Covered = true
252 return []File{f.file("R", 0, 1, 1)}
253 }, "R", nil},
254 {"covered_warm_session_in_run", func(f *fixture) []File {
255 f.req.Holder.Admissions = append(f.req.Holder.Admissions,
256 hostclaim.AdmissionRecord{Generation: "R", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "warm", Covered: true})
257 return []File{f.pin}
258 }, "g0", nil},
259 {"absent_marker_origin", func(f *fixture) []File {
260 f.origin()
261 f.req.Holder.Generation = "R"
262 f.req.Holder.Admissions[0].Predecessor = hostclaim.RootGeneration
263 return []File{f.file("", 0, 1, 1)}
264 }, hostclaim.RootGeneration, nil},
265 {"foreign_predecessor_region_marker_fallback", func(f *fixture) []File {
266 f.chain("g0")
267 f.req.Holder.Admissions[0].PredecessorRegion = "us"
268 return []File{f.pin}
269 }, "g0", nil},
270 }
271 for _, tt := range tests {
272 t.Run(tt.name, func(t *testing.T) {
273 f := newFixture(t)
274 f.req.Holder.Generation = "R"
275 f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
276 {Generation: "T", Predecessor: "g0", Region: "us", PredecessorRegion: "ie", Mode: "cold"},
277 {Generation: "R", Predecessor: "T", Region: "ie", PredecessorRegion: "us", Mode: "cold", Coverage: "clean-return", Covered: true},
278 }
279 want := tt.setup(f)
280 before := f.b.Snapshot()
281 holderBefore, _ := json.Marshal(f.req.Holder)
282 got, err := Select(t.Context(), f.store, f.req)
283 holderAfter, _ := json.Marshal(f.req.Holder)
284 if f.b.Snapshot() != before || string(holderAfter) != string(holderBefore) {
285 t.Fatal("selection mutated inputs or storage")
286 }
287 if !errors.Is(err, tt.want) {
288 t.Fatalf("got %+v / %v, want %v", got, err, tt.want)
289 }
290 if tt.want != nil {
291 if got != nil {
292 t.Fatal("partial plan")
293 }
294 return
295 }
296 if got == nil || got.Generation != tt.gen || !reflect.DeepEqual(got.Files, want) ||
297 got.Root == nil || *got.Root != want[0] || got.MaxTXID != want[len(want)-1].MaxTXID ||
298 got.MarkerVersion != f.markerVersion {
299 t.Fatalf("got %+v, want %s %+v", got, tt.gen, want)
300 }
301 if tt.gen == f.marker.Op && (got.Pin == nil || *got.Pin != f.marker.Pin) {
302 t.Fatalf("marker pin not preserved: %+v", got.Pin)
303 }
304 })
305 }
306}
307
308func TestRefusals(t *testing.T) {
309 tests := []struct {
310 name string
311 setup func(*fixture)
312 want error
313 }{
314 {"marker_holder_disagreement", func(f *fixture) { f.req.Holder.Generation = "unknown" }, ErrDisagreement},
315 {"nonempty_unexplained_disagreement", func(f *fixture) { f.req.Holder.Generation = "unknown"; f.file("unknown", 0, 1, 1) }, ErrDisagreement},
316 {"covered_local_marker", func(f *fixture) { f.chain("g0"); f.req.Holder.Admissions[0].Covered = true }, ErrCovered},
317 {"broken_chain", func(f *fixture) { f.chain("g0"); f.req.Holder.Admissions[1].Predecessor = "missing" }, ErrDisagreement},
318 {"cyclic_chain", func(f *fixture) { f.chain("g0"); f.req.Holder.Admissions[0].Predecessor = "g2" }, ErrDisagreement},
319 {"incomplete_listing", func(f *fixture) { f.b.FailListAfterPages = 1 }, ErrListing},
320 {"partial_data_with_error", func(f *fixture) {
321 f.store = altered{Reader: f.b, list: func(vs []lease.ObjectVersion) ([]lease.ObjectVersion, error) { return vs, io.ErrUnexpectedEOF }}
322 }, ErrListing},
323 {"incomplete_unfinished_listing", func(f *fixture) { f.chain("g0"); f.b.FailListAfterPages = 1 }, ErrListing},
324 {"pinned_version_missing", func(f *fixture) {
325 f.b.Put("data", f.pin.Key, header(f.t, 1, 11))
326 if err := f.b.DeleteVersion(f.t.Context(), "data", f.pin.Key, f.pin.VersionID); err != nil {
327 f.t.Fatal(err)
328 }
329 }, ErrPinnedMissing},
330 {"pinned_version_vanished_after_listing", func(f *fixture) {
331 f.store = altered{Reader: f.b, open: func(context.Context, string, string, string) (io.ReadCloser, error) {
332 return nil, lease.ErrMirrorNotFound
333 }}
334 }, ErrPinnedMissing},
335 {"pinned_header_mismatch", func(f *fixture) {
336 f.marker.SnapshotVersion = f.b.Put("data", f.pin.Key, header(f.t, 1, 12))
337 f.publish()
338 }, ErrPinnedMismatch},
339 {"pinned_key_mismatch", func(f *fixture) { f.marker.SnapshotTXID = 12; f.publish() }, ErrPinnedMismatch},
340 {"pinned_delete_marker", func(f *fixture) { f.marker.SnapshotVersion = f.b.BareDelete("data", f.pin.Key); f.publish() }, ErrDeleteMarker},
341 {"selected_delete_marker_on_open", func(f *fixture) {
342 f.store = altered{Reader: f.b, open: func(context.Context, string, string, string) (io.ReadCloser, error) {
343 return nil, lease.ErrIsDeleteMarker
344 }}
345 }, ErrDeleteMarker},
346 {"marker_delete_marker", func(f *fixture) { f.b.BareDelete("data", f.req.Holder.Location+"/generation.json") }, ErrDeleteMarker},
347 {"marker_read_error", func(f *fixture) {
348 f.b.FailOn = func(op, key string) error {
349 if op == "get" {
350 return io.ErrUnexpectedEOF
351 }
352 return nil
353 }
354 }, ErrMarker},
355 {"marker_disappears", func(f *fixture) {
356 f.b.FailOn = func(op, key string) error {
357 if op == "get" {
358 return lease.ErrMirrorNotFound
359 }
360 return nil
361 }
362 }, ErrMarker},
363 {"marker_incarnation", func(f *fixture) { f.marker.Incarnation = "previous"; f.publish() }, ErrIncarnation},
364 {"marker_host", func(f *fixture) { f.marker.Host = "other"; f.publish() }, ErrMarker},
365 {"malformed_marker", func(f *fixture) { f.b.Put("data", f.req.Holder.Location+"/generation.json", []byte("{")) }, ErrMarker},
366 {"trailing_marker_json", func(f *fixture) {
367 b, _ := json.Marshal(f.marker)
368 f.b.Put("data", f.req.Holder.Location+"/generation.json", append(b, []byte(" {}")...))
369 }, ErrMarker},
370 {"empty_pin_version", func(f *fixture) { f.marker.SnapshotVersion = ""; f.publish() }, ErrVersionID},
371 {"null_pin_version", func(f *fixture) { f.marker.SnapshotVersion = "null"; f.publish() }, ErrVersionID},
372 {"empty_current_version", func(f *fixture) { alterVersion(f, "") }, ErrVersionID},
373 {"null_current_version", func(f *fixture) { alterVersion(f, "null") }, ErrVersionID},
374 {"tombstoned_unfinished_is_not_empty", func(f *fixture) {
375 f.chain("g0")
376 dead := f.file("g2", 0, 1, 1)
377 f.b.BareDelete("data", dead.Key)
378 }, ErrEmpty},
379 {"metadata_only_unfinished", func(f *fixture) {
380 f.chain("g0")
381 f.b.Put("data", f.req.Holder.Location+"/g-g2/build-intent.json", []byte("{}"))
382 }, ErrEmpty},
383 {"empty_published_generation", func(f *fixture) {
384 f.marker.Op = "empty"
385 f.marker.Pin = Pin{}
386 f.req.Holder.Generation = "empty"
387 f.publish()
388 }, ErrEmpty},
389 {"bad_successor_no_fallback", func(f *fixture) { r := f.file("g0", 9, 1, 20); f.b.Put("data", r.Key, header(f.t, 1, 19)) }, ErrRoot},
390 {"missing_snapshot_txid", func(f *fixture) {
391 b, _ := json.Marshal(f.marker)
392 var fields map[string]any
393 json.Unmarshal(b, &fields)
394 delete(fields, "snapshot_txid")
395 b, _ = json.Marshal(fields)
396 f.b.Put("data", f.req.Holder.Location+"/generation.json", b)
397 }, ErrMarker},
398 {"malformed_current_ltx", func(f *fixture) { f.b.Put("data", f.req.Holder.Location+"/g-g0/0000/not-a-range.ltx", []byte("bad")) }, ErrListing},
399 {"invalid_snapshot_min", func(f *fixture) { f.file("g0", 9, 2, 20) }, ErrRoot},
400 {"gap_without_root", func(f *fixture) { f.origin(); f.file("", 0, 2, 2) }, ErrGap},
401 {"location_escape", func(f *fixture) { f.req.Holder.Location = "instances/h/../other" }, ErrInput},
402 {"holder_root_sentinel", func(f *fixture) { f.req.Holder.Generation = hostclaim.RootGeneration }, ErrInput},
403 }
404 for _, tt := range tests {
405 t.Run(tt.name, func(t *testing.T) {
406 f := newFixture(t)
407 tt.setup(f)
408 before := f.b.Snapshot()
409 got, err := Select(t.Context(), f.store, f.req)
410 if got != nil || !errors.Is(err, tt.want) {
411 t.Fatalf("got %+v / %v, want %v", got, err, tt.want)
412 }
413 if f.b.Snapshot() != before {
414 t.Fatal("mutated storage")
415 }
416 })
417 }
418}
419
420func TestContinuationRootRefusals(t *testing.T) {
421 for _, gen := range []string{"", "fresh"} {
422 name := gen
423 if name == "" {
424 name = "origin"
425 }
426 for level := 0; level < 9; level++ {
427 for _, corruption := range []string{"malformed_header", "truncated_file", "header_range_mismatch"} {
428 t.Run(fmt.Sprintf("%s/level_%d/%s", name, level, corruption), func(t *testing.T) {
429 f := newFixture(t)
430 if gen == "" {
431 f.origin()
432 } else {
433 f.marker.Op = gen
434 f.marker.Pin = Pin{}
435 f.req.Holder.Generation = gen
436 f.publish()
437 }
438 root := f.file(gen, level, 1, 1)
439 body := header(t, 1, 1)
440 switch corruption {
441 case "malformed_header":
442 body[0] ^= 0xff
443 case "truncated_file":
444 body = body[:len(body)-1]
445 case "header_range_mismatch":
446 body = header(t, 1, 2)
447 }
448 f.b.Put("data", root.Key, body)
449 before := f.b.Snapshot()
450 got, err := Select(t.Context(), f.store, f.req)
451 if got != nil || !errors.Is(err, ErrRoot) {
452 t.Fatalf("got %+v / %v, want nil / %v", got, err, ErrRoot)
453 }
454 if f.b.Snapshot() != before {
455 t.Fatal("selection mutated storage")
456 }
457 })
458 }
459 }
460 }
461}
462
463func TestMaxLevel0TXID(t *testing.T) {
464 tests := []struct {
465 name string
466 setup func(*fixture)
467 wantLevel0 uint64
468 wantRestored uint64
469 }{
470 {"current_anchor", func(f *fixture) {}, 11, 11},
471 {"current_continuation", func(f *fixture) {
472 f.file("g0", 0, 12, 12)
473 f.file("g0", 0, 13, 13)
474 }, 13, 13},
475 {"noncurrent_data", func(f *fixture) {
476 old := f.file("g0", 0, 30, 30)
477 f.b.BareDelete("data", old.Key)
478 }, 11, 11},
479 {"delete_marker_only", func(f *fixture) {
480 old := f.file("g0", 0, 40, 40)
481 f.b.BareDelete("data", old.Key)
482 if err := f.b.DeleteVersion(f.t.Context(), "data", old.Key, old.VersionID); err != nil {
483 f.t.Fatal(err)
484 }
485 }, 11, 11},
486 {"newer_restored_snapshot", func(f *fixture) {
487 f.file("g0", 9, 1, 30)
488 }, 11, 30},
489 {"no_current_level0", func(f *fixture) {
490 anchor := f.file("g0", 0, 11, 11)
491 f.b.BareDelete("data", anchor.Key)
492 }, 0, 11},
493 }
494 for _, entry := range []string{"ordinary", "prepared"} {
495 for _, tt := range tests {
496 t.Run(entry+"/"+tt.name, func(t *testing.T) {
497 f := newFixture(t)
498 tt.setup(f)
499 var got *Plan
500 var err error
501 if entry == "ordinary" {
502 got, err = Select(t.Context(), f.store, f.req)
503 } else {
504 p := f.marker.Pin
505 got, err = SelectPrepared(t.Context(), f.store, f.req, PreparedPin{
506 Generation: "g0", SnapshotKey: p.SnapshotKey, SnapshotVersion: p.SnapshotVersion,
507 SnapshotTXID: p.SnapshotTXID, SnapshotCRC64: p.SnapshotCRC64,
508 })
509 }
510 if err != nil || got == nil {
511 t.Fatalf("got %+v / %v, want a plan", got, err)
512 }
513 if got.MaxLevel0TXID != tt.wantLevel0 || got.MaxTXID != tt.wantRestored {
514 t.Fatalf("MaxLevel0TXID=%d, MaxTXID=%d; want %d, %d",
515 got.MaxLevel0TXID, got.MaxTXID, tt.wantLevel0, tt.wantRestored)
516 }
517 })
518 }
519 }
520}
521
522func alterVersion(f *fixture, id string) {
523 f.file("g0", 0, 12, 12)
524 f.store = altered{Reader: f.b, list: func(vs []lease.ObjectVersion) ([]lease.ObjectVersion, error) {
525 for i := range vs {
526 if strings.Contains(vs[i].Key, "0000/000000000000000c-") {
527 vs[i].VersionID = id
528 }
529 }
530 return vs, nil
531 }}
532}
533
534func TestSelectionDetails(t *testing.T) {
535 tests := []struct {
536 name string
537 setup func(*fixture) []File
538 }{
539 {"pinned_noncurrent_duplicate", func(f *fixture) []File {
540 f.b.Put("data", f.pin.Key, header(f.t, 1, 11))
541 return []File{f.pin}
542 }},
543 {"pinned_under_retention_tombstone", func(f *fixture) []File {
544 f.b.BareDelete("data", f.pin.Key)
545 return []File{f.pin}
546 }},
547 {"retention_tombstones_ignored", func(f *fixture) []File {
548 f.b.BareDelete("data", f.file("g0", 0, 13, 13).Key)
549 return []File{f.pin}
550 }},
551 {"max_then_min_then_level", func(f *fixture) []File {
552 f.file("g0", 8, 12, 19)
553 f.file("g0", 8, 12, 20)
554 f.file("g0", 1, 10, 20)
555 return []File{f.pin, f.file("g0", 2, 10, 20)}
556 }},
557 {"origin_snapshot_and_isolation", func(f *fixture) []File {
558 f.origin()
559 f.file("other", 9, 1, 999)
560 f.b.Put("data", f.req.Holder.Location+"/child/0009/0000000000000001-000000000000ffff.ltx", header(f.t, 1, 65535))
561 return []File{f.file("", 9, 1, 20), f.file("", 0, 21, 21)}
562 }},
563 {"legacy_location", func(f *fixture) []File {
564 f.req.Holder.Location = "instances/h"
565 f.req.Holder.Generation = ""
566 return []File{f.file("", 0, 1, 1)}
567 }},
568 {"fresh_zero_pin_marker", func(f *fixture) []File {
569 f.marker.Op = "fresh"
570 f.marker.Pin = Pin{}
571 f.req.Holder.Generation = "fresh"
572 f.publish()
573 return []File{f.file("fresh", 0, 1, 1)}
574 }},
575 {"warm_record_in_unfinished_run", func(f *fixture) []File {
576 f.chain("g0")
577 f.req.Holder.Admissions = append(f.req.Holder.Admissions, hostclaim.AdmissionRecord{Generation: "g2", Predecessor: "g2", Region: "ie", PredecessorRegion: "ie", Mode: "warm"})
578 return []File{f.pin}
579 }},
580 {"root_advance_past_missing_old_pin", func(f *fixture) []File {
581 if err := f.b.DeleteVersion(f.t.Context(), "data", f.pin.Key, f.pin.VersionID); err != nil {
582 f.t.Fatal(err)
583 }
584 return []File{f.file("g0", 9, 1, 30)}
585 }},
586 }
587 for _, tt := range tests {
588 t.Run(tt.name, func(t *testing.T) {
589 f := newFixture(t)
590 want := tt.setup(f)
591 got, err := Select(t.Context(), f.store, f.req)
592 if err != nil || !reflect.DeepEqual(got.Files, want) {
593 t.Fatalf("got %+v / %v, want %+v", got, err, want)
594 }
595 })
596 }
597}
598
599// Port of resumeOpen's prepared branch (protocol_test.go:658) and
600// restorePinned/restoreWith (:183/:187), moves §5.1 / §7.14.
601func TestPreparedRestore(t *testing.T) {
602 tests := []struct {
603 name string
604 setup func(*fixture, *PreparedPin, File) []File
605 want error
606 ordinaryRefuses bool
607 }{
608 {"absent_target_marker_covered_source", func(f *fixture, p *PreparedPin, root File) []File {
609 f.origin()
610 f.req.Holder.Generation = "g2"
611 return []File{root, f.file("g2", 1, 29, 33)}
612 }, nil, true},
613 {"absent_target_marker_trimmed_source", func(f *fixture, p *PreparedPin, root File) []File {
614 f.origin()
615 f.req.Holder.Generation = "g2"
616 f.req.Holder.Admissions = f.req.Holder.Admissions[1:]
617 return []File{root}
618 }, nil, true},
619 {"unpublished_historical_pin", func(f *fixture, p *PreparedPin, root File) []File {
620 f.b.Put("data", root.Key, header(f.t, 1, 30))
621 return []File{root}
622 }, nil, false},
623 {"marker_pin_disagreement_ignored", func(f *fixture, p *PreparedPin, root File) []File {
624 f.marker.Op = "g2" // Its old g0 pin cannot override the authorized pin.
625 f.publish()
626 return []File{root}
627 }, nil, false},
628 {"malformed_marker_ignored", func(f *fixture, p *PreparedPin, root File) []File {
629 f.b.Put("data", f.req.Holder.Location+"/generation.json", []byte("{"))
630 return []File{root}
631 }, nil, false},
632 {"marker_tombstone_ignored", func(f *fixture, p *PreparedPin, root File) []File {
633 f.b.BareDelete("data", f.req.Holder.Location+"/generation.json")
634 return []File{root}
635 }, nil, false},
636 {"missing_prepared_root_no_current_key_fallback", func(f *fixture, p *PreparedPin, root File) []File {
637 f.b.Put("data", root.Key, header(f.t, 1, 30))
638 deletePreparedRoot(f, root)
639 return nil
640 }, ErrPinnedMissing, false},
641 {"empty_prepared_generation_never_older_baseline", func(f *fixture, p *PreparedPin, root File) []File {
642 f.chain("g0")
643 deletePreparedRoot(f, root)
644 versions, err := f.b.ListVersions(f.t.Context(), "data", prefix(f.req.Holder.Location+"/", "g2"))
645 if err != nil {
646 f.t.Fatal(err)
647 }
648 for _, v := range versions {
649 if err := f.b.DeleteVersion(f.t.Context(), "data", v.Key, v.VersionID); err != nil {
650 f.t.Fatal(err)
651 }
652 }
653 baseline, err := Select(f.t.Context(), f.b, f.req)
654 if err != nil || baseline.Generation != "g0" {
655 f.t.Fatalf("ordinary baseline: %+v / %v", baseline, err)
656 }
657 return nil
658 }, ErrPinnedMissing, false},
659 {"zero_txid_prepared_pin_refuses", func(f *fixture, p *PreparedPin, root File) []File {
660 *p = PreparedPin{Generation: "empty"}
661 return nil
662 }, ErrPinnedMismatch, false},
663 {"prepared_pin_key_mismatch", func(f *fixture, p *PreparedPin, root File) []File {
664 p.SnapshotTXID++
665 return nil
666 }, ErrPinnedMismatch, false},
667 {"prepared_pin_other_generation", func(f *fixture, p *PreparedPin, root File) []File {
668 p.SnapshotKey = f.pin.Key
669 p.SnapshotVersion = f.pin.VersionID
670 p.SnapshotTXID = f.pin.MaxTXID
671 return nil
672 }, ErrPinnedMismatch, false},
673 {"prepared_pin_header_mismatch", func(f *fixture, p *PreparedPin, root File) []File {
674 p.SnapshotVersion = f.b.Put("data", root.Key, header(f.t, 1, 29))
675 return nil
676 }, ErrPinnedMismatch, false},
677 {"prepared_pin_empty_version", func(f *fixture, p *PreparedPin, root File) []File {
678 p.SnapshotVersion = ""
679 return nil
680 }, ErrVersionID, false},
681 {"prepared_pin_deleted_version", func(f *fixture, p *PreparedPin, root File) []File {
682 p.SnapshotVersion = f.b.BareDelete("data", root.Key)
683 return nil
684 }, ErrDeleteMarker, false},
685 {"prepared_root_advance_past_missing_pin", func(f *fixture, p *PreparedPin, root File) []File {
686 deletePreparedRoot(f, root)
687 f.file("other", 9, 1, 999)
688 return []File{f.file("g2", 9, 1, 40), f.file("g2", 0, 41, 41)}
689 }, nil, false},
690 {"prepared_bad_successor_no_fallback", func(f *fixture, p *PreparedPin, root File) []File {
691 r := f.file("g2", 9, 1, 40)
692 f.b.Put("data", r.Key, header(f.t, 1, 39))
693 return nil
694 }, ErrRoot, false},
695 {"prepared_incomplete_listing", func(f *fixture, p *PreparedPin, root File) []File {
696 f.b.FailListAfterPages = 1
697 return nil
698 }, ErrListing, false},
699 }
700 for _, tt := range tests {
701 t.Run(tt.name, func(t *testing.T) {
702 f := newFixture(t)
703 f.req.Holder.Generation = "g2"
704 f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
705 {Generation: "source", Predecessor: "older", Region: "us", PredecessorRegion: "us", Mode: "cold", Covered: true},
706 {Generation: "g2", Predecessor: "source", Region: "ie", PredecessorRegion: "us", Mode: "cold"},
707 }
708 root := f.file("g2", 9, 1, 30)
709 f.file("g2", 0, 30, 30)
710 prepared := PreparedPin{Generation: "g2", SnapshotKey: root.Key, SnapshotVersion: root.VersionID, SnapshotTXID: 30, SnapshotCRC64: 123}
711 want := tt.setup(f, &prepared, root)
712 if tt.ordinaryRefuses {
713 got, err := Select(t.Context(), f.store, f.req)
714 if got != nil || !errors.Is(err, ErrDisagreement) {
715 t.Fatalf("ordinary selection: %+v / %v", got, err)
716 }
717 }
718 before := f.b.Snapshot()
719 holderBefore, _ := json.Marshal(f.req.Holder)
720 f.b.FailOn = func(op, key string) error {
721 if op != "list" && op != "get" {
722 t.Fatalf("mutation %s %s", op, key)
723 }
724 return nil
725 }
726 reader := preparedReader{Reader: f.store, t: t, base: prefix(f.req.Holder.Location+"/", prepared.Generation)}
727 got, err := SelectPrepared(t.Context(), reader, f.req, prepared)
728 holderAfter, _ := json.Marshal(f.req.Holder)
729 if f.b.Snapshot() != before || string(holderBefore) != string(holderAfter) {
730 t.Fatal("selection mutated inputs or storage")
731 }
732 if !errors.Is(err, tt.want) {
733 t.Fatalf("got %+v / %v, want %v", got, err, tt.want)
734 }
735 if tt.want != nil {
736 if got != nil {
737 t.Fatal("partial plan")
738 }
739 return
740 }
741 wantPin := Pin{SnapshotKey: prepared.SnapshotKey, SnapshotVersion: prepared.SnapshotVersion, SnapshotTXID: prepared.SnapshotTXID, SnapshotCRC64: prepared.SnapshotCRC64}
742 if got == nil || got.Generation != prepared.Generation || !reflect.DeepEqual(got.Files, want) ||
743 got.Root == nil || *got.Root != want[0] || got.MaxTXID != want[len(want)-1].MaxTXID ||
744 got.Marker != nil || got.MarkerVersion != "" || got.Pin == nil || *got.Pin != wantPin {
745 t.Fatalf("got %+v, want prepared %s %+v", got, prepared.Generation, want)
746 }
747 })
748 }
749}
750
751func deletePreparedRoot(f *fixture, root File) {
752 f.t.Helper()
753 if err := f.b.DeleteVersion(f.t.Context(), "data", root.Key, root.VersionID); err != nil {
754 f.t.Fatal(err)
755 }
756}
757
758type preparedReader struct {
759 Reader
760 t *testing.T
761 base string
762}
763
764func (s preparedReader) ListVersions(ctx context.Context, bucket, base string) ([]lease.ObjectVersion, error) {
765 if base != s.base {
766 s.t.Fatalf("prepared listing escaped generation: %s", base)
767 }
768 return s.Reader.ListVersions(ctx, bucket, base)
769}
770func (s preparedReader) GetVersion(context.Context, string, string, string) ([]byte, error) {
771 s.t.Fatal("prepared restore must not read a marker")
772 return nil, ErrMarker
773}
774func (s preparedReader) OpenVersion(ctx context.Context, bucket, key, id string) (io.ReadCloser, error) {
775 if !strings.HasPrefix(key, s.base) {
776 s.t.Fatalf("prepared open escaped generation: %s", key)
777 }
778 return s.Reader.OpenVersion(ctx, bucket, key, id)
779}
780