Cloudflare D1 Sharding in Practice: Breaking Through the 10GB Limit and the Single-Writer Bottleneck

If you’ve ever run a multi-tenant SaaS or write-heavy event-logging service on Cloudflare D1, you eventually hit this wall: a hard 10GB limit per database, combined with the fact that every write goes through a single writer that processes them sequentially. With Postgres or MySQL you’d just add read replicas or reach for a partitioning extension. D1 is different by design — it’s fundamentally “many SQLite instances distributed across the edge,” so the whole approach has to change.

This post covers everything you need to actually put sharding into production: how to tell when indexing alone is enough versus when you genuinely need to shard, code for tenant-ID-based routing built on Worker + KV, a procedure for rebalancing shards with zero downtime, and strategies for working around cross-shard aggregation.

Key takeaways

• D1 caps out at 10GB (Paid) / 500MB (Free) per database, with an account-wide ceiling of 1TB (Paid) / 5GB (Free), and each database has a single writer processing writes sequentially.
• Read-heavy bottlenecks should be solved with Sessions API–based read replicas first, not sharding — read replicas do nothing for write throughput.
• Only consider sharding when (1) DB size is approaching 10GB or (2) write contention is driving up latency. Before that, check indexes, batch writes, and cold-data archiving first.
• For shard mapping, a KV + control-DB directory mapping is far easier to rebalance than a hash-based scheme.
• Cross-shard JOINs and transactions aren’t supported, so aggregation has to be worked around with fan-out queries or a separate rollup store.

Where the real bottlenecks come from: D1’s 10GB hard limit and single-writer architecture

D1’s constraints split into two layers: the size limit on an individual database and the storage limit across the whole account. Per Cloudflare’s official docs, the numbers look like this:

The important asymmetry here is that “the account-wide capacity is generous, but any individual database fills up fast.” On the Paid plan you can use up to 1TB across your whole account, but if you’ve crammed 100 tenants into a single database, that database is still capped at 10GB. In other words, the real bottleneck isn’t the account limit — it’s the “10GB per database” wall. Services with tables that just keep growing — event logs, audit tables — tend to hit this wall sooner than expected.

The second constraint is the single-writer architecture. D1 runs on the SQLite engine, and as the docs state explicitly, “each database processes queries sequentially on a single thread.” The commonly cited theoretical ceiling is around 1,000 queries per second for short (~1ms) queries, but real production queries mix index scans, joins, and write transactions, so latency starts climbing well below that number in practice. The real issue is that this is a concurrency problem, not just a throughput problem. While tenant A’s heavy write transaction is running, tenant B’s requests against the same database sit in a queue. In a multi-tenant setup, that turns directly into a noisy-neighbor problem, where one tenant’s traffic spike drags up latency for every other tenant.

The typical signals that reveal this bottleneck in practice:

• INSERT/UPDATE p95 latency spikes 3–5x above normal during peak hours, even though indexes are already in place.
• API responses for other tenants slow down while one large tenant’s batch job is running.
• DB size reported by wrangler d1 info has crossed 8–9GB and keeps trending up.
• Bulk insert operations start throwing too many SQL variables or query timeout errors.

Deciding when you actually need to shard: read-heavy vs. write-heavy workloads

Sharding is expensive. It means building a routing layer, a rebalancing procedure, and workarounds for cross-shard aggregation — treating “just split it up to be safe” as the default is over-engineering. In practice, you need to diagnose the nature of your workload first.

If the bottleneck is read-heavy, sharding isn’t the answer

If slowness comes from a surge in read traffic, D1 offers Sessions API–based read replicas. Requests get distributed to geographically nearby read replicas, cutting latency, and since multiple replicas can serve reads concurrently, read throughput improves too. But as the official docs state, “all write queries are still routed to the primary database only” — read replicas don’t improve write throughput at all. So reaching for sharding first when the workload is read-heavy is solving things in the wrong order — exhaust read replicas, KV/Cache API caching, and query optimization first.

