Move the HTTP, shard and billing test support beside their owners; state the three largest owners' remaining contracts; remove 72 inherited warnings

docs/quality/owners.json

@@ -1815,6 +1815,126 @@
       "path": "src/postings.rs",
       "reason": "Postings codec and span planner tests; compile beside the same private codec in the server, fuzz and invariant harness crates so the owner stays within the file-size limit; no copied codec or public test API.",
       "syntax": "path = \"postings/tests.rs\""
+    },
+    {
+      "category": "effect",
+      "count": 1,
+      "owner": "crate::TouchJournal::start",
+      "path": "src/touch.rs",
+      "reason": "Per-stream touch flusher started under the journal map lock and ended by the journal's own close; no supervisor is reachable at creation and the task owns nothing beyond its journal.",
+      "syntax": "tokio::spawn"
+    },
+    {
+      "category": "effect",
+      "count": 1,
+      "owner": "crate::TaskSupervisor::launch_shutdown",
+      "path": "src/tasks/shutdown.rs",
+      "reason": "The one shutdown driver: it joins the supervised set and publishes the report every observer waits on, so it cannot itself be supervised.",
+      "syntax": "tokio::spawn"
+    },
+    {
+      "category": "effect",
+      "count": 1,
+      "owner": "crate::main",
+      "path": "src/main.rs",
+      "reason": "Process entry reads the Tokio worker floor once before the runtime exists; no service configuration or credentials are inferred.",
+      "syntax": "std::env::var"
+    },
+    {
+      "category": "global",
+      "count": 1,
+      "owner": "crate::reg::M",
+      "path": "src/failpoints.rs",
+      "reason": "The DST failpoint registry: one process-wide lazily initialised map keyed by point and stream name, armed and read only by the DST program; a per-runtime copy would let a parked point outlive the runtime that armed it.",
+      "syntax": "std :: sync :: OnceLock < Registry >"
+    },
+    {
+      "category": "effect",
+      "count": 1,
+      "owner": "crate::reset_refuses_while_in_flight",
+      "path": "src/dst/trace_store_tests/streams.rs",
+      "reason": "Trace-store regression leaks one abandoned list stream on purpose so its lifetime guard never runs and reset must refuse; nothing else escapes the test.",
+      "syntax": "std::mem::forget"
+    },
+    {
+      "category": "unresolved-glob",
+      "count": 1,
+      "owner": "crate",
+      "path": "src/http/test_support.rs",
+      "reason": "HTTP fixture adapters import their enclosing production owner; the compiler resolves the parent exports and the adapters only forward scenario calls.",
+      "syntax": "compiler-resolved import; syntax cannot infer exports"
+    },
+    {
+      "category": "unresolved-glob",
+      "count": 1,
+      "owner": "crate",
+      "path": "src/shard/test_support.rs",
+      "reason": "Shard test helpers import their enclosing production owner; the compiler resolves the parent exports and the helpers only encode fixtures and arm scan faults.",
+      "syntax": "compiler-resolved import; syntax cannot infer exports"
+    },
+    {
+      "category": "global",
+      "count": 1,
+      "owner": "crate::dirty_scan_faults::M",
+      "path": "src/shard/test_support.rs",
+      "reason": "Test-only fault injection for the durable dirty-index scan, keyed by shard prefix so concurrent tests cannot poison each other; nothing outside cfg(test) reads it.",
+      "syntax": "std :: sync :: OnceLock < Mutex < HashMap < String , u32 > > >"
+    },
+    {
+      "category": "global",
+      "count": 1,
+      "owner": "crate::BILLING_CLOCK_OVERRIDE",
+      "path": "src/billing/test_clock.rs",
+      "reason": "Test-only billing clock override the month-sensitive tests inject; nothing outside cfg(test) reads it.",
+      "syntax": "std :: sync :: atomic :: AtomicI64"
+    },
+    {
+      "category": "global",
+      "count": 1,
+      "owner": "crate::billing_clock_lock::L",
+      "path": "src/billing/test_clock.rs",
+      "reason": "Test-only lock serialising the month-sensitive tests around the injected clock; nothing outside cfg(test) reads it.",
+      "syntax": "std :: sync :: OnceLock < tokio :: sync :: RwLock < () > >"
+    },
+    {
+      "category": "effect",
+      "count": 1,
+      "owner": "crate::serve_h1",
+      "path": "src/http.rs",
+      "reason": "Each accepted HTTP/1 connection is served by a task the listener's own JoinSet owns and joins at shutdown.",
+      "syntax": "tokio::task::JoinSet::spawn"
+    },
+    {
+      "category": "effect",
+      "count": 1,
+      "owner": "crate::router",
+      "path": "src/http.rs",
+      "reason": "The debug abort route spawns one bare task that sleeps briefly and aborts the process; nothing outlives it.",
+      "syntax": "tokio::spawn"
+    },
+    {
+      "category": "by-path-module",
+      "count": 1,
+      "owner": "crate::test_support",
+      "path": "src/http.rs",
+      "reason": "HTTP fixture adapters; declared by path like the surface's read adapter so the file name says what it holds; no copied surface or public test API.",
+      "syntax": "path = \"http/test_support.rs\""
+    },
+    {
+      "category": "by-path-module",
+      "count": 1,
+      "owner": "crate::tests",
+      "path": "src/http.rs",
+      "reason": "HTTP surface unit tests; declared by path like the surface's read adapter; no copied surface or public test API.",
+      "syntax": "path = \"http/tests.rs\""
+    },
+    {
+      "category": "unresolved-glob",
+      "count": 1,
+      "owner": "crate",
+      "path": "src/http/tests.rs",
+      "reason": "HTTP surface test module imports its enclosing production owner; the compiler resolves the parent exports and the fixtures pin the interval cursor.",
+      "syntax": "compiler-resolved import; syntax cannot infer exports"
     }
   ]
 }

