Entity Sharding

datum-cluster-sharding is Datum's counterpart to Akka Cluster Sharding: it spreads a large population of stateful entities across the nodes of a datum-cluster, routes messages to an entity by id no matter which node currently owns it, and moves entities between nodes when membership changes. You address an entity through an EntityRef and never track where it lives.

cargo add datum-cluster-sharding

It builds directly on the pieces the other guides describe: datum-cluster membership and its oldest-member coordinator convention, and the long-lived DCP node sessions of the cluster-aware datum-agent. Entities themselves are ordinary Ractor actors.

Quickstart

use datum_agent::{
    ClusterAgent, ClusterAgentConfig,
    dcp::{DcpJobFactories, DcpServerConfig, DcpTcpServerConfig},
};
use datum_cluster::ClusterConfig;
use datum_cluster_sharding::{
    EntityContext, ReplyPort, Sharding, ShardingConfig, ShardingResult,
};

// Entity messages cross node boundaries, so they are serializable. An `ask`
// message carries a sharding `ReplyPort` that the entity completes.
#[derive(Serialize, Deserialize)]
enum CounterMsg {
    Add(u64),
    Get(ReplyPort<u64>),
}

// A cluster-aware agent bundles membership, DCP node sessions, and the job
// registry. One loopback node here; add `seed_nodes` to the ClusterConfig to
// form a real cluster.
let agent = ClusterAgent::start(
    ClusterAgentConfig {
        cluster: ClusterConfig::new("orders-1"),
        dcp: DcpServerConfig {
            tcp: Some(DcpTcpServerConfig {
                addr: "127.0.0.1:0".parse().expect("loopback addr"),
            }),
            ..DcpServerConfig::default()
        },
        ..ClusterAgentConfig::default()
    },
    DcpJobFactories::new(),
)
.await
.expect("cluster agent starts");

// Install one sharding region on this node.
let sharding = Sharding::init(
    &agent,
    ShardingConfig {
        num_shards: 64,
        ..ShardingConfig::default()
    },
)
.expect("sharding region");

// Register an entity type. The factory rebuilds the entity's behavior every
// time it is (re)spawned — on first message, after a handler panic, or after
// the shard moves to this node.
sharding
    .register_entity_type("counter", |_spawn: EntityContext| {
        let mut total = 0_u64;
        move |_context: &EntityContext, message: CounterMsg| -> ShardingResult<()> {
            match message {
                CounterMsg::Add(amount) => total += amount,
                CounterMsg::Get(reply) => {
                    let _ = reply.send(total);
                }
            }
            Ok(())
        }
    })
    .await
    .expect("entity type registered");

// Wait until this node is the shard coordinator so shards can be placed.
while !agent
    .cluster()
    .current_state()
    .is_placement_coordinator(agent.cluster().node_id())
{
    tokio::time::sleep(Duration::from_millis(10)).await;
}

// Address an entity by id; the region routes to whichever node owns the
// shard that `cart-42` hashes to. The same EntityRef with awaited sends
// preserves per-entity order.
let cart = sharding.entity_ref::<CounterMsg>("counter", "cart-42");
cart.tell(CounterMsg::Add(3)).await.expect("tell add");
let total = cart
    .ask(Duration::from_secs(5), CounterMsg::Get)
    .await
    .expect("ask get");
assert_eq!(total, 3);

sharding.shutdown().await;
agent.shutdown().await.expect("agent shuts down");

Sharding::init installs one region per node on a running ClusterAgentHandle. You register entity types by name, then hand out EntityRefs. On a single node every shard is owned locally; add seed nodes and start more agents and the same EntityRef code routes across the cluster unchanged.

Envelopes and extractors

Sharding needs two ids for every message: which entity it targets, and which shard that entity belongs to. Datum mirrors Akka's entity/shard extractors.

pub struct ShardEnvelope<M> { pub entity_id: String, pub message: M }

pub trait ShardExtractor<M> {
    fn entity_id<'a>(&self, envelope: &'a ShardEnvelope<M>) -> &'a str { &envelope.entity_id }
    fn shard_id(&self, entity_id: &str) -> String;
}

The default, DefaultShardExtractor::new(num_shards), hashes the entity id with Datum's stable FNV-1a 64-bit hash and takes it modulo the shard count. entity_ref(type_name, entity_id) uses the region's configured default extractor; entity_ref_with_extractor(type_name, entity_id, extractor) takes a custom one when your ids carry their own shard key. The shard count is a fixed partitioning of the id space — pick it up front (ShardingConfig::num_shards, default 128); it is not resized at runtime.

Entities and the EntityRef

An entity type is a name plus a factory. The factory receives an EntityContext (type_name + entity_id) and returns the entity's EntityBehavior<M> — a handler run inside one Ractor actor. A plain FnMut(&EntityContext, M) -> ShardingResult<()> closure is an EntityBehavior via a blanket impl, as in the quickstart; implement the trait directly when you want a named struct with fields.

