[fix] Fix to avoid request dropping in the accept queue

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
 

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",

tests/test_accept_probe.py

@@ -0,0 +1,257 @@
+# 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 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
+
+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, 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 if drops is None else drops,
+    )
+
+
+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 "dropped an incoming 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("dropped an incoming 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("dropped an incoming 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()
+
+    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
+
+
+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,
+    )
+
+
+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