keywatch 0.1.1

Async channel that coalesces per-key Add/Delete updates with an optional cooldown.
Documentation

Key-aware update channel with cooldown coalescing and minimal diff delivery.

A keywatch channel helps keep a conceptual key -> value store on a receiver in sync with a sender by emitting only the minimal sequence of change events required for convergence. Updates are per-key and can be either Add(value) (insert / upsert) or Delete (removal). The channel aggressively coalesces redundant intermediate updates while still preserving enough ordering to let the receiver reconstruct the latest state.

Core behaviors:

  • Two kinds of Add coalescing:
    1. Pre-delivery coalescing: multiple Adds for the same key sent before the first one is received collapse so only the last value is ever seen (even with zero cooldown). This keeps the pending set minimal when the receiver is idle or busy elsewhere.
    2. Cooldown coalescing: after an Add is delivered, further Adds within the cooldown window are withheld (only the most recent retained) and then emitted exactly once when the window elapses.
  • Deletes are never delayed by cooldown; they preempt any pending Add for their key.
  • Redundant Deletes are deduplicated: at most one Delete is emitted per present key until an Add recreates it; Deletes for unknown keys are suppressed.
  • Clear emits a single Delete for each key currently known to the receiver (and cancels any pending Add values in transit).
  • Ordering: Updates are yielded in FIFO order of their effective send time. Regular updates take their actual send call time; a cooled Add occupies a slot whose effective time is the cooldown expiry (later replacement Adds for that key before expiry only mutate the value in that slot). Thus matured cooled Adds appear in the order their cooldown slots were first created, interleaved with other keys' updates whose effective times arrive earlier.

Example distinguishing the two coalescing modes:

# #[tokio::main(flavor = "current_thread")] async fn main() {
use std::time::Duration;
use keywatch::{channel, Update};

// 1. Pre-delivery coalescing (zero cooldown): burst before first receive.
let (tx0, mut rx0) = channel::<&'static str, i32>(Duration::ZERO);
for v in 1..=5 { tx0.send("k", Update::Add(v)).unwrap(); }
// Only the last value (5) is observed immediately.
assert_eq!(rx0.recv().await, Some(("k", Update::Add(5))));

// 2. Cooldown coalescing (non-zero cooldown): subsequent burst after an observed Add.
let cooldown = Duration::from_millis(80);
let (tx, mut rx) = channel::<&'static str, i32>(cooldown);
tx.send("k", Update::Add(10)).unwrap();
assert_eq!(rx.recv().await, Some(("k", Update::Add(10))));
for v in 11..=15 { tx.send("k", Update::Add(v)).unwrap(); }
// The second value is withheld until cooldown expiry; measure elapsed.
let t_start = std::time::Instant::now();
let second = rx.recv().await.unwrap();
let elapsed = t_start.elapsed();
assert_eq!(second, ("k", Update::Add(15)));
assert!(elapsed >= cooldown);
# }