Bound the rate-aware batcher backlog by SimonHeybrock · Pull Request #1271 · scipp/esslivedata

SimonHeybrock · 2026-08-31T10:03:36Z

Motivation

A window advances by at most one batch length per batch() call, so once a service iteration takes longer than the batch length the caller hands over more data than the window can release. The surplus accumulated in _overflow for the life of the process — unbounded, reachable memory that the cyclic collector cannot touch, while the Kafka consumer reported consumer_lag=0, is_healthy=True and batches_dropped=0 throughout, because the backlog sits downstream of the one buffer that reports.

This is what OOM-killed detector_data on TBL: an ~11 s Timepix3 update against the adaptive batcher's 8 s ceiling stranded ~330 messages per iteration, each pinning its ev44 payload through the numpy views in DetectorEvents. #1265 was deployed against the same OOM and did not stop it, because the growth is in live objects rather than cyclic garbage.

#1270 removes this particular capacity problem, but nothing bounds the backlog: any workflow slower than the maximum batch length reproduces the failure, silently.

What this does

Bounds the retained gated overflow and drops the stalest surplus — the same bounded-buffer policy BackgroundMessageSource already applies to the consumer queue (max_queue_size, drop oldest, count, log). Conceptually this is load shedding, with the control loop replaced by a constraint: a bounded buffer drops exactly the surplus, so it self-tunes to the shortfall instead of searching for a rate through discrete levels.

There are two bounds. The primary one is data time: each stream retains at most max(2 batch lengths, 10 s) of backlog — the seconds term is the burst tolerance (a delivery gap that hands over less than it in one poll is replayed in full rather than holed), the batch-length floor keeps ordinary jitter and the adaptive wrapper's escalated windows unshed. The secondary one is bytes (512 MB), a memory backstop for payload sizes the data-time bound cannot express.

Only GATED_STREAM_KINDS reach _overflow, so control, log and context data can never be shed — the selectivity is structural rather than stated.

Shedding keeps the newest messages, and the window reaches them through the ordinary gap-detection jump on the following call: the shed region is empty by construction, so no gated stream contributes to the window and the existing check fires. An earlier revision forced the jump on every shed, bypassing gap recovery's veto (no jump while a gated stream still has messages in the window); a scenario test showed that livelocks delivery when a byte-bound trim of one oversized stream coincides with a peer gating normally — area-detector frames alongside detector events in the same service is exactly that topology. The newest message survives unconditionally, even alone over the byte bound — the eventual jump needs a surviving message to anchor the new window on, so shedding can never empty the backlog.

The policy is stated as a "Freshness policy" section in the module docstring, alongside the existing clock policy, with a pointer from the glossary.

Why data time is the primary bound

A byte cap bounds memory but says nothing about how stale the retained backlog is, and the batcher only catches up at batch_length - iteration_time per call. Retaining a deep backlog therefore buys late data at the price of showing everything late for as long as the replay takes. Reproducing a 120 s stall at 14 MB/s, one shed event in every case:

policy	iteration	dropped	data hole	lag once recovered
bytes only	0.5 s	1162 MB	83 s	25 s
bytes only	0.9 s	1156 MB	83 s	315 s
both bounds	0.5 s	1645 MB	117 s	~2 iterations
both bounds	0.9 s	1639 MB	117 s	~2 iterations

Under the byte cap alone, recovery time depends on how much headroom the loop happens to have, and a loop running near its window stays minutes behind live. Operators steer beam, samples and detector commissioning from this output, where a stale reading is worse than an intermittent hole — and shedding at all is an exception state to fix upstream, not a mode to run in.

A byte budget also buys wildly different tolerances per stream kind (~20 s of ev44, ~1.5 s of area-detector frames); a bound in data time is uniform across instruments, and its batch-length floor follows the adaptive wrapper's escalated windows automatically.

Choosing the bound

