| 1 | // The `carlos db` replica door (spec 2026-09-02 §6). It serves an |
| 2 | // instance's replicated LTX files BY COORDINATE (level plus TXID span) from |
| 3 | // the region data bucket, to a member of the account and nobody else, so the |
| 4 | // CLI can restore a private copy on the member's own machine. The console |
| 5 | // never executes tenant SQL and never holds a restored database. |
| 6 | // |
| 7 | // resolveReplica is the half that makes no S3 call: resolving an instance |
| 8 | // record to one replica prefix, and refusing everything that is not this |
| 9 | // console's to read. Every refusal there happens BEFORE the bucket is |
| 10 | // touched. That is deliberate: a door that tried and mapped AccessDenied |
| 11 | // would leak the attempt, the latency, and eventually a bucket name into a |
| 12 | // log a member can provoke. |
| 13 | |
| 14 | package console |
| 15 | |
| 16 | import ( |
| 17 | "context" |
| 18 | "errors" |
| 19 | "fmt" |
| 20 | "io" |
| 21 | "log" |
| 22 | "net/http" |
| 23 | "os" |
| 24 | "strconv" |
| 25 | "sync" |
| 26 | "time" |
| 27 | |
| 28 | "github.com/aws/smithy-go" |
| 29 | "github.com/superfly/ltx" |
| 30 | |
| 31 | "github.com/carlosframework/platform/internal/activator" |
| 32 | "github.com/carlosframework/platform/internal/dbdoor" |
| 33 | "github.com/carlosframework/platform/internal/dbreads" |
| 34 | "github.com/carlosframework/platform/internal/hostclaim" |
| 35 | "github.com/carlosframework/platform/internal/lease" |
| 36 | "github.com/carlosframework/platform/internal/provision" |
| 37 | "github.com/carlosframework/platform/internal/registry" |
| 38 | ) |
| 39 | |
| 40 | var ( |
| 41 | // ErrResidency: the data is in another region's bucket, or on a |
| 42 | // customer's own fleet box. IAM makes those reads impossible by design, |
| 43 | // so this is the design working rather than a fault to retry. |
| 44 | ErrResidency = errors.New("db: the instance's data is outside this region") |
| 45 | // ErrPooled: a pooled record's data lives on the pool box, the same |
| 46 | // class the instance-move worker refuses. |
| 47 | ErrPooled = errors.New("db: pooled instances have no replica of their own") |
| 48 | // ErrBadHost: path.Join cleans "../x" to "x" rather than refusing it, |
| 49 | // so a host becomes a key segment only after ValidHostShape, and only |
| 50 | // when it is the host the record itself names. |
| 51 | ErrBadHost = errors.New("db: invalid host") |
| 52 | ) |
| 53 | |
| 54 | func (s *Server) resolveReplica(rec provision.Record, host string) (string, error) { |
| 55 | if !provision.ValidHostShape(host) || rec.Host != host { |
| 56 | return "", fmt.Errorf("%w: %q", ErrBadHost, host) |
| 57 | } |
| 58 | if rec.Pool != "" { |
| 59 | return "", fmt.Errorf("%w: %s is in pool %q", ErrPooled, host, rec.Pool) |
| 60 | } |
| 61 | // A fleet placement is checked on its own, not left to the region |
| 62 | // compare: the data is on the customer's box whatever this console's |
| 63 | // CARLOS_REGION says. |
| 64 | if _, fleet := registry.FleetPlacement(rec.Region); fleet { |
| 65 | return "", fmt.Errorf("%w: %s is placed on fleet %q", ErrResidency, host, rec.Region) |
| 66 | } |
| 67 | if rec.Region != s.cfg.Region { |
| 68 | return "", fmt.Errorf("%w: %s is homed in %q, this console serves %q", |
| 69 | ErrResidency, host, rec.Region, s.cfg.Region) |
| 70 | } |
| 71 | return activator.HibernatePrefix(s.cfg.HibernatePrefix, host), nil |
| 72 | } |
| 73 | |
| 74 | // replicaReader is the only part of a litestream replica client the door |
| 75 | // may hold: the two read methods. Holding the narrow interface means no |
| 76 | // handler here can reach WriteLTXFile, DeleteLTXFiles or DeleteAll by |
| 77 | // mistake, whatever the concrete client underneath can do. |
| 78 | type replicaReader interface { |
| 79 | LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) |
| 80 | OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) |
| 81 | } |
| 82 | |
| 83 | // cliDBBurst bounds the db door per TOKEN within the ordinary minute window. |
| 84 | // Its own budget, not the instance verbs' 60 (spec §13.6): one restore is |
| 85 | // ten listings plus one request per planned file, and an app with a long L0 |
| 86 | // tail plans dozens. The number is a ceiling on a stolen token, not a |
| 87 | // policing of honest pulls. |
| 88 | const cliDBBurst = 600 |
| 89 | |
| 90 | // dbAuditAction names the audit row every served file writes. |
| 91 | const dbAuditAction = "db-read" |
| 92 | |
| 93 | // dbRequest is what dbGate resolves for both handlers. |
| 94 | type dbRequest struct { |
| 95 | claim hostclaim.Observed |
| 96 | acct *Account |
| 97 | actor string |
| 98 | app string |
| 99 | host string |
| 100 | prefix string |
| 101 | rd replicaReader |
| 102 | } |
| 103 | |
| 104 | // dbGate is the door's whole gate, in the repo's order: principal (cliRoute |
| 105 | // has already refused every service credential at the zero Set) -> |
| 106 | // membership (byte-identical 404) -> budget (429) -> configuration (503) -> |
| 107 | // the record (404 when absent) -> the class rail and residency (409), none |
| 108 | // of which touches the bucket -> the replica client. ok=false means a |
| 109 | // response has been written. |
| 110 | func (s *Server) dbGate(w http.ResponseWriter, r *http.Request) (dbRequest, bool) { |
| 111 | // Belt to cliRoute's braces: the zero Set already stops every service |
| 112 | // principal before this runs, so this line is only reached by a request |
| 113 | // that did not come through cliRoute at all. |
| 114 | if p, ok := principalFrom(r.Context()); !ok || p.Kind != PrincipalUser { |
| 115 | http.NotFound(w, r) |
| 116 | return dbRequest{}, false |
| 117 | } |
| 118 | actor, acct, app, ok := s.apiAccountApp(w, r) |
| 119 | if !ok { |
| 120 | return dbRequest{}, false |
| 121 | } |
| 122 | if !s.cli.db.allow(hashToken(bearer(r)), time.Now()) { |
| 123 | jsonError(w, http.StatusTooManyRequests, "too many database requests. Wait a minute and try again.") |
| 124 | return dbRequest{}, false |
| 125 | } |
| 126 | if s.cfg.Releases == nil || s.cfg.HibernateBucket == "" { |
| 127 | jsonError(w, http.StatusServiceUnavailable, "this console has no data bucket set up, so it cannot read databases.") |
| 128 | return dbRequest{}, false |
| 129 | } |
| 130 | host := r.PathValue("host") |
| 131 | ctx := r.Context() |
| 132 | src, err := s.resolveDBSource(ctx, acct, actor, app, host) |
| 133 | switch { |
| 134 | case errors.Is(err, os.ErrNotExist), errors.Is(err, ErrBadHost): |
| 135 | http.NotFound(w, r) |
| 136 | return dbRequest{}, false |
| 137 | case errors.Is(err, errSourceClass): |
| 138 | jsonError(w, http.StatusConflict, dbClassRefusal(src.rec)) |
| 139 | return dbRequest{}, false |
| 140 | case errors.Is(err, ErrPooled): |
| 141 | jsonError(w, http.StatusConflict, fmt.Sprintf( |
| 142 | "%s runs in the pool %s, so its data is on the pool's own server. carlos db cannot read it.", host, src.rec.Pool)) |
| 143 | return dbRequest{}, false |
| 144 | case errors.Is(err, ErrResidency): |
| 145 | // Main's sentence unless this console forwards (db region forward |
| 146 | // spec §5.5 step 1): then DBF-1, DBF-2 or the forward marker. |
| 147 | s.dbResidency(w, r, src.rec) |
| 148 | return dbRequest{}, false |
| 149 | case errors.Is(err, errSourceClaim): |
| 150 | g := dbRequest{acct: acct, actor: actor, app: app, host: host} |
| 151 | if s.cfg.DBVersions == nil { |
| 152 | jsonError(w, http.StatusServiceUnavailable, "this console has no versioned database reader") |
| 153 | return dbRequest{}, false |
| 154 | } |
| 155 | s.dbClaimFailure(w, r, g, src.claimErr) |
| 156 | return dbRequest{}, false |
| 157 | case errors.Is(err, errSourceResolve): |
| 158 | log.Printf("console: api db %s/%s/%s: resolve: %v", acct.Sqid, app, host, err) |
| 159 | jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.") |
| 160 | return dbRequest{}, false |
| 161 | case err != nil: |
| 162 | log.Printf("console: api db %s/%s/%s: read record: %v", acct.Sqid, app, host, err) |
| 163 | jsonError(w, http.StatusBadGateway, "could not read the instance. Try again.") |
| 164 | return dbRequest{}, false |
| 165 | } |
| 166 | if s.cfg.HostClaim { |
| 167 | return dbRequest{claim: src.claim, acct: acct, actor: actor, app: app, host: host, prefix: src.prefix}, true |
| 168 | } |
| 169 | rd, err := s.dbReplica(ctx, src.prefix) |
| 170 | if err != nil { |
| 171 | log.Printf("console: api db %s/%s/%s: replica client: %v", acct.Sqid, app, host, err) |
| 172 | jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.") |
| 173 | return dbRequest{}, false |
| 174 | } |
| 175 | return dbRequest{acct: acct, actor: actor, app: app, host: host, prefix: src.prefix, rd: rd}, true |
| 176 | } |
| 177 | |
| 178 | var ( |
| 179 | errSourceClass = errors.New("db: not an instance the platform keeps a replica of") |
| 180 | errSourceClaim = errors.New("db: the instance's claim could not be read") |
| 181 | // errSourceResolve marks a resolveReplica failure that is none of the |
| 182 | // named refusals, so the db door keeps its own log line and answer for |
| 183 | // it apart from a failed record read. |
| 184 | errSourceResolve = errors.New("db: the replica could not be resolved") |
| 185 | ) |
| 186 | |
| 187 | // dbSource is a replica source both the db door and the --from door may |
| 188 | // read (seeded-instances spec §4, codex 17): a record of this account and |
| 189 | // app, an instance with a replica of its own, homed in this console's |
| 190 | // region, and, with claims on, its versioned claim. claimErr is the claim |
| 191 | // read's own error, set beside the errSourceClaim wrap so a caller can hand |
| 192 | // it to dbClaimFailure unwrapped. |
| 193 | type dbSource struct { |
| 194 | rec provision.Record |
| 195 | prefix string |
| 196 | claim hostclaim.Observed |
| 197 | claimErr error |
| 198 | } |
| 199 | |
| 200 | // resolveDBSource holds the source checks dbGate made inline. It does no |
| 201 | // authorization: dbGate keeps its person-only gate, and the --from door |
| 202 | // authorizes through cliRoute's instance role. |
| 203 | func (s *Server) resolveDBSource(ctx context.Context, acct *Account, actor, app, host string) (dbSource, error) { |
| 204 | if !provision.ValidHostShape(host) { |
| 205 | return dbSource{}, fmt.Errorf("%w: %q", ErrBadHost, host) |
| 206 | } |
| 207 | rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host) |
| 208 | if err != nil { |
| 209 | return dbSource{}, err // os.ErrNotExist passes through as itself |
| 210 | } |
| 211 | if msg := dbClassRefusal(*rec); msg != "" { |
| 212 | return dbSource{rec: *rec}, fmt.Errorf("%w: %s", errSourceClass, msg) |
| 213 | } |
| 214 | prefix, err := s.resolveReplica(*rec, host) |
| 215 | if err != nil { |
| 216 | switch { |
| 217 | case errors.Is(err, ErrBadHost), errors.Is(err, ErrPooled), errors.Is(err, ErrResidency): |
| 218 | return dbSource{rec: *rec}, err |
| 219 | } |
| 220 | return dbSource{rec: *rec}, fmt.Errorf("%w: %w", errSourceResolve, err) |
| 221 | } |
| 222 | src := dbSource{rec: *rec, prefix: prefix} |
| 223 | if s.cfg.HostClaim { |
| 224 | if s.cfg.DBVersions == nil { |
| 225 | return src, fmt.Errorf("%w: this console has no versioned database reader", errSourceClaim) |
| 226 | } |
| 227 | g := dbRequest{acct: acct, actor: actor, app: app, host: host} |
| 228 | if src.claim, err = s.dbClaim(ctx, g, *rec); err != nil { |
| 229 | src.claimErr = err |
| 230 | return src, fmt.Errorf("%w: %w", errSourceClaim, err) |
| 231 | } |
| 232 | src.prefix = src.claim.Claim.Holder.Location |
| 233 | } |
| 234 | return src, nil |
| 235 | } |
| 236 | |
| 237 | // dbClassRefusal is the class rail as a sentence, or "" for an |
| 238 | // activator-managed exec instance: the only class whose database the |
| 239 | // platform replicates (spec §1 scope). Same three facts the move doors |
| 240 | // refuse on; the fleet and pool cases are resolveReplica's. The rule lives |
| 241 | // in dbreads.ClassRefusal (db region forward spec §8.2). |
| 242 | func dbClassRefusal(rec provision.Record) string { return dbreads.ClassRefusal(rec) } |
| 243 | |
| 244 | // dbResidencyRefusal names where the data is, in the vocabulary the |
| 245 | // regions door uses (the home region by its CONSOLE_HOME_REGION name). |
| 246 | func (s *Server) dbResidencyRefusal(rec provision.Record) string { |
| 247 | if fleet, ok := registry.FleetPlacement(rec.Region); ok { |
| 248 | 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) |
| 249 | } |
| 250 | return fmt.Sprintf("%s keeps its data in %s. This console can only read data kept in %s.", |
| 251 | rec.Host, s.dbRegionName(rec.Region), s.dbRegionName(s.cfg.Region)) |
| 252 | } |
| 253 | |
| 254 | func (s *Server) dbRegionName(region string) string { |
| 255 | if r := s.homeRegionOut(region); r != "" { |
| 256 | return "region " + r |
| 257 | } |
| 258 | return "the home region" |
| 259 | } |
| 260 | |
| 261 | // dbStoreFailure answers a failed bucket read. AccessDenied is the |
| 262 | // residency refusal as a backstop (spec §6): the role this console runs as |
| 263 | // cannot read that data, which is the design working, not a fault to |
| 264 | // retry. Everything else is logged with its detail (bucket names, request |
| 265 | // ids) and answered with none of it. |
| 266 | func (s *Server) dbStoreFailure(w http.ResponseWriter, g dbRequest, what string, err error) { |
| 267 | log.Printf("console: api db %s/%s/%s: %s: %v", g.acct.Sqid, g.app, g.host, what, err) |
| 268 | var api smithy.APIError |
| 269 | if errors.As(err, &api) && api.ErrorCode() == "AccessDenied" { |
| 270 | jsonError(w, http.StatusConflict, fmt.Sprintf( |
| 271 | "this console is not allowed to read the data for %s. It is kept outside the region this console serves.", g.host)) |
| 272 | return |
| 273 | } |
| 274 | jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.") |
| 275 | } |
| 276 | |
| 277 | // parseDBLevel accepts a bare decimal level in 0..MaxLevel. Anything else |
| 278 | // is not a coordinate. |
| 279 | func parseDBLevel(s string) (int, bool) { |
| 280 | n, err := strconv.ParseUint(s, 10, 8) |
| 281 | if err != nil || n > dbdoor.MaxLevel { |
| 282 | return 0, false |
| 283 | } |
| 284 | return int(n), true |
| 285 | } |
| 286 | |
| 287 | // handleAPIDBList answers one level's listing. An out-of-range level is a |
| 288 | // 404 (it is not a coordinate); a level with nothing in it is 200 and an |
| 289 | // empty list (it is a coordinate that happens to be empty, spec §6.1). |
| 290 | func (s *Server) handleAPIDBList(w http.ResponseWriter, r *http.Request) { |
| 291 | level, ok := parseDBLevel(r.PathValue("level")) |
| 292 | if !ok { |
| 293 | http.NotFound(w, r) |
| 294 | return |
| 295 | } |
| 296 | g, ok := s.dbGate(w, r) |
| 297 | if !ok { |
| 298 | return |
| 299 | } |
| 300 | if s.cfg.HostClaim { |
| 301 | out, err := s.listDBVersions(r.Context(), g, level) |
| 302 | if err != nil { |
| 303 | s.dbClaimFailure(w, r, g, err) |
| 304 | return |
| 305 | } |
| 306 | w.Header().Set("Cache-Control", "no-store") |
| 307 | writeJSON(w, http.StatusOK, out) |
| 308 | return |
| 309 | } |
| 310 | out, err := listReplicaLevel(r.Context(), g.rd, level) |
| 311 | if err != nil { |
| 312 | s.dbStoreFailure(w, g, fmt.Sprintf("list level %d", level), err) |
| 313 | return |
| 314 | } |
| 315 | writeJSON(w, http.StatusOK, out) |
| 316 | } |
| 317 | |
| 318 | // listReplicaLevel lists one level of rd in the door's wire shape, ordered |
| 319 | // by MinTXID, with an empty level as an empty list. The db door and the |
| 320 | // console's own in-process door (replicaDoer, for --from) both list |
| 321 | // through it, so a seeded copy restores from the listing carlos db pull |
| 322 | // reads. The body moved to dbdoor.ListLevel so the box's planner lists |
| 323 | // through it too (db region forward spec §6). |
| 324 | func listReplicaLevel(ctx context.Context, rd replicaReader, level int) (dbdoor.ListResponse, error) { |
| 325 | return dbdoor.ListLevel(ctx, rd, level) |
| 326 | } |
| 327 | |
| 328 | // handleAPIDBGet streams one LTX file by coordinate. The bytes pass through |
| 329 | // the console, never a presigned URL (the edge store door's F8 precedent): |
| 330 | // nothing the request supplies picks the bucket or the key, only a level |
| 331 | // and a span inside the prefix resolveReplica chose. |
| 332 | func (s *Server) handleAPIDBGet(w http.ResponseWriter, r *http.Request) { |
| 333 | level, ok := parseDBLevel(r.PathValue("level")) |
| 334 | if !ok { |
| 335 | http.NotFound(w, r) |
| 336 | return |
| 337 | } |
| 338 | minTXID, maxTXID, err := dbdoor.ParseSpan(r.PathValue("span")) |
| 339 | if err != nil { |
| 340 | http.NotFound(w, r) |
| 341 | return |
| 342 | } |
| 343 | g, ok := s.dbGate(w, r) |
| 344 | if !ok { |
| 345 | return |
| 346 | } |
| 347 | var pinned *dbVersionReader |
| 348 | if s.cfg.HostClaim { |
| 349 | pinned, err = s.pinnedDBReader(r, g, level, minTXID, maxTXID) |
| 350 | if err != nil { |
| 351 | s.dbClaimFailure(w, r, g, err) |
| 352 | return |
| 353 | } |
| 354 | g.rd = pinned |
| 355 | } |
| 356 | rangeHdr := r.Header.Get("Range") |
| 357 | offset, size, err := dbdoor.ParseRange(rangeHdr) |
| 358 | if err != nil { |
| 359 | jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range") |
| 360 | return |
| 361 | } |
| 362 | if pinned != nil { |
| 363 | if _, err := dbVersionRange(pinned.handle.Size, offset, size); err != nil { |
| 364 | jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range") |
| 365 | return |
| 366 | } |
| 367 | } |
| 368 | if !s.dbStreams.acquire(g.actor) { |
| 369 | w.Header().Set("Retry-After", dbStreamRetryAfter) |
| 370 | jsonError(w, http.StatusTooManyRequests, "too many database downloads at once. Wait a few seconds and try again.") |
| 371 | return |
| 372 | } |
| 373 | defer s.dbStreams.release(g.actor) |
| 374 | // The idle watchdog: any stretch of dbStreamIdle with no byte moved |
| 375 | // cancels the bucket read. The write side has its own deadline below, |
| 376 | // renewed per chunk, because a cancelled context does not unblock a |
| 377 | // write to a client that has stopped reading. |
| 378 | idle := s.dbStreamIdle |
| 379 | ctx, cancel := context.WithCancel(r.Context()) |
| 380 | defer cancel() |
| 381 | watchdog := time.AfterFunc(idle, cancel) |
| 382 | defer watchdog.Stop() |
| 383 | rc, err := g.rd.OpenLTXFile(ctx, level, ltx.TXID(minTXID), ltx.TXID(maxTXID), offset, size) |
| 384 | if errors.Is(err, os.ErrNotExist) { |
| 385 | http.NotFound(w, r) |
| 386 | return |
| 387 | } |
| 388 | if err != nil { |
| 389 | s.dbStoreFailure(w, g, fmt.Sprintf("open level %d %s", level, dbdoor.Span(minTXID, maxTXID)), err) |
| 390 | return |
| 391 | } |
| 392 | defer rc.Close() |
| 393 | // One row per successful GET, ranged or not. A ranged read is still a |
| 394 | // read of the file: skipping it would let a caller list the coordinates |
| 395 | // and fetch every file from byte 1 with no row written. A resumed read |
| 396 | // says where it started. |
| 397 | detail := fmt.Sprintf("read database replica file level %d, transactions %d to %d (cli)", level, minTXID, maxTXID) |
| 398 | if offset > 0 { |
| 399 | detail = fmt.Sprintf("read database replica file level %d, transactions %d to %d, from byte %d (cli)", level, minTXID, maxTXID, offset) |
| 400 | } |
| 401 | s.auditActivity(g.acct, g.app, g.actor, Activity{ |
| 402 | Action: dbAuditAction, TargetType: "instance", Target: g.host, Detail: detail, |
| 403 | }) |
| 404 | if pinned != nil { |
| 405 | if err := dbVersionHeaders(w, r, pinned.handle.Size, offset, size); err != nil { |
| 406 | jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range") |
| 407 | return |
| 408 | } |
| 409 | } |
| 410 | w.Header().Set("Content-Type", "application/octet-stream") |
| 411 | w.Header().Set("Cache-Control", "no-store") |
| 412 | status := http.StatusOK |
| 413 | if rangeHdr != "" { |
| 414 | status = http.StatusPartialContent |
| 415 | } |
| 416 | rw := http.NewResponseController(w) |
| 417 | // The deadline is never cleared here. net/http flushes what is still |
| 418 | // buffered (the last bytes, the chunked terminator) in finishRequest, |
| 419 | // after this handler returns, and clears the write deadline itself |
| 420 | // only after that flush. Clearing it on the way out would leave that |
| 421 | // flush unbounded, holding the connection for as long as a client that |
| 422 | // stopped reading near the end cares to wait. |
| 423 | _ = rw.SetWriteDeadline(time.Now().Add(idle)) |
| 424 | w.WriteHeader(status) |
| 425 | err = dbStream(w, rw, rc, watchdog, idle) |
| 426 | // Flush the tail now, under a fresh deadline, so it is done (or cut) |
| 427 | // before the deferred release gives this stream's slot back. What |
| 428 | // finishRequest has left after this is the chunked terminator, still |
| 429 | // under the same deadline. On a failed read the deadline is renewed |
| 430 | // too, so the client gets its truncated body promptly (and resumes) |
| 431 | // rather than a connection cut by a deadline the stall used up. |
| 432 | _ = rw.SetWriteDeadline(time.Now().Add(idle)) |
| 433 | if ferr := rw.Flush(); err == nil { |
| 434 | err = ferr |
| 435 | } |
| 436 | if err != nil { |
| 437 | // The status is already on the wire. The client's resumable reader |
| 438 | // sees a short body against the listed size and reopens at the |
| 439 | // offset it reached. |
| 440 | log.Printf("console: api db %s/%s/%s: stream level %d %s: %v", |
| 441 | g.acct.Sqid, g.app, g.host, level, dbdoor.Span(minTXID, maxTXID), err) |
| 442 | } |
| 443 | } |
| 444 | |
| 445 | // dbStream copies rc to w a chunk at a time, renewing the write deadline |
| 446 | // before each chunk and the read watchdog after it, so a stream ends when |
| 447 | // either side stops moving for idle. |
| 448 | func dbStream(w io.Writer, rw *http.ResponseController, rc io.Reader, watchdog *time.Timer, idle time.Duration) error { |
| 449 | buf := make([]byte, 32<<10) |
| 450 | for { |
| 451 | n, rerr := rc.Read(buf) |
| 452 | if n > 0 { |
| 453 | if err := rw.SetWriteDeadline(time.Now().Add(idle)); err != nil && !errors.Is(err, http.ErrNotSupported) { |
| 454 | return err |
| 455 | } |
| 456 | if _, err := w.Write(buf[:n]); err != nil { |
| 457 | return err |
| 458 | } |
| 459 | watchdog.Reset(idle) |
| 460 | } |
| 461 | if rerr == io.EOF { |
| 462 | return nil |
| 463 | } |
| 464 | if rerr != nil { |
| 465 | return rerr |
| 466 | } |
| 467 | } |
| 468 | } |
| 469 | |
| 470 | // dbReplicaCacheMax bounds the replica-client cache. Each client owns an |
| 471 | // HTTP transport, so one per request would cost a TLS handshake per file and |
| 472 | // leave an idle connection behind each; one per prefix lets a restore's |
| 473 | // dozens of requests share one. Over the bound the map is dropped |
| 474 | // wholesale, the rateLimiter's rule: forgetting clients costs a handshake, |
| 475 | // a walk would cost a lock held across every door. |
| 476 | const dbReplicaCacheMax = 64 |
| 477 | |
| 478 | type dbReplicaCache struct { |
| 479 | mu sync.Mutex |
| 480 | clients map[string]replicaReader |
| 481 | } |
| 482 | |
| 483 | // cachedReadReplica is the production dbReplica: lease.ReadReplicaClient on |
| 484 | // the configured bucket, one per prefix. |
| 485 | func (s *Server) cachedReadReplica(ctx context.Context, prefix string) (replicaReader, error) { |
| 486 | c := &s.dbClients |
| 487 | c.mu.Lock() |
| 488 | defer c.mu.Unlock() |
| 489 | if rd, ok := c.clients[prefix]; ok { |
| 490 | return rd, nil |
| 491 | } |
| 492 | rc, err := lease.ReadReplicaClient(ctx, s.cfg.HibernateBucket, prefix, s.cfg.HibernateRegion) |
| 493 | if err != nil { |
| 494 | return nil, err |
| 495 | } |
| 496 | if c.clients == nil || len(c.clients) >= dbReplicaCacheMax { |
| 497 | c.clients = make(map[string]replicaReader, 8) |
| 498 | } |
| 499 | c.clients[prefix] = rc |
| 500 | return rc, nil |
| 501 | } |
| 502 | |
| 503 | // The db door's stream bounds. The per-minute token budget bounds how many |
| 504 | // requests start, not how many are still running, and a slow or stalled |
| 505 | // download holds a goroutine, an S3 connection and a client connection for |
| 506 | // as long as it lasts. So active file streams are bounded per principal and |
| 507 | // across the console, and refused with 429 and Retry-After when full; and |
| 508 | // each stream must move a byte at least every dbStreamIdle, in both |
| 509 | // directions, or it is cut. The CLI's restore downloads one file at a time |
| 510 | // (dbdoor.RestoreTo), so a person's restore needs one slot. |
| 511 | const ( |
| 512 | dbMaxStreams = 32 |
| 513 | dbMaxStreamsPerPrincipal = 4 |
| 514 | dbStreamIdleDefault = 30 * time.Second |
| 515 | dbStreamRetryAfter = "5" |
| 516 | ) |
| 517 | |
| 518 | type dbStreamLimits struct { |
| 519 | mu sync.Mutex |
| 520 | maxTotal, maxPer int |
| 521 | total int |
| 522 | per map[string]int |
| 523 | } |
| 524 | |
| 525 | func newDBStreamLimits(maxTotal, maxPer int) *dbStreamLimits { |
| 526 | return &dbStreamLimits{maxTotal: maxTotal, maxPer: maxPer, per: map[string]int{}} |
| 527 | } |
| 528 | |
| 529 | // acquire takes a slot for key, or reports that the console or key is full. |
| 530 | func (l *dbStreamLimits) acquire(key string) bool { |
| 531 | l.mu.Lock() |
| 532 | defer l.mu.Unlock() |
| 533 | if l.total >= l.maxTotal || l.per[key] >= l.maxPer { |
| 534 | return false |
| 535 | } |
| 536 | l.total++ |
| 537 | l.per[key]++ |
| 538 | return true |
| 539 | } |
| 540 | |
| 541 | func (l *dbStreamLimits) release(key string) { |
| 542 | l.mu.Lock() |
| 543 | defer l.mu.Unlock() |
| 544 | l.total-- |
| 545 | if l.per[key]--; l.per[key] <= 0 { |
| 546 | delete(l.per, key) |
| 547 | } |
| 548 | } |
| 549 | |
amadan