Sharding & Clustering

Purple8 Hyper Graph scales horizontally with a sharding layer built on top of the single-node GraphEngine. You choose a partitioning strategy — hash, label, or range — and the ShardedGraphEngine coordinates across shards transparently.

Architecture overview

                      ┌──────────────────────────────┐
                      │   ShardedGraphEngine          │
                      │   ┌──────────────────────┐   │
                      │   │ FederatedQueryEngine │   │
                      │   └──────────────────────┘   │
                      └────────────┬─────────────────┘
                ┌───────────────┬──┴──────────────┐
           Shard 0          Shard 1           Shard 2
       GraphEngine()    GraphEngine()     GraphEngine()
       ./data/s0        ./data/s1         ./data/s2

Each shard is a full GraphEngine backed by BrickCoreStorage (Purple8 Cortex) with its own BrickWAL, CortexAutoConfig memory envelope, HNSW index, and KMS config. The FederatedQueryEngine merges results from all shards (scatter-gather) for MATCH and CALL db.vector.search.

Partitioning strategies

Hash partitioning (default)

Entities are distributed by consistent hashing on node ID. Balances evenly across shards. Best for general-purpose workloads.

from purple8_graph.sharding import ShardedGraphEngine, HashPartitioner

engine = ShardedGraphEngine(
    shard_configs=[
        {"path": "./data/shard-0"},
        {"path": "./data/shard-1"},
        {"path": "./data/shard-2"},
    ],
    partitioner=HashPartitioner(),
)

# Usage is identical to a single-node GraphEngine
engine.add_node("Person", {"name": "Alice", "role": "engineer"})
results = engine.query("MATCH (p:Person) RETURN p.name LIMIT 10")

---

Label partitioning

Assigns node labels to specific shards. Best when workloads split clearly along data type lines (e.g., hot financial data on Shard 0, cold archive on Shard 2).

from purple8_graph.sharding import LabelPartitioner

engine = ShardedGraphEngine(
    shard_configs=[
        {"path": "./data/shard-0"},
        {"path": "./data/shard-1"},
        {"path": "./data/shard-2"},
    ],
    partitioner=LabelPartitioner({
        "Transaction":  0,   # all Transaction nodes → shard 0
        "Account":      0,
        "Customer":     1,
        "Document":     2,
        "Attachment":   2,
    }),
)

Cross-shard edges are resolved at query time by the FederatedQueryEngine.

---

Range partitioning

Partitions by a numeric property value range. Best for time-series or ordered data.

from purple8_graph.sharding import RangePartitioner

engine = ShardedGraphEngine(
    shard_configs=[
        {"path": "./data/2022"},
        {"path": "./data/2023"},
        {"path": "./data/2024"},
    ],
    partitioner=RangePartitioner(
        property="year",
        ranges=[(2022, 2022), (2023, 2023), (2024, 2025)],
    ),
)

Federated queries

MATCH and CALL db.vector.search are automatically federated across all shards. The FederatedQueryEngine issues parallel queries and merges results:

# This Cypher runs on all 3 shards and merges the results
results = engine.query("""
    CALL db.vector.search('Document', $vec, 10)
    YIELD node, score
    MATCH (node)-[:AUTHORED_BY]->(author:Person)
    RETURN node.title, author.name, score
    ORDER BY score DESC
    LIMIT 5
""", vec=embedding)

Merge semantics:

• RETURN results are concatenated, then any ORDER BY / LIMIT applied globally
• COUNT(*) is summed across shards
• AVG() is weighted by per-shard row count
• Vector search: top-k is collected from each shard, merged, and top-k re-selected globally

High-availability with Raft

The RaftShardedEngine wraps each shard in a Raft consensus group for leader election and automatic failover. Each shard can have 1, 3, or 5 replicas.

from purple8_graph.raft import RaftShardedEngine
from purple8_graph.sharding import HashPartitioner

engine = RaftShardedEngine(
    shard_groups=[
        {
            # Shard 0 — 3 Raft replicas
            "replicas": [
                {"host": "node-0a.internal", "port": 9010, "path": "/data/s0"},
                {"host": "node-0b.internal", "port": 9010, "path": "/data/s0"},
                {"host": "node-0c.internal", "port": 9010, "path": "/data/s0"},
            ]
        },
        {
            # Shard 1 — 3 Raft replicas
            "replicas": [
                {"host": "node-1a.internal", "port": 9010, "path": "/data/s1"},
                {"host": "node-1b.internal", "port": 9010, "path": "/data/s1"},
                {"host": "node-1c.internal", "port": 9010, "path": "/data/s1"},
            ]
        },
    ],
    partitioner=HashPartitioner(),
    election_timeout_ms=300,
)

