Paying for Retrieval Across Storage and Shards

A query node disappears. Its replacement can reach the durable database, but its cache is empty. Meanwhile, the service has grown from thousands of documents to millions. Does adding more query nodes restore the latency target, or do those nodes compete for the same remote bytes?

We can answer part of that question on paper. Count dependencies, requests and bytes separately. Then check what a complete result requires when those bytes come from several shards. A small calculated latency is useful evidence about a plan’s assumptions; it is insufficient evidence about end-to-end p99.

These costs cross both paths on the primer’s search-system map. Review its execution and resources vocabulary when distinguishing logical work, storage traffic and request dependencies.

This capstone continues the durable-write and snapshot chapter. Your exit artifact is a resource table, a failure contract and an architecture revision you can defend. Reading time covers the prose; allow extra time for the exercises and your own trace.

Evidence boundary: All workloads, capacities, request plans, shard distances and recovery choices below are synthetic or stipulated. Public mechanisms are attributed to passages inspected on October 1, 2026. No engine benchmark, provider-price comparison, fault injection or learner assessment ran. The rolling documentation is not a pinned implementation of turbopuffer’s in-progress v3 migration.

Grow the workload without changing the answer

Our engineering-document service keeps stable doc IDs, versions, tenant membership and live/deleted state. It now stores a title vector and a body vector, each 768-dimensional float32: 768 × 4B = 3072B = 3KiB. The non-vector payload is 12KiB. Storing it once gives a limited stable-layout total of 12 + 2×3 = 18KiB/document.

The following table changes the scale explicitly. Rates are constant for the calculations, not measured service capacity.

The new requirement is a 150ms end-to-end p99 target, including client/service overhead. In a measured latency distribution, p99 is the value at or below which 99% of requests complete. A mean-model sum does not establish that percentile. We also stipulate that 90% of queries touch 1% of tenants and that nodes can be replaced cold. Tenant skew motivates cache-local routing; it supplies no byte cache-hit percentage by itself.

A dense query still selects exactly one field, body or title, with one vector per document in that field. Rank distinct documents by squared Euclidean distance ascending, then doc ID ascending. The tenant comes from trusted identity context. Live state and query predicates are evaluated at the selected snapshot; a trusted current-policy authorization decision precedes content disclosure or external reranking. These requirements survive growth. Faster I/O cannot fix a wrong version, an ineligible hit or an unauthorized disclosure.

Predict the remote critical path

A request asks remote storage for data. A round groups requests that can start together because their addresses are already known. If the answer to request R1 tells us which address to fetch in R2, R2 is dependent: it cannot start with R1. The critical path is the longest dependency chain that must finish before the answer is ready.

This distinction appears in turbopuffer’s architecture account. It describes fetching centroid information, then fetching selected clusters in parallel. Its fuller cold path also includes storage metadata, filtering indexes and recent unindexed data. The planner trades extra dependent roundtrips against fetching more data in an existing round. The document’s example timings are vendor observations, not constants for our service or for every object-storage class.

Consider two smaller synthetic plans. Plan A discovers one address at a time and transfers only what it needs. Plan B already knows three ranges from compatible local routing metadata and eagerly fetches them together, including unused data. This assumption is what removes the dependencies; sending requests concurrently cannot remove a dependency on an unknown address.

In words, A waits for routing, then candidates, then projection; its request sizes sum to 1MiB. B fetches three known ranges at once and selects locally from 4MiB. Both issue three requests. B reduces dependency latency while spending more bandwidth. Neither plan includes freshness/authorization work unless those inputs are already available under the stated contract.

Stipulate 20ms of latency per round, perfect request parallelism within a round, and one effective aggregate transfer rate shared by its requests. Latency and transfer do not overlap in this toy. Thus:

modeled I/O milliseconds = rounds × 20 + total MiB / aggregate MiB/s × 1000

Giving each parallel request the full aggregate rate would count capacity three times. Real concurrency, latency/transfer overlap, retries and queueing need measurements rather than this formula.

Guided change: lose bandwidth or lose the routing cache

