Data-rate shaping

Between "something produces data many times per second" and "the database, the UI, or a single-writer loop consume it" sit small concurrency problems every gateway and simulation tier hand-rolls — each ~100 lines that are easy to get subtly wrong (flush/publish races, shutdown drains, timer and ownership lifecycles). Elarion core ships them once, in the Elarion.Buffering namespace (ADR-0055, ADR-0069). Which one do I want?

All four are BCL-only and need no DI registration (construct them wherever the data path lives — typically an actor, gateway component, or the loop that owns the state). The whole family is loss-tolerant by contract: a crash, a full queue, or a failed flush loses at most recent in-flight data. Data that must not be lost belongs in a handler transaction and the outbox, never in these primitives.

WriteBehindBuffer<T> — batch samples to the database

var buffer = new WriteBehindBuffer<Measurement>(
    async (batch, ct) => await WriteBatchAsync(batch, ct), // e.g. ExecuteInsertAsync on a fresh scope
    new WriteBehindBufferOptions {
        MaxItems = 500,                          // flush when the batch is full…
        FlushInterval = TimeSpan.FromSeconds(5), // …or when the oldest item is this old
    },
    onFlushError: (ex, batch) => logger.LogWarning(ex, "Dropped {Count} samples", batch.Count));

buffer.Add(measurement);        // any thread, never blocks on the flush target
await buffer.FlushAsync(ct);    // explicit flush — delegate failures rethrow here
await buffer.DisposeAsync();    // stops the timer, flushes the tail, never throws

The contract is loss-tolerant samples:

• The buffer is bounded — past Capacity (default 4 × MaxItems) the oldest unflushed item is dropped, so a slow or down database bounds memory instead of growing it. DroppedCount meters the pressure; a climbing value means the flush target can't keep up.
• A failed flush drops its batch rather than retrying it (a poisoned batch must never wedge the pipeline). Explicit FlushAsync rethrows to its caller; background and dispose flushes route to the onFlushError callback — without one they are swallowed, so supply it in anything beyond a prototype.
• Flushes are single-flight: items added while a flush runs coalesce into the next drain pass, so a slow target gets fewer, larger calls — not a stack of concurrent ones.

Samples that must not be lost don't belong here — record them transactionally in the handler and let the outbox do the deferring.

KeyedConflater<TKey, TValue> — latest-wins to the UI

var conflater = new KeyedConflater<string, QuoteUpdated>(
    async (symbol, quote, ct) => await clientEvents.PublishAsync(quote, ClientEventScope.Global, ct),
    new KeyedConflaterOptions { MinInterval = TimeSpan.FromMilliseconds(250) });

conflater.Post("ACME", update);  // any thread, never blocks on the publish target
await conflater.DisposeAsync();  // publishes every pending latest, then drops late posts

Each key emits at most once per MinInterval:

• The first post of an idle key emits immediately (leading edge) — a dashboard shows the first value without waiting a window.
• Posts inside the window conflate: only the newest value survives, and it emits when the window elapses (trailing edge). A quiet key therefore always publishes its final value — conflation never ends on a stale reading.
• Emissions for one key never overlap: a publish slower than the window lowers the effective rate instead of stacking calls. Keys are independent, and idle keys retire automatically, so unbounded key spaces (device ids, symbols) don't leak.
• Delivery is at-most-once, matching the client-event contract: a publish failure drops that emission to the optional onPublishError callback, and the next post heals.

BoundedMpscQueue<T> — commands into a single-writer loop

