remora/crates/remora-stream/tests/kernel.rs

//! Streaming execution kernel acceptance matrix.
//!
//! Synthetic sources and targets exercise the bounded copy loop without any
//! backend fixture. Memory bound evidence is the single reserved payload buffer.

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::time::Duration;

use remora_domain::{FailureClass, FailureDisposition, RuntimeFailureClass};
use remora_stream::{
    CancelToken, CopyConfig, CopyHooks, CopyOutcome, PublicDetails, PublicReason, RetryTiming,
    SafeDetailValue, Sha256Accountant, SourceBody, StreamError, StreamSide, StructuredFailure,
    TargetBody, TargetFinish, TransferCheckpoint, TransferProgress, capped_server_hint,
    classify_stream_error, copy_stream, copy_stream_seeded, full_jitter_delay_with,
    public_message_for_class, saturating_backoff,
};
use tokio::sync::Notify;

const KILOBYTE: u64 = 1024;
const MEGABYTE: u64 = 1024 * KILOBYTE;
const GIGABYTE: u64 = 1024 * MEGABYTE;

/// Patterned source that never materializes the full payload.
struct PatternSource {
    remaining: u64,
    offset: u64,
    fail_after: Option<u64>,
    stall: Option<Duration>,
    stall_after: Option<u64>,
    max_chunk: usize,
    peak_fill: Arc<AtomicUsize>,
    oversize_read: bool,
}

impl PatternSource {
    fn new(total: u64) -> Self {
        Self {
            remaining: total,
            offset: 0,
            fail_after: None,
            stall: None,
            stall_after: None,
            max_chunk: usize::MAX,
            peak_fill: Arc::new(AtomicUsize::new(0)),
            oversize_read: false,
        }
    }

    fn starting_at(total: u64, offset: u64) -> Self {
        let mut source = Self::new(total.saturating_sub(offset));
        source.offset = offset;
        source
    }

    fn fail_after(mut self, bytes: u64) -> Self {
        self.fail_after = Some(bytes);
        self
    }

    fn stall_after(mut self, bytes: u64, stall: Duration) -> Self {
        self.stall_after = Some(bytes);
        self.stall = Some(stall);
        self
    }

    fn max_chunk(mut self, max_chunk: usize) -> Self {
        self.max_chunk = max_chunk;
        self
    }

    fn oversize_read(mut self) -> Self {
        self.oversize_read = true;
        self
    }
}

impl SourceBody for PatternSource {
    async fn read(&mut self, buf: &mut [u8]) -> Result<usize, StreamError> {
        if self.oversize_read {
            return Ok(buf.len().saturating_add(1));
        }
        if let Some(limit) = self.fail_after
            && self.offset >= limit
        {
            return Err(StreamError::transient_source(
                "synthetic mid-stream interruption with token=super-secret-token",
            ));
        }
        if let (Some(after), Some(stall)) = (self.stall_after, self.stall)
            && self.offset >= after
        {
            tokio::time::sleep(stall).await;
        }
        if self.remaining == 0 {
            return Ok(0);
        }
        let want = (buf.len().min(self.max_chunk) as u64).min(self.remaining) as usize;
        for (index, slot) in buf[..want].iter_mut().enumerate() {
            let absolute = self.offset + index as u64;
            *slot = pattern_byte(absolute);
        }
        self.peak_fill.fetch_max(want, Ordering::SeqCst);
        self.offset += want as u64;
        self.remaining -= want as u64;
        Ok(want)
    }
}

fn pattern_byte(offset: u64) -> u8 {
    (offset
        .wrapping_mul(0x9E37_79B9_7F4A_7C15)
        .wrapping_add(0xA24B_AED4_96E9_7C8D)
        >> 56) as u8
}

fn fill_pattern(offset: u64, len: usize) -> Vec<u8> {
    (0..len)
        .map(|index| pattern_byte(offset + index as u64))
        .collect()
}

struct CountingTarget {
    written: u64,
    absolute_base: u64,
    expected: Option<u64>,
    delay_per_write: Option<Duration>,
    stall_first_write: Option<Duration>,
    fail_after: Option<u64>,
    bytes: Option<Vec<u8>>,
    peak_write: Arc<AtomicUsize>,
}