The two terms answer different failure modes. The seconds term (10 s) is sized against delivery gaps, not processing speed: a consumer-group rebalance alone takes seconds, so a bound of two batch lengths would turn every ordinary infrastructure blip into a permanent hole. Retaining a within-bound burst is nearly free — an idle service recovers it by a gap jump plus timeout closes at poll rate, not by the slow crawl of an overloaded loop — so the price of the tolerance is paid only under sustained overload beyond what escalation absorbs, where the steady lag approaches the bound. No pre-production statistics on gap durations exist; max_backlog_s and the drop counters are the instrument for tuning the default.

The batch-length floor (2) exists because the overflow legitimately holds about one batch length of in-transit next-window traffic: a fixed seconds bound below the escalated batch length would shed healthy traffic, so the floor takes over at escalated windows (16 s at the 8 s ceiling).

The floor can stay small because loss does not depend on it. Sweeping the bound against overload severity (14 Hz, 200 iterations, loss = fraction of fed data shed, lag = steady window-to-live distance):

iteration	N=2	N=3	N=5	unbounded
1.00 s	0.00% / 0.07 s	0.00% / 0.07 s	0.00% / 0.07 s	0.00% / 0.07 s
1.05 s	0.27% / 1.6 s	0.23% / 2.5 s	0.17% / 4.6 s	0% / 7.6 s
1.10 s	0.60% / 1.6 s	1.13% / 2.7 s	1.00% / 4.7 s	0% / 15 s
1.20 s	3.35% / 1.7 s	3.26% / 2.7 s	1.03% / 4.5 s	0% / 30 s
2.00 s	24.8% / 1.6 s	24.5% / 2.6 s	24.3% / 4.6 s	0% / 150 s

Data loss is set by the overload ratio, not by the bound — it is the surplus the window cannot release, and it barely moves between N=2 and N=5. The bound only sets the lag, proportionally. A small value therefore buys freshness almost for free. The unbounded column's lag is still climbing at iteration 200; it does not converge.

Nothing is shed at all while the loop keeps up, and an iteration that merely overshoots its window sheds nothing either. With the adaptive wrapper in the loop, a sustained 1.1x overload escalates the window to 1.43 s and sheds nothing — escalation gets there first, which is the intended division of labour.

Why the horizon is per stream

The horizon is anchored on each stream's own frontier, not on the backlog as a whole. A single global horizon conflates backlog depth with the offset between streams: a stream stamped ahead of its peers sets the horizon for all of them and sheds their fresh data as if it were stale. With a 2.5 s offset between two streams under 2x overload:

policy	lost (stream A)	lost (stream B, 2.5 s ahead)
bytes only	17.2%	20.6%
global horizon	27.5%	48.1%

Under the global horizon the ahead stream's frontier sheds stream A's fresh data on top of A's own surplus. The per-stream horizon removes that coupling: each stream loses at most its own overload surplus, and a peer's offset adds nothing — the invariant the peer test pins. (An earlier measurement showing 0.0% for stream A relied on the forced-jump variant that turned out to livelock; the laggard's free ride went with it, and loss now stays with each stream's own overload.)

Each frontier is the stream's plausible_anchor rather than its bare maximum, so a single far-future stray cannot condemn the real backlog behind it.

The byte bound stays global, since memory is, and drops the stalest survivors regardless of stream.

Why the memory backstop is in bytes

Payload sizes at this layer span four orders of magnitude, so no message count works. Measured against a cap of 2000 messages:

payload	per message	× 2000
monitor counts (~1k bins)	~8 KB	16 MB
ev44, modest (5k events)	~40 KB	80 MB
ev44, busy (100k events)	~800 KB	1.6 GB
ad00 frame (4096² uint16)	33.5 MB	67 GB

Sizing covers both shapes that reach the overflow: the nbytes of the numpy arrays in the DetectorEvents/MonitorEvents dataclasses, and underlying_size() of the sc.DataArray that the da00/ad00 adapters deliver for monitor counts and area-detector frames — the adapters convert upstream of the batcher, so no raw ADArray ever reaches it.

