amadan

carlos / platform Public

carlos db: Astra round 4, first-open 429s, Retry-After cap, pragma as a name

- A file's first open now goes through download's retry loop, so a full
  door's 429 waits out Retry-After instead of failing back to litestream,
  whose four quick retries ignore it.
- Retry-After is clamped to 60 before it is multiplied: 18446744074 used
  to wrap to ~290ms, and larger values went negative.
- The pragma guard looks only in statement position (first, or after
  EXPLAIN [QUERY PLAN]); one statement is already enforced. A column or
  alias named "pragma" no longer refuses a read. The 8,624 generated
  disguises are all still refused.

Each change has a test that failed first.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b05ceeef43936db6ee30a46663c48cacac915445 parent f7490ce
6 files changed, +96 −20
  • db.go +20 −10
  • db_test.go +15 −0
  • internal/dbdoor/client.go +2 −1
  • internal/dbdoor/client_test.go +21 −0
  • internal/dbdoor/restore.go +4 −9
  • internal/dbdoor/restore_test.go +34 −0

CI passed — run details

diff --git a/db.go b/db.go
index 1a5a7f39a..8efc220a4 100644
--- a/db.go
+++ b/db.go
@@ -444,21 +444,31 @@ var dbReadPragmas = map[string]bool{
// writable_schema and the rest), since a pragma is the one statement that
// can loosen what follows it.
//
-// The pragma rule is an invariant over the whole statement's tokens, not a
-// prefix check: SQLite skips byte-order marks, comments and whitespace
-// anywhere, and applies a pragma at prepare even under EXPLAIN. So wherever
-// a PRAGMA keyword appears, the pragma it names may be given a value only
-// if it is a read-only one, and only in the (argument) form.
+// The pragma rule reads the statement's tokens, not its raw prefix: SQLite
+// skips byte-order marks, comments and whitespace anywhere, and applies a
+// pragma at prepare even under EXPLAIN. So when the statement is a PRAGMA
+// (first token, or after EXPLAIN [QUERY PLAN]), the pragma it names may be
+// given a value only if it is a read-only one, and only in the (argument)
+// form. Elsewhere "pragma" is an identifier and a read may use it.
func checkQuerySQL(q string) error {
_, rest := splitFirstStatement(q)
if len(sqlTokens(rest)) > 0 {
return errOneStatement
}
toks := sqlTokens(q)
- for i, tk := range toks {
- if tk.kind != sqlWord || tk.text != "pragma" {
- continue
+ // A pragma is a statement, so it can only stand in statement position:
+ // first, or after EXPLAIN [QUERY PLAN]. Anywhere else "pragma" is an
+ // ordinary identifier (a column, an alias). One statement is enforced
+ // above, so there is no later statement position to check.
+ i := 0
+ if i < len(toks) && toks[i].kind == sqlWord && toks[i].text == "explain" {
+ i++
+ if i+1 < len(toks) && toks[i].kind == sqlWord && toks[i].text == "query" &&
+ toks[i+1].kind == sqlWord && toks[i+1].text == "plan" {
+ i += 2
}
+ }
+ if i < len(toks) && toks[i].kind == sqlWord && toks[i].text == "pragma" {
// PRAGMA [schema .] name [= value | (value)]
j := i + 1
name := ""
@@ -471,10 +481,10 @@ func checkQuerySQL(q string) error {
j += 2
}
if j >= len(toks) || toks[j].kind != sqlPunct || (toks[j].text != "=" && toks[j].text != "(") {
- continue // a bare read of the pragma's value
+ return nil // a bare read of the pragma's value
}
if toks[j].text == "(" && dbReadPragmas[name] {
- continue
+ return nil
}
return errPragmaChange
}
diff --git a/db_test.go b/db_test.go
index 94d7c688c..57c398e06 100644
--- a/db_test.go
+++ b/db_test.go
@@ -560,3 +560,18 @@ func TestQueryRefusesEveryDisguiseOfAPragmaChange(t *testing.T) {
}
}
}
+
+// "pragma" is only a statement in statement position. As a column name or
+// alias it is an ordinary identifier, and a read that uses it must run.
+func TestQueryAllowsPragmaAsAnIdentifier(t *testing.T) {
+ for _, q := range []string{
+ "SELECT pragma FROM (SELECT 1 AS pragma)",
+ "SELECT pragma = 1 FROM (SELECT 1 AS pragma)",
+ "SELECT x FROM (SELECT 1 AS x) WHERE x IN (SELECT 1 AS pragma)",
+ "EXPLAIN SELECT pragma = 1 FROM (SELECT 1 AS pragma)",
+ } {
+ if err := checkQuerySQL(q); err != nil {
+ t.Errorf("checkQuerySQL(%q) = %v, want it allowed", q, err)
+ }
+ }
+}
diff --git a/internal/dbdoor/client.go b/internal/dbdoor/client.go
index f14680b71..7632dceff 100644
--- a/internal/dbdoor/client.go
+++ b/internal/dbdoor/client.go
@@ -148,7 +148,8 @@ func doorError(res *http.Response) error {
// 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)
+ // Clamp the count before multiplying: a large one would wrap.
+ e.RetryAfter = time.Duration(min(secs, 60)) * time.Second
}
return e
}
diff --git a/internal/dbdoor/client_test.go b/internal/dbdoor/client_test.go
index 72e23353d..8b3952258 100644
--- a/internal/dbdoor/client_test.go
+++ b/internal/dbdoor/client_test.go
@@ -14,6 +14,7 @@ import (
"strings"
"sync"
"testing"
+ "time"
"github.com/benbjohnson/litestream"
lsfile "github.com/benbjohnson/litestream/file"
@@ -289,3 +290,23 @@ func TestClientSendsCoordinatesAndRange(t *testing.T) {
t.Fatalf("tail %q range %q, want %q %q", gotTail, gotRange, GetTail(3, 16, 32), RangeHeader(512, 100))
}
}
+
+// Retry-After is capped at a minute however large the door's number, and a
+// number too big to multiply into a Duration must not wrap round to a
+// short or negative wait.
+func TestClientCapsRetryAfter(t *testing.T) {
+ for _, v := range []string{"61", "3600", "18446744074", "9223372036", "9223372037"} {
+ do := func(context.Context, string, http.Header) (*http.Response, error) {
+ rec := httptest.NewRecorder()
+ rec.Header().Set("Retry-After", v)
+ rec.WriteHeader(http.StatusTooManyRequests)
+ io.WriteString(rec, `{"error":"busy"}`)
+ return rec.Result(), nil
+ }
+ _, err := NewClient(do).OpenLTXFile(context.Background(), 0, 1, 1, 0, 0)
+ var de *DoorError
+ if !errors.As(err, &de) || de.RetryAfter != time.Minute {
+ t.Errorf("Retry-After: %s gave %v, want the one-minute cap", v, de.RetryAfter)
+ }
+ }
+}
diff --git a/internal/dbdoor/restore.go b/internal/dbdoor/restore.go
index 385e23295..85d5ee802 100644
--- a/internal/dbdoor/restore.go
+++ b/internal/dbdoor/restore.go
@@ -147,16 +147,11 @@ type recorder struct {
}
func (r *recorder) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
- rc, err := r.Client.OpenLTXFile(ctx, level, minTXID, maxTXID, offset, size)
+ // The first open goes through download's loop too, so a full door's
+ // 429 waits out its Retry-After here rather than failing back to
+ // litestream, whose own quick retries ignore it.
+ rc, err := r.download(ctx, nil, level, minTXID, maxTXID, offset, size)
if err != nil {
- if errors.Is(err, os.ErrNotExist) {
- r.mu.Lock()
- r.vanished = true
- r.mu.Unlock()
- }
- return nil, err
- }
- if rc, err = r.download(ctx, rc, level, minTXID, maxTXID, offset, size); err != nil {
return nil, err
}
if offset != 0 {
diff --git a/internal/dbdoor/restore_test.go b/internal/dbdoor/restore_test.go
index 7b51ab40a..3836ed4df 100644
--- a/internal/dbdoor/restore_test.go
+++ b/internal/dbdoor/restore_test.go
@@ -487,3 +487,37 @@ func TestRestoreToStopsWhenCancelledDuringTheCheck(t *testing.T) {
}
assertEmptyDir(t, parent)
}
+
+// The FIRST open of a file meeting a full door (429) waits and asks again,
+// the same as a reopen does, instead of failing straight back to
+// litestream, whose own retries ignore Retry-After.
+func TestRestoreToRetriesABusyFirstOpen(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
+ refused := map[string]int{}
+ 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()
+ n := refused[tail]
+ refused[tail] = n + 1
+ mu.Unlock()
+ if rangeStart(h) == 0 && n < 4 {
+ return answer(http.StatusTooManyRequests, ""), nil
+ }
+ return door.do(ctx, tail, h)
+ }
+ got, err := RestoreTo(context.Background(), NewClient(do), t.TempDir())
+ if err != nil {
+ t.Fatalf("RestoreTo past four busy first opens per file = %v, want it to wait and finish", err)
+ }
+ defer os.RemoveAll(got.Dir)
+ if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows {
+ t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows)
+ }
+}