Contributions from Recoco maintainer 🦀 : 2 bugs in stats.rs + ergonomics improvements

[bashandbone]

Hi! I'm the maintainer of Recoco, a pure-Rust fork of CocoIndex that we publish on crates.io. We track the v1 branch closely — implementing upstream changes as they land, occasionally adapting them for a no-Python environment.

As part of implementing PR #1767 (progress watching API) on our side, I did a detailed Rust-to-Rust comparison of our crates/core against your rust/core. I found two correctness bugs introduced in #1767, a few ergonomics improvements we made that might be worth upstreaming, and some trivial code-quality items.

I'd like to contribute patches for all of this if you are interested. I'll start with the bugs, then follow with individual issues/PRs for the rest if you're interested. Happy to take guidance on your preferred process — file issues first, open PRs directly, etc.

Bug 1: notify_terminated() doesn't persist the termination sentinel

File: rust/core/src/engine/stats.rs
Introduced in: #1767

notify_terminated() sends TERMINATED_VERSION to the watch channel but never writes it into the Mutex<VersionedProcessingStats>. As a result:

Callers who poll via stats.snapshot() or stats_snapshot() (rather than subscribing to the watch channel) never see that the task has terminated — they continue observing the last real version number.
A late update() call after notify_terminated() succeeds silently, mutating the stats after the termination signal was sent. Watch subscribers would then see a real version number again after having seen TERMINATED_VERSION, which violates the sentinel semantics.

// Current: only the watch channel gets TERMINATED_VERSION
pub fn notify_terminated(&self) {
    let _ = self.version_tx.send(TERMINATED_VERSION);
    // inner mutex is never updated ← bug
}

// A late update() after notify_terminated() succeeds:
stats.notify_terminated();
stats.update("comp", |g| g.num_adds += 1);  // no guard → mutates state
assert_eq!(stats.snapshot().version, TERMINATED_VERSION);  // FAILS: returns 1

Fix: Write TERMINATED_VERSION into the mutex inside notify_terminated(), and add an early-return guard in update():

pub fn notify_terminated(&self) {
    let mut guard = self.inner.lock().unwrap();
    guard.version = TERMINATED_VERSION;           // persist to mutex
    let snap = guard.clone();
    drop(guard);
    let _ = self.version_tx.send(TERMINATED_VERSION);
}

pub fn update(&self, operation_name: &str, mutator: impl FnOnce(&mut ProcessingStatsGroup)) {
    let mut guard = self.inner.lock().unwrap();
    if guard.version == TERMINATED_VERSION {       // guard against late updates
        return;
    }
    // ... rest unchanged
}

Bug 2: Watch channel carries only a version number — stats+version are not read atomically (TOCTOU)

File: rust/core/src/engine/stats.rs
Introduced in: #1767

The watch channel sends a u64 version. To read the actual stats, a subscriber must call stats.snapshot() as a separate step. Between rx.changed().await and snapshot(), more update() calls may have run, so the version the watcher saw no longer corresponds to the stats it reads:

// Subscriber:
rx.changed().await.unwrap();
let version = *rx.borrow();          // e.g. 3
let snap = stats.snapshot();         // another update ran → snap.version is now 5
// version ≠ snap.version — incoherent read

The original UpdateHandle::stats_snapshot() returns a VersionedProcessingStats from the mutex (version + stats together), so that call-site is fine — but anyone subscribing via the watch API and then polling stats is exposed to the race.

Fix: Send VersionedProcessingStats (or an equivalent snapshot) through the watch channel instead of just u64. Subscribers then get a coherent (version, stats) pair in one rx.borrow() call, with no auxiliary lock needed:

type WatchPayload = VersionedProcessingStats; // stats + version together

// In update():
let snap = guard.clone();
drop(guard);
let _ = self.version_tx.send(snap);          // full snapshot in channel

// Subscriber:
rx.changed().await.unwrap();
let snap = rx.borrow().clone();   // coherent: snap.version always matches snap.stats

(The extra clone cost on the watch send is negligible at a 1-second reporting interval.)

Tests for both bugs

Both bugs are currently untestable because the relevant test cases are absent. I'd include these with the fix:

#[test]
fn test_snapshot_observes_termination() {
    let stats = ProcessingStats::new();
    stats.update("comp", |g| g.num_adds += 1);
    stats.notify_terminated();
    // Must see TERMINATED_VERSION via snapshot, not just via watch:
    assert_eq!(stats.snapshot().version, TERMINATED_VERSION);
}

#[test]
fn test_update_is_noop_after_termination() {
    let stats = ProcessingStats::new();
    stats.update("comp", |g| g.num_adds += 1);
    stats.notify_terminated();
    stats.update("comp", |g| g.num_adds += 100);  // should be silently ignored
    assert_eq!(stats.snapshot().stats["comp"].num_adds, 1);
    assert_eq!(stats.snapshot().version, TERMINATED_VERSION);
}

Other improvements (lower priority, happy to discuss)

Beyond the bugs, we made several smaller improvements while implementing this engine. Happy to contribute these as separate PRs if useful, or just leave them here as reference:

1. UpdateHandle → IntoFuture (app.rs): Implementing std::future::IntoFuture lets callers write handle.await? instead of handle.result().await?. Small ergonomics win, no breaking change.

2. UpdateHandle watch API (app.rs): Exposing a watch::Receiver<VersionedProcessingStats> directly on UpdateHandle (via a watch_stats() method) makes the subscribe-and-loop pattern idiomatic without requiring access to the raw ProcessingStats object.

3. DeclaredTargetState → DeclaredEffect (context.rs, execution.rs): The struct represents an action/effect to apply to a target, not a target state value. The rename reads more accurately at call sites throughout execution.rs. (Total: 2 files, ~8 occurrences.)

4. FnCallMemoEntry default via #[derive] (context.rs): Replace the manual Default impl with #[derive(Default)] + #[default] on the Pending variant — cleaner and self-documenting.

5. Minor idioms (stable_path.rs, target_state_path.rs): A few places use verbose UFCS (::method(...)) where the method syntax (value.method(...)) works fine. Also write!(f, "{}", uuid.to_string()) can be write!(f, "{}", uuid).

6. Batcher::new() in utils: We removed the explicit BatchQueue argument from Batcher::new() since it's an implementation detail callers shouldn't need to construct. If this is something you'd want, it would require a change in cocoindex_utils.

Thanks for CocoIndex — the incremental dataflow model is excellent, and we've enjoyed working with the v1 engine. Happy to prepare any of the above as PRs once we've synced on what's wanted.