src/warehouse/workspace_events.rs
//! Iceberg writer/reader for the **workspace-populate STEP stream**
//! (`workspace_events`, EPIC #21 POPULATE 6-8).
//!
//! Where [`super::clone_events`] records the git *clone/fetch* outcome per
//! member, this records the per-STEP outcome of a
//! [`crate::monitor::republish_with`] POPULATE pass — one row per
//! `(member, step)` plus a workspace-level `populate` summary row (member `*`).
//! It makes "what did populate actually do, and did each phase succeed" a
//! first-class, persisted, remotely-readable stream instead of scattered
//! `eprintln!`s.
//!
//! Steps: `snapshot` · `deep-scan` · `test-surface` · `arch-generate` ·
//! `populate` (the workspace-level roll-up). Status: `ok` | `error` |
//! `skipped`. Same shape and read pattern as `clone_events`, so the fat/thin
//! renderers reuse one code path.
//!
//! Write path: [`WorkspaceEventLog`] accumulates rows during a populate pass;
//! [`append_workspace_events`] flushes them in one Iceberg snapshot.
//! Best-effort + non-fatal — a logging-row failure never aborts a populate.
//!
//! Read path: [`query_workspace_events`] scans the table, scopes by workspace,
//! newest-first.
//!
//! Schema: [`super::iceberg_schema::workspace_events`].
use std::sync::Arc;
use anyhow::{Result, anyhow};
use arrow::array::{Array, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use chrono::{DateTime, TimeZone, Utc};
use iceberg::arrow::schema_to_arrow_schema;
use super::iceberg::{IcebergWarehouse, TABLE_WORKSPACE_EVENTS, append_batch};
// Column indices match `iceberg_schema::workspace_events` field order.
const COL_TS_MICROS: usize = 0;
const COL_WORKSPACE: usize = 1;
const COL_MEMBER: usize = 2;
const COL_STEP: usize = 3;
const COL_STATUS: usize = 4;
const COL_DETAIL: usize = 5;
const COL_ELAPSED_MS: usize = 6;
/// Member sentinel for a workspace-level (not per-member) event row.
pub const WORKSPACE_MEMBER: &str = "*";
/// Canonical populate-step tags (the `step` column).
pub mod step {
/// Tantivy code-index (re)build over the members' checkouts (monitor step 2).
/// Recorded only when the walk found ZERO files across present members — a
/// broken walk (classically symlinked FAT members, which the walker does not
/// follow), never an idle one. A green populate over an empty search index
/// is the exact fake-green this tag exists to prevent.
pub const INDEX_BUILD: &str = "index-build";
/// Index snapshot into Iceberg (monitor step 3).
pub const SNAPSHOT: &str = "snapshot";
/// Transitive dep-closure deep scan (monitor step 5, opt-in).
pub const DEEP_SCAN: &str = "deep-scan";
/// Test-surface discovery/logging (monitor step 6).
pub const TEST_SURFACE: &str = "test-surface";
/// Architecture-wiring generate + historize (monitor step 7 — folded in).
pub const ARCH_GENERATE: &str = "arch-generate";
/// `Cargo.toml` FEATURE-ACTIVATION graph parse → `feature_dep_edges`
/// (monitor step 7d). The one populate step whose whole output is written
/// under a SINGLE snapshot id for all members — see the step's comment.
pub const FEATURE_GRAPH: &str = "feature-graph";
/// Semantic-vector embed + HNSW fill (monitor step 3b). SKIPPED loudly when
/// the binary was built without an embedder (`embed-tract`/`embed-ort`), so a
/// hollow-Mímir populate is never silently green.
pub const EMBED: &str = "embed";
/// RESOLVED (SCIP) call-edge ingest (monitor step 1b). SKIPPED loudly when the
/// binary was built without `--features scip`.
pub const SCIP: &str = "scip";
/// The workspace-level populate roll-up (monitor step 8, member `*`).
pub const POPULATE: &str = "populate";
}
/// Canonical status tags (the `status` column).
pub mod status {
pub const OK: &str = "ok";
pub const ERROR: &str = "error";
pub const SKIPPED: &str = "skipped";
}
/// One row of the `workspace_events` table — a single populate STEP's outcome
/// for one member (or the workspace-level `populate` roll-up, member `*`).
///
/// `serde` derives let a server ship the rows to a thin client (mirrors
/// [`super::clone_events::CloneEventRow`]).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct WorkspaceEventRow {
/// Event time, microseconds since the Unix epoch (UTC).
pub ts_micros: i64,
/// The workspace this populate belonged to.
pub workspace: String,
/// The member the step ran for, or [`WORKSPACE_MEMBER`] (`*`) for a
/// workspace-level step (`populate`, or a whole-workspace `deep-scan`).
pub member: String,
/// The populate step (see [`step`]).
pub step: String,
/// `ok` | `error` | `skipped` (see [`status`]).
pub status: String,
/// Free-form detail: on `ok` a summary (snapshot id, node/edge counts, …);
/// on `error` the error chain (`{e:#}`). Stored as "" never null.
pub detail: String,
/// Wall-clock duration of the step, milliseconds.
pub elapsed_ms: i64,
}
/// An accumulating populate-step log. The populate pass pushes one row per step
/// as it runs, then [`flush`](Self::flush) appends them in a single Iceberg
/// snapshot at the end (best-effort, non-fatal).
#[derive(Debug, Default)]
pub struct WorkspaceEventLog {
workspace: String,
rows: Vec<WorkspaceEventRow>,
}
impl WorkspaceEventLog {
pub fn new(workspace: impl Into<String>) -> Self {
Self {
workspace: workspace.into(),
rows: Vec::new(),
}
}
/// Record one step outcome (stamped now, UTC micros).
pub fn record(
&mut self,
member: impl Into<String>,
step_: &str,
status_: &str,
detail: impl Into<String>,
elapsed_ms: i64,
) {
self.rows.push(WorkspaceEventRow {
ts_micros: Utc::now().timestamp_micros(),
workspace: self.workspace.clone(),
member: member.into(),
step: step_.to_string(),
status: status_.to_string(),
detail: detail.into(),
elapsed_ms,
});
}
/// Record a workspace-level step (member `*`).
pub fn record_workspace(
&mut self,
step_: &str,
status_: &str,
detail: impl Into<String>,
elapsed_ms: i64,
) {
self.record(WORKSPACE_MEMBER, step_, status_, detail, elapsed_ms);
}
/// The rows accumulated so far.
pub fn rows(&self) -> &[WorkspaceEventRow] {
&self.rows
}
/// True when a `member`/`step` pair recorded an `error`.
pub fn any_error(&self) -> bool {
self.rows.iter().any(|r| r.status == status::ERROR)
}
/// Append the accumulated rows to `workspace_events` in one snapshot.
/// Best-effort + **non-fatal**: a warehouse hiccup is logged and swallowed
/// so a step-logging failure never aborts an otherwise-successful populate.
pub fn flush(&self, wh: &IcebergWarehouse) {
if self.rows.is_empty() {
return;
}
if let Err(e) = wh.block_on(append_workspace_events(wh, &self.rows)) {
eprintln!(
" ⚠ workspace_events: dropped {} populate-step row(s) for `{}` (non-fatal): {e:#}",
self.rows.len(),
self.workspace,
);
}
}
}
/// Append a batch of `workspace_events` rows (one Iceberg snapshot). Mirrors
/// [`super::clone_events::append_clone_events`].
pub async fn append_workspace_events(
wh: &IcebergWarehouse,
rows: &[WorkspaceEventRow],
) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
// Blank/torn metadata JSON self-heals (drop+recreate the 0-row table).
let table = wh
.load_table_for_append(
TABLE_WORKSPACE_EVENTS,
super::iceberg_schema::workspace_events()?,
&[],
)
.await?;
let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(
TimestampMicrosecondArray::from(rows.iter().map(|r| r.ts_micros).collect::<Vec<_>>())
.with_timezone("+00:00"),
),
Arc::new(StringArray::from(
rows.iter().map(|r| r.workspace.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.member.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.step.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.status.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.detail.clone()).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
rows.iter().map(|r| r.elapsed_ms).collect::<Vec<_>>(),
)),
];
let batch = RecordBatch::try_new(arrow_schema, cols)?;
append_batch(wh.catalog(), table, batch).await?;
Ok(())
}
/// Which workspace events to read.
#[derive(Debug, Clone)]
pub enum WorkspaceEventSelector {
/// Every event whose `workspace` matches.
Workspace(String),
/// Everything in the table.
All,
}
/// Read workspace events, scoped by `sel`, returned **newest-first**
/// (`ts_micros` descending). Iceberg gives no scan order, so the reader sorts.
pub async fn query_workspace_events(
wh: &IcebergWarehouse,
sel: &WorkspaceEventSelector,
) -> Result<Vec<WorkspaceEventRow>> {
let batches: Vec<RecordBatch> =
super::iceberg::load_and_read_all(wh, TABLE_WORKSPACE_EVENTS).await?;
let mut out: Vec<WorkspaceEventRow> = Vec::new();
for b in &batches {
let ts = col_ts(b, COL_TS_MICROS)?;
let workspace = col_str(b, COL_WORKSPACE)?;
let member = col_str(b, COL_MEMBER)?;
let step_ = col_str(b, COL_STEP)?;
let st = col_str(b, COL_STATUS)?;
let detail = col_str(b, COL_DETAIL)?;
let elapsed = col_i64(b, COL_ELAPSED_MS)?;
for i in 0..b.num_rows() {
let row = WorkspaceEventRow {
ts_micros: ts.value(i),
workspace: workspace.value(i).to_string(),
member: member.value(i).to_string(),
step: step_.value(i).to_string(),
status: st.value(i).to_string(),
detail: detail.value(i).to_string(),
elapsed_ms: elapsed.value(i),
};
let keep = match sel {
WorkspaceEventSelector::Workspace(w) => &row.workspace == w,
WorkspaceEventSelector::All => true,
};
if keep {
out.push(row);
}
}
}
out.sort_by(|a, b| b.ts_micros.cmp(&a.ts_micros));
Ok(out)
}
/// Render populate step events human-readably: newest first, each line
/// `mark step member — detail (elapsed)`.
pub fn render_events(rows: &[WorkspaceEventRow]) -> String {
if rows.is_empty() {
return "(no workspace populate events recorded)\n".to_string();
}
let mut out = String::new();
for r in rows {
let mark = match r.status.as_str() {
status::OK => "✓",
status::ERROR => "✗",
status::SKIPPED => "·",
_ => "?",
};
let when = ts_to_rfc3339(r.ts_micros);
let detail = if r.detail.is_empty() {
String::new()
} else {
format!(" — {}", r.detail)
};
out.push_str(&format!(
"{mark} {when} {step:<14} {member}{detail} ({elapsed}ms)\n",
step = r.step,
member = r.member,
elapsed = r.elapsed_ms,
));
}
out
}
// ─── column helpers ─────────────────────────────────────────────────────
fn col_str(b: &RecordBatch, idx: usize) -> Result<&StringArray> {
b.column(idx)
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| anyhow!("workspace_events col {idx} is not StringArray"))
}
fn col_i64(b: &RecordBatch, idx: usize) -> Result<&Int64Array> {
b.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| anyhow!("workspace_events col {idx} is not Int64Array"))
}
fn col_ts(b: &RecordBatch, idx: usize) -> Result<&TimestampMicrosecondArray> {
b.column(idx)
.as_any()
.downcast_ref::<TimestampMicrosecondArray>()
.ok_or_else(|| anyhow!("workspace_events col {idx} is not TimestampMicrosecondArray"))
}
/// Format `ts_micros` as RFC3339 (UTC).
pub fn ts_to_rfc3339(ts_micros: i64) -> String {
let dt: DateTime<Utc> = Utc
.timestamp_micros(ts_micros)
.single()
.unwrap_or_else(Utc::now);
dt.to_rfc3339()
}
#[cfg(test)]
mod tests {
use super::*;
/// Inject-and-assert: a populate pass logs per-step + a workspace roll-up,
/// flushes once, and reads back newest-first, scoped by workspace, with the
/// (member, step, status, detail) round-tripping exactly.
#[test]
fn populate_steps_persist_and_read_back_scoped() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let mut log = WorkspaceEventLog::new("nordisk");
log.record("facett", step::SNAPSHOT, status::OK, "snapshot-7", 12);
log.record(
"facett",
step::ARCH_GENERATE,
status::OK,
"nodes=40 edges=88",
30,
);
log.record(
"korp",
step::ARCH_GENERATE,
status::ERROR,
"scan failed: boom",
5,
);
log.record_workspace(step::POPULATE, status::ERROR, "1/2 members clean", 47);
assert!(log.any_error());
log.flush(&wh);
let rows = wh
.block_on(query_workspace_events(
&wh,
&WorkspaceEventSelector::Workspace("nordisk".into()),
))
.unwrap();
assert_eq!(rows.len(), 4, "one row per recorded step");
// The failed member/step is first-class and its error round-trips.
let failed: Vec<_> = rows.iter().filter(|r| r.status == status::ERROR).collect();
assert_eq!(failed.len(), 2, "korp arch + the populate roll-up");
let korp = rows.iter().find(|r| r.member == "korp").unwrap();
assert_eq!(korp.step, "arch-generate");
assert!(
korp.detail.contains("boom"),
"error chain readable: {}",
korp.detail
);
// The workspace-level roll-up row (member `*`).
let rollup = rows.iter().find(|r| r.member == WORKSPACE_MEMBER).unwrap();
assert_eq!(rollup.step, "populate");
assert_eq!(rollup.status, "error");
// Another workspace's scope sees nothing.
let other = wh
.block_on(query_workspace_events(
&wh,
&WorkspaceEventSelector::Workspace("holger".into()),
))
.unwrap();
assert!(other.is_empty(), "workspace scope isolates events");
// Human render + JSON RPC contract.
let txt = render_events(&rows);
assert!(txt.contains("arch-generate") && txt.contains("korp") && txt.contains('✗'));
let json = serde_json::to_string(&rows).unwrap();
let back: Vec<WorkspaceEventRow> = serde_json::from_str(&json).unwrap();
assert_eq!(back, rows);
}
#[test]
fn empty_log_flush_is_a_noop() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let log = WorkspaceEventLog::new("nordisk");
log.flush(&wh); // must not error / create rows
let rows = wh
.block_on(query_workspace_events(&wh, &WorkspaceEventSelector::All))
.unwrap();
assert!(rows.is_empty());
}
}