knowledge_runtime/projection/
rebuild.rs

1//! Projection rebuild driver interface (KR-003).
2//!
3//! The [`ProjectionTracker`](super::lifecycle::ProjectionTracker) is **observability only**.
4//! It records projection health transitions but does NOT execute, schedule, or
5//! retry rebuilds. External callers must implement the [`RebuildDriver`] trait
6//! to drive rebuild execution and report outcomes back to the tracker.
7//!
8//! ## Why a trait instead of built-in execution?
9//!
10//! Rebuild execution depends on:
11//! - Which upstream data sources to re-query
12//! - How to batch/throttle rebuilds
13//! - Retry and error handling policies
14//! - Concurrency and resource management
15//!
16//! These concerns are owned by the orchestrator, not the runtime. The runtime
17//! provides projection health inspection so the driver can make informed
18//! decisions, but never initiates work autonomously.
19//!
20//! ## Phase status: KR-003 architecture closure
21//!
22//! This trait is the explicit architecture answer to "who drives rebuilds?"
23//! The tracker is observability; this trait is the execution contract.
24
25use crate::error::RuntimeError;
26use crate::ids::ProjectionId;
27use crate::projection::lifecycle::{ProjectionHealth, ProjectionTracker, ProjectionVersion};
28
29/// External rebuild driver trait (KR-003).
30///
31/// Implement this trait in your orchestration layer to drive projection
32/// rebuilds based on tracker health state. The runtime never calls this
33/// trait itself — it is for external callers that poll projection health
34/// and decide when to rebuild.
35///
36/// ## Example flow
37///
38/// ```text
39/// 1. Orchestrator polls tracker.health(&id) -> Stale
40/// 2. Orchestrator calls driver.rebuild(&id) -> Ok(outcome)
41/// 3. Orchestrator calls runtime.record_projection_build(id, ...) or
42///    runtime.record_projection_failure(id, error)
43/// 4. Tracker state transitions to Healthy or remains Stale
44/// ```
45#[allow(async_fn_in_trait)]
46pub trait RebuildDriver {
47    /// Execute a rebuild for the given projection.
48    ///
49    /// Returns a [`RebuildOutcome`] on success, or an error if the rebuild
50    /// could not be attempted. The caller is responsible for reporting the
51    /// outcome back to the tracker via `record_projection_build()` or
52    /// `record_projection_failure()`.
53    async fn rebuild(&self, id: &ProjectionId) -> Result<RebuildOutcome, RuntimeError>;
54
55    /// Check whether this driver can handle a rebuild for the given projection.
56    ///
57    /// Returns `false` if the projection kind/scope is not supported by this
58    /// driver. The orchestrator can use this to skip unsupported projections
59    /// without error.
60    fn can_rebuild(&self, id: &ProjectionId) -> bool;
61}
62
63/// Outcome of a successful rebuild.
64#[derive(Debug, Clone)]
65pub struct RebuildOutcome {
66    /// How many source items were processed.
67    pub source_count: usize,
68    /// How long the rebuild took, in milliseconds.
69    pub build_duration_ms: u64,
70    /// Version metadata for the rebuilt projection.
71    pub version: Option<ProjectionVersion>,
72}
73
74/// Drive rebuilds for all stale projections using the provided driver.
75///
76/// This is a convenience function that polls the tracker for stale projections
77/// and calls the driver for each one. It reports outcomes back to the tracker
78/// automatically.
79///
80/// Returns the number of projections that were successfully rebuilt.
81pub async fn rebuild_stale<D: RebuildDriver>(
82    tracker: &mut ProjectionTracker,
83    driver: &D,
84) -> Result<usize, RuntimeError> {
85    let stale_ids: Vec<ProjectionId> = tracker
86        .all_ids()
87        .filter(|id| tracker.health(id) == ProjectionHealth::Stale)
88        .cloned()
89        .collect();
90
91    let mut rebuilt = 0;
92    for id in stale_ids {
93        if !driver.can_rebuild(&id) {
94            continue;
95        }
96        match driver.rebuild(&id).await {
97            Ok(outcome) => {
98                tracker.record_build(
99                    id,
100                    outcome.source_count,
101                    outcome.build_duration_ms,
102                    outcome.version,
103                );
104                rebuilt += 1;
105            }
106            Err(e) => {
107                tracker.record_failure(id, format!("{e}"));
108            }
109        }
110    }
111    Ok(rebuilt)
112}
knowledge_runtime/projection/rebuild.rs

knowledge_runtime/projection/
rebuild.rs
