Lines
99.17 %
Functions
100 %
Branches
use super::*;
use rpc::ScriptCtx;
use rpc::ScriptLimits;
use sqlx::types::Uuid;
#[derive(Debug)]
enum Sent {
Data(String),
Finished,
}
struct MpscSink {
tx: mpsc::UnboundedSender<Sent>,
#[async_trait::async_trait]
impl ResponseSink for MpscSink {
async fn send(&self, bytes: Vec<u8>) {
let _ = self.tx.send(Sent::Data(String::from_utf8(bytes).unwrap()));
async fn finish(&self) {
let _ = self.tx.send(Sent::Finished);
fn new_state() -> (EvalChannelState, mpsc::UnboundedReceiver<Sent>) {
state_with_ctx(ScriptCtx::new(Uuid::new_v4()))
fn state_with_ctx(ctx: ScriptCtx) -> (EvalChannelState, mpsc::UnboundedReceiver<Sent>) {
let (tx, rx) = mpsc::unbounded_channel();
let sink = Box::new(MpscSink { tx });
let state = EvalChannelState::with_session(Session::new(ctx).expect("session"), sink);
(state, rx)
/// Collects `expected` responses, awaiting each rather than
/// busy-polling a fixed budget so a slow CI runner never spuriously
/// yields fewer (a genuine hang still fails via the per-response
/// timeout). After the expected count arrives it keeps draining on a
/// short grace window, so an over-emission regression (an extra
/// frame) is collected and trips the caller's count assertion. The
/// state is still alive here, so a finish is always a regression.
async fn drain_count(rx: &mut mpsc::UnboundedReceiver<Sent>, expected: usize) -> Vec<String> {
let mut out = Vec::new();
loop {
let budget = if out.len() < expected {
std::time::Duration::from_secs(10)
} else {
std::time::Duration::from_millis(100)
};
match tokio::time::timeout(budget, rx.recv()).await {
Ok(Some(Sent::Data(s))) => out.push(s),
Ok(Some(Sent::Finished)) => panic!("channel finished while its state is alive"),
Ok(None) | Err(_) => break,
out
/// Everything the worker emits until it exits and drops the sink.
async fn until_worker_exits(rx: &mut mpsc::UnboundedReceiver<Sent>) -> Vec<Sent> {
while let Ok(Some(event)) =
tokio::time::timeout(std::time::Duration::from_secs(10), rx.recv()).await
{
out.push(event);
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn empty_feed_produces_no_responses() {
let (mut state, mut rx) = new_state();
state.feed(b"").unwrap();
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
assert!(rx.try_recv().is_err());
async fn complete_frame_yields_response_with_trailing_newline() {
state
.feed(b"(:id 1 :form (rpc-protocol-version))\n")
.unwrap();
let responses = drain_count(&mut rx, 1).await;
assert_eq!(responses.len(), 1, "got {responses:?}");
let r = &responses[0];
assert!(r.starts_with("(:id 1"), "{r:?}");
assert!(r.ends_with('\n'), "{r:?}");
async fn two_concatenated_frames_yield_two_responses() {
.feed(b"(:id 3 :form (rpc-protocol-version))\n(:id 4 :form (rpc-protocol-version))\n")
let responses = drain_count(&mut rx, 2).await;
assert_eq!(responses.len(), 2);
assert!(responses[0].contains(":id 3"));
assert!(responses[1].contains(":id 4"));
async fn dropping_the_state_answers_queued_forms_then_finishes() {
.feed(b"(:id 1 :form (rpc-protocol-version))\n(:id 2 :form (rpc-protocol-version))\n")
drop(state);
let events = until_worker_exits(&mut rx).await;
assert!(
matches!(
&events[..],
[Sent::Data(first), Sent::Data(second), Sent::Finished]
if first.starts_with("(:id 1") && second.starts_with("(:id 2")
),
"{events:?}"
);
async fn dropping_the_state_mid_frame_only_finishes() {
state.feed(b"(:id 1 :form (rpc-protocol").unwrap();
assert!(matches!(&events[..], [Sent::Finished]), "{events:?}");
async fn etx_byte_in_stream_is_stripped() {
// Send `(form)<ETX>` — the ETX must NOT pollute the decoder
// (which would refuse the byte as non-paren content) and must
// NOT prevent the frame from completing.
let mut bytes = b"(:id 5 :form (rpc-protocol-version))\n".to_vec();
bytes.push(ETX);
state.feed(&bytes).unwrap();
assert!(responses[0].contains(":id 5"));
async fn etx_byte_alone_bumps_epoch_without_decoding_a_frame() {
// Just an ETX with no surrounding form should not produce
// any output — the bumper fires, but there's nothing to
// cancel since no eval is in flight.
state.feed(&[ETX]).unwrap();
async fn etx_byte_after_form_cancels_inflight_eval() {
// End-to-end cancel scenario through the eval-channel layer:
// feed a long-running form, then feed an ETX byte on the same
// channel. `feed` both bumps the epoch and advances the
// interrupt generation, so the cancel is observed whether the
// eval is already inside Wasm (epoch trap) or still in the
// worker-pickup / compile window (pre-start check_interrupt).
//
// Fuel is set effectively unbounded and the loop runs billions
// of iterations (the bound is just under i32::MAX, since the
// loop counter is an Index / i32), so the form can NEVER
// terminate on its own (neither a natural value nor a
// fuel-exhaustion `:code runtime`) within the test window.
// `:code interrupted` is therefore the only possible outcome —
// the assertion pins exactly that, which is what makes this a
// real regression guard for the cancel path rather than a test
// the fuel cap can satisfy on its own.
let ctx = ScriptCtx::new(Uuid::new_v4()).with_limits(ScriptLimits {
fuel: u64::MAX,
..ScriptLimits::default()
});
let (mut state, mut rx) = state_with_ctx(ctx);
.feed(b"(:id 9 :form (do ((i 0 (+ i 1))) ((>= i 2000000000) i)))\n")
// Yield so the worker picks the form up and starts running.
assert!(r.contains(":id 9"), "{r:?}");
assert!(r.contains(":code interrupted"), "{r:?}");
#[tokio::test(flavor = "current_thread")]
async fn feed_rejects_non_utf8() {
let (mut state, _rx) = new_state();
let invalid: [u8; 2] = [0xC0, 0xC1];
assert!(state.feed(&invalid).is_err());