docs/quality/source-allowances.json

@@ -1554,62 +1554,6 @@
       "path": "bench/awsbench/src/main.rs",
       "syntax": "allow (unreachable_code)"
     },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate::dlq_and_settle",
-      "path": "src/application/consumer/delivery.rs",
-      "syntax": "allow (clippy :: too_many_arguments)"
-    },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate::ReadPlan < 'a >::segment",
-      "path": "src/application/read.rs",
-      "syntax": "allow (clippy :: too_many_arguments)"
-    },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate::TEST_ASSERT_KEYED_DENSE",
-      "path": "src/application/read.rs",
-      "syntax": "allow (clippy :: too_many_arguments)"
-    },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate::read_merged",
-      "path": "src/application/read.rs",
-      "syntax": "allow (clippy :: too_many_arguments)"
-    },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate::read_stitched",
-      "path": "src/application/read.rs",
-      "syntax": "allow (clippy :: too_many_arguments)"
-    },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate::WALK_CLOSE_SUBMITS_SNAPSHOT",
-      "path": "src/dst/tests/runtime_sweep.rs",
-      "syntax": "allow (non_snake_case)"
-    },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate::TraceState::event",
-      "path": "src/dst/trace_store.rs",
-      "syntax": "allow (clippy :: too_many_arguments)"
-    },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate::read",
-      "path": "src/history/canonical_span.rs",
-      "syntax": "allow (clippy :: too_many_arguments)"
-    },
     {
       "category": "exception",
       "count": 1,
@@ -1624,27 +1568,6 @@
       "path": "src/http.rs",
       "syntax": "allow (clippy :: result_large_err , reason = \"transport boundary returns Axum wire response directly; application errors stay compact\")"
     },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate::ReadParams::sig",
-      "path": "src/http.rs",
-      "syntax": "allow (dead_code)"
-    },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate::raise_nofile",
-      "path": "src/http.rs",
-      "syntax": "allow (clippy :: too_many_arguments)"
-    },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate::sse_response",
-      "path": "src/http.rs",
-      "syntax": "allow (clippy :: too_many_arguments)"
-    },
     {
       "category": "exception",
       "count": 1,
@@ -1729,13 +1652,6 @@
       "path": "src/product.rs",
       "syntax": "allow (clippy :: result_large_err , reason = \"transport boundary returns Axum wire response directly; application errors stay compact\")"
     },
-    {
-      "category": "exception",
-      "count": 1,
-      "owner": "crate",
-      "path": "src/project_policy.rs",
-      "syntax": "allow (dead_code)"
-    },
     {
       "category": "exception",
       "count": 1,
@@ -1890,13 +1806,6 @@
       "path": "src/billing.rs",
       "syntax": "std :: sync :: atomic :: AtomicU64"
     },
-    {
-      "category": "global",
-      "count": 1,
-      "owner": "crate::BILLING_CLOCK_OVERRIDE",
-      "path": "src/billing.rs",
-      "syntax": "std :: sync :: atomic :: AtomicI64"
-    },
     {
       "category": "global",
       "count": 1,
@@ -1925,13 +1834,6 @@
       "path": "src/billing.rs",
       "syntax": "std :: sync :: atomic :: AtomicU64"
     },