impl CountingTarget {
    fn discarding() -> Self {
        Self {
            written: 0,
            absolute_base: 0,
            expected: None,
            delay_per_write: None,
            stall_first_write: None,
            fail_after: None,
            bytes: None,
            peak_write: Arc::new(AtomicUsize::new(0)),
        }
    }

    fn collecting() -> Self {
        Self {
            written: 0,
            absolute_base: 0,
            expected: None,
            delay_per_write: None,
            stall_first_write: None,
            fail_after: None,
            bytes: Some(Vec::new()),
            peak_write: Arc::new(AtomicUsize::new(0)),
        }
    }

    fn with_delay(mut self, delay: Duration) -> Self {
        self.delay_per_write = Some(delay);
        self
    }

    fn stall_first_write(mut self, stall: Duration) -> Self {
        self.stall_first_write = Some(stall);
        self
    }

    fn fail_after(mut self, bytes: u64) -> Self {
        self.fail_after = Some(bytes);
        self
    }

    fn expect_total(mut self, total: u64) -> Self {
        self.expected = Some(total);
        self
    }

    fn absolute_base(mut self, base: u64) -> Self {
        self.absolute_base = base;
        self
    }
}

impl TargetBody for CountingTarget {
    async fn write_all(&mut self, data: &[u8]) -> Result<(), StreamError> {
        if let Some(limit) = self.fail_after
            && self.written >= limit
        {
            return Err(StreamError::transient_target(
                "target write failed; Authorization: Bearer leaked-token",
            ));
        }
        if let Some(stall) = self.stall_first_write.take() {
            tokio::time::sleep(stall).await;
        }
        if let Some(delay) = self.delay_per_write {
            tokio::time::sleep(delay).await;
        }
        self.peak_write.fetch_max(data.len(), Ordering::SeqCst);
        if let Some(bytes) = &mut self.bytes {
            bytes.extend_from_slice(data);
        } else if let Some(expected) = self.expected {
            if !data.is_empty() {
                let absolute = self.absolute_base + self.written;
                assert_eq!(data[0], pattern_byte(absolute));
                let last = data.len() - 1;
                assert_eq!(data[last], pattern_byte(absolute + last as u64));
            }
            let _ = expected;
        }
        self.written += data.len() as u64;
        Ok(())
    }

    async fn finish(self) -> Result<TargetFinish, StreamError> {
        if let Some(expected) = self.expected {
            assert_eq!(self.written, expected);
        }
        Ok(TargetFinish {
            bytes_written: self.written,
        })
    }
}

#[derive(Default)]
struct RecordingHooks {
    progress_events: Vec<u64>,
    checkpoint_events: Vec<u64>,
    progress_count: Arc<AtomicU64>,
    checkpoint_count: Arc<AtomicU64>,
}

impl CopyHooks for RecordingHooks {
    async fn on_progress(&mut self, progress: &TransferProgress) -> Result<(), StreamError> {
        self.progress_events.push(progress.completed_bytes());
        self.progress_count.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }

    async fn on_checkpoint(&mut self, checkpoint: &TransferCheckpoint) -> Result<(), StreamError> {
        self.checkpoint_events.push(checkpoint.completed_bytes());
        self.checkpoint_count.fetch_add(1, Ordering::SeqCst);
        // Integrity covers prior seed + session; equals absolute completed bytes.
        assert_eq!(
            checkpoint.integrity().bytes_hashed(),
            checkpoint.completed_bytes()
        );
        assert!(checkpoint.integrity().digest().is_some());
        Ok(())
    }
}

fn config_for(total: u64, buffer_size: usize) -> CopyConfig {
    CopyConfig {
        buffer_size,
        progress_interval_bytes: buffer_size as u64,
        checkpoint_interval_bytes: (buffer_size as u64).saturating_mul(4),
        no_progress_timeout: Duration::from_secs(5),
        initial_completed_bytes: 0,
        total_bytes: Some(total),
        completed_members: 0,
        total_members: Some(1),
    }
}

