diff --git a/README.md b/README.md index 5797cad84..ecf488091 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,7 @@ > | Patch | Why | Upstream | > |---|---|---| > | Forget a dropped batch in the seen cache so sync can redeliver it | A batch that fails (e.g. write-permit timeout) stays in `seen`, so its partial versions never fill until restart ([getployz/ployz#3199](https://github.com/getployz/ployz/issues/3199)) | [#577](https://github.com/superfly/corrosion/issues/577), PR [#594](https://github.com/superfly/corrosion/pull/594) | +> | Add `[db] synchronous` to choose how write connections sync the WAL | With the default `NORMAL`, a power loss drops commits Corrosion already broadcast, and peers never send a node its own changes back. `full` syncs each commit before broadcast, in one sync per commit rather than one per page as an inode sync flag does ([getployz/ployz#2925](https://github.com/getployz/ployz/issues/2925)) | not offered yet | > | Build files: `ployz/Dockerfile`, `.github/workflows/ployz-*.yml`, this note | Image publishing and the weekly rebase. The arm64 image sets `JEMALLOC_SYS_WITH_LG_PAGE=14` for 16K-page kernels | not for upstream | # Corrosion diff --git a/crates/corro-agent/src/agent/handlers.rs b/crates/corro-agent/src/agent/handlers.rs index 3afb58805..f6c32856b 100644 --- a/crates/corro-agent/src/agent/handlers.rs +++ b/crates/corro-agent/src/agent/handlers.rs @@ -1710,7 +1710,8 @@ mod tests { println!("temp db: {db_path:?}"); let write_sema = Arc::new(Semaphore::new(1)); - let pool = SplitPool::create(db_path, write_sema.clone(), -1048576).await?; + let pool = + SplitPool::create(db_path, write_sema.clone(), -1048576, Default::default()).await?; { let mut conn = pool.write_priority().await?; diff --git a/crates/corro-agent/src/agent/reaper.rs b/crates/corro-agent/src/agent/reaper.rs index 64a396609..2a64d0aee 100644 --- a/crates/corro-agent/src/agent/reaper.rs +++ b/crates/corro-agent/src/agent/reaper.rs @@ -269,7 +269,7 @@ mod tests { async fn setup_test_db(path: std::path::PathBuf) -> eyre::Result { let write_sema = Arc::new(Semaphore::new(1)); - let pool = SplitPool::create(path, write_sema, -1048576).await?; + let pool = SplitPool::create(path, write_sema, -1048576, Default::default()).await?; // Create test table let conn = pool.write_priority().await?; diff --git a/crates/corro-agent/src/agent/setup.rs b/crates/corro-agent/src/agent/setup.rs index 1a352ade7..0f7933c25 100644 --- a/crates/corro-agent/src/agent/setup.rs +++ b/crates/corro-agent/src/agent/setup.rs @@ -100,7 +100,13 @@ pub async fn setup(conf: Config, tripwire: Tripwire) -> eyre::Result<(Agent, Age let write_sema = Arc::new(Semaphore::new(1)); - let pool = SplitPool::create(&conf.db.path, write_sema.clone(), conf.db.cache_size_kib).await?; + let pool = SplitPool::create( + &conf.db.path, + write_sema.clone(), + conf.db.cache_size_kib, + conf.db.synchronous, + ) + .await?; let clock = Arc::new( uhlc::HLCBuilder::default() diff --git a/crates/corro-types/src/agent.rs b/crates/corro-types/src/agent.rs index 25913fd25..7423aafe8 100644 --- a/crates/corro-types/src/agent.rs +++ b/crates/corro-types/src/agent.rs @@ -39,7 +39,7 @@ use crate::{ }, channel::{bounded, CorroSender}, compress::ZstdDicts, - config::{BroadcastMethod, Config}, + config::{BroadcastMethod, Config, Synchronous}, metrics_tracker::MetricsTracker, pubsub::SubsManager, schema::Schema, @@ -595,6 +595,7 @@ struct SplitPoolInner { path: PathBuf, write_sema: Arc, cache_size_kib: i64, + synchronous: Synchronous, read: SqlitePool, write: SqlitePool, @@ -681,10 +682,13 @@ impl SplitPool { path: P, write_sema: Arc, cache_size_kib: i64, + synchronous: Synchronous, ) -> Result { let rw_pool = sqlite_pool::Config::new(path.as_ref()) .max_size(1) - .create_pool_transform(move |conn| rusqlite_to_crsqlite_write(conn, cache_size_kib))?; + .create_pool_transform(move |conn| { + rusqlite_to_crsqlite_write(conn, cache_size_kib, synchronous) + })?; debug!("built RW pool"); @@ -698,6 +702,7 @@ impl SplitPool { path.as_ref().to_owned(), write_sema, cache_size_kib, + synchronous, ro_pool, rw_pool, )) @@ -707,6 +712,7 @@ impl SplitPool { path: PathBuf, write_sema: Arc, cache_size_kib: i64, + synchronous: Synchronous, read: SqlitePool, write: SqlitePool, ) -> Self { @@ -732,6 +738,7 @@ impl SplitPool { path, write_sema, cache_size_kib, + synchronous, read, write, priority_tx, @@ -781,7 +788,7 @@ impl SplitPool { #[tracing::instrument(skip(self), level = "debug")] pub fn client_dedicated(&self) -> rusqlite::Result { let conn = rusqlite::Connection::open(&self.0.path)?; - let cr_conn = rusqlite_to_crsqlite_write(conn, self.0.cache_size_kib)?; + let cr_conn = rusqlite_to_crsqlite_write(conn, self.0.cache_size_kib, self.0.synchronous)?; trace_heavy_queries(cr_conn.conn())?; Ok(cr_conn) } diff --git a/crates/corro-types/src/config.rs b/crates/corro-types/src/config.rs index 7caaca76c..13d37c5af 100755 --- a/crates/corro-types/src/config.rs +++ b/crates/corro-types/src/config.rs @@ -162,6 +162,29 @@ pub struct DbConfig { /// WARNING: Setting this too low (<100MB) can severely degrade performance. #[serde(default = "default_cache_size_kib")] pub cache_size_kib: i64, + /// SQLite `PRAGMA synchronous` for write connections. Default: normal. + #[serde(default)] + pub synchronous: Synchronous, +} + +/// How write connections sync the WAL. `Full` syncs it at every commit, so a +/// commit survives power loss before it is broadcast; `Normal` syncs only at +/// checkpoints, so the last commits can be lost to power loss. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Synchronous { + #[default] + Normal, + Full, +} + +impl Synchronous { + pub fn pragma(self) -> &'static str { + match self { + Synchronous::Normal => "NORMAL", + Synchronous::Full => "FULL", + } + } } impl DbConfig { @@ -814,6 +837,7 @@ impl ConfigBuilder { schema_paths: self.schema_paths, subscriptions_path: None, cache_size_kib: default_cache_size_kib(), + synchronous: Synchronous::default(), }, api: ApiConfig { bind_addr: self.api_addr, @@ -1045,3 +1069,26 @@ mod tests { assert!(plum_foca::EagerRatios::default().validate().is_ok()); } } + +#[cfg(test)] +mod db_config_tests { + use super::*; + + fn parse(toml: &str) -> DbConfig { + config::Config::builder() + .add_source(config::File::from_str(toml, config::FileFormat::Toml)) + .build() + .unwrap() + .try_deserialize() + .unwrap() + } + + #[test] + fn synchronous_defaults_to_normal_and_reads_full() { + assert_eq!(parse(r#"path = "/db""#).synchronous, Synchronous::Normal); + assert_eq!( + parse("path = \"/db\"\nsynchronous = \"full\"").synchronous, + Synchronous::Full + ); + } +} diff --git a/crates/corro-types/src/pubsub.rs b/crates/corro-types/src/pubsub.rs index d4a51dd2a..96bd31265 100644 --- a/crates/corro-types/src/pubsub.rs +++ b/crates/corro-types/src/pubsub.rs @@ -2589,7 +2589,13 @@ mod tests { let subscriptions_path: Utf8PathBuf = tmpdir.path().join("subs").display().to_string().into(); - let pool = SplitPool::create(db_path, Arc::new(Semaphore::new(1)), -1048576).await?; + let pool = SplitPool::create( + db_path, + Arc::new(Semaphore::new(1)), + -1048576, + Default::default(), + ) + .await?; let clock = Arc::new(uhlc::HLC::default()); { @@ -2709,9 +2715,14 @@ mod tests { let subscriptions_path: Utf8PathBuf = tmpdir.path().join("subs").display().to_string().into(); - let pool = SplitPool::create(&db_path, Arc::new(Semaphore::new(1)), -1048576) - .await - .unwrap(); + let pool = SplitPool::create( + &db_path, + Arc::new(Semaphore::new(1)), + -1048576, + Default::default(), + ) + .await + .unwrap(); let mut conn = pool.write_priority().await.unwrap(); let clock = Arc::new(uhlc::HLC::default()); { diff --git a/crates/corro-types/src/sqlite.rs b/crates/corro-types/src/sqlite.rs index 4d7ee1d51..33cf47ede 100644 --- a/crates/corro-types/src/sqlite.rs +++ b/crates/corro-types/src/sqlite.rs @@ -19,6 +19,7 @@ use thread_local::ThreadLocal; use tracing::{error, info, trace, warn}; use tripwire::Tripwire; +use crate::config::Synchronous; use crate::vtab::unnest::UnnestTab; pub type SqlitePool = sqlite_pool::Pool; @@ -366,6 +367,7 @@ static CRSQL_EXT_DIR: Lazy = Lazy::new(|| { pub fn rusqlite_to_crsqlite_write( conn: rusqlite::Connection, cache_size_kib: i64, + synchronous: Synchronous, ) -> rusqlite::Result { let conn = rusqlite_to_crsqlite(conn)?; conn.execute_batch(&format!( @@ -373,8 +375,10 @@ pub fn rusqlite_to_crsqlite_write( PRAGMA cache_size = {}; PRAGMA temp_store = MEMORY; PRAGMA cache_spill = FALSE; + PRAGMA synchronous = {}; ", - cache_size_kib + cache_size_kib, + synchronous.pragma() ))?; Ok(conn) @@ -565,6 +569,19 @@ mod tests { use super::*; + #[test] + fn write_connections_use_the_configured_synchronous() -> Result<(), Box> + { + let tmpdir = tempfile::TempDir::new()?; + for (synchronous, expected) in [(Synchronous::Normal, 1), (Synchronous::Full, 2)] { + let conn = Connection::open(tmpdir.path().join("test.db"))?; + let conn = rusqlite_to_crsqlite_write(conn, -1048576, synchronous)?; + let actual: i64 = conn.query_row("PRAGMA synchronous", [], |row| row.get(0))?; + assert_eq!(actual, expected, "{synchronous:?}"); + } + Ok(()) + } + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn concurrent_writes() -> Result<(), Box> { let tmpdir = tempfile::TempDir::new()?; diff --git a/doc/config/db.md b/doc/config/db.md index 6d4f52459..a48f36197 100644 --- a/doc/config/db.md +++ b/doc/config/db.md @@ -47,3 +47,14 @@ Larger values can improve write performance noticeably under heavy ingest, at th [db] cache_size_kib = -1048576 # 1 GiB ``` + +#### `db.synchronous` + +SQLite's [`PRAGMA synchronous`](https://www.sqlite.org/pragma.html#pragma_synchronous) for write connections: `"normal"` or `"full"`. + +Defaults to `"normal"`, which syncs the WAL only at checkpoints. A power loss can then drop the last commits, after Corrosion has already broadcast them. `"full"` syncs the WAL at every commit, so a commit is durable before it is broadcast, at the cost of one sync per commit. + +```toml +[db] +synchronous = "full" +```