Lines
60.61 %
Functions
28.57 %
Branches
100 %
//! Brings every tenant database to the current schema.
//!
//! Runs as its own process — a k8s Job, not a serving container. See
//! [`server::fleet`] for why this must never live inside `boot()`.
//! Exit codes are the Job's contract: `0` means every tenant was verified at the
//! current schema, `1` means the run left something undone — a tenant
//! quarantined, one still held by another worker, or a verdict that could not be
//! written — and `2` means the run could not be performed at all. Only `0` may
//! gate a rollout: the other two all mean some customer's database is in a state
//! nobody has confirmed.
use server::bootstrap::ADMIN_USER_ID;
use server::db::DBError;
use server::fleet::migrate_fleet;
use sqlx::types::Uuid;
use std::process::ExitCode;
#[cfg(test)]
mod tests;
/// Every tenant verified at the current schema.
const HEALTHY: u8 = 0;
/// The run finished but left something unconfirmed.
const DEGRADED: u8 = 1;
/// The run could not be performed.
const UNUSABLE: u8 = 2;
/// The Job's contract with Flux, kept separate from the logging so it can be
/// tested: a wrong code here is the difference between a rollout that stops and
/// one that proceeds over a database nobody has confirmed.
fn verdict(outcome: &Result<server::fleet::FleetReport, DBError>) -> u8 {
match outcome {
Ok(report) if report.is_healthy() => HEALTHY,
Ok(_) => DEGRADED,
Err(_) => UNUSABLE,
}
fn report_outcome(outcome: &Result<server::fleet::FleetReport, DBError>) {
Ok(report) if report.is_healthy() => {
log::info!("fleet is at the current schema: {}", report.summary());
Ok(report) => {
log::error!("fleet migration degraded: {}", report.summary());
for (id, reason) in &report.quarantined {
log::error!(" {id} quarantined: {reason}");
for id in &report.unresolved {
log::error!(" {id} was held by another worker and never verified here");
for (id, reason) in &report.unpublished {
log::error!(" {id} has a verdict that could not be recorded: {reason}");
Err(e) => log::error!("fleet migration could not run: {e}"),
#[tokio::main]
async fn main() -> ExitCode {
env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
let outcome = run().await;
report_outcome(&outcome);
ExitCode::from(verdict(&outcome))
async fn run() -> Result<server::fleet::FleetReport, DBError> {
sqlx::any::install_default_drivers();
let admin = ADMIN_USER_ID
.parse::<Uuid>()
.map_err(|e| DBError::FleetWorkerLost(format!("admin id is not a uuid: {e}")))?;
migrate_fleet(admin).await