fix: correct syntax errors and type mismatches in progress watching API

[Copilot]

The progress watching implementation introduced in the previous commits had multiple compilation-blocking errors: orphaned methods outside impl blocks, duplicate mod tests, and incorrect usage of tokio::sync::Mutex as if it were std::sync::Mutex.

Stats (execution/stats.rs)

• Orphaned methods: component_start/component_complete/component_error were placed after the #[cfg(test)] mod tests closing brace instead of inside impl UpdateStats — moved into the correct #[cfg(feature = "persistence")] impl block
• Duplicate test module: Two #[cfg(test)] mod tests blocks merged into one
• Invalid test method calls: Tests called processing.inc() / processing.get() which don't exist on ProcessingCounters — corrected to processing.start(), processing.num_starts.get(), processing.get_in_process()
• Incorrect test assertion: component_counters_update_correctly created 3 endings for 2 starts (producing in_process = -1) — fixed by starting 3 items

Live Updater (execution/live_updater.rs)

• Async mutex misuse: emit_progress_update used match self.recv_state.lock() { Ok(s) => s, Err(p) => p.into_inner() } — tokio::sync::Mutex::lock() returns a future, not a Result. Changed to async fn using .await
• Non-existent field: wait() referenced self.status_rx (doesn't exist as a direct field; lives inside recv_state) — now locks recv_state briefly to clone the receiver before spawning the emitter task
• Wrong field name: status_rx.borrow().active_sources → active_source_idx, with indices mapped to source names via bounds-safe .filter_map/.get()

---

✨ Let Copilot coding agent set things up for you — coding agent works faster and does higher quality work when set up for your repo.

[Copilot]

Just as a heads up, I was blocked by some firewall rules while working on your feedback. Expand below for details.

Warning

Firewall rules blocked me from connecting to one or more addresses (expand for details)

I tried to connect to the following addresses, but was blocked by firewall rules:

• test
  • Triggering command: /home/REDACTED/work/recoco/recoco/target/debug/deps/recoco_core-af5104cd93dc7568 /home/REDACTED/work/recoco/recoco/target/debug/deps/recoco_core-af5104cd93dc7568 --emit=dep-info,metadata,link -C opt-level=3 -C embed-bitcode=no -C codegen-units=256 -C debuginfo=2 -C debug-assertions=on coco�� 8f104189414828/r-plugin-opt=/usr/libexec/gcc/x86_64-linux-gnu/13/lto-wrapper 8f104189414828/b-plugin-opt=-fresolution=/tmp/ccOEvi6h.res bin/rustc s/derive_where-b/home/REDACTED/.rustup/toolchains/stable-x86_64-REDACTED-linux-gnu/bin/rustc s/derive_where-b--crate-name 38c908fe84130.rlsecrecy bin/rustc (dns block)
  • Triggering command: /home/REDACTED/work/recoco/recoco/target/debug/deps/recoco_core-af5104cd93dc7568 /home/REDACTED/work/recoco/recoco/target/debug/deps/recoco_core-af5104cd93dc7568 b312ba6.derive_builder_macro.c22eb12bd181124b-cgu.1.rcgu.o b312ba6.derive_builder_macro.c22eb12bd181124b-cgu.2.rcgu.o --warn=clippy::nursery b312ba6.87gj7iq4ljyv5lrurvk2vj78s.rcgu.o -1949cf8c6b5b557--error-format=json ld/aws-lc-sys-48--json=diagnostic-rendered-ansi,artifacts,future-incompat 4f2335dc0.rlib ib d7.rlib b bin/rustc .rli�� /index.crates.io--error-format=json (dns block)
  • Triggering command: /home/REDACTED/work/recoco/recoco/target/debug/deps/recoco_core-af5104cd93dc7568 /home/REDACTED/work/recoco/recoco/target/debug/deps/recoco_core-af5104cd93dc7568 ld/aws-lc-sys-48-Wl,--version-script=/home/REDACTED/work/recoco/recoco/target/debug/de�� ld/aws-lc-sys-48-Wl,--no-undefined-version bin/rustc (dns block)

If you need me to access, download, or install something from one of these locations, you can either:

• Configure Actions setup steps to set up my environment, which run before the firewall is enabled
• Add the appropriate URLs or hosts to the custom allowlist in this repository's Copilot coding agent settings (admins only)

[Copilot]

Pull request overview

Fixes compilation-blocking issues in the progress watching implementation so flow progress updates and per-component stats compile and behave correctly.

Changes:

• Move UpdateStats::{component_start,component_complete,component_error} back into the correct impl UpdateStats block and consolidate duplicate tests modules.
• Fix stats tests to use the correct ProcessingCounters API and correct the in-process accounting assertion.
• Correct FlowLiveUpdater progress emission by properly awaiting tokio::sync::Mutex::lock(), cloning the watch::Receiver from recv_state, and mapping active source indices to names safely.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
crates/recoco-core/src/execution/stats.rs	Restores correct UpdateStats impl structure and fixes/relocates component stats tests.
crates/recoco-core/src/execution/live_updater.rs	Fixes async mutex usage and corrects active source tracking/name mapping for progress updates.

Comments suppressed due to low confidence (1)

crates/recoco-core/src/execution/live_updater.rs:598

• emit_progress_update keeps the tokio::sync::Mutex<UpdateReceiveState> guard while building and sending the ProgressUpdate. Even though this is small work today, it can block concurrent callers of next_status_updates() longer than necessary. Consider cloning only the needed status fields inside a short lock scope (or make FlowLiveUpdaterStatus: Clone and clone it), then drop the guard before assembling/sending the update.

    async fn emit_progress_update(&self) {
        // Always wait for the recv_state lock so we don't silently drop updates,
        // especially the final completion snapshot.
        let recv_state = self.recv_state.lock().await;

        let status = recv_state.status_rx.borrow();

        let active_sources: Vec<String> = status
            .active_source_idx
            .iter()
            .filter_map(|&idx| {
                self.flow_ctx
                    .flow
                    .flow_instance
                    .import_ops
                    .get(idx)
                    .map(|op| op.name.clone())
            })
            .collect();

        let total_sources = self.stats_per_task.len();
        let completed_sources = total_sources - status.active_source_idx.len();

        let source_stats: Vec<stats::SourceUpdateInfo> = std::iter::zip(
            self.flow_ctx.flow.flow_instance.import_ops.iter(),
            self.stats_per_task.iter(),
        )
        .map(|(import_op, stats)| stats::SourceUpdateInfo {
            source_name: import_op.name.clone(),
            stats: stats.as_ref().clone(),
        })
        .collect();

        let operation_in_process = self
            .operation_in_process_stats
            .get_all_operations_in_process();

        let update = ProgressUpdate {
            active_sources,
            total_sources,
            completed_sources,
            source_stats,
            operation_in_process,
        };

        let _ = self.progress_tx.send(Some(update));
    }

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.