| 1 | // db.go is `carlos db`: a member reads an app's database from its replica, |
| 2 | // with no box access and no AWS credentials (spec 2026-09-02). |
| 3 | // |
| 4 | // It never connects to the running database. The console serves the |
| 5 | // replica's LTX files by coordinate (internal/console/apidb.go); this side |
| 6 | // restores them, with litestream's own restore and a client that has no |
| 7 | // write methods (internal/dbdoor), into a private 0700 directory, proves the |
| 8 | // result with lease.CheckDB, and then either copies it to --out (pull) or |
| 9 | // opens it read-only and runs one query (query). The console never executes |
| 10 | // tenant SQL and never holds a restored database. |
| 11 | // |
| 12 | // Read-only is enforced, not assumed (spec §4): mode=ro and query_only. A |
| 13 | // copy is harmless to write to; the point is that nobody, person or agent, |
| 14 | // can come away believing a write reached production. |
| 15 | // |
| 16 | // Unit 1 of three (spec §13): ls and its freshness fields, then shell, |
| 17 | // follow. |
| 18 | package main |
| 19 | |
| 20 | import ( |
| 21 | "context" |
| 22 | "errors" |
| 23 | "flag" |
| 24 | "fmt" |
| 25 | "io" |
| 26 | "log" |
| 27 | "os" |
| 28 | "os/signal" |
| 29 | "slices" |
| 30 | "sort" |
| 31 | "strings" |
| 32 | "syscall" |
| 33 | "time" |
| 34 | |
| 35 | "github.com/carlosframework/platform/internal/dbdoor" |
| 36 | "github.com/carlosframework/platform/internal/dbquery" |
| 37 | "github.com/carlosframework/platform/internal/dbreads" |
| 38 | ) |
| 39 | |
| 40 | const dbUsage = `usage: carlos db <verb> --app <a> [--instance <host>] [flags] |
| 41 | |
| 42 | pull --out <path> restore a copy of the instance's database and save it |
| 43 | at <path>, which must not exist yet (saved 0600) |
| 44 | query --sql <text> restore a copy and run one read-only query against it; |
| 45 | rows go to stdout, tab-separated, with a header row |
| 46 | |
| 47 | Add --at <time> to either verb to read the database as it was at that |
| 48 | time, for example --at 2026-10-07T14:00:00Z. The copy holds every change |
| 49 | replicated before then, so it can be a little older than the time you give. |
| 50 | How far back you can go depends on how much history the platform keeps. |
| 51 | |
| 52 | Both read the platform's replica of the database, never the live one, and |
| 53 | nothing you do to the copy reaches the instance. --instance names which of |
| 54 | the app's instances to read; it may be left out when the app has one. The |
| 55 | usual --account / --account-id / --console / --destination apply. |
| 56 | ` |
| 57 | |
| 58 | func runDB(args []string) { |
| 59 | if len(args) == 0 { |
| 60 | fmt.Fprint(os.Stderr, dbUsage) |
| 61 | os.Exit(2) |
| 62 | } |
| 63 | switch args[0] { |
| 64 | case "pull": |
| 65 | runDBPull(args[1:]) |
| 66 | case "query": |
| 67 | runDBQuery(args[1:]) |
| 68 | case "-h", "--help", "help": |
| 69 | fmt.Fprint(stdout, dbUsage) |
| 70 | default: |
| 71 | fmt.Fprintf(os.Stderr, "carlos db: unknown verb %q\n%s", args[0], dbUsage) |
| 72 | os.Exit(2) |
| 73 | } |
| 74 | } |
| 75 | |
| 76 | // dbTarget is what both verbs resolve first: the session, the account, the |
| 77 | // app and the one instance the verb reads. |
| 78 | type dbTarget struct { |
| 79 | sess *apiSession |
| 80 | acct apiAcct |
| 81 | app string |
| 82 | host string |
| 83 | // at is --at: the copy holds what was replicated before it. Zero is |
| 84 | // the newest copy. |
| 85 | at time.Time |
| 86 | } |
| 87 | |
| 88 | // dbFlags registers the flags both verbs share and returns the resolver. |
| 89 | // Status lines go to stderr: query's stdout is its rows, and must pipe |
| 90 | // cleanly into whatever reads them. |
| 91 | func dbFlags(fs *flag.FlagSet, verb string) func() dbTarget { |
| 92 | app := appFlag(fs, "app whose database to read") |
| 93 | instance := fs.String("instance", "", "the instance whose database to read (may be left out when the app has one)") |
| 94 | acctRaw, acctIDRaw, _ := accountFlags(fs) |
| 95 | consoleFlag := consoleOption(fs, "console to read through (default: the only logged-in one)") |
| 96 | atRaw := fs.String("at", "", "read the database as it was at this time (RFC 3339, for example 2026-10-07T14:00:00Z)") |
| 97 | return func() dbTarget { |
| 98 | // Before any network: a time that can't be read is the person's |
| 99 | // to fix, and costs nothing to refuse. |
| 100 | at, err := parseDBAt(*atRaw, time.Now()) |
| 101 | if err != nil { |
| 102 | log.Fatalf("carlos db %s: %v", verb, err) |
| 103 | } |
| 104 | sess := requireDBAPISession(*consoleFlag) |
| 105 | printUsing(os.Stderr, sess) |
| 106 | acct := resolvedAccount(sess, acctRaw, acctIDRaw) |
| 107 | name := resolvedApp(sess, acct, strings.TrimSpace(*app), dbUsage) |
| 108 | host, err := resolveDBInstance(context.Background(), sess, acct, name, strings.TrimSpace(*instance)) |
| 109 | if err != nil { |
| 110 | log.Fatalf("carlos db %s: %v", verb, err) |
| 111 | } |
| 112 | return dbTarget{sess: sess, acct: acct, app: name, host: host, at: at} |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | // parseDBAt reads --at. Empty is the zero time, the newest copy. A time |
| 117 | // must carry its zone: a bare clock time would be read in whichever zone |
| 118 | // this shell is in, and the copy would silently be hours off. A time after |
| 119 | // now is refused rather than quietly answered with the newest copy, so |
| 120 | // nobody holds a copy labelled with a moment that hasn't happened. |
| 121 | func parseDBAt(raw string, now time.Time) (time.Time, error) { |
| 122 | raw = strings.TrimSpace(raw) |
| 123 | if raw == "" { |
| 124 | return time.Time{}, nil |
| 125 | } |
| 126 | at, err := time.Parse(time.RFC3339, raw) |
| 127 | if err != nil { |
| 128 | return time.Time{}, fmt.Errorf("--at %q is not a time. Write it in RFC 3339 with a time zone, for example 2026-10-07T14:00:00Z for UTC", raw) |
| 129 | } |
| 130 | if at.After(now) { |
| 131 | return time.Time{}, fmt.Errorf("--at %s is in the future. Leave --at out to read the newest copy", raw) |
| 132 | } |
| 133 | // 0001-01-01T00:00:00Z parses to Go's zero time, which means "no |
| 134 | // --at". It is before any history, so it is asked as the earliest |
| 135 | // time instead, and gets the before-history answer. |
| 136 | if at.Before(dbdoor.EarliestAsOf) { |
| 137 | at = dbdoor.EarliestAsOf |
| 138 | } |
| 139 | return at, nil |
| 140 | } |
| 141 | |
| 142 | // requireDBAPISession resolves signed-in mode, refusing direct bucket mode |
| 143 | // by name: the door is the console's, and an operator holding the bucket |
| 144 | // already has a box. |
| 145 | func requireDBAPISession(consoleFlag string) *apiSession { |
| 146 | sess, err := resolveMode(consoleFlag) |
| 147 | if err != nil { |
| 148 | log.Fatal(err) |
| 149 | } |
| 150 | if sess == nil { |
| 151 | log.Fatal("carlos db works through a console, and this shell is in direct bucket mode " + |
| 152 | "(CARLOS_DEPLOYMENT_BUCKET or CARLOS_DEPLOYMENT_DIR is set). Unset it and run: carlos auth login --console <url>.") |
| 153 | } |
| 154 | return sess |
| 155 | } |
| 156 | |
| 157 | // resolveDBInstance is --instance, or the app's only instance when it has |
| 158 | // exactly one. Several is a refusal naming them: guessing would hand |
| 159 | // someone the wrong data while telling them it is the right data (spec §7). |
| 160 | func resolveDBInstance(ctx context.Context, sess *apiSession, acct apiAcct, app, host string) (string, error) { |
| 161 | if host != "" { |
| 162 | return host, nil |
| 163 | } |
| 164 | recs, truncated, _, err := sess.instanceList(ctx, acct, app, "", "") |
| 165 | if err != nil { |
| 166 | return "", fmt.Errorf("could not list the app's instances to pick one: %w", err) |
| 167 | } |
| 168 | if len(recs) == 1 && !truncated { |
| 169 | return recs[0].Host, nil |
| 170 | } |
| 171 | if len(recs) == 0 { |
| 172 | return "", fmt.Errorf("%s has no instances, so there is no database to read", app) |
| 173 | } |
| 174 | hosts := make([]string, 0, len(recs)) |
| 175 | for _, r := range recs { |
| 176 | hosts = append(hosts, r.Host) |
| 177 | } |
| 178 | sort.Strings(hosts) |
| 179 | more := "" |
| 180 | if truncated { |
| 181 | more = ", and more" |
| 182 | } |
| 183 | return "", fmt.Errorf("%s has %d instances. Name one with --instance: %s%s", app, len(recs), strings.Join(hosts, ", "), more) |
| 184 | } |
| 185 | |
| 186 | // dbRestore restores the target's replica into a private directory under |
| 187 | // the system temp dir. The caller removes Restored.Dir. |
| 188 | // |
| 189 | // A door refusal carrying the db region forward marker comes back as an |
| 190 | // *errDBForward, so a verb can switch to the forwarded read (dbforward.go). |
| 191 | func dbRestore(ctx context.Context, t dbTarget) (*dbdoor.Restored, error) { |
| 192 | r, err := dbdoor.RestoreAsOf(ctx, t.sess.dbClient(t.acct, t.app, t.host), "", t.at) |
| 193 | if err != nil { |
| 194 | var door *dbdoor.DoorError |
| 195 | if errors.As(err, &door) && door.Forward != nil { |
| 196 | return nil, &errDBForward{door: door} |
| 197 | } |
| 198 | return nil, dbRestoreError(t.host, err) |
| 199 | } |
| 200 | return r, nil |
| 201 | } |
| 202 | |
| 203 | // dbRestoreError turns a failed restore into the sentence a member needs. |
| 204 | // A refusal from the door carries the console's own words; the others are |
| 205 | // named here so none of them reads as a transient "no replica found". |
| 206 | func dbRestoreError(host string, err error) error { |
| 207 | var door *dbdoor.DoorError |
| 208 | var before *dbdoor.BeforeHistoryError |
| 209 | switch { |
| 210 | case errors.As(err, &before): |
| 211 | return fmt.Errorf("%s's replica does not go back that far. The oldest point it keeps is %s", host, dbTime(before.Oldest)) |
| 212 | case errors.Is(err, dbdoor.ErrNoTimes): |
| 213 | return fmt.Errorf("%s's replica can't be read as of a time yet. Leave --at out to read the newest copy", host) |
| 214 | case errors.As(err, &door) && door.Status == 404: |
| 215 | return fmt.Errorf("no instance %s that you can read", host) |
| 216 | case errors.As(err, &door): |
| 217 | return errors.New(door.Error()) |
| 218 | case errors.Is(err, dbdoor.ErrNothingReplicated): |
| 219 | return fmt.Errorf("the platform has no copy of %s's database yet. Nothing has been replicated. "+ |
| 220 | "If the app replicates its own database (replicate: \"app\"), the platform never will", host) |
| 221 | case errors.Is(err, dbdoor.ErrCheckFailed): |
| 222 | return fmt.Errorf("the copy of %s's database failed its integrity check, so it was not kept: %v", host, err) |
| 223 | case errors.Is(err, os.ErrNotExist): |
| 224 | return fmt.Errorf("the replica of %s changed while it was being read, twice. Try again", host) |
| 225 | case errors.Is(err, context.Canceled): |
| 226 | return errors.New("stopped") |
| 227 | } |
| 228 | return fmt.Errorf("could not restore %s's database: %w", host, err) |
| 229 | } |
| 230 | |
| 231 | // dbAsOf renders the copy's newest transaction and its time (R1). |
| 232 | func dbAsOf(r *dbdoor.Restored) string { |
| 233 | if r.At.IsZero() { |
| 234 | return fmt.Sprintf("transaction %d", r.TXID) |
| 235 | } |
| 236 | return fmt.Sprintf("transaction %d, written %s", r.TXID, dbTime(r.At)) |
| 237 | } |
| 238 | |
| 239 | func dbTime(t time.Time) string { return t.UTC().Format("2006-01-02 15:04:05 UTC") } |
| 240 | |
| 241 | // dbPullHolds is pull's line saying what the copy holds; with --at it also |
| 242 | // names the requested time, since the copy can be older than it. |
| 243 | func dbPullHolds(r *dbdoor.Restored, at time.Time) string { |
| 244 | if at.IsZero() { |
| 245 | return fmt.Sprintf("It holds every change up to %s.\n", dbAsOf(r)) |
| 246 | } |
| 247 | return fmt.Sprintf("It holds every change up to %s, the last one replicated before %s.\n", dbAsOf(r), dbTime(at)) |
| 248 | } |
| 249 | |
| 250 | // dbQueryStatus is query's stderr line before the rows. |
| 251 | func dbQueryStatus(host string, r *dbdoor.Restored, at time.Time) string { |
| 252 | if at.IsZero() { |
| 253 | return fmt.Sprintf("Querying a copy of %s's database, up to %s. Nothing you run here reaches %s.\n", host, dbAsOf(r), host) |
| 254 | } |
| 255 | return fmt.Sprintf("Querying a copy of %s's database as it was before %s, up to %s. Nothing you run here reaches %s.\n", host, dbTime(at), dbAsOf(r), host) |
| 256 | } |
| 257 | |
| 258 | func runDBPull(args []string) { |
| 259 | fs := flag.NewFlagSet("db pull", flag.ExitOnError) |
| 260 | target := dbFlags(fs, "pull") |
| 261 | out := fs.String("out", "", "where to save the copy; must not exist yet") |
| 262 | fs.Usage = flagUsage(fs, dbUsage) |
| 263 | fs.Parse(args) |
| 264 | if strings.TrimSpace(*out) == "" { |
| 265 | fmt.Fprint(os.Stderr, "carlos db pull: --out is required: name a path for the copy\n\n"+dbUsage) |
| 266 | os.Exit(2) |
| 267 | } |
| 268 | if _, err := os.Lstat(*out); err == nil { |
| 269 | fmt.Fprintf(os.Stderr, "carlos db pull: %s already exists. Choose a new path for --out\n", *out) |
| 270 | os.Exit(2) |
| 271 | } |
| 272 | t := target() |
| 273 | ctx, stop := dbSignalContext() |
| 274 | defer stop() |
| 275 | if err := dbPull(ctx, t, *out); err != nil { |
| 276 | stop() |
| 277 | log.Fatalf("carlos db pull: %v", err) |
| 278 | } |
| 279 | } |
| 280 | |
| 281 | // dbPull restores, copies to out, and removes the private directory on |
| 282 | // every path. It returns rather than exits so that cleanup always runs. |
| 283 | func dbPull(ctx context.Context, t dbTarget, out string) error { |
| 284 | r, err := dbRestore(ctx, t) |
| 285 | var fwd *errDBForward |
| 286 | if errors.As(err, &fwd) { |
| 287 | // The copy's files come from the region that holds them (db region |
| 288 | // forward spec §6), when the console forwards pulls there. |
| 289 | if !slices.Contains(fwd.door.Forward.Ops, dbreads.OpPull) { |
| 290 | // A console that forwards queries and not pulls (plan 2's): DBF-8. |
| 291 | return fmt.Errorf(dbfNoPull, t.host, fwd.door.Forward.Region) |
| 292 | } |
| 293 | r, err = dbPullForward(ctx, t, *fwd.door.Forward) |
| 294 | } |
| 295 | if err != nil { |
| 296 | return err |
| 297 | } |
| 298 | defer os.RemoveAll(r.Dir) |
| 299 | if err := copyOut(ctx, r.Path, out); err != nil { |
| 300 | return err |
| 301 | } |
| 302 | fmt.Fprintf(stdout, "Saved a copy of %s's database to %s.\n", t.host, out) |
| 303 | fmt.Fprint(stdout, dbPullHolds(r, t.at)) |
| 304 | fmt.Fprintf(stdout, "It is a copy. Nothing you do to it reaches %s.\n", t.host) |
| 305 | return nil |
| 306 | } |
| 307 | |
| 308 | // copyOut copies the restored database to out, 0600, refusing an existing |
| 309 | // out. A copy and never os.Rename: the temp directory and the user's |
| 310 | // destination are routinely on different filesystems, where a rename |
| 311 | // fails. A failed copy leaves nothing behind at out. |
| 312 | func copyOut(ctx context.Context, src, out string) (err error) { |
| 313 | // Nothing is published once the person has asked to stop. |
| 314 | if err := ctx.Err(); err != nil { |
| 315 | return err |
| 316 | } |
| 317 | in, err := os.Open(src) |
| 318 | if err != nil { |
| 319 | return err |
| 320 | } |
| 321 | defer in.Close() |
| 322 | f, err := os.OpenFile(out, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) |
| 323 | if err != nil { |
| 324 | if errors.Is(err, os.ErrExist) { |
| 325 | return fmt.Errorf("%s already exists. Choose a new path for --out", out) |
| 326 | } |
| 327 | return err |
| 328 | } |
| 329 | defer func() { |
| 330 | if err != nil { |
| 331 | f.Close() |
| 332 | os.Remove(out) |
| 333 | } |
| 334 | }() |
| 335 | // The mode passed to OpenFile is filtered by the umask; 0600 is what |
| 336 | // the spec promises, so it is set, not hoped for. |
| 337 | if err = f.Chmod(0o600); err != nil { |
| 338 | return err |
| 339 | } |
| 340 | if _, err = io.Copy(f, ctxReader{ctx, in}); err != nil { |
| 341 | return err |
| 342 | } |
| 343 | if err = f.Sync(); err != nil { |
| 344 | return err |
| 345 | } |
| 346 | if err = f.Close(); err != nil { |
| 347 | return err |
| 348 | } |
| 349 | // A stop that lands as the copy finishes still withdraws it: the |
| 350 | // caller reports success only for a copy that was not interrupted. |
| 351 | if err = ctx.Err(); err != nil { |
| 352 | os.Remove(out) |
| 353 | return err |
| 354 | } |
| 355 | return nil |
| 356 | } |
| 357 | |
| 358 | // ctxReader stops a copy at the next read once ctx is done. |
| 359 | type ctxReader struct { |
| 360 | ctx context.Context |
| 361 | r io.Reader |
| 362 | } |
| 363 | |
| 364 | func (c ctxReader) Read(p []byte) (int, error) { |
| 365 | if err := c.ctx.Err(); err != nil { |
| 366 | return 0, err |
| 367 | } |
| 368 | return c.r.Read(p) |
| 369 | } |
| 370 | |
| 371 | func runDBQuery(args []string) { |
| 372 | fs := flag.NewFlagSet("db query", flag.ExitOnError) |
| 373 | target := dbFlags(fs, "query") |
| 374 | sqlText := fs.String("sql", "", "the query to run") |
| 375 | fs.Usage = flagUsage(fs, dbUsage) |
| 376 | fs.Parse(args) |
| 377 | if strings.TrimSpace(*sqlText) == "" { |
| 378 | fmt.Fprint(os.Stderr, "carlos db query: --sql is required\n\n"+dbUsage) |
| 379 | os.Exit(2) |
| 380 | } |
| 381 | t := target() |
| 382 | ctx, stop := dbSignalContext() |
| 383 | defer stop() |
| 384 | if err := dbQuery(ctx, t, *sqlText); err != nil { |
| 385 | stop() |
| 386 | log.Fatalf("carlos db query: %v", err) |
| 387 | } |
| 388 | } |
| 389 | |
| 390 | func dbQuery(ctx context.Context, t dbTarget, sqlText string) error { |
| 391 | r, err := dbRestore(ctx, t) |
| 392 | var fwd *errDBForward |
| 393 | if errors.As(err, &fwd) { |
| 394 | // The instance is read in its own region (db region forward spec |
| 395 | // §5.5), when the console forwards queries there. |
| 396 | if !slices.Contains(fwd.door.Forward.Ops, dbreads.OpQuery) { |
| 397 | return errors.New(fwd.door.Error()) |
| 398 | } |
| 399 | return dbQueryForward(ctx, t, sqlText, *fwd.door.Forward) |
| 400 | } |
| 401 | if err != nil { |
| 402 | return err |
| 403 | } |
| 404 | defer os.RemoveAll(r.Dir) |
| 405 | fmt.Fprint(os.Stderr, dbQueryStatus(t.host, r, t.at)) |
| 406 | return runQueryContext(ctx, stdout, r.Path, sqlText) |
| 407 | } |
| 408 | |
| 409 | // The query runner and its guards live in internal/dbquery (spec 2026-10-22 |
| 410 | // §8.3), so the CLI and the db-read worker run one copy of them. These |
| 411 | // names stay for db.go's callers and tests. |
| 412 | var ( |
| 413 | errOneStatement = dbquery.ErrOneStatement |
| 414 | errPragmaChange = dbquery.ErrPragmaChange |
| 415 | ) |
| 416 | |
| 417 | func checkQuerySQL(q string) error { return dbquery.CheckSQL(q) } |
| 418 | |
| 419 | func dbQueryCell(v any) string { return dbquery.Cell(v) } |
| 420 | |
| 421 | func runQuery(w io.Writer, path, sqlText string) error { |
| 422 | return runQueryContext(context.Background(), w, path, sqlText) |
| 423 | } |
| 424 | |
| 425 | // runQueryContext reads nothing itself: the open and the query are |
| 426 | // dbquery.RunContext's. |
| 427 | func runQueryContext(ctx context.Context, w io.Writer, path, sqlText string) error { |
| 428 | return dbquery.RunContext(ctx, w, path, sqlText, dbquery.Options{}) |
| 429 | } |
| 430 | |
| 431 | // dbStopSignals end a db verb early. Each one cancels the context instead of |
| 432 | // killing the process, so the deferred removal of the private directory |
| 433 | // always runs: Ctrl-C, a closed terminal (SIGHUP), a supervisor's SIGTERM, |
| 434 | // and SIGPIPE. The last is the one that matters most: `carlos db query ... |
| 435 | // | head` closes stdout early, and with SIGPIPE left to the runtime Go |
| 436 | // kills the process on the next write, which would leave a plaintext copy |
| 437 | // of production in the temp directory. Caught, the write returns EPIPE |
| 438 | // instead and the query loop stops on it. |
| 439 | var dbStopSignals = []os.Signal{os.Interrupt, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGPIPE} |
| 440 | |
| 441 | func dbSignalContext() (context.Context, context.CancelFunc) { |
| 442 | return signal.NotifyContext(context.Background(), dbStopSignals...) |
| 443 | } |
| 444 | |
amadan