coven_database/store/
store_reads.rs

1use super::{host_sql_reads::HostSqlReads, SqlReadContext};
2use crate::{DbError, QueryDependencies};
3use coven_foundation::bounded_workers::BoundedWorkers;
4use rusqlite::{Connection, OpenFlags};
5use std::{num::NonZeroUsize, path::Path};
6
7const WORKERS: usize = 4;
8const QUEUED: NonZeroUsize = NonZeroUsize::new(64).expect("positive queue capacity");
9
10/// Application reads over owned read-only connections, with a separate bounded
11/// executor for processing owned results after their transaction ends.
12#[derive(Clone)]
13pub struct StoreReads {
14    connections: BoundedWorkers<Connection>,
15    processing: BoundedWorkers<()>,
16}
17
18impl StoreReads {
19    /// Open application readers after the store's schema has been validated.
20    /// All connections open before any worker or read capability is exposed.
21    pub fn open(path: &Path) -> Result<Self, DbError> {
22        let connections = (0..WORKERS)
23            .map(|_| {
24                let connection = Connection::open_with_flags(
25                    path,
26                    OpenFlags::SQLITE_OPEN_READ_ONLY
27                        | OpenFlags::SQLITE_OPEN_NO_MUTEX
28                        | OpenFlags::SQLITE_OPEN_URI,
29                )?;
30                connection.pragma_update(None, "foreign_keys", "ON")?;
31                Ok(connection)
32            })
33            .collect::<rusqlite::Result<Vec<_>>>()
34            .map_err(DbError::from)?;
35        Ok(Self {
36            connections: BoundedWorkers::start(connections, QUEUED, "coven-read")
37                .map_err(|e| DbError::context("start read workers", e))?,
38            processing: BoundedWorkers::start(vec![(); WORKERS], QUEUED, "coven-read-processing")
39                .map_err(|e| DbError::context("start read processing workers", e))?,
40        })
41    }
42
43    pub async fn read<F, R, E>(&self, read: F) -> Result<Result<R, E>, DbError>
44    where
45        F: for<'connection> FnOnce(SqlReadContext<'connection>) -> Result<R, E> + Send + 'static,
46        R: Send + 'static,
47        E: Send + 'static,
48    {
49        self.connections
50            .call(move |connection| HostSqlReads::new(connection).read(read))
51            .await
52    }
53
54    pub async fn read_tracked<F, R, E>(
55        &self,
56        read: F,
57    ) -> Result<(Result<R, E>, QueryDependencies), DbError>
58    where
59        F: for<'connection> FnOnce(SqlReadContext<'connection>) -> Result<R, E> + Send + 'static,
60        R: Send + 'static,
61        E: Send + 'static,
62    {
63        self.connections
64            .call(move |connection| HostSqlReads::new(connection).read_tracked(read))
65            .await
66    }
67
68    /// Run owned result processing without occupying a read connection.
69    pub async fn process<F, R>(&self, process: F) -> R
70    where
71        F: FnOnce() -> R + Send + 'static,
72        R: Send + 'static,
73    {
74        self.processing.call(move |()| process()).await
75    }
76}
coven_database/store/store_reads.rs

coven_database/store/
store_reads.rs
