Cluster Monitor

Cluster Monitor

Roadmap, not a currently supported deployment mode. Multi-node operation is not implemented as a production feature in commit c42e19eba. This page deliberately previews the planned architecture. Do not point several MemCP processes at the same writable database or shard today.

MemCP currently manages databases and shards inside one process. Local files, S3-compatible object storage, and optional Ceph/RADOS provide persistence backends; they do not by themselves provide a distributed database coordinator. The dashboard and process monitor observe one running instance and are not a cluster-control plane.

Intended use cases

The original cluster-monitor design targeted two complementary forms of scale-out:

• Database placement: different tenants or databases can be assigned to different nodes.
• Shard placement: shards of one large table can be distributed across several nodes.
• Mixed placement: frequently shared reference data can coexist with independently owned writable shards.

These remain architectural goals rather than documented production features.

Existing local resource model

MemCP already exposes a SharedResource lifecycle for lazily loaded resources:

State	Current local meaning	Possible cluster meaning
COLD	The resource is not loaded in memory.	No node currently holds a local readable copy.
SHARED	The resource is acquired for reading.	One or more nodes may hold a read-only copy.
WRITE	The resource is acquired exclusively for mutation.	Exactly one fenced owner may modify the resource.

Callers acquire capabilities through GetRead() or GetExclusive() and release them through the returned function. The current implementation tracks local lifetime and exclusivity. Its interface is deliberately capable of evolving toward remote loading or routing, but it does not currently negotiate ownership between processes.

Planned cluster modes

The design work distinguishes three modes so distributed machinery does not slow down small installations:

Mode	Catalog and coordination	Intended scale
Single node	Current local registries and persistence	Lowest latency and operational complexity
Low-scale cluster	Replicated membership plus a CRUSH-assigned directory for active tables and shards	A bounded number of nodes and catalog objects in one data center
High-scale cluster	Partitioned catalog relations, stable handles, lazy local materialization and bounded working-set caches	Very large numbers of databases, tables and shards without every node holding global registries

Switching catalog modes is planned as an explicit migration, not an automatic threshold. Single-node operation remains a first-class fast path.

Planned low-scale coordination model

The current design in the project roadmap replaces the earlier ZooKeeper sketch with a leaderless, CRUSH-directed directory and MOESI-inspired cache states:

• every table and shard receives a deterministic directory node from its stable identity and the current CRUSH membership map;
• the directory node coordinates ownership but does not have to cache the data itself;
• directory state exists only for active objects and records shared holders, an exclusive holder and waiting writers;
• local cache states are Modified, Exclusive, Shared or absent/Invalid;
• reads may retain several shared copies, while mutation requires one fenced exclusive owner;
• invalidation targets only actual sharers instead of broadcasting every shard event to every node;
• RADOS is the planned persistent source of truth for this mode; cache ownership and durable storage are separate concerns.

The low-scale proposal assumes nodes in one data center and does not attempt to remain writable through an arbitrary network partition. A temporarily unreachable member causes affected work to wait or time out; permanent isolation advances the membership epoch and reassigns directory responsibility. This limitation must remain visible rather than being described as general geo-distributed consensus.

The former wiki text proposed ZooKeeper paths such as /memcp/resources/<resource-id>/owner. That design has been superseded: the roadmap explicitly aims for no ZooKeeper, Raft, gossip service or additional coordination product.

Query routing and distributed execution

Every node is intended to accept and coordinate a query. A database may have a preferred node as a cache-locality and load-balancing hint, never as the only execution point. HTTP can redirect suitable requests; a MySQL connection would require transparent tunnelling or local coordination because the protocol has no database-node redirect.

The existing two-stage scan/reduce architecture is a useful foundation for remote work. The planned flow sends a serializable filter/map/reduce program to a node holding relevant shards, performs filtering and partial aggregation beside the data, and returns compact partial states. SUM, COUNT, MIN and MAX combine naturally; AVG transports SUM and COUNT, while variance needs an algebraic partial state. GROUP BY should combine several aggregates in one remote pass instead of issuing one RPC per aggregate.

Ordered scans produce local sorted runs or bounded top-k states. Intermediate nodes merge them hierarchically and apply range braking; a single coordinator should not receive every candidate row. Cancellation, deadlines, visibility snapshots, code serialization and version compatibility are required parts of the wire contract.

High-scale catalog roadmap

At very large catalog sizes, a complete database/table list on every node becomes the bottleneck even if data shards are distributed. The proposed DistributedCatalog therefore uses:

• stable database/table IDs with generations instead of persisted process pointers;
• partitioned name and ID indexes plus versioned table definitions and shard-manifest trees;
• bounded node-local caches containing only the active working set;
• paginated database, table, process and administration listings;
• scoped settings and partitioned users/grants rather than one growing mutable global object;
• range leases for AUTO_INCREMENT, partitioned unique-key ownership and routed foreign-key reference indexes;
• partitioned jobs, logs, metrics and transaction intents with epochs and idempotent takeover;
• hierarchical GROUP BY, DISTINCT, UNION, ORDER BY and top-k reduction without a global heap or hash map.

The hard scaling objective is that one node's RAM and background work grow with its working set and CRUSH share, not with the total number of databases or tables in the cluster.

Planned implementation sequence

1. Introduce a common catalog contract while preserving the current local fast path.
2. Add stable handles, generations, immutable definitions and version-aware plan-cache keys.
3. Implement membership epochs, CRUSH directory assignment and authenticated node transport.
4. Add binary Scheme/value framing and remote unordered scan/reduce.
5. Add exclusive ownership, invalidation, write forwarding and durable handover.
6. Implement distributed ordered scans, grouped partial aggregation and cancellation.
7. Add partitioned catalogs, lazy handles, paginated administration and high-scale system relations.
8. Provide migration, failure injection, backup/restore, rolling-upgrade and scale tests before declaring production support.

No roadmap item has a promised release date. The design is intentionally published early so users can evaluate direction and contributors can discuss invariants before interfaces become stable.

Requirements before production support

A multi-node implementation must define and test at least:

• authenticated node identity and encrypted transport;
• leases, fencing tokens, expiry, and split-brain prevention;
• discovery and membership changes;
• shard/database placement and rebalancing;
• query routing, cancellation, retries, and partial-failure behavior;
• transaction boundaries across remotely owned data;
• WAL ownership, replay, rebuild, and handover ordering;
• cache invalidation and generation publication;
• behavior during coordinator, node, network, and backend outages;
• backup, restore, rolling upgrade, and downgrade procedures;
• observability for ownership, lease age, routing, lag, and failed handovers.

Until those contracts exist, deploy each writable MemCP database under one server process and use the selected persistence backend according to Persistency and Performance Guarantees. See Storage Backends and Shards, RecordIDs, Main Storage, Delta Storage.