Leader election happens within election_timeout_ms of a leader failure. Writes are replicated synchronously (quorum-based). Reads are served from the leader by default; use read_consistency="eventual" to serve from any replica.

Primary / replica (WAL streaming) with automated failover

The recommended topology for read scaling and two-host availability without Raft: one primary, N replicas. The primary streams its WAL to each replica over a dedicated port (8101); replicas apply it to their own storage and serve reads. Any replica can be promoted.

# Primary
P8G_REPLICATION_ROLE=primary
P8G_REPLICATION_BIND_HOST=0.0.0.0
P8G_REPLICATION_TLS=1
P8G_REPLICATION_TLS_CERT=/run/tls/primary.pem
P8G_REPLICATION_TLS_KEY=/run/tls/primary.key
P8G_REPLICATION_TLS_CA=/run/tls/ca.pem
P8G_REPLICATION_MODE=sync              # success requires the configured follower acknowledgements
P8G_REPLICATION_SYNC_REPLICAS=1
P8G_REPLICATION_SYNC_TIMEOUT_S=5
P8G_TENANT_REPLICATION_MODE=auto       # auto | manual on replicas
P8G_FOLLOWER_ADDRESSES=replica-1:8101
P8G_REPLICATION_AUTH_TOKEN=<shared secret>

# Replica
P8G_REPLICATION_ROLE=follower
P8G_REPLICATION_TLS=1
P8G_REPLICATION_BIND_HOST=0.0.0.0
P8G_REPLICATION_TLS_CERT=/run/tls/replica.pem
P8G_REPLICATION_TLS_KEY=/run/tls/replica.key
P8G_REPLICATION_TLS_CA=/run/tls/ca.pem
P8G_PRIMARY_HOST=primary
P8G_PRIMARY_PORT=8101
P8G_REPLICATION_AUTH_TOKEN=<shared secret>
P8G_FAILOVER_ENABLED=1                 # run the failover watchdog here
P8G_FAILOVER_PRIMARY_ADDRESS=primary:8080
P8G_NODE_ADDRESS=replica-1:8080
P8G_FAILOVER_TARGET_BACKEND=noop       # elbv2 on AWS

Replication fails before starting network threads if authentication or TLS is missing. The default listener binds loopback; cross-host listeners must be configured explicitly. Promoted followers retain their authentication, TLS, binding, and synchronous-ack settings.

Synchronous write confirmation

sync requires the configured number of follower acknowledgements; the primary never reduces that number to match the connected followers. No followers, insufficient followers, a disconnect, or an acknowledgement timeout cannot produce a successful confirmation. Invalid replication roles or modes fail startup rather than silently selecting standalone or asynchronous operation. Synchronous replica counts must be positive integers and the timeout must be finite and positive.

The local mutation may already have committed when confirmation fails:

• Embedded Python writes raise ReplicationAckError.
• REST returns HTTP 503, a readable detail, and a structured replication object with reason="replication_ack_unconfirmed", local_write_may_have_committed=true, and blind_retry_safe=false.
• The HTTP Python SDK raises ReplicationUnconfirmedError; its detail contains the replication object.
• MCP returns a protocol error (isError=true) preserving the REST error details.

Do not blindly retry: reconcile the affected records first. A transaction that encounters this uncertainty is marked failed and propagates the error; that is not confirmation of rollback. Batch publication buffers all already-written records before waiting on the final sequence, so a failed first acknowledgement does not omit the rest of that batch. Local indexing and deletion cleanup finish before confirmation is requested.

Authenticated operators can inspect GET /replication/status: sync_required reports the configured requirement and last_sync_failure retains the latest unconfirmed sequence, required and received acknowledgement counts, and retry warning. async remains nonblocking and does not promise follower confirmation.

Tenant control plane and routing

Tenant definitions travel in a separate control message, not the graph WAL. The definition includes the tenant ID, name, isolation mode, quota, status, account type and capability flags. It excludes users, credentials, IAM/SSO role mappings and arbitrary tenant metadata. Configure identity integration separately on each deployment; this path does not copy it or replace the IdP.

Graph mutations carry their physical storage scope. Existing and newly opened tenant engines join the same replication stream automatically:

• Dedicated tenants route to their own storage, never the default tenant.
• Shared tenants retain their existing namespaced IDs; the follower does not prefix them a second time.
• Sharded tenants retain shard identity. Cross-shard edge copies finish locally before their acknowledgement barrier, including when confirmation fails.
• Closing one tenant engine does not stop replication for other tenants.

