Watch notifications

Deep dive

How it works

Parts:

• revision counter — one global u64, +1 per write (per transaction in etcd), the only ordering
• event history — ring of recent (rev, key, PUT|DELETE, optional prev value), plus compacted (oldest gone)
• watcher registry — key or prefix, next_rev; kept in a synced and an unsynced group
• prefix or interval index — event key to its watchers without scanning all
• client cache — snapshot, last_rev, resume logic

1. Client lists the prefix; the server returns the items and revision R.
2. It opens watch(prefix, from = R + 1). The server answers Compacted if from <= compacted.
3. Live watchers (from beyond the head) are registered and get each new event from the index. Behind ones replay history first, then join them.
4. Client applies events in order, storing last_rev; a revision <= last_rev is dropped.
5. Disconnect: reopen at last_rev + 1; Compacted means clear the cache, go to 1.
6. Idle watchers receive periodic empty progress marks that move last_rev forward.

 write -> [rev++ | history ring ... compacted|  r+1 r+2 ... head]
                          index: prefix -> watchers
   list@R -------------> watch(from R+1) --events--> client (last_rev)

Code

main checks no gap or duplicate, resume, compaction and prefix filtering.

use std::collections::{BTreeMap, VecDeque};

#[derive(Clone, Debug, PartialEq)]
pub struct Event { pub rev: u64, pub key: String, pub val: Option<String> } // None = DELETE

#[derive(Debug, PartialEq)]
pub enum WatchErr { Compacted(u64) } // oldest revision still replayable is compacted + 1

pub struct Store {
    rev: u64,
    data: BTreeMap<String, String>,
    history: VecDeque<Event>, // ring of recent events, ascending rev
    cap: usize,
    compacted: u64, // events with rev <= compacted are gone
}
impl Store {
    pub fn new(cap: usize) -> Self {
        Store { rev: 0, data: BTreeMap::new(), history: VecDeque::new(), cap, compacted: 0 }
    }
    fn record(&mut self, key: &str, val: Option<String>) {
        self.rev += 1; // one revision per write
        self.history.push_back(Event { rev: self.rev, key: key.into(), val });
        while self.history.len() > self.cap {
            self.compacted = self.history.pop_front().unwrap().rev;
        }
    }
    pub fn put(&mut self, key: &str, val: &str) {
        self.data.insert(key.into(), val.into());
        self.record(key, Some(val.into()));
    }
    pub fn delete(&mut self, key: &str) {
        if self.data.remove(key).is_some() { self.record(key, None); }
    }
    /// Snapshot of a prefix plus the revision it is consistent with.
    pub fn list(&self, prefix: &str) -> (u64, BTreeMap<String, String>) {
        let m = self.data.range(prefix.to_string()..)
            .take_while(|(k, _)| k.starts_with(prefix))
            .map(|(k, v)| (k.clone(), v.clone())).collect();
        (self.rev, m)
    }
    pub fn watch(&self, prefix: &str, from_rev: u64) -> Result<Watcher, WatchErr> {
        if from_rev <= self.compacted { return Err(WatchErr::Compacted(self.compacted)); }
        Ok(Watcher { prefix: prefix.into(), next_rev: from_rev })
    }
}
pub struct Watcher { prefix: String, pub next_rev: u64 }
impl Watcher {
    /// Matching events with rev >= next_rev, in order. next_rev then jumps to
    /// store.rev + 1 even if nothing matched (a progress notification).
    pub fn poll(&mut self, s: &Store) -> Result<Vec<Event>, WatchErr> {
        if self.next_rev <= s.compacted { return Err(WatchErr::Compacted(s.compacted)); }
        let start = s.history.partition_point(|e| e.rev < self.next_rev);
        let out = s.history.iter().skip(start)
            .filter(|e| e.key.starts_with(&self.prefix)).cloned().collect();
        self.next_rev = s.rev + 1;
        Ok(out)
    }
}
/// Prefix index: an event key walks its own prefixes, not all watchers.
#[derive(Default)]
pub struct Hub { by_prefix: BTreeMap<String, Vec<usize>> }
impl Hub {
    pub fn add(&mut self, id: usize, prefix: &str) { self.by_prefix.entry(prefix.into()).or_default().push(id); }
    pub fn matching(&self, key: &str) -> Vec<usize> {
        let mut out = Vec::new();
        for i in (0..=key.len()).filter(|&i| key.is_char_boundary(i)) {
            if let Some(ids) = self.by_prefix.get(&key[..i]) { out.extend(ids); }
        }
        out
    }
}
fn apply(view: &mut BTreeMap<String, String>, evs: &[Event], last: &mut u64) {
    for e in evs {
        assert!(e.rev > *last); // strictly increasing: no duplicate, no reorder
        *last = e.rev;
        match &e.val { Some(v) => view.insert(e.key.clone(), v.clone()), None => view.remove(&e.key) };
    }
}
fn main() {
    let mut s = Store::new(1000);
    for i in 0..20 { s.put(&format!("/svc/{}", i % 7), &format!("v{i}")); s.put("/cfg/a", "x"); }
    let (r, mut view) = s.list("/svc/");
    s.put("/svc/new", "n"); s.delete("/svc/1"); s.put("/cfg/z", "other"); // between list and watch
    let mut w = s.watch("/svc/", r + 1).unwrap();
    s.put("/svc/2", "late");
    let (mut last, evs) = (r, w.poll(&s).unwrap());
    assert!(evs.len() == 3 && evs.iter().all(|e| e.key.starts_with("/svc/")));
    apply(&mut view, &evs, &mut last);
    assert_eq!(view, s.list("/svc/").1); // snapshot + stream == final state
    let seen = w.next_rev; // "disconnect", then resume from the last seen revision
    s.put("/svc/a", "1"); s.put("/svc/a", "2"); s.delete("/svc/a");
    let mut w2 = s.watch("/svc/", seen).unwrap();
    apply(&mut view, &w2.poll(&s).unwrap(), &mut last);
    assert_eq!(view, s.list("/svc/").1);
    assert!(w2.poll(&s).unwrap().is_empty()); // nothing replayed twice
    let mut t = Store::new(5); // tiny history: compaction is quick
    for i in 0..20 { t.put(&format!("/k/{}", i % 4), &i.to_string()); }
    assert_eq!(t.watch("/", 1).err(), Some(WatchErr::Compacted(15)));
    let (r, _) = t.list("/"); // recover: relist, then watch r + 1
    assert!(t.watch("/", r + 1).is_ok());
}

