feat: support read-only sqlite databases (#16)

Author: losfair

sqlite/backend.rs

@@ -22,6 +22,7 @@ use denokv_proto::VALUE_ENCODING_V8;
 use rand::Rng;
 use rand::RngCore;
 use rusqlite::params;
+use rusqlite::DatabaseName;
 use rusqlite::OptionalExtension;
 use rusqlite::Transaction;
 use uuid::Uuid;
@@ -139,6 +140,7 @@ pub struct SqliteBackend {
   rng: Box<dyn RngCore + Send>,
   pub dequeue_notify: Arc<tokio::sync::Notify>,
   pub messages_running: HashSet<QueueMessageId>,
+  pub readonly: bool,
 }
 
 impl SqliteBackend {
@@ -164,15 +166,20 @@ impl SqliteBackend {
     dequeue_notify: Arc<tokio::sync::Notify>,
     rng: Box<dyn RngCore + Send>,
   ) -> Result<Self, anyhow::Error> {
-    conn.pragma_update(None, "journal_mode", "wal")?;
-
+    let readonly = conn.is_readonly(DatabaseName::Main)?;
     let mut this = Self {
       conn,
       rng,
       dequeue_notify,
       messages_running: HashSet::new(),
+      readonly,
     };
 
+    if readonly {
+      return Ok(this);
+    }
+
+    this.conn.pragma_update(None, "journal_mode", "wal")?;
     this.run_tx(|tx, _| {
       tx.execute(STATEMENT_CREATE_MIGRATION_TABLE, [])?;
 
@@ -246,6 +253,12 @@ impl SqliteBackend {
     &mut self,
     write: AtomicWrite,
   ) -> Result<Option<CommitResult>, anyhow::Error> {
+    if self.readonly {
+      return Err(
+        TypeError("Cannot write to a read-only database".to_string()).into(),
+      );
+    }
+
     let (has_enqueues, commit_result) = self.run_tx(|tx, rng| {
       for check in &write.checks {
         let real_versionstamp = tx

sqlite/lib.rs

@@ -114,14 +114,20 @@ async fn sqlite_thread(
   mut request_rx: tokio::sync::mpsc::Receiver<SqliteRequest>,
 ) {
   let start = utc_now();
-  if let Err(err) = backend.queue_cleanup() {
-    panic!("KV queue cleanup failed: {err}");
+  if !backend.readonly {
+    if let Err(err) = backend.queue_cleanup() {
+      panic!("KV queue cleanup failed: {err}");
+    }
   }
   let mut dequeue_channels: VecDeque<oneshot::Sender<DequeuedMessage>> =
     VecDeque::with_capacity(4);
-  let queue_next_ready = match backend.queue_next_ready() {
-    Ok(queue_next_ready) => queue_next_ready,
-    Err(err) => panic!("KV queue_next_ready failed: {err}"),
+  let queue_next_ready = if backend.readonly {
+    None
+  } else {
+    match backend.queue_next_ready() {
+      Ok(queue_next_ready) => queue_next_ready,
+      Err(err) => panic!("KV queue_next_ready failed: {err}"),
+    }
   };
   let mut queue_dequeue_deadline = compute_deadline_with_max_and_jitter(
     queue_next_ready,
@@ -132,9 +138,13 @@ async fn sqlite_thread(
     start + duration_with_jitter(QUEUE_RUNNING_CLEANUP_INTERVAL);
   let mut queue_keepalive_deadline =
     start + duration_with_jitter(QUEUE_MESSAGE_DEADLINE_UPDATE_INTERVAL);
-  let expiry_next_ready = match backend.collect_expired() {
-    Ok(expiry_next_ready) => expiry_next_ready,
-    Err(err) => panic!("KV collect expired failed: {err}"),
+  let expiry_next_ready = if backend.readonly {
+    None
+  } else {
+    match backend.collect_expired() {
+      Ok(expiry_next_ready) => expiry_next_ready,
+      Err(err) => panic!("KV collect expired failed: {err}"),
+    }
   };
   let mut expiry_deadline = compute_deadline_with_max_and_jitter(
     expiry_next_ready,
@@ -149,12 +159,14 @@ async fn sqlite_thread(
     if !dequeue_channels.is_empty() {
       closest_deadline = closest_deadline.min(queue_dequeue_deadline);
     }
-    let timer = if let Ok(time_to_closest_deadline) =
+    let timer = if backend.readonly {
+      Either::Left(futures::future::pending::<()>())
+    } else if let Ok(time_to_closest_deadline) =
       closest_deadline.signed_duration_since(now).to_std()
     {
-      Either::Left(tokio::time::sleep(time_to_closest_deadline))
+      Either::Right(Either::Left(tokio::time::sleep(time_to_closest_deadline)))
     } else {
-      Either::Right(futures::future::ready(()))
+      Either::Right(Either::Right(futures::future::ready(())))
     };
     select! {
       biased;