diff --git a/swarm-controller/src/review_classifier/review.rs b/swarm-controller/src/review_classifier/review.rs
index 9c84556a..f5d549bf 100644
--- a/swarm-controller/src/review_classifier/review.rs
+++ b/swarm-controller/src/review_classifier/review.rs
@@ -442,19 +442,37 @@ pub(crate) fn spawn_queue(state: &AppState) {
tokio::time::sleep(TIMING.pull_wait).await;
};
let queue = Queue(Some(client.clone()));
- consume(
- &client,
- TIMING,
- &reviews,
- Arc::clone(&forge),
+ let handles = Handles {
+ forge: Arc::clone(&forge),
queue,
- forge,
+ poster: forge,
runs,
- )
- .await;
+ };
+ consume(&client, TIMING, &reviews, handles).await;
});
}
+/// What a review reaches outside the queue: the forge it reads the PR and
+/// rules from, the classifier it runs on, the poster for its review, and the
+/// store that records it.
+struct Handles {
+ forge: Arc,
+ queue: Q,
+ poster: Arc
,
+ runs: Runs,
+}
+
+impl ,
- runs: Runs,
+ handles: Handles ,
- runs: Runs,
+ handles: Handles Clone for Handles {
+ fn clone(&self) -> Self {
+ Self {
+ forge: Arc::clone(&self.forge),
+ queue: self.queue.clone(),
+ poster: Arc::clone(&self.poster),
+ runs: self.runs.clone(),
+ }
+ }
+}
+
/// Take requests from [`REQUEST_STREAM`] one at a time, each once a run
/// place is free, and review each in a task of its own. Never returns;
/// dropping it stops every review it started, and the queue delivers their
@@ -467,10 +485,7 @@ async fn consume(
client: &async_nats::Client,
timing: Timing,
reviews: &Arc,
- queue: Q,
- poster: Arc,
) where
S: FileSource + PullSource + Send + Sync + 'static,
Q: RunSink + SettingsSource,
@@ -493,7 +508,7 @@ async fn consume(
tracing::error!(error = %e, "a requested review's task ended early");
}
}
- match queue.settings().await {
+ match handles.queue.settings().await {
Ok(settings) => reviews.resize(settings.max_concurrent_reviews),
Err(e) => tracing::warn!(
error = %format!("{e:#}"),
@@ -520,10 +535,7 @@ async fn consume(
slot,
timing,
Arc::clone(reviews),
- Arc::clone(&forge),
- queue.clone(),
- Arc::clone(&poster),
- runs.clone(),
+ handles.clone(),
));
}
}
@@ -549,19 +561,12 @@ async fn next(
/// Review the request in `message`, holding `slot`, and ack it once its
/// record is written.
-#[expect(
- clippy::too_many_arguments,
- reason = "each is one thing a review reaches"
-)]
async fn take(
message: jetstream::Message,
slot: Slot,
timing: Timing,
reviews: Arc,
- queue: Q,
- poster: Arc,
) where
S: FileSource + PullSource + Send + Sync + 'static,
Q: RunSink + SettingsSource,
@@ -595,7 +600,13 @@ async fn take(
}
return;
};
- let work = review(&*forge, &*forge, &queue, &*poster, &runs, &request, slot);
+ let Handles {
+ forge,
+ queue,
+ poster,
+ runs,
+ } = &handles;
+ let work = review(&**forge, &**forge, queue, &**poster, runs, &request, slot);
in_progress(&message, timing.ack_wait / 3, work).await;
drop(claim);
if let Err(e) = message.ack().await {
@@ -1875,16 +1886,13 @@ mod tests {
let poster = Arc::clone(&self.rig.poster);
let store = self.rig.store.clone();
tokio::spawn(async move {
- consume(
- &client,
- TEST_TIMING,
- &reviews,
- Arc::clone(&forge),
- gate,
+ let handles = Handles {
+ forge,
+ queue: gate,
poster,
- store,
- )
- .await;
+ runs: store,
+ };
+ consume(&client, TEST_TIMING, &reviews, handles).await;
})
}