refactor(peer): split bulk request dispatch

Move file-chunk and Stream Install branches into focused helpers so the new admission path remains within the workspace's strict Clippy limits without changing request behavior.

Test Plan:
- just clippy
- git diff --check
This commit is contained in:
2026-09-12 12:59:05 +02:00
parent b3174deabf
commit 01d92b1922
2 changed files with 68 additions and 36 deletions
+1 -2
View File
@@ -430,8 +430,7 @@ async fn run_server_body(
if !has_child_capacity(connection_tasks.len(), MAX_ESTABLISHED_CONNECTIONS) {
log::warn!(
"Closing excess peer connection from {remote_addr} at application limit {}",
MAX_ESTABLISHED_CONNECTIONS,
"Closing excess peer connection from {remote_addr} at application limit {MAX_ESTABLISHED_CONNECTIONS}",
);
connection.close(application::Error::UNKNOWN);
continue;
+43 -10
View File
@@ -300,13 +300,43 @@ async fn dispatch_request(
.schedule_hint(StateDomain::CallToPlay, hint, source_ip);
DispatchResult::close(framed_tx)
}
Request::GetGameFileChunk {
request @ Request::GetGameFileChunk { .. } => {
dispatch_file_chunk(ctx, request, framed_tx, stream_shutdown).await
}
Request::StreamInstall {
game_id,
content_id,
} => {
dispatch_stream_install(
ctx,
game_id,
content_id,
framed_tx,
stream_shutdown,
admission,
origin,
)
.await
}
}
}
async fn dispatch_file_chunk(
ctx: &PeerCtx,
request: Request,
framed_tx: ResponseWriter,
stream_shutdown: &CancellationToken,
) -> DispatchResult {
let Request::GetGameFileChunk {
game_id,
content_id,
relative_path,
offset,
length,
} => {
} = request
else {
unreachable!("file-chunk dispatcher received a different request")
};
match handle_file_chunk_request(
ctx,
game_id,
@@ -323,12 +353,17 @@ async fn dispatch_request(
ChunkDispatch::Reset(writer) => DispatchResult::reset(writer),
}
}
Request::StreamInstall {
game_id,
content_id,
} => {
let Some(mut stream_install_admission) = admission.try_acquire_stream_install(origin)
else {
async fn dispatch_stream_install(
ctx: &PeerCtx,
game_id: String,
content_id: lanspread_db::content_manifest::ContentId,
framed_tx: ResponseWriter,
stream_shutdown: &CancellationToken,
admission: &ServerAdmission,
origin: ObservedOrigin,
) -> DispatchResult {
let Some(mut stream_install_admission) = admission.try_acquire_stream_install(origin) else {
let mut tx = framed_tx.into_inner();
let _ = tx.reset(application::Error::UNKNOWN);
return DispatchResult::reset(FramedWrite::new(tx, control_codec()));
@@ -346,8 +381,6 @@ async fn dispatch_request(
drop(stream_install_admission);
DispatchResult::close(writer)
}
}
}
fn reset_response_writer(framed_tx: ResponseWriter, label: &str) -> DispatchResult {
let mut tx = framed_tx.into_inner();