Predict which plan wins at 100MiB/s. Then reduce the aggregate rate to 10MiB/s. Finally restore 100MiB/s but make B’s routing metadata cold: one additional dependent metadata round is required. For this sensitivity case only, its byte size is stipulated negligible, so B keeps 4MiB. Count its two rounds before checking the answer.

Turn document changes into a resource budget

The SlateDB overview offers an accessible object-storage LSM example. A log-structured merge tree, or LSM, buffers key/value changes, persists sorted tables and merges tables through compaction. Its write-ahead log supports crash recovery; its manifest identifies published files and recovery state. This is SlateDB’s documented design, not proof that turbopuffer uses SlateDB.

Those mechanisms explain why raw payload bytes are only one layer. Updating a 3KiB vector may serialize keys/versions, append a log record, change ANN membership, update retained cluster-local summaries, flush tables and later rewrite data during compaction. Stable doc-ID text/tenant postings can remain unchanged when their logical values do. A real tenant or text change still changes those indexes. The identity chapter’s reference trace supplies that distinction.

Keep the layers separate: logical changes; KV-serialized material; WAL/flush/compaction bytes; remote requests/bytes; CPU; and end-to-end latency. The following saved inputs count uncompressed vector/payload material only. Vector-only updates are logical patches; their 3KiB excludes version/key metadata and any required reconstruction of the complete document state.

Use binary units: 1KiB is 1024B; 1MiB is 1024KiB; 1GiB is 1024MiB. Before adding indexes, replicas, compression or metadata, 50,000 documents occupy about 0.8583GiB and 50 million about 858.3069GiB. These are model quantities, not a cache-sizing recommendation.

If every grown-workload update replaces all 18KiB, raw ingest is 2000 × 18 / 1024 = 35.15625MiB/s. Our mixed workload instead uses 25% full replacements and 75% single-vector patches. Its mean is 0.25×18 + 0.75×3 = 6.75KiB/update, or 13.18359375MiB/s. Raw bytes have changed even though the update count has not.

During a 120s indexer pause, 2000 × 120 = 240,000 updates accumulate. A restored indexer serving 2,500 updates/s must also handle the incoming 2,000, leaving 500/s for catch-up. Drain time is 240,000 / 500 = 480s. That capacity assumes the same update-cost mix and constant rates. If service equals arrival rate, the backlog never drains; if it is lower, backlog grows.

The backlog contains 240,000 × 6.75 × 1024 = 1,658,880,000B of stipulated raw material. That is an aggregate across this service, not a per-query tail or a per-namespace backlog. The bytes alone do not determine the recent-write scan cost for a tenant. Index publication, shard/tenant placement and query mode determine what each query must reconcile.

Charge candidate work and projection separately

Partition the corpus into four document shards: disjoint sets with one owner for each document and both of its vector fields. Stipulate that each query reads 1,024 selected-field candidates per shard, at 3KiB/vector. Candidate work totals 4 × 1024 × 3KiB = 12MiB. After global selection, fetch ten 12KiB payloads: 120KiB, or 0.1171875MiB. Together they request 12.1171875MiB/query, before routing, filtering, tail and index metadata.

Candidate bytes buy ranking work; projection bytes buy requested output fields. Sending candidates from every shard does not require projecting every local candidate. Global selection can avoid unnecessary payload fetches, at the possible cost of a dependent projection round. The fixed 1,024 candidate budget describes work, not a proof of exact-neighbor coverage.

At 1,000 queries/s with a stipulated byte-weighted cache hit rate of 95%, expected remote reads are 1000 × 12.1171875 × 0.05 = 605.859375MiB/s. That already exceeds the 512MiB/s aggregate read budget. Writes, compaction and cache bandwidth have additional budgets. Parallel shard reads can finish sooner for one request while saturating a shared resource across many requests.

A byte hit rate measures avoided remote bytes divided by requested bytes. A count of queries, objects or requests that hit cache is a different statistic. Nor does “90% of queries touch 1% of tenants” imply 95% byte hits: those tenants might have large working sets, cold payloads, frequent changes or poor admission/eviction behavior. Measure hit/miss bytes by data type and tenant distribution.

