Introduction

magic_sync keeps partial copies of PostgreSQL tables on SQLite replicas, and brings back what the replicas write.

Picture a chain of shops. The head office runs PostgreSQL. Every cash register runs SQLite and must keep selling when the network is down. The registers hold the catalog; each register writes receipts that the head office must end up with; and each prints a footer of its own, which the head office may ask it to change — a setting, which the head office wishes and the register carries out (Settings and wishes). That is the system this book builds, one step at a time, and it is also the example you can run in five minutes.

How the book is written

Every chapter of Build your own has two kinds of block:

You write — code that is yours: copy it, change it.

What happens — what the engine does with it. You can skip these and come back when you want to know.

Under the hood is the engine from the ground up, for when you want to understand all of it: the idea it rests on, the life of a change, every failure and how it heals, and the protocol a replica in any language follows.

The idea, in one paragraph

Most sync engines replay a log of events. This one compares states: for every replica, the server keeps what that replica holds, works out what it should hold, and sends the difference. A change is a reason to compare again, never an event that must be delivered. That one choice is why a replica can be away for a week and catch up from the current state, why a failing row does not block the ones behind it, and why nobody ever repairs data by hand.

Run the shop

The example is two small programs: the head office (PostgreSQL) and a register (SQLite), each with a page in the browser. You will start one head office and two registers, edit the catalog, take a register offline, sell while it is offline, and bring it back online.

You need Rust 1.85 or later, Node 22.13 or later with pnpm, and PostgreSQL 14 or later (Docker brings one). Every command below runs from the root of a clone:

git clone https://github.com/rsansores/magic_sync
cd magic_sync

1. A database for the head office

With Docker:

docker compose -f examples/shop/docker-compose.yml up -d

That starts PostgreSQL on port 5433 with the user, password and database shop, which is what the head office connects to unless you set DATABASE_URL. Without Docker Compose, the same with plain Docker:

docker run -d --name shop-postgres -p 5433:5432 \
  -e POSTGRES_USER=shop -e POSTGRES_PASSWORD=shop -e POSTGRES_DB=shop postgres:17

2. The head office

Its page is a Vue app (examples/shop/web) over the engine’s dashboard, which is built from this repository first:

pnpm --dir dashboard install
pnpm --dir dashboard build
pnpm --dir examples/shop/web install
pnpm --dir examples/shop/web build

Then the head office itself:

cargo run --manifest-path examples/shop/server/Cargo.toml

Open http://127.0.0.1:8080. The first time, it installs everything it needs in the database: its own tables, and the engine’s.

3. Two registers

Each in a terminal of its own, from the root of the clone:

cargo run --manifest-path examples/shop/register/Cargo.toml -- --name "Register 1" --port 8081

cargo run --manifest-path examples/shop/register/Cargo.toml -- --name "Register 2" --port 8082

Each introduces itself to the head office, which provisions it, and appears on the head office’s page with its own page inside. A register keeps its database in the directory it was started from (register-8081.db): started there again, it is the same register.

Both programs listen on 127.0.0.1 only: the example has no login. To open it from another machine (over a VPN, say), bind them to every interface:

cargo run --manifest-path examples/shop/server/Cargo.toml -- --bind 0.0.0.0:8080
cargo run --manifest-path examples/shop/register/Cargo.toml -- --name "Register 1" --port 8081 --bind 0.0.0.0

Ctrl-C stops either one, and loses nothing it was doing.

4. Things to try

• Add a product at the head office. Both registers have it within a second, and the dashboard stays green: everything is in sync.

• Take Register 2 offline (its Go offline button). Sell a few things on it: it keeps selling, and says how many receipts are not at the head office yet. Change a price and add a product at the head office: the dashboard turns amber, catching up, and after half a minute says Register 2 is not answering. Click it to see how far along it is, table by table.

• Bring it back online. Its receipts reach the head office, it catches up with the catalog, in order, and the dashboard turns green again.

• Restart the head office while a register is offline and a product has changed. What the head office queued for the register is in its database: back online, the register gets it.

• Ask where a product is. who holds it? next to a product lists each register and where that product stands on it: holds it, on its way, set aside. That is the question to answer next to a row, in your own listings.

• Break something. Start a register with a bug that refuses one product:

  cargo run --manifest-path examples/shop/register/Cargo.toml -- --name "Register 3" --port 8083 --refuse "Tea"
  

  Add Tea at the head office: Register 3 turns red, needs a fix, the Problems tab says why, and every other product still reaches it. Stop it and start it again without --refuse — the fix deployed — and within seconds the head office’s retry gets Tea through and the page is green.

• Lose a register. break the network under a register makes every send to it fail, as a network would: it turns unreachable, and the head office tries it again, waiting longer each time. restore the network, and it catches up.

• Delete a product a register has sold. That register keeps the product for its receipts — it retires it: out of the list, still there for what points at it — and the other deletes it.

• Wish a register another footer. Each register prints a footer of its own under its receipts, and Remote update of footers through wishes at the head office shows each one as the register reports it, green: in sync. change one and wish another: the row turns amber, the register’s footer struck through beside the wished one, on its way — and within a second the register prints it, and the row is green again, showing what the register now reports. Turn the register’s printer off first and the wish waits — the printer is off — until it is back on. Wish one longer than the printer’s 40 characters and the register refuses it: the row turns red and says why, and the register’s footer stands; dismiss lets go of the wish. Change the footer on the register itself while the head office’s form is open, then wish: the wish was based on the footer the page showed, so it is not applied — superseded — and the register’s own stands. Take a register offline and wish: it is on its way until the register is back. Every time, both pages end showing the same footer (Settings and wishes).

The rest of the book builds this from scratch.

Set up a project

A system kept in sync by magic_sync is at least two programs: the server, on PostgreSQL, and a replica, on SQLite, that runs on every device. They share one thing — the tables that travel, as Rust structs — and almost nothing else. The shop is laid out so:

shop/
├── rows/       the tables, as both sides hold them: a library
├── server/     the head office: a program, and a workspace of its own
└── register/   a register: a program, and a workspace of its own

A build of magic_sync is one side: the postgres feature builds the server’s, the sqlite feature a replica’s. A build with both is refused — the crate does not compile — and so is a build with neither. Everything in this chapter follows from that.

The rows: one crate, no side of its own

You write a library for the rows and their codecs, which depends on magic_sync without picking a side, and lets each program pick one through a feature of its own:

[package]
name = "shop_rows"
version = "0.0.0"
edition = "2021"
publish = false

[dependencies]
# No side of its own: each program that uses the rows picks one, below.
# Outside this repository, leave `path` out.
magic_sync = { version = "0.1.1", path = "../../../magic_sync" }
chrono = { version = "0.4", features = ["serde"] }
serde = { version = "1", features = ["derive"] }
sqlx = { version = "0.8", features = ["runtime-tokio", "uuid", "chrono"] }
uuid = { version = "1", features = ["serde", "v7"] }

[features]
# The server builds the rows with `server`, a register with `register`: what
# only one side has (where the registers are) is behind its feature.
server = ["magic_sync/postgres"]
register = ["magic_sync/sqlite"]

What only one side has is behind its feature: where the server finds the replicas (magic_sync::plan::Replicas) exists only on the server, so the shop’s REGISTERS is #[cfg(feature = "server")]. The rows themselves and the codec list compile on either side (The rows, codecs).

The server

You write a program with the postgres side, and the rows built for it:

[package]
name = "shop_server"
version = "0.0.0"
edition = "2021"
publish = false

# A workspace of its own: a build is the server's side or a replica's, never
# both, and Cargo builds a workspace's members together.
[workspace]

