alkgit 0.1.0

//! The async→blocking pack-source bridge (ADR-018 §4, ADR-005 §1):
//! the substrate piece between the push machine's command flush and
//! [`GitPackIngest::prepare`](crate::GitPackIngest::prepare)'s boxed
//! `io::Read + Send` pack end.
//!
//! The substrates surrender the pack tail as a raw async reader
//! ([`RawTail`](crate::duplex::RawTail) — the duplex session's read
//! half past the command flush, the stateless POST body past the
//! command invocation's flush); [`bridge_pack_source`] turns that into
//! the blocking end `prepare` consumes, the machine hands it in, and
//! the ingestion (impl-side, `spawn_blocking`) reads the pack from it:
//!
//! - **Bounded by construction** (ADR-005 §1's bounded-channel rule): a
//!   relay task reads bounded chunks from the async tail and sends them
//!   over [`PACK_SOURCE_CHANNEL_CAPACITY`]-bounded mpsc; the blocking
//!   end parks on the channel when the relay falls behind, and the
//!   relay parks on the channel when the blocking ingestion falls
//!   behind — backpressure is bounded end to end, no unbounded buffer
//!   ever grows.
//! - **The relay-break discipline** (the wave-2 sideband relay's
//!   break-before-join): a downstream failure (the ingestion gone —
//!   the blocking end dropped) breaks the relay before any join, so
//!   nothing parks forever; a transport failure (the tail errored)
//!   breaks it too. The [`PackSourceBridge::relay`] handle exists so
//!   the machine can join after the fact; the bridge's failures
//!   surface on the blocking end as the io errors an ingestion turns
//!   into its `unpack` ng leg.
//! - **The wire-side pack-size budget** (ADR-009's `max_pack_size`): the
//!   relay counts the streamed bytes and fails closed — a breach errors
//!   the bridge *before* the budget-exceeding bytes are delivered
//!   (nothing past the budget line ever reaches the ingestion). The
//!   typed failure lands in [`PackSourceFault`] as the client-fault
//!   budget class ([`SubstrateError::Protocol`]: 413-mapped at the door
//!   on stateless, session error on duplex); the blocking end surfaces
//!   the same failure as an io error. `Limits` still flows into
//!   `prepare` — the impl-side defense is the impl's
//!   (`core/w3-pack-ingest-doubles`' counting-reader posture); this
//!   relay is the wire-side enforcement point.
//!
//! Front-door-blind (convention 15): the bridge is generic over the
//! async reader; no http/door/channel type appears anywhere here.

use std::io;
use std::sync::Arc;

use futures::io::{AsyncRead, AsyncReadExt};
use parking_lot::Mutex as ParkingMutex;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;

use crate::limits::Limits;
use crate::substrate::SubstrateError;

/// Bounded capacity of the pack-source bridge's channel (ADR-005 §1's
/// bounded rule, the sideband sink's posture mirrored onto the inbound
/// stream): either side that falls behind parks — no unbounded buffer.
pub const PACK_SOURCE_CHANNEL_CAPACITY: usize = 4;

/// The relay's per-read chunk cap: reads through the tail request at
/// most this many bytes, and each channel message carries at most this
/// many — the buffer bound (ADR-009's bounded-resources posture:
/// budgets bound aggregate work, internal buffering stays capped).
pub const PACK_SOURCE_CHUNK_SIZE: usize = 64 * 1024;

/// The typed-failure handle the relay records through (the machine's
/// observable half of the failure taxonomy): a wire-side pack-size
/// breach lands here as [`SubstrateError::Protocol`] (the client-fault
/// budget class — 413-mapped at the door on stateless, session error on
/// duplex, ADR-009), a transport failure as [`SubstrateError::Io`]. The
/// handle clones cheaply (shared slot) so the machine can hold it while
/// the boxed end moves into `prepare`.
///
/// Contract stamp: ADR-009 (the fail-closed breach behavior and its
/// client-fault class), ADR-013 §3 (the streaming failure surfaces into
/// the ingestion that turns it into the ng leg).
#[derive(Clone, Default)]
pub struct PackSourceFault(Arc<ParkingMutex<Option<SubstrateError>>>);

