# `concurrent/` — M9 stream-native concurrency primitives
This module hosts Datum's M9 concurrency primitives. Read the repository root
[`CLAUDE.md`](../../../../CLAUDE.md) and
[`roadmap/M9-concurrency-primitives.md`](../../../../roadmap/M9-concurrency-primitives.md)
before editing this module; M9 design rules are settled decisions.
## Layout
- `mod.rs` — module docs and re-exports only.
- `signal.rs` — `Signal<T>`: latest-value state cell, lock-free `get()`, caller-thread writes,
control-plane actor, and a coalesced `changes()` feed.
- `subscription.rs` — `Subscription<T>`: same state-cell core, caller-thread writes through a
sequence-claimed ring, with a bounded lossless-by-default `changes()` feed and explicit overflow
policy.
- `channel.rs` — `Channel<T>`: closeable bounded MPSC source over a lock-free `ArrayQueue` with
producer parking outside the hot path.
- `topic.rs` — `Topic<T>`: many-publisher/many-subscriber broadcast. A control actor owns
subscribe/unsubscribe/close and publishes `ArcSwap` subscriber snapshots; publishers claim a
single global sequence turn and enqueue directly into per-subscriber lock-free slots.
## Invariants
- `Signal`/`Subscription` use a Ractor actor only for the control plane: subscribe, unsubscribe,
close/terminal, and publishing registry snapshots through `ArcSwap`.
- `get()` is synchronous and lock-free: it is an `ArcSwap` mirror read, not an actor message.
Scalar hot loops should prefer `get_cloned()` to use `ArcSwap::load()`'s guard path and avoid
`Arc` refcount contention.
- State writes preserve the M9 ordering rule on the caller thread: apply transition, store mirror,
publish/deliver wake state, then return to the caller. Backpressured subscription writes park the
producer outside the actor.
- Subscribe publishes the slot-table snapshot before the feed can observe later values. For
`Signal`, slots read the global `mirror + published_sequence` snapshot and keep only a consumed
cursor plus parked flag; seed-then-overwrite is idempotent because the first pull reads the latest
global snapshot. For `Subscription`, the seeded sequence sets the subscriber cursor and later
values are consumed from the ring at `S + 1`.
- `update(f)` is a CAS-style read-modify-write; `f` may be re-invoked if the CAS loses a race.
- `changes()` returns an ordinary `Source` blueprint. Each materialization registers a fresh
subscriber slot with the actor and unregisters it on stream drop.
- Post-close subscriptions deterministically emit the current final snapshot and then complete.
- No public `Versioned<T>` wrapper is introduced in M9. Internal sequence numbers exist only to
order ring publication and subscribe seeding.
- `Topic::subscribe()` returns an ordinary `Source` blueprint. Each materialization registers a
fresh slot; each slot sees only elements published after that registration. Publishing with zero
subscribers succeeds and drops the element, matching FS2 `Topic`.
- `Topic` publish ordering is global. Publishers claim sequence turns before touching subscriber
slots, so every subscriber that receives two elements observes them in the same order.
- `Topic` overflow is per subscriber: `Backpressure` waits for every active slot in the publish
snapshot to have room; `Sliding` drops that slot's oldest queued element before enqueueing the
new one; `Dropping` skips the new element for full slots. Other subscribers are unaffected.
- Topic subscriber wakeups must stay edge-triggered: wake only on empty-to-non-empty or when the
subscriber has parked at the queue head; subscribers drain in bounded 256-element batches.