# Realtime watches
A watch is an **in-memory, at-most-once** subscription to a collection's
mutations. It is the complement of the durable change stream (T11/T12), and the
two exist for opposite reasons:
| Guarantee | at-least-once, survives a crash | at-most-once, dies with the process |
| Backlog when nobody reads | accumulates in a durable queue | nothing is produced at all |
| Cost with no consumer | a queue row per event | one `Acquire` load |
| Write path | becomes transaction-only | unchanged |
| Use it for | work that must happen | work that is worth doing *now* |
The durable half must not lose an event. The realtime half must not cost
anything when nobody is listening.
```rust
use armdb::{Config, NoHook, RapiraCodec, TypedTree, WatchHook, WatchKind};
let tree = TypedTree::open_hooked(
path,
Config::balanced().build(),
RapiraCodec,
WatchHook::<Row, NoHook, true>::with_old(NoHook),
)?;
let mut rx = tree.watch(Default::default());
tree.put(&key, Row(1))?;
let event = rx.recv_blocking()?;
assert_eq!(event.kind(), WatchKind::Insert);
assert_eq!(event.new()?, Some(Row(1)));
```
`WatchHook` wraps any hook, so a watch composes with TTL, projections, counters
and claims: `WatchHook::with_old(existing_hook)`. `new_only(...)` is the same
thing without old values — cheaper for the Var families, where retaining the old
value costs a read.
## What you get, and what you must do about it
Delivery is **at-most-once**, per subscriber, with a bounded ring
(`WatchOptions::capacity`, default 1024) and drop-oldest eviction. A dropped
event is never silent: the next `recv` returns `WatchError::Lagged(n)` at the
position of the gap, and the events on either side of it still arrive in order.
So the consumer recipe is fixed, and both halves are load-bearing:
1. **`watch()` first, then scan.** A receiver sees mutations from the moment it
registers; rows written earlier are not in the stream and must come from the
scan.
2. **On `Lagged`, scan again.** The stream is now missing rows, and only a scan
can tell you which.
```rust
let mut rx = map.watch(Default::default());
let mut mirror = scan(&map); // step 1
loop {
match rx.recv_blocking() {
Ok(event) => apply(&mut mirror, &map, event.key()),
Err(WatchError::Lagged(_)) => mirror = scan(&map), // step 2
Err(WatchError::Closed) => break,
Err(_) => unreachable!("blocking recv yields neither Empty nor Timeout"),
}
}
```
**An event is a pointer, not a snapshot.** Note what `apply` receives: the key,
not the payload. A rebuild that applies `event.new()` converges only while
writes have stopped — against a live writer, an older event's payload can
overwrite a newer row. Read the row and apply what you find. The payloads are
there for the cases where they *are* the answer (a delete's old value, an audit
log), not as a substitute for the row.
Because a Map's scan is `iter_view()`, **a watched Map must be opened with
`Config::iterable(true)`** — the default is `false`. It is refused at open with
`DbError::WatchRequiresIterable` rather than at `watch()`, so subscribing itself
cannot fail.
### Outcomes
| `Ok(event)` | the next event in order |
| `WatchError::Empty` | nothing queued right now (`try_recv` only) |
| `WatchError::Timeout` | deadline passed with nothing queued (`recv_timeout` only) |
| `WatchError::Lagged(n)` | `n` events were dropped at this position; not terminal |
| `WatchError::Closed` | the database shut down; the ring is drained first |
Precedence is by position, not by kind: a queued event outranks an expired
deadline, and a gap is reported when the reader reaches it, so a `Lagged` never
jumps ahead of events that survived it.
## Ordering
Within one shard, an event never precedes the visibility of its row: it is
published under the same lock that made the row visible. Per key, therefore,
events arrive in the order the mutations committed.
Across keys, and across collections, there is no ordering. Nor are transaction
boundaries visible: a `db.tx()` touching three collections delivers three
independent events, and a subscriber cannot tell they were one commit. If you
need that, you need the durable change stream.
Wakeups happen **after** every relevant lock is released — the shard lock for a
facade write, and for a transaction, after every participant has finished. A
consumer woken by an event can therefore read the collection immediately without
deadlocking, and a slow consumer can never hold up a writer.
## Cost
The design's premise is that a watch you are not using costs nothing:
- **No watch on the collection**: `WATCHES` is a compile-time `false`, and the
whole path is eliminated at monomorphisation.
- **A watch with no receiver**: one `Acquire` load on the subscriber count. No
allocation, no old-value read, no event.
- **A watch with a receiver**: one `Arc` per event, shared by every subscriber,
plus the old value's cost in `with_old` mode. For the Typed, Const and Zero
families the old value is in RAM; the Var families must read it.
`new_only` exists for that last line. Note that it saves nothing when the inner
hook already needs old values — a `Projected` carrier does — because
`NEEDS_OLD_VALUE` is the union of the two.
A subscription can never fail a write and never waits for consumer progress: a
full ring drops an event, it does not block. What it *does* cost the write path
when a receiver is live is the event build, inside the critical section.
## Lifecycle
`watch()` is infallible and may be called any number of times; each receiver is
independent, has its own ring, and is not `Clone`. Dropping a receiver
deregisters it; the last one leaving returns the collection to the
zero-subscriber fast path.
Subscriptions are owned by the **database**, not by the collection handle:
`Db::shutdown` — and therefore `Drop` — closes every channel, after the TTL
sweeper is joined so an in-flight sweep cannot publish into a closed channel.
`Db::close` is non-terminal and leaves receivers live. A closed receiver drains
what it already holds before it reports `Closed`.
There is one ordering guarantee around registration worth naming, because the
recipe above depends on it: when `watch()` returns, every write that decided not
to publish to this receiver has already made its row visible. That is what makes
"scan after subscribing" complete — a write is in the stream, or in the scan,
never in neither. The cost is that `watch()` briefly waits out the mutations
already in flight when it registers.
## Boundaries
- **Six typed managed families only** — `TypedTree`/`TypedMap`,
`ZeroTree`/`ZeroMap`, `VarTypedTree`/`VarTypedMap`. The byte-level
`ConstTree`/`ConstMap`/`VarTree`/`VarMap` openers take an untyped hook and
cannot decode an event; `watch()` does not exist for them.
- **`LsmTree` is rejected at open** with `DbError::WatchUnsupported`. It is not
a participant in the mutation seam, so an attached watch would silently never
emit.
- **A follower emits nothing.** Replication apply does not run the hook seam, so
a replica's rows change without events. Watch the leader.
- **Migration and crash replay emit nothing.** Both run at open, before the
collection is published, so no receiver can exist yet.
- **No cross-collection or cross-shard ordering, and no transaction
boundaries** — see [Ordering](#ordering).
- **No persistence, no replay, no backlog.** A subscriber that was not there
did not miss an event; there was no event.