Greptile Summary

This PR fixes a bug where a lite streamer would exit due to the dormancy timeout while active follow subscribers were still connected. It introduces a FollowGuard RAII type and FollowReceiver wrapper so the run-loop can gate the dormancy branch on active_followers == 0, and a FollowerDropped wake-up message to re-arm the timer as soon as the last subscriber disconnects.

Key changes:

• Added AtomicUsize active_followers on Streamer and Arc-shared into each FollowGuard
• FollowGuard::new increments the counter; FollowGuard::drop decrements it and sends Message::FollowerDropped when it reaches zero
• FollowReceiver wraps broadcast::Receiver with a FollowGuard, so the lifecycle of the guard is tied to the subscriber lifetime
• The tokio::select! dormancy branch gains an if active_followers == 0 precondition, preventing idle shutdown while followers are alive
• run is refactored into run_with_dormant_timeout to accept an injectable timeout, enabling the new deterministic regression test
• Two style-level suggestions were left: (1) the regression test uses real wall-clock sleep delays instead of tokio's paused-time facilities, which could be flaky under CI load; (2) Ordering::Relaxed on the atomic operations is technically sound due to the happens-before chain provided by the mpsc channel, but Acquire/Release would be more self-documenting

Confidence Score: 4/5

• Safe to merge; the fix is logically correct with only minor style concerns around test timing and memory ordering semantics.
• The core logic is sound: the RAII guard pattern cleanly tracks follower lifetime, the FollowerDropped wake-up correctly re-arms the dormancy timer, and the regression test validates the intended behavior. The only concerns are non-blocking style suggestions (real-time test delays and Relaxed ordering that works but is implicit). No logic errors or race conditions were found.
• No files require special attention; both comments are style suggestions on lite/src/backend/streamer.rs.

Important Files Changed

Sequence Diagram

sequenceDiagram
    participant C as Client (read.rs)
    participant SC as StreamerClient
    participant S as Streamer (run loop)
    participant FG as FollowGuard

    C->>SC: follow(start_seq_num)
    SC->>S: Message::Follow
    S->>FG: FollowGuard::new() [active_followers += 1]
    S-->>SC: Ok(FollowReceiver { _guard, rx })
    SC-->>C: Ok(FollowReceiver)

    note over S: dormancy select branch<br/>disabled (active_followers != 0)

    C->>C: follow_rx.recv() [streaming records ...]

    C->>FG: drop(FollowReceiver)
    FG->>FG: fetch_sub → active_followers = 0
    FG->>S: Message::FollowerDropped (wake-up)
    note over FG: guard dropped

    S->>S: receive FollowerDropped (no-op handler)
    S->>S: loop top: load active_followers == 0
    S->>S: reset dormancy timer (dormant_timeout)
    note over S: dormancy branch now enabled
    S->>S: dormancy fires → break (streamer exits)

_{Last reviewed commit: 6bac2e8}