[dependencies]
# The server's side. Outside this repository, leave `path` out.
magic_sync = { version = "0.1.1", path = "../../../magic_sync", features = ["postgres"] }
# The rows both sides share, built for the server.
shop_rows = { path = "../rows", features = ["server"] }
axum = "0.8"
serde = { version = "1", features = ["derive"] }
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres", "uuid", "chrono"] }
tokio = { version = "1", features = ["full"] }
tower-http = { version = "0.6", features = ["fs"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
uuid = { version = "1", features = ["serde", "v7"] }

sqlx is the engine’s database layer: the pool you hand it is sqlx’s, and so are the transactions and connections it takes. Enable its postgres driver and a runtime; the engine runs on tokio.

A replica

You write a program with the sqlite side:

[package]
name = "shop_register"
version = "0.0.0"
edition = "2021"
publish = false

# A workspace of its own: a build is the server's side or a replica's, never
# both, and Cargo builds a workspace's members together.
[workspace]

[dependencies]
# A replica's side. Outside this repository, leave `path` out.
magic_sync = { version = "0.1.1", path = "../../../magic_sync", features = ["sqlite"] }
# The rows both sides share, built for a replica.
shop_rows = { path = "../rows", features = ["register"] }
axum = "0.8"
chrono = "0.4"
reqwest = { version = "0.12", default-features = false, features = ["json"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite", "uuid", "chrono"] }
tokio = { version = "1", features = ["full"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
uuid = { version = "1", features = ["serde", "v7"] }

One workspace per side

Cargo builds the members of a workspace together and turns on, for a crate they share, every feature any of them asks for. Put the server and a replica in one workspace and the first cargo build, cargo test or editor check that builds both builds magic_sync with both sides, and the crate refuses to compile, saying so.

So each program is a workspace of its own: an empty [workspace] table in its manifest, as above, or one workspace for everything on the server and another for everything on the replica. The rows are a path dependency of both, and belong to neither. Each side’s tests run in its own workspace, and a test never needs the other side’s build: the server’s tests drive the server, a replica’s tests drive the replica.

A replica that is not written in Rust needs none of this: it follows the replica protocol, and only the server uses the crate.

Outside this repository

The shop reaches the engine by path, because it lives next to it; a project of yours depends on the published crates, and leaves path out:

magic_sync = { version = "0.1.0", features = ["postgres"] }

The rest of this part writes what goes in these crates: the tables, the rows, the codecs, then each side.

The tables

The shop has three tables on the head office’s PostgreSQL:

• registers — the replicas. The engine reads which registers exist, and which of them should receive anything, from here.
• products — written at the head office, held by every register.
• receipts — written by a register, filed at the head office.

And the same tables, minus registers, on every register’s SQLite.

On the server

You write your tables, as you always do (examples/shop/server/schema.sql):

-- The registers: the replicas, and which of them receive anything.
CREATE TABLE IF NOT EXISTS shop.registers (
    id     UUID PRIMARY KEY,
    name   TEXT NOT NULL,
    url    TEXT NOT NULL,
    active BOOLEAN NOT NULL DEFAULT true
);

-- Written here, held by every register.
CREATE TABLE IF NOT EXISTS shop.products (
    id          UUID PRIMARY KEY,
    name        TEXT NOT NULL,
    price_cents BIGINT NOT NULL,
    updated_at  TIMESTAMPTZ NOT NULL
);

-- Written by a register, filed here.
CREATE TABLE IF NOT EXISTS shop.receipts (
    id          UUID PRIMARY KEY,
    register_id UUID NOT NULL,
    product_id  UUID NOT NULL,
    quantity    INTEGER NOT NULL,
    total_cents BIGINT NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL
);

-- Each register's footer, as it last said: a setting the head office may
-- wish another.
CREATE TABLE IF NOT EXISTS shop.footers (
    register_id UUID PRIMARY KEY,
    text        TEXT NOT NULL,
    changed_at  TIMESTAMPTZ NOT NULL
);

products.updated_at is a column like any other: a register takes what the head office sends, whatever its version, because the order of the messages — not a column — says which is the latest. A column that orders versions (#[row(version)]) belongs to a table a register edits and the head office files, such as a record a register corrects, or a setting of its own — the footer each register prints, which the head office may wish another (Settings and wishes).

The engine’s part: installed for you, or in your migrations

The engine keeps its own tables in a sync schema, and puts a trigger on every table that travels down, noting each change in the writing transaction — so a schema of yours by that name needs another: the server refuses to install into it, or to read it as its own. There are two ways to get them there.

You write, if you have migrations — the way a system in production should — the engine’s schema in one of them: its steps, magic_sync::plan::migration(1) up to migration(SERVER_SCHEMA_VERSION), one SQL script each, in order. Then the triggers, in the migration that creates the tables, as you would their indexes. You do not write them by hand: Engine::triggers prints them, from the same declarations the server starts from, one CREATE TRIGGER per line:

pub async fn print_triggers(pool: &PgPool) -> Result<(), Error> {
    let mut conn = pool.acquire().await?;
    println!(
        "{}",
        Engine::triggers(&mut conn, &codecs(), REGISTERS).await?
    );
    println!(
        "{}",
        Engine::indexes(&mut conn, &codecs(), REGISTERS).await?
    );
    Ok(())
}

For the shop, it prints:

CREATE TRIGGER sync_note_row AFTER INSERT OR UPDATE OR DELETE ON "shop"."products" FOR EACH ROW EXECUTE FUNCTION sync.note_row('products', '', '1', 'id', '0');
CREATE TRIGGER sync_refuse_truncate BEFORE TRUNCATE ON "shop"."products" FOR EACH STATEMENT EXECUTE FUNCTION sync.refuse_truncate();
CREATE TRIGGER sync_note_row AFTER INSERT OR UPDATE OR DELETE ON "sync"."sync_wishes" FOR EACH ROW EXECUTE FUNCTION sync.note_row('sync_wishes', '', '1', 'id', '0', 'answered', 'outcome', 'reason', 'answered_at');
CREATE TRIGGER sync_refuse_truncate BEFORE TRUNCATE ON "sync"."sync_wishes" FOR EACH STATEMENT EXECUTE FUNCTION sync.refuse_truncate();

The arguments say how the table is noted: its name, its partition column (see The server; '' for none), its key’s columns in the order the row declares them, the columns that point at other mirrored tables (kept of the row as it was, so a parent it no longer points at can be let go), and the columns replicas own (in a table written on both sides; see Recipes) — each list after its length. Print them again when a declaration changes.

What happens: forget one, write one with other arguments, or disable it, and the server refuses to start, naming the table and printing the trigger it needs. Leave a trigger on a table no row declares, and it refuses too: its changes would be noted and never sent.

The planner looks rows up by more than their keys: by the columns that point at another mirrored table, by a row’s replica or partition column, by the column a table follows, and a group’s members by their row and their group, its assignments by their group and their replica. On a large table, a lookup no index serves reads the whole table on every pass, so the server needs an index for each: one of yours serves when it leads with those columns, in any order. Engine::indexes prints, one CREATE INDEX per line, those your database does not have — write them in the migration beside the triggers.

What happens: forget one, and the server refuses to start, printing the indexes it needs.

Mind: a trigger as printed does not fire in a session whose session_replication_role is replica — a logical replication subscription applying changes, a restore told to disable triggers — so what such a session writes reaches no replica. If one writes your mirrored tables, enable their triggers always (ALTER TABLE … ENABLE ALWAYS TRIGGER sync_note_row, and the same for sync_refuse_truncate and sync_note_group): the server takes a trigger enabled always, and refuses one enabled for replication sessions only.

You write, if you have no migrations — the shop’s way — nothing: build the server with Server::install instead of Server::new (The server). Every time it starts, in one transaction and one process at a time, it brings the sync schema to the version this build needs, writes every trigger that is missing or not as the engine needs it, and creates every index it needs that no index serves. Run it before anything writes a mirrored table: a write before its trigger exists is never noted.

Mind: Server::install creates an index with a plain CREATE INDEX, which holds writes to the table until it is built — on a large table, a long time. Print them with Engine::indexes and create them yourself first, CREATE INDEX CONCURRENTLY (outside a transaction): an index that serves is left as it is.

On a register

You write the same tables, with one more column on each table that comes down and that a register’s own rows point at (examples/shop/register/schema.sql):

CREATE TABLE IF NOT EXISTS products (
    id          BLOB PRIMARY KEY NOT NULL,
    name        TEXT NOT NULL,
    price_cents INTEGER NOT NULL,
    updated_at  TEXT NOT NULL,
    retired_at  TEXT
);

CREATE TABLE IF NOT EXISTS receipts (
    id          BLOB PRIMARY KEY NOT NULL,
    register_id BLOB NOT NULL,
    product_id  BLOB NOT NULL REFERENCES products (id),
    quantity    INTEGER NOT NULL,
    total_cents INTEGER NOT NULL,
    created_at  TEXT NOT NULL
);

-- What it prints under every receipt: its own, which the head office may
-- wish another.
CREATE TABLE IF NOT EXISTS footers (
    register_id BLOB PRIMARY KEY NOT NULL,
    text        TEXT NOT NULL,
    changed_at  TEXT NOT NULL
);

What happens: when the head office deletes a product a receipt here points at, the register cannot delete it without breaking the receipt. It sets retired_at instead: the product stays for the receipt, and your screens leave retired rows out. Offered again, it is live again. Once nothing points at it any more, it goes, with the next message the register applies.

A retired row keeps its values only for what points at it: a unique index on such a table leaves the retired rows out, or a product deleted and made again with the same code would be refused here while the old one stays. check refuses one that does not:

CREATE UNIQUE INDEX products_code ON products (code) WHERE retired_at IS NULL;

After every message it applies, in a transaction of its own, the replica looks for the retired rows that nothing keeps any more — no row of any table, live or retired, points at them — and lets them go, children first: on a table of many rows, an index on its retired column keeps that cheap.

Same two ways for the engine’s part: your migrations run the engine’s steps, magic_sync::replica::migration(1) up to migration(REPLICA_SCHEMA_VERSION) in order, and the script magic_sync::replica::triggers writes for your codecs — print them once and paste them into the migration that creates the tables, and the triggers again when a table is added or changes how it travels:

pub fn print_triggers() -> Result<(), magic_sync::Error> {
    println!("{}", magic_sync::replica::triggers(&codecs())?);
    Ok(())
}

or, without migrations, magic_sync::replica::install applies the engine’s tables and its triggers in one transaction, every time the replica starts.

The triggers note every insert, update and delete of a table that goes up (an update that changes the key notes the old key too), refuse a write the engine did not make to a table the server sends, let a row of a two-way table begin only on the server, and refuse an update of history (a record without a version). The refusals wait until the replica knows which it is (A replica). Replica::new runs magic_sync::replica::check, which refuses a database whose triggers are not exactly these; a test of yours runs it too, over the database your migrations build.

The rows

Each table that travels is a Rust struct whose fields are exactly its columns, and which says how it travels. The same struct serves both sides.

You write (examples/shop/rows/src/lib.rs):

/// A product: written at the head office, held by every register.
#[derive(sqlx::FromRow, Clone, Debug, Serialize, Deserialize, magic_sync::Row)]
#[row(
    table = "products",
    schema = "shop",
    mirror = "catalog",
    sync = "broadcast",
    retired = "retired_at"
)]
pub struct Product {
    pub id: Uuid,
    pub name: String,
    pub price_cents: i64,
    pub updated_at: DateTime<Utc>,
}

/// A receipt: written by a register, filed at the head office.
#[derive(sqlx::FromRow, Clone, Debug, Serialize, Deserialize, magic_sync::Row)]
#[row(table = "receipts", schema = "shop", mirror = "record")]
pub struct Receipt {
    pub id: Uuid,
    pub register_id: Uuid,
    pub product_id: Uuid,
    pub quantity: i32,
    pub total_cents: i64,
    #[row(created)]
    pub created_at: DateTime<Utc>,
}

Which way it travels: mirror

Which replicas hold it: sync

For a table that goes down:

What happens: you never declared that a receipt needs its product. The engine reads the foreign keys from the database: a replica sent a row is sent every row it points at, and keeps those while anything of its own points at them. A table with no sync rule still reaches every replica that needs it.

The rest

The shop’s rows use a few more: schema, the table’s PostgreSQL schema; retired, the column a register sets on a product it lets go of but must keep; #[row(version)], the column of a record a register edits whose value only moves forward; and #[row(created)], a column written once. Every key and flag the derive takes — a key of several columns, a view rows are read through, the columns one side leaves out, the columns replicas own — is in the derive’s reference, and Recipes puts most of them to work.

Fields follow sqlx::FromRow: #[sqlx(rename = "…")] names the column, #[sqlx(skip)] is no column, #[sqlx(flatten)] brings in a part’s columns (the part derives magic_sync::Columns).

Keeping the struct and the table together

What happens: every time the server or a replica starts, the engine holds each table against its row on that database — column for column, type for type, nullability for nullability, and the primary key — and refuses to start on one that differs, naming every difference. A column added on one side only, or to the struct only, never runs. The columns a row leaves out on purpose (retired, unmirrored) are not compared.

A test of yours that builds the server, or a replica, over the database your migrations build is all it takes to see it before a deploy.

The derive checks the rest where you write it: a key that is not a field, a column named twice, an attribute a table of its kind cannot have (a sync rule on a record, a version on a table the server sends down), a key or version of a type the engine cannot read — each is a compile error pointing at what says it.

How a row travels: codecs

A codec is a table’s row type, and how its rows are turned into bytes and back. The engine holds every table’s codec in one list, and never names a table itself.

You write the list, parents before children (examples/shop/rows/src/lib.rs):

/// A receipt is filed under the register that sent it, whatever it says.
fn place_receipt(receipt: &mut Receipt, register: Uuid, _: &()) {
    receipt.register_id = register;
}

/// Every table that travels, parents first.
pub fn codecs() -> Vec<Codec<()>> {
    vec![
        Codec::json::<Product>(None),
        Codec::json::<Receipt>(Some(place_receipt)),
        Codec::json::<Footer>(None),
    ]
}

JSON or protobuf

Codec::json sends a row as a JSON object of its columns, each value as its field’s serde writes it — a 64-bit integer as its decimal text, which a reader holding numbers as doubles would otherwise round: nothing else to declare, as long as serde writes each field under its own name. A field with #[serde(rename = "…")], a rename_all on the row, or a part brought in with #[sqlx(flatten)] that serde nests instead of flattening (give the field #[serde(flatten)] too) makes every row of the table fail to encode, the first time one travels. That is the simple way, and the one to start with.

Codec::new sends it as a protobuf message of your own, with a mapping each way — smaller on the wire, and a schema a replica in another language can generate code from:

pub fn product() -> Codec<()> {
    Codec::new::<Product, ProductRow>(product_to_wire, product_from_wire, None)
}

where ProductRow is a prost message:

/// A product on the wire.
#[derive(Clone, PartialEq, prost::Message)]
pub struct ProductRow {
    #[prost(string, tag = "1")]
    pub id: String,
    #[prost(string, tag = "2")]
    pub name: String,
    #[prost(int64, tag = "3")]
    pub price_cents: i64,
    #[prost(string, tag = "4")]
    pub updated_at: String,
}

and the two functions map the row to it and back, failing with a magic_sync::wire::MapError, which any error converts into with ?:

fn product_to_wire(p: Product) -> Result<ProductRow, MapError> {
    Ok(ProductRow {
        id: p.id.to_string(),
        name: p.name,
        price_cents: p.price_cents,
        updated_at: p.updated_at.to_rfc3339(),
    })
}

fn product_from_wire(m: ProductRow) -> Result<Product, MapError> {
    Ok(Product {
        id: Uuid::parse_str(&m.id)?,
        name: m.name,
        price_cents: m.price_cents,
        updated_at: DateTime::parse_from_rfc3339(&m.updated_at)?.with_timezone(&Utc),
    })
}

A row that does not encode is set aside on the server, for every replica that was to get it; one that does not decode on a replica is refused there, and reported. Either comes back once a fix lets it through.

What the server knows about a sender: P and place

Codec<P> carries a type P of yours: what the server knows about a replica that sends something up (here (): nothing). Every row a replica uploads goes through its codec’s place function with the replica’s id and that knowledge, so a receipt is filed under the register that sent it — the transport says which — whatever the row itself claims.

A multi-company shop would make P the register’s company, read when the upload arrives (Senders), and stamp it on every receipt.

The server

The server’s side is one value, a Server, that you build once and run in a task of its own. It needs four things from you: the codecs, where the replicas are, how bytes reach a replica, and what you know about a replica that sends something up.

Where the replicas are

You write (examples/shop/rows/src/lib.rs):

/// The registers, as the engine finds them on the server.
#[cfg(feature = "server")]
pub const REGISTERS: magic_sync::plan::Replicas =
    magic_sync::plan::Replicas::new("shop.registers", "registers").live("active");

shop.registers is a table or a view with an id column; registers is its name as foreign keys name it; active is the condition a register that should receive anything meets. A register that becomes active is provisioned (Provisioning); one that stops being active is forgotten.

Replicas::partitioned_by("company_id") names a column that splits rows and replicas into groups no row crosses — one per customer of a multi-customer system, say: a broadcast row reaches every live replica of its partition (Recipes). The shop has one company, so none.

How bytes reach a replica: the transport

The engine hands you each message as bytes, for one replica, and you deliver it. How is yours — a broker, HTTP, a queue of your own — but a transport must provide what the replica protocol requires of it: nothing it took is lost, a replica’s messages arrive in order and one at a time, a message comes again until it is acknowledged, and an acknowledgement names its message. The engine’s guarantees rest on it.

You write one method: send these bytes to that replica. The shop’s transport is a queue per register in the head office’s own database, which a register reads over HTTP (examples/shop/server/src/main.rs):

/// The head office's way to its registers: a queue per register, in the
/// database (`shop.deliveries`). The registers whose line is cut are a demo
/// of a network failing: sending to one fails.
#[derive(Clone)]
struct Deliveries {
    pool: PgPool,
    /// Wakes the registers waiting for their next message.
    arrived: Arc<Notify>,
    cut: Arc<Mutex<HashSet<Uuid>>>,
}

impl Transport for Deliveries {
    async fn send(&self, register: Uuid, message: Vec<u8>, id: &str) -> Result<(), Error> {
        if self.is_cut(register) {
            // A retry may get it through: the network's failure, not the
            // message's.
            return Err(Error::transport(true, "the line to it is cut".to_owned()));
        }
        // The same message sent twice is queued once: `id` names it.
        sqlx::query(
            "INSERT INTO shop.deliveries (register_id, id, message) VALUES ($1, $2, $3) \
             ON CONFLICT (register_id, id) DO NOTHING",
        )
        .bind(register)
        .bind(id)
        .bind(message)
        .execute(&self.pool)
        .await?;
        self.arrived.notify_waiters();
        Ok(())
    }
}

id names the message: the shop gives it to the register with the message, and the register acknowledges it by that name, which takes that message — and no other — out of the queue:

/// The register's next message, named in the `message-id` header, and left
/// at the head of its queue until the register acknowledges it. When there is
/// none, the register is answered once one arrives, or after a while that
/// there is none (204).
async fn next(State(app): State<App>, Path(register): Path<Uuid>) -> Result<Response, Failure> {
    if app.deliveries.is_cut(register) {
        return Ok(StatusCode::SERVICE_UNAVAILABLE.into_response());
    }
    let deadline = tokio::time::Instant::now() + WAIT_FOR_NEXT;
    loop {
        // Listening before looking: a message queued in between still wakes
        // this request.
        let arrived = app.deliveries.arrived.notified();
        tokio::pin!(arrived);
        arrived.as_mut().enable();
        let head: Option<(String, Vec<u8>)> = sqlx::query_as(
            "SELECT id, message FROM shop.deliveries WHERE register_id = $1 ORDER BY seq LIMIT 1",
        )
        .bind(register)
        .fetch_optional(&app.pool)
        .await?;
        if let Some((id, message)) = head {
            return Ok(([(MESSAGE_ID, id)], message).into_response());
        }
        tokio::select! {
            () = arrived => {}
            () = tokio::time::sleep_until(deadline) => break,
            () = stopped(app.stopping.clone()) => break,
        }
    }
    Ok(StatusCode::NO_CONTENT.into_response())
}

/// The register applied the message `id` and committed it: it leaves the
/// queue, and the next one is given. An acknowledgement names its message, so
/// one that comes twice, or late, never takes another message out.
async fn ack(
    State(app): State<App>,
    Path((register, id)): Path<(Uuid, String)>,
) -> Result<StatusCode, Failure> {
    sqlx::query("DELETE FROM shop.deliveries WHERE register_id = $1 AND id = $2")
        .bind(register)
        .bind(id)
        .execute(&app.pool)
        .await?;
    Ok(StatusCode::NO_CONTENT)
}

With NATS JetStream, the stream is the queue, and a durable pull consumer per replica with max_ack_pending = 1 hands out one message at a time.

When sending fails, say whether a retry may help:

/// The network: a retry may get the message through.
pub fn unreachable(e: impl Display) -> Error {
    Error::transport(true, e.to_string())
}

/// This message, never: the transport will not take it.
pub fn refused(e: impl Display) -> Error {
    Error::transport(false, e.to_string())
}

What happens: either way, that replica waits out a backoff while every other replica carries on; the message stays first in its outbox, with everything after it, and goes again, in order, when the backoff ends. A refusal shows in its last_error; a message refused for good again and again, or an outbox grown past its bound, is dropped and the replica provisioned again — an alarm. The same message always goes under the same id. A send that does not answer within send_timeout is a transient failure. The engine never hands a transport a message larger than its limit: a row too big for any message is set aside instead.

Building and running it

You write:

    let deliveries = Deliveries {
        pool: pool.clone(),
        arrived: Arc::new(Notify::new()),
        cut: Arc::default(),
    };
    let mut settings = Settings::default();
    settings.idle = Duration::from_millis(200);
    // A fix deployed (a register restarted without its bug) gets through
    // within seconds, not the hour a real server waits.
    settings.retry_every = Duration::from_secs(10);
    // No migrations of its own: the engine sets up what the database lacks.
    let server = Arc::new(
        Server::install(
            pool.clone(),
            codecs(),
            REGISTERS,
            deliveries.clone(),
            Anyone::new(pool.clone()),
            settings,
        )
        .await?,
    );
    let (stop, stopping) = watch::channel(false);
    let relay = tokio::spawn({
        let server = server.clone();
        let stopping = stopping.clone();
        async move { server.run(stopped(stopping)).await }
    });

Server::install sets up what the database lacks (The tables); with migrations of your own, use Server::new with the same arguments, which refuses to start on a database that is not set up. run returns as soon as the future it is given resolves — the shop’s, on Ctrl-C. A pass it cuts short is planned and sent again by the next relay, as after a crash:

    // On Ctrl-C: no new request, the registers waiting for a message are
    // answered, and the relay stops — a pass it cuts short runs again.
    axum::serve(listener, router)
        .with_graceful_shutdown(async move {
            interrupted().await;
            stop.send_replace(true);
        })
        .await?;
    relay.await?;
    pool.close().await;

What happens in run, for as long as your process lives:

• It takes a PostgreSQL advisory lock. Run as many server processes as you like: one of them runs the relay, the others stand by and take over if it goes away. The lock is a session’s, so the pool you hand it reaches PostgreSQL directly or through a pooler in session mode: through one in transaction mode (PgBouncer’s default) a pass would run on a server process that does not hold the lock, and the relay refuses to.
• While it holds the lock, it reads the noted changes, plans what each replica should hold against what it holds, and — in one transaction on the lock’s own connection, so a process that lost the lock cannot record over the one that took it — records what each replica will hold and writes the messages that carry the difference to the replica’s outbox. Then it sends every outbox through your transport.
• When it takes the lock — usually after a deploy — and every hour, it tries again whatever it set aside: notes it could not plan, rows it could not send, uploads it could not file. A deploy that fixed the cause lets them through.
• Every few minutes it logs what needs someone: a backlog that is not draining, rows in quarantine. Each of those means wait (the network) or deploy a fix (a bug), never go fix the data.

Settings holds the knobs: message size limits, notes per pass, how often it retries and prunes, how long a send may take and how long a replica it failed to waits, how long a replica may leave what it was sent unconfirmed before it is provisioned again, and role — a role every transaction of the engine’s own assumes, when your connecting role has less than it needs. Besides the sync schema, the engine needs to create temporary tables (the database’s TEMPORARY privilege, which every role has unless revoked): the relay works out each pass’s plan in them, so a large one is never in its memory whole. Server::new refuses to start without it. Change what you need of the defaults:

pub fn settings() -> Result<Settings, Error> {
    let mut settings = Settings::default();
    settings.limits = Limits::new(200, 256 * 1024)?; // rows and bytes your transport takes
    settings.role = Some("sync_engine".to_owned());
    Ok(settings)
}

What replicas send up

You write a route that hands every message from a replica to the server, saying which replica it came from — from your transport, never from the message itself:

/// What a register sends up, filed under the register the path names, and
/// answered once filed. An error is one worth sending again (503): the
/// engine keeps whatever a resend could not cure, and answers it.
async fn up(
    State(app): State<App>,
    Path(register): Path<Uuid>,
    body: Bytes,
) -> Result<StatusCode, Failure> {
    if app.deliveries.is_cut(register) {
        return Err(Failure(
            StatusCode::SERVICE_UNAVAILABLE,
            "the line is cut".to_owned(),
        ));
    }
    app.server.receive(register, &body).await?;
    Ok(StatusCode::NO_CONTENT)
}

An error means the message must come again (the database is away, or it speaks a later protocol than this build); filing it again is harmless. Any other failure is no reason to send it again: its rows are kept, it is answered, and they are filed again after a deploy. The shop answers an error with the status that says whether to try again, which its registers read the same way:

/// A failure, answered with the status that tells the caller whether asking
/// again may help: 503 when the database (or what the engine waits on) is
/// away, 500 when the same request would fail the same way.
struct Failure(StatusCode, String);

impl From<Error> for Failure {
    fn from(e: Error) -> Self {
        let status = if e.is_transient() {
            StatusCode::SERVICE_UNAVAILABLE
        } else {
            StatusCode::INTERNAL_SERVER_ERROR
        };
        Failure(status, e.to_string())
    }
}

impl From<sqlx::Error> for Failure {
    fn from(e: sqlx::Error) -> Self {
        Error::from(e).into()
    }
}

impl IntoResponse for Failure {
    fn into_response(self) -> Response {
        (self.0, self.1).into_response()
    }
}

What happens: the rows are filed in a transaction, each by its codec’s rules — a record kept from its first delivery, the replica’s columns of a both table taken, a catalog refused — under the replica the message came from. After the commit, the replica is told its upload was filed, through your transport. A row the server cannot file is kept as it came, and filed again after a deploy. The same message also carries what the replica could not take, what it applied, and what it holds.

What you know about a sender: Senders

Anyone files every replica’s uploads as they are, in a plain transaction.

You write, when your database cares who is writing — a role, a tenant setting your row policies read — or you want to stamp uploads with what you know of the sender:

/// What the server knows of a register that sends something up: its company.
pub struct Company(pub Uuid);

pub struct Companies(pub PgPool);

impl Senders for Companies {
    type Known = Company; // the `P` of `Codec<P>`

    async fn known(&self, register: Uuid) -> Result<Option<Company>, Error> {
        let company = sqlx::query_scalar("SELECT company_id FROM shop.registers WHERE id = $1")
            .bind(register)
            .fetch_optional(&self.0)
            .await?;
        Ok(company.map(Company))
    }

    async fn filing(&self, company: &Company) -> Result<Transaction<'static, Postgres>, Error> {
        let mut tx = self.0.begin().await?;
        // What the row policies read: who is writing.
        sqlx::query("SELECT set_config('shop.company', $1, true)")
            .bind(company.0.to_string())
            .execute(&mut *tx)
            .await?;
        Ok(tx)
    }
}

A register for which known answers None is one the server does not know: its messages are dropped, with a log.

A replica

A replica’s side is one value too, a Replica: you hand it every message that arrives, and run its upward loop in a task of its own.

Building it

You write (examples/shop/register/src/main.rs):

    // No migrations of its own: the engine sets up what the database lacks,
    // including the trigger that notes every receipt written.
    replica::install(&mut *pool.acquire().await?, &codecs()).await?;

    let (online, plugged) = watch::channel(true);
    let client = reqwest::Client::new();
    let printer = Printer {
        on: Arc::new(AtomicBool::new(true)),
    };
    let mut settings = Settings::default();
    settings.idle = Duration::from_millis(300);
    // What carries out the head office's wishes of its footer.
    settings.wishes = Wishes::new().on::<Footer>(printer.clone());
    let replica = Arc::new(
        Replica::new(
            pool.clone(),
            codecs(),
            me,
            Http {
                client: client.clone(),
                server: server.clone(),
                online: plugged.clone(),
            },
            settings,
        )
        .await?,
    );
    let (stop, stopping) = watch::channel(false);
    let up = tokio::spawn({
        let replica = replica.clone();
        let stopping = stopping.clone();
        async move { replica.run(stopped(stopping)).await }
    });
    let down = tokio::spawn(receive(
        client.clone(),
        server.clone(),
        me,
        replica.clone(),
        plugged,
        stopping,
    ));

me says which replica this is: a Uuid the server knows (a row of shop.registers). When a replica does not know its id until it is set up, implement Identity instead, answering None until it does: nothing goes up until then.

What happens: until the replica first runs knowing its id — built with it, or sending or applying a message once Identity answers — its database refuses none of your writes: you are setting it up. A placeholder of its own row, say, renamed once it learns its id, with the rows of its own that point at it. What you wrote to the server’s tables is not noted, and the provisioning sets it straight: the rows the server sends overwrite the ones of the same key, and the ones it does not send go. What you wrote to the replica’s own tables is noted by the key it has by then, and goes up once the replica knows which it is. From then on, the server’s tables are the server’s.

run sends up until the future it is given resolves, finishing what it is sending.

Replica::new refuses a connection the engine cannot keep its word on, and so does every message it applies, on whichever connection of the pool: commits that are not durable (synchronous below FULL, which sqlx’s pools keep by default), since a message the replica acknowledged must still be applied after a power cut; a journal that cannot roll back (journal_mode OFF or MEMORY, of a database with a file; one in memory has no crash to survive), since a message is applied whole or not at all; and triggers that miss the rows a conflict replaces (recursive_triggers, off by default), since a row an INSERT OR REPLACE deletes must go up as gone. Set them in the pool’s options, so every connection has them:

    let pool = SqlitePool::connect_with(
        SqliteConnectOptions::new()
            .filename(&file)
            .create_if_missing(true)
            .foreign_keys(true)
            .pragma("recursive_triggers", "ON"),
    )
    .await?;

Taking what the server sends

You write a loop over your transport: hand each message to receive, and acknowledge it to the transport only after receive returned Ok. The shop’s asks the head office for the next message, which answers as soon as there is one, and acknowledges it by its id:

/// Takes the head office's messages one at a time: each applied and
/// committed, then acknowledged by its id — never before.
async fn receive(
    client: reqwest::Client,
    server: String,
    me: Uuid,
    replica: Arc<ShopReplica>,
    plugged: watch::Receiver<bool>,
    stopping: watch::Receiver<bool>,
) {
    loop {
        let next = tokio::select! {
            next = next_while_plugged(&client, &server, me, plugged.clone()) => next,
            () = stopped(stopping.clone()) => return,
        };
        let (id, message) = match next {
            Ok(Some(next)) => next,
            Ok(None) => continue,
            Err(e) => {
                tracing::warn!("no message from the head office: {e}");
                tokio::time::sleep(AFTER_A_FAILURE).await;
                continue;
            }
        };
        match replica.receive(&message).await {
            Ok(received) => {
                if let Received::Applied(tables) = &received {
                    tracing::info!(?tables, "applied");
                }
                // Lost, the acknowledgement is harmless: the message comes
                // again, and changes nothing the second time.
                let acked = client
                    .post(format!("{server}/sync/{me}/ack/{id}"))
                    .send()
                    .await
                    .and_then(reqwest::Response::error_for_status);
                if let Err(e) = acked {
                    tracing::warn!("not acknowledged, and comes again: {e}");
                }
            }
            Err(e) => {
                // Not acknowledged: it comes again.
                tracing::warn!("not applied, and comes again: {e}");
                tokio::time::sleep(AFTER_A_FAILURE).await;
            }
        }
    }
}

/// The next message and its id, once the register is plugged in; `None` when
/// the head office had none for a while, or the register was unplugged
/// while it waited.
async fn next_while_plugged(
    client: &reqwest::Client,
    server: &str,
    me: Uuid,
    mut plugged: watch::Receiver<bool>,
) -> Result<Option<(String, Vec<u8>)>, String> {
    // An error is the sender gone: the register is stopping.
    let _ = plugged.wait_for(|plugged| *plugged).await;
    tokio::select! {
        next = take(client, server, me) => next,
        _ = plugged.wait_for(|plugged| !*plugged) => Ok(None),
    }
}

/// Asks the head office for the next message, which it answers as soon as
/// there is one.
async fn take(
    client: &reqwest::Client,
    server: &str,
    me: Uuid,
) -> Result<Option<(String, Vec<u8>)>, String> {
    let response = client
        .get(format!("{server}/sync/{me}/next"))
        .send()
        .await
        .and_then(reqwest::Response::error_for_status)
        .map_err(|e| e.to_string())?;
    if response.status() == reqwest::StatusCode::NO_CONTENT {
        return Ok(None);
    }
    let id = response
        .headers()
        .get("message-id")
        .and_then(|id| id.to_str().ok())
        .ok_or("a message without an id")?
        .to_owned();
    let message = response.bytes().await.map_err(|e| e.to_string())?;
    Ok(Some((id, message.to_vec())))
}

What happens: the message is applied in one transaction, and committed before receive returns. A message at or below the last one applied is a redelivery and changes nothing. A row this replica cannot take — it does not decode, a constraint refuses it — is left out and reported, and the rest applies: one bad row never blocks the queue. An error means the replica’s own storage failed, or the message speaks a later protocol than this build: don’t acknowledge, and it comes again.

receive answers what it did — Received::Applied with the keys of every row it applied, per table — so you can react: the shop’s loop logs it; a screen would refresh when products arrived.

Sending things up: the uplink

You write one method: send these bytes to the server. It is a transport too, and provides the same; the shop’s is a POST to the head office, which answers once it filed what came:

/// The way up: a POST to the head office, unless the register is unplugged.
struct Http {
    client: reqwest::Client,
    server: String,
    online: watch::Receiver<bool>,
}

impl Uplink for Http {
    async fn send(
        &self,
        register: Uuid,
        _lane: &'static str,
        message: Vec<u8>,
    ) -> Result<(), Error> {
        if !*self.online.borrow() {
            return Err(unreachable("unplugged"));
        }
        let response = self
            .client
            .post(format!("{}/sync/{register}/up", self.server))
            .body(message)
            .send()
            .await
            .map_err(unreachable)?;
        let status = response.status();
        if status.is_success() {
            return Ok(());
        }
        // A server error, or one that asks to come back later, may go
        // through next time; any other refusal is of this message, for good.
        Err(Error::transport(
            status.is_server_error()
                || status == reqwest::StatusCode::TOO_MANY_REQUESTS
                || status == reqwest::StatusCode::REQUEST_TIMEOUT,
            format!("the head office answered {status}"),
        ))
    }
}

/// The head office out of reach: a retry may get through.
fn unreachable(e: impl std::fmt::Display) -> Error {
    Error::transport(true, e.to_string())
}

What happens in run: whenever there is something to say — a receipt written, a message applied, an hourly fingerprint due — it gathers it into one message and sends it through your uplink: the rows written since, as they are now (a receipt edited twice goes up once), the rows it could not take, the last message it applied, and its fingerprints. Once the uplink took it, that is recorded; a receipt is forgotten only when the server says it filed it, and sent again if the server never answers. When the uplink fails for a while, it backs off and tries again; nothing is lost. When it refuses a message for good, the rows it carried are set aside, counted in the backlog, until the replica runs again with a build that may send them.

After writing something you want to see go up at once, call replica.nudge(); otherwise it goes on the loop’s next turn.

Lanes

A replica that writes a table often — positions, readings — can give it a lane of its own, so it never waits behind the rest, nor the rest behind it:

pub fn settings() -> Settings {
    let mut settings = Settings::default();
    settings.lanes = vec![
        Lane::new("main", &[]), // everything else, and what the engine says
        Lane::new("telemetry", &["positions"]),
    ];
    settings
}

Your uplink gets the lane’s name and sends each on its own stream, each kept in order. Each lane runs on its own: one whose uplink fails waits out its own backoff, and one whose send hangs is given up on after send_timeout (30 seconds unless you set it), while the others go on.

Settings and wishes

A setting is a value a replica holds and acts on: a module’s configuration, a screen’s brightness, the footer a register prints. Two parties want to write it. The replica’s user, or its hardware, says what the setting is; the head office says what it wants it to be. Neither is a merge of the other, and the engine never merges: the replica owns the row, and the server wishes its desired columns other values, which the replica carries out — or refuses, or finds it changed first.

Once the system settles — every message applied, every write filed — the server and the replica show the same value of every setting, and the same answer to every wish, whichever side’s write ruled.

A digital twin

So the server holds two halves of every setting: what the replica reports it is — the row, as the replica last wrote it — and what the server desires it to be — the wish, with where it stands. That pair is what IoT platforms call a device twin or device shadow: desired state on one side, reported state on the other, and the device reconciling the two. Here the twin is rows in your own PostgreSQL, travelling the way every other row does: offline-first, in order, confirmed, and set straight after a restore of either side. There is no other service to run.

What it unlocks:

• Remote configuration. A screen’s brightness, a printer’s footer, a scale’s tare, a module’s parameters — set at the head office, carried out on the replica when it can be: now, after it comes back online, or once its hardware answers.
• A configuration page that tells the truth. Beside each value, what the replica reports and what was wished, with where the wish stands: on its way, waiting and why (the module is off), applied, refused and why (out of range), or superseded (changed on the replica first). The shop’s head office shows its registers’ footers this way.
• Fleet-wide changes you can watch. Wish the same value on many replicas, and Engine::wishes and the dashboard count them where they stand: what is done, what waits, what was refused.
• Hardware in the loop. The replica carries a wish out through your handler, which talks to the device. It answers what the device answered, and a device that is away has the wish wait and be offered again.
• Local changes that are not lost. With if_unchanged, a technician’s change on site wins over a head office edit based on what was there before.
• Who asked. A wish carries a note of yours (who, or why), and its answer carries when it was given.

The row

You write, the row the register owns:

/// One of a register's settings, as the register holds it: its value is
/// the register's, and the head office may wish it another.
#[derive(sqlx::FromRow, Clone, magic_sync::Row)]
#[row(
    table = "settings",
    schema = "shop",
    mirror = "setting",
    key = "register_id, id",
    origin = "register_id",
    apply = "if_unchanged"
)]
pub struct RegisterSetting {
    pub register_id: Uuid,
    pub id: Uuid,
    pub name: String,
    #[row(desired)]
    pub value: String,
    #[row(version)]
    pub changed_at: DateTime<Utc>,
}

A setting goes up as a record does, with three things more:

The tables are yours, as for any row; the wishes live in a table of the engine’s, sync_wishes, which its schema’s second step creates on each side and its triggers cover — print them again (Engine::triggers, replica::triggers) once you add a setting. A name beginning sync_ is the engine’s on either side: no table of yours may take one.

Wishing, on the server

You write, where an operator sets a value:

/// An operator sets a register's setting from the page that showed it: the
/// row as the page read it, with the value they typed. The register carries
/// the wish out only while the setting is still what the operator saw.
pub async fn set_from_the_page<S: Senders, T: Transport>(
    server: &Server<S, T>,
    seen: RegisterSetting,
    value: String,
    operator: &str,
) -> Result<Wished, Error> {
    let mut tx = server.begin().await?;
    let wished = server
        .desire(&mut tx, RegisterSetting { value, ..seen }, Some(operator))
        .await?;
    // … the host's own audit row, in the same transaction …
    tx.commit().await?;
    Ok(wished)
}

desire takes the row as the operator saw it, with the values they want: a wish is its desired columns, and with if_unchanged its version is the one the wish is based on. It writes the wish in your transaction, beside whatever else you write; the next pass sends it to the register. Wishing the row again replaces its wish, and is carried out again; withdraw takes it back.

What happens: the wish is a row of sync_wishes, written on both sides: the server owns what is wished, the replica its answer. So it travels as every row does — recorded before it is sent, confirmed, provisioned again with the rest — and a register that runs a build without wishes refuses it, which shows on the dashboard as a row set aside for it.

Carrying it out, on a replica

You write, one handler per setting table, in the replica’s settings:

/// The screen's brightness is a value the hardware has to take: a wish of
/// it is carried out there, and the engine records what the setting then is.
pub struct Screen;

impl CarryOut<RegisterSetting> for Screen {
    async fn carry_out(&self, wish: Wish<RegisterSetting>) -> Outcome {
        if wish.wanted.name != "brightness" {
            // Every other setting lives only in the database.
            return Outcome::Applied;
        }
        let Ok(level) = wish.wanted.value.parse::<u8>() else {
            return Outcome::Refused(format!("`{}` is no brightness", wish.wanted.value));
        };
        match set_backlight(level) {
            Ok(()) => Outcome::Applied,
            Err(why) => Outcome::Later(why),
        }
    }
}

pub fn wishes(mut settings: Settings) -> Settings {
    settings.wishes = Wishes::new().on::<RegisterSetting>(Screen);
    settings
}

The engine offers the handler each wish once the message that brought it is committed, with no transaction open, in a task of its own: it may take its time — up to wishes_timeout — and talk to the hardware. It is handed only a wish of a row the replica holds (one of a row it does not hold yet waits for the row), and, under if_unchanged, only while the row is at the version the wish is based on. It answers:

• Applied — the engine writes the wished columns onto the row, its version moved on, and records the answer in the same transaction; both go up;
• Refused(why) — final for this wish; the row stays as it is;
• Later(why) — the wish waits, and is offered again: when you call Replica::offer_wishes (the hardware came back), after the next message that changes a wish, at start, and every wishes_every while one waits.

A handler that panics, or does not answer within wishes_timeout, has the wish wait too; one wish’s trouble never holds back another’s. A write the database refuses (a check of yours on the setting) answers the wish refused, with the database’s reason.

The shop’s registers carry out the head office’s wishes of their receipt footer through their printer (examples/shop/register/src/main.rs):

/// The receipt printer, which takes a new footer — when it is on, and when
/// the footer fits its 40 characters.
#[derive(Clone)]
struct Printer {
    on: Arc<AtomicBool>,
}

impl CarryOut<Footer> for Printer {
    async fn carry_out(&self, wish: Wish<Footer>) -> Outcome {
        let text = &wish.wanted.text;
        if text.chars().count() > 40 {
            return Outcome::Refused(format!(
                "\"{text}\" is longer than the printer's 40 characters"
            ));
        }
        if !self.on.load(Ordering::Relaxed) {
            return Outcome::Later("the printer is off".to_owned());
        }
        // A real printer is sent the footer here. `Applied`: the engine
        // writes it into `footers`, and tells the head office.
        Outcome::Applied
    }
}

A setting that lives only in the database needs no code: Direct answers Applied, and the engine writes it. Replica::new refuses a setting with no handler.

Mind: a replica that stops after the handler ran and before the answer was recorded offers the same wish again; so does one restored from an older copy. Carrying the same wish out twice must leave the setting as once does.

Strategies: who wins

With always, every wish is carried out: the last one wins over a change the register’s user made before it arrived, and a change made after it wins until the next wish. With if_unchanged, a wish is carried out only while the register still holds the version the operator saw; otherwise it is superseded — not carried out, the register’s value stands, and the operator sees why. Either way the replica decides, once, and both sides then show what it decided.

The server files a setting as it does a record, the newer version winning. A register that is provisioned again sends its settings and its answers again, and those the server takes whatever their version — they are what the register holds now. A register’s digests — hourly by default (Settings::digests_every) — fingerprint its settings and its answers too, so one the server finds apart from what it records — the server restored from an older copy, say — is provisioned again. The snapshot names the settings the server records it holding, and the register deletes, as its own write, any it no longer holds: so the server lets go of a setting row only as the register’s delete of it is filed. Either side restored from an older copy, both still end showing the same value, once a digest finds them apart.

Where a wish stands

pub async fn where_the_wishes_stand<S: Senders, T: Transport>(
    server: &Server<S, T>,
    pool: &PgPool,
    settings: &[RegisterSetting],
) -> Result<(), Error> {
    let mut conn = pool.acquire().await?;
    let keys: Vec<_> = settings.iter().map(Row::key).collect();
    for wish in server.wishes::<RegisterSetting>(&mut conn, &keys).await? {
        let state = match wish.state {
            WishState::Pending => "on its way",
            WishState::Waiting => "waiting for the register",
            WishState::Applied => "applied",
            WishState::Refused => "refused",
            WishState::Superseded => "superseded: the register changed it first",
            _ => "in a state of a later version",
        };
        println!(
            "{} → {}: {state} {:?}",
            wish.wanted.name, wish.wanted.value, wish.reason
        );
    }
    Ok(())
}

pending until the register answers this wish; then applied, refused, superseded, or waiting with the handler’s reason — the module that does not answer, say, which the head office could not see otherwise.

The dashboard counts each replica’s wishes by where they stand, and Engine::wishes lists a scope’s, those that need attention first, for WishList to show (The dashboard).

Provisioning a replica

A new replica holds nothing. Provisioning sends it everything it should hold, then tells it to let go of anything else it has.

You write nothing for a new replica: before every pass the relay looks at the replicas, and provisions each one it sees live for the first time — or live again, or in another partition than before. One that stops being live, or whose row is gone, is forgotten and sent nothing until it is live again. The shop’s head office only writes the register’s row when it introduces itself.

To provision one again on demand, in any transaction of yours:

/// An operator's "send it everything again".
pub async fn send_everything_again(pool: &PgPool, register: Uuid) -> Result<(), Error> {
    let mut tx = pool.begin().await?;
    magic_sync::plan::note_resync(&mut tx, register, None).await?;
    tx.commit().await?;
    Ok(())
}

The last argument of note_resync is the replica’s partition, when you use partitions. A “send it everything again” button for an operator is that function.

What happens: the next pass forgets what the server thought the replica held, and sends everything its rules say it should hold, numbered as one snapshot, followed by an end marker. On the end marker, the replica deletes (or retires) every row of those tables the snapshot did not name. A newer snapshot supersedes one still in progress, and what the replica was owed before it is dropped: the snapshot carries it.

You rarely need it to repair anything. The server provisions a replica again by itself when what it holds may be wrong: when it says it missed a message (each names the one before it), or applied one this server never numbered — its database restored from an older copy, which provisions every replica again; when, every hour, the fingerprint it reports of what it holds does not match what the server recorded; and when it leaves the last messages it was sent unconfirmed for long (Settings::unconfirmed_after) while nothing newer goes to it — the transport may have lost them.

Progress and health

Every read here takes a Scope — Scope::Replicas(&[a, b]), a whole Scope::Partition(id), or Scope::All — and counts only live replicas. That is how a host shows each audience its own: a customer of a multi-customer system gets its partition, an operator gets everything, and a user who may see some replicas gets the list of those. The engine knows replicas only by id; put your own names on them.

Each read is one snapshot: every figure in an answer was true at the same moment. Called on a connection outside a transaction, a read opens a REPEATABLE READ one of its own; called inside yours, yours must be REPEATABLE READ (or SERIALIZABLE), and a read refuses a READ COMMITTED one, which would see each statement’s own moment.

Every answer derives serde::Serialize: return it from a handler as it is, and a dashboard reads the JSON. Each carries contract, the version of that JSON (status::CONTRACT), and a dashboard refuses an answer of a contract it does not read.

The overview: is everything where it should be?

You write (server.engine() hands out the engine):

pub async fn overview(engine: &Engine, pool: &PgPool, customer: Uuid) -> Result<(), Error> {
    let mut conn = pool.acquire().await?;
    let overview = engine
        .overview(&mut conn, Scope::Partition(customer), 20)
        .await?;
    // Whether nothing waits for a fix.
    println!("healthy: {}", overview.health.healthy);
    // How many replicas are in each state.
    println!(
        "{} synced, {} catching up",
        overview.states.synced, overview.states.catching_up
    );
    // The 20 that need attention most, worst first.
    for r in &overview.replicas.first {
        let why = match r.state {
            ReplicaState::Synced => "holds what it should; all it wrote is filed".to_owned(),
            ReplicaState::CatchingUp => format!("{} changes on their way", r.pending),
            ReplicaState::Unreachable => format!("{} sends failed: {:?}", r.failures, r.last_error),
            ReplicaState::AwaitingFix => format!("{} rows set aside", r.quarantined),
            _ => "a state of a later version".to_owned(),
        };
        println!("{}: {why}", r.replica);
    }
    // Which tables are not where they should be yet, and what they wait on.
    for t in overview.tables.iter().filter(|t| !t.synced) {
        println!(
            "{}: {} pending, {} in flight, {} set aside, {} not filed",
            t.table, t.pending, t.in_flight, t.quarantined, t.unfiled
        );
    }
    Ok(())
}

What happens: the overview never counts the rows a replica should hold. It reads what the engine records — the changes not yet planned, the messages a replica has not confirmed, what is set aside — so it costs a few milliseconds for 200 replicas holding 3 million rows while they are caught up, and about a fifth of a second at worst, when every row is on its way (the engine’s bench_the_dashboard_over_200_units_holding_3m_rows, on a 32-core workstation). Whether a replica is synced is exact all the same: a change it lacks is either noted and not yet planned, or in a message it has not said it applied, or set aside. A test holds the engine to it under a failing network: a replica called synced holds exactly what it should.

Its size does not grow with the fleet. Every replica is counted in overview.states, and overview.replicas lists the worst of them up to the limit you pass — those waiting for a fix, then those out of reach, then those catching up, the longest silent first — with total uncut. The rest are a page away: engine().listing(&mut conn, scope, Some(ReplicaState::Synced), Page::new(offset, limit)) lists them in the same order, in one state or in all. overview.now is the server’s time when it read: every time in the answer is the server’s, and how long ago one was is measured from it, never from the reader’s clock.

A replica is waiting for a fix (AwaitingFix) only for what is its own: a row set aside for it, an upload of its that waits in the inbox, a change of it alone (its provisioning, say) that the planner set aside. A change the planner set aside that may concern every replica is one problem, listed once in the health; each replica it may concern is catching up on it.

overview.health lists what the engine waits on over the scope, cut at the limit you pass (total is uncut): parked notes, rows in quarantine and why, uploads in the inbox and why, replicas a send keeps failing to. Each means one of two things — wait (the network) or deploy a fix (a bug) — and none of them ever means go fix the data by hand. healthy is whether nothing waits for a fix: a replica out of reach is the network’s to settle, listed in failing but not unhealthy; what waits for it waits in its outbox (owed on its overview), not on the planner, so oldest, the age of the oldest change not yet planned, does not grow while it is away. engine().health(&mut conn, scope, limit) answers it alone; the server’s runner logs it every few minutes.

How far along one replica is

The overview says whether; progress says how far: “holds 480 of 500”.

pub async fn progress(engine: &Engine, pool: &PgPool, register: Uuid) -> Result<(), Error> {
    let mut conn = pool.acquire().await?;
    let progress = engine
        .progress(&mut conn, Scope::Replicas(&[register]))
        .await?;
    for register in progress.replicas {
        for t in register.tables {
            println!(
                "{}: {} of {} ({} on their way, {} set aside)",
                t.table,
                t.holds - t.outdated,
                t.should,
                t.in_flight,
                t.quarantined
            );
        }
    }
    Ok(())
}

What happens: a row counts as held only once the replica confirmed it applied the message that carried it. Until then it is in_flight; a row it was sent and should no longer hold is to_remove, confirmed or not. So the numbers never run ahead of the replica: “holds 480 of 500” means 480 are on it, now. Counting what a replica should hold reads its rules over the tables: ask it of a few replicas at a time, not of a fleet.

Where is one row?

pub async fn who_holds(engine: &Engine, pool: &PgPool, product: &Product) -> Result<(), Error> {
    let mut conn = pool.acquire().await?;
    let row = engine
        .row_progress(&mut conn, Scope::All, "products", &product.key_text())
        .await?;
    for place in row.places {
        let state = match place.state {
            RowState::Holds => "holds it",
            RowState::Outdated => "holds an older version",
            RowState::OnItsWay => "has it on its way",
            RowState::Missing => "lacks it",
            RowState::Leaving => "is letting it go",
            RowState::Quarantined => "has it set aside",
            _ => "is in a state of a later version",
        };
        println!("{} {state}", place.replica);
    }
    Ok(())
}

The other direction

On a replica:

    let mut conn = app.pool.acquire().await.map_err(fail)?;
    let up = replica::progress(&mut conn)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    let unsent = up.iter().map(|t| t.unsent).sum();
    let unfiled = up.iter().map(|t| t.unfiled).sum();

That is the shop register’s “2 receipts not at the head office yet”. The replica tells the server the same, on its main lane, whenever it changes — with the rows it sends, at once when it starts or empties, and at most every Outbox::backlog_every while it grows. The server shows it as the replica’s backlog, as of when it was said; a replica with one is still catching up.

The dashboard

The engine answers what a dashboard asks (Progress and health); magic_sync_dashboard, on npm, draws it in Vue 3. In between are your handlers: the dashboard fetches nothing, and the engine serves no HTTP. That is on purpose — who may see which replicas is your system’s question, asked behind your login, and the engine only takes the answer as a scope.

The handlers

You write the reads, in whatever your server is built with. The shop’s, in axum (examples/shop/server/src/main.rs):

// What the dashboard reads: the engine's answers, as they are. A shop with a
// login would ask them over the registers the caller may see
// (`Scope::Replicas`); this one has none, so every register. Each read is one
// snapshot of its own, so each takes a connection, not a transaction.

async fn overview(State(app): State<App>) -> Result<Json<Overview>, Failure> {
    let mut conn = app.pool.acquire().await?;
    let overview = app
        .server
        .engine()
        .overview(&mut conn, Scope::All, 20)
        .await?;
    Ok(Json(overview))
}

#[derive(Deserialize)]
struct Page {
    state: Option<ReplicaState>,
    #[serde(default)]
    offset: u32,
}

async fn listing(
    State(app): State<App>,
    Query(page): Query<Page>,
) -> Result<Json<Listing>, Failure> {
    let mut conn = app.pool.acquire().await?;
    let listing = app
        .server
        .engine()
        .listing(
            &mut conn,
            Scope::All,
            page.state,
            magic_sync::status::Page::new(page.offset, 20),
        )
        .await?;
    Ok(Json(listing))
}

async fn progress(
    State(app): State<App>,
    Path(register): Path<Uuid>,
) -> Result<Json<Progress>, Failure> {
    let mut conn = app.pool.acquire().await?;
    let progress = app
        .server
        .engine()
        .progress(&mut conn, Scope::Replicas(&[register]))
        .await?;
    Ok(Json(progress))
}

async fn row(
    State(app): State<App>,
    Path((table, key)): Path<(String, String)>,
) -> Result<Json<RowProgress>, Failure> {
    let mut conn = app.pool.acquire().await?;
    let places = app
        .server
        .engine()
        .row_progress(&mut conn, Scope::All, &table, &key)
        .await
        .map_err(|e| {
            if e.is_transient() {
                Failure::from(e)
            } else {
                // A table that does not travel, a key that is not one: the
                // caller's.
                Failure(StatusCode::BAD_REQUEST, e.to_string())
            }
        })?;
    Ok(Json(places))
}
async fn wishes(State(app): State<App>) -> Result<Json<WishListing>, Failure> {
    let mut conn = app.pool.acquire().await?;
    let wishes = app
        .server
        .engine()
        .wishes(
            &mut conn,
            Scope::All,
            None,
            magic_sync::status::Page::new(0, 50),
        )
        .await?;
    Ok(Json(wishes))
}

The shop has no login, so it asks over every register (Scope::All). A system with one answers for the caller instead: a customer’s partition (Scope::Partition), or the replicas this user may see (Scope::Replicas(&ids)). Each read takes a connection and reads one snapshot of its own. The answers serialize as they are; return them untouched, and the dashboard of the same version reads them — an answer of another contract version it refuses, saying so, rather than draw it wrong.

What happens: the overview reads only what the engine records, so it answers in milliseconds for hundreds of replicas that are caught up — a fifth of a second at worst, while a whole fleet has rows on their way — and the page can ask every few seconds. It counts every replica but lists only the worst twenty; the listing pages through the rest, in one state or all. Progress counts what one replica should hold: ask it for the one being looked at, not for the fleet.

The page

pnpm add magic_sync_dashboard

You write the fetching, and your names for what the engine knows by id. The shop’s (examples/shop/web/src/App.vue), where api is a fetch of each handler above (examples/shop/web/src/api.ts):

import { computed, onUnmounted, ref, watch } from 'vue'
import {
  RowPlaces,
  SyncDashboard,
  type Listing,
  type Overview,
  type Progress,
  type ReplicaState,
  type RowProgress,
} from 'magic_sync_dashboard'
import { api, type FooterTwin, type Product, type ShopState } from './api'

const state = ref<ShopState | null>(null)
const overview = ref<Overview | null>(null)
const selected = ref<string | null>(null)
const progress = ref<Progress | null>(null)
// The replicas listed, a page at a time, in the state picked.
const filter = ref<ReplicaState | null>(null)
const offset = ref(0)
const listing = ref<Listing | null>(null)
// The product whose registers are shown, and where it is on each.
const holding = ref<string | null>(null)
const places = ref<RowProgress | null>(null)
const error = ref('')

// The dashboard's names for what the engine knows by id.
const names = computed(() => Object.fromEntries((state.value?.registers ?? []).map((r) => [r.id, r.name])))
const labels = { products: 'Products', receipts: 'Receipts' }

async function refresh() {
  try {
    ;[state.value, overview.value, listing.value] = await Promise.all([
      api.state(),
      api.overview(),
      api.listing(filter.value, offset.value),
    ])
    progress.value = selected.value ? await api.progress(selected.value) : null
    places.value = holding.value ? await api.row('products', holding.value) : null
    error.value = ''
  } catch (e) {
    error.value = String(e)
  }
}
refresh()
const timer = setInterval(refresh, 1000)
onUnmounted(() => clearInterval(timer))
watch([selected, holding, filter, offset], refresh)

        <!-- A register unplugged is "not answering" after 30 s, not 15 min: it is a demo. -->
        <SyncDashboard
          v-model:selected="selected"
          v-model:filter="filter"
          v-model:offset="offset"
          :overview="overview"
          :listing="listing"
          :progress="progress"
          :page-size="20"
          :names="names"
          :labels="labels"
          :silent-after="30_000"
        />

and its styles, once, where the app starts (examples/shop/web/src/main.ts):

import 'magic_sync_dashboard/styles.css'

What happens: a banner says in one line where things stand — green, everything is in sync, most of the time; amber while changes are on their way or a replica is out of reach; red when something waits for a fix — then the replicas, worst first, the tables, and the problems, each with its reason. Picking a state, or turning a page, sets filter or offset; you fetch that page of the listing. Without a listing, the dashboard shows the overview’s worst and says how many there are in all. Clicking a replica sets selected; you fetch its progress, and it shows table by table. How long ago something happened is measured from the server’s now in the answer, moved on by the time passed in the page — the browser’s clock is never compared with the server’s.

Where one row is

“Which registers hold this product?” is asked next to the product, in your own listing — not on the dashboard, which knows rows only by key. The shop puts a who holds it? link on each product: it asks the row handler with the product’s id (places, above), and shows the answer with RowPlaces, under the product:

            <tr v-if="holding === p.id">
              <td colspan="3"><RowPlaces :places="places" :names="names" /></td>
            </tr>

What happens: each register, and where the row stands on it — holds it, holds an older version, on its way, letting it go, or set aside. In a system with many listings, the same link on each listing of a table that travels answers “which units hold this operator”, “which hold this order”.

Wishes

A replica’s line says how many wishes of its settings it has not carried out, is waiting to, refused, or found changed there first (Settings and wishes); a replica with one it has not answered is catching up. One waiting is answered — the replica tries it again by itself, and its reason says what it waits for — so it does not keep the replica from synced. Engine::wishes lists them — those that need attention first — for WishList to show, each with its replica’s answer and why.

Its look and its words

Its look is CSS variables (--msd-accent, --msd-ok, …) you may set to your own tokens on any ancestor; the defaults keep every text at a contrast of 4.5:1 or more, which a test of the stylesheet holds them to. Its words are English or Spanish, following your vue-i18n locale, each a whole sentence whose counts take the locale’s plural forms. A system with a design system of its own skips the ready-made component and draws the same answers with its own parts, through useSyncDashboard and the sentence functions of magic_sync_dashboard/core: the package’s README says how.

One version, both sides

The dashboard’s types are generated from the engine’s (status.rs’s test writes dashboard/src/core/contract/types.ts), and the engine’s own test writes what it answers, every state and every kind of problem, into the dashboard’s fixtures, which its type-check holds field for field to those types. A field renamed on one side fails the other’s build. Release and upgrade the crate and the package together: they carry the same version.

Recipes

A table that extends another one

A product’s nutrition facts, keyed by the product, should go wherever the product goes:

/// A product's nutrition facts: wherever the product goes.
#[derive(sqlx::FromRow, Clone, magic_sync::Row)]
#[row(
    table = "nutrition",
    schema = "shop",
    mirror = "catalog",
    key = "product_id",
    sync = "follows(product_id)"
)]
pub struct Nutrition {
    pub product_id: Uuid,
    pub calories: i32,
}

