amadan

carlos / platform Public

dbdoor: a replica client with no write methods

The CLI's litestream.ReplicaClient over the db door (spec 2026-09-02 §6,
§12). It lists and opens LTX files by coordinate and has no code that
writes a replica: WriteLTXFile, DeleteLTXFiles and DeleteAll return an
error and nothing else. It is not a ReplicaClientV3. A test restores a
real replica through it and counts every method call, so a litestream
upgrade that started calling a mutating method on the restore path fails
here rather than relying on the error.

The client works in coordinate tails through a Doer the CLI supplies, so
the console URL, bearer and account segment stay the CLI's. A 404 on a
file is os.ErrNotExist; other refusals carry the console's sentence.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude bd9b6d2006bedd549b2501fd8b0189011cca4238 parent e6d358e
2 files changed, +436 −0
  • internal/dbdoor/client.go +145 −0
  • internal/dbdoor/client_test.go +291 −0
diff --git a/internal/dbdoor/client.go b/internal/dbdoor/client.go
new file mode 100644
index 000000000..8c3e9246a
--- /dev/null
+++ b/internal/dbdoor/client.go
@@ -0,0 +1,145 @@
+package dbdoor
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "net/http"
+ "os"
+ "strings"
+
+ "github.com/benbjohnson/litestream"
+ "github.com/superfly/ltx"
+)
+
+// Doer sends one GET to the door for one instance. tail is relative to the
+// instance's door root (ListTail or GetTail); the caller owns the console
+// URL, the bearer and the account segment, which is why the client never
+// sees them. The caller's response body is the client's to close.
+type Doer func(ctx context.Context, tail string, header http.Header) (*http.Response, error)
+
+// Client is the CLI's litestream.ReplicaClient over the db door (spec
+// 2026-09-02 §6, "The read-only replica client"), the way internal/fleetapi
+// implements release.Store over the edge store door: the member has no AWS
+// credentials, by design.
+//
+// It supplies the five methods a restore needs (Type, Init, LTXFiles,
+// OpenLTXFile, SetLogger). The three that mutate a replica return
+// errReadOnly and have no other body. It must never implement
+// litestream.ReplicaClientV3, or Restore would take the v0.3 path.
+type Client struct {
+ do Doer
+ logger *slog.Logger
+}
+
+// NewClient returns a read-only replica client that reads through do.
+func NewClient(do Doer) *Client {
+ return &Client{do: do, logger: slog.Default()}
+}
+
+// errReadOnly is not a guard that can be switched off: these three methods
+// have no other body. The two-replicators hazard internal/caps and
+// internal/lease both warn about is unreachable from this client by
+// construction, and a future edit that wanted to reach it would have to add
+// the capability first.
+var errReadOnly = errors.New("dbdoor: this client is read-only; it restores replicas and never writes them")
+
+// DoorError is a refusal from the door, carrying the console's own sentence
+// (its {"error"} body) so the CLI can show it as written.
+type DoorError struct {
+ Status int
+ Message string
+}
+
+func (e *DoorError) Error() string {
+ if e.Message == "" {
+ return fmt.Sprintf("the console answered %d", e.Status)
+ }
+ return e.Message
+}
+
+func (c *Client) Type() string { return "carlos-db" }
+
+// Init is a no-op: the transport needs no setup, and litestream requires
+// it to be idempotent.
+func (c *Client) Init(context.Context) error { return nil }
+
+func (c *Client) SetLogger(l *slog.Logger) { c.logger = l }
+
+// LTXFiles lists one level. seek is filtered here (wire specific 3), with
+// litestream's own rule: files whose MinTXID is below seek are skipped.
+// useMetadata is ignored; nothing in a non-timestamp restore reads
+// CreatedAt except debug logs.
+func (c *Client) LTXFiles(ctx context.Context, level int, seek ltx.TXID, _ bool) (ltx.FileIterator, error) {
+ res, err := c.do(ctx, ListTail(level), http.Header{})
+ if err != nil {
+ return nil, err
+ }
+ defer res.Body.Close()
+ if res.StatusCode != http.StatusOK {
+ return nil, doorError(res)
+ }
+ var list ListResponse
+ if err := json.NewDecoder(io.LimitReader(res.Body, 16<<20)).Decode(&list); err != nil {
+ return nil, fmt.Errorf("dbdoor: unreadable listing for level %d: %w", level, err)
+ }
+ infos := make([]*ltx.FileInfo, 0, len(list.Files))
+ for _, f := range list.Files {
+ if ltx.TXID(f.MinTXID) < seek {
+ continue
+ }
+ infos = append(infos, &ltx.FileInfo{
+ Level: level, MinTXID: ltx.TXID(f.MinTXID), MaxTXID: ltx.TXID(f.MaxTXID),
+ Size: f.Size, CreatedAt: f.CreatedAt,
+ })
+ }
+ return ltx.NewFileInfoSliceIterator(infos), nil
+}
+
+// OpenLTXFile opens one file by coordinate, sending the S3 client's exact
+// Range form. A 404 is os.ErrNotExist (wire specific 5): that is what lets
+// RestoreTo tell a file that vanished under the plan from a refusal.
+func (c *Client) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
+ h := http.Header{}
+ h.Set("Range", RangeHeader(offset, size))
+ res, err := c.do(ctx, GetTail(level, uint64(minTXID), uint64(maxTXID)), h)
+ if err != nil {
+ return nil, err
+ }
+ switch res.StatusCode {
+ case http.StatusOK, http.StatusPartialContent:
+ return res.Body, nil
+ case http.StatusNotFound:
+ res.Body.Close()
+ return nil, fmt.Errorf("dbdoor: level %d %s: %w", level, Span(uint64(minTXID), uint64(maxTXID)), os.ErrNotExist)
+ default:
+ defer res.Body.Close()
+ return nil, doorError(res)
+ }
+}
+
+func (c *Client) WriteLTXFile(context.Context, int, ltx.TXID, ltx.TXID, io.Reader) (*ltx.FileInfo, error) {
+ return nil, errReadOnly
+}
+
+func (c *Client) DeleteLTXFiles(context.Context, []*ltx.FileInfo) error { return errReadOnly }
+
+func (c *Client) DeleteAll(context.Context) error { return errReadOnly }
+
+// doorError reads a refusal's {"error"} sentence. A body that is not that
+// shape (the stdlib's 404 text, a proxy's page) keeps only the status.
+func doorError(res *http.Response) error {
+ raw, _ := io.ReadAll(io.LimitReader(res.Body, 64<<10))
+ 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)}
+ }
+ return &DoorError{Status: res.StatusCode}
+}
+
+var _ litestream.ReplicaClient = (*Client)(nil)
diff --git a/internal/dbdoor/client_test.go b/internal/dbdoor/client_test.go
new file mode 100644
index 000000000..72e23353d
--- /dev/null
+++ b/internal/dbdoor/client_test.go
@@ -0,0 +1,291 @@
+package dbdoor
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "sync"
+ "testing"
+
+ "github.com/benbjohnson/litestream"
+ lsfile "github.com/benbjohnson/litestream/file"
+ "github.com/superfly/ltx"
+
+ "github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
+)
+
+var _ litestream.ReplicaClient = (*Client)(nil)
+
+// noNetwork fails the test if the client ever reaches for the wire.
+func noNetwork(t *testing.T) Doer {
+ return func(context.Context, string, http.Header) (*http.Response, error) {
+ t.Error("the client made a request")
+ return nil, errors.New("no network in this test")
+ }
+}
+
+// The claim the whole design rests on: this client cannot corrupt a
+// replica, because it has no code that writes to one. And it refuses
+// without touching the network.
+func TestClientMutatingMethodsAlwaysError(t *testing.T) {
+ c := NewClient(noNetwork(t))
+ ctx := context.Background()
+ if _, err := c.WriteLTXFile(ctx, 0, 1, 2, strings.NewReader("x")); err == nil {
+ t.Error("WriteLTXFile returned nil; this client must never write")
+ }
+ if err := c.DeleteLTXFiles(ctx, []*ltx.FileInfo{{Level: 0, MinTXID: 1, MaxTXID: 1}}); err == nil {
+ t.Error("DeleteLTXFiles returned nil; this client must never delete")
+ }
+ if err := c.DeleteAll(ctx); err == nil {
+ t.Error("DeleteAll returned nil; this client must never delete")
+ }
+}
+
+// Restore type-asserts for ReplicaClientV3 and would take the v0.3 path if
+// this satisfied it. It must not.
+func TestClientIsNotAV3ReplicaClient(t *testing.T) {
+ var c any = NewClient(noNetwork(t))
+ if _, ok := c.(litestream.ReplicaClientV3); ok {
+ t.Fatal("client satisfies ReplicaClientV3; Restore would take the v0.3 path")
+ }
+}
+
+// testDoor serves a file replica over the door's wire shape: the same
+// coordinates, the same Range forms, the same 404 for a missing file. It
+// stands in for the console here; internal/console tests the real door
+// with this package's client.
+type testDoor struct {
+ t *testing.T
+ rc *lsfile.ReplicaClient
+ mu sync.Mutex
+ lists map[int]int // level -> listing count
+ gone map[string]bool
+ opened int
+}
+
+func newTestDoor(t *testing.T, dir string) *testDoor {
+ return &testDoor{t: t, rc: lsfile.NewReplicaClient(dir), lists: map[int]int{}, gone: map[string]bool{}}
+}
+
+func (d *testDoor) do(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ rec := httptest.NewRecorder()
+ d.serve(ctx, rec, tail, h)
+ return rec.Result(), nil
+}
+
+func (d *testDoor) serve(ctx context.Context, w http.ResponseWriter, tail string, h http.Header) {
+ parts := strings.Split(strings.TrimPrefix(tail, "/ltx/"), "/")
+ level, err := strconv.Atoi(parts[0])
+ if err != nil {
+ http.NotFound(w, nil)
+ return
+ }
+ if len(parts) == 1 {
+ d.mu.Lock()
+ d.lists[level]++
+ d.mu.Unlock()
+ itr, _ := d.rc.LTXFiles(ctx, level, 0, false)
+ out := ListResponse{Files: []LTXFile{}}
+ for itr.Next() {
+ f := itr.Item()
+ out.Files = append(out.Files, LTXFile{Level: level, MinTXID: uint64(f.MinTXID), MaxTXID: uint64(f.MaxTXID), Size: f.Size, CreatedAt: f.CreatedAt})
+ }
+ itr.Close()
+ json.NewEncoder(w).Encode(out)
+ return
+ }
+ min, max, err := ParseSpan(parts[1])
+ if err != nil {
+ http.NotFound(w, nil)
+ return
+ }
+ d.mu.Lock()
+ gone := d.gone[parts[1]]
+ if gone {
+ delete(d.gone, parts[1]) // gone once: a fresh plan finds it again
+ }
+ d.opened++
+ d.mu.Unlock()
+ if gone {
+ http.NotFound(w, nil)
+ return
+ }
+ off, size, err := ParseRange(h.Get("Range"))
+ if err != nil {
+ w.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
+ return
+ }
+ rc, err := d.rc.OpenLTXFile(ctx, level, ltx.TXID(min), ltx.TXID(max), off, size)
+ if errors.Is(err, os.ErrNotExist) {
+ http.NotFound(w, nil)
+ return
+ }
+ if err != nil {
+ w.WriteHeader(http.StatusInternalServerError)
+ return
+ }
+ defer rc.Close()
+ w.WriteHeader(http.StatusPartialContent)
+ io.Copy(w, rc)
+}
+
+// counting wraps the client and counts every method litestream calls on it.
+type counting struct {
+ *Client
+ mu sync.Mutex
+ counts map[string]int
+}
+
+func (c *counting) inc(m string) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.counts[m]++
+}
+
+func (c *counting) WriteLTXFile(ctx context.Context, level int, min, max ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
+ c.inc("WriteLTXFile")
+ return c.Client.WriteLTXFile(ctx, level, min, max, r)
+}
+
+func (c *counting) DeleteLTXFiles(ctx context.Context, a []*ltx.FileInfo) error {
+ c.inc("DeleteLTXFiles")
+ return c.Client.DeleteLTXFiles(ctx, a)
+}
+
+func (c *counting) DeleteAll(ctx context.Context) error {
+ c.inc("DeleteAll")
+ return c.Client.DeleteAll(ctx)
+}
+
+func (c *counting) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
+ c.inc("LTXFiles")
+ return c.Client.LTXFiles(ctx, level, seek, useMetadata)
+}
+
+func (c *counting) OpenLTXFile(ctx context.Context, level int, min, max ltx.TXID, offset, size int64) (io.ReadCloser, error) {
+ c.inc("OpenLTXFile")
+ return c.Client.OpenLTXFile(ctx, level, min, max, offset, size)
+}
+
+// The methods erroring is not enough: a restore that CALLED one would be
+// relying on an error path. This asserts a full restore never reaches them,
+// which is what catches a litestream upgrade changing the restore path.
+func TestRestoreNeverCallsAMutatingMethod(t *testing.T) {
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 3)
+ door := newTestDoor(t, dir)
+ c := &counting{Client: NewClient(door.do), counts: map[string]int{}}
+ out := filepath.Join(t.TempDir(), "out.db")
+ opt := litestream.NewRestoreOptions()
+ opt.OutputPath = out
+ if err := litestream.NewReplicaWithClient(nil, c).Restore(context.Background(), opt); err != nil {
+ t.Fatal(err)
+ }
+ if n := c.counts["WriteLTXFile"] + c.counts["DeleteLTXFiles"] + c.counts["DeleteAll"]; n != 0 {
+ t.Fatalf("restore called a mutating method: %v", c.counts)
+ }
+ if c.counts["LTXFiles"] == 0 || c.counts["OpenLTXFile"] == 0 {
+ t.Fatalf("restore did not go through the client at all: %v", c.counts)
+ }
+ if rows := countRows(t, out); rows != 3*dbdoortest.Rows {
+ t.Fatalf("restored %d rows, want %d", rows, 3*dbdoortest.Rows)
+ }
+}
+
+func countRows(t *testing.T, path string) int {
+ t.Helper()
+ db, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ var n int
+ if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
+ t.Fatal(err)
+ }
+ return n
+}
+
+// seek is not on the wire (wire specific 3): the client filters locally,
+// with litestream's own rule (MinTXID >= seek).
+func TestClientFiltersSeekLocally(t *testing.T) {
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 3)
+ c := NewClient(newTestDoor(t, dir).do)
+ all := collect(t, c, 0, 0)
+ if len(all) < 2 {
+ t.Fatalf("want several files, got %d", len(all))
+ }
+ seek := all[1].MinTXID
+ got := collect(t, c, 0, seek)
+ if len(got) != len(all)-1 || got[0].MinTXID != seek {
+ t.Fatalf("seek %d = %d files starting %d, want %d starting %d", seek, len(got), got[0].MinTXID, len(all)-1, seek)
+ }
+}
+
+func collect(t *testing.T, c *Client, level int, seek ltx.TXID) []*ltx.FileInfo {
+ t.Helper()
+ itr, err := c.LTXFiles(context.Background(), level, seek, false)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer itr.Close()
+ var out []*ltx.FileInfo
+ for itr.Next() {
+ out = append(out, itr.Item())
+ }
+ return out
+}
+
+// A 404 on a file is os.ErrNotExist, which is what the whole-Restore retry
+// keys on; any other refusal carries the console's own sentence.
+func TestClientMapsTheDoorsAnswers(t *testing.T) {
+ answer := func(status int, body string) Doer {
+ return func(context.Context, string, http.Header) (*http.Response, error) {
+ rec := httptest.NewRecorder()
+ rec.WriteHeader(status)
+ io.WriteString(rec, body)
+ return rec.Result(), nil
+ }
+ }
+ ctx := context.Background()
+ if _, err := NewClient(answer(404, "404 page not found\n")).OpenLTXFile(ctx, 0, 1, 1, 0, 0); !errors.Is(err, os.ErrNotExist) {
+ t.Errorf("open on 404 = %v, want os.ErrNotExist", err)
+ }
+ _, err := NewClient(answer(409, `{"error":"the data is elsewhere"}`)).LTXFiles(ctx, 0, 0, false)
+ var de *DoorError
+ if !errors.As(err, &de) || de.Status != 409 || de.Message != "the data is elsewhere" {
+ t.Errorf("list on 409 = %v, want a DoorError carrying the sentence", err)
+ }
+ if _, err := NewClient(answer(409, `{"error":"x"}`)).OpenLTXFile(ctx, 0, 1, 1, 0, 0); errors.Is(err, os.ErrNotExist) || !errors.As(err, &de) {
+ t.Errorf("open on 409 = %v, want a DoorError and not ErrNotExist", err)
+ }
+}
+
+// The client sends the S3 client's exact Range header and the coordinate
+// tail, nothing else.
+func TestClientSendsCoordinatesAndRange(t *testing.T) {
+ var gotTail, gotRange string
+ c := NewClient(func(_ context.Context, tail string, h http.Header) (*http.Response, error) {
+ gotTail, gotRange = tail, h.Get("Range")
+ rec := httptest.NewRecorder()
+ rec.WriteHeader(http.StatusPartialContent)
+ return rec.Result(), nil
+ })
+ rc, err := c.OpenLTXFile(context.Background(), 3, 16, 32, 512, 100)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rc.Close()
+ if gotTail != GetTail(3, 16, 32) || gotRange != RangeHeader(512, 100) {
+ t.Fatalf("tail %q range %q, want %q %q", gotTail, gotRange, GetTail(3, 16, 32), RangeHeader(512, 100))
+ }
+}