From 2f99e354376972a1e9b2b5da37b086769b0d2724 Mon Sep 17 00:00:00 2001
From: OutstanderWang <wangweiyanster@gmail.com>
Date: Wed, 2 Sep 2026 23:32:41 +0800
Subject: [PATCH 1/8] [storage] Raise the accept queue and measure where a
 request was lost

Builds on the retry in #171, which established that a request is lost between the
two ends rather than queued behind slow work. Two things that retry does not cover.

Raise the client-facing ROUTER's accept queue from ZMQ's default of 100, which was
just the unset default. A full accept queue is emptied without sending an RST
(tcp_abort_on_overflow=0), leaving the client established and waiting for a reply
nobody will send, while the server never learns the request existed. The observed
hung unit's listen socket had already charged 23 drops, and machine-wide
ListenOverflows equalled ListenDrops, which is the signature of exactly that. It is
a second silent-loss path alongside the unbounded DEALER queue. Tunable via
TQ_STORAGE_ZMQ_BACKLOG; set it to 100 to restore the old value for an A/B run.

Count requests as the worker decodes them, and sample the accept queue when asked.
The per-op counters advance inside monitor.measure(), so they only move once a
request completes and a request that arrived but never finished reads exactly like
one that never arrived; the diagnostic probe cannot tell those apart either. An
arrival count next to the completion histograms does, and a shortfall against the
caller's send count localizes the loss to one side. The queue-depth probe is opt-in
via TQ_ACCEPT_PROBE_INTERVAL and off by default: it shells out to ss on a timer, and
depth has to be sampled sub-second because the queue drains in milliseconds, which
is why inspecting a unit after it hung always read zero.

Tests cover the probe's peak and delta arithmetic and the levels it logs at.

Signed-off-by: OutstanderWang <wangweiyanster@gmail.com>
---
 tests/test_accept_probe.py               | 150 +++++++++++++++
 transfer_queue/storage/simple_storage.py |  57 ++++++
 transfer_queue/utils/accept_probe.py     | 226 +++++++++++++++++++++++
 3 files changed, 433 insertions(+)
 create mode 100644 tests/test_accept_probe.py
 create mode 100644 transfer_queue/utils/accept_probe.py

diff --git a/tests/test_accept_probe.py b/tests/test_accept_probe.py
new file mode 100644
index 00000000..249ecd78
--- /dev/null
+++ b/tests/test_accept_probe.py
@@ -0,0 +1,150 @@
+# Copyright 2025 Huawei Technologies Co., Ltd. All Rights Reserved.
+# Copyright 2025 The TransferQueue Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests for the accept-queue probe.
+
+A silently dropped connection leaves the client in ESTABLISHED with no reply and the
+unit's worker idle, which is what post-mortem inspection of a hang actually showed.
+The probe turns that guess into a measurement, so these tests pin the arithmetic it
+reports and the levels it logs at.
+"""
+
+from transfer_queue.utils.accept_probe import (
+    AcceptQueueProbe,
+    AcceptQueueSample,
+    AcceptQueueStats,
+    _read_listen_overflows,
+    sample_accept_queue,
+)
+
+
+def _sample(recv_q: int, backlog: int = 100, sk_drops: int = 0, overflows: int = 0) -> AcceptQueueSample:
+    return AcceptQueueSample(
+        timestamp=0.0,
+        recv_q=recv_q,
+        backlog=backlog,
+        sk_drops=sk_drops,
+        listen_overflows=overflows,
+        listen_drops=overflows,
+    )
+
+
+def _probe() -> AcceptQueueProbe:
+    return AcceptQueueProbe(port=34513, owner_id="TQ_STORAGE_UNIT_test", interval_s=0.01)
+
+
+def test_utilization_is_depth_over_backlog():
+    assert _sample(50, backlog=100).utilization == 0.5
+
+
+def test_zero_backlog_reports_zero_utilization_not_zero_division():
+    """A socket read before bind reports backlog 0; that must not raise."""
+    assert _sample(5, backlog=0).utilization == 0.0
+
+
+def test_full_queue_reports_full_utilization():
+    assert _sample(100, backlog=100).utilization == 1.0
+
+
+def test_empty_stats_report_zero_deltas():
+    stats = AcceptQueueStats(port=1234)
+
+    assert stats.sk_drops_delta == 0
+    assert stats.overflow_delta == 0
+
+
+def test_drop_delta_is_last_minus_first():
+    """Deltas are what matter: the absolute counters carry the whole uptime's history."""
+    stats = AcceptQueueStats(port=1234)
+    stats.first_sample = _sample(0, sk_drops=23, overflows=1775)
+    stats.last_sample = _sample(0, sk_drops=31, overflows=1790)
+
+    assert stats.sk_drops_delta == 8
+    assert stats.overflow_delta == 15
+
+
+def test_describe_names_port_and_deltas():
+    stats = AcceptQueueStats(port=34513, backlog=100, peak_recv_q=97)
+    stats.first_sample = _sample(0, sk_drops=23)
+    stats.last_sample = _sample(0, sk_drops=24)
+
+    text = stats.describe()
+
+    assert "port=34513" in text
+    assert "peak_recv_q=97" in text
+    assert "sk_drops_delta=1" in text
+
+
+def test_peak_tracks_highest_depth_not_latest():
+    """The burst is the signal; sampling after it drains reads zero."""
+    probe = _probe()
+
+    probe._record(_sample(10))
+    probe._record(_sample(97))
+    probe._record(_sample(3))
+
+    assert probe.stats.peak_recv_q == 97
+
+
+def test_first_sample_is_retained_as_delta_baseline():
+    probe = _probe()
+
+    probe._record(_sample(1, sk_drops=23))
+    probe._record(_sample(2, sk_drops=25))
+
+    assert probe.stats.first_sample.sk_drops == 23
+    assert probe.stats.sk_drops_delta == 2
+
+
+def test_drop_increase_is_logged_at_error(caplog):
+    """A drop is the direct evidence, so it must not be buried at debug level."""
+    probe = _probe()
+    probe._record(_sample(0, sk_drops=23))
+
+    with caplog.at_level("ERROR"):
+        probe._record(_sample(100, sk_drops=24))
+
+    assert "accept queue dropped a connection" in caplog.text
+
+
+def test_near_full_queue_warns_once(caplog):
+    """Repeating the warning every 100ms would flood the log of a 512-node job."""
+    probe = _probe()
+
+    with caplog.at_level("WARNING"):
+        probe._record(_sample(60))
+        probe._record(_sample(70))
+
+    assert caplog.text.count("reached") == 1
+
+
+def test_no_warning_below_threshold():
+    probe = _probe()
+
+    probe._record(_sample(10))
+
+    assert probe._warned is False
+
+
+def test_listen_overflows_returns_two_non_negative_ints():
+    """Reads the real host, so assert shape rather than a specific value."""
+    overflows, drops = _read_listen_overflows()
+
+    assert overflows >= 0
+    assert drops >= 0
+
+
+def test_sampling_an_unused_port_returns_none():
+    assert sample_accept_queue(1) is None
diff --git a/transfer_queue/storage/simple_storage.py b/transfer_queue/storage/simple_storage.py
index cd0d2b4b..2b363602 100644
--- a/transfer_queue/storage/simple_storage.py
+++ b/transfer_queue/storage/simple_storage.py
@@ -52,6 +52,14 @@
 # Marks a GET_ERROR reply as "the key is gone" so the caller can tell it apart from a real fault.
 KEY_NOT_FOUND_MARKER = "TQKeyNotFound"
 
+# Accept-queue depth for the client-facing ROUTER, well above ZMQ's default of 100 because a
+# full accept queue is drained without an RST and so loses connections silently.
+TQ_STORAGE_ZMQ_BACKLOG = int(os.environ.get("TQ_STORAGE_ZMQ_BACKLOG", 4096))
+
+# Sampling period for the accept-queue probe, in seconds. 0 disables it. Sub-second because
+# the queue drains in milliseconds, so a reading taken after a hang is always zero.
+TQ_ACCEPT_PROBE_INTERVAL = float(os.environ.get("TQ_ACCEPT_PROBE_INTERVAL", 0))
+
 
 class StorageKeyNotFoundError(KeyError):
     """Raised when a requested global index is absent from a storage unit.
@@ -168,6 +176,13 @@ class SimpleStorageUnit:
         zmq_server_info: ZMQ connection information for clients.
     """
 
