diff --git a/CHANGELOG.md b/CHANGELOG.md index 40fa9310..3490ab76 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,8 @@ ### Changes - Log sidecar: request events with status 400 and above (node and ingress) and the `db_failure_5xx` diagnostic event are capped per minute for each sidecar instance: 10 for each 5xx class (`db_failure_5xx`, `node_request_5xx`, `ingress_request_5xx`) and 3 for each 4xx class (`node_request_4xx`, `ingress_request_4xx`). When a cap holds events back, the sidecar emits one `errors_dropped` `node_warning` event per class per minute with exactly the fields `class` (one of those five names), `dropped` (integer, 1 or more) and `window_s` (60). The reserve for the capped classes is at most 585,403,200 bytes per 30 days per sidecar instance (the caps at worst-valid sizes, plus one `errors_dropped` event per class per minute), inside a 600,000,000-byte reserve. Every other event stays unsampled, including slow requests with a status below 400. Each cap is one Vector `lua` transform per class. The shipped fields are pseudonymous or fixed vocabulary; no SQL, ids, DIDs, paths or messages. The reviewed N0 compose template is `8df7e0f2…` (raw `192dfb7e…`); the lineage with the earlier values is in `docs/releasing.md`. The hosted proof's 500-request burst and its control sends use slow status-200 requests, which no cap covers. The volume policy and arithmetic are in `docs/runbooks/tc-959-n0-log-shipping.md`. Details: TC-1843. +- SQL local durability and reservation before commit (`docs/sql-local-durability.md`). Postgres stays the master copy. File-backed SQL databases run `PRAGMA synchronous = FULL` (`[global.storage.sql] synchronous = "full"|"normal"`, env `TINYCLOUD_STORAGE__SQL__SYNCHRONOUS`; `normal` selects the earlier fsync behaviour), and the directories and renames that make a database file or a hydrated cache visible are fsynced. A write is checked before COMMIT and rolled back whole if it would pass the owner's remaining storage budget (`402`), and, when configured, the per-database cap (`max_database_bytes`, `507 sql_database_size_limit`) or the disk headroom (`min_free_bytes`, bounded by the WAL and checkpoint growth the transaction can cause and shared by all databases on the node, `507 sql_disk_headroom`). `max_database_bytes` and `min_free_bytes` default to `0` (off) and are set by config. A COMMIT that fails on disk answers `500 sql_commit_failed` and the node re-hydrates the database from Postgres. A write that fails because the disk is full answers `507 sql_disk_headroom`, whatever `min_free_bytes` is, not the owner's quota or the size limit. Every write request runs in one transaction (a single statement, a batch, a nested write (`WITH … INSERT`) and a request that carries a `PRAGMA` included), so such a request counts two more native SQL calls in request telemetry, and all of them get the same owner budget, size cap and disk gate. A request that wrote nothing is not gated and reserves nothing. The growth of the main file that the next checkpoint will cause stays held in the disk reservation from the commit until a checkpoint runs or the database's connection closes. No migration. Details: TC-1491. +- `MemorySink` caps its size hint at 64 MiB and treats a failed reservation as no hint. Details: TC-1491. ## [1.22.0] - 2026-10-23 diff --git a/CLAUDE.md b/CLAUDE.md index 7c4f51fe..fb540588 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -175,6 +175,9 @@ All use the `TINYCLOUD_` prefix: | `TINYCLOUD_STORAGE_LIMIT` | Storage quota | `10 MiB` | | `TINYCLOUD_KEYS_SECRET` | Key derivation secret | Base64URL string | | `TINYCLOUD_SPACES_ALLOWLIST` | Allowlist endpoint | `http://localhost:10000` | +| `TINYCLOUD_STORAGE__SQL__SYNCHRONOUS` | SQL `PRAGMA synchronous` for file-backed databases: `full` (default) or `normal` (rollback flag) | `full` | +| `TINYCLOUD_STORAGE__SQL__MAX_DATABASE_BYTES` | Per-database SQL size cap; `0` (default) = off. Sam's decision (card `sql-storage-report-read`) | `1 GiB` | +| `TINYCLOUD_STORAGE__SQL__MIN_FREE_BYTES` | Disk kept free after in-flight SQL writes; `0` (default) = check off. Sam's decision (card `sql-storage-report-read`) | `512 MiB` | | `ROCKET_MAX_BLOCKING` | Rocket's blocking-thread-pool cap (default 512). `ROCKET_`-prefixed, not `TINYCLOUD_`, because Rocket builds its tokio runtime from its own Figment before TinyCloud's config is loaded — `TINYCLOUD_MAX_BLOCKING` is silently ignored for this setting. | `1024` | ### Database Support diff --git a/Cargo.lock b/Cargo.lock index 5059c79b..013fd222 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9898,6 +9898,7 @@ dependencies = [ "hmac 0.12.1", "iri-string", "k256", + "libc", "libp2p", "multihash-derive 0.9.1", "pin-project", diff --git a/docs/request-telemetry.md b/docs/request-telemetry.md index d3a3a0e7..dc5609f2 100644 --- a/docs/request-telemetry.md +++ b/docs/request-telemetry.md @@ -90,7 +90,7 @@ SQLx query timing follows the execution/logger lifetime, including network I/O a | Quota/admin | In-memory overrides and size mirrors do not invent queries; usage/hosted-space enumeration and any actual pool work are counted | | SQL/DuckDB identity/fence/alias resolution and durable artifact load/save/CAS/delta | Shared pool, including PostgreSQL bulk scopes; cache hits perform only their actual work | | Request-awaited read audit | Separate measured batch context, fanned out before durable receipts; details below | -| User SQL rusqlite actor | Context on every work message; actual statement/schema/batch/growth-control SQL (a multi-statement request also counts the `SAVEPOINT` and `RELEASE`/`ROLLBACK` of its request transaction, TC-1491), prepare/constraint errors, cold-open PRAGMAs, size/promotion/checkpoint SQL | +| User SQL rusqlite actor | Context on every work message; actual statement/schema/batch/growth-control SQL (a write request also counts the `SAVEPOINT` and `RELEASE`/`ROLLBACK` of its request transaction, TC-1491; the reservation's own PRAGMAs and checks are not counted), prepare/constraint errors, cold-open PRAGMAs, size/promotion/checkpoint SQL | | SQL publication | The current withdrawn publication handler executes no SQL; it is not an uninstrumented database path | | Optional DuckDB actor | Context on every message and the initiating cold-open worker; JSON/Arrow queries, schema/batch/transaction/growth/catalog SQL, locked security settings and promotion/copy/describe SQL | | DuckDB import/cold export blocking workers | Explicit worker context for guarded admission/validation/checkpoint/copy SQL; artifact repository work is measured separately by SQLx | diff --git a/docs/sql-local-durability.md b/docs/sql-local-durability.md new file mode 100644 index 00000000..fd6ce868 --- /dev/null +++ b/docs/sql-local-durability.md @@ -0,0 +1,89 @@ +# SQL local durability and reservation before commit (TC-1491, slice 1a) + +Postgres stays the master copy of every SQL database. This slice makes the node's +**local** commit durable and refuses, before COMMIT, a write that would not fit. +It adds no migration, no schema change and no new storage. + +## What a request does + +A write request runs in one transaction (an outermost `SAVEPOINT`, so every +write request, whether a single statement, a `Batch` or an `Execute` with `schema` statements, is all-or-nothing): + +1. **Limits.** `PRAGMA max_page_count = min(per-database cap, current pages + owner budget / page size)`. SQLite enforces it itself, mid-statement, so the bound is exact. Not applied to the `Forbid` path (budget 0), which keeps its own guard. +2. The statements run. +3. **Pre-commit gate** (`precommit_gate` in `sql/database.rs`). + - Owner budget and cap are re-checked against the page growth of the open transaction. + - Disk: reserve `unspilled WAL bound + pending checkpoint growth` in the node-wide `DiskBudget`, where pending checkpoint growth is the growth of the main file that the next checkpoint will cause (all pages committed since the last checkpoint). A request that wrote nothing reserves nothing. The reservation succeeds only if `free bytes - other in-flight reservations - this reservation >= min_free_bytes`. +4. `COMMIT` (`RELEASE` of the outermost savepoint). With `synchronous = FULL` the WAL is fsynced before it returns. +5. The reservation shrinks to the checkpoint growth (the WAL bytes are on disk now). The in-flight part ends at COMMIT and is released by `SqlService` when the request's outcome is known; the checkpoint growth stays held until a checkpoint runs or the connection closes. + +Any error before COMMIT is a `ROLLBACK` of the whole transaction. Nothing is committed and then reported as failed. + +### Why the disk bound is sound for index, trigger and foreign-key amplification + +At COMMIT SQLite appends one WAL frame (`page size + 24` bytes) per dirty page that has not already spilled to the WAL. Dirty pages are a subset of the connection's page cache, so `SQLITE_DBSTATUS_CACHE_USED` (heap bytes of the cache) bounds them, whatever the SQL text or payload was: index entries, the pages of every b-tree touched, freelist, overflow and trigger-written pages are all just pages. A page dirtied twice in a transaction still costs one frame. WAL bytes that already spilled are on disk and already out of the free space the gate reads. The main file grows at the next checkpoint by the transaction's page growth, which is why that term is reserved and held until a checkpoint runs. + +`the_unspilled_wal_bound_covers_what_commit_appends` checks the bound against real commits (1 to 100 000 rows, six indexes, plus an update of half the rows). + +What is **not** exact: `free bytes` is a `statvfs` snapshot, so another process on the same filesystem can still take the space after the check (the `min_free_bytes` floor is the margin for that). The reservation covers this process's concurrent databases only. The progress handler that stops a statement early only watches free space, not the cache. + +If free space cannot be read the write is refused (`Internal`), never allowed. + +## fsync policy + +- `PRAGMA synchronous` is set on every file-backed connection: `FULL` by default, in `open_connection` and, before the copy, in `promote_to_file`. With WAL, `FULL` fsyncs the WAL on every COMMIT. `NORMAL` (the rollback flag) fsyncs only at checkpoints. +- SQLite itself fsyncs the directory when it first creates a `-wal` file (the unix VFS `UNIXFILE_DIRSYNC` flag on a new WAL), but not for the main database file. So the node fsyncs: the directories it creates for a database (`create_dir_all_durable`), the directory after `promote_to_file` creates the database file, and the directory after the renames in `CacheFileSink::commit` (the parts are already `sync_all`'d before the rename). +- Deleting stale cache files before a hydration is not directory-synced: Postgres is master and the cache is re-creatable. +- `FULL` is necessary, not sufficient. It does not prove the CVM volume honours flushes. +- Databases below `memory_threshold` (10 MiB) are in-memory SQLite: there is no local file to fsync, and their durability is the Postgres save, as before. Store-master (1b) needs them file-backed. + +## Outcomes in this phase + +| | meaning | +|---|---| +| **Rejected** | Rolled back before COMMIT: no effect on the local file, Postgres or any later snapshot. Reservation refusals (`sql_database_size_limit`, `sql_disk_headroom`, owner budget `402`), `SQLITE_FULL`, any statement error (the whole request), and a COMMIT that returns an error. | +| **Committed** | The local COMMIT returned and, for a file-backed database with `FULL`, its WAL fsync did; and the Postgres save succeeded. Only then is success returned. | +| **Unknown** | Unchanged TC-1274 `outcome_unknown`: the Postgres save was sent and its outcome is not known. | + +Unchanged on purpose: a failed Postgres save discards the local state. A local commit that was never released (the process died after COMMIT, before the response) is dropped when the service next re-hydrates from Postgres, which is correct while Postgres is master. + +**COMMIT error.** If COMMIT's fsync fails after the WAL frames were written, SQLite reports the error and the request is reported as rejected, but the commit can still appear when the file is next opened. That is harmless here: the node discards the local state (`CommitFailed`) and re-hydrates from Postgres, which never saw the write. Under store-master (1b and later) it is an outcome-unknown case: keep the local WAL and answer `outcome_unknown`. + +A kill -9 test cannot inject "after COMMIT, before its fsync completed"; `before_commit`, `after_commit` and `before_response` seams cover the cuts that exist. SIGKILL does not drop the page cache, so the tests prove the commit protocol, not power loss. + +## Settings + +`[global.storage.sql]`; env `TINYCLOUD_STORAGE__SQL__` (the legacy single-underscore form works for names without an underscore). + +| key | default | meaning | +|---|---|---| +| `synchronous` | `"full"` | `"full"` or `"normal"`; the rollback flag. This is the durability fix, not a limit. | +| `max_database_bytes` | `0` (no cap) | per-database size cap, enforced by SQLite's `max_page_count` | +| `min_free_bytes` | `0` (check off) | disk kept free after all in-flight reservations; `0` disables the gate, the reservation and the free-space checks before COMMIT (a full-disk error still reads free space once, see Errors) | + +**The two limits are off by default.** They are Sam's decision (card `sql-storage-report-read`) and are set by config once he answers; an unmeasured cap could refuse real users' databases. The mechanism is fully tested with explicit values. What stays on by default besides `synchronous`: the transaction per write request and the owner-budget limit below. + +**Owner budget is not switchable and can refuse a write today's code allows.** `max_page_count` is set from the owner's remaining quota (`limit - used`, the same `storage_check` read as before). Today a write that overshoots a non-zero remainder is allowed once and only the next one is refused with 402; now the overshooting write itself is refused with 402 before COMMIT. The comparison is the database's page growth against the remainder, the unit the quota already charges (logical database size); growth that frees pages elsewhere in the file is not counted, so it is never stricter than the charge. + +## Errors + +| error | HTTP | body prefix | +|---|---|---| +| per-database size cap | 507 | `sql_database_size_limit:` | +| disk headroom | 507 | `sql_disk_headroom:` | +| owner budget (existing `StorageWouldGrow`) | 402 | the TC-619 quota body | +| local COMMIT failed | 500 | `sql_commit_failed:` | + +507, not 413: the SQL client maps 413 to `SQL_RESPONSE_TOO_LARGE`, the KV and quota clients read it as an exhausted storage plan, and TC-1012's 413 is the request body cap. + +A write that fails because the disk is full is answered as `sql_disk_headroom` (507) in every configuration. When SQLite reports the database full, the node reads the free space once; if it is below a fixed floor of 16 database pages, the disk is the cause, whatever `min_free_bytes` is set to. Otherwise the owner budget (402) or the per-database cap (`sql_database_size_limit`, 507) is the cause. + +## Scope notes + +**What counts as a write.** `Execute`, `Batch` and `ExecuteStatement` are always wrapped (a read inside a batch is wrapped too; only telemetry notices). A single `Query` is wrapped, gated and reserved when `parser::classify_sql` says it is not read-only. That classification covers writes nested in a query tree (TC-1585): a `WITH … INSERT` or `WITH … UPDATE` is a write, so it gets the transaction, the owner budget, the size cap and the pre-commit gate like any other write. The answer does not depend on the caller's abilities; a read-tier grant is only a shortcut (its writes are refused anyway). The route uses the same classifier (`sql_request_is_write`) to decide whether to read the owner's budget, and calls the `*_authorized` service entry points (`execute_with_budget_authorized`, `execute_without_growth_authorized`, `execute_authorized`), so the budget path carries the per-statement authority of TC-1581. `needs_transaction` is the one place this is decided for a request. + +**PRAGMA.** A request that carries a `PRAGMA` (admin only) runs in the same transaction as any other write request, and is limited, gated and reserved like one. The actor's connection reset after a request that held a non-report `PRAGMA` (TC-1586) runs after the request, after the transaction has ended. `PRAGMA synchronous` is not on the admin allow-list, so it cannot be read or set through a request. + +**Retire order.** When that reset fails, the actor is retired (deregistered before the error reply, TC-1594 tracks closing the connection first). The pre-commit gate and the reservation release are on the same path as before: the reservation is released by a drop guard in `SqlService::execute_locked` when the request returns, whatever the actor did, so the retire order does not change it. A request that triggers the reset held a `PRAGMA`; its transaction and its reservation have ended before the reset runs. The actor also retires when a connection is still inside a transaction after a request. + +`Publication` commands are not gated either. DuckDB is untouched. diff --git a/test/m1-realdata-e2e/Cargo.lock b/test/m1-realdata-e2e/Cargo.lock index 45689d33..db1ca850 100644 --- a/test/m1-realdata-e2e/Cargo.lock +++ b/test/m1-realdata-e2e/Cargo.lock @@ -7986,6 +7986,7 @@ dependencies = [ "hkdf", "hmac 0.12.1", "k256", + "libc", "libp2p", "multihash-derive 0.9.1", "pin-project", diff --git a/test/w5-policy-runtime-node-e2e/Cargo.lock b/test/w5-policy-runtime-node-e2e/Cargo.lock index 74e38f9f..beffcffc 100644 --- a/test/w5-policy-runtime-node-e2e/Cargo.lock +++ b/test/w5-policy-runtime-node-e2e/Cargo.lock @@ -8434,6 +8434,7 @@ dependencies = [ "hkdf", "hmac 0.12.1", "k256", + "libc", "libp2p", "multihash-derive 0.9.1", "pin-project", diff --git a/tinycloud-core/Cargo.toml b/tinycloud-core/Cargo.toml index 04683d8b..0cda3535 100644 --- a/tinycloud-core/Cargo.toml +++ b/tinycloud-core/Cargo.toml @@ -48,6 +48,7 @@ serde_ipld_dagcbor = "0.3" ucan-capabilities-object.workspace = true multihash-derive = "0.9" rusqlite.workspace = true +libc = "0.2" sqlparser.workspace = true tracing.workspace = true tokio = { workspace = true, features = ["sync", "time"] } diff --git a/tinycloud-core/src/database_artifacts.rs b/tinycloud-core/src/database_artifacts.rs index 933bb62d..f06936c5 100644 --- a/tinycloud-core/src/database_artifacts.rs +++ b/tinycloud-core/src/database_artifacts.rs @@ -198,6 +198,9 @@ pub trait ArtifactSink: Send { async fn finish_part(&mut self) -> Result<(), DatabaseArtifactError>; } +/// The most [`MemorySink`] allocates ahead of the bytes it is given. +const MAX_SIZE_HINT_BYTES: u64 = 64 * 1024 * 1024; + /// 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)] @@ -228,7 +231,11 @@ impl ArtifactSink for MemorySink { ArtifactPart::Wal => self.wal.as_mut(), }; if let Some(target) = target { - target.reserve_exact(usize::try_from(bytes).unwrap_or(0)); + // The hint comes from stored metadata. A corrupt value must not + // abort the node (`reserve_exact` panics or aborts on overflow or + // a failed allocation), so cap it and treat a refusal as no hint. + let hint = usize::try_from(bytes.min(MAX_SIZE_HINT_BYTES)).unwrap_or(0); + let _ = target.try_reserve_exact(hint); } } @@ -341,6 +348,7 @@ impl CacheFileSink { 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`. + let mut dirs = std::collections::BTreeSet::new(); while let Some((temp, path)) = self.staged.first().cloned() { if let Err(error) = tokio::fs::rename(&temp, &path).await { self.discard(); @@ -348,6 +356,14 @@ impl CacheFileSink { } release_staging(&temp); self.staged.remove(0); + if let Some(parent) = path.parent() { + dirs.insert(parent.to_path_buf()); + } + } + // The parts were synced before the rename; the rename itself is only + // durable once its directory is (TC-1491). + for dir in dirs { + tokio::fs::File::open(&dir).await?.sync_all().await?; } Ok(()) } @@ -2236,4 +2252,35 @@ mod tests { .iter() .all(|n| !n.contains(&format!(".tmp.{}.", std::process::id())))); } + + /// A corrupt size in the artifact metadata is only a hint: it must not + /// abort or panic the node, and the bytes still arrive intact. + #[tokio::test] + async fn tc1491_memory_sink_survives_an_absurd_size_hint() { + for hint in [ + u64::MAX, + usize::MAX as u64, + (isize::MAX as u64) + 1, + 1 << 62, + 1 << 40, + ] { + let mut sink = MemorySink::default(); + sink.begin_part(ArtifactPart::Checkpoint).await.unwrap(); + sink.size_hint(ArtifactPart::Checkpoint, hint); + sink.write(b"SQLite format 3\0").await.unwrap(); + sink.write(b"checkpoint").await.unwrap(); + sink.finish_part().await.unwrap(); + sink.begin_part(ArtifactPart::Wal).await.unwrap(); + sink.size_hint(ArtifactPart::Wal, hint); + sink.write(b"wal").await.unwrap(); + sink.finish_part().await.unwrap(); + assert_eq!(sink.checkpoint, b"SQLite format 3\0checkpoint", "{hint}"); + assert_eq!(sink.wal.as_deref(), Some(&b"wal"[..]), "{hint}"); + assert!( + sink.checkpoint.capacity() <= 64 * 1024 * 1024 + 64, + "a hint of {hint} reserved {} bytes", + sink.checkpoint.capacity() + ); + } + } } diff --git a/tinycloud-core/src/sql/database.rs b/tinycloud-core/src/sql/database.rs index 6f0602b3..bd428a24 100644 --- a/tinycloud-core/src/sql/database.rs +++ b/tinycloud-core/src/sql/database.rs @@ -6,7 +6,7 @@ //! SQL query time. use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicU64, AtomicU8, Ordering}; use std::sync::Arc; use dashmap::DashMap; @@ -19,6 +19,10 @@ use super::{ authority::{record_ddl_without_schema, SqlAuthority}, authorizer::{PragmaTracker, StatementScope}, caveats::SqlCaveats, + durability::{ + cache_used_bytes, unspilled_wal_bound, DurabilityPolicy, HeldGuard, ReservationRefused, + WriteProbe, + }, parser, storage::{self, StorageMode}, types::*, @@ -59,6 +63,11 @@ enum DbMessage { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum GrowthPolicy { Allow, + /// The owner's storage budget has this many bytes left: the write may grow + /// the database by at most that. SQLite enforces it (`max_page_count`), so + /// a write that would pass it is rolled back before COMMIT with + /// [`SqlError::StorageWouldGrow`] instead of overshooting by one write. + Budget(u64), /// The space is full: run the request, but undo it and fail with /// [`SqlError::StorageWouldGrow`] if the database grew. No-op DDL, /// `DELETE` and `DROP` free pages instead of adding them, so they pass. @@ -77,7 +86,7 @@ pub struct DatabaseHandle { impl DatabaseHandle { /// Close the connection before acknowledging that its local files can be removed. - pub(crate) async fn shutdown(&self) -> Result<(), SqlError> { + pub async fn shutdown(&self) -> Result<(), SqlError> { let (response_tx, response_rx) = oneshot::channel(); self.tx .send(( @@ -195,6 +204,26 @@ pub fn spawn_actor( base_path: String, memory_threshold: u64, databases: Arc>, +) -> DatabaseHandle { + spawn_actor_with_policy( + space_id, + db_name, + base_path, + memory_threshold, + databases, + DurabilityPolicy::default(), + ) +} + +/// [`spawn_actor`] with the node's durability policy (TC-1491): the +/// `synchronous` level, the size cap and the disk headroom. +pub fn spawn_actor_with_policy( + space_id: String, + db_name: String, + base_path: String, + memory_threshold: u64, + databases: Arc>, + policy: DurabilityPolicy, ) -> DatabaseHandle { let (tx, mut rx) = mpsc::channel::<(Option>, DbMessage)>(32); let id = NEXT_ACTOR_ID.fetch_add(1, Ordering::Relaxed); @@ -214,7 +243,7 @@ pub fn spawn_actor( }; let mut conn = { let _context = request_telemetry::enter(initial_context); - let Ok(connection) = storage::open_connection(&mode) else { + let Ok(connection) = storage::open_connection(&mode, policy.synchronous) else { tracing::error!("Failed to open database"); databases.remove_if(&(space_id, db_name), |_, handle| handle.id == id); return; @@ -222,6 +251,11 @@ pub fn spawn_actor( connection }; + // What this database holds of the disk budget is dropped when its + // connection is gone, however the actor ends. + let key = format!("{space_id}/{db_name}"); + let held_guard = HeldGuard::new(policy.budget.clone(), key.clone()); + let shutdown_response = loop { // Block on receiving with timeout let (context, msg) = @@ -261,12 +295,23 @@ pub fn spawn_actor( } => { let pragmas = PragmaTracker::default(); authority.denied_arm().clear(); + let ctx = WriteContext { + policy: &policy, + key: &key, + dir: existing_ancestor(&file_path), + file_backed: matches!(mode, StorageMode::File(_)), + remaining: match growth { + GrowthPolicy::Budget(remaining) => Some(remaining), + _ => None, + }, + file: &file_path, + }; let result = match growth { - GrowthPolicy::Allow => handle_message_authorized( - &conn, &request, &caveats, &authority, &pragmas, + GrowthPolicy::Allow | GrowthPolicy::Budget(_) => handle_message_ctx( + &conn, &request, &caveats, &authority, &pragmas, &ctx, ), - GrowthPolicy::Forbid => handle_message_without_growth_authorized( - &conn, &request, &caveats, &authority, &pragmas, + GrowthPolicy::Forbid => handle_message_without_growth_ctx( + &conn, &request, &caveats, &authority, &pragmas, &ctx, ), }; // A statement the authorizer refused for a caveat fails the @@ -314,20 +359,48 @@ pub fn spawn_actor( if result.is_ok() && matches!(mode, StorageMode::InMemory) { if let Ok(size) = storage::database_size(&conn) { if size > memory_threshold { - match storage::promote_to_file(&conn, &file_path) { - Ok(new_conn) => { - conn = new_conn; - mode = StorageMode::File(file_path.clone()); - tracing::info!("Promoted database to file storage"); - } - Err(_) => { - tracing::error!("Failed to promote database to file"); + // The copy needs the whole database on disk, + // next to what in-flight writes still need. + let room = if policy.min_free_bytes == 0 { + Ok(()) + } else { + policy.budget.reserve( + &key, + size, + 0, + policy.free_space.as_ref(), + existing_ancestor(&file_path), + policy.min_free_bytes, + ) + }; + if room.is_err() { + tracing::warn!( + "Not enough free disk to promote a database; keeping it in memory" + ); + } else { + let promoted = storage::promote_to_file( + &conn, + &file_path, + policy.synchronous, + ); + // The copy is on disk now. + policy.budget.release(&key); + match promoted { + Ok(new_conn) => { + conn = new_conn; + mode = StorageMode::File(file_path.clone()); + tracing::info!("Promoted database to file storage"); + } + Err(_) => { + tracing::error!("Failed to promote database to file"); + } } } } } } + seam(&policy.seams.before_response); let _ = response_tx.send(result); } DbMessage::Export { response_tx } => { @@ -336,6 +409,11 @@ pub fn spawn_actor( } DbMessage::Checkpoint { response_tx } => { let result = handle_checkpoint(&conn, &mode, &file_path); + if result.is_ok() { + // The WAL is in the main file now: its growth is on + // the disk and no longer pending. + policy.budget.forget(&key); + } let _ = response_tx.send(result); } DbMessage::Wal { response_tx } => { @@ -348,6 +426,7 @@ pub fn spawn_actor( // SQLite can checkpoint its WAL on close. Finish that before removing // the registry entry or allowing a service to replace the local files. drop(conn); + drop(held_guard); // Deregister only OUR OWN entry. A replacement actor may already have // been hydrated and registered for this key (the dead-actor respawn in @@ -366,6 +445,14 @@ pub fn spawn_actor( DatabaseHandle { id, tx } } +/// The nearest existing ancestor of `path`: where `statvfs` can be asked how +/// much room the filesystem has before the file or its directory exists. +fn existing_ancestor(path: &Path) -> &Path { + path.ancestors() + .find(|ancestor| !ancestor.as_os_str().is_empty() && ancestor.exists()) + .unwrap_or_else(|| Path::new(".")) +} + fn handle_wal(mode: &StorageMode, file_path: &Path) -> Result>, SqlError> { if !matches!(mode, StorageMode::File(_)) { return Ok(None); @@ -433,34 +520,22 @@ fn handle_checkpoint( const GUARD_BEGIN: &str = "SAVEPOINT tinycloud_storage_guard"; const GUARD_KEEP: &str = "RELEASE tinycloud_storage_guard"; -const GUARD_UNDO: &str = "ROLLBACK TO tinycloud_storage_guard; RELEASE tinycloud_storage_guard"; +/// The guard's savepoint is the outermost transaction, so undoing the request +/// is a ROLLBACK: it writes nothing, where `ROLLBACK TO` + `RELEASE` would +/// commit an empty write transaction (a WAL frame for the header). +const GUARD_UNDO: &str = "ROLLBACK"; /// Run `request` inside a savepoint and keep its effects only if the /// database did not grow (`PRAGMA page_count` reflects uncommitted pages on /// this connection). Every failed guarded request is rolled back, including /// schema entries and batch statements that ran before a later rejection. -#[cfg(test)] -fn handle_message_without_growth( - conn: &rusqlite::Connection, - request: &SqlRequest, - caveats: &Option, - ability: &str, -) -> Result { - handle_message_without_growth_authorized( - conn, - request, - caveats, - &SqlAuthority::from(ability), - &PragmaTracker::default(), - ) -} - -fn handle_message_without_growth_authorized( +fn handle_message_without_growth_ctx( conn: &rusqlite::Connection, request: &SqlRequest, caveats: &Option, authority: &SqlAuthority, pragmas: &PragmaTracker, + ctx: &WriteContext<'_>, ) -> Result { let before = page_count(conn)?; { @@ -468,7 +543,7 @@ fn handle_message_without_growth_authorized( conn.execute_batch(GUARD_BEGIN) .map_err(|e| SqlError::Internal(e.to_string()))?; } - let result = handle_message_authorized(conn, request, caveats, authority, pragmas); + let result = handle_message_ctx(conn, request, caveats, authority, pragmas, ctx); // An early error return can leave the request's authorizer installed. conn.authorizer(None::) -> Authorization>); // ROLLBACK conflict handling can already have undone the transaction. @@ -476,11 +551,31 @@ fn handle_message_without_growth_authorized( return result; } let grew = page_count(conn).map(|after| after > before); + // The disk check before COMMIT, for a request that is about to be kept + // and has written something. `Some` is what stays held afterwards. + let gate = match (&result, &grew) { + (Ok(_), Ok(false)) if WriteProbe::of(conn).writing() => baseline(conn) + .and_then(|base| { + precommit_gate( + conn, + ctx, + &Baseline { + pages: u64::try_from(before).unwrap_or(0), + page_size: base.page_size, + }, + ) + }) + .map(Some), + _ => Ok(None), + }; let end = match grew { - Ok(false) if result.is_ok() => GUARD_KEEP, + Ok(false) if result.is_ok() && gate.is_ok() => GUARD_KEEP, // Undo earlier statements on any error, even without page growth. _ => GUARD_UNDO, }; + if end == GUARD_KEEP { + seam(&ctx.policy.seams.before_commit); + } let end_result = { let _timer = QueryTimer::new(); conn.execute_batch(end) @@ -494,10 +589,25 @@ fn handle_message_without_growth_authorized( conn.execute_batch(GUARD_UNDO) .map_err(|e| SqlError::Internal(e.to_string()))?; } - return Err(SqlError::Sqlite(error.to_string())); - } + ctx.policy.budget.release(ctx.key); + return Err(commit_error(&error)); + } + let held = match gate { + Ok(held) => held, + Err(error) => { + ctx.policy.budget.release(ctx.key); + return Err(error); + } + }; match grew? { - false => result, + false => { + seam(&ctx.policy.seams.after_commit); + // The WAL bytes are on disk; the checkpoint growth is what stays. + if let Some(pending) = held { + ctx.policy.budget.hold(ctx.key, pending); + } + result + } true => Err(SqlError::StorageWouldGrow), } } @@ -508,29 +618,275 @@ fn page_count(conn: &rusqlite::Connection) -> Result { .map_err(|e| SqlError::Internal(e.to_string())) } +/// The unit tests drive a bare connection: no disk, no budget. +#[cfg(test)] +fn handle_message( + conn: &rusqlite::Connection, + request: &SqlRequest, + caveats: &Option, + ability: &str, +) -> Result { + let policy = DurabilityPolicy::default(); + let ctx = WriteContext { + policy: &policy, + key: "test/test", + dir: Path::new("."), + file_backed: false, + remaining: None, + file: Path::new(""), + }; + handle_message_ctx( + conn, + request, + caveats, + &SqlAuthority::from(ability), + &PragmaTracker::default(), + &ctx, + ) +} + +#[cfg(test)] +fn handle_message_without_growth( + conn: &rusqlite::Connection, + request: &SqlRequest, + caveats: &Option, + ability: &str, +) -> Result { + let policy = DurabilityPolicy::default(); + let ctx = WriteContext { + policy: &policy, + key: "test/test", + dir: Path::new("."), + file_backed: false, + remaining: None, + file: Path::new(""), + }; + handle_message_without_growth_ctx( + conn, + request, + caveats, + &SqlAuthority::from(ability), + &PragmaTracker::default(), + &ctx, + ) +} + const REQUEST_BEGIN: &str = "SAVEPOINT tinycloud_request"; const REQUEST_KEEP: &str = "RELEASE tinycloud_request"; +/// SQLite's default `max_page_count`. +const UNLIMITED_PAGES: u64 = 4_294_967_294; +/// Opcodes between the disk checks of a running statement. +const PROGRESS_INTERVAL_OPS: std::os::raw::c_int = 10_000; +/// What the progress handler saw: the disk below its floor, or free space it +/// could not read. +const TRIPPED_NONE: u8 = 0; +const TRIPPED_EXHAUSTED: u8 = 1; +const TRIPPED_UNKNOWN: u8 = 2; + +/// What a write request needs to be bounded before it commits (TC-1491). +struct WriteContext<'a> { + policy: &'a DurabilityPolicy, + /// `space/db`, the key of this database's reservation. + key: &'a str, + /// An existing directory on the filesystem holding the database. + dir: &'a Path, + /// True once the database is a file (and so has a WAL and uses the disk). + file_backed: bool, + /// Bytes the owner's storage budget still allows this database to grow + /// by, or `None` for no budget. + remaining: Option, + /// The database's main file (it may not exist yet). + file: &'a Path, +} + +/// The state a request starts from. +struct Baseline { + pages: u64, + page_size: u64, +} + +fn pragma_u64(conn: &rusqlite::Connection, pragma: &str) -> Result { + conn.query_row(&format!("PRAGMA {pragma}"), [], |row| row.get::<_, i64>(0)) + .map(|value| u64::try_from(value).unwrap_or(0)) + .map_err(|e| SqlError::Internal(e.to_string())) +} + +fn baseline(conn: &rusqlite::Connection) -> Result { + Ok(Baseline { + pages: pragma_u64(conn, "page_count")?, + page_size: pragma_u64(conn, "page_size")?, + }) +} + +fn set_max_pages(conn: &rusqlite::Connection, pages: u64) -> Result<(), SqlError> { + conn.query_row(&format!("PRAGMA max_page_count = {pages}"), [], |_| Ok(())) + .map_err(|e| SqlError::Internal(e.to_string())) +} + +/// The page limit SQLite enforces for this request, and whether the owner's +/// budget (rather than the per-database cap) is the tighter of the two. SQLite +/// never sets it below the current size, so a database already over the cap +/// can still run writes that do not grow it. +fn page_limit(ctx: &WriteContext<'_>, baseline: &Baseline) -> (u64, bool) { + let page_size = baseline.page_size.max(1); + let cap = match ctx.policy.max_database_bytes { + 0 => UNLIMITED_PAGES, + bytes => (bytes / page_size).max(1), + }; + match ctx.remaining { + Some(remaining) => { + let budget = baseline.pages.saturating_add(remaining / page_size); + (budget.min(cap), budget < cap) + } + None => (cap, false), + } +} + +fn is_full_error(error: &SqlError) -> bool { + matches!(error, SqlError::Sqlite(m) | SqlError::SchemaError(m) if m.contains("database or disk is full")) +} + +/// A disk with fewer free bytes than this many database pages cannot take +/// another write. Below it, a `SQLITE_FULL` is the disk's (ENOSPC) whatever +/// the headroom setting: a fixed, small floor, so that no configuration is +/// needed for a full disk to be reported as one. +const DISK_FULL_FLOOR_PAGES: u64 = 16; + +/// What a `SQLITE_FULL` means: the size limit SQLite enforces (`max_page_count`) +/// or the disk itself (ENOSPC). The page count cannot tell them apart, since +/// SQLite has already undone the failed statement by the time it is read; the +/// free-space reading can, and it is read here whatever `min_free_bytes` is +/// (one `statvfs`, on this error path only): less than [`DISK_FULL_FLOOR_PAGES`] +/// pages free, or less than the headroom floor next to what other writes +/// reserved, is the disk. A full disk is never reported as the owner's quota +/// or the database's size limit. +fn classify_full(ctx: &WriteContext<'_>, baseline: &Baseline, budget_binding: bool) -> SqlError { + if ctx.policy.max_database_bytes == 0 && !budget_binding { + // No limit of ours can have fired, so the disk really is full. + return SqlError::DiskHeadroom; + } + let min_free = ctx.policy.min_free_bytes; + let disk_exhausted = ctx.file_backed + && match ctx.policy.free_space.free_bytes(ctx.dir) { + Ok(free) => { + free < DISK_FULL_FLOOR_PAGES.saturating_mul(baseline.page_size) + || (min_free > 0 + && free + < ctx + .policy + .budget + .reserved_excluding(ctx.key) + .saturating_add(min_free)) + } + // Unreadable: with the headroom check on, refuse like the gate does. + Err(_) => min_free > 0, + }; + match (disk_exhausted, budget_binding) { + (true, _) => SqlError::DiskHeadroom, + (false, true) => SqlError::StorageWouldGrow, + (false, false) => SqlError::DatabaseSizeLimit, + } +} + +/// Free space could not be read: the headroom check cannot answer, so the +/// request is refused. One error for every place that reads it. +fn free_space_unreadable() -> SqlError { + SqlError::Internal("could not read the free disk space".into()) +} + +/// Bytes the next checkpoint will grow the main file by: the pages the +/// database has beyond the file's own size, all of them committed since the +/// last checkpoint and held in the WAL until then. Read from the file, so it +/// needs no state of its own and counts every earlier commit, not only this +/// request's. +fn pending_checkpoint_bytes(ctx: &WriteContext<'_>, pages: u64, page_size: u64) -> u64 { + let file_pages = std::fs::metadata(ctx.file) + .map(|meta| meta.len() / page_size.max(1)) + .unwrap_or(pages); + pages.saturating_sub(file_pages).saturating_mul(page_size) +} + +/// The last check before COMMIT, for a request that has written. Rolling back +/// on an error is the caller's. On success, returns the bytes of checkpoint +/// growth to keep held once the commit is done (the rest of the reservation +/// ends with the commit). +/// +/// The disk bound: the pages COMMIT still writes to the WAL are the dirty +/// pages that have not spilled yet, and those are all in the connection's +/// page cache, so `cache_used` bounds them whatever the SQL payload was +/// (index, trigger and foreign-key writes, freelist and overflow pages are +/// all just pages). A later checkpoint grows the main file by every page +/// committed since the last one ([`pending_checkpoint_bytes`]), which is held +/// until a checkpoint runs. WAL bytes already spilled are on disk and already +/// out of the free space the check reads. +fn precommit_gate( + conn: &rusqlite::Connection, + ctx: &WriteContext<'_>, + baseline: &Baseline, +) -> Result { + let now = pragma_u64(conn, "page_count")?; + let growth_pages = now.saturating_sub(baseline.pages); + let growth_bytes = growth_pages.saturating_mul(baseline.page_size); + if ctx + .remaining + .is_some_and(|remaining| growth_bytes > remaining) + { + return Err(SqlError::StorageWouldGrow); + } + if growth_pages > 0 + && ctx.policy.max_database_bytes > 0 + && now.saturating_mul(baseline.page_size) > ctx.policy.max_database_bytes + { + return Err(SqlError::DatabaseSizeLimit); + } + if !ctx.file_backed || ctx.policy.min_free_bytes == 0 { + return Ok(0); + } + let wal_bound = unspilled_wal_bound(cache_used_bytes(conn), baseline.page_size); + let pending = pending_checkpoint_bytes(ctx, now, baseline.page_size); + match ctx.policy.budget.reserve( + ctx.key, + wal_bound, + pending, + ctx.policy.free_space.as_ref(), + ctx.dir, + ctx.policy.min_free_bytes, + ) { + Ok(()) => Ok(pending), + Err(ReservationRefused::Headroom) => Err(SqlError::DiskHeadroom), + Err(ReservationRefused::Unknown) => Err(free_space_unreadable()), + } +} + +fn seam(hook: &Option>) { + if let Some(hook) = hook { + hook(); + } +} /// Whether `request` runs in a transaction of its own (TC-1491). /// -/// Every request that can write does, so it is all-or-nothing whatever its +/// Every request that can write does, so the checks before COMMIT have a +/// transaction to roll back, and the request is all-or-nothing whatever its /// conflict clauses say: a statement that stops with `OR FAIL` or /// `ON CONFLICT FAIL` keeps the rows it changed first when it runs on its /// own, and a request with several statements keeps its earlier ones. -/// A request that carries a `PRAGMA` (admin only) is wrapped like any other. -/// Every pragma the authorizer admits is transactional (`user_version` and -/// `application_id` are undone by the rollback; the rest are reports), and -/// the authorizer's allow-list is what keeps connection-level settings out of -/// the transaction. The actor's TC-1586 connection reset runs after the -/// transaction has ended. +/// A request that carries a `PRAGMA` (admin only) is wrapped like any other, +/// and so is limited, gated and reserved like any write. Every pragma the +/// authorizer admits is transactional (`user_version` and `application_id` +/// are undone by the rollback; the rest are reports), and the authorizer's +/// allow-list is what keeps connection-level settings out of the +/// transaction. The actor's TC-1586 connection reset runs after the +/// transaction has ended. Not wrapped: a query that only reads. /// /// Whether a query writes is `parser::classify_sql` (TC-1585), which sees -/// writes nested in a query tree: a `WITH … INSERT` is wrapped like any other -/// write. It does not depend on the caller's abilities (a read-tier grant is -/// only a shortcut: its writes are refused anyway), and a statement that does -/// not parse is left to `handle_request_authorized` to report. This is the one -/// place that decides it; `Execute`, `Batch` and `ExecuteStatement` are wrapped -/// whatever they contain, a read in a batch included. +/// writes nested in a query tree: a `WITH … INSERT` is wrapped, gated and +/// reserved like any other write. It does not depend on the caller's +/// abilities (a read-tier grant is only a shortcut: its writes are refused +/// anyway), and a statement that does not parse is left to +/// `handle_request_authorized` to report. This is the one place that decides +/// it; `Execute`, `Batch` and `ExecuteStatement` are wrapped whatever they +/// contain, a read in a batch included. fn needs_transaction( request: &SqlRequest, caveats: &Option, @@ -574,22 +930,6 @@ impl DdlLog { } } -#[cfg(test)] -fn handle_message( - conn: &rusqlite::Connection, - request: &SqlRequest, - caveats: &Option, - ability: &str, -) -> Result { - handle_message_authorized( - conn, - request, - caveats, - &SqlAuthority::from(ability), - &PragmaTracker::default(), - ) -} - /// Run `request` as one transaction when it needs one: every statement is /// kept or none is. /// @@ -602,22 +942,94 @@ fn handle_message( /// and outside the actor's TC-1586 connection reset, which runs after the /// transaction has ended. /// -/// When a transaction is already open, [`handle_message_without_growth_authorized`]'s +/// When a transaction is already open, [`handle_message_without_growth_ctx`]'s /// guard has it and already undoes the whole request on any error (reporting /// growth ahead of the SQL error), so this adds nothing there. /// +/// Order, for a write: limits (SQLite's `max_page_count`: the per-database +/// cap and the owner's remaining budget), the statements, the pre-commit gate +/// ([`precommit_gate`]), COMMIT. Any failure before COMMIT is a ROLLBACK. +/// /// If the transaction is still open after the final `ROLLBACK`, the actor /// retires the connection instead of serving another request on it. -fn handle_message_authorized( +fn handle_message_ctx( conn: &rusqlite::Connection, request: &SqlRequest, caveats: &Option, authority: &SqlAuthority, pragmas: &PragmaTracker, + ctx: &WriteContext<'_>, ) -> Result { if !conn.is_autocommit() || !needs_transaction(request, caveats, authority) { return handle_request_authorized(conn, request, caveats, authority, pragmas); } + let base = baseline(conn)?; + set_max_pages(conn, page_limit(ctx, &base).0)?; + let tripped = Arc::new(AtomicU8::new(TRIPPED_NONE)); + let disk_checked = ctx.file_backed && ctx.policy.min_free_bytes > 0; + if disk_checked { + let (budget, free_space) = (ctx.policy.budget.clone(), ctx.policy.free_space.clone()); + let (key, dir) = (ctx.key.to_string(), ctx.dir.to_path_buf()); + let (min_free, tripped) = (ctx.policy.min_free_bytes, tripped.clone()); + let probe = WriteProbe::of(conn); + // Stops a statement early once the disk is already too full; the + // authoritative check is `precommit_gate`. A transaction that has + // written nothing yet needs no room, so reads are never stopped. + conn.progress_handler( + PROGRESS_INTERVAL_OPS, + Some(move || { + if !probe.writing() { + return false; + } + let Ok(free) = free_space.free_bytes(&dir) else { + tripped.store(TRIPPED_UNKNOWN, Ordering::Relaxed); + return true; + }; + let others = budget.reserved_excluding(&key); + let exhausted = free < others.saturating_add(min_free); + if exhausted { + tripped.store(TRIPPED_EXHAUSTED, Ordering::Relaxed); + } + exhausted + }), + ); + } + let outcome = run_transaction(conn, request, caveats, authority, pragmas, ctx, &base); + if disk_checked { + conn.progress_handler(0, None:: bool>); + } + let outcome = outcome.map_err(|error| match tripped.load(Ordering::Relaxed) { + TRIPPED_EXHAUSTED => SqlError::DiskHeadroom, + TRIPPED_UNKNOWN => free_space_unreadable(), + _ => error, + }); + // A failed request keeps nothing, so nothing of it stays reserved. + if outcome.is_err() { + ctx.policy.budget.release(ctx.key); + } + let _ = set_max_pages(conn, UNLIMITED_PAGES); + outcome.map(|(result, held)| { + // Committed: its WAL bytes are on disk now. What stays held, until a + // checkpoint runs, is the checkpoint growth. A request that wrote + // nothing changes nothing here. + if let Some(pending) = held { + ctx.policy.budget.hold(ctx.key, pending); + } + result + }) +} + +/// The transaction itself; on `Err` it has been rolled back. On success, the +/// checkpoint growth to keep held, if the request wrote anything. +fn run_transaction( + conn: &rusqlite::Connection, + request: &SqlRequest, + caveats: &Option, + authority: &SqlAuthority, + pragmas: &PragmaTracker, + ctx: &WriteContext<'_>, + base: &Baseline, +) -> Result<(SqlExecutionResult, Option), SqlError> { { let _timer = QueryTimer::new(); conn.execute_batch(REQUEST_BEGIN) @@ -626,32 +1038,71 @@ fn handle_message_authorized( let result = handle_request_authorized(conn, request, caveats, authority, pragmas); // An early error return can leave the request's authorizer installed. conn.authorizer(None::) -> Authorization>); - // SQLite can already have undone the transaction (ROLLBACK conflict - // handling); there is then nothing left to undo. - if result.is_err() && conn.is_autocommit() { - return result; - } - // A failure rolls the whole transaction back. `ROLLBACK TO` + `RELEASE` - // would instead commit an empty write transaction and rewrite the header. - let end = if result.is_ok() { - REQUEST_KEEP - } else { - "ROLLBACK" + // The error is surfaced before the rollback so the caller can still read + // the uncommitted page count to tell a size limit from a full disk. + // A request that took no write transaction (reads only, in a batch or an + // `execute`) wrote nothing: there is nothing to check, reserve or hold. + let result = result.and_then(|ok| { + if WriteProbe::of(conn).writing() { + precommit_gate(conn, ctx, base).map(|pending| (ok, Some(pending))) + } else { + Ok((ok, None)) + } + }); + let (result, held) = match result { + Ok(ok) => ok, + Err(error) => { + // SQLite can already have undone the transaction (ROLLBACK + // conflict handling); there is then nothing left to undo. + let full = is_full_error(&error); + let classified = full.then(|| classify_full(ctx, base, page_limit(ctx, base).1)); + if !conn.is_autocommit() { + // A failure rolls the whole transaction back. `ROLLBACK TO` + + // `RELEASE` would instead commit an empty write transaction + // and rewrite the header. + let _timer = QueryTimer::new(); + let _ = conn.execute_batch("ROLLBACK"); + } + return Err(classified.unwrap_or(error)); + } }; - let end_result = { + seam(&ctx.policy.seams.before_commit); + // RELEASE of the outermost savepoint is the COMMIT. + let commit = { let _timer = QueryTimer::new(); - conn.execute_batch(end) + conn.execute_batch(REQUEST_KEEP) }; - if let Err(error) = end_result { - // RELEASE of the outermost savepoint is the COMMIT. A deferred - // constraint can fail it and leave the writes and the savepoint open. + if let Err(error) = commit { if !conn.is_autocommit() { let _timer = QueryTimer::new(); let _ = conn.execute_batch("ROLLBACK"); } - return result.and(Err(SqlError::Sqlite(error.to_string()))); + return Err(commit_error(&error)); + } + seam(&ctx.policy.seams.after_commit); + Ok((result, held)) +} + +/// COMMIT failed. A deferred constraint is the request's own fault; anything +/// else is the disk. +/// +/// If COMMIT's fsync fails after the WAL frames were written, SQLite reports +/// the error and rolls the transaction back in memory, but the frames are +/// already in the file and, if their commit mark reached it, the commit can +/// reappear when the database is next opened — although we reported the +/// request as rejected. That is harmless while Postgres is master: the caller +/// discards the local state and re-hydrates from Postgres, which never saw the +/// write. Under store-master (1b+) it is an outcome-unknown case: the local +/// WAL must be kept and the request answered `outcome_unknown`, not rejected. +fn commit_error(error: &rusqlite::Error) -> SqlError { + match error.sqlite_error_code() { + Some(rusqlite::ErrorCode::ConstraintViolation) => SqlError::Sqlite(error.to_string()), + Some(rusqlite::ErrorCode::DiskFull) => SqlError::DiskHeadroom, + code => { + tracing::error!(?code, "Local SQL commit failed"); + SqlError::CommitFailed + } } - result } fn handle_request_authorized( @@ -1057,6 +1508,7 @@ fn execute_statement( #[cfg(test)] mod tests { use super::*; + use crate::sql::durability::Synchronous; #[tokio::test] async fn tc626_review_release_escape_cannot_survive_checkpoint() { @@ -1148,7 +1600,8 @@ mod tests { "BEGIN", ] { for prefix in ["", "PRAGMA user_version = 1; /* boundary; */ "] { - let conn = storage::open_connection(&StorageMode::InMemory).unwrap(); + let conn = + storage::open_connection(&StorageMode::InMemory, Synchronous::Full).unwrap(); conn.execute_batch( "CREATE TABLE data (value TEXT); INSERT INTO data VALUES ('before')", ) @@ -1186,7 +1639,7 @@ mod tests { #[test] fn tc626_review_single_pragma_preserves_quoted_semicolons_and_comments() { - let conn = storage::open_connection(&StorageMode::InMemory).unwrap(); + let conn = storage::open_connection(&StorageMode::InMemory, Synchronous::Full).unwrap(); conn.execute_batch("CREATE TABLE \"quoted;name\" (value TEXT)") .unwrap(); let result = handle_message_without_growth( @@ -1226,7 +1679,7 @@ mod tests { #[test] fn growth_guard_rolls_back_failed_deferred_commit() { - let conn = storage::open_connection(&StorageMode::InMemory).unwrap(); + let conn = storage::open_connection(&StorageMode::InMemory, Synchronous::Full).unwrap(); conn.execute_batch( "CREATE TABLE parents (id INTEGER PRIMARY KEY); CREATE TABLE children ( @@ -1661,16 +2114,10 @@ mod publication_rollback_tests { assert_eq!(scalar(&actor, "SELECT count(*) FROM t").await, "1"); assert_eq!(scalar(&actor, "SELECT v FROM t").await, "keep"); - // The `Forbid` guard predates this change and ends a rejected - // request with ROLLBACK TO + RELEASE, which commits an empty - // transaction and bumps the header's change counter; its - // content is checked above, its bytes only for `Allow`. - if growth == GrowthPolicy::Allow { - assert!( - actor.checkpoint().await.unwrap() == before, - "a rejected batch must not change a single byte of the checkpoint" - ); - } + assert!( + actor.checkpoint().await.unwrap() == before, + "a rejected batch must not change a single byte of the checkpoint" + ); } } } @@ -1884,7 +2331,8 @@ mod publication_rollback_tests { let dir = tempfile::tempdir().unwrap(); let file = dir.path().join("fail.db"); for mode in [StorageMode::InMemory, StorageMode::File(file.clone())] { - let conn = storage::open_connection(&mode).unwrap(); + let conn = + storage::open_connection(&mode, crate::sql::durability::Synchronous::Full).unwrap(); // `ON CONFLICT FAIL` on a column is outside what the SQL parser // accepts, so the schema is created on the connection directly. conn.execute_batch( @@ -2047,7 +2495,23 @@ mod ddl_authority_tests { caveats: &Option, authority: &SqlAuthority, ) -> Result { - handle_message_authorized(conn, request, caveats, authority, &PragmaTracker::default()) + let policy = DurabilityPolicy::default(); + let ctx = WriteContext { + policy: &policy, + key: "test/test", + dir: Path::new("."), + file_backed: false, + remaining: None, + file: Path::new(""), + }; + handle_message_ctx( + conn, + request, + caveats, + authority, + &PragmaTracker::default(), + &ctx, + ) } fn table_exists(conn: &rusqlite::Connection, name: &str) -> bool { @@ -2527,7 +2991,8 @@ mod ddl_authority_tests { StorageMode::InMemory, StorageMode::File(dir.path().join("reset.db")), ] { - let conn = storage::open_connection(&mode).unwrap(); + let conn = + storage::open_connection(&mode, crate::sql::durability::Synchronous::Full).unwrap(); conn.execute_batch("CREATE TABLE t (a)").unwrap(); let read = |pragma: &str| -> String { conn.query_row(&format!("PRAGMA {pragma}"), [], |row| { diff --git a/tinycloud-core/src/sql/durability.rs b/tinycloud-core/src/sql/durability.rs new file mode 100644 index 00000000..86ddce1c --- /dev/null +++ b/tinycloud-core/src/sql/durability.rs @@ -0,0 +1,406 @@ +//! Local commit durability and the reservation made before a write commits +//! (TC-1491, slice 1a). The contract, the invariants and what is and is not +//! guaranteed are in `docs/sql-local-durability.md`. + +use std::collections::HashMap; +use std::path::Path; +use std::sync::{Arc, Mutex}; + +use rusqlite::Connection; + +use super::types::SqlError; + +/// `PRAGMA synchronous` for file-backed databases. +#[derive( + Debug, Clone, Copy, PartialEq, Eq, Hash, Default, serde::Serialize, serde::Deserialize, +)] +#[serde(rename_all = "lowercase")] +pub enum Synchronous { + /// WAL fsync on every COMMIT: a commit that returned survives power loss + /// (given a volume that honours flushes). + #[default] + Full, + /// WAL fsync only at checkpoints: a power loss can drop the most recent + /// commits. The rollback flag for `Full`. + Normal, +} + +impl Synchronous { + pub fn pragma_value(self) -> &'static str { + match self { + Synchronous::Full => "FULL", + Synchronous::Normal => "NORMAL", + } + } +} + +/// Free space on the filesystem holding the SQL cache. A seam so tests can +/// exhaust it without filling a disk. +pub trait FreeSpace: Send + Sync { + fn free_bytes(&self, path: &Path) -> std::io::Result; +} + +/// `statvfs`: the bytes an unprivileged process can still allocate. +#[derive(Debug, Default)] +pub struct StatvfsFreeSpace; + +impl FreeSpace for StatvfsFreeSpace { + fn free_bytes(&self, path: &Path) -> std::io::Result { + use std::os::unix::ffi::OsStrExt; + let c_path = std::ffi::CString::new(path.as_os_str().as_bytes()) + .map_err(|_| std::io::Error::from(std::io::ErrorKind::InvalidInput))?; + let mut stat = std::mem::MaybeUninit::::uninit(); + // SAFETY: `c_path` is a valid NUL-terminated string and `stat` is a + // valid out-pointer; it is only read after statvfs reports success. + let rc = unsafe { libc::statvfs(c_path.as_ptr(), stat.as_mut_ptr()) }; + if rc != 0 { + return Err(std::io::Error::last_os_error()); + } + // SAFETY: statvfs succeeded, so it initialised the struct. + let stat = unsafe { stat.assume_init() }; + // The field widths differ by platform (u32 on macOS). + #[allow(clippy::unnecessary_cast)] + let free = (stat.f_bavail as u64).saturating_mul(stat.f_frsize as u64); + Ok(free) + } +} + +/// Why a reservation was refused. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReservationRefused { + /// Free space minus every in-flight reservation and the reservation asked + /// for would fall below the floor. + Headroom, + /// Free space could not be read; refusing is the conservative answer. + Unknown, +} + +/// What one database holds of the [`DiskBudget`]. +#[derive(Debug, Default, Clone, Copy)] +struct Held { + /// The WAL bytes of the write in flight and the checkpoint growth it + /// causes; released when the request's outcome is known. + inflight: u64, + /// The growth of the main file the next checkpoint will cause: pages + /// committed since the last one, which are in the WAL until then. Held + /// until a checkpoint runs or the database's connection closes, because + /// the disk does not have those bytes until then and nothing else pays for + /// them (the save to Postgres ships the WAL, not a checkpoint, while the + /// WAL is small). + pending: u64, +} + +impl Held { + fn total(self) -> u64 { + self.inflight.saturating_add(self.pending) + } +} + +/// The disk bytes in-flight writes of every database on this node still need. +/// One per node, shared by all actors, so two databases cannot each take the +/// same free bytes. +#[derive(Debug, Default)] +pub struct DiskBudget { + held: Mutex>, +} + +impl DiskBudget { + /// Set `key`'s reservation to `inflight` bytes for the write in flight and + /// `pending` bytes of checkpoint growth, if the disk can still hold both + /// next to everyone else's. Leaves the old reservation when refusing. + pub fn reserve( + &self, + key: &str, + inflight: u64, + pending: u64, + free_space: &dyn FreeSpace, + dir: &Path, + min_free_bytes: u64, + ) -> Result<(), ReservationRefused> { + let mut held = self.held.lock().unwrap_or_else(|e| e.into_inner()); + let others: u64 = held + .iter() + .filter(|(other, _)| other.as_str() != key) + .map(|(_, held)| held.total()) + .fold(0, u64::saturating_add); + let free = free_space + .free_bytes(dir) + .map_err(|_| ReservationRefused::Unknown)?; + let needed = others + .saturating_add(inflight) + .saturating_add(pending) + .saturating_add(min_free_bytes); + if free < needed { + return Err(ReservationRefused::Headroom); + } + held.insert(key.to_string(), Held { inflight, pending }); + Ok(()) + } + + /// The write is committed: its WAL bytes are on disk. Keep `pending` bytes + /// of checkpoint growth held, without a free-space check. + pub fn hold(&self, key: &str, pending: u64) { + let mut held = self.held.lock().unwrap_or_else(|e| e.into_inner()); + if pending == 0 { + held.remove(key); + } else { + held.insert( + key.to_string(), + Held { + inflight: 0, + pending, + }, + ); + } + } + + /// The write in flight is over (refused, failed or saved): drop what it + /// reserved. The checkpoint growth of commits that stay is still held. + pub fn release(&self, key: &str) { + let mut held = self.held.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(entry) = held.get_mut(key) { + entry.inflight = 0; + if entry.pending == 0 { + held.remove(key); + } + } + } + + /// Drop everything held for `key`: a checkpoint has run, or the + /// database's connection is gone, and what is on disk is on disk. + pub fn forget(&self, key: &str) { + self.held + .lock() + .unwrap_or_else(|e| e.into_inner()) + .remove(key); + } + + /// Bytes reserved by every database but `key`. + pub fn reserved_excluding(&self, key: &str) -> u64 { + self.held + .lock() + .unwrap_or_else(|e| e.into_inner()) + .iter() + .filter(|(other, _)| other.as_str() != key) + .map(|(_, held)| held.total()) + .fold(0, u64::saturating_add) + } + + /// Bytes currently reserved by all databases. + pub fn reserved(&self) -> u64 { + self.held + .lock() + .unwrap_or_else(|e| e.into_inner()) + .values() + .fold(0, |sum, held| sum.saturating_add(held.total())) + } +} + +/// Releases a database's in-flight reservation when the request's outcome is +/// known. +pub struct ReservationGuard { + budget: Arc, + key: String, +} + +impl ReservationGuard { + pub fn new(budget: Arc, key: String) -> Self { + Self { budget, key } + } +} + +impl Drop for ReservationGuard { + fn drop(&mut self) { + self.budget.release(&self.key); + } +} + +/// Forgets everything a database holds of the [`DiskBudget`] when its actor +/// ends, however it ends. +pub struct HeldGuard { + budget: Arc, + key: String, +} + +impl HeldGuard { + pub fn new(budget: Arc, key: String) -> Self { + Self { budget, key } + } +} + +impl Drop for HeldGuard { + fn drop(&mut self) { + self.budget.forget(&self.key); + } +} + +/// Whether the main database has a write transaction open on a connection, +/// asked of SQLite itself (`sqlite3_txn_state`). A request that never took one +/// wrote nothing: it needs no room on the disk and nothing to reserve. It is +/// set by the first statement that writes (an `UPDATE` that matches no row +/// included) and stays until the transaction ends. +pub struct WriteProbe(*mut rusqlite::ffi::sqlite3); + +// SAFETY: a probe is only used while the connection it was made from is alive +// and on the thread that runs the request (SQLite calls the progress handler on +// the thread running the statement), and `sqlite3_txn_state` only reads the +// connection's own state. +unsafe impl Send for WriteProbe {} + +impl WriteProbe { + pub fn of(conn: &Connection) -> Self { + // SAFETY: the pointer is only dereferenced by SQLite, through + // `writing`, while `conn` is alive (see the `Send` note above). + Self(unsafe { conn.handle() }) + } + + pub fn writing(&self) -> bool { + // SAFETY: see `WriteProbe::of`; "main" is a NUL-terminated literal. + let state = unsafe { rusqlite::ffi::sqlite3_txn_state(self.0, c"main".as_ptr()) }; + state == rusqlite::ffi::SQLITE_TXN_WRITE + } +} + +/// Test seams at the commit boundary. A seam only observes: the kill -9 +/// tests abort the process from one. +#[derive(Clone, Default)] +pub struct CommitSeams { + /// After every check passed, immediately before COMMIT. + pub before_commit: Option>, + /// After COMMIT returned (and, for `Full`, its fsync), before the actor + /// answers. + pub after_commit: Option>, + /// Just before the actor sends its response. + pub before_response: Option>, +} + +impl std::fmt::Debug for CommitSeams { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("CommitSeams").finish_non_exhaustive() + } +} + +/// 0 turns the per-database cap off. Off by default: the limit is a product +/// decision (card `sql-storage-report-read`), set by config once decided. +pub const DEFAULT_MAX_DATABASE_BYTES: u64 = 0; +/// 0 turns the disk headroom check off (no reservation, no free-space reads). +/// Off by default for the same reason. +pub const DEFAULT_MIN_FREE_BYTES: u64 = 0; + +/// Everything the actors need to make a commit durable and to refuse a write +/// that would not fit. +#[derive(Clone)] +pub struct DurabilityPolicy { + pub synchronous: Synchronous, + /// Per-database size cap, enforced by SQLite itself through + /// `max_page_count`. 0 = no cap. + pub max_database_bytes: u64, + /// Free bytes that must remain after every in-flight write's worst case. + /// 0 = the disk headroom check is off. + pub min_free_bytes: u64, + pub budget: Arc, + pub free_space: Arc, + pub seams: CommitSeams, +} + +impl Default for DurabilityPolicy { + fn default() -> Self { + Self { + synchronous: Synchronous::Full, + max_database_bytes: DEFAULT_MAX_DATABASE_BYTES, + min_free_bytes: DEFAULT_MIN_FREE_BYTES, + budget: Arc::new(DiskBudget::default()), + free_space: Arc::new(StatvfsFreeSpace), + seams: CommitSeams::default(), + } + } +} + +impl std::fmt::Debug for DurabilityPolicy { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DurabilityPolicy") + .field("synchronous", &self.synchronous) + .field("max_database_bytes", &self.max_database_bytes) + .field("min_free_bytes", &self.min_free_bytes) + .finish_non_exhaustive() + } +} + +/// `fsync` a directory so the entries created or renamed in it survive a +/// crash. SQLite syncs the directory itself when it first creates a `-wal` +/// file (`UNIXFILE_DIRSYNC` in its unix VFS), but not for the main file. +pub fn sync_dir(dir: &Path) -> std::io::Result<()> { + std::fs::File::open(dir)?.sync_all() +} + +/// `create_dir_all`, then make every directory it created durable: each new +/// directory and the existing directory that now holds the first of them. +pub fn create_dir_all_durable(path: &Path) -> std::io::Result<()> { + let mut missing = Vec::new(); + let mut cursor = Some(path); + while let Some(dir) = cursor { + if dir.as_os_str().is_empty() || dir.exists() { + break; + } + missing.push(dir); + cursor = dir.parent(); + } + if missing.is_empty() { + return Ok(()); + } + std::fs::create_dir_all(path)?; + for dir in missing.iter().rev() { + sync_dir(dir)?; + if let Some(parent) = dir.parent().filter(|p| !p.as_os_str().is_empty()) { + sync_dir(parent)?; + } + } + Ok(()) +} + +/// Apply `synchronous` to a file-backed connection. +pub fn set_synchronous(conn: &Connection, synchronous: Synchronous) -> Result<(), SqlError> { + conn.pragma_update(None, "synchronous", synchronous.pragma_value()) + .map_err(|e| SqlError::Internal(e.to_string())) +} + +/// Current heap bytes of this connection's page cache. Every page SQLite has +/// yet to write to the WAL at COMMIT is in it. +pub fn cache_used_bytes(conn: &Connection) -> u64 { + let mut current: std::os::raw::c_int = 0; + let mut high: std::os::raw::c_int = 0; + // SAFETY: the handle is valid for the borrow of `conn`, and the two + // out-pointers are valid for the call. `sqlite3_db_status` only reads the + // connection's own counters. + let rc = unsafe { + rusqlite::ffi::sqlite3_db_status( + conn.handle(), + rusqlite::ffi::SQLITE_DBSTATUS_CACHE_USED, + &mut current, + &mut high, + 0, + ) + }; + if rc != rusqlite::ffi::SQLITE_OK { + // Unknown: the high-water mark is the largest it has been. + return u64::try_from(high.max(0)).unwrap_or(0); + } + u64::try_from(current.max(0)).unwrap_or(0) +} + +/// SQLite's per-frame WAL overhead. +pub const WAL_FRAME_HEADER_BYTES: u64 = 24; + +/// Upper bound on the bytes COMMIT still appends to the WAL: one frame per +/// page in the cache. Dirty pages are a subset of cached pages, so this +/// bounds the writes of index, trigger, foreign-key, freelist and overflow +/// amplification alike, whatever the SQL payload was. A page dirtied twice +/// in one transaction still costs one frame. +pub fn unspilled_wal_bound(cache_used: u64, page_size: u64) -> u64 { + if page_size == 0 { + return cache_used; + } + cache_used + .div_ceil(page_size) + .saturating_mul(page_size + WAL_FRAME_HEADER_BYTES) +} diff --git a/tinycloud-core/src/sql/mod.rs b/tinycloud-core/src/sql/mod.rs index c6794394..755ff4c6 100644 --- a/tinycloud-core/src/sql/mod.rs +++ b/tinycloud-core/src/sql/mod.rs @@ -3,6 +3,7 @@ pub mod authorizer; pub mod caveats; pub mod database; pub mod delegated_caveats; +pub mod durability; pub mod parser; pub mod service; pub mod storage; diff --git a/tinycloud-core/src/sql/service.rs b/tinycloud-core/src/sql/service.rs index 4c2f09b0..0442e4ee 100644 --- a/tinycloud-core/src/sql/service.rs +++ b/tinycloud-core/src/sql/service.rs @@ -27,7 +27,8 @@ use crate::{request_cancel, request_telemetry}; use super::{ authority::{DdlPolicy, SqlAuthority}, caveats::SqlCaveats, - database::{spawn_actor, DatabaseHandle, GrowthPolicy}, + database::{spawn_actor_with_policy, DatabaseHandle, GrowthPolicy}, + durability::{DurabilityPolicy, ReservationGuard}, types::*, }; @@ -161,6 +162,7 @@ pub struct SqlService { memory_threshold: u64, artifact_repository: Arc, bounds: SqlWaitBounds, + durability: DurabilityPolicy, ddl_policy: DdlPolicy, } @@ -179,10 +181,18 @@ impl SqlService { memory_threshold, artifact_repository, bounds: SqlWaitBounds::default(), + durability: DurabilityPolicy::default(), ddl_policy: DdlPolicy::default(), } } + /// The node's local durability policy (TC-1491): `synchronous`, the + /// per-database size cap and the disk headroom. + pub fn with_durability(mut self, durability: DurabilityPolicy) -> Self { + self.durability = durability; + self + } + /// Whether a schema-changing statement needs `tinycloud.sql/schema` /// (TC-1581). Defaults to [`DdlPolicy::Enforce`]. pub fn with_ddl_policy(mut self, policy: DdlPolicy) -> Self { @@ -279,6 +289,52 @@ impl SqlService { .await } + /// Execute with the owner's remaining storage budget: a write that would + /// grow the database by more than `remaining_bytes` is rolled back before + /// COMMIT with [`SqlError::StorageWouldGrow`]. + pub async fn execute_with_budget( + &self, + space: &SpaceId, + db_name: &str, + request: SqlRequest, + caveats: Option, + ability: String, + remaining_bytes: u64, + ) -> Result { + self.execute_with_budget_authorized( + space, + db_name, + request, + caveats, + SqlAuthority::from(ability), + remaining_bytes, + ) + .await + } + + /// As [`Self::execute_with_budget`], judging each statement against every + /// SQL ability the invocation holds (TC-1581). This is the entry point + /// `/invoke` uses for a write inside the owner's remaining budget. + pub async fn execute_with_budget_authorized( + &self, + space: &SpaceId, + db_name: &str, + request: SqlRequest, + caveats: Option, + authority: SqlAuthority, + remaining_bytes: u64, + ) -> Result { + self.execute_with_growth( + space, + db_name, + request, + caveats, + authority.with_ddl_policy(self.ddl_policy), + GrowthPolicy::Budget(remaining_bytes), + ) + .await + } + async fn execute_with_growth( &self, space: &SpaceId, @@ -310,6 +366,12 @@ impl SqlService { growth: GrowthPolicy, ) -> Result { let key = (space.to_string(), db_name.to_string()); + // Whatever the request still holds of the disk budget is released once + // its outcome is known: after the durable save, or its failure. + let _reservation = ReservationGuard::new( + self.durability.budget.clone(), + format!("{}/{}", key.0, key.1), + ); let mut handle = self.handle(space, db_name).await?; begin_write()?; @@ -331,8 +393,11 @@ impl SqlService { let result = match result { Ok(result) => result, Err(error) => { - if growth == GrowthPolicy::Forbid - && self.expectation(&key) == ArtifactExpectation::Absent + if (growth == GrowthPolicy::Forbid + && self.expectation(&key) == ArtifactExpectation::Absent) + // The disk failed the commit: its WAL is not to be trusted + // (Postgres never saw the write), so start over from it. + || matches!(error, SqlError::CommitFailed) { self.discard_local_state(&key).await?; } @@ -750,12 +815,13 @@ impl SqlService { self.databases .entry((space.to_string(), db_name.to_string())) .or_insert_with(|| { - spawn_actor( + spawn_actor_with_policy( space.to_string(), db_name.to_string(), self.base_path.clone(), self.memory_threshold, self.databases.clone(), + self.durability.clone(), ) }); HydrationOutcome::Ready @@ -861,7 +927,7 @@ impl SqlService { // adding a database page. Full spaces replace checkpoint + WAL, but // must also bound the resulting image against the durable charge: // live state can include partial writes from an earlier failed request. - let wal = if growth == GrowthPolicy::Allow { + let wal = if growth != GrowthPolicy::Forbid { handle.wal().await? } else { None @@ -4781,6 +4847,186 @@ mod tests { ); } } + + /// TC-1491: the budget and the disk reservation cost no Postgres round + /// trip, and a refused write costs none at all. + #[tokio::test] + async fn tc1491_reservation_adds_no_repository_round_trips() { + for memory_threshold in [u64::MAX, 0] { + let mut sequences = Vec::new(); + for budget in [None, Some(1 << 30)] { + let calls = Arc::new(std::sync::Mutex::new(Vec::new())); + let cache = TempDir::new().unwrap(); + let service = SqlService::new( + cache.path().to_string_lossy().into_owned(), + memory_threshold, + Arc::new(CallCountingRepository { + inner: artifact_repository().await, + calls: calls.clone(), + }), + ); + let space = test_space_id("sql-reservation-trips"); + let write = |sql: &str| { + let request = SqlRequest::Execute { + schema: None, + sql: sql.to_string(), + params: Vec::new(), + }; + let ability = "tinycloud.sql/admin".to_string(); + let space = space.clone(); + let service = service.clone(); + async move { + match budget { + None => { + service + .execute(&space, "main", request, None, ability) + .await + } + Some(left) => { + service + .execute_with_budget( + &space, "main", request, None, ability, left, + ) + .await + } + } + } + }; + write("CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)") + .await + .unwrap(); + write("INSERT INTO t (v) VALUES (zeroblob(1000))") + .await + .unwrap(); + sequences.push(calls.lock().unwrap().clone()); + } + assert_eq!( + sequences[0], sequences[1], + "a budget changes which repository calls a write makes" + ); + // One persist per write, after the initial hydration lookup. + let persists = sequences[0] + .iter() + .filter(|call| matches!(**call, "save" | "save_delta")) + .count(); + assert_eq!(persists, 2, "{:?}", sequences[0]); + } + + // A write the budget refuses reaches Postgres not at all, and leaves the + // local database as it was. + let calls = Arc::new(std::sync::Mutex::new(Vec::new())); + let cache = TempDir::new().unwrap(); + let service = SqlService::new( + cache.path().to_string_lossy().into_owned(), + 0, + Arc::new(CallCountingRepository { + inner: artifact_repository().await, + calls: calls.clone(), + }), + ); + let space = test_space_id("sql-reservation-refused"); + let request = |sql: &str| SqlRequest::Execute { + schema: None, + sql: sql.to_string(), + params: Vec::new(), + }; + service + .execute( + &space, + "main", + request("CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)"), + None, + "tinycloud.sql/admin".into(), + ) + .await + .unwrap(); + calls.lock().unwrap().clear(); + let refused = service + .execute_with_budget( + &space, + "main", + request("INSERT INTO t (v) VALUES (zeroblob(500000))"), + None, + "tinycloud.sql/write".into(), + 4096, + ) + .await; + assert!( + matches!(refused, Err(SqlError::StorageWouldGrow)), + "{refused:?}" + ); + assert!( + calls.lock().unwrap().is_empty(), + "{:?}", + calls.lock().unwrap() + ); + assert!(rows_of(&service, &space, "SELECT id FROM t") + .await + .is_empty()); + assert_eq!(service.durability.budget.reserved(), 0); + } + + /// TC-1491: with the disk headroom check on, a successful write holds + /// nothing once its save is done, except the growth the next checkpoint + /// will cause. A save of a small WAL ships the WAL and does not + /// checkpoint, so that growth stays held until a checkpoint runs. + #[tokio::test] + async fn tc1491_a_saved_write_holds_only_its_pending_checkpoint_growth() { + let cache = TempDir::new().unwrap(); + let durability = crate::sql::durability::DurabilityPolicy { + min_free_bytes: 1, + ..Default::default() + }; + let service = SqlService::new( + cache.path().to_string_lossy().into_owned(), + 0, + artifact_repository().await, + ) + .with_durability(durability); + let space = test_space_id("sql-reservation-pending"); + let write = |sql: &str| { + service.execute_with_budget( + &space, + "main", + SqlRequest::Execute { + schema: None, + sql: sql.to_string(), + params: Vec::new(), + }, + None, + "tinycloud.sql/admin".into(), + 1 << 30, + ) + }; + let budget = &service.durability.budget; + + write("CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)") + .await + .unwrap(); + assert_eq!(budget.reserved(), 0); + + // Saved as a WAL delta: no checkpoint, so the growth is still held. + write("INSERT INTO t (v) VALUES (zeroblob(400000))") + .await + .unwrap(); + let held = budget.reserved(); + assert!(held >= 390_000, "held {held}"); + write("INSERT INTO t (v) VALUES (zeroblob(10))") + .await + .unwrap(); + assert!(budget.reserved() >= 390_000); + + // A checkpoint puts it on the disk; a write that adds no page then + // leaves nothing held after its save. + let handle = service.handle(&space, "main").await.unwrap(); + handle.checkpoint().await.unwrap(); + assert_eq!(budget.reserved(), 0); + write("INSERT INTO t (v) VALUES (zeroblob(10))") + .await + .unwrap(); + assert_eq!(budget.reserved(), 0); + } + /// TC-1491 + TC-1585: a `WITH … INSERT` is a write, so in a batch it is /// one of the statements that must be kept or undone with the rest. A /// later failure leaves nothing of it locally, writes nothing to Postgres, diff --git a/tinycloud-core/src/sql/storage.rs b/tinycloud-core/src/sql/storage.rs index ae0de8f4..d0e9661a 100644 --- a/tinycloud-core/src/sql/storage.rs +++ b/tinycloud-core/src/sql/storage.rs @@ -4,6 +4,7 @@ use rusqlite::Connection; use crate::request_telemetry::QueryTimer; +use super::durability::{create_dir_all_durable, set_synchronous, sync_dir, Synchronous}; use super::types::SqlError; #[derive(Debug, Clone)] @@ -12,14 +13,17 @@ pub enum StorageMode { File(PathBuf), } -pub fn open_connection(mode: &StorageMode) -> Result { +pub fn open_connection( + mode: &StorageMode, + synchronous: Synchronous, +) -> Result { let conn = match mode { StorageMode::InMemory => { Connection::open_in_memory().map_err(|e| SqlError::Internal(e.to_string()))? } StorageMode::File(path) => { if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent).map_err(|e| SqlError::Internal(e.to_string()))?; + create_dir_all_durable(parent).map_err(|e| SqlError::Internal(e.to_string()))?; } Connection::open(path).map_err(|e| SqlError::Internal(e.to_string()))? } @@ -27,6 +31,8 @@ pub fn open_connection(mode: &StorageMode) -> Result { // Enable WAL mode for file-backed databases if matches!(mode, StorageMode::File(_)) { + // Before the journal mode: the WAL switch is itself a write. + set_synchronous(&conn, synchronous)?; { let _timer = QueryTimer::new(); conn.pragma_update(None, "journal_mode", "wal") @@ -63,9 +69,13 @@ pub fn database_size(conn: &Connection) -> Result { Ok(page_count * page_size) } -pub fn promote_to_file(conn: &Connection, path: &PathBuf) -> Result { +pub fn promote_to_file( + conn: &Connection, + path: &PathBuf, + synchronous: Synchronous, +) -> Result { if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent).map_err(|e| SqlError::Internal(e.to_string()))?; + create_dir_all_durable(parent).map_err(|e| SqlError::Internal(e.to_string()))?; } // rusqlite appends the file path, which holds the space id, to open @@ -77,6 +87,9 @@ pub fn promote_to_file(conn: &Connection, path: &PathBuf) -> Result Result 0); let guarded = request("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"); diff --git a/tinycloud-core/tests/sql_local_durability.rs b/tinycloud-core/tests/sql_local_durability.rs new file mode 100644 index 00000000..a03e9897 --- /dev/null +++ b/tinycloud-core/tests/sql_local_durability.rs @@ -0,0 +1,1186 @@ +//! TC-1491 slice 1a: local commit durability and the reservation made before +//! COMMIT. See `docs/sql-local-durability.md` for the contract these pin. + +use std::io::{BufRead, BufReader}; +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; + +use dashmap::DashMap; +use tinycloud_core::sql::database::{spawn_actor_with_policy, DatabaseHandle, GrowthPolicy}; +use tinycloud_core::sql::durability::{ + create_dir_all_durable, CommitSeams, DurabilityPolicy, FreeSpace, Synchronous, +}; +use tinycloud_core::sql::storage::{open_connection, promote_to_file, StorageMode}; +use tinycloud_core::sql::{SqlError, SqlExecutionResult, SqlRequest, SqlResponse, SqlValue}; + +const SPACE: &str = "space"; +const DB: &str = "durable"; + +/// Free space that tests set by hand. +#[derive(Debug)] +struct FakeFree(AtomicU64); + +impl FakeFree { + fn new(bytes: u64) -> Arc { + Arc::new(Self(AtomicU64::new(bytes))) + } + fn set(&self, bytes: u64) { + self.0.store(bytes, Ordering::SeqCst); + } +} + +impl FreeSpace for FakeFree { + fn free_bytes(&self, _: &Path) -> std::io::Result { + Ok(self.0.load(Ordering::SeqCst)) + } +} + +/// A policy with the headroom check ON (`min_free_bytes` is 0 = off by default, +/// so the tests that exercise the mechanism pass an explicit value). +fn policy_with(free: &Arc, min_free_bytes: u64) -> DurabilityPolicy { + assert!(min_free_bytes > 0); + DurabilityPolicy { + min_free_bytes, + free_space: free.clone(), + ..Default::default() + } +} + +/// A file-backed actor: with a zero threshold the first write promotes it. +fn actor(dir: &Path, db: &str, policy: DurabilityPolicy) -> DatabaseHandle { + spawn_actor_with_policy( + SPACE.into(), + db.into(), + dir.to_str().unwrap().into(), + 0, + Arc::new(DashMap::new()), + policy, + ) +} + +fn exec(sql: &str) -> SqlRequest { + SqlRequest::Execute { + schema: None, + sql: sql.into(), + params: vec![], + } +} + +fn batch(statements: &[&str]) -> SqlRequest { + SqlRequest::Batch { + statements: statements + .iter() + .map(|sql| tinycloud_core::sql::types::SqlStatement { + sql: (*sql).into(), + params: vec![], + }) + .collect(), + } +} + +async fn run( + actor: &DatabaseHandle, + request: SqlRequest, + growth: GrowthPolicy, +) -> Result { + actor + .execute(request, None, "tinycloud.sql/admin".into(), growth) + .await +} + +async fn ok(actor: &DatabaseHandle, sql: &str) { + run(actor, exec(sql), GrowthPolicy::Allow) + .await + .unwrap_or_else(|error| panic!("{sql}: {error}")); +} + +async fn rows(actor: &DatabaseHandle, sql: &str) -> Vec> { + match run( + actor, + SqlRequest::Query { + sql: sql.into(), + params: vec![], + max_rows: None, + max_bytes: None, + }, + GrowthPolicy::Allow, + ) + .await + .unwrap() + .response + { + SqlResponse::Query(query) => query.rows, + other => panic!("expected rows, got {other:?}"), + } +} + +async fn count(actor: &DatabaseHandle, table: &str) -> i64 { + match rows(actor, &format!("SELECT count(*) FROM {table}")).await[0][0] { + SqlValue::Integer(n) => n, + ref other => panic!("{other:?}"), + } +} + +/// Six indexes: a short statement that dirties many pages. (The SQL parser +/// admits no triggers; indexes are the amplification a caller can cause.) +const AMPLIFIED_SCHEMA: &[&str] = &[ + "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT, b TEXT, c TEXT, d TEXT, e TEXT, f TEXT)", + "CREATE INDEX t_a ON t (a)", + "CREATE INDEX t_b ON t (b)", + "CREATE INDEX t_c ON t (c)", + "CREATE INDEX t_d ON t (d)", + "CREATE INDEX t_e ON t (e)", + "CREATE INDEX t_f ON t (f)", +]; +/// ~300 bytes of SQL text; adds 20 000 rows to seven b-trees. +const AMPLIFIED_INSERT: &str = "INSERT INTO t (a, b, c, d, e, f) \ + SELECT hex(randomblob(12)), hex(randomblob(12)), hex(randomblob(12)), \ + hex(randomblob(12)), hex(randomblob(12)), hex(randomblob(12)) \ + FROM seed s1, seed s2 LIMIT 20000"; + +/// 256 rows to cross-join, built by doubling. +async fn seed(actor: &DatabaseHandle) { + ok(actor, "CREATE TABLE seed (x INTEGER)").await; + ok(actor, "INSERT INTO seed VALUES (1)").await; + for _ in 0..8 { + ok(actor, "INSERT INTO seed SELECT x FROM seed").await; + } +} + +fn wal_len(dir: &Path, db: &str) -> u64 { + std::fs::metadata(dir.join(SPACE).join(format!("{db}.db-wal"))) + .map(|m| m.len()) + .unwrap_or(0) +} + +#[test] +fn synchronous_is_full_on_every_file_backed_connection_and_normal_on_request() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("a").join("b").join("x.db"); + let read = |conn: &rusqlite::Connection| -> i64 { + conn.query_row("PRAGMA synchronous", [], |row| row.get(0)) + .unwrap() + }; + // PRAGMA synchronous: 1 = NORMAL, 2 = FULL. + let opened = open_connection(&StorageMode::File(path.clone()), Synchronous::Full).unwrap(); + assert_eq!(read(&opened), 2); + let normal = open_connection(&StorageMode::File(path.clone()), Synchronous::Normal).unwrap(); + assert_eq!(read(&normal), 1); + drop((opened, normal)); + + let memory = open_connection(&StorageMode::InMemory, Synchronous::Full).unwrap(); + memory.execute_batch("CREATE TABLE t (v)").unwrap(); + for (setting, expected) in [(Synchronous::Full, 2), (Synchronous::Normal, 1)] { + let target = dir.path().join(format!("promoted-{expected}.db")); + let promoted = promote_to_file(&memory, &target, setting).unwrap(); + assert_eq!(read(&promoted), expected); + let mode: String = promoted + .query_row("PRAGMA journal_mode", [], |row| row.get(0)) + .unwrap(); + assert_eq!(mode, "wal"); + } +} + +/// An actor runs under either `synchronous` setting through promotion to a +/// file and a reopen. The setting itself is asserted on the connections in +/// `synchronous_is_full_on_every_file_backed_connection_and_normal_on_request`: +/// since TC-1586 an admin can no longer read `PRAGMA synchronous` through the +/// actor (it is not on the PRAGMA allow-list). +#[tokio::test] +async fn an_actor_runs_under_either_synchronous_setting_after_promotion_and_reopen() { + for setting in [Synchronous::Full, Synchronous::Normal] { + let dir = tempfile::tempdir().unwrap(); + let policy = DurabilityPolicy { + synchronous: setting, + ..Default::default() + }; + let first = actor(dir.path(), DB, policy.clone()); + ok(&first, "CREATE TABLE t (v INTEGER)").await; // promotes to a file + ok(&first, "INSERT INTO t VALUES (1)").await; + assert_eq!( + rows(&first, "PRAGMA journal_mode").await[0][0], + SqlValue::Text("wal".into()) + ); + first.shutdown().await.unwrap(); + let reopened = actor(dir.path(), DB, policy); + assert_eq!(count(&reopened, "t").await, 1); + } +} + +#[tokio::test] +async fn an_amplified_transaction_is_rejected_before_commit_and_leaves_no_effects() { + let dir = tempfile::tempdir().unwrap(); + // 1 MiB free, no floor: the SQL text is a few hundred bytes, but the pages + // it dirties are several MiB. + let free = FakeFree::new(1 << 20); + let a = actor(dir.path(), DB, policy_with(&free, 1)); + free.set(64 << 20); + for statement in AMPLIFIED_SCHEMA { + ok(&a, statement).await; + } + seed(&a).await; + ok(&a, "INSERT INTO t (a, b) VALUES ('keep', 'keep')").await; + assert!(AMPLIFIED_INSERT.len() < 600); + + let checkpoint_before = a.checkpoint().await.unwrap(); + let export_before = a.export().await.unwrap(); + + free.set(1 << 20); + let refused = run(&a, exec(AMPLIFIED_INSERT), GrowthPolicy::Allow).await; + assert!( + matches!(refused, Err(SqlError::DiskHeadroom)), + "{refused:?}" + ); + + // No effect anywhere: rows, the snapshot and export. + free.set(64 << 20); + assert_eq!(count(&a, "t").await, 1); + assert_eq!(a.checkpoint().await.unwrap(), checkpoint_before); + assert_eq!(a.export().await.unwrap(), export_before); + + // The same statement fits when the disk has room. + free.set(1 << 30); + ok(&a, AMPLIFIED_INSERT).await; + assert_eq!(count(&a, "t").await, 20_001); +} + +#[tokio::test] +async fn a_batch_is_all_or_nothing_under_each_kind_of_rejection() { + let dir = tempfile::tempdir().unwrap(); + let free = FakeFree::new(64 << 20); + let policy = DurabilityPolicy { + // 32 pages of 4 KiB. + max_database_bytes: 128 * 1024, + min_free_bytes: 1, + free_space: free.clone(), + ..Default::default() + }; + let a = actor(dir.path(), DB, policy); + for statement in AMPLIFIED_SCHEMA { + ok(&a, statement).await; + } + let baseline_checkpoint = a.checkpoint().await.unwrap(); + + // 1. Size cap: the first statements apply, the last one passes the cap. + let over_cap = run( + &a, + batch(&[ + "INSERT INTO t (a) VALUES ('first')", + "INSERT INTO t (a) VALUES ('second')", + "INSERT INTO t (a) VALUES (hex(zeroblob(200000)))", + ]), + GrowthPolicy::Allow, + ) + .await; + assert!( + matches!(over_cap, Err(SqlError::DatabaseSizeLimit)), + "{over_cap:?}" + ); + assert_eq!(count(&a, "t").await, 0); + + // 2. Disk headroom at the gate. + free.set(1 << 10); + let no_room = run( + &a, + batch(&[ + "INSERT INTO t (a) VALUES ('first')", + "INSERT INTO t (a) VALUES ('second')", + ]), + GrowthPolicy::Allow, + ) + .await; + assert!( + matches!(no_room, Err(SqlError::DiskHeadroom)), + "{no_room:?}" + ); + free.set(64 << 20); + assert_eq!(count(&a, "t").await, 0); + + // 3. The owner's budget. + let over_budget = run( + &a, + batch(&[ + "INSERT INTO t (a) VALUES ('first')", + "INSERT INTO t (a) VALUES (hex(zeroblob(40000)))", + ]), + GrowthPolicy::Budget(8192), + ) + .await; + assert!( + matches!(over_budget, Err(SqlError::StorageWouldGrow)), + "{over_budget:?}" + ); + assert_eq!(count(&a, "t").await, 0); + assert_eq!(a.checkpoint().await.unwrap(), baseline_checkpoint); + + // 4. A statement error after earlier statements. + let broken = run( + &a, + batch(&[ + "INSERT INTO t (a) VALUES ('first')", + "INSERT INTO nope VALUES (1)", + ]), + GrowthPolicy::Allow, + ) + .await; + assert!(broken.is_err()); + assert_eq!(count(&a, "t").await, 0); + assert_eq!(a.checkpoint().await.unwrap(), baseline_checkpoint); + + // And the same batch passes when nothing is wrong. + ok(&a, "INSERT INTO t (a) VALUES ('kept')").await; + assert_eq!(count(&a, "t").await, 1); +} + +/// TC-1585: a `WITH … INSERT` is a query to the parser but a write to the +/// node. It gets the transaction, the owner's budget, the size cap and the +/// pre-commit disk gate, and a refusal leaves nothing: no rows, no checkpoint +/// change, no reservation. +#[tokio::test] +async fn a_nested_write_is_bounded_before_commit_like_any_other_write() { + const CTE_WRITE: &str = + "WITH c AS (SELECT hex(zeroblob(40000)) AS v) INSERT INTO t (a) SELECT v FROM c"; + let query = |sql: &str| SqlRequest::Query { + sql: sql.into(), + params: vec![], + max_rows: None, + max_bytes: None, + }; + let dir = tempfile::tempdir().unwrap(); + let free = FakeFree::new(64 << 20); + let policy = DurabilityPolicy { + // 32 pages of 4 KiB. + max_database_bytes: 128 * 1024, + min_free_bytes: 1, + free_space: free.clone(), + ..Default::default() + }; + let a = actor(dir.path(), DB, policy.clone()); + for statement in AMPLIFIED_SCHEMA { + ok(&a, statement).await; + } + let baseline_checkpoint = a.checkpoint().await.unwrap(); + let wal_before = wal_len(dir.path(), DB); + + // The owner's budget. + let over_budget = run(&a, query(CTE_WRITE), GrowthPolicy::Budget(8192)).await; + assert!( + matches!(over_budget, Err(SqlError::StorageWouldGrow)), + "{over_budget:?}" + ); + // The per-database cap. + let over_cap = run( + &a, + query("WITH c AS (SELECT hex(zeroblob(200000)) AS v) INSERT INTO t (a) SELECT v FROM c"), + GrowthPolicy::Allow, + ) + .await; + assert!( + matches!(over_cap, Err(SqlError::DatabaseSizeLimit)), + "{over_cap:?}" + ); + // The disk gate, at the pre-commit check. + free.set(1 << 10); + let no_room = run( + &a, + query("WITH c AS (SELECT hex(zeroblob(2000)) AS v) INSERT INTO t (a) SELECT v FROM c"), + GrowthPolicy::Allow, + ) + .await; + assert!( + matches!(no_room, Err(SqlError::DiskHeadroom)), + "{no_room:?}" + ); + free.set(64 << 20); + + assert_eq!(count(&a, "t").await, 0); + assert_eq!(a.checkpoint().await.unwrap(), baseline_checkpoint); + assert_eq!(wal_len(dir.path(), DB), wal_before); + assert_eq!(policy.budget.reserved(), 0); + + // The same write is kept when it fits. + let fits = run( + &a, + query("WITH c AS (SELECT 'small' AS v) INSERT INTO t (a) SELECT v FROM c"), + GrowthPolicy::Budget(1 << 20), + ) + .await; + assert!(fits.is_ok(), "{fits:?}"); + assert_eq!(count(&a, "t").await, 1); +} + +#[tokio::test] +async fn the_size_cap_and_owner_budget_are_enforced_by_sqlite_and_shrinking_writes_pass() { + let dir = tempfile::tempdir().unwrap(); + let free = FakeFree::new(64 << 20); + let policy = DurabilityPolicy { + max_database_bytes: 64 * 4096, + min_free_bytes: 1, + free_space: free.clone(), + ..Default::default() + }; + let a = actor(dir.path(), DB, policy); + ok(&a, "CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)").await; + + // Per-database cap: pages grow until the 64th, then the write is refused + // whole, whatever its size. + let mut stored = 0; + loop { + let result = run( + &a, + exec("INSERT INTO t (v) VALUES (zeroblob(30000))"), + GrowthPolicy::Allow, + ) + .await; + match result { + Ok(_) => stored += 1, + Err(SqlError::DatabaseSizeLimit) => break, + Err(other) => panic!("{other:?}"), + } + assert!(stored < 64, "the cap never bit"); + } + assert!(stored >= 5, "stored only {stored}"); + assert_eq!(count(&a, "t").await, stored); + let size = a.export().await.unwrap().len(); + assert!(size <= 64 * 4096, "database grew to {size}"); + + // At the cap, a write that frees pages still passes. + ok(&a, "DELETE FROM t WHERE id <= 2").await; + assert_eq!(count(&a, "t").await, stored - 2); + + // Owner budget: one page left means a write needing several is refused, + // a small one that fits in free pages is not. + let refused = run( + &a, + exec("INSERT INTO t (v) VALUES (zeroblob(100000))"), + GrowthPolicy::Budget(4096), + ) + .await; + assert!( + matches!(refused, Err(SqlError::StorageWouldGrow)), + "{refused:?}" + ); + let small = run( + &a, + exec("INSERT INTO t (v) VALUES (zeroblob(100))"), + GrowthPolicy::Budget(4096), + ) + .await; + assert!(small.is_ok(), "{small:?}"); + // A delete under a budget never needs a page, and it deletes: the rows + // above id 3 go, those up to it stay. + assert!(count(&a, "t").await > 1); + let shrink = run( + &a, + exec("DELETE FROM t WHERE id > 3"), + GrowthPolicy::Budget(1), + ) + .await; + assert!(shrink.is_ok(), "{shrink:?}"); + assert_eq!(count(&a, "t").await, 1); +} + +/// The shipped defaults are off: no cap, no headroom check, no reservation +/// (the limits are Sam's decision, card `sql-storage-report-read`). `synchronous` +/// is the durability fix, not a limit, and defaults to FULL. +#[tokio::test] +async fn the_limits_are_off_by_default_but_the_same_policy_enforces_when_set() { + let defaults = DurabilityPolicy::default(); + assert_eq!(defaults.max_database_bytes, 0); + assert_eq!(defaults.min_free_bytes, 0); + assert_eq!(defaults.synchronous, Synchronous::Full); + + let dir = tempfile::tempdir().unwrap(); + // A disk that reports no free space at all, and a blob bigger than any cap + // below: with the defaults nothing refuses and nothing is reserved. + let free = FakeFree::new(0); + let off = DurabilityPolicy { + free_space: free.clone(), + ..Default::default() + }; + let a = actor(dir.path(), "off", off.clone()); + ok(&a, "CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)").await; + ok(&a, "INSERT INTO t (v) VALUES (zeroblob(3000000))").await; + assert_eq!(count(&a, "t").await, 1); + assert_eq!(off.budget.reserved(), 0, "nothing is reserved when off"); + + // The same writes under explicit limits are refused. + let on = DurabilityPolicy { + max_database_bytes: 1 << 20, + min_free_bytes: 1, + free_space: free.clone(), + ..Default::default() + }; + let b = actor(dir.path(), "on", on.clone()); + free.set(1 << 30); + ok(&b, "CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)").await; + let capped = run( + &b, + exec("INSERT INTO t (v) VALUES (zeroblob(3000000))"), + GrowthPolicy::Allow, + ) + .await; + assert!( + matches!(capped, Err(SqlError::DatabaseSizeLimit)), + "{capped:?}" + ); + free.set(0); + let no_room = run( + &b, + exec("INSERT INTO t (v) VALUES (zeroblob(10))"), + GrowthPolicy::Allow, + ) + .await; + assert!( + matches!(no_room, Err(SqlError::DiskHeadroom)), + "{no_room:?}" + ); +} + +#[tokio::test] +async fn disk_headroom_exhaustion_refuses_writes_and_recovers_without_residue() { + let dir = tempfile::tempdir().unwrap(); + let free = FakeFree::new(64 << 20); + let min_free = 8 << 20; + let a = actor(dir.path(), DB, policy_with(&free, min_free)); + ok(&a, "CREATE TABLE t (v TEXT)").await; + + // The floor alone refuses: free space below it, whatever the write. + free.set(min_free - 1); + let refused = run(&a, exec("INSERT INTO t VALUES ('x')"), GrowthPolicy::Allow).await; + assert!( + matches!(refused, Err(SqlError::DiskHeadroom)), + "{refused:?}" + ); + // Reads and writes that touch nothing still answer. + assert_eq!(count(&a, "t").await, 0); + + free.set(64 << 20); + ok(&a, "INSERT INTO t VALUES ('x')").await; + assert_eq!(count(&a, "t").await, 1); +} + +#[tokio::test] +async fn concurrent_databases_share_one_reservation_budget() { + let dir = tempfile::tempdir().unwrap(); + let free = FakeFree::new(64 << 20); + let policy = policy_with(&free, 1); + let a = actor(dir.path(), "a", policy.clone()); + let b = actor(dir.path(), "b", policy.clone()); + ok(&a, "CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)").await; + ok(&b, "CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)").await; + + // `a` commits a large write. Its checkpoint growth stays held until a + // checkpoint runs, not only for the length of the request. + ok(&a, "INSERT INTO t (v) VALUES (zeroblob(400000))").await; + let held = policy.budget.reserved(); + assert!(held >= 390_000, "held {held}"); + assert_eq!(policy.budget.reserved_excluding("space/a"), 0); + // A later, small commit of `a` does not forget the growth of the earlier + // one: what is held follows the file, not the request. + ok(&a, "INSERT INTO t (v) VALUES (zeroblob(10))").await; + let held = policy.budget.reserved(); + assert!(held >= 390_000, "held {held}"); + + // 300 KB free: `b`'s small write fits alone, not next to `a`'s 400 KB. + free.set(300_000); + let refused = run( + &b, + exec("INSERT INTO t (v) VALUES (zeroblob(100))"), + GrowthPolicy::Allow, + ) + .await; + assert!( + matches!(refused, Err(SqlError::DiskHeadroom)), + "{refused:?}" + ); + assert_eq!(count(&b, "t").await, 0); + assert_eq!(policy.budget.reserved_excluding("space/a"), 0); + + // `a`'s checkpoint puts that growth on the disk: nothing of it is held. + a.checkpoint().await.unwrap(); + assert_eq!(policy.budget.reserved(), 0); + ok(&b, "INSERT INTO t (v) VALUES (zeroblob(100))").await; + assert_eq!(count(&b, "t").await, 1); + // A write that adds no page leaves nothing held, refused or done. + assert_eq!(policy.budget.reserved(), 0); + + // The reservation also ends with the database's connection. + free.set(64 << 20); + ok(&a, "INSERT INTO t (v) VALUES (zeroblob(400000))").await; + assert!(policy.budget.reserved() >= 390_000); + a.shutdown().await.unwrap(); + assert_eq!(policy.budget.reserved(), 0); +} + +#[tokio::test] +async fn a_database_over_the_cap_still_runs_writes_that_do_not_grow_it() { + let dir = tempfile::tempdir().unwrap(); + let free = FakeFree::new(64 << 20); + let big = actor(dir.path(), DB, policy_with(&free, 1)); + ok(&big, "CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)").await; + for _ in 0..10 { + ok(&big, "INSERT INTO t (v) VALUES (zeroblob(30000))").await; + } + big.shutdown().await.unwrap(); + // The cap is now below the file's size. + let small_cap = DurabilityPolicy { + max_database_bytes: 4096, + ..policy_with(&free, 1) + }; + let a = actor(dir.path(), DB, small_cap); + ok(&a, "UPDATE t SET v = zeroblob(10) WHERE id = 1").await; + ok(&a, "DELETE FROM t WHERE id = 2").await; + let grow = run( + &a, + exec("INSERT INTO t (v) VALUES (zeroblob(300000))"), + GrowthPolicy::Allow, + ) + .await; + assert!(matches!(grow, Err(SqlError::DatabaseSizeLimit)), "{grow:?}"); +} + +/// A `SQLITE_FULL` that comes from a full disk is reported as the disk, in the +/// default configuration too (`min_free_bytes` = 0): never as the owner's +/// quota or the database's size limit. The free-space reading is the fake one; +/// a real ENOSPC cannot be provoked here. +#[tokio::test] +async fn a_full_disk_is_not_reported_as_the_owners_quota_or_the_size_limit() { + let dir = tempfile::tempdir().unwrap(); + let free = FakeFree::new(64 << 20); + type Is = fn(&SqlError) -> bool; + let cases: [(&str, u64, GrowthPolicy, Is); 2] = [ + // One page of the owner's budget: the write is over it. + ("budget", 0, GrowthPolicy::Budget(4096), |error| { + matches!(error, SqlError::StorageWouldGrow) + }), + ("cap", 64 * 4096, GrowthPolicy::Allow, |error| { + matches!(error, SqlError::DatabaseSizeLimit) + }), + ]; + for (name, max_database_bytes, growth, is_the_limit) in cases { + let policy = DurabilityPolicy { + max_database_bytes, + free_space: free.clone(), + ..Default::default() + }; + assert_eq!(policy.min_free_bytes, 0); + let a = actor(dir.path(), name, policy.clone()); + ok(&a, "CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)").await; + let SqlValue::Integer(page_size) = rows(&a, "PRAGMA page_size").await[0][0] else { + panic!("page_size"); + }; + let floor = 16 * page_size as u64; + let too_big = "INSERT INTO t (v) VALUES (zeroblob(500000))"; + + // Room on the disk: the limit that fired is the one reported. + free.set(64 << 20); + let refused = run(&a, exec(too_big), growth).await.unwrap_err(); + assert!(is_the_limit(&refused), "{name}: {refused:?}"); + + // No room (below a few pages): the disk, for the same write. + for left in [0, floor - 1] { + free.set(left); + let refused = run(&a, exec(too_big), growth).await; + assert!( + matches!(refused, Err(SqlError::DiskHeadroom)), + "{name}, {left} bytes free: {refused:?}" + ); + } + // Over the floor it is a limit again. + free.set(floor); + let refused = run(&a, exec(too_big), growth).await.unwrap_err(); + assert!(is_the_limit(&refused), "{name}: {refused:?}"); + assert_eq!(count(&a, "t").await, 0); + assert_eq!(policy.budget.reserved(), 0); + } +} + +/// A request that wrote nothing needs no room: with the disk below its floor, +/// a wrapped request that only reads answers (the pre-commit gate and the +/// progress handler both let it through), a request that writes does not. +/// (`batch` runs its statements without reading their rows, so the reads here +/// are the ones a wrapped request can carry: `schema` entries of an `execute`, +/// a named statement, a `PRAGMA` report.) +#[tokio::test] +async fn reads_in_a_wrapped_request_are_neither_gated_nor_reserved() { + const SCAN: &str = "SELECT 1 FROM seed s1, seed s2 WHERE s1.x + s2.x < 0"; + let dir = tempfile::tempdir().unwrap(); + let free = FakeFree::new(64 << 20); + let min_free = 8 << 20; + let policy = policy_with(&free, min_free); + let a = actor(dir.path(), DB, policy.clone()); + ok(&a, "CREATE TABLE t (v TEXT)").await; + seed(&a).await; + let checkpoint_before = a.checkpoint().await.unwrap(); + let reads_then = |sql: &str| SqlRequest::Execute { + // The scan crosses several progress-handler intervals. A DROP of a + // table that is not there takes no write transaction. + schema: Some(vec![SCAN.into(), "SELECT count(*) FROM t".into()]), + sql: sql.into(), + params: vec![], + }; + let named = Some(tinycloud_core::sql::SqlCaveats { + statements: Some(vec![tinycloud_core::sql::caveats::PreparedStatement { + name: "scan".into(), + sql: SCAN.into(), + }]), + ..Default::default() + }); + + free.set(min_free - 1); + for growth in [GrowthPolicy::Allow, GrowthPolicy::Budget(1 << 20)] { + let reads = run(&a, reads_then("DROP TABLE IF EXISTS not_there"), growth).await; + assert!(reads.is_ok(), "{reads:?}"); + } + let named_read = a + .execute( + SqlRequest::ExecuteStatement { + name: "scan".into(), + params: vec![], + }, + named, + "tinycloud.sql/admin".into(), + GrowthPolicy::Allow, + ) + .await; + assert!(named_read.is_ok(), "{named_read:?}"); + // A report of the schema is wrapped (it can carry a PRAGMA) and reads. + assert!(!rows(&a, "PRAGMA table_info(t)").await.is_empty()); + assert_eq!(policy.budget.reserved(), 0); + + // A write is refused on the same disk: after the reads, before them, or + // in the middle of a long scan. + for request in [ + reads_then("INSERT INTO t VALUES ('x')"), + SqlRequest::Execute { + schema: Some(vec!["INSERT INTO t VALUES ('x')".into(), SCAN.into()]), + sql: "DROP TABLE IF EXISTS not_there".into(), + params: vec![], + }, + exec("INSERT INTO t VALUES ('x')"), + ] { + let refused = run(&a, request, GrowthPolicy::Allow).await; + assert!( + matches!(refused, Err(SqlError::DiskHeadroom)), + "{refused:?}" + ); + } + assert_eq!(count(&a, "t").await, 0); + assert_eq!(a.checkpoint().await.unwrap(), checkpoint_before); + assert_eq!(policy.budget.reserved(), 0); +} + +/// A request that carries a `PRAGMA` (admin only) is limited, gated and +/// reserved like any write: the owner's budget, the size cap and the disk +/// gate refuse it, and a refusal leaves nothing, the header value included. +#[tokio::test] +async fn a_request_with_a_pragma_is_bounded_like_any_write() { + let dir = tempfile::tempdir().unwrap(); + let free = FakeFree::new(64 << 20); + let policy = DurabilityPolicy { + // 32 pages of 4 KiB. + max_database_bytes: 128 * 1024, + min_free_bytes: 1, + free_space: free.clone(), + ..Default::default() + }; + let a = actor(dir.path(), DB, policy.clone()); + for statement in AMPLIFIED_SCHEMA { + ok(&a, statement).await; + } + let checkpoint_before = a.checkpoint().await.unwrap(); + let big = "INSERT INTO t (a) VALUES (hex(zeroblob(200000)))"; + let small = "INSERT INTO t (a) VALUES (hex(zeroblob(2000)))"; + let with_pragma = |statement: &str| batch(&["PRAGMA user_version = 3", statement]); + + let over_cap = run(&a, with_pragma(big), GrowthPolicy::Allow).await; + assert!( + matches!(over_cap, Err(SqlError::DatabaseSizeLimit)), + "{over_cap:?}" + ); + let over_budget = run(&a, with_pragma(big), GrowthPolicy::Budget(8192)).await; + assert!( + matches!(over_budget, Err(SqlError::StorageWouldGrow)), + "{over_budget:?}" + ); + free.set(1 << 10); + let no_room = run(&a, with_pragma(small), GrowthPolicy::Allow).await; + assert!( + matches!(no_room, Err(SqlError::DiskHeadroom)), + "{no_room:?}" + ); + // A `query` that sets the header value is a write to the gate as well. + let query_no_room = run( + &a, + SqlRequest::Query { + sql: "PRAGMA user_version = 5".into(), + params: vec![], + max_rows: None, + max_bytes: None, + }, + GrowthPolicy::Allow, + ) + .await; + assert!( + matches!(query_no_room, Err(SqlError::DiskHeadroom)), + "{query_no_room:?}" + ); + free.set(64 << 20); + + assert_eq!(count(&a, "t").await, 0); + assert_eq!( + rows(&a, "PRAGMA user_version").await[0][0], + SqlValue::Integer(0) + ); + assert_eq!(a.checkpoint().await.unwrap(), checkpoint_before); + assert_eq!(policy.budget.reserved(), 0); + + // The same request is kept when it fits. + run(&a, with_pragma(small), GrowthPolicy::Budget(1 << 20)) + .await + .unwrap(); + assert_eq!(count(&a, "t").await, 1); + assert_eq!( + rows(&a, "PRAGMA user_version").await[0][0], + SqlValue::Integer(3) + ); +} + +/// The size limit is enforced by SQLite while a statement runs +/// (`max_page_count`), not only by the check before COMMIT: a statement far +/// bigger than the cap stops at the cap, so the WAL it leaves behind is +/// bounded by the cap and not by the statement. +#[tokio::test] +async fn the_size_cap_stops_a_statement_while_it_runs() { + let dir = tempfile::tempdir().unwrap(); + let free = FakeFree::new(64 << 20); + let cap = 128 * 1024; + let policy = DurabilityPolicy { + max_database_bytes: cap, + min_free_bytes: 1, + free_space: free.clone(), + ..Default::default() + }; + let a = actor(dir.path(), DB, policy); + ok(&a, "CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)").await; + let wal_before = wal_len(dir.path(), DB); + + // ~10 MB of text, eighty times the cap. With the check at COMMIT alone, + // the transaction would spill all of it to the WAL first. + let result = run( + &a, + exec("INSERT INTO t (v) VALUES (hex(zeroblob(5000000)))"), + GrowthPolicy::Allow, + ) + .await; + assert!( + matches!(result, Err(SqlError::DatabaseSizeLimit)), + "{result:?}" + ); + let spilled = wal_len(dir.path(), DB) - wal_before; + assert!( + spilled <= 2 * cap, + "the failed statement left {spilled} bytes in the WAL" + ); + assert_eq!(count(&a, "t").await, 0); +} + +#[test] +fn directories_created_for_a_database_are_made_durable() { + let dir = tempfile::tempdir().unwrap(); + let nested = dir.path().join("sql").join("space").join("deeper"); + create_dir_all_durable(&nested).unwrap(); + assert!(nested.is_dir()); + // Idempotent on an existing path. + create_dir_all_durable(&nested).unwrap(); +} + +/// The bound the gate relies on: whatever a transaction dirtied, COMMIT +/// appends no more to the WAL than one frame per page left in the page cache. +#[test] +fn the_unspilled_wal_bound_covers_what_commit_appends() { + use tinycloud_core::sql::durability::{cache_used_bytes, unspilled_wal_bound}; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("bound.db"); + let conn = open_connection(&StorageMode::File(path.clone()), Synchronous::Full).unwrap(); + let wal = path.with_extension("db-wal"); + let wal_len = || std::fs::metadata(&wal).map(|m| m.len()).unwrap_or(0); + let page_size: u64 = conn + .query_row("PRAGMA page_size", [], |r| r.get(0)) + .unwrap(); + conn.execute_batch( + "CREATE TABLE t (id INTEGER PRIMARY KEY, a, b, c, d); \ + CREATE INDEX ta ON t (a); CREATE INDEX tb ON t (b); \ + CREATE INDEX tc ON t (c); CREATE INDEX td ON t (d); \ + CREATE TABLE seed (x); INSERT INTO seed VALUES (1);", + ) + .unwrap(); + for _ in 0..9 { + conn.execute_batch("INSERT INTO seed SELECT x FROM seed") + .unwrap(); + } + let mut observed_spill = false; + for rows in [1, 50, 1_000, 10_000, 100_000] { + conn.execute_batch("SAVEPOINT s").unwrap(); + let at_savepoint = wal_len(); + conn.execute( + "INSERT INTO t (a, b, c, d) SELECT hex(randomblob(10)), hex(randomblob(10)), \ + hex(randomblob(10)), hex(randomblob(10)) FROM seed s1, seed s2 LIMIT ?1", + [rows], + ) + .unwrap(); + // Rewriting existing pages counts too. + conn.execute("UPDATE t SET a = hex(randomblob(10)) WHERE id % 2 = 0", []) + .unwrap(); + let before = wal_len(); + let bound = unspilled_wal_bound(cache_used_bytes(&conn), page_size); + observed_spill |= before > at_savepoint; + conn.execute_batch("RELEASE s").unwrap(); + let appended = wal_len() - before; + assert!( + appended <= bound, + "{rows} rows: COMMIT appended {appended} bytes, bound was {bound}" + ); + } + assert!( + observed_spill, + "the cache never spilled; the test proved little" + ); +} + +/// Indicative only (not a benchmark gate): p50/p99 of a small committed write +/// with `synchronous` NORMAL and FULL. Run with +/// `TC1491_LATENCY_DIR= cargo test -p tinycloud-core +/// --test sql_local_durability latency -- --ignored --nocapture`. +#[test] +#[ignore] +fn synchronous_latency_indicative() { + let base = std::env::var("TC1491_LATENCY_DIR").expect("TC1491_LATENCY_DIR"); + for round in 0..2 { + for setting in [Synchronous::Normal, Synchronous::Full] { + let dir = tempfile::tempdir_in(&base).unwrap(); + let conn = + open_connection(&StorageMode::File(dir.path().join("lat.db")), setting).unwrap(); + conn.execute_batch("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)") + .unwrap(); + let mut micros = Vec::with_capacity(1000); + for i in 0..1000 { + let start = std::time::Instant::now(); + conn.execute("INSERT INTO t (v) VALUES (?1)", [format!("value-{i}")]) + .unwrap(); + micros.push(start.elapsed().as_micros() as u64); + } + micros.sort_unstable(); + println!( + "round {round} {setting:?}: p50 {} us, p99 {} us, max {} us", + micros[499], micros[989], micros[999] + ); + } + } +} + +// ---- kill -9 --------------------------------------------------------------- + +/// Where the child process dies. +#[derive(Clone, Copy, Debug)] +enum Kill { + BeforeCommit, + AfterCommit, + BeforeResponse, +} + +impl Kill { + fn name(self) -> &'static str { + match self { + Kill::BeforeCommit => "before_commit", + Kill::AfterCommit => "after_commit", + Kill::BeforeResponse => "before_response", + } + } +} + +const CHILD_ENV: &str = "TC1491_KILL_CHILD"; +/// Request number (1-based, counted over commits) at which the child dies. +const KILL_AT: u64 = 6; + +fn sigkill_self() -> ! { + // SAFETY: plain syscall on our own pid. + unsafe { libc::kill(libc::getpid(), libc::SIGKILL) }; + std::process::abort() +} + +/// Entry point of the child process; a no-op test in the normal run. +#[test] +fn kill_child_entry() { + let Ok(spec) = std::env::var(CHILD_ENV) else { + return; + }; + let mut parts = spec.split('|'); + let dir = PathBuf::from(parts.next().unwrap()); + let kill = parts.next().unwrap().to_string(); + + // Commits are counted when their `after_commit` seam runs (the setup + // CREATE TABLE is the first). The child dies in the sixth. + let commits = Arc::new(AtomicU64::new(0)); + let fresh = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let seam = |f: Box| Some(Arc::new(f) as Arc); + let before_commit = { + let (commits, kill) = (commits.clone(), kill.clone()); + seam(Box::new(move || { + if kill == "before_commit" && commits.load(Ordering::SeqCst) + 1 == KILL_AT { + sigkill_self(); + } + })) + }; + let after_commit = { + let (commits, fresh, kill) = (commits.clone(), fresh.clone(), kill.clone()); + seam(Box::new(move || { + let n = commits.fetch_add(1, Ordering::SeqCst) + 1; + fresh.store(true, Ordering::SeqCst); + if kill == "after_commit" && n == KILL_AT { + sigkill_self(); + } + })) + }; + let before_response = { + let (commits, fresh, kill) = (commits.clone(), fresh.clone(), kill.clone()); + seam(Box::new(move || { + // Only the response that follows a commit; rejected requests and + // reads answer without one. + if fresh.swap(false, Ordering::SeqCst) + && kill == "before_response" + && commits.load(Ordering::SeqCst) == KILL_AT + { + sigkill_self(); + } + })) + }; + let policy = DurabilityPolicy { + seams: CommitSeams { + before_commit, + after_commit, + before_response, + }, + ..Default::default() + }; + + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let a = actor(&dir, DB, policy); + run( + &a, + exec("CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER NOT NULL)"), + GrowthPolicy::Allow, + ) + .await + .unwrap(); + println!("SETUP"); + for i in 1..=20_i64 { + // Every third request is rejected after its first statement ran. + let request = if i % 3 == 0 { + batch(&[ + &format!("INSERT INTO t (id, v) VALUES ({}, {})", 1000 + i, 1000 + i), + "INSERT INTO t (id, v) VALUES (NULL, NULL)", + ]) + } else { + exec(&format!("INSERT INTO t (id, v) VALUES ({i}, {i})")) + }; + let result = run(&a, request, GrowthPolicy::Allow).await; + match result { + Ok(_) => println!("ACK {i}"), + Err(_) => println!("REJECTED {i}"), + } + } + println!("DONE"); + }); +} + +fn run_child(dir: &Path, kill: Kill) -> Vec { + let mut child = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "kill_child_entry", + "--nocapture", + "--test-threads=1", + ]) + .env(CHILD_ENV, format!("{}|{}", dir.display(), kill.name())) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let lines: Vec = BufReader::new(child.stdout.take().unwrap()) + .lines() + .map_while(Result::ok) + .collect(); + let status = child.wait().unwrap(); + use std::os::unix::process::ExitStatusExt; + assert_eq!( + status.signal(), + Some(libc::SIGKILL), + "the child must die by SIGKILL: {status:?} {lines:?}" + ); + lines +} + +#[tokio::test(flavor = "multi_thread")] +async fn kill_minus_nine_at_each_commit_boundary_loses_no_acknowledged_write_and_keeps_no_rejected_one( +) { + for kill in [Kill::BeforeCommit, Kill::AfterCommit, Kill::BeforeResponse] { + let dir = tempfile::tempdir().unwrap(); + let lines = run_child(dir.path(), kill); + assert!( + lines.iter().any(|line| line.ends_with("SETUP")), + "{kill:?} {lines:?}" + ); + let acked: Vec = lines + .iter() + .filter_map(|l| l.strip_prefix("ACK ")) + .map(|n| n.parse().unwrap()) + .collect(); + let rejected: Vec = lines + .iter() + .filter_map(|l| l.strip_prefix("REJECTED ")) + .map(|n| n.parse().unwrap()) + .collect(); + assert!(!lines.contains(&"DONE".to_string()), "{kill:?}"); + assert!( + !rejected.is_empty() && acked.len() >= 3, + "{kill:?} {lines:?}" + ); + + // Reopen what the dead process left. + let a = actor(dir.path(), DB, DurabilityPolicy::default()); + let present: Vec = rows(&a, "SELECT id FROM t ORDER BY id") + .await + .into_iter() + .map(|row| match row[0] { + SqlValue::Integer(n) => n, + ref other => panic!("{other:?}"), + }) + .collect(); + for id in &acked { + assert!(present.contains(id), "{kill:?}: acknowledged {id} is lost"); + } + for id in &rejected { + assert!( + !present.contains(id) && !present.contains(&(1000 + id)), + "{kill:?}: rejected {id} (or its batch prefix) survived" + ); + } + // Whatever else is there is the single request in flight when the + // process died: committed locally, never acknowledged ("unknown"). + let unexplained: Vec<_> = present.iter().filter(|id| !acked.contains(id)).collect(); + match kill { + Kill::BeforeCommit => assert!(unexplained.is_empty(), "{kill:?} {unexplained:?}"), + Kill::AfterCommit | Kill::BeforeResponse => { + assert_eq!(unexplained.len(), 1, "{kill:?} {unexplained:?}") + } + } + } +} diff --git a/tinycloud-node-server/src/config.rs b/tinycloud-node-server/src/config.rs index e2deada2..47f6cbe3 100644 --- a/tinycloud-node-server/src/config.rs +++ b/tinycloud-node-server/src/config.rs @@ -1331,6 +1331,24 @@ pub struct SqlStorageConfig { /// callers that gave up on it, so the database still warms. #[serde(default = "default_sql_hydration_max_ms")] pub hydration_max_ms: u64, + /// `PRAGMA synchronous` of file-backed databases (TC-1491): `full` makes + /// every local COMMIT fsync the WAL before it returns, `normal` fsyncs + /// only at checkpoints. `normal` is the rollback flag. + #[serde(default)] + pub synchronous: tinycloud_core::sql::durability::Synchronous, + /// Per-database size cap (TC-1491). A write that would grow a database past + /// it is rolled back before COMMIT (`507 sql_database_size_limit`). + /// **0 (the default) = no cap.** The limit is Sam's decision (card + /// `sql-storage-report-read`); set it by config once he answers. + #[serde(default = "default_sql_max_database_bytes")] + pub max_database_bytes: ByteUnit, + /// Free disk that must remain after the worst-case growth of every write + /// in flight (TC-1491); a write that would eat into it is rolled back + /// before COMMIT (`507 sql_disk_headroom`). **0 (the default) = the check + /// is off** (no reservation, no free-space reads). The limit is Sam's + /// decision (card `sql-storage-report-read`). + #[serde(default = "default_sql_min_free_bytes")] + pub min_free_bytes: ByteUnit, /// Whether a schema-changing statement (CREATE, ALTER, DROP of tables, /// views and indexes) needs `tinycloud.sql/schema` (TC-1581). `sql/write` /// covers data statements only. `"enforce"` (the default) refuses the @@ -1340,6 +1358,14 @@ pub struct SqlStorageConfig { pub ddl_requires_schema: DdlPolicy, } +fn default_sql_max_database_bytes() -> ByteUnit { + ByteUnit::Byte(0) +} + +fn default_sql_min_free_bytes() -> ByteUnit { + ByteUnit::Byte(0) +} + fn default_sql_memory_threshold() -> ByteUnit { ByteUnit::Mebibyte(10) } @@ -1361,6 +1387,16 @@ fn default_sql_hydration_max_ms() -> u64 { } impl SqlStorageConfig { + /// The local durability policy the SQL actors run under. + pub fn durability(&self) -> tinycloud_core::sql::durability::DurabilityPolicy { + tinycloud_core::sql::durability::DurabilityPolicy { + synchronous: self.synchronous, + max_database_bytes: self.max_database_bytes.as_u64(), + min_free_bytes: self.min_free_bytes.as_u64(), + ..Default::default() + } + } + pub fn wait_bounds(&self) -> tinycloud_core::sql::SqlWaitBounds { tinycloud_core::sql::SqlWaitBounds { lock_wait: Duration::from_millis(self.lock_wait_timeout_ms), @@ -1380,6 +1416,9 @@ impl Default for SqlStorageConfig { hydration_wait_timeout_ms: default_sql_hydration_wait_timeout_ms(), artifact_transfer_timeout_ms: default_sql_artifact_transfer_timeout_ms(), hydration_max_ms: default_sql_hydration_max_ms(), + synchronous: Default::default(), + max_database_bytes: default_sql_max_database_bytes(), + min_free_bytes: default_sql_min_free_bytes(), ddl_requires_schema: DdlPolicy::default(), } } diff --git a/tinycloud-node-server/src/lib.rs b/tinycloud-node-server/src/lib.rs index b0c3982a..5d035bfb 100644 --- a/tinycloud-node-server/src/lib.rs +++ b/tinycloud-node-server/src/lib.rs @@ -861,6 +861,7 @@ pub async fn app_with_control( database_artifact_repository.clone(), ) .with_wait_bounds(tinycloud_config.storage.sql.wait_bounds()) + .with_durability(tinycloud_config.storage.sql.durability()) .with_ddl_policy(tinycloud_config.storage.sql.ddl_requires_schema); let mut policy_v3_runtime = diff --git a/tinycloud-node-server/src/main.rs b/tinycloud-node-server/src/main.rs index ecc58178..361f22a0 100644 --- a/tinycloud-node-server/src/main.rs +++ b/tinycloud-node-server/src/main.rs @@ -272,6 +272,51 @@ mod tests { assert_eq!(loaded.retention.history_days.get("free"), Some(&7)); } + #[test] + fn sql_durability_settings_default_and_take_env_overrides() { + use tinycloud_core::sql::durability::Synchronous; + let _lock = lock_env(); + let names = [ + "TINYCLOUD_STORAGE__SQL__SYNCHRONOUS", + "TINYCLOUD_STORAGE_SQL_SYNCHRONOUS", + "TINYCLOUD_STORAGE__SQL__MAX_DATABASE_BYTES", + "TINYCLOUD_STORAGE__SQL__MIN_FREE_BYTES", + ]; + let _unset: Vec<_> = names.iter().map(|name| EnvVarGuard::unset(name)).collect(); + let extract = || { + build_config_figment() + .extract::() + .expect("config should parse") + .storage + .sql + }; + + let sql = extract(); + assert_eq!(sql.synchronous, Synchronous::Full, "full by default"); + assert_eq!(sql.max_database_bytes.as_u64(), 0, "no cap by default"); + assert_eq!( + sql.min_free_bytes.as_u64(), + 0, + "headroom check off by default" + ); + let policy = sql.durability(); + assert_eq!(policy.synchronous, Synchronous::Full); + assert_eq!(policy.max_database_bytes, 0); + assert_eq!(policy.min_free_bytes, 0); + + let _canonical = EnvVarGuard::set("TINYCLOUD_STORAGE__SQL__SYNCHRONOUS", "normal"); + let _max = EnvVarGuard::set("TINYCLOUD_STORAGE__SQL__MAX_DATABASE_BYTES", "2 GiB"); + let _free = EnvVarGuard::set("TINYCLOUD_STORAGE__SQL__MIN_FREE_BYTES", "1 GiB"); + let sql = extract(); + assert_eq!(sql.synchronous, Synchronous::Normal); + assert_eq!(sql.max_database_bytes.as_u64(), 2 << 30); + assert_eq!(sql.min_free_bytes.as_u64(), 1 << 30); + drop(_canonical); + + let _legacy = EnvVarGuard::set("TINYCLOUD_STORAGE_SQL_SYNCHRONOUS", "normal"); + assert_eq!(extract().synchronous, Synchronous::Normal); + } + #[test] fn canonical_double_underscore_loads_storage_database() { let _lock = lock_env(); diff --git a/tinycloud-node-server/src/routes/mod.rs b/tinycloud-node-server/src/routes/mod.rs index baf996b9..a64668f5 100644 --- a/tinycloud-node-server/src/routes/mod.rs +++ b/tinycloud-node-server/src/routes/mod.rs @@ -2859,13 +2859,16 @@ async fn handle_sql_invoke( // available when content storage is full. let grows_content = publication_command.is_none() && sql_request_is_write(&sql_request, &exec_caveats, ability); - let full_space = if grows_content { - storage_check(space, tinycloud, config, quota_cache) - .await? - .filter(|check| check.remaining().is_none()) + // One read, as before: it now also carries the budget a write may still + // use, so the service can refuse an overshooting write before COMMIT + // instead of one write late (TC-1491). No extra database round trip. + let storage = if grows_content { + storage_check(space, tinycloud, config, quota_cache).await? } else { None }; + let full_space = storage.filter(|check| check.remaining().is_none()); + let remaining_budget = storage.and_then(StorageCheck::remaining); let execute_start = Instant::now(); if publication_command .as_ref() @@ -2889,6 +2892,17 @@ async fn handle_sql_invoke( authority, ) .await + } else if let Some(remaining) = remaining_budget { + sql_service + .execute_with_budget_authorized( + space, + &db_name, + sql_request, + exec_caveats, + authority, + remaining, + ) + .await } else { sql_service .execute_authorized(space, &db_name, sql_request, exec_caveats, authority) @@ -2904,7 +2918,7 @@ async fn handle_sql_invoke( execute_start.elapsed(), ); let response = execute_result.map_err(|e| -> InvokeError { - match (&e, full_space) { + match (&e, storage) { (SqlError::StorageWouldGrow, Some(check)) => check .reject(StorageRejectionKind::QuotaExceeded, space) .into(), @@ -3390,6 +3404,10 @@ pub(crate) fn sql_error_to_status(err: &SqlError) -> Status { SqlError::ResponseTooLarge(..) | SqlError::RowLimitExceeded(_) => Status::new(413), SqlError::QuotaExceeded => Status::new(429), SqlError::StorageWouldGrow => Status::new(402), + // 507, not 413: the SQL client reads 413 as `SQL_RESPONSE_TOO_LARGE` + // and the KV and quota clients as an exhausted storage plan, and + // TC-1012's 413 is the request body cap. Neither is true here. + SqlError::DatabaseSizeLimit | SqlError::DiskHeadroom => Status::new(507), SqlError::InvalidStatement(_) => Status::BadRequest, SqlError::SchemaError(_) => Status::BadRequest, SqlError::ReadOnlyViolation | SqlError::CaveatViolation(_) => Status::Forbidden, @@ -3400,6 +3418,7 @@ pub(crate) fn sql_error_to_status(err: &SqlError) -> Status { tinycloud_core::request_telemetry::note_failure(FailureNote::internal(Service::Sql)); Status::InternalServerError } + SqlError::CommitFailed => Status::InternalServerError, SqlError::Unavailable => Status::ServiceUnavailable, SqlError::OutcomeUnknown => Status::InternalServerError, SqlError::Busy(_) | SqlError::RequestTimeout => Status::ServiceUnavailable, @@ -7830,6 +7849,33 @@ mod tests { // ---- T2: write-class SQL 402 gate --------------------------------- + #[test] + fn tc1491_reservation_refusals_are_507_with_distinct_codes() { + assert_eq!( + sql_error_to_status(&SqlError::DatabaseSizeLimit), + Status::new(507) + ); + assert_eq!( + sql_error_to_status(&SqlError::DiskHeadroom), + Status::new(507) + ); + assert_eq!( + sql_error_to_status(&SqlError::StorageWouldGrow), + Status::new(402) + ); + assert_eq!( + sql_error_to_status(&SqlError::CommitFailed), + Status::InternalServerError + ); + let size = SqlError::DatabaseSizeLimit.to_string(); + let disk = SqlError::DiskHeadroom.to_string(); + assert!(size.starts_with("sql_database_size_limit:"), "{size}"); + assert!(disk.starts_with("sql_disk_headroom:"), "{disk}"); + assert!(SqlError::CommitFailed + .to_string() + .starts_with("sql_commit_failed:")); + } + #[tokio::test] async fn sql_request_is_write_classifies_variants() { use tinycloud_core::sql::types::SqlStatement; @@ -8738,6 +8784,85 @@ mod tests { Ok(()) } + /// TC-1585 + TC-1491 over `/invoke`: a `WITH … INSERT` inside the owner's + /// remaining budget (not a full space) takes the budget path like a plain + /// write: a growth past the budget is refused before COMMIT with 402 and + /// changes nothing, a small one passes. + #[tokio::test] + async fn tc1585_cte_wrapped_write_over_the_remaining_budget_is_402_and_leaves_nothing( + ) -> Result<()> { + use rocket::data::ByteUnit; + use rocket::http::{ContentType, Header, Status}; + use rocket::local::asynchronous::Client; + + let setup = metered_sql_http_setup("tc1585-cte-budget").await?; + // Room for a page or two, not for 64 KiB. + let limit = ByteUnit::Byte(setup.used + 8 * 1024); + let mut nonce = 0u32; + let mut header = |ability: &str| -> Result { + nonce += 1; + sql_invocation_header(&setup, ability, &format!("urn:uuid:tc1585b-{nonce}")) + }; + let big = header("tinycloud.sql/write")?; + let small = header("tinycloud.sql/write")?; + let count = header("tinycloud.sql/read")?; + + let client = Client::tracked(metered_sql_rocket(setup, limit)).await?; + let send = |auth: &String, request: &SqlRequest| { + let client = &client; + let auth = auth.clone(); + let body = serde_json::to_string(request); + async move { + let response = client + .post("/invoke") + .header(Header::new("Authorization", auth)) + .header(ContentType::JSON) + .body(body.expect("request serializes")) + .dispatch() + .await; + let status = response.status(); + (status, response.into_string().await.unwrap_or_default()) + } + }; + let cte_insert = |params: Vec| SqlRequest::Execute { + sql: + "WITH c AS (SELECT ? AS body) INSERT INTO labels (label, val) SELECT body, 1 FROM c" + .to_string(), + params, + schema: None, + }; + + let (status, body) = send( + &big, + &cte_insert(vec![SqlValue::Text("g".repeat(64 * 1024))]), + ) + .await; + assert_eq!(status, Status::new(402), "{body}"); + assert!(body.contains("storage_quota_exceeded"), "{body}"); + + let (status, body) = send(&small, &cte_insert(vec![SqlValue::Text("ok".into())])).await; + assert_eq!(status, Status::Ok, "{body}"); + + // Only the small write is there: the seeded row plus one. + let (status, body) = send( + &count, + &SqlRequest::Query { + sql: "SELECT count(*) FROM labels".to_string(), + params: vec![], + max_rows: None, + max_bytes: None, + }, + ) + .await; + assert_eq!(status, Status::Ok, "{body}"); + assert_eq!( + serde_json::from_str::(&body)?["rows"], + serde_json::json!([[2]]), + "{body}" + ); + Ok(()) + } + /// TC-1585: a CTE-wrapped write reports the table it changes, so the route /// publishes the same write event as for a plain write (the events are /// built from the service's `write_targets`; before the fix these writes diff --git a/tinycloud.toml b/tinycloud.toml index 504fcd2c..1c231a3f 100644 --- a/tinycloud.toml +++ b/tinycloud.toml @@ -26,6 +26,13 @@ cors = true # hydration_wait_timeout_ms = 30000 # artifact_transfer_timeout_ms = 45000 # hydration_max_ms = 600000 # a hydration keeps warming after its callers give up + ## Local commit durability and the checks before COMMIT (TC-1491). + ## synchronous: "full" fsyncs the WAL on every COMMIT; "normal" is the rollback flag. + ## The two limits are off by default (0 = off): they are Sam's decision (card + ## sql-storage-report-read); set them here once he answers. + # synchronous = "full" + # max_database_bytes = 0 # per database; over it: 507 sql_database_size_limit + # min_free_bytes = 0 # disk kept free; short of it: 507 sql_disk_headroom ## Schema-changing SQL (CREATE/ALTER/DROP of tables, views and indexes) needs ## tinycloud.sql/schema (TC-1581); tinycloud.sql/write covers data statements. ## "enforce" (default) refuses the statement, "log" runs it and records it