#[tokio::test]
async fn small_transfer_completes_with_known_digest() {
    let payload = b"remora-stream-kernel";
    let source = SliceSource {
        data: payload.to_vec(),
        position: 0,
    };
    let target = CountingTarget::collecting();
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();
    let outcome = copy_stream(
        source,
        target,
        config_for(payload.len() as u64, 64),
        &cancel,
        &mut hooks,
    )
    .await;

    let CopyOutcome::Completed(result) = outcome else {
        panic!("expected completion: {outcome:?}");
    };
    assert_eq!(result.progress.completed_bytes(), payload.len() as u64);
    assert_eq!(result.progress.session_bytes(), payload.len() as u64);
    assert_eq!(result.target.bytes_written, payload.len() as u64);
    // Requested 64 is clamped up to MIN_BUFFER_SIZE.
    assert_eq!(result.buffer_len, CopyConfig::MIN_BUFFER_SIZE);
    assert!(result.buffer_capacity >= result.buffer_len);
    assert!(result.buffer_capacity < result.buffer_len.saturating_add(4096));
    assert_eq!(
        result.integrity.digest().map(|d| d.as_str()),
        Some(sha256_hex(payload).as_str())
    );
    assert!(!hooks.progress_events.is_empty());
}

struct SliceSource {
    data: Vec<u8>,
    position: usize,
}

impl SourceBody for SliceSource {
    async fn read(&mut self, buf: &mut [u8]) -> Result<usize, StreamError> {
        if self.position >= self.data.len() {
            return Ok(0);
        }
        let n = buf.len().min(self.data.len() - self.position);
        buf[..n].copy_from_slice(&self.data[self.position..self.position + n]);
        self.position += n;
        Ok(n)
    }
}

fn sha256_hex(data: &[u8]) -> String {
    use sha2::{Digest, Sha256};
    let digest = Sha256::digest(data);
    let mut out = String::from("sha256:");
    for byte in digest {
        use std::fmt::Write as _;
        let _ = write!(out, "{byte:02x}");
    }
    out
}

#[tokio::test]
async fn multi_gigabyte_stream_keeps_bounded_memory() {
    // Two gigabytes of synthetic patterned data. The kernel holds one buffer.
    let total = 2 * GIGABYTE;
    let buffer_size = 256 * 1024;
    let source = PatternSource::new(total).max_chunk(buffer_size);
    let peak_fill = Arc::clone(&source.peak_fill);
    let target = CountingTarget::discarding().expect_total(total);
    let peak_write = Arc::clone(&target.peak_write);
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();

    let outcome = copy_stream(
        source,
        target,
        config_for(total, buffer_size),
        &cancel,
        &mut hooks,
    )
    .await;

    let CopyOutcome::Completed(result) = outcome else {
        panic!("expected multi-GB completion: {outcome:?}");
    };
    assert_eq!(result.progress.completed_bytes(), total);
    assert_eq!(result.target.bytes_written, total);
    assert_eq!(result.integrity.bytes_hashed(), total);
    assert!(result.integrity.digest().is_some());

    // Bounded memory evidence: one configured buffer, no growth beyond page slack.
    assert_eq!(result.buffer_len, buffer_size);
    assert!(result.buffer_capacity >= buffer_size);
    assert!(result.buffer_capacity < buffer_size.saturating_add(4096));
    assert!(peak_fill.load(Ordering::SeqCst) <= buffer_size);
    assert!(peak_write.load(Ordering::SeqCst) <= buffer_size);
    assert!(hooks.progress_count.load(Ordering::SeqCst) > 0);
    assert!(hooks.checkpoint_count.load(Ordering::SeqCst) > 0);
}

#[tokio::test]
async fn mid_stream_interruption_classifies_and_preserves_progress() {
    let total = 8 * MEGABYTE;
    let fail_after = 3 * MEGABYTE;
    let buffer_size = 64 * 1024;
    let source = PatternSource::new(total).fail_after(fail_after);
    let target = CountingTarget::discarding();
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();

    let outcome = copy_stream(
        source,
        target,
        config_for(total, buffer_size),
        &cancel,
        &mut hooks,
    )
    .await;

    let CopyOutcome::Failed {
        failure,
        progress,
        integrity,
        buffer_len,
        buffer_capacity,
    } = outcome
    else {
        panic!("expected failure: {outcome:?}");
    };

    assert_eq!(failure.class(), FailureClass::TransientSource);
    assert_eq!(failure.disposition(), Some(FailureDisposition::Retryable));
    assert_eq!(
        failure.runtime_class(),
        Some(RuntimeFailureClass::TransientSource)
    );
    // Public projection must not carry the synthetic secret token.
    assert!(!failure.message().contains("super-secret-token"));
    assert!(!format!("{failure:?}").contains("super-secret-token"));
    assert!(progress.completed_bytes() >= fail_after.saturating_sub(buffer_size as u64));
    assert!(progress.completed_bytes() < total);
    assert_eq!(integrity.bytes_hashed(), progress.completed_bytes());
    assert_eq!(buffer_len, buffer_size);
    assert!(buffer_capacity >= buffer_len);
}

