amadan

carlos / platform Public

dbdoor: wait out a busy reopen without losing the bytes downloaded

When reopening the rest of a file met a transport error, a 503 or a 429,
download deleted the partial file and litestream started the file again
from its first byte. Those reopens are now retried in place, keeping the
file and the offset: exponential backoff from 250ms up to 8s, a 429's
Retry-After honoured (seconds, capped at a minute, now carried on
DoorError), and every wait ends the moment the context does. Only
attempts that deliver no byte count toward the give-up limit of five;
any progress resets it. Other refusals (403, 404, 409) are final as
before, and a vanished file still marks the restore for a fresh plan.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 5c5351145b3dcf5b95d1c1f2389c4161cc362bdc parent cd15382
3 files changed, +249 −20
  • internal/dbdoor/client.go +14 −3
  • internal/dbdoor/restore.go +69 −17
  • internal/dbdoor/restore_test.go +166 −0
diff --git a/internal/dbdoor/client.go b/internal/dbdoor/client.go
index 8c3e9246a..f14680b71 100644
--- a/internal/dbdoor/client.go
+++ b/internal/dbdoor/client.go
@@ -9,7 +9,9 @@ import (
"log/slog"
"net/http"
"os"
+ "strconv"
"strings"
+ "time"
"github.com/benbjohnson/litestream"
"github.com/superfly/ltx"
@@ -52,6 +54,9 @@ var errReadOnly = errors.New("dbdoor: this client is read-only; it restores repl
type DoorError struct {
Status int
Message string
+ // RetryAfter is the door's Retry-After, in whole seconds, when it sent
+ // one (a 429 from a full stream bound does); zero otherwise.
+ RetryAfter time.Duration
}
func (e *DoorError) Error() string {
@@ -136,10 +141,16 @@ func doorError(res *http.Response) error {
var body struct {
Error string `json:"error"`
}
- if json.Unmarshal(raw, &body) == nil && strings.TrimSpace(body.Error) != "" {
- return &DoorError{Status: res.StatusCode, Message: strings.TrimSpace(body.Error)}
+ e := &DoorError{Status: res.StatusCode}
+ if json.Unmarshal(raw, &body) == nil {
+ e.Message = strings.TrimSpace(body.Error)
}
- return &DoorError{Status: res.StatusCode}
+ // Seconds only (the door never sends a date), capped so a hostile or
+ // broken value cannot park a download for hours.
+ if secs, err := strconv.Atoi(strings.TrimSpace(res.Header.Get("Retry-After"))); err == nil && secs > 0 {
+ e.RetryAfter = min(time.Duration(secs)*time.Second, time.Minute)
+ }
+ return e
}
var _ litestream.ReplicaClient = (*Client)(nil)
diff --git a/internal/dbdoor/restore.go b/internal/dbdoor/restore.go
index 859d0f43f..95f66afad 100644
--- a/internal/dbdoor/restore.go
+++ b/internal/dbdoor/restore.go
@@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"io"
+ "net/http"
"os"
"path/filepath"
"sync"
@@ -192,17 +193,21 @@ func (r *recorder) download(ctx context.Context, rc io.ReadCloser, level int, mi
return nil, err
}
var written int64
- for stalls := 0; ; {
- n, cerr := io.Copy(f, rc)
- rc.Close()
- written += n
- if cerr == nil {
- break
- }
- if n > 0 {
- stalls = 0
- } else if stalls++; stalls >= dbDownloadMaxStalls {
- return fail(cerr)
+ stalls, wait := 0, dbDownloadBackoff
+ for {
+ if rc != nil {
+ n, cerr := io.Copy(f, rc)
+ rc.Close()
+ rc = nil
+ written += n
+ if cerr == nil {
+ break
+ }
+ if n > 0 {
+ stalls, wait = 0, dbDownloadBackoff
+ } else if stalls++; stalls >= dbDownloadMaxStalls {
+ return fail(cerr)
+ }
}
if ctx.Err() != nil {
return fail(ctx.Err())
@@ -213,13 +218,38 @@ func (r *recorder) download(ctx context.Context, rc io.ReadCloser, level int, mi
break
}
}
- if rc, err = r.Client.OpenLTXFile(ctx, level, minTXID, maxTXID, offset+written, rest); err != nil {
- if errors.Is(err, os.ErrNotExist) {
- r.mu.Lock()
- r.vanished = true
- r.mu.Unlock()
- }
+ next, err := r.Client.OpenLTXFile(ctx, level, minTXID, maxTXID, offset+written, rest)
+ if err == nil {
+ rc = next
+ continue
+ }
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ r.mu.Lock()
+ r.vanished = true
+ r.mu.Unlock()
return fail(err)
+ case ctx.Err() != nil:
+ return fail(ctx.Err())
+ case !transientOpen(err):
+ return fail(err)
+ }
+ // A busy door or a dropped connection: wait and ask again for the
+ // same remaining range, keeping every byte already here. It counts
+ // as an attempt that delivered nothing.
+ if stalls++; stalls >= dbDownloadMaxStalls {
+ return fail(err)
+ }
+ d := wait
+ var door *DoorError
+ if errors.As(err, &door) && door.RetryAfter > 0 {
+ d = door.RetryAfter
+ }
+ wait = min(wait*2, dbDownloadBackoffMax)
+ select {
+ case <-ctx.Done():
+ return fail(ctx.Err())
+ case <-time.After(d):
}
}
if _, err := f.Seek(0, io.SeekStart); err != nil {
@@ -283,3 +313,25 @@ func (t *headerTee) Read(p []byte) (int, error) {
}
return n, err
}
+
+// dbDownloadBackoff is the first wait before retrying a reopen that met a
+// busy door or a dropped connection; each further wait doubles, up to
+// dbDownloadBackoffMax. A 429's Retry-After overrides it.
+var dbDownloadBackoff = 250 * time.Millisecond
+
+const dbDownloadBackoffMax = 8 * time.Second
+
+// transientOpen reports a reopen failure worth waiting out: the door busy
+// or briefly unavailable (429, 502, 503, 504), or no answer at all (a
+// transport error). Any other refusal from the door is final.
+func transientOpen(err error) bool {
+ var door *DoorError
+ if errors.As(err, &door) {
+ switch door.Status {
+ case http.StatusTooManyRequests, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout:
+ return true
+ }
+ return false
+ }
+ return true
+}
diff --git a/internal/dbdoor/restore_test.go b/internal/dbdoor/restore_test.go
index 6cad1ae8a..81d91b0aa 100644
--- a/internal/dbdoor/restore_test.go
+++ b/internal/dbdoor/restore_test.go
@@ -5,8 +5,10 @@ import (
"errors"
"io"
"net/http"
+ "net/http/httptest"
"os"
"path/filepath"
+ "strconv"
"strings"
"sync"
"sync/atomic"
@@ -297,3 +299,167 @@ func TestRestoreToStopsWhenCancelledAfterTheDownloads(t *testing.T) {
}
assertEmptyDir(t, parent)
}
+
+// answer is a canned door response.
+func answer(status int, retryAfter string) *http.Response {
+ rec := httptest.NewRecorder()
+ if retryAfter != "" {
+ rec.Header().Set("Retry-After", retryAfter)
+ }
+ rec.WriteHeader(status)
+ io.WriteString(rec, `{"error":"busy"}`)
+ return rec.Result()
+}
+
+// rangeStart is the offset a request's Range header asks for.
+func rangeStart(h http.Header) int64 {
+ off, _, _ := ParseRange(h.Get("Range"))
+ return off
+}
+
+// A reopen of the rest of a file that meets a busy door (503, or 429 with
+// Retry-After) is retried in place, keeping the bytes already downloaded,
+// rather than thrown back to litestream, which would start the file again.
+func TestRestoreToRetriesABusyReopenWithoutLosingProgress(t *testing.T) {
+ defer func(old time.Duration) { dbDownloadBackoff = old }(dbDownloadBackoff)
+ dbDownloadBackoff = time.Millisecond
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 2)
+ door := newTestDoor(t, src)
+ var mu sync.Mutex
+ busy := map[string]bool{}
+ var restarts atomic.Int32
+ reached := map[string]int64{}
+ do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ if !strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") {
+ return door.do(ctx, tail, h)
+ }
+ off := rangeStart(h)
+ key := tail + "@" + strconv.FormatInt(off, 10)
+ mu.Lock()
+ if off < reached[tail] {
+ restarts.Add(1)
+ }
+ reached[tail] = max(reached[tail], off)
+ first := !busy[key]
+ busy[key] = true
+ mu.Unlock()
+ if off > 0 && first {
+ return answer(http.StatusServiceUnavailable, ""), nil
+ }
+ res, err := door.do(ctx, tail, h)
+ if err == nil {
+ res.Body = &flakyBody{ReadCloser: res.Body, left: 100}
+ }
+ return res, err
+ }
+ got, err := RestoreTo(context.Background(), NewClient(do), t.TempDir())
+ if err != nil {
+ t.Fatalf("RestoreTo past busy reopens = %v, want it to retry in place and finish", err)
+ }
+ defer os.RemoveAll(got.Dir)
+ if n := restarts.Load(); n != 0 {
+ t.Fatalf("%d requests went back to an earlier offset: progress was thrown away", n)
+ }
+ if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows {
+ t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows)
+ }
+}
+
+// A 429's Retry-After is honoured before the next attempt.
+func TestRestoreToHonoursRetryAfter(t *testing.T) {
+ defer func(old time.Duration) { dbDownloadBackoff = old }(dbDownloadBackoff)
+ dbDownloadBackoff = time.Millisecond
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 1)
+ door := newTestDoor(t, src)
+ var mu sync.Mutex
+ var refusedAt, nextAt time.Time
+ do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ if !strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") {
+ return door.do(ctx, tail, h)
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ if rangeStart(h) > 0 && refusedAt.IsZero() {
+ refusedAt = time.Now()
+ return answer(http.StatusTooManyRequests, "1"), nil
+ }
+ if !refusedAt.IsZero() && nextAt.IsZero() {
+ nextAt = time.Now()
+ }
+ res, err := door.do(ctx, tail, h)
+ if err == nil && rangeStart(h) == 0 {
+ res.Body = &flakyBody{ReadCloser: res.Body, left: 100}
+ }
+ return res, err
+ }
+ got, err := RestoreTo(context.Background(), NewClient(do), t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(got.Dir)
+ if refusedAt.IsZero() || nextAt.Sub(refusedAt) < 900*time.Millisecond {
+ t.Fatalf("retried %v after a 429 with Retry-After: 1", nextAt.Sub(refusedAt))
+ }
+}
+
+// The wait between attempts ends the moment the person stops.
+func TestRestoreToBackoffIsCancellable(t *testing.T) {
+ defer func(old time.Duration) { dbDownloadBackoff = old }(dbDownloadBackoff)
+ dbDownloadBackoff = time.Hour
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 1)
+ door := newTestDoor(t, src)
+ do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ if strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") && rangeStart(h) > 0 {
+ return answer(http.StatusServiceUnavailable, ""), nil
+ }
+ res, err := door.do(ctx, tail, h)
+ if err == nil && strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") {
+ res.Body = &flakyBody{ReadCloser: res.Body, left: 100}
+ }
+ return res, err
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ time.AfterFunc(200*time.Millisecond, cancel)
+ start := time.Now()
+ parent := t.TempDir()
+ if _, err := RestoreTo(ctx, NewClient(do), parent); !errors.Is(err, context.Canceled) {
+ t.Fatalf("RestoreTo cancelled during a backoff = %v, want context.Canceled", err)
+ }
+ if d := time.Since(start); d > 5*time.Second {
+ t.Fatalf("the stop took %v to land", d)
+ }
+ assertEmptyDir(t, parent)
+}
+
+// A door that stays busy is given up on: only attempts that deliver no
+// byte count, and there are a bounded number of them.
+func TestRestoreToGivesUpOnAPermanentlyBusyDoor(t *testing.T) {
+ defer func(old time.Duration) { dbDownloadBackoff = old }(dbDownloadBackoff)
+ dbDownloadBackoff = time.Millisecond
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 1)
+ door := newTestDoor(t, src)
+ var reopens atomic.Int32
+ do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ if strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") && rangeStart(h) > 0 {
+ reopens.Add(1)
+ return answer(http.StatusServiceUnavailable, ""), nil
+ }
+ res, err := door.do(ctx, tail, h)
+ if err == nil && strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") {
+ res.Body = &flakyBody{ReadCloser: res.Body, left: 100}
+ }
+ return res, err
+ }
+ parent := t.TempDir()
+ if _, err := RestoreTo(context.Background(), NewClient(do), parent); err == nil {
+ t.Fatal("RestoreTo against a door that never recovers succeeded")
+ }
+ if n := reopens.Load(); n == 0 || n > 100 {
+ t.Fatalf("%d reopens against a permanently busy door", n)
+ }
+ assertEmptyDir(t, parent)
+}