# Singleflight Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Add a Go-style `singleflight::Group` that deduplicates concurrent calls keyed by an arbitrary key, sharing one result among all duplicate callers.

**Architecture:** A `Group<K, V>` wraps `Arc<GroupInner>` (like `WaitGroup`) holding a `Mutex<HashMap<K, Arc<Call<V>>>>`. The first caller for a key is the *leader* and awaits its own future; concurrent callers are *followers* that wait on an `event_listener::Event` and clone the leader's result. A drop-guard guarantees that a cancelled or panicking leader removes its map entry and wakes waiters (which then retry), so no waiter can hang. `do_chan` spawns the work via `go::spawn` and returns a `chan::Receiver`.

**Tech Stack:** Rust (edition 2024), `event_listener` (already a dep), `async-channel` (via `go::chan`), `go::spawn`. Runtime-agnostic; tests run on tokio, async-std, and smol via the existing `run_async!` macro.

---

### Task 1: Core `Group` with `do_` and `forget`

**Files:**
- Create: `src/singleflight.rs`
- Modify: `src/lib.rs` (add `pub mod singleflight;`)
- Test: `tests/integration.rs` (append tests)

- [ ] **Step 1: Write the failing tests**

Append to `tests/integration.rs`:

```rust
// ===========================================================================
// Singleflight tests
// ===========================================================================

use std::sync::atomic::{AtomicUsize, Ordering};

#[test]
fn singleflight_single_call() {
    run_async!(async {
        let g = go::singleflight::Group::<&str, i32>::new();
        let (v, shared) = g.do_("k", async { 5 }).await;
        assert_eq!(v, 5);
        assert!(!shared, "a lone call is never shared");
    });
}

#[test]
fn singleflight_dedups_concurrent_calls() {
    run_async!(async {
        let g = go::singleflight::Group::<&str, i32>::new();
        let counter = std::sync::Arc::new(AtomicUsize::new(0));

        let mut handles = Vec::new();
        for _ in 0..10 {
            let g = g.clone();
            let counter = counter.clone();
            handles.push(go::spawn(async move {
                g.do_("key", async move {
                    counter.fetch_add(1, Ordering::SeqCst);
                    go::sleep(std::time::Duration::from_millis(100)).await;
                    7
                })
                .await
            }));
        }

        let mut shared_count = 0;
        for h in handles {
            let (v, shared) = h.await.unwrap();
            assert_eq!(v, 7);
            if shared {
                shared_count += 1;
            }
        }

        assert_eq!(
            counter.load(Ordering::SeqCst),
            1,
            "the underlying future must run exactly once"
        );
        assert!(shared_count >= 1, "followers must observe shared == true");
    });
}

#[test]
fn singleflight_distinct_keys_do_not_dedup() {
    run_async!(async {
        let g = go::singleflight::Group::<&str, i32>::new();
        let counter = std::sync::Arc::new(AtomicUsize::new(0));

        let g1 = g.clone();
        let c1 = counter.clone();
        let h1 = go::spawn(async move {
            g1.do_("a", async move {
                c1.fetch_add(1, Ordering::SeqCst);
                go::sleep(std::time::Duration::from_millis(50)).await;
                1
            })
            .await
        });

        let g2 = g.clone();
        let c2 = counter.clone();
        let h2 = go::spawn(async move {
            g2.do_("b", async move {
                c2.fetch_add(1, Ordering::SeqCst);
                go::sleep(std::time::Duration::from_millis(50)).await;
                2
            })
            .await
        });

        let (va, _) = h1.await.unwrap();
        let (vb, _) = h2.await.unwrap();
        assert_eq!(va, 1);
        assert_eq!(vb, 2);
        assert_eq!(counter.load(Ordering::SeqCst), 2);
    });
}

#[test]
fn singleflight_forget_lets_a_new_call_through() {
    run_async!(async {
        let g = go::singleflight::Group::<&str, i32>::new();
        let counter = std::sync::Arc::new(AtomicUsize::new(0));

        // Leader 1 is in flight (sleeping).
        let g1 = g.clone();
        let c1 = counter.clone();
        let h1 = go::spawn(async move {
            g1.do_("k", async move {
                c1.fetch_add(1, Ordering::SeqCst);
                go::sleep(std::time::Duration::from_millis(100)).await;
                1
            })
            .await
        });

        // Let leader 1 register in the map.
        go::sleep(std::time::Duration::from_millis(20)).await;

        // Forget the in-flight key so the next call does NOT dedup with it.
        g.forget(&"k");

        let g2 = g.clone();
        let c2 = counter.clone();
        let h2 = go::spawn(async move {
            g2.do_("k", async move {
                c2.fetch_add(1, Ordering::SeqCst);
                go::sleep(std::time::Duration::from_millis(100)).await;
                2
            })
            .await
        });

        let (v1, _) = h1.await.unwrap();
        let (v2, _) = h2.await.unwrap();
        assert_eq!(v1, 1);
        assert_eq!(v2, 2);
        assert_eq!(
            counter.load(Ordering::SeqCst),
            2,
            "forget must allow a second concurrent execution"
        );
    });
}

#[test]
fn singleflight_abandoned_leader_does_not_hang() {
    run_async!(async {
        let g = go::singleflight::Group::<&str, i32>::new();

        // Cancel the leader's future by letting a timeout elapse first.
        let r = go::timeout(
            std::time::Duration::from_millis(20),
            g.do_("k", async {
                go::sleep(std::time::Duration::from_millis(500)).await;
                1
            }),
        )
        .await;
        assert!(r.is_err(), "the leader future should have been cancelled");

        // A fresh call on the same key must still complete (no stuck entry).
        let (v, _) = g.do_("k", async { 42 }).await;
        assert_eq!(v, 42);
    });
}
```

