| 1 | package dbdoor |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "io" |
| 9 | "net/http" |
| 10 | "net/http/httptest" |
| 11 | "os" |
| 12 | "path/filepath" |
| 13 | "strconv" |
| 14 | "strings" |
| 15 | "sync" |
| 16 | "testing" |
| 17 | "time" |
| 18 | |
| 19 | "github.com/benbjohnson/litestream" |
| 20 | lsfile "github.com/benbjohnson/litestream/file" |
| 21 | "github.com/superfly/ltx" |
| 22 | |
| 23 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 24 | ) |
| 25 | |
| 26 | var _ litestream.ReplicaClient = (*Client)(nil) |
| 27 | |
| 28 | // noNetwork fails the test if the client ever reaches for the wire. |
| 29 | func noNetwork(t *testing.T) Doer { |
| 30 | return func(context.Context, string, http.Header) (*http.Response, error) { |
| 31 | t.Error("the client made a request") |
| 32 | return nil, errors.New("no network in this test") |
| 33 | } |
| 34 | } |
| 35 | |
| 36 | // The claim the whole design rests on: this client cannot corrupt a |
| 37 | // replica, because it has no code that writes to one. And it refuses |
| 38 | // without touching the network. |
| 39 | func TestClientMutatingMethodsAlwaysError(t *testing.T) { |
| 40 | c := NewClient(noNetwork(t)) |
| 41 | ctx := context.Background() |
| 42 | if _, err := c.WriteLTXFile(ctx, 0, 1, 2, strings.NewReader("x")); err == nil { |
| 43 | t.Error("WriteLTXFile returned nil; this client must never write") |
| 44 | } |
| 45 | if err := c.DeleteLTXFiles(ctx, []*ltx.FileInfo{{Level: 0, MinTXID: 1, MaxTXID: 1}}); err == nil { |
| 46 | t.Error("DeleteLTXFiles returned nil; this client must never delete") |
| 47 | } |
| 48 | if err := c.DeleteAll(ctx); err == nil { |
| 49 | t.Error("DeleteAll returned nil; this client must never delete") |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | // Restore type-asserts for ReplicaClientV3 and would take the v0.3 path if |
| 54 | // this satisfied it. It must not. |
| 55 | func TestClientIsNotAV3ReplicaClient(t *testing.T) { |
| 56 | var c any = NewClient(noNetwork(t)) |
| 57 | if _, ok := c.(litestream.ReplicaClientV3); ok { |
| 58 | t.Fatal("client satisfies ReplicaClientV3; Restore would take the v0.3 path") |
| 59 | } |
| 60 | } |
| 61 | |
| 62 | // testDoor serves a file replica over the door's wire shape: the same |
| 63 | // coordinates, the same Range forms, the same 404 for a missing file. It |
| 64 | // stands in for the console here; internal/console tests the real door |
| 65 | // with this package's client. |
| 66 | type testDoor struct { |
| 67 | t *testing.T |
| 68 | rc *lsfile.ReplicaClient |
| 69 | mu sync.Mutex |
| 70 | lists map[int]int // level -> listing count |
| 71 | gone map[string]bool |
| 72 | opened int |
| 73 | } |
| 74 | |
| 75 | func newTestDoor(t *testing.T, dir string) *testDoor { |
| 76 | return &testDoor{t: t, rc: lsfile.NewReplicaClient(dir), lists: map[int]int{}, gone: map[string]bool{}} |
| 77 | } |
| 78 | |
| 79 | func (d *testDoor) do(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 80 | rec := httptest.NewRecorder() |
| 81 | d.serve(ctx, rec, tail, h) |
| 82 | return rec.Result(), nil |
| 83 | } |
| 84 | |
| 85 | func (d *testDoor) serve(ctx context.Context, w http.ResponseWriter, tail string, h http.Header) { |
| 86 | parts := strings.Split(strings.TrimPrefix(tail, "/ltx/"), "/") |
| 87 | level, err := strconv.Atoi(parts[0]) |
| 88 | if err != nil { |
| 89 | http.NotFound(w, nil) |
| 90 | return |
| 91 | } |
| 92 | if len(parts) == 1 { |
| 93 | d.mu.Lock() |
| 94 | d.lists[level]++ |
| 95 | d.mu.Unlock() |
| 96 | itr, _ := d.rc.LTXFiles(ctx, level, 0, false) |
| 97 | out := ListResponse{Files: []LTXFile{}} |
| 98 | for itr.Next() { |
| 99 | f := itr.Item() |
| 100 | out.Files = append(out.Files, LTXFile{Level: level, MinTXID: uint64(f.MinTXID), MaxTXID: uint64(f.MaxTXID), Size: f.Size, CreatedAt: f.CreatedAt}) |
| 101 | } |
| 102 | itr.Close() |
| 103 | json.NewEncoder(w).Encode(out) |
| 104 | return |
| 105 | } |
| 106 | min, max, err := ParseSpan(parts[1]) |
| 107 | if err != nil { |
| 108 | http.NotFound(w, nil) |
| 109 | return |
| 110 | } |
| 111 | d.mu.Lock() |
| 112 | gone := d.gone[parts[1]] |
| 113 | if gone { |
| 114 | delete(d.gone, parts[1]) // gone once: a fresh plan finds it again |
| 115 | } |
| 116 | d.opened++ |
| 117 | d.mu.Unlock() |
| 118 | if gone { |
| 119 | http.NotFound(w, nil) |
| 120 | return |
| 121 | } |
| 122 | off, size, err := ParseRange(h.Get("Range")) |
| 123 | if err != nil { |
| 124 | w.WriteHeader(http.StatusRequestedRangeNotSatisfiable) |
| 125 | return |
| 126 | } |
| 127 | rc, err := d.rc.OpenLTXFile(ctx, level, ltx.TXID(min), ltx.TXID(max), off, size) |
| 128 | if errors.Is(err, os.ErrNotExist) { |
| 129 | http.NotFound(w, nil) |
| 130 | return |
| 131 | } |
| 132 | if err != nil { |
| 133 | w.WriteHeader(http.StatusInternalServerError) |
| 134 | return |
| 135 | } |
| 136 | defer rc.Close() |
| 137 | w.WriteHeader(http.StatusPartialContent) |
| 138 | io.Copy(w, rc) |
| 139 | } |
| 140 | |
| 141 | // counting wraps the client and counts every method litestream calls on it. |
| 142 | type counting struct { |
| 143 | *Client |
| 144 | mu sync.Mutex |
| 145 | counts map[string]int |
| 146 | } |
| 147 | |
| 148 | func (c *counting) inc(m string) { |
| 149 | c.mu.Lock() |
| 150 | defer c.mu.Unlock() |
| 151 | c.counts[m]++ |
| 152 | } |
| 153 | |
| 154 | func (c *counting) WriteLTXFile(ctx context.Context, level int, min, max ltx.TXID, r io.Reader) (*ltx.FileInfo, error) { |
| 155 | c.inc("WriteLTXFile") |
| 156 | return c.Client.WriteLTXFile(ctx, level, min, max, r) |
| 157 | } |
| 158 | |
| 159 | func (c *counting) DeleteLTXFiles(ctx context.Context, a []*ltx.FileInfo) error { |
| 160 | c.inc("DeleteLTXFiles") |
| 161 | return c.Client.DeleteLTXFiles(ctx, a) |
| 162 | } |
| 163 | |
| 164 | func (c *counting) DeleteAll(ctx context.Context) error { |
| 165 | c.inc("DeleteAll") |
| 166 | return c.Client.DeleteAll(ctx) |
| 167 | } |
| 168 | |
| 169 | func (c *counting) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) { |
| 170 | c.inc("LTXFiles") |
| 171 | return c.Client.LTXFiles(ctx, level, seek, useMetadata) |
| 172 | } |
| 173 | |
| 174 | func (c *counting) OpenLTXFile(ctx context.Context, level int, min, max ltx.TXID, offset, size int64) (io.ReadCloser, error) { |
| 175 | c.inc("OpenLTXFile") |
| 176 | return c.Client.OpenLTXFile(ctx, level, min, max, offset, size) |
| 177 | } |
| 178 | |
| 179 | // The methods erroring is not enough: a restore that CALLED one would be |
| 180 | // relying on an error path. This asserts a full restore never reaches them, |
| 181 | // which is what catches a litestream upgrade changing the restore path. |
| 182 | func TestRestoreNeverCallsAMutatingMethod(t *testing.T) { |
| 183 | dir := t.TempDir() |
| 184 | dbdoortest.Replicate(t, dir, 3) |
| 185 | door := newTestDoor(t, dir) |
| 186 | c := &counting{Client: NewClient(door.do), counts: map[string]int{}} |
| 187 | out := filepath.Join(t.TempDir(), "out.db") |
| 188 | opt := litestream.NewRestoreOptions() |
| 189 | opt.OutputPath = out |
| 190 | if err := litestream.NewReplicaWithClient(nil, c).Restore(context.Background(), opt); err != nil { |
| 191 | t.Fatal(err) |
| 192 | } |
| 193 | if n := c.counts["WriteLTXFile"] + c.counts["DeleteLTXFiles"] + c.counts["DeleteAll"]; n != 0 { |
| 194 | t.Fatalf("restore called a mutating method: %v", c.counts) |
| 195 | } |
| 196 | if c.counts["LTXFiles"] == 0 || c.counts["OpenLTXFile"] == 0 { |
| 197 | t.Fatalf("restore did not go through the client at all: %v", c.counts) |
| 198 | } |
| 199 | if rows := countRows(t, out); rows != 3*dbdoortest.Rows { |
| 200 | t.Fatalf("restored %d rows, want %d", rows, 3*dbdoortest.Rows) |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | func countRows(t *testing.T, path string) int { |
| 205 | t.Helper() |
| 206 | db, err := sql.Open("sqlite", "file:"+path+"?mode=ro") |
| 207 | if err != nil { |
| 208 | t.Fatal(err) |
| 209 | } |
| 210 | defer db.Close() |
| 211 | var n int |
| 212 | if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil { |
| 213 | t.Fatal(err) |
| 214 | } |
| 215 | return n |
| 216 | } |
| 217 | |
| 218 | // seek is not on the wire (wire specific 3): the client filters locally, |
| 219 | // with litestream's own rule (MinTXID >= seek). |
| 220 | func TestClientFiltersSeekLocally(t *testing.T) { |
| 221 | dir := t.TempDir() |
| 222 | dbdoortest.Replicate(t, dir, 3) |
| 223 | c := NewClient(newTestDoor(t, dir).do) |
| 224 | all := collect(t, c, 0, 0) |
| 225 | if len(all) < 2 { |
| 226 | t.Fatalf("want several files, got %d", len(all)) |
| 227 | } |
| 228 | seek := all[1].MinTXID |
| 229 | got := collect(t, c, 0, seek) |
| 230 | if len(got) != len(all)-1 || got[0].MinTXID != seek { |
| 231 | t.Fatalf("seek %d = %d files starting %d, want %d starting %d", seek, len(got), got[0].MinTXID, len(all)-1, seek) |
| 232 | } |
| 233 | } |
| 234 | |
| 235 | func collect(t *testing.T, c *Client, level int, seek ltx.TXID) []*ltx.FileInfo { |
| 236 | t.Helper() |
| 237 | itr, err := c.LTXFiles(context.Background(), level, seek, false) |
| 238 | if err != nil { |
| 239 | t.Fatal(err) |
| 240 | } |
| 241 | defer itr.Close() |
| 242 | var out []*ltx.FileInfo |
| 243 | for itr.Next() { |
| 244 | out = append(out, itr.Item()) |
| 245 | } |
| 246 | return out |
| 247 | } |
| 248 | |
| 249 | // A 404 on a file is os.ErrNotExist, which is what the whole-Restore retry |
| 250 | // keys on; any other refusal carries the console's own sentence. |
| 251 | func TestClientMapsTheDoorsAnswers(t *testing.T) { |
| 252 | answer := func(status int, body string) Doer { |
| 253 | return func(context.Context, string, http.Header) (*http.Response, error) { |
| 254 | rec := httptest.NewRecorder() |
| 255 | rec.WriteHeader(status) |
| 256 | io.WriteString(rec, body) |
| 257 | return rec.Result(), nil |
| 258 | } |
| 259 | } |
| 260 | ctx := context.Background() |
| 261 | if _, err := NewClient(answer(404, "404 page not found\n")).OpenLTXFile(ctx, 0, 1, 1, 0, 0); !errors.Is(err, os.ErrNotExist) { |
| 262 | t.Errorf("open on 404 = %v, want os.ErrNotExist", err) |
| 263 | } |
| 264 | _, err := NewClient(answer(409, `{"error":"the data is elsewhere"}`)).LTXFiles(ctx, 0, 0, false) |
| 265 | var de *DoorError |
| 266 | if !errors.As(err, &de) || de.Status != 409 || de.Message != "the data is elsewhere" { |
| 267 | t.Errorf("list on 409 = %v, want a DoorError carrying the sentence", err) |
| 268 | } |
| 269 | if _, err := NewClient(answer(409, `{"error":"x"}`)).OpenLTXFile(ctx, 0, 1, 1, 0, 0); errors.Is(err, os.ErrNotExist) || !errors.As(err, &de) { |
| 270 | t.Errorf("open on 409 = %v, want a DoorError and not ErrNotExist", err) |
| 271 | } |
| 272 | } |
| 273 | |
| 274 | // The client sends the S3 client's exact Range header and the coordinate |
| 275 | // tail, nothing else. |
| 276 | func TestClientSendsCoordinatesAndRange(t *testing.T) { |
| 277 | var gotTail, gotRange string |
| 278 | c := NewClient(func(_ context.Context, tail string, h http.Header) (*http.Response, error) { |
| 279 | gotTail, gotRange = tail, h.Get("Range") |
| 280 | rec := httptest.NewRecorder() |
| 281 | rec.WriteHeader(http.StatusPartialContent) |
| 282 | return rec.Result(), nil |
| 283 | }) |
| 284 | rc, err := c.OpenLTXFile(context.Background(), 3, 16, 32, 512, 100) |
| 285 | if err != nil { |
| 286 | t.Fatal(err) |
| 287 | } |
| 288 | rc.Close() |
| 289 | if gotTail != GetTail(3, 16, 32) || gotRange != RangeHeader(512, 100) { |
| 290 | t.Fatalf("tail %q range %q, want %q %q", gotTail, gotRange, GetTail(3, 16, 32), RangeHeader(512, 100)) |
| 291 | } |
| 292 | } |
| 293 | |
| 294 | // Retry-After is capped at a minute however large the door's number, and a |
| 295 | // number too big to multiply into a Duration must not wrap round to a |
| 296 | // short or negative wait. |
| 297 | func TestClientCapsRetryAfter(t *testing.T) { |
| 298 | for _, v := range []string{"61", "3600", "18446744074", "9223372036", "9223372037"} { |
| 299 | do := func(context.Context, string, http.Header) (*http.Response, error) { |
| 300 | rec := httptest.NewRecorder() |
| 301 | rec.Header().Set("Retry-After", v) |
| 302 | rec.WriteHeader(http.StatusTooManyRequests) |
| 303 | io.WriteString(rec, `{"error":"busy"}`) |
| 304 | return rec.Result(), nil |
| 305 | } |
| 306 | _, err := NewClient(do).OpenLTXFile(context.Background(), 0, 1, 1, 0, 0) |
| 307 | var de *DoorError |
| 308 | if !errors.As(err, &de) || de.RetryAfter != time.Minute { |
| 309 | t.Errorf("Retry-After: %s gave %v, want the one-minute cap", v, de.RetryAfter) |
| 310 | } |
| 311 | } |
| 312 | } |
| 313 | |
amadan