amadan

carlos / platform Public

dbdoor: the compatible handle protocol in the client (claim Plan 7 Task 1)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude bc75435a368b464a9f2346cccb0bfd22b8ab3c9e parent 76e0f52
4 files changed, +181 −5
  • internal/dbdoor/client.go +20 −4
  • internal/dbdoor/client_handles.go +65 −0
  • internal/dbdoor/client_handles_test.go +93 −0
  • internal/dbdoor/coords.go +3 −1
diff --git a/internal/dbdoor/client.go b/internal/dbdoor/client.go
index 7632dceff..65d637f2d 100644
--- a/internal/dbdoor/client.go
+++ b/internal/dbdoor/client.go
@@ -33,8 +33,9 @@ type Doer func(ctx context.Context, tail string, header http.Header) (*http.Resp
// 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
+ handles dbHandleState
+ do Doer
+ logger *slog.Logger
}
// NewClient returns a read-only replica client that reads through do.
@@ -91,8 +92,12 @@ func (c *Client) LTXFiles(ctx context.Context, level int, seek ltx.TXID, _ bool)
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 {
+ files, err := c.rememberFiles(level, list)
+ if err != nil {
+ return nil, err
+ }
+ infos := make([]*ltx.FileInfo, 0, len(files))
+ for _, f := range files {
if ltx.TXID(f.MinTXID) < seek {
continue
}
@@ -109,6 +114,13 @@ func (c *Client) LTXFiles(ctx context.Context, level int, seek ltx.TXID, _ bool)
// 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{}
+ handle, err := c.fileHandle(level, uint64(minTXID), uint64(maxTXID))
+ if err != nil {
+ return nil, err
+ }
+ if handle != "" {
+ h.Set(HandleHeader, handle)
+ }
h.Set("Range", RangeHeader(offset, size))
res, err := c.do(ctx, GetTail(level, uint64(minTXID), uint64(maxTXID)), h)
if err != nil {
@@ -116,6 +128,10 @@ func (c *Client) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID lt
}
switch res.StatusCode {
case http.StatusOK, http.StatusPartialContent:
+ if handle != "" && res.Header.Get(HandleHeader) != handle {
+ res.Body.Close()
+ return nil, errors.New("dbdoor: console did not enforce the pinned handle")
+ }
return res.Body, nil
case http.StatusNotFound:
res.Body.Close()
diff --git a/internal/dbdoor/client_handles.go b/internal/dbdoor/client_handles.go
new file mode 100644
index 000000000..4ee7e8f20
--- /dev/null
+++ b/internal/dbdoor/client_handles.go
@@ -0,0 +1,65 @@
+package dbdoor
+
+import (
+ "errors"
+ "fmt"
+ "sync"
+)
+
+const HandleHeader = "X-Carlos-DB-Handle"
+
+type dbHandleState struct {
+ mu sync.Mutex
+ pinned bool
+ files map[string]LTXFile
+}
+
+func (c *Client) rememberFiles(level int, list ListResponse) ([]LTXFile, error) {
+ c.handles.mu.Lock()
+ defer c.handles.mu.Unlock()
+ if list.Protocol != 0 && list.Protocol != 2 {
+ return nil, errors.New("dbdoor: unsupported listing protocol")
+ }
+ if c.handles.pinned && list.Protocol != 2 {
+ return nil, errors.New("dbdoor: pinned listing protocol disappeared")
+ }
+ if list.Protocol == 0 {
+ return list.Files, nil
+ }
+ files := make(map[string]LTXFile, len(c.handles.files)+len(list.Files))
+ for k, v := range c.handles.files {
+ files[k] = v
+ }
+ seen := map[string]bool{}
+ out := make([]LTXFile, 0, len(list.Files))
+ for _, f := range list.Files {
+ k := GetTail(level, f.MinTXID, f.MaxTXID)
+ if f.Level != level || f.MinTXID == 0 || f.MaxTXID < f.MinTXID || f.Handle == "" || f.Size < 0 || seen[k] {
+ return nil, errors.New("dbdoor: invalid pinned listing")
+ }
+ seen[k] = true
+ if old, ok := files[k]; ok {
+ if old.Size != f.Size {
+ return nil, errors.New("dbdoor: pinned file size changed; retry the database pull")
+ }
+ f = old // Preserve first handle and its metadata through retries/resumes.
+ }
+ files[k] = f
+ out = append(out, f)
+ }
+ c.handles.files = files
+ c.handles.pinned = true
+ return out, nil
+}
+
+func (c *Client) fileHandle(level int, lo, hi uint64) (string, error) {
+ c.handles.mu.Lock()
+ defer c.handles.mu.Unlock()
+ if !c.handles.pinned {
+ return "", nil
+ }
+ if f, ok := c.handles.files[GetTail(level, lo, hi)]; ok {
+ return f.Handle, nil
+ }
+ return "", fmt.Errorf("dbdoor: coordinate has no pinned handle: %d %s", level, Span(lo, hi))
+}
diff --git a/internal/dbdoor/client_handles_test.go b/internal/dbdoor/client_handles_test.go
new file mode 100644
index 000000000..cc0c5af00
--- /dev/null
+++ b/internal/dbdoor/client_handles_test.go
@@ -0,0 +1,93 @@
+package dbdoor
+
+import (
+ "context"
+ "encoding/json"
+ "io"
+ "net/http"
+ "strings"
+ "testing"
+
+ "github.com/superfly/ltx"
+)
+
+func TestHandleProtocolCompatibility(t *testing.T) {
+ for _, mode := range []string{"legacy", "pinned", "downgraded-get"} {
+ t.Run(mode, func(t *testing.T) {
+ c := NewClient(func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ out := &http.Response{StatusCode: 200, Header: http.Header{}}
+ if tail == ListTail(0) {
+ list := ListResponse{Files: []LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: 3}}}
+ if mode != "legacy" {
+ list.Protocol = 2
+ list.Files[0].Handle = "opaque"
+ }
+ b, _ := json.Marshal(list)
+ out.Body = io.NopCloser(strings.NewReader(string(b)))
+ } else {
+ want := ""
+ if mode != "legacy" {
+ want = "opaque"
+ }
+ if h.Get(HandleHeader) != want || h.Get("Range") != "bytes=1-2" {
+ t.Fatal(tail, h)
+ }
+ if mode == "pinned" {
+ out.Header.Set(HandleHeader, want)
+ }
+ out.Body = io.NopCloser(strings.NewReader("bc"))
+ }
+ return out, nil
+ })
+ it, err := c.LTXFiles(t.Context(), 0, 0, false)
+ if err != nil {
+ t.Fatal(err)
+ }
+ it.Close()
+ rc, err := c.OpenLTXFile(t.Context(), 0, ltx.TXID(1), ltx.TXID(1), 1, 2)
+ if mode == "downgraded-get" {
+ if err == nil {
+ rc.Close()
+ t.Fatal("accepted unpinned body")
+ }
+ return
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ b, err := io.ReadAll(rc)
+ rc.Close()
+ if err != nil || string(b) != "bc" {
+ t.Fatal(string(b), err)
+ }
+ })
+ }
+}
+
+func TestHandleProtocolNeverFallsBack(t *testing.T) {
+ c := NewClient(nil)
+ first := ListResponse{Protocol: 2, Files: []LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: 3, Handle: "first"}}}
+ if _, err := c.rememberFiles(0, first); err != nil {
+ t.Fatal(err)
+ }
+ again := ListResponse{Protocol: 2, Files: []LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: 3, Handle: "second"}}}
+ files, err := c.rememberFiles(0, again)
+ if err != nil || files[0].Handle != "first" {
+ t.Fatal(files, err)
+ }
+ again.Files[0].Size = 4
+ if _, err := c.rememberFiles(0, again); err == nil {
+ t.Fatal("metadata changed")
+ }
+ if _, err := c.rememberFiles(0, ListResponse{}); err == nil {
+ t.Fatal("listing downgraded")
+ }
+ if _, err := c.fileHandle(0, 2, 2); err == nil {
+ t.Fatal("missing handle fell back")
+ }
+ for _, bad := range []ListResponse{{Protocol: 3}, {Protocol: 2, Files: []LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1}}}} {
+ if _, err := NewClient(nil).rememberFiles(0, bad); err == nil {
+ t.Fatal("bad listing", bad)
+ }
+ }
+}
diff --git a/internal/dbdoor/coords.go b/internal/dbdoor/coords.go
index 67aba1ced..689b02d2f 100644
--- a/internal/dbdoor/coords.go
+++ b/internal/dbdoor/coords.go
@@ -32,6 +32,7 @@ const MaxLevel = 9
// MaxTXID are MANDATORY: Restore rejects Size < ltx.HeaderSize and
// ResumableReader uses Size to detect a premature EOF.
type LTXFile struct {
+ Handle string `json:"handle,omitempty"`
Level int `json:"level"`
MinTXID uint64 `json:"min_txid"`
MaxTXID uint64 `json:"max_txid"`
@@ -42,7 +43,8 @@ type LTXFile struct {
// ListResponse is ordered by MinTXID ASCENDING: restoreLevelCursor assumes
// filename order, which S3 gives because the names are zero-padded hex.
type ListResponse struct {
- Files []LTXFile `json:"files"`
+ Protocol int `json:"protocol,omitempty"`
+ Files []LTXFile `json:"files"`
}
// Base is the door's root for one instance: everything below it is a