- [ ] **Step 2: Run the tests to verify they fail**

Run: `cargo test --test integration singleflight 2>&1 | head -30`
Expected: FAIL — compile error `failed to resolve: could not find 'singleflight' in 'go'`.

- [ ] **Step 3: Create `src/singleflight.rs`**

Create `src/singleflight.rs` with this exact content:

```rust
//! Deduplicate concurrent calls keyed by an arbitrary key.
//!
//! Equivalent to Go's `golang.org/x/sync/singleflight`: while a call for a
//! given key is in flight, concurrent callers for the same key wait and receive
//! a clone of the single in-flight result instead of running the work again.

use std::collections::HashMap;
use std::future::Future;
use std::hash::Hash;
use std::sync::{Arc, Mutex};

use event_listener::Event;

struct CallState<V> {
    dups: usize,
    result: Option<V>,
}

struct Call<V> {
    event: Event,
    state: Mutex<CallState<V>>,
}

struct GroupInner<K, V> {
    calls: Mutex<HashMap<K, Arc<Call<V>>>>,
}

/// Deduplicates concurrent calls so that only one execution is in flight for a
/// given key at a time; duplicate callers receive a clone of the shared result.
pub struct Group<K, V> {
    inner: Arc<GroupInner<K, V>>,
}

impl<K, V> Clone for Group<K, V> {
    fn clone(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
        }
    }
}

impl<K, V> Default for Group<K, V> {
    fn default() -> Self {
        Self::new()
    }
}

enum Role<V> {
    Leader(Arc<Call<V>>),
    Follower(Arc<Call<V>>),
}

// Removes the in-flight entry and wakes waiters if the leader's future is
// dropped (cancelled) or panics before publishing a result. Owns a clone of the
// key (rather than borrowing it) so the `do_` future stays `Send` without
// requiring `K: Sync`.
struct LeaderGuard<'a, K: Eq + Hash, V> {
    inner: &'a GroupInner<K, V>,
    key: K,
    call: &'a Arc<Call<V>>,
    done: bool,
}

impl<'a, K: Eq + Hash, V> Drop for LeaderGuard<'a, K, V> {
    fn drop(&mut self) {
        if !self.done {
            self.inner.calls.lock().unwrap().remove(&self.key);
            self.call.event.notify(usize::MAX);
        }
    }
}

impl<K, V> Group<K, V> {
    /// Create an empty group.
    pub fn new() -> Self {
        Self {
            inner: Arc::new(GroupInner {
                calls: Mutex::new(HashMap::new()),
            }),
        }
    }
}

impl<K, V> Group<K, V>
where
    K: Eq + Hash + Clone,
    V: Clone,
{
    /// Execute `fut`, deduplicating against any concurrent call for `key`.
    ///
    /// The first caller for a key (the leader) runs `fut`; concurrent callers
    /// for the same key wait and receive a clone of the leader's result.
    /// Returns `(value, shared)`, where `shared` is `true` if the value was
    /// shared with at least one other concurrent caller.
    ///
    /// If the leader's future is cancelled (dropped) or panics before
    /// producing a value, the in-flight call is abandoned: waiting callers
    /// retry, and one of them becomes the new leader.
    ///
    /// # Examples
    ///
    /// ```
    /// use go::singleflight::Group;
    ///
    /// async fn dedup(group: &Group<&'static str, i32>) {
    ///     let (value, shared) = group.do_("key", async { 42 }).await;
    ///     assert_eq!(value, 42);
    ///     let _ = shared;
    /// }
    /// ```
    pub async fn do_<F>(&self, key: K, fut: F) -> (V, bool)
    where
        F: Future<Output = V>,
    {
        let mut fut = Some(fut);
        loop {
            let role = {
                let mut calls = self.inner.calls.lock().unwrap();
                match calls.get(&key) {
                    Some(call) => {
                        call.state.lock().unwrap().dups += 1;
                        Role::Follower(Arc::clone(call))
                    }
                    None => {
                        let call = Arc::new(Call {
                            event: Event::new(),
                            state: Mutex::new(CallState {
                                dups: 0,
                                result: None,
                            }),
                        });
                        calls.insert(key.clone(), Arc::clone(&call));
                        Role::Leader(call)
                    }
                }
            };

            match role {
                Role::Follower(call) => {
                    let listener = call.event.listen();
                    if let Some(v) = call.state.lock().unwrap().result.clone() {
                        return (v, true);
                    }
                    listener.await;
                    if let Some(v) = call.state.lock().unwrap().result.clone() {
                        return (v, true);
                    }
                    // Leader abandoned the call without a result; retry as a
                    // candidate leader.
                    continue;
                }
                Role::Leader(call) => {
                    let mut guard = LeaderGuard {
                        inner: &self.inner,
                        key: key.clone(),
                        call: &call,
                        done: false,
                    };
                    let value = fut
                        .take()
                        .expect("leader future already consumed")
                        .await;
                    let shared = {
                        let mut calls = self.inner.calls.lock().unwrap();
                        calls.remove(&key);
                        let mut state = call.state.lock().unwrap();
                        state.result = Some(value.clone());
                        state.dups > 0
                    };
                    guard.done = true;
                    call.event.notify(usize::MAX);
                    return (value, shared);
                }
            }
        }
    }

    /// Forget an in-flight key so the next `do_` call starts a fresh execution.
    ///
    /// Callers already waiting on the forgotten call still receive its result
    /// when the original leader finishes; callers arriving after `forget`
    /// create a new leader. Forgetting an absent key is a no-op.
    pub fn forget(&self, key: &K) {
        self.inner.calls.lock().unwrap().remove(key);
    }
}
```

- [ ] **Step 4: Wire the module into `src/lib.rs`**

In `src/lib.rs`, add the module declaration next to `pub mod chan;`. Change:

```rust
pub mod chan;
mod waitgroup;
mod select;
```

to:

```rust
pub mod chan;
mod waitgroup;
mod select;
pub mod singleflight;
```

- [ ] **Step 5: Run the tests on the default backend**

Run: `cargo test --test integration singleflight 2>&1 | tail -20`
Expected: PASS — all five `singleflight_*` tests pass.

- [ ] **Step 6: Run the tests on all three backends**

Run:
```bash
cargo test --test integration singleflight && \
cargo test --no-default-features --features rt-async-std --test integration singleflight && \
cargo test --no-default-features --features rt-smol --test integration singleflight
```
Expected: PASS on all three.

- [ ] **Step 7: Commit**

```bash
git add src/singleflight.rs src/lib.rs tests/integration.rs
git commit -m "feat: add singleflight Group with do_ and forget"
```

---

### Task 2: `do_chan` channel variant

**Files:**
- Modify: `src/singleflight.rs` (add `do_chan` to the `impl Group` block)
- Test: `tests/integration.rs` (append one test)

- [ ] **Step 1: Write the failing test**

Append to `tests/integration.rs`:

```rust
#[test]
fn singleflight_do_chan_yields_result() {
    run_async!(async {
        let g = go::singleflight::Group::<&str, i32>::new();
        let rx = g.do_chan("k", async { 99 });
        let (v, _shared) = rx.recv().await.unwrap();
        assert_eq!(v, 99);
    });
}
```

- [ ] **Step 2: Run the test to verify it fails**

Run: `cargo test --test integration singleflight_do_chan 2>&1 | head -20`
Expected: FAIL — `no method named 'do_chan' found`.

- [ ] **Step 3: Add `do_chan`**

In `src/singleflight.rs`, inside the `impl<K, V> Group<K, V> where K: Eq + Hash + Clone, V: Clone` block, add this method directly after `do_` (before `forget`):

```rust
    /// Like [`do_`](Self::do_) but runs the work on the active runtime and
    /// returns a receiver that yields `(value, shared)` exactly once.
    ///
    /// Requires `Send + 'static` because it spawns onto the runtime.
    pub fn do_chan<F>(&self, key: K, fut: F) -> crate::chan::Receiver<(V, bool)>
    where
        F: Future<Output = V> + Send + 'static,
        K: Send + 'static,
        V: Send + 'static,
    {
        let (tx, rx) = crate::chan::bounded::<(V, bool)>(1);
        let group = self.clone();
        crate::spawn(async move {
            let result = group.do_(key, fut).await;
            let _ = tx.send(result).await;
        });
        rx
    }
