Thoughtbot Engineer Tess Griffin Unveils Database Sharding Strategies

Sharding Is Hard: Let's Make It Easier! - Tess Griffin — Ruby on Rails

Ruby on Rails

Effective database sharding requires structuring data boundaries around natural operational entities so that complex cross-shard joins and distributed transactions rarely occur.

AI-assisted summary of Sharding Is Hard: Let's Make It Easier! - Tess Griffin — original source. Check the original for full context.

Source published:

Key ideas

  • Sharding becomes necessary when primary database write throughput saturates the largest available hardware, as read replicas can horizontally scale read traffic but do not relieve write bottlenecks. (0:10)
  • Selecting an effective shard key depends on domain boundaries; partitioning by clinic rather than patient or practitioner isolates nearly all transactional writes and joins to a single machine while satisfying regulatory isolation. (4:21)
  • Explicit shard mapping tables introduce a routing lookup dependency but enable zero-disruption tenant rebalancing and dedicated hardware assignment that mathematical hashing cannot provide. (8:27)
  • Database proxies like PgDog can abstract the operational overhead of two-phase commits and scatter-gather fan-outs, but distributed transactions remain inherently slow and fragile compared to clean boundary modeling. (18:52)

Detailed notes

## Background Presented by Tess Griffin (thoughtbot) and published on YouTube by the Ruby on Rails channel on October 1, 2026, this presentation addresses how to scale out when a database outgrows vertical hardware limits. Based on a multi-clinic healthcare application, the talk explores selecting an effective shard key, routing queries via native Rails features or proxies like PgDog, and handling cross-shard joins and transactions under the guiding principle that effective sharding depends on clean boundaries.

## TL;DR When write throughput exhausts vertical scaling and read replicas, horizontal sharding distributes rows across discrete database nodes using a designated shard key. Evaluating a multi-clinic healthcare platform, Griffin demonstrates why sharding on a tenant boundary (`clinic_id`) confines writes, joins, and transactions locally, avoiding performance-degrading scatter-gather operations. Query routing, background jobs, and online resharding can be handled natively inside Rails or delegated to an external database proxy like PgDog. While scatter-gather reads and two-phase commit protocols exist for unavoidable multi-shard operations, their high latency and locking penalties mean true success lies in modeling data boundaries so cross-shard operations remain exceptionally rare.

## Atmosphere & style Griffin delivers a pragmatic, clear-eyed conference presentation that balances dry technical humor—such as examining a generic e-commerce app named 'Napify' or diagnosing databases actively 'on fire'—with rigorous architectural realism. The pacing is reassuring and educational, guiding viewers from the panic of ballooning $500,000 hosting bills through methodical trade-off evaluations of foreign keys and distributed protocols. It feels like an experienced senior engineer demystifying an intimidating rite of architectural passage, trading heroic distributed-systems complexity for grounded, disciplined domain modeling.

## Bulletin - [00:10](https://www.youtube.com/watch?v=yLoqg7UkryU&t=10s) Sharding becomes necessary when primary database write throughput saturates the largest available hardware, as read replicas can horizontally scale read traffic but do not relieve write bottlenecks. - [04:21](https://www.youtube.com/watch?v=yLoqg7UkryU&t=261s) Selecting an effective shard key depends on domain boundaries; partitioning by clinic rather than patient or practitioner isolates nearly all transactional writes and joins to a single machine while satisfying regulatory isolation. - [08:27](https://www.youtube.com/watch?v=yLoqg7UkryU&t=507s) Explicit shard mapping tables introduce a routing lookup dependency but enable zero-disruption tenant rebalancing and dedicated hardware assignment that mathematical hashing cannot provide. - [18:52](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1132s) Database proxies like PgDog can abstract the operational overhead of two-phase commits and scatter-gather fan-outs, but distributed transactions remain inherently slow and fragile compared to clean boundary modeling.

## Key takeaways - [00:10](https://www.youtube.com/watch?v=yLoqg7UkryU&t=10s) Vertical scaling hits hard physical and financial limits where larger instances either do not exist or become cost-prohibitive, forcing systems to partition data across multiple machines. - [00:10](https://www.youtube.com/watch?v=yLoqg7UkryU&t=10s) Read replicas fail to resolve write saturation because every replica maintains a complete copy of the data and must replicate every write processed by the primary instance. - [03:08](https://www.youtube.com/watch?v=yLoqg7UkryU&t=188s) Queries lacking the designated shard key trigger scatter-gather execution across all shards, multiplying database load and eliminating the performance advantages of sharding. - [04:21](https://www.youtube.com/watch?v=yLoqg7UkryU&t=261s) Evaluating candidate shard keys requires analyzing foreign key relationships; selecting an entity that does not bound related tables results in constant cross-shard joins. - [04:21](https://www.youtube.com/watch?v=yLoqg7UkryU&t=261s) Replicating static or global reference tables across every shard preserves local join capabilities without leaking foreign key lookups across network boundaries. - [08:27](https://www.youtube.com/watch?v=yLoqg7UkryU&t=507s) Computed modulo hashing makes fleet expansion difficult because changing the divisor forces a widespread rebalancing of tenants across every database instance. - [08:27](https://www.youtube.com/watch?v=yLoqg7UkryU&t=507s) Storing shard assignments in an explicit mapping table treats tenant placement as mutable data, allowing operators to isolate noisy tenants onto dedicated hardware via simple row updates. - [11:28](https://www.youtube.com/watch?v=yLoqg7UkryU&t=688s) Production data migration pipelines require careful rate throttling across historical backfills and change-data-capture catch-up replication to avoid crashing the live primary database. - [12:33](https://www.youtube.com/watch?v=yLoqg7UkryU&t=753s) Native Rails sharding supports multi-database routing via `connects_to`, but ad-hoc usage causes shard-routing logic to sprawl uncontrollably across controllers and services. - [14:49](https://www.youtube.com/watch?v=yLoqg7UkryU&t=889s) Background jobs should centralize shard switching within base class lifecycle hooks like `around_perform`, while queue backend tables must remain on a centralized, unsharded database. - [18:12](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1092s) Cross-shard relational joins cannot occur within a single database engine and require scatter-gather coordination to query instances in parallel and merge result sets in memory. - [18:52](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1132s) PostgreSQL supports two-phase commits via `PREPARE TRANSACTION`, but coordinator crashes before the commit phase leave orphaned row locks holding database resources indefinitely. - [18:52](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1132s) While database proxies can manage distributed two-phase commit recovery protocols, they cannot eliminate the latency and blocking costs inherent to multi-node network transactions. - [21:44](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1304s) Successful sharding is fundamentally a data modeling problem: establishing clean tenant boundaries eliminates the need to rely on complex distributed transactions.

