ruststream 0.7.5

//! Integration tests for the unified failure policy: handler panics and decode failures, settled
//! by the per-subscriber `on_failure(panic = .., decode = ..)` policy. Driven over `MemoryBroker`.
#![cfg(all(
    feature = "macros",
    feature = "memory",
    feature = "json",
    feature = "testing"
))]

mod common;

use common::{Order, Wire};

use ruststream::memory::prelude::*;
use std::future::Future;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};

use ruststream::runtime::{BlanketLayer, Handler, RustStreamError};
use ruststream::testing::{Outcome, TestApp};
use serde::{Deserialize, Serialize};

/// Default policy: a panic fails fast. Used by `handler_panic_fails_fast_and_run_returns_err`.
#[subscriber("boom")]
async fn boom(order: &Order) -> HandlerOutcome {
    // The test publishes ids other than u32::MAX, so this assertion always fails (panics); the
    // trailing expression keeps the body typed as HandlerOutcome.
    assert_eq!(order.id, u32::MAX, "handler exploded");
    HandlerOutcome::ack()
}

/// `panic = drop` settles the offending message and keeps consuming. The poison id is 0.
#[subscriber("dropping", on_failure(panic = drop))]
async fn dropping(order: &Order) -> HandlerOutcome {
    assert!(order.id != 0, "poison order must panic");
    HandlerOutcome::ack()
}

/// `decode = fail_fast` tears the service down on a payload that cannot decode.
#[subscriber("decodeff", on_failure(decode = fail_fast))]
async fn decode_ff(_order: &Order) -> HandlerOutcome {
    HandlerOutcome::ack()
}

/// `decode = skip` acks past a payload that cannot decode and keeps consuming.
#[subscriber("skipping", on_failure(decode = skip))]
async fn skipping(_order: &Order) -> HandlerOutcome {
    HandlerOutcome::ack()
}

/// A batch handler under an explicit `panic = fail_fast`.
#[subscriber("batchboom", on_failure(panic = fail_fast))]
async fn batch_boom(orders: &[Order]) -> HandlerOutcome {
    // The test always delivers a non-empty batch, so this assertion always fails (panics).
    assert!(orders.is_empty(), "batch handler exploded");
    HandlerOutcome::ack()
}

/// What the publishing handlers below answer with. The id alone is what the assertions read, so
/// the reply is that one field under a declared name.
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema, Outgoing)]
struct Acked(u32);

/// A publishing handler: exercises the single-message decode-failure path (default `decode = drop`).
#[subscriber("rpcd", reply("rpcd.out"))]
async fn rpcd(order: &Order) -> Acked {
    Acked(order.id)
}

/// A plain batch handler: exercises the per-element batch decode-failure path.
#[subscriber("bd")]
async fn bd(orders: &[Order]) -> HandlerOutcome {
    let _ = orders;
    HandlerOutcome::ack()
}

/// A batch publishing handler: exercises the batch-publishing decode-failure path.
#[subscriber("bpd", reply("bpd.out"))]
async fn bpd(orders: &[Order]) -> Vec<Acked> {
    orders.iter().map(|o| Acked(o.id)).collect()
}