If the bottleneck is write-heavy, there aren’t many alternatives to sharding

It’s a different story when the problem is concurrent write contention or a single database approaching the 10GB ceiling. Read replicas help with neither — one primary is still processing all writes sequentially, and one file is still hitting the 10GB wall. In this case, manual sharding by tenant or entity is effectively the only path to horizontal scaling.

A checklist for making the call:

• Is your largest database at 70%+ of the 10GB limit with no sign of the growth curve flattening?
• Does write p95 latency during peak hours run 2x or more above baseline? (Make sure you’re looking at write latency, not read latency.)
• Do both symptoms above persist even after you’ve already applied indexing, batch writes, and cold-data archiving?
• Is there enough variance in per-tenant data volume (e.g., the top 5% of tenants account for 60% of total size) that isolating specific tenants would actually help?

If three or more of these four apply, it’s time to move sharding into the design phase. If most answers are “no,” the indexing checklist in the final section below should buy you several more months.

Designing tenant-ID-based routing: a Worker + KV shard-mapping table

There are two approaches to shard keys: hash-based and directory (mapping-table) based. Hash-based (shard = hash(tenant_id) % N) is simple to implement, but changing the number of shards or isolating and moving a single tenant requires rehashing at scale. A directory-based mapping, on the other hand, records facts like “this tenant currently lives on shard-3” in a separate table, which makes it possible to do partial rebalancing — moving exactly one tenant to a different shard without touching anything else. In multi-tenant environments, directory-based mapping is almost always the recommended choice.

The architecture has three layers:

1. Control DB: the single source of truth for shard mapping. A small, dedicated D1 database holding just one shard_map table.
2. KV cache: querying the control DB on every request would make D1 itself the bottleneck, so mapping results are cached in Workers KV as a read-through cache.
3. Shard DBs: the N D1 databases that actually hold tenant data.

# wrangler.toml
name = "multitenant-api"
main = "src/index.ts"
compatibility_date = "2025-01-01"

[[d1_databases]]
binding = "CONTROL_DB"
database_name = "control-db"
database_id = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

[[d1_databases]]
binding = "SHARD_0"
database_name = "tenant-shard-0"
database_id = "xxxxxxxx-0000-xxxx-xxxx-xxxxxxxxxxxx"

[[d1_databases]]
binding = "SHARD_1"
database_name = "tenant-shard-1"
database_id = "xxxxxxxx-1111-xxxx-xxxx-xxxxxxxxxxxx"

[[kv_namespaces]]
binding = "SHARD_MAP_KV"
id = "yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy"

Control DB schema:

CREATE TABLE shard_map (
  tenant_id   TEXT PRIMARY KEY,
  shard_id    TEXT NOT NULL,       -- 'SHARD_0', 'SHARD_1' ...
  status      TEXT NOT NULL DEFAULT 'ACTIVE', -- ACTIVE | MIGRATING | DONE
  target_shard_id TEXT,            -- 리밸런싱 중일 때만 채워짐
  updated_at  INTEGER NOT NULL
);
CREATE INDEX idx_shard_map_status ON shard_map(status);

The routing logic checks KV first, and only falls back to the control DB on a cache miss, writing the result back through to the cache:

// src/shard-router.ts
type Env = {
  CONTROL_DB: D1Database;
  SHARD_MAP_KV: KVNamespace;
  SHARD_0: D1Database;
  SHARD_1: D1Database;
  [key: string]: any;
};

interface ShardEntry {
  shardId: string;
  status: "ACTIVE" | "MIGRATING" | "DONE";
  targetShardId?: string;
}

