Signal & Subscription

Signal<T> and Subscription<T> are Datum's two state-cell primitives. Both hold a single latest value that any thread can read with a lock-free snapshot, and both expose the history of that value as an ordinary changes() Source. They share one core and differ in exactly one thing: what happens to a change when a subscriber is slower than the writer.

• Signal<T> — a coalesced feed. A slow subscriber sees the newest value and may skip intermediate ones. Writers are never held back. Mirrors FS2 SignallingRef and Tokio watch.
• Subscription<T> — a lossless feed. Every accepted change reaches every subscriber, in order, subject to an explicit overflow policy. Mirrors ZIO SubscriptionRef.

If you only skim one thing: use Signal when "latest wins", use Subscription when "every transition matters". The rest of this page is about making that choice well and using each one correctly.

This page goes deep on the two state cells. For the one-page tour of all four M9 primitives (Signal, Subscription, Channel, Topic), see Concurrency Primitives.

The mental model

Think of a whiteboard on a wall. Anyone walking past can glance at it and read the current value — instantly, without asking anyone's permission. That glance is get(): a synchronous, lock-free read of the latest snapshot. One person at a time updates the board; that's set() / update().

The difference between the two primitives is what a camera pointed at the board records:

• With a Signal, the camera only takes a photo when someone looks. If the value changed three times between two glances, the camera records only the latest — the intermediate states are gone. This is coalescing, and for a "current status" board it is exactly what you want.
• With a Subscription, the camera records every write in order. If a viewer falls behind, the writer either waits for them, or you have told the camera in advance that it may drop or fail — there is no silent loss.

When to reach for which

You want to track…	Losing an intermediate value is…	Use
Live config, feature flags	Fine — only the current setting matters	Signal
A health / readiness flag	Fine — you act on the current state	Signal
A progress gauge or metric	Fine — you read the latest number	Signal
Leader / current-owner state	Fine — stale intermediates are noise	Signal
An audit trail of state changes	A bug — every transition is a record	Subscription
A state machine others must replay exactly	A bug — skipping a state corrupts the replay	Subscription
Committed offsets / sequence numbers	A bug — a skipped offset is lost work	Subscription

How these differ from hubs and queues

Datum already ships dynamic operators for element movement. State cells solve a different problem:

• MergeHub / BroadcastHub and Channel<T> move elements from producers to consumers. Each element is a discrete item of work. There is no "current value" to read out-of-band.
• Signal / Subscription model shared state. There is always a current value — even with zero subscribers — and get() reads it without touching the stream at all. The changes() feed is a view of that state over time, seeded with the current value on subscribe.

Rule of thumb: if a reader would ever ask "what is the value right now?" outside of a stream, you want a state cell. If everything is "process each item as it flows by", you want a hub, a Topic, or a Channel.

Why they are built this way

Two distinct types, not one subscribe(mode)

Coalesced and lossless are not two settings of one primitive — they have sharply different memory, backpressure, and failure semantics. A coalesced feed is a single conflating slot and never backpressures the writer. A lossless feed is a bounded per-subscriber buffer that must do something under pressure — wait, drop, or fail. Hiding that behind one innocent-looking subscribe() call would let a caller pick "lossless" without realizing they had also picked "my writer can now be blocked by a slow reader." The M9 design deliberately makes it two named types so the trade-off is visible at the call site (maintainer-confirmed after review: both are wanted, as different access options for real use cases).

The two-plane rule (why writes are cheap)

Both primitives use a two-plane design:

• The actor control plane (a Ractor actor) owns only the low-rate, correctness-critical work: subscribe, unsubscribe, close, terminal delivery, and publishing the subscriber registry as an ArcSwap snapshot.
• The lock-free data plane carries the hot path: set / update run on the caller thread, publish the ArcSwap value mirror, wake any parked subscribers, and return. get() is a plain ArcSwap load.

This matters because an actor message is not free. Ractor boxes every delivered message, and Datum's own diagnostic (measured over 100,000 transitions, in the M9 benchmark record) puts a full actor round-trip — enqueue + wake + drain a no-op handler — at 435 ns and ~999 B per message. The two-plane caller-thread write measures 192 ns and 56 B with no subscribers. If set sent an actor message per write, that ~1 KB allocation would land on the hottest path, once per transition, times the fan-out. Because state changes happen at transition rate (not element rate), keeping the actor for control only — and never for the per-write path — is what makes Datum's signal propagation faster than FS2 and ZIO on both wall-clock and CPU across the whole 1→1024 sweep.

