| 1 | package lease |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "database/sql" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "net/url" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | lsfile "github.com/benbjohnson/litestream/file" |
| 16 | "modernc.org/sqlite" |
| 17 | ) |
| 18 | |
| 19 | func TestCheckerClosedCheckpointFinalizationProbe(t *testing.T) { |
| 20 | backend := &lockStore{} |
| 21 | h := localHeld(t, backend, 30*time.Second) |
| 22 | rn := h.rn |
| 23 | ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) |
| 24 | defer cancel() |
| 25 | path := filepath.Join(t.TempDir(), "source.db") |
| 26 | seedDB(t, path) |
| 27 | replica := t.TempDir() |
| 28 | h.replicaClient = lsfile.NewReplicaClient(replica) |
| 29 | receipt, err := h.Checkpoint(ctx, path) |
| 30 | if err != nil { |
| 31 | t.Fatal(err) |
| 32 | } |
| 33 | if receipt.CompletedAt.IsZero() || rn != h.rn { |
| 34 | t.Fatal("checkpoint ownership") |
| 35 | } |
| 36 | for _, suffix := range []string{"-wal", "-shm"} { |
| 37 | s, e := os.Stat(path + suffix) |
| 38 | if e != nil { |
| 39 | t.Fatalf("expected PERSIST_WAL sidecar%s: %v", suffix, e) |
| 40 | } |
| 41 | t.Logf("after Litestream Close%s size=%d", suffix, s.Size()) |
| 42 | } |
| 43 | u := url.URL{Scheme: "file", Path: path} |
| 44 | q := u.Query() |
| 45 | q.Set("mode", "rw") |
| 46 | u.RawQuery = q.Encode() |
| 47 | d, err := sql.Open("sqlite", u.String()) |
| 48 | if err != nil { |
| 49 | t.Fatal(err) |
| 50 | } |
| 51 | d.SetMaxOpenConns(1) |
| 52 | d.SetMaxIdleConns(1) |
| 53 | conn, err := d.Conn(ctx) |
| 54 | if err != nil { |
| 55 | t.Fatal(err) |
| 56 | } |
| 57 | if err = conn.Raw(func(raw any) error { |
| 58 | f, ok := raw.(sqlite.FileControl) |
| 59 | if !ok { |
| 60 | return fmt.Errorf("no file control") |
| 61 | } |
| 62 | before, e := f.FileControlPersistWAL("main", -1) |
| 63 | if e != nil { |
| 64 | return e |
| 65 | } |
| 66 | t.Logf("ordinary PERSIST_WAL before explicit clear=%d", before) |
| 67 | after, e := f.FileControlPersistWAL("main", 0) |
| 68 | if e != nil { |
| 69 | return e |
| 70 | } |
| 71 | if after != 0 { |
| 72 | return fmt.Errorf("persist remains%d", after) |
| 73 | } |
| 74 | return nil |
| 75 | }); err != nil { |
| 76 | t.Fatal(err) |
| 77 | } |
| 78 | var integrity string |
| 79 | if err = conn.QueryRowContext(ctx, "PRAGMA integrity_check").Scan(&integrity); err != nil || integrity != "ok" { |
| 80 | t.Fatalf("integrity=%q err=%v", integrity, err) |
| 81 | } |
| 82 | var busy, log, done int |
| 83 | if err = conn.QueryRowContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)").Scan(&busy, &log, &done); err != nil || busy != 0 || log != 0 || done != 0 { |
| 84 | t.Fatalf("checkpoint %d %d %d err=%v", busy, log, done, err) |
| 85 | } |
| 86 | if err = conn.Close(); err != nil { |
| 87 | t.Fatal(err) |
| 88 | } |
| 89 | if err = d.Close(); err != nil { |
| 90 | t.Fatal(err) |
| 91 | } |
| 92 | absent := func() { |
| 93 | t.Helper() |
| 94 | for _, suffix := range []string{"-wal", "-shm", "-journal"} { |
| 95 | if _, e := os.Lstat(path + suffix); !os.IsNotExist(e) { |
| 96 | t.Fatalf("sidecar remains%s err=%v", suffix, e) |
| 97 | } |
| 98 | } |
| 99 | } |
| 100 | absent() |
| 101 | if rn != h.rn { |
| 102 | t.Fatal("renewer changed") |
| 103 | } |
| 104 | if _, err = h.SafeUntil(); err != nil { |
| 105 | t.Fatal(err) |
| 106 | } |
| 107 | if _, deletes := backend.counts(); deletes != 0 { |
| 108 | t.Fatal("lease released") |
| 109 | } |
| 110 | if peer, e := acquireTTL(ctx, "test-bucket", "one.example", "test-region", 30*time.Second); e == nil { |
| 111 | peer.Release(ctx) |
| 112 | t.Fatal("peer acquired") |
| 113 | } |
| 114 | pin, err := os.Open(path) |
| 115 | if err != nil { |
| 116 | t.Fatal(err) |
| 117 | } |
| 118 | defer pin.Close() |
| 119 | beforeInfo, err := pin.Stat() |
| 120 | if err != nil { |
| 121 | t.Fatal(err) |
| 122 | } |
| 123 | hash := func() [32]byte { |
| 124 | t.Helper() |
| 125 | if _, e := pin.Seek(0, 0); e != nil { |
| 126 | t.Fatal(e) |
| 127 | } |
| 128 | h := sha256.New() |
| 129 | if _, e := io.Copy(h, pin); e != nil { |
| 130 | t.Fatal(e) |
| 131 | } |
| 132 | var out [32]byte |
| 133 | copy(out[:], h.Sum(nil)) |
| 134 | return out |
| 135 | } |
| 136 | before := hash() |
| 137 | q.Set("mode", "ro") |
| 138 | q.Set("immutable", "1") |
| 139 | u.RawQuery = q.Encode() |
| 140 | reader, err := sql.Open("sqlite", u.String()) |
| 141 | if err != nil { |
| 142 | t.Fatal(err) |
| 143 | } |
| 144 | reader.SetMaxOpenConns(1) |
| 145 | var count int |
| 146 | if err = reader.QueryRowContext(ctx, "SELECT count(*) FROM t").Scan(&count); err != nil || count != 50 { |
| 147 | t.Fatalf("immutable rows=%d err=%v", count, err) |
| 148 | } |
| 149 | if err = reader.Close(); err != nil { |
| 150 | t.Fatal(err) |
| 151 | } |
| 152 | after := hash() |
| 153 | afterInfo, err := pin.Stat() |
| 154 | if err != nil { |
| 155 | t.Fatal(err) |
| 156 | } |
| 157 | if before != after || !os.SameFile(beforeInfo, afterInfo) { |
| 158 | t.Fatal("immutable check changed main") |
| 159 | } |
| 160 | absent() |
| 161 | if count := restoreCount(t, replica, filepath.Join(t.TempDir(), "restored.db")); count != 50 { |
| 162 | t.Fatalf("replica rows=%d", count) |
| 163 | } |
| 164 | if _, err = h.SafeUntil(); err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | if _, deletes := backend.counts(); deletes != 0 { |
| 168 | t.Fatal("prep/check released lease") |
| 169 | } |
| 170 | t.Logf("PASS same Held, SQLite cleanup without unlink, immutable read preserves main SHA256=%x and absent sidecars; restored rows=50", after) |
| 171 | } |
| 172 | |
amadan