512 MB is sized against a 32 GB production host shared by all backend services and their workflow/job data. The backlog absorbs transients rather than storing data, so it takes a small slice — every backend service shedding at once costs ~2 GB, ~6% of the host, which matters given #1264 saw detector_data alone reach 29.9 GB. It is the binding bound only for frame-sized payloads: ten seconds of 4096² frames is gigabytes, so the byte cap caps area-detector retention at ~15 frames while the data-time bound governs everything else.

It is deliberately not raised for DREAM. DREAM is specified at up to 1e8 events/s across all banks, ~763 MB/s, or ~3.4 MB per ev44 message at 16 banks and 14 Hz. At that rate there is no useful amount of buffering: every second behind costs another 763 MB, so a service that cannot keep up cannot be rescued by a deeper buffer. Raising the cap would buy fractions of a second while adding directly to a footprint that something else already dominates.

Reporting

drain_metrics() on MessageBatcher returns the deepest backlog observed in data time and the messages and bytes shed, all per metrics interval — the same semantics as the counters in the periodic consumer_metrics line — and the processor emits them in its periodic processor_metrics line. Cumulative totals remain available as total_dropped_messages/total_dropped_bytes, following BackgroundMessageSource's naming. The batcher is the only place this lag is visible: the consumer reports no lag for data it has already handed over, and per-stream ingest lag is measured before batching. max_backlog_s is a leading indicator — it rises during any stall, whether or not shedding follows. The throttled batcher_backlog_shedding warning (at most once per 60 s, since shedding is intermittent by nature) now carries both limits, the observed peak, and the cumulative totals.

This covers items 1 and 2 of #378. Item 3, propagating the drop rate to the dashboard, is not done here — these are backend logs only, and ServiceStatus is unchanged — so #378 should stay open.

What this does not bound

The retained overflow is the smaller half of batcher memory at high rates. Peak residency in healthy, keeping-up operation, split by where it sits:

banks	batch length	active window
4	1.0 s	545 MB
4	8.0 s	5450 MB
16	1.0 s	545 MB
16	8.0 s	5450 MB

The active window — the per-stream buckets holding one batch length of traffic — is ten times the overflow cap at the escalated batch length, and nothing bounds it against a memory budget. It is worse than a static figure: AdaptiveMessageBatcher escalates the batch length because processing is slow, and residency scales linearly with batch length, so the footprint grows 8x exactly when the service is already struggling.

This is pre-existing and orthogonal to the leak fixed here, so it is out of scope for this PR, but it is the term that matters at DREAM scale. Addressing it means capping max_level against a memory budget rather than a fixed 3, bounding total batcher residency rather than only the overflow, or shedding upstream at the consumer — a design decision worth taking on its own.

Behaviour worth knowing

A producer or broker that resumes after a silence and delivers more than the data-time bound in a single poll is shed down to it (10 s at base batch lengths, two batch lengths once escalation exceeds that); a shorter gap is replayed in full. Beyond the bound the batcher does not distinguish resumption from a stall, and under this policy staying live is the intended call; the drop counters make it visible.
The bound applies only to gridded gated streams. A stream too slow to fill a slot per batch (e.g. an area detector at 1 Hz against a 1 s batch) never reaches _overflow at all — it is drained into the batch at each close and cannot accumulate — so the metric reads zero for it by construction.
With an inter-stream offset larger than the bound under sustained overload, the ahead stream bears the extra shedding (each stream still loses its own overload surplus), because the window can only be in one place and rides the laggard. That is inherent to a single window rather than to the bound; the previous policy grew memory instead of choosing. Realistic ESS offsets (pulse-stamped detectors) are far below one batch length.

Test plan