Entities are spawned on first message and live on whichever node owns their shard. You reach one through an EntityRef<M>:

• entity_ref.tell(message).await — send one message.
• entity_ref.ask(timeout, make_message).await — send a message carrying a sharding ReplyPort<R> and await the reply. make_message is FnOnce(ReplyPort<R>) -> M, so a tuple-variant constructor like CounterMsg::Get is exactly the right shape. The entity calls reply.send(value); the port serializes as an opaque target and routes the reply back across a node boundary transparently.
• entity_ref.entity_id() / entity_ref.shard_id() — the resolved ids (the shard id is computed synchronously by the extractor).

A handler panic restarts the entity, not the shard: the entity actor restarts and the shard and its other entities keep running.

Delivery semantics — stated honestly

These are the guarantees v0.10 actually provides. Read them before you design on top of sharding.

• At-most-once per attempt. A tell/ask is delivered at most once. There is no built-in redelivery: if a send fails (timeout, a node lost mid-flight), it is your retry that re-sends, and a retry can cause the entity to process the message more than once. The benchmark harness counts those duplicate processings honestly rather than hiding them.
• Per-entity ordering, per sender. Messages to one entity from one sender that awaits each tell/ask on the same EntityRef are processed in send order. Concurrent senders, or fire- and-forget sends that don't await, are not ordered against each other.
• No persistence. The allocation table (which node owns which shard) is in-memory, maintained by the coordinator and replicated best-effort to peer regions over DCP node sessions. Entity state is in-memory and is not migrated when a shard moves — a rebalanced or respawned entity starts from its factory again. Remember-entities re-spawns entity ids after a move or restart; it does not carry their state. Durable entity state (event-sourcing / snapshots) is not part of v0.10.

The coordinator and shard allocation

Sharding reuses datum-cluster's placement coordinator: the oldest reachable Up member, by incarnation then node id. The coordinator allocates each shard to a node with a least-shards policy (the allocation spreads shards so per-node counts differ by at most one), records an allocation generation per shard, and replicates the table to peer regions.

While a shard is unallocated — first use, or just after a move — the owning region buffers envelopes for it, bounded by ShardingConfig::allocation_buffer. Overflow surfaces as ShardingError::BufferOverflow { type_name, shard_id } rather than growing without limit. The envelope/reply hot path runs over long-lived per-peer DCP shard pipes (batched, correlation-id replies); only the rare allocation miss falls back to request/response.

Coordinator takeover. If the coordinator node is lost, the new oldest member rebuilds the allocation table from surviving regions, keeps the highest generation per shard, and reallocates any shard still pointing at a non-eligible member. The same split-brain caveat as membership applies: there is no fencing, so under a partition with timeout downing more than one node can transiently act as coordinator.

Rebalance

The coordinator moves shards for two reasons, exposed as RebalanceReason and recorded per round (handle.rebalance_rounds() returns Vec<RebalanceRound> of ShardMovements while this node was coordinator):

Reason	When	Capped?
DeadOwner	the previous owner is no longer an eligible shard owner (node left/downed)	no — dead-owner reallocation is never throttled
GracefulSpread	a live shard is moved to spread load after a node joins	yes — at most ShardingConfig::rebalance_per_round moves per round

During a move the shard's region buffers in-flight messages (and passivation requests) through the same bounded handoff buffer, drains after handoff_drain_delay, and preserves per-entity order across the move for a single awaiting sender. Messages to entities on other, non-moving shards are unaffected.

Passivation

Passivation stops an idle entity to free memory; the next message respawns it on demand. Two triggers:

• Idle timeout — set ShardingConfig::passivation_idle_timeout and an entity with no traffic for that long is passivated automatically.
• Explicit — entity_ref.passivate().await stops the entity now. If its shard is moving, the passivation is ordered through the same handoff buffer as user messages.

Passivation stops the entity actor and removes it from its owner's runtime; the next tell/ask spawns a fresh entity from the factory.

Remember-entities

By default an entity exists only while it has a live actor: after passivation or a node restart, an id is gone until the next message for it arrives. Remember-entities records started entity ids so they can be re-spawned automatically when their shard starts on a node — after rebalance, restart recovery, or coordinator takeover — without waiting for a triggering message.

Enable it with a store and register the type through the remembered variant:

use std::sync::Arc;
use datum_cluster_sharding::{FileStore, RememberEntitiesConfig, ShardingConfig};

let config = ShardingConfig {
    remember_entities: RememberEntitiesConfig::with_store(Arc::new(FileStore::new("/var/lib/datum/shards"))),
    ..ShardingConfig::default()
};
// then, per type:
sharding.register_remembered_entity_type("counter", factory).await?;

The store SPI