#[tokio::test]
async fn backpressure_still_completes_with_bounded_buffer() {
    let total = 512 * KILOBYTE;
    let buffer_size = 16 * 1024;
    let source = PatternSource::new(total).max_chunk(buffer_size);
    let target = CountingTarget::discarding()
        .expect_total(total)
        .with_delay(Duration::from_millis(1));
    let peak_write = Arc::clone(&target.peak_write);
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();

    let outcome = copy_stream(
        source,
        target,
        config_for(total, buffer_size),
        &cancel,
        &mut hooks,
    )
    .await;

    let CopyOutcome::Completed(result) = outcome else {
        panic!("expected completion under backpressure: {outcome:?}");
    };
    assert_eq!(result.progress.completed_bytes(), total);
    assert!(peak_write.load(Ordering::SeqCst) <= buffer_size);
}

#[tokio::test]
async fn no_progress_timeout_maps_to_transient_source() {
    let total = MEGABYTE;
    let buffer_size = 64 * 1024;
    let source = PatternSource::new(total).stall_after(0, Duration::from_millis(200));
    let target = CountingTarget::discarding();
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();
    let mut config = config_for(total, buffer_size);
    config.no_progress_timeout = Duration::from_millis(50);

    let outcome = copy_stream(source, target, config, &cancel, &mut hooks).await;

    let CopyOutcome::Failed {
        failure, progress, ..
    } = outcome
    else {
        panic!("expected no-progress failure: {outcome:?}");
    };
    assert_eq!(failure.class(), FailureClass::TransientSource);
    assert_eq!(
        failure.safe_details().get("reason"),
        Some(&SafeDetailValue::String("no_progress".into()))
    );
    assert_eq!(progress.completed_bytes(), 0);
}

#[tokio::test]
async fn no_progress_timeout_maps_to_transient_target() {
    let total = MEGABYTE;
    let buffer_size = 64 * 1024;
    let source = PatternSource::new(total).max_chunk(buffer_size);
    let target = CountingTarget::discarding().stall_first_write(Duration::from_millis(200));
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();
    let mut config = config_for(total, buffer_size);
    config.no_progress_timeout = Duration::from_millis(50);

    let outcome = copy_stream(source, target, config, &cancel, &mut hooks).await;

    let CopyOutcome::Failed {
        failure, progress, ..
    } = outcome
    else {
        panic!("expected target no-progress failure: {outcome:?}");
    };
    assert_eq!(failure.class(), FailureClass::TransientTarget);
    assert_eq!(
        failure.safe_details().get("reason"),
        Some(&SafeDetailValue::String("no_progress".into()))
    );
    assert_eq!(
        failure.safe_details().get("side"),
        Some(&SafeDetailValue::String("target".into()))
    );
    assert_eq!(progress.completed_bytes(), 0);
}

