Add [db] synchronous to choose how write connections sync the WAL

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

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?;

crates/corro-agent/src/agent/reaper.rs

@@ -269,7 +269,7 @@ mod tests {
 
     async fn setup_test_db(path: std::path::PathBuf) -> eyre::Result<SplitPool> {
         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?;

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()

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<Semaphore>,
     cache_size_kib: i64,
+    synchronous: Synchronous,
 
     read: SqlitePool,
     write: SqlitePool,
@@ -681,10 +682,13 @@ impl SplitPool {
         path: P,
         write_sema: Arc<Semaphore>,
         cache_size_kib: i64,
+        synchronous: Synchronous,
     ) -> Result<Self, SplitPoolCreateError> {
         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<Semaphore>,
         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<CrConn> {
         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)
     }

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
+        );
+    }
+}

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());
         {

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<CrConn>;
@@ -366,15 +367,18 @@ static CRSQL_EXT_DIR: Lazy<TempDir> = Lazy::new(|| {
 pub fn rusqlite_to_crsqlite_write(
     conn: rusqlite::Connection,
     cache_size_kib: i64,
+    synchronous: Synchronous,
 ) -> rusqlite::Result<CrConn> {
     let conn = rusqlite_to_crsqlite(conn)?;
     conn.execute_batch(&format!(
         "
         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<dyn std::error::Error>>
+    {
+        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<dyn std::error::Error>> {
         let tmpdir = tempfile::TempDir::new()?;

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"
+```