Choose the replica's provisioning policy through P8G_TENANT_REPLICATION_MODE or ReplicationConfig.tenant_replication_mode:

Mode	Behaviour
auto (default)	Mirror primary-approved definitions and create missing tenants. Do not independently charge or recheck tenant-license slots. Existing local identity configuration and custom metadata stay local.
manual	Require matching tenant IDs, isolation, quota, status, account type and capabilities. Missing or mismatched tenants remain unready; no graph writes are silently queued or redirected to another tenant.

Local operator-created tenants still use TenantManager.create_tenant() and its normal license-cap checks in either mode. For manual provisioning, provision through that SDK before starting the replica, using the primary's approved configuration. Do not edit the live registry through another process.

An isolation mismatch is not an automatic storage migration: correct it explicitly before admitting the replica. Retired mirrored tenants are suspended; their data is not physically erased by this control plane.

Readiness and operations:

• GET /replication/status exposes aggregate tenant_replication readiness, revision, mode, unready count and promotion_ready.
• GET /replication/tenants requires admin:settings and identifies individual missing or mismatched tenants. Public health does not disclose these IDs.
• GET /health reports degraded status for unready configuration, an outstanding fresh-handshake gate, or replication application failure. The failover health probe respects these signals.
• Promotion fails closed while configuration is unready. A received catalog is fsync'd before confirmation; disk or engine-open failures are not success. A saved catalog restores configuration after restart, but a fresh primary tenant handshake is required before promotion. Readiness is checked again after follower workers stop; if that final check fails, the node remains a stopped follower until repaired and restarted.
• Tenant-aware streams reject legacy peers without the control-plane protocol; they never fall back to default-tenant application. An advertised cursor ahead of the primary's known sequence also requires resynchronization. Explicit protocol rejection invalidates prior readiness; ordinary network loss alone does not discard a previously confirmed catalog.

Operational scope: Tenant routing, provisioning, and synchronous acknowledgement checks govern live streaming writes and replica promotion gates. Readiness reflects the received primary catalog; if a replica has been network-isolated for an extended window, verify sequence convergence and catalog parity via GET /replication/status and GET /replication/tenants before manual intervention.

Automated promotion uses verified HTTPS without redirects. Advertised API addresses must therefore serve HTTPS with certificates trusted by P8G_REPLICATION_TLS_CA (or the system trust store), either directly or through the deployment's existing TLS ingress.

For isolated local development only, P8G_ENV=development together with P8G_REPLICATION_ALLOW_INSECURE_LOCAL=1 permits plaintext/tokenless WAL traffic when both the listener and follower destination are loopback. This exception cannot expose a network interface; automated promotion still requires a token.

Behaviour:

