Skip to content
This repository was archived by the owner on Jun 16, 2026. It is now read-only.

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

Merged
Adam Poulemanos (bashandbone) merged 2 commits into from
Mar 19, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
71ce4a9
Initial plan
Copilot Mar 19, 2026
334e000
fix: correct syntax errors, duplicate test module, and type errors in…
Copilot Mar 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 25 additions & 15 deletions crates/recoco-core/src/execution/live_updater.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -549,25 +549,23 @@ impl FlowLiveUpdater {
}

/// Emit a progress update to all subscribers.
fn emit_progress_update(&self) {
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 = match self.recv_state.lock() {
Ok(state) => state,
// If the mutex is poisoned, recover the inner state and still emit
// a best-effort progress update instead of silently skipping it.
Err(poisoned) => poisoned.into_inner(),
};
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()
.map(|&idx| {
self.flow_ctx.flow.flow_instance.import_ops[idx]
.name
.clone()
.filter_map(|&idx| {
self.flow_ctx
.flow
.flow_instance
.import_ops
.get(idx)
.map(|op| op.name.clone())
})
.collect();

Expand Down Expand Up @@ -608,7 +606,7 @@ impl FlowLiveUpdater {
let operation_in_process_stats = self.operation_in_process_stats.clone();

// Clone the status receiver so we can derive active_sources for periodic updates.
let mut status_rx = self.status_rx.clone();
let status_rx = self.recv_state.lock().await.status_rx.clone();

let progress_emitter = tokio::spawn(async move {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
Expand Down Expand Up @@ -639,8 +637,20 @@ impl FlowLiveUpdater {
let operation_in_process =
operation_in_process_stats.get_all_operations_in_process();

// Derive active_sources from the latest FlowLiveUpdaterStatus.
let active_sources = status_rx.borrow().active_sources.clone();
// Derive active_sources names from the latest FlowLiveUpdaterStatus.
let active_sources: Vec<String> = status_rx
.borrow()
.active_source_idx
.iter()
.filter_map(|&idx| {
flow_ctx
.flow
.flow_instance
.import_ops
.get(idx)
.map(|op| op.name.clone())
})
.collect();

let update = ProgressUpdate {
active_sources,
Expand Down Expand Up @@ -678,7 +688,7 @@ impl FlowLiveUpdater {
}

// Emit final progress update
self.emit_progress_update();
self.emit_progress_update().await;

progress_emitter.abort();
Ok(())
Expand Down
239 changes: 119 additions & 120 deletions crates/recoco-core/src/execution/stats.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -254,126 +254,6 @@ impl UpdateStats {
.or_default()
.clone()
}
}

#[cfg(test)]
mod tests {
use super::*;
use serde_json::Value;

#[test]
fn component_counters_update_correctly() {
let stats = UpdateStats::default();

// Populate per-component stats by writing into the by_component map.
{
let mut components = stats.by_component.write().unwrap();
let comp_stats = components.entry("comp_a".to_string()).or_default();
comp_stats.num_processed.inc(3);
comp_stats.num_errors.inc(1);
comp_stats.processing.inc(2);
}

let components = stats.by_component.read().unwrap();
let comp_stats = components.get("comp_a").expect("component stats should exist");
assert_eq!(comp_stats.num_processed.get(), 3);
assert_eq!(comp_stats.num_errors.get(), 1);
assert_eq!(comp_stats.processing.get(), 2);
}

#[test]
fn merge_combines_component_stats() {
let base = UpdateStats::default();
let delta = UpdateStats::default();

{
let mut base_components = base.by_component.write().unwrap();
let base_comp = base_components.entry("comp_a".to_string()).or_default();
base_comp.num_processed.inc(5);
base_comp.num_errors.inc(2);
base_comp.processing.inc(1);
}

{
let mut delta_components = delta.by_component.write().unwrap();
let delta_comp_a = delta_components.entry("comp_a".to_string()).or_default();
delta_comp_a.num_processed.inc(4);
delta_comp_a.num_errors.inc(3);
delta_comp_a.processing.inc(2);

let delta_comp_b = delta_components.entry("comp_b".to_string()).or_default();
delta_comp_b.num_processed.inc(7);
delta_comp_b.num_errors.inc(0);
delta_comp_b.processing.inc(5);
}

base.merge(&delta);

let components = base.by_component.read().unwrap();
let comp_a = components.get("comp_a").expect("comp_a should exist after merge");
assert_eq!(comp_a.num_processed.get(), 5 + 4);
assert_eq!(comp_a.num_errors.get(), 2 + 3);
assert_eq!(comp_a.processing.get(), 1 + 2);

let comp_b = components.get("comp_b").expect("comp_b should be added by merge");
assert_eq!(comp_b.num_processed.get(), 7);
assert_eq!(comp_b.num_errors.get(), 0);
assert_eq!(comp_b.processing.get(), 5);
}

#[test]
fn serialization_omits_empty_by_component() {
let stats = UpdateStats::default();

let value = serde_json::to_value(&stats).expect("serialization should succeed");
let obj = value.as_object().expect("stats should serialize to JSON object");

// When there are no per-component stats, by_component should be omitted if
// #[serde(skip_serializing_if = "HashMap::is_empty")] is used.
assert!(
!obj.contains_key("by_component"),
"by_component should be omitted when empty"
);
}

#[test]
fn serialization_includes_non_empty_by_component() {
let stats = UpdateStats::default();
{
let mut components = stats.by_component.write().unwrap();
let comp_stats = components.entry("comp_a".to_string()).or_default();
comp_stats.num_processed.inc(1);
comp_stats.num_errors.inc(0);
comp_stats.processing.inc(0);
}

let value = serde_json::to_value(&stats).expect("serialization should succeed");
let obj = value.as_object().expect("stats should serialize to JSON object");

let by_component_value = obj
.get("by_component")
.expect("by_component should be present when non-empty");

let by_component_obj = by_component_value
.as_object()
.expect("by_component should be a JSON object");

let comp_a_value = by_component_obj
.get("comp_a")
.expect("comp_a entry should be present");

let comp_a_obj = comp_a_value
.as_object()
.expect("component stats should serialize to JSON object");

// We don't assert exact field names beyond presence to keep this robust to
// unrelated serialization changes, but we ensure some counters are present.
assert!(
comp_a_obj.contains_key("num_processed"),
"component stats should include num_processed"
);
}
}

/// Record component processing start.
pub fn component_start(&self, component_name: &str, count: i64) {
Expand Down Expand Up @@ -938,4 +818,123 @@ mod tests {

assert_eq!(op_stats.get_operation_in_process_count("test_op"), 10); // 15-5
}

#[cfg(feature = "persistence")]
#[test]
fn component_counters_update_correctly() {
let stats = UpdateStats::default();

// Start 3 items, complete 2, error 1 → all accounted for, in_process = 0.
stats.component_start("comp_a", 3);
stats.component_complete("comp_a", 2);
stats.component_error("comp_a", 1);

let components = stats.by_component.read().unwrap();
let comp_stats = components
.get("comp_a")
.expect("component stats should exist");
assert_eq!(comp_stats.num_processed.get(), 2);
assert_eq!(comp_stats.num_errors.get(), 1);
// processing: 3 started, 3 ended (2 complete + 1 error)
assert_eq!(comp_stats.processing.get_in_process(), 0);
}

#[cfg(feature = "persistence")]
#[test]
fn merge_combines_component_stats() {
let base = UpdateStats::default();
let delta = UpdateStats::default();

{
let mut base_components = base.by_component.write().unwrap();
let base_comp = base_components.entry("comp_a".to_string()).or_default();
base_comp.num_processed.inc(5);
base_comp.num_errors.inc(2);
base_comp.processing.start(1);
}

{
let mut delta_components = delta.by_component.write().unwrap();
let delta_comp_a = delta_components.entry("comp_a".to_string()).or_default();
delta_comp_a.num_processed.inc(4);
delta_comp_a.num_errors.inc(3);
delta_comp_a.processing.start(2);

let delta_comp_b = delta_components.entry("comp_b".to_string()).or_default();
delta_comp_b.num_processed.inc(7);
delta_comp_b.num_errors.inc(0);
delta_comp_b.processing.start(5);
}

base.merge(&delta);

let components = base.by_component.read().unwrap();
let comp_a = components
.get("comp_a")
.expect("comp_a should exist after merge");
assert_eq!(comp_a.num_processed.get(), 5 + 4);
assert_eq!(comp_a.num_errors.get(), 2 + 3);
assert_eq!(comp_a.processing.num_starts.get(), 1 + 2);

let comp_b = components
.get("comp_b")
.expect("comp_b should be added by merge");
assert_eq!(comp_b.num_processed.get(), 7);
assert_eq!(comp_b.num_errors.get(), 0);
assert_eq!(comp_b.processing.num_starts.get(), 5);
}

#[cfg(feature = "persistence")]
#[test]
fn serialization_omits_empty_by_component() {
let stats = UpdateStats::default();

let value = serde_json::to_value(&stats).expect("serialization should succeed");
let obj = value
.as_object()
.expect("stats should serialize to JSON object");

// When there are no per-component stats, by_component should be omitted.
assert!(
!obj.contains_key("by_component"),
"by_component should be omitted when empty"
);
}

#[cfg(feature = "persistence")]
#[test]
fn serialization_includes_non_empty_by_component() {
let stats = UpdateStats::default();
{
let mut components = stats.by_component.write().unwrap();
let comp_stats = components.entry("comp_a".to_string()).or_default();
comp_stats.num_processed.inc(1);
}

let value = serde_json::to_value(&stats).expect("serialization should succeed");
let obj = value
.as_object()
.expect("stats should serialize to JSON object");

let by_component_value = obj
.get("by_component")
.expect("by_component should be present when non-empty");

let by_component_obj = by_component_value
.as_object()
.expect("by_component should be a JSON object");

let comp_a_value = by_component_obj
.get("comp_a")
.expect("comp_a entry should be present");

let comp_a_obj = comp_a_value
.as_object()
.expect("component stats should serialize to JSON object");

assert!(
comp_a_obj.contains_key("num_processed"),
"component stats should include num_processed"
);
}
}
Loading