Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

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.