This resource table exposes two different problems. A cold query transfers 12.1171875 / 100 × 1000 = 121.171875ms worth of data. With three 20ms rounds, its modeled I/O alone is 181.171875ms, above the 150ms target. Hypothetically collapsing that to one round while holding bytes fixed gives 141.171875ms, before CPU, authorization, client overhead and queueing.

The smaller sum does not establish feasibility. Holding bytes fixed while removing dependencies is a sensitivity calculation; it does not construct an achievable plan. Projection ordinarily waits for winners. Eager fetching can remove that wait while increasing bytes, as Plan B showed. Measured cold/warm latency distributions, concurrent throughput and failure/recovery behavior must establish whether a concrete plan meets end-to-end p99.

Prove the shard merge before trusting fanout

Why can each shard return only its local document top-k? Assume every required shard responds, each evaluates the same snapshot and eligibility predicate, documents have one owner, and every shard uses the same exact comparable total order.

Take a document omitted from its shard’s exact top-k. At least k distinct eligible documents in that same shard precede it in the total order. They also precede it globally. The omitted document cannot belong to the global top-k. Therefore, union the local document top-k lists, sort by the same order and take k. Tie handling matters: “precede” means (distance, doc_id), not distance alone.

Use this tiny, separate synthetic snapshot S=42. Every row is live, belongs to trusted tenant A and supplies the selected body field’s saved squared distance. The toy asks for k=2; the earlier resource worksheet asks for k=10.

The local top-2 lists are s0 [1,5], s1 [2,6], s2 [3,7], s3 [4,8]. Their sorted union is [2,1,3,4,5,6,7,8]; the global top-2 is [2,1]. Doc1 wins the boundary tie with doc3. The full-table exact oracle gives the same answer. Taking the best result from just one shard has no such proof.

Guided merge failure: a shard disappears

Predict the observed result with s1 missing. Can the coordinator call it complete because it still has two IDs? Also ask for k=20 with every shard present, then query tenant C. Use the full saved table, rather than assuming a result count proves completeness.

The proof has a narrow boundary. Approximate shard searches can omit better documents; exact global merging cannot repair those omissions. Different snapshots can merge different logical database states. If documents overlap across shards, merge needs deduplication and agreement on each document’s version and score. Overlap alone does not defeat local exact distinct-document top-k sufficiency when that agreement and one total order hold.

Taking vector slots is a different contract: vector top-2 [d1,d1] need not contain document top-2 [d1,d2]. Compound or late-interaction scoring needs a defined document score and aggregation before applying the document proof. Our single-field, one-owner contract keeps both ownership reconciliation and multi-vector aggregation out of this exercise.

Lexical scoring has another boundary. BM25 uses corpus statistics such as document frequency and length normalization. If shards compute different shard-local statistics, their raw scores need not represent one global ranking function. Define shared statistics or a different explicit ranking contract before applying a comparable-order merge argument. Parallelism does not make unlike scores comparable.

Define what survives node loss

turbopuffer’s architecture describes NVMe/memory caches and routing toward a node that already has a namespace cached; any query node can serve any namespace. Its Guarantees page identifies object storage as its stateful dependency and says consistency is preferred over availability when that storage is unreachable, with query configuration able to change the tradeoff. These are named product statements, not a universal failure protocol.

Our model keeps committed data and valid published state authoritative; caches are disposable copies. Losing a cache node changes temperature. Recovery loads a valid base and the committed tail needed for the selected snapshot; it does not promote whichever cached file or highest version happens to survive. S3’s documented consistency model provides strong reads after successful object writes and atomic single-key updates, while explicitly excluding atomic updates across keys. The database’s valid publication/recovery boundary is still required.

The trace’s equivalent in prose is precise. A’s body-vector patch commits for doc17, owned by s2, with both named fields remaining under that owner. It advances the document version while leaving logical text/title/tenant values unchanged. The authority must reconstruct its complete state from the patch; the raw-byte worksheet omits that overhead. At I=40/S=42, the query includes the committed position 41 through the tail until a compatible base incorporates it. Reconciliation suppresses the old vector before ranking. Tenant A, live state and the requested tag predicate determine snapshot eligibility across all required shards. Global winner IDs drive projection from those selected versions, followed by the current-policy disclosure check.