When many producer threads (connection receive continuations, timers) feed one consumer thread (a simulation tick loop, a device writer), ConcurrentQueue<T> allocates segments and is unbounded, and Channel<T> allocates and is async-shaped. BoundedMpscQueue<T> is the third shape: a fixed-capacity, array-backed, lock-free struct queue (Vyukov's bounded algorithm) whose TryEnqueue and TryDequeue never block and never allocate — all slots are preallocated at construction, and capacity rounds up to a power of two:

var commands = new BoundedMpscQueue<MoveCommand>(4096);

// Producers — any thread, e.g. a connection receive continuation:
if (!commands.TryEnqueue(new MoveCommand(entityId, x, y)))
    _dropped++;                       // full = backpressure; this message class is loss-tolerant

// The single consumer — the tick loop:
while (commands.TryDequeue(out var command))
    Apply(command);

The contract:

• Bounded means backpressure, and the policy is yours. A full queue returns false — there are no blocking or waiting APIs. Decide per message class: drop loss-tolerant traffic (movement, telemetry), retry or fail must-land control messages. The queue also owns no loop or thread; the application drains it from whatever tick it already has.
• FIFO per producer. Cross-producer arrival order is slot-claim order (the order producers won the enqueue cursor), which under contention may differ from call order.
• Single consumer. TryDequeue must never run concurrently with itself — one consumer at a time (migrating the consumer between threads is fine as long as calls don't overlap). A debug-only assertion catches violations.
• Reference hygiene. T may carry object references (a command with a payload array); a dequeued slot is cleared so the queue never keeps them alive until the slot's next reuse. For pure value types the clear compiles away.

Composition: the actor telemetry path

The recommended owner is the actor that already serializes the hot state (the write-behind and live-telemetry use cases): it constructs both helpers as activation state, Adds/Posts from its turns, and disposes them in OnDeactivateAsync — one sample stream, two shaped consumers:

public Task RecordAsync(Measurement m) {
    _buffer.Add(m);               // → batched ExecuteInsertAsync every few seconds
    _conflater.Post(m.Sensor, m); // → at most 4 client events/s per sensor (MinInterval = 250 ms)
    return Task.CompletedTask;
}

These four shapes are deliberately the whole surface — windows, joins, or replay are the trigger for a real reactive/streaming library, not for growing these helpers. For ordered, gap-visible streams where element identity matters, use ordered streams instead of conflation.

Below the helpers: dirty-flag-and-sweep for simulation state

Both helpers assume an append-shaped sample stream: every reading is a new fact, produced at gateway rates — hundreds of keys, a few hertz each. A simulation loop is a different workload: thousands of entities mutating at tick rate, where only the latest state of each entity matters and the update path is allocation-sensitive. At that tier, even latest-wins helper machinery is overhead — WriteBehindBuffer queues every update, and KeyedConflater pays a dictionary hit and timer scheduling per key per window. Keep both off a per-update hot path.

The standard shape there is dirty-flag-and-sweep, in application code:

1. An update mutates the entity the loop already owns and sets Dirty = true — field writes, zero allocation, no per-update queue entry.
2. A periodic sweep (tick-aligned or interval) walks a stable iteration structure — a preallocated list or bitset of dirty indices, not LINQ — copies dirty state into a reused batch buffer, clears the flags, and hands the batch off for one batched write.
3. Explicit save points (a session ending, an entity unloading) call the same sweep directly.

The sweep itself allocates (rows, command, parameters) — at flush frequency, not update frequency, which is the point. A crash loses at most one interval: the same loss-tolerant contract as the rest of the family, so the same rule applies — state that must not be lost belongs in a handler transaction and the outbox, not in this loop.

StagedBatchFlusher<TBatch> — the generic half of the handoff

The dirty bit, the iteration structure, and the batch's shape are application business. What is generic — and easy to get subtly wrong — is the ownership/signaling protocol between the sweep and the background write: who may touch the batch when, how the writer signals it's done, what happens when the write throws, and who drains on shutdown. StagedBatchFlusher<TBatch> owns exactly that half and never inspects the batch:

• Ownership protocol. The producer stages into the batch only while IsIdle is true. TrySubmit transfers ownership to the background writer without blocking or allocating; when the write delegate returns or throws, ownership returns and IsIdle becomes true again — with full visibility of the writer's effects, so the producer can safely reuse the batch. An optional reset callback runs at each ownership return so the clear-the-sections logic lives in one place.
• Skip and retry. TrySubmit returning false means a batch is still in flight — stage nothing and let the dirty flags carry the state to the next sweep. That is why there is exactly one batch and no internal queue: flags already bridge a busy interval, so a second buffer would buy latency nobody needs at the cost of ownership complexity.
• Error policy. A throwing write drops that batch's staged content by contract (latest-wins makes this safe — the next sweep re-stages current values), surfaces through onFlushError, and never tears down the loop or leaks ownership.
• Shutdown drains. DisposeAsync completes the in-flight write and writes a batch submitted before disposal, uncancelled — the last state of departed entities — bounded by an optional DisposeTimeout. FlushAsync awaits idleness at explicit save points without disposing.

An actor is the natural owner of the hot state: updates and the sweep are both turns on one mailbox, so the sweep sees a consistent view with no locking, the IsIdle gate is race-free (only the mailbox submits), and the sweep trigger is a facade call from whatever tick the application already has — the simulation loop, a scheduled job, or a hosted service:

[SqlRecord("entity_positions")]
public sealed partial record EntityPositionRow {
    public required Guid Id { get; init; }
    public required float X { get; init; }
    public required float Y { get; init; }
}

// The batch and its sections are application code; the flusher never looks inside.
public sealed class PositionBatch {
    public List<EntityPositionRow> Rows { get; } = [];
}

[Actor]
public sealed class ZoneActor : IActorLifecycle {
    // Mutable by design: hot per-entity state, owned and serialized by the mailbox.
    private sealed class EntitySlot { public Guid Id; public float X; public float Y; public bool Dirty; }

    private readonly Dictionary<Guid, EntitySlot> _entities = [];
    private readonly PositionBatch _batch = new();          // the one preallocated, producer-owned batch
    private readonly StagedBatchFlusher<PositionBatch> _flusher;

    public ZoneActor(ISqlDatabase db, ILogger<ZoneActor> logger) {
        _flusher = new StagedBatchFlusher<PositionBatch>(
            async (batch, ct) => {
                await using var session = await db.OpenSessionAsync(ct);
                await session.InsertManyAsync(batch.Rows,
                    " ON CONFLICT (id) DO UPDATE SET x = EXCLUDED.x, y = EXCLUDED.y", ct);
            },
            onFlushError: (ex, batch) => logger.LogWarning(ex, "Dropped {Count} rows", batch.Rows.Count),
            reset: static batch => batch.Rows.Clear());
    }

    public Task Move(Guid id, float x, float y) {           // hot path: field writes only
        var slot = _entities[id];
        slot.X = x; slot.Y = y; slot.Dirty = true;
        return Task.CompletedTask;
    }

    public Task Sweep() {                                   // tick-aligned, e.g. every 30 s
        if (!_flusher.IsIdle) return Task.CompletedTask;    // writer busy — flags carry to the next sweep

        foreach (var slot in _entities.Values) {
            if (!slot.Dirty) continue;
            slot.Dirty = false;
            _batch.Rows.Add(new EntityPositionRow { Id = slot.Id, X = slot.X, Y = slot.Y });
        }
        if (_batch.Rows.Count > 0) _flusher.TrySubmit(_batch);
        return Task.CompletedTask;
    }

    public async ValueTask OnDeactivateAsync(CancellationToken ct) {
        await Sweep();                                      // stage whatever is still dirty…
        await _flusher.DisposeAsync();                      // …and drain it before the actor goes away
    }
}

The framework's contribution beyond the handoff is the flush target, per host tier:

There is still deliberately no DirtySweep<T> helper. The dirty bit and the iteration structure belong to the application's entity representation, and any framework intermediary between the update and the flag would reintroduce exactly the per-update bookkeeping this shape exists to avoid. New state families — even ones with maximally different persistence semantics, like replace-sets with deletions next to row updates — extend the batch, never the protocol: the framework's seams are the mailbox that serializes the hot state, the flusher that owns the handoff, and the batched write the sweep flushes into.