amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package hostbox
2
3import (
4 "bytes"
5 "context"
6 "database/sql"
7 "encoding/binary"
8 "errors"
9 "fmt"
10 "io"
11 "os"
12 "path/filepath"
13 "strings"
14 "testing"
15
16 "github.com/carlosframework/platform/internal/genselect"
17 "github.com/carlosframework/platform/internal/hostclaim"
18 "github.com/carlosframework/platform/internal/hostwrite"
19 "github.com/carlosframework/platform/internal/lease"
20 "github.com/carlosframework/platform/internal/lease/leasetest"
21 "github.com/superfly/ltx"
22 _ "modernc.org/sqlite"
23)
24
25type restoreReader struct {
26 data map[string][]byte
27 openedKey string
28 openedVers string
29}
30
31func (*restoreReader) ListVersions(context.Context, string, string) ([]lease.ObjectVersion, error) {
32 return nil, errors.New("unexpected listing")
33}
34func (*restoreReader) GetVersion(context.Context, string, string, string) ([]byte, error) {
35 return nil, errors.New("unexpected get")
36}
37func (r *restoreReader) OpenVersion(_ context.Context, _ string, key, version string) (io.ReadCloser, error) {
38 r.openedKey, r.openedVers = key, version
39 b, ok := r.data[key+"@"+version]
40 if !ok {
41 return nil, errors.New("missing immutable version")
42 }
43 return io.NopCloser(bytes.NewReader(b)), nil
44}
45
46func TestLoadPinnedReadsTheSelectedVersionAndRejectsCorruption(t *testing.T) {
47 r := &restoreReader{data: map[string][]byte{
48 "instances/shop/g-one/0009/0000000000000001-0000000000000001.ltx@selected": []byte("corrupt selected object"),
49 "instances/shop/g-one/0009/0000000000000001-0000000000000001.ltx@current": []byte("different current object"),
50 }}
51 p := &genselect.Plan{Generation: "one", MaxTXID: 1, Files: []genselect.File{{
52 Key: "instances/shop/g-one/0009/0000000000000001-0000000000000001.ltx", VersionID: "selected", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len("corrupt selected object")),
53 }}}
54 if _, _, err := loadPinned(context.Background(), r, "bucket", p); err == nil {
55 t.Fatal("loadPinned accepted an object without a valid complete LTX trailer")
56 }
57 if r.openedVers != "selected" {
58 t.Fatalf("opened version %q, want immutable selected version", r.openedVers)
59 }
60}
61
62type restoreWork struct {
63 calls int
64 active bool
65}
66
67func (w *restoreWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
68 w.calls++
69 w.active = true
70 defer func() { w.active = false }()
71 return fn()
72}
73
74func TestColdRestoreRefusesBeforeMutationWithoutSelectedAuthority(t *testing.T) {
75 r := ColdRestore{}
76 if err := r.Install(context.Background(), t.TempDir()+"/db", &genselect.Plan{}); !errors.Is(err, ErrUnsupported) {
77 t.Fatalf("Install error = %v, want ErrUnsupported", err)
78 }
79}
80
81func TestColdRestorePurgesSameGenerationUnderSelectedWork(t *testing.T) {
82 dir := t.TempDir()
83 db := filepath.Join(dir, "shop.db")
84 if err := os.WriteFile(db, []byte("warm database must not be reused"), 0o600); err != nil {
85 t.Fatal(err)
86 }
87 if err := os.WriteFile(db+".gen", []byte("one\n"), 0o600); err != nil {
88 t.Fatal(err)
89 }
90 key, snapshot, crc := snapshotFixture(t, "selected")
91 rd := &restoreReader{data: map[string][]byte{key + "@v1": snapshot}}
92 work := &restoreWork{}
93 err := (ColdRestore{Reader: rd, Bucket: "bucket", Generation: "one", Work: work}).Install(context.Background(), db,
94 &genselect.Plan{Generation: "one", MaxTXID: 1, Pin: &genselect.Pin{SnapshotCRC64: crc}, Files: []genselect.File{{Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len(snapshot))}}})
95 if err != nil {
96 t.Fatalf("Install: %v", err)
97 }
98 if work.calls != 2 {
99 t.Fatalf("selected work calls = %d, want capture and install", work.calls)
100 }
101 dbcheck, err := sql.Open("sqlite", db)
102 if err != nil {
103 t.Fatal(err)
104 }
105 defer dbcheck.Close()
106 var got string
107 if err := dbcheck.QueryRow(`SELECT value FROM fixture`).Scan(&got); err != nil || got != "selected" {
108 t.Fatalf("restored value=%q err=%v", got, err)
109 }
110 if string(mustRead(t, db+".gen")) != "one\n" {
111 t.Fatal("generation marker not installed")
112 }
113}
114
115func TestColdRestorePinsVersionAndRetiresForeignGenerationOutsideWork(t *testing.T) {
116 dir := t.TempDir()
117 db := filepath.Join(dir, "shop.db")
118 if err := os.WriteFile(db, []byte("foreign"), 0o600); err != nil {
119 t.Fatal(err)
120 }
121 if err := os.WriteFile(db+".gen", []byte("old\n"), 0o600); err != nil {
122 t.Fatal(err)
123 }
124 key, v1, crc := snapshotFixture(t, "selected-v1")
125 _, v2, _ := snapshotFixture(t, "current-v2")
126 rd := &restoreReader{data: map[string][]byte{key + "@v1": v1, key + "@v2": v2}}
127 work := &restoreWork{}
128 retired := ""
129 parked := filepath.Join(dir, "parked")
130 err := (ColdRestore{Reader: rd, Bucket: "bucket", Generation: "new", Work: work, RetireArtifacts: func(_ context.Context, g string) error {
131 if work.active {
132 t.Fatal("retirement ran under Work flock")
133 }
134 retired = g
135 if err := os.Mkdir(parked, 0o700); err != nil {
136 return err
137 }
138 if err := os.Rename(db, filepath.Join(parked, "shop.db")); err != nil {
139 return err
140 }
141 return os.Rename(db+".gen", filepath.Join(parked, "shop.db.gen"))
142 }}).Install(context.Background(), db,
143 &genselect.Plan{Generation: "one", MaxTXID: 1, Pin: &genselect.Pin{SnapshotCRC64: crc}, Files: []genselect.File{{Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len(v1))}}})
144 if err != nil {
145 t.Fatal(err)
146 }
147 if retired != "old" {
148 t.Fatalf("retired %q, want old", retired)
149 }
150 if rd.openedVers != "v1" {
151 t.Fatalf("opened %q, want pinned v1", rd.openedVers)
152 }
153 check, err := sql.Open("sqlite", db)
154 if err != nil {
155 t.Fatal(err)
156 }
157 defer check.Close()
158 var got string
159 if err := check.QueryRow(`SELECT value FROM fixture`).Scan(&got); err != nil || got != "selected-v1" {
160 t.Fatalf("restored value=%q err=%v", got, err)
161 }
162 if string(mustRead(t, db+".gen")) != "new\n" {
163 t.Fatalf("installed marker=%q, want admitted generation", mustRead(t, db+".gen"))
164 }
165}
166
167func TestColdRestoreReportsPurgeFailureBeforeInstall(t *testing.T) {
168 dir := t.TempDir()
169 db := filepath.Join(dir, "shop.db")
170 if err := os.Mkdir(db, 0o700); err != nil {
171 t.Fatal(err)
172 }
173 if err := os.WriteFile(filepath.Join(db, "still-here"), []byte("x"), 0o600); err != nil {
174 t.Fatal(err)
175 }
176 key, snapshot, _ := snapshotFixture(t, "never-installed")
177 rd := &restoreReader{data: map[string][]byte{key + "@v1": snapshot}}
178 err := (ColdRestore{Reader: rd, Bucket: "bucket", Generation: "one", Work: &restoreWork{}}).Install(context.Background(), db,
179 &genselect.Plan{Generation: "one", MaxTXID: 1, Files: []genselect.File{{Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len(snapshot))}}})
180 if err == nil || !strings.Contains(err.Error(), "purge "+db) {
181 t.Fatalf("Install error=%v, want checked purge failure", err)
182 }
183 if _, err := os.Stat(filepath.Join(db, "still-here")); err != nil {
184 t.Fatalf("purge failure damaged existing artifact: %v", err)
185 }
186}
187
188func TestColdRestoreInstallEmptyRequiresProofAndWritesOnlyGeneration(t *testing.T) {
189 dir := t.TempDir()
190 db := filepath.Join(dir, "empty.db")
191 work := &restoreWork{}
192 proved := false
193 r := ColdRestore{Generation: "new", Work: work}
194 if err := r.InstallEmpty(context.Background(), db, func(context.Context) error { proved = true; return nil }); err != nil {
195 t.Fatal(err)
196 }
197 if !proved || work.calls != 2 {
198 t.Fatalf("proved=%v guarded calls=%d", proved, work.calls)
199 }
200 if _, err := os.Stat(db); !errors.Is(err, os.ErrNotExist) {
201 t.Fatalf("empty bootstrap wrote DB: %v", err)
202 }
203 if string(mustRead(t, db+".gen")) != "new\n" {
204 t.Fatal("admitted generation missing")
205 }
206}
207
208func TestColdRestoreAcceptsSelectorOverlappingOutput(t *testing.T) {
209 for _, min := range []ltx.TXID{10, 12} {
210 t.Run(fmt.Sprintf("continuation-%d-13", min), func(t *testing.T) {
211 ctx := t.Context()
212 bucket := leasetest.NewVersionedBucket()
213 bucket.AddBucket("data")
214 bucket.PageSize = 1
215 key, snapshot, crc := restoreLTXFixture(t, "snapshot-at-11", 9, 1, 11)
216 version := bucket.Put("data", key, snapshot)
217 continuation, data, _ := restoreLTXFixture(t, "compacted-at-13", 1, min, 13)
218 bucket.Put("data", continuation, data)
219 plan, err := genselect.SelectPrepared(ctx, bucket, genselect.Request{
220 Bucket: "data", Region: "ie", Host: "shop",
221 Holder: hostclaim.Holder{Incarnation: "inc", Location: "instances/shop"},
222 }, genselect.PreparedPin{Generation: "one", SnapshotKey: key, SnapshotVersion: version, SnapshotTXID: 11, SnapshotCRC64: crc})
223 if err != nil {
224 t.Fatal(err)
225 }
226 if len(plan.Files) != 2 || plan.Files[0].MinTXID != 1 || plan.Files[0].MaxTXID != 11 || plan.Files[1].MinTXID != uint64(min) || plan.Files[1].MaxTXID != 13 || plan.MaxTXID != 13 {
227 t.Fatalf("selector output = %+v", plan)
228 }
229 path := filepath.Join(t.TempDir(), "restored.db")
230 if err := (ColdRestore{Reader: bucket, Bucket: "data", Generation: "new", Work: &restoreWork{}}).Install(ctx, path, plan); err != nil {
231 t.Fatalf("restore selector output: %v", err)
232 }
233 db, err := sql.Open("sqlite", path)
234 if err != nil {
235 t.Fatal(err)
236 }
237 defer db.Close()
238 var got string
239 if err := db.QueryRow(`SELECT value FROM fixture`).Scan(&got); err != nil || got != "compacted-at-13" {
240 t.Fatalf("restored value = %q, err = %v", got, err)
241 }
242 if string(mustRead(t, path+".gen")) != "new\n" {
243 t.Fatal("admitted generation marker missing")
244 }
245 })
246 }
247}
248
249func TestLoadPinnedRejectsGapAndNonadvancingRange(t *testing.T) {
250 for _, bounds := range [][2]ltx.TXID{{13, 13}, {10, 11}} {
251 t.Run(fmt.Sprint(bounds), func(t *testing.T) {
252 key, snapshot, _ := restoreLTXFixture(t, "snapshot", 9, 1, 11)
253 continuation, data, _ := restoreLTXFixture(t, "invalid", 1, bounds[0], bounds[1])
254 reader := &restoreReader{data: map[string][]byte{key + "@v1": snapshot, continuation + "@v2": data}}
255 plan := &genselect.Plan{Generation: "one", MaxTXID: uint64(bounds[1]), Files: []genselect.File{
256 {Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 11, Size: int64(len(snapshot))},
257 {Key: continuation, VersionID: "v2", Level: 1, MinTXID: uint64(bounds[0]), MaxTXID: uint64(bounds[1]), Size: int64(len(data))},
258 }}
259 if _, _, err := loadPinned(t.Context(), reader, "data", plan); !errors.Is(err, genselect.ErrGap) {
260 t.Fatalf("invalid continuation accepted: %v", err)
261 }
262 })
263 }
264}
265
266func snapshotFixture(t *testing.T, value string) (string, []byte, uint64) {
267 t.Helper()
268 return restoreLTXFixture(t, value, 9, 1, 1)
269}
270
271func restoreLTXFixture(t *testing.T, value string, level int, min, max ltx.TXID) (string, []byte, uint64) {
272 t.Helper()
273 path := filepath.Join(t.TempDir(), "fixture.db")
274 db, err := sql.Open("sqlite", path)
275 if err != nil {
276 t.Fatal(err)
277 }
278 if _, err = db.Exec(`CREATE TABLE fixture(value TEXT); INSERT INTO fixture VALUES (?)`, value); err != nil {
279 t.Fatal(err)
280 }
281 if err = db.Close(); err != nil {
282 t.Fatal(err)
283 }
284 return ltxFromDBFile(t, path, level, min, max)
285}
286
287// ltxFromDBFile cuts the database file at path into one LTX object and
288// returns its key, bytes and file checksum.
289func ltxFromDBFile(t *testing.T, path string, level int, min, max ltx.TXID) (string, []byte, uint64) {
290 t.Helper()
291 raw := mustRead(t, path)
292 pageSize := int(binary.BigEndian.Uint16(raw[16:18]))
293 if pageSize == 1 {
294 pageSize = 65536
295 }
296 if len(raw)%pageSize != 0 {
297 t.Fatalf("database size %d not page aligned", len(raw))
298 }
299 spec := ltx.FileSpec{Header: ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: uint32(pageSize), Commit: uint32(len(raw) / pageSize), MinTXID: min, MaxTXID: max}}
300 for off := 0; off < len(raw); off += pageSize {
301 pgno := uint32(off/pageSize + 1)
302 if pgno == ltx.LockPgno(uint32(pageSize)) {
303 continue
304 }
305 spec.Pages = append(spec.Pages, ltx.PageSpec{Header: ltx.PageHeader{Pgno: pgno}, Data: append([]byte(nil), raw[off:off+pageSize]...)})
306 }
307 var out bytes.Buffer
308 if _, err := spec.WriteTo(&out); err != nil {
309 t.Fatal(err)
310 }
311 return fmt.Sprintf("instances/shop/g-one/%04x/%s", level, ltx.FormatFilename(min, max)), out.Bytes(), uint64(spec.Trailer.FileChecksum)
312}
313
314func mustRead(t *testing.T, path string) []byte {
315 t.Helper()
316 b, err := os.ReadFile(path)
317 if err != nil {
318 t.Fatal(err)
319 }
320 return b
321}
322
323// The landing writes <dbPath>.gen through the one marker writer this
324// package's cold restore reads (moves Plan 2 ruling G2): the prepared
325// generation's name as plain text, so a later admission's readGeneration
326// accepts it. A name the restore would refuse is refused on the way in.
327func TestGenerationMarkerIsSharedWithTheColdRestore(t *testing.T) {
328 path := filepath.Join(t.TempDir(), "h.db.gen")
329 if err := WriteGenerationMarker(path, "mv_abc"); err != nil {
330 t.Fatal(err)
331 }
332 for _, read := range []func(string) (string, error){ReadGenerationMarker, readGeneration} {
333 if got, err := read(path); err != nil || got != "mv_abc" {
334 t.Fatalf("read back %q, %v", got, err)
335 }
336 }
337 if got, err := ReadGenerationMarker(filepath.Join(t.TempDir(), "absent.gen")); err != nil || got != "" {
338 t.Fatalf("absent: %q, %v", got, err)
339 }
340 for _, bad := range []string{"", ".", "..", "instances/h", "a\nb"} {
341 if err := WriteGenerationMarker(path, bad); err == nil {
342 t.Fatalf("%q was written", bad)
343 }
344 }
345 if got, _ := ReadGenerationMarker(path); got != "mv_abc" {
346 t.Fatalf("a refused write changed the marker: %q", got)
347 }
348}
349
350func TestLoadPinnedNormalizesOnlyAfterOriginalProof(t *testing.T) {
351 key, original, _ := snapshotFixture(t, "checksummed")
352 var spec ltx.FileSpec
353 if _, err := spec.ReadFrom(bytes.NewReader(original)); err != nil {
354 t.Fatal(err)
355 }
356 checksum := ltx.Checksum(0)
357 for _, page := range spec.Pages {
358 checksum ^= ltx.ChecksumPage(page.Header.Pgno, page.Data)
359 }
360 checksum |= ltx.ChecksumFlag
361 spec.Header.Flags = 0
362 spec.Trailer.PostApplyChecksum = checksum
363 var encoded bytes.Buffer
364 if _, err := spec.WriteTo(&encoded); err != nil {
365 t.Fatal(err)
366 }
367 original = append([]byte(nil), encoded.Bytes()...)
368 for _, fault := range []string{"valid", "pin", "file-checksum", "database-checksum"} {
369 t.Run(fault, func(t *testing.T) {
370 body := append([]byte(nil), original...)
371 pin := uint64(spec.Trailer.FileChecksum)
372 switch fault {
373 case "pin":
374 pin ^= 1
375 case "file-checksum":
376 body[len(body)-1] ^= 1
377 case "database-checksum":
378 bad := spec
379 bad.Trailer.PostApplyChecksum ^= 1
380 var out bytes.Buffer
381 if _, err := bad.WriteTo(&out); err != nil {
382 t.Fatal(err)
383 }
384 body, pin = out.Bytes(), uint64(bad.Trailer.FileChecksum)
385 }
386 reader := &restoreReader{data: map[string][]byte{key + "@selected": body}}
387 plan := &genselect.Plan{Generation: "one", MaxTXID: 1, Pin: &genselect.Pin{SnapshotCRC64: pin}, Files: []genselect.File{{Key: key, VersionID: "selected", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len(body))}}}
388 files, final, err := loadPinned(t.Context(), reader, "data", plan)
389 if fault != "valid" {
390 if err == nil {
391 t.Fatal("normalization bypassed original proof")
392 }
393 return
394 }
395 if err != nil {
396 t.Fatal(err)
397 }
398 if final != checksum || len(files) != 1 || !bytes.Equal(reader.data[key+"@selected"], original) {
399 t.Fatal("normalization changed selected bytes or lost database checksum")
400 }
401 var converted ltx.FileSpec
402 if _, err := converted.ReadFrom(bytes.NewReader(files[0].data)); err != nil {
403 t.Fatal(err)
404 }
405 if !converted.Header.NoChecksum() || converted.Header.PreApplyChecksum != 0 || converted.Trailer.PostApplyChecksum != 0 {
406 t.Fatal("incompatible restore stream")
407 }
408 if len(converted.Pages) != len(spec.Pages) {
409 t.Fatal("page loss")
410 }
411 for i := range converted.Pages {
412 if converted.Pages[i].Header != spec.Pages[i].Header || !bytes.Equal(converted.Pages[i].Data, spec.Pages[i].Data) {
413 t.Fatal("normalization changed database pages")
414 }
415 }
416 path := filepath.Join(t.TempDir(), "restored.db")
417 if err := (ColdRestore{Reader: reader, Bucket: "data", Generation: "successor", Work: &restoreWork{}}).Install(t.Context(), path, plan); err != nil {
418 t.Fatal(err)
419 }
420 })
421 }
422}
423