RememberEntitiesStore is the pluggable persistence seam. It records started/stopped ids per type_name/shard_id and lists ids for shard startup; list_shards (default empty) lets a store recover shards after a node restart:

pub trait RememberEntitiesStore: Send + Sync + 'static {
    fn record_entity_started(&self, type_name: &str, shard_id: &str, entity_id: &str) -> …;
    fn record_entity_stopped(&self, type_name: &str, shard_id: &str, entity_id: &str) -> …;
    fn list_entities_for_shard(&self, type_name: &str, shard_id: &str) -> … Vec<String>;
    fn list_shards(&self, type_name: &str) -> … Vec<String> { /* default: empty */ }
}

Two stores ship in the box:

• InMemoryStore — process memory. It survives a shard rebalance when all regions share the same InMemoryStore value, but it does not survive process death. For tests and single-process development.
• FileStore — a per-shard append log at <dir>/<hex(type)>/<hex(shard)>.log. Starts append one record; stops/passivation rewrite the log with the live ids, compacting tombstones immediately. It survives a full node restart.

Passivation removes the id

Passivation removes the entity id from the remember store — matching Akka Cluster Sharding's rule that passivated entities are not remembered. A passivated entity is not respawned on the next shard start; only a new message brings it back. Ordinary actor failure/restart and node stop do not remove the id.

Durability honesty

Store writes go through a bounded write-behind queue (RememberEntitiesConfig::queue_capacity), so recording an id never blocks the entity message hot path. The direct consequence, stated plainly: a delivered message is acknowledged before its remember-entities record is necessarily durable. Even FileStore with FileStoreFsyncPolicy::Always (fsync every append/compaction; the default; OnCompaction and Never relax it) fsyncs inside the write-behind worker — the message ack does not wait for it. Use handle.flush_remember_entities().await to wait for queued writes to drain and become restart-visible when you need that barrier (e.g. before a planned shutdown).

When a store errors, RememberEntitiesFailurePolicy decides availability:

Policy	Behavior
FailOpen (default)	keep routing entity messages; publish a shard-level error event
FailClosed	mark the shard failed; future messages fail with ShardingError::RememberEntities { … }

Either way the failure is observable: handle.subscribe_remember_entities_events() yields RememberEntitiesEvents (type/shard/entity, the RememberEntitiesOperation, the policy, and the error message).

Performance

Numbers are recorded honestly in roadmap/benchmarks/cluster.md. The frozen Pekko target and the Datum result use the same full workload — 1,000,000 messages, 1,000 entities, 3 nodes — with app-level logical-ack correctness counters on both sides (unique_acks == expected_acks, missing_acks == 0).

Sharded entity throughput, 1M / 1k / 3 nodes:

	Wall	Latency p50	Latency p99	CPU
Datum	5,475 ms	0.193 ms	0.977 ms	67.4 µs/op (≈67 s total)
Pekko 1.6.0	48,533 ms	0.252 ms	18.923 ms	≈221 s total (220,610 ms)

Datum beats the frozen Pekko row on wall (~8.9×), p50, p99, and whole-process CPU (~3.3× less), at 182,659 msg/s with a 28.6 MiB RSS peak. (Datum's CPU is measured per-op and multiplied out across the 1,000,000 messages; Pekko's is the whole-process total the harness recorded.)

Rebalance — kill one node, 1M / 1k, 3 → 2 nodes: 341 / 341 affected entities rehomed in 710 ms (Pekko: 13,175 ms), max post-kill latency 752 ms (Pekko: 13,175 ms), first ack after kill 0.239 ms — with the same 1,000,000 / 1,000,000 logical acks and 0 missing.

Remember-entities cost: enabling the in-memory store on the full workload added +2.0% CPU (1.48 µs/op) with wall/latency inside run noise — the write-behind queue keeps store writes off the message hot path.

Per-node CPU/RSS caveat. The Datum figures come from an in-process 3-node runner, so per-node CPU and RSS are not separable — the CPU and RSS numbers above are whole-runner figures, and a per-node resource claim awaits a multi-process Datum harness. Pekko, by contrast, runs three JVMs and the frozen table lists its per-node rows. Compare the whole-process totals with that in mind; the record states this limitation on every affected row.

Limits in v0.10

Sharding is the youngest layer; know its edges. No durable entity state (state is in-memory and not migrated — remember-entities re-spawns ids, not state). At-most-once per attempt with caller-driven retries, so design for idempotent handling or duplicate-tolerant work. The allocation table is in-memory with no persistent store. The shard count is fixed at init. And the coordinator inherits membership's split-brain caveat — a quorum/lease resolver is the named prerequisite for strong singleton claims.

See also

• Membership — the membership, downing, and coordinator layer this builds on.
• Agent — the cluster-aware agent (ClusterAgent) and DCP node sessions.
• Actors Interop — the Ractor model entities run on.