src/code_room/backfill.rs
//! A member's history, asked for from a commit — walked in the background.
//!
//! A member's first consume starts at its tip (decision D2); `--from <sha|date>`
//! asks for the commits before it. Walking them is hours of `cargo check` for
//! a long history, so it is not done on the call that asks:
//!
//! ```text
//! request the door: resolve `from` on the member's branch, write ONE row
//! `backfill_runs { requested, from_sha, base_seq, total }` in the lake
//! job one commit per turn, each turn its own consume (its own hold of
//! the consumer and, on a server, of the workspace's gate):
//! first turn replay from `from_sha` (the member's ordinals keep
//! counting; nothing is deleted)
//! next turns the commit after the offset
//! the offset is the member's head → row `done`
//! ```
//!
//! **The lake is the whole state.** "Open" = the member's last backfill row is
//! `requested`; "started" = its offset's ordinal is past the row's `base_seq`;
//! "where" = its offset. A process that dies in the middle leaves exactly that,
//! and the next one reads it ([`CodeRoom::backfills`]) and goes on with the
//! commit after the offset — there is no queue file and no progress file.
//!
//! While a backfill is open the member belongs to its job: every other run
//! leaves it out (`MemberStatus::Backfilling`), so a sweep never takes a batch
//! of its commits in one hold. A commit pushed to the member meanwhile is the
//! new end of the same walk: it is consumed after the commits before it, in
//! first-parent order, by the same job.
//!
//! What a reader sees meanwhile: the member at the commit its offset is on —
//! a commit of the PAST until the walk is back at the tip (that is what a
//! replay is; the ordinals of the walk are the history asked for).
use std::time::Instant;
use mimir_ingest::{BackfillRow, EventLog, Offset, git, schema};
use serde_json::{Value, json};
use super::consumer_loop::{Member, short};
use super::{CodeRoom, Refusal, State};
/// One member's backfill, as the lake and the member's clone say it is.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Backfill {
pub member: String,
/// The first commit of the history asked for.
pub from_sha: String,
/// `requested` (open) · `done`.
pub status: String,
/// The first replayed commit landed.
pub started: bool,
/// First-parent commits from `from_sha` to the member's tip NOW (it grows
/// when the member is pushed to); the row's own count when the clone
/// cannot be read.
pub total: usize,
/// Of those, consumed.
pub done: usize,
/// The commit the member is on (its offset), when it has one.
pub at: Option<Offset>,
}
impl Backfill {
pub fn is_open(&self) -> bool {
self.status == schema::STATUS_REQUESTED
}
/// `znippy-zoomies: backfill from 1f0c2a9b3d — 7 of 20 commit(s), at seq 9 @ 8a1b2c3d4e`
pub fn line(&self) -> String {
let at = self.at.as_ref().map_or("not consumed yet".to_string(), |o| format!("at seq {} @ {}", o.seq, short(&o.sha)));
match (self.is_open(), self.started) {
(true, true) => format!("{}: backfill from {} — {} of {} commit(s), {at}", self.member, short(&self.from_sha), self.done, self.total),
(true, false) => format!("{}: backfill from {} — queued, {} commit(s) to walk, {at}", self.member, short(&self.from_sha), self.total),
(false, _) => format!("{}: backfill from {} — done ({} commit(s))", self.member, short(&self.from_sha), self.total),
}
}
pub fn to_json(&self) -> Value {
json!({
"member": self.member,
"from": self.from_sha,
"status": if self.is_open() && self.started { "running" } else { self.status.as_str() },
"commits_done": self.done,
"commits_total": self.total,
"at": self.at.as_ref().map(|o| json!({ "seq": o.seq, "sha": o.sha })),
})
}
}
/// What the job does next for a member ([`CodeRoom::backfill_next`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Next {
/// Consume ONE commit of the member: with `from`, the first one (a replay
/// from that commit); without, the one after its offset.
Step { from: Option<String>, backfill: Backfill },
/// The walk is at the member's head: the `done` row is written.
Finished(Backfill),
/// No backfill of the member is open.
Nothing,
}
impl CodeRoom {
fn backfill_of(&self, log: &EventLog, member: &Member, row: &BackfillRow) -> Backfill {
let at = log.block_on(log.offset(&member.name)).ok().flatten();
let started = at.as_ref().is_some_and(|o| o.seq > row.base_seq);
let clone = self.layout.clone_of(&member.name);
// the member's stream is the branch its offset is on
let count = |since: Option<&str>| at.as_ref().and_then(|o| git::first_parent_since(&clone, &o.branch, since).ok()).map(|c| c.len());
let total = git::offset_before(&clone, &row.from_sha).ok().and_then(|before| count(before.as_deref())).unwrap_or(row.total.max(0) as usize);
let left = if !row.is_open() {
0
} else if started {
at.as_ref().and_then(|o| count(Some(&o.sha))).unwrap_or(0).min(total)
} else {
total
};
Backfill { member: member.name.clone(), from_sha: row.from_sha.clone(), status: row.status.clone(), started, total, done: total - left, at }
}
/// The last backfill of every member that ever had one: the open ones
/// with where they are, the finished ones as `done`. Reads the lake and
/// the consumer's clones; never waits for the consumer.
pub fn backfills(&self, members: &[Member]) -> Result<Vec<Backfill>, Refusal> {
let log = self.log()?;
let last = log
.block_on(log.backfill_last())
.map_err(|e| Refusal::new(State::Internal, format!("code room of `{}`: read the backfills: {e:#}", self.workspace)))?;
Ok(members.iter().filter_map(|m| last.get(&m.name).map(|row| self.backfill_of(&log, m, row))).collect())
}
/// **Ask for `member`'s history from `from`** (a commit, ≥ 7 hex digits, or
/// a date `YYYY-MM-DD`). Nothing is consumed here: the commit is resolved
/// on the member's branch and ONE `requested` row is written — the job
/// that walks it ([`CodeRoom::backfill_next`], one commit per turn) is the
/// caller's to start, and to start again after a restart.
///
/// Refused, by state: a `from` that names no first-parent commit of the
/// branch, a member with no clone or no branch (`invalid_argument`), and a
/// member whose backfill is already open — one at a time per member; the
/// answer says where that one is.
///
/// SYNC; waits for a run in flight (the row goes through the consumer's
/// seat). On a server, call it inside the workspace's gate: it syncs the
/// consumer's clone of the member.
pub fn request_backfill(&self, member: &Member, from: &str) -> Result<Backfill, Refusal> {
let invalid = |why: String| Refusal::new(State::InvalidArgument, format!("backfill of `{}` in `{}`: {why}", member.name, self.workspace));
if !member.has_source() {
return Err(invalid(format!("no clone at {} yet — fetch the workspace first", member.source.display())));
}
self.with_consumer(|log, consumer| {
let internal = |what: &str, e: anyhow::Error| Refusal::new(State::Internal, format!("backfill of `{}` in `{}`: {what}: {e:#}", member.name, self.workspace));
let last = log.block_on(log.backfills(Some(&member.name))).map_err(|e| internal("read the backfills", e))?;
if let Some(open) = last.last().filter(|r| r.is_open()) {
return Err(invalid(format!("one is already open — {}", self.backfill_of(log, member, open).line())));
}
let offset = log.block_on(log.offset(&member.name)).map_err(|e| internal("read the offset", e))?;
let branch = match (&offset, member.tracked_branch()) {
(Some(o), _) => o.branch.clone(),
(None, Some((b, _))) => b,
(None, None) => return Err(invalid("the member has no branch to follow (set `branch = \"…\"` for it in the workspace descriptor)".into())),
};
let clone = self.layout.clone_of(&member.name);
if let Some(parent) = clone.parent() {
std::fs::create_dir_all(parent).map_err(|e| internal("create the clone dir", e.into()))?;
}
git::sync_clone(&member.source.to_string_lossy(), &clone).map_err(|e| internal("fetch the consumer's clone", e))?;
let first = git::resolve_from(&clone, &branch, from).map_err(|e| invalid(format!("{e:#}")))?;
let before = git::offset_before(&clone, &first).map_err(|e| internal("resolve the commit before it", e))?;
let total = git::first_parent_since(&clone, &branch, before.as_deref()).map_err(|e| internal("walk the branch", e))?.len();
let tip = git::tip(&clone, &branch).map_err(|e| internal("read the tip", e))?;
let row = BackfillRow {
repo: member.name.clone(),
from_sha: first,
base_seq: offset.as_ref().map_or(0, |o| o.seq),
target_sha: tip,
total: total as i64,
status: schema::STATUS_REQUESTED.to_string(),
detail: Some(format!("asked as `{from}` on branch `{branch}`")),
};
log.block_on(consumer.writer().record_backfill(&row)).map_err(|e| internal("write the request", e))?;
Ok(self.backfill_of(log, member, &row))
})
}
/// What the backfill job does next for `member` (see [`Next`]). When the
/// walk has reached the member's head this WRITES the `done` row — so call
/// it once per turn, and never while a turn's consume is still running.
pub fn backfill_next(&self, member: &Member) -> Result<Next, Refusal> {
let log = self.log()?;
let internal = |what: &str, e: anyhow::Error| Refusal::new(State::Internal, format!("backfill of `{}` in `{}`: {what}: {e:#}", member.name, self.workspace));
let rows = log.block_on(log.backfills(Some(&member.name))).map_err(|e| internal("read the backfills", e))?;
let Some(row) = rows.last().filter(|r| r.is_open()) else { return Ok(Next::Nothing) };
let backfill = self.backfill_of(&log, member, row);
if !backfill.started {
return Ok(Next::Step { from: Some(row.from_sha.clone()), backfill });
}
if backfill.at.as_ref().is_some_and(|o| o.sha != member.head) && backfill.done < backfill.total {
return Ok(Next::Step { from: None, backfill });
}
// at the head (or nothing of the branch is left after the offset)
self.finish_backfill(member, row, backfill.total, None).map(Next::Finished)
}
/// Close `member`'s open backfill with a `done` row (the walk reached the
/// head, or — `why` — it cannot go on and says so).
pub fn finish_backfill(&self, member: &Member, row: &BackfillRow, walked: usize, why: Option<String>) -> Result<Backfill, Refusal> {
let t0 = Instant::now();
self.with_consumer(|log, consumer| {
let done = BackfillRow {
status: schema::STATUS_DONE.to_string(),
total: walked as i64,
target_sha: log.block_on(log.offset(&member.name)).ok().flatten().map_or_else(|| row.target_sha.clone(), |o| o.sha),
detail: why.clone(),
..row.clone()
};
log.block_on(consumer.writer().record_backfill(&done))
.map_err(|e| Refusal::new(State::Internal, format!("backfill of `{}` in `{}`: write the `done` row: {e:#} ({} ms)", member.name, self.workspace, t0.elapsed().as_millis())))?;
Ok(self.backfill_of(log, member, &done))
})
}
}