amadan

carlos / platform Public

console: serve replica LTX files by coordinate, members only

The carlos db door (spec 2026-09-02 §6). Two GET routes list one level's
LTX files and stream one file by level and TXID span, read from the
region data bucket through litestream's own S3 client, so the key layout
comes from the code that writes it. Levels 0-9 answer an empty list
rather than 404, listings ascend by min_txid, and Range follows the S3
client's two forms.

The routes are closed doors (the zero Set): every service credential,
admin included, gets the constant closed-door 403 before any handler
runs. A member who is not in the account, or names no such app or
instance, gets the byte-identical 404. The class rail, another region, a
fleet placement and a pool are refused with a 409 before any replica
client is built; an AccessDenied from S3 maps to the same residency
refusal, and store errors are logged, never returned. Each file opened
writes a db-read audit row. The door has its own token budget.

cmd/console reads the CARLOS_HIBERNATE_* triple through the activator's
ConfigFromEnv, plus CARLOS_REGION. internal/dbdoor/dbdoortest builds real
litestream replicas for tests on both sides of the wire.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e6d358ea95e393c62b190ca724222a12c6a5a8ea parent c85bbd2
9 files changed, +962 −4
  • cmd/console/main.go +31 −2
  • cmd/console/wiring_test.go +32 −0
  • internal/console/apidb.go +305 −0
  • internal/console/apidb_test.go +436 −0
  • internal/console/authcli.go +13 −0
  • internal/console/server.go +7 −0
  • internal/console/svcroutes_test.go +11 −2
  • internal/dbdoor/dbdoortest/replica.go +78 −0
  • internal/dbdoor/dbdoortest/replica_test.go +49 −0
