Make PRAGMA synchronous configurable for a Namespace (#1533)

* Make PRAGMA synchronous configurable for a Namespace

SQLite allows setting synchronous values to any four levels. This
patch provides an option to set the level for a namespace which will
be used for every connection. It defaults to `NORMAL`.

allowed values: https://www.sqlite.org/pragma.html#pragma_synchronous

e.g. to enable:

    curl -X POST http://sqld-admin-endpoint/v1/namespaces/<namespace>/config -H "Content-Type: application/json" -d '{"durability_mode": "strong"}' -v

* Make `durability_mode` field optional in proto

* Set `default` in the enum

Co-authored-by: ad hoc <[email protected]>

---------

Co-authored-by: ad hoc <[email protected]>

Author: avinassh

libsql-replication/proto/metadata.proto

@@ -2,6 +2,13 @@ syntax = "proto3";
 
 package metadata;
 
+enum DurabilityMode {
+    RELAXED = 0;
+    STRONG = 1;
+    EXTRA = 2;
+    OFF = 3;
+}
+
 // Database config used to send db configs over the wire and stored
 // in the meta store.
 message DatabaseConfig {
@@ -12,11 +19,12 @@ message DatabaseConfig {
     // maximum db size (in pages)
     uint64 max_db_pages = 4;
     optional string heartbeat_url = 5;
-    optional string bottomless_db_id = 6; 
+    optional string bottomless_db_id = 6;
     optional string jwt_key = 7;
     optional uint64 txn_timeout_s = 8;
     bool allow_attach = 9;
     optional uint64 max_row_size = 10;
     optional bool shared_schema = 11;
     optional string shared_schema_name = 12;
+    optional DurabilityMode durability_mode = 13;
 }

libsql-replication/src/generated/metadata.rs

@@ -30,4 +30,38 @@ pub struct DatabaseConfig {
     pub shared_schema: ::core::option::Option<bool>,
     #[prost(string, optional, tag = "12")]
     pub shared_schema_name: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(enumeration = "DurabilityMode", optional, tag = "13")]
+    pub durability_mode: ::core::option::Option<i32>,
+}
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum DurabilityMode {
+    Relaxed = 0,
+    Strong = 1,
+    Extra = 2,
+    Off = 3,
+}
+impl DurabilityMode {
+    /// String value of the enum field names used in the ProtoBuf definition.
+    ///
+    /// The values are not transformed in any way and thus are considered stable
+    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+    pub fn as_str_name(&self) -> &'static str {
+        match self {
+            DurabilityMode::Relaxed => "RELAXED",
+            DurabilityMode::Strong => "STRONG",
+            DurabilityMode::Extra => "EXTRA",
+            DurabilityMode::Off => "OFF",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "RELAXED" => Some(Self::Relaxed),
+            "STRONG" => Some(Self::Strong),
+            "EXTRA" => Some(Self::Extra),
+            "OFF" => Some(Self::Off),
+            _ => None,
+        }
+    }
 }

libsql-server/src/connection/config.rs

@@ -1,12 +1,16 @@
 use crate::{namespace::NamespaceName, LIBSQL_PAGE_SIZE};
 use bytesize::mb;
+use rusqlite::types::ToSqlOutput;
+use rusqlite::ToSql;
+use serde::{Deserialize, Serialize};
+use std::fmt::Display;
+use std::str::FromStr;
 use url::Url;
 
+use super::TXN_TIMEOUT;
 use libsql_replication::rpc::metadata;
 use tokio::time::Duration;
 
-use super::TXN_TIMEOUT;
-
 #[derive(Debug, Clone, serde::Deserialize)]
 pub struct DatabaseConfig {
     pub block_reads: bool,
@@ -29,6 +33,8 @@ pub struct DatabaseConfig {
     pub is_shared_schema: bool,
     #[serde(default)]
     pub shared_schema_name: Option<NamespaceName>,
+    #[serde(default)]
+    pub durability_mode: DurabilityMode,
 }
 
 const fn default_max_size() -> u64 {
@@ -54,6 +60,7 @@ impl Default for DatabaseConfig {
             max_row_size: default_max_row_size(),
             is_shared_schema: false,
             shared_schema_name: None,
+            durability_mode: DurabilityMode::default(),
         }
     }
 }
@@ -77,6 +84,10 @@ impl From<&metadata::DatabaseConfig> for DatabaseConfig {
                 .shared_schema_name
                 .clone()
                 .map(NamespaceName::new_unchecked),
+            durability_mode: match value.durability_mode {
+                None => DurabilityMode::default(),
+                Some(m) => DurabilityMode::from(metadata::DurabilityMode::try_from(m)),
+            },
         }
     }
 }