+    # Requests counted the moment the worker decodes one, independent of whether it completes.
+    # Class-level defaults so a unit built without __init__ (tests drive the worker loop
+    # directly) still counts instead of raising. See the increment site for why they exist.
+    _requests_arrived = 0
+    _arrivals_by_op: dict[str, int] = {}
+    _accept_probe = None
+
     def __init__(self, storage_unit_size: int | None = None):
         """Initialize a SimpleStorageUnit with the specified size.
 
@@ -180,6 +195,11 @@ def __init__(self, storage_unit_size: int | None = None):
 
         self.storage_data = StorageUnitData(self.storage_unit_size)
 
+        # Own copies so counts stay per unit; the class-level defaults above only exist for
+        # instances built without __init__.
+        self._requests_arrived = 0
+        self._arrivals_by_op = {}
+
         # Internal communication address for proxy and workers
         self._inproc_addr = f"inproc://simple_storage_workers_{self.storage_unit_id}"
 
@@ -219,6 +239,10 @@ def _init_zmq_socket(self) -> None:
 
         # Frontend: ROUTER for receiving client requests
         self.put_get_socket = create_zmq_socket(self.zmq_context, zmq.ROUTER, self._node_ip)
+        # An overflowing accept queue is drained silently (tcp_abort_on_overflow=0), so it
+        # surfaces only as a client stuck in ESTABLISHED waiting for a reply that never
+        # comes. Env-tunable so an A/B run can restore ZMQ's default of 100.
+        self.put_get_socket.setsockopt(zmq.BACKLOG, TQ_STORAGE_ZMQ_BACKLOG)
 
         while True:
             try:
@@ -229,6 +253,18 @@ def _init_zmq_socket(self) -> None:
                 logger.warning(f"[{self.storage_unit_id}]: Try to bind ZMQ sockets failed, retrying...")
                 continue
 
+        if TQ_ACCEPT_PROBE_INTERVAL > 0:
+            # Imported lazily: the probe shells out to ``ss`` on a timer, so a run that has
+            # not asked for it should not even load the module.
+            from transfer_queue.utils.accept_probe import AcceptQueueProbe
+
+            self._accept_probe = AcceptQueueProbe(
+                port=self._put_get_socket_port,
+                owner_id=str(self.storage_unit_id),
+                interval_s=TQ_ACCEPT_PROBE_INTERVAL,
+            )
+            self._accept_probe.start()
+
         # Backend: DEALER for worker communication (connected via zmq.proxy)
         self.worker_socket = create_zmq_socket(self.zmq_context, zmq.DEALER, self._node_ip)
         self.worker_socket.bind(self._inproc_addr)
@@ -348,6 +384,11 @@ def _worker_routine(self) -> None:
                 started = time.perf_counter()
 
                 try:
+                    # Counted on arrival, unlike op_stats which only advances on completion, so a
+                    # gap between the two isolates requests that arrived and never finished.
+                    self._requests_arrived += 1
+                    self._arrivals_by_op[str(operation)] = self._arrivals_by_op.get(str(operation), 0) + 1
+
                     logger.debug(f"[{self.storage_unit_id}]: worker received operation: {operation}")
 
                     # Process request
@@ -582,8 +623,24 @@ def _handle_get_metrics(self) -> ZMQMessage:
             "capacity": self.storage_unit_size,
             "active_keys": self.storage_data.active_key_count,
             "process_rss_bytes": process_rss,
+            # Reported next to but separately from op_stats below, which is derived from
+            # completion-time histograms: a gap between the two is a request that arrived and
+            # never finished, which the diagnostic probe cannot otherwise distinguish.
+            "requests_arrived": self._requests_arrived,
+            "arrivals_by_op": dict(self._arrivals_by_op),
         }
 
+        if self._accept_probe is not None:
+            stats = self._accept_probe.stats
+            metrics["accept_queue"] = {
+                "backlog": stats.backlog,
+                "peak_recv_q": stats.peak_recv_q,
+                "peak_utilization": stats.peak_utilization,
+                "sk_drops_delta": stats.sk_drops_delta,
+                "listen_overflow_delta": stats.overflow_delta,
+                "samples": stats.samples,
+            }
+
         # Include per-operation stats if Prometheus metrics are enabled
         if self._metrics is not None:
             op_stats = {}
diff --git a/transfer_queue/utils/accept_probe.py b/transfer_queue/utils/accept_probe.py
new file mode 100644
index 00000000..d7d2bb8a
--- /dev/null
+++ b/transfer_queue/utils/accept_probe.py
@@ -0,0 +1,226 @@
+"""Accept-queue instrumentation for the storage-unit ROUTER socket.
+
+A get request was observed leaving the manager (TCP counted the bytes and the peer
+ACKed them) while the unit's worker never saw it: its GET_DATA counter did not move
+and it kept polling. The suspected drop point is the listen socket's accept queue,
+which ZMQ leaves at its default backlog of 100 and which the kernel empties
+silently (``tcp_abort_on_overflow=0``), so neither end raises an error.
+
+This module samples the kernel's own view of that queue so the guess becomes a
+measurement:
+
+* ``Recv-Q`` on a listening socket is the number of established-but-not-yet-accepted
+  connections, i.e. the live queue depth.
+* ``ListenOverflows`` / ``ListenDrops`` count queue-full events machine-wide.
+* ``sk_drops`` counts drops charged to one specific socket.
+
+Sampling runs in a thread because the queue drains in milliseconds; a value read
+after a hang has already returned to zero, which is exactly what earlier
+post-mortem inspection saw.
+"""
+
+from __future__ import annotations
+
+import re
+import subprocess
+import threading
+import time
+from dataclasses import dataclass, field
+
+from transfer_queue.utils.logging_utils import get_logger
+
+logger = get_logger(__name__)
+
+_RECVQ_RE = re.compile(r"^LISTEN\s+(\d+)\s+(\d+)\s+\S*:(\d+)\s", re.M)
+_DROPS_RE = re.compile(r"\bd(\d+)\b")
+
+
+@dataclass
+class AcceptQueueSample:
+    """One reading of the accept queue for a single listening port."""
+
+    timestamp: float
+    recv_q: int
+    backlog: int
+    sk_drops: int
+    listen_overflows: int
+    listen_drops: int
+
+    @property
+    def utilization(self) -> float:
+        """Queue depth as a fraction of the configured backlog."""
+        return self.recv_q / self.backlog if self.backlog else 0.0
+
+
+@dataclass
+class AcceptQueueStats:
+    """Peak and delta summary over a sampling window."""
+
+    port: int
+    samples: int = 0
+    peak_recv_q: int = 0
+    peak_utilization: float = 0.0
+    backlog: int = 0
+    first_sample: AcceptQueueSample | None = None
+    last_sample: AcceptQueueSample | None = None
+    peak_history: list[tuple[float, int]] = field(default_factory=list)
+
+    @property
+    def sk_drops_delta(self) -> int:
+        if self.first_sample is None or self.last_sample is None:
+            return 0
+        return self.last_sample.sk_drops - self.first_sample.sk_drops
+
+    @property
+    def overflow_delta(self) -> int:
+        if self.first_sample is None or self.last_sample is None:
+            return 0
+        return self.last_sample.listen_overflows - self.first_sample.listen_overflows
+
+    def describe(self) -> str:
+        return (
+            f"port={self.port} samples={self.samples} backlog={self.backlog} "
+            f"peak_recv_q={self.peak_recv_q} peak_util={self.peak_utilization:.1%} "
+            f"sk_drops_delta={self.sk_drops_delta} listen_overflow_delta={self.overflow_delta}"
+        )
+
+
+def _read_listen_socket(port: int) -> tuple[int, int, int] | None:
+    """Return (recv_q, backlog, sk_drops) for the listening socket on ``port``."""
+    try:
+        out = subprocess.run(["ss", "-lntm"], capture_output=True, text=True, timeout=5, check=False).stdout
+    except (OSError, subprocess.SubprocessError) as exc:
+        logger.debug(f"accept-probe: ss failed: {exc}")
+        return None
+
+    for match in _RECVQ_RE.finditer(out):
+        recv_q, backlog, found_port = (int(g) for g in match.groups())
+        if found_port != port:
+            continue
+        # skmem lives on the continuation line right after the match.
+        tail = out[match.end() : match.end() + 400]
+        drops_match = _DROPS_RE.search(tail.split("\n")[1] if "\n" in tail else "")
+        return recv_q, backlog, int(drops_match.group(1)) if drops_match else 0
+    return None
+
+
+def _read_listen_overflows() -> tuple[int, int]:
+    """Return machine-wide (ListenOverflows, ListenDrops) from /proc/net/netstat."""
+    try:
+        with open("/proc/net/netstat") as handle:
+            lines = handle.read().splitlines()
+    except OSError:
+        return 0, 0
+
+    for i, line in enumerate(lines):
+        if not line.startswith("TcpExt:") or "ListenOverflows" not in line:
+            continue
+        keys = line.split()
+        values = lines[i + 1].split()
+        try:
+            return (
+                int(values[keys.index("ListenOverflows")]),
+                int(values[keys.index("ListenDrops")]),
+            )
+        except (ValueError, IndexError):
+            return 0, 0
+    return 0, 0
+
+
+def sample_accept_queue(port: int) -> AcceptQueueSample | None:
+    """Take one accept-queue reading for ``port``, or None if it cannot be read."""
+    listen = _read_listen_socket(port)
+    if listen is None:
+        return None
+    recv_q, backlog, sk_drops = listen
+    overflows, drops = _read_listen_overflows()
+    return AcceptQueueSample(
+        timestamp=time.time(),
+        recv_q=recv_q,
+        backlog=backlog,
+        sk_drops=sk_drops,
+        listen_overflows=overflows,
+        listen_drops=drops,
+    )
+
+
+class AcceptQueueProbe:
+    """Sample one listening port's accept queue from a background thread.
+
+    Args:
+        port: Listening port to watch.
+        owner_id: Identifier used in log lines (the storage unit id).
+        interval_s: Seconds between samples. Keep well under a second; the queue
+            drains in milliseconds and a slower cadence misses the burst entirely.
+        warn_utilization: Log a warning the first time depth reaches this fraction
+            of the backlog, so a near-miss is visible before any drop happens.
+    """
+
+    def __init__(
+        self,
+        port: int,
+        owner_id: str,
+        interval_s: float = 0.1,
+        warn_utilization: float = 0.5,
+    ) -> None:
+        self.port = port
+        self.owner_id = owner_id
+        self.interval_s = interval_s
+        self.warn_utilization = warn_utilization
+        self.stats = AcceptQueueStats(port=port)
+        self._stop = threading.Event()
+        self._thread: threading.Thread | None = None
+        self._warned = False
+
+    def start(self) -> None:
+        if self._thread is not None:
+            return
+        self._thread = threading.Thread(target=self._run, name=f"AcceptQueueProbe-{self.owner_id}", daemon=True)
+        self._thread.start()
+        logger.info(f"[{self.owner_id}]: accept-queue probe started on port {self.port} (interval={self.interval_s}s)")
+
+    def stop(self) -> None:
+        self._stop.set()
+        if self._thread is not None:
+            self._thread.join(timeout=5)
+            self._thread = None
+        logger.info(f"[{self.owner_id}]: accept-queue probe stopped. {self.stats.describe()}")
+
+    def _run(self) -> None:
+        while not self._stop.is_set():
+            sample = sample_accept_queue(self.port)
+            if sample is not None:
+                self._record(sample)
+            self._stop.wait(self.interval_s)
+
+    def _record(self, sample: AcceptQueueSample) -> None:
+        stats = self.stats
+        stats.samples += 1
+        stats.backlog = sample.backlog or stats.backlog
+        if stats.first_sample is None:
+            stats.first_sample = sample
+        stats.last_sample = sample
+
+        if sample.recv_q > stats.peak_recv_q:
+            stats.peak_recv_q = sample.recv_q
+            stats.peak_utilization = sample.utilization
+            stats.peak_history.append((sample.timestamp, sample.recv_q))
+
+        # A drop charged to this socket is the direct evidence the guess needs, so it
+        # is reported at error level with the depth that produced it.
+        if stats.first_sample is not None and sample.sk_drops > stats.first_sample.sk_drops:
+            logger.error(
+                f"[{self.owner_id}]: accept queue dropped a connection on port {self.port}. "
+                f"recv_q={sample.recv_q}/{sample.backlog} sk_drops={sample.sk_drops} "
+                f"(+{sample.sk_drops - stats.first_sample.sk_drops} since probe start). "
+                f"A silently dropped connection leaves the client in ESTABLISHED with no "
+                f"reply; raise ZMQ_BACKLOG above {sample.backlog}."
+            )
+
+        if not self._warned and sample.utilization >= self.warn_utilization:
+            self._warned = True
+            logger.warning(
+                f"[{self.owner_id}]: accept queue on port {self.port} reached "
+                f"{sample.recv_q}/{sample.backlog} ({sample.utilization:.0%}); "
+                f"connections are queueing faster than they are accepted."
+            )

From 7bd07913f927ff8b0a722e714424f230ba239a2f Mon Sep 17 00:00:00 2001
From: OutstanderWang <wangweiyanster@gmail.com>
Date: Tue, 8 Sep 2026 17:22:15 +0800
Subject: [PATCH 2/8] [fix] Report an accept-queue drop once per drop, not once
 per sample

sk_drops is a cumulative kernel counter, but the check compared it against the
probe's first sample. Once the socket had ever dropped a connection the
condition stayed true for the life of the process: at the 0.1s sampling
interval that is ten errors a second per storage unit, and one long-running
instance had logged over a hundred thousand of them.

The flood also destroyed the signal the probe exists to provide. Repeating the
same cumulative total on every sample says nothing about when the drops
happened, so a handful charged during one resume window is indistinguishable
from an ongoing overflow -- exactly the question the probe was added to answer.

Compare against the previous sample instead, captured before last_sample is
overwritten, and report both deltas so a single line still shows the run total.

Signed-off-by: OutstanderWang <wangweiyanster@gmail.com>
---
 tests/test_accept_probe.py           | 31 ++++++++++++++++++++++++++++
 transfer_queue/utils/accept_probe.py | 11 ++++++----
 2 files changed, 38 insertions(+), 4 deletions(-)

diff --git a/tests/test_accept_probe.py b/tests/test_accept_probe.py
index 249ecd78..45fbf443 100644
--- a/tests/test_accept_probe.py
+++ b/tests/test_accept_probe.py
@@ -119,6 +119,37 @@ def test_drop_increase_is_logged_at_error(caplog):
     assert "accept queue dropped a connection" in caplog.text
 
 
+def test_steady_drop_count_is_logged_once_not_every_sample(caplog):
+    """sk_drops is cumulative, so a past drop must not be re-reported forever.
+
+    Measuring against the probe's first sample made the condition permanently true once
+    the socket had ever dropped a connection: at a 0.1s interval that is ten errors a
+    second for the life of the process, and it erases when the drop actually happened.
+    """
+    probe = _probe()
+    probe._record(_sample(0, sk_drops=12))
+
+    with caplog.at_level("ERROR"):
+        probe._record(_sample(0, sk_drops=13))  # a new drop -- report it
+        for _ in range(20):
+            probe._record(_sample(0, sk_drops=13))  # unchanged -- stay quiet
+
+    assert caplog.text.count("accept queue dropped a connection") == 1
+
+
+def test_each_new_drop_is_reported(caplog):
+    """Quieting the repeat must not swallow genuinely new drops."""
+    probe = _probe()
+    probe._record(_sample(0, sk_drops=12))
+
+    with caplog.at_level("ERROR"):
+        probe._record(_sample(0, sk_drops=13))
+        probe._record(_sample(0, sk_drops=13))
+        probe._record(_sample(0, sk_drops=14))
+
+    assert caplog.text.count("accept queue dropped a connection") == 2
+
+
 def test_near_full_queue_warns_once(caplog):
     """Repeating the warning every 100ms would flood the log of a 512-node job."""
     probe = _probe()
diff --git a/transfer_queue/utils/accept_probe.py b/transfer_queue/utils/accept_probe.py
index d7d2bb8a..07df0316 100644
--- a/transfer_queue/utils/accept_probe.py
+++ b/transfer_queue/utils/accept_probe.py
@@ -197,6 +197,7 @@ def _record(self, sample: AcceptQueueSample) -> None:
         stats = self.stats
         stats.samples += 1
         stats.backlog = sample.backlog or stats.backlog
+        previous = stats.last_sample
         if stats.first_sample is None:
             stats.first_sample = sample
         stats.last_sample = sample
@@ -206,13 +207,15 @@ def _record(self, sample: AcceptQueueSample) -> None:
             stats.peak_utilization = sample.utilization
             stats.peak_history.append((sample.timestamp, sample.recv_q))
 
-        # A drop charged to this socket is the direct evidence the guess needs, so it
-        # is reported at error level with the depth that produced it.
-        if stats.first_sample is not None and sample.sk_drops > stats.first_sample.sk_drops:
+        # sk_drops is a monotonic kernel counter, so this compares against the previous
+        # sample rather than the first: measuring from probe start would keep reporting a
+        # drop that happened once, on every sample, and erase when it actually occurred.
+        if previous is not None and sample.sk_drops > previous.sk_drops:
             logger.error(
                 f"[{self.owner_id}]: accept queue dropped a connection on port {self.port}. "
                 f"recv_q={sample.recv_q}/{sample.backlog} sk_drops={sample.sk_drops} "
-                f"(+{sample.sk_drops - stats.first_sample.sk_drops} since probe start). "
+                f"(+{sample.sk_drops - previous.sk_drops} since the last sample, "
+                f"+{stats.sk_drops_delta} since probe start). "
                 f"A silently dropped connection leaves the client in ESTABLISHED with no "
                 f"reply; raise ZMQ_BACKLOG above {sample.backlog}."
             )

From 9632c5753c45f3c99e84b8cf8805ddde1489f1e0 Mon Sep 17 00:00:00 2001
From: OutstanderWang <wangweiyanster@gmail.com>
Date: Thu, 10 Sep 2026 14:58:34 +0800
Subject: [PATCH 3/8] [fix] Add the license header and missing docstrings to
 accept_probe

The sanity job runs check_license.py and check_docstrings.py over every source
file. accept_probe.py was added without the Apache header, so the license step
failed and the job stopped before reaching the docstring step, which was hiding
five undocumented public members behind the first error.

Comments and docstrings only; no behavior change.

Signed-off-by: OutstanderWang <wangweiyanster@gmail.com>
---
 transfer_queue/utils/accept_probe.py | 20 ++++++++++++++++++++
 1 file changed, 20 insertions(+)

diff --git a/transfer_queue/utils/accept_probe.py b/transfer_queue/utils/accept_probe.py
index 07df0316..f5985514 100644
--- a/transfer_queue/utils/accept_probe.py
+++ b/transfer_queue/utils/accept_probe.py
@@ -1,3 +1,18 @@
+# Copyright 2025 Huawei Technologies Co., Ltd. All Rights Reserved.
+# Copyright 2025 The TransferQueue Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 """Accept-queue instrumentation for the storage-unit ROUTER socket.
 
 A get request was observed leaving the manager (TCP counted the bytes and the peer
@@ -67,17 +82,20 @@ class AcceptQueueStats:
 
     @property
     def sk_drops_delta(self) -> int:
+        """Connections this socket dropped during the window, 0 until two samples exist."""
         if self.first_sample is None or self.last_sample is None:
             return 0
         return self.last_sample.sk_drops - self.first_sample.sk_drops
 
     @property
     def overflow_delta(self) -> int:
+        """Machine-wide accept-queue overflows during the window, 0 until two samples exist."""
         if self.first_sample is None or self.last_sample is None:
             return 0
         return self.last_sample.listen_overflows - self.first_sample.listen_overflows
 
     def describe(self) -> str:
+        """Return a one-line summary of the window, for the probe's shutdown log."""
         return (
             f"port={self.port} samples={self.samples} backlog={self.backlog} "
             f"peak_recv_q={self.peak_recv_q} peak_util={self.peak_utilization:.1%} "
@@ -173,6 +191,7 @@ def __init__(
         self._warned = False
 
     def start(self) -> None:
+        """Start the sampling thread; a second call is a no-op."""
         if self._thread is not None:
             return
         self._thread = threading.Thread(target=self._run, name=f"AcceptQueueProbe-{self.owner_id}", daemon=True)
@@ -180,6 +199,7 @@ def start(self) -> None:
         logger.info(f"[{self.owner_id}]: accept-queue probe started on port {self.port} (interval={self.interval_s}s)")
 
     def stop(self) -> None:
+        """Stop the sampling thread and log the window summary."""
         self._stop.set()
         if self._thread is not None:
             self._thread.join(timeout=5)

From 86ed6113dfa61de39c9ff9ae462f727eb1bc5d4f Mon Sep 17 00:00:00 2001
From: OutstanderWang <wangweiyanster@gmail.com>
Date: Thu, 10 Sep 2026 21:12:31 +0800
Subject: [PATCH 4/8] [fix] Base the timeout diagnosis on arrival counters and
 stop the probe on shutdown

Addresses the two P1 comments from the Codex review on #172.

A successful post-failure probe only shows the unit is serving now, but the
diagnosis reported verdict=request_lost_in_flight unconditionally. A unit that
resumes inside the 10s diagnostic window answers that probe too, having merely
finished the original request late, so the verdict asserted where the request was
lost on evidence that could not support it. It also printed ops= from op_stats,
which _handle_get_metrics only populates when Prometheus is enabled, so the line
read ops={} as if the unit had served nothing.

Decide from the unit's own arrival counter for the failed operation instead: no
arrival means the request never reached the worker, an arrival means it did and
did not finish, and no counter at all now reports only that the unit recovered.
op_stats is reported as unavailable rather than empty when Prometheus is off.

The arrival counters could not actually be correlated as intended. The worker
keyed them by str(operation), which for a (str, Enum) member renders as
"ZMQRequestType.GET_DATA", while op_stats is keyed "GET_DATA", so the two dicts
shared no keys. Key by operation.name: the enum's value is the short wire token
"GET", so .value would not match either.

Separately, nothing ever called AcceptQueueProbe.stop(). The finalizer tore down
only the ZMQ resources, so when a unit was finalized without its process exiting
the daemon thread kept spawning ss on a timer and the window summary was never
logged. Stop it in _shutdown_resources, before the ZMQ teardown.

Signed-off-by: OutstanderWang <wangweiyanster@gmail.com>
---
 tests/test_accept_probe.py                    | 38 +++++++++++
 tests/test_storage_request_retry.py           | 23 ++++++-
 .../managers/simple_storage_manager.py        | 65 ++++++++++++++++---
 transfer_queue/storage/simple_storage.py      | 13 +++-
 4 files changed, 127 insertions(+), 12 deletions(-)

diff --git a/tests/test_accept_probe.py b/tests/test_accept_probe.py
index 45fbf443..344a3933 100644
--- a/tests/test_accept_probe.py
+++ b/tests/test_accept_probe.py
@@ -21,6 +21,8 @@
 reports and the levels it logs at.
 """
 
+import threading
+
 from transfer_queue.utils.accept_probe import (
     AcceptQueueProbe,
     AcceptQueueSample,
@@ -179,3 +181,39 @@ def test_listen_overflows_returns_two_non_negative_ints():
 
 def test_sampling_an_unused_port_returns_none():
     assert sample_accept_queue(1) is None
+
+
+def test_unit_shutdown_stops_the_probe():
+    """Nothing else calls stop(), so the sampling thread would outlive the unit."""
+    from unittest.mock import MagicMock
+
+    from transfer_queue.storage.simple_storage import SimpleStorageUnit
+
+    unit_class = SimpleStorageUnit.__ray_metadata__.modified_class
+    probe = MagicMock()
+
+    unit_class._shutdown_resources(
+        shutdown_event=threading.Event(),
+        worker_thread=None,
+        proxy_thread=None,
+        zmq_context=None,
+        put_get_socket=None,
+        accept_probe=probe,
+    )
+
+    probe.stop.assert_called_once()
+
+
+def test_shutdown_without_a_probe_is_a_no_op():
+    """The probe is opt-in, so the default path must not require one."""
+    from transfer_queue.storage.simple_storage import SimpleStorageUnit
+
+    unit_class = SimpleStorageUnit.__ray_metadata__.modified_class
+
+    unit_class._shutdown_resources(
+        shutdown_event=threading.Event(),
+        worker_thread=None,
+        proxy_thread=None,
+        zmq_context=None,
+        put_get_socket=None,
+    )
diff --git a/tests/test_storage_request_retry.py b/tests/test_storage_request_retry.py
index ec6e9476..9b654202 100644
--- a/tests/test_storage_request_retry.py
+++ b/tests/test_storage_request_retry.py
@@ -147,7 +147,12 @@ def test_log_heavy_operation_thresholds(caplog, elapsed, payload_bytes, should_l
 @pytest.mark.parametrize(
     "tcp_result, probe_result, expected",
     [
-        ((None, None), {"active_keys": 4, "op_stats": {"GET_DATA": {"request_count": 1025}}}, "request_lost"),
+        # A serving unit that decoded no such request never received the one that timed out.
+        ((None, None), {"active_keys": 4, "arrivals_by_op": {"GET_DATA": 0}}, "request_lost_in_flight"),
+        # A unit that resumed inside the probe window answers too, having finished it late.
+        ((None, None), {"active_keys": 4, "arrivals_by_op": {"GET_DATA": 9}}, "arrived_but_unfinished"),
+        # Without arrival counters the probe only proves the unit is serving now.
+        ((None, None), {"active_keys": 4}, "unit_serving_again"),
         ((None, None), zmq.error.Again(), "unit_not_serving"),
         (ConnectionRefusedError(), zmq.error.Again(), "tcp=down(ConnectionRefusedError)"),
     ],
@@ -165,11 +170,25 @@ async def test_diagnosis_classifies_the_failure(tcp_result, probe_result, expect
     )
 
     with patch.object(ssm.asyncio, "open_connection", tcp), patch.object(manager, "_probe_storage_unit", probe):
-        diagnosis = await manager._diagnose_storage_unit("unit_a")
+        diagnosis = await manager._diagnose_storage_unit("unit_a", "get")
 
     assert expected in diagnosis
 
 
+def test_diagnosis_does_not_report_empty_op_stats_as_zero_traffic():
+    """op_stats is Prometheus-gated, so an empty dict must not read as 'served nothing'."""
+    described = ssm._describe_unit_state({"requests_arrived": 7, "active_keys": 1})
+
+    assert "completed=unavailable(prometheus_disabled)" in described
+    assert "completed={}" not in described
+
+
+def test_arrival_key_for_get_is_the_enum_name_not_its_wire_value():
+    """The unit keys arrivals by ZMQRequestType.name; the value is the short token 'GET'."""
+    assert ssm._ARRIVAL_KEY_BY_OPERATION["get"] == "GET_DATA"
+    assert ssm._ARRIVAL_KEY_BY_OPERATION["put"] == "PUT_DATA"
+
+
 @pytest.mark.asyncio
 async def test_diagnosis_handles_an_unknown_unit():
     assert "unit_not_registered" in await _manager(with_unit=False)._diagnose_storage_unit("missing")
diff --git a/transfer_queue/storage/managers/simple_storage_manager.py b/transfer_queue/storage/managers/simple_storage_manager.py
index 70765cb9..e04cb49b 100644
--- a/transfer_queue/storage/managers/simple_storage_manager.py
+++ b/transfer_queue/storage/managers/simple_storage_manager.py
@@ -68,6 +68,29 @@ class StorageUnitTimeout(RuntimeError):
     """
 
 
+# Maps the operation name used by the retry path to the storage unit's arrival counter key.
+# Explicit rather than derived: the counters are keyed by ZMQRequestType.name ("GET_DATA"),
+# while the enum's value is the short wire token "GET", so "get".upper() would miss.
+_ARRIVAL_KEY_BY_OPERATION = {"get": "GET_DATA", "put": "PUT_DATA", "clear": "CLEAR_DATA"}
+
+
+def _describe_unit_state(body: dict[str, Any]) -> str:
+    """Summarize a probe response for the failure log."""
+    parts = [
+        f"requests_arrived={body.get('requests_arrived')}",
+        f"active_keys={body.get('active_keys')}",
+        f"rss_gb={body.get('process_rss_bytes', 0) / 2**30:.2f}",
+    ]
+    # Only present when the unit runs with Prometheus enabled; an empty dict here would read
+    # as "served nothing" rather than "not measured".
+    op_stats = body.get("op_stats") or {}
+    if op_stats:
+        parts.append(f"completed={ {op: stats.get('request_count') for op, stats in op_stats.items()} }")
+    else:
+        parts.append("completed=unavailable(prometheus_disabled)")
+    return f"({' '.join(parts)})"
+
+
 _SU_SUBDIR = "simple_storage"
 _SU_INFO_FILE = "storage_unit_info.json"
 
@@ -239,10 +262,14 @@ async def _probe_storage_unit(self, target_storage_unit: str, socket: zmq.Socket
             raise RuntimeError(f"unexpected probe response type {response_msg.request_type}")
         return response_msg.body
 
-    async def _diagnose_storage_unit(self, target_storage_unit: str) -> str:
-        """Classify a timeout as a lost request, a stuck unit, or an unreachable node.
+    async def _diagnose_storage_unit(self, target_storage_unit: str, operation: str = "") -> str:
+        """Classify a timeout using the unit's own arrival counters.
 
         Returns one log line and never raises: it runs while another failure is being reported.
+
+        Args:
+            target_storage_unit (str): Unit that failed to answer.
+            operation (str): Failed operation as passed to ``_request_with_retry``, e.g. ``get``.
         """
         info = self.storage_unit_infos.get(target_storage_unit)
         if info is None:
@@ -259,17 +286,37 @@ async def _diagnose_storage_unit(self, target_storage_unit: str) -> str:
 
         try:
             body = await self._probe_storage_unit(target_storage_unit=target_storage_unit)
-            op_counts = {op: stats.get("request_count") for op, stats in (body.get("op_stats") or {}).items()}
-            return (
-                f"{tcp} verdict=request_lost_in_flight (unit answered a fresh probe: "
-                f"ops={op_counts} active_keys={body.get('active_keys')} "
-                f"rss_gb={body.get('process_rss_bytes', 0) / 2**30:.2f})"
-            )
         except zmq.error.Again:
             return f"{tcp} verdict=unit_not_serving (no probe answer in {TQ_SIMPLE_STORAGE_PROBE_TIMEOUT}s)"
         except Exception as e:
             return f"{tcp} verdict=unknown (probe failed: {type(e).__name__}: {e})"
 
+        return f"{tcp} {self._verdict_from_counters(body, operation)} {_describe_unit_state(body)}"
+
+    @staticmethod
+    def _verdict_from_counters(body: dict[str, Any], operation: str) -> str:
+        """Decide what a successful probe proves about the request that timed out.
+
+        The probe answering only shows the unit is serving *now*: a unit that resumed inside
+        the diagnostic window answers it too, having merely finished the original request late.
+        The unit's arrival counter is what separates the two, so absent that counter this
+        reports only that the unit recovered rather than asserting where the request went.
+        """
+        arrivals = body.get("arrivals_by_op")
+        op_key = _ARRIVAL_KEY_BY_OPERATION.get(operation)
+        if not isinstance(arrivals, dict) or op_key is None:
+            return "verdict=unit_serving_again (no arrival counters to locate the request)"
+
+        arrived = arrivals.get(op_key)
+        if not isinstance(arrived, int):
+            return f"verdict=unit_serving_again (unit reports no {op_key} arrivals counter)"
+        if arrived == 0:
+            return f"verdict=request_lost_in_flight (unit decoded no {op_key} request at all)"
+        return (
+            f"verdict=arrived_but_unfinished (unit decoded {arrived} {op_key} request(s); "
+            f"the timed-out one reached the worker and did not complete)"
+        )
+
     async def _request_with_retry(
         self,
         operation: str,
@@ -306,7 +353,7 @@ async def _request_with_retry(
                     continue
                 logger.error(
                     f"[{self.storage_manager_id}]: {operation} failed after {attempt} attempts. "
-                    f"{request_context} {e} {await self._diagnose_storage_unit(target_storage_unit)}"
+                    f"{request_context} {e} {await self._diagnose_storage_unit(target_storage_unit, operation)}"
                 )
                 raise
 
diff --git a/transfer_queue/storage/simple_storage.py b/transfer_queue/storage/simple_storage.py
index 2b363602..385d0ab0 100644
--- a/transfer_queue/storage/simple_storage.py
+++ b/transfer_queue/storage/simple_storage.py
@@ -43,6 +43,7 @@
 
 if TYPE_CHECKING:
     from transfer_queue.metrics import TQMetricsExporter
+    from transfer_queue.utils.accept_probe import AcceptQueueProbe
 
 logger = get_logger(__name__)
 
@@ -226,6 +227,7 @@ def __init__(self, storage_unit_size: int | None = None):
             self.proxy_thread,
             self.zmq_context,
             self.put_get_socket,
+            self._accept_probe,
         )
 
     def _init_zmq_socket(self) -> None:
@@ -387,7 +389,10 @@ def _worker_routine(self) -> None:
                     # Counted on arrival, unlike op_stats which only advances on completion, so a
                     # gap between the two isolates requests that arrived and never finished.
                     self._requests_arrived += 1
-                    self._arrivals_by_op[str(operation)] = self._arrivals_by_op.get(str(operation), 0) + 1
+                    # Keyed by name, not str() or value: str() renders as
+                    # "ZMQRequestType.GET_DATA" and the value is the short wire token "GET",
+                    # while op_stats below is keyed "GET_DATA". Only name lets the two join.
+                    self._arrivals_by_op[operation.name] = self._arrivals_by_op.get(operation.name, 0) + 1
 
                     logger.debug(f"[{self.storage_unit_id}]: worker received operation: {operation}")
 
@@ -798,6 +803,7 @@ def _shutdown_resources(
         proxy_thread: Thread | None,
         zmq_context: zmq.Context | None,
         put_get_socket: zmq.Socket | None,
+        accept_probe: "AcceptQueueProbe | None" = None,
     ) -> None:
         """Clean up resources on garbage collection."""
         logger.info("Shutting down SimpleStorageUnit resources...")
@@ -805,6 +811,11 @@ def _shutdown_resources(
         # Signal all threads to stop
         shutdown_event.set()
 
+        # Stop before the ZMQ teardown: the probe samples on its own timer and would keep
+        # spawning `ss` after the unit is gone, and stopping it logs the window summary.
+        if accept_probe is not None:
+            accept_probe.stop()
+
         # Terminate put_get_socket
         if put_get_socket:
             put_get_socket.close(linger=0)

From 9d1ef6ad6c67502f03a78deb5c802badec71c3a0 Mon Sep 17 00:00:00 2001
From: OutstanderWang <wangweiyanster@gmail.com>
Date: Fri, 11 Sep 2026 17:27:36 +0800
Subject: [PATCH 5/8] [storage] Trim comments and merge the diagnosis helper

Review feedback: fold _verdict_from_counters into _describe_unit_state so the
verdict and the state it was drawn from are built in one pass, and drop comments
that restated what the code already says. The arrival-key spelling that a comment
explained is pinned by test_arrival_key_for_get_is_the_enum_name_not_its_wire_value.

Signed-off-by: OutstanderWang <wangweiyanster@gmail.com>
---
 tests/test_storage_request_retry.py           |  2 +-
 .../managers/simple_storage_manager.py        | 57 +++++++++----------
 transfer_queue/storage/simple_storage.py      | 11 ----
 3 files changed, 28 insertions(+), 42 deletions(-)

diff --git a/tests/test_storage_request_retry.py b/tests/test_storage_request_retry.py
index 9b654202..dd207353 100644
--- a/tests/test_storage_request_retry.py
+++ b/tests/test_storage_request_retry.py
@@ -177,7 +177,7 @@ async def test_diagnosis_classifies_the_failure(tcp_result, probe_result, expect
 
 def test_diagnosis_does_not_report_empty_op_stats_as_zero_traffic():
     """op_stats is Prometheus-gated, so an empty dict must not read as 'served nothing'."""
-    described = ssm._describe_unit_state({"requests_arrived": 7, "active_keys": 1})
+    described = ssm._describe_unit_state({"requests_arrived": 7, "active_keys": 1}, "get")
 
     assert "completed=unavailable(prometheus_disabled)" in described
     assert "completed={}" not in described
diff --git a/transfer_queue/storage/managers/simple_storage_manager.py b/transfer_queue/storage/managers/simple_storage_manager.py
index e04cb49b..cce8e33a 100644
--- a/transfer_queue/storage/managers/simple_storage_manager.py
+++ b/transfer_queue/storage/managers/simple_storage_manager.py
@@ -74,21 +74,42 @@ class StorageUnitTimeout(RuntimeError):
 _ARRIVAL_KEY_BY_OPERATION = {"get": "GET_DATA", "put": "PUT_DATA", "clear": "CLEAR_DATA"}
 
 
-def _describe_unit_state(body: dict[str, Any]) -> str:
-    """Summarize a probe response for the failure log."""
+def _describe_unit_state(body: dict[str, Any], operation: str) -> str:
+    """Turn a successful probe into a verdict plus the state it was drawn from.
+
+    The probe answering only shows the unit is serving *now*: a unit that resumed inside the
+    diagnostic window answers it too, having merely finished the original request late. The
+    arrival counter for the failed op is what separates the two, so without it this reports
+    only that the unit recovered rather than asserting where the request went.
+    """
+    arrivals = body.get("arrivals_by_op")
+    op_key = _ARRIVAL_KEY_BY_OPERATION.get(operation)
+    if not isinstance(arrivals, dict) or op_key is None:
+        verdict = "verdict=unit_serving_again (no arrival counters to locate the request)"
+    else:
+        arrived = arrivals.get(op_key)
+        if not isinstance(arrived, int):
+            verdict = f"verdict=unit_serving_again (unit reports no {op_key} arrivals counter)"
+        elif arrived == 0:
+            verdict = f"verdict=request_lost_in_flight (unit decoded no {op_key} request at all)"
+        else:
+            verdict = (
+                f"verdict=arrived_but_unfinished (unit decoded {arrived} {op_key} request(s); "
+                f"the timed-out one reached the worker and did not complete)"
+            )
+
     parts = [
         f"requests_arrived={body.get('requests_arrived')}",
         f"active_keys={body.get('active_keys')}",
         f"rss_gb={body.get('process_rss_bytes', 0) / 2**30:.2f}",
     ]
-    # Only present when the unit runs with Prometheus enabled; an empty dict here would read
-    # as "served nothing" rather than "not measured".
     op_stats = body.get("op_stats") or {}
     if op_stats:
         parts.append(f"completed={ {op: stats.get('request_count') for op, stats in op_stats.items()} }")
     else:
+        # op_stats is populated only with Prometheus; an empty dict would read as "served nothing".
         parts.append("completed=unavailable(prometheus_disabled)")
-    return f"({' '.join(parts)})"
+    return f"{verdict} ({' '.join(parts)})"
 
 
 _SU_SUBDIR = "simple_storage"
@@ -291,31 +312,7 @@ async def _diagnose_storage_unit(self, target_storage_unit: str, operation: str
         except Exception as e:
             return f"{tcp} verdict=unknown (probe failed: {type(e).__name__}: {e})"
 
-        return f"{tcp} {self._verdict_from_counters(body, operation)} {_describe_unit_state(body)}"
-
-    @staticmethod
-    def _verdict_from_counters(body: dict[str, Any], operation: str) -> str:
-        """Decide what a successful probe proves about the request that timed out.
-
-        The probe answering only shows the unit is serving *now*: a unit that resumed inside
-        the diagnostic window answers it too, having merely finished the original request late.
-        The unit's arrival counter is what separates the two, so absent that counter this
-        reports only that the unit recovered rather than asserting where the request went.
-        """
-        arrivals = body.get("arrivals_by_op")
-        op_key = _ARRIVAL_KEY_BY_OPERATION.get(operation)
-        if not isinstance(arrivals, dict) or op_key is None:
-            return "verdict=unit_serving_again (no arrival counters to locate the request)"
-
-        arrived = arrivals.get(op_key)
-        if not isinstance(arrived, int):
-            return f"verdict=unit_serving_again (unit reports no {op_key} arrivals counter)"
-        if arrived == 0:
-            return f"verdict=request_lost_in_flight (unit decoded no {op_key} request at all)"
-        return (
-            f"verdict=arrived_but_unfinished (unit decoded {arrived} {op_key} request(s); "
-            f"the timed-out one reached the worker and did not complete)"
-        )
+        return f"{tcp} {_describe_unit_state(body, operation)}"
 
     async def _request_with_retry(
         self,
diff --git a/transfer_queue/storage/simple_storage.py b/transfer_queue/storage/simple_storage.py
index 385d0ab0..5f150247 100644
--- a/transfer_queue/storage/simple_storage.py
+++ b/transfer_queue/storage/simple_storage.py
@@ -177,9 +177,6 @@ class SimpleStorageUnit:
         zmq_server_info: ZMQ connection information for clients.
     """
 
-    # Requests counted the moment the worker decodes one, independent of whether it completes.
-    # Class-level defaults so a unit built without __init__ (tests drive the worker loop
-    # directly) still counts instead of raising. See the increment site for why they exist.
     _requests_arrived = 0
     _arrivals_by_op: dict[str, int] = {}
     _accept_probe = None
@@ -241,9 +238,6 @@ def _init_zmq_socket(self) -> None:
 
         # Frontend: ROUTER for receiving client requests
         self.put_get_socket = create_zmq_socket(self.zmq_context, zmq.ROUTER, self._node_ip)
-        # An overflowing accept queue is drained silently (tcp_abort_on_overflow=0), so it
-        # surfaces only as a client stuck in ESTABLISHED waiting for a reply that never
-        # comes. Env-tunable so an A/B run can restore ZMQ's default of 100.
         self.put_get_socket.setsockopt(zmq.BACKLOG, TQ_STORAGE_ZMQ_BACKLOG)
 
         while True:
@@ -386,12 +380,7 @@ def _worker_routine(self) -> None:
                 started = time.perf_counter()
 
                 try:
-                    # Counted on arrival, unlike op_stats which only advances on completion, so a
-                    # gap between the two isolates requests that arrived and never finished.
                     self._requests_arrived += 1
-                    # Keyed by name, not str() or value: str() renders as
-                    # "ZMQRequestType.GET_DATA" and the value is the short wire token "GET",
-                    # while op_stats below is keyed "GET_DATA". Only name lets the two join.
                     self._arrivals_by_op[operation.name] = self._arrivals_by_op.get(operation.name, 0) + 1
 
                     logger.debug(f"[{self.storage_unit_id}]: worker received operation: {operation}")

From 993d6f9bb3ea6eebc8e1ab94c7aae24a135fb2d3 Mon Sep 17 00:00:00 2001
From: OutstanderWang <wangweiyanster@gmail.com>
Date: Fri, 11 Sep 2026 20:20:44 +0800
Subject: [PATCH 6/8] [storage] Stop the diagnostics from naming causes the
 data cannot establish

Two review findings, both about inference rather than measurement.

The timeout diagnosis read the unit's arrival counter for the failed operation and
reported arrived_but_unfinished when it was nonzero, request_lost_in_flight when it
was zero. Neither follows. The counters are per operation and cumulative, carry no
request id and never decrease, so a nonzero count may be entirely historical: a unit
that had served nine GETs and never saw the tenth still reports nine. The zero case
is no better, because the counters restart with the process, so a unit that restarted
inside the timeout window reports zero for a request that did reach its predecessor.
A probe answering proves only that the unit is serving again, so that is all the
verdict now says; the counters stay in the line as triage input. This also removes
the operation argument and the key map that existed only to support the inference.

The accept-queue probe alerted "accept queue dropped a connection" and advised raising
ZMQ_BACKLOG whenever a listening socket's sk_drops rose. The kernel charges sk_drops
on many establishment failures that are not overflow -- failing to allocate the child
socket under memory pressure, failing to route it, failing to inherit the port -- all
of which reach tcp_listendrop() without touching ListenOverflows (Linux v6.6,
net/ipv4/tcp_ipv4.c exit_overflow/exit_nonewsk/exit). So sk_drops locates the socket,
not the cause. The alert now reports the drop, prints the overflow and non-overflow
deltas beside it, and offers the backlog only as the fix for one candidate.

ListenDrops minus ListenOverflows is the discriminator that was missing, so it is now
computed and exposed in get_metrics: positive means a bigger backlog could not have
prevented every drop in the window.

Also corrects the docstrings. ListenOverflows and ListenDrops are per network
namespace rather than machine-wide, ListenDrops is not a queue-full counter, and
tcp_abort_on_overflow=0 does not strand the connection: it withholds the RST, and the
server's SYN-ACK retransmits let the connection complete if the queue drains within
tcp_synack_retries.

Signed-off-by: OutstanderWang <wangweiyanster@gmail.com>
---
 tests/test_accept_probe.py                    | 56 ++++++++++++++++---
 tests/test_storage_request_retry.py           | 42 ++++++++++----
 .../managers/simple_storage_manager.py        | 51 +++++------------
 transfer_queue/storage/simple_storage.py      |  1 +
 transfer_queue/utils/accept_probe.py          | 50 ++++++++++++-----
 5 files changed, 130 insertions(+), 70 deletions(-)

diff --git a/tests/test_accept_probe.py b/tests/test_accept_probe.py
index 344a3933..a834955c 100644
--- a/tests/test_accept_probe.py
+++ b/tests/test_accept_probe.py
@@ -15,10 +15,10 @@
 
 """Tests for the accept-queue probe.
 
-A silently dropped connection leaves the client in ESTABLISHED with no reply and the
-unit's worker idle, which is what post-mortem inspection of a hang actually showed.
-The probe turns that guess into a measurement, so these tests pin the arithmetic it
-reports and the levels it logs at.
+A connection dropped during establishment is invisible to both ends, and the kernel
+charges the listening socket's sk_drops on several such paths, of which a full accept
+queue is only one. These tests pin the arithmetic the probe reports, the levels it logs
+at, and that it never names a cause its counters cannot establish.
 """
 
 import threading
@@ -32,14 +32,16 @@
 )
 
 
-def _sample(recv_q: int, backlog: int = 100, sk_drops: int = 0, overflows: int = 0) -> AcceptQueueSample:
+def _sample(
+    recv_q: int, backlog: int = 100, sk_drops: int = 0, overflows: int = 0, drops: int | None = None
+) -> AcceptQueueSample:
     return AcceptQueueSample(
         timestamp=0.0,
         recv_q=recv_q,
         backlog=backlog,
         sk_drops=sk_drops,
         listen_overflows=overflows,
-        listen_drops=overflows,
+        listen_drops=overflows if drops is None else drops,
     )
 
 
@@ -118,7 +120,7 @@ def test_drop_increase_is_logged_at_error(caplog):
     with caplog.at_level("ERROR"):
         probe._record(_sample(100, sk_drops=24))
 
-    assert "accept queue dropped a connection" in caplog.text
+    assert "dropped an incoming connection" in caplog.text
 
 
 def test_steady_drop_count_is_logged_once_not_every_sample(caplog):
@@ -136,7 +138,7 @@ def test_steady_drop_count_is_logged_once_not_every_sample(caplog):
         for _ in range(20):
             probe._record(_sample(0, sk_drops=13))  # unchanged -- stay quiet
 
-    assert caplog.text.count("accept queue dropped a connection") == 1
+    assert caplog.text.count("dropped an incoming connection") == 1
 
 
 def test_each_new_drop_is_reported(caplog):
@@ -149,7 +151,7 @@ def test_each_new_drop_is_reported(caplog):
         probe._record(_sample(0, sk_drops=13))
         probe._record(_sample(0, sk_drops=14))
 
-    assert caplog.text.count("accept queue dropped a connection") == 2
+    assert caplog.text.count("dropped an incoming connection") == 2
 
 
 def test_near_full_queue_warns_once(caplog):
@@ -217,3 +219,39 @@ def test_shutdown_without_a_probe_is_a_no_op():
         zmq_context=None,
         put_get_socket=None,
     )
+
+
+def test_non_overflow_drops_are_separated_from_overflows():
+    """ListenDrops counts every establishment failure; only some are queue overflows.
+
+    The kernel charges a listening socket's sk_drops on several paths -- a full accept
+    queue, but also failures to allocate or route the new connection -- so this difference
+    is what says whether a bigger backlog could have helped.
+    """
+    stats = AcceptQueueStats(port=1234)
+    stats.first_sample = _sample(0, overflows=10, drops=20)
+    stats.last_sample = _sample(0, overflows=11, drops=25)
+
+    assert stats.overflow_delta == 1
+    assert stats.non_overflow_drop_delta == 4
+
+
+def test_pure_overflow_window_reports_no_other_drops():
+    stats = AcceptQueueStats(port=1234)
+    stats.first_sample = _sample(0, overflows=10, drops=10)
+    stats.last_sample = _sample(0, overflows=13, drops=13)
+
+    assert stats.overflow_delta == 3
+    assert stats.non_overflow_drop_delta == 0
+
+
+def test_drop_alert_does_not_assert_the_queue_overflowed(caplog):
+    """sk_drops locates the socket, not the cause, so the alert must not name one."""
+    probe = _probe()
+    probe._record(_sample(0, sk_drops=1))
+
+    with caplog.at_level("ERROR"):
+        probe._record(_sample(0, sk_drops=2))
+
+    assert "dropped an incoming connection" in caplog.text
+    assert "accept queue dropped" not in caplog.text
diff --git a/tests/test_storage_request_retry.py b/tests/test_storage_request_retry.py
index dd207353..59f82edc 100644
--- a/tests/test_storage_request_retry.py
+++ b/tests/test_storage_request_retry.py
@@ -147,11 +147,10 @@ def test_log_heavy_operation_thresholds(caplog, elapsed, payload_bytes, should_l
 @pytest.mark.parametrize(
     "tcp_result, probe_result, expected",
     [
-        # A serving unit that decoded no such request never received the one that timed out.
-        ((None, None), {"active_keys": 4, "arrivals_by_op": {"GET_DATA": 0}}, "request_lost_in_flight"),
-        # A unit that resumed inside the probe window answers too, having finished it late.
-        ((None, None), {"active_keys": 4, "arrivals_by_op": {"GET_DATA": 9}}, "arrived_but_unfinished"),
-        # Without arrival counters the probe only proves the unit is serving now.
+        # A probe that answers proves only that the unit is serving again, whatever the
+        # cumulative counters happen to say.
+        ((None, None), {"active_keys": 4, "arrivals_by_op": {"GET_DATA": 0}}, "unit_serving_again"),
+        ((None, None), {"active_keys": 4, "arrivals_by_op": {"GET_DATA": 9}}, "unit_serving_again"),
         ((None, None), {"active_keys": 4}, "unit_serving_again"),
         ((None, None), zmq.error.Again(), "unit_not_serving"),
         (ConnectionRefusedError(), zmq.error.Again(), "tcp=down(ConnectionRefusedError)"),
@@ -170,23 +169,44 @@ async def test_diagnosis_classifies_the_failure(tcp_result, probe_result, expect
     )
 
     with patch.object(ssm.asyncio, "open_connection", tcp), patch.object(manager, "_probe_storage_unit", probe):
-        diagnosis = await manager._diagnose_storage_unit("unit_a", "get")
+        diagnosis = await manager._diagnose_storage_unit("unit_a")
 
     assert expected in diagnosis
 
 
 def test_diagnosis_does_not_report_empty_op_stats_as_zero_traffic():
     """op_stats is Prometheus-gated, so an empty dict must not read as 'served nothing'."""
-    described = ssm._describe_unit_state({"requests_arrived": 7, "active_keys": 1}, "get")
+    described = ssm._describe_unit_state({"requests_arrived": 7, "active_keys": 1})
 
     assert "completed=unavailable(prometheus_disabled)" in described
     assert "completed={}" not in described
 
 
-def test_arrival_key_for_get_is_the_enum_name_not_its_wire_value():
-    """The unit keys arrivals by ZMQRequestType.name; the value is the short token 'GET'."""
-    assert ssm._ARRIVAL_KEY_BY_OPERATION["get"] == "GET_DATA"
-    assert ssm._ARRIVAL_KEY_BY_OPERATION["put"] == "PUT_DATA"
+@pytest.mark.parametrize(
+    "arrivals",
+    [
+        # Nine historical GETs, all already completed: they say nothing about this request.
+        {"GET_DATA": 9},
+        # The unit restarted during the timeout window, so its counters began again at zero.
+        {"GET_DATA": 0},
+        # The request arrived and finished after the caller gave up waiting.
+        {"GET_DATA": 10},
+    ],
+)
+def test_cumulative_counters_never_become_a_claim_about_this_request(arrivals):
+    """Counters are per-op and cumulative, so no value of them locates one request.
+
+    They carry no request id, never decrease, and reset when the unit restarts, so both
+    directions of inference are unsound: a nonzero count may be entirely historical, and a
+    zero one may mean the request reached a previous process.
+    """
+    described = ssm._describe_unit_state({"requests_arrived": sum(arrivals.values()), "arrivals_by_op": arrivals})
+
+    assert "verdict=unit_serving_again" in described
+    assert "arrived_but_unfinished" not in described
+    assert "request_lost_in_flight" not in described
+    # The raw counters stay in the line as triage input, just not as a verdict.
+    assert "arrivals_by_op=" in described
 
 
 @pytest.mark.asyncio
diff --git a/transfer_queue/storage/managers/simple_storage_manager.py b/transfer_queue/storage/managers/simple_storage_manager.py
index cce8e33a..94acbbc6 100644
--- a/transfer_queue/storage/managers/simple_storage_manager.py
+++ b/transfer_queue/storage/managers/simple_storage_manager.py
@@ -68,38 +68,17 @@ class StorageUnitTimeout(RuntimeError):
     """
 
 
-# Maps the operation name used by the retry path to the storage unit's arrival counter key.
-# Explicit rather than derived: the counters are keyed by ZMQRequestType.name ("GET_DATA"),
-# while the enum's value is the short wire token "GET", so "get".upper() would miss.
-_ARRIVAL_KEY_BY_OPERATION = {"get": "GET_DATA", "put": "PUT_DATA", "clear": "CLEAR_DATA"}
+def _describe_unit_state(body: dict[str, Any]) -> str:
+    """Summarize a successful probe: the unit is serving again, plus its own counters.
 
-
-def _describe_unit_state(body: dict[str, Any], operation: str) -> str:
-    """Turn a successful probe into a verdict plus the state it was drawn from.
-
-    The probe answering only shows the unit is serving *now*: a unit that resumed inside the
-    diagnostic window answers it too, having merely finished the original request late. The
-    arrival counter for the failed op is what separates the two, so without it this reports
-    only that the unit recovered rather than asserting where the request went.
+    Deliberately draws no conclusion about the request that timed out. The counters are
+    cumulative per operation and carry no request identity, so they cannot say whether this
+    request arrived: a nonzero count may be entirely historical, and a zero one only means
+    the unit has not decoded that operation since it last started.
     """
-    arrivals = body.get("arrivals_by_op")
-    op_key = _ARRIVAL_KEY_BY_OPERATION.get(operation)
-    if not isinstance(arrivals, dict) or op_key is None:
-        verdict = "verdict=unit_serving_again (no arrival counters to locate the request)"
-    else:
-        arrived = arrivals.get(op_key)
-        if not isinstance(arrived, int):
-            verdict = f"verdict=unit_serving_again (unit reports no {op_key} arrivals counter)"
-        elif arrived == 0:
-            verdict = f"verdict=request_lost_in_flight (unit decoded no {op_key} request at all)"
-        else:
-            verdict = (
-                f"verdict=arrived_but_unfinished (unit decoded {arrived} {op_key} request(s); "
-                f"the timed-out one reached the worker and did not complete)"
-            )
-
     parts = [
         f"requests_arrived={body.get('requests_arrived')}",
+        f"arrivals_by_op={body.get('arrivals_by_op')}",
         f"active_keys={body.get('active_keys')}",
         f"rss_gb={body.get('process_rss_bytes', 0) / 2**30:.2f}",
     ]
@@ -109,7 +88,7 @@ def _describe_unit_state(body: dict[str, Any], operation: str) -> str:
     else:
         # op_stats is populated only with Prometheus; an empty dict would read as "served nothing".
         parts.append("completed=unavailable(prometheus_disabled)")
-    return f"{verdict} ({' '.join(parts)})"
+    return f"verdict=unit_serving_again ({' '.join(parts)})"
 
 
 _SU_SUBDIR = "simple_storage"
@@ -283,14 +262,12 @@ async def _probe_storage_unit(self, target_storage_unit: str, socket: zmq.Socket
             raise RuntimeError(f"unexpected probe response type {response_msg.request_type}")
         return response_msg.body
 
-    async def _diagnose_storage_unit(self, target_storage_unit: str, operation: str = "") -> str:
-        """Classify a timeout using the unit's own arrival counters.
+    async def _diagnose_storage_unit(self, target_storage_unit: str) -> str:
+        """Report whether the unit is reachable and serving after a request to it timed out.
 
         Returns one log line and never raises: it runs while another failure is being reported.
-
-        Args:
-            target_storage_unit (str): Unit that failed to answer.
-            operation (str): Failed operation as passed to ``_request_with_retry``, e.g. ``get``.
+        Says nothing about where the timed-out request went; the unit exposes no per-request
+        state that could establish that.
         """
         info = self.storage_unit_infos.get(target_storage_unit)
         if info is None:
@@ -312,7 +289,7 @@ async def _diagnose_storage_unit(self, target_storage_unit: str, operation: str
         except Exception as e:
             return f"{tcp} verdict=unknown (probe failed: {type(e).__name__}: {e})"
 
-        return f"{tcp} {_describe_unit_state(body, operation)}"
+        return f"{tcp} {_describe_unit_state(body)}"
 
     async def _request_with_retry(
         self,
@@ -350,7 +327,7 @@ async def _request_with_retry(
                     continue
                 logger.error(
                     f"[{self.storage_manager_id}]: {operation} failed after {attempt} attempts. "
-                    f"{request_context} {e} {await self._diagnose_storage_unit(target_storage_unit, operation)}"
+                    f"{request_context} {e} {await self._diagnose_storage_unit(target_storage_unit)}"
                 )
                 raise
 
diff --git a/transfer_queue/storage/simple_storage.py b/transfer_queue/storage/simple_storage.py
index 5f150247..24258b99 100644
--- a/transfer_queue/storage/simple_storage.py
+++ b/transfer_queue/storage/simple_storage.py
@@ -632,6 +632,7 @@ def _handle_get_metrics(self) -> ZMQMessage:
                 "peak_utilization": stats.peak_utilization,
                 "sk_drops_delta": stats.sk_drops_delta,
                 "listen_overflow_delta": stats.overflow_delta,
+                "non_overflow_drop_delta": stats.non_overflow_drop_delta,
                 "samples": stats.samples,
             }
 
diff --git a/transfer_queue/utils/accept_probe.py b/transfer_queue/utils/accept_probe.py
index f5985514..f2bd3ed4 100644
--- a/transfer_queue/utils/accept_probe.py
+++ b/transfer_queue/utils/accept_probe.py
@@ -17,21 +17,29 @@
 
 A get request was observed leaving the manager (TCP counted the bytes and the peer
 ACKed them) while the unit's worker never saw it: its GET_DATA counter did not move
-and it kept polling. The suspected drop point is the listen socket's accept queue,
-which ZMQ leaves at its default backlog of 100 and which the kernel empties
-silently (``tcp_abort_on_overflow=0``), so neither end raises an error.
+and it kept polling. One candidate drop point is the listen socket's accept queue,
+which ZMQ leaves at its default backlog of 100. With the default
+``tcp_abort_on_overflow=0`` an overflowing queue drops the client's final ACK rather
+than sending an RST, so neither end reports an error; the server keeps retransmitting
+its SYN-ACK, and the connection still completes if the queue drains before
+``tcp_synack_retries`` runs out.
 
 This module samples the kernel's own view of that queue so the guess becomes a
 measurement:
 
 * ``Recv-Q`` on a listening socket is the number of established-but-not-yet-accepted
   connections, i.e. the live queue depth.
-* ``ListenOverflows`` / ``ListenDrops`` count queue-full events machine-wide.
-* ``sk_drops`` counts drops charged to one specific socket.
+* ``ListenOverflows`` counts accept-queue overflows, and ``ListenDrops`` counts every
+  connection dropped during establishment, overflow or not. Both are per network
+  namespace, so neither attributes an event to one port.
+* ``sk_drops`` is charged to one specific socket, but the kernel increments it on
+  several establishment failures -- a full queue is only one of them -- so a rise
+  locates the socket, not the cause.
 
 Sampling runs in a thread because the queue drains in milliseconds; a value read
 after a hang has already returned to zero, which is exactly what earlier
-post-mortem inspection saw.
+post-mortem inspection saw. A zero reading therefore cannot rule out an earlier
+burst, which is why the deltas matter alongside the instantaneous depth.
 """
 
 from __future__ import annotations
@@ -89,11 +97,24 @@ def sk_drops_delta(self) -> int:
 
     @property
     def overflow_delta(self) -> int:
-        """Machine-wide accept-queue overflows during the window, 0 until two samples exist."""
+        """Namespace-wide accept-queue overflows during the window, 0 until two samples exist."""
         if self.first_sample is None or self.last_sample is None:
             return 0
         return self.last_sample.listen_overflows - self.first_sample.listen_overflows
 
+    @property
+    def non_overflow_drop_delta(self) -> int:
+        """Namespace-wide establishment drops during the window that were not overflows.
+
+        ListenDrops counts every connection dropped while being established and
+        ListenOverflows only the full-queue ones, so a positive difference is direct
+        evidence that raising the backlog would not have prevented all of them.
+        """
+        if self.first_sample is None or self.last_sample is None:
+            return 0
+        drops = self.last_sample.listen_drops - self.first_sample.listen_drops
+        return drops - self.overflow_delta
+
     def describe(self) -> str:
         """Return a one-line summary of the window, for the probe's shutdown log."""
         return (
@@ -123,7 +144,7 @@ def _read_listen_socket(port: int) -> tuple[int, int, int] | None:
 
 
 def _read_listen_overflows() -> tuple[int, int]:
-    """Return machine-wide (ListenOverflows, ListenDrops) from /proc/net/netstat."""
+    """Return this namespace's (ListenOverflows, ListenDrops) from /proc/net/netstat."""
     try:
         with open("/proc/net/netstat") as handle:
             lines = handle.read().splitlines()
@@ -232,12 +253,15 @@ def _record(self, sample: AcceptQueueSample) -> None:
         # drop that happened once, on every sample, and erase when it actually occurred.
         if previous is not None and sample.sk_drops > previous.sk_drops:
             logger.error(
-                f"[{self.owner_id}]: accept queue dropped a connection on port {self.port}. "
-                f"recv_q={sample.recv_q}/{sample.backlog} sk_drops={sample.sk_drops} "
+                f"[{self.owner_id}]: listening socket on port {self.port} dropped an incoming "
+                f"connection. recv_q={sample.recv_q}/{sample.backlog} sk_drops={sample.sk_drops} "
                 f"(+{sample.sk_drops - previous.sk_drops} since the last sample, "
-                f"+{stats.sk_drops_delta} since probe start). "
-                f"A silently dropped connection leaves the client in ESTABLISHED with no "
-                f"reply; raise ZMQ_BACKLOG above {sample.backlog}."
+                f"+{stats.sk_drops_delta} since probe start); netns since probe start: "
+                f"listen_overflows +{stats.overflow_delta}, other establishment drops "
+                f"+{stats.non_overflow_drop_delta}. The kernel charges sk_drops for several "
+                f"establishment failures, so a full queue is only one candidate: raising "
+                f"ZMQ_BACKLOG above {sample.backlog} helps only if recv_q above and the "
+                f"overflow delta point that way."
             )
 
         if not self._warned and sample.utilization >= self.warn_utilization:

From da8acd0f6cbdab7ac79855d8d0e2ccf75836b15f Mon Sep 17 00:00:00 2001
From: OutstanderWang <wangweiyanster@gmail.com>
Date: Fri, 11 Sep 2026 20:49:03 +0800
Subject: [PATCH 7/8] [metrics] Export the request-loss diagnostics to
 Prometheus and Grafana

The arrival counters and accept-queue probe only reached log strings, so nothing
could chart them or alert on them. Export both through the controller's exporter,
which already collects storage-unit metrics over ZMQ.

Arrivals become tq_storage_requests_arrived and tq_storage_arrivals_by_op, the
latter labelled by op_type like the existing tq_storage_request_ops it is meant to
be read against: arrivals count decode, request_ops counts completion, so a gap
between the two rates is requests arriving and not finishing.

The accept-queue series carry the probe's backlog, peak depth, peak utilization and
drop deltas. They are removed rather than set to zero when the probe is disabled,
following the capacity-is-None precedent, because a zero drop count would otherwise
be indistinguishable from a measured absence of drops.

Overflow and non-overflow drops are exported separately. The kernel charges a
listening socket's sk_drops on several establishment failures, so the split is what
tells a dashboard whether raising the backlog would have helped.

Names omit the _total suffix per the rule in docs/metrics.md: these are Gauges fed
from a remote body, and the reserved suffix breaks label_values() queries.

Adds a Grafana row with those four views, and documents the metrics, the two
environment variables that drive them, and the cumulative-since-probe-start
semantics that make rate() the right operator.

Signed-off-by: OutstanderWang <wangweiyanster@gmail.com>
---
 docs/metrics.md                               | 35 ++++++++
 scripts/grafana_dashboard.json                | 62 ++++++++++++++
 tests/test_metrics.py                         | 78 ++++++++++++++++++
 transfer_queue/metrics.py                     | 81 +++++++++++++++++++
 .../managers/simple_storage_manager.py        |  2 -
 transfer_queue/storage/simple_storage.py      |  4 +-
 6 files changed, 257 insertions(+), 5 deletions(-)

diff --git a/docs/metrics.md b/docs/metrics.md
index 05559dd5..327f6079 100644
--- a/docs/metrics.md
+++ b/docs/metrics.md
@@ -60,6 +60,8 @@ Steps:
 |---------------------|---------|-------------|
 | `TQ_METRICS_COLLECT_INTERVAL` | `10` | Background collection interval (seconds) |
 | `TQ_METRICS_STORAGE_TIMEOUT` | `5` | ZMQ timeout for storage unit queries (seconds) |
+| `TQ_ACCEPT_PROBE_INTERVAL` | `0` (off) | Accept-queue sampling period (seconds); enables the accept-queue metrics below |
+| `TQ_STORAGE_ZMQ_BACKLOG` | `4096` | Accept-queue depth for the storage unit's listening socket |
 
 ## Architecture
 
@@ -140,6 +142,38 @@ Steps:
 | `tq_storage_request_latency_p50` | Gauge | `storage_unit_id`, `op_type` | P50 request latency (seconds) |
 | `tq_storage_request_latency_p99` | Gauge | `storage_unit_id`, `op_type` | P99 request latency (seconds) |
 
+### Storage Request-Loss Diagnostics (collected via ZMQ, exposed on controller)
+
+Arrivals are counted when the worker decodes a request, while `tq_storage_request_ops` only
+advances once one completes, so a sustained gap between them means requests are arriving and
+not finishing. The counters are cumulative per operation and carry no request identity, so
+they characterise a unit, not any individual request.
+
+| Metric | Type | Labels | Description |
+|--------|------|--------|-------------|
+| `tq_storage_requests_arrived` | Gauge | `storage_unit_id` | Requests decoded by the worker, whether or not they completed |
+| `tq_storage_arrivals_by_op` | Gauge | `storage_unit_id`, `op_type` | Same, broken down by operation |
+
+The accept-queue series below exist only when the unit runs with `TQ_ACCEPT_PROBE_INTERVAL`
+set; the series are removed rather than reported as zero when the probe is off, so a missing
+series means "not measured" rather than "no drops". `tq_storage_accept_queue_peak` and the two
+drop counters are cumulative since the probe started, so use `rate()` on them.
+
+| Metric | Type | Labels | Description |
+|--------|------|--------|-------------|
+| `tq_storage_accept_queue_backlog` | Gauge | `storage_unit_id` | Configured accept-queue depth (`TQ_STORAGE_ZMQ_BACKLOG`) |
+| `tq_storage_accept_queue_peak` | Gauge | `storage_unit_id` | Deepest accept-queue occupancy seen by the probe |
+| `tq_storage_accept_queue_peak_utilization_ratio` | Gauge | `storage_unit_id` | Peak occupancy as a fraction of the backlog |
+| `tq_storage_socket_drops` | Gauge | `storage_unit_id` | Connections dropped on this listening socket since the probe started |
+| `tq_storage_listen_overflows` | Gauge | `storage_unit_id` | Namespace-wide accept-queue overflows since the probe started |
+| `tq_storage_listen_other_drops` | Gauge | `storage_unit_id` | Namespace-wide establishment drops that were **not** overflows |
+
+The kernel charges a listening socket's `sk_drops` on several connection-establishment
+failures — a full accept queue, but also failures to allocate or route the new connection — so
+`tq_storage_socket_drops` rising locates the socket, not the cause. Compare the last two
+series: a non-zero `tq_storage_listen_other_drops` means raising the backlog would not have
+prevented every drop in that window.
+
 ### Storage Unit Native Metrics (exposed on each storage unit's own endpoint)
 
 | Metric | Type | Labels | Description |
@@ -163,6 +197,7 @@ The dashboard ([`scripts/grafana_dashboard.json`](../scripts/grafana_dashboard.j
 | **Request Throughput & Latency** | Controller Request Rate (ops/s), Controller Request Latency (repeats per quantile) |
 | **Partition Status** | Samples per Partition, Production Progress, Consumption Progress |
 | **Storage Units** | Utilization Bar Gauge, Active Keys, Capacity vs Active Keys, RSS Memory, Storage Request Rate, Storage Request Latency (repeats per quantile), Produced vs Cleared Samples/s, Active Keys Delta |
+| **Storage Request-Loss Diagnostics** | Arrived vs Completed/s, Accept-Queue Peak vs Backlog, Listening-Socket Drops/s, Overflow vs Other Establishment Drops/s |
 
 ### Template Variables
 
diff --git a/scripts/grafana_dashboard.json b/scripts/grafana_dashboard.json
index 0e6c5341..9f3ee36b 100644
--- a/scripts/grafana_dashboard.json
+++ b/scripts/grafana_dashboard.json
@@ -370,6 +370,68 @@
         { "expr": "sum(tq_storage_active_keys_total)", "legendFormat": "Total Active Keys (all storage units)" }
       ],
       "type": "timeseries"
+    },
+    {
+      "collapsed": false,
+      "gridPos": { "h": 1, "w": 24, "x": 0, "y": 86 },
+      "id": 104,
+      "title": "Storage Request-Loss Diagnostics",
+      "type": "row"
+    },
+    {
+      "datasource": { "type": "prometheus", "uid": "${datasource}" },
+      "description": "Arrivals are counted when the worker decodes a request; completions only when it finishes. A sustained gap means requests are arriving and not completing. Neither series identifies an individual request.",
+      "fieldConfig": { "defaults": { "color": { "mode": "palette-classic" }, "custom": { "fillOpacity": 10, "lineWidth": 2 } } },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 87 },
+      "id": 40,
+      "options": { "legend": { "calcs": ["lastNotNull"], "displayMode": "table", "placement": "bottom" }, "tooltip": { "mode": "multi" } },
+      "title": "Arrived vs Completed (per second)",
+      "targets": [
+        { "expr": "sum by (op_type) (rate(tq_storage_arrivals_by_op{op_type=~\"$op_type\"}[$__rate_interval]))", "legendFormat": "arrived {{ op_type }}" },
+        { "expr": "sum by (op_type) (rate(tq_storage_request_ops{op_type=~\"$op_type\"}[$__rate_interval]))", "legendFormat": "completed {{ op_type }}" }
+      ],
+      "type": "timeseries"
+    },
+    {
+      "datasource": { "type": "prometheus", "uid": "${datasource}" },
+      "description": "Peak is a high-water mark since the probe started, so it never decreases. Only present when TQ_ACCEPT_PROBE_INTERVAL is set.",
+      "fieldConfig": { "defaults": { "color": { "mode": "palette-classic" }, "custom": { "fillOpacity": 10, "lineWidth": 2 } } },
+      "gridPos": { "h": 8, "w": 12, "x": 12, "y": 87 },
+      "id": 41,
+      "options": { "legend": { "calcs": ["lastNotNull"], "displayMode": "table", "placement": "bottom" }, "tooltip": { "mode": "multi" } },
+      "title": "Accept-Queue Peak vs Backlog",
+      "targets": [
+        { "expr": "tq_storage_accept_queue_peak", "legendFormat": "peak {{ storage_unit_id }}" },
+        { "expr": "tq_storage_accept_queue_backlog", "legendFormat": "backlog {{ storage_unit_id }}" }
+      ],
+      "type": "timeseries"
+    },
+    {
+      "datasource": { "type": "prometheus", "uid": "${datasource}" },
+      "description": "The kernel charges sk_drops for several connection-establishment failures, of which a full accept queue is only one, so a rise locates the socket, not the cause.",
+      "fieldConfig": { "defaults": { "color": { "mode": "palette-classic" }, "custom": { "fillOpacity": 10, "lineWidth": 2 } } },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 95 },
+      "id": 42,
+      "options": { "legend": { "calcs": ["lastNotNull"], "displayMode": "table", "placement": "bottom" }, "tooltip": { "mode": "multi" } },
+      "title": "Listening-Socket Drops (per second)",
+      "targets": [
+        { "expr": "sum by (storage_unit_id) (rate(tq_storage_socket_drops[$__rate_interval]))", "legendFormat": "{{ storage_unit_id }}" }
+      ],
+      "type": "timeseries"
+    },
+    {
+      "datasource": { "type": "prometheus", "uid": "${datasource}" },
+      "description": "Namespace-wide, not per port. A non-zero 'other' series means raising the backlog would not have prevented every drop in the window.",
+      "fieldConfig": { "defaults": { "color": { "mode": "palette-classic" }, "custom": { "fillOpacity": 10, "lineWidth": 2 } } },
+      "gridPos": { "h": 8, "w": 12, "x": 12, "y": 95 },
+      "id": 43,
+      "options": { "legend": { "calcs": ["lastNotNull"], "displayMode": "table", "placement": "bottom" }, "tooltip": { "mode": "multi" } },
+      "title": "Overflow vs Other Establishment Drops (per second)",
+      "targets": [
+        { "expr": "sum (rate(tq_storage_listen_overflows[$__rate_interval]))", "legendFormat": "accept-queue overflow" },
+        { "expr": "sum (rate(tq_storage_listen_other_drops[$__rate_interval]))", "legendFormat": "other establishment drops" }
+      ],
+      "type": "timeseries"
     }
   ],
   "refresh": "10s",
diff --git a/tests/test_metrics.py b/tests/test_metrics.py
index a8ac3e2b..66e82a3f 100644
--- a/tests/test_metrics.py
+++ b/tests/test_metrics.py
@@ -98,6 +98,14 @@ def test_all_metrics_are_registered(self):
             "tq_storage_active_keys_total",
             "tq_storage_utilization_ratio",
             "tq_storage_memory_rss_bytes",
+            "tq_storage_requests_arrived",
+            "tq_storage_arrivals_by_op",
+            "tq_storage_accept_queue_backlog",
+            "tq_storage_accept_queue_peak",
+            "tq_storage_accept_queue_peak_utilization_ratio",
+            "tq_storage_socket_drops",
+            "tq_storage_listen_overflows",
+            "tq_storage_listen_other_drops",
         ]
 
         registered = {m.name for m in exporter.registry.collect()}
@@ -315,6 +323,76 @@ def test_storage_metrics_populated_on_success(self):
         assert exporter.storage_utilization.labels(storage_unit_id="SU_001")._value.get() == 0.25
         assert exporter.storage_memory_rss.labels(storage_unit_id="SU_001")._value.get() == 512 * 1024 * 1024
 
+    def test_arrival_counters_are_exported(self):
+        """Arrival counts reach Prometheus, so a dashboard can compare them with completions."""
+        exporter = TQMetricsExporter()
+        fake_su_info = MagicMock()
+        fake_su_info.id = "SU_001"
+        exporter._storage_unit_infos = {"SU_001": fake_su_info}
+        exporter._query_storage_unit = MagicMock(
+            return_value={
+                "storage_unit_id": "SU_001",
+                "capacity": 1000,
+                "active_keys": 1,
+                "requests_arrived": 42,
+                "arrivals_by_op": {"GET_DATA": 30, "PUT_DATA": 12},
+            }
+        )
+
+        exporter.collect_storage_metrics()
+
+        assert exporter.storage_requests_arrived.labels(storage_unit_id="SU_001")._value.get() == 42
+        by_op = exporter.storage_arrivals_by_op
+        assert by_op.labels(storage_unit_id="SU_001", op_type="GET_DATA")._value.get() == 30
+        assert by_op.labels(storage_unit_id="SU_001", op_type="PUT_DATA")._value.get() == 12
+
+    def test_accept_queue_metrics_are_exported(self):
+        """The overflow/non-overflow split is what tells a dashboard if backlog is the issue."""
+        exporter = TQMetricsExporter()
+        fake_su_info = MagicMock()
+        fake_su_info.id = "SU_001"
+        exporter._storage_unit_infos = {"SU_001": fake_su_info}
+        exporter._query_storage_unit = MagicMock(
+            return_value={
+                "storage_unit_id": "SU_001",
+                "capacity": 1000,
+                "active_keys": 1,
+                "accept_queue": {
+                    "backlog": 4096,
+                    "peak_recv_q": 97,
+                    "peak_utilization": 0.02,
+                    "sk_drops_delta": 5,
+                    "listen_overflow_delta": 2,
+                    "non_overflow_drop_delta": 3,
+                },
+            }
+        )
+
+        exporter.collect_storage_metrics()
+
+        label = {"storage_unit_id": "SU_001"}
+        assert exporter.storage_accept_queue_backlog.labels(**label)._value.get() == 4096
+        assert exporter.storage_accept_queue_peak.labels(**label)._value.get() == 97
+        assert exporter.storage_socket_drops.labels(**label)._value.get() == 5
+        assert exporter.storage_listen_overflows.labels(**label)._value.get() == 2
+        assert exporter.storage_listen_other_drops.labels(**label)._value.get() == 3
+
+    def test_accept_queue_series_absent_when_the_probe_is_off(self):
+        """The probe is opt-in; reporting 0 drops would read as 'measured, and none'."""
+        exporter = TQMetricsExporter()
+        fake_su_info = MagicMock()
+        fake_su_info.id = "SU_001"
+        exporter._storage_unit_infos = {"SU_001": fake_su_info}
+        exporter._query_storage_unit = MagicMock(
+            return_value={"storage_unit_id": "SU_001", "capacity": 1000, "active_keys": 1}
+        )
+
+        exporter.collect_storage_metrics()
+
+        exported = {sample.name for metric in exporter.registry.collect() for sample in metric.samples}
+        assert "tq_storage_socket_drops" not in exported
+        assert "tq_storage_accept_queue_backlog" not in exported
+
     def test_storage_metrics_handles_query_failure(self):
         """If a storage unit query fails, other units should still be collected."""
         exporter = TQMetricsExporter()
diff --git a/transfer_queue/metrics.py b/transfer_queue/metrics.py
index 12914585..e179bff7 100644
--- a/transfer_queue/metrics.py
+++ b/transfer_queue/metrics.py
@@ -185,6 +185,63 @@ def _define_controller_metrics(self) -> None:
             "tq_storage_memory_rss_bytes", "Storage unit process RSS memory", ["storage_unit_id"], registry=r
         )
 
+        # ---- Storage-unit request-loss diagnostics ----
+        # Requests counted as the worker decodes them, so comparing this against
+        # tq_storage_request_ops (which advances only on completion) shows requests that
+        # arrived and did not finish. Neither locates an individual request.
+        self.storage_requests_arrived = Gauge(
+            "tq_storage_requests_arrived",
+            "Requests decoded by the storage unit worker, whether or not they completed",
+            ["storage_unit_id"],
+            registry=r,
+        )
+        self.storage_arrivals_by_op = Gauge(
+            "tq_storage_arrivals_by_op",
+            "Requests decoded by the storage unit worker, by operation",
+            ["storage_unit_id", "op_type"],
+            registry=r,
+        )
+
+        # ---- Accept-queue probe (only populated when TQ_ACCEPT_PROBE_INTERVAL > 0) ----
+        self.storage_accept_queue_backlog = Gauge(
+            "tq_storage_accept_queue_backlog",
+            "Configured accept-queue depth of the storage unit's listening socket",
+            ["storage_unit_id"],
+            registry=r,
+        )
+        self.storage_accept_queue_peak = Gauge(
+            "tq_storage_accept_queue_peak",
+            "Deepest accept-queue occupancy seen by the probe",
+            ["storage_unit_id"],
+            registry=r,
+        )
+        self.storage_accept_queue_peak_utilization = Gauge(
+            "tq_storage_accept_queue_peak_utilization_ratio",
+            "Peak accept-queue occupancy as a fraction of the backlog",
+            ["storage_unit_id"],
+            registry=r,
+        )
+        self.storage_socket_drops = Gauge(
+            "tq_storage_socket_drops",
+            "Connections dropped on this listening socket since the probe started",
+            ["storage_unit_id"],
+            registry=r,
+        )
+        # Split so a dashboard can tell whether raising the backlog would have helped: the
+        # kernel charges sk_drops for several establishment failures, not only a full queue.
+        self.storage_listen_overflows = Gauge(
+            "tq_storage_listen_overflows",
+            "Namespace-wide accept-queue overflows since the probe started",
+            ["storage_unit_id"],
+            registry=r,
+        )
+        self.storage_listen_other_drops = Gauge(
+            "tq_storage_listen_other_drops",
+            "Namespace-wide establishment drops that were not accept-queue overflows",
+            ["storage_unit_id"],
+            registry=r,
+        )
+
         # ---- Storage request metrics (collected via ZMQ, exposed as gauges) ----
         # P50/P99 are pre-computed on the storage unit side and sent via ZMQ,
         # avoiding the need to replicate histogram bucket structures (which
@@ -362,6 +419,30 @@ def collect_storage_metrics(self) -> None:
                 self.storage_active_keys.labels(storage_unit_id=label).set(active)
                 self.storage_memory_rss.labels(storage_unit_id=label).set(metrics.get("process_rss_bytes", 0))
 
+                self.storage_requests_arrived.labels(storage_unit_id=label).set(metrics.get("requests_arrived", 0))
+                for op_type, arrived in (metrics.get("arrivals_by_op") or {}).items():
+                    self.storage_arrivals_by_op.labels(storage_unit_id=label, op_type=op_type).set(arrived)
+
+                # Absent unless the unit runs with TQ_ACCEPT_PROBE_INTERVAL set. Drop the series
+                # rather than reporting zero, so a disabled probe is not read as "no drops".
+                accept_queue = metrics.get("accept_queue")
+                accept_gauges = (
+                    (self.storage_accept_queue_backlog, "backlog"),
+                    (self.storage_accept_queue_peak, "peak_recv_q"),
+                    (self.storage_accept_queue_peak_utilization, "peak_utilization"),
+                    (self.storage_socket_drops, "sk_drops_delta"),
+                    (self.storage_listen_overflows, "listen_overflow_delta"),
+                    (self.storage_listen_other_drops, "non_overflow_drop_delta"),
+                )
+                for gauge, key in accept_gauges:
+                    if accept_queue is None:
+                        try:
+                            gauge.remove(label)
+                        except (KeyError, ValueError):
+                            pass
+                    else:
+                        gauge.labels(storage_unit_id=label).set(accept_queue.get(key, 0))
+
                 # Per-operation request stats
                 for op_type, op_data in metrics.get("op_stats", {}).items():
                     self.storage_request_ops.labels(storage_unit_id=label, op_type=op_type).set(
diff --git a/transfer_queue/storage/managers/simple_storage_manager.py b/transfer_queue/storage/managers/simple_storage_manager.py
index 94acbbc6..64868c52 100644
--- a/transfer_queue/storage/managers/simple_storage_manager.py
+++ b/transfer_queue/storage/managers/simple_storage_manager.py
@@ -63,8 +63,6 @@ class StorageUnitTimeout(RuntimeError):
     """A storage unit did not answer within the send/recv timeout.
 
     Distinct from an error the unit reported: only a missing answer is worth a new connection.
-    The message must name the unit, its endpoint and the timeout, because it is what the callers
-    of ``put_data`` and ``get_data`` see, and the retry logs rely on it instead of repeating them.
     """
 
 
diff --git a/transfer_queue/storage/simple_storage.py b/transfer_queue/storage/simple_storage.py
index 24258b99..d71bc991 100644
--- a/transfer_queue/storage/simple_storage.py
+++ b/transfer_queue/storage/simple_storage.py
@@ -617,9 +617,7 @@ def _handle_get_metrics(self) -> ZMQMessage:
             "capacity": self.storage_unit_size,
             "active_keys": self.storage_data.active_key_count,
             "process_rss_bytes": process_rss,
-            # Reported next to but separately from op_stats below, which is derived from
-            # completion-time histograms: a gap between the two is a request that arrived and
-            # never finished, which the diagnostic probe cannot otherwise distinguish.
+            # Counted on arrival; op_stats below only advances on completion.
             "requests_arrived": self._requests_arrived,
             "arrivals_by_op": dict(self._arrivals_by_op),
         }

From f85e1deb5bdcaca8efaa8d2d05892db001152dbe Mon Sep 17 00:00:00 2001
From: OutstanderWang <wangweiyanster@gmail.com>
Date: Fri, 11 Sep 2026 21:21:08 +0800
Subject: [PATCH 8/8] [storage] Trim the comments this branch added

Compress the prose the diagnostics accumulated across review rounds. The
accept_probe module docstring had grown to 28 lines recounting one incident and
restating each kernel counter; it keeps only the two caveats a caller needs to read
the numbers correctly and points at docs/metrics.md for the rest. Several comments
that restated their code are gone, and the invariant docstrings on the diagnosis
path are shortened rather than dropped, since they are what stops the overclaiming
verdict coming back.

Also drops AcceptQueueStats.peak_history, which was appended to on every new peak
and never read.

No behavior change.

Signed-off-by: OutstanderWang <wangweiyanster@gmail.com>
---
 transfer_queue/metrics.py                     | 13 ++---
 .../managers/simple_storage_manager.py        | 11 ++--
 transfer_queue/storage/simple_storage.py      | 15 ++---
 transfer_queue/utils/accept_probe.py          | 58 +++++--------------
 4 files changed, 30 insertions(+), 67 deletions(-)

diff --git a/transfer_queue/metrics.py b/transfer_queue/metrics.py
index e179bff7..e75792b5 100644
--- a/transfer_queue/metrics.py
+++ b/transfer_queue/metrics.py
@@ -186,9 +186,8 @@ def _define_controller_metrics(self) -> None:
         )
 
         # ---- Storage-unit request-loss diagnostics ----
-        # Requests counted as the worker decodes them, so comparing this against
-        # tq_storage_request_ops (which advances only on completion) shows requests that
-        # arrived and did not finish. Neither locates an individual request.
+        # Read against tq_storage_request_ops, which advances only on completion: a gap
+        # between the two is requests arriving and not finishing.
         self.storage_requests_arrived = Gauge(
             "tq_storage_requests_arrived",
             "Requests decoded by the storage unit worker, whether or not they completed",
@@ -227,8 +226,8 @@ def _define_controller_metrics(self) -> None:
             ["storage_unit_id"],
             registry=r,
         )
-        # Split so a dashboard can tell whether raising the backlog would have helped: the
-        # kernel charges sk_drops for several establishment failures, not only a full queue.
+        # Split because sk_drops covers several establishment failures, not only a full
+        # queue; the difference is what says whether a bigger backlog would have helped.
         self.storage_listen_overflows = Gauge(
             "tq_storage_listen_overflows",
             "Namespace-wide accept-queue overflows since the probe started",
@@ -423,8 +422,8 @@ def collect_storage_metrics(self) -> None:
                 for op_type, arrived in (metrics.get("arrivals_by_op") or {}).items():
                     self.storage_arrivals_by_op.labels(storage_unit_id=label, op_type=op_type).set(arrived)
 
-                # Absent unless the unit runs with TQ_ACCEPT_PROBE_INTERVAL set. Drop the series
-                # rather than reporting zero, so a disabled probe is not read as "no drops".
+                # Drop the series rather than report zero when the probe is off, so a
+                # disabled probe is not read as "measured, and no drops".
                 accept_queue = metrics.get("accept_queue")
                 accept_gauges = (
                     (self.storage_accept_queue_backlog, "backlog"),
diff --git a/transfer_queue/storage/managers/simple_storage_manager.py b/transfer_queue/storage/managers/simple_storage_manager.py
index 64868c52..c89d1cfb 100644
--- a/transfer_queue/storage/managers/simple_storage_manager.py
+++ b/transfer_queue/storage/managers/simple_storage_manager.py
@@ -69,10 +69,8 @@ class StorageUnitTimeout(RuntimeError):
 def _describe_unit_state(body: dict[str, Any]) -> str:
     """Summarize a successful probe: the unit is serving again, plus its own counters.
 
-    Deliberately draws no conclusion about the request that timed out. The counters are
-    cumulative per operation and carry no request identity, so they cannot say whether this
-    request arrived: a nonzero count may be entirely historical, and a zero one only means
-    the unit has not decoded that operation since it last started.
+    Draws no conclusion about the timed-out request: the counters are cumulative per
+    operation and carry no request identity, so no value of them locates one request.
     """
     parts = [
         f"requests_arrived={body.get('requests_arrived')}",
@@ -84,7 +82,7 @@ def _describe_unit_state(body: dict[str, Any]) -> str:
     if op_stats:
         parts.append(f"completed={ {op: stats.get('request_count') for op, stats in op_stats.items()} }")
     else:
-        # op_stats is populated only with Prometheus; an empty dict would read as "served nothing".
+        # Only with Prometheus enabled; an empty dict would read as "served nothing".
         parts.append("completed=unavailable(prometheus_disabled)")
     return f"verdict=unit_serving_again ({' '.join(parts)})"
 
@@ -264,8 +262,7 @@ async def _diagnose_storage_unit(self, target_storage_unit: str) -> str:
         """Report whether the unit is reachable and serving after a request to it timed out.
 
         Returns one log line and never raises: it runs while another failure is being reported.
-        Says nothing about where the timed-out request went; the unit exposes no per-request
-        state that could establish that.
+        Says nothing about where that request went; no per-request state exists to show it.
         """
         info = self.storage_unit_infos.get(target_storage_unit)
         if info is None:
diff --git a/transfer_queue/storage/simple_storage.py b/transfer_queue/storage/simple_storage.py
index d71bc991..61c5e9a1 100644
--- a/transfer_queue/storage/simple_storage.py
+++ b/transfer_queue/storage/simple_storage.py
@@ -53,12 +53,10 @@
 # Marks a GET_ERROR reply as "the key is gone" so the caller can tell it apart from a real fault.
 KEY_NOT_FOUND_MARKER = "TQKeyNotFound"
 
-# Accept-queue depth for the client-facing ROUTER, well above ZMQ's default of 100 because a
-# full accept queue is drained without an RST and so loses connections silently.
+# Accept-queue depth for the client-facing ROUTER. A full queue loses connections silently.
 TQ_STORAGE_ZMQ_BACKLOG = int(os.environ.get("TQ_STORAGE_ZMQ_BACKLOG", 4096))
 
-# Sampling period for the accept-queue probe, in seconds. 0 disables it. Sub-second because
-# the queue drains in milliseconds, so a reading taken after a hang is always zero.
+# Accept-queue sampling period in seconds; 0 disables the probe. Keep sub-second.
 TQ_ACCEPT_PROBE_INTERVAL = float(os.environ.get("TQ_ACCEPT_PROBE_INTERVAL", 0))
 
 
@@ -193,8 +191,6 @@ def __init__(self, storage_unit_size: int | None = None):
 
         self.storage_data = StorageUnitData(self.storage_unit_size)
 
-        # Own copies so counts stay per unit; the class-level defaults above only exist for
-        # instances built without __init__.
         self._requests_arrived = 0
         self._arrivals_by_op = {}
 
@@ -250,8 +246,8 @@ def _init_zmq_socket(self) -> None:
                 continue
 
         if TQ_ACCEPT_PROBE_INTERVAL > 0:
-            # Imported lazily: the probe shells out to ``ss`` on a timer, so a run that has
-            # not asked for it should not even load the module.
+            # Lazy: the probe shells out to ``ss`` on a timer, so keep it out of runs that
+            # have not enabled it.
             from transfer_queue.utils.accept_probe import AcceptQueueProbe
 
             self._accept_probe = AcceptQueueProbe(
@@ -799,8 +795,7 @@ def _shutdown_resources(
         # Signal all threads to stop
         shutdown_event.set()
 
-        # Stop before the ZMQ teardown: the probe samples on its own timer and would keep
-        # spawning `ss` after the unit is gone, and stopping it logs the window summary.
+        # Before the ZMQ teardown: the probe runs on its own timer and would outlive the unit.
         if accept_probe is not None:
             accept_probe.stop()
 
diff --git a/transfer_queue/utils/accept_probe.py b/transfer_queue/utils/accept_probe.py
index f2bd3ed4..916955c4 100644
--- a/transfer_queue/utils/accept_probe.py
+++ b/transfer_queue/utils/accept_probe.py
@@ -13,33 +13,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Accept-queue instrumentation for the storage-unit ROUTER socket.
-
-A get request was observed leaving the manager (TCP counted the bytes and the peer
-ACKed them) while the unit's worker never saw it: its GET_DATA counter did not move
-and it kept polling. One candidate drop point is the listen socket's accept queue,
-which ZMQ leaves at its default backlog of 100. With the default
-``tcp_abort_on_overflow=0`` an overflowing queue drops the client's final ACK rather
-than sending an RST, so neither end reports an error; the server keeps retransmitting
-its SYN-ACK, and the connection still completes if the queue drains before
-``tcp_synack_retries`` runs out.
-
-This module samples the kernel's own view of that queue so the guess becomes a
-measurement:
-
-* ``Recv-Q`` on a listening socket is the number of established-but-not-yet-accepted
-  connections, i.e. the live queue depth.
-* ``ListenOverflows`` counts accept-queue overflows, and ``ListenDrops`` counts every
-  connection dropped during establishment, overflow or not. Both are per network
-  namespace, so neither attributes an event to one port.
-* ``sk_drops`` is charged to one specific socket, but the kernel increments it on
-  several establishment failures -- a full queue is only one of them -- so a rise
-  locates the socket, not the cause.
-
-Sampling runs in a thread because the queue drains in milliseconds; a value read
-after a hang has already returned to zero, which is exactly what earlier
-post-mortem inspection saw. A zero reading therefore cannot rule out an earlier
-burst, which is why the deltas matter alongside the instantaneous depth.
+"""Accept-queue sampling for the storage-unit ROUTER socket.
+
+A connection dropped while being established is invisible to both ends, so the only
+way to see it is to read the kernel's own counters. Two caveats govern how callers
+may read what this reports: ``ListenOverflows`` and ``ListenDrops`` are per network
+namespace rather than per port, and the kernel charges a socket's ``sk_drops`` for
+several establishment failures, of which a full queue is only one. See
+``docs/metrics.md`` for the exported metrics.
 """
 
 from __future__ import annotations
@@ -48,7 +29,7 @@
 import subprocess
 import threading
 import time
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 
 from transfer_queue.utils.logging_utils import get_logger
 
@@ -86,7 +67,6 @@ class AcceptQueueStats:
     backlog: int = 0
     first_sample: AcceptQueueSample | None = None
     last_sample: AcceptQueueSample | None = None
-    peak_history: list[tuple[float, int]] = field(default_factory=list)
 
     @property
     def sk_drops_delta(self) -> int:
@@ -104,12 +84,7 @@ def overflow_delta(self) -> int:
 
     @property
     def non_overflow_drop_delta(self) -> int:
-        """Namespace-wide establishment drops during the window that were not overflows.
-
-        ListenDrops counts every connection dropped while being established and
-        ListenOverflows only the full-queue ones, so a positive difference is direct
-        evidence that raising the backlog would not have prevented all of them.
-        """
+        """Establishment drops in this window that were not accept-queue overflows."""
         if self.first_sample is None or self.last_sample is None:
             return 0
         drops = self.last_sample.listen_drops - self.first_sample.listen_drops
@@ -189,10 +164,9 @@ class AcceptQueueProbe:
     Args:
         port: Listening port to watch.
         owner_id: Identifier used in log lines (the storage unit id).
-        interval_s: Seconds between samples. Keep well under a second; the queue
-            drains in milliseconds and a slower cadence misses the burst entirely.
-        warn_utilization: Log a warning the first time depth reaches this fraction
-            of the backlog, so a near-miss is visible before any drop happens.
+        interval_s: Seconds between samples. Keep sub-second; the queue drains in
+            milliseconds, so a slower cadence misses the burst entirely.
+        warn_utilization: Warn once when depth first reaches this fraction of the backlog.
     """
 
     def __init__(
@@ -246,11 +220,9 @@ def _record(self, sample: AcceptQueueSample) -> None:
         if sample.recv_q > stats.peak_recv_q:
             stats.peak_recv_q = sample.recv_q
             stats.peak_utilization = sample.utilization
-            stats.peak_history.append((sample.timestamp, sample.recv_q))
 
-        # sk_drops is a monotonic kernel counter, so this compares against the previous
-        # sample rather than the first: measuring from probe start would keep reporting a
-        # drop that happened once, on every sample, and erase when it actually occurred.
+        # Against the previous sample, not the first: sk_drops is cumulative, so measuring
+        # from probe start would re-report one old drop on every sample.
         if previous is not None and sample.sk_drops > previous.sk_drops:
             logger.error(
                 f"[{self.owner_id}]: listening socket on port {self.port} dropped an incoming "