Skip to main content
Version: v5

Sharding

Added in: v4.4.0 (provisional)

Changed in: v4.5.0 — expanded sharding functionality: Harper now honors write requests with residency information that will not be stored on the local node, and nodes can be declaratively configured as part of a shard.

Harper's replication system supports sharding — storing different data across different subsets of nodes. Single-key reads can fetch records from their owners across shards. Sharding distributes storage and write work across nodes.

A key read for a record not held locally can fetch its body from an owner. This does not provide cross-shard query fan-out or write forwarding. With schema-defined sharding, a non-owner write commits locally and replicates the complete body to its owners through the configured replication routes.

By default (without sharding), Harper replicates all data to all nodes.

Schema-defined sharding​

In v5.4, @sharded hashes primary keys to fixed slots on every worker. Slot counts and the installed slot map belong to the database. Attribute-derived policies and custom expressions are refused pending concurrent-move design (#1178).

Older per-record mechanisms​

The setResidency, setResidencyById, getResidency and residency-list APIs below retain their names. They are the older mechanism, with two approaches:

Dynamic sharding — the location (residency) of records is determined dynamically based on where the record was written, the record's data, or a custom function. Records can be relocated dynamically based on where they are accessed. Residency information is specific to each record.

Static sharding — each node is assigned to a specific numbered shard, and each record is replicated to the nodes in that shard based on the primary key, regardless of where the data was written or accessed. More predictable than dynamic sharding: data location is always determinable from the primary key.

Dynamic Sharding​

The configuration, REST, Operations API and programmatic replicateTo examples below control legacy residency. On @sharded tables, replicateTo does not override the installed map: only the map determines owners. See ownership, reads, and writes.

Replication Count​

The simplest way to limit replication is to configure a per-database replication count. Set replicateTo on a database entry under replication.databases in harper-config.yaml to specify how many additional nodes data should be replicated to:

replication:
  databases:
    - system
    - name: data
      replicateTo: 2

This ensures each record in data is stored on three nodes total (the node that first stored it, plus two others). Because setting replication.databases to a list replaces the '*' default (all databases), include system and every other database the cluster should continue replicating (see Data Selection).

Replication Control via REST Header​

With the REST interface, you can specify replication targets and confirmation requirements per request using the X-Replicate-To header:

Changed in: v5.4.0 — Only a principal with the super_user permission can send X-Replicate-To. Other callers receive HTTP 403 whenever the header has a value, including on read requests and for values such as 0, a node list, or *. Application code can still set replication options on a trusted context; see Programmatic Replication Control.

PUT /MyTable/3
X-Replicate-To: 2;confirm=1
  • 2 — replicate to two additional nodes
  • confirm=1 — wait for confirmation from one additional node before responding

Specify exact destination nodes by hostname:

PUT /MyTable/3
X-Replicate-To: node1,node2

The confirm parameter can be combined with explicit node lists, or with * to require confirmation while keeping the database's configured replication (no per-request residency; passing X-Replicate-To requires super_user permissions, and non-superuser write paths can set context.replicatedConfirmation in a Resource method):

PUT /MyTable/3
X-Replicate-To: *;confirm=1

Replication Control via Operations API​

Specify replicateTo and replicatedConfirmation in the operation body:

Changed in: v5.4.0 — These operation fields require the super_user permission. For operations that accept these fields, a non-super-user caller receives HTTP 403 when either field has a non-null value, including replicateTo: 0 and an empty node list. Application code can still set them; see Programmatic Replication Control.

{
	"operation": "update",
	"schema": "dev",
	"table": "MyTable",
	"hashValues": [3],
	"record": {
		"name": "John Doe"
	},
	"replicateTo": 2,
	"replicatedConfirmation": 1
}

Or specify explicit nodes:

{
	// ...
	"replicateTo": ["node-1", "node-2"],
	// ...
}

Programmatic Replication Control​

Set replicateTo on the context in a resource method. It applies to the writes made after you set it. Here the application, not the client, decides how widely an order is replicated:

export class Order extends tables.Order {
	put(record) {
		const context = this.getContext();
		context.replicateTo = record.total >= 10000 ? 3 : 1; // or an array of node names
		return super.put(record);
	}
}

replicatedConfirmation is read when a transaction starts. A request's transaction has already started when your resource method runs, so setting replicatedConfirmation on its context there has no effect. To wait for peer confirmation from code, pass it in a new context object to transaction(). Do not pass the request's context, or a copy of it: it already holds the request's transaction, so transaction() joins that transaction and the count is ignored.

import { tables, transaction } from 'harper';

await transaction({ replicatedConfirmation: 1, replicateTo: 2 }, async (txn) => {
	await tables.Order.put(order, txn);
});

That transaction commits on its own, separately from any request transaction that is running: if the request fails afterwards, this write stays committed. Set replicateTo on the same object when you need both.

Values your code sets on a context apply for every caller, including users without the super_user permission. Derive them from application logic, not from request input such as headers, query parameters, or the record body: copying a client-supplied value into replicateTo or replicatedConfirmation gives that client the control the super_user check removes.

On @sharded tables, replicateTo does not change placement; the database map determines owners. Explicit replicatedConfirmation counts acknowledgments from other owners for each affected owner set in the first written database. A local temporary replica cannot satisfy it. See ownership, reads, and writes.

Confirmation outcomes​

Changed in: v5.4.0

LMDB now waits for requested peer confirmation, as RocksDB does. This can add peer latency to existing LMDB applications that use confirm or replicatedConfirmation. Explicit confirmation still applies only to the first database written by the transaction.

Confirmation Failure After a Local Commit​

Added in: v5.4.0

A replication confirmation failure can occur after the write has committed locally. When that happens, the rejected commit error and the REST or operations API error body include replicationUnconfirmed: true and retryable: false. Terminal JSON, NDJSON, and SSE stream error records also carry these fields. Every local database write in that commit's linked transaction has finished committing, but the requested peer confirmation was not obtained. This provides the same local durability as a successful write response; it does not add an fsync guarantee.

For example, a REST error for a confirmation timeout can contain:

{
	"type": "error:ServerError",
	"code": "ServerError",
	"title": "Timed out after 900000ms waiting for replication confirmation from 2 node(s) for database \"data\" (received 1)",
	"status": 504,
	"instance": "/MyTable/3",
	"replicationUnconfirmed": true,
	"retryable": false
}

The existing error status, code, and message remain available. An operations error has its usual error field plus code, replicationUnconfirmed: true, and retryable: false. Programmatic errors wrap the original rejection in cause; text-only protocol errors use the new ReplicationConfirmationError class name. Check this flag before compensating or retrying a non-idempotent write, because replaying the request could repeat an insert or apply an update twice. The flag does not return auto-generated record keys; use client-selected identifiers and apply the application's concurrency controls when you need to inspect the written records after a failed response.

The flag covers writes issued to the reported commit, not later application steps that were never reached after its rejection. If an enclosing scope rolls back its pending writes, its error omits the flag while preserving retryable: false because an independent inner write may already have committed.

Pre-commit refusals, including a confirmation count that cannot be met, do not carry this flag. Its absence does not establish that nothing was written: independent database commits can fail partially. A rejected commit ends its transaction scope; start a fresh transaction for subsequent writes.

Confirmation currently applies to the first database a transaction writes to. This outcome flag covers the linked local commits; it does not enable confirmation for later databases.

Static Sharding​

Basic Static Shard Configuration​

Assign a node to a numbered shard in harper-config.yaml:

replication:
  shard: 1

Complete shard-id set​

Added in: v5.4.0

For the first default map of a new schema-defined sharding table, configure replication.shards next to replication.shard:

replication:
  shard: 0
  shards: [0, 1]

replication.shards is the complete set of shard IDs available to the deployment: unique integers from 0 through 65535. Every node uses the same set; order does not matter. Its own replication.shard must be in the list.

The list is required only for a database that has declarative tables and no installed slot map. First activation computes the default map solely from the slot count and the sorted list and installs it as a versioned snapshot. Concurrent installers with equal inputs produce the same snapshot and digest. A database converted with a custom map does not need the list.

Once installed, the snapshot is authoritative. Changing the list never recomputes placement. Adding an ID only makes that shard available for future reassignment; it moves no records.

Setup exchanges include a digest of the list. Nodes with different lists refuse declarative writes on both sides, even after a map is installed. A configured list on one peer and no list on another also differs. A node whose own shard is outside its list, or which needs an initial map but has no list, also refuses placement. The condition appears in cluster_status.sharding and is logged once. Plan for this write refusal during a rolling list change.

Other shard configuration​

Shards can also be assigned per route:

replication:
  routes:
    - hostname: node1
      shard: 1
    - hostname: node2
      shard: 2

Or dynamically via the operations API by including shard in an add_node or set_node operation:

{
	"operation": "add_node",
	"hostname": "node1",
	"shard": 1
}

Once shards are configured, use schema-defined sharding for a schema policy applied on every thread, or the function APIs below for legacy application-controlled placement.

Custom Sharding​

By Record Content (setResidency)​

Define a custom residency function that is called with the full record. Return an array of node hostnames or a shard number.

With this approach, record metadata (including residency information) and indexed properties are replicated to all nodes, but the full record is only stored on the specified nodes, apart from the short-lived replicas described next.

Changed in: v5.4.0 Engine: RocksDB

On RocksDB, a node outside a record's residency that reads the record fetches it from a resident node and keeps the result as a local replica for replicaExpiration seconds (default one hour), renewed by reads after replicaRefreshFraction of that time. The record stays where the residency function placed it: a replica is never written back, replicated onward or counted as a resident, and a resident's write turns it back into a stub so the next read fetches the new version. See the @table arguments. On the deprecated LMDB engine, remote reads are always proxy reads; no local replica is stored and no replica sweep runs.

Return node hostnames:

MyTable.setResidency((record) => {
	return record.id % 2 === 0 ? ['node1'] : ['node2'];
});

Return a shard number (replicates to all nodes in that shard):

MyTable.setResidency((record) => {
	return record.id % 2 === 0 ? 1 : 2;
});

By Primary Key Only (setResidencyById)​

Define a residency function based solely on the primary key. A read from another node is answered by a resident and keeps no local replica. Records (including metadata) are only replicated to the specified nodes — metadata does not need to be replicated everywhere, which allows data to be retrieved without needing access to record data or metadata on the requesting node. This does not hold with dedicated replication threads (replication.threads above 0), which do not run the residency function.

Return a shard number:

MyTable.setResidencyById((id) => {
	return id % 2 === 0 ? 1 : 2;
});

Return node hostnames:

MyTable.setResidencyById((id) => {
	return id % 2 === 0 ? ['node1'] : ['node2'];
});

Disabling Cross-Node Access​

By default, a key read for a record not stored locally fetches it from a node that holds it. To disable this and only return data stored on the local node (including a replica it currently holds), set replicateFrom to false.

Via the operations API:

{
	"operation": "search_by_id",
	"table": "MyTable",
	"ids": [3],
	"replicateFrom": false
}

Via the REST API:

GET /MyTable/3
X-Replicate-From: none

See Also​