#[tokio::test]
async fn cooperative_cancellation_stops_transfer() {
    let total = 64 * MEGABYTE;
    let buffer_size = 64 * 1024;
    let source = PatternSource::new(total).max_chunk(4 * 1024);
    let target = CountingTarget::discarding();
    let cancel = CancelToken::new();
    let cancel_for_task = cancel.clone();
    let notify = Arc::new(Notify::new());
    let notify_wait = Arc::clone(&notify);
    let checkpoints_after_cancel = Arc::new(AtomicU64::new(0));
    let checkpoints_for_hooks = Arc::clone(&checkpoints_after_cancel);

    struct CancelAfterProgress {
        cancel: CancelToken,
        notify: Arc<Notify>,
        armed: bool,
        checkpoints_after_cancel: Arc<AtomicU64>,
    }

    impl CopyHooks for CancelAfterProgress {
        async fn on_progress(&mut self, progress: &TransferProgress) -> Result<(), StreamError> {
            if !self.armed && progress.completed_bytes() > 0 {
                self.armed = true;
                self.cancel.cancel();
                self.notify.notify_one();
            }
            Ok(())
        }

        async fn on_checkpoint(
            &mut self,
            _checkpoint: &TransferCheckpoint,
        ) -> Result<(), StreamError> {
            if self.armed {
                self.checkpoints_after_cancel.fetch_add(1, Ordering::SeqCst);
            }
            Ok(())
        }
    }

    let hooks = CancelAfterProgress {
        cancel: cancel_for_task,
        notify,
        armed: false,
        checkpoints_after_cancel: checkpoints_for_hooks,
    };

    let copy = tokio::spawn(async move {
        copy_stream(
            source,
            target,
            config_for(total, buffer_size),
            &cancel,
            hooks,
        )
        .await
    });

    notify_wait.notified().await;
    let outcome = copy.await.expect("join");

    match outcome {
        CopyOutcome::Cancelled {
            progress,
            integrity,
            ..
        } => {
            assert!(progress.completed_bytes() > 0);
            assert!(progress.completed_bytes() < total);
            assert_eq!(integrity.bytes_hashed(), progress.completed_bytes());
            assert_eq!(checkpoints_after_cancel.load(Ordering::SeqCst), 0);
        }
        other => panic!("expected cancellation: {other:?}"),
    }
}

#[tokio::test]
async fn oversize_source_read_fails_closed() {
    let source = PatternSource::new(MEGABYTE).oversize_read();
    let target = CountingTarget::discarding();
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();

    let outcome = copy_stream(
        source,
        target,
        config_for(MEGABYTE, 64 * 1024),
        &cancel,
        &mut hooks,
    )
    .await;

    let CopyOutcome::Failed {
        failure, progress, ..
    } = outcome
    else {
        panic!("expected oversize-read failure: {outcome:?}");
    };
    assert_eq!(failure.class(), FailureClass::Internal);
    assert_eq!(progress.completed_bytes(), 0);
}

#[tokio::test]
async fn resume_with_seeded_integrity_matches_full_digest() {
    let total = 256 * 1024u64;
    let half = total / 2;
    let buffer_size = 16 * 1024;

    let mut accountant = Sha256Accountant::new();
    let prefix = fill_pattern(0, half as usize);
    accountant.update(&prefix);
    assert_eq!(accountant.bytes_hashed(), half);

    let source = PatternSource::starting_at(total, half).max_chunk(buffer_size);
    let target = CountingTarget::discarding()
        .expect_total(half)
        .absolute_base(half);
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();
    let mut config = config_for(total, buffer_size);
    config.initial_completed_bytes = half;
    config.progress_interval_bytes = 32 * 1024;
    config.checkpoint_interval_bytes = 64 * 1024;

    let outcome = copy_stream_seeded(source, target, config, &cancel, &mut hooks, accountant).await;

    let CopyOutcome::Completed(result) = outcome else {
        panic!("expected resumed completion: {outcome:?}");
    };
    assert_eq!(result.progress.completed_bytes(), total);
    assert_eq!(result.progress.session_bytes(), half);
    assert_eq!(result.target.bytes_written, half);
    assert_eq!(result.integrity.bytes_hashed(), total);

    let full = fill_pattern(0, total as usize);
    assert_eq!(
        result.integrity.digest().map(|d| d.as_str()),
        Some(sha256_hex(&full).as_str())
    );
}

#[tokio::test]
async fn mismatched_integrity_seed_fails_closed() {
    let mut accountant = Sha256Accountant::new();
    accountant.update(b"prefix");
    let source = PatternSource::new(KILOBYTE);
    let target = CountingTarget::discarding();
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();
    let mut config = config_for(KILOBYTE, 4 * 1024);
    config.initial_completed_bytes = 100; // does not match accountant

    let outcome = copy_stream_seeded(source, target, config, &cancel, &mut hooks, accountant).await;

    let CopyOutcome::Failed { failure, .. } = outcome else {
        panic!("expected seed mismatch failure: {outcome:?}");
    };
    assert_eq!(failure.class(), FailureClass::Internal);
}

