orch/engine/subscribers/
unblock.rs

1//! Reacts to Done events — unblocks parent tasks immediately.
2
3use crate::backends::ExternalBackend;
4use crate::engine::events::TaskEvent;
5use crate::engine::tasks::TaskManager;
6use crate::engine::tick;
7use crate::store::TaskStore;
8use std::sync::Arc;
9use tokio::sync::broadcast;
10
11/// Spawn a task that listens for Done events and unblocks parent tasks.
12///
13/// This mirrors the logic in `tick_unblock_parents` but triggers instantly
14/// instead of waiting for the next tick. The tick-loop call remains as a
15/// fallback for any events missed due to subscriber lag.
16pub fn spawn(
17    mut rx: broadcast::Receiver<TaskEvent>,
18    backend: Arc<dyn ExternalBackend>,
19    task_manager: Arc<TaskManager>,
20    store: Arc<TaskStore>,
21    repo: String,
22) {
23    tokio::spawn(async move {
24        loop {
25            match rx.recv().await {
26                Ok(event) if event.new_status == "done" && event.repo == repo => {
27                    tracing::info!(
28                        task_id = event.task_id,
29                        "event-driven parent unblock triggered"
30                    );
31                    if let Err(e) =
32                        tick::tick_unblock_parents(&backend, &task_manager, &store, &repo).await
33                    {
34                        tracing::warn!(?e, "event-driven unblock failed (tick will retry)");
35                    }
36                }
37                Ok(_) => {} // Not a done event or different repo
38                Err(broadcast::error::RecvError::Lagged(n)) => {
39                    tracing::debug!(n, "unblock subscriber lagged, tick will catch up");
40                }
41                Err(_) => break, // Channel closed
42            }
43        }
44    });
45}
orch/engine/subscribers/unblock.rs

orch/engine/subscribers/
unblock.rs
