mj_controller/pollers/
runtime_feed.rs

1use super::*;
2
3pub struct RemoteDashboardWorkerPoller {
4    pub targets: tokio::sync::watch::Sender<Vec<WorkerPollTarget>>,
5    pub updates: SessionManagerUpdates,
6    pub control: SessionManagerControl,
7    pub shutdown: SessionManagerShutdown,
8    pub state: tokio::sync::watch::Receiver<RuntimeStateUpdate>,
9    /// Reviews the daemon is running for this workspace's sessions.
10    pub reviews: tokio::sync::watch::Receiver<Vec<crate::review_host::RuntimeReviewView>>,
11    /// Background events the daemon wants reported once, oldest first.
12    pub notices: tokio::sync::watch::Receiver<Vec<daemon::RuntimeNotice>>,
13    pub config: tokio::sync::watch::Receiver<mj_core::config::Config>,
14}
15
16/// Records and lifecycle ownership must reach the surface in the same frame.
17#[derive(Debug, Clone, Default)]
18pub struct RuntimeStateUpdate {
19    pub workspace_names: std::collections::BTreeMap<String, String>,
20    pub revision: u64,
21    pub records: Vec<SessionRecord>,
22    pub lifecycles: Vec<daemon::RuntimeLifecycleView>,
23    pub moves: Vec<mj_core::state::MoveOperation>,
24    /// Parent/child relations for the sessions in `records`, so a surface can
25    /// keep a daemon-created child out of the real workspace without a full
26    /// state reload.
27    pub subagents: Vec<SubagentRecord>,
28}
29
30/// What a session looked like the last time a view was published for it.
31///
32/// The poller compares this before reading anything, so a session that has not
33/// moved costs one comparison rather than a full transcript load. Nothing here
34/// grows with the transcript: the projection is identified by its ordinal and
35/// digest, and the operational state is bounded by the relay's own command and
36/// configuration surface.
37#[derive(Debug, Clone, PartialEq)]
38pub(super) struct PublishedView {
39    pub(super) projection_ordinal: u64,
40    pub(super) projection_digest: String,
41    pub(super) operational: Option<mj_core::relay::RelayOperationalState>,
42    pub(super) connected: bool,
43    pub(super) error: Option<String>,
44}
45
46impl PublishedView {
47    pub(super) fn of(runtime: &crate::daemon::RuntimeSessionView) -> Self {
48        Self {
49            projection_ordinal: runtime.projection_ordinal,
50            projection_digest: runtime.projection_digest.clone(),
51            operational: runtime.operational.clone(),
52            connected: runtime.connected,
53            error: runtime.error.as_ref().map(|error| format!("{error:?}")),
54        }
55    }
56
57    pub(super) fn matches(&self, runtime: &crate::daemon::RuntimeSessionView) -> bool {
58        *self == Self::of(runtime)
59    }
60}
61
62pub(super) const PROJECTION_CONVERGENCE_RETRIES: u8 = 20;
63pub(super) const PROJECTION_CONVERGENCE_RETRY_DELAY: Duration = Duration::from_millis(50);
64
65#[derive(Debug, Clone, PartialEq, Eq)]
66pub(super) struct ProjectionMismatch {
67    pub(super) published_ordinal: u64,
68    pub(super) published_digest: String,
69    pub(super) durable_ordinal: u64,
70    pub(super) durable_digest: String,
71}
72
73#[derive(Default)]
74pub(super) struct ProjectionConvergence {
75    pub(super) attempts: std::collections::BTreeMap<String, (ProjectionMismatch, u8)>,
76}
77
78impl ProjectionConvergence {
79    pub(super) fn converged(&mut self, session_id: &str) {
80        self.attempts.remove(session_id);
81    }
82
83    /// Give a lifecycle rollback and the daemon's cached relay view a bounded
84    /// window to converge. Repeating the same mismatch eventually reports the
85    /// integrity failure instead of hiding it indefinitely.
86    pub(super) fn should_retry(&mut self, session_id: &str, mismatch: ProjectionMismatch) -> bool {
87        let entry = self
88            .attempts
89            .entry(session_id.to_owned())
90            .or_insert_with(|| (mismatch.clone(), 0));
91        if entry.0 != mismatch {
92            *entry = (mismatch, 0);
93        }
94        entry.1 = entry.1.saturating_add(1);
95        entry.1 <= PROJECTION_CONVERGENCE_RETRIES
96    }
97}
98
99/// Read-only updates shared by the dashboard and workspace preview. A snapshot
100/// precedes its session views, so consumers can establish membership first.
101pub enum RuntimeFeedUpdate {
102    Snapshot(Box<daemon::RuntimeSnapshot>),
103    Session {
104        session_id: String,
105        view: Box<ManagedSessionView>,
106    },
107    Error(String),
108}
109
110/// Dropping a subscription cancels even a pending daemon long poll. The task
111/// owns no writer or relay connection; blocking projection reads are bounded.
112pub struct RuntimeFeed {
113    pub updates: tokio::sync::mpsc::Receiver<RuntimeFeedUpdate>,
114    pub(super) task: tokio::task::JoinHandle<()>,
115}
116
117impl Drop for RuntimeFeed {
118    fn drop(&mut self) {
119        self.task.abort();
120    }
121}
122
123pub(super) type StoredProjection = Option<(MaterializedSession, mj_core::state::ProjectionWindow)>;
124
125pub(super) async fn load_runtime_projection(session_id: String) -> Result<StoredProjection> {
126    static READERS: std::sync::LazyLock<Arc<tokio::sync::Semaphore>> =
127        std::sync::LazyLock::new(|| Arc::new(tokio::sync::Semaphore::new(4)));
128    let permit = Arc::clone(&READERS)
129        .acquire_owned()
130        .await
131        .context("projection readers stopped")?;
132    tokio::task::spawn_blocking(move || {
133        let _permit = permit;
134        let result = crate::database::load_materialized_projection_tail(
135            &session_id,
136            crate::database::PROJECTION_TAIL_ITEMS,
137        );
138        // A blocking SQLite read can outlive cancellation of its subscriber.
139        if let Err(error) = &result {
140            tracing::warn!(%session_id, %error, "could not load runtime projection");
141        }
142        result
143    })
144    .await
145    .context("projection load task failed")?
146}
147
148pub(super) fn spawn_runtime_feed_with<P, PF, L, LF>(
149    workspace_id: String,
150    poll: P,
151    load: L,
152) -> RuntimeFeed
153where
154    P: Fn(String, u64) -> PF + Send + 'static,
155    PF: Future<Output = Result<daemon::RuntimeSnapshot>> + Send,
156    L: Fn(String) -> LF + Clone + Send + 'static,
157    LF: Future<Output = Result<StoredProjection>> + Send + 'static,
158{
159    let (tx, updates) = tokio::sync::mpsc::channel(32);
160    let task = tokio::spawn(async move {
161        let result = run_runtime_feed(workspace_id, poll, load, &tx).await;
162        if let Err(error) = result {
163            let message = format!("Runtime feed stopped: {error:#}");
164            tracing::error!(%message);
165            let _ = tx.send(RuntimeFeedUpdate::Error(message)).await;
166        }
167    });
168    RuntimeFeed { updates, task }
169}
170
171pub(super) async fn run_runtime_feed<P, PF, L, LF>(
172    workspace_id: String,
173    poll: P,
174    load: L,
175    tx: &tokio::sync::mpsc::Sender<RuntimeFeedUpdate>,
176) -> Result<()>
177where
178    P: Fn(String, u64) -> PF,
179    PF: Future<Output = Result<daemon::RuntimeSnapshot>>,
180    L: Fn(String) -> LF + Clone + Send + 'static,
181    LF: Future<Output = Result<StoredProjection>> + Send + 'static,
182{
183    let mut revision = 0;
184    let mut convergence = ProjectionConvergence::default();
185    let mut published = std::collections::BTreeMap::<String, PublishedView>::new();
186    loop {
187        let mut snapshot = match poll(workspace_id.clone(), revision).await {
188            Ok(snapshot) => snapshot,
189            Err(error) => {
190                if tx
191                    .send(RuntimeFeedUpdate::Error(format!(
192                        "Could not refresh sessions: {error:#}"
193                    )))
194                    .await
195                    .is_err()
196                {
197                    return Ok(());
198                }
199                tokio::time::sleep(Duration::from_millis(250)).await;
200                continue;
201            }
202        };
203        let snapshot_revision = snapshot.revision;
204        let sessions = std::mem::take(&mut snapshot.sessions);
205        published.retain(|id, _| sessions.iter().any(|session| &session.session_id == id));
206        convergence
207            .attempts
208            .retain(|id, _| sessions.iter().any(|session| &session.session_id == id));
209        if tx
210            .send(RuntimeFeedUpdate::Snapshot(Box::new(snapshot)))
211            .await
212            .is_err()
213        {
214            return Ok(());
215        }
216        let mut pending = sessions
217            .into_iter()
218            .filter(|runtime| {
219                !published
220                    .get(&runtime.session_id)
221                    .is_some_and(|last| last.matches(runtime))
222            })
223            .collect::<std::collections::VecDeque<_>>();
224        let mut tasks = tokio::task::JoinSet::new();
225        let mut retry = false;
226        while !pending.is_empty() || !tasks.is_empty() {
227            // Independent session reads overlap, without flooding SQLite or
228            // leaving an unbounded number of blocking reads after cancellation.
229            while tasks.len() < 4 {
230                let Some(runtime) = pending.pop_front() else {
231                    break;
232                };
233                let load = load.clone();
234                tasks.spawn(async move {
235                    let stored = if runtime.operational.is_some() {
236                        load(runtime.session_id.clone()).await
237                    } else {
238                        Ok(None)
239                    };
240                    (runtime, stored)
241                });
242            }
243            let Some(result) = tasks.join_next().await else {
244                break;
245            };
246            let (runtime, stored) = result.context("join runtime projection reader")?;
247            let session_id = runtime.session_id.clone();
248            let fingerprint = PublishedView::of(&runtime);
249            let Some(view) = runtime_projection_view(runtime, stored, &mut convergence) else {
250                retry = true;
251                continue;
252            };
253            if view.snapshot.is_some() {
254                published.insert(session_id.clone(), fingerprint);
255            } else {
256                published.remove(&session_id);
257            }
258            if tx
259                .send(RuntimeFeedUpdate::Session {
260                    session_id,
261                    view: Box::new(view),
262                })
263                .await
264                .is_err()
265            {
266                return Ok(());
267            }
268        }
269        if retry {
270            tokio::time::sleep(PROJECTION_CONVERGENCE_RETRY_DELAY).await;
271        } else {
272            revision = revision.max(snapshot_revision);
273        }
274    }
275}
276
277pub(super) fn runtime_projection_view(
278    runtime: daemon::RuntimeSessionView,
279    stored: Result<StoredProjection>,
280    convergence: &mut ProjectionConvergence,
281) -> Option<ManagedSessionView> {
282    let Some(operational) = runtime.operational else {
283        return Some(ManagedSessionView {
284            snapshot: None,
285            connected: runtime.connected,
286            error: runtime.error,
287        });
288    };
289    let detail = match stored {
290        Ok(Some((materialized, window)))
291            if materialized.applied_event_ordinal > runtime.projection_ordinal
292                || (materialized.applied_event_ordinal == runtime.projection_ordinal
293                    && materialized.applied_event_digest == runtime.projection_digest) =>
294        {
295            convergence.converged(&runtime.session_id);
296            return Some(ManagedSessionView {
297                snapshot: Some(ManagedSessionSnapshot {
298                    materialized,
299                    window,
300                    operational,
301                    latest_credential_sync_signal: runtime.latest_credential_sync_signal,
302                    worker_build: None,
303                    subagent_requests: Vec::new(),
304                    subagent_results: Vec::new(),
305                }),
306                connected: runtime.connected,
307                error: runtime.error,
308            });
309        }
310        Ok(Some((materialized, _))) => {
311            let mismatch = ProjectionMismatch {
312                published_ordinal: runtime.projection_ordinal,
313                published_digest: runtime.projection_digest,
314                durable_ordinal: materialized.applied_event_ordinal,
315                durable_digest: materialized.applied_event_digest.clone(),
316            };
317            if convergence.should_retry(&runtime.session_id, mismatch) {
318                return None;
319            }
320            if materialized.applied_event_ordinal < runtime.projection_ordinal {
321                format!(
322                    "daemon published projection {} but SQLite contains only {} after a bounded convergence retry",
323                    runtime.projection_ordinal, materialized.applied_event_ordinal
324                )
325            } else {
326                format!(
327                    "daemon and SQLite projection digests differ at ordinal {} after a bounded convergence retry",
328                    runtime.projection_ordinal
329                )
330            }
331        }
332        Ok(None) => "daemon published a session with no durable projection".into(),
333        Err(error) => format!("load daemon-owned projection: {error:#}"),
334    };
335    Some(ManagedSessionView {
336        snapshot: None,
337        connected: false,
338        error: Some(ViewError::ProjectionIntegrity(detail)),
339    })
340}
mj_controller/pollers/runtime_feed.rs

mj_controller/pollers/
runtime_feed.rs