/// Injects a good order, a payload the decoder rejects, and another good order, in that order.
///
/// Every migrated policy test drives the same three deliveries: what differs is how the middle
/// one is settled and whether the service is still there for the third.
async fn drive_good_bad_good<S: Send + Sync + 'static>(tb: &TestApp<S>, topic: &str) {
    tb.message(&Order { id: 1 })
        .to(topic)
        .publish()
        .await
        .expect("publish");
    tb.message(&Wire::of(b"not json"))
        .to(topic)
        .publish()
        .await
        .expect("publish");
    tb.message(&Order { id: 2 })
        .to(topic)
        .publish()
        .await
        .expect("publish");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn handler_panic_fails_fast_and_run_returns_err() {
    let app =
        RustStream::new(AppInfo::new("boom", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(boom);
        });
    let tb = TestApp::start(app).await.expect("startup failed");

    tb.message(&Order { id: 1 })
        .to("boom")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .subscriber("boom")
        .assert_called_once()
        .panicked();
    tb.assert_shut_down();
    let result = tb.shutdown().await;
    assert!(
        matches!(result, Err(RustStreamError::Dispatch(_))),
        "a fail-fast panic must make run() return a dispatch error, got {result:?}",
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn panic_drop_keeps_the_subscriber_consuming() {
    let app =
        RustStream::new(AppInfo::new("dropping", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(dropping);
        });
    let tb = TestApp::start(app).await.expect("startup failed");

    // A good order, then the poison order that panics (dropped), then a good order that must
    // still be processed.
    for id in [7u32, 0, 9] {
        tb.message(&Order { id })
            .to("dropping")
            .publish()
            .await
            .expect("publish");
    }

    assert_eq!(
        tb.broker::<MemoryBroker>()
            .subscriber("dropping")
            .outcomes(),
        [Outcome::Ack, Outcome::Panicked, Outcome::Ack],
    );
    tb.assert_running();
    let result = tb.shutdown().await;
    assert!(
        result.is_ok(),
        "a dropped panic must not error the run: {result:?}"
    );
}

/// `panic = drop` with no panic in the body: the second delivery panics in the middleware below.
#[subscriber("guarded", on_failure(panic = drop))]
async fn guarded(_order: &Order) -> HandlerOutcome {
    HandlerOutcome::ack()
}

/// A middleware that panics on its second call while building the future, before any of it is
/// polled: a panic the policy has to catch as surely as one inside the handler's body.
#[derive(Clone)]
struct EagerGuard(Arc<AtomicU32>);

struct EagerGuardHandler<H> {
    inner: H,
    calls: Arc<AtomicU32>,
}

impl<M, C, S, H> Handler<M, C, S> for EagerGuardHandler<H>
where
    M: Sync,
    C: Send,
    S: Send + Sync,
    H: Handler<M, C, S>,
{
    fn handle(
        &self,
        msg: &M,
        ctx: &mut Context<'_, C, S>,
    ) -> impl Future<Output = HandlerOutcome> + Send {
        let call = self.calls.fetch_add(1, Ordering::SeqCst);
        assert!(call != 1, "the guard refused the second delivery");
        self.inner.handle(msg, ctx)
    }
}

impl BlanketLayer for EagerGuard {
    fn apply<M, C, S, H>(&self, handler: H) -> impl Handler<M, C, S> + 'static
    where
        M: Send + Sync + 'static,
        C: Send + 'static,
        S: Send + Sync + 'static,
        H: Handler<M, C, S> + 'static,
    {
        EagerGuardHandler {
            inner: handler,
            calls: Arc::clone(&self.0),
        }
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_panic_before_the_handler_future_exists_is_settled_by_the_policy() {
    let app = RustStream::new(AppInfo::new("guarded", "0.1.0"))
        .layer(EagerGuard(Arc::new(AtomicU32::new(0))))
        .with_broker(MemoryBroker::new(), |b| {
            b.include(guarded);
        });
    let tb = TestApp::start(app).await.expect("startup failed");

    for id in [7u32, 8, 9] {
        tb.message(&Order { id })
            .to("guarded")
            .publish()
            .await
            .expect("publish");
    }

    assert_eq!(
        tb.broker::<MemoryBroker>().subscriber("guarded").outcomes(),
        [Outcome::Ack, Outcome::Panicked, Outcome::Ack],
    );
    tb.assert_running();
    tb.shutdown()
        .await
        .expect("a dropped panic must not error the run");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn decode_fail_fast_returns_err() {
    let app =
        RustStream::new(AppInfo::new("decodeff", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(decode_ff);
        });
    let tb = TestApp::start(app).await.expect("startup failed");

    tb.message(&Wire::of(b"not json"))
        .to("decodeff")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .subscriber("decodeff")
        .assert_called_once()
        .assert_last_failed_to_decode();
    tb.assert_shut_down();
    let result = tb.shutdown().await;
    assert!(
        matches!(result, Err(RustStreamError::Dispatch(_))),
        "a fail-fast decode failure must make run() return a dispatch error, got {result:?}",
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn decode_skip_acks_past_bad_input_and_continues() {
    let app =
        RustStream::new(AppInfo::new("skipping", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(skipping);
        });
    let tb = TestApp::start(app).await.expect("startup failed");

    drive_good_bad_good(&tb, "skipping").await;

    // The malformed payload is acked past rather than requeued, and the good order behind it is
    // still processed.
    assert_eq!(
        tb.broker::<MemoryBroker>()
            .subscriber("skipping")
            .outcomes(),
        [Outcome::Ack, Outcome::DecodeFailed, Outcome::Ack],
    );
    tb.assert_running();
    let result = tb.shutdown().await;
    assert!(
        result.is_ok(),
        "a skipped decode failure must not error the run: {result:?}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn publishing_decode_failure_is_dropped_and_continues() {
    let router = Router::<MemoryBroker>::new()
        .include(rpcd)
        .out(Reply, Publish)
        .build();
    let app = RustStream::new(AppInfo::new("rpcd", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("startup failed");

    drive_good_bad_good(&tb, "rpcd").await;

    assert_eq!(
        tb.broker::<MemoryBroker>().subscriber("rpcd").outcomes(),
        [Outcome::Ack, Outcome::DecodeFailed, Outcome::Ack],
    );
    // The element that never decoded published no reply; the two that did, did.
    assert_eq!(
        tb.broker::<MemoryBroker>()
            .published::<Acked>("rpcd.out")
            .decoded(),
        vec![Acked(1), Acked(2)],
    );
    tb.assert_running();
    let result = tb.shutdown().await;
    assert!(
        result.is_ok(),
        "a dropped decode failure must not error the run: {result:?}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn batch_decode_failure_drops_the_bad_element() {
    let app = RustStream::new(AppInfo::new("bd", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(bd.batch(nonzero!(64)));
    });
    let tb = TestApp::start(app).await.expect("startup failed");

    drive_good_bad_good(&tb, "bd").await;

    // The rejected element is settled by the policy and never becomes part of a batch, so the
    // handler saw exactly the two decodable orders.
    let seen: Vec<Order> = tb.broker::<MemoryBroker>().subscriber("bd").received();
    assert_eq!(seen.iter().map(|o| o.id).collect::<Vec<_>>(), [1, 2]);
    tb.broker::<MemoryBroker>()
        .subscriber("bd")
        .assert_called(2)
        .settled(HandlerOutcome::ack());
    tb.assert_running();
    let result = tb.shutdown().await;
    assert!(
        result.is_ok(),
        "a dropped batch element must not error the run: {result:?}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn batch_publishing_decode_failure_is_dropped() {
    let router = Router::<MemoryBroker>::new()
        .include(bpd.batch(nonzero!(64)))
        .out(Reply, Publish)
        .build();
    let app = RustStream::new(AppInfo::new("bpd", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("startup failed");

    drive_good_bad_good(&tb, "bpd").await;

    let seen: Vec<Order> = tb.broker::<MemoryBroker>().subscriber("bpd").received();
    assert_eq!(seen.iter().map(|o| o.id).collect::<Vec<_>>(), [1, 2]);
    assert_eq!(
        tb.broker::<MemoryBroker>()
            .published::<Acked>("bpd.out")
            .decoded(),
        vec![Acked(1), Acked(2)],
    );
    tb.assert_running();
    let result = tb.shutdown().await;
    assert!(
        result.is_ok(),
        "a dropped batch reply element must not error the run: {result:?}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn batch_handler_panic_fails_fast() {
    let app =
        RustStream::new(AppInfo::new("batchboom", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(batch_boom.batch(nonzero!(64)));
        });
    let tb = TestApp::start(app).await.expect("startup failed");

    tb.message(&Order { id: 1 })
        .to("batchboom")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .subscriber("batchboom")
        .assert_called_once()
        .panicked();
    tb.assert_shut_down();
    let result = tb.shutdown().await;
    assert!(
        matches!(result, Err(RustStreamError::Dispatch(_))),
        "a fail-fast batch panic must make run() return a dispatch error, got {result:?}",
    );
}

/// A frame that refuses an empty payload, so one element of a batch can fail to construct.
struct Strict<'a>(&'a [u8]);

impl Deserialized for Strict<'_> {
    type Output<'a> = Strict<'a>;
    type Error = &'static str;

    fn from_payload(payload: &[u8]) -> Result<Strict<'_>, &'static str> {
        if payload.is_empty() {
            return Err("an empty frame");
        }
        Ok(Strict(payload))
    }
}

impl Input for Strict<'_> {
    type Axis = SoloDeserialized<Strict<'static>>;
}

#[subscriber("strict")]
async fn store(frames: &[Strict<'_>]) -> HandlerOutcome {
    let bytes: usize = frames.iter().map(|frame| frame.0.len()).sum();
    tracing::info!(bytes, "stored");
    HandlerOutcome::ack()
}

/// A self-deserializing batch whose element fails to construct still hands the body the rest,
/// and the harness records that call: the frames the body saw, settled by its verdict.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_self_deserializing_batch_is_recorded_around_a_refused_frame() {
    // A retaining bus replays what it holds, so the subscription's first batch carries all three.
    let broker = MemoryBroker::retaining(Retention::Messages(nonzero!(8)));
    let publisher = broker.publisher();
    for frame in [b"first".as_slice(), b"", b"third"] {
        publisher
            .message(&Wire::of(frame))
            .to("strict")
            .publish()
            .await
            .expect("publish");
    }
    let app = RustStream::new(AppInfo::new("strict", "0.1.0")).with_broker(broker, |b| {
        b.include(store.batch(nonzero!(8)).start_at(MemoryPosition::start()));
    });
    let tb = TestApp::start(app).await.expect("startup failed");
    tb.settle().await.expect("the replayed batch settles");

    let handle = tb.broker::<MemoryBroker<Retaining>>();
    let frames = handle.subscriber("strict");
    assert_eq!(
        frames.batches_raw(),
        [[b"first".as_slice(), b"third".as_slice()]],
    );
    frames.settled(HandlerOutcome::ack());
    tb.assert_running();
}

/// A batch that refused every frame never reached the body, so the harness records no call.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_self_deserializing_batch_of_refused_frames_records_no_call() {
    let app =
        RustStream::new(AppInfo::new("strict", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(store.batch(nonzero!(8)));
        });
    let tb = TestApp::start(app).await.expect("startup failed");

    tb.message(&Wire::of(b""))
        .to("strict")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .subscriber("strict")
        .assert_not_called();
    tb.assert_running();
}