```

- [ ] **Step 4: Run the test on the default backend**

Run: `cargo test --test integration singleflight_do_chan 2>&1 | tail -10`
Expected: PASS.

- [ ] **Step 5: Run the full singleflight suite on all three backends**

Run:
```bash
cargo test --test integration singleflight && \
cargo test --no-default-features --features rt-async-std --test integration singleflight && \
cargo test --no-default-features --features rt-smol --test integration singleflight
```
Expected: PASS on all three.

- [ ] **Step 6: Commit**

```bash
git add src/singleflight.rs tests/integration.rs
git commit -m "feat: add singleflight Group::do_chan"
```

---

### Task 3: Documentation wiring

**Files:**
- Modify: `src/lib.rs` (add a bullet to the `## Primitives` list)

- [ ] **Step 1: Add the primitives bullet**

In `src/lib.rs`, in the `## Primitives` doc list, change:

```rust
//! - [`WaitGroup`] — wait for a group of tasks to complete
//! - [`sleep`] / [`timeout`] — async timers
```

to:

```rust
//! - [`WaitGroup`] — wait for a group of tasks to complete
//! - [`sleep`] / [`timeout`] — async timers
//! - [`singleflight::Group`] — deduplicate concurrent calls by key
```

- [ ] **Step 2: Verify the doc example and intra-doc link compile**

Run: `cargo test --doc 2>&1 | tail -15`
Expected: PASS — the `Group::do_` example compiles and runs; no broken-link warnings.

- [ ] **Step 3: Verify docs build cleanly**

Run: `RUSTDOCFLAGS="-D warnings" cargo doc --no-deps 2>&1 | tail -15`
Expected: SUCCESS — no warnings (the `[`singleflight::Group`]` link resolves).

- [ ] **Step 4: Commit**

```bash
git add src/lib.rs
git commit -m "docs: list singleflight::Group in crate primitives"
```

---

## Notes for the implementer

- **Why each caller holds its own `fut`:** followers keep their future unpolled. If the leader is cancelled/panics, a woken follower loops and may become the new leader, awaiting *its own* future. This is what prevents waiters from hanging on an abandoned leader.
- **No `MutexGuard` is held across `.await`** in `do_`. The follower locks only to clone the result (a complete statement) and awaits the `EventListener` with no lock held; the leader publishes its result inside a scoped lock block before notifying. This keeps the future `Send` for `do_chan`.
- **Timing in tests:** the concurrency tests use `go::sleep` (50–100 ms) to keep calls overlapping while followers join. These windows are generous; if a heavily loaded CI machine ever flakes, widen the leader sleep rather than shortening it.
