From 4755731cb31bb98f7914927da33a6549bf9a9496 Mon Sep 17 00:00:00 2001 From: Peter Permenter <41281403+TusanHomichi@users.noreply.github.com> Date: Sat, 12 Sep 2026 01:22:00 +0200 Subject: [PATCH] records(export): stream installation exports without buffering them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Export read every finalized version with `fetch_all` and assembled the whole ZIP in memory before the handler answered, so an installation export cost memory in proportion to its history (#47). `record_export::export_to` now writes the container into any seekable sink as it produces it: one read transaction, a metadata-only pass for the manifest (which must be the container's first entry), then a pass that streams each unit's stored bytes and unit manifest. `export_stream::EntryBuffer` holds one entry and applies the `zip` writer's patches to it, so the destination sees an append-only stream and the bytes stay identical to the buffered archive: no data descriptors, real CRC-32 and sizes in the local header, the same central directory. A patch is classified by position — inside the held entry — so the CRC and size patch that seeks backwards and the ZIP64 extra-field update that seeks forwards both land correctly. The export holds one pooled connection at a time, which is what the review's first blocker needed. Its audit event is written in its own short write transaction, committed before the read snapshot that both content passes share, and that snapshot is a reader only (ADR 0019): an export never reserves the writer while it streams, and exports sharing a pool never hold one connection while waiting for another. A scope that holds nothing to export is refused before anything is recorded, so the audit still attests exactly what it says it does. Delivery ends only on an explicit success. Whether a transfer was complete is a flag the producer sets 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 a complete download. Every other ending fails the body: a production failure, a producer that panicked (a drop guard covers it), a stalled client, and a client that stayed away past both send windows. Proof: streamed and buffered bytes compared across entry sizes including the patch boundary, and the bytes a live socket delivers compared against the buffered export at the instant the delivered manifest states; the installation's peak memory measured in a fresh process against pre-seeded databases (500 units / 8.7 MB archive at 5.1 MiB, the same units at four times the payload at 7.4 MiB, 4,000 units / 69.7 MB at 14.1 MiB), so the payload volume does not drive it while the unit count does; a whole export on a one-connection pool; two exports overlapping on a two-connection pool with ordinary work still progressing; an export observed mid-stream while the installation's own writer commits; a client that outlasts both stall windows and still cannot receive a complete transfer; and unit tests for each terminal case (success, closed channel without success, failure item, guard on drop, stall, departed client). Each of these was falsified against the design it replaces before it was kept. ADR 0014 and the format document state the connection ownership, the memory terms, and what an interrupted transfer means. --- Cargo.lock | 2 + crates/consolebook-server/Cargo.toml | 7 + .../consolebook-server/src/export_stream.rs | 210 +++ crates/consolebook-server/src/exports_http.rs | 566 +++++++- crates/consolebook-server/src/lib.rs | 1 + .../consolebook-server/src/record_export.rs | 471 ++++++- .../consolebook-server/src/trainee_packet.rs | 2 +- .../tests/export_stream_bytes.rs | 1247 +++++++++++++++++ .../consolebook-server/tests/record_export.rs | 36 + docs/decisions/0014-record-export-format.md | 30 +- docs/development.md | 2 +- docs/formats/record-export.md | 15 + web/src/lib/api.ts | 7 + 13 files changed, 2513 insertions(+), 83 deletions(-) create mode 100644 crates/consolebook-server/src/export_stream.rs create mode 100644 crates/consolebook-server/tests/export_stream_bytes.rs 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);