#[tokio::test]
async fn non_zero_initial_without_seed_fails_closed() {
    let source = PatternSource::new(KILOBYTE);
    let target = CountingTarget::discarding();
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();
    let mut config = config_for(KILOBYTE, 4 * 1024);
    config.initial_completed_bytes = 64;

    let outcome = copy_stream(source, target, config, &cancel, &mut hooks).await;
    let CopyOutcome::Failed { failure, .. } = outcome else {
        panic!("expected unseeded resume failure: {outcome:?}");
    };
    assert_eq!(failure.class(), FailureClass::Internal);
}

#[tokio::test]
async fn target_failure_redacts_authorization_header() {
    let total = MEGABYTE;
    let buffer_size = 32 * 1024;
    let source = PatternSource::new(total);
    let target = CountingTarget::discarding().fail_after(128 * 1024);
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();

    let outcome = copy_stream(
        source,
        target,
        config_for(total, buffer_size),
        &cancel,
        &mut hooks,
    )
    .await;

    let CopyOutcome::Failed { failure, .. } = outcome else {
        panic!("expected target failure: {outcome:?}");
    };
    assert_eq!(failure.class(), FailureClass::TransientTarget);
    assert!(!failure.message().contains("leaked-token"));
    assert!(!failure.message().contains("Bearer"));
    assert_eq!(
        failure.message(),
        public_message_for_class(FailureClass::TransientTarget)
    );
}

#[test]
fn structured_failure_mapping_covers_runtime_classes() {
    let cases = [
        (
            StreamError::Authorization {
                side: StreamSide::Source,
            },
            FailureClass::Authorization,
            FailureDisposition::Permanent,
        ),
        (
            StreamError::SourceNotFound,
            FailureClass::SourceNotFound,
            FailureDisposition::Permanent,
        ),
        (
            StreamError::SourceChanged,
            FailureClass::SourceChanged,
            FailureDisposition::Permanent,
        ),
        (
            StreamError::transient_source("x"),
            FailureClass::TransientSource,
            FailureDisposition::Retryable,
        ),
        (
            StreamError::transient_target("x"),
            FailureClass::TransientTarget,
            FailureDisposition::Retryable,
        ),
        (
            StreamError::integrity("x"),
            FailureClass::Integrity,
            FailureDisposition::Retryable,
        ),
        (
            StreamError::Capacity {
                side: StreamSide::Target,
            },
            FailureClass::Capacity,
            FailureDisposition::Retryable,
        ),
        (
            StreamError::internal("stack trace at https://evil.example/path?token=1"),
            FailureClass::Internal,
            FailureDisposition::Retryable,
        ),
        (
            StreamError::Cancelled,
            FailureClass::Cancelled,
            FailureDisposition::Permanent,
        ),
    ];

    for (error, class, disposition) in cases {
        let failure = classify_stream_error(&error);
        assert_eq!(failure.class(), class);
        assert_eq!(failure.disposition(), Some(disposition));
        assert!(!failure.message().contains("https://"));
        assert!(!failure.message().contains("token="));
        assert!(!failure.message().contains("stack trace"));
    }
}

#[test]
fn public_failure_rejects_free_form_secrets() {
    let secret = "synthetic-api-key-4f7c9a";
    // Typed constructor cannot accept free-form message or reason text.
    let failure = StructuredFailure::public(
        FailureClass::TransientSource,
        PublicDetails::new()
            .with_side(StreamSide::Source)
            .with_reason(PublicReason::NoProgress),
    );
    assert_eq!(
        failure.message(),
        public_message_for_class(FailureClass::TransientSource)
    );
    assert!(!failure.message().contains(secret));
    assert_eq!(
        failure.safe_details().get("reason"),
        Some(&SafeDetailValue::String("no_progress".into()))
    );
    assert!(!format!("{failure:?}").contains(secret));
}