TestBacklogShedding: backlog bounded in data time, one deep stall costs bounded lag, peer offset adds no shedding beyond a stream's own surplus, byte shedding of one stream does not disturb a healthy peer, stray future timestamp does not condemn the backlog, memory backstop shortens retention for large payloads, retained backlog bounded under sustained overload, backlog does not grow with time, nothing dropped while keeping up, window keeps up with live data while shedding, log/context stream never dropped, shedding is logged, dropped_bytes tracks dropped payload, metrics report backlog depth and drops, metrics reset on drain, memory backstop sizes scipp payloads (parametrized numpy/sc.DataArray), oversized payloads never shed the entire backlog, delivery hiccup within the seconds bound replays in full, batch floor governs at escalated batch lengths
tests/core tests/services tests/kafka — 988 passed, 1 skipped, 4 xfailed (pre-existing)
Full suite — 4799 passed, 43 skipped, 8 xfailed (pre-existing)
ruff check and ruff format clean

A window advances by at most one batch length per batch() call, so once a service iteration takes longer than the batch length the caller hands over more data than the window can release. The surplus accumulated in _overflow for the life of the process: unbounded, reachable memory that the cyclic collector cannot touch, while the Kafka consumer reported zero lag because the backlog sits downstream of it. This is what OOM-killed detector_data on TBL, where a ~11 s Timepix3 update against the adaptive batcher's 8 s ceiling stranded ~330 messages per iteration, each pinning its ev44 payload. Cap the retained overflow and drop the stalest surplus, the same bounded-buffer policy BackgroundMessageSource already applies to the consumer queue. Only GATED_STREAM_KINDS reach _overflow, so control, log and context data can never be shed. Shedding keeps the newest messages and forces the window jump that a detected gap would, so the window catches up to live traffic instead of crawling through a backlog it can never clear. On the reproduced TBL overload this takes the stranded backlog from 64871 messages and growing to 871 bounded, and delivers more messages than before (211199 vs 160329) because the window no longer falls permanently behind. Reporting is a throttled cumulative warning plus a dropped_messages counter. Propagating the drop rate to the dashboard (#378 item 3) is left open. Refs #378

Payload sizes at this layer span four orders of magnitude: an ev44 chunk is tens of kilobytes, one ad00 area-detector frame (4096x4096 uint16) is 33.5 MB. A message count cannot serve both. The previous cap of 2000 messages was wrong in both directions: it permitted 67 GB of retained ORCA frames, while shedding data on instruments with ten or more gating streams during a stall they would have recovered from -- a 5-batch stall at the escalated 8 s batch length leaves 4380 legitimate messages at ten streams and 17520 at forty. Measure the payload instead and bound the retained bytes. Sizing sums the nbytes of the arrays the payload exposes, which covers both shapes that reach the overflow: DetectorEvents/MonitorEvents dataclasses of numpy arrays, and the ADArray named tuple holding a frame. The default of 512 MB is sized against a 32 GB production host shared by all backend services and their workflow/job data. The backlog absorbs transients rather than storing data, so it takes a small slice: every backend service shedding at once costs ~2 GB. It holds ~13000 ev44 chunks, more than a 40-stream instrument produces in a 20 s stall, and ~15 area-detector frames. The survivable stall shrinks as ingest bandwidth grows, which is the point -- a backlog that cannot be caught up on should be shed, and a byte budget makes that automatic instead of a tuning decision.

The byte cap alone bounds memory but says nothing about how stale the retained backlog is, and the batcher only catches up at batch_length - iteration_time per call. A deep stall therefore left the service replaying old data for minutes when the loop had little headroom (measured: 5 min after a 120 s stall at a 0.9 s iteration), and the tolerated transient varied by three orders of magnitude across stream kinds because a byte budget buys ~20 s of ev44 but ~1.5 s of area-detector frames. Cap the backlog at two batch lengths of data time, keeping the byte cap as a memory backstop. Operators steer beam, samples and detector commissioning from this output, where a stale reading is worse than an intermittent hole; shedding is an exception state to fix upstream, not a mode to run in. Post-stall lag is now bounded by the cap regardless of loop headroom (recovery within two iterations in the same scenario) and the tolerance is uniform across instruments. The horizon is per stream, against each stream's own frontier: a single horizon for the whole backlog conflates depth with the offset between streams, shedding a stream's fresh data because a peer is stamped further ahead. Each frontier is a plausible anchor rather than a bare maximum, so one far-future stray cannot condemn the backlog behind it. The byte bound stays global, since memory is. Also report backlog depth and shedding in the periodic service metrics: the batcher is the only place this lag is visible, since the consumer reports no lag for data it has already handed over and per-stream ingest lag is measured before batching.