impl PackSourceFault {
    fn record(&self, error: SubstrateError) {
        *self.0.lock() = Some(error);
    }

    /// Takes the recorded failure, if any (the machine consults it
    /// after the ingestion outcome to classify the wire failure).
    pub fn take_failure(&self) -> Option<SubstrateError> {
        self.0.lock().take()
    }
}

/// The blocking-side pack end ([`GitPackIngest::prepare`](crate::GitPackIngest::prepare)'s
/// `Box<dyn io::Read + Send>` filler): a `std::io::Read` over the
/// bounded bridge channel. Reads serve the buffered chunk remainder
/// first, then park on the channel (the blocking posture —
/// `futures::executor::block_on` over the recv future, the wave-1/2
/// bridge choice, safe from any blocking thread with or without a
/// runtime context); channel close reads as clean EOF; a relay failure
/// reads as that io error (re-reported on later reads, the cached
/// terminal form — no re-entry into the channel after it).
///
/// Contract stamp: ADR-005 §1 (the bounded bridge the boxed stream
/// terminates), ADR-018 §4 (the `io::Read + Send` end), ADR-009 (the
/// bounded backpressure).
pub struct PackSource {
    rx: mpsc::Receiver<io::Result<Vec<u8>>>,
    chunk: Vec<u8>,
    pos: usize,
    failure: Option<(io::ErrorKind, String)>,
}

impl io::Read for PackSource {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }
        if self.pos < self.chunk.len() {
            let n = buf.len().min(self.chunk.len() - self.pos);
            buf[..n].copy_from_slice(&self.chunk[self.pos..self.pos + n]);
            self.pos += n;
            return Ok(n);
        }
        if let Some((kind, message)) = &self.failure {
            return Err(io::Error::new(*kind, message.clone()));
        }
        match futures::executor::block_on(self.rx.recv()) {
            Some(Ok(chunk)) => {
                if chunk.is_empty() {
                    return Ok(0);
                }
                let n = buf.len().min(chunk.len());
                buf[..n].copy_from_slice(&chunk[..n]);
                self.chunk = chunk;
                self.pos = n;
                Ok(n)
            }
            Some(Err(error)) => {
                let kind = error.kind();
                let message = error.to_string();
                self.failure = Some((kind, message.clone()));
                Err(io::Error::new(kind, message))
            }
            None => Ok(0),
        }
    }
}

/// Bridges the async pack-tail reader into the blocking end
/// `prepare` consumes: spawns the relay task over a bounded channel
/// and returns the bridge triple — the blocking end (box it for the
/// `Box<dyn io::Read + Send>` parameter), the [`PackSourceFault`]
/// handle (clone it into the machine), and the relay join handle (await
/// it only after the ingestion has finished with the blocking end —
/// dropping or exhausting the end is what breaks the relay, the
/// break-before-join discipline).
///
/// Contract stamp: ADR-018 §4 (the boxed `io::Read + Send` end the wire
/// layer constructs), ADR-005 §1 (the bounded-channel substrate
/// structure), ADR-013 §3 (the raw pack tail), ADR-009 (the wire-side
/// `max_pack_size` enforcement, fail-closed).
pub fn bridge_pack_source<R>(tail: R, limits: &Limits) -> PackSourceBridge
where
    R: AsyncRead + Unpin + Send + 'static,
{
    let (tx, rx) = mpsc::channel(PACK_SOURCE_CHANNEL_CAPACITY);
    let fault = PackSourceFault::default();
    let relay = tokio::spawn(relay_pack_chunks(
        tail,
        tx,
        limits.max_pack_size,
        fault.clone(),
    ));
    PackSourceBridge {
        end: PackSource {
            rx,
            chunk: Vec::new(),
            pos: 0,
            failure: None,
        },
        fault,
        relay,
    }
}

/// The bridged pack source ([`bridge_pack_source`]'s output): the
/// blocking end, the typed-failure handle, and the relay's join handle.
pub struct PackSourceBridge {
    /// The blocking `io::Read + Send` end — box it into `prepare`'s
    /// `pack` parameter.
    pub end: PackSource,
    /// The typed-failure handle (the wire-side failure taxonomy).
    pub fault: PackSourceFault,
    /// The relay task: breaks when the blocking end is dropped or
    /// exhausted; join it after the ingestion finished with the end.
    pub relay: JoinHandle<()>,
}