Separate recovery cases from that successful trace:

The Query API’s mode-specific contract says strong queries search all unindexed writes; eventual queries search at most 128MiB and can be stale. Its multi-queries share one snapshot. The Guarantees page explicitly says strong sharded queries use one snapshot across shards while eventual queries may reflect different times per shard. Our exact merge exercise assumes the former kind of shared boundary; choosing stale or mixed-time reads changes the answer contract. Neither contract settles ANN fidelity or disclosure permission.

Independent changed-input capstone

Keep 50 million documents, 1,000 queries/s, 2,000 updates/s, the four-shard work model and 120s pause. Now change byte hits to 90%, full replacements to 50% of updates, and restored indexing service to 2,200 updates/s. One required shard is unavailable during a cold replacement. The 150ms p99 target remains.

Before checking, recalculate the complete four-shard workload after restoration, rather than claiming a capacity win from dropping the failed shard. Write its remote reads, mean update bytes, raw ingest, pending raw bytes, net catch-up and drain time. State the response while the shard is missing and the authority used for recovery. If indexing service falls to 2,000 or 1,900 updates/s, say what happens to the backlog.

Then trace one tenant-scoped body-vector update and one filtered body query through durable storage, base+tail reconciliation, index maintenance, shards, global selection, projection and disclosure. Choose an architecture revision, reject one plausible alternative, and name a measurement that would change your decision. A list of product names or the assertion “cache more” does not satisfy this artifact.

Carry the whole decision forward

The series began with a map of search systems, connecting documents and representations to the read/write paths. Identity and layout then determined what moved and what needed a separate projection. Execution and filtering determined which candidate work was justified. Visibility determined which document versions had authority. Growth now makes the bytes, dependencies and failure contract explicit.

Your final artifact should let another engineer reproduce the resource table and follow one update and query without guessing the snapshot, shard ownership, score order or disclosure point. Include the unmet budget and the experiment that could change the architecture. That is a defensible design decision even before a production capacity claim is possible.

Copy these six fields into one decision note. Fill them progressively, attach each chapter’s output, and revise earlier fields when a requirement or assumption changes. Label worksheet calculations as synthetic; mark physical measurements you have not made as unknown.

1. Workload and answer contract: What changed? Record the query/update mix, selected vector field and scoring order, snapshot eligibility, required freshness and current-policy disclosure separately.
2. Identity and update path: Which stable IDs, payloads, postings and ownership references stay or move? Attach the identity chapter’s reference and byte trace.
3. Query plan and coverage: Attach the lexical skip proof and exact eligible dense baseline. Record the chosen paths and cumulative fallback limits. Keep lexical scores and dense distances separate.
4. Authority and recovery: Record the durable boundary, indexed boundary and selected query snapshot (D/I/S), winning versions and tombstones, the retry receipt, and publication/reclamation conditions. Attach the version trace.
5. Resources and failure response: Separate logical, physical and remote cost layers. Attach the resource table, identify required shards, and declare unavailable/stale response semantics.
6. Decision and falsifier: Choose a revision, reject an alternative, retain unmet constraints, and name the measurement and outcome that would reverse the choice while holding workload and result fidelity fixed.

For selected reading, use turbopuffer Architecture for cache-local routing and its cold request dependencies; Query and Guarantees for named freshness, shard-snapshot and object-store dependency scopes; SlateDB’s overview for WAL, sorted files, manifests and amplification; and S3’s consistency section for single-key atomicity and the multi-key boundary. These rolling documents resolve mechanisms and contracts; their example timings do not measure this workload.

Designing Data-Intensive Applications, second edition, by Martin Kleppmann and Chris Riccomini (March 2026), is a study bridge for partitioning and consistency. Database Internals, Alex Petrov (2019), is a bridge for storage and maintenance. Their full texts were not accessed for this series, so these are topic recommendations. The chapter’s merge proof is general reasoning under its declared assumptions. No full DiskANN-method claim or proprietary consensus/availability protocol is needed to defend it.