`_recover_from_gap` reads as a silence-gap mechanism, but it is also how the window catches up when a call hands over more data time than one batch length -- and the messages the jump lands behind re-route to negative slots and ride along in the next batch rather than being dropped. Both are load-bearing when reasoning about the backlog: the one-batch-per-call figure is a ceiling that binds only inside continuous data, and the data actually lost under overload is well below the surplus the window could not release.

The parameter name is unit-free and the queue element type is only visible in the constructor, so `max_queue_size=1000` reads as a message count when it is a batch count -- up to a hundred times more messages than it appears. Neither says anything about bytes, which is the resource that runs out, and which is why the backlog bound added here needs a byte cap of its own.

…rics Three review findings on the backlog bound, plus its policy statement: - The byte backstop sized every sc.DataArray payload at the 1 KB fallback: the da00/ad00 adapters convert to scipp upstream of the batcher, so AREA_DETECTOR and MONITOR_COUNTS messages never carry the array-exposing shapes the sizing assumed, and a 33.5 MB frame counted as 1 KB -- defeating the bound for the one payload class where it is binding. Scipp payloads are now sized via underlying_size(), and the memory-backstop test is parametrized over numpy and scipp payloads so the shapes the adapters actually deliver are covered. - Shedding could drop the entire overflow when the newest surviving message alone exceeded the byte bound; the forced gap jump then crashed on min() of the empty backlog. The newest message now survives unconditionally -- the jump needs an anchor, and keeping the newest is the policy anyway. - The processor_metrics line mixed per-interval fields (batches, errors) with cumulative drop counters. drain_metrics() now reports per-interval values, matching consumer_metrics; cumulative totals remain available as total_dropped_* and in the throttled shedding warning, following BackgroundMessageSource's naming. The freshness policy (stay live, shed the stalest, gated kinds only) is now stated in the module docstring alongside the clock policy rather than only in the comment on the bound constants, with a pointer from the glossary. The default drain_metrics docstring no longer claims that batchers without the bound hold no backlog -- SimpleMessageBatcher does, and reports zeros here. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Forcing the gap jump on every shed bypassed the veto that gap recovery exists to honour: no jump while any gated stream still has messages in the window. A shed can be a byte-bound trim of one oversized stream while a peer gates normally -- area-detector frames alongside detector events in the same service is exactly that topology -- and the forced jump then rides the trimmed stream's frontier past the peer's live traffic. The peer's gate becomes unsatisfiable, the timeout threshold recedes with every jump, and nothing closes until the wall-clock backstop: a delivery livelock, found by the scenario test added here (2000 of 2000 healthy-peer messages held, zero delivered, across 100 calls). The forced jump is also unnecessary. When a shed does strand the window behind the survivors, the stranded region is empty by construction -- the data was just discarded -- so the ordinary gap check fires on the next call and jumps in one step. Bounded lag arrives one poll iteration late; livelock never. The sustained- overload properties (window keeps up, backlog and lag bounded) hold unchanged with the veto respected, and a deep stall now delivers its first window as a proper batch instead of merging it into a far-future one with an understated start_time. One distribution changes: under overload plus an inter-stream offset, the lagging stream now sheds its own surplus instead of riding the forced jump's negative slots to zero loss while the ahead peer bore everything. The per-stream horizon's guarantee -- an offset alone never causes shedding, and a peer's offset never adds to a stream's own loss -- is unchanged, and is what the reworked peer test pins. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Two batch lengths is below ordinary infrastructure noise: a consumer-group rebalance alone takes seconds, and every delivery gap beyond the bound became a permanent hole in each workflow's data. Retaining more is nearly free in that regime -- an idle service recovers a retained burst by a gap jump plus timeout closes at poll rate, not the slow crawl of an overloaded loop -- so holing it bought nothing. The bound is now the larger of two terms. ``max_backlog_s`` (10 s) expresses the intent: how deep a burst is replayed in full rather than shed. Its price is paid only under sustained overload past what escalation absorbs, where the steady lag approaches the bound. The two-batch-length floor stays because the overflow legitimately holds about a batch length of in-transit next-window traffic: a fixed seconds bound below the escalated batch length would shed healthy traffic, so the floor takes over at escalated windows (16 s at the 8 s ceiling). No pre-production statistics on gap durations exist; the ``max_backlog_s`` metric and drop counters are the instrument for tuning the default. The interval-peak attribute moves to ``_backlog_peak_s`` to free the name for the new parameter. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Two test docstrings still described the forced-jump mechanism that the veto rework removed: the byte-shedding peer test opened by asserting the exact opposite of _shed_backlog's contract, in the very test that pinned the livelock the forced jump caused. Reword both to state the current design -- shedding leaves the jump to gap detection. Also drop _shed_backlog's bool return, which nothing consumed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