#[tokio::test]
async fn target_finish_honors_no_progress_timeout() {
    use std::future::pending;

    struct EmptySource;
    impl SourceBody for EmptySource {
        async fn read(&mut self, _buf: &mut [u8]) -> Result<usize, StreamError> {
            Ok(0)
        }
    }

    struct StalledFinish;
    impl TargetBody for StalledFinish {
        async fn write_all(&mut self, _data: &[u8]) -> Result<(), StreamError> {
            Ok(())
        }
        async fn finish(self) -> Result<TargetFinish, StreamError> {
            pending().await
        }
    }

    let cancel = CancelToken::new();
    let config = CopyConfig {
        no_progress_timeout: Duration::from_millis(20),
        ..CopyConfig::default()
    };
    let outcome = tokio::time::timeout(
        Duration::from_millis(200),
        copy_stream(
            EmptySource,
            StalledFinish,
            config,
            &cancel,
            RecordingHooks::default(),
        ),
    )
    .await
    .expect("finish must not hang past no-progress bound");

    let CopyOutcome::Failed { failure, .. } = outcome else {
        panic!("expected target no-progress on stalled finish: {outcome:?}");
    };
    assert_eq!(failure.class(), FailureClass::TransientTarget);
    assert_eq!(
        failure.safe_details().get("reason"),
        Some(&SafeDetailValue::String("no_progress".into()))
    );
}

#[tokio::test]
async fn target_finish_honors_cancellation() {
    use std::future::pending;

    struct EmptySource;
    impl SourceBody for EmptySource {
        async fn read(&mut self, _buf: &mut [u8]) -> Result<usize, StreamError> {
            Ok(0)
        }
    }

    struct StalledFinish;
    impl TargetBody for StalledFinish {
        async fn write_all(&mut self, _data: &[u8]) -> Result<(), StreamError> {
            Ok(())
        }
        async fn finish(self) -> Result<TargetFinish, StreamError> {
            pending().await
        }
    }

    let cancel = CancelToken::new();
    let cancel_for_task = cancel.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(10)).await;
        cancel_for_task.cancel();
    });

    // Long no-progress so cancellation, not timeout, must win.
    let config = CopyConfig {
        no_progress_timeout: Duration::from_secs(30),
        ..CopyConfig::default()
    };
    let outcome = tokio::time::timeout(
        Duration::from_millis(200),
        copy_stream(
            EmptySource,
            StalledFinish,
            config,
            &cancel,
            RecordingHooks::default(),
        ),
    )
    .await
    .expect("finish must return promptly on cancel");

    assert!(
        matches!(outcome, CopyOutcome::Cancelled { .. }),
        "expected cancellation during finish: {outcome:?}"
    );
}

#[test]
fn retry_timing_helpers_honor_bounds() {
    let timing = RetryTiming {
        base_backoff: Duration::from_millis(50),
        max_backoff: Duration::from_millis(400),
        max_server_hint: Duration::from_secs(2),
        ..RetryTiming::default_bounds()
    };
    assert!(saturating_backoff(0, &timing) <= timing.max_backoff);
    assert!(saturating_backoff(20, &timing) <= timing.max_backoff);
    let delay = full_jitter_delay_with(Duration::from_millis(100), 42);
    assert!(delay <= Duration::from_millis(100));
    assert_eq!(
        capped_server_hint(Some(Duration::from_secs(1)), &timing),
        Some(Duration::from_secs(1))
    );
    assert_eq!(
        capped_server_hint(Some(Duration::from_secs(10)), &timing),
        None
    );
}

#[tokio::test]
async fn progress_and_checkpoint_hooks_fire_on_intervals() {
    let total = 256 * 1024;
    let buffer_size = 16 * 1024;
    let source = PatternSource::new(total);
    let target = CountingTarget::discarding().expect_total(total);
    let cancel = CancelToken::new();
    let mut hooks = RecordingHooks::default();
    let mut config = config_for(total, buffer_size);
    config.progress_interval_bytes = 32 * 1024;
    config.checkpoint_interval_bytes = 64 * 1024;

    let outcome = copy_stream(source, target, config, &cancel, &mut hooks).await;
    assert!(matches!(outcome, CopyOutcome::Completed(_)));
    assert!(hooks.progress_events.len() >= 4);
    assert!(hooks.checkpoint_events.len() >= 2);
    assert!(hooks.progress_events.is_sorted());
    assert!(hooks.checkpoint_events.is_sorted());
    assert_eq!(*hooks.progress_events.last().unwrap(), total);
}