diff --git a/docs/incremental-database-artifacts.md b/docs/incremental-database-artifacts.md index 4224145d..be5e74dc 100644 --- a/docs/incremental-database-artifacts.md +++ b/docs/incremental-database-artifacts.md @@ -61,6 +61,10 @@ Explicit DuckDB exports checkpoint the engine and durably install that checkpoint before returning. SQLite exports use a non-destructive backup, so they do not disturb the WAL baseline. +## Bounded reads (TC-822) + +The node never reads a whole artifact row. Existence and size checks use `load_meta` (metadata columns only); hydration streams `payload` then `delta_payload` in 4 MiB slices into a cache-file sink, verifying the content hashes as it goes and restarting if the row changes mid-read. See `docs/runbooks/tc-822-artifact-payload-storage.md` for the Postgres `STORAGE EXTERNAL` operator step. + ## Observability Every persistence operation emits structured fields: diff --git a/docs/runbooks/tc-822-artifact-payload-storage.md b/docs/runbooks/tc-822-artifact-payload-storage.md new file mode 100644 index 00000000..262d586b --- /dev/null +++ b/docs/runbooks/tc-822-artifact-payload-storage.md @@ -0,0 +1,67 @@ +# TC-822: bounded artifact reads and the Postgres `payload` storage setting + +Tracking: [TC-822](https://linear.app/tinycloud-labs/issue/TC-822), under TC-1491. This is an **operator step on Postgres**, not a migration: it is not run at boot and rewrites no data. The PR that introduces the bounded reads contains no migration or schema change (schema freeze). + +## What the node does now + +The node never reads a whole `database_artifact` row. `load_meta` selects the metadata columns only. Hydration (`read_chunks`) reads `payload`, then `delta_payload`, in 4 MiB `substr(column, offset, length)` slices, each its own statement under the pool's base statement timeout and guarded by `WHERE revision = AND checkpoint_content_hash = `. The slices are hashed as they stream and compared with `checkpoint_content_hash` / `delta_content_hash` / `content_hash`. A slice that finds no row (the artifact was saved or deleted meanwhile) restarts the hydration from fresh metadata, at most twice; any hash or size mismatch fails the hydration and seeds nothing. + +Round trips per artifact hydration: an artifact of at most 8 MiB stored bytes (`BASE_TIMEOUT_PAYLOAD_BYTES`) is `1 + 1` (+1 with a WAL delta); a larger one is `1 + ceil(checkpoint_bytes / 4 MiB) + ceil(delta_bytes / 4 MiB)`. + +## Why Postgres needs `STORAGE EXTERNAL` on `payload` + +`bytea` columns default to `STORAGE EXTENDED`: large values are pglz-compressed in TOAST. `substr` on a compressed value decompresses the whole **prefix** up to `offset + length` inside the server, so slice cost and server memory grow with the offset. Measured on PostgreSQL 14 (`shared_buffers=64MB`, SQLite-like data compressing 3.4x): + +| artifact | slice offset | EXTENDED (compressed) | EXTERNAL | +|---|---|---|---| +| 256 MiB | 250 MB | 360 ms, 325 MB backend RSS | 1 ms, ~4 MB | +| 1000 MiB | 1,044 MB | 1.4 s, 1.3 GB backend RSS | 1 ms, ~7 MB | +| 1000 MiB, all 250 slices | | 214 s | 0.67 s | + +Without the setting, the node stays correct and its own memory stays bounded, but a large, compressed artifact hydrates slowly and costs the database server as much memory as one slice reaches into the value, and with a tight base statement timeout a deep slice can be cancelled (a 500 ms timeout is exceeded past ~100 MB offsets). Apply it before relying on large artifacts. + +## The step + +```sql +SET lock_timeout = '3s'; -- ACCESS EXCLUSIVE, metadata only; retry if it times out +ALTER TABLE database_artifact ALTER COLUMN payload SET STORAGE EXTERNAL; +``` + +- Idempotent: running it again is a no-op. Check with `SELECT attstorage FROM pg_attribute WHERE attrelid = 'database_artifact'::regclass AND attname = 'payload'` (`e` = external, `x` = extended). +- Not a rewrite. **Existing rows keep their compressed form** until their next full checkpoint save, which writes `payload` anew (WAL-only saves touch `delta_payload`, bounded to 8 MiB, and do not need the setting). Measured: `ALTER` alone leaves an existing compressed row as slow as before. +- A schema created fresh by the migrations stays `EXTENDED`; run the statement once per database. +- Do not run it inside a migration or the boot transaction. Do not mass-rewrite rows without a separate decision (an `UPDATE ... SET payload = payload || ''::bytea` rewrites every TOAST chunk and is WAL-heavy). + +## Production plan + +- Run the `ALTER` on the **new US-West** branch **before** the logical-replication initial copy: every row then arrives fresh and uncompressed, sliceable, and no rewrite is ever needed. +- On us-east the `ALTER` alone converts nothing; each artifact converts at its next full checkpoint. Until then a large, compressed artifact hydrates on the slow path (numbers above: ~360 ms and ~325 MB server memory for a slice at 250 MB of a 256 MiB artifact; 214 s to hydrate 1000 MiB). Node RAM stays bounded either way. +- The `ALTER` is DDL: run it once per side, outside the replication-copy-to-GO schema freeze window. This change's merge/deploy must not land inside that window either. This PR contains no migration or schema change. + +## Cost: `EXTERNAL` disables compression + +Compressible rows get larger on disk by `stored_bytes x (ratio - 1)` once they convert (PlanetScale bills above 10 GB; the database is about 6.4 GB in total). Measure the exact figure first with a metadata-only query (it reads the stored datum size and the raw size from the TOAST header; it does not detoast or decompress, verified on a 1000 MiB compressible row: 1 ms, 1 shared buffer hit, no TOAST pages read): + +```sql +SELECT count(*), pg_size_pretty(sum(pg_column_size(payload))) AS stored, + pg_size_pretty(sum(octet_length(payload))) AS raw, + round(sum(octet_length(payload))::numeric / NULLIF(sum(pg_column_size(payload)), 0), 2) AS ratio +FROM database_artifact; +``` + +The production figure is **unmeasured** until ops runs it (no production role here). Scenarios, growth = X x (r - 1) GB, X = stored `payload` GB, r = raw/stored ratio (3.45 is the synthetic pglz ratio of the test data; real SQLite ratios are likely lower), total 6.4 GB before: + +| stored X | r = 1.5 | r = 2.5 | r = 3.45 | +|---|---|---|---| +| 0.5 GB | +0.25 GB | +0.75 GB | +1.2 GB | +| 1 GB | +0.5 GB | +1.5 GB | +2.5 GB | +| 2 GB | +1.0 GB | +3.0 GB | +4.9 GB (11.3 GB total, over 10) | +| 4 GB | +2.0 GB | +6.0 GB (12.4 GB, over 10) | +9.8 GB (16.2 GB, over 10) | + +The 10 GB line is crossed when growth exceeds 3.6 GB: X = 4 GB at r = 2.5 or 3.45, and X = 2 GB at r = 3.45. All other cells stay under it. + +SQLite deployments need nothing, but note that SQLite's `substr` on a BLOB materialises the whole blob inside the SQLite library for every slice (about one blob of transient memory, ~0.3 s per slice per 256 MiB), so the "one slice" memory bound holds for the node's own heap and for Postgres, not for SQLite's internal buffer. + +## Adapting for at-rest encoding (TC-887) + +`ArtifactMeta` would gain the encoding columns; chunking applies to the **stored** bytes, so decoding belongs in the `ArtifactSink` (or a decoding wrapper) and the verified hashes are the stored-bytes hashes. `load_meta` / `read_chunks` documentation says the same. diff --git a/tinycloud-core/src/database_artifacts.rs b/tinycloud-core/src/database_artifacts.rs index 383914ac..933bb62d 100644 --- a/tinycloud-core/src/database_artifacts.rs +++ b/tinycloud-core/src/database_artifacts.rs @@ -1,11 +1,17 @@ use async_trait::async_trait; use sea_orm::{ - sea_query::Expr, ActiveModelTrait, ActiveValue::Set, ColumnTrait, DbErr, EntityTrait, - QueryFilter, QuerySelect, + sea_query::{Expr, Func}, + ActiveModelTrait, + ActiveValue::Set, + ColumnTrait, DbErr, EntityTrait, QueryFilter, QuerySelect, }; use time::{format_description::well_known::Rfc3339, OffsetDateTime}; -use crate::{db_pool::BulkScope, hash::hash, models::database_artifact}; +use crate::{ + db_pool::BulkScope, + hash::{hash, Blake3Hasher}, + models::database_artifact, +}; #[derive(Debug, Clone)] pub struct DatabaseArtifact { @@ -109,29 +115,502 @@ pub enum DatabaseArtifactError { IncrementalPersistenceUnsupported, #[error("database artifact was written from a superseded checkpoint")] StaleLineage, + #[error("database artifact changed while it was being read")] + ArtifactChanged, + #[error("database artifact {part} failed its content hash")] + HashMismatch { part: &'static str }, + #[error("database artifact {part} is {actual} bytes, expected {expected}")] + SizeMismatch { + part: &'static str, + expected: u64, + actual: u64, + }, + /// The sink handed to [`DatabaseArtifactRepository::read_chunks`] refused + /// the bytes; the message is the sink's own. + #[error("{0}")] + Sink(String), +} + +/// The stored metadata of one artifact row: everything but the payload bytes. +/// +/// Reading it never touches the `payload` / `delta_payload` columns, so its +/// cost is independent of the artifact's size. It is what an existence check +/// or a size comparison needs, and what [`DatabaseArtifactRepository::read_chunks`] +/// pins a streamed read to. +/// +/// Shape note for at-rest encoding (TC-887): sizes and hashes here describe the +/// STORED bytes, which today are the plain database file. An encoding column +/// (`payload_encoding`) would be added to this struct, and chunking would keep +/// applying to the stored (encoded) bytes, so decoding belongs in the sink. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ArtifactMeta { + pub revision: i64, + pub size_bytes: i64, + pub content_hash: String, + pub checkpoint_content_hash: String, + pub checkpoint_size_bytes: i64, + pub delta_content_hash: Option, + pub delta_size_bytes: i64, + pub updated_at: String, + pub storage_mode: String, + pub backend: String, +} + +impl ArtifactMeta { + /// Whether the row carries a WAL delta next to its checkpoint. + pub fn has_delta(&self) -> bool { + self.delta_content_hash.is_some() + } +} + +/// Which stored column a streamed byte range belongs to. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ArtifactPart { + Checkpoint, + Wal, +} + +impl ArtifactPart { + pub fn as_str(self) -> &'static str { + match self { + Self::Checkpoint => "checkpoint", + Self::Wal => "wal", + } + } +} + +/// Where [`DatabaseArtifactRepository::read_chunks`] delivers the artifact. +/// +/// Calls for one attempt arrive as `begin_part(Checkpoint)`, `write`*, +/// `finish_part`, then, when the row has a delta, the same for `Wal`. A part +/// with no bytes still gets `begin_part` and `finish_part`. Nothing the sink +/// received is final until the caller's own commit step: a sink must be able to +/// `reset` and take the whole sequence again (a changed row restarts the read), +/// and must publish nothing before its caller says the read succeeded. +#[async_trait] +pub trait ArtifactSink: Send { + /// Discard everything received so far; a new attempt is about to begin. + async fn reset(&mut self) -> Result<(), DatabaseArtifactError>; + async fn begin_part(&mut self, part: ArtifactPart) -> Result<(), DatabaseArtifactError>; + /// The stored size of the part just begun, for sinks that buffer it. + fn size_hint(&mut self, _part: ArtifactPart, _bytes: u64) {} + async fn write(&mut self, bytes: &[u8]) -> Result<(), DatabaseArtifactError>; + async fn finish_part(&mut self) -> Result<(), DatabaseArtifactError>; +} + +/// An [`ArtifactSink`] that keeps the bytes in memory, for callers that return +/// the whole artifact anyway (a DuckDB export) and for tests. +#[derive(Debug, Default)] +pub struct MemorySink { + pub checkpoint: Vec, + pub wal: Option>, + current: Option, +} + +#[async_trait] +impl ArtifactSink for MemorySink { + async fn reset(&mut self) -> Result<(), DatabaseArtifactError> { + *self = Self::default(); + Ok(()) + } + + async fn begin_part(&mut self, part: ArtifactPart) -> Result<(), DatabaseArtifactError> { + self.current = Some(part); + if part == ArtifactPart::Wal { + self.wal = Some(Vec::new()); + } + Ok(()) + } + + fn size_hint(&mut self, part: ArtifactPart, bytes: u64) { + let target = match part { + ArtifactPart::Checkpoint => Some(&mut self.checkpoint), + ArtifactPart::Wal => self.wal.as_mut(), + }; + if let Some(target) = target { + target.reserve_exact(usize::try_from(bytes).unwrap_or(0)); + } + } + + async fn write(&mut self, bytes: &[u8]) -> Result<(), DatabaseArtifactError> { + match self.current { + Some(ArtifactPart::Checkpoint) => self.checkpoint.extend_from_slice(bytes), + Some(ArtifactPart::Wal) => self + .wal + .get_or_insert_with(Vec::new) + .extend_from_slice(bytes), + None => return Err(DatabaseArtifactError::Sink("write outside a part".into())), + } + Ok(()) + } + + async fn finish_part(&mut self) -> Result<(), DatabaseArtifactError> { + self.current = None; + Ok(()) + } +} + +/// An [`ArtifactSink`] that stages the checkpoint and WAL as temp files next +/// to their final paths: nothing is visible at a final path until +/// [`CacheFileSink::commit`], which the caller runs only after +/// [`read_artifact`] succeeded. Dropping the sink (a failed, abandoned or +/// timed-out hydration) removes whatever was staged. +/// +/// The leading bytes of each part are checked with the given validators before +/// the part is accepted, so a payload the database engine would misread (and +/// silently revert to empty) is refused. +pub struct CacheFileSink { + service: &'static str, + label: String, + checkpoint_path: std::path::PathBuf, + wal_path: std::path::PathBuf, + valid_checkpoint: fn(&[u8]) -> bool, + valid_wal: fn(&[u8]) -> bool, + current: Option, + staged: Vec<(std::path::PathBuf, std::path::PathBuf)>, +} + +struct StagedPart { + part: ArtifactPart, + temp: std::path::PathBuf, + file: tokio::fs::File, + validated: bool, +} + +impl CacheFileSink { + pub fn new( + service: &'static str, + label: String, + checkpoint_path: std::path::PathBuf, + wal_path: std::path::PathBuf, + valid_checkpoint: fn(&[u8]) -> bool, + valid_wal: fn(&[u8]) -> bool, + ) -> Self { + Self { + service, + label, + checkpoint_path, + wal_path, + valid_checkpoint, + valid_wal, + current: None, + staged: Vec::new(), + } + } + + fn io(error: std::io::Error) -> DatabaseArtifactError { + DatabaseArtifactError::Sink(error.to_string()) + } + + fn validator(&self, part: ArtifactPart) -> fn(&[u8]) -> bool { + match part { + ArtifactPart::Checkpoint => self.valid_checkpoint, + ArtifactPart::Wal => self.valid_wal, + } + } + + fn validate(&self, part: ArtifactPart, head: &[u8]) -> Result<(), DatabaseArtifactError> { + if (self.validator(part))(head) { + return Ok(()); + } + tracing::error!( + service = self.service, + role = part.as_str(), + "Durable database artifact does not carry the expected file header; refusing to hydrate" + ); + Err(DatabaseArtifactError::Sink(format!( + "database artifact {} for {} does not carry the expected file header", + part.as_str(), + self.label + ))) + } + + fn discard(&mut self) { + if let Some(current) = self.current.take() { + let _ = std::fs::remove_file(¤t.temp); + release_staging(¤t.temp); + } + for (temp, _) in self.staged.drain(..) { + let _ = std::fs::remove_file(&temp); + release_staging(&temp); + } + } + + /// Move the staged files to their final paths. The caller has already + /// removed the stale cache files they replace. + pub async fn commit(&mut self) -> std::io::Result<()> { + // Renamed entries leave `staged`; on failure every one still in it + // (the failing temp included) is removed by `discard`. + while let Some((temp, path)) = self.staged.first().cloned() { + if let Err(error) = tokio::fs::rename(&temp, &path).await { + self.discard(); + return Err(error); + } + release_staging(&temp); + self.staged.remove(0); + } + Ok(()) + } +} + +impl Drop for CacheFileSink { + fn drop(&mut self) { + self.discard(); + } +} + +/// Staging files this process has created and not yet published or removed. +/// The sweep never touches these, so it cannot race a live write whose pid +/// matches a dead process's (pid reuse) or a write of another task. +fn live_staging() -> &'static std::sync::Mutex> { + static LIVE: std::sync::OnceLock< + std::sync::Mutex>, + > = std::sync::OnceLock::new(); + LIVE.get_or_init(Default::default) +} + +/// A temp sibling of `path`, unique per call, registered as live until +/// [`release_staging`]. +pub(crate) fn staging_path(path: &std::path::Path) -> std::path::PathBuf { + static NEXT_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + // Appended to the whole path, never `with_extension`: `main.db` and + // `main.db-wal` must not share a temp name. + let temp = std::path::PathBuf::from(format!( + "{}.tmp.{}.{}", + path.display(), + std::process::id(), + NEXT_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + )); + live_staging() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(temp.clone()); + temp +} + +/// The staging file at `temp` was published or removed: the sweep may take it +/// from here on (a no-op if it is already gone). +pub(crate) fn release_staging(temp: &std::path::Path) { + live_staging() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(temp); +} + +#[cfg(feature = "duckdb")] +/// Releases a [`staging_path`] when dropped, whichever way its writer ends. +pub(crate) struct StagingRelease(pub(crate) std::path::PathBuf); + +#[cfg(feature = "duckdb")] +impl Drop for StagingRelease { + fn drop(&mut self) { + release_staging(&self.0); + } +} + +/// Remove the `.tmp..` files a crashed hydration left beside +/// each of `finals`: exactly this sink's temp-name pattern, and never a file +/// this process still has in flight. Run before a hydration begins, under the +/// database's hydration exclusion. +pub(crate) async fn sweep_stale_staging(finals: &[&std::path::Path]) { + for path in finals { + let (Some(dir), Some(name)) = (path.parent(), path.file_name()) else { + continue; + }; + let prefix = format!("{}.tmp.", name.to_string_lossy()); + let Ok(mut entries) = tokio::fs::read_dir(dir).await else { + continue; + }; + while let Ok(Some(entry)) = entries.next_entry().await { + let file_name = entry.file_name(); + let Some(rest) = file_name + .to_string_lossy() + .strip_prefix(&prefix) + .map(str::to_owned) + else { + continue; + }; + let is_staging = matches!( + rest.split_once('.'), + Some((pid, id)) + if !pid.is_empty() + && !id.is_empty() + && pid.bytes().all(|b| b.is_ascii_digit()) + && id.bytes().all(|b| b.is_ascii_digit()) + ); + let temp = entry.path(); + let live = live_staging() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .contains(&temp); + if is_staging && !live && tokio::fs::remove_file(&temp).await.is_ok() { + tracing::info!("Removed a staging file left by an interrupted hydration"); + } + } + } +} + +#[async_trait] +impl ArtifactSink for CacheFileSink { + async fn reset(&mut self) -> Result<(), DatabaseArtifactError> { + self.discard(); + Ok(()) + } + + async fn begin_part(&mut self, part: ArtifactPart) -> Result<(), DatabaseArtifactError> { + let path = match part { + ArtifactPart::Checkpoint => self.checkpoint_path.clone(), + ArtifactPart::Wal => self.wal_path.clone(), + }; + if let Some(parent) = path.parent() { + tokio::fs::create_dir_all(parent).await.map_err(Self::io)?; + } + let temp = staging_path(&path); + let file = match tokio::fs::File::create(&temp).await { + Ok(file) => file, + Err(error) => { + release_staging(&temp); + return Err(Self::io(error)); + } + }; + self.current = Some(StagedPart { + part, + temp, + file, + validated: false, + }); + Ok(()) + } + + async fn write(&mut self, bytes: &[u8]) -> Result<(), DatabaseArtifactError> { + use tokio::io::AsyncWriteExt; + let Some(current) = self.current.as_ref() else { + return Err(DatabaseArtifactError::Sink("write outside a part".into())); + }; + // The first slice holds every header the validators look at. + if !current.validated { + self.validate(current.part, bytes)?; + } + let current = self.current.as_mut().expect("checked above"); + current.validated = true; + current.file.write_all(bytes).await.map_err(Self::io) + } + + async fn finish_part(&mut self) -> Result<(), DatabaseArtifactError> { + let Some(current) = self.current.as_ref() else { + return Err(DatabaseArtifactError::Sink("no part to finish".into())); + }; + // An empty part never saw a write; it is judged as the empty header. + if !current.validated { + self.validate(current.part, &[])?; + } + let StagedPart { + part, temp, file, .. + } = self.current.take().expect("checked above"); + if let Err(error) = file.sync_all().await { + let _ = tokio::fs::remove_file(&temp).await; + release_staging(&temp); + return Err(Self::io(error)); + } + drop(file); + let path = match part { + ArtifactPart::Checkpoint => self.checkpoint_path.clone(), + ArtifactPart::Wal => self.wal_path.clone(), + }; + self.staged.push((temp, path)); + Ok(()) + } +} + +/// Read an artifact into `sink` in bounded slices, restarting when the row +/// changes underneath the read. +/// +/// Returns the metadata the delivered bytes belong to, or `None` when there is +/// no such artifact (the sink is reset). A concurrent save between two +/// slices surfaces from [`DatabaseArtifactRepository::read_chunks`] as +/// [`DatabaseArtifactError::ArtifactChanged`]; the sink is `reset` and the read +/// restarts from fresh metadata, at most [`ARTIFACT_READ_RETRIES`] times, so a +/// sink never ends up holding a checkpoint of one revision and a WAL of +/// another. +pub async fn read_artifact( + repository: &R, + service: &str, + space: &str, + name: &str, + sink: &mut dyn ArtifactSink, +) -> Result, DatabaseArtifactError> { + let mut retries = 0; + loop { + let Some(meta) = repository.load_meta(service, space, name).await? else { + sink.reset().await?; + return Ok(None); + }; + sink.reset().await?; + match repository + .read_chunks(service, space, name, &meta, sink) + .await + { + Ok(()) => return Ok(Some(meta)), + Err(DatabaseArtifactError::ArtifactChanged) if retries < ARTIFACT_READ_RETRIES => { + retries += 1; + tracing::info!( + service, + retries, + "Database artifact changed mid-read; restarting" + ); + } + Err(error) => return Err(error), + } + } } #[async_trait] pub trait DatabaseArtifactRepository: Send + Sync { - async fn load( + /// The artifact's stored metadata, with no payload bytes read. + /// + /// One cheap statement whatever the artifact's size: use it for existence + /// checks and size comparisons, never `read_chunks`. + async fn load_meta( &self, service: &str, space: &str, name: &str, - ) -> Result, DatabaseArtifactError>; + ) -> Result, DatabaseArtifactError>; - /// [`Self::load`], told the artifact's stored size when the caller knows - /// it, so a repository can skip the bulk statement timeout's extra round - /// trips for a small artifact (TC-1054). `None` means unknown. - async fn load_sized( + /// Stream the stored checkpoint, then the WAL delta when `meta` has one, + /// into `sink` in slices of at most [`ARTIFACT_CHUNK_BYTES`]; an artifact + /// of at most [`BASE_TIMEOUT_PAYLOAD_BYTES`] stored bytes is one slice per + /// part. + /// + /// Peak memory is one slice on PostgreSQL. On SQLite it is one stored blob + /// transiently (the engine materialises the value before `substr` cuts it) + /// and CPU is O(n^2 / chunk) in the artifact size, so the bound there is + /// the part's size, not the slice. Every slice is + /// its own statement (`substr(column, offset, length)`, which runs on + /// both PostgreSQL `bytea` and SQLite `BLOB`) under the pool's base + /// timeout, guarded by `WHERE revision = meta.revision AND + /// checkpoint_content_hash = meta.checkpoint_content_hash`: a row that + /// moved on yields [`DatabaseArtifactError::ArtifactChanged`] instead of + /// bytes from two revisions. Round trips are `ceil(size / chunk)` per + /// non-empty part, on top of the one `load_meta` that produced `meta`. + /// + /// The checkpoint is hashed while it streams and compared to + /// `checkpoint_content_hash`, the WAL to `delta_content_hash`, and the row's + /// logical `content_hash` to what those imply; any mismatch is an error and + /// the caller must discard what the sink received. Prefer [`read_artifact`], + /// which also handles the restart. + /// + /// TC-887: chunking applies to the STORED bytes; an encoded payload would + /// be decoded by the sink (or a decoding wrapper around it), and the + /// hashes verified here would be the stored-bytes hashes. + async fn read_chunks( &self, service: &str, space: &str, name: &str, - _known_size: Option, - ) -> Result, DatabaseArtifactError> { - self.load(service, space, name).await - } + meta: &ArtifactMeta, + sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError>; async fn save( &self, @@ -159,11 +638,19 @@ pub trait DatabaseArtifactRepository: Send + Sync { /// [`RecoveringConnection::begin_bulk`] scope (TC-1054). It matches the SQL /// service's WAL delta bound (8 MiB): the server's work on an 8 MiB value /// (compression, TOAST, WAL) takes well under a second, far inside the 30 s -/// default. Loads use the same bound, judged by the stored size. +/// default. /// /// [`RecoveringConnection::begin_bulk`]: crate::db_pool::RecoveringConnection::begin_bulk pub const BASE_TIMEOUT_PAYLOAD_BYTES: usize = 8 * 1024 * 1024; +/// The slice size of a streamed artifact read. Two slices fit well inside +/// [`BASE_TIMEOUT_PAYLOAD_BYTES`]'s budget, so each runs under the base +/// statement timeout. +pub const ARTIFACT_CHUNK_BYTES: usize = 4 * 1024 * 1024; + +/// How many times [`read_artifact`] restarts a read whose row changed. +pub const ARTIFACT_READ_RETRIES: usize = 2; + /// The stored metadata a full checkpoint save is checked against. struct CurrentArtifact { revision: i64, @@ -223,64 +710,203 @@ impl SeaOrmDatabaseArtifactRepository { } } +struct ArtifactKey<'a> { + service: &'a str, + space: &'a str, + name: &'a str, +} + +impl SeaOrmDatabaseArtifactRepository { + /// Stream one payload column into `sink`, hash it, and return its CID. + async fn read_part( + &self, + key: &ArtifactKey<'_>, + meta: &ArtifactMeta, + part: ArtifactPart, + size_bytes: i64, + expected_hash: &str, + sink: &mut dyn ArtifactSink, + ) -> Result { + let column = match part { + ArtifactPart::Checkpoint => database_artifact::Column::Payload, + ArtifactPart::Wal => database_artifact::Column::DeltaPayload, + }; + let expected_len = u64::try_from(size_bytes).unwrap_or(0); + // An artifact that fits the base-timeout budget is one slice per part: + // no extra round trips for the common small database. + // Gated on the part's own size as well, so inconsistent metadata + // can never ask for a slice past the budget. + let budget = BASE_TIMEOUT_PAYLOAD_BYTES as i64; + let slice_bytes = if meta.size_bytes <= budget && size_bytes <= budget { + expected_len.max(1) + } else { + ARTIFACT_CHUNK_BYTES as u64 + }; + let mut hasher = Blake3Hasher::new(); + sink.begin_part(part).await?; + sink.size_hint(part, expected_len); + let mut offset = 0_u64; + while offset < expected_len { + let len = (expected_len - offset).min(slice_bytes); + // `substr` is 1-based and takes int4 on PostgreSQL. + let start = i32::try_from(offset + 1) + .map_err(|_| DatabaseArtifactError::PayloadTooLarge(expected_len))?; + let slice = database_artifact::Entity::find() + .select_only() + .expr( + Func::cust("substr") + .arg(Expr::col(column)) + .arg(start) + .arg(len as i32), + ) + .filter(database_artifact::Column::Service.eq(key.service)) + .filter(database_artifact::Column::Space.eq(key.space)) + .filter(database_artifact::Column::Name.eq(key.name)) + .filter(database_artifact::Column::Revision.eq(meta.revision)) + .filter( + database_artifact::Column::CheckpointContentHash + .eq(meta.checkpoint_content_hash.as_str()), + ) + .into_tuple::>>() + .one(&self.conn) + .await?; + // No row: the artifact was replaced or deleted since `meta`. + let Some(slice) = slice else { + return Err(DatabaseArtifactError::ArtifactChanged); + }; + let slice = slice.unwrap_or_default(); + if slice.len() as u64 != len { + return Err(DatabaseArtifactError::SizeMismatch { + part: part.as_str(), + expected: expected_len, + actual: offset + slice.len() as u64, + }); + } + hasher.update(&slice); + sink.write(&slice).await?; + offset += len; + } + let actual = hasher.finalize().to_cid(0x55).to_string(); + if actual != expected_hash { + return Err(DatabaseArtifactError::HashMismatch { + part: part.as_str(), + }); + } + sink.finish_part().await?; + Ok(actual) + } +} + #[async_trait] impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { - async fn load( + async fn load_meta( &self, service: &str, space: &str, name: &str, - ) -> Result, DatabaseArtifactError> { - self.load_sized(service, space, name, None).await + ) -> Result, DatabaseArtifactError> { + use database_artifact::Column; + let row = database_artifact::Entity::find_by_id(( + service.to_string(), + space.to_string(), + name.to_string(), + )) + .select_only() + .column(Column::Revision) + .column(Column::SizeBytes) + .column(Column::ContentHash) + .column(Column::CheckpointContentHash) + .column(Column::CheckpointSizeBytes) + .column(Column::DeltaContentHash) + .column(Column::DeltaSizeBytes) + .column(Column::UpdatedAt) + .column(Column::StorageMode) + .column(Column::Backend) + .into_tuple::<( + i64, + i64, + String, + String, + i64, + Option, + i64, + String, + String, + String, + )>() + .one(&self.conn) + .await?; + Ok(row.map( + |( + revision, + size_bytes, + content_hash, + checkpoint_content_hash, + checkpoint_size_bytes, + delta_content_hash, + delta_size_bytes, + updated_at, + storage_mode, + backend, + )| ArtifactMeta { + revision, + size_bytes, + content_hash, + checkpoint_content_hash, + checkpoint_size_bytes, + delta_content_hash, + delta_size_bytes, + updated_at, + storage_mode, + backend, + }, + )) } - async fn load_sized( + async fn read_chunks( &self, service: &str, space: &str, name: &str, - known_size: Option, - ) -> Result, DatabaseArtifactError> { - // Hydration reads the whole checkpoint and WAL in one statement. One - // that may be larger than a delta runs under the bulk statement - // timeout (TC-1054); a known-small one under the base timeout, with - // no extra round trip. - let bulk = known_size.is_none_or(|size| size > BASE_TIMEOUT_PAYLOAD_BYTES as u64); - let find = || { - database_artifact::Entity::find_by_id(( - service.to_string(), - space.to_string(), - name.to_string(), - )) + meta: &ArtifactMeta, + sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError> { + let key = ArtifactKey { + service, + space, + name, }; - let row = if bulk { - self.conn - .bulk_read("load_artifact", |scope| async move { - let row = find().one(&scope).await?; - scope.commit().await?; - Ok(row) - }) - .await - } else { - // A read-only statement on the pool, retried like any other. - find().one(&self.conn).await + let checkpoint = self + .read_part( + &key, + meta, + ArtifactPart::Checkpoint, + meta.checkpoint_size_bytes, + &meta.checkpoint_content_hash, + sink, + ) + .await?; + let logical = match &meta.delta_content_hash { + Some(delta_hash) => { + self.read_part( + &key, + meta, + ArtifactPart::Wal, + meta.delta_size_bytes, + delta_hash, + sink, + ) + .await?; + hash(format!("{checkpoint}:{delta_hash}").as_bytes()) + .to_cid(0x55) + .to_string() + } + None => checkpoint, }; - row.map(|row| { - row.map(|model| DatabaseArtifact { - payload: model.payload, - content_hash: model.content_hash, - revision: model.revision, - size_bytes: model.size_bytes, - updated_at: model.updated_at, - backend: model.backend, - storage_mode: model.storage_mode, - checkpoint_content_hash: model.checkpoint_content_hash, - delta_payload: model.delta_payload, - delta_content_hash: model.delta_content_hash, - delta_size_bytes: model.delta_size_bytes, - }) - }) - .map_err(DatabaseArtifactError::Db) + if logical != meta.content_hash { + return Err(DatabaseArtifactError::HashMismatch { part: "content" }); + } + Ok(()) } async fn save( @@ -575,8 +1201,55 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { } } +/// Test-only whole-artifact read built on the public streaming API, so tests +/// can assert on payload bytes. There is deliberately no such method on the +/// repository: the node never reads a whole row. +#[cfg(test)] +pub(crate) mod test_support { + use super::*; + + #[async_trait] + pub(crate) trait ReadAll { + async fn read_all( + &self, + service: &str, + space: &str, + name: &str, + ) -> Result, DatabaseArtifactError>; + } + + #[async_trait] + impl ReadAll for T { + async fn read_all( + &self, + service: &str, + space: &str, + name: &str, + ) -> Result, DatabaseArtifactError> { + let mut sink = MemorySink::default(); + let Some(meta) = read_artifact(self, service, space, name, &mut sink).await? else { + return Ok(None); + }; + Ok(Some(DatabaseArtifact { + payload: sink.checkpoint, + content_hash: meta.content_hash, + revision: meta.revision, + size_bytes: meta.size_bytes, + updated_at: meta.updated_at, + backend: meta.backend, + storage_mode: meta.storage_mode, + checkpoint_content_hash: meta.checkpoint_content_hash, + delta_payload: sink.wal, + delta_content_hash: meta.delta_content_hash, + delta_size_bytes: meta.delta_size_bytes, + })) + } + } +} + #[cfg(test)] mod tests { + use super::test_support::ReadAll; use super::*; use crate::migrations::Migrator; use sea_orm::{ConnectOptions, Database}; @@ -613,7 +1286,11 @@ mod tests { assert_eq!(delta.size_bytes, 112); assert_eq!(delta.delta_size_bytes, 12); - let loaded = repo.load("sql", "space", "main").await.unwrap().unwrap(); + let loaded = repo + .read_all("sql", "space", "main") + .await + .unwrap() + .unwrap(); assert_eq!(loaded.payload, vec![1; 100]); assert_eq!(loaded.delta_payload, Some(vec![2; 12])); assert_eq!(loaded.storage_mode, "checkpoint+wal"); @@ -627,7 +1304,11 @@ mod tests { ) .await .unwrap(); - let checkpoint = repo.load("sql", "space", "main").await.unwrap().unwrap(); + let checkpoint = repo + .read_all("sql", "space", "main") + .await + .unwrap() + .unwrap(); assert_eq!(checkpoint.size_bytes, 80); assert_eq!(checkpoint.delta_size_bytes, 0); assert_eq!(checkpoint.delta_payload, None); @@ -685,7 +1366,11 @@ mod tests { "expected a lineage rejection, got {err:?}" ); - let loaded = repo.load("sql", "space", "main").await.unwrap().unwrap(); + let loaded = repo + .read_all("sql", "space", "main") + .await + .unwrap() + .unwrap(); assert_eq!(loaded.revision, 2); assert_eq!(loaded.payload, vec![2; 200]); assert_eq!(loaded.delta_payload, None); @@ -741,7 +1426,11 @@ mod tests { "expected a lineage rejection, got {err:?}" ); - let loaded = repo.load("sql", "space", "main").await.unwrap().unwrap(); + let loaded = repo + .read_all("sql", "space", "main") + .await + .unwrap() + .unwrap(); assert_eq!(loaded.revision, 1); assert_eq!(loaded.payload, vec![1; 4096]); } @@ -815,11 +1504,736 @@ mod tests { // The stale writer must not have clobbered: durable state is the winner's // payload at revision 2, never the loser's bytes. - let loaded = repo.load("sql", "space", "main").await.unwrap().unwrap(); + let loaded = repo + .read_all("sql", "space", "main") + .await + .unwrap() + .unwrap(); assert_eq!(loaded.revision, 2); assert_eq!( loaded.payload, winner.payload, "stale writer must not overwrite the winner's checkpoint" ); } + + // ---- bounded streaming reads (TC-822) ---- + + use sea_orm::{ConnectionTrait, DatabaseConnection}; + use std::sync::{Arc, Mutex}; + + const CHUNK: usize = ARTIFACT_CHUNK_BYTES; + + /// The replacement (checkpoint, optional WAL) a disturbance saves. + type Disturb = fn(usize) -> (Vec, Option>); + + fn bytes(len: usize, seed: u8) -> Vec { + (0..len) + .map(|i| (i as u8).wrapping_mul(31).wrapping_add(seed)) + .collect() + } + + /// A migrated in-memory SQLite repository plus a log of every statement + /// it sends, the "counting connection" of the round-trip assertions. + async fn spied_repo() -> ( + SeaOrmDatabaseArtifactRepository, + DatabaseConnection, + Arc>>, + ) { + let mut options = ConnectOptions::new("sqlite::memory:".to_string()); + options.max_connections(1); + let mut conn = Database::connect(options).await.unwrap(); + Migrator::up(&conn, None).await.unwrap(); + let statements = Arc::new(Mutex::new(Vec::new())); + let log = statements.clone(); + conn.set_metric_callback(move |info| { + log.lock().unwrap().push(info.statement.sql.clone()); + }); + ( + SeaOrmDatabaseArtifactRepository::new(conn.clone()), + conn, + statements, + ) + } + + fn ceil_div(len: usize) -> usize { + len.div_ceil(CHUNK) + } + + #[tokio::test] + async fn chunk_boundaries_round_trip_within_the_round_trip_budget() { + let (repo, _conn, statements) = spied_repo().await; + let sizes = [0, 1, CHUNK - 1, CHUNK, CHUNK + 1]; + let deltas = [None, Some(1), Some(CHUNK), Some(CHUNK + 1)]; + for (index, len) in sizes.into_iter().enumerate() { + for (delta_index, delta) in deltas.into_iter().enumerate() { + let name = format!("db-{index}-{delta_index}"); + let payload = bytes(len, 1); + repo.save( + "sql", + "space", + &name, + payload.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let wal = delta.map(|len| bytes(len, 2)); + if let Some(wal) = &wal { + repo.save_delta("sql", "space", &name, wal.clone(), ArtifactExpectation::Any) + .await + .unwrap(); + } + + statements.lock().unwrap().clear(); + let mut sink = MemorySink::default(); + let meta = read_artifact(&repo, "sql", "space", &name, &mut sink) + .await + .unwrap() + .expect("the artifact exists"); + assert_eq!(sink.checkpoint, payload, "{name}: checkpoint"); + assert_eq!(sink.wal, wal, "{name}: wal"); + assert_eq!(meta.has_delta(), wal.is_some(), "{name}"); + + // Every statement is a retryable read: a lost connection costs + // one slice's retry, not the hydration. + for sql in statements.lock().unwrap().iter() { + assert!(crate::db_pool::is_read_only_statement(sql), "{sql}"); + } + // One metadata statement, then one per slice of each part; an + // artifact within the base-timeout budget is one slice a part. + let total = len + delta.unwrap_or(0); + let parts = |part: usize| { + if total <= BASE_TIMEOUT_PAYLOAD_BYTES { + usize::from(part > 0) + } else { + ceil_div(part) + } + }; + let expected = 1 + parts(len) + delta.map_or(0, parts); + assert_eq!( + statements.lock().unwrap().len(), + expected, + "{name}: round trips for {len} + {delta:?} bytes" + ); + } + } + } + + #[tokio::test] + async fn metadata_and_absence_never_select_payload_columns() { + let (repo, _conn, statements) = spied_repo().await; + repo.save( + "sql", + "space", + "main", + bytes(CHUNK + 5, 3), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + repo.save_delta( + "sql", + "space", + "main", + bytes(9, 4), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + + statements.lock().unwrap().clear(); + let meta = repo + .load_meta("sql", "space", "main") + .await + .unwrap() + .unwrap(); + assert_eq!(meta.revision, 2); + assert_eq!(meta.checkpoint_size_bytes, (CHUNK + 5) as i64); + assert_eq!(meta.delta_size_bytes, 9); + assert_eq!(meta.size_bytes, (CHUNK + 14) as i64); + assert_eq!(meta.storage_mode, "checkpoint+wal"); + assert!(repo + .load_meta("sql", "space", "absent") + .await + .unwrap() + .is_none()); + + let seen = statements.lock().unwrap().clone(); + assert_eq!(seen.len(), 2, "{seen:?}"); + assert!( + seen.iter().all(|sql| !sql.contains("payload")), + "a metadata read must not touch payload columns: {seen:?}" + ); + + // Reading an absent artifact through the streaming helper stops at the + // metadata too. + statements.lock().unwrap().clear(); + let mut sink = MemorySink::default(); + assert!(read_artifact(&repo, "sql", "space", "absent", &mut sink) + .await + .unwrap() + .is_none()); + assert_eq!(statements.lock().unwrap().len(), 1); + } + + /// Disturbs the row once per attempt, right after the first slice landed, + /// the way a concurrent save would between two slice statements. + struct RaceSink<'a> { + inner: MemorySink, + repo: &'a SeaOrmDatabaseArtifactRepository, + disturbances_left: usize, + disturbed_this_attempt: bool, + disturb: Disturb, + attempts: usize, + } + + #[async_trait] + impl ArtifactSink for RaceSink<'_> { + async fn reset(&mut self) -> Result<(), DatabaseArtifactError> { + self.attempts += 1; + self.disturbed_this_attempt = false; + self.inner.reset().await + } + async fn begin_part(&mut self, part: ArtifactPart) -> Result<(), DatabaseArtifactError> { + self.inner.begin_part(part).await + } + async fn write(&mut self, bytes: &[u8]) -> Result<(), DatabaseArtifactError> { + self.inner.write(bytes).await?; + if self.disturbances_left > 0 && !self.disturbed_this_attempt { + self.disturbed_this_attempt = true; + self.disturbances_left -= 1; + let (checkpoint, wal) = (self.disturb)(self.disturbances_left); + self.repo + .save("sql", "space", "main", checkpoint, ArtifactExpectation::Any) + .await + .unwrap(); + if let Some(wal) = wal { + self.repo + .save_delta("sql", "space", "main", wal, ArtifactExpectation::Any) + .await + .unwrap(); + } + } + Ok(()) + } + async fn finish_part(&mut self) -> Result<(), DatabaseArtifactError> { + self.inner.finish_part().await + } + } + + fn race_sink( + repo: &SeaOrmDatabaseArtifactRepository, + disturbances: usize, + disturb: Disturb, + ) -> RaceSink<'_> { + RaceSink { + inner: MemorySink::default(), + repo, + disturbances_left: disturbances, + disturbed_this_attempt: false, + disturb, + attempts: 0, + } + } + + #[tokio::test] + async fn a_row_replaced_mid_read_restarts_and_never_mixes_revisions() { + let (repo, _conn, _) = spied_repo().await; + repo.save( + "sql", + "space", + "main", + bytes(2 * CHUNK + 7, 1), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let replacement = bytes(CHUNK + 3, 9); + let wal = bytes(11, 8); + + let mut sink = race_sink(&repo, 1, |_| (bytes(CHUNK + 3, 9), Some(bytes(11, 8)))); + let meta = read_artifact(&repo, "sql", "space", "main", &mut sink) + .await + .unwrap() + .unwrap(); + assert_eq!(sink.attempts, 2, "one restart"); + assert_eq!(meta.revision, 3, "save, save, then the WAL delta"); + assert_eq!(sink.inner.checkpoint, replacement, "only the new revision"); + assert_eq!(sink.inner.wal, Some(wal)); + } + + #[tokio::test] + async fn a_wal_only_change_mid_read_also_restarts() { + let (repo, _conn, _) = spied_repo().await; + repo.save( + "sql", + "space", + "main", + bytes(2 * CHUNK + 1, 1), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + repo.save_delta( + "sql", + "space", + "main", + bytes(5, 2), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + + // The checkpoint hash is untouched by a delta, so the revision guard + // is what catches this. + struct DeltaRace<'a>(RaceSink<'a>, bool); + #[async_trait] + impl ArtifactSink for DeltaRace<'_> { + async fn reset(&mut self) -> Result<(), DatabaseArtifactError> { + self.0.reset().await + } + async fn begin_part( + &mut self, + part: ArtifactPart, + ) -> Result<(), DatabaseArtifactError> { + self.0.begin_part(part).await + } + async fn write(&mut self, bytes: &[u8]) -> Result<(), DatabaseArtifactError> { + self.0.inner.write(bytes).await?; + if !self.1 { + self.1 = true; + self.0 + .repo + .save_delta("sql", "space", "main", vec![7; 6], ArtifactExpectation::Any) + .await + .unwrap(); + } + Ok(()) + } + async fn finish_part(&mut self) -> Result<(), DatabaseArtifactError> { + self.0.finish_part().await + } + } + let mut sink = DeltaRace(race_sink(&repo, 0, |_| unreachable!()), false); + let meta = read_artifact(&repo, "sql", "space", "main", &mut sink) + .await + .unwrap() + .unwrap(); + assert_eq!(meta.revision, 3); + assert_eq!(sink.0.attempts, 2); + assert_eq!(sink.0.inner.wal, Some(vec![7; 6])); + assert_eq!(sink.0.inner.checkpoint, bytes(2 * CHUNK + 1, 1)); + } + + #[tokio::test] + async fn a_row_that_keeps_changing_gives_up_after_the_retries() { + let (repo, _conn, _) = spied_repo().await; + repo.save( + "sql", + "space", + "main", + bytes(2 * CHUNK + 1, 1), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let mut sink = race_sink(&repo, 10, |left| (bytes(2 * CHUNK + 1, left as u8), None)); + let error = read_artifact(&repo, "sql", "space", "main", &mut sink) + .await + .expect_err("every attempt is overtaken"); + assert!( + matches!(error, DatabaseArtifactError::ArtifactChanged), + "{error:?}" + ); + assert_eq!(sink.attempts, 1 + ARTIFACT_READ_RETRIES); + } + + fn dir_names(dir: &std::path::Path) -> Vec { + let mut names: Vec = std::fs::read_dir(dir) + .unwrap() + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .collect(); + names.sort(); + names + } + + fn cache_sink(dir: &std::path::Path) -> CacheFileSink { + CacheFileSink::new( + "sql", + "space/main".to_string(), + dir.join("main.db"), + dir.join("main.db-wal"), + |_| true, + |_| true, + ) + } + + #[tokio::test] + async fn a_cache_file_is_never_seeded_from_a_mixed_read() { + let (repo, _conn, _) = spied_repo().await; + repo.save( + "sql", + "space", + "main", + bytes(2 * CHUNK + 7, 1), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let dir = tempfile::tempdir().unwrap(); + let mut files = cache_sink(dir.path()); + + // Overtaken after the first slice of the old checkpoint reached the + // temp file; the restart must rewrite it from the new revision. + struct Overtaken<'a>( + &'a mut CacheFileSink, + &'a SeaOrmDatabaseArtifactRepository, + bool, + ); + #[async_trait] + impl ArtifactSink for Overtaken<'_> { + async fn reset(&mut self) -> Result<(), DatabaseArtifactError> { + self.0.reset().await + } + async fn begin_part( + &mut self, + part: ArtifactPart, + ) -> Result<(), DatabaseArtifactError> { + self.0.begin_part(part).await + } + async fn write(&mut self, bytes: &[u8]) -> Result<(), DatabaseArtifactError> { + self.0.write(bytes).await?; + if !self.2 { + self.2 = true; + self.1 + .save( + "sql", + "space", + "main", + vec![5; CHUNK + 2], + ArtifactExpectation::Any, + ) + .await + .unwrap(); + } + Ok(()) + } + async fn finish_part(&mut self) -> Result<(), DatabaseArtifactError> { + self.0.finish_part().await + } + } + read_artifact( + &repo, + "sql", + "space", + "main", + &mut Overtaken(&mut files, &repo, false), + ) + .await + .unwrap() + .unwrap(); + assert!( + dir_names(dir.path()) + .iter() + .all(|name| name.contains(".tmp.")), + "nothing is published before commit: {:?}", + dir_names(dir.path()) + ); + files.commit().await.unwrap(); + assert_eq!(dir_names(dir.path()), ["main.db"]); + assert_eq!( + std::fs::read(dir.path().join("main.db")).unwrap(), + vec![5; CHUNK + 2] + ); + } + + #[tokio::test] + async fn corrupt_artifact_bytes_are_rejected_and_nothing_is_seeded() { + let (repo, conn, _) = spied_repo().await; + let len = CHUNK + 10; + repo.save( + "sql", + "space", + "main", + bytes(len, 1), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + repo.save_delta( + "sql", + "space", + "main", + bytes(20, 2), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + + for (what, sql, part) in [ + ( + "checkpoint flipped, same length", + format!("UPDATE database_artifact SET payload = zeroblob({len})"), + "checkpoint", + ), + ( + "wal flipped, same length", + "UPDATE database_artifact SET delta_payload = zeroblob(20)".to_string(), + "wal", + ), + ] { + // Restore a good row before each corruption. + repo.save( + "sql", + "space", + "main", + bytes(len, 1), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + repo.save_delta( + "sql", + "space", + "main", + bytes(20, 2), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + conn.execute_unprepared(&sql).await.unwrap(); + + let dir = tempfile::tempdir().unwrap(); + let mut sink = cache_sink(dir.path()); + let error = read_artifact(&repo, "sql", "space", "main", &mut sink) + .await + .expect_err(what); + assert!( + matches!(&error, DatabaseArtifactError::HashMismatch { part: got } if *got == part), + "{what}: {error:?}" + ); + drop(sink); + assert_eq!( + dir_names(dir.path()), + Vec::::new(), + "{what}: nothing left behind" + ); + } + + // A stored size that disagrees with the bytes is refused too. + repo.save( + "sql", + "space", + "main", + bytes(len, 1), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + conn.execute_unprepared("UPDATE database_artifact SET payload = zeroblob(10)") + .await + .unwrap(); + let error = read_artifact(&repo, "sql", "space", "main", &mut MemorySink::default()) + .await + .expect_err("short payload"); + assert!( + matches!(error, DatabaseArtifactError::SizeMismatch { .. }), + "{error:?}" + ); + } + + #[tokio::test] + async fn the_cache_sink_checks_the_leading_bytes_of_each_part() { + let (repo, _conn, _) = spied_repo().await; + repo.save( + "sql", + "space", + "main", + b"MAGIC-body".to_vec(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let dir = tempfile::tempdir().unwrap(); + let mut sink = CacheFileSink::new( + "sql", + "space/main".to_string(), + dir.path().join("main.db"), + dir.path().join("main.db-wal"), + |head| head.starts_with(b"OTHER"), + |_| true, + ); + let error = read_artifact(&repo, "sql", "space", "main", &mut sink) + .await + .expect_err("wrong header"); + assert!( + error.to_string().contains("expected file header"), + "{error}" + ); + drop(sink); + assert!(dir_names(dir.path()).is_empty()); + } + + #[tokio::test] + async fn postgres_slices_round_trip_at_chunk_boundaries_and_detect_changes() { + let Some(fixture) = crate::test_support::PostgresTestSchema::migrated_through( + "postgres_slices_round_trip_at_chunk_boundaries_and_detect_changes", + None, + ) + .await + else { + return; + }; + let repo = SeaOrmDatabaseArtifactRepository::new(fixture.conn.clone()); + for (index, len) in [0, 1, CHUNK - 1, CHUNK, CHUNK + 1].into_iter().enumerate() { + for (delta_index, delta) in [None, Some(1), Some(CHUNK + 1)].into_iter().enumerate() { + let name = format!("db-{index}-{delta_index}"); + let payload = bytes(len, 1); + repo.save( + "sql", + "space", + &name, + payload.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let wal = delta.map(|len| bytes(len, 2)); + if let Some(wal) = &wal { + repo.save_delta("sql", "space", &name, wal.clone(), ArtifactExpectation::Any) + .await + .unwrap(); + } + let mut sink = MemorySink::default(); + read_artifact(&repo, "sql", "space", &name, &mut sink) + .await + .unwrap() + .unwrap(); + assert_eq!(sink.checkpoint, payload, "{name}: checkpoint"); + assert_eq!(sink.wal, wal, "{name}: wal"); + } + } + + // Mid-read replacement restarts on Postgres too. + repo.save( + "sql", + "space", + "main", + bytes(2 * CHUNK + 7, 1), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let mut sink = race_sink(&repo, 1, |_| (bytes(CHUNK + 3, 9), None)); + read_artifact(&repo, "sql", "space", "main", &mut sink) + .await + .unwrap() + .unwrap(); + assert_eq!(sink.attempts, 2); + assert_eq!(sink.inner.checkpoint, bytes(CHUNK + 3, 9)); + + // And so does corruption. + fixture + .conn + .execute_unprepared("UPDATE database_artifact SET payload = decode(repeat('00', 8), 'hex') || substr(payload, 9) WHERE name = 'main'") + .await + .unwrap(); + let error = read_artifact(&repo, "sql", "space", "main", &mut MemorySink::default()) + .await + .expect_err("corrupt"); + assert!( + matches!(error, DatabaseArtifactError::HashMismatch { .. }), + "{error:?}" + ); + } + + fn names(dir: &std::path::Path) -> Vec { + let mut found: Vec = std::fs::read_dir(dir) + .unwrap() + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .collect(); + found.sort(); + found + } + + async fn stage_both(sink: &mut CacheFileSink) { + for (part, body) in [ + (ArtifactPart::Checkpoint, b"cp"), + (ArtifactPart::Wal, b"wl"), + ] { + sink.begin_part(part).await.unwrap(); + sink.write(body).await.unwrap(); + sink.finish_part().await.unwrap(); + } + } + + #[tokio::test] + async fn a_failed_commit_leaves_no_staged_temp_files() { + for blocked in ["main.db", "main.db-wal"] { + let dir = tempfile::tempdir().unwrap(); + let mut sink = cache_sink(dir.path()); + stage_both(&mut sink).await; + // A non-empty directory at the destination makes the rename fail. + let target = dir.path().join(blocked); + std::fs::create_dir(&target).unwrap(); + std::fs::write(target.join("occupied"), b"x").unwrap(); + + sink.commit().await.unwrap_err(); + + assert!( + names(dir.path()).iter().all(|name| !name.contains(".tmp.")), + "{blocked}: {:?}", + names(dir.path()) + ); + drop(sink); + assert!(names(dir.path()).iter().all(|name| !name.contains(".tmp."))); + } + } + + #[tokio::test] + async fn the_sweep_removes_only_orphaned_staging_files() { + let dir = tempfile::tempdir().unwrap(); + let orphans = ["main.db.tmp.4242.0", "main.db-wal.tmp.4242.7"]; + let kept = [ + "main.db", + "main.db-wal", + "other.db.tmp.4242.0", + "main.db.tmp.abc.def", + "main.db.tmp.1", + "main.db.tmp.", + "main.db.tmpx.1.2", + ]; + for name in orphans.iter().chain(kept.iter()) { + std::fs::write(dir.path().join(name), b"x").unwrap(); + } + // A hydration in progress: one part still streaming, one finished. + let mut live = cache_sink(dir.path()); + live.begin_part(ArtifactPart::Checkpoint).await.unwrap(); + live.write(b"cp").await.unwrap(); + live.finish_part().await.unwrap(); + live.begin_part(ArtifactPart::Wal).await.unwrap(); + let live_files: Vec = names(dir.path()) + .into_iter() + .filter(|name| name.contains(&format!(".tmp.{}.", std::process::id()))) + .collect(); + assert_eq!(live_files.len(), 2, "{live_files:?}"); + + let main = dir.path().join("main.db"); + let wal = dir.path().join("main.db-wal"); + sweep_stale_staging(&[&main, &wal]).await; + + let left = names(dir.path()); + for name in orphans { + assert!(!left.iter().any(|n| n == name), "{name} survived: {left:?}"); + } + for name in kept.iter().map(|n| n.to_string()).chain(live_files) { + assert!(left.contains(&name), "{name} was removed: {left:?}"); + } + + // Once the hydration ends, its files are no longer protected. + drop(live); + assert!(names(dir.path()) + .iter() + .all(|n| !n.contains(&format!(".tmp.{}.", std::process::id())))); + } } diff --git a/tinycloud-core/src/db_pool.rs b/tinycloud-core/src/db_pool.rs index 700271f3..cf1dab56 100644 --- a/tinycloud-core/src/db_pool.rs +++ b/tinycloud-core/src/db_pool.rs @@ -282,7 +282,7 @@ impl RecoveringConnection { /// transaction opened with `SET LOCAL statement_timeout`: three more /// round trips (`BEGIN`, `SET LOCAL`, `COMMIT`), so use it only on /// infrequent paths. Statements in it are not retried after a lost - /// connection; wrap a read in [`Self::bulk_read`] for that. Everywhere + /// connection; a read-only scope is not retried either. Everywhere /// else the scope is the pool itself, statement by statement, under the /// usual retry policy. pub async fn begin_bulk(&self) -> Result { @@ -306,30 +306,6 @@ impl RecoveringConnection { Ok(BulkScope::Transaction(transaction)) } - /// Run `read` in a [`Self::begin_bulk_read`] scope, and run the whole - /// attempt once more if it lost its connection within the retry window, - /// as a read-only statement on the pool would be. The attempt's `BEGIN` is - /// not retried on its own: nested retries would let one request wait for - /// several connection-loss detections. - pub async fn bulk_read(&self, operation: &'static str, read: F) -> Result - where - F: Fn(BulkScope) -> Fut, - Fut: Future>, - { - let Some(timeout) = self.bulk_statement_timeout() else { - // The scope is the pool, whose statements retry themselves. - return read(BulkScope::Pool(self.clone())).await; - }; - self.run(operation, true, || async { - let transaction = self.conn.begin().await?; - transaction - .execute_unprepared(&set_local_statement_timeout(timeout)) - .await?; - read(BulkScope::Transaction(transaction)).await - }) - .await - } - /// The `statement_timeout` a [`Self::begin_bulk`] transaction runs with, or /// `None` when no statement timeout applies to this pool. fn bulk_statement_timeout(&self) -> Option { @@ -985,7 +961,7 @@ const WRITE_KEYWORDS: [&str; 5] = ["INSERT", "UPDATE", "DELETE", "MERGE", "INTO" /// of anything else (`nextval`, `setval`, `pg_advisory_lock`, /// `pg_terminate_backend`, `set_config`, `lo_*`, a user-defined function) /// may have side effects, so it makes the statement non-retryable. -const CALL_ALLOWLIST: [&str; 36] = [ +const CALL_ALLOWLIST: [&str; 37] = [ // keywords "SELECT", "FROM", @@ -1024,6 +1000,7 @@ const CALL_ALLOWLIST: [&str; 36] = [ "COALESCE", "LOWER", "UPPER", + "SUBSTR", ]; #[derive(Debug, PartialEq, Eq)] diff --git a/tinycloud-core/src/duckdb/service.rs b/tinycloud-core/src/duckdb/service.rs index 49e469b5..ac88d42d 100644 --- a/tinycloud-core/src/duckdb/service.rs +++ b/tinycloud-core/src/duckdb/service.rs @@ -1,17 +1,15 @@ use std::{ collections::HashMap, path::{Path, PathBuf}, - sync::{ - atomic::{AtomicU64, Ordering}, - Arc, Weak, - }, + sync::{Arc, Weak}, }; use dashmap::DashMap; use tinycloud_auth::resource::SpaceId; use crate::database_artifacts::{ - ArtifactExpectation, DatabaseArtifactError, DatabaseArtifactRepository, + read_artifact, staging_path, sweep_stale_staging, ArtifactExpectation, CacheFileSink, + DatabaseArtifactError, DatabaseArtifactRepository, MemorySink, StagingRelease, }; use crate::request_telemetry::{self, QueryTimer}; @@ -207,21 +205,28 @@ impl DuckDbService { let operation_lock = Self::database_lock(&self.operation_locks, &key).await; let _operation = operation_lock.lock().await; - let artifact = self - .artifact_repository - .load("duckdb", &key.0, db_name) - .await - .map_err(artifact_error_to_duckdb)? - .ok_or(DuckDbError::DatabaseNotFound)?; + // The export is the response body, so it is buffered whole; the read + // itself still streams in slices. + let mut artifact = MemorySink::default(); + read_artifact( + self.artifact_repository.as_ref(), + "duckdb", + &key.0, + db_name, + &mut artifact, + ) + .await + .map_err(artifact_error_to_duckdb)? + .ok_or(DuckDbError::DatabaseNotFound)?; validate_payload( space, db_name, "checkpoint", - &artifact.payload, + &artifact.checkpoint, is_duckdb_file, )?; - let Some(delta) = artifact.delta_payload else { - return Ok(artifact.payload); + let Some(delta) = artifact.wal else { + return Ok(artifact.checkpoint); }; validate_payload(space, db_name, "wal", &delta, |payload| { !is_duckdb_file(payload) @@ -236,7 +241,7 @@ impl DuckDbService { let _context = request_telemetry::enter(context); let temp = tempfile::tempdir().map_err(|e| DuckDbError::Internal(e.to_string()))?; let path = temp.path().join("export.duckdb"); - std::fs::write(&path, artifact.payload) + std::fs::write(&path, artifact.checkpoint) .map_err(|e| DuckDbError::Internal(e.to_string()))?; std::fs::write(duckdb_wal_path(&path), delta) .map_err(|e| DuckDbError::Internal(e.to_string()))?; @@ -274,7 +279,8 @@ impl DuckDbService { .map_err(|e| DuckDbError::Internal(e.to_string()))?; let final_path = dir.join(format!("{}.duckdb", db_name)); - let temp_path = temp_write_path(&final_path); + let temp_path = staging_path(&final_path); + let _live = StagingRelease(temp_path.clone()); // Write to temp file first tokio::fs::write(&temp_path, data) @@ -414,51 +420,46 @@ impl DuckDbService { async fn hydrate_cache(&self, space: &SpaceId, db_name: &str) -> Result<(), DuckDbError> { let cache_path = self.cache_path(space, db_name); let key = (space.to_string(), db_name.to_string()); - match self - .artifact_repository - .load("duckdb", &space.to_string(), db_name) - .await - .map_err(artifact_error_to_duckdb)? + sweep_stale_staging(&[&cache_path, &duckdb_wal_path(&cache_path)]).await; + let mut sink = CacheFileSink::new( + "duckdb", + format!("{space}/{db_name}"), + cache_path.clone(), + duckdb_wal_path(&cache_path), + is_duckdb_file, + is_not_duckdb_file, + ); + match read_artifact( + self.artifact_repository.as_ref(), + "duckdb", + &space.to_string(), + db_name, + &mut sink, + ) + .await + .map_err(artifact_error_to_duckdb)? { - Some(artifact) => { + Some(meta) => { tracing::info!( service = "duckdb", - revision = artifact.revision, - storage_mode = %artifact.storage_mode, - bytes = artifact.payload.len(), - delta_bytes = artifact.delta_size_bytes, - logical_bytes = artifact.size_bytes, - content_hash = %artifact.content_hash, - checkpoint_content_hash = %artifact.checkpoint_content_hash, + revision = meta.revision, + storage_mode = %meta.storage_mode, + bytes = meta.checkpoint_size_bytes, + delta_bytes = meta.delta_size_bytes, + logical_bytes = meta.size_bytes, + content_hash = %meta.content_hash, + checkpoint_content_hash = %meta.checkpoint_content_hash, "Loaded database artifact" ); - // Refuse to seed the cache with bytes DuckDB would misread. The - // checkpoint must be a database; the WAL, which carries no - // header of its own, must at least not be one — that is the - // shape a crossed hydration writes. - validate_payload( - space, - db_name, - "checkpoint", - &artifact.payload, - is_duckdb_file, - )?; - if let Some(delta) = artifact.delta_payload.as_deref() { - validate_payload(space, db_name, "wal", delta, |payload| { - !is_duckdb_file(payload) - })?; - } - remove_duckdb_cache_files(&cache_path).await?; - write_cache_file(&cache_path, &artifact.payload).await?; - if let Some(delta) = artifact.delta_payload { - write_cache_file(&duckdb_wal_path(&cache_path), &delta).await?; - } + sink.commit() + .await + .map_err(|e| DuckDbError::Internal(e.to_string()))?; self.lineage.insert( key, ArtifactExpectation::Derived { - revision: artifact.revision, - checkpoint_content_hash: artifact.checkpoint_content_hash, + revision: meta.revision, + checkpoint_content_hash: meta.checkpoint_content_hash, }, ); Ok(()) @@ -571,7 +572,7 @@ impl DuckDbService { if without_growth { let durable_bytes = self .artifact_repository - .load("duckdb", &key.0, db_name) + .load_meta("duckdb", &key.0, db_name) .await .map_err(artifact_error_to_duckdb)? .map_or(0, |artifact| artifact.size_bytes.max(0) as u64); @@ -603,6 +604,12 @@ impl DuckDbService { } } +/// The WAL carries no header of its own; it must at least not be a database, +/// which is the shape a crossed hydration writes. +fn is_not_duckdb_file(payload: &[u8]) -> bool { + !is_duckdb_file(payload) +} + /// Whether `payload` carries DuckDB's main-database header. fn is_duckdb_file(payload: &[u8]) -> bool { payload @@ -633,42 +640,6 @@ fn validate_payload( ))) } -/// A temp sibling of `path`, unique per call. -/// -/// The suffix is APPENDED to the whole path. `Path::with_extension` replaces -/// everything after the last `.`, so `main.duckdb` and `main.duckdb.wal` both -/// mapped to `main.duckdb.tmp` — as did `import_db`'s own staging file. The -/// checkpoint and WAL writes of one hydration raced on a single file and -/// cross-consumed each other's bytes, reverting the database silently. -fn temp_write_path(path: &Path) -> PathBuf { - static NEXT_TEMP_ID: AtomicU64 = AtomicU64::new(0); - PathBuf::from(format!( - "{}.tmp.{}.{}", - path.display(), - std::process::id(), - NEXT_TEMP_ID.fetch_add(1, Ordering::Relaxed) - )) -} - -async fn write_cache_file(path: &Path, payload: &[u8]) -> Result<(), DuckDbError> { - if let Some(parent) = path.parent() { - tokio::fs::create_dir_all(parent) - .await - .map_err(|e| DuckDbError::Internal(e.to_string()))?; - } - - let temp_path = temp_write_path(path); - if let Err(e) = tokio::fs::write(&temp_path, payload).await { - let _ = tokio::fs::remove_file(&temp_path).await; - return Err(DuckDbError::Internal(e.to_string())); - } - if let Err(e) = tokio::fs::rename(&temp_path, path).await { - let _ = tokio::fs::remove_file(&temp_path).await; - return Err(DuckDbError::Internal(e.to_string())); - } - Ok(()) -} - async fn remove_duckdb_cache_files(path: &Path) -> Result<(), DuckDbError> { for candidate in [ path.to_path_buf(), @@ -688,6 +659,8 @@ fn duckdb_wal_path(path: &Path) -> PathBuf { PathBuf::from(format!("{}.wal", path.display())) } +/// `DuckDbError` has no retryable-unavailable variant (adding one would reach +/// every route mapping), so an exhausted `ArtifactChanged` stays `Internal`. fn artifact_error_to_duckdb(err: DatabaseArtifactError) -> DuckDbError { DuckDbError::Internal(err.to_string()) } @@ -696,7 +669,7 @@ fn artifact_error_to_duckdb(err: DatabaseArtifactError) -> DuckDbError { mod tests { use super::*; use crate::{ - database_artifacts::SeaOrmDatabaseArtifactRepository, + database_artifacts::{test_support::ReadAll, SeaOrmDatabaseArtifactRepository}, migrations::Migrator, sea_orm::{ConnectOptions, Database}, sea_orm_migration::MigratorTrait, @@ -788,7 +761,7 @@ mod tests { assert_eq!(sizes.space_total(&space).await, 0); assert!(matches!(exported, Err(DuckDbError::DatabaseNotFound))); assert!(repo - .load("duckdb", &space.to_string(), name) + .read_all("duckdb", &space.to_string(), name) .await .unwrap() .is_none()); @@ -847,7 +820,7 @@ mod tests { assert_eq!(sizes.space_total(&space).await, 0); assert!(matches!(exported, Err(DuckDbError::DatabaseNotFound))); assert!(repo - .load("duckdb", &space.to_string(), "absent") + .read_all("duckdb", &space.to_string(), "absent") .await .unwrap() .is_none()); @@ -888,7 +861,7 @@ mod tests { .expect("dropping an absent table is a successful no-op"); assert_eq!(sizes.space_total(&space).await, 0); assert!(repo - .load("duckdb", &space.to_string(), name) + .read_all("duckdb", &space.to_string(), name) .await .unwrap() .is_none()); @@ -933,7 +906,7 @@ mod tests { .await .unwrap(); let before = repo - .load("duckdb", &space.to_string(), "main") + .read_all("duckdb", &space.to_string(), "main") .await .unwrap() .unwrap(); @@ -973,7 +946,7 @@ mod tests { "{rejected:?}" ); let after = repo - .load("duckdb", &space.to_string(), "main") + .read_all("duckdb", &space.to_string(), "main") .await .unwrap() .unwrap(); @@ -1174,7 +1147,7 @@ mod tests { .await .unwrap(); let artifact = repo - .load("duckdb", &space.to_string(), "analytics") + .read_all("duckdb", &space.to_string(), "analytics") .await .unwrap() .unwrap(); @@ -1205,7 +1178,7 @@ mod tests { .unwrap(); assert_eq!(names, "one"); let after = repo - .load("duckdb", &space.to_string(), "analytics") + .read_all("duckdb", &space.to_string(), "analytics") .await .unwrap() .unwrap(); diff --git a/tinycloud-core/src/sql/service.rs b/tinycloud-core/src/sql/service.rs index f569c0ae..f9c634e4 100644 --- a/tinycloud-core/src/sql/service.rs +++ b/tinycloud-core/src/sql/service.rs @@ -3,7 +3,7 @@ use std::{ future::Future, path::{Path, PathBuf}, sync::{ - atomic::{AtomicU64, AtomicU8, Ordering}, + atomic::{AtomicU8, Ordering}, Arc, Weak, }, time::Duration, @@ -15,7 +15,8 @@ use tokio::{sync::OwnedMutexGuard, time::Instant}; use tracing::Instrument; use crate::database_artifacts::{ - ArtifactExpectation, DatabaseArtifact, DatabaseArtifactError, DatabaseArtifactRepository, + read_artifact, sweep_stale_staging, ArtifactExpectation, ArtifactMeta, CacheFileSink, + DatabaseArtifactError, DatabaseArtifactRepository, }; use crate::db_pool::{is_transient_error, track_writes, transient_error_class}; use crate::{request_cancel, request_telemetry}; @@ -521,7 +522,7 @@ impl SqlService { || SqlBusyStage::ArtifactTransfer, self.bounds.artifact_transfer, self.artifact_repository - .load("sql", &space.to_string(), db_name), + .load_meta("sql", &space.to_string(), db_name), ) .await? .map_err(artifact_error_to_sql)? @@ -680,18 +681,34 @@ impl SqlService { (HydrationFlight { phase, outcome }, start) } - /// Load the durable artifact, seed the cache from it and spawn the actor. - /// Only the load is bounded (`hydration_max`): the cache writes after it - /// are local and must finish once begun. + /// Stream the durable artifact into the cache, seed it and spawn the actor. + /// Only the streaming read is bounded (`hydration_max`): it stages temp + /// files that a timeout simply drops, while the commit after it is local + /// and must finish once begun. async fn hydrate(&self, space: &SpaceId, db_name: &str, phase: &AtomicU8) -> HydrationOutcome { + let cache_path = self.cache_path(space, db_name); + sweep_stale_staging(&[&cache_path, &sql_wal_path(&cache_path)]).await; + let mut sink = CacheFileSink::new( + "sql", + format!("{space}/{db_name}"), + cache_path.clone(), + sql_wal_path(&cache_path), + is_sqlite_database, + is_sqlite_wal, + ); let loaded = tokio::time::timeout( self.bounds.hydration_max, - self.artifact_repository - .load("sql", &space.to_string(), db_name), + read_artifact( + self.artifact_repository.as_ref(), + "sql", + &space.to_string(), + db_name, + &mut sink, + ), ) .await; - let artifact = match loaded { - Ok(Ok(artifact)) => artifact, + let meta = match loaded { + Ok(Ok(meta)) => meta, Ok(Err(error)) => return hydration_failure(artifact_error_to_sql(error)), Err(_) => { tracing::warn!( @@ -704,7 +721,7 @@ impl SqlService { } }; phase.store(PHASE_WRITING, Ordering::Release); - if let Err(error) = self.seed_cache(space, db_name, artifact).await { + if let Err(error) = self.seed_cache(space, db_name, meta, sink).await { return hydration_failure(error); } self.databases @@ -721,53 +738,39 @@ impl SqlService { HydrationOutcome::Ready } + /// Replace the cache files with the verified artifact staged in `sink` + /// (or with nothing, when there is no artifact). async fn seed_cache( &self, space: &SpaceId, db_name: &str, - artifact: Option, + meta: Option, + mut sink: CacheFileSink, ) -> Result<(), SqlError> { let cache_path = self.cache_path(space, db_name); let key = (space.to_string(), db_name.to_string()); - match artifact { - Some(artifact) => { + match meta { + Some(meta) => { tracing::info!( service = "sql", - revision = artifact.revision, - storage_mode = %artifact.storage_mode, - bytes = artifact.payload.len(), - delta_bytes = artifact.delta_size_bytes, - logical_bytes = artifact.size_bytes, - content_hash = %artifact.content_hash, - checkpoint_content_hash = %artifact.checkpoint_content_hash, + revision = meta.revision, + storage_mode = %meta.storage_mode, + bytes = meta.checkpoint_size_bytes, + delta_bytes = meta.delta_size_bytes, + logical_bytes = meta.size_bytes, + content_hash = %meta.content_hash, + checkpoint_content_hash = %meta.checkpoint_content_hash, "Loaded database artifact" ); - // Refuse to seed the cache with bytes SQLite would misread. A - // checkpoint that is not a database opens as a fresh empty one, - // and a WAL that is not a WAL is ignored outright — both revert - // the database silently, and the next write makes the reverted - // state durable. - validate_payload( - space, - db_name, - "checkpoint", - &artifact.payload, - is_sqlite_database, - )?; - if let Some(delta) = artifact.delta_payload.as_deref() { - validate_payload(space, db_name, "wal", delta, is_sqlite_wal)?; - } - remove_sql_cache_files(&cache_path).await?; - write_cache_file(&cache_path, &artifact.payload).await?; - if let Some(delta) = artifact.delta_payload { - write_cache_file(&sql_wal_path(&cache_path), &delta).await?; - } + sink.commit() + .await + .map_err(|e| SqlError::Internal(e.to_string()))?; self.lineage.insert( key, ArtifactExpectation::Derived { - revision: artifact.revision, - checkpoint_content_hash: artifact.checkpoint_content_hash, + revision: meta.revision, + checkpoint_content_hash: meta.checkpoint_content_hash, }, ); Ok(()) @@ -885,7 +888,7 @@ impl SqlService { .bounded( || SqlBusyStage::ArtifactTransfer, self.bounds.artifact_transfer, - self.artifact_repository.load("sql", &key.0, db_name), + self.artifact_repository.load_meta("sql", &key.0, db_name), ) .await? .map_err(artifact_error_to_sql)? @@ -939,66 +942,6 @@ fn is_sqlite_wal(payload: &[u8]) -> bool { .any(|magic| payload.starts_with(magic)) } -/// Reject a hydration payload whose leading bytes are not the format the file -/// it is about to become is read as. -fn validate_payload( - space: &SpaceId, - db_name: &str, - role: &'static str, - payload: &[u8], - is_valid: impl Fn(&[u8]) -> bool, -) -> Result<(), SqlError> { - if is_valid(payload) { - return Ok(()); - } - tracing::error!( - service = "sql", - role, - bytes = payload.len(), - "Durable database artifact does not carry the expected file header; refusing to hydrate" - ); - Err(SqlError::Internal(format!( - "database artifact {role} for {space}/{db_name} does not carry the expected file header" - ))) -} - -/// A temp sibling of `path`, unique per call. -/// -/// The suffix is APPENDED to the whole path. `Path::with_extension` replaces -/// everything after the last `.`, so `main.db` and `main.db-wal` both mapped to -/// `main.db.tmp`: the checkpoint and WAL writes of one hydration raced on a -/// single file and cross-consumed each other's bytes, leaving either a WAL that -/// SQLite ignores (silent revert to the checkpoint) or a "database" with no -/// SQLite header (silent revert to empty). -fn temp_write_path(path: &Path) -> PathBuf { - static NEXT_TEMP_ID: AtomicU64 = AtomicU64::new(0); - PathBuf::from(format!( - "{}.tmp.{}.{}", - path.display(), - std::process::id(), - NEXT_TEMP_ID.fetch_add(1, Ordering::Relaxed) - )) -} - -async fn write_cache_file(path: &Path, payload: &[u8]) -> Result<(), SqlError> { - if let Some(parent) = path.parent() { - tokio::fs::create_dir_all(parent) - .await - .map_err(|e| SqlError::Internal(e.to_string()))?; - } - - let temp_path = temp_write_path(path); - if let Err(e) = tokio::fs::write(&temp_path, payload).await { - let _ = tokio::fs::remove_file(&temp_path).await; - return Err(SqlError::Internal(e.to_string())); - } - if let Err(e) = tokio::fs::rename(&temp_path, path).await { - let _ = tokio::fs::remove_file(&temp_path).await; - return Err(SqlError::Internal(e.to_string())); - } - Ok(()) -} - async fn remove_sql_cache_files(path: &Path) -> Result<(), SqlError> { for candidate in [ path.to_path_buf(), @@ -1028,6 +971,16 @@ fn artifact_error_to_sql(err: DatabaseArtifactError) -> SqlError { ); SqlError::Unavailable } + // The row kept changing across every restart. Nothing was written, so + // a fresh request is safe: the same answer as any contended store. + DatabaseArtifactError::ArtifactChanged => { + tracing::warn!( + service = "sql", + error_class = "artifact_changed", + "Database artifact kept changing while it was read" + ); + SqlError::Unavailable + } err => SqlError::Internal(err.to_string()), } } @@ -1105,14 +1058,19 @@ fn log_busy(stage: SqlBusyStage, bound: Duration, waited: Duration) { #[cfg(test)] mod tests { use super::*; + use crate::database_artifacts::staging_path as temp_write_path; use crate::{ - database_artifacts::{DatabaseArtifact, DeltaSave, SeaOrmDatabaseArtifactRepository}, + database_artifacts::{ + test_support::ReadAll, ArtifactMeta, ArtifactSink, DatabaseArtifact, DeltaSave, + SeaOrmDatabaseArtifactRepository, + }, migrations::Migrator, sea_orm::{ConnectOptions, Database, DbErr}, sea_orm_migration::MigratorTrait, sql_sizes::{SizeTrackingArtifactRepository, SqlSizes}, }; use async_trait::async_trait; + use std::sync::atomic::AtomicU64; use tempfile::TempDir; use tinycloud_auth::{ resolver::DID_METHODS, @@ -1197,7 +1155,7 @@ mod tests { )); assert_eq!(sizes.space_total(&space).await, 0); assert!(repo - .load("sql", &space.to_string(), name) + .read_all("sql", &space.to_string(), name) .await .unwrap() .is_none()); @@ -1245,7 +1203,7 @@ mod tests { )); assert_eq!(sizes.space_total(&space).await, 0); assert!(repo - .load("sql", &space.to_string(), "main") + .read_all("sql", &space.to_string(), "main") .await .unwrap() .is_none()); @@ -1284,7 +1242,7 @@ mod tests { .expect("dropping an absent table is a successful no-op"); assert_eq!(sizes.space_total(&space).await, 0); assert!(repo - .load("sql", &space.to_string(), name) + .read_all("sql", &space.to_string(), name) .await .unwrap() .is_none()); @@ -1326,7 +1284,7 @@ mod tests { .await .unwrap(); let before = repo - .load("sql", &space.to_string(), "main") + .read_all("sql", &space.to_string(), "main") .await .unwrap() .unwrap(); @@ -1367,7 +1325,7 @@ mod tests { "{rejected:?}" ); let after = repo - .load("sql", &space.to_string(), "main") + .read_all("sql", &space.to_string(), "main") .await .unwrap() .unwrap(); @@ -2087,7 +2045,7 @@ mod tests { .unwrap(); } let mut charged_bytes = repo - .load("sql", &space.to_string(), "main") + .read_all("sql", &space.to_string(), "main") .await .unwrap() .unwrap() @@ -2103,7 +2061,7 @@ mod tests { .await .unwrap_or_else(|e| panic!("{name}: non-growing write must pass: {e}")); let artifact = repo - .load("sql", &space.to_string(), "main") + .read_all("sql", &space.to_string(), "main") .await .unwrap() .unwrap(); @@ -2188,7 +2146,7 @@ mod tests { .await .unwrap_or_else(|e| panic!("{name}: DROP must pass: {e}")); let artifact = repo - .load("sql", &space.to_string(), "main") + .read_all("sql", &space.to_string(), "main") .await .unwrap() .unwrap(); @@ -2252,7 +2210,7 @@ mod tests { .await .unwrap(); let charged_bytes = repo - .load("sql", &space.to_string(), "main") + .read_all("sql", &space.to_string(), "main") .await .unwrap() .unwrap() @@ -2303,7 +2261,7 @@ mod tests { .await .unwrap(); let artifact = repo - .load("sql", &space.to_string(), "main") + .read_all("sql", &space.to_string(), "main") .await .unwrap() .unwrap(); @@ -2528,7 +2486,7 @@ mod tests { .unwrap(); let artifact = repo - .load("sql", &space.to_string(), "main") + .read_all("sql", &space.to_string(), "main") .await .unwrap() .unwrap(); @@ -2548,7 +2506,7 @@ mod tests { // still has to contain both acknowledged writes. service.export(&space, "main").await.unwrap(); let after_export = repo - .load("sql", &space.to_string(), "main") + .read_all("sql", &space.to_string(), "main") .await .unwrap() .unwrap(); @@ -2710,14 +2668,27 @@ mod tests { #[async_trait] impl DatabaseArtifactRepository for CountingArtifactRepository { - async fn load( + async fn load_meta( &self, service: &str, space: &str, name: &str, - ) -> Result, DatabaseArtifactError> { + ) -> Result, DatabaseArtifactError> { self.loads.fetch_add(1, Ordering::Relaxed); - self.inner.load(service, space, name).await + self.inner.load_meta(service, space, name).await + } + + async fn read_chunks( + &self, + service: &str, + space: &str, + name: &str, + meta: &ArtifactMeta, + sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError> { + self.inner + .read_chunks(service, space, name, meta, sink) + .await } async fn save( @@ -2860,13 +2831,26 @@ mod tests { #[async_trait] impl DatabaseArtifactRepository for LostReplyArtifactRepository { - async fn load( + async fn load_meta( + &self, + service: &str, + space: &str, + name: &str, + ) -> Result, DatabaseArtifactError> { + self.inner.load_meta(service, space, name).await + } + + async fn read_chunks( &self, service: &str, space: &str, name: &str, - ) -> Result, DatabaseArtifactError> { - self.inner.load(service, space, name).await + meta: &ArtifactMeta, + sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError> { + self.inner + .read_chunks(service, space, name, meta, sink) + .await } async fn save( @@ -2962,14 +2946,27 @@ mod tests { #[async_trait] impl DatabaseArtifactRepository for FlakyArtifactRepository { - async fn load( + async fn load_meta( &self, service: &str, space: &str, name: &str, - ) -> Result, DatabaseArtifactError> { + ) -> Result, DatabaseArtifactError> { self.fail()?; - self.inner.load(service, space, name).await + self.inner.load_meta(service, space, name).await + } + + async fn read_chunks( + &self, + service: &str, + space: &str, + name: &str, + meta: &ArtifactMeta, + sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError> { + self.inner + .read_chunks(service, space, name, meta, sink) + .await } async fn save( @@ -3117,15 +3114,26 @@ mod tests { #[async_trait] impl DatabaseArtifactRepository for FailingArtifactRepository { - async fn load( + async fn load_meta( &self, _service: &str, _space: &str, _name: &str, - ) -> Result, DatabaseArtifactError> { + ) -> Result, DatabaseArtifactError> { Ok(None) } + async fn read_chunks( + &self, + _service: &str, + _space: &str, + _name: &str, + _meta: &ArtifactMeta, + _sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError> { + Ok(()) + } + async fn save( &self, _service: &str, @@ -3298,12 +3306,12 @@ mod tests { #[async_trait] impl DatabaseArtifactRepository for GatedArtifactRepository { - async fn load( + async fn load_meta( &self, service: &str, space: &str, name: &str, - ) -> Result, DatabaseArtifactError> { + ) -> Result, DatabaseArtifactError> { self.loads.fetch_add(1, Ordering::SeqCst); Self::pass(&self.load_open).await; if self.fail_loads.load(Ordering::SeqCst) { @@ -3311,7 +3319,20 @@ mod tests { sea_orm::error::ConnAcquireErr::Timeout, ))); } - self.inner.load(service, space, name).await + self.inner.load_meta(service, space, name).await + } + + async fn read_chunks( + &self, + service: &str, + space: &str, + name: &str, + meta: &ArtifactMeta, + sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError> { + self.inner + .read_chunks(service, space, name, meta, sink) + .await } async fn save( @@ -3402,8 +3423,16 @@ mod tests { async fn tc1274_a_held_operation_lock_answers_sql_busy_lock_wait() { let (repo, space) = seeded("sql-busy-lock").await; let cache = TempDir::new().unwrap(); - let service = SqlService::new(cache.path().to_string_lossy().into_owned(), u64::MAX, repo) - .with_wait_bounds(wait_bounds(200)); + // Warm both databases under the default bounds first: the hydration + // (disk, fsync) is not what this test asserts, only the lock wait, + // which must not depend on how fast the disk is. + let service = SqlService::new(cache.path().to_string_lossy().into_owned(), u64::MAX, repo); + read(&service, &space).await.unwrap(); + service + .execute(&space, "b", insert("w"), None, "tinycloud.sql/admin".into()) + .await + .unwrap(); + let service = service.with_wait_bounds(wait_bounds(200)); let held = service.operation_lock(&space, "a").await.lock_owned().await; let err = tokio::time::timeout(PATIENCE, read(&service, &space)) @@ -3425,6 +3454,25 @@ mod tests { assert_eq!(rows(read(&service, &space).await.unwrap()), 1); } + #[tokio::test] + async fn a_hydration_sweeps_temp_files_a_crashed_one_left_behind() { + let (repo, space) = seeded("sql-sweep-orphans").await; + let cache = TempDir::new().unwrap(); + let dir = cache.path().join(space.to_string()); + std::fs::create_dir_all(&dir).unwrap(); + // pid 1 never matches this process's live files: a crashed hydration's. + let orphan = dir.join("a.db.tmp.1.0"); + let orphan_wal = dir.join("a.db-wal.tmp.1.1"); + let unrelated = dir.join("b.db.tmp.1.0"); + for file in [&orphan, &orphan_wal, &unrelated] { + std::fs::write(file, b"x").unwrap(); + } + let service = SqlService::new(cache.path().to_string_lossy().into_owned(), u64::MAX, repo); + assert_eq!(rows(read(&service, &space).await.unwrap()), 1); + assert!(!orphan.exists() && !orphan_wal.exists()); + assert!(unrelated.exists(), "another database's file is not ours"); + } + #[tokio::test] async fn tc1274_a_stalled_hydration_answers_busy_and_still_warms_once() { let (inner, space) = seeded("sql-busy-hydration").await; @@ -3748,6 +3796,9 @@ mod tests { ); assert!(service.hydrations.lock().unwrap().is_empty()); + // The fresh hydration does real disk work (fsync); give it the + // default bound so only the gated first attempt is bound-limited. + let service = service.with_wait_bounds(SqlWaitBounds::default()); repo.load_open.send_replace(true); assert_eq!(rows(read(&service, &space).await.unwrap()), 1); assert_eq!(repo.loads(), 2, "the next request hydrated afresh"); @@ -3820,16 +3871,29 @@ mod tests { #[async_trait] impl DatabaseArtifactRepository for PanickingArtifactRepository { - async fn load( + async fn load_meta( &self, service: &str, space: &str, name: &str, - ) -> Result, DatabaseArtifactError> { + ) -> Result, DatabaseArtifactError> { if !self.panicked.swap(true, Ordering::SeqCst) { panic!("injected hydration panic"); } - self.inner.load(service, space, name).await + self.inner.load_meta(service, space, name).await + } + + async fn read_chunks( + &self, + service: &str, + space: &str, + name: &str, + meta: &ArtifactMeta, + sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError> { + self.inner + .read_chunks(service, space, name, meta, sink) + .await } async fn save( @@ -3910,4 +3974,284 @@ mod tests { assert!(!context.latch(), "the request was cut, not latched"); assert_eq!(rows(read(&service, &space).await.unwrap()), 1); } + + /// Counts metadata reads and the payload bytes delivered to sinks. + struct ByteSpyRepository { + inner: Arc, + metas: AtomicU64, + chunk_reads: AtomicU64, + payload_bytes: Arc, + } + + struct CountingSink<'a>(&'a mut dyn ArtifactSink, Arc); + + #[async_trait] + impl ArtifactSink for CountingSink<'_> { + async fn reset(&mut self) -> Result<(), DatabaseArtifactError> { + self.0.reset().await + } + async fn begin_part( + &mut self, + part: crate::database_artifacts::ArtifactPart, + ) -> Result<(), DatabaseArtifactError> { + self.0.begin_part(part).await + } + async fn write(&mut self, bytes: &[u8]) -> Result<(), DatabaseArtifactError> { + self.1.fetch_add(bytes.len() as u64, Ordering::SeqCst); + self.0.write(bytes).await + } + async fn finish_part(&mut self) -> Result<(), DatabaseArtifactError> { + self.0.finish_part().await + } + } + + #[async_trait] + impl DatabaseArtifactRepository for ByteSpyRepository { + async fn load_meta( + &self, + service: &str, + space: &str, + name: &str, + ) -> Result, DatabaseArtifactError> { + self.metas.fetch_add(1, Ordering::SeqCst); + self.inner.load_meta(service, space, name).await + } + + async fn read_chunks( + &self, + service: &str, + space: &str, + name: &str, + meta: &ArtifactMeta, + sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError> { + self.chunk_reads.fetch_add(1, Ordering::SeqCst); + let mut counting = CountingSink(sink, Arc::clone(&self.payload_bytes)); + self.inner + .read_chunks(service, space, name, meta, &mut counting) + .await + } + + async fn save( + &self, + service: &str, + space: &str, + name: &str, + payload: Vec, + expected: ArtifactExpectation, + ) -> Result { + self.inner + .save(service, space, name, payload, expected) + .await + } + + async fn save_delta( + &self, + service: &str, + space: &str, + name: &str, + payload: Vec, + expected: ArtifactExpectation, + ) -> Result { + self.inner + .save_delta(service, space, name, payload, expected) + .await + } + } + + #[tokio::test] + async fn forbid_growth_checks_and_export_existence_read_no_payload_bytes() { + let spy = Arc::new(ByteSpyRepository { + inner: artifact_repository().await, + metas: AtomicU64::new(0), + chunk_reads: AtomicU64::new(0), + payload_bytes: Arc::new(AtomicU64::new(0)), + }); + let cache = TempDir::new().unwrap(); + let space = test_space_id("sql-zero-payload"); + let service = SqlService::new( + cache.path().to_string_lossy().into_owned(), + u64::MAX, + spy.clone(), + ); + let write = |sql: &str| SqlRequest::Execute { + schema: Some(vec![ + "CREATE TABLE IF NOT EXISTS notes (id INTEGER PRIMARY KEY, body TEXT)".into(), + ]), + sql: sql.into(), + params: vec![], + }; + service + .execute( + &space, + "main", + write("INSERT INTO notes VALUES (1, 'kept')"), + None, + "tinycloud.sql/admin".into(), + ) + .await + .unwrap(); + let reset = || { + spy.metas.store(0, Ordering::SeqCst); + spy.chunk_reads.store(0, Ordering::SeqCst); + spy.payload_bytes.store(0, Ordering::SeqCst); + }; + + // Export of a live, present database: existence from metadata alone. + reset(); + assert!(!service.export(&space, "main").await.unwrap().is_empty()); + assert_eq!(spy.metas.load(Ordering::SeqCst), 1); + assert_eq!(spy.chunk_reads.load(Ordering::SeqCst), 0); + assert_eq!(spy.payload_bytes.load(Ordering::SeqCst), 0); + + // Export of a missing database. + reset(); + assert!(matches!( + service.export(&space, "absent").await, + Err(SqlError::DatabaseNotFound) + )); + assert_eq!(spy.payload_bytes.load(Ordering::SeqCst), 0); + assert_eq!(spy.chunk_reads.load(Ordering::SeqCst), 0); + + // A write that would grow a full space: the durable size comes from + // metadata, not from reading the checkpoint back. + reset(); + let rejected = service + .execute_without_growth( + &space, + "main", + SqlRequest::Execute { + schema: None, + sql: "INSERT INTO notes VALUES (2, ?)".into(), + params: vec![SqlValue::Text("x".repeat(256 * 1024))], + }, + None, + "tinycloud.sql/write".into(), + ) + .await; + assert!(matches!(rejected, Err(SqlError::StorageWouldGrow))); + assert_eq!(spy.chunk_reads.load(Ordering::SeqCst), 0); + assert_eq!(spy.payload_bytes.load(Ordering::SeqCst), 0); + + // A write that fits: it reaches the Forbid size comparison, which is + // answered from metadata. (Warm the actor first; hydration does read.) + service + .execute( + &space, + "main", + SqlRequest::Query { + sql: "SELECT COUNT(*) FROM notes".into(), + params: vec![], + max_rows: None, + max_bytes: None, + }, + None, + "tinycloud.sql/read".into(), + ) + .await + .unwrap(); + reset(); + service + .execute_without_growth( + &space, + "main", + SqlRequest::Execute { + schema: None, + sql: "UPDATE notes SET body = 'kept'".into(), + params: vec![], + }, + None, + "tinycloud.sql/write".into(), + ) + .await + .unwrap(); + assert!( + spy.metas.load(Ordering::SeqCst) >= 1, + "the Forbid path consulted the durable size" + ); + assert_eq!(spy.chunk_reads.load(Ordering::SeqCst), 0); + assert_eq!(spy.payload_bytes.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn hydration_rejects_a_corrupt_artifact_and_seeds_nothing() { + use sea_orm::ConnectionTrait; + let db = Database::connect(ConnectOptions::new("sqlite::memory:".to_string())) + .await + .unwrap(); + Migrator::up(&db, None).await.unwrap(); + let repo = Arc::new(SeaOrmDatabaseArtifactRepository::new(db.clone())); + let space = test_space_id("sql-corrupt-hydration"); + let seed_cache = TempDir::new().unwrap(); + let seed = SqlService::new( + seed_cache.path().to_string_lossy().into_owned(), + u64::MAX, + repo.clone(), + ); + seed.execute( + &space, + "main", + SqlRequest::Execute { + schema: Some(vec!["CREATE TABLE t (id INTEGER)".into()]), + sql: "INSERT INTO t VALUES (1)".into(), + params: vec![], + }, + None, + "tinycloud.sql/admin".into(), + ) + .await + .unwrap(); + // Keep the SQLite header so only the content hash can catch it. + let stored = repo + .read_all("sql", &space.to_string(), "main") + .await + .unwrap() + .unwrap(); + let mut flipped = stored.payload.clone(); + *flipped.last_mut().unwrap() ^= 0xff; + db.execute(sea_orm::Statement::from_sql_and_values( + sea_orm::DbBackend::Sqlite, + "UPDATE database_artifact SET payload = ?", + [flipped.into()], + )) + .await + .unwrap(); + + let cold_cache = TempDir::new().unwrap(); + let cold = SqlService::new( + cold_cache.path().to_string_lossy().into_owned(), + u64::MAX, + repo, + ); + let result = cold + .execute( + &space, + "main", + SqlRequest::Query { + sql: "SELECT COUNT(*) FROM t".into(), + params: vec![], + max_rows: None, + max_bytes: None, + }, + None, + "tinycloud.sql/read".into(), + ) + .await; + assert!(result.is_err(), "a corrupt artifact must not hydrate"); + let leftovers: Vec<_> = walkdir_files(cold_cache.path()); + assert!(leftovers.is_empty(), "nothing seeded: {leftovers:?}"); + } + + fn walkdir_files(dir: &Path) -> Vec { + let mut found = Vec::new(); + for entry in std::fs::read_dir(dir).into_iter().flatten().flatten() { + let path = entry.path(); + if path.is_dir() { + found.extend(walkdir_files(&path)); + } else { + found.push(path); + } + } + found + } } diff --git a/tinycloud-core/src/sql_sizes.rs b/tinycloud-core/src/sql_sizes.rs index 812447d6..34ab520a 100644 --- a/tinycloud-core/src/sql_sizes.rs +++ b/tinycloud-core/src/sql_sizes.rs @@ -5,8 +5,8 @@ use tinycloud_auth::resource::SpaceId; use tokio::sync::RwLock; use crate::database_artifacts::{ - ArtifactExpectation, DatabaseArtifact, DatabaseArtifactError, DatabaseArtifactRepository, - DeltaSave, + ArtifactExpectation, ArtifactMeta, ArtifactSink, DatabaseArtifact, DatabaseArtifactError, + DatabaseArtifactRepository, DeltaSave, }; /// Process-local mirror of durable SQL artifact sizes, keyed by space then @@ -16,11 +16,7 @@ use crate::database_artifacts::{ /// single writer, the Postgres compose runs one node container. #[derive(Debug, Clone, Default)] #[allow(clippy::type_complexity)] -pub struct SqlSizes( - Arc>>>, - /// Set once [`SqlSizes::seed_from`] has loaded every artifact's size. - Arc, -); +pub struct SqlSizes(Arc>>>); impl SqlSizes { pub fn new() -> Self { @@ -56,30 +52,9 @@ impl SqlSizes { .insert((service, name), size.max(0) as u64); } *self.0.write().await = fresh; // overwrite the map wholesale with DB truth - self.1.store(true, std::sync::atomic::Ordering::Release); Ok(()) } - /// The stored size of one artifact as far as this node knows: its - /// recorded size, `Some(0)` for one absent from a seeded mirror (this - /// node is the artifacts' only writer), `None` before the mirror is - /// seeded. Stale only after a save whose outcome was unknown, by at most - /// that save's delta. - pub async fn known_size(&self, service: &str, space: &str, name: &str) -> Option { - if !self.1.load(std::sync::atomic::Ordering::Acquire) { - return None; - } - Some( - self.0 - .read() - .await - .get(space) - .and_then(|inner| inner.get(&(service.to_string(), name.to_string()))) - .copied() - .unwrap_or(0), - ) - } - /// Overwrite (not accumulate) the size for one (service, space, db-name). pub async fn update(&self, service: &str, space: &str, name: &str, size_bytes: u64) { self.0 @@ -122,27 +97,25 @@ impl SizeTrackingArtifactRepository { #[async_trait] impl DatabaseArtifactRepository for SizeTrackingArtifactRepository { - async fn load( + async fn load_meta( &self, service: &str, space: &str, name: &str, - ) -> Result, DatabaseArtifactError> { - let known_size = self.sizes.known_size(service, space, name).await; - self.inner - .load_sized(service, space, name, known_size) - .await + ) -> Result, DatabaseArtifactError> { + self.inner.load_meta(service, space, name).await } - async fn load_sized( + async fn read_chunks( &self, service: &str, space: &str, name: &str, - known_size: Option, - ) -> Result, DatabaseArtifactError> { + meta: &ArtifactMeta, + sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError> { self.inner - .load_sized(service, space, name, known_size) + .read_chunks(service, space, name, meta, sink) .await } @@ -197,24 +170,6 @@ mod tests { ssi::{dids::DIDBuf, jwk::JWK}, }; - #[tokio::test] - async fn known_sizes_are_unknown_until_seeded() { - let db = Database::connect(ConnectOptions::new("sqlite::memory:".to_string())) - .await - .unwrap(); - Migrator::up(&db, None).await.unwrap(); - let sizes = SqlSizes::new(); - assert_eq!(sizes.known_size("sql", "space", "db").await, None); - sizes.seed_from(&db).await.unwrap(); - assert_eq!( - sizes.known_size("sql", "space", "db").await, - Some(0), - "absent" - ); - sizes.update("sql", "space", "db", 9 << 20).await; - assert_eq!(sizes.known_size("sql", "space", "db").await, Some(9 << 20)); - } - fn test_space_id(name: &str) -> SpaceId { let jwk = JWK::generate_ed25519().unwrap(); let did: DIDBuf = DID_METHODS.generate(&jwk, "key").unwrap(); diff --git a/tinycloud-core/tests/artifact_bounded_read.rs b/tinycloud-core/tests/artifact_bounded_read.rs new file mode 100644 index 00000000..6d5ae21d --- /dev/null +++ b/tinycloud-core/tests/artifact_bounded_read.rs @@ -0,0 +1,130 @@ +//! TC-822: hydrating a large artifact streams it in slices, so the Rust heap +//! the node holds for it does not grow with the artifact. +//! +//! The row is inserted with `zeroblob` (no 256 MiB buffer in the test) and its +//! hash is computed incrementally. A tracking allocator records the peak heap +//! growth during the read. SQLite's own `substr(blob, ..)` materialises the +//! blob inside the SQLite library (C allocator, not counted here); the +//! Postgres equivalent is measured separately (see the PR). + +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use sea_orm::{ConnectOptions, ConnectionTrait, Database, DbBackend, Statement}; +use tinycloud_core::{ + database_artifacts::{ + read_artifact, CacheFileSink, SeaOrmDatabaseArtifactRepository, ARTIFACT_CHUNK_BYTES, + }, + hash::Blake3Hasher, + migrations::Migrator, + sea_orm_migration::MigratorTrait, +}; + +struct Tracking; +static CURRENT: AtomicUsize = AtomicUsize::new(0); +static PEAK: AtomicUsize = AtomicUsize::new(0); + +unsafe impl GlobalAlloc for Tracking { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let ptr = System.alloc(layout); + if !ptr.is_null() { + let now = CURRENT.fetch_add(layout.size(), Ordering::Relaxed) + layout.size(); + PEAK.fetch_max(now, Ordering::Relaxed); + } + ptr + } + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + System.dealloc(ptr, layout); + CURRENT.fetch_sub(layout.size(), Ordering::Relaxed); + } + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { + let new = System.realloc(ptr, layout, new_size); + if !new.is_null() { + if new_size >= layout.size() { + let now = CURRENT.fetch_add(new_size - layout.size(), Ordering::Relaxed) + new_size + - layout.size(); + PEAK.fetch_max(now, Ordering::Relaxed); + } else { + CURRENT.fetch_sub(layout.size() - new_size, Ordering::Relaxed); + } + } + new + } +} + +#[global_allocator] +static ALLOC: Tracking = Tracking; + +const ARTIFACT_BYTES: usize = 256 * 1024 * 1024; + +#[tokio::test] +async fn a_256_mib_artifact_hydrates_with_a_bounded_heap() { + let dir = tempfile::tempdir().unwrap(); + let mut options = ConnectOptions::new(format!( + "sqlite:{}?mode=rwc", + dir.path().join("caps.db").display() + )); + options.max_connections(1); + let conn = Database::connect(options).await.unwrap(); + Migrator::up(&conn, None).await.unwrap(); + + let mut hasher = Blake3Hasher::new(); + let zeros = vec![0_u8; ARTIFACT_CHUNK_BYTES]; + for _ in 0..ARTIFACT_BYTES / ARTIFACT_CHUNK_BYTES { + hasher.update(&zeros); + } + drop(zeros); + let hash = hasher.finalize().to_cid(0x55).to_string(); + conn.execute(Statement::from_sql_and_values( + DbBackend::Sqlite, + "INSERT INTO database_artifact (service, space, name, revision, content_hash, payload, \ + size_bytes, backend, storage_mode, created_at, updated_at, checkpoint_size_bytes, \ + checkpoint_content_hash, delta_payload, delta_content_hash, delta_size_bytes) \ + VALUES ('sql', 'space', 'big', 1, ?, zeroblob(?), ?, 'storage.database', \ + 'database-blob', 'now', 'now', ?, ?, NULL, NULL, 0)", + [ + hash.clone().into(), + (ARTIFACT_BYTES as i64).into(), + (ARTIFACT_BYTES as i64).into(), + (ARTIFACT_BYTES as i64).into(), + hash.into(), + ], + )) + .await + .unwrap(); + + let repo = SeaOrmDatabaseArtifactRepository::new(conn); + let cache = tempfile::tempdir().unwrap(); + let mut sink = CacheFileSink::new( + "sql", + "space/big".to_string(), + cache.path().join("big.db"), + cache.path().join("big.db-wal"), + |_| true, + |_| true, + ); + + let baseline = CURRENT.load(Ordering::Relaxed); + PEAK.store(baseline, Ordering::Relaxed); + let meta = read_artifact(&repo, "sql", "space", "big", &mut sink) + .await + .unwrap() + .expect("the artifact exists"); + let growth = PEAK.load(Ordering::Relaxed).saturating_sub(baseline); + sink.commit().await.unwrap(); + + assert_eq!(meta.checkpoint_size_bytes as usize, ARTIFACT_BYTES); + assert_eq!( + std::fs::metadata(cache.path().join("big.db")) + .unwrap() + .len() as usize, + ARTIFACT_BYTES + ); + // A few slice-sized buffers (row decode, copy, write), never the artifact. + assert!( + growth < 6 * ARTIFACT_CHUNK_BYTES, + "heap grew {} MiB hydrating a {} MiB artifact", + growth >> 20, + ARTIFACT_BYTES >> 20 + ); +} diff --git a/tinycloud-core/tests/duckdb_diagnostics_policy.rs b/tinycloud-core/tests/duckdb_diagnostics_policy.rs index 730ecc78..eed3fabd 100644 --- a/tinycloud-core/tests/duckdb_diagnostics_policy.rs +++ b/tinycloud-core/tests/duckdb_diagnostics_policy.rs @@ -131,6 +131,53 @@ fn tenant_diagnostics_cannot_escape_process_streams_and_normal_sql_still_works() assert!(String::from_utf8_lossy(&output.stderr).contains("tc1304_stderr_capture_control")); } +/// Whole-artifact read for assertions, built on the streaming API: the +/// repository itself has no whole-row read. +#[async_trait::async_trait] +trait ReadAll { + async fn read_all( + &self, + service: &str, + space: &str, + name: &str, + ) -> Result< + Option, + tinycloud_core::database_artifacts::DatabaseArtifactError, + >; +} + +#[async_trait::async_trait] +impl ReadAll for T { + async fn read_all( + &self, + service: &str, + space: &str, + name: &str, + ) -> Result< + Option, + tinycloud_core::database_artifacts::DatabaseArtifactError, + > { + use tinycloud_core::database_artifacts::{read_artifact, DatabaseArtifact, MemorySink}; + let mut sink = MemorySink::default(); + let Some(meta) = read_artifact(self, service, space, name, &mut sink).await? else { + return Ok(None); + }; + Ok(Some(DatabaseArtifact { + payload: sink.checkpoint, + content_hash: meta.content_hash, + revision: meta.revision, + size_bytes: meta.size_bytes, + updated_at: meta.updated_at, + backend: meta.backend, + storage_mode: meta.storage_mode, + checkpoint_content_hash: meta.checkpoint_content_hash, + delta_payload: sink.wal, + delta_content_hash: meta.delta_content_hash, + delta_size_bytes: meta.delta_size_bytes, + })) + } +} + #[tokio::test] async fn invocation_template_is_not_authority_to_execute_ddl_with_read_ability() { let directory = tempfile::tempdir().unwrap(); @@ -188,7 +235,7 @@ async fn invocation_template_is_not_authority_to_execute_ddl_with_read_ability() async fn with_prefixed_writes_require_authority_and_persist_query_write_targets() { use tinycloud_auth::resource::SpaceId; use tinycloud_core::{ - database_artifacts::{DatabaseArtifactRepository, SeaOrmDatabaseArtifactRepository}, + database_artifacts::SeaOrmDatabaseArtifactRepository, duckdb::{service::DuckDbService, types::DuckDbError}, migrations::Migrator, sea_orm::{ConnectOptions, Database}, @@ -285,7 +332,7 @@ async fn with_prefixed_writes_require_authority_and_persist_query_write_targets( let before_rows = rows(&service, &space, data_sql).await; let before_catalog = rows(&service, &space, catalog_sql).await; let before_artifact = repository - .load("duckdb", &space.to_string(), "nested") + .read_all("duckdb", &space.to_string(), "nested") .await .unwrap() .unwrap(); @@ -341,7 +388,7 @@ async fn with_prefixed_writes_require_authority_and_persist_query_write_targets( assert_eq!(rows(&service, &space, data_sql).await, before_rows); assert_eq!(rows(&service, &space, catalog_sql).await, before_catalog); let denied_artifact = repository - .load("duckdb", &space.to_string(), "nested") + .read_all("duckdb", &space.to_string(), "nested") .await .unwrap() .unwrap(); @@ -382,7 +429,7 @@ async fn with_prefixed_writes_require_authority_and_persist_query_write_targets( } }; let prior = repository - .load("duckdb", &space.to_string(), "nested") + .read_all("duckdb", &space.to_string(), "nested") .await .unwrap() .unwrap(); @@ -415,7 +462,7 @@ async fn with_prefixed_writes_require_authority_and_persist_query_write_targets( next_value += 1; assert_eq!(rows(&service, &space, data_sql).await, expected); let persisted = repository - .load("duckdb", &space.to_string(), "nested") + .read_all("duckdb", &space.to_string(), "nested") .await .unwrap() .unwrap(); @@ -622,7 +669,7 @@ async fn executing_explain_template_still_requires_the_wrapped_write_ability() { async fn restricted_invocations_fail_closed_on_unsupported_sql_shapes_without_changing_data() { use tinycloud_auth::resource::SpaceId; use tinycloud_core::{ - database_artifacts::{DatabaseArtifactRepository, SeaOrmDatabaseArtifactRepository}, + database_artifacts::SeaOrmDatabaseArtifactRepository, duckdb::{ service::DuckDbService, types::{DuckDbError, DuckDbExecutionResult}, @@ -754,7 +801,7 @@ async fn restricted_invocations_fail_closed_on_unsupported_sql_shapes_without_ch let before_rows = query_rows(run(&service, &space, snapshot_sql, None, READ).await); let before_catalog = query_rows(run(&service, &space, catalog_sql, None, READ).await); let before_artifact = repository - .load("duckdb", &space.to_string(), "restrictions") + .read_all("duckdb", &space.to_string(), "restrictions") .await .unwrap() .unwrap(); @@ -859,7 +906,7 @@ async fn restricted_invocations_fail_closed_on_unsupported_sql_shapes_without_ch before_catalog ); let after_artifact = repository - .load("duckdb", &space.to_string(), "restrictions") + .read_all("duckdb", &space.to_string(), "restrictions") .await .unwrap() .unwrap(); diff --git a/tinycloud-node-server/src/lib.rs b/tinycloud-node-server/src/lib.rs index 4a65c225..1958da3c 100644 --- a/tinycloud-node-server/src/lib.rs +++ b/tinycloud-node-server/src/lib.rs @@ -1924,18 +1924,29 @@ mod retry_after_fairing_tests { #[async_trait::async_trait] impl tinycloud_core::database_artifacts::DatabaseArtifactRepository for StalledArtifacts { - async fn load( + async fn load_meta( &self, _: &str, _: &str, _: &str, ) -> Result< - Option, + Option, tinycloud_core::database_artifacts::DatabaseArtifactError, > { std::future::pending().await } + async fn read_chunks( + &self, + _: &str, + _: &str, + _: &str, + _: &tinycloud_core::database_artifacts::ArtifactMeta, + _: &mut dyn tinycloud_core::database_artifacts::ArtifactSink, + ) -> Result<(), tinycloud_core::database_artifacts::DatabaseArtifactError> { + std::future::pending().await + } + async fn save( &self, _: &str, diff --git a/tinycloud-node-server/tests/postgres_pool_timeouts.rs b/tinycloud-node-server/tests/postgres_pool_timeouts.rs index d1a151fe..ccbfb965 100644 --- a/tinycloud-node-server/tests/postgres_pool_timeouts.rs +++ b/tinycloud-node-server/tests/postgres_pool_timeouts.rs @@ -48,7 +48,7 @@ use tinycloud_auth::resource::SpaceId; use tinycloud_auth::ssi::dids::DIDBuf; use tinycloud_auth::ssi::jwk::JWK; use tinycloud_core::database_artifacts::{ - DatabaseArtifactError, DatabaseArtifactRepository, SeaOrmDatabaseArtifactRepository, + DatabaseArtifactError, DatabaseArtifactRepository, MemorySink, SeaOrmDatabaseArtifactRepository, }; use tinycloud_core::db_pool::{is_transient_error, transient_error_class, StatementTimeoutMode}; use tinycloud_core::sea_orm::{ @@ -57,7 +57,6 @@ use tinycloud_core::sea_orm::{ }; use tinycloud_core::sql::{SqlError, SqlRequest, SqlResponse, SqlService, SqlValue}; use tinycloud_core::RecoveringConnection; -use tinycloud_core::{SizeTrackingArtifactRepository, SqlSizes}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; @@ -561,17 +560,16 @@ async fn postgres_large_artifacts_persist_under_default_timeouts() { schema.drop().await; } -/// A large hydration whose server-side work outlasts the base statement -/// timeout succeeds, because the size mirror marks the artifact large and -/// its load runs under the bulk timeout; the same load forced under the base -/// timeout is cancelled (Opus's M6). The artifact is read through a view -/// that sleeps 1 s per row, standing in for detoasting and decompressing a -/// large payload. (A slow network does not exercise this: the artifact is -/// one row, and PostgreSQL does not cancel a statement while it is writing -/// a row to the client.) +/// Hydration is a series of 4 MiB slice statements, each under the BASE +/// statement timeout (TC-822): a large artifact hydrates through a pool whose +/// base timeout is far below the whole-artifact transfer, and a slice whose +/// server-side work outlasts the base timeout is cancelled as a transient +/// error rather than holding the pool. The slow case reads the artifact +/// through a view that sleeps 1 s per row, standing in for detoasting a +/// large payload. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn postgres_large_hydration_runs_under_the_bulk_timeout() { - let Some(url) = postgres_url("postgres_large_hydration_runs_under_the_bulk_timeout") else { +async fn postgres_hydration_slices_run_under_the_base_timeout() { + let Some(url) = postgres_url("postgres_hydration_slices_run_under_the_base_timeout") else { return; }; let schema = Schema::new(&url, "bulkload").await; @@ -585,42 +583,51 @@ async fn postgres_large_hydration_runs_under_the_bulk_timeout() { write(&seed, &space, "a", 4 * 1024 * 1024).await.unwrap(); } } - let name = &schema.name; - schema - .admin - .execute_unprepared(&format!( - "ALTER TABLE {name}.database_artifact RENAME TO database_artifact_rows; \ - CREATE FUNCTION {name}.slow_row() RETURNS boolean LANGUAGE plpgsql AS \ - $$ BEGIN PERFORM pg_sleep(1); RETURN true; END $$; \ - CREATE VIEW {name}.database_artifact AS \ - SELECT * FROM {name}.database_artifact_rows WHERE {name}.slow_row();" - )) - .await - .unwrap(); - let strict = pool(schema.url(&url, None), |config| { config.database.statement_timeout_ms = 500; config.database.bulk_statement_timeout_ms = 10_000; }) .await; - let sizes = SqlSizes::new(); - sizes.seed_from(&direct).await.unwrap(); let repository = Arc::new(SeaOrmDatabaseArtifactRepository::new(strict.clone())); let cache = tempfile::TempDir::new().unwrap(); let service = SqlService::new( cache.path().to_string_lossy().into_owned(), u64::MAX, - Arc::new(SizeTrackingArtifactRepository::new( - repository.clone(), - sizes, - )), + repository.clone(), + ); + assert_eq!( + count(&service, &space, "a").await.unwrap(), + 3, + "a multi-slice artifact hydrates under the 500 ms base timeout" ); - assert_eq!(count(&service, &space, "a").await.unwrap(), 3); + let meta = repository + .load_meta("sql", &space.to_string(), "a") + .await + .unwrap() + .expect("the artifact exists"); + let name = &schema.name; + schema + .admin + .execute_unprepared(&format!( + "ALTER TABLE {name}.database_artifact RENAME TO database_artifact_rows; \ + CREATE FUNCTION {name}.slow_row() RETURNS boolean LANGUAGE plpgsql AS \ + $$ BEGIN PERFORM pg_sleep(1); RETURN true; END $$; \ + CREATE VIEW {name}.database_artifact AS \ + SELECT * FROM {name}.database_artifact_rows WHERE {name}.slow_row();" + )) + .await + .unwrap(); let error = repository - .load_sized("sql", &space.to_string(), "a", Some(0)) + .read_chunks( + "sql", + &space.to_string(), + "a", + &meta, + &mut MemorySink::default(), + ) .await - .expect_err("the same load under the 500 ms base timeout is cancelled"); + .expect_err("a slice that outlasts the 500 ms base timeout is cancelled"); match &error { DatabaseArtifactError::Db(error) => { assert_eq!( @@ -991,19 +998,21 @@ async fn blackhole_mid_hydration_waits_for_one_detection_only() { assert!(elapsed < Duration::from_secs(6), "{elapsed:?}"); } -/// The case CI found: the hydration's `BEGIN` draws an idle connection that -/// died silently, and the load on the next connection is cut mid-transfer. -/// The attempt is retried as a whole, its `BEGIN` never on its own, so the -/// request waits for at most two detections, not three. +/// The case CI found: a hydration statement draws an idle connection that +/// died silently, and a later slice on the next connection is cut +/// mid-transfer. Hydration is a series of independently retried statements +/// (TC-822), so the request waits for at most three detections: the dead idle +/// connection (the metadata statement, retried once), then the cut slice and +/// its one retry. Never more: each statement retries once, within the window. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn blackhole_dead_idle_connection_then_cut_hydration_waits_two_detections_at_most() { +async fn blackhole_dead_idle_connection_then_cut_hydration_waits_three_detections_at_most() { let Some(url) = blackhole_url( - "blackhole_dead_idle_connection_then_cut_hydration_waits_two_detections_at_most", + "blackhole_dead_idle_connection_then_cut_hydration_waits_three_detections_at_most", ) else { return; }; let (elapsed, _) = mid_hydration_blackhole_with(&url, "bhidlecut", |_| {}, true).await; - assert!(elapsed < Duration::from_secs(9), "{elapsed:?}"); + assert!(elapsed < Duration::from_secs(14), "{elapsed:?}"); } /// Black-hole every connection streaming db `a`'s hydration on a cold node,