Orders for one register

An order is taken by one register and goes to that register only:

/// An order: to the register it is assigned to.
#[derive(sqlx::FromRow, Clone, magic_sync::Row)]
#[row(
    table = "orders",
    schema = "shop",
    mirror = "catalog",
    sync = "assigned(register_id)"
)]
pub struct Order {
    pub id: Uuid,
    pub register_id: Option<Uuid>,
    pub product_id: Uuid,
    pub quantity: i32,
}

The order’s product goes with it — no declaration: the foreign key says so.

Give a register a whole list at once

Customers belong to lists, and a register is given lists. A list of 10,000 customers given to a register is one row in register_lists:

CREATE TABLE shop.lists (id UUID PRIMARY KEY);
CREATE TABLE shop.list_members (
    list_id UUID REFERENCES shop.lists (id), customer_id UUID REFERENCES shop.customers (id),
    PRIMARY KEY (list_id, customer_id));
CREATE TABLE shop.register_lists (
    list_id UUID REFERENCES shop.lists (id), register_id UUID REFERENCES shop.registers (id),
    PRIMARY KEY (list_id, register_id));

/// A customer: to every register given a list it is on.
#[derive(sqlx::FromRow, Clone, magic_sync::Row)]
#[row(
    table = "customers",
    schema = "shop",
    mirror = "catalog",
    sync = "grouped(members = list_members, assigned = register_lists)"
)]
pub struct Customer {
    pub id: Uuid,
    pub name: String,
}

