fix: count HTTP refusals and preserve replaced routes at the final sequence

quest/m1/README.md

@@ -135,7 +135,6 @@ above them.
 - [A/V clock](/quest/m1/av-clock.md) - the audio playhead drives Sync.reference while audio plays, through per-track sync handles
 - [Plan: watch worker](/quest/m1/plan-watch-worker.md) - prototype an invisible page worker against app-spawned workers, and land the jank harness that decides
 - [Watch worker](/quest/m1/watch-worker.md) - watch playback runs in a worker onto an OffscreenCanvas, so main-thread jank never stalls video or audio
-- [Rust papercuts](/quest/m1/papercuts-rs.md) - HTTP refusals are counted, and a `u64::MAX` resume is unbounded
 - [Duration parser defaults](/quest/m1/cli-duration-defaults.md) - a moq-tokio duration set in code beats the parser default on a parsed or merged config, zero included, after #5350
 - [Publish channel count](/quest/m1/publish-audio-channel-count.md) - forcing a channel count on an Audio.Capture stops costing the subscriber gaps of silence, and the media fixture's cold-start audio-tone flake is fixed at its cause
 - [JS request deadline](/quest/m1/js-request-deadline.md) - each JS request, PUBLISH_NAMESPACE included, has one fatal 10 s timer from create to answer and fails fast without stream credit, so nothing queues or retries

quest/m2/http-fetch-stats.md

@@ -42,8 +42,6 @@ older reader ignores and a newer one defaults to zero.
 
 ## Related
 
-- [Rust papercuts](/quest/m1/papercuts-rs.md) - counts `/fetch` refusals in
-  `/metrics`; this counts what an admitted fetch serves
 - [Bitrate claim](/quest/m2/rate-claim.md) - HTTP refuses capped tokens, so
   stats are HTTP's only byte account
 - [Session outcomes](/quest/m1/session-outcomes.md) - the session counters

rs/moq-net/src/model/resume.rs

@@ -267,13 +267,19 @@ impl Copy {
 			start: max_some(reader.start, self.floor),
 			..reader.clone()
 		};
-		if let Some(until) = self.until {
-			let end = until.and_then(Position::after_group).unwrap_or_default();
+		if let Some(end) = self.end().map(Position::group) {
 			wire.end = Some(wire.end.map_or(end, |requested| requested.min(end)));
 		}
 		wire
 	}
 
