Function collect_bounded_response

nodedb/src/control/server/dispatch_utils.rs:25–65

Drain a dispatched request's bounded response channel, enforcing a total-payload byte ceiling across streamed partials. Returns the final Response (non-streaming: pass-through; streaming: concatenated payload) or an error if the channel closed without a final chunk or if the accumulated payload would exceed the ceiling.

(
    rx: &mut tokio::sync::mpsc::Receiver<Response>,
    max_result_bytes: usize,
)

/// concatenated payload) or an error if the channel closed without a
/// final chunk or if the accumulated payload would exceed the ceiling.
pub(crate) async fn collect_bounded_response(
    rx: &mut tokio::sync::mpsc::Receiver<Response>,
    max_result_bytes: usize,
) -> Result<Response, DispatchCollectError> {
    let mut combined_payload: Vec<u8> = Vec::new();
    let mut final_response_meta: Option<Response> = None;
    let mut final_streaming = false;

    loop {
        let Some(resp) = rx.recv().await else { break };
        if resp.partial {
            combined_payload.extend_from_slice(&resp.payload);
            if combined_payload.len() > max_result_bytes {
                return Err(DispatchCollectError::OverBudget {
                    bytes: combined_payload.len(),
                });
            }
        } else if combined_payload.is_empty() {
            return Ok(resp);
        } else {
            combined_payload.extend_from_slice(&resp.payload);
            if combined_payload.len() > max_result_bytes {
                return Err(DispatchCollectError::OverBudget {
                    bytes: combined_payload.len(),
                });
            }
            final_response_meta = Some(resp);
            final_streaming = true;
            break;
        }
    }

    if final_streaming {
        let meta = final_response_meta.expect("final_streaming ⇒ meta set");
        return Ok(Response {
            payload: Payload::from_vec(combined_payload),
            ..meta
        });
    }
    Err(DispatchCollectError::ChannelClosed)
}

/// Current wall-clock time as milliseconds since Unix epoch.
///