From aad78ac831e4081a9ce8da1fdf37248443bae2a4 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Fri, 27 Feb 2026 11:44:25 +0000 Subject: [PATCH 01/10] Add load shedding for backend message processing (#378) When the backend can't keep up with the Kafka message stream, the new LoadShedder selectively drops bulk event data (detector events, monitor events/counts, area detector) while preserving control messages and f144 logs. Detection uses consecutive non-None batcher results as the overload signal, with hysteresis to prevent oscillation. - New LoadShedder class with 50% subsampling when active - ServiceStatus gains is_shedding and messages_dropped fields - x5f2 serialization updated for new fields (backward-compatible defaults) - OrchestratingProcessor wires in shedding before the batcher - Dashboard shows SHEDDING badge (amber) and dropped message count - Load shedding is enabled by default but can be disabled for testing Prompt: Implement the following plan: Load Shedding for Backend Message Processing (#378) Co-Authored-By: Claude Opus 4.6 --- src/ess/livedata/core/job.py | 2 + src/ess/livedata/core/load_shedder.py | 101 ++++++++++ .../livedata/core/orchestrating_processor.py | 17 ++ .../widgets/backend_status_widget.py | 22 ++- src/ess/livedata/kafka/x5f2_compat.py | 10 + tests/core/load_shedder_test.py | 181 ++++++++++++++++++ tests/helpers/livedata_app.py | 4 +- tests/kafka/status_message_test.py | 27 +++ 8 files changed, 361 insertions(+), 3 deletions(-) create mode 100644 src/ess/livedata/core/load_shedder.py create mode 100644 tests/core/load_shedder_test.py diff --git a/src/ess/livedata/core/job.py b/src/ess/livedata/core/job.py index 1d0a6e161..5795f4929 100644 --- a/src/ess/livedata/core/job.py +++ b/src/ess/livedata/core/job.py @@ -122,6 +122,8 @@ class ServiceStatus: active_job_count: int messages_processed: int error: str | None = None + is_shedding: bool = False + messages_dropped: int = 0 def _add_time_coords( diff --git a/src/ess/livedata/core/load_shedder.py b/src/ess/livedata/core/load_shedder.py new file mode 100644 index 000000000..1bf2be9c0 --- /dev/null +++ b/src/ess/livedata/core/load_shedder.py @@ -0,0 +1,101 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2025 Scipp contributors (https://github.com/scipp) +"""Load shedding for backend message processing. + +When the backend can't keep up with the Kafka message stream, the LoadShedder +selectively drops bulk event data while preserving control messages and f144 logs. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from .message import Message, StreamKind + +DROPPABLE_KINDS = frozenset( + { + StreamKind.DETECTOR_EVENTS, + StreamKind.MONITOR_EVENTS, + StreamKind.MONITOR_COUNTS, + StreamKind.AREA_DETECTOR, + } +) + +# Consecutive non-None batcher results before entering shedding mode +_ACTIVATION_THRESHOLD = 5 +# Consecutive idle (None) batcher results before exiting shedding mode +_DEACTIVATION_THRESHOLD = 3 + + +@dataclass(frozen=True, slots=True) +class LoadShedderState: + """Snapshot of load shedder state for status reporting.""" + + is_shedding: bool + messages_dropped: int + + +class LoadShedder: + """Selectively drops bulk event data when the backend falls behind. + + Detection uses consecutive non-None batcher results as the overload signal. + When active, keeps every 2nd droppable message (50% reduction). + """ + + def __init__(self) -> None: + self._consecutive_batches: int = 0 + self._consecutive_idle: int = 0 + self._is_shedding: bool = False + self._messages_dropped: int = 0 + self._subsample_counter: int = 0 + + @property + def state(self) -> LoadShedderState: + return LoadShedderState( + is_shedding=self._is_shedding, + messages_dropped=self._messages_dropped, + ) + + def report_batch_result(self, batch_produced: bool) -> None: + """Update overload detection counters after a batcher cycle. + + Parameters + ---------- + batch_produced: + True if the batcher returned a batch (non-None), False if idle (None). + """ + if batch_produced: + self._consecutive_batches += 1 + self._consecutive_idle = 0 + if ( + not self._is_shedding + and self._consecutive_batches >= _ACTIVATION_THRESHOLD + ): + self._is_shedding = True + else: + self._consecutive_idle += 1 + self._consecutive_batches = 0 + if self._is_shedding and self._consecutive_idle >= _DEACTIVATION_THRESHOLD: + self._is_shedding = False + self._subsample_counter = 0 + + def shed(self, messages: list[Message]) -> list[Message]: + """Filter messages when shedding is active. + + When inactive, returns all messages unchanged. + When active, drops every other droppable message (50% reduction). + Non-droppable messages (control, f144 logs) are always preserved. + """ + if not self._is_shedding: + return messages + result: list[Message] = [] + for msg in messages: + if msg.stream.kind not in DROPPABLE_KINDS: + result.append(msg) + else: + self._subsample_counter += 1 + if self._subsample_counter % 2 == 0: + result.append(msg) + else: + self._messages_dropped += 1 + return result diff --git a/src/ess/livedata/core/orchestrating_processor.py b/src/ess/livedata/core/orchestrating_processor.py index 00f3fe00e..fb4193f54 100644 --- a/src/ess/livedata/core/orchestrating_processor.py +++ b/src/ess/livedata/core/orchestrating_processor.py @@ -14,6 +14,7 @@ from .job import JobResult, JobStatus, ServiceState, ServiceStatus from .job_manager import JobFactory, JobManager, WorkflowData from .job_manager_adapter import JobManagerAdapter +from .load_shedder import LoadShedder from .message import ( COMMANDS_STREAM_ID, STATUS_STREAM_ID, @@ -89,6 +90,7 @@ def __init__( sink: MessageSink[Tout], preprocessor_factory: PreprocessorFactory[Tin, Tout], message_batcher: MessageBatcher | None = None, + enable_load_shedding: bool = True, ) -> None: self._source = source self._sink = sink @@ -100,6 +102,7 @@ def __init__( self._config_processor = ConfigProcessor( job_manager_adapter=self._job_manager_adapter ) + self._load_shedder = LoadShedder() if enable_load_shedding else None self._last_status_update: int | None = None self._status_update_interval = 2_000_000_000 # 2 seconds @@ -143,7 +146,11 @@ def process(self) -> None: self._report_status() + if self._load_shedder is not None: + data_messages = self._load_shedder.shed(data_messages) message_batch = self._message_batcher.batch(data_messages) + if self._load_shedder is not None: + self._load_shedder.report_batch_result(message_batch is not None) if message_batch is None: self._empty_batches += 1 self._maybe_log_metrics() @@ -222,6 +229,7 @@ def _report_status(self) -> None: def _get_service_status(self, job_statuses: list[JobStatus]) -> ServiceStatus: """Get the current service status for heartbeat publishing.""" + shedder = self._load_shedder return ServiceStatus( instrument=self._instrument, namespace=self._namespace, @@ -231,6 +239,10 @@ def _get_service_status(self, job_statuses: list[JobStatus]) -> ServiceStatus: active_job_count=len(job_statuses), messages_processed=self._messages_processed, error=self._service_error, + is_shedding=shedder.state.is_shedding if shedder is not None else False, + messages_dropped=( + shedder.state.messages_dropped if shedder is not None else 0 + ), ) def _maybe_log_metrics(self) -> None: @@ -242,6 +254,7 @@ def _maybe_log_metrics(self) -> None: if timestamp - self._last_metrics_time >= self._metrics_interval: active_jobs = len(self._job_manager.active_jobs) + shedder = self._load_shedder logger.info( 'processor_metrics', messages=self._messages_processed, @@ -249,6 +262,10 @@ def _maybe_log_metrics(self) -> None: empty_batches=self._empty_batches, active_jobs=active_jobs, errors=self._errors_since_last_metrics, + shedding=shedder.state.is_shedding if shedder is not None else False, + messages_dropped=( + shedder.state.messages_dropped if shedder is not None else 0 + ), interval_seconds=(timestamp - self._last_metrics_time) / 1e9, ) # Reset counters (except messages_processed which is cumulative for service) diff --git a/src/ess/livedata/dashboard/widgets/backend_status_widget.py b/src/ess/livedata/dashboard/widgets/backend_status_widget.py index bf92eb24f..53b809bcf 100644 --- a/src/ess/livedata/dashboard/widgets/backend_status_widget.py +++ b/src/ess/livedata/dashboard/widgets/backend_status_widget.py @@ -26,6 +26,7 @@ class WorkerUIConstants: } DEFAULT_COLOR = "#6c757d" STALE_COLOR = "#dc3545" # Red for unexpectedly disappeared workers + SHEDDING_COLOR = "#ff8c00" # Amber for load shedding # Sizes NAMESPACE_WIDTH = 200 @@ -126,12 +127,14 @@ def __init__( self.update(status, is_stale, last_seen_seconds_ago) def _get_status_color(self, status: ServiceStatus, is_stale: bool) -> str: - """Get color for worker state, considering staleness.""" + """Get color for worker state, considering staleness and load shedding.""" if is_stale: # Graceful shutdown (inferred from timed-out stopping): show gray if status.state == ServiceState.stopping: return WorkerUIConstants.COLORS[ServiceState.stopped] return WorkerUIConstants.STALE_COLOR + if status.is_shedding: + return WorkerUIConstants.SHEDDING_COLOR return WorkerUIConstants.COLORS.get( status.state, WorkerUIConstants.DEFAULT_COLOR ) @@ -165,6 +168,8 @@ def update( # Distinguish graceful shutdown from unexpected disappearance is_graceful = status.state == ServiceState.stopping status_text = "STOPPED" if is_graceful else "STALE" + elif status.is_shedding: + status_text = "SHEDDING" else: status_text = status.state.value.upper() status_style = self._create_status_style(status_color) @@ -187,7 +192,13 @@ def update( # Stats jobs_text = f"Jobs: {status.active_job_count}" msgs_text = f"Msgs: {_format_messages(status.messages_processed)}" - self._stats_pane.object = f"{jobs_text} | {msgs_text}" + stats_parts = [jobs_text, msgs_text] + if status.messages_dropped > 0: + stats_parts.append( + f'' + f"Dropped: {_format_messages(status.messages_dropped)}" + ) + self._stats_pane.object = f"{' | '.join(stats_parts)}" def _calculate_uptime(self, started_at_ns: int) -> float: """Calculate uptime in seconds from started_at timestamp.""" @@ -278,6 +289,7 @@ def _format_summary(self) -> str: stopped_count = 0 stale_count = 0 error_count = 0 + shedding_count = 0 for worker_key, status in self._service_registry.worker_statuses.items(): is_stale = self._service_registry.is_status_stale(worker_key) @@ -288,6 +300,8 @@ def _format_summary(self) -> str: stopped_count += 1 else: stale_count += 1 + elif status.is_shedding: + shedding_count += 1 elif status.state == ServiceState.starting: starting_count += 1 elif status.state == ServiceState.running: @@ -310,6 +324,10 @@ def _span(color: str, count: int, label: str) -> str: ) if running_count: parts.append(_span(colors[ServiceState.running], running_count, "running")) + if shedding_count: + parts.append( + _span(WorkerUIConstants.SHEDDING_COLOR, shedding_count, "shedding") + ) if stopping_count: parts.append( _span(colors[ServiceState.stopping], stopping_count, "stopping") diff --git a/src/ess/livedata/kafka/x5f2_compat.py b/src/ess/livedata/kafka/x5f2_compat.py index fc2bcd36d..deb2fb960 100644 --- a/src/ess/livedata/kafka/x5f2_compat.py +++ b/src/ess/livedata/kafka/x5f2_compat.py @@ -197,6 +197,12 @@ class ServiceStatusPayload(pydantic.BaseModel): description="Total messages processed since startup" ) error: str | None = pydantic.Field(default=None, description="Error message if any") + is_shedding: bool = pydantic.Field( + default=False, description="Whether load shedding is active" + ) + messages_dropped: int = pydantic.Field( + default=0, description="Cumulative messages dropped by load shedding" + ) class ServiceStatusJSON(pydantic.BaseModel): @@ -282,6 +288,8 @@ def from_service_status( active_job_count=status.active_job_count, messages_processed=status.messages_processed, error=status.error, + is_shedding=status.is_shedding, + messages_dropped=status.messages_dropped, ), ), ) @@ -298,6 +306,8 @@ def to_service_status(self) -> ServiceStatus: active_job_count=message.active_job_count, messages_processed=message.messages_processed, error=message.error, + is_shedding=message.is_shedding, + messages_dropped=message.messages_dropped, ) diff --git a/tests/core/load_shedder_test.py b/tests/core/load_shedder_test.py new file mode 100644 index 000000000..54a66b704 --- /dev/null +++ b/tests/core/load_shedder_test.py @@ -0,0 +1,181 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2025 Scipp contributors (https://github.com/scipp) + +import pytest + +from ess.livedata.core.load_shedder import ( + _ACTIVATION_THRESHOLD, + _DEACTIVATION_THRESHOLD, + DROPPABLE_KINDS, + LoadShedder, +) +from ess.livedata.core.message import Message, StreamId, StreamKind + + +def _make_message(kind: StreamKind, name: str = "src") -> Message: + return Message(timestamp=0, stream=StreamId(kind=kind, name=name), value=b"") + + +class TestLoadShedderInitialState: + def test_not_shedding_initially(self): + shedder = LoadShedder() + assert shedder.state.is_shedding is False + + def test_zero_dropped_initially(self): + shedder = LoadShedder() + assert shedder.state.messages_dropped == 0 + + +class TestLoadShedderActivation: + def test_activates_after_consecutive_batches(self): + shedder = LoadShedder() + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + assert shedder.state.is_shedding is True + + def test_does_not_activate_below_threshold(self): + shedder = LoadShedder() + for _ in range(_ACTIVATION_THRESHOLD - 1): + shedder.report_batch_result(batch_produced=True) + assert shedder.state.is_shedding is False + + def test_idle_cycle_resets_consecutive_count(self): + shedder = LoadShedder() + for _ in range(_ACTIVATION_THRESHOLD - 1): + shedder.report_batch_result(batch_produced=True) + shedder.report_batch_result(batch_produced=False) + # Restart counting — should not activate after fewer than threshold + for _ in range(_ACTIVATION_THRESHOLD - 1): + shedder.report_batch_result(batch_produced=True) + assert shedder.state.is_shedding is False + + +class TestLoadShedderDeactivation: + @pytest.fixture + def active_shedder(self): + shedder = LoadShedder() + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + assert shedder.state.is_shedding is True + return shedder + + def test_deactivates_after_consecutive_idle(self, active_shedder): + for _ in range(_DEACTIVATION_THRESHOLD): + active_shedder.report_batch_result(batch_produced=False) + assert active_shedder.state.is_shedding is False + + def test_does_not_deactivate_below_threshold(self, active_shedder): + for _ in range(_DEACTIVATION_THRESHOLD - 1): + active_shedder.report_batch_result(batch_produced=False) + assert active_shedder.state.is_shedding is True + + def test_batch_resets_idle_count(self, active_shedder): + for _ in range(_DEACTIVATION_THRESHOLD - 1): + active_shedder.report_batch_result(batch_produced=False) + active_shedder.report_batch_result(batch_produced=True) + # Restart idle counting + for _ in range(_DEACTIVATION_THRESHOLD - 1): + active_shedder.report_batch_result(batch_produced=False) + assert active_shedder.state.is_shedding is True + + +class TestLoadShedderShed: + def test_passes_everything_when_inactive(self): + shedder = LoadShedder() + messages = [ + _make_message(StreamKind.DETECTOR_EVENTS), + _make_message(StreamKind.LOG), + _make_message(StreamKind.MONITOR_EVENTS), + ] + result = shedder.shed(messages) + assert result == messages + + def test_preserves_non_droppable_when_active(self): + shedder = LoadShedder() + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + + non_droppable_kinds = [ + StreamKind.LOG, + StreamKind.LIVEDATA_COMMANDS, + StreamKind.LIVEDATA_RESPONSES, + StreamKind.LIVEDATA_DATA, + StreamKind.LIVEDATA_ROI, + StreamKind.LIVEDATA_STATUS, + StreamKind.UNKNOWN, + ] + messages = [_make_message(kind) for kind in non_droppable_kinds] + result = shedder.shed(messages) + assert result == messages + + def test_drops_roughly_half_of_droppable_when_active(self): + shedder = LoadShedder() + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + + messages = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(100)] + result = shedder.shed(messages) + assert len(result) == 50 + + def test_dropped_count_accuracy(self): + shedder = LoadShedder() + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + + messages = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(100)] + shedder.shed(messages) + assert shedder.state.messages_dropped == 50 + + def test_dropped_count_is_cumulative(self): + shedder = LoadShedder() + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + + batch = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(10)] + shedder.shed(batch) + shedder.shed(batch) + assert shedder.state.messages_dropped == 10 + + def test_all_droppable_kinds_are_shed(self): + shedder = LoadShedder() + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + + for kind in DROPPABLE_KINDS: + messages = [_make_message(kind) for _ in range(10)] + before = shedder.state.messages_dropped + result = shedder.shed(messages) + assert len(result) < len(messages), f"{kind} was not shed" + assert shedder.state.messages_dropped > before + + def test_mixed_messages_preserves_non_droppable(self): + shedder = LoadShedder() + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + + log_msg = _make_message(StreamKind.LOG) + cmd_msg = _make_message(StreamKind.LIVEDATA_COMMANDS) + det_msgs = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(10)] + messages = [log_msg, *det_msgs, cmd_msg] + + result = shedder.shed(messages) + assert log_msg in result + assert cmd_msg in result + + +class TestLoadShedderState: + def test_state_reflects_shedding(self): + shedder = LoadShedder() + assert shedder.state.is_shedding is False + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + assert shedder.state.is_shedding is True + + def test_state_is_snapshot(self): + shedder = LoadShedder() + state = shedder.state + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + # Original snapshot unchanged (frozen dataclass) + assert state.is_shedding is False + assert shedder.state.is_shedding is True diff --git a/tests/helpers/livedata_app.py b/tests/helpers/livedata_app.py index 291f06c9a..d1ad9e396 100644 --- a/tests/helpers/livedata_app.py +++ b/tests/helpers/livedata_app.py @@ -82,7 +82,9 @@ def from_service_builder( consumer = FakeConsumer() if use_naive_message_batcher: builder._processor_cls = partial( - OrchestratingProcessor, message_batcher=NaiveMessageBatcher() + OrchestratingProcessor, + message_batcher=NaiveMessageBatcher(), + enable_load_shedding=False, ) service = builder.from_consumer( consumer=consumer, diff --git a/tests/kafka/status_message_test.py b/tests/kafka/status_message_test.py index 10e88dc2f..3fa775f40 100644 --- a/tests/kafka/status_message_test.py +++ b/tests/kafka/status_message_test.py @@ -781,6 +781,24 @@ def test_to_service_status(self): assert converted.active_job_count == original.active_job_count assert converted.messages_processed == original.messages_processed + def test_round_trip_with_shedding_fields(self): + """Test that shedding fields survive model round-trip.""" + original = make_service_status(is_shedding=True, messages_dropped=42) + msg = ServiceStatusMessage.from_service_status(original) + converted = msg.to_service_status() + + assert converted.is_shedding is True + assert converted.messages_dropped == 42 + + def test_round_trip_defaults_shedding_fields(self): + """Test that shedding fields default gracefully.""" + original = make_service_status() + msg = ServiceStatusMessage.from_service_status(original) + converted = msg.to_service_status() + + assert converted.is_shedding is False + assert converted.messages_dropped == 0 + class TestServiceStatusX5F2Integration: """Test service status x5f2 serialization/deserialization.""" @@ -808,6 +826,15 @@ def test_service_status_x5f2_round_trip(self): assert converted.messages_processed == original.messages_processed assert converted.error == original.error + def test_service_status_x5f2_round_trip_with_shedding(self): + """Test x5f2 round-trip includes load shedding fields.""" + original = make_service_status(is_shedding=True, messages_dropped=1234) + x5f2_data = service_status_to_x5f2(original) + converted = x5f2_to_service_status(x5f2_data) + + assert converted.is_shedding is True + assert converted.messages_dropped == 1234 + def test_service_status_x5f2_with_error(self): """Test x5f2 round-trip with error message.""" original = make_service_status( From 35b6cd72fa41533fede6e8ad173975756e70790a Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Fri, 27 Feb 2026 14:14:58 +0000 Subject: [PATCH 02/10] Show drop rate as rolling 60s percentage in dashboard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the cumulative dropped-messages counter with a rolling 60-second window that tracks both dropped and eligible (total droppable) counts. The dashboard now shows "Dropped: 38% (62/164)" — percentage plus raw counts over the recent window — giving an actionable, stable signal instead of an ever-growing number. Implementation uses time-bucketed counters (10 buckets × 6s) inside LoadShedder, with a fake-clock injection point for deterministic testing. Eligible messages are counted even when shedding is inactive, so the rate correctly reflects the fraction being shed. Prompt: Show drop fraction/percentage and use a rolling time window instead of cumulative counters. Co-Authored-By: Claude Opus 4.6 --- src/ess/livedata/core/job.py | 1 + src/ess/livedata/core/load_shedder.py | 73 +++++++- .../livedata/core/orchestrating_processor.py | 18 +- .../widgets/backend_status_widget.py | 16 +- src/ess/livedata/kafka/x5f2_compat.py | 7 +- tests/core/load_shedder_test.py | 172 +++++++++++++----- tests/kafka/status_message_test.py | 11 +- 7 files changed, 232 insertions(+), 66 deletions(-) diff --git a/src/ess/livedata/core/job.py b/src/ess/livedata/core/job.py index 5795f4929..9137b00bf 100644 --- a/src/ess/livedata/core/job.py +++ b/src/ess/livedata/core/job.py @@ -124,6 +124,7 @@ class ServiceStatus: error: str | None = None is_shedding: bool = False messages_dropped: int = 0 + messages_eligible: int = 0 def _add_time_coords( diff --git a/src/ess/livedata/core/load_shedder.py b/src/ess/livedata/core/load_shedder.py index 1bf2be9c0..9a353da13 100644 --- a/src/ess/livedata/core/load_shedder.py +++ b/src/ess/livedata/core/load_shedder.py @@ -8,6 +8,8 @@ from __future__ import annotations +import time +from collections.abc import Callable from dataclasses import dataclass from .message import Message, StreamKind @@ -26,6 +28,9 @@ # Consecutive idle (None) batcher results before exiting shedding mode _DEACTIVATION_THRESHOLD = 3 +_N_BUCKETS = 10 +_BUCKET_DURATION_S = 6.0 # 10 buckets x 6s = 60s rolling window + @dataclass(frozen=True, slots=True) class LoadShedderState: @@ -33,6 +38,56 @@ class LoadShedderState: is_shedding: bool messages_dropped: int + messages_eligible: int + + +class _RollingCounter: + """Pair of (dropped, eligible) counters over a fixed rolling time window. + + The window is divided into fixed-size time buckets. Buckets older than the + window are discarded when the counter is advanced to the current time. + """ + + def __init__( + self, + n_buckets: int = _N_BUCKETS, + bucket_duration_s: float = _BUCKET_DURATION_S, + clock: Callable[[], float] = time.monotonic, + ) -> None: + self._n_buckets = n_buckets + self._bucket_duration_s = bucket_duration_s + self._clock = clock + self._dropped = [0] * n_buckets + self._eligible = [0] * n_buckets + self._current_bucket: int = 0 + self._last_time: float = clock() + + def _advance(self) -> None: + """Advance to the current time, zeroing any expired buckets.""" + now = self._clock() + elapsed = now - self._last_time + steps = int(elapsed / self._bucket_duration_s) + if steps <= 0: + return + # Cap: if we've been idle longer than the full window, just clear everything + steps = min(steps, self._n_buckets) + for i in range(1, steps + 1): + bucket = (self._current_bucket + i) % self._n_buckets + self._dropped[bucket] = 0 + self._eligible[bucket] = 0 + self._current_bucket = (self._current_bucket + steps) % self._n_buckets + self._last_time += steps * self._bucket_duration_s + + def record(self, *, dropped: int, eligible: int) -> None: + """Record counts into the current bucket.""" + self._advance() + self._dropped[self._current_bucket] += dropped + self._eligible[self._current_bucket] += eligible + + def totals(self) -> tuple[int, int]: + """Return (dropped, eligible) summed over the rolling window.""" + self._advance() + return sum(self._dropped), sum(self._eligible) class LoadShedder: @@ -40,20 +95,23 @@ class LoadShedder: Detection uses consecutive non-None batcher results as the overload signal. When active, keeps every 2nd droppable message (50% reduction). + Drop statistics are tracked over a rolling 60-second window. """ - def __init__(self) -> None: + def __init__(self, *, clock: Callable[[], float] = time.monotonic) -> None: self._consecutive_batches: int = 0 self._consecutive_idle: int = 0 self._is_shedding: bool = False - self._messages_dropped: int = 0 self._subsample_counter: int = 0 + self._rolling = _RollingCounter(clock=clock) @property def state(self) -> LoadShedderState: + dropped, eligible = self._rolling.totals() return LoadShedderState( is_shedding=self._is_shedding, - messages_dropped=self._messages_dropped, + messages_dropped=dropped, + messages_eligible=eligible, ) def report_batch_result(self, batch_produced: bool) -> None: @@ -85,9 +143,15 @@ def shed(self, messages: list[Message]) -> list[Message]: When inactive, returns all messages unchanged. When active, drops every other droppable message (50% reduction). Non-droppable messages (control, f144 logs) are always preserved. + + Both active and inactive calls record eligible message counts into the + rolling window so the drop rate reflects what fraction is being shed. """ + eligible = sum(1 for m in messages if m.stream.kind in DROPPABLE_KINDS) if not self._is_shedding: + self._rolling.record(dropped=0, eligible=eligible) return messages + dropped = 0 result: list[Message] = [] for msg in messages: if msg.stream.kind not in DROPPABLE_KINDS: @@ -97,5 +161,6 @@ def shed(self, messages: list[Message]) -> list[Message]: if self._subsample_counter % 2 == 0: result.append(msg) else: - self._messages_dropped += 1 + dropped += 1 + self._rolling.record(dropped=dropped, eligible=eligible) return result diff --git a/src/ess/livedata/core/orchestrating_processor.py b/src/ess/livedata/core/orchestrating_processor.py index fb4193f54..e07a56d27 100644 --- a/src/ess/livedata/core/orchestrating_processor.py +++ b/src/ess/livedata/core/orchestrating_processor.py @@ -229,7 +229,7 @@ def _report_status(self) -> None: def _get_service_status(self, job_statuses: list[JobStatus]) -> ServiceStatus: """Get the current service status for heartbeat publishing.""" - shedder = self._load_shedder + shedder_state = self._load_shedder.state if self._load_shedder else None return ServiceStatus( instrument=self._instrument, namespace=self._namespace, @@ -239,10 +239,9 @@ def _get_service_status(self, job_statuses: list[JobStatus]) -> ServiceStatus: active_job_count=len(job_statuses), messages_processed=self._messages_processed, error=self._service_error, - is_shedding=shedder.state.is_shedding if shedder is not None else False, - messages_dropped=( - shedder.state.messages_dropped if shedder is not None else 0 - ), + is_shedding=shedder_state.is_shedding if shedder_state else False, + messages_dropped=shedder_state.messages_dropped if shedder_state else 0, + messages_eligible=shedder_state.messages_eligible if shedder_state else 0, ) def _maybe_log_metrics(self) -> None: @@ -254,7 +253,7 @@ def _maybe_log_metrics(self) -> None: if timestamp - self._last_metrics_time >= self._metrics_interval: active_jobs = len(self._job_manager.active_jobs) - shedder = self._load_shedder + shedder_state = self._load_shedder.state if self._load_shedder else None logger.info( 'processor_metrics', messages=self._messages_processed, @@ -262,9 +261,12 @@ def _maybe_log_metrics(self) -> None: empty_batches=self._empty_batches, active_jobs=active_jobs, errors=self._errors_since_last_metrics, - shedding=shedder.state.is_shedding if shedder is not None else False, + shedding=shedder_state.is_shedding if shedder_state else False, messages_dropped=( - shedder.state.messages_dropped if shedder is not None else 0 + shedder_state.messages_dropped if shedder_state else 0 + ), + messages_eligible=( + shedder_state.messages_eligible if shedder_state else 0 ), interval_seconds=(timestamp - self._last_metrics_time) / 1e9, ) diff --git a/src/ess/livedata/dashboard/widgets/backend_status_widget.py b/src/ess/livedata/dashboard/widgets/backend_status_widget.py index 53b809bcf..7ac727bcd 100644 --- a/src/ess/livedata/dashboard/widgets/backend_status_widget.py +++ b/src/ess/livedata/dashboard/widgets/backend_status_widget.py @@ -74,6 +74,17 @@ def _format_messages(count: int) -> str: return f"{count / 1_000_000:.1f}M" +def _format_drop_rate(dropped: int, eligible: int) -> str: + """Format drop rate as percentage with counts over the rolling window.""" + if eligible == 0: + return f"Dropped: {_format_messages(dropped)}" + pct = 100 * dropped / eligible + return ( + f"Dropped: {pct:.0f}% " + f"({_format_messages(dropped)}/{_format_messages(eligible)})" + ) + + class WorkerStatusRow: """Widget to display the status of a single backend worker. @@ -194,9 +205,12 @@ def update( msgs_text = f"Msgs: {_format_messages(status.messages_processed)}" stats_parts = [jobs_text, msgs_text] if status.messages_dropped > 0: + drop_text = _format_drop_rate( + status.messages_dropped, status.messages_eligible + ) stats_parts.append( f'' - f"Dropped: {_format_messages(status.messages_dropped)}" + f"{drop_text}" ) self._stats_pane.object = f"{' | '.join(stats_parts)}" diff --git a/src/ess/livedata/kafka/x5f2_compat.py b/src/ess/livedata/kafka/x5f2_compat.py index deb2fb960..9934f0c0f 100644 --- a/src/ess/livedata/kafka/x5f2_compat.py +++ b/src/ess/livedata/kafka/x5f2_compat.py @@ -201,7 +201,10 @@ class ServiceStatusPayload(pydantic.BaseModel): default=False, description="Whether load shedding is active" ) messages_dropped: int = pydantic.Field( - default=0, description="Cumulative messages dropped by load shedding" + default=0, description="Messages dropped in the rolling window" + ) + messages_eligible: int = pydantic.Field( + default=0, description="Droppable messages seen in the rolling window" ) @@ -290,6 +293,7 @@ def from_service_status( error=status.error, is_shedding=status.is_shedding, messages_dropped=status.messages_dropped, + messages_eligible=status.messages_eligible, ), ), ) @@ -308,6 +312,7 @@ def to_service_status(self) -> ServiceStatus: error=message.error, is_shedding=message.is_shedding, messages_dropped=message.messages_dropped, + messages_eligible=message.messages_eligible, ) diff --git a/tests/core/load_shedder_test.py b/tests/core/load_shedder_test.py index 54a66b704..a39e3d2bf 100644 --- a/tests/core/load_shedder_test.py +++ b/tests/core/load_shedder_test.py @@ -5,7 +5,9 @@ from ess.livedata.core.load_shedder import ( _ACTIVATION_THRESHOLD, + _BUCKET_DURATION_S, _DEACTIVATION_THRESHOLD, + _N_BUCKETS, DROPPABLE_KINDS, LoadShedder, ) @@ -16,31 +18,58 @@ def _make_message(kind: StreamKind, name: str = "src") -> Message: return Message(timestamp=0, stream=StreamId(kind=kind, name=name), value=b"") +class FakeClock: + """Deterministic clock for testing the rolling window.""" + + def __init__(self, start: float = 0.0) -> None: + self._time = start + + def __call__(self) -> float: + return self._time + + def advance(self, seconds: float) -> None: + self._time += seconds + + +def _make_shedder(clock: FakeClock | None = None) -> LoadShedder: + if clock is None: + clock = FakeClock() + return LoadShedder(clock=clock) + + +def _activate(shedder: LoadShedder) -> None: + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + assert shedder.state.is_shedding is True + + class TestLoadShedderInitialState: def test_not_shedding_initially(self): - shedder = LoadShedder() + shedder = _make_shedder() assert shedder.state.is_shedding is False def test_zero_dropped_initially(self): - shedder = LoadShedder() + shedder = _make_shedder() assert shedder.state.messages_dropped == 0 + def test_zero_eligible_initially(self): + shedder = _make_shedder() + assert shedder.state.messages_eligible == 0 + class TestLoadShedderActivation: def test_activates_after_consecutive_batches(self): - shedder = LoadShedder() - for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) - assert shedder.state.is_shedding is True + shedder = _make_shedder() + _activate(shedder) def test_does_not_activate_below_threshold(self): - shedder = LoadShedder() + shedder = _make_shedder() for _ in range(_ACTIVATION_THRESHOLD - 1): shedder.report_batch_result(batch_produced=True) assert shedder.state.is_shedding is False def test_idle_cycle_resets_consecutive_count(self): - shedder = LoadShedder() + shedder = _make_shedder() for _ in range(_ACTIVATION_THRESHOLD - 1): shedder.report_batch_result(batch_produced=True) shedder.report_batch_result(batch_produced=False) @@ -53,10 +82,8 @@ def test_idle_cycle_resets_consecutive_count(self): class TestLoadShedderDeactivation: @pytest.fixture def active_shedder(self): - shedder = LoadShedder() - for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) - assert shedder.state.is_shedding is True + shedder = _make_shedder() + _activate(shedder) return shedder def test_deactivates_after_consecutive_idle(self, active_shedder): @@ -81,7 +108,7 @@ def test_batch_resets_idle_count(self, active_shedder): class TestLoadShedderShed: def test_passes_everything_when_inactive(self): - shedder = LoadShedder() + shedder = _make_shedder() messages = [ _make_message(StreamKind.DETECTOR_EVENTS), _make_message(StreamKind.LOG), @@ -91,9 +118,8 @@ def test_passes_everything_when_inactive(self): assert result == messages def test_preserves_non_droppable_when_active(self): - shedder = LoadShedder() - for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) + shedder = _make_shedder() + _activate(shedder) non_droppable_kinds = [ StreamKind.LOG, @@ -109,37 +135,16 @@ def test_preserves_non_droppable_when_active(self): assert result == messages def test_drops_roughly_half_of_droppable_when_active(self): - shedder = LoadShedder() - for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) + shedder = _make_shedder() + _activate(shedder) messages = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(100)] result = shedder.shed(messages) assert len(result) == 50 - def test_dropped_count_accuracy(self): - shedder = LoadShedder() - for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) - - messages = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(100)] - shedder.shed(messages) - assert shedder.state.messages_dropped == 50 - - def test_dropped_count_is_cumulative(self): - shedder = LoadShedder() - for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) - - batch = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(10)] - shedder.shed(batch) - shedder.shed(batch) - assert shedder.state.messages_dropped == 10 - def test_all_droppable_kinds_are_shed(self): - shedder = LoadShedder() - for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) + shedder = _make_shedder() + _activate(shedder) for kind in DROPPABLE_KINDS: messages = [_make_message(kind) for _ in range(10)] @@ -149,9 +154,8 @@ def test_all_droppable_kinds_are_shed(self): assert shedder.state.messages_dropped > before def test_mixed_messages_preserves_non_droppable(self): - shedder = LoadShedder() - for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) + shedder = _make_shedder() + _activate(shedder) log_msg = _make_message(StreamKind.LOG) cmd_msg = _make_message(StreamKind.LIVEDATA_COMMANDS) @@ -163,19 +167,87 @@ def test_mixed_messages_preserves_non_droppable(self): assert cmd_msg in result +class TestRollingWindow: + def test_dropped_count_within_window(self): + clock = FakeClock() + shedder = _make_shedder(clock) + _activate(shedder) + + messages = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(100)] + shedder.shed(messages) + assert shedder.state.messages_dropped == 50 + assert shedder.state.messages_eligible == 100 + + def test_counts_accumulate_across_calls_in_same_bucket(self): + clock = FakeClock() + shedder = _make_shedder(clock) + _activate(shedder) + + batch = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(10)] + shedder.shed(batch) + shedder.shed(batch) + assert shedder.state.messages_dropped == 10 + assert shedder.state.messages_eligible == 20 + + def test_counts_decay_after_window_expires(self): + clock = FakeClock() + shedder = _make_shedder(clock) + _activate(shedder) + + messages = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(100)] + shedder.shed(messages) + assert shedder.state.messages_dropped == 50 + + # Advance past the full window + clock.advance(_N_BUCKETS * _BUCKET_DURATION_S + 1) + assert shedder.state.messages_dropped == 0 + assert shedder.state.messages_eligible == 0 + + def test_partial_window_decay(self): + clock = FakeClock() + shedder = _make_shedder(clock) + _activate(shedder) + + # Record in bucket 0 + batch = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(10)] + shedder.shed(batch) + dropped_first = shedder.state.messages_dropped + + # Advance to a new bucket and record more + clock.advance(_BUCKET_DURATION_S) + shedder.shed(batch) + assert shedder.state.messages_dropped > dropped_first + + # Advance so the first bucket expires but not the second + clock.advance((_N_BUCKETS - 1) * _BUCKET_DURATION_S) + state = shedder.state + # Only the second bucket's data should remain + assert state.messages_dropped == 5 + assert state.messages_eligible == 10 + + def test_eligible_tracked_when_not_shedding(self): + """Even when not shedding, eligible messages are counted.""" + clock = FakeClock() + shedder = _make_shedder(clock) + # Not activated — no shedding + messages = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(20)] + shedder.shed(messages) + state = shedder.state + assert state.messages_dropped == 0 + assert state.messages_eligible == 20 + + class TestLoadShedderState: def test_state_reflects_shedding(self): - shedder = LoadShedder() + shedder = _make_shedder() assert shedder.state.is_shedding is False - for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) + _activate(shedder) assert shedder.state.is_shedding is True def test_state_is_snapshot(self): - shedder = LoadShedder() + shedder = _make_shedder() state = shedder.state - for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) + _activate(shedder) # Original snapshot unchanged (frozen dataclass) assert state.is_shedding is False assert shedder.state.is_shedding is True diff --git a/tests/kafka/status_message_test.py b/tests/kafka/status_message_test.py index 3fa775f40..dd8310ac7 100644 --- a/tests/kafka/status_message_test.py +++ b/tests/kafka/status_message_test.py @@ -783,12 +783,15 @@ def test_to_service_status(self): def test_round_trip_with_shedding_fields(self): """Test that shedding fields survive model round-trip.""" - original = make_service_status(is_shedding=True, messages_dropped=42) + original = make_service_status( + is_shedding=True, messages_dropped=42, messages_eligible=100 + ) msg = ServiceStatusMessage.from_service_status(original) converted = msg.to_service_status() assert converted.is_shedding is True assert converted.messages_dropped == 42 + assert converted.messages_eligible == 100 def test_round_trip_defaults_shedding_fields(self): """Test that shedding fields default gracefully.""" @@ -798,6 +801,7 @@ def test_round_trip_defaults_shedding_fields(self): assert converted.is_shedding is False assert converted.messages_dropped == 0 + assert converted.messages_eligible == 0 class TestServiceStatusX5F2Integration: @@ -828,12 +832,15 @@ def test_service_status_x5f2_round_trip(self): def test_service_status_x5f2_round_trip_with_shedding(self): """Test x5f2 round-trip includes load shedding fields.""" - original = make_service_status(is_shedding=True, messages_dropped=1234) + original = make_service_status( + is_shedding=True, messages_dropped=1234, messages_eligible=3000 + ) x5f2_data = service_status_to_x5f2(original) converted = x5f2_to_service_status(x5f2_data) assert converted.is_shedding is True assert converted.messages_dropped == 1234 + assert converted.messages_eligible == 3000 def test_service_status_x5f2_with_error(self): """Test x5f2 round-trip with error message.""" From 1fe6dcfa06f09dfe1d625738e682ea65907667ee Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Fri, 27 Feb 2026 14:22:57 +0000 Subject: [PATCH 03/10] Simplify drop rate display to percentage only The raw counts (62/164) next to the cumulative Msgs count was confusing since they cover different time windows. Show just "Dropped: 50%" which is unambiguous alongside the total "Msgs" throughput counter. Prompt: Remove raw counts from the drop rate display, just show percentage. Co-Authored-By: Claude Opus 4.6 --- .../livedata/dashboard/widgets/backend_status_widget.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/src/ess/livedata/dashboard/widgets/backend_status_widget.py b/src/ess/livedata/dashboard/widgets/backend_status_widget.py index 7ac727bcd..06fa8c617 100644 --- a/src/ess/livedata/dashboard/widgets/backend_status_widget.py +++ b/src/ess/livedata/dashboard/widgets/backend_status_widget.py @@ -75,14 +75,11 @@ def _format_messages(count: int) -> str: def _format_drop_rate(dropped: int, eligible: int) -> str: - """Format drop rate as percentage with counts over the rolling window.""" + """Format drop rate as percentage over the rolling window.""" if eligible == 0: - return f"Dropped: {_format_messages(dropped)}" + return "Dropped: <1%" pct = 100 * dropped / eligible - return ( - f"Dropped: {pct:.0f}% " - f"({_format_messages(dropped)}/{_format_messages(eligible)})" - ) + return f"Dropped: {pct:.0f}%" class WorkerStatusRow: From 09482fd445f81695b2b470833ce08474a2f212ec Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Fri, 27 Feb 2026 14:24:16 +0000 Subject: [PATCH 04/10] Use "Dropping" instead of "Dropped" for active drop rate "Dropping: 50%" reads as a current rate, which better matches the rolling-window semantics than "Dropped" which implies a past event. Prompt: Change Dropped->Dropping to indicate current rate. Also update PR. Co-Authored-By: Claude Opus 4.6 --- src/ess/livedata/dashboard/widgets/backend_status_widget.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/ess/livedata/dashboard/widgets/backend_status_widget.py b/src/ess/livedata/dashboard/widgets/backend_status_widget.py index 06fa8c617..f099ade77 100644 --- a/src/ess/livedata/dashboard/widgets/backend_status_widget.py +++ b/src/ess/livedata/dashboard/widgets/backend_status_widget.py @@ -77,9 +77,9 @@ def _format_messages(count: int) -> str: def _format_drop_rate(dropped: int, eligible: int) -> str: """Format drop rate as percentage over the rolling window.""" if eligible == 0: - return "Dropped: <1%" + return "Dropping: <1%" pct = 100 * dropped / eligible - return f"Dropped: {pct:.0f}%" + return f"Dropping: {pct:.0f}%" class WorkerStatusRow: From 3ccf294c577f7c0833d3a4e4d8047da32690ec7a Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Fri, 27 Feb 2026 14:37:20 +0000 Subject: [PATCH 05/10] Add multi-level load shedding with exponential drop rates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The single-level shedder (fixed 50% drop) could not recover when the backend was overloaded by more than 2x. Replace the boolean on/off with escalating levels where each level keeps 1/2^N of droppable messages (level 1 = 50%, level 2 = 75%, level 3 = 87.5%, …). Escalation and de-escalation each require the same consecutive-cycle thresholds as before, applied per level step. Wire the new shedding_level field through ServiceStatus, x5f2 serialization, and structured metrics logging. The dashboard badge remains "SHEDDING" since the existing drop-rate percentage already conveys severity. Prompt: Help me understand the shedding strategy added in this branch. PR says it drops every other message, but what if that is not enough? Follow-up: Multi-level shedding sounds good, is it simple to implement? Follow-up: No need to cap, dropping 9/10 or more is still reasonable. Would it make sense to use sth. like 1/2->1/4->1/8->...? Follow-up: Note that I don't think there is a need to report the shedding level like "SHEDDING L2" if we report the drop rate? The two are redundant in a sense. Co-Authored-By: Claude Opus 4.6 --- src/ess/livedata/core/job.py | 1 + src/ess/livedata/core/load_shedder.py | 35 ++++---- .../livedata/core/orchestrating_processor.py | 2 + src/ess/livedata/kafka/x5f2_compat.py | 5 ++ tests/core/load_shedder_test.py | 81 +++++++++++++++++++ tests/kafka/status_message_test.py | 13 ++- 6 files changed, 121 insertions(+), 16 deletions(-) diff --git a/src/ess/livedata/core/job.py b/src/ess/livedata/core/job.py index 9137b00bf..f113a347a 100644 --- a/src/ess/livedata/core/job.py +++ b/src/ess/livedata/core/job.py @@ -123,6 +123,7 @@ class ServiceStatus: messages_processed: int error: str | None = None is_shedding: bool = False + shedding_level: int = 0 messages_dropped: int = 0 messages_eligible: int = 0 diff --git a/src/ess/livedata/core/load_shedder.py b/src/ess/livedata/core/load_shedder.py index 9a353da13..b67914dc4 100644 --- a/src/ess/livedata/core/load_shedder.py +++ b/src/ess/livedata/core/load_shedder.py @@ -37,6 +37,7 @@ class LoadShedderState: """Snapshot of load shedder state for status reporting.""" is_shedding: bool + shedding_level: int messages_dropped: int messages_eligible: int @@ -94,14 +95,18 @@ class LoadShedder: """Selectively drops bulk event data when the backend falls behind. Detection uses consecutive non-None batcher results as the overload signal. - When active, keeps every 2nd droppable message (50% reduction). + Shedding uses exponential levels: level N keeps every ``2**N``-th droppable + message. Each level handles a 2x increase in overload (level 1 = 50% drop, + level 2 = 75%, level 3 = 87.5%, …). The level escalates by 1 after + ``_ACTIVATION_THRESHOLD`` consecutive non-idle batcher cycles and + de-escalates by 1 after ``_DEACTIVATION_THRESHOLD`` consecutive idle cycles. Drop statistics are tracked over a rolling 60-second window. """ def __init__(self, *, clock: Callable[[], float] = time.monotonic) -> None: self._consecutive_batches: int = 0 self._consecutive_idle: int = 0 - self._is_shedding: bool = False + self._level: int = 0 self._subsample_counter: int = 0 self._rolling = _RollingCounter(clock=clock) @@ -109,7 +114,8 @@ def __init__(self, *, clock: Callable[[], float] = time.monotonic) -> None: def state(self) -> LoadShedderState: dropped, eligible = self._rolling.totals() return LoadShedderState( - is_shedding=self._is_shedding, + is_shedding=self._level > 0, + shedding_level=self._level, messages_dropped=dropped, messages_eligible=eligible, ) @@ -125,32 +131,33 @@ def report_batch_result(self, batch_produced: bool) -> None: if batch_produced: self._consecutive_batches += 1 self._consecutive_idle = 0 - if ( - not self._is_shedding - and self._consecutive_batches >= _ACTIVATION_THRESHOLD - ): - self._is_shedding = True + if self._consecutive_batches >= _ACTIVATION_THRESHOLD: + self._level += 1 + self._consecutive_batches = 0 else: self._consecutive_idle += 1 self._consecutive_batches = 0 - if self._is_shedding and self._consecutive_idle >= _DEACTIVATION_THRESHOLD: - self._is_shedding = False - self._subsample_counter = 0 + if self._level > 0 and self._consecutive_idle >= _DEACTIVATION_THRESHOLD: + self._level -= 1 + self._consecutive_idle = 0 + if self._level == 0: + self._subsample_counter = 0 def shed(self, messages: list[Message]) -> list[Message]: """Filter messages when shedding is active. When inactive, returns all messages unchanged. - When active, drops every other droppable message (50% reduction). + When active, keeps every ``2**level``-th droppable message. Non-droppable messages (control, f144 logs) are always preserved. Both active and inactive calls record eligible message counts into the rolling window so the drop rate reflects what fraction is being shed. """ eligible = sum(1 for m in messages if m.stream.kind in DROPPABLE_KINDS) - if not self._is_shedding: + if self._level == 0: self._rolling.record(dropped=0, eligible=eligible) return messages + keep_every = 2**self._level dropped = 0 result: list[Message] = [] for msg in messages: @@ -158,7 +165,7 @@ def shed(self, messages: list[Message]) -> list[Message]: result.append(msg) else: self._subsample_counter += 1 - if self._subsample_counter % 2 == 0: + if self._subsample_counter % keep_every == 0: result.append(msg) else: dropped += 1 diff --git a/src/ess/livedata/core/orchestrating_processor.py b/src/ess/livedata/core/orchestrating_processor.py index e07a56d27..d342c4199 100644 --- a/src/ess/livedata/core/orchestrating_processor.py +++ b/src/ess/livedata/core/orchestrating_processor.py @@ -240,6 +240,7 @@ def _get_service_status(self, job_statuses: list[JobStatus]) -> ServiceStatus: messages_processed=self._messages_processed, error=self._service_error, is_shedding=shedder_state.is_shedding if shedder_state else False, + shedding_level=shedder_state.shedding_level if shedder_state else 0, messages_dropped=shedder_state.messages_dropped if shedder_state else 0, messages_eligible=shedder_state.messages_eligible if shedder_state else 0, ) @@ -262,6 +263,7 @@ def _maybe_log_metrics(self) -> None: active_jobs=active_jobs, errors=self._errors_since_last_metrics, shedding=shedder_state.is_shedding if shedder_state else False, + shedding_level=(shedder_state.shedding_level if shedder_state else 0), messages_dropped=( shedder_state.messages_dropped if shedder_state else 0 ), diff --git a/src/ess/livedata/kafka/x5f2_compat.py b/src/ess/livedata/kafka/x5f2_compat.py index 9934f0c0f..fa459e082 100644 --- a/src/ess/livedata/kafka/x5f2_compat.py +++ b/src/ess/livedata/kafka/x5f2_compat.py @@ -200,6 +200,9 @@ class ServiceStatusPayload(pydantic.BaseModel): is_shedding: bool = pydantic.Field( default=False, description="Whether load shedding is active" ) + shedding_level: int = pydantic.Field( + default=0, description="Current shedding level (0=off, N=keep 1/2^N)" + ) messages_dropped: int = pydantic.Field( default=0, description="Messages dropped in the rolling window" ) @@ -292,6 +295,7 @@ def from_service_status( messages_processed=status.messages_processed, error=status.error, is_shedding=status.is_shedding, + shedding_level=status.shedding_level, messages_dropped=status.messages_dropped, messages_eligible=status.messages_eligible, ), @@ -311,6 +315,7 @@ def to_service_status(self) -> ServiceStatus: messages_processed=message.messages_processed, error=message.error, is_shedding=message.is_shedding, + shedding_level=message.shedding_level, messages_dropped=message.messages_dropped, messages_eligible=message.messages_eligible, ) diff --git a/tests/core/load_shedder_test.py b/tests/core/load_shedder_test.py index a39e3d2bf..ae0073af3 100644 --- a/tests/core/load_shedder_test.py +++ b/tests/core/load_shedder_test.py @@ -244,6 +244,12 @@ def test_state_reflects_shedding(self): _activate(shedder) assert shedder.state.is_shedding is True + def test_state_reports_shedding_level(self): + shedder = _make_shedder() + assert shedder.state.shedding_level == 0 + _activate(shedder) + assert shedder.state.shedding_level == 1 + def test_state_is_snapshot(self): shedder = _make_shedder() state = shedder.state @@ -251,3 +257,78 @@ def test_state_is_snapshot(self): # Original snapshot unchanged (frozen dataclass) assert state.is_shedding is False assert shedder.state.is_shedding is True + + +def _escalate_to(shedder: LoadShedder, level: int) -> None: + """Escalate the shedder to the given level.""" + for _ in range(level): + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + assert shedder.state.shedding_level == level + + +def _deescalate_by(shedder: LoadShedder, steps: int) -> None: + """De-escalate the shedder by the given number of steps.""" + for _ in range(steps): + for _ in range(_DEACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=False) + + +class TestMultiLevelEscalation: + def test_escalates_to_level_2(self): + shedder = _make_shedder() + _escalate_to(shedder, 2) + assert shedder.state.shedding_level == 2 + + def test_escalates_to_level_3(self): + shedder = _make_shedder() + _escalate_to(shedder, 3) + assert shedder.state.shedding_level == 3 + + def test_escalation_requires_threshold_per_level(self): + shedder = _make_shedder() + _escalate_to(shedder, 1) + # Not enough batches for next level + for _ in range(_ACTIVATION_THRESHOLD - 1): + shedder.report_batch_result(batch_produced=True) + assert shedder.state.shedding_level == 1 + + +class TestMultiLevelDeescalation: + def test_deescalates_one_level_at_a_time(self): + shedder = _make_shedder() + _escalate_to(shedder, 3) + _deescalate_by(shedder, 1) + assert shedder.state.shedding_level == 2 + + def test_deescalates_to_zero(self): + shedder = _make_shedder() + _escalate_to(shedder, 2) + _deescalate_by(shedder, 2) + assert shedder.state.shedding_level == 0 + assert shedder.state.is_shedding is False + + def test_deescalation_requires_threshold_per_level(self): + shedder = _make_shedder() + _escalate_to(shedder, 2) + for _ in range(_DEACTIVATION_THRESHOLD - 1): + shedder.report_batch_result(batch_produced=False) + assert shedder.state.shedding_level == 2 + + +class TestMultiLevelDropRates: + @pytest.mark.parametrize( + ("level", "expected_kept"), + [ + (1, 128), # keep 1/2 of 256 + (2, 64), # keep 1/4 of 256 + (3, 32), # keep 1/8 of 256 + (4, 16), # keep 1/16 of 256 + ], + ) + def test_drop_rate_at_level(self, level, expected_kept): + shedder = _make_shedder() + _escalate_to(shedder, level) + messages = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(256)] + result = shedder.shed(messages) + assert len(result) == expected_kept diff --git a/tests/kafka/status_message_test.py b/tests/kafka/status_message_test.py index dd8310ac7..f202277ba 100644 --- a/tests/kafka/status_message_test.py +++ b/tests/kafka/status_message_test.py @@ -784,12 +784,16 @@ def test_to_service_status(self): def test_round_trip_with_shedding_fields(self): """Test that shedding fields survive model round-trip.""" original = make_service_status( - is_shedding=True, messages_dropped=42, messages_eligible=100 + is_shedding=True, + shedding_level=2, + messages_dropped=42, + messages_eligible=100, ) msg = ServiceStatusMessage.from_service_status(original) converted = msg.to_service_status() assert converted.is_shedding is True + assert converted.shedding_level == 2 assert converted.messages_dropped == 42 assert converted.messages_eligible == 100 @@ -800,6 +804,7 @@ def test_round_trip_defaults_shedding_fields(self): converted = msg.to_service_status() assert converted.is_shedding is False + assert converted.shedding_level == 0 assert converted.messages_dropped == 0 assert converted.messages_eligible == 0 @@ -833,12 +838,16 @@ def test_service_status_x5f2_round_trip(self): def test_service_status_x5f2_round_trip_with_shedding(self): """Test x5f2 round-trip includes load shedding fields.""" original = make_service_status( - is_shedding=True, messages_dropped=1234, messages_eligible=3000 + is_shedding=True, + shedding_level=3, + messages_dropped=1234, + messages_eligible=3000, ) x5f2_data = service_status_to_x5f2(original) converted = x5f2_to_service_status(x5f2_data) assert converted.is_shedding is True + assert converted.shedding_level == 3 assert converted.messages_dropped == 1234 assert converted.messages_eligible == 3000 From 364f1cde46c30eba59f426ea33c31c490813356f Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 2 Mar 2026 06:41:01 +0000 Subject: [PATCH 06/10] Log shedding level transitions at warning severity Escalation, de-escalation, and full stop are discrete events worth surfacing above the periodic INFO metrics. Each log line includes the new level and keep rate for quick diagnosis. Prompt: I wonder if we should increase the logging severity when we shed? Follow-up: Maybe you are right that the periodic metric logging should stay at INFO level, but changes in shedding level might be worth a warning? Co-Authored-By: Claude Opus 4.6 --- src/ess/livedata/core/load_shedder.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/ess/livedata/core/load_shedder.py b/src/ess/livedata/core/load_shedder.py index b67914dc4..c7bcafd43 100644 --- a/src/ess/livedata/core/load_shedder.py +++ b/src/ess/livedata/core/load_shedder.py @@ -12,8 +12,12 @@ from collections.abc import Callable from dataclasses import dataclass +import structlog + from .message import Message, StreamKind +logger = structlog.get_logger(__name__) + DROPPABLE_KINDS = frozenset( { StreamKind.DETECTOR_EVENTS, @@ -134,6 +138,12 @@ def report_batch_result(self, batch_produced: bool) -> None: if self._consecutive_batches >= _ACTIVATION_THRESHOLD: self._level += 1 self._consecutive_batches = 0 + keep_rate = 100 / 2**self._level + logger.warning( + 'shedding_escalated', + level=self._level, + keep_rate=f"{keep_rate:.1f}%", + ) else: self._consecutive_idle += 1 self._consecutive_batches = 0 @@ -142,6 +152,14 @@ def report_batch_result(self, batch_produced: bool) -> None: self._consecutive_idle = 0 if self._level == 0: self._subsample_counter = 0 + logger.warning('shedding_stopped') + else: + keep_rate = 100 / 2**self._level + logger.warning( + 'shedding_deescalated', + level=self._level, + keep_rate=f"{keep_rate:.1f}%", + ) def shed(self, messages: list[Message]) -> list[Message]: """Filter messages when shedding is active. From b652d1f98fb4e2e8ca25997584453b3ff0073e1f Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 2 Mar 2026 06:55:13 +0000 Subject: [PATCH 07/10] Cap shedding at level 3 and fix keep-rate display Without a cap, high shedding levels make processing cycles near-instant, causing the "consecutive non-None batches" overload signal to fire rapidly in a positive feedback loop. Level 3 (keep 1/8, 87.5% drop) handles up to 8x overload while still producing usable data (~1-2 messages per source per batch at typical rates of 14 msg/s/source with 5-10 sources per worker). Also switch keep-rate log format from percentage (which underflows to 0.0% at high levels) to a fraction like "1/8". Prompt: What is that rapid escalation then deescalation? and 0.0% is not useful. Follow-up: In practice we expect each source to produce 14 messages/second, and each worker would process 5-10 sources. Leaving aside that the current indiscriminate shedding is an issue, what might a good max level be? Co-Authored-By: Claude Opus 4.6 --- src/ess/livedata/core/load_shedder.py | 20 ++++++++++++-------- tests/core/load_shedder_test.py | 15 +++++++++++---- 2 files changed, 23 insertions(+), 12 deletions(-) diff --git a/src/ess/livedata/core/load_shedder.py b/src/ess/livedata/core/load_shedder.py index c7bcafd43..34b72d5fd 100644 --- a/src/ess/livedata/core/load_shedder.py +++ b/src/ess/livedata/core/load_shedder.py @@ -32,6 +32,8 @@ # Consecutive idle (None) batcher results before exiting shedding mode _DEACTIVATION_THRESHOLD = 3 +_MAX_LEVEL = 3 + _N_BUCKETS = 10 _BUCKET_DURATION_S = 6.0 # 10 buckets x 6s = 60s rolling window @@ -101,9 +103,10 @@ class LoadShedder: Detection uses consecutive non-None batcher results as the overload signal. Shedding uses exponential levels: level N keeps every ``2**N``-th droppable message. Each level handles a 2x increase in overload (level 1 = 50% drop, - level 2 = 75%, level 3 = 87.5%, …). The level escalates by 1 after - ``_ACTIVATION_THRESHOLD`` consecutive non-idle batcher cycles and - de-escalates by 1 after ``_DEACTIVATION_THRESHOLD`` consecutive idle cycles. + level 2 = 75%, level 3 = 87.5%). The level escalates by 1 after + ``_ACTIVATION_THRESHOLD`` consecutive non-idle batcher cycles, up to + ``_MAX_LEVEL``, and de-escalates by 1 after ``_DEACTIVATION_THRESHOLD`` + consecutive idle cycles. Drop statistics are tracked over a rolling 60-second window. """ @@ -135,14 +138,16 @@ def report_batch_result(self, batch_produced: bool) -> None: if batch_produced: self._consecutive_batches += 1 self._consecutive_idle = 0 - if self._consecutive_batches >= _ACTIVATION_THRESHOLD: + if ( + self._consecutive_batches >= _ACTIVATION_THRESHOLD + and self._level < _MAX_LEVEL + ): self._level += 1 self._consecutive_batches = 0 - keep_rate = 100 / 2**self._level logger.warning( 'shedding_escalated', level=self._level, - keep_rate=f"{keep_rate:.1f}%", + keeping=f"1/{2**self._level}", ) else: self._consecutive_idle += 1 @@ -154,11 +159,10 @@ def report_batch_result(self, batch_produced: bool) -> None: self._subsample_counter = 0 logger.warning('shedding_stopped') else: - keep_rate = 100 / 2**self._level logger.warning( 'shedding_deescalated', level=self._level, - keep_rate=f"{keep_rate:.1f}%", + keeping=f"1/{2**self._level}", ) def shed(self, messages: list[Message]) -> list[Message]: diff --git a/tests/core/load_shedder_test.py b/tests/core/load_shedder_test.py index ae0073af3..5f06973c0 100644 --- a/tests/core/load_shedder_test.py +++ b/tests/core/load_shedder_test.py @@ -7,6 +7,7 @@ _ACTIVATION_THRESHOLD, _BUCKET_DURATION_S, _DEACTIVATION_THRESHOLD, + _MAX_LEVEL, _N_BUCKETS, DROPPABLE_KINDS, LoadShedder, @@ -280,10 +281,17 @@ def test_escalates_to_level_2(self): _escalate_to(shedder, 2) assert shedder.state.shedding_level == 2 - def test_escalates_to_level_3(self): + def test_escalates_to_max_level(self): shedder = _make_shedder() - _escalate_to(shedder, 3) - assert shedder.state.shedding_level == 3 + _escalate_to(shedder, _MAX_LEVEL) + assert shedder.state.shedding_level == _MAX_LEVEL + + def test_does_not_escalate_beyond_max_level(self): + shedder = _make_shedder() + _escalate_to(shedder, _MAX_LEVEL) + for _ in range(_ACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_produced=True) + assert shedder.state.shedding_level == _MAX_LEVEL def test_escalation_requires_threshold_per_level(self): shedder = _make_shedder() @@ -323,7 +331,6 @@ class TestMultiLevelDropRates: (1, 128), # keep 1/2 of 256 (2, 64), # keep 1/4 of 256 (3, 32), # keep 1/8 of 256 - (4, 16), # keep 1/16 of 256 ], ) def test_drop_rate_at_level(self, level, expected_kept): From 4b842a236fcb820c3039766ac10d3b27dddf8778 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 2 Mar 2026 08:17:24 +0000 Subject: [PATCH 08/10] Fix load shedding false positive on batcher timestamp catch-up The SimpleMessageBatcher emits empty batches (non-None with zero messages) when message timestamps jump forward after a pause between measurements. These were incorrectly counted as overload signals, triggering shedding during normal operation. Changed report_batch_result to accept batch_message_count (int) instead of batch_produced (bool). Empty batches and no-batch cycles are both treated as idle, so only genuine data-carrying batches contribute to the activation threshold. Added documentation explaining why consecutive non-empty batches are a reliable overload signal (given the batcher's 1-second time windows and the ~10 ms processing loop) and why empty batches must be excluded. Prompt: "Please use a new worktree to help me think through #739 (branch load-shedding). I am particularly worried about whether the overload-detection strategy is safe in the sense that I want to be really sure that shedding does not get activated unless we actually have to. Are there scenarios where the system is operating normally but we nevertheless enter shedding?" Follow-up: "I was more worried about the batcher genuinely never returning None because the upstream message rate is much higher than our processing frequency. Is the shedding strategy only working as a coincidence given how the current batcher works?" Follow-up: "We should fix this (and add tests) and also document clearly what you explained above in docstrings and code comments." Co-Authored-By: Claude Opus 4.6 --- src/ess/livedata/core/load_shedder.py | 48 +++++++++-- .../livedata/core/orchestrating_processor.py | 5 +- tests/core/load_shedder_test.py | 83 +++++++++++++++---- 3 files changed, 112 insertions(+), 24 deletions(-) diff --git a/src/ess/livedata/core/load_shedder.py b/src/ess/livedata/core/load_shedder.py index 34b72d5fd..be7b19805 100644 --- a/src/ess/livedata/core/load_shedder.py +++ b/src/ess/livedata/core/load_shedder.py @@ -4,6 +4,20 @@ When the backend can't keep up with the Kafka message stream, the LoadShedder selectively drops bulk event data while preserving control messages and f144 logs. + +Overload detection relies on the ``SimpleMessageBatcher`` producing consecutive +non-empty batches. Under normal operation, the batcher uses 1-second time windows +aligned to message timestamps. Because the processing loop runs at ~10 ms intervals, +each cycle fetches only ~10 ms worth of messages — well within the current window — +so ``batch()`` returns None roughly 99 out of 100 calls. A non-None result means +messages have crossed a window boundary. Consecutive non-None results mean the +processor could not drain the window before the next boundary arrived, i.e., it is +falling behind real-time. + +Empty batches (non-None but with zero messages) are excluded from the overload signal. +The batcher emits these when message timestamps jump forward (e.g., after a pause +between measurement runs) to step through the gap one window at a time. These do not +indicate overload and must not trigger shedding. """ from __future__ import annotations @@ -27,9 +41,10 @@ } ) -# Consecutive non-None batcher results before entering shedding mode +# Consecutive non-empty batcher results before entering shedding mode. +# With 1-second batch windows this means ~5 seconds of sustained overload. _ACTIVATION_THRESHOLD = 5 -# Consecutive idle (None) batcher results before exiting shedding mode +# Consecutive idle cycles (no batch, or empty batch) before de-escalating one level. _DEACTIVATION_THRESHOLD = 3 _MAX_LEVEL = 3 @@ -100,13 +115,24 @@ def totals(self) -> tuple[int, int]: class LoadShedder: """Selectively drops bulk event data when the backend falls behind. - Detection uses consecutive non-None batcher results as the overload signal. + Overload detection counts consecutive non-empty batches produced by the + message batcher. A non-empty batch means messages crossed a time-window + boundary, which happens approximately once per batch window under normal + load. Consecutive non-empty batches mean the processor is not keeping up: + by the time one batch is processed, enough new messages have arrived to + immediately complete the next window. + + Empty batches (non-None result with zero messages) are explicitly excluded. + The ``SimpleMessageBatcher`` emits these to step through timestamp gaps + (e.g., after a pause between measurements) and they do not indicate load. + Shedding uses exponential levels: level N keeps every ``2**N``-th droppable message. Each level handles a 2x increase in overload (level 1 = 50% drop, level 2 = 75%, level 3 = 87.5%). The level escalates by 1 after - ``_ACTIVATION_THRESHOLD`` consecutive non-idle batcher cycles, up to + ``_ACTIVATION_THRESHOLD`` consecutive non-empty batches, up to ``_MAX_LEVEL``, and de-escalates by 1 after ``_DEACTIVATION_THRESHOLD`` consecutive idle cycles. + Drop statistics are tracked over a rolling 60-second window. """ @@ -127,15 +153,21 @@ def state(self) -> LoadShedderState: messages_eligible=eligible, ) - def report_batch_result(self, batch_produced: bool) -> None: + def report_batch_result(self, batch_message_count: int) -> None: """Update overload detection counters after a batcher cycle. + Only batches with at least one message count toward the activation + threshold. Empty batches (zero messages) are treated as idle because + they arise from the batcher stepping through timestamp gaps, not from + genuine overload. + Parameters ---------- - batch_produced: - True if the batcher returned a batch (non-None), False if idle (None). + batch_message_count: + Number of messages in the batch returned by the batcher, or 0 if + the batcher returned None (no batch) or an empty batch. """ - if batch_produced: + if batch_message_count > 0: self._consecutive_batches += 1 self._consecutive_idle = 0 if ( diff --git a/src/ess/livedata/core/orchestrating_processor.py b/src/ess/livedata/core/orchestrating_processor.py index d342c4199..f869ee9b4 100644 --- a/src/ess/livedata/core/orchestrating_processor.py +++ b/src/ess/livedata/core/orchestrating_processor.py @@ -150,7 +150,10 @@ def process(self) -> None: data_messages = self._load_shedder.shed(data_messages) message_batch = self._message_batcher.batch(data_messages) if self._load_shedder is not None: - self._load_shedder.report_batch_result(message_batch is not None) + # Empty batches (from batcher timestamp catch-up) are not an + # overload signal — only count batches that carry data. + count = len(message_batch.messages) if message_batch is not None else 0 + self._load_shedder.report_batch_result(count) if message_batch is None: self._empty_batches += 1 self._maybe_log_metrics() diff --git a/tests/core/load_shedder_test.py b/tests/core/load_shedder_test.py index 5f06973c0..118fb518c 100644 --- a/tests/core/load_shedder_test.py +++ b/tests/core/load_shedder_test.py @@ -40,7 +40,7 @@ def _make_shedder(clock: FakeClock | None = None) -> LoadShedder: def _activate(shedder: LoadShedder) -> None: for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) + shedder.report_batch_result(batch_message_count=10) assert shedder.state.is_shedding is True @@ -66,17 +66,17 @@ def test_activates_after_consecutive_batches(self): def test_does_not_activate_below_threshold(self): shedder = _make_shedder() for _ in range(_ACTIVATION_THRESHOLD - 1): - shedder.report_batch_result(batch_produced=True) + shedder.report_batch_result(batch_message_count=10) assert shedder.state.is_shedding is False def test_idle_cycle_resets_consecutive_count(self): shedder = _make_shedder() for _ in range(_ACTIVATION_THRESHOLD - 1): - shedder.report_batch_result(batch_produced=True) - shedder.report_batch_result(batch_produced=False) + shedder.report_batch_result(batch_message_count=10) + shedder.report_batch_result(batch_message_count=0) # Restart counting — should not activate after fewer than threshold for _ in range(_ACTIVATION_THRESHOLD - 1): - shedder.report_batch_result(batch_produced=True) + shedder.report_batch_result(batch_message_count=10) assert shedder.state.is_shedding is False @@ -89,21 +89,21 @@ def active_shedder(self): def test_deactivates_after_consecutive_idle(self, active_shedder): for _ in range(_DEACTIVATION_THRESHOLD): - active_shedder.report_batch_result(batch_produced=False) + active_shedder.report_batch_result(batch_message_count=0) assert active_shedder.state.is_shedding is False def test_does_not_deactivate_below_threshold(self, active_shedder): for _ in range(_DEACTIVATION_THRESHOLD - 1): - active_shedder.report_batch_result(batch_produced=False) + active_shedder.report_batch_result(batch_message_count=0) assert active_shedder.state.is_shedding is True def test_batch_resets_idle_count(self, active_shedder): for _ in range(_DEACTIVATION_THRESHOLD - 1): - active_shedder.report_batch_result(batch_produced=False) - active_shedder.report_batch_result(batch_produced=True) + active_shedder.report_batch_result(batch_message_count=0) + active_shedder.report_batch_result(batch_message_count=10) # Restart idle counting for _ in range(_DEACTIVATION_THRESHOLD - 1): - active_shedder.report_batch_result(batch_produced=False) + active_shedder.report_batch_result(batch_message_count=0) assert active_shedder.state.is_shedding is True @@ -264,7 +264,7 @@ def _escalate_to(shedder: LoadShedder, level: int) -> None: """Escalate the shedder to the given level.""" for _ in range(level): for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) + shedder.report_batch_result(batch_message_count=10) assert shedder.state.shedding_level == level @@ -272,7 +272,7 @@ def _deescalate_by(shedder: LoadShedder, steps: int) -> None: """De-escalate the shedder by the given number of steps.""" for _ in range(steps): for _ in range(_DEACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=False) + shedder.report_batch_result(batch_message_count=0) class TestMultiLevelEscalation: @@ -290,7 +290,7 @@ def test_does_not_escalate_beyond_max_level(self): shedder = _make_shedder() _escalate_to(shedder, _MAX_LEVEL) for _ in range(_ACTIVATION_THRESHOLD): - shedder.report_batch_result(batch_produced=True) + shedder.report_batch_result(batch_message_count=10) assert shedder.state.shedding_level == _MAX_LEVEL def test_escalation_requires_threshold_per_level(self): @@ -298,7 +298,7 @@ def test_escalation_requires_threshold_per_level(self): _escalate_to(shedder, 1) # Not enough batches for next level for _ in range(_ACTIVATION_THRESHOLD - 1): - shedder.report_batch_result(batch_produced=True) + shedder.report_batch_result(batch_message_count=10) assert shedder.state.shedding_level == 1 @@ -320,7 +320,7 @@ def test_deescalation_requires_threshold_per_level(self): shedder = _make_shedder() _escalate_to(shedder, 2) for _ in range(_DEACTIVATION_THRESHOLD - 1): - shedder.report_batch_result(batch_produced=False) + shedder.report_batch_result(batch_message_count=0) assert shedder.state.shedding_level == 2 @@ -339,3 +339,56 @@ def test_drop_rate_at_level(self, level, expected_kept): messages = [_make_message(StreamKind.DETECTOR_EVENTS) for _ in range(256)] result = shedder.shed(messages) assert len(result) == expected_kept + + +class TestEmptyBatchesIgnored: + """Empty batches from batcher timestamp catch-up must not trigger shedding. + + The SimpleMessageBatcher emits empty batches (non-None with 0 messages) when + message timestamps jump forward (e.g., after a pause between measurement runs). + These are reported as batch_message_count=0 and must be treated as idle cycles. + """ + + def test_consecutive_empty_batches_do_not_activate(self): + shedder = _make_shedder() + for _ in range(_ACTIVATION_THRESHOLD + 5): + shedder.report_batch_result(batch_message_count=0) + assert shedder.state.is_shedding is False + + def test_empty_batches_reset_consecutive_count(self): + """An empty batch between data batches resets the overload counter.""" + shedder = _make_shedder() + for _ in range(_ACTIVATION_THRESHOLD - 1): + shedder.report_batch_result(batch_message_count=10) + # Empty batch interrupts the streak + shedder.report_batch_result(batch_message_count=0) + for _ in range(_ACTIVATION_THRESHOLD - 1): + shedder.report_batch_result(batch_message_count=10) + assert shedder.state.is_shedding is False + + def test_empty_batches_count_toward_deactivation(self): + """Empty batches count as idle and contribute to de-escalation.""" + shedder = _make_shedder() + _activate(shedder) + for _ in range(_DEACTIVATION_THRESHOLD): + shedder.report_batch_result(batch_message_count=0) + assert shedder.state.is_shedding is False + + def test_timestamp_gap_scenario(self): + """Simulate the batcher catch-up after a 5+ second timestamp gap. + + The batcher emits one empty batch per window to step through the gap. + None of these should trigger shedding. + """ + shedder = _make_shedder() + # Normal operation: occasional data batches with idle cycles between + shedder.report_batch_result(batch_message_count=50) + for _ in range(5): + shedder.report_batch_result(batch_message_count=0) + # Gap: 7 consecutive empty batches (batcher catching up through gap) + for _ in range(7): + shedder.report_batch_result(batch_message_count=0) + assert shedder.state.is_shedding is False + # Normal operation resumes with one data batch + shedder.report_batch_result(batch_message_count=50) + assert shedder.state.is_shedding is False From 0bad11277bd49819949e38d567f6125138e0dc81 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 2 Mar 2026 08:24:01 +0000 Subject: [PATCH 09/10] Fix misleading module docstring about processing cycle timing The docstring claimed each cycle fetches "only ~10 ms worth of messages" based on the poll interval. Under real load, processing itself takes hundreds of milliseconds, so the relevant question is whether the total cycle (fetch + process + publish) fits within the 1-second batch window. Reworded to describe the actual mechanism. Co-Authored-By: Claude Opus 4.6 --- src/ess/livedata/core/load_shedder.py | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/src/ess/livedata/core/load_shedder.py b/src/ess/livedata/core/load_shedder.py index be7b19805..424a2c97f 100644 --- a/src/ess/livedata/core/load_shedder.py +++ b/src/ess/livedata/core/load_shedder.py @@ -6,13 +6,14 @@ selectively drops bulk event data while preserving control messages and f144 logs. Overload detection relies on the ``SimpleMessageBatcher`` producing consecutive -non-empty batches. Under normal operation, the batcher uses 1-second time windows -aligned to message timestamps. Because the processing loop runs at ~10 ms intervals, -each cycle fetches only ~10 ms worth of messages — well within the current window — -so ``batch()`` returns None roughly 99 out of 100 calls. A non-None result means -messages have crossed a window boundary. Consecutive non-None results mean the -processor could not drain the window before the next boundary arrived, i.e., it is -falling behind real-time. +non-empty batches. The batcher uses 1-second time windows aligned to message +timestamps: ``batch()`` returns None while all incoming messages fall within the +current window, and returns a non-None batch only when a message crosses the window +boundary. Under normal load, the total processing cycle (fetch → preprocess → +workflow → publish) completes well within one batch window, so the messages fetched +in the next cycle still fall within the same window — ``batch()`` returns None. +Consecutive non-None results mean the processing cycle consistently takes longer +than the batch window, so messages accumulate past the next boundary on every call. Empty batches (non-None but with zero messages) are excluded from the overload signal. The batcher emits these when message timestamps jump forward (e.g., after a pause From f8c760da71028ffb711ecc9a66ee731bfc9b45d8 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 2 Mar 2026 08:29:24 +0000 Subject: [PATCH 10/10] Document effective capacity threshold from idle sleep The 100 ms idle sleep in process() means the total cycle time is processing_time + N*100ms, so shedding activates at ~90% utilization rather than exactly 100%. This is a desirable safety margin. Co-Authored-By: Claude Opus 4.6 --- src/ess/livedata/core/load_shedder.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/ess/livedata/core/load_shedder.py b/src/ess/livedata/core/load_shedder.py index 424a2c97f..77828fe74 100644 --- a/src/ess/livedata/core/load_shedder.py +++ b/src/ess/livedata/core/load_shedder.py @@ -15,6 +15,15 @@ Consecutive non-None results mean the processing cycle consistently takes longer than the batch window, so messages accumulate past the next boundary on every call. +The effective capacity threshold is slightly below 100% because of the 100 ms idle +sleep in ``OrchestratingProcessor.process()``. When the batcher returns None (no +boundary crossed yet), the processor sleeps 100 ms before the next poll. This means +the total cycle time is ``processing_time + N * 100 ms`` (where N ≥ 1 idle cycles). +With 1-second batch windows, shedding can activate when processing alone takes +roughly 900 ms or more — about 90% utilization. This built-in safety margin is +desirable: a system at 90%+ utilization has almost no headroom for traffic bursts +or GC pauses. + Empty batches (non-None but with zero messages) are excluded from the overload signal. The batcher emits these when message timestamps jump forward (e.g., after a pause between measurement runs) to step through the gap one window at a time. These do not