-    {
-      "category": "global",
-      "count": 1,
-      "owner": "crate::billing_clock_lock::L",
-      "path": "src/billing.rs",
-      "syntax": "std :: sync :: OnceLock < tokio :: sync :: RwLock < () > >"
-    },
     {
       "category": "global",
       "count": 1,
@@ -1988,13 +1890,6 @@
       "path": "src/failpoints.rs",
       "syntax": "std :: sync :: OnceLock < tokio :: sync :: Notify >"
     },
-    {
-      "category": "global",
-      "count": 1,
-      "owner": "crate::reg::M",
-      "path": "src/failpoints.rs",
-      "syntax": "std :: sync :: OnceLock < Mutex < HashMap < (Fp , String) , FpState > > >"
-    },
     {
       "category": "global",
       "count": 1,
@@ -2247,13 +2142,6 @@
       "path": "src/shard.rs",
       "syntax": "std :: sync :: OnceLock < Mutex < HashMap < String , () > > >"
     },
-    {
-      "category": "global",
-      "count": 1,
-      "owner": "crate::dirty_scan_faults::M",
-      "path": "src/shard.rs",
-      "syntax": "std :: sync :: OnceLock < Mutex < HashMap < String , u32 > > >"
-    },
     {
       "category": "global",
       "count": 1,
@@ -3864,13 +3752,6 @@
       "path": "src/history/postings_validation_tests.rs",
       "syntax": "compiler-resolved import; syntax cannot infer exports"
     },
-    {
-      "category": "unresolved-glob",
-      "count": 1,
-      "owner": "crate::tests",
-      "path": "src/http.rs",
-      "syntax": "compiler-resolved import; syntax cannot infer exports"
-    },
     {
       "category": "unresolved-glob",
       "count": 1,

docs/refactor/test-inventory.json

@@ -2160,7 +2160,7 @@
     "attributes": [
       "#[tokio::test(flavor = \"current_thread\", start_paused = true)]"
     ],
-    "function_sha256": "ebd8cf2db4bf1a26a03716d8da99e98671cb514d9f8298a69702f50987631166",
+    "function_sha256": "c19966727f38f37212c94748fdbc1737679222c79fa8a2810079c4740890a7c9",
     "mechanisms": [
       "STORE_LATENCY"
     ],
@@ -2811,7 +2811,7 @@
     "attributes": [
       "#[tokio::test(flavor = \"multi_thread\", worker_threads = 4)]"
     ],
-    "function_sha256": "6a83c50969fe771c13aa8f46f7fbbd163ebcf3a12191a9a8c96cd5797ede4a0e",
+    "function_sha256": "b81632e4d007350aa2784b7a0aa34f3358a943f4102aa5d84e63220658d12f78",
     "mechanisms": [],
     "configuration": [],
     "file": "src/dst/tests/livefeed_history.rs",
@@ -4743,18 +4743,6 @@
     "file": "src/dst/trace_store_tests/streams.rs",
     "scenarios": []
   },