MridulS

Requesting changes based on adversarial checks against the current head. The focused suite is green (209 passed, 1 expected xfail), but the first two inline findings reproduce cases where the new freshness/memory policy either preserves a poisoned future timestamp over valid traffic or retains shed-eligible bulk data entirely outside the bound. The remaining comments cover byte accounting, a vacuous regression test, and lower-priority API/observability details.

MridulS · 2026-09-03T12:09:04Z

+                self._backlog_peak_s, (frontier - stamps[0]).to_seconds()
+            )
+            horizons[stream] = frontier - bound
+        kept = [item for item in sized if item[1].timestamp >= horizons[item[1].stream]]


[P1] Do not let timestamps rejected by plausible_anchor win retention. This lower-only cutoff keeps every disconnected timestamp after frontier, and the following oldest-first byte trim therefore sacrifices valid traffic before a future outlier. With a converged 20 Hz batcher and a 10 KB cap, two seconds of valid 100-byte traffic plus one 20 KB message at +600 s drops all 20 valid overflow messages and retains only the outlier; the next gap recovery moves the data clock roughly 600 seconds forward. At 1 Hz it happens even without byte pressure because one valid overflow pulse and one future stray tie, plausible_anchor selects the later group, and the valid pulse is time-shed. Please treat timestamps outside the selected plausible component separately, and add both sparse/tied and byte-pressure regressions.

Good catch on the byte-pressure case, that was a real regression: a stray ahead of the frontier now goes first under byte pressure, before any oldest-first trimming of the connected traffic (832345b, with the 20 Hz / 10 KB scenario as a regression test).

The sparse tie case I have left as is, deliberately. I tried exempting every message outside the plausible component from the time bound, which handles the tie, but it regresses the peer-offset test: a 21-message group that had fallen 3 s behind a 40-message group of the same stream was retained instead of shed, and then anchored the next gap jump. Structurally that stale minority group and the lone valid pulse are the same shape (a disconnected group behind the frontier); only the group sizes differ. The cost of the tie case is one message, and the +600 s jump that follows is pre-existing (it happens without shedding too, once the valid traffic drains) and is corrected by the stall backstop. Telling a majority from a tie inside the shedding logic is possible but buys little for a contrived input, so I would rather not add it.

MridulS · 2026-09-03T12:09:04Z

