Sharding

Sharding partitions a collection's data across multiple replica sets ("shards") and routes operations to the right shards via mongos. The data side lives under src/mongo/db/s/ and src/mongo/db/global_catalog/; the router side is at src/mongo/s/.

Purpose

Sharding adds:

• Horizontal scale — both for storage and write throughput.
• Data locality — by zoning shard-key ranges to specific shards (e.g. by region).
• Rebalancing — the balancer moves chunks between shards to keep them roughly equal.
• Online resharding — rewriting a collection's shard key while the cluster keeps serving traffic.

Cluster topology

graph LR
    subgraph Routers
        Mongos1[mongos]
        Mongos2[mongos]
    end
    Mongos1 -.metadata.-> CSRS[Config Server<br/>Replica Set]
    Mongos2 -.metadata.-> CSRS

    subgraph Shards
        S1[Shard A<br/>replica set]
        S2[Shard B<br/>replica set]
        S3[Shard C<br/>replica set]
    end

    Mongos1 --> S1
    Mongos1 --> S2
    Mongos2 --> S2
    Mongos2 --> S3
    S1 -.balancer.-> S2

The config server replica set (CSRS) stores cluster metadata in a set of config.* collections (config.collections, config.chunks, config.databases, config.shards, …).

Chunks and the routing table

A sharded collection's key space is divided into chunks. Each chunk has a [min, max) range and an owning shard. The ChunkManager materializes the routing table for a collection from config.chunks. The mapping for the entire cluster lives in:

• config.databases — primary shard for each database.
• config.collections — sharding metadata for each sharded collection.
• config.chunks — the per-chunk placement and version information.
• config.shards — per-shard host list and tags.
• config.tags — zone definitions.

mongos and shard-local mongod instances consume this metadata via the catalog cache (src/mongo/s/catalog_cache.h) and the shard-server-side catalog cache (src/mongo/db/s/catalog_cache_loader.h).

Routing a request

sequenceDiagram
    participant Client
    participant Router as mongos
    participant Cache as CatalogCache
    participant Shard as Shard primary

    Client->>Router: insert/find/update OP_MSG
    Router->>Cache: getCollectionRoutingInfo(nss)
    Cache-->>Router: ChunkManager (snapshot at V_n)
    Router->>Shard: command + shardVersion=V_n
    Shard->>Shard: validate shardVersion
    alt version matches
        Shard-->>Router: result
        Router-->>Client: aggregate result
    else stale shardVersion
        Shard-->>Router: StaleConfig(V_correct)
        Router->>Cache: refresh(nss)
        Router->>Shard: retry with V_correct
    end

Every command targeted to a shard carries a shardVersion token. If the shard is on a different version than the router, the shard returns StaleConfig and the router refreshes its cache and retries. This keeps the cluster eventually consistent without requiring strong synchronization.

Migrations

The balancer (running on the CSRS primary) periodically scans for imbalance and orders chunk migrations:

stateDiagram-v2
    [*] --> Scheduled
    Scheduled --> CloningPhase1: source begins clone
    CloningPhase1 --> CloningPhase2: catch up writes
    CloningPhase2 --> CommitPhase: critical section
    CommitPhase --> Cleanup: ownership transferred
    Cleanup --> [*]: orphans deleted
    Scheduled --> Aborted
    CloningPhase1 --> Aborted
    CloningPhase2 --> Aborted
    CommitPhase --> Aborted

Migration source/destination state machines are in src/mongo/db/s/. Specifically:

• src/mongo/db/s/migration_source_manager.cpp — donor side.
• src/mongo/db/s/migration_destination_manager.cpp — recipient side.
• src/mongo/db/s/range_deleter* — cleans up orphans on the donor after migration.
• src/mongo/db/s/balancer/ — scheduling decisions on the CSRS.

DDL coordinators

Sharded DDL (create, drop, rename, shardCollection, reshardCollection) is run by DDL coordinators under src/mongo/db/global_catalog/ddl/. Each coordinator is a primary-only service that:

1. Persists its state in a coordinator collection.
2. Drives the operation through phases (prepare, fan-out, commit).
3. Survives primary failover by resuming from its persisted state.
4. Coordinates with shards via _shardsvr* commands.

Resharding

Resharding rewrites an already-sharded collection's data into a new shard key without taking the collection offline. The data path is in src/mongo/db/s/resharding/; the coordinator and metadata live under src/mongo/db/global_catalog/ddl/. The router-side surface is in src/mongo/s/resharding/. The protocol has cloning, applier, and critical-section phases similar to a migration but at the collection-wide scale, with explicit recovery for primary failover.

Cluster transactions

Multi-shard transactions are routed by the TransactionRouter on mongos. See Transactions and the mongos page. The shard side is in src/mongo/db/transaction/ and src/mongo/db/repl/.

Cluster parameters

A cluster parameter is a parameter persisted in the config server and propagated to all shards and mongoses. Used for cluster-wide knobs like defaultMaxTimeMS. Implementation is split across src/mongo/db/cluster_parameters.cpp and IDL declarations under **/*_cluster_parameter.idl.

Key source files

File	Purpose
src/mongo/s/catalog_cache.h	Router-side routing-table cache.
src/mongo/s/transaction_router.cpp	Cluster transactions.
src/mongo/s/async_requests_sender.cpp	Multi-shard fan-out.
src/mongo/db/s/migration_source_manager.cpp	Migration donor.
src/mongo/db/s/migration_destination_manager.cpp	Migration recipient.
src/mongo/db/s/balancer/	Balancer.
src/mongo/db/s/resharding/	Resharding data path.
src/mongo/db/global_catalog/ddl/	Sharded DDL coordinators.
src/mongo/db/global_catalog/sharding_catalog_client.cpp	Read/write helpers for config.* collections.
src/mongo/db/sharding_environment/	Shard-side runtime: shard identity, sharding state.

Integration points

• Replication — the CSRS is itself a replica set, and migrations rely on durable oplog application.
• Transactions — cluster transactions span shards via the TransactionRouter.
• Op observer — emits the special oplog entries used to coordinate migrations and resharding.
• Catalog cache — both mongos and shards keep their own caches, with similar but not identical APIs.

Entry points for modification

Most sharded-DDL changes start at a coordinator under src/mongo/db/global_catalog/ddl/. Routing-cache changes are concentrated in src/mongo/s/catalog_cache.h and the loaders. Migration logic is split between donor and recipient under src/mongo/db/s/. The balancer's scheduling lives at src/mongo/db/s/balancer/. The sharding resmoke suite is the primary CI gate; many tests under jstests/sharding/ rely on fail points to deterministically reproduce migration races.