## Detailed walkthrough ### [00:10](https://www.youtube.com/watch?v=yLoqg7UkryU&t=10s) The Limits of Vertical Scaling and When to Shard

Tess Griffin, a senior developer at thoughtbot, opens the talk by diagnosing a classic infrastructure nightmare: an application that has lived comfortably on a single database for years, growing steadily until routine vertical scaling hits a wall. Teams instinctively keep upgrading the hardware—making the box "bigger and bigger and bigger"—until the hosting bill tops $500,000 a month and the database still buckles under the load. In this state, once-instantaneous queries crawl, the slightest traffic spike threatens to set the entire system on fire, and further vertical scaling is off the table simply because larger instances either do not exist or cost an impossible fortune.

The first step in triage is measurement. Teams need to interrogate Postgres directly to understand whether the workload is read-heavy or write-heavy, identify which specific tables absorb the bulk of the write volume, and inspect which tables have grown massive. If read traffic is the culprit, there is an easier, cheaper remedy: read replicas. A replica provides a live copy of the dataset to handle read queries, allowing reads to scale horizontally simply by provisioning more servers. However, standard SQL architectures deliberately trade slower writes for faster reads; every read replica still maintains a full copy of every row, and the primary database alone must process every write. When massive tables bog down every associated read, write, and index, and write throughput has saturated the largest available box, vertical scaling is exhausted.

Horizontal scaling through sharding is the solution. Rather than relying on a single monolithic machine, sharding breaks a single database into multiple distinct databases, spreading rows across them so that every individual machine handles only a slice of total writes and stores only a fraction of the total dataset.

- **Vertical Scaling Limitations**: Upgrading hardware eventually reaches physical or financial limits where "the next size up doesn't exist." - **Read Replicas**: Distribute read load effectively, but do nothing to alleviate primary write bottlenecks since every write still routes to one primary. - **The Core Mechanic of Sharding**: Splitting rows across distinct machines using a dedicated column value—the **shard key**.

---