+        call and jumps in one step.  Bounded lag arrives one call late;
+        livelock never.
+        """
+        if not self._overflow:


[P1] The byte backstop misses retained gated messages in _future. A new or otherwise ungridded bulk stream is routed to _future before it is observed, so this early return bypasses all limits and metrics. With max_backlog_bytes=100, 100 distinct 8 KB AREA_DETECTOR messages at window.end + 0.1 s leave 800 KB in _future, zero drops, and zero backlog metrics. Repeating messages there grows memory indefinitely; advancing the fake wall clock past the stall threshold does not recover because _buffered_messages() also excludes _future. Please include shed-eligible gated _future traffic in the bound/stall accounting (without shedding non-gated context), or otherwise ensure this buffer cannot persist unbounded.

Fair concern, but I could only reproduce this with a frozen timestamp: 100 messages all stamped at exactly window.end + 0.1 s. With distinct advancing timestamps _future never accumulates (alone or alongside a healthy gated stream under a slow loop), because every HWM advance closes the window and drains it, and an ungridded stream is by definition too sparse to fill a batch, so it holds at most about three messages. The frozen-timestamp producer grows the active-window bucket unboundedly in exactly the same way, with no recovery either, and that is pre-existing: it is the active-window bound that the description scopes out, and it needs the same fix (a bound on total batcher residency rather than on _overflow). Including _future in the stall check would not bound memory here either, since stall recovery re-places the window at that timestamp and the messages then grow the bucket instead. I will open a follow-up issue for the active-window bound rather than widen this PR.

Follow-ups: the residency bound is tracked in #1275, and the stuck-clock symptom (bucket and _future variants) is fixed in #1284.

MridulS · 2026-09-03T12:09:05Z

+        parts = getattr(value, '__dict__', {}).values()
+    total = 0
+    for part in parts:
+        part_nbytes = getattr(part, 'nbytes', None)


[P2] Count the backing allocation pinned by NumPy views, not only the visible view. This is arbitrarily low for slices, and it affects the production monitor adapter: a 100,000-event non-pixellated MonitorEvents retained an 800,096-byte ev44 FlatBuffer through time_of_arrival.base, while this function reported 400,000 bytes. The nominal 512 MiB cap can therefore retain roughly 1 GiB and dropped_bytes is understated too. Please count unique root/base allocations (avoiding double-counting shared bases), and cover the real MonitorEvents representation in the sizing tests.

MridulS · 2026-09-03T12:09:05Z

+        fed, delivered, _, _ = self._run(50, batcher)
+        early = fed - delivered - batcher.total_dropped_messages
+        fed, delivered, _, _ = self._run(200, batcher)
+        late = fed - delivered - batcher.total_dropped_messages


[P2/testing] This comparison is vacuous because _run() restarts timestamps and local feed/delivery counters, while total_dropped_messages remains cumulative on the reused batcher. Exact-head values are early=21 and late=-940, although 41 messages are actually retained, so the one-sided assertion necessarily passes. Continue the timestamp/counters, subtract a drop delta, or compare the actual retained buffers so this protects the no-growth invariant.

MridulS · 2026-09-03T12:09:05Z

        self._future: list[Message[Any]] = []
        self._clock = clock
        self._last_close_wall = clock()
+        self._max_backlog_s = max_backlog_s


[P3] Validate the new limits at construction. max_backlog_s=NaN or inf constructs successfully and fails only on the first overflow when converted to Duration; setting both time limits negative places the horizon after the frontier and drops the complete overflow, contradicting the newest-survivor invariant. Please reject unsupported non-finite/negative values up front (while retaining zero if it is intentionally supported).

MridulS · 2026-09-03T12:09:05Z

+        pending replacement by the rate-aware batcher and deliberately does
+        not implement the backlog bound.
+        """
+        return BatcherMetrics(max_backlog_s=0.0, dropped_messages=0, dropped_bytes=0)


[P2/observability] Returning zero is indistinguishable from a measured healthy backlog for the supported --batcher=simple fallback. In a reproduction it retained 1,981 messages spanning 99 seconds while this snapshot reported max_backlog_s=0. The docstring acknowledges the limitation, but the emitted processor_metrics record does not. Please report its future-queue depth or represent unsupported metrics as unavailable rather than a false zero.

Agreed it is a false zero, but the simple batcher is a fallback slated for replacement by the rate-aware one, and its future queue is the thing that replacement removes. Reporting "unavailable" would mean an optional field or a sentinel in the metrics record for one batcher on its way out, so I would rather leave the docstring caveat and not touch the record shape.

MridulS · 2026-09-03T12:09:05Z

