| 1 | package dbdoor |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "io" |
| 7 | "net/http" |
| 8 | "net/http/httptest" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strconv" |
| 12 | "strings" |
| 13 | "sync" |
| 14 | "sync/atomic" |
| 15 | "testing" |
| 16 | "time" |
| 17 | |
| 18 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 19 | ) |
| 20 | |
| 21 | // litestream creates <out>.tmp itself with os.Create, at the process umask, |
| 22 | // so a 0600 file cannot be pre-created for it. The containing directory is |
| 23 | // the only thing that can be private, and it must also confine the -wal |
| 24 | // and -shm a read-only open still needs. |
| 25 | func TestRestoreToUsesAPrivateDirectory(t *testing.T) { |
| 26 | src := t.TempDir() |
| 27 | dbdoortest.Replicate(t, src, 2) |
| 28 | parent := t.TempDir() |
| 29 | got, err := RestoreTo(context.Background(), NewClient(newTestDoor(t, src).do), parent) |
| 30 | if err != nil { |
| 31 | t.Fatal(err) |
| 32 | } |
| 33 | defer os.RemoveAll(got.Dir) |
| 34 | if filepath.Dir(got.Path) != got.Dir || filepath.Dir(got.Dir) != parent { |
| 35 | t.Fatalf("path %s, dir %s: want the copy inside a fresh directory under %s", got.Path, got.Dir, parent) |
| 36 | } |
| 37 | fi, err := os.Stat(got.Dir) |
| 38 | if err != nil { |
| 39 | t.Fatal(err) |
| 40 | } |
| 41 | if perm := fi.Mode().Perm(); perm != 0o700 { |
| 42 | t.Fatalf("restore directory is %04o, want 0700", perm) |
| 43 | } |
| 44 | if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows { |
| 45 | t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows) |
| 46 | } |
| 47 | } |
| 48 | |
| 49 | // R1: the handover names the newest replicated transaction and its time, |
| 50 | // and the time is the LTX header's, the time of the write itself. |
| 51 | func TestRestoreToReportsTheNewestTransaction(t *testing.T) { |
| 52 | src := t.TempDir() |
| 53 | before := time.Now().Add(-time.Second) |
| 54 | dbdoortest.Replicate(t, src, 3) |
| 55 | door := newTestDoor(t, src) |
| 56 | c := NewClient(door.do) |
| 57 | files := collect(t, c, 0, 0) |
| 58 | want := files[len(files)-1].MaxTXID |
| 59 | got, err := RestoreTo(context.Background(), c, t.TempDir()) |
| 60 | if err != nil { |
| 61 | t.Fatal(err) |
| 62 | } |
| 63 | defer os.RemoveAll(got.Dir) |
| 64 | if got.TXID != uint64(want) { |
| 65 | t.Errorf("TXID = %d, want %d", got.TXID, want) |
| 66 | } |
| 67 | if got.At.Before(before) || got.At.After(time.Now().Add(time.Second)) { |
| 68 | t.Errorf("At = %v, want the time of the last write (around now)", got.At) |
| 69 | } |
| 70 | } |
| 71 | |
| 72 | // A planned file can vanish under the restore when the writer's L0 |
| 73 | // retention fires. ResumableReader treats ErrNotExist as FATAL and does not |
| 74 | // retry, so the retry has to wrap the whole Restore: a fresh plan is the |
| 75 | // only recovery. |
| 76 | func TestRestoreToRetriesTheWholeRestoreOnce(t *testing.T) { |
| 77 | src := t.TempDir() |
| 78 | dbdoortest.Replicate(t, src, 2) |
| 79 | door := newTestDoor(t, src) |
| 80 | first := collect(t, NewClient(door.do), 0, 0)[0] |
| 81 | door.gone[Span(uint64(first.MinTXID), uint64(first.MaxTXID))] = true |
| 82 | door.lists = map[int]int{} |
| 83 | got, err := RestoreTo(context.Background(), NewClient(door.do), t.TempDir()) |
| 84 | if err != nil { |
| 85 | t.Fatalf("RestoreTo with a file vanishing once = %v, want success on a fresh plan", err) |
| 86 | } |
| 87 | defer os.RemoveAll(got.Dir) |
| 88 | if door.lists[9] != 2 { |
| 89 | t.Fatalf("snapshot level listed %d times, want 2 (one plan per Restore)", door.lists[9]) |
| 90 | } |
| 91 | if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows { |
| 92 | t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows) |
| 93 | } |
| 94 | } |
| 95 | |
| 96 | // Once, not forever: a file that keeps vanishing is a failure to report, |
| 97 | // and the private directory goes with it. |
| 98 | func TestRestoreToRetriesOnlyOnce(t *testing.T) { |
| 99 | src := t.TempDir() |
| 100 | dbdoortest.Replicate(t, src, 2) |
| 101 | door := newTestDoor(t, src) |
| 102 | first := collect(t, NewClient(door.do), 0, 0)[0] |
| 103 | span := Span(uint64(first.MinTXID), uint64(first.MaxTXID)) |
| 104 | c := NewClient(func(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 105 | door.mu.Lock() |
| 106 | door.gone[span] = true |
| 107 | door.mu.Unlock() |
| 108 | return door.do(ctx, tail, h) |
| 109 | }) |
| 110 | parent := t.TempDir() |
| 111 | if _, err := RestoreTo(context.Background(), c, parent); !errors.Is(err, os.ErrNotExist) { |
| 112 | t.Fatalf("RestoreTo = %v, want the vanished file reported", err) |
| 113 | } |
| 114 | assertEmptyDir(t, parent) |
| 115 | } |
| 116 | |
| 117 | // "Nothing replicated yet" is its own answer, for both of the errors |
| 118 | // litestream gives it (spec §10), and leaves nothing behind. |
| 119 | func TestRestoreToNamesAnEmptyReplica(t *testing.T) { |
| 120 | parent := t.TempDir() |
| 121 | _, err := RestoreTo(context.Background(), NewClient(newTestDoor(t, t.TempDir()).do), parent) |
| 122 | if !errors.Is(err, ErrNothingReplicated) { |
| 123 | t.Fatalf("RestoreTo of an empty replica = %v, want ErrNothingReplicated", err) |
| 124 | } |
| 125 | assertEmptyDir(t, parent) |
| 126 | } |
| 127 | |
| 128 | // A copy that fails lease.CheckDB is never handed over (spec §10). A |
| 129 | // replica litestream wrote always holds its own _litestream_seq table and |
| 130 | // a corrupt one fails inside Restore, so the check's refusal is injected: |
| 131 | // what is under test is that RestoreTo honours it. |
| 132 | func TestRestoreToRefusesACopyThatFailsTheCheck(t *testing.T) { |
| 133 | src := t.TempDir() |
| 134 | dbdoortest.Replicate(t, src, 1) |
| 135 | var checked string |
| 136 | defer func(old func(context.Context, string) error) { checkDB = old }(checkDB) |
| 137 | checkDB = func(_ context.Context, path string) error { |
| 138 | checked = path |
| 139 | return errors.New("integrity_check: page 2 is never used") |
| 140 | } |
| 141 | parent := t.TempDir() |
| 142 | _, err := RestoreTo(context.Background(), NewClient(newTestDoor(t, src).do), parent) |
| 143 | if !errors.Is(err, ErrCheckFailed) { |
| 144 | t.Fatalf("RestoreTo of a copy failing its check = %v, want ErrCheckFailed", err) |
| 145 | } |
| 146 | if filepath.Base(checked) != restoredName { |
| 147 | t.Fatalf("checked %q, want the restored copy", checked) |
| 148 | } |
| 149 | assertEmptyDir(t, parent) |
| 150 | } |
| 151 | |
| 152 | func assertEmptyDir(t *testing.T, dir string) { |
| 153 | t.Helper() |
| 154 | ents, err := os.ReadDir(dir) |
| 155 | if err != nil { |
| 156 | t.Fatal(err) |
| 157 | } |
| 158 | if len(ents) != 0 { |
| 159 | t.Fatalf("%s holds %d entries after a failed restore, want none (no copy of production left behind)", dir, len(ents)) |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | // countingBody tracks how many door responses are open at once. |
| 164 | type countingBody struct { |
| 165 | io.ReadCloser |
| 166 | open *atomic.Int32 |
| 167 | once sync.Once |
| 168 | } |
| 169 | |
| 170 | func (b *countingBody) Close() error { |
| 171 | b.once.Do(func() { b.open.Add(-1) }) |
| 172 | return b.ReadCloser.Close() |
| 173 | } |
| 174 | |
| 175 | // The door bounds a person's concurrent file streams, and litestream's |
| 176 | // compactor holds every planned file open until the restore ends. Streaming |
| 177 | // straight through would need one slot per file, so RestoreTo downloads each |
| 178 | // file whole into its private directory before handing it on: one stream at |
| 179 | // a time, whatever the plan's length. |
| 180 | func TestRestoreToHoldsOneStreamAtATime(t *testing.T) { |
| 181 | src := t.TempDir() |
| 182 | dbdoortest.Replicate(t, src, 4) |
| 183 | door := newTestDoor(t, src) |
| 184 | var open, peak atomic.Int32 |
| 185 | do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 186 | res, err := door.do(ctx, tail, h) |
| 187 | if err != nil || !strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") { |
| 188 | return res, err // listings are read and closed at once |
| 189 | } |
| 190 | n := open.Add(1) |
| 191 | for p := peak.Load(); n > p && !peak.CompareAndSwap(p, n); p = peak.Load() { |
| 192 | } |
| 193 | res.Body = &countingBody{ReadCloser: res.Body, open: &open} |
| 194 | return res, nil |
| 195 | } |
| 196 | got, err := RestoreTo(context.Background(), NewClient(do), t.TempDir()) |
| 197 | if err != nil { |
| 198 | t.Fatal(err) |
| 199 | } |
| 200 | defer os.RemoveAll(got.Dir) |
| 201 | if p := peak.Load(); p != 1 { |
| 202 | t.Fatalf("restore held %d file streams open at once, want 1", p) |
| 203 | } |
| 204 | if rows := countRows(t, got.Path); rows != 4*dbdoortest.Rows { |
| 205 | t.Fatalf("restored %d rows, want %d", rows, 4*dbdoortest.Rows) |
| 206 | } |
| 207 | if ents, _ := os.ReadDir(got.Dir); len(ents) > 3 { |
| 208 | t.Fatalf("downloaded files left in the private directory: %v", ents) |
| 209 | } |
| 210 | } |
| 211 | |
| 212 | // flakyBody hands over at most limit bytes and then fails the way a reset |
| 213 | // connection does. |
| 214 | type flakyBody struct { |
| 215 | io.ReadCloser |
| 216 | left int |
| 217 | } |
| 218 | |
| 219 | func (b *flakyBody) Read(p []byte) (int, error) { |
| 220 | if b.left <= 0 { |
| 221 | return 0, io.ErrUnexpectedEOF |
| 222 | } |
| 223 | if len(p) > b.left { |
| 224 | p = p[:b.left] |
| 225 | } |
| 226 | n, err := b.ReadCloser.Read(p) |
| 227 | b.left -= n |
| 228 | return n, err |
| 229 | } |
| 230 | |
| 231 | // flaky wraps a door so every file response breaks after limit bytes. |
| 232 | func flaky(door *testDoor, limit int, gets *atomic.Int32) Doer { |
| 233 | return func(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 234 | res, err := door.do(ctx, tail, h) |
| 235 | if err != nil || !strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") { |
| 236 | return res, err |
| 237 | } |
| 238 | gets.Add(1) |
| 239 | res.Body = &flakyBody{ReadCloser: res.Body, left: limit} |
| 240 | return res, nil |
| 241 | } |
| 242 | } |
| 243 | |
| 244 | // Over a link that resets every few hundred bytes, a download keeps what it |
| 245 | // has and asks for the rest, so a file larger than any one connection |
| 246 | // survives still arrives. |
| 247 | func TestRestoreToResumesAFileAcrossResets(t *testing.T) { |
| 248 | src := t.TempDir() |
| 249 | dbdoortest.Replicate(t, src, 2) |
| 250 | var gets atomic.Int32 |
| 251 | got, err := RestoreTo(context.Background(), NewClient(flaky(newTestDoor(t, src), 100, &gets)), t.TempDir()) |
| 252 | if err != nil { |
| 253 | t.Fatalf("RestoreTo over a resetting link = %v, want it to resume and finish", err) |
| 254 | } |
| 255 | defer os.RemoveAll(got.Dir) |
| 256 | if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows { |
| 257 | t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows) |
| 258 | } |
| 259 | } |
| 260 | |
| 261 | // A link that delivers nothing is given up on after a bounded number of |
| 262 | // attempts, not retried forever. |
| 263 | func TestRestoreToGivesUpOnALinkThatMakesNoProgress(t *testing.T) { |
| 264 | src := t.TempDir() |
| 265 | dbdoortest.Replicate(t, src, 1) |
| 266 | var gets atomic.Int32 |
| 267 | parent := t.TempDir() |
| 268 | if _, err := RestoreTo(context.Background(), NewClient(flaky(newTestDoor(t, src), 0, &gets)), parent); err == nil { |
| 269 | t.Fatal("RestoreTo over a dead link succeeded") |
| 270 | } |
| 271 | if n := gets.Load(); n > 40 { |
| 272 | t.Fatalf("made %d attempts at a link that never delivered a byte", n) |
| 273 | } |
| 274 | assertEmptyDir(t, parent) |
| 275 | } |
| 276 | |
| 277 | // Ctrl-C after the last file has arrived still stops the restore. The |
| 278 | // local half (litestream's compaction over the downloaded files) does not |
| 279 | // check the context itself, so the downloaded files do, and RestoreTo |
| 280 | // checks again before CheckDB. Nothing is handed over, nothing left. |
| 281 | func TestRestoreToStopsWhenCancelledAfterTheDownloads(t *testing.T) { |
| 282 | src := t.TempDir() |
| 283 | dbdoortest.Replicate(t, src, 3) |
| 284 | door := newTestDoor(t, src) |
| 285 | files := len(collect(t, NewClient(door.do), 0, 0)) |
| 286 | ctx, cancel := context.WithCancel(context.Background()) |
| 287 | defer cancel() |
| 288 | var gets atomic.Int32 |
| 289 | do := func(c context.Context, tail string, h http.Header) (*http.Response, error) { |
| 290 | res, err := door.do(c, tail, h) |
| 291 | if err == nil && strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") && int(gets.Add(1)) == files { |
| 292 | cancel() // the last download is in hand; only local work is left |
| 293 | } |
| 294 | return res, err |
| 295 | } |
| 296 | parent := t.TempDir() |
| 297 | if _, err := RestoreTo(ctx, NewClient(do), parent); !errors.Is(err, context.Canceled) { |
| 298 | t.Fatalf("RestoreTo cancelled after its downloads = %v, want context.Canceled", err) |
| 299 | } |
| 300 | assertEmptyDir(t, parent) |
| 301 | } |
| 302 | |
| 303 | // answer is a canned door response. |
| 304 | func answer(status int, retryAfter string) *http.Response { |
| 305 | rec := httptest.NewRecorder() |
| 306 | if retryAfter != "" { |
| 307 | rec.Header().Set("Retry-After", retryAfter) |
| 308 | } |
| 309 | rec.WriteHeader(status) |
| 310 | io.WriteString(rec, `{"error":"busy"}`) |
| 311 | return rec.Result() |
| 312 | } |
| 313 | |
| 314 | // rangeStart is the offset a request's Range header asks for. |
| 315 | func rangeStart(h http.Header) int64 { |
| 316 | off, _, _ := ParseRange(h.Get("Range")) |
| 317 | return off |
| 318 | } |
| 319 | |
| 320 | // A reopen of the rest of a file that meets a busy door (503, or 429 with |
| 321 | // Retry-After) is retried in place, keeping the bytes already downloaded, |
| 322 | // rather than thrown back to litestream, which would start the file again. |
| 323 | func TestRestoreToRetriesABusyReopenWithoutLosingProgress(t *testing.T) { |
| 324 | defer func(old time.Duration) { dbDownloadBackoff = old }(dbDownloadBackoff) |
| 325 | dbDownloadBackoff = time.Millisecond |
| 326 | src := t.TempDir() |
| 327 | dbdoortest.Replicate(t, src, 2) |
| 328 | door := newTestDoor(t, src) |
| 329 | var mu sync.Mutex |
| 330 | busy := map[string]bool{} |
| 331 | var restarts atomic.Int32 |
| 332 | reached := map[string]int64{} |
| 333 | do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 334 | if !strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") { |
| 335 | return door.do(ctx, tail, h) |
| 336 | } |
| 337 | off := rangeStart(h) |
| 338 | key := tail + "@" + strconv.FormatInt(off, 10) |
| 339 | mu.Lock() |
| 340 | if off < reached[tail] { |
| 341 | restarts.Add(1) |
| 342 | } |
| 343 | reached[tail] = max(reached[tail], off) |
| 344 | first := !busy[key] |
| 345 | busy[key] = true |
| 346 | mu.Unlock() |
| 347 | if off > 0 && first { |
| 348 | return answer(http.StatusServiceUnavailable, ""), nil |
| 349 | } |
| 350 | res, err := door.do(ctx, tail, h) |
| 351 | if err == nil { |
| 352 | res.Body = &flakyBody{ReadCloser: res.Body, left: 100} |
| 353 | } |
| 354 | return res, err |
| 355 | } |
| 356 | got, err := RestoreTo(context.Background(), NewClient(do), t.TempDir()) |
| 357 | if err != nil { |
| 358 | t.Fatalf("RestoreTo past busy reopens = %v, want it to retry in place and finish", err) |
| 359 | } |
| 360 | defer os.RemoveAll(got.Dir) |
| 361 | if n := restarts.Load(); n != 0 { |
| 362 | t.Fatalf("%d requests went back to an earlier offset: progress was thrown away", n) |
| 363 | } |
| 364 | if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows { |
| 365 | t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows) |
| 366 | } |
| 367 | } |
| 368 | |
| 369 | // A 429's Retry-After is honoured before the next attempt. |
| 370 | func TestRestoreToHonoursRetryAfter(t *testing.T) { |
| 371 | defer func(old time.Duration) { dbDownloadBackoff = old }(dbDownloadBackoff) |
| 372 | dbDownloadBackoff = time.Millisecond |
| 373 | src := t.TempDir() |
| 374 | dbdoortest.Replicate(t, src, 1) |
| 375 | door := newTestDoor(t, src) |
| 376 | var mu sync.Mutex |
| 377 | var refusedAt, nextAt time.Time |
| 378 | do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 379 | if !strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") { |
| 380 | return door.do(ctx, tail, h) |
| 381 | } |
| 382 | mu.Lock() |
| 383 | defer mu.Unlock() |
| 384 | if rangeStart(h) > 0 && refusedAt.IsZero() { |
| 385 | refusedAt = time.Now() |
| 386 | return answer(http.StatusTooManyRequests, "1"), nil |
| 387 | } |
| 388 | if !refusedAt.IsZero() && nextAt.IsZero() { |
| 389 | nextAt = time.Now() |
| 390 | } |
| 391 | res, err := door.do(ctx, tail, h) |
| 392 | if err == nil && rangeStart(h) == 0 { |
| 393 | res.Body = &flakyBody{ReadCloser: res.Body, left: 100} |
| 394 | } |
| 395 | return res, err |
| 396 | } |
| 397 | got, err := RestoreTo(context.Background(), NewClient(do), t.TempDir()) |
| 398 | if err != nil { |
| 399 | t.Fatal(err) |
| 400 | } |
| 401 | defer os.RemoveAll(got.Dir) |
| 402 | if refusedAt.IsZero() || nextAt.Sub(refusedAt) < 900*time.Millisecond { |
| 403 | t.Fatalf("retried %v after a 429 with Retry-After: 1", nextAt.Sub(refusedAt)) |
| 404 | } |
| 405 | } |
| 406 | |
| 407 | // The wait between attempts ends the moment the person stops. |
| 408 | func TestRestoreToBackoffIsCancellable(t *testing.T) { |
| 409 | defer func(old time.Duration) { dbDownloadBackoff = old }(dbDownloadBackoff) |
| 410 | dbDownloadBackoff = time.Hour |
| 411 | src := t.TempDir() |
| 412 | dbdoortest.Replicate(t, src, 1) |
| 413 | door := newTestDoor(t, src) |
| 414 | do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 415 | if strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") && rangeStart(h) > 0 { |
| 416 | return answer(http.StatusServiceUnavailable, ""), nil |
| 417 | } |
| 418 | res, err := door.do(ctx, tail, h) |
| 419 | if err == nil && strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") { |
| 420 | res.Body = &flakyBody{ReadCloser: res.Body, left: 100} |
| 421 | } |
| 422 | return res, err |
| 423 | } |
| 424 | ctx, cancel := context.WithCancel(context.Background()) |
| 425 | time.AfterFunc(200*time.Millisecond, cancel) |
| 426 | start := time.Now() |
| 427 | parent := t.TempDir() |
| 428 | if _, err := RestoreTo(ctx, NewClient(do), parent); !errors.Is(err, context.Canceled) { |
| 429 | t.Fatalf("RestoreTo cancelled during a backoff = %v, want context.Canceled", err) |
| 430 | } |
| 431 | if d := time.Since(start); d > 5*time.Second { |
| 432 | t.Fatalf("the stop took %v to land", d) |
| 433 | } |
| 434 | assertEmptyDir(t, parent) |
| 435 | } |
| 436 | |
| 437 | // A door that stays busy is given up on: only attempts that deliver no |
| 438 | // byte count, and there are a bounded number of them. |
| 439 | func TestRestoreToGivesUpOnAPermanentlyBusyDoor(t *testing.T) { |
| 440 | defer func(old time.Duration) { dbDownloadBackoff = old }(dbDownloadBackoff) |
| 441 | dbDownloadBackoff = time.Millisecond |
| 442 | src := t.TempDir() |
| 443 | dbdoortest.Replicate(t, src, 1) |
| 444 | door := newTestDoor(t, src) |
| 445 | var reopens atomic.Int32 |
| 446 | do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 447 | if strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") && rangeStart(h) > 0 { |
| 448 | reopens.Add(1) |
| 449 | return answer(http.StatusServiceUnavailable, ""), nil |
| 450 | } |
| 451 | res, err := door.do(ctx, tail, h) |
| 452 | if err == nil && strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") { |
| 453 | res.Body = &flakyBody{ReadCloser: res.Body, left: 100} |
| 454 | } |
| 455 | return res, err |
| 456 | } |
| 457 | parent := t.TempDir() |
| 458 | if _, err := RestoreTo(context.Background(), NewClient(do), parent); err == nil { |
| 459 | t.Fatal("RestoreTo against a door that never recovers succeeded") |
| 460 | } |
| 461 | if n := reopens.Load(); n == 0 || n > 100 { |
| 462 | t.Fatalf("%d reopens against a permanently busy door", n) |
| 463 | } |
| 464 | assertEmptyDir(t, parent) |
| 465 | } |
| 466 | |
| 467 | // A stop that lands while the copy is being checked still withdraws it: |
| 468 | // RestoreTo checks the context after CheckDB and hands nothing over. |
| 469 | func TestRestoreToStopsWhenCancelledDuringTheCheck(t *testing.T) { |
| 470 | src := t.TempDir() |
| 471 | dbdoortest.Replicate(t, src, 1) |
| 472 | ctx, cancel := context.WithCancel(context.Background()) |
| 473 | defer cancel() |
| 474 | defer func(old func(context.Context, string) error) { checkDB = old }(checkDB) |
| 475 | var sawCtx bool |
| 476 | checkDB = func(c context.Context, path string) error { |
| 477 | sawCtx = c == ctx |
| 478 | cancel() // the stop arrives as the check finishes |
| 479 | return nil |
| 480 | } |
| 481 | parent := t.TempDir() |
| 482 | if _, err := RestoreTo(ctx, NewClient(newTestDoor(t, src).do), parent); !errors.Is(err, context.Canceled) { |
| 483 | t.Fatalf("RestoreTo cancelled during the check = %v, want context.Canceled", err) |
| 484 | } |
| 485 | if !sawCtx { |
| 486 | t.Fatal("the check did not get the restore's context") |
| 487 | } |
| 488 | assertEmptyDir(t, parent) |
| 489 | } |
| 490 | |
| 491 | // The FIRST open of a file meeting a full door (429) waits and asks again, |
| 492 | // the same as a reopen does, instead of failing straight back to |
| 493 | // litestream, whose own retries ignore Retry-After. |
| 494 | func TestRestoreToRetriesABusyFirstOpen(t *testing.T) { |
| 495 | defer func(old time.Duration) { dbDownloadBackoff = old }(dbDownloadBackoff) |
| 496 | dbDownloadBackoff = time.Millisecond |
| 497 | src := t.TempDir() |
| 498 | dbdoortest.Replicate(t, src, 2) |
| 499 | door := newTestDoor(t, src) |
| 500 | var mu sync.Mutex |
| 501 | refused := map[string]int{} |
| 502 | do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 503 | if !strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") { |
| 504 | return door.do(ctx, tail, h) |
| 505 | } |
| 506 | mu.Lock() |
| 507 | n := refused[tail] |
| 508 | refused[tail] = n + 1 |
| 509 | mu.Unlock() |
| 510 | if rangeStart(h) == 0 && n < 4 { |
| 511 | return answer(http.StatusTooManyRequests, ""), nil |
| 512 | } |
| 513 | return door.do(ctx, tail, h) |
| 514 | } |
| 515 | got, err := RestoreTo(context.Background(), NewClient(do), t.TempDir()) |
| 516 | if err != nil { |
| 517 | t.Fatalf("RestoreTo past four busy first opens per file = %v, want it to wait and finish", err) |
| 518 | } |
| 519 | defer os.RemoveAll(got.Dir) |
| 520 | if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows { |
| 521 | t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows) |
| 522 | } |
| 523 | } |
| 524 | |
| 525 | func TestDownloadTempFileFailure(t *testing.T) { |
| 526 | for _, supplied := range []bool{false, true} { |
| 527 | t.Run(strconv.FormatBool(supplied), func(t *testing.T) { |
| 528 | var rc io.ReadCloser |
| 529 | var open atomic.Int32 |
| 530 | if supplied { |
| 531 | open.Store(1) |
| 532 | rc = &countingBody{ReadCloser: io.NopCloser(strings.NewReader("unused")), open: &open} |
| 533 | } |
| 534 | r := &recorder{dir: filepath.Join(t.TempDir(), "missing")} |
| 535 | got, err := r.download(t.Context(), rc, 0, 1, 1, 0, 0) |
| 536 | if got != nil || !errors.Is(err, os.ErrNotExist) { |
| 537 | t.Fatalf("download = %v, %v", got, err) |
| 538 | } |
| 539 | if open.Load() != 0 { |
| 540 | t.Fatal("supplied reader leaked") |
| 541 | } |
| 542 | }) |
| 543 | } |
| 544 | } |
| 545 | |
amadan