wireshift-core 0.1.1

#![allow(missing_docs)]

use std::sync::mpsc::Sender;
use std::sync::Mutex;
use std::time::Duration;

use wireshift_core::backend::{
    Backend, BackendCompletion, BackendFactory, BackendKind, BackendSubmission, BoxedBackend,
    CancellationHandle,
};
use wireshift_core::error::{Error, Result};
use wireshift_core::op::CompletionPayload;
use wireshift_core::ops::Nop;
use wireshift_core::strategy::BatchDrain;
use wireshift_core::{Ring, RingConfig};

/// Backend that emits exactly one valid completion for the first submission and
/// then drops its completion sender, forcing the *next* `complete()` to observe
/// a disconnected channel: a non-timeout fatal backend error mid-batch. This is
/// the exact shape a dead backend thread produces partway through a drain.
struct OneThenDisconnectBackend {
    tx: Mutex<Option<Sender<BackendCompletion>>>,
}

impl Backend for OneThenDisconnectBackend {
    fn kind(&self) -> BackendKind {
        BackendKind::Fallback
    }

    fn submit(&self, submission: BackendSubmission) -> Result<()> {
        let mut guard = self.tx.lock().expect("mock completion tx mutex");
        if let Some(tx) = guard.take() {
            // Deliver one good completion for this request, then let `tx` drop at
            // the end of scope so the channel disconnects. The next drain poll
            // must surface that disconnect rather than silently break the batch.
            tx.send(BackendCompletion {
                id: submission.id,
                result: Ok(CompletionPayload::Unit),
            })
            .expect("send mock completion");
        }
        Ok(())
    }

    fn cancel(&self, _cancellation: CancellationHandle) -> Result<()> {
        Ok(())
    }

    fn shutdown(&self) -> Result<()> {
        Ok(())
    }
}

#[derive(Default, Clone)]
struct OneThenDisconnectFactory;

impl BackendFactory for OneThenDisconnectFactory {
    fn create(
        &self,
        _config: &RingConfig,
        completion_tx: Sender<BackendCompletion>,
    ) -> Result<BoxedBackend> {
        Ok(Box::new(OneThenDisconnectBackend {
            tx: Mutex::new(Some(completion_tx)),
        }))
    }
}

#[test]
fn batch_drain_surfaces_non_timeout_error_after_the_first_completion() {
    let ring: Ring<OneThenDisconnectFactory> = Ring::new(RingConfig::default()).expect("ring");

    // Register one inflight op. Submitting drives the mock, which queues the
    // completion and disconnects the channel. Keep the handle alive so the
    // inflight entry survives until the drain routes it.
    let _req = ring.submit(Nop).expect("submit nop");

    let drain = BatchDrain::new(ring.clone(), 8);
    let results = drain.drain(Some(Duration::from_millis(100)));

    // First poll routes the queued completion; the second poll hits the
    // disconnected channel. That non-timeout error must be surfaced (Law 10),
    // not silently swallowed by the old `Err(_) => break` that discarded every
    // non-first error and hid fatal backend failures mid-batch.
    assert_eq!(
        results.len(),
        2,
        "expected the ok completion plus the surfaced disconnect error, got {results:?}"
    );
    assert!(
        results[0].is_ok(),
        "first completion should route ok, got {:?}",
        results[0]
    );
    let err = results[1]
        .as_ref()
        .expect_err("second poll must surface the disconnect error, not break quietly");
    assert!(
        matches!(err, Error::Completion { .. }),
        "expected a non-timeout completion error, got {err:?}"
    );
    assert!(
        err.to_string().contains("disconnected"),
        "error should name the channel disconnect: {err}"
    );
}