amadan

carlos / platform Public

console, dbdoor: the claim-bracketed pull door (claim Plan 7)

Under CARLOS_HOST_CLAIM the console brackets each database listing with
two claim reads (O1/O2), selects the holder generation's files through
internal/genselect, and returns console-signed handles naming exact
object versions (the pinned snapshot's CRC-64 included); file reads open
exactly that version, ranged reads included, with no current-key
fallback. The client keeps handles and verifies the pinned snapshot.
With the flag unset the door is unchanged. Inert until the floor.

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Squashed from feat/hostclaim-door into main.
Co-Authored-By: claude <claude~paul@keymail.dev>
codex pushed by claude e112199786d0de810593e966c3d41335d0a02110 parent b37f73e
23 files changed, +2274 −26
  • cmd/console/hostclaim_test.go +20 −0
  • cmd/console/main.go +15 −0
  • infra/modules/region-host/main.tf +8 −1
  • infra/modules/region-remote-host/main.tf +8 −1
  • internal/console/apidb.go +47 −0
  • internal/console/apidbclaim.go +305 −0
  • internal/console/apidbclaim_generation_test.go +164 −0
  • internal/console/apidbclaim_http_test.go +301 −0
  • internal/console/apidbclaim_r3_test.go +221 −0
  • internal/console/apidbclaim_range_test.go +198 −0
  • internal/console/apidbclaim_test.go +386 −0
  • internal/console/server.go +14 −4
  • internal/dbdoor/client.go +35 −5
  • internal/dbdoor/client_handles.go +92 −0
  • internal/dbdoor/client_handles_test.go +168 −0
  • internal/dbdoor/coords.go +12 −8
  • internal/dbdoor/restore.go +43 −6
  • internal/dbdoor/restore_test.go +21 −0
  • internal/lease/leasetest/bucket.go +27 −0
  • internal/lease/leasetest/contract.go +85 −0
  • internal/lease/leasetest/s3server.go +25 −0
  • internal/lease/versioned.go +28 −1
  • internal/lease/versioned_test.go +51 −0

CI passed — run details

diff --git a/cmd/console/hostclaim_test.go b/cmd/console/hostclaim_test.go
new file mode 100644
index 000000000..0972abea4
--- /dev/null
+++ b/cmd/console/hostclaim_test.go
@@ -0,0 +1,20 @@
+package main
+
+import "testing"
+
+func TestHostClaimEnabled(t *testing.T) {
+ for _, v := range []string{"", " ", "1", "true", "0"} {
+ calls := 0
+ got := hostClaimEnabled(func(k string) string {
+ calls++
+ if k != "CARLOS_HOST_CLAIM" {
+ t.Fatal(k)
+ }
+ return v
+ })
+ want := v != "" && v != " "
+ if got != want || calls != 1 {
+ t.Fatalf("%q: %v calls %d", v, got, calls)
+ }
+ }
+}
diff --git a/cmd/console/main.go b/cmd/console/main.go
index e2ee59ee9..804cfc5ad 100644
--- a/cmd/console/main.go
+++ b/cmd/console/main.go
@@ -73,6 +73,7 @@ import (
"crypto/ed25519"
"flag"
"fmt"
+ "github.com/carlosframework/platform/internal/lease"
"log"
"net"
"net/http"
@@ -269,6 +270,14 @@ func serve() {
// same one the server got.
rel := openStore()
dbBucket, dbRegion, dbPrefix, placement := dbDoorWiring()
+ hostClaim := hostClaimEnabled(os.Getenv)
+ var dbVersions lease.VersionedStore
+ if hostClaim && dbBucket != "" {
+ dbVersions, err = lease.NewVersionedS3(context.Background(), map[string]string{dbBucket: dbRegion})
+ if err != nil {
+ log.Fatalf("console: versioned database reader: %v", err)
+ }
+ }
// Before the server exists: every door that seals reads this switch.
appconfig.SetBoundSealing(secretsEnvelopeV2())
srv, err := console.NewServer(console.Config{
@@ -286,6 +295,8 @@ func serve() {
GeoIP: openGeoIP(strings.TrimSpace(os.Getenv("CONSOLE_GEOIP_DB"))),
// The carlos db door's data bucket and this box's placement
// (dbDoorWiring).
+ HostClaim: hostClaim,
+ DBVersions: dbVersions,
HibernateBucket: dbBucket,
HibernateRegion: dbRegion,
HibernatePrefix: dbPrefix,
@@ -1176,3 +1187,7 @@ func dbDoorWiring() (bucket, region, prefix, placement string) {
}
return cfg.Bucket, cfg.Region, cfg.Prefix, placement
}
+
+func hostClaimEnabled(getenv func(string) string) bool {
+ return strings.TrimSpace(getenv("CARLOS_HOST_CLAIM")) != ""
+}
diff --git a/infra/modules/region-host/main.tf b/infra/modules/region-host/main.tf
index d95461840..b58458f7d 100644
--- a/infra/modules/region-host/main.tf
+++ b/infra/modules/region-host/main.tf
@@ -179,11 +179,18 @@ resource "aws_iam_role_policy" "host_buckets" {
# moves/* grants are unchanged: the destination region retires its
# own staging (spec N13).
{
+ Sid = "PullDoorObjectVersions"
Effect = "Allow"
- Action = ["s3:GetObjectVersion", "s3:DeleteObjectVersion"]
+ Action = ["s3:GetObjectVersion"]
Resource = ["${var.data_bucket_arn}/*"]
},
{
+ Effect = "Allow"
+ Action = ["s3:DeleteObjectVersion"]
+ Resource = ["${var.data_bucket_arn}/*"]
+ },
+ {
+ Sid = "PullDoorVersionList"
Effect = "Allow"
Action = ["s3:ListBucketVersions"]
Resource = [var.data_bucket_arn]
diff --git a/infra/modules/region-remote-host/main.tf b/infra/modules/region-remote-host/main.tf
index d86783977..a5e4a203d 100644
--- a/infra/modules/region-remote-host/main.tf
+++ b/infra/modules/region-remote-host/main.tf
@@ -96,11 +96,18 @@ resource "aws_iam_user_policy" "host" {
# moves/* grants are unchanged: the destination region retires its
# own staging (spec N13).
{
+ Sid = "PullDoorObjectVersions"
Effect = "Allow"
- Action = ["s3:GetObjectVersion", "s3:DeleteObjectVersion"]
+ Action = ["s3:GetObjectVersion"]
Resource = ["${aws_s3_bucket.data.arn}/*"]
},
{
+ Effect = "Allow"
+ Action = ["s3:DeleteObjectVersion"]
+ Resource = ["${aws_s3_bucket.data.arn}/*"]
+ },
+ {
+ Sid = "PullDoorVersionList"
Effect = "Allow"
Action = ["s3:ListBucketVersions"]
Resource = [aws_s3_bucket.data.arn]
diff --git a/internal/console/apidb.go b/internal/console/apidb.go
index bf1a783d0..d474a3b88 100644
--- a/internal/console/apidb.go
+++ b/internal/console/apidb.go
@@ -31,6 +31,7 @@ import (
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
@@ -91,6 +92,7 @@ const dbAuditAction = "db-read"
// dbRequest is what dbGate resolves for both handlers.
type dbRequest struct {
+ claim hostclaim.Observed
acct *Account
actor string
app string
@@ -162,6 +164,20 @@ func (s *Server) dbGate(w http.ResponseWriter, r *http.Request) (dbRequest, bool
jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.")
return dbRequest{}, false
}
+ if s.cfg.HostClaim {
+ g := dbRequest{acct: acct, actor: actor, app: app, host: host}
+ if s.cfg.DBVersions == nil {
+ jsonError(w, http.StatusServiceUnavailable, "this console has no versioned database reader")
+ return dbRequest{}, false
+ }
+ g.claim, err = s.dbClaim(ctx, g, *rec)
+ if err != nil {
+ s.dbClaimFailure(w, r, g, err)
+ return dbRequest{}, false
+ }
+ g.prefix = g.claim.Claim.Holder.Location
+ return g, true
+ }
rd, err := s.dbReplica(ctx, prefix)
if err != nil {
log.Printf("console: api db %s/%s/%s: replica client: %v", acct.Sqid, app, host, err)
@@ -244,6 +260,16 @@ func (s *Server) handleAPIDBList(w http.ResponseWriter, r *http.Request) {
if !ok {
return
}
+ if s.cfg.HostClaim {
+ out, err := s.listDBVersions(r.Context(), g, level)
+ if err != nil {
+ s.dbClaimFailure(w, r, g, err)
+ return
+ }
+ w.Header().Set("Cache-Control", "no-store")
+ writeJSON(w, http.StatusOK, out)
+ return
+ }
itr, err := g.rd.LTXFiles(r.Context(), level, 0, false)
if err != nil {
s.dbStoreFailure(w, g, fmt.Sprintf("list level %d", level), err)
@@ -288,12 +314,27 @@ func (s *Server) handleAPIDBGet(w http.ResponseWriter, r *http.Request) {
if !ok {
return
}
+ var pinned *dbVersionReader
+ if s.cfg.HostClaim {
+ pinned, err = s.pinnedDBReader(r, g, level, minTXID, maxTXID)
+ if err != nil {
+ s.dbClaimFailure(w, r, g, err)
+ return
+ }
+ g.rd = pinned
+ }
rangeHdr := r.Header.Get("Range")
offset, size, err := dbdoor.ParseRange(rangeHdr)
if err != nil {
jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range")
return
}
+ if pinned != nil {
+ if _, err := dbVersionRange(pinned.handle.Size, offset, size); err != nil {
+ jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range")
+ return
+ }
+ }
if !s.dbStreams.acquire(g.actor) {
w.Header().Set("Retry-After", dbStreamRetryAfter)
jsonError(w, http.StatusTooManyRequests, "too many database downloads at once. Wait a few seconds and try again.")
@@ -330,6 +371,12 @@ func (s *Server) handleAPIDBGet(w http.ResponseWriter, r *http.Request) {
s.auditActivity(g.acct, g.app, g.actor, Activity{
Action: dbAuditAction, TargetType: "instance", Target: g.host, Detail: detail,
})
+ if pinned != nil {
+ if err := dbVersionHeaders(w, r, pinned.handle.Size, offset, size); err != nil {
+ jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range")
+ return
+ }
+ }
w.Header().Set("Content-Type", "application/octet-stream")
w.Header().Set("Cache-Control", "no-store")
status := http.StatusOK
diff --git a/internal/console/apidbclaim.go b/internal/console/apidbclaim.go
new file mode 100644
index 000000000..fbd685ead
--- /dev/null
+++ b/internal/console/apidbclaim.go
@@ -0,0 +1,305 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "crypto/ecdsa"
+ "crypto/hmac"
+ "crypto/sha256"
+ "encoding/base64"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "os"
+ "reflect"
+ "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"
+ "github.com/superfly/ltx"
+)
+
+const dbHandleTTL = 15 * time.Minute
+
+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")
+ errDBHandle = errors.New("invalid database file handle")
+ errDBHandleExpired = errors.New("database file handle expired; retry the database pull")
+)
+
+// dbFileHandle is encoded as this ordered JSON struct, never as a map.
+// Account and App bind the capability to the authenticated route as well.
+type dbFileHandle struct {
+ V int `json:"v"`
+ Host string `json:"host"`
+ Location string `json:"location"`
+ Key string `json:"key"`
+ VersionID string `json:"version_id"`
+ Incarnation string `json:"incarnation"`
+ Expiry int64 `json:"expiry"`
+ Account string `json:"account"`
+ App string `json:"app"`
+ Size int64 `json:"size"`
+ SnapshotCRC64 *uint64 `json:"snapshot_crc64,omitempty"`
+}
+
+func dbHandleSecret(key *ecdsa.PrivateKey) []byte {
+ mac := hmac.New(sha256.New, key.D.FillBytes(make([]byte, 32)))
+ mac.Write([]byte("carlos-db-handle-key-v1\x00"))
+ return mac.Sum(nil)
+}
+
+func dbHandleMAC(secret, body []byte) []byte {
+ mac := hmac.New(sha256.New, secret)
+ mac.Write([]byte("carlos-db-file-v1\x00"))
+ mac.Write(body)
+ return mac.Sum(nil)
+}
+
+func signDBHandle(secret []byte, h dbFileHandle) (string, error) {
+ if len(secret) != sha256.Size {
+ return "", errDBHandle
+ }
+ body, err := json.Marshal(h)
+ if err != nil {
+ return "", err
+ }
+ return base64.RawURLEncoding.EncodeToString(body) + "." + base64.RawURLEncoding.EncodeToString(dbHandleMAC(secret, body)), nil
+}
+
+func verifyDBHandle(secret []byte, token string, now time.Time) (dbFileHandle, error) {
+ var h dbFileHandle
+ if len(secret) != sha256.Size || len(token) > 8192 {
+ return h, errDBHandle
+ }
+ a, b, ok := strings.Cut(token, ".")
+ if !ok {
+ return h, errDBHandle
+ }
+ body, err := base64.RawURLEncoding.DecodeString(a)
+ if err != nil {
+ return h, errDBHandle
+ }
+ sig, err := base64.RawURLEncoding.DecodeString(b)
+ if err != nil || !hmac.Equal(sig, dbHandleMAC(secret, body)) {
+ return h, errDBHandle
+ }
+ dec := json.NewDecoder(bytes.NewReader(body))
+ dec.DisallowUnknownFields()
+ if err := dec.Decode(&h); err != nil {
+ return dbFileHandle{}, errDBHandle
+ }
+ canonical, err := json.Marshal(h)
+ if err != nil || !bytes.Equal(body, canonical) || h.V != 1 || h.VersionID == "" || h.VersionID == "null" || h.Size < 0 {
+ return dbFileHandle{}, errDBHandle
+ }
+ if now.Unix() >= h.Expiry {
+ return dbFileHandle{}, errDBHandleExpired
+ }
+ return h, nil
+}
+
+func dbLevelPrefix(location string, level int) string {
+ // litestream/s3 uses hex level directories, unlike its file replica.
+ return fmt.Sprintf("%s/%04x/", location, level)
+}
+
+func dbVersionKey(location string, level int, lo, hi uint64) string {
+ return dbLevelPrefix(location, level) + ltx.FormatFilename(ltx.TXID(lo), ltx.TXID(hi))
+}
+
+func (s *Server) dbClaim(ctx context.Context, g dbRequest, rec provision.Record) (hostclaim.Observed, error) {
+ obs, err := hostclaim.Read(ctx, s.cfg.Releases, g.host)
+ if err != nil {
+ return hostclaim.Observed{}, err
+ }
+ if obs.Absent {
+ return obs, errDBNotAdopted
+ }
+ h := obs.Claim.Holder
+ if h == nil {
+ return obs, errDBClaimClosed
+ }
+ inc, err := hostclaim.Incarnation(g.acct.Sqid, g.app, g.host, rec.CreatedAt)
+ if err != nil {
+ return obs, err
+ }
+ if h.Family != "instance" || h.Account != g.acct.Sqid || h.App != g.app || h.Key != provision.RecordKey(g.acct.Sqid, g.app, g.host) || h.Incarnation != inc || !h.CreatedAt.Equal(rec.CreatedAt) {
+ return obs, os.ErrNotExist
+ }
+ switch obs.Claim.Op.Kind {
+ case hostclaim.KindDeleting, hostclaim.KindParking, hostclaim.KindPurging:
+ return obs, errDBClaimClosed
+ }
+ if h.Location == "" {
+ return obs, errDBClaimClosed
+ }
+ return obs, nil
+}
+
+func sameDBAdmission(a, b hostclaim.Observed) bool {
+ if a.Absent || b.Absent || a.Claim == nil || b.Claim == nil {
+ return false
+ }
+ // §6.5 permits unrelated chore revisions; compare all holder/op content,
+ // including stamp, session, admissions, journal and execution ownership.
+ return a.Host == b.Host && reflect.DeepEqual(a.Claim.Holder, b.Claim.Holder) && reflect.DeepEqual(a.Claim.Op, b.Claim.Op)
+}
+
+func (s *Server) dbClaimFailure(w http.ResponseWriter, r *http.Request, g dbRequest, err error) {
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ http.NotFound(w, r)
+ 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):
+ jsonError(w, http.StatusGone, err.Error())
+ case errors.Is(err, errDBHandle):
+ jsonError(w, http.StatusBadRequest, err.Error())
+ default:
+ s.dbStoreFailure(w, g, "claim/version read", err)
+ }
+}
+
+func (cfg Config) dbClaimRegion() string {
+ for _, region := range []string{cfg.Region, cfg.HomeRegion, cfg.HibernateRegion} {
+ if region = strings.TrimSpace(region); region != "" {
+ return region
+ }
+ }
+ return ""
+}
+
+func (s *Server) listDBVersions(ctx context.Context, g dbRequest, level int) (dbdoor.ListResponse, error) {
+ // 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
+ plan, err := genselect.Select(ctx, s.cfg.DBVersions, genselect.Request{
+ Bucket: s.cfg.HibernateBucket, Region: s.cfg.dbClaimRegion(), Host: g.host, Holder: *h,
+ })
+ if err != nil {
+ return dbdoor.ListResponse{}, errDBClaimChanged
+ }
+ for _, f := range plan.Files {
+ var crc *uint64
+ if plan.Pin != nil && f.Key == plan.Pin.SnapshotKey && f.VersionID == plan.Pin.SnapshotVersion {
+ crc = &plan.Pin.SnapshotCRC64
+ }
+ 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, SnapshotCRC64: crc})
+ if err != nil {
+ return dbdoor.ListResponse{}, err
+ }
+ out.Files = append(out.Files, dbdoor.LTXFile{Level: f.Level, MinTXID: f.MinTXID, MaxTXID: f.MaxTXID, Size: f.Size, Handle: token, SnapshotCRC64: crc})
+ }
+ o2, err := hostclaim.Read(ctx, s.cfg.Releases, g.host)
+ if err != nil {
+ return dbdoor.ListResponse{}, err
+ }
+ if !sameDBAdmission(g.claim, o2) {
+ return dbdoor.ListResponse{}, errDBClaimChanged
+ }
+ 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 || !dbHandleCoordinate(h, level, lo, hi) {
+ return nil, errDBHandle
+ }
+ return &dbVersionReader{store: s.cfg.DBVersions, bucket: s.cfg.HibernateBucket, handle: h}, nil
+}
+
+// This adapter keeps the existing stream budget, watchdog and audit path.
+// Its only object read names the signed version, including on every resume.
+type dbVersionReader struct {
+ store lease.VersionedStore
+ bucket string
+ handle dbFileHandle
+}
+
+func (v *dbVersionReader) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) {
+ return nil, errDBHandle
+}
+func (v *dbVersionReader) OpenLTXFile(ctx context.Context, level int, lo, hi ltx.TXID, offset, size int64) (io.ReadCloser, error) {
+ if !dbHandleCoordinate(v.handle, level, uint64(lo), uint64(hi)) {
+ return nil, errDBHandle
+ }
+ n, err := dbVersionRange(v.handle.Size, offset, size)
+ if err != nil {
+ return nil, err
+ }
+ var rc io.ReadCloser
+ ranged, ok := v.store.(lease.VersionRangeOpener)
+ if ok && n > 0 {
+ rc, err = ranged.OpenVersionRange(ctx, v.bucket, v.handle.Key, v.handle.VersionID, offset, n)
+ } else {
+ rc, err = v.store.OpenVersion(ctx, v.bucket, v.handle.Key, v.handle.VersionID)
+ }
+ if errors.Is(err, lease.ErrMirrorNotFound) {
+ return nil, os.ErrNotExist
+ }
+ if err != nil {
+ return nil, err
+ }
+ if !ok {
+ if _, err = io.CopyN(io.Discard, rc, offset); err != nil {
+ rc.Close()
+ return nil, err
+ }
+ }
+ return &dbRangeReader{Reader: io.LimitReader(rc, n), Closer: rc}, nil
+}
+
+type dbRangeReader struct {
+ io.Reader
+ io.Closer
+}
+
+func dbVersionRange(total, offset, size int64) (int64, error) {
+ if total < 0 || offset < 0 || size < 0 || offset > total || (offset == total && total != 0) {
+ return 0, errDBHandle
+ }
+ n := total - offset
+ if size != 0 && size < n {
+ n = size
+ }
+ return n, nil
+}
+func dbVersionHeaders(w http.ResponseWriter, r *http.Request, total, offset, size int64) error {
+ n, err := dbVersionRange(total, offset, size)
+ if err != nil {
+ return err
+ }
+ w.Header().Set("Content-Length", strconv.FormatInt(n, 10))
+ w.Header().Set(dbdoor.HandleHeader, r.Header.Get(dbdoor.HandleHeader))
+ if r.Header.Get("Range") != "" && n > 0 {
+ w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+n-1, total))
+ }
+ return nil
+}
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
new file mode 100644
index 000000000..7c461700a
--- /dev/null
+++ b/internal/console/apidbclaim_http_test.go
@@ -0,0 +1,301 @@
+package console
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "io/fs"
+ "net/http"
+ "os"
+ "path/filepath"
+ "strconv"
+ "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"
+ "github.com/carlosframework/platform/internal/svcrole"
+)
+
+func enableDoorPlan(t *testing.T, fx *dbFx) *leasetest.VersionedBucket {
+ t.Helper()
+ rec, err := provision.GetRecord(t.Context(), fx.rel, "bab", "titogo", dbTestHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ inc, err := hostclaim.Incarnation("bab", "titogo", dbTestHost, rec.CreatedAt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c := hostclaim.Claim{V: 2, Host: dbTestHost, Gen: 1, Holder: &hostclaim.Holder{Family: "instance", Key: provision.RecordKey("bab", "titogo", dbTestHost), Account: "bab", App: "titogo", CreatedAt: rec.CreatedAt, Incarnation: inc, Location: "instances/" + dbTestHost}, Op: hostclaim.Op{Kind: hostclaim.KindServing}}
+ raw, _ := json.Marshal(c)
+ key, _ := hostclaim.Key(dbTestHost)
+ if _, err := fx.rel.Put(t.Context(), key, raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ 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()
+ if err != nil {
+ t.Fatal(err)
+ }
+ fx.srv.dbHandleKey = dbHandleSecret(secret)
+ return m
+}
+
+func TestDoorPlanHTTPCompatibilityAndRestore(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 {
+ return err
+ }
+ if d.IsDir() {
+ return nil
+ }
+ rel, err := filepath.Rel(root, p)
+ if err != nil {
+ return err
+ }
+ parts := strings.Split(filepath.ToSlash(rel), "/")
+ if len(parts) != 2 {
+ return fmt.Errorf("unexpected file: %s", rel)
+ }
+ level, err := strconv.Atoi(parts[0])
+ if err != nil {
+ return err
+ }
+ b, err := os.ReadFile(p)
+ if err != nil {
+ return err
+ }
+ model.Put("data-test", dbLevelPrefix(location, level)+parts[1], b)
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ r, err := http.NewRequestWithContext(ctx, "GET", fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil)
+ if err != nil {
+ return nil, err
+ }
+ r.Header = h.Clone()
+ r.Header.Set("Authorization", "Bearer "+fx.member)
+ return http.DefaultClient.Do(r)
+ }
+ got, err := dbdoor.RestoreTo(t.Context(), dbdoor.NewClient(do), t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(got.Dir)
+ db, err := sql.Open("sqlite", "file:"+got.Path+"?mode=ro")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ var rows int
+ if err := db.QueryRow("SELECT count(*) FROM t").Scan(&rows); err != nil {
+ t.Fatal(err)
+ }
+ if rows != 3*dbdoortest.Rows {
+ t.Fatal(rows)
+ }
+ // The old CLI ignores additive fields and sends no handle.
+ res, err := do(t.Context(), dbdoor.ListTail(0), http.Header{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ var list dbdoor.ListResponse
+ if err := json.NewDecoder(res.Body).Decode(&list); err != nil {
+ t.Fatal(err)
+ }
+ res.Body.Close()
+ if len(list.Files) == 0 {
+ t.Fatal("empty L0")
+ }
+ f := list.Files[0]
+ res, err = do(t.Context(), dbdoor.GetTail(f.Level, f.MinTXID, f.MaxTXID), http.Header{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ io.Copy(io.Discard, res.Body)
+ res.Body.Close()
+ want := http.StatusOK
+ if enabled {
+ want = http.StatusBadRequest
+ }
+ if res.StatusCode != want {
+ t.Fatalf("old CLI: %d want %d", res.StatusCode, want)
+ }
+ if !enabled && list.Protocol != 0 {
+ t.Fatal("flag-off wire changed")
+ }
+ if enabled && fx.opens.Load() != 0 {
+ t.Fatal("pinned door opened unversioned replica")
+ }
+ })
+ }
+}
+
+func TestDoorPlanHTTPAuthorizationAndMissingClaim(t *testing.T) {
+ fx := dbFixture(t)
+ model := enableDoorPlan(t, fx)
+ model.FailOn = func(op, key string) error { t.Fatalf("unauthorized data access %s %s", op, key); return nil }
+ stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "door-plan-stranger")
+ status, h, body := rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), stranger, "")
+ assertNotFound(t, "stranger", canonicalNotFound(t), status, h, body)
+ admin := serviceToken(t, fx.st, OpsAccountID, "door-plan-admin", "titogo", svcrole.Admin)
+ status, _, _ = rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), admin, "")
+ if status != http.StatusForbidden {
+ t.Fatal(status)
+ }
+ // A second existing record with no claim must not auto-adopt here.
+ other := "other.go.tito.io"
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: other})
+ status, h, body = rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", other, 0)), fx.member, "")
+ if status != 409 || h.Get("Retry-After") != "1" || !strings.Contains(body, "host not adopted") {
+ t.Fatal(status, h, body)
+ }
+}
+
+func TestDoorPlanHTTPTruncationResumeAndFlagOff(t *testing.T) {
+ fx := dbFixture(t)
+ model := enableDoorPlan(t, fx)
+ key := dbVersionKey("instances/"+dbTestHost, 0, 1, 1)
+ 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) == 2 {
+ return &dbRangeReader{Reader: io.LimitReader(rc, 3), Closer: rc}, nil
+ }
+ return rc, nil
+ }}
+ do := func(tail string, h http.Header) *http.Response {
+ t.Helper()
+ r, err := http.NewRequest("GET", fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r.Header = h.Clone()
+ r.Header.Set("Authorization", "Bearer "+fx.member)
+ res, err := http.DefaultClient.Do(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return res
+ }
+ res := do(dbdoor.ListTail(0), http.Header{})
+ var list dbdoor.ListResponse
+ if err := json.NewDecoder(res.Body).Decode(&list); err != nil {
+ t.Fatal(err)
+ }
+ res.Body.Close()
+ if len(list.Files) != 1 {
+ t.Fatal(list)
+ }
+ f := list.Files[0]
+ h := http.Header{}
+ h.Set(dbdoor.HandleHeader, f.Handle)
+ 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) != string(payload[:3]) || res.Header.Get(dbdoor.HandleHeader) != f.Handle {
+ t.Fatal(string(first), err, res.Header)
+ }
+ model.Put("data-test", key, []byte("replacement"))
+ h.Set("Range", "bytes=3-")
+ 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...)) != string(payload) {
+ t.Fatal(string(rest), err, res.StatusCode)
+ }
+ events, err := fx.st.Events(OpsAccountID, "titogo", 10)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ranged := false
+ for _, e := range events {
+ if e.Action == dbAuditAction && e.Target == dbTestHost && strings.Contains(e.Detail, "byte 3") {
+ ranged = true
+ }
+ }
+ if !ranged {
+ t.Fatal("missing pinned range audit", events)
+ }
+ // A flag-off console ignores the handle header exactly as today (R1):
+ // it serves the coordinate and never echoes the handle, so a v2 client
+ // rejects the response (Task 1's acknowledgement check).
+ // Keep startup configuration immutable while either server is running.
+ legacy := dbFixture(t)
+ legacy.replicate(t, dbTestHost, 1)
+ lc := lsfile.NewReplicaClient(filepath.Join(legacy.root, "instances", dbTestHost))
+ lf := listFiles(t, lc, 0)[0]
+ req, err := http.NewRequest("GET", legacy.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(lf.MinTXID), uint64(lf.MaxTXID))), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Header = h.Clone()
+ req.Header.Del("Range") // a whole-file GET, so success is 200, not 206
+ req.Header.Set("Authorization", "Bearer "+legacy.member)
+ res, err = http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ io.Copy(io.Discard, res.Body)
+ res.Body.Close()
+ if res.StatusCode != 200 {
+ t.Fatal("flag-off changed its answer to a request carrying a handle header", res.StatusCode)
+ }
+ if res.Header.Get(dbdoor.HandleHeader) != "" {
+ t.Fatal("flag-off console echoed a handle it cannot enforce")
+ }
+}
diff --git a/internal/console/apidbclaim_r3_test.go b/internal/console/apidbclaim_r3_test.go
new file mode 100644
index 000000000..42634773a
--- /dev/null
+++ b/internal/console/apidbclaim_r3_test.go
@@ -0,0 +1,221 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "database/sql"
+ "encoding/json"
+ "io"
+ "net/http"
+ "os"
+ "path/filepath"
+ "strings"
+ "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/superfly/ltx"
+)
+
+func TestDoorPlanUnnamedHomeRegion(t *testing.T) {
+ for _, unfinished := range []bool{false, true} {
+ t.Run(map[bool]string{false: "origin", true: "unfinished"}[unfinished], func(t *testing.T) {
+ fx := dbFixture(t)
+ model := enableDoorPlan(t, fx)
+ fx.srv.cfg.HomeRegion = ""
+ fx.srv.cfg.HibernateRegion = "eu-west-1"
+ if fx.srv.cfg.Region != "" {
+ t.Fatal("fixture has a placement")
+ }
+ if unfinished {
+ obs, err := hostclaim.Read(t.Context(), fx.rel, dbTestHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c := cloneDoorClaim(obs.Claim)
+ c.Holder.Generation = "next"
+ c.Holder.Admissions = []hostclaim.AdmissionRecord{{Generation: "next", Predecessor: hostclaim.RootGeneration, Region: "eu-west-1", PredecessorRegion: "eu-west-1", Mode: "cold"}}
+ writeDoorClaim(t, fx.srv, dbRequest{host: dbTestHost}, c)
+ }
+ model.Put("data-test", dbVersionKey("instances/"+dbTestHost, 0, 1, 1), doorLTXHeader(t, 1, 1))
+ status, body := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), fx.member)
+ if status != http.StatusOK {
+ t.Fatalf("unnamed home listing: %d %s", status, body)
+ }
+ })
+ }
+}
+
+func TestDoorPlanRegionStartup(t *testing.T) {
+ fx := dbFixture(t)
+ for _, mode := range []string{"unresolved", "bucket-region", "named-home", "placement", "flag-off", "door-off"} {
+ t.Run(mode, func(t *testing.T) {
+ cfg := fx.srv.cfg
+ cfg.HostClaim = true
+ cfg.Region, cfg.HomeRegion, cfg.HibernateRegion = "", "", ""
+ switch mode {
+ case "bucket-region":
+ cfg.HibernateRegion = "eu-west-1"
+ case "named-home":
+ cfg.HomeRegion = "ie"
+ case "placement":
+ cfg.Region = "uk"
+ case "flag-off":
+ cfg.HostClaim = false
+ case "door-off":
+ cfg.HibernateBucket = ""
+ }
+ _, err := NewServer(cfg, fx.st)
+ if mode == "unresolved" {
+ if err == nil || !strings.Contains(err.Error(), "region") {
+ t.Fatalf("startup without a region: %v", err)
+ }
+ } else if err != nil {
+ t.Fatal(err)
+ }
+ })
+ }
+}
+
+func TestDoorPlanPinnedSnapshotCRC(t *testing.T) {
+ for _, mode := range []string{"match", "mismatch", "zero", "historical", "successor"} {
+ t.Run(mode, func(t *testing.T) {
+ fx := dbFixture(t)
+ fx.replicate(t, dbTestHost, 2)
+ replica := lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost))
+ var readers []io.Reader
+ for _, f := range listFiles(t, replica, 0) {
+ b, err := os.ReadFile(replica.LTXFilePath(0, f.MinTXID, f.MaxTXID))
+ if err != nil {
+ t.Fatal(err)
+ }
+ readers = append(readers, bytes.NewReader(b))
+ }
+ var snapshot bytes.Buffer
+ compactor, err := ltx.NewCompactor(&snapshot, readers)
+ if err != nil {
+ t.Fatal(err)
+ }
+ compactor.HeaderFlags = ltx.HeaderFlagNoChecksum
+ if err := compactor.Compact(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ dec := ltx.NewDecoder(bytes.NewReader(snapshot.Bytes()))
+ if err := dec.Verify(); err != nil {
+ t.Fatal(err)
+ }
+ crc := uint64(dec.Trailer().FileChecksum)
+ maxTXID := uint64(dec.Header().MaxTXID)
+ model := enableDoorPlan(t, fx)
+ obs, err := hostclaim.Read(t.Context(), fx.rel, dbTestHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c := cloneDoorClaim(obs.Claim)
+ c.Holder.Generation = "live"
+ loc := c.Holder.Location
+ key := dbVersionKey(loc+"/g-live", 9, 1, maxTXID)
+ id := model.Put("data-test", key, snapshot.Bytes())
+ pin := genselect.Pin{SnapshotKey: key, SnapshotVersion: id, SnapshotTXID: maxTXID, SnapshotCRC64: crc}
+ switch mode {
+ case "mismatch":
+ pin.SnapshotCRC64 ^= 1
+ case "zero":
+ pin.SnapshotCRC64 = 0
+ case "historical":
+ model.Put("data-test", key, []byte("late replacement"))
+ case "successor":
+ pin.SnapshotKey = dbVersionKey(loc+"/g-live", 9, 1, 1)
+ pin.SnapshotTXID = 1
+ pin.SnapshotVersion = "retained-marker-only"
+ pin.SnapshotCRC64 ^= 1
+ }
+ marker := genselect.Marker{Op: "live", Host: dbTestHost, Incarnation: c.Holder.Incarnation, Box: "box", InstalledAt: time.Unix(1, 0).UTC(), Pin: pin}
+ raw, err := json.Marshal(marker)
+ if err != nil {
+ t.Fatal(err)
+ }
+ model.Put("data-test", loc+"/generation.json", raw)
+ writeDoorClaim(t, fx.srv, dbRequest{host: dbTestHost}, c)
+ do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ req, err := http.NewRequestWithContext(ctx, "GET", fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil)
+ if err != nil {
+ return nil, err
+ }
+ req.Header = h.Clone()
+ req.Header.Set("Authorization", "Bearer "+fx.member)
+ return http.DefaultClient.Do(req)
+ }
+ parent := t.TempDir()
+ got, err := dbdoor.RestoreTo(t.Context(), dbdoor.NewClient(do), parent)
+ if mode == "mismatch" || mode == "zero" {
+ if got != nil {
+ os.RemoveAll(got.Dir)
+ t.Fatal("restored a valid snapshot whose CRC differs from the marker")
+ }
+ if err == nil || !strings.Contains(err.Error(), dbdoor.ErrCheckFailed.Error()) {
+ t.Fatalf("want member-safe check refusal: %v", err)
+ }
+ entries, readErr := os.ReadDir(parent)
+ if readErr != nil || len(entries) != 0 {
+ t.Fatalf("failed restore left files: %v %v", entries, readErr)
+ }
+ } else {
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(got.Dir)
+ db, err := sql.Open("sqlite", "file:"+got.Path+"?mode=ro")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ var rows int
+ if err := db.QueryRow("SELECT count(*) FROM t").Scan(&rows); err != nil || rows != 2*dbdoortest.Rows {
+ t.Fatalf("rows %d: %v", rows, err)
+ }
+ }
+ res, err := do(t.Context(), dbdoor.ListTail(9), http.Header{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer res.Body.Close()
+ var list struct {
+ Files []struct {
+ Handle string
+ SnapshotCRC64 *uint64 `json:"snapshot_crc64"`
+ }
+ }
+ if err := json.NewDecoder(res.Body).Decode(&list); err != nil || len(list.Files) != 1 {
+ t.Fatalf("listing: %+v %v", list, err)
+ }
+ h, err := verifyDBHandle(fx.srv.dbHandleKey, list.Files[0].Handle, time.Now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ raw, err = json.Marshal(h)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var signed struct {
+ SnapshotCRC64 *uint64 `json:"snapshot_crc64"`
+ }
+ if err := json.Unmarshal(raw, &signed); err != nil {
+ t.Fatal(err)
+ }
+ for _, value := range []*uint64{list.Files[0].SnapshotCRC64, signed.SnapshotCRC64} {
+ if mode == "successor" {
+ if value != nil {
+ t.Fatal("successor inherited old pin CRC")
+ }
+ } else if value == nil || *value != pin.SnapshotCRC64 {
+ t.Fatalf("lost expected CRC: %v, want %d", value, pin.SnapshotCRC64)
+ }
+ }
+ })
+ }
+}
diff --git a/internal/console/apidbclaim_range_test.go b/internal/console/apidbclaim_range_test.go
new file mode 100644
index 000000000..883e2bf8c
--- /dev/null
+++ b/internal/console/apidbclaim_range_test.go
@@ -0,0 +1,198 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "os"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+type doorRangeHook struct {
+ doorOpenHook
+ openRange func(context.Context, string, string, string, int64, int64) (io.ReadCloser, error)
+}
+
+func (v doorRangeHook) OpenVersionRange(ctx context.Context, b, k, id string, offset, length int64) (io.ReadCloser, error) {
+ return v.openRange(ctx, b, k, id, offset, length)
+}
+
+type doorCountedReader struct {
+ io.ReadCloser
+ ctx context.Context
+ read *atomic.Int64
+ delay time.Duration
+}
+
+func (r *doorCountedReader) Read(p []byte) (int, error) {
+ if r.delay > 0 {
+ timer := time.NewTimer(r.delay)
+ defer timer.Stop()
+ select {
+ case <-r.ctx.Done():
+ return 0, r.ctx.Err()
+ case <-timer.C:
+ }
+ if len(p) > 4096 {
+ p = p[:4096]
+ }
+ }
+ n, err := r.ReadCloser.Read(p)
+ r.read.Add(int64(n))
+ return n, err
+}
+
+func TestDoorPlanDeepRangeReadsOnlyRequestedBytes(t *testing.T) {
+ for _, ranged := range []bool{false, true} {
+ t.Run(fmt.Sprint(ranged), func(t *testing.T) {
+ s, g, _, model := doorPlanFixture(t, true)
+ const offset = 4 << 20
+ payload := append(bytes.Repeat([]byte("p"), offset), []byte("original")...)
+ key := dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1)
+ id := model.Put("data", key, payload)
+ model.Put("data", key, []byte("replacement"))
+ var read atomic.Int64
+ base := doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, v string) (io.ReadCloser, error) {
+ rc, err := model.OpenVersion(ctx, b, k, v)
+ return &doorCountedReader{ReadCloser: rc, read: &read}, err
+ }}
+ var store lease.VersionedStore = base
+ if ranged {
+ store = doorRangeHook{doorOpenHook: base, openRange: func(ctx context.Context, b, k, v string, off, length int64) (io.ReadCloser, error) {
+ if b != "data" || k != key || v != id || off != offset || length != 3 {
+ t.Fatalf("range arguments = %s %s %s %d %d", b, k, v, off, length)
+ }
+ rc, err := model.OpenVersionRange(ctx, b, k, v, off, length)
+ return &doorCountedReader{ReadCloser: rc, read: &read}, err
+ }}
+ }
+ rd := &dbVersionReader{store: store, bucket: s.cfg.HibernateBucket, handle: dbFileHandle{
+ Location: g.claim.Claim.Holder.Location, Key: key, VersionID: id, Size: int64(len(payload)),
+ }}
+ rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, offset, 3)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := io.ReadAll(rc)
+ rc.Close()
+ if err != nil || string(body) != "ori" {
+ t.Fatalf("body = %q, %v", body, err)
+ }
+ want := int64(offset + 3)
+ if ranged {
+ want = 3
+ }
+ if got := read.Load(); got != want {
+ t.Fatalf("storage read %d bytes; want %d", got, want)
+ }
+ })
+ }
+}
+
+func TestDoorPlanRangeErrorsNeverFallBack(t *testing.T) {
+ for _, failure := range []error{lease.ErrMirrorNotFound, lease.ErrIsDeleteMarker, context.Canceled, errors.New("offline")} {
+ t.Run(failure.Error(), func(t *testing.T) {
+ s, g, _, model := doorPlanFixture(t, true)
+ key := dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1)
+ model.Put("data", key, []byte("replacement"))
+ var fullOpens int
+ store := doorRangeHook{
+ doorOpenHook: doorOpenHook{VersionedStore: model, open: func(context.Context, string, string, string) (io.ReadCloser, error) {
+ fullOpens++
+ return io.NopCloser(bytes.NewReader([]byte("fallback"))), nil
+ }},
+ openRange: func(context.Context, string, string, string, int64, int64) (io.ReadCloser, error) {
+ return nil, failure
+ },
+ }
+ rd := &dbVersionReader{store: store, bucket: s.cfg.HibernateBucket, handle: dbFileHandle{
+ Location: g.claim.Claim.Holder.Location, Key: key, VersionID: "selected", Size: 8,
+ }}
+ rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, 2, 3)
+ if rc != nil {
+ rc.Close()
+ }
+ want := failure
+ if errors.Is(failure, lease.ErrMirrorNotFound) {
+ want = os.ErrNotExist
+ }
+ if !errors.Is(err, want) || fullOpens != 0 {
+ t.Fatalf("range error = %v; want %v, full opens = %d", err, want, fullOpens)
+ }
+ })
+ }
+}
+
+func TestDoorPlanHTTPDeepRangeKeepsWatchdogAlive(t *testing.T) {
+ fx := dbFixture(t)
+ model := enableDoorPlan(t, fx)
+ 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
+ fx.srv.dbStreamIdle = 250 * time.Millisecond
+ fx.srv.cfg.DBVersions = doorRangeHook{
+ doorOpenHook: doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, v string) (io.ReadCloser, error) {
+ fullOpens.Add(1)
+ rc, err := model.OpenVersion(ctx, b, k, v)
+ return &doorCountedReader{ReadCloser: rc, ctx: ctx, read: &read, delay: 10 * time.Millisecond}, err
+ }},
+ openRange: func(ctx context.Context, b, k, v string, off, length int64) (io.ReadCloser, error) {
+ rangeOpens.Add(1)
+ if v != id || off != offset || length != int64(len(want)) {
+ return nil, fmt.Errorf("unexpected range %s %d %d", v, off, length)
+ }
+ rc, err := model.OpenVersionRange(ctx, b, k, v, off, length)
+ return &doorCountedReader{ReadCloser: rc, ctx: ctx, read: &read, delay: 10 * time.Millisecond}, err
+ },
+ }
+ status, _, body := rawReq(t, http.MethodGet, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), fx.member, "")
+ var list dbdoor.ListResponse
+ 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()
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Header.Set("Authorization", "Bearer "+fx.member)
+ req.Header.Set(dbdoor.HandleHeader, list.Files[0].Handle)
+ req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset))
+ start := time.Now()
+ 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, want) {
+ t.Fatalf("range = status %d, %d bytes, %v; want 206 and %d bytes", res.StatusCode, len(got), err, len(want))
+ }
+ if time.Since(start) <= 2*fx.srv.dbStreamIdle {
+ t.Fatal("stream did not outlive two watchdog intervals")
+ }
+ if read.Load() != int64(len(want)) || fullOpens.Load() != 0 || rangeOpens.Load() != 1 {
+ t.Fatalf("read %d bytes, full opens %d, range opens %d", read.Load(), fullOpens.Load(), rangeOpens.Load())
+ }
+ if res.Header.Get("Content-Range") != fmt.Sprintf("bytes %d-%d/%d", offset, len(payload)-1, len(payload)) || res.Header.Get(dbdoor.HandleHeader) != list.Files[0].Handle {
+ t.Fatalf("range headers = %v", res.Header)
+ }
+ fx.waitIdle(t, time.Second)
+}
diff --git a/internal/console/apidbclaim_test.go b/internal/console/apidbclaim_test.go
new file mode 100644
index 000000000..560258018
--- /dev/null
+++ b/internal/console/apidbclaim_test.go
@@ -0,0 +1,386 @@
+package console
+
+import (
+ "context"
+ "crypto/ecdsa"
+ "crypto/elliptic"
+ "crypto/rand"
+ "encoding/json"
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type doorVersionHook struct {
+ lease.VersionedStore
+ after func()
+}
+
+func (v doorVersionHook) ListVersions(ctx context.Context, b, p string) ([]lease.ObjectVersion, error) {
+ out, err := v.VersionedStore.ListVersions(ctx, b, p)
+ if v.after != nil {
+ v.after()
+ }
+ return out, err
+}
+
+type doorReadFault struct {
+ release.Store
+ err error
+}
+
+func (s *doorReadFault) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if s.err != nil {
+ return nil, "", s.err
+ }
+ return s.Store.Get(ctx, key)
+}
+
+func doorPlanFixture(t *testing.T, legacy bool) (*Server, dbRequest, provision.Record, *leasetest.VersionedBucket) {
+ t.Helper()
+ rec := provision.Record{Host: "shop.example", Account: "bab", App: "jam"}
+ inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ loc := "instances/" + rec.Host
+ if !legacy {
+ loc += "/" + inc
+ }
+ claim := hostclaim.Claim{V: 2, Host: rec.Host, Gen: 1, Holder: &hostclaim.Holder{Family: "instance", Key: provision.RecordKey(rec.Account, rec.App, rec.Host), Account: rec.Account, App: rec.App, Incarnation: inc, Location: loc}, Op: hostclaim.Op{Kind: hostclaim.KindServing}}
+ store := release.NewDirStore(t.TempDir())
+ data, _ := json.Marshal(claim)
+ key, _ := hostclaim.Key(rec.Host)
+ if _, err := store.Put(t.Context(), key, data, ""); err != nil {
+ t.Fatal(err)
+ }
+ model := leasetest.NewVersionedBucket()
+ model.AddBucket("data")
+ 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 {
+ t.Fatal(err)
+ }
+ return s, g, rec, model
+}
+
+func writeDoorClaim(t *testing.T, s *Server, g dbRequest, c *hostclaim.Claim) {
+ t.Helper()
+ key, _ := hostclaim.Key(g.host)
+ _, etag, err := s.cfg.Releases.Get(t.Context(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ b, err := json.Marshal(c)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err = s.cfg.Releases.Put(t.Context(), key, b, etag); err != nil {
+ t.Fatal(err)
+ }
+}
+func cloneDoorClaim(c *hostclaim.Claim) *hostclaim.Claim {
+ b, _ := json.Marshal(c)
+ var out hostclaim.Claim
+ _ = json.Unmarshal(b, &out)
+ return &out
+}
+
+func TestDoorPlanHandles(t *testing.T) {
+ key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ secret := dbHandleSecret(key)
+ h := dbFileHandle{V: 1, Host: "shop.example", Location: "instances/shop.example/inc", Key: "instances/shop.example/inc/0000/a", VersionID: "v+/?", Incarnation: "inc", Expiry: 200, Account: "bab", App: "jam", Size: 6}
+ token, err := signDBHandle(secret, h)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, err := verifyDBHandle(dbHandleSecret(key), token, time.Unix(199, 0))
+ if err != nil || got != h {
+ t.Fatal(got, err)
+ }
+ for _, bad := range []string{"", token + "x", "x" + token, strings.Repeat("x", 8193)} {
+ if _, err := verifyDBHandle(secret, bad, time.Unix(199, 0)); !errors.Is(err, errDBHandle) {
+ t.Fatal(bad, err)
+ }
+ }
+ if _, err := verifyDBHandle([]byte(strings.Repeat("x", 32)), token, time.Unix(199, 0)); !errors.Is(err, errDBHandle) {
+ t.Fatal(err)
+ }
+ if _, err := verifyDBHandle(secret, token, time.Unix(200, 0)); !errors.Is(err, errDBHandleExpired) {
+ t.Fatal(err)
+ }
+}
+
+func TestDoorPlanListingAndPinnedOpen(t *testing.T) {
+ for _, legacy := range []bool{false, true} {
+ t.Run(map[bool]string{true: "legacy", false: "scoped"}[legacy], func(t *testing.T) {
+ s, g, _, m := doorPlanFixture(t, legacy)
+ loc := g.claim.Claim.Holder.Location
+ key := dbVersionKey(loc, 0, 1, 1)
+ m.Put("data", key, []byte("older"))
+ 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)
+ m.Put("data", loc+"/lock.json", []byte("secret"))
+ m.Put("data", dbVersionKey(loc+"/other-incarnation", 0, 3, 3), []byte("other"))
+ m.Put("data", dbVersionKey(loc+"-sibling", 0, 4, 4), []byte("sibling"))
+ m.Put("data", dbVersionKey(loc+"/g-old", 0, 5, 5), []byte("generation"))
+ m.PageSize = 1
+ out, err := s.listDBVersions(t.Context(), g, 0)
+ if err != nil || out.Protocol != 2 || len(out.Files) != 1 {
+ t.Fatalf("%+v %v", out, err)
+ }
+ token := out.Files[0].Handle
+ h, err := verifyDBHandle(s.dbHandleKey, token, time.Now())
+ if err != nil || h.VersionID != id || h.Location != loc {
+ t.Fatal(h, err)
+ }
+ req := httptest.NewRequest(http.MethodGet, "/", nil)
+ req.Header.Set(dbdoor.HandleHeader, token)
+ rd, err := s.pinnedDBReader(req, g, 0, 1, 1)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Stall after admission, then overwrite the identical legacy key and
+ // advance the holder to a new incarnation before the physical open.
+ m.Put("data", key, []byte("replacement"))
+ next := cloneDoorClaim(g.claim.Claim)
+ next.Gen++
+ next.Holder.CreatedAt = time.Unix(1, 0)
+ 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, 102, 3)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := io.ReadAll(rc)
+ rc.Close()
+ if err != nil || string(body) != "igi" {
+ t.Fatalf("%q %v", body, err)
+ }
+ g2 := g
+ g2.claim.Claim = next
+ if _, err := s.pinnedDBReader(req, g2, 0, 1, 1); err == nil {
+ t.Fatal("new incarnation accepted old handle")
+ }
+ bad := g
+ bad.host = "other.example"
+ if _, err := s.pinnedDBReader(req, bad, 0, 1, 1); err == nil {
+ t.Fatal("cross host")
+ }
+ if _, err := s.pinnedDBReader(req, g, 0, 2, 2); err == nil {
+ t.Fatal("cross coordinate")
+ }
+ if err := m.DeleteVersion(t.Context(), "data", key, id); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, 0, 0); !errors.Is(err, os.ErrNotExist) {
+ t.Fatal(err)
+ }
+ })
+ }
+}
+
+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), doorLTXHeader(t, 1, 1))
+ fault := &doorReadFault{Store: s.cfg.Releases}
+ s.cfg.Releases = fault
+ s.cfg.DBVersions = doorVersionHook{VersionedStore: m, after: func() {
+ next := cloneDoorClaim(g.claim.Claim)
+ next.Gen++
+ switch change {
+ case "holder":
+ next.Holder.Stamp = next.Gen
+ case "op":
+ next.Op = hostclaim.Op{Kind: hostclaim.KindDeleting, ID: "delete"}
+ case "same-gen-op":
+ next.Gen--
+ next.Op.Exec = "another-box"
+ case "chore":
+ next.Touched.Complete = true
+ case "absent":
+ fault.err = os.ErrNotExist
+ return
+ case "read-error":
+ fault.err = errors.New("offline")
+ return
+ case "malformed":
+ next.V = 999
+ }
+ writeDoorClaim(t, s, g, next)
+ }}
+ out, err := s.listDBVersions(t.Context(), g, 0)
+ if change == "chore" {
+ if err != nil || len(out.Files) != 1 {
+ t.Fatal(out, err)
+ }
+ } else if err == nil || len(out.Files) != 0 {
+ t.Fatalf("partial/success: %+v %v", out, err)
+ }
+ })
+ }
+}
+
+func TestDoorPlanClaimRefusals(t *testing.T) {
+ 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 }
+ c := cloneDoorClaim(g.claim.Claim)
+ want := errDBClaimClosed
+ switch change {
+ case "absent":
+ s.cfg.Releases = release.NewDirStore(t.TempDir())
+ want = errDBNotAdopted
+ case "free":
+ c.Holder = nil
+ c.Op = hostclaim.Op{}
+ case "deleting", "parking", "purging":
+ c.Op = hostclaim.Op{Kind: change, ID: "x"}
+ case "owner":
+ g.app = "other"
+ want = os.ErrNotExist
+ case "incarnation":
+ rec.CreatedAt = time.Unix(1, 0)
+ want = os.ErrNotExist
+ case "location":
+ c.Holder.Location = ""
+ want = hostclaim.ErrMalformed
+ case "malformed":
+ c.V = 99
+ want = hostclaim.ErrMalformed
+ case "read-error":
+ want = errors.New("offline")
+ s.cfg.Releases = &doorReadFault{Store: s.cfg.Releases, err: want}
+ }
+ if change != "absent" && change != "read-error" {
+ writeDoorClaim(t, s, dbRequest{host: rec.Host}, c)
+ }
+ if _, err := s.dbClaim(t.Context(), g, rec); !errors.Is(err, want) {
+ t.Fatalf("%v want %v", err, want)
+ }
+ })
+ }
+}
+
+func TestDoorPlanListingFailureAndRanges(t *testing.T) {
+ s, g, _, m := doorPlanFixture(t, true)
+ for _, lo := range []uint64{1, 2, 3} {
+ m.Put("data", dbVersionKey(g.claim.Claim.Holder.Location, 0, lo, lo), []byte("sixsix"))
+ }
+ m.PageSize = 1
+ m.FailListAfterPages = 1
+ if out, err := s.listDBVersions(t.Context(), g, 0); err == nil || len(out.Files) != 0 {
+ t.Fatal(out, err)
+ }
+ for _, tc := range []struct {
+ offset, size, want int64
+ ok bool
+ }{{0, 0, 6, true}, {2, 0, 4, true}, {2, 2, 2, true}, {2, 100, 4, true}, {6, 0, 0, false}, {-1, 0, 0, false}, {0, -1, 0, false}} {
+ n, err := dbVersionRange(6, tc.offset, tc.size)
+ if (err == nil) != tc.ok || n != tc.want {
+ t.Fatal(tc, n, err)
+ }
+ }
+ w := httptest.NewRecorder()
+ r := httptest.NewRequest("GET", "/", nil)
+ r.Header.Set("Range", "bytes=2-3")
+ r.Header.Set(dbdoor.HandleHeader, "signed")
+ if err := dbVersionHeaders(w, r, 6, 2, 2); err != nil {
+ t.Fatal(err)
+ }
+ if w.Header().Get("Content-Length") != "2" || w.Header().Get("Content-Range") != "bytes 2-3/6" || w.Header().Get(dbdoor.HandleHeader) != "signed" {
+ t.Fatal(w.Header())
+ }
+}
+
+type doorOpenHook struct {
+ lease.VersionedStore
+ open func(context.Context, string, string, string) (io.ReadCloser, error)
+}
+
+func (v doorOpenHook) OpenVersion(ctx context.Context, b, k, id string) (io.ReadCloser, error) {
+ return v.open(ctx, b, k, id)
+}
+
+type doorTrackedReader struct {
+ io.Reader
+ closed bool
+}
+
+func (r *doorTrackedReader) Close() error { r.closed = true; return nil }
+
+type doorCancelledReader struct{ ctx context.Context }
+
+func (r doorCancelledReader) Read([]byte) (int, error) { return 0, r.ctx.Err() }
+
+func TestDoorPlanRangeReaderCloses(t *testing.T) {
+ for _, mode := range []string{"complete", "short-skip", "cancelled-skip"} {
+ t.Run(mode, func(t *testing.T) {
+ s, g, _, m := doorPlanFixture(t, true)
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ tracked := &doorTrackedReader{Reader: strings.NewReader("abcdef")}
+ if mode == "short-skip" {
+ tracked.Reader = strings.NewReader("a")
+ }
+ if mode == "cancelled-skip" {
+ cancel()
+ tracked.Reader = doorCancelledReader{ctx: ctx}
+ }
+ key := dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1)
+ s.cfg.DBVersions = doorOpenHook{VersionedStore: m, open: func(ctx context.Context, b, k, id string) (io.ReadCloser, error) {
+ if b != "data" || k != key || id != "selected" {
+ t.Fatal(b, k, id)
+ }
+ return tracked, nil
+ }}
+ rd := &dbVersionReader{store: s.cfg.DBVersions, bucket: "data", handle: dbFileHandle{Location: g.claim.Claim.Holder.Location, Key: key, VersionID: "selected", Size: 6}}
+ rc, err := rd.OpenLTXFile(ctx, 0, 1, 1, 2, 2)
+ if mode != "complete" {
+ if err == nil || !tracked.closed {
+ t.Fatal(err, tracked.closed)
+ }
+ if mode == "cancelled-skip" && !errors.Is(err, context.Canceled) {
+ t.Fatal(err)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ b, err := io.ReadAll(rc)
+ if err != nil || string(b) != "cd" {
+ t.Fatal(string(b), err)
+ }
+ if tracked.closed {
+ t.Fatal("closed before caller finished")
+ }
+ rc.Close()
+ if !tracked.closed {
+ t.Fatal("reader leaked")
+ }
+ })
+ }
+}
diff --git a/internal/console/server.go b/internal/console/server.go
index 7d1093876..3f87a9364 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -13,6 +13,7 @@ import (
"encoding/json"
"errors"
"fmt"
+ "github.com/carlosframework/platform/internal/lease"
"html/template"
"log"
"math"
@@ -455,8 +456,10 @@ const (
// Config is everything the server reads its world through.
type Config struct {
- Origin string // the console's public origin — the OAuth client_id
- Keymail string // keymail base URL
+ HostClaim bool
+ DBVersions lease.VersionedStore
+ Origin string // the console's public origin — the OAuth client_id
+ Keymail string // keymail base URL
// Registry is the path to the platform registry DB. Read-mostly (routes,
// domains, alerts), but fleets.go's fleet/box/placement doors (customer
// fleets plan, Task 7) genuinely write through it via openRegistry
@@ -741,8 +744,9 @@ type Config struct {
// Server is the console's http.Handler.
type Server struct {
- cfg Config
- store *Store
+ dbHandleKey []byte
+ cfg Config
+ store *Store
// appRegionOf and beforeClaimInsert are TEST SEAMS (nil in
// production): a store failure on the app-region read (C20), and a
// competing insert landing after the region was resolved.
@@ -1083,6 +1087,9 @@ func normalizeDelegationZone(raw string) string {
// NewServer wires a server over an opened Store, loading (or minting) the
// console's Keymail signing key.
func NewServer(cfg Config, st *Store) (*Server, error) {
+ if cfg.HostClaim && cfg.HibernateBucket != "" && cfg.dbClaimRegion() == "" {
+ return nil, fmt.Errorf("CARLOS_HOST_CLAIM: %s", "region is required")
+ }
cfg.Origin = strings.TrimSuffix(cfg.Origin, "/")
// originHost is what a RedirectHosts entry must never equal: redirecting
// the console's own origin to itself is a loop, not a retirement.
@@ -1191,6 +1198,9 @@ func NewServer(cfg Config, st *Store) (*Server, error) {
// CLI identity: the device-code login `carlos auth login` drives, the
// approval page, and the tokens an owner can see and kill (authcli.go).
// The db door's production replica reader (apidb.go); tests repoint it.
+ if cfg.HostClaim {
+ s.dbHandleKey = dbHandleSecret(key)
+ }
s.dbReplica = s.cachedReadReplica
s.dbStreams = newDBStreamLimits(dbMaxStreams, dbMaxStreamsPerPrincipal)
s.dbStreamIdle = dbStreamIdleDefault
diff --git a/internal/dbdoor/client.go b/internal/dbdoor/client.go
index 7632dceff..fb54df849 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.
@@ -79,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
@@ -91,8 +106,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 {
+ 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 {
continue
}
@@ -101,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
@@ -109,6 +128,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 +142,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..002095564
--- /dev/null
+++ b/internal/dbdoor/client_handles.go
@@ -0,0 +1,92 @@
+package dbdoor
+
+import (
+ "errors"
+ "fmt"
+ "sync"
+)
+
+const HandleHeader = "X-Carlos-DB-Handle"
+
+type dbHandleState struct {
+ listMu sync.Mutex
+ mu sync.Mutex
+ plan []LTXFile
+ 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.Plan && list.Protocol != 2 {
+ return nil, errors.New("dbdoor: invalid pinned listing")
+ }
+ 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(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
+ 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)
+ }
+ 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
+}
+
+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))
+}
+
+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
new file mode 100644
index 000000000..308b25e8a
--- /dev/null
+++ b/internal/dbdoor/client_handles_test.go
@@ -0,0 +1,168 @@
+package dbdoor
+
+import (
+ "context"
+ "encoding/json"
+ "io"
+ "net/http"
+ "strconv"
+ "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)
+ }
+ }
+}
+
+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()
+}
+
+func TestHandleProtocolRetainsSnapshotCRC(t *testing.T) {
+ for _, first := range []string{"0", "12345678901234567890"} {
+ t.Run(first, func(t *testing.T) {
+ c := NewClient(nil)
+ for _, crc := range []string{first, "7", "null"} {
+ var list ListResponse
+ raw := `{"protocol":2,"files":[{"level":9,"min_txid":1,"max_txid":11,"size":100,"handle":"` + crc + `","snapshot_crc64":` + crc + `}]}`
+ if err := json.Unmarshal([]byte(raw), &list); err != nil {
+ t.Fatal(err)
+ }
+ files, err := c.rememberFiles(9, list)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if files[0].Handle != first || files[0].SnapshotCRC64 == nil {
+ t.Fatalf("lost first handle metadata: %+v", files[0])
+ }
+ want, err := strconv.ParseUint(first, 10, 64)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if *files[0].SnapshotCRC64 != want {
+ t.Fatalf("CRC changed to %d", *files[0].SnapshotCRC64)
+ }
+ }
+ })
+ }
+}
diff --git a/internal/dbdoor/coords.go b/internal/dbdoor/coords.go
index 67aba1ced..a6da38a86 100644
--- a/internal/dbdoor/coords.go
+++ b/internal/dbdoor/coords.go
@@ -32,17 +32,21 @@ const MaxLevel = 9
// MaxTXID are MANDATORY: Restore rejects Size < ltx.HeaderSize and
// ResumableReader uses Size to detect a premature EOF.
type LTXFile struct {
- Level int `json:"level"`
- MinTXID uint64 `json:"min_txid"`
- MaxTXID uint64 `json:"max_txid"`
- Size int64 `json:"size"`
- CreatedAt time.Time `json:"created_at"`
+ Handle string `json:"handle,omitempty"`
+ SnapshotCRC64 *uint64 `json:"snapshot_crc64,omitempty"`
+ Level int `json:"level"`
+ MinTXID uint64 `json:"min_txid"`
+ MaxTXID uint64 `json:"max_txid"`
+ Size int64 `json:"size"`
+ 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 {
- Files []LTXFile `json:"files"`
+ Protocol int `json:"protocol,omitempty"`
+ Plan bool `json:"plan,omitempty"`
+ Files []LTXFile `json:"files"`
}
// Base is the door's root for one instance: everything below it is a
diff --git a/internal/dbdoor/restore.go b/internal/dbdoor/restore.go
index 85d5ee802..35a3ac947 100644
--- a/internal/dbdoor/restore.go
+++ b/internal/dbdoor/restore.go
@@ -81,6 +81,16 @@ func RestoreTo(ctx context.Context, c *Client, parent string) (*Restored, error)
break
}
removeCopy(out)
+ rec.mu.Lock()
+ checkFailed := rec.checkFailed
+ rec.mu.Unlock()
+ if checkFailed {
+ os.RemoveAll(dir)
+ if cerr := ctx.Err(); cerr != nil {
+ return nil, cerr
+ }
+ return nil, ErrCheckFailed
+ }
vanished := errors.Is(err, os.ErrNotExist) || rec.sawVanish()
if attempt == 0 && vanished && ctx.Err() == nil {
continue
@@ -139,11 +149,12 @@ func removeCopy(out string) {
// newest file opened is the copy's "as of" (R1).
type recorder struct {
*Client
- dir string // the private directory each file is downloaded into
- mu sync.Mutex
- vanished bool
- maxTXID ltx.TXID
- header *bytes.Buffer // first ltx.HeaderSize bytes of the maxTXID file
+ dir string // the private directory each file is downloaded into
+ mu sync.Mutex
+ vanished bool
+ checkFailed bool
+ maxTXID ltx.TXID
+ header *bytes.Buffer // first ltx.HeaderSize bytes of the maxTXID file
}
func (r *recorder) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
@@ -187,7 +198,9 @@ const dbDownloadMaxStalls = 5
func (r *recorder) download(ctx context.Context, rc io.ReadCloser, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
f, err := os.CreateTemp(r.dir, "ltx-*")
if err != nil {
- rc.Close()
+ if rc != nil {
+ rc.Close()
+ }
return nil, err
}
fail := func(err error) (io.ReadCloser, error) {
@@ -258,9 +271,33 @@ func (r *recorder) download(ctx context.Context, rc io.ReadCloser, level int, mi
if _, err := f.Seek(0, io.SeekStart); err != nil {
return fail(err)
}
+ if err := r.verifyPinnedSnapshot(ctx, f, level, minTXID, maxTXID, offset, written); err != nil {
+ r.mu.Lock()
+ r.checkFailed = true
+ r.mu.Unlock()
+ return fail(err)
+ }
return &downloaded{ctx: ctx, File: f}, nil
}
+func (r *recorder) verifyPinnedSnapshot(ctx context.Context, f *os.File, level int, lo, hi ltx.TXID, offset, written int64) error {
+ r.handles.mu.Lock()
+ meta := r.handles.files[GetTail(level, uint64(lo), uint64(hi))]
+ r.handles.mu.Unlock()
+ if meta.SnapshotCRC64 == nil {
+ return nil
+ }
+ if offset != 0 || written != meta.Size {
+ return ErrCheckFailed
+ }
+ dec := ltx.NewDecoder(&downloaded{ctx: ctx, File: f})
+ if err := dec.Verify(); err != nil || uint64(dec.Trailer().FileChecksum) != *meta.SnapshotCRC64 {
+ return ErrCheckFailed
+ }
+ _, err := f.Seek(0, io.SeekStart)
+ return err
+}
+
// downloaded is a file read once and removed on Close.
type downloaded struct {
ctx context.Context
diff --git a/internal/dbdoor/restore_test.go b/internal/dbdoor/restore_test.go
index 3836ed4df..2be975f52 100644
--- a/internal/dbdoor/restore_test.go
+++ b/internal/dbdoor/restore_test.go
@@ -521,3 +521,24 @@ func TestRestoreToRetriesABusyFirstOpen(t *testing.T) {
t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows)
}
}
+
+func TestDownloadTempFileFailure(t *testing.T) {
+ for _, supplied := range []bool{false, true} {
+ t.Run(strconv.FormatBool(supplied), func(t *testing.T) {
+ var rc io.ReadCloser
+ var open atomic.Int32
+ if supplied {
+ open.Store(1)
+ rc = &countingBody{ReadCloser: io.NopCloser(strings.NewReader("unused")), open: &open}
+ }
+ r := &recorder{dir: filepath.Join(t.TempDir(), "missing")}
+ got, err := r.download(t.Context(), rc, 0, 1, 1, 0, 0)
+ if got != nil || !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("download = %v, %v", got, err)
+ }
+ if open.Load() != 0 {
+ t.Fatal("supplied reader leaked")
+ }
+ })
+ }
+}
diff --git a/internal/lease/leasetest/bucket.go b/internal/lease/leasetest/bucket.go
index fe13f4d98..94450f5e0 100644
--- a/internal/lease/leasetest/bucket.go
+++ b/internal/lease/leasetest/bucket.go
@@ -254,6 +254,31 @@ func (f *VersionedBucket) OpenVersion(ctx context.Context, bucket, key, versionI
return io.NopCloser(bytes.NewReader(b)), nil
}
+func versionRange(body []byte, offset, length int64) ([]byte, error) {
+ if offset < 0 || length == 0 || (length > 0 && offset > (1<<63-1)-(length-1)) || offset >= int64(len(body)) {
+ return nil, fmt.Errorf("leasetest: invalid version range")
+ }
+ n := int64(len(body)) - offset
+ if length > 0 && length < n {
+ n = length
+ }
+ return body[offset : offset+n], nil
+}
+
+func (f *VersionedBucket) OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ b, err := f.read(bucket, key, versionID)
+ if err != nil {
+ return nil, err
+ }
+ b, err = versionRange(b, offset, length)
+ if err != nil {
+ return nil, err
+ }
+ return io.NopCloser(bytes.NewReader(b)), nil
+}
+
func (f *VersionedBucket) copyLocked(srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
if srcVersionID == "" {
return "", "", lease.ErrNoVersionID
@@ -338,3 +363,5 @@ func (f *VersionedBucket) PutIfAbsent(ctx context.Context, bucket, key string, b
}
var _ lease.VersionedStore = (*VersionedBucket)(nil)
+
+var _ lease.VersionRangeOpener = (*VersionedBucket)(nil)
diff --git a/internal/lease/leasetest/contract.go b/internal/lease/leasetest/contract.go
index d1a823408..235d75859 100644
--- a/internal/lease/leasetest/contract.go
+++ b/internal/lease/leasetest/contract.go
@@ -105,6 +105,91 @@ func RunVersionedStoreContract(t *testing.T, mk Maker) {
}
})
+ t.Run("range by exact version", func(t *testing.T) {
+ st, m, b, served := mk(t)
+ ranged, ok := st.(lease.VersionRangeOpener)
+ if !ok {
+ t.Skip("store does not support ranged version reads")
+ }
+ key := "dir/a b+c%?d"
+ id := m.Put(b, key, []byte("original"))
+ m.Put(b, key, []byte("replacement"))
+ marker := m.BareDelete(b, key)
+ for _, tc := range []struct {
+ name string
+ offset, length int64
+ want string
+ }{
+ {"bounded", 2, 3, "igi"},
+ {"to end", 3, -1, "ginal"},
+ {"past end", 6, 100, "al"},
+ {"whole", 0, -1, "original"},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ rc, err := ranged.OpenVersionRange(ctx, b, key, id, tc.offset, tc.length)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := io.ReadAll(rc)
+ closeErr := rc.Close()
+ if err != nil || closeErr != nil || string(body) != tc.want {
+ t.Fatalf("range = %q, %v, close %v; want %q", body, err, closeErr, tc.want)
+ }
+ })
+ }
+ for _, tc := range []struct {
+ name, bucket, version string
+ offset, length int64
+ want error
+ local bool
+ }{
+ {"missing version ID", b, "", 0, 1, lease.ErrNoVersionID, true},
+ {"negative offset", b, id, -1, 1, nil, true},
+ {"zero length", b, id, 0, 0, nil, true},
+ {"overflow", b, id, 2, 1<<63 - 1, nil, true},
+ {"unsatisfiable", b, id, 8, -1, nil, false},
+ {"missing version", b, "missing", 0, 1, lease.ErrMirrorNotFound, false},
+ {"marker", b, marker, 0, 1, lease.ErrIsDeleteMarker, false},
+ {"missing bucket", "nobucket", id, 0, 1, nil, false},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ before := 0
+ if served != nil {
+ before = served()
+ }
+ rc, err := ranged.OpenVersionRange(ctx, tc.bucket, key, tc.version, tc.offset, tc.length)
+ if rc != nil {
+ rc.Close()
+ }
+ if err == nil || (tc.want != nil && !errors.Is(err, tc.want)) {
+ t.Fatalf("range error = %v; want %v", err, tc.want)
+ }
+ if tc.want == nil && errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Fatalf("range error was treated as missing version: %v", err)
+ }
+ if tc.local && served != nil && served() != before {
+ t.Fatal("invalid range sent a request")
+ }
+ })
+ }
+ if err := st.DeleteVersion(ctx, b, key, id); err != nil {
+ t.Fatal(err)
+ }
+ if rc, err := ranged.OpenVersionRange(ctx, b, key, id, 2, 3); !errors.Is(err, lease.ErrMirrorNotFound) {
+ if rc != nil {
+ rc.Close()
+ }
+ t.Fatalf("deleted version range = %v", err)
+ }
+ m.FailOn = func(op, key string) error { return errors.New("offline") }
+ if rc, err := ranged.OpenVersionRange(ctx, b, key, id, 2, 3); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
+ if rc != nil {
+ rc.Close()
+ }
+ t.Fatalf("failed range = %v", err)
+ }
+ })
+
t.Run("copy by exact version; create-only; markers", func(t *testing.T) {
st, m, b, _ := mk(t)
k := "src/a b+c%?d"
diff --git a/internal/lease/leasetest/s3server.go b/internal/lease/leasetest/s3server.go
index f0877aeb5..92cba0b9d 100644
--- a/internal/lease/leasetest/s3server.go
+++ b/internal/lease/leasetest/s3server.go
@@ -3,9 +3,11 @@ package leasetest
import (
"encoding/xml"
"errors"
+ "fmt"
"io"
"net/http"
"net/url"
+ "strconv"
"strings"
"sync/atomic"
"time"
@@ -110,6 +112,29 @@ func NewS3Server(f *VersionedBucket, serves ...string) (http.Handler, func() int
writeErr(w, err)
return
}
+ if hdr := r.Header.Get("Range"); hdr != "" {
+ span, ok := strings.CutPrefix(hdr, "bytes=")
+ start, end, dash := strings.Cut(span, "-")
+ offset, startErr := strconv.ParseInt(start, 10, 64)
+ length := int64(-1)
+ if end != "" {
+ last, endErr := strconv.ParseInt(end, 10, 64)
+ if endErr != nil || last < offset || last-offset == 1<<63-1 {
+ s3err(w, http.StatusRequestedRangeNotSatisfiable, "InvalidRange")
+ return
+ }
+ length = last - offset + 1
+ }
+ total := len(body)
+ body, err = versionRange(body, offset, length)
+ if !ok || !dash || startErr != nil || err != nil {
+ s3err(w, http.StatusRequestedRangeNotSatisfiable, "InvalidRange")
+ return
+ }
+ w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+int64(len(body))-1, total))
+ w.Header().Set("Content-Length", strconv.Itoa(len(body)))
+ w.WriteHeader(http.StatusPartialContent)
+ }
w.Write(body)
case r.Method == http.MethodPut && r.Header.Get("x-amz-copy-source") != "":
f.serveCopy(w, r, bucket, key)
diff --git a/internal/lease/versioned.go b/internal/lease/versioned.go
index 54377d71b..c07bd037e 100644
--- a/internal/lease/versioned.go
+++ b/internal/lease/versioned.go
@@ -63,6 +63,13 @@ type VersionedStore interface {
PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (versionID string, err error)
}
+// VersionRangeOpener optionally streams a range of one exact object version.
+// A negative length reads to the end; a positive length bounds the read.
+// Zero length and negative offsets are invalid. Errors match OpenVersion.
+type VersionRangeOpener interface {
+ OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error)
+}
+
// versionedS3 binds every bucket to its own regional client explicitly.
// An unbound bucket is an error, never a guess.
type versionedS3 struct {
@@ -164,6 +171,24 @@ func mapReadErr(err error) error {
}
func (s *versionedS3) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
+ return s.openVersion(ctx, bucket, key, versionID, nil)
+}
+
+func (s *versionedS3) OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error) {
+ if versionID == "" {
+ return nil, ErrNoVersionID
+ }
+ if offset < 0 || length == 0 || (length > 0 && offset > (1<<63-1)-(length-1)) {
+ return nil, errors.New("lease: invalid version range")
+ }
+ rangeHeader := fmt.Sprintf("bytes=%d-", offset)
+ if length > 0 {
+ rangeHeader += fmt.Sprint(offset + length - 1)
+ }
+ return s.openVersion(ctx, bucket, key, versionID, &rangeHeader)
+}
+
+func (s *versionedS3) openVersion(ctx context.Context, bucket, key, versionID string, byteRange *string) (io.ReadCloser, error) {
if versionID == "" {
return nil, ErrNoVersionID
}
@@ -171,7 +196,7 @@ func (s *versionedS3) OpenVersion(ctx context.Context, bucket, key, versionID st
if err != nil {
return nil, err
}
- out, err := c.GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID)})
+ out, err := c.GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID), Range: byteRange})
if err != nil {
if m := mapReadErr(err); m != err {
return nil, m
@@ -266,3 +291,5 @@ func (s *versionedS3) PutIfAbsent(ctx context.Context, bucket, key string, body
}
var _ VersionedStore = (*versionedS3)(nil)
+
+var _ VersionRangeOpener = (*versionedS3)(nil)
diff --git a/internal/lease/versioned_test.go b/internal/lease/versioned_test.go
index a7d415af3..2d7ad8f1f 100644
--- a/internal/lease/versioned_test.go
+++ b/internal/lease/versioned_test.go
@@ -2,6 +2,8 @@ package lease_test
import (
"context"
+ "io"
+ "net/http"
"net/http/httptest"
"testing"
@@ -95,3 +97,52 @@ func TestVersionedS3CopyEscapesLikeAWS(t *testing.T) {
t.Fatalf("copied %q, want the pinned old version", cur)
}
}
+
+func TestVersionedS3RangeWire(t *testing.T) {
+ m := leasetest.NewVersionedBucket()
+ m.AddBucket("data")
+ key := "dir/a b+c%?d"
+ id := m.Put("data", key, []byte("original"))
+ m.Put("data", key, []byte("replacement"))
+ handler, _ := leasetest.NewS3Server(m, "data")
+ for _, tc := range []struct {
+ length int64
+ header, contentRange, body string
+ }{
+ {3, "bytes=2-4", "bytes 2-4/8", "igi"},
+ {-1, "bytes=2-", "bytes 2-7/8", "iginal"},
+ } {
+ t.Run(tc.header, func(t *testing.T) {
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Method != http.MethodGet || r.URL.Query().Get("versionId") != id || r.Header.Get("Range") != tc.header {
+ t.Errorf("request = %s %s, Range %q", r.Method, r.URL, r.Header.Get("Range"))
+ }
+ rec := httptest.NewRecorder()
+ handler.ServeHTTP(rec, r)
+ if rec.Code != http.StatusPartialContent || rec.Header().Get("Content-Range") != tc.contentRange || rec.Body.String() != tc.body {
+ t.Errorf("response = %d %v %q", rec.Code, rec.Header(), rec.Body.String())
+ }
+ for k, values := range rec.Header() {
+ w.Header()[k] = values
+ }
+ w.WriteHeader(rec.Code)
+ w.Write(rec.Body.Bytes())
+ }))
+ defer srv.Close()
+ st := lease.NewVersionedS3WithClients(map[string]*awss3.Client{"data": wireClient(t, srv.URL)})
+ ranged, ok := st.(lease.VersionRangeOpener)
+ if !ok {
+ t.Fatal("S3 store lacks ranged version reads")
+ }
+ rc, err := ranged.OpenVersionRange(t.Context(), "data", key, id, 2, tc.length)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := io.ReadAll(rc)
+ rc.Close()
+ if err != nil || string(body) != tc.body {
+ t.Fatalf("read = %q, %v", body, err)
+ }
+ })
+ }
+}