/// The bridge relay: reads bounded chunks from the async tail into the
/// bounded channel until EOF; fails closed on the [`Limits::max_pack_size`]
/// budget (before the breaching bytes are delivered) and on a transport
/// failure; breaks on a downstream failure instead of parking forever
/// (the wave-2 sideband relay's break-before-join discipline).
async fn relay_pack_chunks<R>(
    mut tail: R,
    tx: mpsc::Sender<io::Result<Vec<u8>>>,
    budget: u64,
    fault: PackSourceFault,
) where
    R: AsyncRead + Unpin,
{
    let mut chunk = vec![0u8; PACK_SOURCE_CHUNK_SIZE];
    let mut relayed: u64 = 0;
    loop {
        match tail.read(&mut chunk).await {
            Ok(0) => break,
            Ok(n) => {
                if relayed.saturating_add(n as u64) > budget {
                    let message = format!(
                        "the pack exceeds the max_pack_size budget ({budget}) — fail closed (ADR-009)"
                    );
                    fault.record(SubstrateError::Protocol(message.clone()));
                    let _ = tx
                        .send(Err(io::Error::new(io::ErrorKind::InvalidData, message)))
                        .await;
                    break;
                }
                relayed += n as u64;
                if tx.send(Ok(chunk[..n].to_vec())).await.is_err() {
                    // The ingestion side is gone: break the relay before
                    // any join — nothing parks forever.
                    break;
                }
            }
            Err(error) => {
                let kind = error.kind();
                let message = error.to_string();
                fault.record(SubstrateError::Io(io::Error::new(kind, message.clone())));
                let _ = tx.send(Err(io::Error::new(kind, message))).await;
                break;
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use bytes::Bytes;
    use futures::stream::{self};
    use std::io::Read as _;
    use std::time::Duration;
    use tokio::time::timeout;

    type BodyChunks = stream::Iter<std::vec::IntoIter<io::Result<Bytes>>>;
    type TestTail = crate::stateless::BodyReader<BodyChunks, io::Error>;

    fn body_reader(items: Vec<io::Result<Bytes>>) -> TestTail {
        crate::stateless::BodyReader::new(stream::iter(items))
    }

    fn tail(chunks: Vec<Vec<u8>>) -> TestTail {
        body_reader(
            chunks
                .into_iter()
                .map(|bytes| io::Result::Ok(Bytes::from(bytes)))
                .collect(),
        )
    }

    fn failing_tail(chunks: Vec<Vec<u8>>) -> TestTail {
        let mut items: Vec<io::Result<Bytes>> = chunks
            .into_iter()
            .map(|bytes| io::Result::Ok(Bytes::from(bytes)))
            .collect();
        items.push(io::Result::Err(io::Error::other("the body failed")));
        body_reader(items)
    }

    // The blocking reads must not block the test thread forever: the
    // relay runs on the multi-thread runtime while the test thread
    // parks on the channel (the blocking posture the ingestion uses).
    fn multi_thread() -> tokio::runtime::Runtime {
        tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .unwrap()
    }

    #[test]
    fn tail_bytes_reach_the_blocking_end_in_order_to_eof() {
        let rt = multi_thread();
        rt.block_on(async {
            let data: Vec<u8> = (0..200_000u32).map(|i| (i % 251) as u8).collect();
            let chunks = data.chunks(17_333).map(<[u8]>::to_vec).collect();
            let bridge = bridge_pack_source(tail(chunks), &Limits::default());
            let PackSourceBridge {
                mut end,
                fault,
                relay,
            } = bridge;
            assert!(fault.take_failure().is_none());

            let mut got = Vec::new();
            end.read_to_end(&mut got).unwrap();
            assert_eq!(got, data, "in order, byte-exact, to clean EOF");
            relay.await.unwrap();
            assert!(fault.take_failure().is_none(), "EOF is no failure");
        });
    }

    /// One bridge, bounded pieces: small reads across chunk boundaries
    /// keep the order (the buffered remainder serves first).
    #[test]
    fn small_reads_keep_the_order_across_chunk_boundaries() {
        let rt = multi_thread();
        rt.block_on(async {
            let data: Vec<u8> = (0..5000u32).map(|i| (i % 7) as u8).collect();
            let chunks = vec![data[..3000].to_vec(), data[3000..].to_vec()];
            let bridge = bridge_pack_source(tail(chunks), &Limits::default());
            let PackSourceBridge { mut end, relay, .. } = bridge;

            for expected_window in data.chunks(137) {
                let mut got = vec![0u8; expected_window.len()];
                end.read_exact(&mut got).unwrap();
                assert_eq!(got, expected_window);
            }
            let mut eof = [0u8; 1];
            assert_eq!(end.read(&mut eof).unwrap(), 0, "clean EOF");
            relay.await.unwrap();
        });
    }

    /// A slow ingestion parks the relay on the bounded channel: the
    /// reader drains a dribble at a time, everything arrives in order,
    /// and the relay only completes after the drain (the parked posture,
    /// observed through the join).
    #[test]
    fn a_slow_reader_gets_bounded_in_order_chunks_and_the_relay_parks_to_the_end() {
        let rt = multi_thread();
        rt.block_on(async {
            let data: Vec<u8> = vec![0u8; PACK_SOURCE_CHUNK_SIZE * 3 + 7];
            let bridge = bridge_pack_source(tail(vec![data.clone()]), &Limits::default());
            let PackSourceBridge {
                mut end,
                fault,
                relay,
            } = bridge;
            assert!(fault.take_failure().is_none());

            let mut joined = tokio::spawn(relay);

            let mut total = 0usize;
            loop {
                let mut got = vec![0u8; 997];
                let n = end.read(&mut got).unwrap();
                if n == 0 {
                    break;
                }
                assert!(n <= PACK_SOURCE_CHUNK_SIZE, "the per-chunk cap holds");
                total += n;
            }
            assert_eq!(total, data.len(), "all bytes arrived, in order");
            let _ = timeout(Duration::from_secs(5), &mut joined)
                .await
                .expect("the relay finished after the drain")
                .unwrap();
        });
    }

    /// A downstream failure (the ingestion gone — the blocking end
    /// dropped mid-stream) breaks the relay before any join: nothing
    /// parks forever.
    #[test]
    fn a_downstream_failure_breaks_the_relay() {
        let rt = multi_thread();
        rt.block_on(async {
            let data: Vec<u8> = vec![0u8; PACK_SOURCE_CHUNK_SIZE * 20];
            let bridge = bridge_pack_source(tail(vec![data]), &Limits::default());
            let PackSourceBridge {
                mut end,
                fault,
                relay,
            } = bridge;

            let mut got = vec![0u8; 100];
            end.read_exact(&mut got).unwrap();
            drop(end);

            timeout(Duration::from_secs(5), relay)
                .await
                .expect("the relay breaks promptly after the downstream is gone")
                .unwrap();
            // Dropping the end is the machine's own action, not a bridge
            // failure — the fault slot stays empty.
            assert!(fault.take_failure().is_none());
        });
    }

    /// The wire-side pack-size budget fails closed (ADR-009): the breach
    /// errors the bridge *before* the budget-exceeding bytes are
    /// delivered — the ingestion sees under-budget bytes only, the io
    /// error names the budget, later reads re-report it (no re-entry),
    /// and the typed failure is the client-fault budget class
    /// ([`SubstrateError::Protocol`]).
    #[test]
    fn the_pack_size_budget_fails_closed_before_the_breach_bytes() {
        let rt = multi_thread();
        rt.block_on(async {
            let data: Vec<u8> = (0..10_000u32).map(|i| (i % 251) as u8).collect();
            let chunks = data.chunks(300).map(<[u8]>::to_vec).collect();
            let limits = Limits {
                max_pack_size: 1000,
                ..Limits::default()
            };
            let bridge = bridge_pack_source(tail(chunks), &limits);
            let PackSourceBridge {
                mut end,
                fault,
                relay,
            } = bridge;

            let mut before = Vec::new();
            let mut got = vec![0u8; 4096];
            loop {
                match end.read(&mut got) {
                    Ok(0) => break,
                    Ok(n) => before.extend_from_slice(&got[..n]),
                    Err(_) => break,
                }
            }
            assert!(
                !before.is_empty() && before.len() <= 1000,
                "some under-budget bytes delivered, none past the line: {}",
                before.len()
            );
            let error = end.read(&mut got).unwrap_err();
            assert!(
                error.to_string().contains("max_pack_size"),
                "the io error names the budget: {error}"
            );
            let error = end.read(&mut got).unwrap_err();
            assert!(error.to_string().contains("max_pack_size"));
            let failure = fault.take_failure().expect("the typed failure landed");
            match failure {
                SubstrateError::Protocol(message) => {
                    assert!(message.contains("max_pack_size"), "{message}");
                }
                other => panic!("the budget breach is the client-fault class: {other:?}"),
            }
            timeout(Duration::from_secs(5), relay)
                .await
                .expect("the relay ended at the budget breach")
                .unwrap();
        });
    }

    /// A transport failure (the tail errored mid-stream) surfaces on
    /// the blocking end as its io error and breaks the relay; the typed
    /// failure is [`SubstrateError::Io`].
    #[test]
    fn a_transport_failure_surfaces_on_the_blocking_end() {
        let rt = multi_thread();
        rt.block_on(async {
            let data: Vec<u8> = vec![b'x'; 2048];
            let bridge = bridge_pack_source(failing_tail(vec![data]), &Limits::default());
            let PackSourceBridge {
                mut end,
                fault,
                relay,
            } = bridge;

            let mut got = vec![0u8; 4096];
            match end.read(&mut got) {
                Ok(n) if n <= 2048 && got[..n].iter().all(|b| *b == b'x') => {}
                Ok(n) => panic!("unexpected clean read before the failure: {n}"),
                Err(_) => {}
            }
            let error = end
                .read(&mut got)
                .expect_err("the transport failure surfaces on the blocking end");
            assert!(error.to_string().contains("the body failed"), "{error}");
            let failure = fault.take_failure().expect("the typed failure landed");
            match failure {
                SubstrateError::Io(message) => {
                    assert!(
                        message.to_string().contains("the body failed"),
                        "{message:?}"
                    );
                }
                other => panic!("a transport failure records as Io: {other:?}"),
            }
            timeout(Duration::from_secs(5), relay)
                .await
                .expect("the relay ended at the failure")
                .unwrap();
        });
    }

    /// An empty tail EOFs immediately: Ok(0), the relay completes
    /// without any failure.
    #[test]
    fn an_empty_tail_reads_as_clean_eof() {
        let rt = multi_thread();
        rt.block_on(async {
            let bridge = bridge_pack_source(tail(Vec::new()), &Limits::default());
            let PackSourceBridge {
                mut end,
                fault,
                relay,
            } = bridge;
            let mut got = [0u8; 16];
            assert_eq!(end.read(&mut got).unwrap(), 0);
            relay.await.unwrap();
            assert!(fault.take_failure().is_none());
        });
    }

    /// The relay's reads cap at [`PACK_SOURCE_CHUNK_SIZE`] (the buffer
    /// bound): the delivered chunks never exceed the cap, even when the
    /// underlying reader offers one huge chunk.
    #[test]
    fn the_relay_chunks_at_the_cap() {
        let rt = multi_thread();
        rt.block_on(async {
            let data: Vec<u8> = vec![0u8; PACK_SOURCE_CHUNK_SIZE * 4 + 33];
            let bridge = bridge_pack_source(tail(vec![data.clone()]), &Limits::default());
            let PackSourceBridge { mut end, relay, .. } = bridge;
            let mut total = 0u64;
            let mut biggest = 0usize;
            let mut got = vec![0u8; 1 << 20];
            loop {
                let n = end.read(&mut got).unwrap();
                if n == 0 {
                    break;
                }
                biggest = biggest.max(n);
                total += n as u64;
            }
            assert_eq!(total, data.len() as u64);
            assert!(
                biggest <= PACK_SOURCE_CHUNK_SIZE,
                "each delivered chunk respects the cap: {biggest}"
            );
            relay.await.unwrap();
        });
    }
}