diff --git a/Cargo.lock b/Cargo.lock index d0697d5..243a6a1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -324,7 +324,9 @@ dependencies = [ "axum", "axum-extra", "clap", + "futures-util", "getrandom 0.3.4", + "http-body", "http-body-util", "jiff", "mime_guess", diff --git a/crates/consolebook-server/Cargo.toml b/crates/consolebook-server/Cargo.toml index 5cfbfa5..a808df5 100644 --- a/crates/consolebook-server/Cargo.toml +++ b/crates/consolebook-server/Cargo.toml @@ -11,6 +11,13 @@ anyhow = "1" argon2 = "0.5" axum = "0.8" axum-extra = { version = "0.10", features = ["cookie"] } +# Streaming the export reads rows one at a time (`TryStreamExt::try_next`) +# so a unit's bytes are released as it is written (#47). +futures-util = { version = "0.3", default-features = false, features = ["std"] } +# The streamed export's response body: an incremental body over a bounded +# channel, so delivery never buffers the archive. +http-body = "1" + clap = { version = "4", features = ["derive", "env"] } getrandom = "0.3" mime_guess = "2" diff --git a/crates/consolebook-server/src/export_stream.rs b/crates/consolebook-server/src/export_stream.rs new file mode 100644 index 0000000..f086212 --- /dev/null +++ b/crates/consolebook-server/src/export_stream.rs @@ -0,0 +1,210 @@ +//! A `Write + Seek` sink that lets the export archive stream to a +//! non-seekable destination while staying byte-identical to the buffered +//! archive (#47). +//! +//! `docs/formats/record-export.md` fixes the container's bytes: stored +//! entries, a fixed entry order, and every entry's modification time and +//! permissions. A `zip` writer over a non-seekable sink would not produce +//! those bytes — it cannot go back and fill in an entry's CRC-32 and +//! sizes, so it writes zeros in the local header, sets the data-descriptor +//! flag, and appends a 16-byte descriptor after the payload +//! (`zip::write::ZipWriter::new_stream`). Verification would still accept +//! such an archive, but byte compatibility with the shipped exporter would +//! be gone, and issue #47 requires both. +//! +//! Writing one entry with `ZipWriter::new` produces this sequence: +//! +//! ```text +//! seek(stream_position) the entry's header_start +//! write(local header) magic, fields, name, extra field +//! write(payload) the record's stored bytes +//! seek(header_start + 14) back to the CRC-32 field +//! write(crc32) write(sizes) the patch, three short writes +//! seek(file_end) forward again, to the archive's end +//! ``` +//! +//! A patch reaches into bytes the destination has already been given, so +//! the entry it belongs to cannot be handed on before the patch is known. +//! This sink therefore holds the entry being written — its local header +//! and its payload — and applies a patch to it in place: +//! +//! - an entry begins when a write starts with the ZIP local-header magic; +//! the previous entry is complete by then and is released; +//! - a seek that lands **inside the entry being held** is a patch, in +//! whichever direction it came from: the CRC-32 and size patch seeks +//! backwards to the local header's fixed fields, and a ZIP64 entry's +//! extra-field update seeks forwards to it. Both write into the held +//! bytes at the position asked for; +//! - a seek to or past the entry's end closes it and hands it to the +//! destination as one append. +//! +//! Peak memory is one entry — a record's stored bytes and its local +//! header — plus, while the container finalizes, the central directory. +//! Both terms are named in ADR 0014's costs; neither is the corpus, and +//! the entry term is bounded by the largest stored record rather than by +//! the history's size. + +use std::io::{self, Seek, SeekFrom, Write}; + +/// The first bytes of a ZIP local file header (APPNOTE 6.3), which mark +/// where one entry's bytes end and the next begins. +const LOCAL_HEADER_MAGIC: [u8; 4] = [0x50, 0x4b, 0x03, 0x04]; + +/// A seekable view of an append-only `Write` sink: one entry of memory, +/// real positions, and no rewinding. +pub struct EntryBuffer { + sink: W, + /// The entry being written: its local header, then its payload. + entry: Vec, + /// Where the entry being written began. + entry_at: u64, + /// Whether an entry is open. + open: bool, + /// Where the writer's next write logically lands. + cursor: u64, + /// The destination's end: everything already sent. + delivered: u64, + /// Whether the writer is writing into the open entry rather than + /// appending to it: the local-header or ZIP64-extra-field patch. + patching: bool, + /// The most bytes one entry ever held. + peak_entry: usize, +} + +impl EntryBuffer { + /// Wraps `sink`. + #[must_use] + pub fn new(sink: W) -> Self { + Self { + sink, + entry: Vec::new(), + entry_at: 0, + open: false, + cursor: 0, + delivered: 0, + patching: false, + peak_entry: 0, + } + } + + /// The most bytes held for one entry: its local header and payload, + /// and at finalization the central directory that follows the last + /// entry. `0` before anything is written. + #[must_use] + pub fn peak_entry_bytes(&self) -> usize { + self.peak_entry + } + + fn remember_peak(&mut self) { + self.peak_entry = self.peak_entry.max(self.entry.len()); + } + + /// Hands the open entry to the destination and closes it. + fn release_entry(&mut self) -> io::Result<()> { + if !self.open { + return Ok(()); + } + self.remember_peak(); + let entry = std::mem::take(&mut self.entry); + // The destination only ever appends; a completed entry is written + // where its stream has reached, which is where it began. + self.sink.seek(SeekFrom::Start(self.delivered))?; + self.sink.write_all(&entry)?; + self.delivered = self.entry_at + entry.len() as u64; + self.patching = false; + self.open = false; + Ok(()) + } + + /// The underlying sink, once the container is complete. An entry left + /// open by an unfinished container is released first, so no buffered + /// byte is dropped silently. + pub fn into_inner(mut self) -> io::Result { + self.release_entry()?; + Ok(self.sink) + } +} + +impl Write for EntryBuffer { + fn write(&mut self, buf: &[u8]) -> io::Result { + if self.patching { + // The writer sought into the entry it is still writing; its + // bytes belong at the position it asked for, which the entry + // already reaches or is grown to reach. + let offset = + usize::try_from(self.cursor.saturating_sub(self.entry_at)).unwrap_or(usize::MAX); + let end = offset.checked_add(buf.len()).ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "entry position overflow") + })?; + if end > self.entry.len() { + self.entry.resize(end, 0); + } + self.entry[offset..end].copy_from_slice(buf); + self.cursor += buf.len() as u64; + self.remember_peak(); + return Ok(buf.len()); + } + if !self.open || (self.cursor >= self.delivered && buf.starts_with(&LOCAL_HEADER_MAGIC)) { + // A local file header begins an entry; the previous one is + // complete now and goes out. + self.release_entry()?; + self.open = true; + self.entry_at = self.cursor; + } + self.entry.extend_from_slice(buf); + self.cursor += buf.len() as u64; + self.remember_peak(); + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + self.sink.flush() + } + + fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result { + let mut total = 0; + for buf in bufs { + total += self.write(buf)?; + } + Ok(total) + } +} + +impl Seek for EntryBuffer { + fn seek(&mut self, pos: SeekFrom) -> io::Result { + match pos { + SeekFrom::Start(at) => { + let entry_end = self.entry_at.saturating_add(self.entry.len() as u64); + // Inside the open entry is where the container's patches + // land; anywhere else is an ordinary move. + self.patching = self.open && at >= self.entry_at && at < entry_end; + self.cursor = at; + Ok(at) + } + SeekFrom::Current(by) => { + let at = i64::try_from(self.cursor) + .ok() + .and_then(|base| base.checked_add(by)) + .filter(|at| *at >= 0) + .ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "negative position") + })?; + self.seek(SeekFrom::Start(at.cast_unsigned())) + } + SeekFrom::End(by) => { + if by != 0 { + return Err(io::Error::new( + io::ErrorKind::Unsupported, + "seeking backwards from the end of a streamed archive", + )); + } + // The container is complete: the last entry — which by now + // carries the central directory and the end record — goes + // out, and the stream ends where the archive does. + self.release_entry()?; + self.cursor = self.delivered; + Ok(self.delivered) + } + } + } +} diff --git a/crates/consolebook-server/src/exports_http.rs b/crates/consolebook-server/src/exports_http.rs index 12ff5ce..0cd8fd5 100644 --- a/crates/consolebook-server/src/exports_http.rs +++ b/crates/consolebook-server/src/exports_http.rs @@ -5,16 +5,53 @@ //! `record_export`; handlers translate refusals into stable error codes //! and deliver the documented archive bytes as attachments. +use std::io::Write as _; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::task::{Context as TaskContext, Poll}; +use std::time::Duration; + use axum::Router; +use axum::body::{Body, Bytes}; use axum::extract::{Path, State}; use axum::http::{StatusCode, header}; use axum::response::{IntoResponse, Json, Response}; use axum::routing::get; +use http_body::Frame; +use time::OffsetDateTime; +use tokio::sync::{mpsc, oneshot}; +use crate::export_stream::EntryBuffer; use crate::http::{ApiError, AppState, CurrentUser}; use crate::record_export::{self, ExportRefusal, Scope}; use crate::trainee_packet::{self, PacketRefusal}; +/// How many chunks may wait between the archive's producer and the +/// response body. The bound is the backpressure: a client that stops +/// reading blocks the producer at this many chunks rather than letting it +/// run ahead, and a client that disconnects closes the channel, which +/// stops the producer and releases its connection and read transaction. +const BODY_CHUNKS: usize = 8; +/// How long one chunk may wait for room in the response queue, and how long +/// the failure detail may wait for room of its own. The queue is full for +/// this long only when the client has stopped reading altogether, so +/// reaching it ends the export: the transfer is ended without its final +/// chunk, which the client reads as an incomplete download rather than as a +/// complete export. A client that keeps making progress — even slowly — +/// never reaches it. +/// +/// Public because a transport-level test has to outlast both windows to +/// prove that losing the failure detail still loses the transfer; the unit +/// tests inject their own, much shorter, deadlines instead. +pub const EXPORT_STALL_LIMIT: Duration = Duration::from_secs(10); +/// How long the preflight may take before the request fails. This covers +/// authorization, the metadata pass, and the audit: the time before the +/// response can still carry a typed answer. It deliberately excludes the +/// payload pass, which has no deadline of its own beyond the per-chunk +/// stall guard. +const PREFLIGHT_LIMIT: Duration = Duration::from_secs(30); + pub(crate) fn routes() -> Router { Router::new() .route("/api/drafts/{id}/export", get(export_record)) @@ -54,24 +91,391 @@ fn export_refusal(refusal: ExportRefusal) -> ApiError { "nothing_to_export", "this scope holds no finalized version; an export never claims completeness it lacks", ), + ExportRefusal::ExportFailed => ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "export_failed", + "the export could not be produced or delivered; nothing complete was sent", + ), } } -/// The archive as a download: exactly the documented bytes. +/// The archive as a download, streamed: the documented bytes are written +/// to the response as they are produced, so the whole corpus of stored +/// payloads is never held (#47; ADR 0014 costs). What still scales with +/// the installation is its unit count, not its bytes: the entry metadata +/// the archive manifest lists, the serialized manifest itself, and the +/// container's central directory are each O(units), and the database +/// driver buffers a bounded number of rows per query. The browser's own +/// download helper buffers the response it saves, separately. +/// +/// A typed refusal is answered before the response starts. After it +/// starts, the body carries the archive and nothing else: a failure has +/// no status line left to change, so it ends the transfer without its +/// final chunk. The client reads that as an incomplete download, never as +/// a complete export — the bytes of a complete archive are exactly the +/// bytes the container writer produced, and a truncated stream cannot be +/// mistaken for them. async fn deliver(state: &AppState, actor_user_id: i64, scope: Scope) -> Result { - match record_export::export(&state.pool, actor_user_id, scope).await? { - Ok(export) => { - let disposition = format!("attachment; filename=\"{}\"", export.file_name); - Ok(( - [ - (header::CONTENT_TYPE, "application/zip".to_owned()), - (header::CONTENT_DISPOSITION, disposition), - ], - export.bytes, + let exported_at = OffsetDateTime::now_utc().unix_timestamp(); + let (started_tx, started_rx) = oneshot::channel(); + let (body_tx, body_rx) = mpsc::channel::(BODY_CHUNKS); + let failure_tx = body_tx.clone(); + let completion = Completion::new(); + let producer_completion = completion.clone(); + let pool = state.pool.clone(); + // The archive is produced on a blocking thread: it streams SQLite rows + // and writes ZIP bytes, and a slow client must not stall the runtime. + // The handle is dropped: the response body drives the producer, and a + // `spawn_blocking` task cannot be cancelled once it runs. + let _work = tokio::task::spawn_blocking(move || { + // The container writer patches each entry's local header after its + // payload, so it writes through the sink that holds the entry and + // hands the destination an append-only stream (#47). + let sink = EntryBuffer::new(ChunkSink::new(body_tx)); + let runtime = tokio::runtime::Handle::current(); + let mut started = Some(started_tx); + // A producer that goes away without finishing — a panic, or a + // return this code forgot — must not leave the body to end + // cleanly, which would look like a complete download. + let mut guard = FailureGuard::new(failure_tx.clone()); + let produced = runtime.block_on(record_export::export_to( + &pool, + actor_user_id, + scope, + exported_at, + sink, + |produced| { + // The scope is authorized, read, and audited: the response + // may start, and only a failure before this point can still + // reach the client as a typed answer. + if let Some(started) = started.take() { + drop(started.send(Ok(produced))); + } + guard.arm(); + }, + )); + // The destination's tail is handed on here, on the thread that owns + // it: the archive is complete only once every byte is out. + let (outcome, failure) = match produced { + Ok((outcome, mut sink)) => { + let flushed = sink.flush(); + drop(sink); + (outcome, flushed.err().map(|err| err.to_string())) + } + Err(err) => { + // A client that left is not a production failure: nobody is + // left to read a signal, and the body simply closes. Any + // other failure — including a client that stopped reading + // altogether — ends the transfer incomplete, so it is never + // saved as a whole export (#47). + let client_gone = err + .chain() + .filter_map(|cause| cause.downcast_ref::()) + .any(|io| { + matches!( + io.kind(), + std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionAborted + ) + }); + if !client_gone { + tracing::error!(error = %err, "producing a record export failed"); + } + ( + Err(ExportRefusal::ExportFailed), + (!client_gone).then(|| err.to_string()), + ) + } + }; + // A typed refusal that never started a response is the caller's to + // answer. A failure after the response started has no status line + // left, so it goes to the body as the signal that ends the transfer + // without its final chunk: a send that is awaited rather than + // dropped, because a dropped future would never travel and the + // download would end cleanly. + if let (Err(refusal), Some(started)) = (&outcome, started.take()) { + drop(started.send(Err(*refusal))); + } + // Success is a fact only when the archive was produced and its tail + // flushed. Everything else leaves the completion unachieved, so the + // body fails closed when the channel closes, whatever the queue had + // room for. + let complete = outcome.is_ok() && failure.is_none(); + if let Some(detail) = failure { + signal_failure(&failure_tx, detail); + } + if complete { + producer_completion.achieved(); + } + guard.disarm(); + // The receiver is gone when the client left before the response + // started; the export is over either way. + drop(started); + }); + let produced = match tokio::time::timeout(PREFLIGHT_LIMIT, started_rx).await { + Ok(Ok(Ok(produced))) => produced, + Ok(Ok(Err(refusal))) => return Err(export_refusal(refusal)), + Ok(Err(_)) | Err(_) => { + // A `spawn_blocking` task cannot be cancelled once it runs; the + // producer stops on its own at its first send, because the + // response body's receiver is dropped with this return. + return Err(ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "export_failed", + "the export could not be produced", + )); + } + }; + let disposition = format!("attachment; filename=\"{}\"", produced.file_name); + Response::builder() + .header(header::CONTENT_TYPE, "application/zip") + .header(header::CONTENT_DISPOSITION, disposition) + .body(Body::new(ExportBody { + body: body_rx, + completion, + })) + .map_err(|err| { + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "export_failed", + format!("the export response could not be built: {err}"), ) - .into_response()) + }) +} + +/// One message of a streamed export body: bytes, or the failure that ends +/// the download short of a complete archive. +enum StreamItem { + Chunk(Bytes), + /// The failure that ended the archive short, when the queue had room + /// for it. Only the detail: whether the transfer was complete is + /// [`Completion`]'s answer, which needs no room at all. + Failed(String), +} + +/// Whether the producer finished a complete archive. +/// +/// This is deliberately not the channel: a failure that cannot be queued — +/// because the client stopped reading and the queue is full — would leave +/// channel closure indistinguishable from success, and a client that +/// resumes reading would be handed a truncated prefix that looks complete. +/// Only an explicit success permits a clean end of body; anything else +/// fails closed. +#[derive(Clone)] +struct Completion(Arc); + +impl Completion { + fn new() -> Self { + Self(Arc::new(AtomicBool::new(false))) + } + + /// The archive was produced and its tail flushed. + fn achieved(&self) { + self.0.store(true, Ordering::SeqCst); + } + + fn is_achieved(&self) -> bool { + self.0.load(Ordering::SeqCst) + } +} + +/// Hands the failure to the response body, so the transfer ends without +/// its final chunk rather than looking complete. This is a blocking send +/// with a bound: the producer is already on a blocking thread, and the +/// alternative — an unpolled `send` future — would drop the signal and end +/// the download cleanly. +fn signal_failure(sender: &mpsc::Sender, detail: String) { + let deadline = std::time::Instant::now() + EXPORT_STALL_LIMIT; + if send_before(sender, StreamItem::Failed(detail), deadline).is_err() { + tracing::error!("the export's failure could not reach the response body"); + } +} + +/// Why one item never reached the response body. +enum SendOutcome { + /// The receiver is gone: the client left. + ClientGone, + /// The queue stayed full until the deadline: the client stopped + /// reading, so the transfer is over either way. + Stalled, +} + +/// Hands one item to the response body, waiting for room until `deadline`. +/// A full queue means a client that is not reading; the wait is a sleep on +/// a blocking thread, so no runtime worker is stalled and no queue grows. +fn send_before( + sender: &mpsc::Sender, + mut item: StreamItem, + deadline: std::time::Instant, +) -> std::result::Result<(), SendOutcome> { + loop { + item = match sender.try_send(item) { + Ok(()) => return Ok(()), + Err(mpsc::error::TrySendError::Closed(_)) => return Err(SendOutcome::ClientGone), + Err(mpsc::error::TrySendError::Full(item)) => item, + }; + if std::time::Instant::now() >= deadline { + return Err(SendOutcome::Stalled); + } + std::thread::sleep(Duration::from_millis(1)); + } +} + +/// Ends a started download incomplete when the producer goes away without +/// completing it — the path a panic takes, where dropping the channel +/// alone would look like a clean end of body. +struct FailureGuard { + sender: mpsc::Sender, + armed: bool, +} + +impl FailureGuard { + fn new(sender: mpsc::Sender) -> Self { + Self { + sender, + armed: false, + } + } + + /// Arms the guard once the response has started, so only a failure the + /// client can still see is signalled. + fn arm(&mut self) { + self.armed = true; + } + + /// The producer reached a known end: complete, refused, or a failure + /// it signalled itself. + fn disarm(&mut self) { + self.armed = false; + } +} + +impl Drop for FailureGuard { + fn drop(&mut self) { + if self.armed { + signal_failure( + &self.sender, + "the export ended before it was complete".to_owned(), + ); + } + } +} + +/// The response body of a streamed export. A failure ends the stream +/// without a clean end-of-body, so the transfer is incomplete rather than +/// an apparently successful partial file (issue #47). +struct ExportBody { + body: mpsc::Receiver, + completion: Completion, +} + +impl http_body::Body for ExportBody { + type Data = Bytes; + type Error = std::io::Error; + + fn poll_frame( + mut self: Pin<&mut Self>, + cx: &mut TaskContext<'_>, + ) -> Poll, Self::Error>>> { + match self.body.poll_recv(cx) { + Poll::Ready(Some(StreamItem::Chunk(bytes))) => { + Poll::Ready(Some(Ok(Frame::data(bytes)))) + } + Poll::Ready(Some(StreamItem::Failed(detail))) => Poll::Ready(Some(Err( + std::io::Error::other(format!("the export could not be completed: {detail}")), + ))), + // The stream ended. Unless the producer said it finished a + // complete archive, that end is a failure, however clean the + // channel closure looked. + Poll::Ready(None) => Poll::Ready(if self.completion.is_achieved() { + None + } else { + Some(Err(std::io::Error::other( + "the export ended before it was complete", + ))) + }), + Poll::Pending => Poll::Pending, + } + } +} + +/// The archive's destination: an append-only stream of numbered pieces +/// over the response body. The container writer hands it finished entries, +/// so it never rewrites a byte; the channel's bound is the backpressure, +/// and a closed channel ends the export instead of buffering it for nobody. +struct ChunkSink { + sender: mpsc::Sender, + /// How many bytes have been handed to the response: the position the + /// destination has reached. + sent: u64, +} + +impl ChunkSink { + fn new(sender: mpsc::Sender) -> Self { + Self { sender, sent: 0 } + } + + /// Hands one chunk to the response body, waiting while the queue is + /// full so the producer stays inside the channel's bound rather than + /// growing a queue or stalling a runtime worker. The wait is bounded: + /// a client that stops reading altogether loses the transfer, which it + /// reads as an incomplete download, and a client that keeps making + /// progress — however slowly — never notices the bound. + fn send_bytes(&mut self, bytes: Vec) -> std::io::Result<()> { + let deadline = std::time::Instant::now() + EXPORT_STALL_LIMIT; + self.send_within(bytes, deadline) + } + + /// The same hand-off with the caller's deadline, so the stall bound is + /// the caller's to choose. + fn send_within(&mut self, bytes: Vec, deadline: std::time::Instant) -> std::io::Result<()> { + let size = bytes.len() as u64; + match send_before( + &self.sender, + StreamItem::Chunk(Bytes::from(bytes)), + deadline, + ) { + Ok(()) => { + self.sent += size; + Ok(()) + } + Err(SendOutcome::ClientGone) => Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "the client left", + )), + // Not the client's departure: the transfer ends incomplete and + // the client can see that it did. + Err(SendOutcome::Stalled) => Err(std::io::Error::other("the client stopped reading")), + } + } +} + +impl std::io::Write for ChunkSink { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.send_bytes(buf.to_vec())?; + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl std::io::Seek for ChunkSink { + fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result { + use std::io::SeekFrom; + match pos { + SeekFrom::Current(0) | SeekFrom::End(0) => Ok(self.sent), + SeekFrom::Start(at) if at == self.sent => Ok(at), + SeekFrom::Start(at) => Err(std::io::Error::other(format!( + "a streamed export cannot rewrite position {at}" + ))), + other => Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + format!("a streamed export is append-only ({other:?})"), + )), } - Err(refusal) => Err(export_refusal(refusal)), } } @@ -176,3 +580,141 @@ async fn my_enrollments( Err(refusal) => Err(packet_refusal(refusal)), } } + +#[cfg(test)] +mod tests { + use super::*; + use http_body::Body as _; + + /// Polls one frame of a body, so the terminal cases can be asserted. + async fn frame(body: &mut ExportBody) -> Option, std::io::Error>> { + std::future::poll_fn(|cx| Pin::new(&mut *body).poll_frame(cx)).await + } + + /// A body whose producer finished a complete archive ends cleanly. + #[tokio::test] + async fn an_achieved_completion_ends_the_body() { + let (sender, receiver) = mpsc::channel::(BODY_CHUNKS); + let completion = Completion::new(); + let mut body = ExportBody { + body: receiver, + completion: completion.clone(), + }; + sender + .send(StreamItem::Chunk(Bytes::from_static(b"PK"))) + .await + .expect("queue a chunk"); + completion.achieved(); + drop(sender); + assert!(matches!(frame(&mut body).await, Some(Ok(_)))); + assert!(frame(&mut body).await.is_none()); + } + + /// A body whose producer went away without finishing fails closed, even + /// though the channel closed exactly as a successful run's would: this + /// is what a client that stopped reading long enough to lose the + /// failure detail must still be told. + #[tokio::test] + async fn a_closed_channel_without_completion_fails_the_body() { + let (sender, receiver) = mpsc::channel::(BODY_CHUNKS); + let mut body = ExportBody { + body: receiver, + completion: Completion::new(), + }; + sender + .send(StreamItem::Chunk(Bytes::from_static(b"PK"))) + .await + .expect("queue a chunk"); + drop(sender); + assert!(matches!(frame(&mut body).await, Some(Ok(_)))); + let end = frame(&mut body).await; + assert!(matches!(end, Some(Err(_))), "{end:?}"); + } + + /// A failure after the response started must reach the body as an + /// error: a body that simply ended would look to every client like a + /// complete download of a short file (issue #47). + #[tokio::test] + async fn a_failure_item_ends_the_body_with_an_error() { + let (sender, receiver) = mpsc::channel::(BODY_CHUNKS); + let mut body = ExportBody { + body: receiver, + completion: Completion::new(), + }; + sender + .send(StreamItem::Chunk(Bytes::from_static(b"PK\x03\x04"))) + .await + .expect("queue the first chunk"); + signal_failure(&sender, "an invented production failure".to_owned()); + drop(sender); + + assert!(matches!(frame(&mut body).await, Some(Ok(_)))); + let failure = frame(&mut body).await; + assert!(matches!(failure, Some(Err(_))), "{failure:?}"); + } + + /// The guard is what a panic relies on: without it, a producer that + /// goes away mid-download would drop the channel and the body would end + /// cleanly. + #[tokio::test] + async fn a_producer_that_goes_away_after_starting_signals_a_failure() { + let (sender, mut receiver) = mpsc::channel::(BODY_CHUNKS); + let mut guard = FailureGuard::new(sender.clone()); + guard.arm(); + drop(guard); + drop(sender); + assert!(matches!(receiver.recv().await, Some(StreamItem::Failed(_)))); + } + + /// A guard that was disarmed — a complete export, a typed refusal, or a + /// failure the producer signalled itself — sends nothing. + #[tokio::test] + async fn a_disarmed_producer_sends_nothing() { + let (sender, mut receiver) = mpsc::channel::(BODY_CHUNKS); + let mut guard = FailureGuard::new(sender.clone()); + guard.arm(); + guard.disarm(); + drop(guard); + drop(sender); + assert!(receiver.recv().await.is_none()); + } + + /// A stall is not the client's departure: it must be classified as a + /// failure, because the client is still there to see the difference. + #[tokio::test] + async fn a_stalled_send_is_not_a_client_departure() { + let (sender, mut receiver) = mpsc::channel::(1); + sender + .send(StreamItem::Chunk(Bytes::from_static(b"first"))) + .await + .expect("fill the queue"); + let mut sink = ChunkSink::new(sender); + let started = std::time::Instant::now(); + let err = sink + .send_within(b"second".to_vec(), started + Duration::from_millis(50)) + .expect_err("the queue is full and nobody reads it"); + assert_ne!(err.kind(), std::io::ErrorKind::BrokenPipe, "{err:?}"); + assert!( + started.elapsed() >= Duration::from_millis(50), + "the send gave up after {:?}", + started.elapsed() + ); + assert!(receiver.try_recv().is_ok()); + } + + /// A client that left is reported as exactly that, so the producer can + /// tell "nobody is listening" from "this failed". + #[tokio::test] + async fn a_departed_client_is_a_broken_pipe() { + let (sender, receiver) = mpsc::channel::(1); + drop(receiver); + let mut sink = ChunkSink::new(sender); + let err = sink + .send_within( + b"bytes".to_vec(), + std::time::Instant::now() + Duration::from_secs(1), + ) + .expect_err("the receiver is gone"); + assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe, "{err:?}"); + } +} diff --git a/crates/consolebook-server/src/lib.rs b/crates/consolebook-server/src/lib.rs index 59ee108..13b844f 100644 --- a/crates/consolebook-server/src/lib.rs +++ b/crates/consolebook-server/src/lib.rs @@ -18,6 +18,7 @@ pub mod draft_review; pub mod drafts_http; pub mod enrollments; pub mod evaluation_drafts; +pub mod export_stream; pub mod export_verify; pub mod exports_http; pub mod finalization; diff --git a/crates/consolebook-server/src/record_export.rs b/crates/consolebook-server/src/record_export.rs index 72cb963..5146f93 100644 --- a/crates/consolebook-server/src/record_export.rs +++ b/crates/consolebook-server/src/record_export.rs @@ -13,9 +13,10 @@ //! `export_verify`'s, which reads the manifests defined here. use std::fmt; -use std::io::{Cursor, Write}; +use std::io::{Cursor, Seek, Write}; use anyhow::{Context, Result, anyhow}; +use futures_util::TryStreamExt; use serde::{Deserialize, Serialize}; use sqlx::{Row, SqliteConnection, SqlitePool}; use time::OffsetDateTime; @@ -84,6 +85,10 @@ pub enum ExportRefusal { /// The scope exists but holds no finalized version; an empty /// archive is never presented as a complete export. NothingToExport, + /// Producing or delivering the archive failed after the scope was + /// known to hold versions. This is not an empty scope, and it is + /// never presented as one. + ExportFailed, } /// One unit as the archive manifest lists it. @@ -138,6 +143,16 @@ pub struct Export { pub unit_count: usize, } +/// What a produced archive states about itself. The bytes are not here: +/// a streamed export writes them as it produces them. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ArchiveProduced { + /// The documented download name, `consolebook--.zip`. + pub file_name: String, + pub exported_at: i64, + pub unit_count: usize, +} + /// The unit directory for one version: `records/{record_id}/v{n}`. #[must_use] pub fn unit_path(record_id: i64, version_number: i64) -> String { @@ -161,52 +176,194 @@ pub async fn export( .await } -/// Exports `scope` stamped with `exported_at` (UTC unix seconds). The -/// archive is a pure function of the scope's rows and this instant. -pub async fn export_at( +/// Exports `scope` stamped with `exported_at` (UTC unix seconds), +/// writing the container into `sink` as it is produced (#47). +/// +/// The archive is a pure function of the scope's rows and this instant, +/// and its bytes are the same whichever sink receives them. Two passes +/// over one read transaction produce it: the first reads every unit's +/// stored fingerprints — not its bytes — so the archive manifest, the +/// container's first entry, can be written before any record byte is +/// read; the second streams each unit's stored bytes and its unit +/// manifest straight into the sink. What is held is the entry metadata +/// (one small record per unit, O(units), and the serialized manifest that +/// lists it), one unit's bytes at a time, and the container's central +/// directory — never the corpus of payloads. +/// +/// The export never holds more than one pooled connection at a time: the +/// audit is written in its own short write transaction, which commits +/// before the read snapshot is taken, and that snapshot is a reader from +/// beginning to end. An export therefore reserves no writer while it +/// streams (ADR 0019), and exports sharing a pool never wait for a +/// connection another export holds. +/// +/// The audit precedes both content passes, so a scope with nothing to +/// export is refused before anything is recorded, and a typed refusal +/// still reaches the caller as one. A recorded `record_exported` attests +/// that the installation produced this export from the state at this +/// instant; it does not attest that the operator received or saved the +/// file, and a delivery that fails part way leaves the record it already +/// wrote — while the client sees an incomplete transfer rather than a +/// complete export. +/// +/// `ready` is called with what the archive states about itself once the +/// scope is authorized, the rows are read, and the export is audited: a +/// streamed delivery starts its response there and reports a failure after +/// that point as an incomplete transfer, never as a complete export. +pub async fn export_to( pool: &SqlitePool, actor_user_id: i64, scope: Scope, exported_at: i64, -) -> Result> { + sink: W, + ready: F, +) -> Result<(std::result::Result, W)> { let audited = match authorize(pool, actor_user_id, scope).await? { Ok(audited) => audited, - Err(refusal) => return Ok(Err(refusal)), + // The sink comes back with the refusal so the caller can flush and + // close whatever destination it handed over. + Err(refusal) => return Ok((Err(refusal), sink)), }; - let mut conn = pool.acquire().await.context("acquiring connection")?; - let rows = collect(&mut conn, scope).await?; - drop(conn); - if rows.is_empty() { - return Ok(Err(match scope { - Scope::Version { .. } => ExportRefusal::NoSuchVersion, - Scope::Record { .. } | Scope::Enrollment { .. } | Scope::Installation => { - ExportRefusal::NothingToExport - } - })); + // The audit is written first, by itself, on one connection: it must be + // committed before the first byte, it must never be written for a scope + // that holds nothing to export, and it must not be part of the + // transaction the payloads stream from. `storage::write_tx` reserves + // the writer for the length of two short statements (ADR 0019). + { + let mut tx = storage::write_tx(pool) + .await + .context("beginning audit write")?; + if !scope_has_units(&mut tx, scope).await? { + // Nothing to export is never audited, and never exported empty. + tx.rollback().await.context("rolling back empty export")?; + return Ok((Err(nothing_to_export(scope)), sink)); + } + audit_export(&mut *tx, actor_user_id, &audited).await?; + tx.commit().await.context("committing audit write")?; } - let installation_id = storage::installation_id(pool).await?; - let unit_count = rows.len(); - let bytes = build_archive(&installation_id, exported_at, scope, rows)?; - // The export is audited once it exists: actor and subject, never - // content (docs/records-integrity.md). - match audited.subject { - Some(subject) => { - audit::record_for_subject( - pool, - EventKind::RecordExported, - Some(actor_user_id), - audited.trainee, - subject, - ) - .await?; + // One read transaction for both content passes: the manifest and the + // payloads describe the same committed state and the same export + // instant. It is a reader from beginning to end — the audit is already + // committed — so a download, however slow, reserves no writer, and the + // export holds exactly one connection throughout. + let mut tx = pool.begin().await.context("beginning export read")?; + let installation_id = storage::installation_id(&mut *tx).await?; + let read = async { + let units = collect_meta(&mut tx, scope).await?; + if units.is_empty() { + // The scope held units when it was audited, and a finalized + // version is immutable and removed only by an authorized + // disposition, which this installation does not yet perform. + // Rather than export emptiness the audit does not describe, + // the export fails: the caller may retry, and the record + // stands for the attempt. + return Err(anyhow!( + "the audited scope no longer holds a finalized version to export" + )); } - None => audit::record(pool, EventKind::RecordExported, Some(actor_user_id), None).await?, + let produced = ArchiveProduced { + file_name: file_name(scope, exported_at)?, + exported_at, + unit_count: units.len(), + }; + // The scope is authorized, the rows are read, and the export is + // recorded: the delivery may start now. + ready(produced.clone()); + let manifest = archive_manifest(&installation_id, exported_at, scope, &units); + let mut writer = ArchiveWriter::new(sink, exported_at)?; + writer.add(ARCHIVE_MANIFEST_PATH, &canonical_json(&manifest)?)?; + write_units_streaming(&mut tx, &mut writer, scope, &installation_id, exported_at).await?; + let sink = writer.into_sink()?; + Ok((produced, sink)) } - Ok(Ok(Export { - file_name: file_name(scope, exported_at)?, - bytes, + .await; + let (produced, sink) = match read { + Ok((produced, sink)) => (produced, sink), + Err(err) => return Err(err), + }; + tx.commit().await.context("ending export read")?; + Ok((Ok(produced), sink)) +} + +/// The refusal an empty scope earns: a version scope that names a missing +/// version, anything else with no finalized version at all. +fn nothing_to_export(scope: Scope) -> ExportRefusal { + match scope { + Scope::Version { .. } => ExportRefusal::NoSuchVersion, + Scope::Record { .. } | Scope::Enrollment { .. } | Scope::Installation => { + ExportRefusal::NothingToExport + } + } +} + +/// Whether the scope holds at least one finalized version, read in the +/// caller's transaction so the answer and the audit below it describe one +/// state. +async fn scope_has_units(conn: &mut SqliteConnection, scope: Scope) -> Result { + let found: Option = match scope { + Scope::Version { + record_id, + version_number, + } => sqlx::query_scalar( + "SELECT v.id FROM evaluation_version v + WHERE v.evaluation_record_id = ?1 AND v.version_number = ?2 LIMIT 1", + ) + .bind(record_id) + .bind(version_number) + .fetch_optional(&mut *conn) + .await + .context("checking the version scope")?, + Scope::Record { record_id } => sqlx::query_scalar( + "SELECT v.id FROM evaluation_version v + WHERE v.evaluation_record_id = ?1 LIMIT 1", + ) + .bind(record_id) + .fetch_optional(&mut *conn) + .await + .context("checking the record scope")?, + Scope::Enrollment { enrollment_id } => sqlx::query_scalar( + "SELECT v.id FROM evaluation_version v + JOIN evaluation_record r ON r.id = v.evaluation_record_id + WHERE r.enrollment_id = ?1 LIMIT 1", + ) + .bind(enrollment_id) + .fetch_optional(&mut *conn) + .await + .context("checking the enrollment scope")?, + Scope::Installation => sqlx::query_scalar("SELECT v.id FROM evaluation_version v LIMIT 1") + .fetch_optional(&mut *conn) + .await + .context("checking the installation scope")?, + }; + Ok(found.is_some()) +} + +/// Exports `scope` stamped with `exported_at` (UTC unix seconds). The +/// archive is a pure function of the scope's rows and this instant. +pub async fn export_at( + pool: &SqlitePool, + actor_user_id: i64, + scope: Scope, + exported_at: i64, +) -> Result> { + let (produced, bytes) = export_to( + pool, + actor_user_id, + scope, exported_at, - unit_count, + Cursor::new(Vec::new()), + |_| {}, + ) + .await?; + let produced = match produced { + Ok(produced) => produced, + Err(refusal) => return Ok(Err(refusal)), + }; + Ok(Ok(Export { + file_name: produced.file_name, + bytes: bytes.into_inner(), + exported_at: produced.exported_at, + unit_count: produced.unit_count, })) } @@ -247,6 +404,43 @@ struct Audited { trainee: Option, } +/// Records one export in the caller's transaction. +/// +/// Deliberately not the transaction the payloads stream from: an audit row +/// committed with the payload pass would hold a write reservation for the +/// whole download (ADR 0019). The caller runs this in its own short write +/// transaction, before the read snapshot is taken, so the record is +/// committed before the first byte — and a scope that holds nothing to +/// export never reaches it. +async fn audit_export<'e>( + executor: impl sqlx::Executor<'e, Database = sqlx::Sqlite>, + actor_user_id: i64, + audited: &Audited, +) -> Result<()> { + match audited.subject { + Some(subject) => { + audit::record_for_subject( + executor, + EventKind::RecordExported, + Some(actor_user_id), + audited.trainee, + subject, + ) + .await?; + } + None => { + audit::record( + executor, + EventKind::RecordExported, + Some(actor_user_id), + None, + ) + .await?; + } + } + Ok(()) +} + /// The scope's read rule, as the typed contract it already is elsewhere /// (ADR 0010): the record read rule for a version or record, the /// training-history read rule for an enrollment, and the explicit @@ -336,6 +530,24 @@ macro_rules! unit_query { }; } +/// The same rows and order, without the stored bytes: what the archive +/// manifest needs and nothing that would make the metadata pass read the +/// corpus. +macro_rules! unit_meta_query { + ($where:literal) => { + concat!( + "SELECT v.evaluation_record_id AS record_id, v.version_number, + v.record_schema, v.content_hash, + v.chain_hash, p.content_hash AS predecessor_content_hash + FROM evaluation_version v + LEFT JOIN evaluation_version p ON p.id = v.predecessor_id + JOIN evaluation_record r ON r.id = v.evaluation_record_id ", + $where, + " ORDER BY v.evaluation_record_id, v.version_number" + ) + }; +} + pub(crate) async fn collect(conn: &mut SqliteConnection, scope: Scope) -> Result> { let rows = match scope { Scope::Version { @@ -394,44 +606,174 @@ pub(crate) fn unit_entries(rows: &[VersionRow]) -> Vec { .collect() } -/// Writes the container exactly as docs/formats/record-export.md lays -/// it out: manifest first, then units in order, stored entries, the -/// export instant as every entry's modification time, `0644`. Rows are -/// consumed so each version's bytes are released once written; the -/// archive is the one copy held to the end (#47 tracks streaming it). -fn build_archive( +/// One stored version's metadata, without its bytes: everything both +/// manifests need. The export reads this first so the archive manifest — +/// which must be the container's first entry — can be written before any +/// record byte is read, while the payloads themselves stay out of memory +/// (#47). +type UnitMeta = UnitEntry; + +/// The metadata pass: every unit of the scope in archive order, with the +/// stored fingerprints but not the stored bytes. The metadata columns are +/// selected on their own — a `canonical_bytes` read here would double the +/// corpus I/O and hold a blob at a time for a length and a CRC-32 the +/// container writer computes for itself. Rows are consumed one at a time, +/// so what this pass holds is the one small record per unit that the +/// archive manifest has to carry. +pub(crate) async fn collect_meta( + conn: &mut SqliteConnection, + scope: Scope, +) -> Result> { + let mut rows = match scope { + Scope::Version { + record_id, + version_number, + } => sqlx::query(unit_meta_query!( + "WHERE v.evaluation_record_id = ?1 AND v.version_number = ?2" + )) + .bind(record_id) + .bind(version_number) + .fetch(&mut *conn), + Scope::Record { record_id } => { + sqlx::query(unit_meta_query!("WHERE v.evaluation_record_id = ?1")) + .bind(record_id) + .fetch(&mut *conn) + } + Scope::Enrollment { enrollment_id } => { + sqlx::query(unit_meta_query!("WHERE r.enrollment_id = ?1")) + .bind(enrollment_id) + .fetch(&mut *conn) + } + Scope::Installation => sqlx::query(unit_meta_query!("")).fetch(&mut *conn), + }; + let mut units = Vec::new(); + while let Some(row) = rows + .try_next() + .await + .context("reading finalized versions")? + { + units.push(UnitMeta { + path: unit_path(row.get("record_id"), row.get("version_number")), + record_id: row.get("record_id"), + version_number: row.get("version_number"), + record_schema: row.get("record_schema"), + content_hash: row.get("content_hash"), + chain_hash: row.get("chain_hash"), + predecessor_content_hash: row.get("predecessor_content_hash"), + }); + } + Ok(units) +} + +/// The payload pass: streams each unit's stored bytes and its unit +/// manifest into an already-started archive, in archive order. Peak +/// memory is one version's bytes at a time. +pub(crate) async fn write_units_streaming( + conn: &mut SqliteConnection, + writer: &mut ArchiveWriter, + scope: Scope, + installation_id: &str, + exported_at: i64, +) -> Result<()> { + let mut rows = match scope { + Scope::Version { + record_id, + version_number, + } => sqlx::query(unit_query!( + "WHERE v.evaluation_record_id = ?1 AND v.version_number = ?2" + )) + .bind(record_id) + .bind(version_number) + .fetch(&mut *conn), + Scope::Record { record_id } => { + sqlx::query(unit_query!("WHERE v.evaluation_record_id = ?1")) + .bind(record_id) + .fetch(&mut *conn) + } + Scope::Enrollment { enrollment_id } => { + sqlx::query(unit_query!("WHERE r.enrollment_id = ?1")) + .bind(enrollment_id) + .fetch(&mut *conn) + } + Scope::Installation => sqlx::query(unit_query!("")).fetch(&mut *conn), + }; + while let Some(row) = rows + .try_next() + .await + .context("reading finalized versions")? + { + let record_id: i64 = row.get("record_id"); + let version_number: i64 = row.get("version_number"); + let bytes: Vec = row.get("canonical_bytes"); + let path = unit_path(record_id, version_number); + writer.add(&format!("{path}/{RECORD_FILE}"), &bytes)?; + drop(bytes); + let unit = UnitManifest { + format: UNIT_FORMAT.to_owned(), + format_version: FORMAT_VERSION, + installation_id: installation_id.to_owned(), + exported_at, + record_id, + version_number, + record_schema: row.get("record_schema"), + content_hash: row.get("content_hash"), + chain_hash: row.get("chain_hash"), + predecessor_content_hash: row.get("predecessor_content_hash"), + }; + writer.add( + &format!("{path}/{UNIT_MANIFEST_FILE}"), + &canonical_json(&unit)?, + )?; + } + Ok(()) +} + +/// The archive manifest for `units`, in the container's entry order. +pub(crate) fn archive_manifest( installation_id: &str, exported_at: i64, scope: Scope, - rows: Vec, -) -> Result> { - let units = unit_entries(&rows); - let manifest = ArchiveManifest { + units: &[UnitEntry], +) -> ArchiveManifest { + ArchiveManifest { format: ARCHIVE_FORMAT.to_owned(), format_version: FORMAT_VERSION, installation_id: installation_id.to_owned(), exported_at, scope, - units, - }; - let mut writer = ArchiveWriter::new(exported_at)?; - writer.add(ARCHIVE_MANIFEST_PATH, &canonical_json(&manifest)?)?; - writer.add_units(installation_id, exported_at, rows, &manifest.units)?; - writer.finish() + units: units.to_vec(), + } } /// The container writer every export shares: stored entries, the export /// instant as each entry's modification time, `0644`, entries in the -/// order they are added. -pub(crate) struct ArchiveWriter { - writer: ZipWriter>>, +/// order they are added. The sink need only be seekable; a streamed +/// export passes [`crate::export_stream::EntryBuffer`], which holds one +/// local header at a time and yields the same bytes (#47). +pub(crate) struct ArchiveWriter { + writer: ZipWriter, options: SimpleFileOptions, } -impl ArchiveWriter { - pub(crate) fn new(exported_at: i64) -> Result { +impl ArchiveWriter>> { + /// The in-memory writer the packet and the compatibility tests use. + pub(crate) fn in_memory(exported_at: i64) -> Result { + Self::new(Cursor::new(Vec::new()), exported_at) + } + + pub(crate) fn finish(self) -> Result> { + let cursor = self + .writer + .finish() + .context("finishing the export archive")?; + Ok(cursor.into_inner()) + } +} + +impl ArchiveWriter { + pub(crate) fn new(sink: W, exported_at: i64) -> Result { Ok(Self { - writer: ZipWriter::new(Cursor::new(Vec::new())), + writer: ZipWriter::new(sink), options: SimpleFileOptions::default() .compression_method(CompressionMethod::Stored) .last_modified_time(dos_time(exported_at)?) @@ -439,6 +781,11 @@ impl ArchiveWriter { }) } + /// Finishes the container, returning the sink it wrote into. + pub(crate) fn into_sink(self) -> Result { + self.writer.finish().context("finishing the export archive") + } + pub(crate) fn add(&mut self, name: &str, bytes: &[u8]) -> Result<()> { self.writer .start_file(name, self.options) @@ -479,14 +826,6 @@ impl ArchiveWriter { } Ok(()) } - - pub(crate) fn finish(self) -> Result> { - let cursor = self - .writer - .finish() - .context("finishing the export archive")?; - Ok(cursor.into_inner()) - } } /// Manifests are canonical JSON under the record format's subset, so diff --git a/crates/consolebook-server/src/trainee_packet.rs b/crates/consolebook-server/src/trainee_packet.rs index 54cf255..5c83826 100644 --- a/crates/consolebook-server/src/trainee_packet.rs +++ b/crates/consolebook-server/src/trainee_packet.rs @@ -601,7 +601,7 @@ pub async fn export_at( }) .collect(), }; - let mut writer = ArchiveWriter::new(exported_at)?; + let mut writer = ArchiveWriter::in_memory(exported_at)?; writer.add(ARCHIVE_MANIFEST_PATH, &canonical_json(&manifest)?)?; writer.add_units(&installation_id, exported_at, rows, &manifest.units)?; for (kind, bytes) in &documents { diff --git a/crates/consolebook-server/tests/export_stream_bytes.rs b/crates/consolebook-server/tests/export_stream_bytes.rs new file mode 100644 index 0000000..ea586e6 --- /dev/null +++ b/crates/consolebook-server/tests/export_stream_bytes.rs @@ -0,0 +1,1247 @@ +//! The streamed container is byte-identical to the buffered one (#47). +#![allow(clippy::cast_possible_truncation)] +//! +//! The streaming sink holds each entry — its local header and its payload — +//! and applies the writer's CRC-32 and size patch to those held bytes +//! before releasing the entry. If that bookkeeping were wrong the two +//! archives would differ — in the local header, in a data descriptor, or in +//! the central directory — so the comparison is the contract. + +use std::io::Write; + +use consolebook_server::export_stream::EntryBuffer; + +fn options() -> zip::write::SimpleFileOptions { + zip::write::SimpleFileOptions::default() + .compression_method(zip::CompressionMethod::Stored) + .last_modified_time( + zip::DateTime::from_date_and_time(2026, 9, 1, 19, 0, 0).expect("valid instant"), + ) + .unix_permissions(0o644) +} + +fn buffered(entries: &[(String, Vec)]) -> Vec { + let mut writer = zip::ZipWriter::new(std::io::Cursor::new(Vec::new())); + for (name, bytes) in entries { + writer.start_file(name.clone(), options()).expect("start"); + writer.write_all(bytes).expect("write"); + } + writer.finish().expect("finish").into_inner() +} + +fn streamed(entries: &[(String, Vec)]) -> (Vec, usize) { + let sink = EntryBuffer::new(std::io::Cursor::new(Vec::new())); + let mut writer = zip::ZipWriter::new(sink); + for (name, bytes) in entries { + writer.start_file(name.clone(), options()).expect("start"); + writer.write_all(bytes).expect("write"); + } + let sink = writer.finish().expect("finish"); + let peak = sink.peak_entry_bytes(); + let bytes = sink.into_inner().expect("release").into_inner(); + (bytes, peak) +} + +fn entries_of(sizes: &[usize]) -> Vec<(String, Vec)> { + sizes + .iter() + .enumerate() + .map(|(index, size)| { + ( + format!("records/{index}/v1/record.json"), + (0..*size).map(|at| (at % 251) as u8).collect(), + ) + }) + .collect() +} + +#[test] +fn a_streamed_archive_is_byte_identical_to_a_buffered_one() { + // Sizes chosen to cross the local-header boundary, a write boundary, + // and a realistic record. + for sizes in [ + vec![0], + vec![1], + vec![13], + vec![14], + vec![15], + vec![1024], + vec![0, 1, 13, 14, 15, 4096], + vec![65_536, 3, 300_000], + ] { + let entries = entries_of(&sizes); + let reference = buffered(&entries); + let (produced, peak) = streamed(&entries); + assert_eq!( + produced, reference, + "sizes {sizes:?} produced different container bytes" + ); + // The sink really held an entry (so the patches had somewhere to + // land) and held no more than the largest one plus its header. + let largest = sizes.iter().copied().max().expect("a size"); + assert!( + peak >= largest, + "sizes {sizes:?} retained only {peak} bytes, less than the largest entry" + ); + assert!( + peak <= largest + 1024, + "sizes {sizes:?} retained {peak} bytes for a {largest}-byte entry" + ); + } +} + +#[test] +fn a_streamed_archive_carries_no_data_descriptors() { + let entries = entries_of(&[2048, 7, 65_536]); + let (produced, _) = streamed(&entries); + let reference = buffered(&entries); + // Every local header carries flag bit 3 only when the writer had to + // defer CRC-32 and sizes to a trailing descriptor, and walking the + // archive by its own size fields only works when they were patched. + let mut at = 0usize; + let mut local_headers = 0; + while at + 30 <= produced.len() && produced[at..at + 4] == [0x50, 0x4b, 0x03, 0x04] { + let flags = u16::from_le_bytes([produced[at + 6], produced[at + 7]]); + assert_eq!(flags & 0x0008, 0, "local header at {at} uses a descriptor"); + let crc = u32::from_le_bytes([ + produced[at + 14], + produced[at + 15], + produced[at + 16], + produced[at + 17], + ]); + let size = u32::from_le_bytes([ + produced[at + 18], + produced[at + 19], + produced[at + 20], + produced[at + 21], + ]); + let uncompressed = u32::from_le_bytes([ + produced[at + 22], + produced[at + 23], + produced[at + 24], + produced[at + 25], + ]); + assert_ne!(crc, 0, "local header at {at} was never patched"); + assert_eq!( + size as usize, + entries[local_headers].1.len(), + "local header at {at} carries the wrong size" + ); + assert_eq!(uncompressed, size, "stored entry at {at} disagrees on size"); + // The whole header, patch included, is the buffered writer's. + assert_eq!( + &produced[at..at + 30], + &reference[at..at + 30], + "local header at {at} differs from the buffered archive" + ); + let name_len = usize::from(u16::from_le_bytes([produced[at + 26], produced[at + 27]])); + let extra_len = usize::from(u16::from_le_bytes([produced[at + 28], produced[at + 29]])); + at += 30 + name_len + extra_len + size as usize; + local_headers += 1; + } + assert_eq!(local_headers, entries.len(), "not every entry was walked"); + assert_eq!(produced, reference); +} + +// ---------------------------------------------------------------- #47 proof +// +// The streamed export's memory is bounded by its entry metadata, one +// record, and the container's directory rather than by the corpus. The +// corpora below are far larger than the bound, the sink discards what it +// receives, and the peak is read from a process that holds nothing else. + +use std::io::Seek; + +use consolebook_server::capabilities::RoleBundle; +use consolebook_server::data_dir::DataDir; +use consolebook_server::programs::{ + self, AnchorDef, CompetencyDef, FormCompetencyDef, FormDef, NarrativeDef, PolicyDef, + RecordType, ScaleDef, ScaleKind, TaskDef, VersionContent, +}; +use consolebook_server::record_export::{self, Scope}; +use consolebook_server::{assignments, enrollments, setup, storage}; +use sqlx::SqlitePool; + +const PAYLOAD: usize = 16 * 1024; +const EXPORTED_AT: i64 = 1_788_289_200; + +fn content() -> VersionContent { + VersionContent { + name: "Example County CTO Program".to_owned(), + label: "2026 rev A".to_owned(), + description: "Invented program for the streaming proof.".to_owned(), + phases: Vec::new(), + phase_transitions: Vec::new(), + competencies: vec![CompetencyDef { + category: "Call processing".to_owned(), + name: "Emergency Call Interrogation".to_owned(), + description: "Obtains and verifies location, callback, and nature.".to_owned(), + tasks: vec![TaskDef { + prompt: "Processes an invented structure-fire call.".to_owned(), + citations: Vec::new(), + }], + citations: Vec::new(), + }], + rating_scales: vec![ScaleDef { + name: "Standard 1-7".to_owned(), + kind: ScaleKind::AnchoredNumeric, + min_value: Some(1), + max_value: Some(7), + anchors: vec![AnchorDef { + value: 4, + label: "Meets standards".to_owned(), + definition: "To the invented standard.".to_owned(), + }], + }], + rating_modifiers: Vec::new(), + evaluation_forms: vec![FormDef { + record_type: RecordType::DailyReport, + name: "Daily Observation Report".to_owned(), + instructions: "Rate observed performance.".to_owned(), + competencies: vec![FormCompetencyDef { + competency: "Emergency Call Interrogation".to_owned(), + rating_scale: "Standard 1-7".to_owned(), + }], + narratives: vec![NarrativeDef { + prompt: "Most acceptable performance.".to_owned(), + required: false, + }], + }], + citations: Vec::new(), + finalization_policy: PolicyDef { + review_approved: false, + required_narratives: false, + ratings_complete: false, + }, + } +} + +/// The peak resident set this process has reached, from its own status. +/// `None` where the platform does not report one, so the measurement is +/// skipped rather than failing the suite. +fn peak_kib() -> Option { + let status = std::fs::read_to_string("/proc/self/status").ok()?; + status + .lines() + .find_map(|line| line.strip_prefix("VmHWM:")) + .and_then(|value| value.split_whitespace().next()) + .and_then(|value| value.parse().ok()) +} + +/// Runs `measure` in a fresh process against the database the fixture just +/// wrote, so the peak it reports belongs to the export rather than to the +/// corpus the fixture allocated. Returns `(units, archive_bytes, +/// peak_growth_kib)`. +fn measure_in_fresh_process(database: &std::path::Path, actor: i64) -> (i64, u64, u64) { + let output = std::process::Command::new(std::env::current_exe().expect("test binary")) + .args([ + "--ignored", + "--exact", + "export_memory_probe_child", + "--nocapture", + "--test-threads=1", + ]) + .env(PROBE_DB, database) + .env(PROBE_ACTOR, actor.to_string()) + .output() + .expect("run the probe"); + let text = String::from_utf8_lossy(&output.stdout); + assert!( + output.status.success(), + "the probe failed: {text}{}", + String::from_utf8_lossy(&output.stderr) + ); + // The child is a test binary too, so the measurement shares its line + // with the harness's own report of the test. + let line = text + .lines() + .find_map(|line| line.find("probe units=").map(|at| &line[at..])) + .unwrap_or_else(|| panic!("no measurement in {text}")); + let field = |name: &str| -> u64 { + line.split_whitespace() + .find_map(|part| part.strip_prefix(name)) + .unwrap_or_else(|| panic!("{name} missing from {line:?}")) + .parse() + .expect("a number") + }; + ( + field("units=").cast_signed(), + field("archive_bytes="), + field("peak_growth_kib="), + ) +} + +const PROBE_DB: &str = "CONSOLEBOOK_EXPORT_PROBE_DB"; +const PROBE_ACTOR: &str = "CONSOLEBOOK_EXPORT_PROBE_ACTOR"; + +/// The measured side of the memory proof: one installation export, in a +/// process that holds nothing but the database. +#[test] +#[ignore = "spawned by the memory proof with a seeded database"] +fn export_memory_probe_child() { + let Some(database) = std::env::var_os(PROBE_DB) else { + return; + }; + let actor: i64 = std::env::var(PROBE_ACTOR) + .expect("the probe actor") + .parse() + .expect("a number"); + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .expect("runtime"); + runtime.block_on(async move { + let pool = storage::open(std::path::Path::new(&database)) + .await + .expect("open"); + let Some(before) = peak_kib() else { + println!("probe unavailable"); + return; + }; + let mut sink = Discard { + bytes: 0, + position: 0, + }; + let (produced, returned) = record_export::export_to( + &pool, + actor, + Scope::Installation, + EXPORTED_AT, + &mut sink, + |_| {}, + ) + .await + .expect("call"); + let produced = produced.expect("exported"); + let after = peak_kib().expect("a reported peak"); + println!( + "probe units={} archive_bytes={} peak_growth_kib={}", + produced.unit_count, + returned.bytes, + after.saturating_sub(before) + ); + pool.close().await; + }); +} + +/// A scratch installation with a published program, an enrolled trainee, +/// an assigned trainer, one daily record, and `count` finalized versions +/// of `size` bytes each, exported in a fresh process. +async fn measured(count: i64, size: usize) -> Option<(i64, u64, u64)> { + if peak_kib().is_none() { + eprintln!("this platform reports no process peak: the memory bound is skipped"); + return None; + } + let (tmp, pool, admin_id, record_id) = installed().await; + seed_versions(&pool, admin_id, record_id, count, size).await; + let database = pool.connect_options().get_filename().to_owned(); + pool.close().await; + let measured = measure_in_fresh_process(std::path::Path::new(&database), admin_id); + drop(tmp); + Some(measured) +} + +/// The installation export's memory is bounded by the entry metadata, one +/// record, and the container's directory — never by the corpus — and every +/// byte reaches the sink. The three exports below separate the two terms: +/// `small` and `heavy` share a unit count and differ fourfold in payload, +/// and `wide` shares `small`'s payload and differs fourfold in units. +#[tokio::test(flavor = "multi_thread")] +async fn an_installation_export_is_bounded_by_its_entries_not_by_the_corpus() { + let small = measured(500, PAYLOAD).await.expect("a measured peak"); + let heavy = measured(500, 4 * PAYLOAD).await.expect("a measured peak"); + let wide = measured(4_000, PAYLOAD).await.expect("a measured peak"); + eprintln!("small: {small:?}\nheavy: {heavy:?}\nwide: {wide:?}"); + // The measurement is real, and the corpora really differ. + assert_eq!(small.0, 500); + assert_eq!(heavy.0, 500); + assert_eq!(wide.0, 4_000); + assert!( + heavy.1 > 3 * small.1, + "the heavy corpus is only {} bytes against {}", + heavy.1, + small.1 + ); + assert!( + wide.1 > 3 * small.1, + "the wide corpus is only {} bytes against {}", + wide.1, + small.1 + ); + // An absolute bound: buffering even the smallest corpus would exceed + // it, and the sink discards what it receives. + assert!( + small.2 < 16 * 1024, + "the export grew by {} KiB for a {} byte archive", + small.2, + small.1 + ); + // Four times the payload at the same unit count: memory does not + // follow the corpus. + assert!( + heavy.2 < small.2 + 4 * 1024, + "four times the payload grew the peak from {} KiB to {} KiB", + small.2, + heavy.2 + ); + // Eight times the units: the entries' metadata, their JSON manifest, + // and the container's directory grow — a per-unit constant of a few + // kilobytes, which ADR 0014 names as the format's one linear term — + // while the corpus, which is eight times larger again, does not. + assert!( + (wide.2 - small.2) * 1024 < 4 * 1024 * (wide.0 - small.0).cast_unsigned(), + "the peak grew by {} KiB for {} more units", + wide.2 - small.2, + wide.0 - small.0 + ); + assert!( + wide.2 * 4 < wide.1 / 1024, + "the export held {} KiB of a {} byte archive", + wide.2, + wide.1 + ); +} + +/// A scratch installation with a published program, an enrolled trainee, an +/// assigned trainer, and one daily record. Returns the pool and the ids the +/// corpus builders need. +async fn installed() -> (tempfile::TempDir, SqlitePool, i64, i64) { + let tmp = tempfile::tempdir().expect("temp dir"); + let data_dir = DataDir::new(tmp.path().join("data")); + data_dir.ensure_layout().expect("layout"); + let pool = storage::open(&data_dir.database()).await.expect("open"); + let code = setup::issue_setup_code(&pool) + .await + .expect("issue") + .expect("uninitialized") + .0; + let admin_id = setup::initialize( + &pool, + &code.raw, + "Example County Communications", + "avery.admin", + "Avery Admin", + "invented-passphrase-1", + ) + .await + .expect("initialize") + .expect("accepted"); + let program_id = programs::create_program(&pool, admin_id, "Example County CTO Program") + .await + .expect("create") + .expect("accepted"); + let version_id = programs::create_version(&pool, admin_id, program_id, &content()) + .await + .expect("create") + .expect("accepted"); + programs::publish_version(&pool, admin_id, version_id) + .await + .expect("publish") + .expect("accepted"); + let trainee = consolebook_server::users::create_with_reset_code( + &pool, + admin_id, + "taylor.trainee", + "Taylor Trainee", + "", + "", + RoleBundle::Trainee, + ) + .await + .expect("create") + .expect("accepted"); + let author = consolebook_server::users::create_with_reset_code( + &pool, + admin_id, + "jordan.trainer", + "Jordan Trainer", + "", + "", + RoleBundle::Trainer, + ) + .await + .expect("create") + .expect("accepted"); + let enrollment_id = enrollments::enroll(&pool, admin_id, version_id, trainee.id) + .await + .expect("enroll") + .expect("enrolled"); + assignments::create(&pool, admin_id, enrollment_id, author.id) + .await + .expect("assign") + .expect("assigned"); + let record_id: i64 = sqlx::query_scalar( + "INSERT INTO evaluation_record + (enrollment_id, program_version_id, evaluation_form_id, owner_user_id, + revision, created_at, created_by) + SELECT ?1, ?2, f.id, ?3, 0, ?4, ?3 + FROM evaluation_form f + WHERE f.program_version_id = ?2 AND f.record_type = 'daily_report' + RETURNING id", + ) + .bind(enrollment_id) + .bind(version_id) + .bind(author.id) + .bind(1_788_289_200_i64) + .fetch_one(&pool) + .await + .expect("insert record"); + (tmp, pool, admin_id, record_id) +} + +/// A corpus of `count` chained finalized versions, each `size` bytes. +async fn seed_versions(pool: &SqlitePool, admin_id: i64, record_id: i64, count: i64, size: usize) { + let payload = vec![b'x'; size]; + let mut tx = storage::write_tx(pool).await.expect("write tx"); + let mut predecessor: Option = None; + for number in 1..=count { + if number > 1 { + sqlx::query( + "INSERT INTO amendment + (evaluation_record_id, predecessor_version_id, reason, + opened_by, opened_by_display_name, opened_at, + opened_after_event_id, opened_after_decision_id) + VALUES (?1, ?2, 'Invented correction.', ?3, 'Avery Admin', ?4, 0, 0)", + ) + .bind(record_id) + .bind(predecessor) + .bind(admin_id) + .bind(1_788_289_200_i64) + .execute(&mut *tx) + .await + .expect("insert amendment"); + } + let id: i64 = sqlx::query_scalar( + "INSERT INTO evaluation_version + (evaluation_record_id, version_number, record_schema, canonical_bytes, + content_hash, chain_hash, predecessor_id, finalized_by, finalized_at) + VALUES (?1, ?2, 2, ?3, ?4, ?4, ?5, ?6, ?7) + RETURNING id", + ) + .bind(record_id) + .bind(number) + .bind(&payload) + .bind("a".repeat(64)) + .bind(predecessor) + .bind(admin_id) + .bind(1_788_289_200_i64) + .fetch_one(&mut *tx) + .await + .expect("insert version"); + predecessor = Some(id); + } + tx.commit().await.expect("commit"); +} + +/// A sink that holds nothing and counts what it is given. Seeking is +/// accepted because the container writer patches each entry's local +/// header; a sink that keeps no bytes has nothing to rewrite. +struct Discard { + bytes: u64, + position: u64, +} + +impl Write for Discard { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + if self.position == self.bytes { + self.bytes += buf.len() as u64; + } + self.position += buf.len() as u64; + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl Seek for Discard { + fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result { + self.position = match pos { + std::io::SeekFrom::Start(at) => at, + std::io::SeekFrom::End(0) => self.bytes, + std::io::SeekFrom::Current(0) => self.position, + other => { + return Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + format!("discarding sink ({other:?})"), + )); + } + }; + Ok(self.position) + } +} + +/// A destination that holds the first entry it is given until the test +/// releases it, so an export can be observed while it is mid-stream. +struct Gate { + position: u64, + entered: std::sync::mpsc::Sender<()>, + release: std::sync::mpsc::Receiver<()>, + held: bool, +} + +impl Write for Gate { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + if !self.held { + self.held = true; + let _ = self.entered.send(()); + let _ = self + .release + .recv_timeout(std::time::Duration::from_secs(60)); + } + self.position += buf.len() as u64; + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl Seek for Gate { + fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result { + self.position = match pos { + std::io::SeekFrom::Start(at) => at, + std::io::SeekFrom::End(0) | std::io::SeekFrom::Current(0) => self.position, + other => { + return Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + format!("gated sink ({other:?})"), + )); + } + }; + Ok(self.position) + } +} + +/// An export that is mid-stream reserves no writer: the installation's +/// other writers keep working while a download is in flight, because the +/// export holds a read snapshot and its audit is its own committed +/// statement rather than part of that transaction (ADR 0019; #47). +#[tokio::test(flavor = "multi_thread")] +async fn a_streaming_export_holds_no_write_reservation() { + let (_tmp, pool, admin_id, record_id) = installed().await; + seed_versions(&pool, admin_id, record_id, 20, PAYLOAD).await; + + let (entered_tx, entered_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let export_pool = pool.clone(); + let exporting = tokio::task::spawn_blocking(move || { + let sink = Gate { + position: 0, + entered: entered_tx, + release: release_rx, + held: false, + }; + tokio::runtime::Handle::current().block_on(async move { + record_export::export_to( + &export_pool, + admin_id, + Scope::Installation, + EXPORTED_AT, + sink, + |_| {}, + ) + .await + }) + }); + // The export is now past its audit and mid-payload. + entered_rx + .recv_timeout(std::time::Duration::from_secs(30)) + .expect("the export reached its first entry"); + + // A writer takes the installation's write reservation and commits. + let started = std::time::Instant::now(); + consolebook_server::audit::record( + &pool, + consolebook_server::audit::EventKind::RecordExported, + Some(admin_id), + None, + ) + .await + .expect("a write while an export streams"); + let waited = started.elapsed(); + assert!( + waited < std::time::Duration::from_secs(1), + "the writer waited {waited:?} for an export that holds no reservation" + ); + + release_tx.send(()).expect("release the export"); + let (produced, _sink) = exporting.await.expect("join").expect("call"); + assert!(produced.is_ok(), "{produced:?}"); + pool.close().await; +} + +/// A pool over the seeded database with exactly `connections` connections, +/// so the acquisition boundary is the one under test. +async fn pool_of(database: &std::path::Path, connections: u32) -> SqlitePool { + sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(connections) + .acquire_timeout(std::time::Duration::from_secs(5)) + .connect_with( + sqlx::sqlite::SqliteConnectOptions::new() + .filename(database) + .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) + .foreign_keys(true) + .busy_timeout(std::time::Duration::from_secs(1)), + ) + .await + .expect("pool") +} + +/// Closes the fixture pool and hands back a pool of exactly `connections` +/// over the same database. +async fn reopened(database: &std::path::Path, connections: u32) -> SqlitePool { + pool_of(database, connections).await +} + +/// An export never needs two connections at once, so a pool of one serves +/// it: the audit's own short write transaction ends before the read +/// snapshot begins. An export that held a read transaction while waiting +/// for a second connection would time out here. +#[tokio::test(flavor = "multi_thread")] +async fn a_one_connection_pool_serves_a_whole_export() { + let (_tmp, pool, admin_id, record_id) = installed().await; + seed_versions(&pool, admin_id, record_id, 8, PAYLOAD).await; + let database = pool.connect_options().get_filename().to_owned(); + pool.close().await; + let single = reopened(std::path::Path::new(&database), 1).await; + + let mut sink = Discard { + bytes: 0, + position: 0, + }; + let (produced, returned) = record_export::export_to( + &single, + admin_id, + Scope::Installation, + EXPORTED_AT, + &mut sink, + |_| {}, + ) + .await + .expect("call"); + let produced = produced.expect("exported"); + assert_eq!(produced.unit_count, 8); + assert!(returned.bytes > 8 * PAYLOAD as u64); + single.close().await; +} + +/// Two exports overlapping on a two-connection pool both reach their +/// stream and finish: each holds one connection at a time, so neither waits +/// for a connection the other holds. An export that held its read +/// transaction while acquiring the audit's connection would leave both +/// waiting until their acquisition timeouts. +#[tokio::test(flavor = "multi_thread")] +async fn overlapping_exports_on_a_two_connection_pool_both_complete() { + let (_tmp, pool, admin_id, record_id) = installed().await; + seed_versions(&pool, admin_id, record_id, 8, PAYLOAD).await; + let database = pool.connect_options().get_filename().to_owned(); + pool.close().await; + let small = reopened(std::path::Path::new(&database), 2).await; + + let first = gate_of(); + let second = gate_of(); + let running: Vec<_> = [first.gate, second.gate] + .into_iter() + .map(|gate| { + let pool = small.clone(); + tokio::task::spawn_blocking(move || { + tokio::runtime::Handle::current().block_on(async move { + record_export::export_to( + &pool, + admin_id, + Scope::Installation, + EXPORTED_AT, + gate, + |_| {}, + ) + .await + }) + }) + }) + .collect(); + // Both exports reach the stream while the other is still open, which is + // the state that needs two connections. + first + .arrived_rx + .recv_timeout(std::time::Duration::from_secs(30)) + .expect("first export streams"); + second + .arrived_rx + .recv_timeout(std::time::Duration::from_secs(30)) + .expect("second export streams"); + first.release.send(()).expect("release the first"); + second.release.send(()).expect("release the second"); + for export in running { + let (produced, _sink) = export.await.expect("join").expect("call"); + assert_eq!(produced.expect("exported").unit_count, 8); + } + small.close().await; +} + +/// A destination that holds its first entry until the test releases it. +struct ExportGate { + position: u64, + released: std::sync::mpsc::Receiver<()>, + entered: std::sync::mpsc::Sender<()>, + held: bool, +} + +impl Write for ExportGate { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + if !self.held { + self.held = true; + let _ = self.entered.send(()); + let _ = self + .released + .recv_timeout(std::time::Duration::from_secs(60)); + } + self.position += buf.len() as u64; + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl Seek for ExportGate { + fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result { + self.position = match pos { + std::io::SeekFrom::Start(at) => at, + std::io::SeekFrom::End(0) | std::io::SeekFrom::Current(0) => self.position, + other => { + return Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + format!("gated sink ({other:?})"), + )); + } + }; + Ok(self.position) + } +} + +/// One held export: its sink, the signal that it reached the stream, and +/// the release it is waiting for. +struct GateHandle { + gate: ExportGate, + arrived_rx: std::sync::mpsc::Receiver<()>, + release: std::sync::mpsc::Sender<()>, +} + +fn gate_of() -> GateHandle { + let (entered_tx, arrived_rx) = std::sync::mpsc::channel(); + let (release, released) = std::sync::mpsc::channel(); + GateHandle { + gate: ExportGate { + position: 0, + released, + entered: entered_tx, + held: false, + }, + arrived_rx, + release, + } +} + +/// An export mid-stream holds one connection, and the installation's other +/// work keeps running on the rest: the audit is already committed, so the +/// download reserves no writer and no second connection. +#[tokio::test(flavor = "multi_thread")] +async fn ordinary_work_progresses_while_an_export_streams() { + let (_tmp, pool, admin_id, record_id) = installed().await; + seed_versions(&pool, admin_id, record_id, 20, PAYLOAD).await; + let database = pool.connect_options().get_filename().to_owned(); + pool.close().await; + let small = reopened(std::path::Path::new(&database), 2).await; + + let held = gate_of(); + let export_pool = small.clone(); + let running = tokio::task::spawn_blocking(move || { + tokio::runtime::Handle::current().block_on(async move { + record_export::export_to( + &export_pool, + admin_id, + Scope::Installation, + EXPORTED_AT, + held.gate, + |_| {}, + ) + .await + }) + }); + // The export is on the wire, holding its one connection for the read + // snapshot; everything below runs on the other. + held.arrived_rx + .recv_timeout(std::time::Duration::from_secs(30)) + .expect("the export reached its stream"); + + let started = std::time::Instant::now(); + consolebook_server::audit::record( + &small, + consolebook_server::audit::EventKind::RecordExported, + Some(admin_id), + None, + ) + .await + .expect("an ordinary write while an export streams"); + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM evaluation_version") + .fetch_one(&small) + .await + .expect("an ordinary read while an export streams"); + assert_eq!(count, 20); + let waited = started.elapsed(); + assert!( + waited < std::time::Duration::from_secs(1), + "ordinary work waited {waited:?} for a streaming export" + ); + + held.release.send(()).expect("release the export"); + let (produced, _sink) = running.await.expect("join").expect("call"); + assert_eq!(produced.expect("exported").unit_count, 20); + small.close().await; +} + +/// A destination that refuses to accept past `cap`, the way a client that +/// stops reading does once the response's queue is full. +struct Capped { + cap: u64, + bytes: u64, + position: u64, +} + +impl Write for Capped { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + if self.position >= self.cap { + return Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "the client left", + )); + } + if self.position == self.bytes { + self.bytes += buf.len() as u64; + } + self.position += buf.len() as u64; + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl Seek for Capped { + fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result { + self.position = match pos { + std::io::SeekFrom::Start(at) => at, + std::io::SeekFrom::End(0) => self.bytes, + std::io::SeekFrom::Current(0) => self.position, + other => { + return Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + format!("capped sink ({other:?})"), + )); + } + }; + Ok(self.position) + } +} + +/// A destination that stops accepting, the way the response body does once +/// its bounded queue is full and the client is not reading. +#[tokio::test(flavor = "multi_thread")] +async fn a_destination_that_stops_accepting_ends_the_export_instead_of_buffering_it() { + let (_tmp, pool, admin_id, record_id) = installed().await; + seed_versions(&pool, admin_id, record_id, 20, PAYLOAD).await; + + // The client stops reading after a couple of records. + let mut sink = Capped { + cap: 2 * PAYLOAD as u64, + bytes: 0, + position: 0, + }; + let outcome = record_export::export_to( + &pool, + admin_id, + Scope::Installation, + EXPORTED_AT, + &mut sink, + |_| {}, + ) + .await; + let err = match outcome { + Ok((produced, _returned)) => { + produced.expect_err("a stopped destination is not a complete export"); + panic!("the export completed into a destination that stopped accepting"); + } + Err(err) => err, + }; + // The failure is the client's departure, surfaced rather than hidden. + assert!( + err.chain() + .filter_map(|cause| cause.downcast_ref::()) + .any(|io| io.kind() == std::io::ErrorKind::BrokenPipe), + "{err:?}" + ); + // The producer stopped where the client stopped: it did not run on, + // holding the rest of the corpus for a reader that had left. + assert!( + sink.bytes <= 4 * PAYLOAD as u64, + "the sink accepted {} bytes after its cap", + sink.bytes + ); + pool.close().await; +} + +/// The response carries the documented bytes: a client reading from the +/// socket receives the same archive the buffered exporter produces for the +/// same scope and instant, verified from the wire alone. +#[tokio::test(flavor = "multi_thread")] +async fn the_export_response_carries_the_buffered_archive_over_the_wire() { + let (_tmp, pool, admin_id, record_id) = installed().await; + seed_versions(&pool, admin_id, record_id, 8, 256 * 1024).await; + let seeded: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM evaluation_version") + .fetch_one(&pool) + .await + .expect("count"); + assert_eq!(seeded, 8, "the corpus is present before the request"); + + // A live listener, so the response travels over a real socket. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("addr"); + let app = + consolebook_server::http::router(consolebook_server::http::AppState { pool: pool.clone() }); + let serving = tokio::spawn(async move { + let _ = axum::serve(listener, app).await; + }); + + let login = live_login(&addr, "avery.admin").await; + let mut client = tokio::net::TcpStream::connect(addr).await.expect("connect"); + let request = format!( + "GET /api/exports/records HTTP/1.1\r\nHost: {addr}\r\nCookie: {}={}\r\nConnection: close\r\n\r\n", + consolebook_server::http::SESSION_COOKIE, + login + ); + tokio::io::AsyncWriteExt::write_all(&mut client, request.as_bytes()) + .await + .expect("write request"); + let mut response = Vec::new(); + tokio::io::AsyncReadExt::read_to_end(&mut client, &mut response) + .await + .expect("read response"); + serving.abort(); + + let (headers, body) = split_response(&response); + let detail = String::from_utf8_lossy(body).into_owned(); + assert!(headers.starts_with("http/1.1 200"), "{headers} {detail}"); + assert!(headers.contains("application/zip"), "{headers}"); + assert!( + headers.contains("content-disposition: attachment"), + "{headers}" + ); + // The body is chunked and has no length: it is produced as it is + // written, not assembled and measured first. + assert!(headers.contains("transfer-encoding: chunked"), "{headers}"); + assert!(!headers.contains("content-length"), "{headers}"); + assert!(chunked_is_complete(body), "the transfer did not finish"); + let body = dechunk(body); + // The container is complete and holds every unit. (The corpus here is + // synthetic bytes, so the record-level checks are the fixtures' job; + // this proof is that the transfer carried the whole container.) + let report = consolebook_server::export_verify::verify_archive(&body); + assert_eq!(report.units.len(), 8, "{report:?}"); + assert!(report.findings.is_empty(), "{report:?}"); + assert_eq!(report.scope, Some(Scope::Installation)); + for unit in &report.units { + assert!( + !unit.findings.iter().any(|finding| matches!( + finding, + consolebook_server::export_verify::Finding::MissingEntry { .. } + )), + "unit {} is incomplete: {:?}", + unit.path, + unit.findings + ); + } + + // The delivered bytes are the buffered archive's, for the instant the + // archive itself states: the streamed path is not merely verifiable, it + // is byte-for-byte the shipped exporter's output. + let exported_at = archive_instant(&body); + let again = record_export::export_at(&pool, admin_id, Scope::Installation, exported_at) + .await + .expect("call") + .expect("exported"); + assert_eq!( + body, + again.bytes, + "the response carried {} bytes; the buffered export for the same instant is {}", + body.len(), + again.bytes.len() + ); + pool.close().await; +} + +/// A client that stops reading long enough to lose the failure detail must +/// still lose the transfer. The queue stays full past the data send's +/// deadline **and** past the failure marker's own deadline, so the marker +/// is never queued; the client then resumes reading, and what it receives +/// must be an incomplete transfer rather than a clean end of body carrying +/// a truncated prefix. +#[tokio::test(flavor = "multi_thread")] +async fn a_client_that_outlasts_every_send_window_never_sees_a_complete_transfer() { + // Larger than the response queue, the socket buffers, and the runtime's + // own buffering together, so a client that reads nothing cannot be + // handed the whole archive from memory. + let (_tmp, pool, admin_id, record_id) = installed().await; + seed_versions(&pool, admin_id, record_id, 40, 1024 * 1024).await; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("addr"); + let app = + consolebook_server::http::router(consolebook_server::http::AppState { pool: pool.clone() }); + let serving = tokio::spawn(async move { + let _ = axum::serve(listener, app).await; + }); + let login = live_login(&addr, "avery.admin").await; + let mut client = tokio::net::TcpStream::connect(addr).await.expect("connect"); + let request = format!( + "GET /api/exports/records HTTP/1.1\r\nHost: {addr}\r\nCookie: {}={}\r\nConnection: close\r\n\r\n", + consolebook_server::http::SESSION_COOKIE, + login + ); + tokio::io::AsyncWriteExt::write_all(&mut client, request.as_bytes()) + .await + .expect("write request"); + + // Nothing is read while both windows expire: the data send gives up, + // and the failure marker has no room to be queued either. + let both_windows = 2 * consolebook_server::exports_http::EXPORT_STALL_LIMIT + + std::time::Duration::from_secs(5); + tokio::time::sleep(both_windows).await; + + // Resuming now: everything the failure had already queued drains, and + // then the stream must end in an error, not in a clean end of body. + let mut response = Vec::new(); + let read = tokio::io::AsyncReadExt::read_to_end(&mut client, &mut response).await; + serving.abort(); + let (headers, body) = split_response(&response); + assert!(headers.starts_with("http/1.1 200"), "{headers}"); + assert!( + !chunked_is_complete(body), + "the client outlasted both send windows and still saw a complete transfer of {} bytes", + body.len() + ); + let delivered = dechunk(body); + let complete = record_export::export_at(&pool, admin_id, Scope::Installation, EXPORTED_AT) + .await + .expect("call") + .expect("exported"); + assert!( + delivered.len() < complete.bytes.len(), + "the stalled client received the whole {} byte archive (read {read:?})", + complete.bytes.len() + ); + pool.close().await; +} + +/// The instant the delivered archive states for itself, from its manifest. +fn archive_instant(bytes: &[u8]) -> i64 { + let mut archive = + zip::ZipArchive::new(std::io::Cursor::new(bytes.to_vec())).expect("readable archive"); + let mut file = archive + .by_name("manifest.json") + .expect("the archive manifest is the first entry"); + let mut text = String::new(); + std::io::Read::read_to_string(&mut file, &mut text).expect("read manifest"); + let manifest: serde_json::Value = serde_json::from_str(&text).expect("manifest json"); + manifest["exported_at"].as_i64().expect("exported_at") +} + +/// Signs in over the live listener and returns the session cookie value. +async fn live_login(addr: &std::net::SocketAddr, username: &str) -> String { + let mut client = tokio::net::TcpStream::connect(addr).await.expect("connect"); + let body = format!("{{\"username\":\"{username}\",\"password\":\"invented-passphrase-1\"}}"); + let request = format!( + "POST /api/auth/login HTTP/1.1\r\nHost: {addr}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + tokio::io::AsyncWriteExt::write_all(&mut client, request.as_bytes()) + .await + .expect("write login"); + let mut response = Vec::new(); + tokio::io::AsyncReadExt::read_to_end(&mut client, &mut response) + .await + .expect("read login"); + let text = String::from_utf8_lossy(&response).into_owned(); + let (headers, _) = split_response(&response); + assert!(headers.starts_with("http/1.1 200"), "{text}"); + let value = headers + .lines() + .find_map(|line| line.strip_prefix("set-cookie: ")) + .and_then(|cookie| cookie.split(';').next()) + .and_then(|pair| pair.split_once('=')) + .map(|(_, value)| value.to_owned()); + value.unwrap_or_else(|| panic!("no session cookie in {headers}")) +} + +/// A chunked response body as the bytes it carries. A body that was cut +/// short yields the bytes it did carry, so a truncated transfer can be +/// inspected rather than panicking. +fn dechunk(body: &[u8]) -> Vec { + let mut out = Vec::new(); + let mut at = 0; + while let Some(end) = body[at..].windows(2).position(|window| window == b"\r\n") { + let size_line = String::from_utf8_lossy(&body[at..at + end]).into_owned(); + let Ok(size) = usize::from_str_radix(size_line.split(';').next().unwrap_or(""), 16) else { + break; + }; + at += end + 2; + if size == 0 || at + size > body.len() { + break; + } + out.extend_from_slice(&body[at..at + size]); + at += size + 2; + } + out +} + +/// Whether a chunked body carries its terminal zero-length chunk. A +/// transfer that ends without it was cut short, which is how a client +/// distinguishes an incomplete download from a complete one. +fn chunked_is_complete(body: &[u8]) -> bool { + let mut at = 0; + while let Some(end) = body[at..].windows(2).position(|window| window == b"\r\n") { + let size_line = String::from_utf8_lossy(&body[at..at + end]).into_owned(); + let Ok(size) = usize::from_str_radix(size_line.split(';').next().unwrap_or(""), 16) else { + return false; + }; + at += end + 2; + if size == 0 { + return true; + } + if at + size + 2 > body.len() { + return false; + } + at += size + 2; + } + false +} + +/// The response's header block and its body. +fn split_response(response: &[u8]) -> (String, &[u8]) { + let at = response + .windows(4) + .position(|window| window == b"\r\n\r\n") + .expect("header end"); + ( + String::from_utf8_lossy(&response[..at]).to_lowercase(), + &response[at + 4..], + ) +} diff --git a/crates/consolebook-server/tests/record_export.rs b/crates/consolebook-server/tests/record_export.rs index 171127e..cc8c96f 100644 --- a/crates/consolebook-server/tests/record_export.rs +++ b/crates/consolebook-server/tests/record_export.rs @@ -1450,6 +1450,42 @@ async fn export_api_delivers_the_documented_bytes() { assert_eq!(status, StatusCode::OK); assert!(export_verify::verify_archive(&bytes).verified()); + // The archive states the instant it was produced at, and re-exporting + // this scope at that instant reproduces the delivered bytes exactly: + // the streamed delivery and the tested producer are one path, byte for + // byte. (`export_stream_bytes.rs` walks the same comparison over a + // larger corpus and the installation scope.) + let (status, _, bytes) = raw_get( + fx.app(), + &format!("/api/drafts/{}/versions/2/export", s.record_id), + &casey, + ) + .await; + assert_eq!(status, StatusCode::OK); + let exported_at = export_verify::verify_archive(&bytes) + .exported_at + .expect("the archive's instant"); + let (produced, again) = record_export::export_to( + &fx.pool, + s.casey_id, + Scope::Version { + record_id: s.record_id, + version_number: 2, + }, + exported_at, + std::io::Cursor::new(Vec::new()), + |_| {}, + ) + .await + .expect("call"); + let produced = produced.expect("exported"); + assert_eq!(produced.unit_count, 1); + assert_eq!( + again.into_inner(), + bytes, + "the delivered bytes differ from the buffered export at the same instant" + ); + // The installation scope answers to export_records only. let (status, _, body) = raw_get(fx.app(), "/api/exports/records", &casey).await; assert_eq!(status, StatusCode::FORBIDDEN); diff --git a/docs/decisions/0014-record-export-format.md b/docs/decisions/0014-record-export-format.md index f3aa8ce..14a2c8c 100644 --- a/docs/decisions/0014-record-export-format.md +++ b/docs/decisions/0014-record-export-format.md @@ -122,9 +122,33 @@ may export, and what verification claims (#45; Milestone 5 slice 1). hash but cannot prove the predecessor's bytes — reported as *not in export*, never inferred; and - on-demand exports mean nothing on disk to dispose of and nothing to - resume: an installation export is one response, assembled in memory - while it is produced, so a very large history costs memory in - proportion until exports stream (tracked separately). + resume: an export is one response, and it is now streamed — written to + the response as it is produced (#47). What still scales with an + installation is its unit count, not its bytes: the unit metadata the + archive manifest lists, the serialized manifest, and the container's + central directory are each O(units), and the database driver buffers a + bounded number of rows per query. The corpus of stored payloads is + never held, one unit's bytes are held at a time, and the browser's own + download helper buffers the response it saves, separately; +- the delivery and its failure share one bounded channel between the + producer and the response. A client that stops reading stops the + producer instead of growing a queue, and a client that stops reading + for longer than the stall bound loses the transfer. Whether a transfer + was complete is an explicit fact the producer records only when the + archive was produced and its tail flushed — not the channel closing, + which a client that outlasted the failure signal would otherwise read + as success. Anything else ends the body in an error: an incomplete + download the verifier refuses, never a complete export; +- the export holds one pooled connection at a time. Its audit event is + written in its own short write transaction, committed before the read + snapshot the manifest and payloads share, and that snapshot is a reader + only (ADR 0019). Exports sharing a connection pool therefore never hold + one connection while waiting for another, and an export never reserves + the writer for the length of a download; +- the audit event records the export the installation produced from the + state at its recorded instant, not the operator's receipt: a delivery + that fails part way leaves the record, and a scope that holds nothing to + export is refused before any record is written. ## Rejected alternatives diff --git a/docs/development.md b/docs/development.md index c3b6b0a..0e43394 100644 --- a/docs/development.md +++ b/docs/development.md @@ -22,7 +22,7 @@ tests show what is implemented. [Roadmap](roadmap.md) owns milestone status. | Finalization and canonical bytes | `finalization.rs`, `canonical.rs`, `record_envelope.rs` | [Integrity](records-integrity.md), [ADR 0011](decisions/0011-canonical-record-format-and-finalization.md) | | Acknowledgments and amendments | `acknowledgments.rs`, `amendments.rs` | [Domain model](domain-model.md), [ADR 0012](decisions/0012-amendment-reopening-state-machine.md) | | Summaries and signoffs | `summaries.rs`, `task_signoffs.rs` | [ADR 0013](decisions/0013-weekly-summaries-and-task-signoffs.md), [ADR 0021](decisions/0021-trainee-signoff-history-read.md) | -| Record exports | `record_export.rs`, `export_verify.rs`, `zip_container.rs` | [ADR 0014](decisions/0014-record-export-format.md), [Export format](formats/record-export.md) | +| Record exports | `record_export.rs`, `export_stream.rs`, `export_verify.rs`, `zip_container.rs` | [ADR 0014](decisions/0014-record-export-format.md), [Export format](formats/record-export.md) | | Trainee packets | `trainee_packet.rs`, `packet_verify.rs` | [ADR 0015](decisions/0015-trainee-packet.md), [ADR 0017](decisions/0017-packet-pin-timeline-verification.md), [Packet format](formats/trainee-packet.md) | | Retention policy and holds | `retention.rs`, `retention/`, `retention_http.rs` | [ADR 0020](decisions/0020-retention-policy-and-hold-administration.md), [Operator guide](retention.md); disposition execution remains [#64](https://github.com/FieldmouseWorks/consolebook/issues/64) | | Web shell and HTTP | `http.rs`, `web_assets.rs`, `notices.rs`, domain `*_http.rs` modules | [ADR 0005](decisions/0005-embedded-web-interface.md), web map below | diff --git a/docs/formats/record-export.md b/docs/formats/record-export.md index 7ba7b47..4de7226 100644 --- a/docs/formats/record-export.md +++ b/docs/formats/record-export.md @@ -245,6 +245,21 @@ export after a tool has repacked it, which is what an operator asking byte-identical to a fresh export is a byte comparison, not a verification finding. +The container is unchanged by how an installation delivers it. A +producing installation writes the archive into the response as it is +produced, so a transfer that ends early leaves an incomplete file: it +lacks the central directory, the verifier reports it as unreadable rather +than as a smaller valid export, and the transfer ends in an error rather +than in the terminal chunk that would have made it look complete. A +complete transfer is an explicit fact the producer records only when the +whole archive was produced and its tail flushed; every other ending — a +production failure, a producer that panicked, a client that stopped +reading long enough to lose the failure signal — fails the body. The audit event +records the export the installation produced from the state at its +recorded instant; it is not a receipt for a completed download, so a +failed delivery is a transfer the operator repeats rather than a record +the installation un-writes. + ## What the archive does not carry - **Drafts.** An unfinalized record is not a record; scopes contain diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 9a8e557..c0c8243 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -1184,6 +1184,13 @@ export function installationExportPath(): string { * Fetches an export archive and hands it to the browser as a download, so * a refusal surfaces as an error instead of a saved error document. * Returns the server's file name. + * + * The server streams its archive, but this path still reads the whole + * response before handing the browser a Blob. That keeps the operator's + * refusal handling intact — a typed refusal arrives as a JSON body before + * anything is saved, and a transfer that fails part way rejects rather + * than saving a truncated file — at the cost of holding the archive in the + * browser while it downloads (#47). */ export async function downloadExport(path: string): Promise { const response = await fetch(path);