export async function resolveShard(
  tenantId: string,
  env: Env
): Promise<D1Database> {
  const cacheKey = `shard:${tenantId}`;
  const cached = await env.SHARD_MAP_KV.get<ShardEntry>(cacheKey, "json");

  let entry: ShardEntry | null = cached;

  if (!entry) {
    const row = await env.CONTROL_DB
      .prepare(
        "SELECT shard_id, status, target_shard_id FROM shard_map WHERE tenant_id = ?"
      )
      .bind(tenantId)
      .first<{ shard_id: string; status: string; target_shard_id: string | null }>();

    if (!row) {
      throw new Error(`Unknown tenant: ${tenantId}`);
    }

    entry = {
      shardId: row.shard_id,
      status: row.status as ShardEntry["status"],
      targetShardId: row.target_shard_id ?? undefined,
    };

    // 읽기 전용 캐시. 리밸런싱 중에는 TTL을 짧게 둔다.
    const ttl = entry.status === "MIGRATING" ? 30 : 300;
    await env.SHARD_MAP_KV.put(cacheKey, JSON.stringify(entry), {
      expirationTtl: ttl,
    });
  }

  // MIGRATING 상태면 라이팅은 아직 원본 샤드로 보낸다 (다음 섹션 참고).
  const activeShardId = entry.status === "DONE" && entry.targetShardId
    ? entry.targetShardId
    : entry.shardId;

  const db = env[activeShardId] as D1Database | undefined;
  if (!db) throw new Error(`Shard binding not found: ${activeShardId}`);
  return db;
}

When onboarding a new tenant, add logic that picks whichever shard currently has the most headroom, based on size and load. Simple round-robin assignment can pile large tenants onto the same shard, so it’s better to factor in expected data volume (e.g., plan tier) at onboarding time — that reduces how often you’ll need to rebalance later.

Moving data between shards and rebalancing with zero downtime

D1 doesn’t offer native cross-database replication or an online migration API, so rebalancing has to be driven as an explicit state machine at the application layer. Here’s the procedure, using the shard_map.status column (ACTIVE → MIGRATING → DONE):

1. Pick a migration target: once size/load monitoring shows a shard crossing a threshold (say, 8GB, or a single tenant accounting for 40%+ of that shard’s writes), pick a tenant to move. Being able to pick just one tenant out of many is exactly the advantage of directory-based mapping.
2. Flip status to MIGRATING: update that tenant’s status to MIGRATING in shard_map and set target_shard_id to the destination shard. From this point, the application keeps writing to the original shard while simultaneously publishing the same write events asynchronously via Cloudflare Queues (dual write).
3. Bulk copy: page through SELECT reads of that tenant’s rows on the source shard and push them into the destination shard with db.batch(). Size your batches around the per-invocation query limit (50 on Free / 1,000 on Paid) and the 100-bound-parameter limit. If there’s a lot of data and latency is less of a concern, it’s simpler to go the offline route: dump with wrangler d1 export --output=tenant.sql, filter it down, and import into the destination with wrangler d1 execute (keeping in mind the 5GB file-import cap).
4. Replay the queue: after the bulk copy finishes, replay the events queued up during migration (writes that happened in the meantime) against the destination shard, in order, to catch up.
5. Verify consistency: compare row counts between source and destination, and checksum key tables (e.g., SELECT COUNT(*), SUM(amount) FROM orders WHERE tenant_id = ?). If anything’s off, retry steps 3–4.
6. Cut over: once verification passes, flip shard_map.status to DONE and update shard_id to the destination in a single transaction. From this moment on, new writes go only to the destination shard.
7. Wait out the cache: keep the original shard alive as a read-only fallback until the cached KV routing entry expires (this is why the code above sets a short 30-second TTL for the MIGRATING state). Build in at least a 1–2 minute grace period to account for global KV propagation delay.
8. Clean up: once the grace period has passed, delete that tenant’s data from the original shard to reclaim space.

The essence of this procedure is “dual write plus a gradual cutover driven by cache TTL,” not “stop everything and move it.” Rebalancing with zero downtime ultimately means accepting the cost of temporarily writing to two places at once.

