recall_echo/graph/
store.rs

1//! SurrealDB store — embedded (kv-surrealkv) or server (WebSocket), selected
2//! at runtime via the `[graph] mode` config key (`embedded` | `server`).
3//!
4//! Both engines are compiled in by default and dispatched through
5//! `surrealdb::engine::any`, so switching backends is a config change,
6//! not a rebuild.
7//!
8//! Concurrency: the embedded SurrealKV backend takes a process-exclusive
9//! file lock. Concurrent access from a second process fails with
10//! [`GraphError::Locked`] after a bounded retry. Server mode (or the serve
11//! daemon) is the supported way to share one store between processes.
12
13use std::path::Path;
14use std::time::Duration;
15
16use surrealdb::engine::any::Any;
17use surrealdb::Surreal;
18
19use super::error::GraphError;
20
21pub type Db = Any;
22
23/// How many times to retry opening an embedded store that is locked by
24/// another process, and the base backoff between attempts (doubled each try).
25///
26/// The common case is a daemon that has just been asked to stop and is
27/// releasing the store, which takes single-digit milliseconds: start far
28/// below that and spend the same total budget (~3.8s) on more attempts.
29const LOCK_RETRY_ATTEMPTS: u32 = 8;
30const LOCK_RETRY_BASE: Duration = Duration::from_millis(15);
31
32/// Connection config for server mode.
33#[derive(Clone)]
34pub struct ServerConfig {
35    pub url: String,
36    pub username: String,
37    pub password: String,
38    pub namespace: String,
39    pub database: String,
40}
41
42impl std::fmt::Debug for ServerConfig {
43    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
44        f.debug_struct("ServerConfig")
45            .field("url", &self.url)
46            .field("username", &self.username)
47            .field("password", &"[REDACTED]")
48            .field("namespace", &self.namespace)
49            .field("database", &self.database)
50            .finish()
51    }
52}
53
54/// True if a SurrealDB error indicates the embedded store's process-exclusive
55/// file lock is held by another process.
56fn is_lock_error(err: &surrealdb::Error) -> bool {
57    is_lock_message(&err.to_string().to_lowercase())
58}
59
60fn is_lock_message(msg: &str) -> bool {
61    msg.contains("lock") && (msg.contains("already") || msg.contains("held"))
62}
63
64/// Open (or create) a SurrealDB embedded store at the given path.
65///
66/// Retries briefly if another process holds the store lock, then fails with
67/// [`GraphError::Locked`] carrying an actionable message.
68pub async fn open(path: &Path) -> Result<Surreal<Db>, GraphError> {
69    let surreal_path = path.join("surreal");
70    std::fs::create_dir_all(&surreal_path)?;
71
72    let path_str = surreal_path.to_str().ok_or_else(|| {
73        GraphError::Io(std::io::Error::new(
74            std::io::ErrorKind::InvalidData,
75            "graph store path contains non-UTF8 characters",
76        ))
77    })?;
78
79    let endpoint = format!("surrealkv://{path_str}");
80    let mut attempt: u32 = 0;
81    let db: Surreal<Db> = loop {
82        match surrealdb::engine::any::connect(&endpoint).await {
83            Ok(db) => break db,
84            Err(e) if is_lock_error(&e) && attempt < LOCK_RETRY_ATTEMPTS => {
85                attempt += 1;
86                tokio::time::sleep(LOCK_RETRY_BASE * 2u32.pow(attempt - 1)).await;
87            }
88            Err(e) if is_lock_error(&e) => {
89                return Err(GraphError::Locked(format!(
90                    "graph store at {} is locked by another process. The embedded \
91                     backend allows one process at a time — retried {} times. \
92                     Another recall-echo command (or the serve daemon) is using it; \
93                     wait for it to finish, or use server mode to share the store.",
94                    surreal_path.display(),
95                    LOCK_RETRY_ATTEMPTS
96                )));
97            }
98            Err(e) => return Err(e.into()),
99        }
100    };
101    db.use_ns("recall").use_db("graph").await?;
102
103    Ok(db)
104}
105
106/// Connect to a SurrealDB server (e.g. `ws://localhost:8787`).
107pub async fn connect(config: &ServerConfig) -> Result<Surreal<Db>, GraphError> {
108    let db = surrealdb::engine::any::connect(&config.url).await?;
109    db.signin(surrealdb::opt::auth::Database {
110        namespace: config.namespace.clone(),
111        database: config.database.clone(),
112        username: config.username.clone(),
113        password: config.password.clone(),
114    })
115    .await?;
116    db.use_ns(&config.namespace)
117        .use_db(&config.database)
118        .await?;
119
120    Ok(db)
121}
122
123/// Initialize the graph schema. Idempotent — safe to call on every open.
124pub async fn init_schema(db: &Surreal<Db>) -> Result<(), GraphError> {
125    db.query(
126        r#"
127        DEFINE TABLE IF NOT EXISTS entity SCHEMAFULL;
128        DEFINE FIELD IF NOT EXISTS name         ON entity TYPE string;
129        DEFINE FIELD IF NOT EXISTS entity_type  ON entity TYPE string;
130        DEFINE FIELD IF NOT EXISTS abstract     ON entity TYPE string;
131        DEFINE FIELD IF NOT EXISTS overview     ON entity TYPE string;
132        DEFINE FIELD IF NOT EXISTS content      ON entity TYPE option<string>;
133        DEFINE FIELD IF NOT EXISTS attributes ON entity TYPE option<object> FLEXIBLE;
134        DEFINE FIELD IF NOT EXISTS embedding    ON entity TYPE option<array<float>>;
135        DEFINE FIELD IF NOT EXISTS mutable      ON entity TYPE bool DEFAULT true;
136        DEFINE FIELD IF NOT EXISTS access_count ON entity TYPE int DEFAULT 0;
137        DEFINE FIELD IF NOT EXISTS utility_score    ON entity TYPE float DEFAULT 0.5;
138        DEFINE FIELD IF NOT EXISTS utility_updates  ON entity TYPE int DEFAULT 0;
139        DEFINE FIELD IF NOT EXISTS created_at   ON entity TYPE datetime DEFAULT time::now();
140        DEFINE FIELD IF NOT EXISTS updated_at   ON entity TYPE datetime DEFAULT time::now();
141        DEFINE FIELD IF NOT EXISTS source       ON entity TYPE option<string>;
142
143        DEFINE INDEX IF NOT EXISTS entity_name   ON entity FIELDS name;
144        DEFINE INDEX IF NOT EXISTS entity_type   ON entity FIELDS entity_type;
145        DEFINE INDEX IF NOT EXISTS entity_vector ON entity FIELDS embedding HNSW DIMENSION 384 DIST COSINE;
146
147        -- Pipeline attribute indexes
148        DEFINE INDEX IF NOT EXISTS entity_pipeline_stage  ON entity FIELDS attributes.pipeline_stage;
149        DEFINE INDEX IF NOT EXISTS entity_pipeline_status ON entity FIELDS attributes.pipeline_status;
150
151        DEFINE TABLE IF NOT EXISTS relates_to SCHEMAFULL TYPE RELATION;
152        DEFINE FIELD IF NOT EXISTS rel_type    ON relates_to TYPE string;
153        DEFINE FIELD IF NOT EXISTS description ON relates_to TYPE option<string>;
154        DEFINE FIELD IF NOT EXISTS valid_from  ON relates_to TYPE datetime DEFAULT time::now();
155        DEFINE FIELD IF NOT EXISTS valid_until ON relates_to TYPE option<datetime>;
156        DEFINE FIELD IF NOT EXISTS confidence  ON relates_to TYPE float DEFAULT 1.0;
157        DEFINE FIELD IF NOT EXISTS last_reinforced ON relates_to TYPE option<datetime>;
158        DEFINE FIELD IF NOT EXISTS source      ON relates_to TYPE option<string>;
159
160        DEFINE INDEX IF NOT EXISTS rel_type_idx ON relates_to FIELDS rel_type;
161
162        DEFINE TABLE IF NOT EXISTS episode SCHEMAFULL;
163        DEFINE FIELD IF NOT EXISTS session_id  ON episode TYPE string;
164        DEFINE FIELD IF NOT EXISTS timestamp   ON episode TYPE datetime DEFAULT time::now();
165        DEFINE FIELD IF NOT EXISTS abstract    ON episode TYPE string;
166        DEFINE FIELD IF NOT EXISTS overview    ON episode TYPE option<string>;
167        DEFINE FIELD IF NOT EXISTS content     ON episode TYPE option<string>;
168        DEFINE FIELD IF NOT EXISTS embedding   ON episode TYPE option<array<float>>;
169        DEFINE FIELD IF NOT EXISTS log_number  ON episode TYPE option<int>;
170        DEFINE FIELD IF NOT EXISTS extracted  ON episode TYPE bool DEFAULT false;
171
172        DEFINE INDEX IF NOT EXISTS episode_session ON episode FIELDS session_id;
173        DEFINE INDEX IF NOT EXISTS episode_time    ON episode FIELDS timestamp;
174        DEFINE INDEX IF NOT EXISTS episode_vector  ON episode FIELDS embedding HNSW DIMENSION 384 DIST COSINE;
175
176        DEFINE TABLE IF NOT EXISTS contributed_to SCHEMAFULL TYPE RELATION;
177        DEFINE FIELD IF NOT EXISTS outcome_result ON contributed_to TYPE string;
178        DEFINE FIELD IF NOT EXISTS was_used       ON contributed_to TYPE bool DEFAULT true;
179        DEFINE FIELD IF NOT EXISTS session_id     ON contributed_to TYPE string;
180        DEFINE FIELD IF NOT EXISTS timestamp      ON contributed_to TYPE datetime DEFAULT time::now();
181
182        DEFINE INDEX IF NOT EXISTS ct_session ON contributed_to FIELDS session_id;
183        "#,
184    )
185    .await?
186    .check()?;
187
188    Ok(())
189}
190
191#[cfg(test)]
192mod tests {
193    use super::*;
194
195    #[test]
196    fn lock_message_detected() {
197        assert!(is_lock_message(
198            "database: the database at /x/surreal/lock is already locked by another process"
199        ));
200        assert!(is_lock_message("file lock held by another process"));
201    }
202
203    #[test]
204    fn non_lock_messages_pass_through() {
205        assert!(!is_lock_message("connection refused"));
206        assert!(!is_lock_message("lockstep protocol mismatch")); // 'lock' without already/held
207        assert!(!is_lock_message("table entity already exists"));
208    }
209}
recall_echo/graph/store.rs

recall_echo/graph/
store.rs