Taking a list away from a register is deleting one row; the engine works out which customers to take back, and keeps those another list still gives it. The members table points at the grouped table by its key — one column or several — and both group tables point at the table of groups. Each group table gets a trigger of its own (sync_note_group), which Engine::triggers prints with the others.

A table both sides write

A register reports its own status on a row the head office owns. Mark the columns the register owns: they go up, the rest come down, and a change to only the register’s columns is not sent back down:

/// A register's state: the head office writes its label, the register its
/// battery.
#[derive(sqlx::FromRow, Clone, magic_sync::Row)]
#[row(
    table = "register_state",
    schema = "shop",
    mirror = "both",
    sync = "assigned(id)"
)]
pub struct RegisterState {
    pub id: Uuid,
    pub label: String,
    #[row(replica_owned)]
    pub battery: Option<i32>,
    pub updated_at: DateTime<Utc>,
}

The row begins on the server: a register edits only its own columns of a row it holds.

A setting both sides want to change

The head office wants a register’s screen brightness at 70; the register’s user turned it to 40. Two writers, one setting. The register is right about what the setting is, and the head office about what it wants: declare the register’s row a setting, and let the head office wish it other values — Settings and wishes is the whole of it.

Archived rows replicas should not hold

Keep a view of the rows that should travel, and name it:

CREATE VIEW shop.live_products AS SELECT * FROM shop.products WHERE archived_at IS NULL;

/// A product, as replicas hold it: never an archived one.
#[derive(sqlx::FromRow, Clone, magic_sync::Row)]
#[row(
    table = "products",
    schema = "shop",
    mirror = "catalog",
    sync = "broadcast",
    active = "shop.live_products",
    unmirrored = "archived_at"
)]
pub struct LiveProduct {
    pub id: Uuid,
    pub name: String,
    pub updated_at: DateTime<Utc>,
}

Archiving a product takes it back from every register — retired where a receipt still points at it. The view decides what is sent for its own sake, not what may travel: an archived product a row the register holds points at still goes there, as that row’s parent, and leaves once nothing points at it. Leave active out and archived products travel like any other, with their archived_at as data, for the register to show as archived.

Several companies on one server

Give rows and replicas a partition column, and name it:

use magic_sync::plan::Replicas;

pub const REGISTERS: Replicas = Replicas::new("shop.registers", "registers")
    .live("active")
    .partitioned_by("company_id");

Every table that goes down carries company_id; a broadcast product reaches only its company’s registers. The triggers Engine::triggers prints note it.

PostgreSQL can keep a row from pointing into another company: give each parent a unique key on (company_id, id), and point at it by both columns.

CREATE TABLE shop.regions (
    id UUID PRIMARY KEY, company_id UUID NOT NULL, name TEXT NOT NULL,
    UNIQUE (company_id, id));
CREATE TABLE shop.stores (
    id UUID PRIMARY KEY, company_id UUID NOT NULL, region_id UUID NOT NULL,
    FOREIGN KEY (company_id, region_id) REFERENCES shop.regions (company_id, id));

What happens: the engine reads the partition column on both sides of a foreign key as “the same company”, which no row crosses anyway: a store points at its region by region_id, so sync = "follows(region_id)" works, and a region goes wherever a store that points at it goes. A key made only of company_id — into a table of companies, say — points at no parent. A parent whose primary key holds company_id — a table partitioned by it — is pointed at by every column of the key, company_id too.

Upgrading

Two things outlive a build of magic_sync: the engine’s tables in each database, and the protocol the server and its replicas speak. A release says in its changelog when it changes either, and what follows is how each changes without losing anything.

The engine’s schema

Each side’s database holds the engine’s own tables at a version of that side’s own: SERVER_SCHEMA_VERSION for the server’s, and REPLICA_SCHEMA_VERSION for a replica’s. They count different steps and move independently: a release that changes only the server’s tables leaves the replicas’ version where it was, and the other way round. A build runs against a database at its own version only: Server::new and Replica::new refuse any other, naming both.

A release that changes one side’s tables raises that side’s version and brings the step from each earlier version with it: one script per version, applied in order, in one transaction — never edited once released. So the engine’s tables move forward one step at a time, and never back: a database a later build upgraded is refused by an earlier one rather than misread.

• Without migrations of your own, starting the new build is the upgrade. Server::install brings the server’s database from whatever version it is at to this build’s, in one transaction and one process at a time (an advisory lock), before the server starts; replica::install does the same on a replica, in one transaction.
• With migrations, the upgrade is a migration of yours that runs the release’s step: plan::migration(n) on the server, replica::migration(n) on a replica, for version n — each also records the version, and your first migration ran step 1: the changelog names the step. Print the triggers again (The tables) in the same migration: a release may write them differently, and the server refuses to start while one differs.

On the server, stop every process of the earlier build before the database is upgraded, then start the new one: a process keeps running against the tables it started with, and does not know what the step changed. The replicas keep working meanwhile — they are built to be offline — and catch up when the server is back.

Each replica upgrades its own database when its new build starts, whenever that is: a fleet is never upgraded at once, and does not need to be.

Your declarations

What decides which rows each replica holds is yours: the replicas’ relation, and per table its key, its sync rule, the tables it points at, its groups and its columns. The relay remembers a digest of them (sync.version.declared); the first pass of a build whose declarations differ — a table routed otherwise, mirrored for the first time or no more, a column added — provisions every live replica again, and lets go of what it kept of a table it mirrors no more. That is an alarm in the log, and a deploy’s worth of traffic: change declarations on purpose. A change to how a codec encodes a row that leaves its columns alone is not seen: provision the replicas yourself (Provisioning).

On a replica, what it wrote of a table its build does not send up — one that no longer goes up, or not yet, after a rollback — stays, set aside and counted in the backlog, and goes up once a build that sends it runs: rolling a replica back and forward again loses none of its writes. An earlier build keeps the note triggers a later one wrote for a table it does not know, so what the host writes there while it runs is noted too. A table that no longer travels for good keeps them as well: drop its sync_note_… triggers in a migration of yours.

The protocol

Every message carries the protocol it speaks, PROTOCOL: a new kind of message, or a field whose meaning changes, raises it; a new field does not (the protocol, section 4.0).

A side that receives a message of a later protocol than its own does not act on it: a replica leaves it unacknowledged, and everything behind it waits until the replica is updated; the server answers a replica’s upload with an error, and the replica sends it again until the server is updated. Nothing is lost either way, and the order of the upgrade only decides which direction waits. A release that raises PROTOCOL says in its changelog which side to upgrade first.

The dashboard

magic_sync_dashboard reads the JSON the engine answers, of the version it carries (status::CONTRACT), and refuses another rather than draw it wrong. The crate and the package are released together, at the same version: upgrade them together.

How it works

This chapter explains the engine from the ground up: the idea it rests on, what a table declares, what happens to a change from the moment it is written to the moment a replica holds it, how every kind of failure is handled, and what the host application has to provide. Read it in order; each section leans on the ones before it.

1. The idea: compare states, do not replay a log

Most sync engines replay a log. Every change becomes an event, the events are sent in order, and the replica re-applies them. That design breaks in familiar ways:

• A lost or skipped event is a permanent divergence. If an event fails and is skipped, nothing will ever send it again, because the log has moved on.
• A failing event blocks everything behind it. Order matters, so one bad event stops the queue for everyone (the classic “poison message”).
• Repair is manual. Someone has to find what was lost and replay it by hand.

This engine does not replay anything. For every replica, the server keeps a record of which rows that replica holds (the held set). When something changes, the engine works out which rows the replica should hold now (the wanted set) and sends only the difference:

send   = wanted − held          (rows to hold, or that changed)
delete = held − wanted          (rows to let go)

Then it records the new held set, and the messages that carry the difference. A change is never “an event that must be delivered”; it is a reason to compare two sets again. That single choice is what makes everything else in this document possible:

• A row a replica did not take, or that the replica refused, is not counted as held. The next comparison finds it missing and sends it again — as it is then, not as it was. (A message the transport has not taken yet waits as it was written, in the order it was written, until it does; a later change to the row follows it.)
• Nothing needs to be replayed in order. If a replica misses messages, the next comparison catches it up from the current state.
• Repair is a comparison too. Provisioning a replica from scratch is “compare against an empty held set”.

Payloads are always rebuilt from the current rows. The engine never stores the content of a change, only that something changed: what it keeps of a message is the message itself, until the transport has taken it: an outbox, not a log to replay.

2. Words

3. What a table declares

A table is declared once, as the Rust struct that is its row, implementing row::Row. The declaration answers two questions.

Which way does it travel? (Mirror)

Which replicas hold which rows? (sync, for tables that go down)

Referenced is what keeps a replica consistent. A row always travels with the rows it points at: if a replica is sent an order that points at a customer, it is sent the customer too, even though customers have no rule of their own. And a parent is only taken back once nothing the replica still holds points at it. The engine reads the foreign keys from the database catalogue, so this needs no declaration: the schema already says it.

The rest of a declaration

#[derive(Row)] turns them into the row’s implementation of row::Row (The rows), and is the only way to implement it: what it writes is checked where it is written, and again against the database when either side starts.

The codec

A table travels in its own encoding: JSON, an object of its columns (Codec::json), or a protobuf message of its own with a mapping each way (Codec::new). A wire::Codec erases the row type so the engine can hold every table in one list. The engine never names a table; it dispatches on the table name through the codecs.

4. The life of a change, server to replica

