Move the fleet, session and tenant tests beside their owners; state the fleet loop's, the session's and the postings cache's remaining contracts; remove 88 inherited warnings

docs/quality/owners.json

@@ -1815,6 +1815,86 @@
       "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/fleet/tests.rs",
+      "reason": "Fleet test module imports its enclosing production owner; the compiler resolves the parent exports and the fixtures pin peer choice, the return-home budget and victim selection.",
+      "syntax": "compiler-resolved import; syntax cannot infer exports"
+    },
+    {
+      "category": "unresolved-glob",
+      "count": 1,
+      "owner": "crate",
+      "path": "src/sse/session/tests.rs",
+      "reason": "SSE session test module imports its enclosing production owner; the compiler resolves the parent exports and the fixtures pin cursor and segment naming.",
+      "syntax": "compiler-resolved import; syntax cannot infer exports"
+    },
+    {
+      "category": "unresolved-glob",
+      "count": 1,
+      "owner": "crate",
+      "path": "src/tenant/tests.rs",
+      "reason": "Tenant test module imports its enclosing production owner; the compiler resolves the parent exports and the fixtures pin the identity grammar.",
+      "syntax": "compiler-resolved import; syntax cannot infer exports"
+    },
+    {
+      "category": "by-path-module",
+      "count": 1,
+      "owner": "crate::tests",
+      "path": "src/tenant.rs",
+      "reason": "Tenant grammar tests; declared by path because the fuzz and invariant harness crates compile the tenant module by path; no copied grammar or public test API.",
+      "syntax": "path = \"tenant/tests.rs\""
+    },
+    {
+      "category": "effect",
+      "count": 1,
+      "owner": "crate::serve",
+      "path": "src/sse/session.rs",
+      "reason": "The SSE driver task is owned by the subscription it drives and ends with the session; a supervised driver would tie a request-scoped task to the runtime supervisor.",
+      "syntax": "tokio::spawn"
     }
   ]
 }

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,
@@ -1729,20 +1673,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,
-      "owner": "crate::serve",
-      "path": "src/sse/session.rs",
-      "syntax": "allow (clippy :: too_many_arguments)"
-    },
     {
       "category": "exception",
       "count": 1,
@@ -1988,13 +1918,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,
@@ -3773,13 +3696,6 @@
       "path": "src/deployment_bearer.rs",
       "syntax": "compiler-resolved import; syntax cannot infer exports"
     },
-    {
-      "category": "unresolved-glob",
-      "count": 1,
-      "owner": "crate::tests",
-      "path": "src/fleet.rs",
-      "syntax": "compiler-resolved import; syntax cannot infer exports"
-    },
     {
       "category": "unresolved-glob",
       "count": 1,
@@ -4151,13 +4067,6 @@
       "path": "src/sse/service.rs",
       "syntax": "compiler-resolved import; syntax cannot infer exports"
     },
-    {
-      "category": "unresolved-glob",
-      "count": 1,
-      "owner": "crate::tests",
-      "path": "src/sse/session.rs",
-      "syntax": "compiler-resolved import; syntax cannot infer exports"
-    },
     {
       "category": "unresolved-glob",
       "count": 1,
@@ -4186,13 +4095,6 @@
       "path": "src/telemetry_batch.rs",
       "syntax": "compiler-resolved import; syntax cannot infer exports"
     },
-    {
-      "category": "unresolved-glob",
-      "count": 1,
-      "owner": "crate::tests",
-      "path": "src/tenant.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 fleet sse_session; do
   case "$owner" in
     postings_codec) file=src/postings.rs; filter=postings:: ;;
     postings) file=src/postings/validated.rs; filter=postings:: ;;
@@ -41,6 +41,13 @@ 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:: ;;
+    fleet) file=src/fleet.rs; filter=fleet:: ;;
+    sse_session) file=src/sse/session.rs; filter="sse:: dst_tests::sse_delivery:: dst_tests::livefeed_swap::" ;;
+    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 +62,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" == fleet || "$owner" == sse_session ]]; 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

src/application/consumer.rs

@@ -154,6 +154,10 @@ pub(crate) struct AuthorizedConsumerContext {
     record: crate::queue::ConsumerRecord,
 }
 impl ConsumerService {
+    #[expect(
+        clippy::too_many_arguments,
+        reason = "ConsumerService::authorize; authorization takes the tenant, stream, consumer and credential parts separately as the handler parsed them; a request struct would exist only for this signature"
+    )]
     pub(crate) async fn authorize(
         self: &Arc<Self>,
         sref: &crate::tenant::TenantStreamRef,
@@ -439,6 +443,10 @@ async fn load_consumer_record(
     }
 }
 
+#[expect(
+    clippy::too_many_lines,
+    reason = "put; the consumer put validates the configuration, decides create-versus-update and publishes the record in one conditional-write sequence; splitting it would separate the validation from the write it guards"
+)]
 pub(crate) async fn put(
     context: AuthorizedStreamContext,
     doc: ConfigInput,
@@ -647,6 +655,10 @@ pub(crate) struct QueuePosition {
 }
 /// Internal callers provide the incarnation-bound target only after fleet authorization.
 impl ConsumerService {
+    #[expect(
+        clippy::too_many_arguments,
+        reason = "ConsumerService::sweep_local; the local sweep takes the descriptor, segment, consumer and generation parts separately as the sweep resolved them; a sweep struct would exist only for this signature"
+    )]
     pub(crate) async fn sweep_local(
         &self,
         target: InternalTarget,
@@ -698,6 +710,10 @@ impl ConsumerService {
             }
         }
     }
+    #[expect(
+        clippy::unwrap_used,
+        reason = "ConsumerService::queue_position; a poisoned stream state may hold a half-advanced durable frontier; recovering it could report a queue position past a length never made durable"
+    )]
     pub(crate) async fn queue_position(
         &self,
         target: InternalTarget,

src/application/consumer/deletion.rs

@@ -9,6 +9,14 @@ use crate::application::consumer_remote::relay_sweep_segment;
 use crate::application::read_remote::InternalTarget;
 use std::sync::Arc;
 
+#[expect(
+    clippy::too_many_arguments,
+    reason = "delete; consumer deletion takes the tenant, stream, consumer, generation and fencing parts separately as the handler resolved them; a delete struct would exist only for this signature"
+)]
+#[expect(
+    clippy::too_many_lines,
+    reason = "delete; the fence, the tombstone and the per-segment cleanup are one deletion whose resumption depends on which step landed; splitting them would separate the steps from the resumption they order"
+)]
 pub(crate) async fn delete(
     state: Arc<ConsumerService>,
     sref: crate::tenant::TenantStreamRef,
@@ -160,6 +168,10 @@ pub(crate) async fn delete(
     })
 }
 
+#[expect(
+    clippy::too_many_lines,
+    reason = "resume_deletion; resumption replays the per-segment cleanup a fenced deletion left behind in the order the debt records; splitting it would separate the steps from the debt that orders them"
+)]
 async fn resume_deletion(
     context: AuthorizedDeletionContext,
 ) -> Result<DeleteOutcome, ConsumerFailure> {