diff --git a/cmd/console/main.go b/cmd/console/main.go
index 8b5a8e789..4a3b04776 100644
--- a/cmd/console/main.go
+++ b/cmd/console/main.go
@@ -82,6 +82,7 @@ import (
"strings"
"time"
+ "github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/appconfig"
"github.com/carlosframework/platform/internal/console"
"github.com/carlosframework/platform/internal/dnsrec"
@@ -265,6 +266,7 @@ func serve() {
// One release.Store, named, because the senders sweep below needs the
// same one the server got.
rel := openStore()
+ dbBucket, dbRegion, dbPrefix, placement := dbDoorWiring()
srv, err := console.NewServer(console.Config{
Origin: origin,
Keymail: envOr("CONSOLE_KEYMAIL", "https://keymail.dev"),
@@ -274,8 +276,14 @@ func serve() {
// opens it with — /fleet/architecture draws the box -> bucket edge
// from it. Empty on a CARLOS_DEPLOYMENT_DIR deployment, which
// correctly draws no bucket node.
- Bucket: os.Getenv("CARLOS_DEPLOYMENT_BUCKET"),
- HomeRegion: strings.TrimSpace(os.Getenv("CONSOLE_HOME_REGION")),
+ Bucket: os.Getenv("CARLOS_DEPLOYMENT_BUCKET"),
+ HomeRegion: strings.TrimSpace(os.Getenv("CONSOLE_HOME_REGION")),
+ // The carlos db door's data bucket and this box's placement
+ // (dbDoorWiring).
+ HibernateBucket: dbBucket,
+ HibernateRegion: dbRegion,
+ HibernatePrefix: dbPrefix,
+ Region: placement,
ReleasePub: pub,
Key: key,
EdgeIPs: edgeIPs(),
@@ -1046,3 +1054,24 @@ func envOr(k, def string) string {
}
return def
}
+
+// dbDoorWiring reads what the `carlos db` door needs (internal/console
+// apidb.go; spec 2026-09-02 §6): the region data bucket triple, parsed by
+// the activator's own ConfigFromEnv so the console reads exactly the
+// prefixes the agent on this box writes, plus CARLOS_REGION, this box's
+// placement, which an instance record's region must equal. An empty bucket
+// leaves the door answering 503. A config the agent itself would refuse
+// turns the door off rather than failing console boot: the console's job is
+// not to police the agent's idle timer.
+func dbDoorWiring() (bucket, region, prefix, placement string) {
+ placement = strings.TrimSpace(os.Getenv("CARLOS_REGION"))
+ cfg, err := activator.ConfigFromEnv()
+ if err != nil {
+ log.Printf("console: %v; the carlos db door is off", err)
+ return "", "", "", placement
+ }
+ if cfg == nil {
+ return "", "", "", placement
+ }
+ return cfg.Bucket, cfg.Region, cfg.Prefix, placement
+}
diff --git a/cmd/console/wiring_test.go b/cmd/console/wiring_test.go
index 531256af5..415e59a8f 100644
--- a/cmd/console/wiring_test.go
+++ b/cmd/console/wiring_test.go
@@ -64,3 +64,35 @@ func TestScannerWiring(t *testing.T) {
}
})
}
+
+// The db door reads the agent's own CARLOS_HIBERNATE_* triple, parsed the
+// agent's way (trimmed, prefix slashes stripped), so the console reads
+// exactly the prefixes the activator beside it writes.
+func TestDBDoorWiringReadsTheAgentsTriple(t *testing.T) {
+ t.Setenv("CARLOS_HIBERNATE_BUCKET", " data-ie ")
+ t.Setenv("CARLOS_HIBERNATE_REGION", "eu-west-1")
+ t.Setenv("CARLOS_HIBERNATE_PREFIX", "/instances/")
+ t.Setenv("CARLOS_HIBERNATE_IDLE", "")
+ t.Setenv("CARLOS_REGION", " uk ")
+ bucket, region, prefix, placement := dbDoorWiring()
+ if bucket != "data-ie" || region != "eu-west-1" || prefix != "instances" || placement != "uk" {
+ t.Fatalf("= %q %q %q %q", bucket, region, prefix, placement)
+ }
+}
+
+// A bad idle duration is the agent's problem, not a reason to fail console
+// boot; it turns the door off instead, loudly.
+func TestDBDoorWiringIsOffOnABadAgentConfig(t *testing.T) {
+ t.Setenv("CARLOS_HIBERNATE_BUCKET", "data-ie")
+ t.Setenv("CARLOS_HIBERNATE_IDLE", "soon")
+ if bucket, _, _, _ := dbDoorWiring(); bucket != "" {
+ t.Fatalf("bucket = %q, want the door off", bucket)
+ }
+}
+
+func TestDBDoorWiringIsOffWithoutABucket(t *testing.T) {
+ t.Setenv("CARLOS_HIBERNATE_BUCKET", "")
+ if bucket, _, _, _ := dbDoorWiring(); bucket != "" {
+ t.Fatalf("bucket = %q, want the door off", bucket)
+ }
+}
diff --git a/internal/console/apidb.go b/internal/console/apidb.go
index ea29755e6..4f4cd8542 100644
--- a/internal/console/apidb.go
+++ b/internal/console/apidb.go
@@ -14,10 +14,24 @@
package console
import (
+ "context"
"errors"
"fmt"
+ "io"
+ "log"
+ "net/http"
+ "os"
+ "sort"
+ "strconv"
+ "sync"
+ "time"
+
+ "github.com/aws/smithy-go"
+ "github.com/superfly/ltx"
"github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
)
@@ -55,3 +69,294 @@ func (s *Server) resolveReplica(rec provision.Record, host string) (string, erro
}
return activator.HibernatePrefix(s.cfg.HibernatePrefix, host), nil
}
+
+// replicaReader is the only part of a litestream replica client the door
+// may hold: the two read methods. Holding the narrow interface means no
+// handler here can reach WriteLTXFile, DeleteLTXFiles or DeleteAll by
+// mistake, whatever the concrete client underneath can do.
+type replicaReader interface {
+ LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error)
+ OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error)
+}
+
+// cliDBBurst bounds the db door per TOKEN within the ordinary minute window.
+// Its own budget, not the instance verbs' 60 (spec §13.6): one restore is
+// ten listings plus one request per planned file, and an app with a long L0
+// tail plans dozens. The number is a ceiling on a stolen token, not a
+// policing of honest pulls.
+const cliDBBurst = 600
+
+// dbAuditAction names the audit row every served file writes.
+const dbAuditAction = "db-read"
+
+// dbRequest is what dbGate resolves for both handlers.
+type dbRequest struct {
+ acct *Account
+ actor string
+ app string
+ host string
+ prefix string
+ rd replicaReader
+}
+
+// dbGate is the door's whole gate, in the repo's order: principal (cliRoute
+// has already refused every service credential at the zero Set) ->
+// membership (byte-identical 404) -> budget (429) -> configuration (503) ->
+// the record (404 when absent) -> the class rail and residency (409), none
+// of which touches the bucket -> the replica client. ok=false means a
+// response has been written.
+func (s *Server) dbGate(w http.ResponseWriter, r *http.Request) (dbRequest, bool) {
+ // Belt to cliRoute's braces: the zero Set already stops every service
+ // principal before this runs, so this line is only reached by a request
+ // that did not come through cliRoute at all.
+ if p, ok := principalFrom(r.Context()); !ok || p.Kind != PrincipalUser {
+ http.NotFound(w, r)
+ return dbRequest{}, false
+ }
+ actor, acct, app, ok := s.apiAccountApp(w, r)
+ if !ok {
+ return dbRequest{}, false
+ }
+ if !s.cli.db.allow(hashToken(bearer(r)), time.Now()) {
+ jsonError(w, http.StatusTooManyRequests, "too many database requests. Wait a minute and try again.")
+ return dbRequest{}, false
+ }
+ if s.cfg.Releases == nil || s.cfg.HibernateBucket == "" {
+ jsonError(w, http.StatusServiceUnavailable, "this console has no data bucket set up, so it cannot read databases.")
+ return dbRequest{}, false
+ }
+ host := r.PathValue("host")
+ if !provision.ValidHostShape(host) {
+ http.NotFound(w, r)
+ return dbRequest{}, false
+ }
+ ctx := r.Context()
+ rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host)
+ if errors.Is(err, os.ErrNotExist) {
+ http.NotFound(w, r)
+ return dbRequest{}, false
+ }
+ if err != nil {
+ log.Printf("console: api db %s/%s/%s: read record: %v", acct.Sqid, app, host, err)
+ jsonError(w, http.StatusBadGateway, "could not read the instance. Try again.")
+ return dbRequest{}, false
+ }
+ if msg := dbClassRefusal(*rec); msg != "" {
+ jsonError(w, http.StatusConflict, msg)
+ return dbRequest{}, false
+ }
+ prefix, err := s.resolveReplica(*rec, host)
+ switch {
+ case errors.Is(err, ErrBadHost):
+ http.NotFound(w, r)
+ return dbRequest{}, false
+ case errors.Is(err, ErrPooled):
+ jsonError(w, http.StatusConflict, fmt.Sprintf(
+ "%s runs in the pool %s, so its data is on the pool's own server. carlos db cannot read it.", host, rec.Pool))
+ return dbRequest{}, false
+ case errors.Is(err, ErrResidency):
+ jsonError(w, http.StatusConflict, s.dbResidencyRefusal(*rec))
+ return dbRequest{}, false
+ case err != nil:
+ log.Printf("console: api db %s/%s/%s: resolve: %v", acct.Sqid, app, host, err)
+ jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.")
+ return dbRequest{}, false
+ }
+ 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)
+ jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.")
+ return dbRequest{}, false
+ }
+ return dbRequest{acct: acct, actor: actor, app: app, host: host, prefix: prefix, rd: rd}, true
+}
+
+// dbClassRefusal is the class rail as a sentence, or "" for an
+// activator-managed exec instance: the only class whose database the
+// platform replicates (spec §1 scope). Same three facts the move doors
+// refuse on; the fleet and pool cases are resolveReplica's.
+func dbClassRefusal(rec provision.Record) string {
+ if k := rec.Kind; k != "" && k != "instance" {
+ return fmt.Sprintf("%s is a %s route. The platform keeps a database copy only for instances.", rec.Host, k)
+ }
+ if rec.Addr != "" || len(rec.Addrs) > 0 {
+ return fmt.Sprintf("%s is served from its own address, so the platform does not manage its database.", rec.Host)
+ }
+ if !rec.HibernateOrDefault() {
+ return fmt.Sprintf("%s does not sleep, so the platform does not keep a copy of its database.", rec.Host)
+ }
+ return ""
+}
+
+// dbResidencyRefusal names where the data is, in the vocabulary the
+// regions door uses (the home region by its CONSOLE_HOME_REGION name).
+func (s *Server) dbResidencyRefusal(rec provision.Record) string {
+ if fleet, ok := registry.FleetPlacement(rec.Region); ok {
+ return fmt.Sprintf("%s runs on your fleet %s, so its data stays on that fleet's own server. carlos db cannot read it.", rec.Host, fleet)
+ }
+ return fmt.Sprintf("%s keeps its data in %s. This console can only read data kept in %s.",
+ rec.Host, s.dbRegionName(rec.Region), s.dbRegionName(s.cfg.Region))
+}
+
+func (s *Server) dbRegionName(region string) string {
+ if r := s.homeRegionOut(region); r != "" {
+ return "region " + r
+ }
+ return "the home region"
+}
+
+// dbStoreFailure answers a failed bucket read. AccessDenied is the
+// residency refusal as a backstop (spec §6): the role this console runs as
+// cannot read that data, which is the design working, not a fault to
+// retry. Everything else is logged with its detail (bucket names, request
+// ids) and answered with none of it.
+func (s *Server) dbStoreFailure(w http.ResponseWriter, g dbRequest, what string, err error) {
+ log.Printf("console: api db %s/%s/%s: %s: %v", g.acct.Sqid, g.app, g.host, what, err)
+ var api smithy.APIError
+ if errors.As(err, &api) && api.ErrorCode() == "AccessDenied" {
+ jsonError(w, http.StatusConflict, fmt.Sprintf(
+ "this console is not allowed to read the data for %s. It is kept outside the region this console serves.", g.host))
+ return
+ }
+ jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.")
+}
+
+// parseDBLevel accepts a bare decimal level in 0..MaxLevel. Anything else
+// is not a coordinate.
+func parseDBLevel(s string) (int, bool) {
+ n, err := strconv.ParseUint(s, 10, 8)
+ if err != nil || n > dbdoor.MaxLevel {
+ return 0, false
+ }
+ return int(n), true
+}
+
+// handleAPIDBList answers one level's listing. An out-of-range level is a
+// 404 (it is not a coordinate); a level with nothing in it is 200 and an
+// empty list (it is a coordinate that happens to be empty, spec §6.1).
+func (s *Server) handleAPIDBList(w http.ResponseWriter, r *http.Request) {
+ level, ok := parseDBLevel(r.PathValue("level"))
+ if !ok {
+ http.NotFound(w, r)
+ return
+ }
+ g, ok := s.dbGate(w, r)
+ if !ok {
+ 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)
+ return
+ }
+ out := dbdoor.ListResponse{Files: []dbdoor.LTXFile{}}
+ for itr.Next() {
+ f := itr.Item()
+ out.Files = append(out.Files, dbdoor.LTXFile{
+ Level: level, MinTXID: uint64(f.MinTXID), MaxTXID: uint64(f.MaxTXID),
+ Size: f.Size, CreatedAt: f.CreatedAt,
+ })
+ }
+ err = itr.Err()
+ if cerr := itr.Close(); err == nil {
+ err = cerr
+ }
+ if err != nil {
+ s.dbStoreFailure(w, g, fmt.Sprintf("list level %d", level), err)
+ return
+ }
+ sort.Slice(out.Files, func(i, j int) bool { return out.Files[i].MinTXID < out.Files[j].MinTXID })
+ writeJSON(w, http.StatusOK, out)
+}
+
+// handleAPIDBGet streams one LTX file by coordinate. The bytes pass through
+// the console, never a presigned URL (the edge store door's F8 precedent):
+// nothing the request supplies picks the bucket or the key, only a level
+// and a span inside the prefix resolveReplica chose.
+func (s *Server) handleAPIDBGet(w http.ResponseWriter, r *http.Request) {
+ level, ok := parseDBLevel(r.PathValue("level"))
+ if !ok {
+ http.NotFound(w, r)
+ return
+ }
+ minTXID, maxTXID, err := dbdoor.ParseSpan(r.PathValue("span"))
+ if err != nil {
+ http.NotFound(w, r)
+ return
+ }
+ g, ok := s.dbGate(w, r)
+ if !ok {
+ return
+ }
+ rangeHdr := r.Header.Get("Range")
+ offset, size, err := dbdoor.ParseRange(rangeHdr)
+ if err != nil {
+ jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range")
+ return
+ }
+ rc, err := g.rd.OpenLTXFile(r.Context(), level, ltx.TXID(minTXID), ltx.TXID(maxTXID), offset, size)
+ if errors.Is(err, os.ErrNotExist) {
+ http.NotFound(w, r)
+ return
+ }
+ if err != nil {
+ s.dbStoreFailure(w, g, fmt.Sprintf("open level %d %s", level, dbdoor.Span(minTXID, maxTXID)), err)
+ return
+ }
+ defer rc.Close()
+ // One row per file opened from its start; a resumed read of the same
+ // file is the same open continuing, not a second one.
+ if offset == 0 {
+ s.auditActivity(g.acct, g.app, g.actor, Activity{
+ Action: dbAuditAction, TargetType: "instance", Target: g.host,
+ Detail: fmt.Sprintf("read database replica file level %d, transactions %d to %d (cli)", level, minTXID, maxTXID),
+ })
+ }
+ w.Header().Set("Content-Type", "application/octet-stream")
+ w.Header().Set("Cache-Control", "no-store")
+ status := http.StatusOK
+ if rangeHdr != "" {
+ status = http.StatusPartialContent
+ }
+ w.WriteHeader(status)
+ if _, err := io.Copy(w, rc); err != nil {
+ // The status is already on the wire. The client's resumable reader
+ // sees a short body against the listed size and reopens at the
+ // offset it reached.
+ log.Printf("console: api db %s/%s/%s: stream level %d %s: %v",
+ g.acct.Sqid, g.app, g.host, level, dbdoor.Span(minTXID, maxTXID), err)
+ }
+}
+
+// dbReplicaCacheMax bounds the replica-client cache. Each client owns an
+// HTTP transport, so one per request would cost a TLS handshake per file and
+// leave an idle connection behind each; one per prefix lets a restore's
+// dozens of requests share one. Over the bound the map is dropped
+// wholesale, the rateLimiter's rule: forgetting clients costs a handshake,
+// a walk would cost a lock held across every door.
+const dbReplicaCacheMax = 64
+
+type dbReplicaCache struct {
+ mu sync.Mutex
+ clients map[string]replicaReader
+}
+
+// cachedReadReplica is the production dbReplica: lease.ReadReplicaClient on
+// the configured bucket, one per prefix.
+func (s *Server) cachedReadReplica(ctx context.Context, prefix string) (replicaReader, error) {
+ c := &s.dbClients
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ if rd, ok := c.clients[prefix]; ok {
+ return rd, nil
+ }
+ rc, err := lease.ReadReplicaClient(ctx, s.cfg.HibernateBucket, prefix, s.cfg.HibernateRegion)
+ if err != nil {
+ return nil, err
+ }
+ if c.clients == nil || len(c.clients) >= dbReplicaCacheMax {
+ c.clients = make(map[string]replicaReader, 8)
+ }
+ c.clients[prefix] = rc
+ return rc, nil
+}
diff --git a/internal/console/apidb_test.go b/internal/console/apidb_test.go
index cc5fea1ea..12e3d9625 100644
--- a/internal/console/apidb_test.go
+++ b/internal/console/apidb_test.go
@@ -1,10 +1,29 @@
package console
import (
+ "context"
+ "encoding/json"
"errors"
+ "fmt"
+ "io"
+ "net/http"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync/atomic"
"testing"
+ "time"
+ "github.com/aws/smithy-go"
+ lsfile "github.com/benbjohnson/litestream/file"
+ lss3 "github.com/benbjohnson/litestream/s3"
+ "github.com/superfly/ltx"
+
+ "github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/svcrole"
)
// Every one of these is refused BEFORE any S3 call. A door that reached the
@@ -109,3 +128,420 @@ func TestResolveReplicaTreatsEmptyRegionsAsAMatch(t *testing.T) {
t.Fatalf("single-region deployment refused: %v", err)
}
}
+
+// ---- the door handlers (Task 4) ----------------------------------------------
+
+const dbTestHost = "acme.go.tito.io"
+
+type dbFx struct {
+ *consoleFixture
+ member string
+ root string // the fake data bucket: one file replica per prefix
+ opens *atomic.Int32 // how many times the door reached for a replica client
+}
+
+// dbFixture is an in-region, hibernating exec instance of titogo whose
+// replica is a real litestream replica on disk, served through the door's
+// replica seam exactly as the S3 client would serve it.
+func dbFixture(t *testing.T) *dbFx {
+ t.Helper()
+ fx, member := instanceFixture(t, "code-db")
+ root := t.TempDir()
+ fx.srv.cfg.HibernateBucket = "data-test"
+ fx.srv.cfg.HibernatePrefix = "instances"
+ opens := new(atomic.Int32)
+ fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
+ opens.Add(1)
+ return lsfile.NewReplicaClient(filepath.Join(root, prefix)), nil
+ }
+ putDBRecord(t, fx, provision.Record{Host: dbTestHost})
+ return &dbFx{consoleFixture: fx, member: member, root: root, opens: opens}
+}
+
+// putDBRecord writes an instance record for titogo with the fields a
+// hibernating exec instance has, overridden by whatever rec sets.
+func putDBRecord(t *testing.T, fx *consoleFixture, rec provision.Record) {
+ t.Helper()
+ rec.App, rec.Account = "titogo", release.OpsAccount
+ if rec.Hibernate == nil {
+ yes := true
+ rec.Hibernate = &yes
+ }
+ body, err := json.Marshal(rec)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fx.rel.Put(context.Background(), provision.RecordKey(rec.Account, rec.App, rec.Host), body, ""); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (fx *dbFx) replicate(t *testing.T, host string, batches int) string {
+ t.Helper()
+ return dbdoortest.Replicate(t, filepath.Join(fx.root, "instances", host), batches)
+}
+
+func (fx *dbFx) url(p string) string { return fx.ts.URL + p }
+
+// A level the activator never writes is EMPTY, not missing. A 404 here
+// stops CalcRestorePlan at the first gap and the restore fails.
+func TestAPIDBListAnswersEmptyForUnwrittenLevels(t *testing.T) {
+ fx := dbFixture(t)
+ for level := 0; level <= dbdoor.MaxLevel; level++ {
+ status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, level)), fx.member)
+ if status != http.StatusOK {
+ t.Fatalf("level %d = %d %s, want 200 with an empty list", level, status, out)
+ }
+ var res dbdoor.ListResponse
+ if err := json.Unmarshal([]byte(out), &res); err != nil || res.Files == nil || len(res.Files) != 0 {
+ t.Fatalf("level %d = %s, want {\"files\":[]}", level, out)
+ }
+ }
+}
+
+// Wire specifics 2 and 4: ascending by min_txid, and size, min_txid and
+// max_txid on every entry.
+func TestAPIDBListAscendsAndCarriesTheMandatoryFields(t *testing.T) {
+ fx := dbFixture(t)
+ fx.replicate(t, dbTestHost, 3)
+ status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), fx.member)
+ if status != http.StatusOK {
+ t.Fatalf("list = %d %s", status, out)
+ }
+ var res dbdoor.ListResponse
+ if err := json.Unmarshal([]byte(out), &res); err != nil {
+ t.Fatal(err)
+ }
+ if len(res.Files) < 2 {
+ t.Fatalf("list = %s, want several L0 files", out)
+ }
+ for i, f := range res.Files {
+ if f.Size <= 0 || f.MinTXID == 0 || f.MaxTXID < f.MinTXID || f.Level != 0 {
+ t.Errorf("entry %d = %+v, want level 0 with size, min_txid and max_txid", i, f)
+ }
+ if i > 0 && f.MinTXID <= res.Files[i-1].MinTXID {
+ t.Errorf("entries %d and %d are not ascending by min_txid: %s", i-1, i, out)
+ }
+ }
+ for _, field := range []string{`"size"`, `"min_txid"`, `"max_txid"`} {
+ if !strings.Contains(out, field) {
+ t.Errorf("list body lacks %s: %s", field, out)
+ }
+ }
+}
+
+// get streams the file by coordinate, byte-for-byte, and honours the two
+// Range forms litestream's client sends.
+func TestAPIDBGetServesTheFileAndHonoursRange(t *testing.T) {
+ fx := dbFixture(t)
+ fx.replicate(t, dbTestHost, 1)
+ c := lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost))
+ files := listFiles(t, c, 0)
+ f := files[0]
+ want, err := os.ReadFile(c.LTXFilePath(0, f.MinTXID, f.MaxTXID))
+ if err != nil {
+ t.Fatal(err)
+ }
+ u := fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(f.MinTXID), uint64(f.MaxTXID)))
+ status, _, out := rawReq(t, http.MethodGet, u, fx.member, "")
+ if status != http.StatusOK || out != string(want) {
+ t.Fatalf("get = %d, %d bytes; want 200 and the file's %d bytes", status, len(out), len(want))
+ }
+ for _, rc := range []struct{ offset, size int64 }{{10, 0}, {10, 20}} {
+ req, _ := http.NewRequest(http.MethodGet, u, nil)
+ req.Header.Set("Authorization", "Bearer "+fx.member)
+ req.Header.Set("Range", dbdoor.RangeHeader(rc.offset, rc.size))
+ res, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, _ := io.ReadAll(res.Body)
+ res.Body.Close()
+ end := int64(len(want))
+ if rc.size > 0 {
+ end = rc.offset + rc.size
+ }
+ if res.StatusCode != http.StatusPartialContent || string(got) != string(want[rc.offset:end]) {
+ t.Errorf("range %+v = %d, %d bytes; want 206 and bytes %d-%d", rc, res.StatusCode, len(got), rc.offset, end)
+ }
+ }
+}
+
+func listFiles(t *testing.T, c *lsfile.ReplicaClient, level int) []*ltx.FileInfo {
+ t.Helper()
+ itr, err := c.LTXFiles(context.Background(), level, 0, false)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer itr.Close()
+ var out []*ltx.FileInfo
+ for itr.Next() {
+ out = append(out, itr.Item())
+ }
+ if len(out) == 0 {
+ t.Fatal("no files at that level")
+ }
+ return out
+}
+
+// A planned file that has gone is a 404 the client maps to os.ErrNotExist
+// (wire specific 5), which is what makes the whole-Restore retry work.
+func TestAPIDBGetOfAMissingFileIsA404(t *testing.T) {
+ fx := dbFixture(t)
+ fx.replicate(t, dbTestHost, 1)
+ status, _ := getRaw(t, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 900, 901)), fx.member)
+ if status != http.StatusNotFound {
+ t.Fatalf("get of a missing coordinate = %d, want 404", status)
+ }
+}
+
+// Coordinate addressing exists to make lock.json unreachable. This is the
+// test that keeps the door from decaying into an object proxy.
+func TestAPIDBServesNoKeyOutsideItsCoordinates(t *testing.T) {
+ fx := dbFixture(t)
+ dir := filepath.Join(fx.root, "instances", dbTestHost)
+ fx.replicate(t, dbTestHost, 1)
+ if err := os.WriteFile(filepath.Join(dir, "lock.json"), []byte(`{"owner":"box-secret"}`), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ base := "/api/cli/apps/bab/titogo/db/" + dbTestHost
+ for _, bad := range []string{
+ base + "/ltx/0/../../lock.json",
+ base + "/lock.json",
+ base + "/ltx/lock.json",
+ base + "/ltx/0/lock.json",
+ base + "/ltx/10",
+ base + "/ltx/-1",
+ base + "/ltx/0/1",
+ base + "/ltx/0/0-1",
+ base + "/ltx/0/2-1",
+ base + "/ltx/10/1-2",
+ "/api/cli/apps/bab/titogo/db/..%2F..%2Fetc/ltx/0",
+ "/api/cli/apps/bab/titogo/db/" + "b.go.tito.io" + "/ltx/0",
+ } {
+ status, _, out := rawReq(t, http.MethodGet, fx.url(bad), fx.member, "")
+ if status == http.StatusOK || status == http.StatusPartialContent || strings.Contains(out, "box-secret") {
+ t.Errorf("%s = %d %s, want a refusal", bad, status, out)
+ }
+ }
+}
+
+// Not a member, no such app, no such instance: the same bytes http.NotFound
+// writes, matching instanceGate, so the door is no oracle for any of them.
+func TestAPIDBStrangersMeetTheByteIdentical404(t *testing.T) {
+ fx := dbFixture(t)
+ want := canonicalNotFound(t)
+ stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-db-stranger")
+ for name, c := range map[string]struct{ tok, path string }{
+ "not a member": {stranger, dbdoor.ListPath("bab", "titogo", dbTestHost, 0)},
+ "no such app": {fx.member, dbdoor.ListPath("bab", "nosuch", dbTestHost, 0)},
+ "no such instance": {fx.member, dbdoor.ListPath("bab", "titogo", "nobody.go.tito.io", 0)},
+ "no such account": {fx.member, dbdoor.ListPath("zzzz", "titogo", dbTestHost, 0)},
+ "get, not a member": {stranger, dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)},
+ } {
+ status, h, body := rawReq(t, http.MethodGet, fx.url(c.path), c.tok, "")
+ assertNotFound(t, name, want, status, h, body)
+ }
+ if n := fx.opens.Load(); n != 0 {
+ t.Fatalf("the door reached for a replica %d times for callers it refuses", n)
+ }
+}
+
+// Every service principal, every role including admin, and the legacy app
+// token meet the closed-door refusal before any handler runs: the same bytes
+// whether or not the app and the instance exist, so it says nothing about
+// THIS app (Ruling, Task 4: the repo's zero-Set closed door, not a 404).
+func TestAPIDBRefusesEveryServicePrincipalTheSameWay(t *testing.T) {
+ fx := dbFixture(t)
+ fx.replicate(t, dbTestHost, 1)
+ appTok, err := fx.st.NewAppToken(OpsAccountID, "titogo", "alice@keymail.dev", "signup", time.Now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ toks := map[string]string{"legacy app token": appTok}
+ for _, role := range []svcrole.Role{svcrole.Publish, svcrole.Instance, svcrole.Operate, svcrole.Admin} {
+ toks[string(role)] = serviceToken(t, fx.st, OpsAccountID, "svc-"+string(role), "titogo", role)
+ }
+ var first string
+ for name, tok := range toks {
+ for _, p := range []string{
+ dbdoor.ListPath("bab", "titogo", dbTestHost, 0),
+ dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1),
+ dbdoor.ListPath("bab", "nosuch", "nobody.go.tito.io", 0),
+ } {
+ status, _, body := rawReq(t, http.MethodGet, fx.url(p), tok, "")
+ if status != http.StatusForbidden || !strings.Contains(body, "closed to service credentials") {
+ t.Errorf("%s at %s = %d %s, want the closed-door refusal", name, p, status, body)
+ }
+ if first == "" {
+ first = body
+ } else if body != first {
+ t.Errorf("%s at %s: refusal varies (%q vs %q)", name, p, body, first)
+ }
+ }
+ }
+ if n := fx.opens.Load(); n != 0 {
+ t.Fatalf("the door reached for a replica %d times for service principals", n)
+ }
+}
+
+// The residency and pooled refusals are answered with no replica client
+// ever built: a 409 naming the cause, before the bucket.
+func TestAPIDBResidencyRefusalsNeverReachTheBucket(t *testing.T) {
+ fx := dbFixture(t)
+ fx.srv.cfg.Region = "ie"
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: "uk.go.tito.io", Region: "uk"})
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: "fleet.go.tito.io", Region: "fleet/acme"})
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: "pool.go.tito.io", Region: "ie", Pool: "tito"})
+ for host, want := range map[string]string{
+ "uk.go.tito.io": "uk",
+ "fleet.go.tito.io": "fleet",
+ "pool.go.tito.io": "pool",
+ dbTestHost: "",
+ } {
+ status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", host, 0)), fx.member)
+ if status != http.StatusConflict || !strings.Contains(out, want) {
+ t.Errorf("%s = %d %s, want a 409 naming %q", host, status, out, want)
+ }
+ }
+ if n := fx.opens.Load(); n != 0 {
+ t.Fatalf("the door reached for a replica %d times before refusing on residency", n)
+ }
+}
+
+// Only an activator-managed exec instance has a replica the platform
+// writes. Everything else is refused by class, before the bucket.
+func TestAPIDBRefusesRoutesTheActivatorDoesNotReplicate(t *testing.T) {
+ fx := dbFixture(t)
+ no := false
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: "svc.go.tito.io", Kind: "service"})
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: "addr.go.tito.io", Addr: "10.0.0.1:8080"})
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: "awake.go.tito.io", Hibernate: &no})
+ for _, host := range []string{"svc.go.tito.io", "addr.go.tito.io", "awake.go.tito.io"} {
+ status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", host, 0)), fx.member)
+ if status != http.StatusConflict {
+ t.Errorf("%s = %d %s, want a 409 refusal", host, status, out)
+ }
+ }
+ if n := fx.opens.Load(); n != 0 {
+ t.Fatalf("the door reached for a replica %d times for routes with none", n)
+ }
+}
+
+// No data bucket configured is a 503, never a door that fails open.
+func TestAPIDBWithoutADataBucketIs503(t *testing.T) {
+ fx := dbFixture(t)
+ fx.srv.cfg.HibernateBucket = ""
+ status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), fx.member)
+ if status != http.StatusServiceUnavailable {
+ t.Fatalf("no bucket = %d %s, want 503", status, out)
+ }
+ if fx.opens.Load() != 0 {
+ t.Fatal("the door reached for a replica with no bucket configured")
+ }
+}
+
+// S3 refusing the read is the residency refusal as a backstop (spec §6),
+// never a generic 502 — and the store's own error text, which carries the
+// bucket name, never reaches the caller.
+func TestAPIDBAccessDeniedIsTheResidencyRefusal(t *testing.T) {
+ fx := dbFixture(t)
+ fx.srv.dbReplica = func(context.Context, string) (replicaReader, error) {
+ return deniedReplica{}, nil
+ }
+ for _, p := range []string{
+ dbdoor.ListPath("bab", "titogo", dbTestHost, 0),
+ dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1),
+ } {
+ status, out := getRaw(t, fx.url(p), fx.member)
+ if status != http.StatusConflict || strings.Contains(out, "data-test") || strings.Contains(out, "secret-bucket") {
+ t.Errorf("%s = %d %s, want the residency 409 with no bucket name", p, status, out)
+ }
+ }
+}
+
+type deniedReplica struct{}
+
+func (deniedReplica) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) {
+ return nil, &smithy.GenericAPIError{Code: "AccessDenied", Message: "no access to secret-bucket"}
+}
+
+func (deniedReplica) OpenLTXFile(context.Context, int, ltx.TXID, ltx.TXID, int64, int64) (io.ReadCloser, error) {
+ return nil, &smithy.GenericAPIError{Code: "AccessDenied", Message: "no access to secret-bucket"}
+}
+
+// Every successful open of a file writes an audit row naming the instance
+// and the TXID span served (spec §6 "Who may open it").
+func TestAPIDBGetIsAudited(t *testing.T) {
+ fx := dbFixture(t)
+ fx.replicate(t, dbTestHost, 1)
+ f := listFiles(t, lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost)), 0)[0]
+ if status, _ := getRaw(t, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(f.MinTXID), uint64(f.MaxTXID))), fx.member); status != http.StatusOK {
+ t.Fatalf("get = %d", status)
+ }
+ evs, err := fx.st.Events(OpsAccountID, "titogo", 10)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, e := range evs {
+ if e.Action == "db-read" && e.Target == dbTestHost && e.Address == "alice@keymail.dev" &&
+ strings.Contains(e.Detail, fmt.Sprint(uint64(f.MaxTXID))) {
+ return
+ }
+ }
+ t.Fatalf("no db-read audit row naming %s and txid %d: %+v", dbTestHost, f.MaxTXID, evs)
+}
+
+// One replica client per prefix: a restore's dozens of requests share one
+// transport rather than paying a handshake each. And the production reader
+// carries the configured bucket and the resolved prefix, nothing a request
+// supplied.
+func TestCachedReadReplicaIsOnePerPrefix(t *testing.T) {
+ t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
+ s := &Server{}
+ s.cfg.HibernateBucket = "data-ie"
+ s.cfg.HibernateRegion = "eu-west-1"
+ ctx := context.Background()
+ a1, err := s.cachedReadReplica(ctx, "instances/a.example.com")
+ if err != nil {
+ t.Fatal(err)
+ }
+ a2, _ := s.cachedReadReplica(ctx, "instances/a.example.com")
+ b, _ := s.cachedReadReplica(ctx, "instances/b.example.com")
+ if a1 != a2 {
+ t.Error("the same prefix built two clients")
+ }
+ if a1 == b {
+ t.Error("two prefixes shared one client")
+ }
+ c, ok := a1.(*lss3.ReplicaClient)
+ if !ok || c.Bucket != "data-ie" || c.Path != "instances/a.example.com" || c.Region != "eu-west-1" {
+ t.Fatalf("client = %#v, want the configured bucket, region and the resolved prefix", a1)
+ }
+}
+
+// The door has its own token-keyed budget, separate from the instance
+// verbs': spending it does not touch the instance budget, and it applies
+// only after membership, so a stranger cannot learn anything from a 429.
+func TestAPIDBHasItsOwnBudget(t *testing.T) {
+ fx := dbFixture(t)
+ fx.srv.cli.db = newRateLimiter(2, cliRateWindow)
+ u := fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0))
+ for i := 0; i < 2; i++ {
+ if status, out := getRaw(t, u, fx.member); status != http.StatusOK {
+ t.Fatalf("request %d = %d %s", i, status, out)
+ }
+ }
+ if status, _ := getRaw(t, u, fx.member); status != http.StatusTooManyRequests {
+ t.Fatalf("third request = %d, want 429", status)
+ }
+ if !fx.srv.cli.instances.allow(hashToken(fx.member), time.Now()) {
+ t.Fatal("the db door spent the instance verbs' budget")
+ }
+ stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-db-budget")
+ for i := 0; i < 3; i++ {
+ if status, _ := getRaw(t, u, stranger); status != http.StatusNotFound {
+ t.Fatalf("stranger request %d = %d, want the 404 before any budget", i, status)
+ }
+ }
+}
diff --git a/internal/console/authcli.go b/internal/console/authcli.go
index 9fdd2e11c..7d09d555c 100644
--- a/internal/console/authcli.go
+++ b/internal/console/authcli.go
@@ -322,6 +322,10 @@ func retryAfter(w http.ResponseWriter, d time.Duration) {
// Server so a test gets a fresh set per fixture.
type cliLimits struct {
start, poll, pollIP, guessAddr, guessIP, write, ship, instances, create *rateLimiter
+ // db is the `carlos db` door's own budget, keyed by token (cliDBBurst,
+ // apidb.go): a restore is dozens of reads, a different shape from the
+ // instance verbs.
+ db *rateLimiter
// svcCreate, svcWrite and svcShip are the SERVICE side of create/write/
// ship — separate counters, separate keys (principalKey's NAME, not the
// person's ADDRESS or the ship-person's token hash), so a credential's
@@ -361,6 +365,7 @@ func newCLILimits() *cliLimits {
// most of this work is not a person, so the address-keyed write
// budget is the wrong principal for it in both directions.
instances: newRateLimiter(cliInstanceBurst, cliRateWindow),
+ db: newRateLimiter(cliDBBurst, cliRateWindow),
// Create gets its own budget on its own (day-long) window, keyed by
// address — the stolen-token ceiling handleAPICreateApp explains.
create: newRateLimiter(cliCreateBurst, cliCreateWindow),
@@ -627,6 +632,14 @@ func (s *Server) registerCLIRoutes() {
s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move/cut", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceMoveCut)
s.cliRoute("DELETE /api/cli/apps/{account}/{app}/instances/{host}/move", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceMoveAbort)
s.cliRoute("DELETE /api/cli/apps/{account}/{app}/instances/{host}", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceDelete)
+ // The `carlos db` replica door (apidb.go; spec 2026-09-02 §6): an
+ // instance's replicated LTX files, by coordinate, for a member to
+ // restore on their own machine. The zero Set, deliberately: the payload
+ // is the whole customer database, and no legitimate credential needs it
+ // (an app reads its own database live, and CI has no business pulling
+ // production data). admin is refused with the rest.
+ s.cliRoute("GET /api/cli/apps/{account}/{app}/db/{host}/ltx/{level}", svcrole.Set{}, s.handleAPIDBList)
+ s.cliRoute("GET /api/cli/apps/{account}/{app}/db/{host}/ltx/{level}/{span}", svcrole.Set{}, s.handleAPIDBGet)
// The zero Set, deliberately: a credential may never mint a credential,
// including itself. This is the escalation cut (spec §3). It takes no
// authority away — handleAPIInstanceToken gates on apiAccountApp, which
diff --git a/internal/console/server.go b/internal/console/server.go
index 7f91d2001..f3e7e9f9e 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -724,6 +724,11 @@ type Server struct {
instances *instanceFetcher
secure bool // Secure cookies iff the origin is https
cli *cliLimits // rate limits on the CLI-identity endpoints (authcli.go)
+ // dbReplica opens the read side of one instance's replica for the
+ // `carlos db` door (apidb.go). NewServer points it at a small cache of
+ // lease.ReadReplicaClient values; tests point it at a file replica.
+ dbReplica func(ctx context.Context, prefix string) (replicaReader, error)
+ dbClients dbReplicaCache
// cliRoles is pattern -> allowed roles, recorded by cliRoute as it
// registers each /api/cli/ door (authcli.go). Written once, during
// registerCLIRoutes, before the server serves anything, and only read
@@ -1147,6 +1152,8 @@ func NewServer(cfg Config, st *Store) (*Server, error) {
s.mux.HandleFunc("/fleet/economics/tier", s.handleEconomicsTierSave)
// 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.
+ s.dbReplica = s.cachedReadReplica
s.registerCLIRoutes()
// The email API's ten doors (apimail.go, tenant-email task 17) —
// registered separately from registerCLIRoutes because this task's file
diff --git a/internal/console/svcroutes_test.go b/internal/console/svcroutes_test.go
index ced377726..b966f2d64 100644
--- a/internal/console/svcroutes_test.go
+++ b/internal/console/svcroutes_test.go
@@ -168,6 +168,13 @@ var closedDoors = []string{
// whose rules it never met. Shut for the same reason the manifest door
// is — an app may not rewrite the bounds it runs inside.
"POST /api/cli/apps/{account}/{app}/instances/{host}/channel",
+ // The `carlos db` replica door (apidb.go) serves the whole customer
+ // database, file by file. No credential the app holds needs it: the app
+ // reads its own database live, and CI has no business pulling
+ // production data. Shut to every role, admin included (spec 2026-09-02
+ // §6, "Who may open it").
+ "GET /api/cli/apps/{account}/{app}/db/{host}/ltx/{level}",
+ "GET /api/cli/apps/{account}/{app}/db/{host}/ltx/{level}/{span}",
// The build probe is what `carlos deploy`'s watch TRUSTS in place of the
// X-Carlos-Version header — the header being the thing that reports a
// release before any process has restarted. A credential that could
@@ -783,6 +790,8 @@ func TestEveryClosedDoorRefusesAServiceCredential(t *testing.T) {
"{name}", "hub",
"{host}", "web.go.tito.io",
"{id}", "deadbeefdeadbeef",
+ "{level}", "0",
+ "{span}", "1-1",
).Replace(path)
if strings.Contains(u, "{") {
t.Fatalf("closedDoors entry %q has a wildcard this test does not fill", pattern)
@@ -800,8 +809,8 @@ func TestEveryClosedDoorRefusesAServiceCredential(t *testing.T) {
}
seen++
}
- if seen != len(closedDoors) || seen != 34 {
- t.Errorf("walked %d closed doors, want all %d (and 34 today: instance-token, three door-key doors, the routes door, instances-manifest, sidecars-manifest, the route-channel move, six fleets, four services, two email credentials, email pause, the build probe, the live tenancy read, five system doors, two deployment-wide antivirus, and four store-entitlement doors)",
+ if seen != len(closedDoors) || seen != 36 {
+ t.Errorf("walked %d closed doors, want all %d (and 36 today: instance-token, three door-key doors, the routes door, instances-manifest, sidecars-manifest, the route-channel move, two db replica doors, six fleets, four services, two email credentials, email pause, the build probe, the live tenancy read, five system doors, two deployment-wide antivirus, and four store-entitlement doors)",
seen, len(closedDoors))
}
}
diff --git a/internal/dbdoor/dbdoortest/replica.go b/internal/dbdoor/dbdoortest/replica.go
new file mode 100644
index 000000000..c3dce59b6
--- /dev/null
+++ b/internal/dbdoor/dbdoortest/replica.go
@@ -0,0 +1,78 @@
+// Package dbdoortest builds real litestream replicas for the db door's tests,
+// on both sides of the wire (internal/console serves one, internal/dbdoor and
+// the CLI restore one). A fake replica would prove only that the fake agrees
+// with itself; these are written by litestream's own replication code, in
+// the layout its file client reads, so a restore through the door is the
+// restore production runs.
+//
+// Test support only: nothing outside a _test.go file may import it.
+package dbdoortest
+
+import (
+ "context"
+ "database/sql"
+ "fmt"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/benbjohnson/litestream"
+ lsfile "github.com/benbjohnson/litestream/file"
+ _ "modernc.org/sqlite"
+)
+
+// Rows is how many rows each batch inserts into table t.
+const Rows = 10
+
+// Replicate creates a WAL-mode SQLite database with a table t(id, body) and
+// replicates it into dir in `batches` separate syncs, so the replica holds
+// several L0 files rather than one. It returns the source database's path;
+// the source holds batches*Rows rows when it returns.
+func Replicate(t testing.TB, dir string, batches int) string {
+ t.Helper()
+ src := filepath.Join(t.TempDir(), "source.db")
+ db, err := sql.Open("sqlite", src)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ if _, err := db.Exec(`PRAGMA journal_mode=WAL; CREATE TABLE t (id INTEGER PRIMARY KEY, body TEXT)`); err != nil {
+ t.Fatal(err)
+ }
+ for b := 0; b < batches; b++ {
+ for i := 0; i < Rows; i++ {
+ if _, err := db.Exec(`INSERT INTO t (body) VALUES (?)`, fmt.Sprintf("row-%d-%d", b, i)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ Sync(t, src, dir)
+ }
+ return src
+}
+
+// Sync replicates the source's committed state into dir once, the way the
+// lease package's Checkpoint does: open, one explicit sync, close.
+func Sync(t testing.TB, src, dir string) {
+ t.Helper()
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer cancel()
+ ldb := litestream.NewDB(src)
+ ldb.SetMetaPath(filepath.Join(filepath.Dir(src), ".dbdoortest-litestream"))
+ ldb.MonitorInterval = 0
+ ldb.Replica = litestream.NewReplicaWithClient(ldb, lsfile.NewReplicaClient(dir))
+ ldb.Replica.MonitorEnabled = false
+ store := litestream.NewStore([]*litestream.DB{ldb}, litestream.CompactionLevels{{Level: 0}})
+ store.CompactionMonitorEnabled = false
+ store.L0RetentionCheckInterval = 0
+ store.SetRetentionEnabled(false)
+ if err := store.Open(ctx); err != nil {
+ t.Fatal(err)
+ }
+ syncErr := ldb.SyncAndWait(ctx)
+ if err := store.Close(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if syncErr != nil {
+ t.Fatal(syncErr)
+ }
+}
diff --git a/internal/dbdoor/dbdoortest/replica_test.go b/internal/dbdoor/dbdoortest/replica_test.go
new file mode 100644
index 000000000..b48813c9c
--- /dev/null
+++ b/internal/dbdoor/dbdoortest/replica_test.go
@@ -0,0 +1,49 @@
+package dbdoortest
+
+import (
+ "context"
+ "database/sql"
+ "path/filepath"
+ "testing"
+
+ "github.com/benbjohnson/litestream"
+ lsfile "github.com/benbjohnson/litestream/file"
+)
+
+// The helper is only worth having if litestream itself can restore what it
+// wrote, with every batch present, from more than one L0 file.
+func TestReplicateWritesARestorableMultiFileReplica(t *testing.T) {
+ dir := t.TempDir()
+ Replicate(t, dir, 3)
+ c := lsfile.NewReplicaClient(dir)
+ itr, err := c.LTXFiles(context.Background(), 0, 0, false)
+ if err != nil {
+ t.Fatal(err)
+ }
+ n := 0
+ for itr.Next() {
+ n++
+ }
+ itr.Close()
+ if n < 2 {
+ t.Fatalf("replica holds %d L0 files, want several", n)
+ }
+ out := filepath.Join(t.TempDir(), "out.db")
+ opt := litestream.NewRestoreOptions()
+ opt.OutputPath = out
+ if err := litestream.NewReplicaWithClient(nil, c).Restore(context.Background(), opt); err != nil {
+ t.Fatal(err)
+ }
+ db, err := sql.Open("sqlite", "file:"+out+"?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*Rows {
+ t.Fatalf("restored %d rows, want %d", rows, 3*Rows)
+ }
+}