Here is everything that happens when someone renames a product on the server, until every replica that should hold it holds the new name.

 UPDATE products …                          (the host's transaction)
   └─ trigger: INSERT INTO sync.notes        (same transaction: atomic)

 relay pass
   0. observe  (what became of the replicas: live, moved, gone)
   1. record   (one REPEATABLE READ transaction on the relay lock's connection)
        read pending notes ──▶ plan ──▶ load and encode rows, a chunk at a time
        ──▶ per replica: pack into messages, number them
        ──▶ sync.outbox ← the messages;  sync.held ← what they carry;  notes ← done
   2. publish  (no transaction open)
        per replica: its outbox, in order ──▶ transport.send ──▶ out of the outbox

 replica
   apply the message in one transaction ──▶ ack

4.1 Noting

Every mirrored table has the engine’s triggers, as Engine::triggers writes them (The tables). Every insert, delete, and update that changes a column the replicas do not own writes a row to sync.notes: the table as the engine knows it (a partitioned table by its own name, not its partition’s), the row’s key in the order the row declares it, the columns it pointed at other mirrored rows with before the change, and the id of the transaction. A key that changed notes the old key too, in the partition it was in, and a row that moved to another partition is noted in the one it left as well. Because the note is written by a trigger in the same transaction, no write path can forget to note, and a rolled-back change leaves no note. TRUNCATE is refused on a mirrored table, because it would remove rows without a note.

Group tables get a trigger too: a member joining or leaving a group, or a group given to or taken from a replica, is noted the same way. The server refuses to start while a trigger is missing, disabled, or not as it would write it.

4.2 Planning

Before each pass the relay compares the replicas (Replicas::new and live) with what it saw last (sync.replicas.live, partition_id): a replica that became live, or moved to another partition, is provisioned again from scratch; one that stopped being live, or is gone, is forgotten — nothing recorded as held by it, nothing sent it.

The relay reads up to a batch of pending notes, oldest first, in a REPEATABLE READ transaction: the notes, the plan, the rows it loads and what it records are one snapshot, so a row is never sent as it is after a write the plan did not see — a parent it now points at left behind. A note this build cannot read (a later build’s) is skipped by the query, and holds back nothing. It reads every note that is not done, instead of “every note after the last one I read”, because PostgreSQL assigns ids before commit: a transaction that started first can commit last with a lower id. A cursor would skip it; a “not done” query never does.

What a replica should hold is the least set of rows closed under three rules: the rows its tables’ rules send it; every row one of those points at (a parent goes with its child); every row that follows one of those. For the notes it read, the planner works out how that set moved, around what changed — never the whole of it but for a replica provisioned from scratch:

1. Rows in question. Of the rows a replica holds, those a change may have taken out of its set: a changed row its rules no longer send it (a Referenced or following row: any that changed), the parents a changed row pointed at before, rows that cannot go. Then, closed over what they point at and what follows them, the rows of its that leaned on those. Every row it holds outside them stays as recorded.
2. What it keeps. The rows its rules send it; the rows in question that a row it holds outside them points at, or that follow one it holds; then, a round at a time from the rows the last round added, what those need: the parents of a row it lacks or that is in question (a follower points at its parent too), and the followers of one it lacks or that is in question.
3. The difference. What it keeps and lacks — or holds as a version older than a change of the batch, or refused — is sent; the rows in question it does not keep are taken back.

What a row a replica holds points at is what the version it holds points at: the row as it is, unless it changed since — then what the first change after that version recorded it pointed at before, whether that change waits beyond the batch, is parked, or was planned while the replica kept the older version for something that still needed it. That version’s parents stay as long as it does. Rows that cannot go (section 4.3) are never kept, nor any row that points at one; a plan that would send such a row runs again with it among them, and a row set aside for pointing at one is tried again once that row changes.

The plan is worked out in the database, in temporary tables of the relay’s connection, set by set, and read a page at a time by what records it: a plan, however many rows it sends — a snapshot of a large replica, a group of thousands given to many replicas — is never in the relay’s memory whole. Every note of the batch is planned together, in one plan: a part planned apart could undo what another sends — take back a parent a child planned elsewhere still needs.

What a replica’s version holds. Every plan reads one snapshot of the database. When a replica is recorded as holding a row, the newest note of that row the snapshot sees is recorded with it. Writers of one row take turns on its lock, so a row’s notes are numbered in the order its changes committed: a later note of it is a change the replica lacks, and sends the row again; an earlier one is a change it holds, and sends nothing. A change committed after the plan read — by a transaction that started earlier, or took its transaction id first — is never missed, and a change the replica already received is never sent twice. The numbers are a sequence’s, so a database dumped and restored, or moved to another cluster, keeps them: its transaction ids start over, and nothing depends on them. (Logical replication carries no sequences: a database moved by it has them set, as for every serial column, before its relay runs.)

4.3 Loading

Every row the delta sends is read once and encoded once by its codec, however many replicas get it, in the plan’s snapshot, a chunk of rows at a time: what the relay holds in memory is a chunk, and one message being filled per replica, however much it sends. Filling many replicas’ messages at once, it ends them early past a bound — never between a table’s deletes and its upserts, which a unique value moving from one row to another needs in one message. A row that cannot go — it does not encode, or no message could carry it — is planned around: the plan runs again with it blocked, so it is not sent, what points at it is not sent either, and a replica holding an older version of it has that version taken back with whatever points at it there. A replica never keeps a version the planner can no longer reason about; each of those rows is quarantined and comes back once a fix lets it through.

4.4 Recording, then publishing

Each replica’s rows are packed into messages (envelope::Down) under the transport’s size limits: parents’ rows first, then deletes, children first. Every message goes out as one envelope::ToReplica, whatever it carries, so a transport moves bytes and never looks inside. A row too large for any message is set aside at loading, like one that does not encode (section 6). Each row travels with its key, so a replica that cannot even decode a row can still say which one it was.

Every message is numbered, and chained. The relay draws one number per message from a PostgreSQL sequence, and gives them out in the order the messages go. A sequence never goes backwards and is not rolled back with a transaction, so no two messages ever share a number. A replica applies nothing at or below the last number it applied: a message delivered twice changes nothing. Each message also carries a random nonce, and names the one sent the replica before it, by number and nonce (previous, previous_nonce); a replica applies a message only when it follows the last one it applied, so one that follows a message it never got — lost on the way, or the replica’s database restored from an older copy — is a gap: it applies nothing more until a snapshot begins, tells the server the newest message it could not follow, and is provisioned again, unless a snapshot that began after that message is already on its way. Messages carry the epoch too, a random id of the server’s installation: a new epoch always begins with a snapshot.

Every message is sealed for its replica: an HMAC of its replica, epoch, number and nonce under a secret of the epoch (sync.epoch). The replica gives the seal back, as it came, when it says what it applied, so the server knows it sent that message. A server database restored from an older copy numbers again from where the copy left off, but draws other nonces: its replicas apply none of it, and say what they applied — a message the server sealed and has no record of sending. The server then takes a new epoch and provisions every replica again; one restored from before its last new epoch hears of an epoch it never had, from replicas it last heard follow its own, and provisions each again. A replica that says it applied less than it said before was restored itself, and is provisioned again.

A provisioning (a snapshot) sends everything the replica should hold, then an envelope::EndSnapshot. On the end, the replica deletes every row of those tables the snapshot did not name. A snapshot replaces what the replica was owed: its outbox is dropped when the snapshot is written.

Recorded before it is sent. Each message is written to its replica’s outbox (sync.outbox) as it is filled, and what it carries is recorded in sync.held with its number, in the same transaction that marks the notes done. Only once that commits does the relay publish: each replica’s outbox, in order, a message leaving it once the transport took it — up to Settings::concurrency replicas at once, so one slow to take its messages holds back no other. So the server decides what a replica will hold before anything goes, and nothing goes that is not recorded: a relay that dies after the commit loses nothing — whoever runs the relay next sends the outbox — and one that dies before it sent nothing. A message keeps its number, and its transport id, however often it goes; a database restored from an older copy numbers its next messages again, and each of those has an id of its own ({epoch}-{sequence}-{nonce}). What a message carries is stored once, in sync.bodies, however many replicas’ messages carry it — the rows of a change every replica gets are one body — under each replica’s own head and tail: a protobuf message is its fields one after the other, so the bytes the transport is given are the message’s own.

The record transaction runs on the connection that holds the relay’s lock. If that connection is lost, PostgreSQL releases the lock and rolls back the record together. So a relay that lost its lock can never commit over the one that took it.

A replica the transport cannot reach keeps its messages in its outbox, in order, while every other one is sent to: nothing is planned again for it. Should its outbox outgrow outbox_messages or outbox_bytes behind any snapshot it begins with, it is dropped and the replica provisioned again instead — an alarm, in its last_error.

4.5 Applying, on the replica, and saying so

The replica applies a message in one transaction (replica::apply):

• Each row overwrites its copy, except the columns the replica owns, whatever its version: the chain orders the server’s messages, so the last applied is the server’s latest, even after a restore or a clock set back.
• A row it had retired is offered again and becomes live.
• A row it is told to let go of is deleted or, if a row of its own still points at it, retired (its RETIRED column set). A sale keeps its customer. A retired row is let go of for good once nothing that stays points at it, children first: after the message is applied, in a transaction of its own, so a row that cannot go never makes a message run again.

Then it tells the server the last message it applied — its number, its nonce and the seal it came with. Until it does, what the server recorded in 4.4 is only on its way: every held row remembers the message that carried it, and counts as held once the replica has acknowledged that message. This is what keeps the progress figures (section 6) honest: they count what the replica confirmed, never what was merely sent.

5. The life of a change, replica to server

A replica writes its own rows (records, and its columns of two-way tables). A trigger notes each write in the writing transaction (sync_notes: one note per row — the table and the key — counting its writes), so no write path can forget, and a row written a thousand times is one note. The replica’s upward loop gathers the noted rows as they are now — a row edited twice goes up once, with its last version — into an envelope::Up, and hands it to the host’s uplink. A row that is gone by then goes as its key, but only for a table keyed by its replica (ORIGIN): history a replica pruned before it went up is not the server’s to delete.

Every send of an upload draws a random number of its own, a resend included, stamped on its notes in the transaction that makes it. The server answers every upload it filed with a Filed message, after its transaction commits, and the replica lets go of the notes that send carried — a note sent again since waits for the answer to that send, and a row written since goes up again, so a late answer never forgets a write it did not carry. An answer that arrives before the replica recorded the send counts all the same; one meant for another replica, or for this one before it was restored, names nothing of its. An upload the server does not answer goes again after a while. An upload stays within the transport’s size; a row that can never fit, or that the transport refuses for good, is set aside and counted in the backlog until the replica runs again — found by sending half of what was refused, and half again, and blamed only when the transport took some other message meanwhile; a transport that refuses everything sets nothing aside and is asked again after a backoff. So the replica always knows what it wrote that has not reached the server yet.

The same Up carries everything else a replica has to tell the server: the rows it could not take, the last message it applied, and, hourly, its digests. A replica may split its writes into lanes — a table written often on a stream of its own — and the first lane carries what the engine itself says.

The server files an upload under the replica the message came from, and what the server knows about it, never what the rows claim:

• a record is kept from its first delivery, or the newest report wins when it has a VERSION;
• a two-way table takes only the columns the replica owns, and only of a row the server has and records the replica holding: an edit of one it forgot the replica holds while it provisions it again waits; one of a row it deleted or is taking back is dropped, the server’s change winning; one of a row the replica was never sent is refused — a replica never creates a two-way row;
• a catalog is refused, because replicas do not write the server’s tables;
• a replica deletes only its own rows, and the server lets go of its copy of one only where the replica had applied every message that carried the row: otherwise the replica may have taken the row again from a message still on its way, and the server’s own delete follows it (the protocol, Appendix A.3).

The host passes its own knowledge of the sender as a value of its choosing (the P in Codec<P>), and each codec may stamp rows with it.

6. When something fails

This is the contract the engine is built around:

An operator fixes the infrastructure or deploys a fix. Nobody repairs data by hand.

Every failure is one of two kinds (Error::is_transient):

• Transient: the infrastructure failed. The database is unreachable, the transport is down, a deadlock happened. Retrying the same work unchanged may succeed.
• Permanent: the data or the code failed. A row that does not encode, a value a constraint refuses, a bug. Retrying unchanged fails the same way.

The transient side is the listed one: an error nobody listed is treated as permanent, so it is set aside and shows, instead of being retried forever.

A transient failure stops the pass, and the same pass runs again. Nothing is lost: the notes are still pending and nothing was recorded. Order is preserved because everything waits.

A permanent failure is narrowed down to what it is about and set aside. Nothing else waits for it, and it gets through by itself once the cause is fixed:

The backstop: digests

Every so often, each replica reports a digest of what it holds of every table: how many rows, and a fingerprint of their keys (FNV-1a over the sorted keys; rows it retired or could not take are left out). The server compares it with what it records the replica holds (leaving out the rows the replica refused, as it does). It only compares once the replica has applied every message recorded for it, so a message on its way is never mistaken for drift. If they differ, the replica is provisioned again from scratch.

So even a failure nobody predicted, or one misclassified as harmless, ends with the replica holding exactly what it should, without anyone noticing it first.

Progress: what each replica holds, honestly

Every read below takes a scope — some replicas, a partition, or all of them — and counts only live replicas, so a host shows each audience its own.

plan::Engine::overview answers first whether everything is where it should be. Per replica, one state:

It never counts the rows a replica should hold: it reads only what the engine records, so it costs little for a fleet — a few milliseconds for 200 replicas holding 3 million rows, while they are caught up; the rows still on their way are counted one by one, about a fifth of a second when every replica has all of them unconfirmed (bench_the_dashboard_over_200_units_holding_3m_rows, on a 32-core workstation) — and it is exact all the same. A change a replica lacks is in one of three places — a note not yet planned, a message it has not confirmed, or set aside — and each is recorded. It counts every replica per state and lists the worst up to a limit; plan::Engine::listing pages through the rest. Per table, the same figures over the scope’s replicas say which tables are not where they should be yet. Every read is one snapshot, and carries the server’s now.

plan::Engine::progress answers how far along: for a few replicas, per table:

holds - outdated of should is the replica’s progress bar, the way a music library shows “40 of 500 songs on this tablet”. It cannot run ahead of the replica, because nothing counts until the replica says it applied it. plan::Engine::row_progress answers the other question, “where is this row?”: per replica, whether it holds it, holds an older version, has it on its way, lacks it, is letting it go, or has it set aside.

The replica answers the opposite direction with replica::progress: per table, rows it wrote that are not sent yet, and rows sent the server has not filed yet. It tells the server the same whenever it changes, and the server shows it as the replica’s backlog.

What an operator sees

plan::Engine::health answers what the engine waits on over a scope: how many notes are pending and since when, the parked notes, the rows in quarantine and why, the uploads in the inbox and why, and the replicas a send keeps failing to. Each of these means one of two things: wait (the infrastructure is down), or deploy a fix (a bug). Never “go fix the data”. When nothing waits for a fix, the engine is healthy — the normal state; a replica out of reach is listed, and waited for.

7. Guarantees, and what they rest on

The engine guarantees that:

1. No committed change is lost. It is noted in the same transaction, and a note is only done in the transaction that records what it asked for and writes the messages that carry it, which are sent until the transport takes them.
2. A replica converges to what it should hold. At-least-once delivery, idempotent application, and the digest backstop.
3. A replica never holds a row without the rows it points at.
4. A replica never goes back in time. The messages of an epoch are chained: the replica applies each once, in order, so the one applied last is the server’s latest.
5. One failure holds back only what it is about.

They rest on:

• One relay at a time. The server’s runner holds a PostgreSQL advisory lock while it relays, and records each pass on that lock’s connection.
• A transport that keeps its contract (the protocol, section 2). Without it, a failed message could be retried after the ones behind it, or lost until the safety net of section 6 catches it.
• Transactions. record and file run in a transaction (they use savepoints); apply runs in one so a message applies whole or not at all.
• A monotonic version column, on a record a replica edits: the replica must stamp it with a clock that never goes behind the value it replaces, or its newer report is filed as older and dropped.

8. What is the host’s, and what is the engine’s

The host writes its tables, a struct per table, the codec list, where the replicas are, how bytes travel each way (a run::server::Transport and a run::replica::Uplink), and, when it needs to, what it knows about a sender (run::server::Senders). Build your own walks through each.

Everything else is the runners’ (run): on the server, the lock, the passes and their backoff, retrying what was set aside, pruning, the alarms, filing uploads and answering them; on a replica, applying each message in one transaction, and the upward loop. The step functions they are built on — Relay::record/publish, replica::apply, replica::outbox and the rest — are the engine’s own: a host drives the engine through the runners.

9. The parts

The derive lives in its own crate, magic_sync_derive, re-exported as magic_sync::Row and magic_sync::Columns.

The crate is compiled for one backend at a time: postgres for the server, sqlite for a replica.

10. How it is tested

• The planner is driven by a seeded random walk of writes over a schema with every strategy. After each step, every modelled replica must hold exactly what its strategies say, and never a row without its parent.
• Each failure in section 6 has its own test, against a fake transport and a codec with a deliberate bug that a “deploy” fixes.
• A storm runs the same walk through the real relay while the transport fails a fifth of the sends — some for good — and loses messages; the relay dies between its record and its sends, or loses its lock mid-pass; replicas take part of what waits for them, die before acknowledging, or are not heard. Replicas are modelled from the protocol (src/model.rs): epochs, chaining and gaps, snapshots, rejections. Each time the system settles, every replica holds exactly what the engine records; once the faults stop and the fix is deployed, exactly what it should, with provisionings rare.
• Races and restores: a transaction that commits after a plan read past it, a pass recorded and never sent, a lock lost mid-pass, a send cut short before a row leaves the replica’s scope, a message lost, a server or a replica restored from an older copy.
• The replica side runs on SQLite: messages from the past, rows it cannot decode, tables it does not have, digests.

Run them with cargo test -p magic_sync --features postgres (PostgreSQL through DATABASE_URL; each test gets a database of its own) and cargo test -p magic_sync --features sqlite.

The replica protocol

What a replica must do to be kept in sync by this engine, whatever language it is written in and whatever database it keeps its rows in. The engine’s own replica (src/replica.rs, on SQLite) is one implementation of it; this document is the contract that one follows, and any other must.

The messages are in proto/sync.proto. Generate code for your language from it; the engine’s own types are generated from it too, and a test fails while they differ.

The words must, must not and should mean what they mean in RFCs.

Sections 1 to 10 are what a replica does, and the whole of what an implementation needs. What the server does with what a replica sends is in Appendix A: it explains the rules, and no replica has to implement any of it.

1. The parties

• The server owns the truth. It decides which rows each replica should hold, sends the difference, and records what each replica holds.
• A replica holds a copy of some of the server’s tables, and may write tables of its own that go up to the server. It is identified by a UUID the server knows.
• The transport carries messages between them. The engine does not depend on which one, but it requires what section 2 lists.

A replica never decides what it should hold. It applies what it is sent, reports what it could not apply, and reports what it holds. It follows one server installation: two that send it messages each take the other’s for a new epoch, and provision it again and again.

2. What the transport must provide

The server sends one kind of message, ToReplica, which carries a Down, an EndSnapshot or a Filed. A replica sends one kind, Up, on one or more lanes (section 7). A transport carries bytes and nothing else: it never needs to know what is inside. Whatever it is — a broker, HTTP over a queue of your own, files on a USB stick — it must provide, in both directions:

1. Nothing it took is lost. Once the transport accepted a message, it delivers it: while the other end is unreachable, and across a restart of either end or of the transport itself. It stores, then forwards.
2. In order. A replica’s messages are delivered in the order the server sent them, and a replica’s Ups on one lane reach the server in the order it sent them.
3. One at a time, down. The next message to a replica is not delivered until the replica acknowledged the previous one (5.1).
4. At least once. A message that was not acknowledged is delivered again. Each side therefore treats every message as one it may have seen before (4.1, 7).
5. An acknowledgement names its message. It releases that message and no other: one that arrives twice, or late, never releases a message the replica has not applied.

With NATS JetStream, that is a stream on file storage, and per replica a durable pull consumer with explicit acknowledgements and max_ack_pending = 1; up, a stream the server consumes per lane in order.

A message the transport carries has a size limit of its own; the server keeps every message within it, and a replica must keep every Up within it (section 7).

A transport acknowledgement (section 5.1) is what makes the transport deliver the next message. It is not the same as telling the server what was applied (section 5.2).

A transport that loses a message anyway costs time, not data: the next message names the one before it, so the replica knows it missed one and is provisioned again (4.1); a replica that leaves messages unconfirmed while nothing newer goes to it is provisioned again (A.1), an upload the server did not answer is sent again (section 7), and a digest that disagrees provisions the replica again (section 8). That is the engine’s safety net, not a licence: until it acts, the replica holds less than it should.

3. Rows and keys

Each table travels in its own encoding, agreed between the server and the replica for that table: either a protobuf message with one field per column, or a JSON object of its columns (the column names as keys). The engine’s messages carry those rows as opaque bytes, so this protocol never changes when a table does.

A row’s key is its primary key, as text, in key order:

• a UUID part: lowercase, hyphenated (0196…-…);
• an integer part: decimal;
• the parts joined by | where a single text is needed (a1b2…|3).

This is the one form keys are compared in, and every key a side writes into a message must be in it. A side reading a key parses each part as its column types it and writes it back in this form before using it (a UUID that came uppercase or without hyphens is the same key); a key that does not parse is malformed.

A table that travels as JSON is a JSON object with one member per column, named as the column, each value as the column’s type writes it in JSON — except a 64-bit integer, which is its decimal text ("9007199254740993"): a JSON number would be rounded by a reader that holds numbers as doubles. A decimal is its text, a UUID its canonical form, an instant RFC 3339.

4. Applying what the server sends

4.0 The protocol version

Every ToReplica and every Up carries protocol, the version of this document its sender speaks (now 1).

• A replica must set protocol on every Up it sends.
• A replica that receives a ToReplica of a later protocol than its own must not act on it and must not acknowledge it to the transport: it waits to be updated, and the message is delivered again then. Every message behind it waits too, which is right, because they speak the same version.
• The server does the same with an Up of a later protocol than its own, and drops an Up of protocol 0 as malformed.

A message that is malformed — bytes that do not decode as a ToReplica; one of protocol 0 (unset), or of the replica’s own protocol or earlier whose body it does not know, or that has none; numbers no server sends (4.1) — carries nothing a replica can act on: it must apply none of it, should log it, and must acknowledge it, so it does not hold back the queue. The digests (section 8) find whatever it carried, if it mattered.

What a server sends, field by field — anything else is a number no server sends, and the corpus of malformed messages (tests/golden/malformed.txt) has a case of each:

• protocol: 1 to the replica’s own. Cases: no_protocol.
• sequence: 1 to 2^63 − 1. Cases: sequence_zero, sequence_too_large.
• nonce: 1 to 2^63 − 1. Cases: no_nonce, nonce_too_large.
• previous: below sequence. Cases: follows_itself, follows_a_later_one.
• previous_nonce: 0 with previous 0, and otherwise 1 to 2^63 − 1. Cases: follows_without_its_nonce, follows_nothing_with_a_nonce, follows_a_nonce_too_large.
• epoch: a UUID. Cases: epoch_no_uuid.
• snapshot_id: a UUID; empty on a message of no snapshot. Cases: snapshot_no_uuid, ends_a_snapshot_of_no_id.
• mac: 16 bytes. Cases: no_mac, mac_too_short, mac_too_long.
• keys: one per upsert. Cases: keys_not_one_per_row.
• tables, deletes, upserts, uploads: no more rows — upserts and deletes together — and no more tables, or uploads, than the replica takes in a message: the server’s limits, or more (500 each by default). Cases: more_rows_than_a_message_carries, more_tables_than_a_message_carries, ends_more_tables_than_a_message_carries, files_more_uploads_than_a_message_carries.

A row’s own fields are the row’s to answer for, not the message’s: a key that does not parse, a row that does not decode, a table the replica does not hold — the row is rejected, or its delete ignored (4.1), and the rest applied (a_delete_of_a_key_that_does_not_parse, a_row_beside_another_key).

What changes the protocol’s number: a new kind of body, or a field whose meaning changes. What does not: a new field. A side must ignore fields it does not know, so a message with a field added means to a side that predates it what it meant before.

4.1 Down: rows to hold, rows to let go

A replica must apply a Down in one transaction: either all of it, or none of it.

The number comes first. Every Down carries a sequence and an epoch, a nonce of its own — a number the server draws at random for it — and previous and previous_nonce: the sequence and nonce of the message the server sent this replica before this one, in this epoch — both 0 on a message that begins a snapshot (4.2), which follows nothing. It carries a mac too, the server’s seal, which the replica keeps as it came and gives back (5.2). The replica keeps the (epoch, sequence, nonce, mac) of the last message it applied, and checks, in order:

• sequence 0 or above 2^63 − 1, an epoch that is not a UUID, a previous at or past sequence, a nonce 0 or above 2^63 − 1, a mac of other than 16 bytes, a previous_nonce above 2^63 − 1, a previous of 0 with a previous_nonce, or another previous without one, or a previous of 0 on a message that begins no snapshot: no server numbers a message so; nor sends a table whose keys are not one per upsert. The message is malformed (4.0).
• the same epoch and a sequence at or below the last one: the message changes nothing. The replica must not apply it, and must acknowledge it to the transport: it is a redelivery, or the server’s database was restored from an older copy and is numbering again — its nonce then is another, but the replica need not tell: either way it should tell the server the last message it applied again (5.2), and a restored server finds out from that.
• previous 0: the message begins a snapshot, and the replica applies it, whatever it applied before — of this epoch, or of another, which is a server installed again or restored: it keeps that epoch’s numbers from now on — and whatever gap it had. A new epoch always begins with a snapshot.
• otherwise the message must follow the last one applied: previous and previous_nonce are its sequence and nonce, of the same epoch. If not, the replica missed one — lost on the way, or the replica’s own database restored from an older copy, or the server’s numbering again — and has a gap. Once it has a gap, it applies nothing but a message that begins a snapshot: not even one that follows the last applied. It must not apply the message, must acknowledge it to the transport, and keeps the sequence of the newest message it could not follow, which it says to the server (5.2) until a snapshot begins. The server answers with one.
• a message that follows the last applied and names a snapshot other than the one in progress — or none, while one is in progress, or one while none is — is one no server sends: malformed (4.0, 4.2).

Then the replica applies tables in the order given. The server puts parents’ rows before their children’s, and children’s deletes before their parents’, so in that order no row ever points at a row the replica lacks. It keeps a table’s deletes in the message that carries its upserts: the rows from a table’s first upsert to its last delete are split across messages only when they do not fit one on their own. So a unique value one row lets go of is there, in the same message, for another row to take.

For each Table:

A table the replica holds from the server is one the server writes: a catalog, or a table written on both sides. Rows of any other table — a record of the replica’s own, a table it does not know — are rejected, and their deletes ignored.

1. Deletes, first: a value a row lets go of — a unique one — is then free for an upsert of the same part to take. The keys in deletes are let go of together:

   • a row that a row staying for the replica points at — a row of its own, through any foreign key: one that would refuse the delete, checked at once or at commit, and one that would cascade or set the pointing row’s columns, which would change a row the server never sent; or a row it retired — is retired instead (4.3), and so, in turn, are the rows a row retired so points at among them. Where the table has nothing to retire it by, the row is kept, and a rejection recorded (section 6);
   • every other is deleted, together, so rows that point only at one another — a tree, a row that points at itself — go whatever the order of their keys. One the database refuses to let go — a trigger, or a row of the server’s that stays and points at it — is kept, and a rejection recorded.

2. Upserts. Each entry of upserts is one row, encoded as the table’s message; keys[i] is the key of upserts[i], and there must be one key per upsert. A row whose own key is not the one beside it is no row the server meant: the replica rejects it under the key beside it. For each row:

   • insert it, or overwrite the stored row with the same key, whatever its version: the messages of an epoch are ordered by their chain (4.1), so the one applied last is the server’s latest, even when its version is older than the stored one — a server restored from a copy, or whose clock was set back, sends what it has;
   • columns the replica owns in a table written by both sides are not overwritten;
   • a row the replica had retired (4.3) becomes live again;
   • a row of its own it deleted, in a table keyed by the replica, whose delete the server has not filed yet (4.4), is not taken: the server sent it before it heard of the delete. The replica records it held back — a rejection marked held_back (section 6) — so the server does not count it held, and sends it again should it keep the row.

   A key both let go of and upserted in one part is held: its upsert comes after. A key upserted twice in one part is the last row beside it.

A row the database refuses is tried again once the rest of its part is in, and once the whole message is — every part of it, the deletes of later parts included — each while that takes one more: it may point at a row after it, or take a unique value a row after it, or a delete after it, gives up. One that a later part lets go of is not tried again: the server let go of it; nor is one a later part sends again: the later row is the one, and the earlier is dropped, neither taken nor rejected. What is still refused then is rejected (section 6). Two rows that swap a unique value cannot be taken one at a time — the database checks each row as it is written — so both are rejected, and keep what they held, until the server sends them otherwise.

A row the replica cannot take, or let go of, does not fail the message. If a row does not decode, if the replica’s database refuses it for its data (a constraint, a trigger), or if the replica does not know the table at all, the replica must leave that row as it was, apply everything else, and record a rejection (section 6) in the same transaction. Nothing a message carries fails it — not even a row whose refusal rolls the whole transaction back (a trigger’s RAISE(ROLLBACK)), nor one a foreign key checked at commit refuses — because the row it points at is not held, or because rows point at a value of it the message changed or let go of: a constraint that fails the commit is the replica finding such a row, never a failure of its storage. The replica applies the message again, row by row, finds the rows at fault — by the database’s own check of its keys — and applies it once more with them rejected before they are tried; and should the database still refuse it for what no row of it explains, it applies it with every row of it rejected, so that the rest of what it says stands. Anything else the database answers of a row — a trigger’s own error included — refuses that row. Only a failure of the replica’s own storage (the database is unavailable, the disk is full, the file is damaged) fails the message: then the replica must not acknowledge it, and it will be delivered again.

4.2 Snapshots

When the server provisions a replica from scratch, it sends everything the replica should hold in Down messages that carry a snapshot_id, then an EndSnapshot naming the tables the snapshot covered. A snapshot of nothing is its EndSnapshot alone.

The snapshot_id is a UUID, as the epoch is; a message naming anything else is malformed (4.1).

• A snapshot begins with its first message, the one whose previous is 0 — the EndSnapshot itself, for a snapshot of nothing. It supersedes the snapshot in progress, if any: the replica forgets what that one named. The messages after it follow it, as any do (4.1).
• While applying a Down of the snapshot in progress, the replica must remember the key of every row it carried, per table — the ones it refused as well: a refused row keeps whatever the replica had of it.
• On EndSnapshot, the replica must let go of every row of the named tables that the snapshot did not name (delete, or retire as in 4.3), children before parents: in reverse of the order the tables are listed. A row it refused was named, so it is never swept. The rejections it recorded for those tables before the snapshot began are forgotten — the snapshot sent again what it should hold — and the ones it recorded since stand. Then the snapshot is no longer in progress.
• A part of a snapshot’s Down may carry recorded: keys of a setting the server records the replica holding, which no other field of the part concerns. Of those, the replica must note each row it does not hold as its own delete (section 7), so the server lets go of it as it files the delete; one it holds it has noted already, as the snapshot began.
• The messages of a snapshot follow one another, and nothing else comes between them: a message that follows the last applied and names a snapshot other than the one in progress — one that ended, or was superseded — or names none while one is in progress, is one no server sends: malformed (4.0).
• EndSnapshot carries sequence, epoch, nonce, previous, previous_nonce and mac, and follows the same rules as Down (4.1). It needs no count of what the snapshot sent: its messages are chained, and one lost is a gap.

4.3 Retired rows

A replica that must let go of a row that one of its own rows still points at (a receipt that names an order) keeps the row and marks it retired: a column the replica owns, set to when it was retired. Retired rows are out of everything the replica offers its users, but stay for what points at them, and only that long: once nothing that stays points at them — a live row or a retired one, of any table — the replica lets go of them, children first. It looks after every message it applies, once that message is applied and apart from it, so a row it cannot let go of never holds a message back; and it says nothing of one it cannot. A retired row holds none of its unique values against a live one. A retired row the server sends again becomes live.

A row of a table written on both sides that the replica edited, and whose edit the server has not filed yet (section 7), is retired too when it is let go of, rather than deleted: the edit still goes up, as the row is. Once the server files it (4.4), the replica lets go of the row. Where the table has nothing to retire it by, the replica refuses to let it go (section 6), and the server sends the delete again later.

4.4 Filed

The server answers each upload (section 7) once it has filed it. On Filed, the replica must forget the notes those uploads carried — the ones whose latest send carried that number, and no other: a note sent again since is forgotten only on the answer to that send, and a row written since the send goes up again. A number none of its sends carries — meant for another replica, or for this one before it was restored — names nothing. The answer may arrive before the replica recorded that the transport took the upload: it counts all the same.

4.5 Wishes

A setting is a table of the replica’s own whose rows go up as a record’s with a version do (section 7) — a value the replica holds and acts on: a module’s configuration, a screen’s brightness — some columns of which the server may wish other values. Which columns a wish sets, and which column is the version, is the table’s declaration, agreed between the server and the replica as the table’s encoding is (section 3). A replica that holds settings holds the server’s wishes in a table of the engine’s, sync_wishes: a table written on both sides (4.1), its rows encoded as the Wish message, its key the wish’s id. A replica that holds none needs none of this section: it rejects the rows of sync_wishes as those of any table it does not know.

A wish is one row of a setting — table, key — and what the server wants it to hold: wanted, the row encoded as table’s own, of which the replica takes the columns a wish sets, and nothing else. One row has one wish at most; a new one replaces it, under a new version, a number the server draws at random: only whether an answer is of the wish the replica holds matters, never which version is the later. The server owns id to note; the replica owns the answer — answered, the version it answers, outcome, reason and answered_at — which it must not take from a message: a wish it takes that it did not hold is one it has not answered.

Once the message that brought a wish is committed — never inside the transaction that applies a message, since carrying one out may take time and reach outside the database — and while no snapshot is in progress (4.2: its end may let go of a wish), the replica carries out every wish it holds, not retired (4.3), that it has not answered, or answered waiting. It takes them one at a time, each read again just before, so a wish replaced or withdrawn meanwhile is carried out as it is then, or not at all:

1. It reads key as table’s key (section 3) and wanted as table’s row. A table it holds as no setting, or a row it cannot read — a table or an encoding this build does not know — it answers waiting, and carries out with a build that can. A key that is not the table’s, a row whose key is not key, or a row of another replica than itself (by the setting’s origin) it answers refused.
2. A row it does not hold it answers waiting (it holds no such row yet), and carries the wish out once it does: it never writes a row a wish names.
3. With if_unchanged, the wish is based on the version wanted carries: if the replica holds the row at another version, it answers superseded, and does not carry it out — the replica’s value stands. Versions are compared as text: an integer decimal; an instant in UTC, RFC 3339, truncated toward the past to the microsecond, always six fractional digits and Z (2026-10-12T10:00:00.123456Z for .1234567).
4. Otherwise it carries it out, however the setting takes it, and answers applied — then, in one transaction, it writes the wished columns onto the row as it holds it then, its version moved past what it was (an integer one more; an instant now, truncated so, or a microsecond past the old one when the clock is behind), and records the answer: a version that cannot move (an integer at its largest), or a write its database refuses, is answered refused, with why, and the row left as it was; refused, with why — the row as it was — final for that wish; or waiting, with why, and carries it out again later.

An answer is recorded only while the wish the replica holds is still the one it answers (the same version, not retired); a wish replaced meanwhile is carried out next, and the row keeps what carrying out the last one wrote. A waiting answer is recorded when it is new — another reason — not every time it waits again. The answer and the row it wrote are noted like any write of the replica’s (section 7): the answer as an edit of the columns the replica owns of a two-way row.

A replica that stops after carrying a wish out and before it records the answer carries it out again: carrying out the same wish twice must leave the setting as once does.

5. Acknowledging

There are two acknowledgements, and a replica gives both.

5.1 To the transport

After the transaction that applied a message commits, the replica acknowledges the message to the transport, which then delivers the next one. Never before the commit: a replica that acknowledged and then failed to commit would never be sent that message again.

5.2 To the server: what was applied

The server counts a row as held by a replica only once the replica says it applied the message that carried it. Until then, the row is “on its way”. That is what makes the server’s progress figures true.

The replica must tell the server the (epoch, sequence, nonce, mac) of the last message it applied, as that message carried them, in an Up on its first lane (section 7) that carries them in epoch, sequence, nonce and mac. It may do so after every message, or once for several (a replica that applied messages 5, 6 and 7 reports 7). It should do so within seconds of applying, and again whenever it applied something new, or was delivered a message at or below the last it applied. sequence 0 says it has applied nothing yet.

A replica with a gap (4.1) must say so: gap, the sequence of the newest message it could not follow, on every Up of its first lane, with its last applied, until a snapshot begins; and it must send one at once whenever that sequence grows. The server answers with a snapshot (A.2).

6. Rejections

A row the replica could not take, or could not let go of (4.1), is a rejection: the table, the key — required, in the form of section 3: the one the row came beside, keys[i], or its delete’s — why, and the (epoch, sequence) of the message that carried the row, or its delete — for a row the end of a snapshot could not let go of, the EndSnapshot’s; for a row held retired for the replica’s edits (4.3) that it could not let go of once they were filed, the message that let go of it. A row whose key cannot be told is logged and not reported. A row held back (4.1) is a rejection marked held_back: nothing is at fault, so the server raises no alarm over it; the rules below hold for it all the same.

• The replica must record a rejection in the same transaction that left the row out, so it cannot be lost.
• It must report each rejection to the server in an Up (rejected), no later than the Up that says it applied the message that refused it (5.2): the server counts the rows a message carried as held once it hears the message was applied.
• It must keep the rejection until it takes the row (a later message carries the row and it applies) or lets go of it (a delete, or a snapshot that does not name it): a digest (section 8) leaves rejected rows out, as the server leaves out the rows recorded refused. It reports only the rejections of the epoch it follows; a snapshot of a new epoch clears the others — those of the tables it names as it ends, and those of a table it does not hold from the server, which no snapshot names. A rejection of a row it holds no copy of — of a table it does not hold from the server, or a row it never took — leaves nothing out of a digest: it may forget that one once reported, and the engine does, so what it keeps of rejections is bounded by what it holds.
• It keeps what the server said within bounds: a reason may be cut (to 512 bytes), and rows of a table it does not know, named or keyed longer than that, are logged rather than kept — no server sends such a table. So any rejection fits a message up — the backlog (section 7) waiting for the next one when it would leave the first rejection no room — and never holds back the confirmation. Rejections count against a message’s rows as the rows it carries do.

7. Sending rows up

For a table the replica writes and the server mirrors:

1. Note every insert, update and delete in the writing transaction: the table and the key — on an update that changes the key, the old key too, which is gone. (A trigger is the only way no write path can forget.) What the replica writes while applying what the server sent is not noted: it is the server’s already. A row of a table written on both sides begins on the server; the replica edits only its own columns of it, and only those edits go up: it must refuse an insert of one, an update of a column it does not own, and a delete of one but a row keyed by itself, in a table keyed by the replica. A record without a version is history, written once. A setting is noted as a record with a version is; and in the transaction that begins a snapshot (4.2) the replica must note again every row of its settings, and every answer it gave to a wish it holds (4.5): the server’s record of them may be behind — it, or the replica, restored from an older copy — and is set straight by what the replica says it holds now (A.3). Such a note is re-stated until the upload that carries it is filed, a later write of the row notwithstanding: its row goes up in a part of its own, restated set.
2. Upload noted rows, as they are at upload time, in an Up:
   • each row in tables[].upserts, encoded as its table’s message;
   • a row that is gone by now, as its key in tables[].deletes, but only for a table keyed by the replica that wrote it: history a replica pruned is not the server’s to delete;
   • one key per row in tables[].keys;
   • tables[].restated set on a part of re-stated rows only: a table’s rows go in two parts when some are re-stated and some are not, and a delete always goes in a part without it;
   • upserts parents first, in the order the tables point at one another, then deletes children first;
   • in uploads, the number of this upload: a fresh one on every send, resends included, drawn at random from 1 to 2^63 − 1, so an answer meant for another replica, or for this replica before it was restored from an older copy, never names its notes;
   • within the transport’s size limit: what does not fit waits for the next upload. A row that can never fit is set aside — kept, counted in the backlog, never sent until the replica runs again with a build that may send it. An Up the transport refuses for good is sent again with half its rows and rejections, and so on, until what it refuses is alone: that row is set aside, that rejection taken as told (it must not hold back the confirmation of what was applied), and nothing beside them — but only if the transport took some other message of that stream meanwhile: a transport that refuses every message says nothing of any one row, and the replica sets nothing aside, raises the alarm and asks again after a backoff.
3. Wait for Filed (4.4) naming that upload, then forget its notes. An upload not answered after a while (minutes) should be sent again; the server files the same rows idempotently.

A replica may say what it wrote and the server has not filed yet, in Up.backlog: per table, rows not sent yet (unsent), rows sent and not yet Filed (unfiled), and rows set aside (set_aside). Say it when it changes — at once when it empties — and not on every write; say it again now and then while it is not empty, and to a server whose epoch changed, either of which may have lost it. An Up without backlog leaves the last one standing (A.3). A replica that never says leaves the server with nothing to show, and nothing breaks.

A replica may send its writes on several lanes — streams of Ups the transport keeps apart, each carrying the writes of some of its tables, so a table that writes often never waits behind the rest. Every table that goes up is on exactly one lane. Rejections, the last applied message, digests and the backlog go on the first. The transport keeps each lane in order (section 2): the backlog an earlier Up carried must not arrive after a later one’s.

One Up may carry several of these at once: rows, rejections, digests, the backlog, and the last applied message.

8. Digests

From time to time (hourly is reasonable), a replica should report what it holds of every table the server sends it, in Up.digests, together with the (epoch, sequence, nonce, mac) of the last message it applied (5.2). It must leave out a table it wrote rows of that the server has not filed (Filed, 4.4); the reason is given at the end of this section.

• table: the table;
• count: how many rows it holds, leaving out rows it retired (4.3) and rows it rejected (6);
• fingerprint: FNV-1a, 64 bits, over the keys of those rows as text (section 3), sorted (byte order), each followed by a newline (\n):

hash = 0xcbf29ce484222325
for each key, sorted:
    for each byte of key + "\n":
        hash = hash XOR byte
        hash = hash × 0x100000001b3   (mod 2^64)

Test vectors: no keys → 0xcbf29ce484222325; the single key a → 0x089bdc07b544e7b2.

It should report its settings and its answers as well, which the server records as the replica uploads them: a server restored from an older copy records what the replica said before, and no message it sends shows it. Their lines are fingerprinted the same way, and say more than the key:

• a setting: each row it holds, as its key, a tab (\t), and its version as text (4.5);
• sync_wishes: each wish it holds, left out as above, as its key, a tab, the version it answered, a tab, and the outcome — each empty for a wish it has not answered.

Test vectors: the setting line a\t1 alone → 0x6f375e81eb6a6700; the unanswered wish w (w\t\t) → 0xba0c6af6f14e26d2; w applied at version 7 (w\t7\tapplied) → 0x2e00e679fcb6ffe0.

A replica must leave out of its digests every table of which it wrote rows the server has not filed yet (section 7): what the server records of it is about to change, and a comparison now would find them apart for nothing: the server records a row the replica deleted as held until it files the delete, and then records it held no more only where the replica had applied every message that carries the row (Appendix A.3). That holds from when it reads them to when it sends them: a table written in between — a write that may go up in the very message — goes without its digest too.

9. A session, start to finish

server                                         replica
  │  (every arrow down is a ToReplica carrying the message named)
  │  Down {snapshot s1, seq 101, previous 0, products…}
  │                                          ──▶  begins s1, applies, remembers its keys
  │  Down {snapshot s1, seq 102, previous 101, orders…}
  │                                          ──▶  applies
  │  EndSnapshot {s1, seq 103, previous 102, [products,orders]}
  │                                          ──▶  sweeps what s1 did not name
  │  ◀── Up {epoch E, sequence 103}               "applied up to 103"
  │      progress: holds 3,200 of 3,200
  │
  │  (a product is renamed)
  │  Down {seq 104, previous 103, products: 1 row}
  │                                          ──▶  applies
  │  ◀── Up {epoch E, sequence 104}
  │
  │  (a row the replica refuses: a check it has, the server lacks)
  │  Down {seq 105, previous 104, orders: 2 rows}
  │                                          ──▶  applies one, rejects one
  │  ◀── Up {sequence 105, rejected [orders 9f…: "check failed"]}
  │      the row is set aside for this replica
  │
  │                                               (writes two receipts)
  │  ◀── Up {tables [receipts: 2 rows], uploads [57]}
  │      filed
  │  Filed {uploads [57]}                    ──▶  forgets the notes upload 57 carried
  │
  │  ◀── Up {digests […], epoch E, sequence 105}   (hourly)
  │      matches what the server records: nothing to do

10. Conformance checklist

The engine’s repository holds test vectors, each line a name, what the other side makes of the message, and the message in hex; each file’s header says how to read it.

• tests/golden/to_replica.txt: what the server sends one replica, step by step — a provisioning of several messages, the first delivered again, updates, a row the replica’s own check refuses, a delete of a row a row of its own points at, a snapshot cut short and the one that supersedes it, a message lost on the way, the gap after it and what comes during the gap, the answer to an upload, a version older than the one held, a field the replica does not know, a new epoch, a wish of a setting the replica holds — carried through a provisioning — and its withdrawal. Each message says what the replica makes of it (applied, ended, filed, already, gap, or lost on the way), and each step is followed by a holds line: what the replica then holds, as JSON.
• tests/golden/malformed.txt: messages no server sends — and the few that set the scene between them — each with what a replica makes of it, malformed for most.
• tests/golden/to_server.txt: what a replica sends up after the steps of to_replica.txt before its upload is answered — its rows, its rejections and what it applied; its digests; the same with a seal the server never made — each with what the server makes of it.
• tests/golden/to_server_verdicts.txt: uploads a replica sends after a provisioning, a change and a take-back it applied — a refused delete with what it applied, an edit of a two-way row, a delete of a row of its own, a refused sweep, a row of its own held back, an answer to a wish and to one withdrawn, a setting and an older version of it, a gap, what it applied said again lower, of another epoch, and past anything sent — each with what the server makes of it.

A replica implementation is complete when:

• it reads every message as a ToReplica, leaves one of a later protocol unacknowledged, and sets protocol on every Up;
• it applies a Down in one transaction, tables in the order given;
• it ignores a message at or below the last applied in the same epoch, and says again what it applied; acknowledges without applying a malformed message (4.0, 4.1);
• it applies a message only when it follows the last applied — sequence and nonce — or begins a snapshot; one that follows another is a gap, after which it applies nothing until a snapshot begins, and it says the newest message it could not follow on its first lane, again whenever that grows (4.1, 5.2);
• it overwrites a row with what the server sends, whatever its version;
• it never overwrites the columns it owns;
• it retires a row it cannot delete, and revives a retired row sent again;
• it applies a table’s deletes before its upserts, lets go of the deletes together, and tries a refused upsert again once the rest are in (4.1);
• it begins a snapshot with the message that follows nothing — an EndSnapshot included — and sweeps on its end, in reverse table order;
• it leaves as it was and records, in the same transaction, a row it cannot take or let go of;
• it reports rejections, each with the message that refused it, before it says it applied that message, and keeps them until it takes the row or lets it go — a row of its own whose delete is not filed among them, marked held_back (4.1);
• it acknowledges the transport only after committing;
• it reports the last applied (epoch, sequence, nonce, mac) to the server, as the message carried them;
• it notes its own writes in the writing transaction, uploads them as they are within the transport’s limit, gives every send a fresh random number, and forgets on Filed only the notes that send carried, as it carried them;
• it reports digests with the fingerprint above (optional, recommended);
• it says its backlog when it changes (optional, recommended);
• if it holds settings: it holds their wishes as a two-way table, takes no answer from the server, carries each out after the message that brought it is committed — superseded, unread or applied as 4.5 says — and records the answer, with the row it wrote, in one transaction; notes every setting and every answer again when a snapshot begins, and sends them re-stated until filed (section 7); deletes each row a snapshot records it holding that it does not (4.2); its digests, if it reports them, include its settings and its answers (section 8).

Appendix A. What the server does with it

Nothing here is a duty of a replica. It is what the server does with what a replica sends, written down so that a replica’s author knows why the rules above are what they are, and what to expect when a replica breaks one. The sections it explains are named at each part.

A.1 What the server guarantees in return

• Every change committed on the server reaches every replica that should hold the row, as the row is at sending time.
• A row is sent with every row it points at that the replica lacks, and is taken back only once nothing the replica keeps points at it.
• Every message carries a number and a random nonce of its own, and the numbers a replica is sent rise within an epoch in the order it is sent them; each names the one before it, number and nonce, so a replica can tell it missed one — even when a restored server numbers a message as one it applied. What a message carries is recorded before it is sent, and a message that the transport did not take is sent again, under the same number and nonce, before anything after it.
• A replica that leaves the last messages it was sent unconfirmed, while nothing newer goes to it, is provisioned again after a while: the transport may have lost them. One that says it missed a message is provisioned again at once.
• A server whose database was restored from an older copy finds out from the first replica that says it applied a message the server sealed and that copy has no record of sending it, and provisions every replica again, under a new epoch. A replica restored from an older copy finds out from the next message it is sent, and says so; on a server with nothing to send it, the server finds out from what it says it applied, which is less than it said before.
• A row the server cannot send to a replica, or that a replica rejected, is set aside and sent again after a fix: it is never silently dropped.
• An upload is answered with Filed only after the transaction that filed it committed.

A.2 What the server makes of “what I applied” (section 5.2)

The server checks the seal: one it did not make — a replica that lies, or a bug — is ignored, and resets nothing. It ignores a report that does not parse, and one of another epoch — unless it last heard the replica apply a message of its own epoch: a replica only goes forward, so it follows an epoch this database never had, which was restored from a copy older than that epoch began. The server provisions that replica again, unless a snapshot written for it since is on its way. A sealed one it judges against what it recorded sending the replica:

• a message past the last it recorded sending it — or, having recorded none since it last provisioned it, past every number it drew — or one at the sequence of the last it sent it, or of the last it heard applied, with another nonce: the server sealed it, and has no record of it. Its own database was restored from a copy older than what the replica applied. It raises an alarm — whoever runs it should know the database went back — takes a new epoch, with a new secret, drops what it was about to send, and provisions every replica again;
• a message older than the last it heard the replica applied: the replica was restored from an older copy, and holds less than the server records. The server provisions it again, unless a snapshot written for it since is on its way;
• otherwise the server counts the rows of every message up to it as held.

A replica with a gap (4.1) is provisioned again, unless a snapshot that began after that message is on its way: a snapshot that began before it is one the replica lost the beginning of, or applied before it was restored from an older copy. A gap past every message the server sent the replica is none it could have missed: malformed, and left out.

Neither a gap nor a message older than the last applied carries a seal the server can check, so what a replica says of itself provisions it again at most once a minute, and twice as long between each time while it keeps saying so, up to 32 minutes; one quiet for 64 minutes starts over. One that says it sooner is provisioned again once it may: the server notes it for then.

A.3 How the server files an upload (section 7)

The server files an upload under the replica it came from (the transport says which), whatever the rows themselves claim: a row of a table keyed by its replica that names another replica there is malformed, and left out. A row of a table written on both sides is an edit of the columns the replica owns, filed only as an update of a row the server has and records the replica holding. One it does not record the replica holding while it provisions the replica again — the provisioning pending, or written and not confirmed — is kept, and filed once it does; otherwise, one of a row the replica no longer holds — the server deleted it, or is taking it back — is left out: the server’s wins. One of a row it never sent the replica is malformed, and left out. A row the server cannot file, or a delete its database refuses, is kept on the server, not lost, and filed later; so are an upload’s rows whose filing fails as a whole for a reason a resend cannot cure. Either still counts as filed for Filed. A replica’s delete of a row of its own is a fact whether the server’s database lets the row go or refuses to: the server records the replica holding it no more, and sends it again a row it kept — but for a row that a message the replica had not applied when it sent the delete carries (the last it says it applied, or else the last the server heard it apply). That one the server still records held: the replica takes that message’s version if it hears the delete filed first, and the server’s own delete of the row, if it let it go, follows; otherwise the replica holds it back (4.1), and its rejection, which comes no later than it says it applied the message, records the row refused (A.4). A delete filed later, from what the server kept, leaves the row held the same way: a delete of a row a replica does not hold changes nothing there. The server keeps, of each row, the newest upload of it: a later upload of the row replaces what it kept, or drops it once filed, so what is filed later never rolls the row back. A record with a version is filed only over an older version of it, and so is a setting — a value the replica holds and acts on — but one in a part restated (section 7), which is filed whatever its version: it is what the replica holds now, even when a replica restored from an older copy says an older version; a delete is filed whether its part is restated or not. The server keeps a row only with its key in keys: one without is malformed, and left out. What the upload says of the replica — rejections, the last applied message, a gap, digests, the backlog — is filed apart from its rows, so neither holds back the other. Deletes of a table not keyed by the replica, and keys that are not the table’s, are malformed and left out. An upload beyond the server’s limits — more bytes than a message takes, or more rows and rejections together than one carries: a rejection counts as a row — is malformed: dropped, unanswered, so it waits on the replica until a server that takes it runs. A replica’s limits are the server’s, or less. So is an upload from a sender that is no replica of the server’s.

The server keeps the last backlog it was told, with the time, and shows the replica as catching up while it is not empty.

A.4 How the server records a rejection (section 6)

On a rejection of the last message that carried the row, the server records the row refused: still held for every decision that takes rows back — a delete still goes when the row leaves the replica’s scope, and a refused delete is sent again — and not held for progress, or for digests. The row is set aside for that replica, and sent again, or its delete, when the relay restarts or on a timer, which is after a deploy on either side has had the chance to fix the cause; it is taken once the replica applies a later message that carries it. A rejection of a delete re-records the row the server took back. A row a replica held back (held_back) is recorded the same way and raises no alarm — nothing is at fault, and a fix is awaited for nothing: the retry sends it again, should the server keep the row. A rejection of a message other than the last one that carried the row, or of another epoch, names a version that no longer matters, or none it was sent, and is left out. The key is read as its table’s key (section 3) whatever its spelling. A rejection of anything else — a table the server does not send, no key or one that is not the table’s, a message never sent the replica — is malformed and left out. Nobody repairs the row by hand.

A.5 How the server compares digests (section 8)

The server compares a digest only when the replica has applied every message recorded for it (the same epoch and last sequence, sealed by the server), and against what it recorded as of that message: when it has sent the replica more meanwhile, it waits for the next digests. It compares a setting’s against the rows of it it holds of that replica, and those of sync_wishes against the answers it keeps — neither while it keeps rows of that table the replica uploaded and it could not file yet, as the replica leaves out a table it wrote rows of that are not filed. If they differ, it provisions the replica again from scratch (4.2): the replica re-states its settings and its answers as the snapshot begins (section 7), and deletes those the snapshot records it holding that it does not. Digests that differ again within a day of the provisioning they caused are a bug no provisioning cures: the server logs them and does not act on them. A digest of a table the server neither sends the replica nor holds as a setting — the replica’s records, a later build’s table, one it no longer mirrors — is skipped. Two digests of one table are malformed: none of that Up’s digests is compared. The server compares a replica’s digests at most once every ten minutes: those that come sooner are not compared. A replica that skips digests loses only this safety net.

A.6 How the server keeps wishes (section 4.5)

The server writes a wish of a setting row when its host asks, in the host’s transaction, under a version it draws at random — so a server restored from an older copy wishes under other numbers than it did — and sends it to the replica the row belongs to (the setting’s origin), as it sends any row: one withdrawn is deleted, and taken back as a row is. It never sends an answer down. It files an answer as the edit of a two-way row it is (A.3): only while it records the replica holding the wish — so an answer to a wish withdrawn is left out — and keeps it beside the wish; a wish stands as its answer says only while that answer is of its version: pending until then. A final answer — applied, refused, superseded — stands against a later upload of another answer to the same version, which the transport may have kept back, but for one re-stated (section 7). It keeps the row wished apart from any encoding, and encodes it as it sends the wish: the setting row it holds, or the one wished when it holds none, with the wished columns and the version the wish is based on laid over it. Table names beginning sync_ are the engine’s: no host table may take one.

Questions

Why one backend per build?

The server plans in PostgreSQL and a replica applies in SQLite: the two halves share the row types and the codecs, and almost nothing else. Building the crate for one side keeps every query in it checked against the database it runs on. Your server enables postgres, your replica sqlite, each in a workspace of its own, and the crate of rows both share picks neither (Set up a project).

Can two sides write the same row?

A table goes down, goes up, or splits its columns between the two (mirror = "both", #[row(replica_owned)]). Two writers on the same column of the same row is not a case the engine resolves. That is deliberate: in the systems it was built for, every fact has one owner, and merging two owners’ edits is a business decision, not a sync one.

When the server must change a value a replica owns — its configuration, a device’s parameters — that is a setting, and the server wishes the replica another value rather than writing it. The replica stays the one writer: it carries the wish out, refuses it, or finds it changed first, and both sides end showing what it decided — a digital twin, desired against reported (Settings and wishes).

What if a replica is offline for a month?

Nothing special. What it should hold is compared with what it holds when it is back, and the difference is sent, as it is then — not a month of events. What the replica wrote meanwhile goes up as it is then.

What if my transport loses a message?

It must not: the protocol asks a transport to lose nothing it took. If one does anyway, the engine heals, late rather than never. The server counts nothing as held that the replica did not confirm, so the rows stay on their way; a replica that leaves its last messages unconfirmed while nothing newer goes to it is provisioned again after unconfirmed_after; an upload the server never answered is sent again; and the hourly fingerprint catches whatever else went missing, and provisions the replica again.

Does the replica have to be written in Rust?

No. The replica protocol is what a replica must do, and the messages are a .proto file. The engine’s own replica is one implementation of it.

What does “nobody repairs data by hand” mean?

Every failure is either the infrastructure (it waits, and retries) or a bug or data no declaration allows (it is narrowed down to the row it is about, set aside, and retried after a deploy). Neither one ever needs someone to edit rows. Every number the engine reports means wait or deploy a fix.

How big can it get?

The planner works on sets in SQL, not row by row: giving a list of 10,000 customers to a register is one row written, and one plan. The engine’s benchmarks measure the operations that move many rows at once, and the dashboard’s reads over 200 replicas holding 3 million rows; CONTRIBUTING.md says how to run them on your own hardware.

The API

A build of magic_sync is one side, and so is its API reference: each lists what a host of that side can name, built from the same sources as this book.

• The server’s — feature postgres: Server, the planner’s Engine and its status reads, the transport.
• A replica’s — feature sqlite: Replica, replica::install and check, the uplink.

What both sides share — the rows, the codecs, the derive, the errors — is in both. docs.rs has the server’s, for every published version.