+	/// The first group past where this copy was replaced, if any: one replaced before
+	/// any group ends at 0, and one replaced at `u64::MAX` has no end.
+	fn end(&self) -> Option<u64> {
+		self.until
+			.and_then(|last| last.map_or(Some(0), |last| last.checked_add(1)))
+	}
+
 	fn update(&mut self, reader: &Subscription) {
 		let wire = self.wire(reader);
 		let _ = match &mut self.sub {
@@ -513,7 +519,7 @@ impl Reader {
 			if copy.done.is_some() {
 				return false;
 			}
-			let until = copy.until;
+			let end = copy.end();
 			let Poll::Ready(Ok(sub)) = copy.poll_ready(groups, &kio::Waiter::noop()) else {
 				return false;
 			};
@@ -522,7 +528,7 @@ impl Reader {
 			// cap here, since unread groups become eligible again when it rises.
 			sub.has_unread_group(&track::Unread {
 				start: groups.0,
-				end: until.and_then(|until| until.map_or(Some(0), |last| last.checked_add(1))),
+				end,
 				delivered,
 			})
 		});
@@ -1546,6 +1552,27 @@ mod test {
 		assert_eq!(recv(&mut sub).sequence, 6);
 	}
 
+	#[test]
+	fn a_replaced_copy_at_the_last_sequence_has_no_overflow_cap() {
+		for (last, end) in [(Some(u64::MAX), None), (None, Some(Position::default()))] {
+			let routes = Producer::new();
+			let a = copy();
+			routes.serve(a.consume());
+			let subscription = kio::Producer::new(Subscription::default());
+			let sub = routes.consume().subscribe(subscription);
+			let mut reader = sub.reader.lock().unwrap();
+			let old = &mut reader.copies[0];
+			old.until = Some(last);
+			assert_eq!(old.wire(&Subscription::default()).end, end);
+			// A reader's narrower bound survives an unbounded replacement cap.
+			let requested = Subscription {
+				end: Some(Position::group(7)),
+				..Default::default()
+			};
+			assert_eq!(old.wire(&requested).end, Some(end.unwrap_or(Position::group(7))));
+		}
+	}
+
 	#[test]
 	#[cfg(feature = "async")]
 	fn a_replaced_route_finishes_only_the_groups_it_had() {

rs/moq-relay/src/web.rs

@@ -935,7 +935,11 @@ async fn admit_http(
 	request.server_name = request_host(uri, headers);
 	request.remote = Some(remote.0);
 	request.tls = mtls.and_then(|Extension(MtlsPeer(identity))| auth::peer(&identity));
-	state.auth.admit(request).await
+	state
+		.auth
+		.admit(request)
+		.await
+		.inspect_err(|err| state.cluster.refusals.record(err.into()))
 }
 
 /// Serve the announced broadcasts for a given prefix.
@@ -957,6 +961,7 @@ async fn serve_announced(
 
 	let lease = admit_http(&state, prefix, query, &uri, &headers, remote, mtls).await?;
 	let Some(origin) = state.cluster.subscriber(lease.token()) else {
+		state.cluster.refusals.record(crate::refusals::Refusal::Forbidden);
 		return Err(StatusCode::UNAUTHORIZED.into());
 	};
 
@@ -1006,6 +1011,7 @@ async fn serve_fetch(
 	let broadcast = lease.token().root.to_string();
 
 	let Some(origin) = state.cluster.subscriber(lease.token()) else {
+		state.cluster.refusals.record(crate::refusals::Refusal::Forbidden);
 		return Err(StatusCode::UNAUTHORIZED.into());
 	};
 
@@ -1126,6 +1132,75 @@ mod tests {
 	use std::io::Write;
 	use tempfile::TempDir;
 
+	#[tokio::test(start_paused = true)]
+	async fn http_refusals_count_at_both_endpoints() {
+		use crate::refusals::Refusal;
+		let dir = TempDir::new().unwrap();
+		let (_, cert, _) = make_certs(&dir);
+		let certificates = moq_tokio::tls::Certificates::from_pem(&std::fs::read(cert).unwrap()).unwrap();
+		for fetch in [false, true] {
+			for publish_only in [false, true] {
+				let auth = if publish_only {
+					crate::auth::Config {
+						public_publish: vec![moq_auth::Pattern::all()],
+						..Default::default()
+					}
+					.init("test", &moq_tokio::tls::Connect::default(), false)
+					.unwrap()
+				} else {
+					auth::Auth::refuse("test")
+				};
+				let cluster = cluster::Cluster::new(crate::cluster::Options::default()).unwrap();
+				let web = Web::new(auth, cluster, certificates.clone(), Config::default());
+				let state = web.state;
+				let remote = ConnectInfo(crate::listener::Peer("127.0.0.1:1234".parse().unwrap()));
+				let status = if fetch {
+					serve_fetch(
+						Path("room/video".to_owned()),
+						Query(FetchParams {
+							auth: AuthQuery { jwt: None },
+							group: FetchGroup::Latest,
+						}),
+						None,
+						remote,
+						"/fetch/room/video".parse().unwrap(),
+						http::HeaderMap::new(),
+						State(state.clone()),
+					)
+					.await
+					.map(|_| ())
+					.into_response()
+					.status()
+				} else {
+					serve_announced(
+						Some(Path("room".to_owned())),
+						Query(AuthQuery { jwt: None }),
+						None,
+						remote,
+						"/announced/room".parse().unwrap(),
+						http::HeaderMap::new(),
+						State(state.clone()),
+					)
+					.await
+					.map(|_| ())
+					.into_response()
+					.status()
+				};
+				assert_eq!(status, StatusCode::UNAUTHORIZED);
+				let reason = if publish_only {
+					Refusal::Forbidden
+				} else {
+					Refusal::Refused
+				};
+				assert_eq!(
+					state.cluster.refusals.count(reason),
+					1,
+					"fetch={fetch}, publish_only={publish_only}"
+				);
+			}
+		}
+	}
+
 	fn make_certs(dir: &TempDir) -> (PathBuf, PathBuf, PathBuf) {
 		make_named_certs(dir, "server", "localhost")
 	}