@@ -96,6 +107,80 @@ impl From<&DatabaseConfig> for metadata::DatabaseConfig {
             max_row_size: Some(value.max_row_size),
             shared_schema: Some(value.is_shared_schema),
             shared_schema_name: value.shared_schema_name.as_ref().map(|s| s.to_string()),
+            durability_mode: Some(metadata::DurabilityMode::from(value.durability_mode).into()),
+        }
+    }
+}
+
+/// Durability mode specifies the `PRAGMA SYNCHRONOUS` setting for the connection
+#[derive(PartialEq, Clone, Copy, Debug, Deserialize, Serialize, Default)]
+#[serde(rename_all = "lowercase")]
+pub enum DurabilityMode {
+    Extra,
+    Strong,
+    #[default]
+    Relaxed,
+    Off,
+}
+
+impl ToSql for DurabilityMode {
+    fn to_sql(&self) -> rusqlite::Result<ToSqlOutput<'_>> {
+        match self {
+            DurabilityMode::Extra => Ok(ToSqlOutput::from("extra")),
+            DurabilityMode::Strong => Ok(ToSqlOutput::from("full")),
+            DurabilityMode::Relaxed => Ok(ToSqlOutput::from("normal")),
+            DurabilityMode::Off => Ok(ToSqlOutput::from("off")),
+        }
+    }
+}
+
+impl FromStr for DurabilityMode {
+    type Err = ();
+
+    fn from_str(input: &str) -> Result<DurabilityMode, Self::Err> {
+        match input {
+            "extra" => Ok(DurabilityMode::Extra),
+            "strong" => Ok(DurabilityMode::Strong),
+            "relaxed" => Ok(DurabilityMode::Relaxed),
+            "off" => Ok(DurabilityMode::Off),
+            _ => Err(()),
+        }
+    }
+}
+
+impl Display for DurabilityMode {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let m = match self {
+            DurabilityMode::Extra => "extra",
+            DurabilityMode::Strong => "strong",
+            DurabilityMode::Relaxed => "relaxed",
+            DurabilityMode::Off => "off",
+        };
+        write!(f, "{m}")
+    }
+}
+
+impl From<DurabilityMode> for metadata::DurabilityMode {
+    fn from(value: DurabilityMode) -> Self {
+        match value {
+            DurabilityMode::Relaxed => metadata::DurabilityMode::Relaxed,
+            DurabilityMode::Strong => metadata::DurabilityMode::Strong,
+            DurabilityMode::Extra => metadata::DurabilityMode::Extra,
+            DurabilityMode::Off => metadata::DurabilityMode::Off,
+        }
+    }
+}
+
+impl From<Result<metadata::DurabilityMode, prost::DecodeError>> for DurabilityMode {
+    fn from(value: Result<metadata::DurabilityMode, prost::DecodeError>) -> Self {
+        match value {
+            Ok(mode) => match mode {
+                metadata::DurabilityMode::Relaxed => DurabilityMode::Relaxed,
+                metadata::DurabilityMode::Strong => DurabilityMode::Strong,
+                metadata::DurabilityMode::Extra => DurabilityMode::Extra,
+                metadata::DurabilityMode::Off => DurabilityMode::Off,
+            },
+            Err(_) => DurabilityMode::default(),
         }
     }
 }

libsql-server/src/connection/libsql.rs

@@ -455,6 +455,9 @@ impl<W: Wal> Connection<W> {
 
         let config = config_store.get();
         conn.pragma_update(None, "max_page_count", config.max_db_pages)?;
+        tracing::info!("setting PRAGMA synchronous to {}", config.durability_mode);
+        conn.pragma_update(None, "synchronous", config.durability_mode)?;
+
         conn.set_limit(
             rusqlite::limits::Limit::SQLITE_LIMIT_LENGTH,
             config.max_row_size as i32,

libsql-server/src/http/admin/mod.rs

@@ -22,7 +22,7 @@ use tower_http::trace::DefaultOnResponse;
 use url::Url;
 
 use crate::auth::parse_jwt_keys;
-use crate::connection::config::DatabaseConfig;
+use crate::connection::config::{DatabaseConfig, DurabilityMode};
 use crate::error::{Error, LoadDumpError};
 use crate::hrana;
 use crate::namespace::{DumpStream, NamespaceName, NamespaceStore, RestoreOption};
@@ -197,6 +197,7 @@ async fn handle_get_config<C: Connector>(
         jwt_key: config.jwt_key.clone(),
         allow_attach: config.allow_attach,
         txn_timeout_s: config.txn_timeout.map(|d| d.as_secs() as u64),
+        durability_mode: Some(config.durability_mode),
     };
     Ok(Json(resp))
 }
@@ -244,6 +245,8 @@ struct HttpDatabaseConfig {
     allow_attach: bool,
     #[serde(default)]
     txn_timeout_s: Option<u64>,
+    #[serde(default)]
+    durability_mode: Option<DurabilityMode>,
 }
 
 async fn handle_post_config<C>(
@@ -272,6 +275,9 @@ async fn handle_post_config<C>(
         config.heartbeat_url = Some(Url::parse(&url)?);
     }
     config.jwt_key = req.jwt_key;
+    if let Some(mode) = req.durability_mode {
+        config.durability_mode = mode;
+    }
 
     store.store(config).await?;
 
@@ -295,6 +301,8 @@ struct CreateNamespaceReq {
     shared_schema_name: Option<NamespaceName>,
     #[serde(default)]
     allow_attach: bool,
+    #[serde(default)]
+    durability_mode: Option<DurabilityMode>,
 }
 
 async fn handle_create_namespace<C: Connector>(
@@ -346,6 +354,7 @@ async fn handle_create_namespace<C: Connector>(
     if let Some(max_db_size) = req.max_db_size {
         config.max_db_pages = max_db_size.as_u64() / LIBSQL_PAGE_SIZE;
     }
+    config.durability_mode = req.durability_mode.unwrap_or(DurabilityMode::default());
 
     app_state.namespaces.create(namespace, dump, config).await?;