vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

review classifier: group the queue consumer's handles into one struct

Browse source 
take and consume carried the forge, the classifier queue, the review poster and the run store as four separate parameters, which took take to eight and needed a too_many_arguments expectation. They travel together everywhere they are used, so they are now one Handles value, cloned per taken request. take has five parameters and no lint attribute. No behaviour change.
This commit is contained in:
atlas 2026-10-22 06:52:35 +02:00
commit efbcc3dce6
1 changed files with 43 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

78
swarm-controller/src/review_classifier/review.rs
View file

@ -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<S, Q, P> {
forge: Arc<S>,
queue: Q,
poster: Arc<P>,
runs: Runs,
}
impl<S, Q: Clone, P> Clone for Handles<S, Q, P> {
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<S, Q, P>(
client: &async_nats::Client,
timing: Timing,
reviews: &Arc<Reviews>,
forge: Arc<S>,
queue: Q,
poster: Arc<P>,
runs: Runs,
handles: Handles<S, Q, P>,
) where
S: FileSource + PullSource + Send + Sync + 'static,
Q: RunSink + SettingsSource,
@ -493,7 +508,7 @@ async fn consume<S, Q, P>(
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<S, Q, P>(
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<S, Q, P>(
message: jetstream::Message,
slot: Slot,
timing: Timing,
reviews: Arc<Reviews>,
forge: Arc<S>,
queue: Q,
poster: Arc<P>,
runs: Runs,
handles: Handles<S, Q, P>,
) where
S: FileSource + PullSource + Send + Sync + 'static,
Q: RunSink + SettingsSource,
@ -595,7 +600,13 @@ async fn take<S, Q, P>(
}
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;
})
}