Complexity and constants

Write: O(1) amortized plus one event per watcher it matches. Opening a watch: O(log H) to locate from in history, then O(replayed events). Finding watchers: O(len(key) · log W) with a prefix map, O(W) scanning. Memory: history ring x event size, shared by all watchers; per watcher a cursor and a bounded send buffer. Toy, measured: replaying a 100k-event history to a prefix watcher took 0.28 ms.

Efficiency

• Cost: per event, one index lookup plus a push per matching watcher; the bound is send bandwidth to slow clients. Measured (Apple M4 Pro, 1,000 events): scan 1.6 µs / 15.6 µs / 159 µs per event at 1k / 10k / 100k distinct-prefix watchers; prefix index 0.5 / 0.9 / 2.1 µs.
• Levers:
  • range index instead of scanning watchers — 17x at 10k, 76x at 100k (measured) — many watchers on few keys each (etcd keeps range watchers in an interval tree, source).
  • one shared history, per-watcher cursor — memory independent of watchers (etcd watchableStore, source) — many readers of the same events.
  • batch events per revision (one message per transaction) — fewer sends and wakeups — multi-key writes.
  • slow watcher moves to catch-up: etcd gives each a 128-event channel, and a full one is parked and retried (source); coalesce or cancel — one slow client must not stall writes.
  • progress marks — resume stays inside the window instead of a re-list (etcd progress_notify; Kubernetes bookmarks) — rarely changing prefixes.
• Naive → standard → state of the art:
  • poll the whole range every N seconds · O(items) per poll per client, change latency up to N.
  • per-watcher queue, scan all watchers per event · O(W) per event, memory per watcher.
  • shared history, interval or prefix index, synced and unsynced groups, progress marks · O(log W + matches) per event.

Implementation notes

• Make the revision and the data one atomic step (MVCC read at R); a list that is not tied to R breaks the join.
• Size history by time: it must cover a reconnect plus a restart; a longer window costs memory, a shorter one costs re-lists.
• After a compaction or failover every client re-lists at once: jitter reconnects (backoff).
• Keep per-watcher buffers bounded; decide at overflow: cancel with an error or coalesce to latest-per-key.
• Follower reads: serve a watch from a replica only if it has applied up to from (linearizable reads).

Pitfalls

• Watching from now after a list loses writes in between; always R + 1.
• A prefix watcher sees revision gaps (other keys wrote them); gaps are not loss.
• Edge-only consumers drift after a loss; react to state, not deltas (reconciliation).
• ZooKeeper classic watches are one-time triggers: events between firing and re-registering are lost, so re-read the node on each event.
• Consul blocking queries return the changed state, not events; reset the index if it goes backwards.

Variants and alternatives

In production

• Orchestration: the Kubernetes API server serves list and watch on resourceVersion; client-go informers and kube-rs watcher relist on 410 for controllers (reconciliation).
• Service discovery and config: etcd watches (server-side synced/unsynced groups); Consul blocking queries.
• Shared-state apps: lobby or document views can follow the same list-then-watch on their own revision counter (pattern).
• Config fleets and agents use watches as their change feed.

Implementations

Sources

• etcd, API guarantees and Watch API · watchable_store.go
• Kubernetes, API concepts: efficient detection of changes
• Apache ZooKeeper, Programmer's Guide: watches
• HashiCorp, Consul blocking queries