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.