The trade-off you accept: there is still an actor behind each primitive. Subscribe/close pay one control-plane message (a few hundred nanoseconds), and if you drop the last handle the actor stops and any live changes() feed fails with StreamError::ActorTerminated. In practice this is invisible — you keep a handle for as long as the feeds run — but it is why these are handles, not plain values.

The contracts, versus FS2 and ZIO

• get() never blocks and never allocates a message. It is ArcSwap::load_full() — a synchronous, immutable Arc<T> snapshot. This matches Tokio watch::borrow() and is cheaper than a Ref.get that has to run through an effect system.
• No version counters, no MVCC wrapper. ZIO-style designs sometimes expose a versioned value to close a get-then-subscribe race in poll-based APIs. Datum's delivery is push-based and the actor serializes subscribe against the registry, so there is no such race to close — subscribe() publishes the slot-table snapshot before the feed can pull, and the first pull reads the latest global snapshot. The version counter was deliberately not added (YAGNI: it only helps a poll API that does not exist).
• T is logically immutable. A new state is a new Arc<T>; old readers keep their old snapshot until they drop it, so reads are never torn. Datum cannot stop you from putting a Mutex or a Cell inside T, but you should treat any value you place in a state cell as immutable. Mutating shared interior state defeats the snapshot model.

Gentle examples

Every snippet below is imported from a cargo test–verified integration test.

A Signal: reads and a coalesced feed

Start with the whole Signal surface in one place — construct, read, write, subscribe, close:

use datum::{Signal, testkit::TestSink};

// A Signal is a latest-value state cell. It starts at an initial value.
let status = Signal::new("idle").unwrap();

// Reads are synchronous, lock-free snapshots — no actor message.
// `get()` returns an owned `Arc<T>`; `get_cloned()` clones the value.
assert_eq!(*status.get(), "idle");
assert_eq!(status.get_cloned(), "idle");

// An acked `set()` publishes before it returns, so you read your own write.
status.set("running").unwrap();
assert_eq!(status.get_cloned(), "running");

// `changes()` is an ordinary Source: the current value, then a *coalesced* feed.
let sink = status.changes().run_with(TestSink::probe()).unwrap();
sink.request(1);
sink.assert_next("running"); // a subscriber always starts at the current value

// A slow subscriber sees the newest value and may skip intermediate states.
status.set("draining").unwrap();
status.set("stopping").unwrap();
sink.request(1);
sink.assert_next("stopping"); // "draining" was coalesced away

// Closing re-emits the final value, then completes the feed.
status.close_with("stopped").unwrap();
sink.request(2);
sink.assert_next("stopped");
sink.expect_complete();

Two things to notice. First, changes() starts at the current value ("running"), not at the value the signal was born with — a subscriber is never blind to the present. Second, "draining" is coalesced away: the feed jumps straight to "stopping" because that write landed before the subscriber pulled. For a status board, that is correct behavior, not lost data.

A Signal as a gauge, read by non-stream code

A state cell is useful even with no subscribers at all. Here a Signal is pure shared state — a worker advances it, and ordinary (non-stream) code reads the latest value. This one uses the prelude glob import instead of naming each type:

use datum::prelude::*;

// A progress gauge: a Signal used purely as shared mutable state.
let progress = Signal::new(0_u32).unwrap();

// A worker advances it with `update()` — a CAS-style read-modify-write.
for _ in 0..3 {
    progress.update(|done| done + 1).unwrap();
}

// Non-stream code reads the gauge with a lock-free snapshot: no subscription,
// no actor hop, callable from anywhere that holds a clone of the handle.
assert_eq!(progress.get_cloned(), 3);

update() is a compare-and-swap read-modify-write: it reads the current value, applies your function, and installs the result. If a concurrent writer wins the race, your closure is re-invoked on the newer value — the same contract as an atomic fetch_update. Prefer update() over get()-then-set() whenever the new value depends on the old one.

A Subscription: every accepted change

Subscription has the identical read side, but its feed is lossless. Note the constructor takes a capacity and an overflow policy. Because the feed is an ordinary Source, it composes with filter / map / fold like any other:

use datum::{Sink, Subscription, SubscriptionOverflow};

// A Subscription is the same state cell, but its feed is *lossless*: with the
// Backpressure policy, every accepted change is delivered in order, no gaps.
let offset = Subscription::new(0_u64, 16, SubscriptionOverflow::Backpressure).unwrap();

// The read side is identical to Signal: acked writes are read-your-writes.
offset.set(1).unwrap();
assert_eq!(offset.get_cloned(), 1);

