amadan

carlos / platform Public

console, dbdoor: the door lists genselect's plan for the holder's generation (Plan 7 follow-up)

Replaces the refusal of a non-empty holder.generation with
generation-aware selection through internal/genselect; refusals map to a
retryable 409; handles still pin exact versions.
Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude ce9aef972af5f9fe738962ea0b345e7c4570e246 parent 3d143b2
9 files changed, +326 −63
  • internal/console/apidbclaim.go +25 −39
  • internal/console/apidbclaim_generation_test.go +164 −0
  • internal/console/apidbclaim_http_test.go +33 −7
  • internal/console/apidbclaim_range_test.go +3 −0
  • internal/console/apidbclaim_test.go +5 −8
  • internal/dbdoor/client.go +19 −5
  • internal/dbdoor/client_handles.go +29 −2
  • internal/dbdoor/client_handles_test.go +45 −0
  • internal/dbdoor/coords.go +3 −2

CI passed — run details

diff --git a/internal/console/apidbclaim.go b/internal/console/apidbclaim.go
index 964d66466..f7e71edec 100644
--- a/internal/console/apidbclaim.go
+++ b/internal/console/apidbclaim.go
@@ -14,12 +14,12 @@ import (
"net/http"
"os"
"reflect"
- "sort"
"strconv"
"strings"
"time"
"github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/genselect"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/provision"
@@ -32,7 +32,6 @@ var (
errDBNotAdopted = errors.New("host not adopted")
errDBClaimChanged = errors.New("host claim changed; retry the database pull")
errDBClaimClosed = errors.New("host is not accepting database reads")
- errDBGeneration = errors.New("database generation requires a generation-aware reader")
errDBHandle = errors.New("invalid database file handle")
errDBHandleExpired = errors.New("database file handle expired; retry the database pull")
)
@@ -143,11 +142,6 @@ func (s *Server) dbClaim(ctx context.Context, g dbRequest, rec provision.Record)
if h.Location == "" {
return obs, errDBClaimClosed
}
- // The moves generation restore contract is not provided by Plans 1/2.
- // Never present root files as the installed generation's database.
- if h.Generation != "" {
- return obs, errDBGeneration
- }
return obs, nil
}
@@ -164,7 +158,7 @@ func (s *Server) dbClaimFailure(w http.ResponseWriter, r *http.Request, g dbRequ
switch {
case errors.Is(err, os.ErrNotExist):
http.NotFound(w, r)
- case errors.Is(err, errDBNotAdopted), errors.Is(err, errDBClaimChanged), errors.Is(err, errDBClaimClosed), errors.Is(err, errDBGeneration):
+ case errors.Is(err, errDBNotAdopted), errors.Is(err, errDBClaimChanged), errors.Is(err, errDBClaimClosed):
w.Header().Set("Retry-After", "1")
jsonError(w, http.StatusConflict, err.Error())
case errors.Is(err, errDBHandleExpired):
@@ -177,35 +171,21 @@ func (s *Server) dbClaimFailure(w http.ResponseWriter, r *http.Request, g dbRequ
}
func (s *Server) listDBVersions(ctx context.Context, g dbRequest, level int) (dbdoor.ListResponse, error) {
- out := dbdoor.ListResponse{Protocol: 2, Files: []dbdoor.LTXFile{}}
+ // Select the complete baseline once; the client retains it across levels.
+ out := dbdoor.ListResponse{Protocol: 2, Plan: true, Files: []dbdoor.LTXFile{}}
h := g.claim.Claim.Holder
- prefix := dbLevelPrefix(h.Location, level)
- versions, err := s.cfg.DBVersions.ListVersions(ctx, s.cfg.HibernateBucket, prefix)
+ plan, err := genselect.Select(ctx, s.cfg.DBVersions, genselect.Request{
+ Bucket: s.cfg.HibernateBucket, Region: s.homeRegionOut(s.cfg.Region), Host: g.host, Holder: *h,
+ })
if err != nil {
- return dbdoor.ListResponse{}, err
+ return dbdoor.ListResponse{}, errDBClaimChanged
}
- seen := map[string]bool{}
- for _, v := range versions {
- if !strings.HasPrefix(v.Key, prefix) {
- return dbdoor.ListResponse{}, errors.New("db: malformed version listing")
- }
- if !v.IsLatest || v.DeleteMarker {
- continue
- }
- name := strings.TrimPrefix(v.Key, prefix)
- lo, hi, err := ltx.ParseFilename(name)
- if err != nil || lo == 0 || hi < lo || v.Key != dbVersionKey(h.Location, level, uint64(lo), uint64(hi)) {
- continue
- }
- if v.VersionID == "" || v.VersionID == "null" || v.Size < 0 || seen[v.Key] {
- return dbdoor.ListResponse{}, errors.New("db: malformed version listing")
- }
- seen[v.Key] = true
- token, err := signDBHandle(s.dbHandleKey, dbFileHandle{V: 1, Host: g.host, Location: h.Location, Key: v.Key, VersionID: v.VersionID, Incarnation: h.Incarnation, Expiry: time.Now().Add(dbHandleTTL).Unix(), Account: g.acct.Sqid, App: g.app, Size: v.Size})
+ for _, f := range plan.Files {
+ token, err := signDBHandle(s.dbHandleKey, dbFileHandle{V: 1, Host: g.host, Location: h.Location, Key: f.Key, VersionID: f.VersionID, Incarnation: h.Incarnation, Expiry: time.Now().Add(dbHandleTTL).Unix(), Account: g.acct.Sqid, App: g.app, Size: f.Size})
if err != nil {
return dbdoor.ListResponse{}, err
}
- out.Files = append(out.Files, dbdoor.LTXFile{Level: level, MinTXID: uint64(lo), MaxTXID: uint64(hi), Size: v.Size, CreatedAt: v.LastModified, Handle: token})
+ out.Files = append(out.Files, dbdoor.LTXFile{Level: f.Level, MinTXID: f.MinTXID, MaxTXID: f.MaxTXID, Size: f.Size, Handle: token})
}
o2, err := hostclaim.Read(ctx, s.cfg.Releases, g.host)
if err != nil {
@@ -214,22 +194,28 @@ func (s *Server) listDBVersions(ctx context.Context, g dbRequest, level int) (db
if !sameDBAdmission(g.claim, o2) {
return dbdoor.ListResponse{}, errDBClaimChanged
}
- sort.Slice(out.Files, func(i, j int) bool {
- if out.Files[i].MinTXID == out.Files[j].MinTXID {
- return out.Files[i].MaxTXID < out.Files[j].MaxTXID
- }
- return out.Files[i].MinTXID < out.Files[j].MinTXID
- })
return out, nil
}
+func dbHandleCoordinate(h dbFileHandle, level int, lo, hi uint64) bool {
+ if h.Key == dbVersionKey(h.Location, level, lo, hi) {
+ return true
+ }
+ rel, ok := strings.CutPrefix(h.Key, h.Location+"/g-")
+ if !ok {
+ return false
+ }
+ gen, tail, ok := strings.Cut(rel, "/")
+ return ok && gen != "" && gen != "." && gen != ".." && gen != hostclaim.RootGeneration && !strings.ContainsAny(gen, "\\\x00") && tail == dbVersionKey("", level, lo, hi)[1:]
+}
+
func (s *Server) pinnedDBReader(r *http.Request, g dbRequest, level int, lo, hi uint64) (*dbVersionReader, error) {
h, err := verifyDBHandle(s.dbHandleKey, r.Header.Get(dbdoor.HandleHeader), time.Now())
if err != nil {
return nil, err
}
holder := g.claim.Claim.Holder
- if h.Host != g.host || h.Account != g.acct.Sqid || h.App != g.app || h.Location != holder.Location || h.Incarnation != holder.Incarnation || h.Key != dbVersionKey(h.Location, level, lo, hi) {
+ if h.Host != g.host || h.Account != g.acct.Sqid || h.App != g.app || h.Location != holder.Location || h.Incarnation != holder.Incarnation || !dbHandleCoordinate(h, level, lo, hi) {
return nil, errDBHandle
}
return &dbVersionReader{store: s.cfg.DBVersions, bucket: s.cfg.HibernateBucket, handle: h}, nil
@@ -247,7 +233,7 @@ func (v *dbVersionReader) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.Fi
return nil, errDBHandle
}
func (v *dbVersionReader) OpenLTXFile(ctx context.Context, level int, lo, hi ltx.TXID, offset, size int64) (io.ReadCloser, error) {
- if v.handle.Key != dbVersionKey(v.handle.Location, level, uint64(lo), uint64(hi)) {
+ if !dbHandleCoordinate(v.handle, level, uint64(lo), uint64(hi)) {
return nil, errDBHandle
}
n, err := dbVersionRange(v.handle.Size, offset, size)
diff --git a/internal/console/apidbclaim_generation_test.go b/internal/console/apidbclaim_generation_test.go
new file mode 100644
index 000000000..2f935db56
--- /dev/null
+++ b/internal/console/apidbclaim_generation_test.go
@@ -0,0 +1,164 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "io"
+ "net/http"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/superfly/ltx"
+)
+
+func doorLTXHeader(t *testing.T, lo, hi uint64) []byte {
+ t.Helper()
+ h := ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, MinTXID: ltx.TXID(lo), MaxTXID: ltx.TXID(hi)}
+ b, err := h.MarshalBinary()
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b
+}
+
+func TestDoorPlanHTTPGenerationSelection(t *testing.T) {
+ for _, mode := range []string{"origin", "published", "unfinished", "unfinished-origin", "gap", "missing-pin", "partial-list", "disagreement", "bad-header", "holder-change", "op-change"} {
+ t.Run(mode, func(t *testing.T) {
+ fx := dbFixture(t)
+ model := enableDoorPlan(t, fx)
+ fx.srv.cfg.HomeRegion = "ie"
+ obs, err := hostclaim.Read(t.Context(), fx.rel, dbTestHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c := cloneDoorClaim(obs.Claim)
+ loc := c.Holder.Location
+ gen := "live"
+ if mode == "origin" || mode == "unfinished-origin" {
+ gen = ""
+ }
+ c.Holder.Generation = gen
+ base := loc
+ if gen != "" {
+ base += "/g-" + gen
+ }
+ put := func(level int, lo, hi uint64) genselect.File {
+ key := dbVersionKey(base, level, lo, hi)
+ b := doorLTXHeader(t, lo, hi)
+ id := model.Put("data-test", key, b)
+ return genselect.File{Key: key, VersionID: id, Level: level, MinTXID: lo, MaxTXID: hi, Size: int64(len(b))}
+ }
+ root := put(9, 1, 11)
+ cont := put(1, 10, 13)
+ put(0, 11, 11)
+ put(0, 12, 12)
+ if gen != "" {
+ marker := genselect.Marker{Op: gen, Host: dbTestHost, Incarnation: c.Holder.Incarnation, Box: "box", InstalledAt: time.Unix(1, 0).UTC(), Pin: genselect.Pin{SnapshotKey: root.Key, SnapshotVersion: root.VersionID, SnapshotTXID: 11}}
+ raw, err := json.Marshal(marker)
+ if err != nil {
+ t.Fatal(err)
+ }
+ model.Put("data-test", loc+"/generation.json", raw)
+ model.Put("data-test", root.Key, []byte("late duplicate must never be selected"))
+ model.Put("data-test", dbVersionKey(loc, 9, 1, 99), doorLTXHeader(t, 1, 99))
+ }
+ switch mode {
+ case "unfinished", "unfinished-origin":
+ predecessor := gen
+ if predecessor == "" {
+ predecessor = hostclaim.RootGeneration
+ }
+ c.Holder.Generation = "next"
+ c.Holder.Admissions = []hostclaim.AdmissionRecord{{Generation: "next", Predecessor: predecessor, Region: "ie", PredecessorRegion: "ie", Mode: "cold"}}
+ case "gap":
+ model.BareDelete("data-test", cont.Key)
+ model.BareDelete("data-test", dbVersionKey(base, 0, 12, 12))
+ put(0, 13, 13)
+ case "missing-pin":
+ if err := model.DeleteVersion(t.Context(), "data-test", root.Key, root.VersionID); err != nil {
+ t.Fatal(err)
+ }
+ case "partial-list":
+ model.PageSize, model.FailListAfterPages = 1, 1
+ case "disagreement":
+ c.Holder.Generation = "unknown"
+ case "bad-header":
+ model.Put("data-test", dbVersionKey(base, 9, 1, 20), []byte("broken successor"))
+ }
+ writeDoorClaim(t, fx.srv, dbRequest{host: dbTestHost}, c)
+ if mode == "holder-change" || mode == "op-change" {
+ fx.srv.cfg.DBVersions = doorOpenHook{VersionedStore: model, open: func(ctx context.Context, bucket, key, version string) (io.ReadCloser, error) {
+ rc, err := model.OpenVersion(ctx, bucket, key, version)
+ next := cloneDoorClaim(c)
+ if mode == "holder-change" {
+ next.Holder.Session++
+ } else {
+ next.Op.Exec = "changed"
+ }
+ writeDoorClaim(t, fx.srv, dbRequest{host: dbTestHost}, next)
+ return rc, err
+ }}
+ }
+ status, h, body := rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 9)), fx.member, "")
+ if mode != "origin" && mode != "published" && mode != "unfinished" && mode != "unfinished-origin" {
+ if status != 409 || h.Get("Retry-After") != "1" || strings.Contains(body, "files") || strings.Contains(body, "instances/") || strings.Contains(body, "genselect:") {
+ t.Fatalf("refusal: %d %v %s", status, h, body)
+ }
+ return
+ }
+ if status != 200 {
+ t.Fatalf("list: %d %s", status, body)
+ }
+ var list dbdoor.ListResponse
+ if err := json.Unmarshal([]byte(body), &list); err != nil {
+ t.Fatal(err)
+ }
+ if list.Protocol != 2 || len(list.Files) != 2 {
+ t.Fatalf("list = %+v", list)
+ }
+ for i, want := range []genselect.File{root, cont} {
+ f := list.Files[i]
+ handle, err := verifyDBHandle(fx.srv.dbHandleKey, f.Handle, time.Now())
+ if err != nil || handle.Key != want.Key || handle.VersionID != want.VersionID || handle.Size != want.Size || f.Size != want.Size || f.Level != want.Level || f.MinTXID != want.MinTXID || f.MaxTXID != want.MaxTXID {
+ t.Fatalf("file %d: %+v %+v %v; want %+v", i, f, handle, err, want)
+ }
+ model.Put("data-test", want.Key, []byte("overwritten after listing"))
+ req, err := http.NewRequest("GET", fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, f.Level, f.MinTXID, f.MaxTXID)), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Header.Set("Authorization", "Bearer "+fx.member)
+ req.Header.Set(dbdoor.HandleHeader, f.Handle)
+ req.Header.Set("Range", "bytes=2-7")
+ res, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, err := io.ReadAll(res.Body)
+ res.Body.Close()
+ if err != nil || res.StatusCode != 206 || !bytes.Equal(got, doorLTXHeader(t, want.MinTXID, want.MaxTXID)[2:8]) || res.Header.Get(dbdoor.HandleHeader) != f.Handle {
+ t.Fatalf("get: %d %x %v", res.StatusCode, got, err)
+ }
+ if err := model.DeleteVersion(t.Context(), "data-test", want.Key, want.VersionID); err != nil {
+ t.Fatal(err)
+ }
+ res, err = http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ res.Body.Close()
+ if res.StatusCode != 404 {
+ t.Fatalf("missing version fell back: %d", res.StatusCode)
+ }
+ }
+ if fx.opens.Load() != 0 {
+ t.Fatal("opened unversioned replica")
+ }
+ })
+ }
+}
diff --git a/internal/console/apidbclaim_http_test.go b/internal/console/apidbclaim_http_test.go
index bde54e1b0..7c461700a 100644
--- a/internal/console/apidbclaim_http_test.go
+++ b/internal/console/apidbclaim_http_test.go
@@ -15,11 +15,13 @@ import (
"strings"
"sync/atomic"
"testing"
+ "time"
lsfile "github.com/benbjohnson/litestream/file"
"github.com/carlosframework/platform/internal/dbdoor"
"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
+ "github.com/carlosframework/platform/internal/genselect"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/lease/leasetest"
"github.com/carlosframework/platform/internal/provision"
@@ -44,6 +46,7 @@ func enableDoorPlan(t *testing.T, fx *dbFx) *leasetest.VersionedBucket {
}
m := leasetest.NewVersionedBucket()
m.AddBucket("data-test")
+ fx.srv.cfg.HomeRegion = "ie"
fx.srv.cfg.HostClaim = true
fx.srv.cfg.DBVersions = m
secret, err := fx.st.SigningKey()
@@ -55,12 +58,34 @@ func enableDoorPlan(t *testing.T, fx *dbFx) *leasetest.VersionedBucket {
}
func TestDoorPlanHTTPCompatibilityAndRestore(t *testing.T) {
- for _, enabled := range []bool{false, true} {
- t.Run(fmt.Sprint(enabled), func(t *testing.T) {
+ for _, mode := range []string{"legacy", "origin", "published", "unfinished"} {
+ t.Run(mode, func(t *testing.T) {
+ enabled := mode != "legacy"
fx := dbFixture(t)
fx.replicate(t, dbTestHost, 3)
if enabled {
model := enableDoorPlan(t, fx)
+ location := "instances/" + dbTestHost
+ if mode == "published" || mode == "unfinished" {
+ obs, err := hostclaim.Read(t.Context(), fx.rel, dbTestHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c := cloneDoorClaim(obs.Claim)
+ c.Holder.Generation = "live"
+ marker := genselect.Marker{Op: "live", Host: dbTestHost, Incarnation: c.Holder.Incarnation, Box: "box", InstalledAt: time.Unix(1, 0).UTC()}
+ raw, err := json.Marshal(marker)
+ if err != nil {
+ t.Fatal(err)
+ }
+ model.Put("data-test", location+"/generation.json", raw)
+ if mode == "unfinished" {
+ c.Holder.Generation = "next"
+ c.Holder.Admissions = []hostclaim.AdmissionRecord{{Generation: "next", Predecessor: "live", Region: "ie", PredecessorRegion: "ie", Mode: "cold"}}
+ }
+ writeDoorClaim(t, fx.srv, dbRequest{host: dbTestHost}, c)
+ location += "/g-live"
+ }
root := filepath.Join(fx.root, "instances", dbTestHost, "ltx")
err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
if err != nil {
@@ -85,7 +110,7 @@ func TestDoorPlanHTTPCompatibilityAndRestore(t *testing.T) {
if err != nil {
return err
}
- model.Put("data-test", dbLevelPrefix("instances/"+dbTestHost, level)+parts[1], b)
+ model.Put("data-test", dbLevelPrefix(location, level)+parts[1], b)
return nil
})
if err != nil {
@@ -180,14 +205,15 @@ func TestDoorPlanHTTPTruncationResumeAndFlagOff(t *testing.T) {
fx := dbFixture(t)
model := enableDoorPlan(t, fx)
key := dbVersionKey("instances/"+dbTestHost, 0, 1, 1)
- model.Put("data-test", key, []byte("original"))
+ payload := append(doorLTXHeader(t, 1, 1), []byte("original")...)
+ model.Put("data-test", key, payload)
var opens atomic.Int32
fx.srv.cfg.DBVersions = doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, id string) (io.ReadCloser, error) {
rc, err := model.OpenVersion(ctx, b, k, id)
if err != nil {
return nil, err
}
- if opens.Add(1) == 1 {
+ if opens.Add(1) == 2 {
return &dbRangeReader{Reader: io.LimitReader(rc, 3), Closer: rc}, nil
}
return rc, nil
@@ -221,7 +247,7 @@ func TestDoorPlanHTTPTruncationResumeAndFlagOff(t *testing.T) {
res = do(dbdoor.GetTail(0, 1, 1), h)
first, err := io.ReadAll(res.Body)
res.Body.Close()
- if !errors.Is(err, io.ErrUnexpectedEOF) || string(first) != "ori" || res.Header.Get(dbdoor.HandleHeader) != f.Handle {
+ if !errors.Is(err, io.ErrUnexpectedEOF) || string(first) != string(payload[:3]) || res.Header.Get(dbdoor.HandleHeader) != f.Handle {
t.Fatal(string(first), err, res.Header)
}
model.Put("data-test", key, []byte("replacement"))
@@ -229,7 +255,7 @@ func TestDoorPlanHTTPTruncationResumeAndFlagOff(t *testing.T) {
res = do(dbdoor.GetTail(0, 1, 1), h)
rest, err := io.ReadAll(res.Body)
res.Body.Close()
- if err != nil || res.StatusCode != 206 || string(append(first, rest...)) != "original" {
+ if err != nil || res.StatusCode != 206 || string(append(first, rest...)) != string(payload) {
t.Fatal(string(rest), err, res.StatusCode)
}
events, err := fx.st.Events(OpsAccountID, "titogo", 10)
diff --git a/internal/console/apidbclaim_range_test.go b/internal/console/apidbclaim_range_test.go
index 937c44ab6..883e2bf8c 100644
--- a/internal/console/apidbclaim_range_test.go
+++ b/internal/console/apidbclaim_range_test.go
@@ -138,6 +138,7 @@ func TestDoorPlanHTTPDeepRangeKeepsWatchdogAlive(t *testing.T) {
const offset = 4 << 20
want := bytes.Repeat([]byte("tail"), 64<<10)
payload := append(bytes.Repeat([]byte("p"), offset), want...)
+ copy(payload, doorLTXHeader(t, 1, 1))
key := dbVersionKey("instances/"+dbTestHost, 0, 1, 1)
id := model.Put("data-test", key, payload)
var read, fullOpens, rangeOpens atomic.Int64
@@ -162,6 +163,8 @@ func TestDoorPlanHTTPDeepRangeKeepsWatchdogAlive(t *testing.T) {
if err := json.Unmarshal([]byte(body), &list); err != nil || status != 200 || len(list.Files) != 1 {
t.Fatalf("list = %d %s, %v", status, body, err)
}
+ read.Store(0)
+ fullOpens.Store(0)
model.Put("data-test", key, []byte("replacement"))
ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
defer cancel()
diff --git a/internal/console/apidbclaim_test.go b/internal/console/apidbclaim_test.go
index 6c5639012..560258018 100644
--- a/internal/console/apidbclaim_test.go
+++ b/internal/console/apidbclaim_test.go
@@ -68,7 +68,7 @@ func doorPlanFixture(t *testing.T, legacy bool) (*Server, dbRequest, provision.R
}
model := leasetest.NewVersionedBucket()
model.AddBucket("data")
- s := &Server{cfg: Config{HostClaim: true, Releases: store, HibernateBucket: "data", DBVersions: model}, dbHandleKey: []byte(strings.Repeat("s", 32))}
+ s := &Server{cfg: Config{HomeRegion: "ie", HostClaim: true, Releases: store, HibernateBucket: "data", DBVersions: model}, dbHandleKey: []byte(strings.Repeat("s", 32))}
g := dbRequest{acct: &Account{Sqid: "bab"}, app: "jam", host: rec.Host}
g.claim, err = s.dbClaim(t.Context(), g, rec)
if err != nil {
@@ -134,7 +134,7 @@ func TestDoorPlanListingAndPinnedOpen(t *testing.T) {
loc := g.claim.Claim.Holder.Location
key := dbVersionKey(loc, 0, 1, 1)
m.Put("data", key, []byte("older"))
- id := m.Put("data", key, []byte("original"))
+ id := m.Put("data", key, append(doorLTXHeader(t, 1, 1), []byte("original")...))
deleted := dbVersionKey(loc, 0, 2, 2)
m.Put("data", deleted, []byte("hidden"))
m.BareDelete("data", deleted)
@@ -167,7 +167,7 @@ func TestDoorPlanListingAndPinnedOpen(t *testing.T) {
next.Holder.Incarnation, _ = hostclaim.Incarnation("bab", "jam", g.host, next.Holder.CreatedAt)
next.Holder.Location = "instances/" + g.host + "/" + next.Holder.Incarnation
writeDoorClaim(t, s, g, next)
- rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, 2, 3)
+ rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, 102, 3)
if err != nil {
t.Fatal(err)
}
@@ -203,7 +203,7 @@ func TestDoorPlanBracket(t *testing.T) {
for _, change := range []string{"holder", "op", "same-gen-op", "chore", "absent", "malformed", "read-error"} {
t.Run(change, func(t *testing.T) {
s, g, _, m := doorPlanFixture(t, false)
- m.Put("data", dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1), []byte("old"))
+ m.Put("data", dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1), doorLTXHeader(t, 1, 1))
fault := &doorReadFault{Store: s.cfg.Releases}
s.cfg.Releases = fault
s.cfg.DBVersions = doorVersionHook{VersionedStore: m, after: func() {
@@ -243,7 +243,7 @@ func TestDoorPlanBracket(t *testing.T) {
}
func TestDoorPlanClaimRefusals(t *testing.T) {
- for _, change := range []string{"absent", "free", "deleting", "parking", "purging", "owner", "incarnation", "location", "generation", "malformed", "read-error"} {
+ for _, change := range []string{"absent", "free", "deleting", "parking", "purging", "owner", "incarnation", "location", "malformed", "read-error"} {
t.Run(change, func(t *testing.T) {
s, g, rec, m := doorPlanFixture(t, true)
m.FailOn = func(op, key string) error { t.Fatalf("refusal reached data bucket: %s %s", op, key); return nil }
@@ -267,9 +267,6 @@ func TestDoorPlanClaimRefusals(t *testing.T) {
case "location":
c.Holder.Location = ""
want = hostclaim.ErrMalformed
- case "generation":
- c.Holder.Generation = "g-current"
- want = errDBGeneration
case "malformed":
c.V = 99
want = hostclaim.ErrMalformed
diff --git a/internal/dbdoor/client.go b/internal/dbdoor/client.go
index 65d637f2d..fb54df849 100644
--- a/internal/dbdoor/client.go
+++ b/internal/dbdoor/client.go
@@ -80,6 +80,20 @@ func (c *Client) SetLogger(l *slog.Logger) { c.logger = l }
// 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) {
+ c.handles.listMu.Lock()
+ defer c.handles.listMu.Unlock()
+ files, ok := c.planFiles(level)
+ if !ok {
+ var err error
+ files, err = c.listFiles(ctx, level)
+ if err != nil {
+ return nil, err
+ }
+ }
+ return fileIterator(files, level, seek), nil
+}
+
+func (c *Client) listFiles(ctx context.Context, level int) ([]LTXFile, error) {
res, err := c.do(ctx, ListTail(level), http.Header{})
if err != nil {
return nil, err
@@ -92,10 +106,10 @@ 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)
}
- files, err := c.rememberFiles(level, list)
- if err != nil {
- return nil, err
- }
+ return c.rememberFiles(level, list)
+}
+
+func fileIterator(files []LTXFile, level int, seek ltx.TXID) ltx.FileIterator {
infos := make([]*ltx.FileInfo, 0, len(files))
for _, f := range files {
if ltx.TXID(f.MinTXID) < seek {
@@ -106,7 +120,7 @@ func (c *Client) LTXFiles(ctx context.Context, level int, seek ltx.TXID, _ bool)
Size: f.Size, CreatedAt: f.CreatedAt,
})
}
- return ltx.NewFileInfoSliceIterator(infos), nil
+ return ltx.NewFileInfoSliceIterator(infos)
}
// OpenLTXFile opens one file by coordinate, sending the S3 client's exact
diff --git a/internal/dbdoor/client_handles.go b/internal/dbdoor/client_handles.go
index 4ee7e8f20..002095564 100644
--- a/internal/dbdoor/client_handles.go
+++ b/internal/dbdoor/client_handles.go
@@ -9,7 +9,9 @@ import (
const HandleHeader = "X-Carlos-DB-Handle"
type dbHandleState struct {
+ listMu sync.Mutex
mu sync.Mutex
+ plan []LTXFile
pinned bool
files map[string]LTXFile
}
@@ -23,6 +25,9 @@ func (c *Client) rememberFiles(level int, list ListResponse) ([]LTXFile, error)
if c.handles.pinned && list.Protocol != 2 {
return nil, errors.New("dbdoor: pinned listing protocol disappeared")
}
+ if list.Plan && list.Protocol != 2 {
+ return nil, errors.New("dbdoor: invalid pinned listing")
+ }
if list.Protocol == 0 {
return list.Files, nil
}
@@ -33,8 +38,8 @@ func (c *Client) rememberFiles(level int, list ListResponse) ([]LTXFile, error)
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] {
+ k := GetTail(f.Level, f.MinTXID, f.MaxTXID)
+ if (!list.Plan && f.Level != level) || f.Level < 0 || f.Level > MaxLevel || 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
@@ -47,6 +52,12 @@ func (c *Client) rememberFiles(level int, list ListResponse) ([]LTXFile, error)
files[k] = f
out = append(out, f)
}
+ if list.Plan {
+ if c.handles.plan == nil {
+ c.handles.plan = out
+ }
+ out = filesAtLevel(c.handles.plan, level)
+ }
c.handles.files = files
c.handles.pinned = true
return out, nil
@@ -63,3 +74,19 @@ func (c *Client) fileHandle(level int, lo, hi uint64) (string, error) {
}
return "", fmt.Errorf("dbdoor: coordinate has no pinned handle: %d %s", level, Span(lo, hi))
}
+
+func filesAtLevel(files []LTXFile, level int) []LTXFile {
+ out := make([]LTXFile, 0)
+ for _, f := range files {
+ if f.Level == level {
+ out = append(out, f)
+ }
+ }
+ return out
+}
+
+func (c *Client) planFiles(level int) ([]LTXFile, bool) {
+ c.handles.mu.Lock()
+ defer c.handles.mu.Unlock()
+ return filesAtLevel(c.handles.plan, level), c.handles.plan != nil
+}
diff --git a/internal/dbdoor/client_handles_test.go b/internal/dbdoor/client_handles_test.go
index cc0c5af00..52fccd044 100644
--- a/internal/dbdoor/client_handles_test.go
+++ b/internal/dbdoor/client_handles_test.go
@@ -91,3 +91,48 @@ func TestHandleProtocolNeverFallsBack(t *testing.T) {
}
}
}
+
+func TestHandleProtocolCompletePlan(t *testing.T) {
+ lists := 0
+ c := NewClient(func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ if strings.Count(tail, "/") == 2 {
+ lists++
+ if lists != 1 {
+ t.Fatal("relisted during one restore")
+ }
+ return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"protocol":2,"plan":true,"files":[{"level":9,"min_txid":1,"max_txid":11,"size":100,"handle":"root"},{"level":1,"min_txid":10,"max_txid":13,"size":100,"handle":"continuation"}]}`))}, nil
+ }
+ if h.Get(HandleHeader) != "continuation" {
+ t.Fatal("lost selected handle", h)
+ }
+ return &http.Response{StatusCode: 206, Header: h.Clone(), Body: io.NopCloser(strings.NewReader("bytes"))}, nil
+ })
+ for _, level := range []int{9, 8, 1, 0, 9} {
+ it, err := c.LTXFiles(t.Context(), level, 0, false)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var count int
+ for it.Next() {
+ count++
+ if it.Item().Level != level {
+ t.Fatal(it.Item())
+ }
+ }
+ if err := it.Close(); err != nil {
+ t.Fatal(err)
+ }
+ want := 0
+ if level == 9 || level == 1 {
+ want = 1
+ }
+ if count != want {
+ t.Fatalf("level %d: %d files, want %d", level, count, want)
+ }
+ }
+ rc, err := c.OpenLTXFile(t.Context(), 1, 10, 13, 0, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rc.Close()
+}
diff --git a/internal/dbdoor/coords.go b/internal/dbdoor/coords.go
index 689b02d2f..444bf4c3b 100644
--- a/internal/dbdoor/coords.go
+++ b/internal/dbdoor/coords.go
@@ -40,10 +40,11 @@ type LTXFile struct {
CreatedAt time.Time `json:"created_at"`
}
-// ListResponse is ordered by MinTXID ASCENDING: restoreLevelCursor assumes
-// filename order, which S3 gives because the names are zero-padded hex.
+// ListResponse contains one ordered level, or a complete restore plan with
+// its root first. Clients retain a complete plan across level requests.
type ListResponse struct {
Protocol int `json:"protocol,omitempty"`
+ Plan bool `json:"plan,omitempty"`
Files []LTXFile `json:"files"`
}