Aspect	Detail
Reads on replicas	Full HTTP API. Eventually consistent; GET /health returns replication_lag (ops behind primary) and replication_role.
Writes on replicas	503 + Retry-After — fail closed, never diverge. Route write verbs to the primary at the load balancer.
Failover	The replica probes the primary's /health every P8G_FAILOVER_INTERVAL_S; after P8G_FAILOVER_UNHEALTHY_THRESHOLD failures it promotes itself (POST /replication/promote, authenticated with the replication token) and calls the P8G_FAILOVER_TARGET_BACKEND hook to repoint the load balancer. Cooldown P8G_FAILOVER_COOLDOWN_S.
Startup gate	Failures are not counted until the primary has been seen healthy once, or P8G_FAILOVER_STARTUP_GRACE_S (default 300 s) has elapsed — so a replica that boots alongside a still-starting primary does not promote itself against it.
Split-brain	Promotion claims a new replication epoch; a stale primary that reconnects is fenced. With >1 replica, enable P8G_RAFT_ENABLED so the election guard and STONITH apply.
Rejoin	Restart the old primary with P8G_REPLICATION_ROLE=follower pointing at the new primary. Wipe its data directory first (or restore from the new primary's latest backup): the old volume holds a diverged WAL tail from writes accepted before fencing, and the epoch check will reject it.

Surviving instance replacement (slot-based role discovery)

Promotion repoints the load balancer, but it does nothing about the old primary's launch template. If that instance is later replaced by its Auto Scaling group — crash, patching, scale event — the new instance boots from the same template and, if the template hard-codes ROLE=primary, comes back as a second primary. Nothing alerts; the split brain is only noticed when data diverges.

Purple8 closes this with a slot record. A slot is a stable identity for an ASG position (a / b), independent of role, instance ID or hostname:

1. Every node is started with P8G_NODE_SLOT=<its slot> and P8G_FAILOVER_PRIMARY_SLOT_PARAM=<SSM parameter>.
2. Instance user data reads the parameter at every boot and derives its role from it: my slot == recorded slot → primary, else follower. If the parameter cannot be read the boot fails closed (exit non-zero, ASG retries) rather than guessing.
3. On promotion the watchdog writes the new primary's slot to the parameter (SsmPrimarySlotRecorder) and, if configured, UPSERTs a stable CNAME such as primary.<instance>.internal to slot-<new>.… (Route53PrimaryAliasRecorder) for consumers that were given a fixed hostname rather than the load balancer.
4. The primary runs PrimarySlotFence: it polls the same parameter and, the moment it stops naming this node's slot, freezes every write with 503 — covering the window where the primary's instance was replaced while it was primary and the new instance read the parameter before the replica promoted.

The result: any instance in either ASG can be terminated at any time and the topology converges to exactly one writable primary without operator intervention. Rebuilt old-primary volumes still need the wipe/restore step above before they resync.

The AWS reference stack (deployment/cloudformation/aws/) wires all of this — SSM parameter, Route 53 alias, IAM, IMDSv2 hop limit for the container, fail-closed user data — with AutoFailover=1.

Full variable list: Environment variables → Automated failover. AWS reference stack with method-based ALB routing and the ELBv2 hook: deployment/cloudformation/aws/.

Snapshot-poll read replicas (legacy)

An older read-replica subsystem (replica.py) pulls periodic full snapshots from the primary instead of streaming the WAL:

P8G_REPLICA_MODE=replica
P8G_PRIMARY_URL=http://primary-host:8100

It cannot be promoted and has higher lag. Prefer the WAL-streaming topology above for new deployments.

Docker Compose: 3-shard cluster

services:
  shard-0:
    image: purple8/graph:latest
    environment:
      SHARD_ID: "0"
      SHARD_PEERS: "shard-1:9010,shard-2:9010"
      P8G_CLUSTER_MODE: "true"
      JWT_SECRET: "${JWT_SECRET}"
    volumes:
      - shard0_data:/data
    ports:
      - "8010:8100"

  shard-1:
    image: purple8/graph:latest
    environment:
      SHARD_ID: "1"
      SHARD_PEERS: "shard-0:9010,shard-2:9010"
      P8G_CLUSTER_MODE: "true"
      JWT_SECRET: "${JWT_SECRET}"
    volumes:
      - shard1_data:/data

  shard-2:
    image: purple8/graph:latest
    environment:
      SHARD_ID: "2"
      SHARD_PEERS: "shard-0:9010,shard-1:9010"
      P8G_CLUSTER_MODE: "true"
      JWT_SECRET: "${JWT_SECRET}"
    volumes:
      - shard2_data:/data

volumes:
  shard0_data:
  shard1_data:
  shard2_data:

Consistency model

Operation	Guarantee
Single-node write	Synchronous BrickWAL (fdatasync) commit + BrickCoreStorage (Cortex) & HNSW update
Raft write	Quorum commit (majority of replicas)
Read from leader	Read-your-writes
Read from replica	Eventual consistency (configurable lag)
Federated MATCH	Eventual consistency across shards
Federated vector search	Approximate global top-k

Six-Sigma Operational Controls

Cluster and primary/replica deployments include four built-in operational safeguards for production storage reliability:

1. Streaming gzip/JSONL Backups with Magic & Size Guards: Cluster and shard backups stream directly as gzip-compressed JSONL archives, verifying the 4-byte P8KM envelope encryption header on encrypted blobs and enforcing the P8G_BACKUP_MAX_ARCHIVE_MB guard on archive size.
2. Online Storage Bit-Rot Scrubber: Run POST /admin/storage/scrub (or engine.scrub_integrity()) on any primary, replica, or shard to validate BrickCoreStorage and BrickWAL checksums online without stopping traffic.
3. Disk High-Watermark Write Admission Controller: Each node enforces P8G_WRITE_ADMISSION_DISK_MAX_PCT=95.0 (default 95.0%), rejecting new writes with a clean backpressure error before a shard or primary volume can exhaust disk space during BrickWAL append or seal cycles.
4. Zero-Spike Health Probing: GET /admin/health inspects only currently active, opened tenant engines in memory — avoiding cold-tenant disk wakeups or latency spikes on multi-tenant clusters.