// `changes()` is an ordinary Source and composes with map/filter/fold.
let completion = offset
    .changes()
    .filter(|value| value % 2 == 0)
    .map(|value| value * 10)
    .run_with(Sink::collect())
    .unwrap();

offset.set(2).unwrap();
offset.set(3).unwrap();
offset.close_with(4).unwrap();

// Seed 1, then every change (2, 3, 4) — kept even values, each scaled by 10.
// Nothing is coalesced away: 2 and 4 survive the filter deterministically.
assert_eq!(completion.wait().unwrap(), vec![20, 40]);

Contrast this directly with the Signal example: here 2 is not coalesced away. Under Backpressure, the feed carries the seed and every accepted change in total order (1, 2, 3, 4), so the even-filter deterministically keeps 2 and 4. That determinism is the whole point of choosing Subscription.

Using them properly

get() snapshot semantics

get() returns an owned Arc<T> — cheap to hold and share, and shared by reference-count with the value currently in the cell. get_cloned() (available when T: Clone) clones the value out through ArcSwap's guarded load path, avoiding an Arc refcount bump; prefer it in scalar hot loops (u64, bool, small Copy types). Either way, the value you get is a snapshot: it is a consistent view of one moment, and it can be stale the instant you hold it if another writer publishes. Do not build check-then-act logic on get() + set() — that is a race. Use update() for atomic read-modify-write.

Acked set() vs fire-and-forget set_eventually()

• set() / update() are acked: they publish the new value to the mirror before returning, so you get read-your-writes — a get() on the same thread immediately afterward sees your write (as signal-basics above asserts). They also return StreamResult<()>, so a write to a closed cell surfaces as an Err you can handle with ?.
• set_eventually() / update_eventually() are the fire-and-forget names, contracted as eventually visible. In today's two-plane build they also publish on the caller thread before returning, but you should not depend on synchronous visibility through them — reach for the acked set() when you need read-your-writes or immediate error reporting.

Choosing a Subscription overflow policy

A lossless feed with a bounded buffer must do something when a subscriber falls a full buffer behind. That "something" is the policy you pass to Subscription::new (capacity must be > 0, or the constructor panics):

Policy	Under pressure	Lossless?	Choose it when
SubscriptionOverflow::Backpressure	The writer parks (on its own thread) until every active subscriber has ring capacity	Yes	Every change must be delivered and it is acceptable for a slow reader to slow the writer
SubscriptionOverflow::DropNew	The transition is applied, but a full subscriber skips the new feed item	No	The state must advance and dropping is an explicit, understood part of the contract
SubscriptionOverflow::Fail	The transition is applied, a full subscriber's feed fails after its accepted items drain, and the writer gets an Err	No	Overflow is a fault you want surfaced loudly rather than absorbed

Backpressure is the default mental model for "lossless". The key thing to internalize: it makes a slow subscriber able to backpressure the writer. If your writer must never block — an unbounded write rate feeding a slow subscriber — then Backpressure will park it. In that situation you must consciously choose to lose data (DropNew) or to fail (Fail); you cannot have all three of "unbounded writes", "lossless", and "never block the writer" at once.

Here is DropNew engaging deterministically. The subscriber reads the seed, then stops pulling; a second change arrives while its 1-slot buffer is full and is dropped for that subscriber only:

use datum::{Subscription, SubscriptionOverflow, testkit::TestSink};

// A capacity-1 buffer with DropNew: a full subscriber skips the newest change
// instead of backpressuring the writer.
let cell = Subscription::new(0_u64, 1, SubscriptionOverflow::DropNew).unwrap();
let sink = cell.changes().run_with(TestSink::probe()).unwrap();

// The subscriber starts at the current value but does not pull again yet.
sink.request(1);
sink.assert_next(0);

// `1` fills the 1-slot buffer; `2` arrives full, so DropNew skips it here.
cell.set(1).unwrap();
cell.set(2).unwrap();
cell.close_with(9).unwrap();

sink.request(3);
sink.assert_next(1); // the buffered change
sink.assert_next(9); // the final value — `2` was dropped for this subscriber
sink.expect_complete();

Under DropNew and Fail, other subscribers that are keeping up are unaffected — overflow is evaluated per subscriber.

Pitfalls