[03:08](https://www.youtube.com/watch?v=yLoqg7UkryU&t=188s) The Shard Key and the Ideal Case: "Napify"

To shard successfully, an application must determine how rows are distributed among instances by designating a specific column as the shard key. The value of this column dictates where a row lives. When incoming queries include the shard key, the database layer routes them directly to the single relevant shard. Conversely, if a query lacks the shard key, it must scatter across all shards simultaneously: a single query becomes three or more, every machine halts to evaluate it, and most return nothing. Scatter-gather queries quickly destroy any performance gains sharding was supposed to provide.

```mermaid flowchart TD accTitle: Shard Key Query Routing accDescr: Routing behavior comparing a query with a shard key targeting one shard versus a keyless query broadcasting to all shards.

subgraph DirectRouting["Targeted Query (With Shard Key)"] Q1["Query with Shard Key"] --> S1["Target Shard"] end

subgraph ScatterRouting["Scatter-Gather (Without Shard Key)"] Q2["Query without Shard Key"] Q2 --> SA["Shard A (No Match)"] Q2 --> SB["Shard B (Row Found)"] Q2 --> SC["Shard C (No Match)"] end ```

Griffin introduces a clean baseline scenario dubbed "Napify" (noting with a grin that "any resemblance to other storefront platforms is coincidental"). An e-commerce platform represents an ideal sharding problem because the domain boundaries are naturally hierarchical: almost every database table hangs cleanly off of a shop. Drawing a conceptual boundary around a single `shop_id` captures virtually all related transactional records without any messy dependencies spilling outside the line. In that architecture, `shop_id` serves as a clean shard key, while cross-cutting concerns like admin users, billing tiers, and feature flags comfortably remain behind on a shared global primary database.

- **Targeted Query**: Contains the shard key; executes against exactly one machine. - **Scatter-Gather Query**: Lacks the shard key; executes across every shard in parallel, multiplying load and nullifying the benefits of sharding. - **Clean Tenancy**: In platforms like Napify, drawing a boundary around a single shop isolates almost all operational tables with zero external leaks.

---

[04:21](https://www.youtube.com/watch?v=yLoqg7UkryU&t=261s) Evaluating Shard Keys for a Healthcare App

The talk shifts to a messy real-world problem: an application called "TheirChart," an original platform serving hundreds of clinics from a single Postgres database that is currently "on fire." The application never planned for sharding, and its two biggest operational bottlenecks are the `visits` and `audit_logs` tables. Because the platform must maintain HIPAA compliance, every access of sensitive data generates an audit entry, causing `audit_logs` to swell uncontrollably. To resolve the bottleneck, the chosen shard key must split both `visits` and `audit_logs`. Because a table can only be partitioned by a column it actually contains, Griffin evaluates the four ID columns present on a `visit`: `patient_id`, `practitioner_id`, `visit_id`, and `clinic_id`.

```mermaid flowchart LR accTitle: Visit Table Foreign Key Candidates accDescr: The four candidate keys evaluated for sharding the visits and audit logs tables.

V["Visit Record"] V --> P["patient_id"] V --> PR["practitioner_id"] V --> VI["visit_id"] V --> C["clinic_id (Chosen Key)"] ```

The speaker tests each potential candidate by making a conceptual cut through the schema to see what breaks: 1. **Patient (`patient_id`)**: While intuitively tempting because the application revolves around patients, patients often visit multiple clinics. Cutting across patients leaves half of the application's queries unable to route effectively, while other operations do not carry a patient context at all, triggering constant scatter-gather queries. 2. **Practitioner (`practitioner_id`)**: Practitioners often work across multiple clinics, and very few ancillary tables relate directly to practitioner records. A key that connects to almost nothing leaves most queries stranded with no shard routing information. 3. **Visit ID (`visit_id`)**: Sharding directly on the primary key of the burning table would evenly distribute visit writes. However, a visit owns almost nothing and touches everything; almost all contextual details live in other tables, forcing every subsequent query to perform expensive cross-shard joins. 4. **Clinic (`clinic_id`)**: Sharding on the clinic isolates writes cleanly within bounded boundaries. Almost every write generated by a clinic stays inside its own silo, meaning joins and transactions remain simple, single-database operations on one machine.

To handle the few cross-cutting edges with clinic sharding, remedies—which are universal across all clinics—are replicated across every shard so queries can join against them locally. Meanwhile, global administrative functions like billing remain on the primary database. Furthermore, this boundary aligns neatly with real-world healthcare compliance: because each clinic operates as an independent covered entity, keeping patient records isolated per clinic satisfies legal privacy boundaries while providing blast-radius containment if a shard fails.

- **Blast Radius Protection**: If a shard suffers downtime or security compromise, the fallout is restricted entirely to that shard's clinics rather than crashing the whole platform. - **Replicated Tables**: Static reference tables like `remedies` are copied to all shards to ensure joins remain local. - **Primary Data**: Global data disconnected from clinic workflows (such as billing) stays centralized on the primary.

---

[08:27](https://www.youtube.com/watch?v=yLoqg7UkryU&t=507s) Assigning Shards: Computed Hashes vs. Stored Shard Maps

Once `clinic_id` is chosen, the architecture faces two distinct implementation requirements: migrating the data onto individual shards and routing subsequent queries accurately. These tasks can be handled within Rails itself or delegated to an external SQL proxy such as PgDog for Postgres, or Vitess / PlanetScale for MySQL. Before any data can migrate, however, every clinic must be assigned a home shard. Systems accomplish this either by computing the mapping on the fly or storing it in a routing table.

```mermaid flowchart TD accTitle: Shard Assignment Approaches accDescr: Comparison between computed hash routing and stored shard map routing.

subgraph Computed["Approach 1: Computed (Hashing)"] CID1["clinic_id"] --> HASH["clinic_id % n"] HASH --> SHARD1["Shard N"] end

subgraph Stored["Approach 2: Stored Shard Map"] CID2["clinic_id"] --> MAP[("shard_map Table")] MAP --> SHARD2["Assigned Shard (Explicit)"] end ```

The computed approach runs the `clinic_id` through a hashing function, such as a naive modulo operation (`clinic_id % n`, where `n` is the total shard count). Griffin cautions that simple modulo hashing is ill-advised in production because altering `n` reshuffles virtually every clinic across different shards. While more sophisticated consistent hashing algorithms exist—Griffin points attendees toward Martin Kleppmann's book, *Designing Data-Intensive Applications*, for an in-depth treatment—any computed approach leaves placement entirely to the mathematical function. Teams cannot manually rebalance or isolate an individual tenant.

The alternative approach stores the shard mapping in an explicit database table with one row per clinic. When a `clinic_id` is provided, the table returns its designated shard. Storing the mapping treats placement as mutable data, granting operators full control: - **Tenant Isolation**: A massive clinic overwhelming its neighbors can be moved to a dedicated hardware shard with a single row update. - **Architectural Trade-Off**: Every database lookup must query this mapping table first. If the mapping table goes down, access to all underlying shards fails. - **Practical Mitigation**: The shard map table is tiny, reads are easily cached, and writes occur infrequently (only when clinics are onboarded or migrated).

---

[11:28](https://www.youtube.com/watch?v=yLoqg7UkryU&t=688s) Migrating Data: DIY Pipelines vs. Proxy-Based Resharding

Moving live production data onto newly partitioned shards presents substantial operational risk. Doing the migration in-house requires building custom tooling: developers must backfill existing records, run catch-up replication to capture writes that landed during the backfill process, and orchestrate an atomic cutover. The copy operations must be throttled carefully; otherwise, the migration itself can exhaust database capacity and bring down production.

Alternatively, teams can offload this operational burden to a database proxy like PgDog. With PgDog, engineers configure a Postgres publication and initiate a reshard command. The application continues running and serving traffic while PgDog handles backfilling, change data capture catch-up, and cutover behind the scenes. The foundational migration steps remain identical, but using a proxy transfers the execution risk and automation to battle-tested infrastructure.

- **DIY Migration Steps**: Backfill historical data, replicate live catch-up writes, and coordinate final cutover—all with rate throttling to prevent database failure. - **Proxy Automation**: Running PgDog's resharding provisions empty shard databases and populates them using Postgres publications while the application operates uninterrupted.

---

[12:33](https://www.youtube.com/watch?v=yLoqg7UkryU&t=753s) Routing Queries: Rails Native Sharding and the Sprawl Problem

When an incoming request hits the stack, the system must decide which shard to query. This routing decision can either live at the proxy layer—where the application connects to what appears to be a single standard database while the proxy routes the SQL statements—or directly inside Rails. Rails has supported multiple databases since version 6.0 and introduced native sharding support in Rails 6.1, functioning identically on Postgres and MySQL.

Configuring Rails begins in `config/database.yml`, where each shard is defined with its own distinct connection pool and dedicated read replica. In the domain layer, an abstract application record binds the shards together using Rails' `connects_to` directive, configuring both a `writing` role and a `reading` role for every shard. Tenant-scoped models inherit from this base class, while universally shared reference tables (`remedies`) and global services (`billing`) retain their respective configurations.

To execute queries against a tenant, Rails relies on a clean connection-switching primitive:

```ruby shard_name = ShardMap.shard_for(clinic_id) ActiveRecord::Base.connected_to(shard: shard_name, role::writing) do # All queries inside this block execute against the clinic's shard @clinic.visits.all end ```

Rails answers only one specific question: *Which shard do I talk to?* However, as Griffin highlights, this introduces a subtle architectural challenge: code sprawl. While the shard key is technically just a single column value, the moment an application shards, every query requires that context. Developers find themselves wrapping database access in `connected_to` blocks inside the `VisitsController`, repeating the pattern across background workers like `ReminderJob`, and watching shard routing logic spread throughout the codebase.

- **Rails Version Milestones**: Multiple database support arrived in Rails 6.0; first-class sharding arrived in Rails 6.1. - **Connection Configuration**: Configured via `database.yml` and wired using `connects_to(shards: {... })` on an abstract model. - **Context Creep**: Sharding requires the shard key everywhere; without strict abstraction boundaries, `connected_to` blocks quickly proliferate across controllers, service objects, and background jobs.

[14:49](https://www.youtube.com/watch?v=yLoqg7UkryU&t=889s) Routing Queries and Background Jobs

When preparing an application for sharding, the biggest architectural hazard is letting routing logic bleed across the entire codebase. If every controller, model, or service has to independently figure out which shard a clinic belongs to, developers are guaranteed to miss places. The cardinal rule of sharding is containment: keep routing in exactly one place. Out of the box, Ruby on Rails provides built-in configuration for multi-database and sharded environments, establishing a single source of truth that knows how to map a request to a shard.

For teams wanting to take separation of concerns a step further, query routing can be moved entirely outside the application layer and handed over to an intelligent database proxy such as PgDog. With a proxy, the Rails application simply passes the shard key, and the proxy takes care of connection routing under the hood. There is no in-app shard map to maintain and no resolver to write. As far as the Rails code is concerned, it goes back to believing it is speaking to a single, standard database while still reaping the horizontal scaling benefits of sharding.

```mermaid flowchart TD accTitle: Shard Routing in Rails Application vs Database Proxy accDescr: Diagram contrasting in-app shard routing via Rails connected_to with proxy-based routing via PgDog.

subgraph AppApproach["Application-Level Routing (Rails)"] direction TB A["Incoming Request / Job"] --> B["Rails Shard Resolver / Map"] B -->|"ActiveRecord::Base.connected_to"| C[("Target Shard")] end

subgraph ProxyApproach["Proxy-Level Routing (e.g., PgDog)"] direction TB D["Incoming Request / Job"] --> E["Set Shard Key (e.g. SET LOCAL)"] E --> F["Database Proxy"] F --> G[("Target Shard")] end ```

The exact same discipline must be applied to background processing. Rather than scattering shard-awareness across dozens of workers, background jobs should be contained through `ApplicationJob`: - **Job Arguments:** Pass the routing key (`clinic_id`) explicitly into the job arguments. - **Single Hook in ApplicationJob:** Implement an `around_perform` callback in `ApplicationJob` that extracts the `clinic_id` and wraps execution in ActiveRecord’s `connected_to(shard:...)`. - **Fail Fast:** If an engineer enqueues a job without a clinic identifier, the base class immediately raises an error instead of letting the job blindly execute against the wrong shard or fallback database. - **Proxy Pattern with SQL:** When using a database proxy, the worker connects to the standard proxy endpoint and sets the sharding key within a transaction using `SET LOCAL sharding_key =...` in the `around_perform` block. - **Queue Database Placement:** Queue backends like Solid Queue are database-backed and must never be sharded across tenant shards; their tables should remain on the primary database or inside an independent, dedicated database (the same isolation principles apply when utilizing Redis-backed queues like Sidekiq).

[17:16](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1036s) Resharding and Handling Growth

A sharding architecture must anticipate the day when the original set of shards runs out of capacity. How an application handles expansion depends heavily on how tenant allocation was designed from the beginning. If the system relies on a mathematical hashing function to determine shard placement, changing the modulus or hash algorithm forces widespread data migration across the entire fleet.

By contrast, maintaining an explicit shard mapping table requires more work up front, but it "pays for itself the first time you need to add more shards." In an explicit lookup model, when an individual tenant becomes massive, an engineer can spin up a dedicated database shard, point that specific clinic’s ID to the new shard in the map table, and migrate only that clinic’s records without disturbing neighboring tenants. When using a database proxy, resharding can often occur dynamically while the application continues serving live traffic, avoiding application-level mapping bugs because the Rails app never managed an explicit mapping table to begin with.

[18:12](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1092s) Cross-Shard Joins and Scatter-Gather

Because choosing a good shard key keeps all of a single clinic's related records co-located on the same physical database instance, the overwhelming majority of relational SQL joins execute locally and never cross shard boundaries. However, certain queries—such as cross-organization reporting, administrative analytics, or global aggregations—unavoidably span multiple databases.

Because a database engine cannot perform a relational join across separate physical database instances in a single round trip, applications must resort to the "scatter-gather" pattern: - **Scatter:** The application loops through every shard in the fleet and dispatches the query to each database in parallel or sequence. - **Gather:** The application collects the independent result sets into memory and aggregates, sums, or merges the answers within Ruby. - **Proxy Fan-Out:** An intelligent proxy can handle scatter-gather transparently. The developer writes a single standard SQL query; the proxy fans the read out to every shard across the cluster, merges the rows or aggregate totals, and returns a single unified result set to Rails.

[18:52](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1132s) Cross-Shard Transactions and Two-Phase Commit

While distributed reads can be aggregated with scatter-gather, cross-shard writes present a much harsher architectural challenge. On a single database node, transactions are strictly atomic: a write either completely succeeds or completely fails as an indivisible unit. Once data is split across machines that communicate over a network, true atomicity in the traditional sense disappears due to the fundamental constraints of distributed systems. Engineers face an unavoidable choice: either block and wait for every single machine to agree before returning, or return an immediate answer and let records catch up asynchronously later.

```mermaid flowchart TD accTitle: Two-Phase Commit Protocol Flow accDescr: Step-by-step visualization of the Prepare and Commit phases in a two-phase commit protocol across multiple database shards.

subgraph Phase1["Phase 1: Prepare"] C1["Coordinator"] -->|"PREPARE TRANSACTION"| S1["Shard 1 (Locks Row)"] C1 -->|"PREPARE TRANSACTION"| S2["Shard 2 (Locks Row)"] S1 -->|"Promise: OK"| C1 S2 -->|"Promise: OK"| C1 end

subgraph Phase2["Phase 2: Commit"] C1 -->|"COMMIT PREPARED"| S1 C1 -->|"COMMIT PREPARED"| S2 end ```

When multi-shard consistency is strictly required, the system must utilize a two-phase commit (2PC) protocol: - **Phase 1 (Prepare):** The coordinator asks every participating shard whether it can execute the operation. Each shard locks the required rows, reserves state, and returns an explicit promise that it can commit. - **Phase 2 (Commit):** Only if every single shard responds affirmatively does the coordinator issue the final commit instruction. If even one shard fails or times out, the coordinator instructs all shards to roll back. - **The Cost of 2PC:** Two-phase commit is an inherently blocking protocol where the entire transaction moves at the speed of the slowest participating shard. - **Coordinator Failure and Orphaned Locks:** PostgreSQL provides built-in support for `PREPARE TRANSACTION`, which allows prepared state to survive database crashes. However, PostgreSQL will not automatically clean up prepared transactions if the coordinating application crashes midway through the handshake. If the coordinator dies before issuing the commit, participating shards are left stranded holding active row locks indefinitely. - **Hand-Rolling vs. Proxies:** Hand-rolling a 2PC coordinator requires writing extensive recovery plumbing, health checkers, and cleanup daemons to detect and resolve stranded locks. A database proxy can execute the two-phase commit protocol automatically behind a standard transaction interface, saving developers from building distributed transaction machinery from scratch.

Even with a proxy handling the protocol, the underlying physical costs remain: locks are held across network round trips, and latency is high. The proxy enforces an all-or-nothing outcome across shards, but "what it can't give you is atomicity. That left when data split." The only sustainable strategy is to choose a sharding key that ensures cross-shard writes account for only an infinitesimal fraction of overall application traffic.

[21:44](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1304s) The Core Lesson: Good Sharding Is Good Boundaries

A database proxy can hide the operational friction of distributed architecture, but it cannot bend the laws of physics. Cross-shard joins remain fundamentally slower than local joins, and two-phase commits remain inherently fragile and resource-heavy. Winning at sharding does not mean heroically engineering your way through cross-shard complexities; success comes from structuring data boundaries so that complex cross-shard edge cases rarely happen in the first place.

When using native Rails sharding, the application team directly owns and maintains every single moving part: schema migrations, shard lookup mappings, routing resolvers, scatter-gather aggregations, and distributed transactions. Introducing a database proxy relieves the application of that plumbing burden, allowing Rails to treat the database cluster as a single logical entity while running resharding workflows online.

Ultimately, sharding is won or lost at the data modeling phase: - Do not attempt to defeat sharding by "working harder on the hard parts." - Arrange models and foreign keys so that each tenant's entire operational footprint lives self-contained in one physical location. - Shard on the natural boundary that is easiest to cleanly separate. - "Good sharding is good boundaries."

## Notable quotes > [00:10](https://www.youtube.com/watch?v=yLoqg7UkryU&t=10s) bigger and bigger and bigger — Tess Griffin

> [00:10](https://www.youtube.com/watch?v=yLoqg7UkryU&t=10s) the next size up doesn't exist. — Tess Griffin

> [03:08](https://www.youtube.com/watch?v=yLoqg7UkryU&t=188s) any resemblance to other storefront platforms is coincidental — Tess Griffin

> [04:21](https://www.youtube.com/watch?v=yLoqg7UkryU&t=261s) on fire — Tess Griffin

> [12:33](https://www.youtube.com/watch?v=yLoqg7UkryU&t=753s) Which shard do I talk to? — Tess Griffin

> [17:16](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1036s) pays for itself the first time you need to add more shards. — Tess Griffin

> [18:52](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1132s) what it can't give you is atomicity. That left when data split. — Tess Griffin

> [21:44](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1304s) working harder on the hard parts. — Tess Griffin

> [21:44](https://www.youtube.com/watch?v=yLoqg7UkryU&t=1304s) Good sharding is good boundaries. — Tess Griffin

## Mentioned Tess Griffin, thoughtbot, Ruby on Rails, Active Record, PostgreSQL, MySQL, PgDog, Vitess, PlanetScale, Martin Kleppmann, Designing Data-Intensive Applications, Solid Queue, Sidekiq, Shopify, HIPAA

## Video description When your database has outgrown the biggest box you can buy, sharding is the next step. Using a multi-clinic healthcare app as the example, Tess Griffin (thoughtbot) covers how to choose a shard key and how to route queries using Rails' built-in sharding or a proxy like PgDog. It also covers what to do about cross-shard joins and transactions. The core lesson is that good sharding comes down to good boundaries.

*** Thank you Shopify for sponsoring the editing and post-production of these videos. Check out insights from the Engineering team at: https://shopify.engineering/

#RailsWorld #RubyOnRails #Postgres #Sharding #Databases #Scaling

[Source](https://www.youtube.com/watch?v=yLoqg7UkryU)

中文简报

Thoughtbot工程师Tess Griffin解析数据库分片策略 高效的数据库分片需要围绕天然业务实体构建数据边界,从而极力避免复杂的跨分片连接与分布式事务。 当写入吞吐量耗尽了垂直扩容和只读副本的潜力时,水平分片可以通过指定的分片键将数据行分布到各个独立的数据库节点上。Griffin以一个多诊所医疗平台为例,阐明了为何按租户边界(`clinic_id`)进行分片能够将写入、连接和事务局限在本地节点,从而避免导致性能下降的分散-聚集操作。查询路由、后台作业和在线重新分片既可以在Rails内部原生处理,也可以委托给如PgDog等外部数据库代理。尽管针对不可避免的多分片操作存在分散-聚集读取和两阶段提交协议,但它们带来的高延迟与锁开销意味着,分片真正成功的关键在于合理建模数据边界,使跨分片操作极为罕见。 当主数据库的写入吞吐量使最大可用硬件饱和时,分片就变得必不可少,因为只读副本虽能水平扩展读取流量,却无法缓解写入瓶颈。 (0:10) 选择有效的分片键取决于业务领域边界;按诊所而非患者或从业人员进行分区,能将几乎所有的事务性写入和连接隔离在单台机器上,同时满足合规层面的隔离要求。 (4:21) 显式分片映射表虽然引入了路由查找依赖,但能够实现数学哈希无法提供的无中断租户再平衡和专用硬件分配。 (8:27) 像PgDog这样的数据库代理可以抽离两阶段提交和分散-聚集扇出的运维开销,但与清晰的边界建模相比,分布式事务本质上仍然缓慢且脆弱。 (18:52)

Source quotes

[0:10] Tess Griffin: bigger and bigger and bigger [0:10] Tess Griffin: the next size up doesn't exist. [3:08] Tess Griffin: any resemblance to other storefront platforms is coincidental [4:21] Tess Griffin: on fire [12:33] Tess Griffin: Which shard do I talk to? [17:16] Tess Griffin: pays for itself the first time you need to add more shards. [18:52] Tess Griffin: what it can't give you is atomicity. That left when data split. [21:44] Tess Griffin: working harder on the hard parts. [21:44] Tess Griffin: Good sharding is good boundaries.

Original transcript

[0:06] [music] [0:10] Hi everyone. This is sharting is hard. [0:14] Let's make it easier. [0:16] My name is Tess Griffin and I'm a senior [0:19] developer at ThoughtBot. Now, [0:22] let's say your app has had one database [0:25] for years and as you grew, you did the [0:28] obvious thing. You made [snorts] it [0:29] bigger and bigger and bigger and now [0:33] you're in for a bill is $500,000 a month [0:38] and it's not even keeping up. Queries [0:41] that used to be instant now crawl a [0:44] light breeze and it falls over. [0:48] You get a spike in traffic and suddenly [0:50] your entire database is on fire and you [0:53] have no idea why. And you can't buy a [0:56] bigger box. it doesn't exist or is too [0:59] expensive. What now? [1:02] The first thing you want to do is [1:04] measure. You want to determine why the [1:08] database is falling over. [1:10] You can ask Postgress itself. [1:13] Are we read heavy or write heavy? Which [1:17] tables take the rights? [1:21] Which tables got huge? If there are a [1:24] few big tables taking most of those [1:27] writes, those are the tables you'll want [1:30] to target. [1:32] If the reads are slow, there's a cheaper [1:35] fix. Most read load can leave the [1:38] database entirely. [1:41] A replica is a live copy of your [1:43] database that serves reads. Writes still [1:46] go to one place, the primary. [1:50] reads scale by adding boxes, but every [1:53] replica still holds every row and the [1:56] primary still takes every right. [2:00] A single database has one primary and [2:03] that primary takes every right. SQL [2:06] databases make this trade on purpose. [2:09] They trade slower writes, so reads can [2:12] be fast. If it's the rights that are [2:14] slow, there's nowhere else to send them [2:16] with one database. [2:19] A table can get so big that everything [2:22] it touches slows down. Reads, writes, [2:25] even indexes get slow. [2:28] We already tried one fix, vertical [2:31] scaling, a bigger box until the next [2:34] size up doesn't exist. [2:37] The other direction is horizontal, more [2:40] machines. Nothing has to be bigger. [2:43] There are just more of them. [2:46] Sharding is horizontal scaling applied [2:49] to your database. It can fix both slow [2:52] writes and giant tables [2:56] one database into several and spread the [2:59] rows across them. Every machine takes a [3:02] slice of the rights and holds a slice of [3:05] the rows. To split, you have to decide [3:08] which rows go to which machine. You pick [3:12] one column and that column's value [3:15] decides where each row lives. [3:23] So that column is your shard key. [3:27] A query that carries the shard key goes [3:29] to one shard. A query that doesn't goes [3:33] to all of them. [3:35] One query becomes three. Every machine [3:38] stops to answer and two of them had [3:40] nothing. [3:41] So each of those and sharding has bought [3:44] you nothing. [3:46] Meet Napify. Any resemblance to other [3:49] storefront platforms is coincidental. [3:54] This is an ideal case for sharding. A [3:57] storefront platform. Every table hangs [4:00] off of the shop. [4:03] Draw a circle around one shop and [4:05] nothing sticks out. The shard key is [4:09] shop ID. And the things outside of your [4:12] box, admins, plans, feature flags, they [4:16] don't belong to any shop. Those stay [4:18] behind on your primary. [4:21] And this is our app. It is currently on [4:24] fire. [4:26] It's not ideal. [4:29] It's called their chart. Also completely [4:32] original platform. Hundreds of clinics, [4:35] one Postgress database. [4:39] Nobody considered sharding before now. [4:41] Visits and audit logs are the tables [4:44] currently falling over and audit logs [4:48] are huge for us because we want to be [4:50] HIPPA compliant and so every access of [4:53] data requires a log. This is the schema [4:57] that we need to shard. [5:00] We need to determine the shard key. The [5:03] key has to split the biggest tables on [5:05] fire which are visits and audit logs. [5:08] And a table can only be split by a [5:11] column it carries. A visit has four IDs [5:15] on it. The patient, the practitioner, [5:18] the clinic, and its own. Those are our [5:21] candidates. For each one, we'll try to [5:26] cut and see what breaks. [5:29] The tempting shard key is the patient, [5:32] the thing the whole app is about. But [5:34] look at what breaks. Every red line is a [5:39] query without the key, which visits [5:41] every shard. [5:44] The patients belong to more than one [5:46] clinic. So half your queries wouldn't [5:49] know which patient to route by and the [5:51] other half don't carry a patient at all. [5:55] What about the practitioner? It's the [5:58] same kind of problem. Practitioners work [6:00] at more than one clinic and almost [6:03] nothing connects to a practitioner. [6:06] If almost nothing connects to your key, [6:09] almost no query knows which shard to go [6:11] to. [6:14] What if we shard on the burning table [6:16] itself? [6:17] Splitting by visit ID spreads the fire [6:20] evenly. And some places do do this. But [6:25] for our case, a visit owns almost [6:28] nothing and touches everything. [6:32] Everything for the visit lives somewhere [6:34] else. Every query now crosses shards. [6:39] You can split the brewing table and [6:41] still not pick the the best shard key. [6:47] shard on the data that's easiest to [6:50] separate for their chart. That's the [6:53] clinic. Every right a clinic makes stays [6:57] inside the circle. [6:59] One edge crosses treatments point at [7:03] remedies. [7:04] But remedies are the same for every [7:06] clinic. So what you can do is copy the [7:09] remedies to every shard. So that way [7:12] your queries still stay local. billing [7:16] doesn't really have anything to do with [7:18] the clinics. So that can stay on the [7:20] primary. [7:22] So the fire splits with this cut. So [7:28] audit logs and visits carry clinic ID. [7:32] So each shard only takes its own audit [7:36] logs and visits which spreads out the [7:39] fire to our new shards. [7:43] Everything a clinic does happens on one [7:46] machine. Joins work. Transactions are [7:49] plain single database transactions. If a [7:52] query misfires or a shard is [7:54] compromised, damage stops at that [7:57] shard's clinics, not all of them. [8:01] So patients can belong to more than one [8:05] clinic, but we want to keep their [8:07] records separate. Each clinic is its own [8:11] covered entity. So it keeps its own [8:14] patient records. [8:18] If you need true single tenant isolation [8:21] for somebody, you can give them a shard [8:23] just to themselves. [8:27] So you've chosen clinic ID as the shard [8:30] key. Now we have a couple jobs left. We [8:33] need to move the data onto the shards [8:36] and we need to route every query to the [8:38] correct shard. You can do this in Rails [8:41] or you can use a proxy. [8:45] A proxy is a box that sits between your [8:49] app and the database and speaks plain [8:52] SQL. So your app still talks to one [8:55] thing while the proxy spreads the data [8:58] across the shards. In Postgress there is [9:03] PG Dog which is an open source proxy on [9:06] my SQL there is the test and planet [9:11] scale is the managed version. So your [9:14] database.yaml and rails points at it [9:17] instead of at Postgress. [9:22] So before any data moves every clinic [9:26] needs an assigned shard. There are two [9:29] ways you can assign the shard. You can [9:31] compute it or you can store it. [9:36] So the first option is to compute it. [9:38] You run the clinic ID through a hashing [9:42] function. So this first example is [9:46] simply clinic ID modulo n. N being the [9:51] number of shards that you've spun up. [9:54] likely don't actually want to use this [9:56] one in production because if you change [9:59] n then every clinic will move to a [10:03] different shard. [10:05] But whatever you choose the hashing [10:08] function to be the mapping [snorts] [10:10] works the same way. The function decides [10:13] where the clinic lives, not you. You [10:16] can't move just one clinic. [10:20] The other option is to store it a table [10:24] with one row per clinic. A clinic goes [10:28] in and the shard comes out. The mapping [10:31] is data now. And that's the point. You [10:34] choose the placements. Do you have a [10:36] huge clinic dragging down its whole [10:39] shard? Move it to a shard of its own [10:42] with one update. [10:44] The trade-off would be now every lookup [10:48] goes through this one table. If for some [10:51] reason this table is unreachable, then [10:53] every shard behind it is also [10:55] unreachable, [10:58] but it's tiny and it only gets written [11:01] to when a clinic is added or moved. For [11:04] the rest of the code samples, we'll be [11:06] using this stored shard map. [11:12] For an in-depth explanation of different [11:15] hashing functions you can use, I [11:18] recommend Martin Kleman's book, [11:20] Designing Data inensive Applications. [11:23] For our examples, we're going to be [11:25] storing the mapping. [11:28] So every clinic has a shard now, but we [11:31] need to actually move the data. [11:35] You can do this all yourself. It's [11:37] definitely a lot of work. You have to [11:41] backfill the rows, catch up the rice [11:44] that did that, that the rights that came [11:46] in while you did the back fill, and then [11:49] you cut over. Each one is tricky. The [11:54] copy has to be throttled or the copy [11:56] itself can take down your database. [11:59] This is a pretty complex part of the [12:01] project [12:04] or you can give it to the proxy. You [12:07] create a publication then run reshard. [12:11] The app keeps moving while the data [12:14] moves. So you provision the empty [12:17] databases which are your shards and then [12:20] PG Dog fills them. The steps didn't go [12:23] away. the back fill, the catchup, the [12:26] cut over all still happen. The [12:29] difference is who runs them. [12:33] Okay, now that we've migrated the data, [12:36] the next job is when a request comes in, [12:41] which shard do we talk to? [12:45] So either your application picks, Rails [12:48] knows about the shards and chooses on [12:51] every request or the proxy picks, the [12:54] app talks to it like one ordinary [12:57] database and the proxy routes. [13:01] We'll start at the app level. Rails has [13:04] supported multiple databases since 6.0 [13:07] and sharding since 6.1. It works the [13:11] same on Postgress or MySQL. [13:15] Step one is config. Every shard is an [13:19] entry in database.yaml. [13:22] Each its own server. Each gets a read [13:25] replica. [13:27] Then one abstract model wires the shards [13:31] together [snorts] with connects to a [13:35] writer and a reading replica for each [13:38] and the tenant scoped models inherited. [13:42] remedies are the same on every shard. [13:46] Anything that isn't a clinics like [13:48] billing stays on the primary [13:53] and this is the primitive. You ask shard [13:57] map which shard the clinic lives on and [14:00] wrap your query and connect it to [14:03] everything inside runs against that [14:06] shard. That's how you reach one clinic's [14:09] data. [14:11] Rails answers exactly one question. [14:14] Which shard do I talk to? [14:18] Your shard key is one detail, but the [14:21] moment you shard, every query needs it. [14:24] That requires a clinic. It wants to [14:27] spread. You want to keep it in one [14:29] place. [14:32] So you need to reach a clinic shard in [14:35] the visits controller. So you open it [14:37] right here. [14:40] Then again in reminder job. [snorts] So [14:43] you also open it here and eventually it [14:49] spreads everywhere. [laughter] [14:51] Um when you're setting up your data for [14:56] charting, you have to find where every [15:00] where you find every clinic and you if [15:03] [snorts] you do it this way, you would [15:04] have to add this logic to every place. [15:07] and it's really easy to miss one. [15:11] So you want to contain it in one place [15:14] and luckily Rails already gives you for [15:18] Rails already gives you that. [15:21] Rails shifts configuration for shard [15:24] support out of the box. [15:26] This is the one place that knows how to [15:30] find a shard. This is the only place [15:34] you can go one step further and move [15:37] routing out of the app completely onto [15:39] the proxy. Hands it the shard key and it [15:43] does the routing. You no longer have the [15:47] shard map to maintain. You don't have to [15:50] write the resolver. Your app goes back [15:52] to believing that there's one database [15:55] while you get the benefits of sharding. [15:59] Now what about background jobs? The same [16:03] rule handle it in one place. [16:06] You put clinic ID in the jobs arguments. [16:10] Then a single around perform in [16:13] application job resolves the shards and [16:16] wraps perform in connected to one place [16:20] decides individual jobs never mention [16:23] shards. And if someone incues a job [16:26] without a clinic, it'll raise instead of [16:28] looking at the wrong shard. [16:32] With the proxy, every process connects [16:35] to the same endpoint, the same around [16:38] perform, but instead of connected to, we [16:41] we pass it this SQL set local and we set [16:46] the sharding key. This is inside of a [16:49] transaction. [16:52] So solid Q is database backed. You want [16:56] to keep its tables on the primary or [16:59] give them their own database. You don't [17:01] want to shard this. The set local is the [17:04] same as with sidekick. [17:08] Whether you use Rails or the proxy, the [17:10] sharding logic should live in one place. [17:16] What happens if you outgrow the shards [17:18] that you spun up originally? In Rails, [17:21] you need to handle that. If you use the [17:24] hashing function, [17:27] that will decide what moves. [17:30] If you store it, if you stored [17:34] everything on your shard map, you can [17:37] just move the clinics that you choose. [17:40] The table was more work to set up front, [17:43] but it pays for itself the first time [17:45] you need to add more shards because [17:47] let's say you have a really big clinic [17:49] that needs its own shard. You can simply [17:52] spin up a new one, assign it the ID of [17:56] the new shard, and you're done. [18:01] The proxy can reshard while the app [18:03] keeps serving. There's no mapping to get [18:06] wrong because you never wrote one if you [18:09] use the proxy. [18:12] Because the shard key keeps the clinic's [18:14] data together, most joins will never [18:18] have to cross a shard. [18:22] The joins that do need to cross can't be [18:24] done in one go. So you ask each shard [18:28] and merge the answers yourself. This is [18:30] called scatter gather. You loop the [18:33] shards, run the same query on each and [18:36] add the numbers up. [18:40] [snorts] So a proxy will do this for [18:43] you. It fans out the cross shard reads. [18:46] You write one query, it hits every shard [18:50] and merges. [18:52] A few writes do cross shards. So [18:57] transactions on one database are atomic. [19:01] The right is only ever seen as a single [19:03] action. Can you keep a transaction [19:07] atomic across shards? That's actually [19:10] not possible in a distributed system. [19:14] Once your data lives on machines that [19:16] talk over a network, you have a choice. [19:19] You either wait for every machine to [19:22] agree before anyone answers or you [19:26] answer now and have something that [19:28] catches up to it later. You can't have [19:30] both. [19:34] In a two-phase commit, you ask every [19:37] shard, hey, can you do this? [19:40] [snorts] Each locks what it needs and [19:43] promises yes. Only if everyone promises [19:48] does anyone commit. This is the wait for [19:52] everyone answer. [19:54] It stays transactional, but everyone [19:57] waits on the slowest shard. [19:59] And if the coordinator dies mid [20:03] handshake, the shards are left in doubt. [20:06] Each has promised and can't take it back [20:10] and it sits holding its locks. The two a [20:14] two-phase commit is a blocking protocol. [20:21] So the protocol is the smallest part. [20:25] Postgress gives you prepare transaction [20:30] and a prepare transaction survives [20:32] crashes. [20:34] So the hard part of hand rolling this [20:37] yourself is the plumbing and everything [20:40] around it. If your code crashes between [20:43] the prepare and commit, every shard is [20:47] stuck holding his locks and Postgress [20:49] will not clean that up for you. [20:52] You have to write what does the [20:55] plumbing, what notices that things are [20:57] stuck and something you you have to [21:00] write the thing that goes and finishes [21:02] that job [21:06] or the proxy can do this for you. You [21:09] write one plain transaction, it runs the [21:12] two-phase commit. The same guarantee, [21:16] the same costs, the locks are still [21:18] real. You just didn't have to build the [21:21] plumbing. So the proxy gives you all or [21:25] nothing across shards. It runs it for [21:27] you. What it can't give you is [21:30] animicity. [21:32] That left when data split. A good shard [21:36] key means you only have to do this for a [21:39] very small percent of your rights. [21:44] A proxy can hide the hard parts. It [21:47] doesn't remove them. A cross shar join [21:50] is still slow. A two-phase commit is [21:53] still fragile. The wind was never making [21:57] the hard parts easy. It's arranging [21:59] things so you rarely have to handle [22:02] this. [22:05] In Rails, you own all the pieces. The [22:08] migration, the mapping, the routing, the [22:10] joins, the transactions. Every piece is [22:13] yours to keep correct and keep running. [22:16] With a proxy, you can hand that work [22:19] over and your app still thinks it's on [22:21] one database. [22:23] You get the benefits of sharding without [22:26] all the plumbing. And that's pretty [22:28] cool. [22:31] You don't beat sharding by working [22:33] harder on the hard parts. You want to [22:37] structure things in a way that you avoid [22:40] them as much as possible. [22:42] Arrange your data so each thing's work [22:45] is in one place and the hard parts [22:48] barely come up. Get that right and most [22:52] of sharding's pain never happens. [22:56] Shard on the data that's easiest to [22:58] separate. Good sharding is good [23:00] boundaries. Thank you. [23:05] [music]