-  {
-    "name": "reset_refuses_while_in_flight",
-    "attributes": [
-      "#[tokio::test]",
-      "#[should_panic(expected = \"while operations are active\")]"
-    ],
-    "function_sha256": "8dbca7724f55e7f2f2d9c3a7fdad5701e038a944e2bdd39d259839d152c0ce70",
-    "mechanisms": [],
-    "configuration": [],
-    "file": "src/dst/trace_store_tests/streams.rs",
-    "scenarios": []
-  },
   {
     "name": "restart_invalidates_process_local_touch_cursors",
     "attributes": [
@@ -5535,7 +5523,7 @@
     "attributes": [
       "#[tokio::test(flavor = \"multi_thread\", worker_threads = 4)]"
     ],
-    "function_sha256": "a546f8f58f27f9b678a3b98253c10f7b695e3f7f36ed75f9e2772ae8ee0d0007",
+    "function_sha256": "db24635d68ea5f36ee93262da9a9e3df35767a2f5e3245972145871d0dfecfad",
     "mechanisms": [],
     "configuration": [
       "let store = FaultStore::uniform(inner.clone(), 103, FaultPlan::new(0, 0, 0));"

examples/readertest.rs

@@ -52,6 +52,10 @@ impl slatedb::BlockTransformer for TestTransformer {
     clippy::disallowed_methods,
     reason = "main; the example reads its one path argument from the process environment; routing it through the server's configuration would couple an example to the runtime"
 )]
+#[expect(
+    clippy::excessive_nesting,
+    reason = "main; the example nests the scan loop inside the per-range task it times; flattening it would separate the count from the range it measures"
+)]
 async fn main() -> anyhow::Result<()> {
     let path = std::env::args()
         .nth(1)

scripts/quality/mutations.sh

@@ -16,7 +16,7 @@ BASE=$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["merge_ba
 git diff --no-ext-diff --binary --src-prefix="a/$PREFIX" --dst-prefix="b/$PREFIX" "$BASE" -- > "$QUALITY_MUTANTS_OUT/harness-pr.diff"
 TOTAL=0
 REGISTERED=(src/bin/pilot/benchmark.rs src/bin/pilot/benchmark/config.rs src/bin/pilot/benchmark/window.rs src/bin/pilot/generator.rs src/bin/pilot/generator/membership.rs)
-for owner in postings_codec postings batch retained quota cursors queue rollup_allocation rollup_storage tasks touch read_accumulator read_spool shard_directory history_partition ops scaler postings_cache sharddir crypto tail_ring shard bootstrap read_request http_read queue_cleanup transaction_append record; do
+for owner in postings_codec postings batch retained quota cursors queue rollup_allocation rollup_storage tasks touch read_accumulator read_spool shard_directory history_partition ops scaler postings_cache sharddir crypto tail_ring shard bootstrap read_request http_read queue_cleanup transaction_append record tasks_shutdown runtime runtime_telemetry product_cursor quota_registry http; do
   case "$owner" in
     postings_codec) file=src/postings.rs; filter=postings:: ;;
     postings) file=src/postings/validated.rs; filter=postings:: ;;
@@ -41,6 +41,12 @@ for owner in postings_codec postings batch retained quota cursors queue rollup_a
     queue_cleanup) file=src/shard/transaction/queue/cleanup.rs; filter=shard:: ;;
     transaction_append) file=src/shard/transaction/append.rs; filter=shard:: ;;
     record) file=src/shard/record.rs; filter=shard:: ;;
+    http) file=src/http.rs; filter=http:: ;;
+    tasks_shutdown) file=src/tasks/shutdown.rs; filter=tasks:: ;;
+    runtime) file=src/runtime.rs; filter=runtime:: ;;
+    runtime_telemetry) file=src/runtime/telemetry.rs; filter=runtime:: ;;
+    product_cursor) file=src/product_cursor.rs; filter=product_cursor:: ;;
+    quota_registry) file=src/quota.rs; filter=quota:: ;;
     postings_cache) file=src/postings_cache.rs; filter=postings_cache:: ;;
     sharddir) file=src/sharddir.rs; filter=sharddir:: ;;
     crypto) file=src/crypto.rs; filter=crypto:: ;;
@@ -55,7 +61,7 @@ for owner in postings_codec postings batch retained quota cursors queue rollup_a
   package=streams-quality-invariants
   mutation_file="$PREFIX$file"
   mutation_diff="$QUALITY_MUTANTS_OUT/harness-pr.diff"
-  if [[ "$owner" == tasks || "$owner" == touch || "$owner" == read_accumulator || "$owner" == read_spool || "$owner" == shard_directory || "$owner" == history_partition || "$owner" == ops || "$owner" == scaler || "$owner" == postings_cache || "$owner" == sharddir || "$owner" == crypto || "$owner" == tail_ring || "$owner" == shard || "$owner" == bootstrap || "$owner" == read_request || "$owner" == http_read || "$owner" == queue_cleanup || "$owner" == transaction_append || "$owner" == record ]]; then
+  if [[ "$owner" == tasks || "$owner" == touch || "$owner" == read_accumulator || "$owner" == read_spool || "$owner" == shard_directory || "$owner" == history_partition || "$owner" == ops || "$owner" == scaler || "$owner" == postings_cache || "$owner" == sharddir || "$owner" == crypto || "$owner" == tail_ring || "$owner" == shard || "$owner" == bootstrap || "$owner" == read_request || "$owner" == http_read || "$owner" == queue_cleanup || "$owner" == transaction_append || "$owner" == record || "$owner" == tasks_shutdown || "$owner" == runtime || "$owner" == runtime_telemetry || "$owner" == product_cursor || "$owner" == quota_registry || "$owner" == http ]]; then
     # These owners use actual service clocks, task handles and storage types.
     # Keep their code and tests in the service crate without substitute models.
     package=streams-slate