ScaleMesh

Several Propus servers acting as one: connect a server from the Control Center and the rest happens on its own. Apps keep their connection URL, SQL, SDKs and keys.

Status: phase 1 (the cluster) is built, and so is data spread over the servers: in a cluster of two or more servers every database is spread (its rows and indexes in 256 slots dealt over the servers, each server's share in its own file). New databases are spread when they are made; databases made before the cluster are moved onto it on their own, while apps keep writing. Transactions across servers commit everywhere or nowhere, even when a server stops mid-commit; BEGIN … COMMIT, the change feed, the audit log and search work on a spread database as on one server. When a server joins, slots move to it until every server holds about the same share; removing a server first moves its slots to the others. Every data slot has a copy on a second server, and a server that stops is taken over by its copies. A spread database runs its requests on its home server (the one it was made on): any server takes its HTTP requests and passes them on to the home (the key is checked there; a transaction begun that way follows it); WebSockets and the Postgres protocol connect to the home. The home itself has no copy yet. Each step ships only after it passes its tests on a Docker cluster of real Propus servers; tested so far with four servers.

#Connecting a server

1. Control Center › Cluster › Add a server shows one command, valid for an hour and for one server:

   curl -fsSLko propus-install.sh https://db.example.com/install.sh && echo '<sha256>  propus-install.sh' | sha256sum -c - && sudo bash propus-install.sh --join 'propus-join:…'
   

2. Run it on the new machine. The installer is checked against its SHA-256 before it runs; it downloads this cluster's own Propus build from the server that made the invitation, checks it against the SHA-256 the invitation carries, installs it, and joins.

3. The server appears in the Cluster page within seconds, with its health.

The invitation carries the cluster's address, a one-time token and the fingerprint of the cluster's certificate authority, so the new server trusts only this cluster and the cluster accepts only this server. Servers talk to each other over mutual TLS with certificates from the cluster's own authority, made when the first invitation is created. Nothing is copied by hand.

An already installed server joins with sudo propus join 'propus-join:…'.

#How the cluster works (phase 1)

• Raft. The servers keep one replicated record of the cluster (members, invitations, the certificate authority) with Raft: pre-vote, check quorum and leader stickiness keep a returning server from unseating a working leader. An odd number of servers vote (1, 3 or 5), so any minority can be down without stalling the cluster; the others follow without a vote, which keeps decisions quick. A new server joins as a learner and votes once it has caught up. Any majority of the voters keeps the cluster working.
• Connections. Servers talk over TCP port 7960 (PROPUS_MESH_LISTEN), mutual TLS with certificates from the cluster's authority. A server's address is found from its join connection, or set with PROPUS_MESH_PUBLIC.
• Health. Every server reports its version, uptime, CPU, memory, free disk and databases every two seconds; one not heard from for six seconds shows as unreachable.
• Restarts. A server that restarts catches up from the leader; when every server restarts at once (a power cut, an upgrade), each finds the others from its saved log and they elect a leader again.
• Leaving. Removing a server revokes its certificate; it is told, stops taking part and runs alone again (its cluster files are kept aside).
• Requests between servers are sent again until answered (the leader may change on the way) and applied once, by request id.
• API. GET /v1/cluster, GET /v1/cluster/databases (spread databases and the keys each server holds), POST /v1/cluster/invites, DELETE /v1/cluster/members/{id} (server admin); GET /install.sh and GET /v1/cluster/binary?token= for joining servers.

#Moving data

• A database made before the cluster. Once there are two servers, each server's own databases are moved onto the cluster: the database's file becomes that server's share (every slot), writes to the old file are refused from that moment with 40001 (drivers and SDKs run the transaction again, now over the cluster), and the slots then spread like any other database's. The system database never moves.
• Even spread. Every 3 seconds the home server of each spread database checks its slots against the servers taking data and moves up to 32 slots from the server with the most to the one with the fewest, until each holds its share. Slot 0 (the catalog and settings) stays on the home.
• How a slot moves. The new owner copies the slots while writes go on; the slots are frozen for a moment (writes to them wait and retry), the keys changed during the copy are copied again, and one Raft decision hands the slots over and raises the layout's version; the old owner then drops them. A request made with an older layout is refused with 40001 and runs again with the new one, so no write goes to the wrong server. A move cut off by a restart is given up and planned again; one that fails is tried again after 30 seconds, then less and less often (at most every 15 minutes), so it never keeps a server busy.
• Removing a server that holds data. The server is marked Draining: it takes no new slots, its slots move to the others, and the cluster removes it once it holds nothing. Slots go first to servers that don't keep their copy; when the servers left keep every copy (two servers, say), the one taking a slot lets its copy go first and a new copy is made after, so a drain always ends. A server holding no data leaves at once.
• Seeing it. Control Center › Cluster shows each server's slots and, per spread database, its home, layout version, the move under way and the keys each server holds. GET /v1/cluster/databases returns the same.

#What happens on its own (all phases)

The Control Center shows every choice it made (each table's key, where each slot lives) and lets you change one, but nothing has to be set.

#Design of the data layer (phases 2–6)

The SQL engine does not change. What changes is the storage under it: in a cluster, a database's keys (rows, index entries, the catalog) are spread over the servers, and a transaction runs on whichever server took the request, reading from and writing to the servers that hold each key. Every SQL feature (joins, triggers, constraints, row-level security, sequences) works as on one server because the same engine runs it.

• Slots. A database's key space is split into 256 slots by a hash of each key. The catalog and settings live in slot 0 (the database's home). The cluster state (Raft) holds which server owns each slot.
• Transactions (optimistic). The server running a transaction reads from the slot owners, records every key and range it reads, and keeps its writes aside (the same overlay as parallel writes on one server).
• Turns, as on one server. On one server a transaction holds the database's writer, so conflicting writes wait instead of failing; a spread database gives the same turns. An interactive transaction (BEGIN … COMMIT) waits for a turn alone and keeps it to COMMIT; single requests write side by side, and one that loses a race runs again alone, where it cannot lose (for a second after, writes queue for that at once instead of losing too); reads take no turn, and a read that keeps seeing changes mid-way (a long report over busy rows) gets a turn alone after a few tries, so it finishes instead of giving up. A slot move's hand-over takes a turn alone, so no transaction sees slots change under it. Apps see the same waits as on one server and never a 40001 from a conflict one server would have queued. (Commits are still checked: turns decide who waits, not what is correct.)
• Commit (two-phase). The servers holding written or read slots each lock the keys, check that nothing the transaction read has changed, and vote. If all vote yes, the decision is recorded durably in the transaction's record on its home server, then sent to every participant, which applies its writes. A read that meets a key locked by a prepared transaction waits for its decision, so no reader sees half of a commit.
• Crashes. A participant records its prepared writes durably before voting; after a restart it asks the transaction's record whether to commit or abort. A transaction with no recorded decision after a timeout is aborted by whoever finds it.
• Sequences. Blocks of numbers are taken from the home slot and handed out locally.
• Moving slots (phase 5). A slot is copied to its new owner, then catches up from its changes, then ownership moves in one Raft decision.
• Copies (phase 6). Every data slot (1–255) has a copy on a second server, made in the background as a slot move is (copied live, frozen a moment, caught up, one Raft decision). From then on each commit writes the owner and the copy in the same two-phase commit, so the copy holds every answered write. When a server has been silent 3 seconds past being marked offline, the leader hands its slots to their copies in one Raft decision (a new layout version) and the copies it kept are dropped; the cluster then makes new copies on its own, onto the servers that are left or onto the server when it returns (it rejoins holding nothing and is filled again). Slot 0 (the catalog, settings, change feed and audit log) has no copy: it stays on the home.

#When a server fails

Tested on Docker under a social-network load (tests/mesh/faults.sh, docs/testing.md): a server switched off, one killed, one cut off the network, the leader stopped, a server joining and one being removed, with copies made again after each fault and every business rule checked at the end (no request failed, all rules held, nothing lost or doubled).

• A server stops, is killed or is cut off: its copies take over within a few seconds. Requests that were waiting on it run again on the new layout (a transaction in BEGIN … COMMIT gets 40001, which drivers run again); nothing fails, nothing is lost or doubled. When it comes back it holds nothing, takes copies and slots again, and every slot has a copy again within about a minute.
• The leader stops or is cut off: the others elect a new one within seconds; requests go on.
• A server is removed: it drains (above) and leaves.
• The whole cluster restarts: the servers find each other from their saved logs and elect a leader.
• Not yet: the home stops. A spread database's requests run on its home server (slot 0, with the catalog and the change feed, has no copy), so while the home is off, that database waits for it. Copying the home and forwarding requests from any server come next.

#What to expect

• A query that names the shard key is as fast as on one server; a query across everything asks every server and is slower.
• A transaction that touches two servers waits for both.
• Writes that touch the same rows go one at a time, as on one server, and each read from another server costs a network round trip (about 1 ms between containers on one machine), so a workload whose every transaction updates the same few rows (an e-commerce checkout posting to the same ledger accounts) runs slower spread than on one server: measured with the Control Center's 2,000-user e-commerce test on one 4-core machine, 55 transactions a second on one server and 17 on five (no errors either way). Spreading pays off on separate machines and for writes to different rows.
• More copies make the cluster survive failures; more servers raise write and storage capacity.

#Phases

1. Cluster (built) — Raft membership, the cluster's certificate authority, invitations, --join, health, removing a server, the Cluster page.
2. Shard keys and placement — automatic keys, slots, placement, routing writes and single-shard reads.
3. Queries across servers — scatter and merge, joins on the shard key, reference tables.
4. Transactions across servers (built) — two-phase commit with recovery after a crash; BEGIN … COMMIT over the cluster.
5. Rebalancing (built) — moving slots live when servers join or leave; databases made before the cluster moved onto it.
6. Copies and failover (built, except the home) — a copy of each data slot on another server, automatic takeover, copies made again on their own. Next: the home's catalog, keys and change feed copied, so the home can fail over too.