+
+# Secondary hard bound, on memory.  The data-time bound alone does not bound
+# bytes: payload sizes span four orders of magnitude, and one ad00
+# area-detector frame (4096x4096 uint16) is 33.5 MB, so two batch lengths of


[P3/docs] This uses the wire dtype, not the representation retained by the batcher. Ad00ToScippAdapter widens uint16 to int32 before batching, so a 4096x4096 frame is about 67.1 MB (plus object overhead), and 512 MiB holds roughly seven frames rather than fifteen. Please correct the sizing rationale, or reconsider the limit if fifteen retained frames is the intended tolerance.

The byte trim shed oldest-first, so a message stamped ahead of its stream's frontier survived while the valid traffic behind it was sacrificed, and the eventual gap jump then had only the stray to anchor on. Messages ahead of the frontier are disconnected by construction, so they go first under byte pressure. MonitorEvents keeps only the time-of-arrival view of its ev44 buffer visible but pins the whole buffer; sizing by the view understated the retained bytes twofold. Numpy arrays are now sized by the allocation they view, counted once per allocation so DetectorEvents' two views do not double up. Reject non-finite or negative backlog limits at construction, since a negative bound empties the overflow and NaN fails only on first use. The no-growth test compared a cumulative drop counter against counters that restarted per run, so it could not fail; it now continues the timeline and compares the held backlog. The ad00 sizing comment used the wire dtype; the adapter widens uint16 to int32, so a frame is 67 MB and the cap holds ~7, not ~15. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

SimonHeybrock mentioned this pull request Aug 31, 2026

Add load shedding for backend message processing #739

Draft

4 tasks

SimonHeybrock added 4 commits August 31, 2026 10:29

SimonHeybrock mentioned this pull request Sep 1, 2026

Add periodic cyclic GC to the service loop to stop unbounded memory growth #1265

Closed

SimonHeybrock and others added 3 commits September 1, 2026 07:37

SimonHeybrock marked this pull request as ready for review September 1, 2026 08:43

SimonHeybrock and others added 2 commits September 1, 2026 10:43

Merge branch 'main' into 378-bound-batcher-backlog

a09a387

SimonHeybrock mentioned this pull request Sep 1, 2026

Active-window batcher residency is unbounded and escalation multiplies it #1275

Open

MridulS requested changes Sep 3, 2026

View reviewed changes

This was referenced Sep 3, 2026

Bound active-window residency in the rate-aware batcher #1283

Closed

Deliver traffic whose timestamp has stopped advancing #1284

Open

SimonHeybrock requested a review from MridulS September 4, 2026 13:50

Navigation Menu

Uh oh!

Bound the rate-aware batcher backlog - #1271

Bound the rate-aware batcher backlog#1271
SimonHeybrock wants to merge 11 commits into
mainfrom
378-bound-batcher-backlog

SimonHeybrock commented Aug 31, 2026 •

edited

Loading

Uh oh!

MridulS left a comment

Uh oh!

MridulS Sep 3, 2026

Uh oh!

SimonHeybrock Sep 3, 2026

Uh oh!

MridulS Sep 3, 2026

Uh oh!

SimonHeybrock Sep 3, 2026

Uh oh!

SimonHeybrock Sep 3, 2026 •

edited

Loading

Uh oh!

MridulS Sep 3, 2026

Uh oh!

MridulS Sep 3, 2026

Uh oh!

MridulS Sep 3, 2026

Uh oh!

MridulS Sep 3, 2026

Uh oh!

SimonHeybrock Sep 3, 2026

Uh oh!

MridulS Sep 3, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

SimonHeybrock commented Aug 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

What this does

Why data time is the primary bound

Choosing the bound

Why the horizon is per stream

Why the memory backstop is in bytes

Reporting

What this does not bound

Behaviour worth knowing

Test plan

Uh oh!

MridulS left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SimonHeybrock Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

SimonHeybrock commented Aug 31, 2026 •

edited

Loading

SimonHeybrock Sep 3, 2026 •

edited

Loading