amadan

carlos / platform Public

console, dbdoor: bucket-region fallback, pinned snapshot CRC verified on restore, nil-reader guard (Plan 7 fix round 3)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude fffd806df5b2570b14b3194e5fcb8a668ee30750 parent ce9aef9
7 files changed, +352 −25
  • internal/console/apidbclaim.go +27 −13
  • internal/console/apidbclaim_r3_test.go +221 −0
  • internal/console/server.go +3 −0
  • internal/dbdoor/client_handles_test.go +30 −0
  • internal/dbdoor/coords.go +7 −6
  • internal/dbdoor/restore.go +43 −6
  • internal/dbdoor/restore_test.go +21 −0

CI passed — run details

diff --git a/internal/console/apidbclaim.go b/internal/console/apidbclaim.go
index f7e71edec..fbd685ead 100644
--- a/internal/console/apidbclaim.go
+++ b/internal/console/apidbclaim.go
@@ -39,16 +39,17 @@ var (
// 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"`
+ 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 {
@@ -170,22 +171,35 @@ func (s *Server) dbClaimFailure(w http.ResponseWriter, r *http.Request, g dbRequ
}
}
+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.homeRegionOut(s.cfg.Region), Host: g.host, Holder: *h,
+ 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 {
- 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})
+ 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})
+ 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 {
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/server.go b/internal/console/server.go
index 44994910b..9e1fb9cdc 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -1087,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.
diff --git a/internal/dbdoor/client_handles_test.go b/internal/dbdoor/client_handles_test.go
index 52fccd044..308b25e8a 100644
--- a/internal/dbdoor/client_handles_test.go
+++ b/internal/dbdoor/client_handles_test.go
@@ -5,6 +5,7 @@ import (
"encoding/json"
"io"
"net/http"
+ "strconv"
"strings"
"testing"
@@ -136,3 +137,32 @@ func TestHandleProtocolCompletePlan(t *testing.T) {
}
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 444bf4c3b..a6da38a86 100644
--- a/internal/dbdoor/coords.go
+++ b/internal/dbdoor/coords.go
@@ -32,12 +32,13 @@ const MaxLevel = 9
// MaxTXID are MANDATORY: Restore rejects Size < ltx.HeaderSize and
// ResumableReader uses Size to detect a premature EOF.
type LTXFile struct {
- Handle string `json:"handle,omitempty"`
- Level int `json:"level"`
- MinTXID uint64 `json:"min_txid"`
- MaxTXID uint64 `json:"max_txid"`
- 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 contains one ordered level, or a complete restore plan with
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")
+ }
+ })
+ }
+}