| 1 | // dbforward.go is the CLI's half of a forwarded `carlos db query` and |
| 2 | // `carlos db pull` (spec |
| 3 | // docs/superpowers/specs/2026-10-22-db-region-forward-design.md §5.5, §6): |
| 4 | // when the db door answers with the forward marker, the CLI makes an |
| 5 | // ephemeral key, holds one request to the console's reads route open, keeps |
| 6 | // every sealed chunk (in memory for a query, spooled to a private directory |
| 7 | // for a pull), and prints or saves nothing until a done result's summary |
| 8 | // checks out against the bytes (codex 6). |
| 9 | package main |
| 10 | |
| 11 | import ( |
| 12 | "bufio" |
| 13 | "bytes" |
| 14 | "context" |
| 15 | "crypto/ecdh" |
| 16 | "crypto/sha256" |
| 17 | "encoding/hex" |
| 18 | "encoding/json" |
| 19 | "errors" |
| 20 | "fmt" |
| 21 | "io" |
| 22 | "net/http" |
| 23 | "net/url" |
| 24 | "os" |
| 25 | "path/filepath" |
| 26 | "strconv" |
| 27 | "strings" |
| 28 | "time" |
| 29 | "unicode/utf8" |
| 30 | |
| 31 | "github.com/benbjohnson/litestream" |
| 32 | lsfile "github.com/benbjohnson/litestream/file" |
| 33 | "github.com/superfly/ltx" |
| 34 | |
| 35 | "github.com/carlosframework/platform/internal/dbdoor" |
| 36 | "github.com/carlosframework/platform/internal/dbreads" |
| 37 | ) |
| 38 | |
| 39 | // The forwarded read's CLI sentences (copy record DBF-3 to DBF-8, DBF-11 to |
| 40 | // DBF-13, DBF-15, DBF-16, DBF-32), approved by Luna on 2026-10-23. |
| 41 | const ( |
| 42 | dbfNotPicked = "no server in region %s started the read within 60 seconds. try again in a few minutes." // DBF-3 |
| 43 | dbfTooLarge = "the database for %s is %s once restored. carlos db can query at most %s in region %s." // DBF-4 |
| 44 | dbfNoRoom = "the server in region %s does not have room to restore %s's database right now. try again later." // DBF-5 |
| 45 | dbfOutputCap = "the result was cut off at 64 MB. narrow the query with WHERE or LIMIT and run it again." // DBF-6 |
| 46 | dbfTimeout = "the query ran for 5 minutes and was stopped. narrow it and run it again." // DBF-7 |
| 47 | dbfNoPull = "%s keeps its data in region %s. carlos db pull cannot fetch a copy from there yet. carlos db query can read it." // DBF-8 |
| 48 | dbfReading = "reading a copy of %s's database in region %s." // DBF-11 |
| 49 | dbfUnfinished = "the read of %s's database in region %s did not finish, so nothing was printed. try again." // DBF-12 |
| 50 | dbfUnchecked = "the answer from region %s was invalid, so carlos db printed nothing. try again." // DBF-13 |
| 51 | dbfChanged = "%s changed while the read was waiting. try again." // DBF-15 |
| 52 | dbfMovingCLI = "%s is moving between regions. read its database once the move has finished." // DBF-16 |
| 53 | dbfNotUTF8 = "this query cannot be sent to region %s, because it contains invalid UTF-8 or the replacement character U+FFFD. fix the query and run it again." // DBF-32 |
| 54 | ) |
| 55 | |
| 56 | // The forwarded pull's CLI sentences (copy record DBF-17, DBF-18, DBF-24), |
| 57 | // approved by Luna on 2026-10-23. |
| 58 | const ( |
| 59 | dbfPullTooLarge = "the database for %s has %s of files to fetch. carlos db pull can fetch at most %s from region %s." // DBF-17 |
| 60 | dbfFetching = "fetching a copy of %s's database from region %s." // DBF-18 |
| 61 | dbfPullNoRoom = "the server in region %s does not have room to prepare a copy of %s's database right now. try again later." // DBF-24 |
| 62 | ) |
| 63 | |
| 64 | // errDBForward is the db door's forward marker, handed back by dbRestore so |
| 65 | // a verb can switch to the forwarded read. |
| 66 | type errDBForward struct { |
| 67 | door *dbdoor.DoorError |
| 68 | } |
| 69 | |
| 70 | func (e *errDBForward) Error() string { return e.door.Error() } |
| 71 | |
| 72 | type dbForwardAnswer struct { |
| 73 | rows []byte |
| 74 | summary dbreads.Summary |
| 75 | capped bool |
| 76 | } |
| 77 | |
| 78 | // dbQueryForward is spec §5.5 steps 2 and 8 for a query. |
| 79 | func dbQueryForward(ctx context.Context, t dbTarget, sqlText string, fwd dbdoor.Forward) error { |
| 80 | // json.Marshal would send U+FFFD for bytes that are not UTF-8, and the |
| 81 | // box would run SQL nobody typed; the console refuses U+FFFD for the |
| 82 | // same reason (plan review finding 36). First, so no other check can |
| 83 | // read the bytes differently. |
| 84 | if !utf8.ValidString(sqlText) || strings.ContainsRune(sqlText, utf8.RuneError) { |
| 85 | return fmt.Errorf(dbfNotUTF8, fwd.Region) |
| 86 | } |
| 87 | if err := checkQuerySQL(sqlText); err != nil { |
| 88 | return err |
| 89 | } |
| 90 | ctx, cancel := context.WithTimeout(ctx, dbreads.CLIBound) |
| 91 | defer cancel() |
| 92 | key, err := dbreads.NewReplyKey() |
| 93 | if err != nil { |
| 94 | return err |
| 95 | } |
| 96 | at := "" |
| 97 | if !t.at.IsZero() { |
| 98 | at = t.at.UTC().Format(time.RFC3339Nano) |
| 99 | } |
| 100 | body, err := json.Marshal(dbreads.ReadBody{Op: dbreads.OpQuery, SQL: sqlText, At: at, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())}) |
| 101 | if err != nil { |
| 102 | return err |
| 103 | } |
| 104 | res, err := t.sess.c.doStream(ctx, http.MethodPost, apiPath(t.acct, t.app, "/db/"+url.PathEscape(t.host)+"/reads"), |
| 105 | bytes.NewReader(body), int64(len(body)), http.Header{"Content-Type": {"application/json"}}) |
| 106 | if err != nil { |
| 107 | if errors.Is(ctx.Err(), context.Canceled) { |
| 108 | return errors.New("stopped") |
| 109 | } |
| 110 | return fmt.Errorf(dbfUnfinished, t.host, fwd.Region) |
| 111 | } |
| 112 | defer res.Body.Close() |
| 113 | if res.StatusCode != http.StatusOK { |
| 114 | raw, _ := io.ReadAll(io.LimitReader(res.Body, 64<<10)) |
| 115 | return apiErrorFromBody(res.StatusCode, raw) |
| 116 | } |
| 117 | sink := &memSink{} |
| 118 | id, result, err := readDBForward(ctx, res.Body, key, t.host, fwd.Region, dbfReading, os.Stderr, sink, newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes)) |
| 119 | if err != nil { |
| 120 | return err |
| 121 | } |
| 122 | ans, err := openDBForward(id, key, sink, *result, t.host, fwd.Region) |
| 123 | if err != nil { |
| 124 | return err |
| 125 | } |
| 126 | fmt.Fprint(os.Stderr, dbQueryStatus(t.host, &dbdoor.Restored{TXID: ans.summary.TXID, At: ans.summary.At}, t.at)) |
| 127 | if _, err := stdout.Write(ans.rows); err != nil { |
| 128 | return err |
| 129 | } |
| 130 | if ans.capped { |
| 131 | return errors.New(dbfOutputCap) |
| 132 | } |
| 133 | return nil |
| 134 | } |
| 135 | |
| 136 | // dbfChunkBound is what a reader keeps before a result arrives (plan review |
| 137 | // finding 37). Every chunk is at least a sealed empty one; only the last |
| 138 | // may be empty, so nothing may follow an empty chunk; there are at most as |
| 139 | // many chunks as a read can make (full ones for the payload, plus one per |
| 140 | // flush interval over the owner's deadline, plus the last); and their |
| 141 | // plaintext, ciphertext less each seal, stays within the payload cap. |
| 142 | type dbfChunkBound struct { |
| 143 | maxBytes, maxChunks int64 |
| 144 | n, plain int64 |
| 145 | sawEmpty bool |
| 146 | } |
| 147 | |
| 148 | func newDBFChunkBound(maxBytes, chunkBytes int64) *dbfChunkBound { |
| 149 | return &dbfChunkBound{maxBytes: maxBytes, |
| 150 | maxChunks: (maxBytes+chunkBytes-1)/chunkBytes + int64(dbreads.OwnerDeadline/dbreads.ChunkEvery) + 1} |
| 151 | } |
| 152 | |
| 153 | // admit says whether a chunk may be kept; false is DBF-13. |
| 154 | func (b *dbfChunkBound) admit(sealed []byte) bool { |
| 155 | size := int64(len(sealed)) - int64(dbreads.SealOverhead) |
| 156 | if size < 0 || b.sawEmpty || b.n >= b.maxChunks || b.plain+size > b.maxBytes { |
| 157 | return false |
| 158 | } |
| 159 | b.n++ |
| 160 | b.plain += size |
| 161 | b.sawEmpty = size == 0 |
| 162 | return true |
| 163 | } |
| 164 | |
| 165 | // dbChunkSink holds a read's sealed chunks until its result arrives: in |
| 166 | // memory for a query (at most 64 MiB), in the pull's private directory for a |
| 167 | // pull (up to its cap). |
| 168 | type dbChunkSink interface { |
| 169 | add(sealed []byte) error |
| 170 | count() int |
| 171 | chunk(i int) ([]byte, error) // 0-based |
| 172 | } |
| 173 | |
| 174 | type memSink struct{ chunks [][]byte } |
| 175 | |
| 176 | func (m *memSink) add(b []byte) error { m.chunks = append(m.chunks, b); return nil } |
| 177 | func (m *memSink) count() int { return len(m.chunks) } |
| 178 | func (m *memSink) chunk(i int) ([]byte, error) { return m.chunks[i], nil } |
| 179 | |
| 180 | // fileSink spools sealed chunks to dir, one file each, 0600 in a 0700 |
| 181 | // directory the caller removes. |
| 182 | type fileSink struct { |
| 183 | dir string |
| 184 | n int |
| 185 | } |
| 186 | |
| 187 | func (f *fileSink) add(b []byte) error { |
| 188 | if f.n == 0 { |
| 189 | if err := os.MkdirAll(f.dir, 0o700); err != nil { |
| 190 | return err |
| 191 | } |
| 192 | } |
| 193 | f.n++ |
| 194 | return dbPullSpoolWrite(filepath.Join(f.dir, strconv.Itoa(f.n)), b, 0o600) |
| 195 | } |
| 196 | |
| 197 | func (f *fileSink) count() int { return f.n } |
| 198 | |
| 199 | func (f *fileSink) chunk(i int) ([]byte, error) { |
| 200 | return os.ReadFile(filepath.Join(f.dir, strconv.Itoa(i+1))) |
| 201 | } |
| 202 | |
| 203 | // drop removes chunk i once its bytes are in the files, so the spool and |
| 204 | // the files are never both whole on the member's disk. |
| 205 | func (f *fileSink) drop(i int) { os.Remove(filepath.Join(f.dir, strconv.Itoa(i+1))) } |
| 206 | |
| 207 | // readDBForward reads the reads route's lines (spec §5.5 steps 6 to 8). It |
| 208 | // says the reading sentence once admitted, keeps every chunk in sequence in |
| 209 | // sink (each one checked against bound before the sink keeps anything, plan |
| 210 | // review finding 37), and returns the read's id and its result unopened. |
| 211 | func readDBForward(ctx context.Context, body io.Reader, key *ecdh.PrivateKey, host, region, reading string, status io.Writer, sink dbChunkSink, bound *dbfChunkBound) (string, *dbreads.Result, error) { |
| 212 | unchecked := fmt.Errorf(dbfUnchecked, region) |
| 213 | sc := bufio.NewScanner(body) |
| 214 | sc.Buffer(make([]byte, 64<<10), dbreads.MaxLineBytes) |
| 215 | var id, ended string |
| 216 | for sc.Scan() { |
| 217 | var l dbreads.Line |
| 218 | if err := json.Unmarshal(sc.Bytes(), &l); err != nil { |
| 219 | return "", nil, unchecked |
| 220 | } |
| 221 | switch { |
| 222 | case l.Admitted != "": |
| 223 | if id != "" || !dbreads.ValidID(l.Admitted) { |
| 224 | return "", nil, unchecked |
| 225 | } |
| 226 | id = l.Admitted |
| 227 | fmt.Fprintln(status, fmt.Sprintf(reading, host, region)) |
| 228 | case l.Chunk != 0: |
| 229 | if id == "" || l.Chunk != int64(sink.count())+1 || !bound.admit(l.Data) { |
| 230 | return "", nil, unchecked |
| 231 | } |
| 232 | if err := sink.add(l.Data); err != nil { |
| 233 | return "", nil, err |
| 234 | } |
| 235 | case l.Result != nil: |
| 236 | if id == "" { |
| 237 | return "", nil, unchecked |
| 238 | } |
| 239 | return id, l.Result, nil |
| 240 | case l.Ended != "": |
| 241 | ended = l.Ended |
| 242 | } |
| 243 | } |
| 244 | if errors.Is(ctx.Err(), context.Canceled) { |
| 245 | return "", nil, errors.New("stopped") |
| 246 | } |
| 247 | if ended == dbreads.EndedNotPicked { |
| 248 | return "", nil, fmt.Errorf(dbfNotPicked, region) |
| 249 | } |
| 250 | return "", nil, fmt.Errorf(dbfUnfinished, host, region) |
| 251 | } |
| 252 | |
| 253 | // openDBForward opens a result: a done one's every chunk and its summary, |
| 254 | // checked against each other before anything is returned; any other state |
| 255 | // mapped to its sentence (spec §5.6, §10). |
| 256 | func openDBForward(id string, key *ecdh.PrivateKey, sink dbChunkSink, res dbreads.Result, host, region string) (dbForwardAnswer, error) { |
| 257 | unchecked := fmt.Errorf(dbfUnchecked, region) |
| 258 | if res.Valid() != nil { |
| 259 | return dbForwardAnswer{}, unchecked |
| 260 | } |
| 261 | s, err := dbreads.OpenSummary(key, id, res.Sealed) |
| 262 | if err != nil { |
| 263 | return dbForwardAnswer{}, unchecked |
| 264 | } |
| 265 | switch res.State { |
| 266 | case dbreads.StateDone: |
| 267 | if res.LastSeq != int64(sink.count()) { |
| 268 | return dbForwardAnswer{}, unchecked |
| 269 | } |
| 270 | var rows bytes.Buffer |
| 271 | for i := 0; i < sink.count(); i++ { |
| 272 | c, err := sink.chunk(i) |
| 273 | if err != nil { |
| 274 | return dbForwardAnswer{}, unchecked |
| 275 | } |
| 276 | seq := int64(i + 1) |
| 277 | plain, err := dbreads.OpenChunk(key, id, seq, dbreads.StreamRows, seq == res.LastSeq, c) |
| 278 | if err != nil { |
| 279 | return dbForwardAnswer{}, unchecked |
| 280 | } |
| 281 | rows.Write(plain) |
| 282 | } |
| 283 | sum := sha256.Sum256(rows.Bytes()) |
| 284 | if int64(rows.Len()) != s.Bytes || hex.EncodeToString(sum[:]) != s.SHA256 { |
| 285 | return dbForwardAnswer{}, unchecked |
| 286 | } |
| 287 | return dbForwardAnswer{rows: rows.Bytes(), summary: s, capped: res.Code == dbreads.CodeOutputCap}, nil |
| 288 | case dbreads.StateRefused: |
| 289 | if res.Code == dbreads.CodeMoving { |
| 290 | return dbForwardAnswer{}, fmt.Errorf(dbfMovingCLI, host) |
| 291 | } |
| 292 | return dbForwardAnswer{}, fmt.Errorf(dbfChanged, host) |
| 293 | case dbreads.StateFailed: |
| 294 | switch res.Code { |
| 295 | case dbreads.CodeNothingReplicated: |
| 296 | return dbForwardAnswer{}, dbRestoreError(host, dbdoor.ErrNothingReplicated) |
| 297 | case dbreads.CodeHistoryGone: |
| 298 | if !s.Oldest.IsZero() { |
| 299 | return dbForwardAnswer{}, dbRestoreError(host, &dbdoor.BeforeHistoryError{Oldest: s.Oldest}) |
| 300 | } |
| 301 | case dbreads.CodeCheckFailed: |
| 302 | return dbForwardAnswer{}, dbRestoreError(host, fmt.Errorf("%w: %s", dbdoor.ErrCheckFailed, s.Error)) |
| 303 | case dbreads.CodeChangedTwice: |
| 304 | return dbForwardAnswer{}, dbRestoreError(host, os.ErrNotExist) |
| 305 | case dbreads.CodeTooLarge: |
| 306 | return dbForwardAnswer{}, fmt.Errorf(dbfTooLarge, host, humanBytes(&s.Restored), humanBytes(&s.Cap), region) |
| 307 | case dbreads.CodeNoRoom: |
| 308 | return dbForwardAnswer{}, fmt.Errorf(dbfNoRoom, region, host) |
| 309 | case dbreads.CodeSQLError: |
| 310 | return dbForwardAnswer{}, errors.New(s.Error) |
| 311 | case dbreads.CodeTimeout: |
| 312 | return dbForwardAnswer{}, errors.New(dbfTimeout) |
| 313 | } |
| 314 | } |
| 315 | return dbForwardAnswer{}, fmt.Errorf(dbfUnfinished, host, region) |
| 316 | } |
| 317 | |
| 318 | // openDBPull opens a pull's result (spec §6, phase 2): a done one's chunks, |
| 319 | // opened in order, are the planned files concatenated; they are checked |
| 320 | // against the summary's count and hash and split by its file list into a |
| 321 | // litestream file replica under dir, which today's restore and check turn |
| 322 | // into the copy. Any other state maps to its sentence. |
| 323 | func openDBPull(ctx context.Context, id string, key *ecdh.PrivateKey, sink dbChunkSink, res dbreads.Result, host, region, dir string) (*dbdoor.Restored, error) { |
| 324 | unchecked := fmt.Errorf(dbfUnchecked, region) |
| 325 | if res.Valid() != nil { |
| 326 | return nil, unchecked |
| 327 | } |
| 328 | s, err := dbreads.OpenSummary(key, id, res.Sealed) |
| 329 | if err != nil { |
| 330 | return nil, unchecked |
| 331 | } |
| 332 | if res.State != dbreads.StateDone { |
| 333 | switch res.Code { |
| 334 | case dbreads.CodeTooLarge: |
| 335 | return nil, fmt.Errorf(dbfPullTooLarge, host, humanBytes(&s.Restored), humanBytes(&s.Cap), region) |
| 336 | case dbreads.CodeNoRoom: |
| 337 | return nil, fmt.Errorf(dbfPullNoRoom, region, host) |
| 338 | } |
| 339 | _, err := openDBForward(id, key, sink, res, host, region) |
| 340 | return nil, err |
| 341 | } |
| 342 | // A done pull is the whole answer: no output cap applies to one. |
| 343 | if res.Code != "" || res.LastSeq != int64(sink.count()) || checkPullManifest(s) != nil { |
| 344 | return nil, unchecked |
| 345 | } |
| 346 | root := filepath.Join(dir, "replica") |
| 347 | h := sha256.New() |
| 348 | fi, written := 0, int64(0) |
| 349 | var head []byte // the current file's first ltx.HeaderSize bytes |
| 350 | var out *os.File |
| 351 | defer func() { |
| 352 | if out != nil { |
| 353 | out.Close() |
| 354 | } |
| 355 | }() |
| 356 | drop, _ := sink.(interface{ drop(int) }) |
| 357 | for i := 0; i < sink.count(); i++ { |
| 358 | // Ctrl-C and the read's deadline stop the rebuild too (Astra round |
| 359 | // 1 on plan 3, finding 2); RestoreLocal's reads check ctx after it. |
| 360 | if err := ctx.Err(); err != nil { |
| 361 | return nil, dbPullStopped(err, host, region) |
| 362 | } |
| 363 | sealed, err := sink.chunk(i) |
| 364 | if err != nil { |
| 365 | return nil, unchecked |
| 366 | } |
| 367 | seq := int64(i + 1) |
| 368 | plain, err := dbreads.OpenChunk(key, id, seq, dbreads.StreamLTX, seq == res.LastSeq, sealed) |
| 369 | if err != nil { |
| 370 | return nil, unchecked |
| 371 | } |
| 372 | h.Write(plain) |
| 373 | for len(plain) > 0 { |
| 374 | if fi >= len(s.Files) { |
| 375 | return nil, unchecked |
| 376 | } |
| 377 | f := s.Files[fi] |
| 378 | if out == nil { |
| 379 | p := litestream.LTXFilePath(root, f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID)) |
| 380 | if err := os.MkdirAll(filepath.Dir(p), 0o700); err != nil { |
| 381 | return nil, err |
| 382 | } |
| 383 | if out, err = os.OpenFile(p, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600); err != nil { |
| 384 | return nil, err |
| 385 | } |
| 386 | } |
| 387 | n := min(int64(len(plain)), f.Size-written) |
| 388 | if _, err := out.Write(plain[:n]); err != nil { |
| 389 | return nil, err |
| 390 | } |
| 391 | if len(head) < ltx.HeaderSize { |
| 392 | head = append(head, plain[:min(n, int64(ltx.HeaderSize-len(head)))]...) |
| 393 | } |
| 394 | plain, written = plain[n:], written+n |
| 395 | if written == f.Size { |
| 396 | if err := out.Close(); err != nil { |
| 397 | out = nil |
| 398 | return nil, err |
| 399 | } |
| 400 | // litestream's restore picks files by name and compacts them |
| 401 | // by header: a file whose header is not its manifest entry |
| 402 | // would restore as a transaction it does not hold (Astra |
| 403 | // round 1 on plan 3, finding 1). |
| 404 | var lh ltx.Header |
| 405 | if lh.UnmarshalBinary(head) != nil || uint64(lh.MinTXID) != f.MinTXID || uint64(lh.MaxTXID) != f.MaxTXID { |
| 406 | out = nil |
| 407 | return nil, unchecked |
| 408 | } |
| 409 | out, fi, written, head = nil, fi+1, 0, nil |
| 410 | } |
| 411 | } |
| 412 | if drop != nil { |
| 413 | drop.drop(i) |
| 414 | } |
| 415 | } |
| 416 | if fi != len(s.Files) || hex.EncodeToString(h.Sum(nil)) != s.SHA256 { |
| 417 | return nil, unchecked |
| 418 | } |
| 419 | db := filepath.Join(dir, "replica.db") |
| 420 | if err := dbdoor.RestoreLocal(ctx, lsfile.NewReplicaClient(root), db, s.TXID); err != nil { |
| 421 | if cerr := ctx.Err(); cerr != nil { |
| 422 | return nil, dbPullStopped(cerr, host, region) |
| 423 | } |
| 424 | return nil, dbRestoreError(host, err) |
| 425 | } |
| 426 | os.RemoveAll(root) |
| 427 | return &dbdoor.Restored{Path: db, Dir: dir, TXID: s.TXID, At: s.At}, nil |
| 428 | } |
| 429 | |
| 430 | // dbPullStopped is a pull that ended on its context: "stopped" for Ctrl-C, |
| 431 | // DBF-12 for the read's own deadline. |
| 432 | func dbPullStopped(err error, host, region string) error { |
| 433 | if errors.Is(err, context.Canceled) { |
| 434 | return errors.New("stopped") |
| 435 | } |
| 436 | return fmt.Errorf(dbfUnfinished, host, region) |
| 437 | } |
| 438 | |
| 439 | // checkPullManifest is the summary's file list checked whole before any |
| 440 | // file is created (plan review finding 22): every size a possible LTX file |
| 441 | // and within the cap, every level one litestream writes, every range a |
| 442 | // range, no file twice, sizes that sum without overflow to the summary's |
| 443 | // byte count, and a last file that ends at its transaction. |
| 444 | func checkPullManifest(s dbreads.Summary) error { |
| 445 | bad := errors.New("the file list does not add up") |
| 446 | if len(s.Files) == 0 || s.Bytes <= 0 || s.Bytes > dbreads.MaxPullBytes { |
| 447 | return bad |
| 448 | } |
| 449 | type coord struct { |
| 450 | level int |
| 451 | min, max uint64 |
| 452 | } |
| 453 | seen := make(map[coord]bool, len(s.Files)) |
| 454 | var sum int64 |
| 455 | for _, f := range s.Files { |
| 456 | c := coord{f.Level, f.MinTXID, f.MaxTXID} |
| 457 | switch { |
| 458 | case f.Size < ltx.HeaderSize+ltx.TrailerSize, f.Size > dbreads.MaxPullBytes, |
| 459 | f.Level < 0, f.Level > litestream.SnapshotLevel, |
| 460 | f.MinTXID < 1, f.MaxTXID < f.MinTXID, seen[c], |
| 461 | sum > dbreads.MaxPullBytes-f.Size: // sum + f.Size would pass the cap |
| 462 | return bad |
| 463 | } |
| 464 | seen[c] = true |
| 465 | sum += f.Size |
| 466 | } |
| 467 | if sum != s.Bytes || s.Files[len(s.Files)-1].MaxTXID != s.TXID { |
| 468 | return bad |
| 469 | } |
| 470 | return nil |
| 471 | } |
| 472 | |
| 473 | // dbPullSpoolWrite writes one spooled chunk: os.WriteFile, a seam so a test |
| 474 | // can fail the spool part way (plan review finding 23). |
| 475 | var dbPullSpoolWrite = os.WriteFile |
| 476 | |
| 477 | // dbPullForward is a pull through the forwarded read (spec §6). The spool, |
| 478 | // the files and the copy live in one private 0700 directory under the system |
| 479 | // temp dir, removed on every path but success, where the caller removes it |
| 480 | // after copyOut, as it does today's restore directory. |
| 481 | func dbPullForward(ctx context.Context, t dbTarget, fwd dbdoor.Forward) (*dbdoor.Restored, error) { |
| 482 | ctx, cancel := context.WithTimeout(ctx, dbreads.CLIBound) |
| 483 | defer cancel() |
| 484 | key, err := dbreads.NewReplyKey() |
| 485 | if err != nil { |
| 486 | return nil, err |
| 487 | } |
| 488 | at := "" |
| 489 | if !t.at.IsZero() { |
| 490 | at = t.at.UTC().Format(time.RFC3339Nano) |
| 491 | } |
| 492 | body, err := json.Marshal(dbreads.ReadBody{Op: dbreads.OpPull, At: at, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())}) |
| 493 | if err != nil { |
| 494 | return nil, err |
| 495 | } |
| 496 | res, err := t.sess.c.doStream(ctx, http.MethodPost, apiPath(t.acct, t.app, "/db/"+url.PathEscape(t.host)+"/reads"), |
| 497 | bytes.NewReader(body), int64(len(body)), http.Header{"Content-Type": {"application/json"}}) |
| 498 | if err != nil { |
| 499 | if errors.Is(ctx.Err(), context.Canceled) { |
| 500 | return nil, errors.New("stopped") |
| 501 | } |
| 502 | return nil, fmt.Errorf(dbfUnfinished, t.host, fwd.Region) |
| 503 | } |
| 504 | defer res.Body.Close() |
| 505 | if res.StatusCode != http.StatusOK { |
| 506 | raw, _ := io.ReadAll(io.LimitReader(res.Body, 64<<10)) |
| 507 | return nil, apiErrorFromBody(res.StatusCode, raw) |
| 508 | } |
| 509 | dir, err := os.MkdirTemp("", "carlos-db-") |
| 510 | if err != nil { |
| 511 | return nil, err |
| 512 | } |
| 513 | // MkdirTemp already creates 0700; say it anyway, as the in-region |
| 514 | // restore does, because the copy's privacy rests on it. |
| 515 | if err := os.Chmod(dir, 0o700); err != nil { |
| 516 | os.RemoveAll(dir) |
| 517 | return nil, err |
| 518 | } |
| 519 | done := false |
| 520 | defer func() { |
| 521 | if !done { |
| 522 | os.RemoveAll(dir) |
| 523 | } |
| 524 | }() |
| 525 | sink := &fileSink{dir: filepath.Join(dir, "sealed")} |
| 526 | id, result, err := readDBForward(ctx, res.Body, key, t.host, fwd.Region, dbfFetching, os.Stderr, sink, newDBFChunkBound(dbreads.MaxPullBytes, dbreads.PullChunkBytes)) |
| 527 | if err != nil { |
| 528 | return nil, err |
| 529 | } |
| 530 | r, err := openDBPull(ctx, id, key, sink, *result, t.host, fwd.Region, dir) |
| 531 | if err != nil { |
| 532 | return nil, err |
| 533 | } |
| 534 | os.RemoveAll(sink.dir) |
| 535 | done = true |
| 536 | return r, nil |
| 537 | } |
| 538 | |
amadan