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

orch/engine/subscribers/
unblock.rs