• Do not use Signal where losing an intermediate value is a bug. Coalescing is silent by design. Audit logs, event sourcing, exact offset tracking, and replayable state machines need every transition — use Subscription. (Conversely, do not reach for Subscription for a status flag: you pay for a bounded buffer and a backpressure decision you did not need.)
• Backpressure couples writer speed to the slowest subscriber. That is the price of lossless. If that coupling is unacceptable, pick DropNew/Fail and accept non-lossless — do not expect a fourth option.
• Keep a handle alive while feeds run. Dropping the last Signal/Subscription handle stops the control actor; any live changes() feed then completes with StreamError::ActorTerminated.
• After close(), writes return Err. A subscriber that arrives after close deterministically receives the final snapshot and then completes — closing is graceful, not abrupt.
• Treat T as immutable. Interior mutability inside T bypasses the snapshot model and reintroduces the data races these primitives exist to remove.
• Dropping a changes() source unsubscribes cleanly — cancelling the stream (e.g. via a KillSwitch or by dropping the completion handle) removes that subscriber and, for a backpressured Subscription, frees the writer.

Integration tips

Because changes() is an ordinary Source and get() is callable from anywhere, state cells drop into the rest of Datum without special ceremony:

• Source / Flow / Sink pipelines. signal.changes() / subscription.changes() are sources you can .map, .filter, .fold, throttle, or feed into any graph. Any operator closure can also call get_cloned() directly to read live state while processing an unrelated element stream.
• Graphs and junctions. Wire a changes() feed into a Merge, Zip, or Broadcast like any other inlet — for example, zip a data stream against a live-config feed so each element carries the config in effect when it was processed.
• Actors. A Signal<bool> makes a natural readiness/health flag between an ActorFlow pipeline and its supervisor: the actor side calls set(false) on a fault, and any watcher reads it with a lock-free get() or reacts on changes().
• KillSwitches. A watcher subscribed to a health Signal<bool> can call switch.shutdown() when the flag flips — pairing a state cell with a KillSwitch gives you a clean, observable stop signal for a running graph.
• datum-net. State cells are in-process handles; to share state across a process boundary, feed a changes() source into a network carrier or StreamRef exactly as you would any other source.

The most common integration is a running pipeline that reads live config. The filter closure reads get_cloned() on every element — a lock-free snapshot that always reflects the latest published value, with no rewiring when the config changes:

use datum::{Signal, Sink, Source};

// A live config cell that a running pipeline reads on every element.
let min_level = Signal::new(2_i64).unwrap();

// The operator closure reads `get_cloned()` — a lock-free snapshot, always the
// latest published value. No rewiring is needed when the config changes.
let cfg = min_level.clone();
let first = Source::from_iter(0_i64..5)
    .filter(move |level| *level >= cfg.get_cloned())
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();
assert_eq!(first, vec![2, 3, 4]);

// Publish new config; the next materialization observes it immediately.
min_level.set(4).unwrap();
let cfg = min_level.clone();
let second = Source::from_iter(0_i64..5)
    .filter(move |level| *level >= cfg.get_cloned())
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();
assert_eq!(second, vec![4]);

Performance

Numbers below come from the same-host M9 comparison record in roadmap/benchmarks/concurrency-primitives.md (captured 2026-07-02). The comparison target is the best JVM row per scenario and sweep point, not one fixed library. Datum passes every frozen wall-clock and CPU target for both state cells.

Scenario family	Sweep	Best JVM target	Datum result
signal_get — hot-path current-value read	1→64 readers	ZIO SubscriptionRef	3.59–7.16× faster wall; 5.61–6.15× lower CPU
signal_propagation — set→observed, coalesced feed	1→1024 subscribers	FS2 SignallingRef.discrete (wall)	2.42–93.49× faster wall; 2.09–343.24× lower CPU
subscription_lossless — every-change delivery	1→256 subscribers	FS2 at 1 sub, ZIO SubscriptionRef.changes at 16→256	1.02–47.23× faster wall; 1.67–271.70× lower CPU

The gap widens with fan-out: at 256 coalesced subscribers a Datum set propagates in ~2.1 ms versus FS2's ~195 ms, at a fraction of the CPU. This is the two-plane rule paying off — subscriber delivery is lock-free and writers scan the subscriber table only when a subscriber is actually parked. The full record includes p50/p99, allocation, peak RSS, correctness counters, and the write-path diagnostic behind each row.

Next steps

• Concurrency Primitives — the one-page tour of all four M9 primitives and how they relate.
• Source, Flow & Sink — the operator surface changes() plugs into.
• Dynamic Streams — KillSwitches and the hubs these state cells complement.
• Backpressure — how Subscription::Backpressure fits Datum's demand model.