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.