Limits on cross-shard queries and strategies for working around aggregation

The problem you’ll hit most often after sharding is cross-shard aggregation — things like “total across all tenants.” Because D1 database bindings are physically separate, neither cross-shard JOINs nor cross-shard transactions are supported. This is a hard constraint with no workaround, so it needs to be a design assumption from day one.

Fan-out queries

If you need real-time results and don’t have too many shards (tens, not hundreds), the simplest approach is to query every shard in parallel and sum the results in the Worker.

async function totalOrdersAcrossShards(env: Env): Promise<number> {
  const shardBindings = ["SHARD_0", "SHARD_1", "SHARD_2"] as const;

  const results = await Promise.all(
    shardBindings.map((key) =>
      (env[key] as D1Database)
        .prepare("SELECT COUNT(*) AS cnt FROM orders")
        .first<{ cnt: number }>()
    )
  );

  return results.reduce((sum, r) => sum + (r?.cnt ?? 0), 0);
}

As the shard count grows, this approach runs into the per-invocation query limit and overall request latency (the slowest shard determines the total response time). Once you’re up to hundreds of shards, fan-out is no longer a good fit for a real-time code path.

Rollup tables + async ETL

For aggregation that doesn’t need second-level freshness — dashboards, reporting — it’s far more stable to pre-aggregate at write time rather than fan out at request time. After each shard’s write transaction completes, publish an “aggregation needs updating” event via Cloudflare Queues; a separate consumer Worker picks up that event and updates rollup rows in a dedicated analytics D1 database (or Analytics Engine, if the data is heavily time-series). That way, the admin dashboard never touches the shards at all — it just queries the single rollup DB.

Use the saga pattern for transactions that span shards

For the rarer case where you need an atomic transaction spanning two shards — moving a resource between tenants, say — D1 doesn’t support distributed transactions, so you have to fall back to a saga pattern built on compensating transactions. Break the operation into idempotent steps (“debit shard A → if that succeeds, credit shard B → if that fails, roll back A”), and assign each step a unique, retry-safe operation ID to prevent duplicate execution.

An indexing and query-optimization checklist to try before you shard

It’s fairly common, once you’ve fully designed a sharding architecture, to realize “actually, fixing the indexes would have been enough.” It’s worth exhausting the checklist below first.

• Check whether indexes are actually being used with EXPLAIN QUERY PLAN. If you see SCAN TABLE, you’re doing a full scan. Build covering indexes around the column combinations that show up together in your WHERE and ORDER BY clauses.
• Select only the columns you need instead of SELECT *. Rows can be up to 2MB, but reading unnecessarily large columns (JSON blobs, etc.) on every query inflates both query time and network payload.
• Batch multi-row writes with db.batch() instead of individual INSERTs. Cutting down round trips alone can meaningfully improve perceived throughput under a single-writer architecture. Just split batches to stay under the 100-bound-parameter and 100KB SQL statement limits.
• Move high-frequency single-row counters and rate limits out of D1. A counter updated several times per second via a direct UPDATE to D1 becomes a single-writer bottleneck on its own. Moving it to Durable Objects or KV frees up D1’s write queue.
• Archive cold data to keep DB size under control. Periodically export infrequently accessed data — old logs, completed orders — to something like R2 and delete it from D1, buying yourself headroom against the 10GB ceiling.
• If the bottleneck is reads, reach for read replicas (Sessions API) and KV/Cache API caching before sharding. As covered earlier, read replicas only help with read throughput, but if reads are the actual problem, that’s often all you need.
• Check for N+1 query patterns. Firing off individual queries inside a loop will hit the per-invocation query limit (50 on Free / 1,000 on Paid) before anything else does. Consolidate with JOIN or an IN (...) clause first.

If you’ve applied this entire checklist and the criteria from earlier (DB size approaching 10GB, write p95 latency spiking) still hold, that’s the point where it’s actually time to build the sharding architecture covered in this post.