Skip to content

Handle low queue TryRecvError and fix test message #252

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
leynos merged 2 commits into from
Aug 3, 2025
Merged

Handle low queue TryRecvError and fix test message #252

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b082433
Handle missing TryRecvError and fix test message
leynos Aug 3, 2025
ce04d1a
Apply formatting
leynos Aug 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 1 addition & 4 deletions examples/metadata_routing.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,10 +60,7 @@ impl FrameMetadata for HeaderSerializer {
struct Ping;

#[derive(bincode::Decode, bincode::Encode)]
#[expect(
dead_code,
reason = "placeholder for demonstration of metadata routing"
)]
// Placeholder for demonstration of metadata routing; not used directly.
struct Pong;

#[tokio::main]
Expand Down
26 changes: 17 additions & 9 deletions src/connection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,10 @@ use std::{
};

use futures::StreamExt;
use tokio::{sync::mpsc, time::Duration};
use tokio::{
sync::mpsc::{self, error::TryRecvError},
time::Duration,
};
use tokio_util::sync::CancellationToken;
use tracing::{info, info_span, warn};

Expand Down Expand Up @@ -393,14 +396,19 @@ where
fn after_high(&mut self, out: &mut Vec<F>, state: &mut ActorState) {
Copy link
Copy Markdown
Contributor

@sourcery-ai sourcery-ai Bot Aug 3, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (complexity): Consider restructuring after_high to use early returns and an if-let for clarity instead of mapping over Option.

You can simplify after_high by reverting the .as_mut().map(…) into an if let Some(…) and pulling your should_yield check up front. That way the reader sees “we only try_recv if there’s a queue and we should yield,” and you don’t construct an intermediate Option<Result<…>>. For example:

fn after_high(&mut self, out: &mut Vec<F>, state: &mut ActorState) {
    self.fairness.after_high();

    // only drain when fairness wants us to yield, and only if we have a receiver
    if !self.fairness.should_yield() {
        return;
    }
    if let Some(rx) = &mut self.low_rx {
        match rx.try_recv() {
            Ok(mut frame) => {
                self.hooks.before_send(&mut frame, &mut self.ctx);
                out.push(frame);
                self.after_low();
            }
            Err(TryRecvError::Disconnected) => {
                // handle the closed queue
                Self::handle_closed_receiver(&mut self.low_rx, state);
            }
            Err(TryRecvError::Empty) => {
                // nothing to do
            }
        }
    }
}

This preserves exactly the same behavior but removes the extra map + inner if let, so the control-flow reads straight down.

self.fairness.after_high();

if self.fairness.should_yield()
&& let Some(rx) = &mut self.low_rx
{
match rx.try_recv() {
Ok(mut frame) => {
self.hooks.before_send(&mut frame, &mut self.ctx);
out.push(frame);
self.after_low();
if self.fairness.should_yield() {
let res = self.low_rx.as_mut().map(mpsc::Receiver::try_recv);
if let Some(res) = res {
match res {
Ok(mut frame) => {
self.hooks.before_send(&mut frame, &mut self.ctx);
out.push(frame);
self.after_low();
}
Err(TryRecvError::Empty) => {}
Err(TryRecvError::Disconnected) => {
Self::handle_closed_receiver(&mut self.low_rx, state);
}
}
}
}
Expand Down
13 changes: 5 additions & 8 deletions src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -507,10 +507,10 @@ async fn process_stream<F, T>(
let peer_addr = stream.peer_addr().ok();
match read_preamble::<_, T>(&mut stream).await {
Ok((preamble, leftover)) => {
if let Some(handler) = on_success.as_ref()
&& let Err(e) = handler(&preamble, &mut stream).await
{
tracing::error!(error = ?e, ?peer_addr, "preamble callback error");
if let Some(handler) = on_success.as_ref() {
if let Err(e) = handler(&preamble, &mut stream).await {
tracing::error!(error = ?e, ?peer_addr, "preamble callback error");
}
}
let stream = RewindStream::new(leftover, stream);
// Hand the connection to the application for processing.
Expand Down Expand Up @@ -558,10 +558,7 @@ mod tests {

/// Test helper preamble carrying no data.
#[derive(Debug, Clone, PartialEq, Encode, Decode)]
#[expect(
dead_code,
reason = "used only in doctest to illustrate an empty preamble"
)]
// Used only in doctest to illustrate an empty preamble.
struct EmptyPreamble;

#[fixture]
Expand Down
2 changes: 1 addition & 1 deletion tests/response.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,7 @@ fn custom_length_roundtrip(
#[tokio::test]
async fn send_response_propagates_write_error() {
let app = WireframeApp::new()
.expect("route registration failed")
.expect("app creation failed")
.frame_processor(LengthPrefixedProcessor::default());

let mut writer = FailingWriter;
Expand Down