recall_echo/graph/
store.rs

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4
5//! SurrealDB store — embedded (kv-surrealkv) or server (WebSocket), selected
6//! at runtime via the `[graph] mode` config key (`embedded` | `server`).
7//!
8//! Both engines are compiled in by default and dispatched through
9//! `surrealdb::engine::any`, so switching backends is a config change,
10//! not a rebuild.
11//!
12//! Concurrency: the embedded SurrealKV backend takes a process-exclusive
13//! file lock. Concurrent access from a second process fails with
14//! [`GraphError::Locked`] after a bounded retry. Server mode (or the serve
15//! daemon) is the supported way to share one store between processes.
16
17use std::path::Path;
18use std::time::Duration;
19
20use surrealdb::engine::any::Any;
21use surrealdb::Surreal;
22
23use super::confidence::PRIOR_CONCENTRATION;
24use super::error::GraphError;
25
26pub type Db = Any;
27
28/// Schema version this build writes. Bumped by every migration.
29///
30/// - `0` — pre-Phase-1: edges carry a bare `confidence` mean.
31/// - `1` — edges carry persisted Beta evidence (`alpha`, `beta`) and a
32///   `self_reinforcements` coherence counter.
33pub const SCHEMA_VERSION: i64 = 1;
34
35/// Record ID of the singleton row holding graph-wide metadata.
36const META_RECORD: &str = "meta:schema";
37
38/// How many times to retry opening an embedded store that is locked by
39/// another process, and the base backoff between attempts (doubled each try).
40///
41/// The common case is a daemon that has just been asked to stop and is
42/// releasing the store, which takes single-digit milliseconds: start far
43/// below that and spend the same total budget (~3.8s) on more attempts.
44const LOCK_RETRY_ATTEMPTS: u32 = 8;
45const LOCK_RETRY_BASE: Duration = Duration::from_millis(15);
46
47/// Connection config for server mode.
48#[derive(Clone)]
49pub struct ServerConfig {
50    pub url: String,
51    pub username: String,
52    pub password: String,
53    pub namespace: String,
54    pub database: String,
55}
56
57impl std::fmt::Debug for ServerConfig {
58    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
59        f.debug_struct("ServerConfig")
60            .field("url", &self.url)
61            .field("username", &self.username)
62            .field("password", &"[REDACTED]")
63            .field("namespace", &self.namespace)
64            .field("database", &self.database)
65            .finish()
66    }
67}
68
69/// True if a SurrealDB error indicates the embedded store's process-exclusive
70/// file lock is held by another process.
71fn is_lock_error(err: &surrealdb::Error) -> bool {
72    is_lock_message(&err.to_string().to_lowercase())
73}
74
75fn is_lock_message(msg: &str) -> bool {
76    msg.contains("lock") && (msg.contains("already") || msg.contains("held"))
77}
78
79/// Open (or create) a SurrealDB embedded store at the given path.
80///
81/// Retries briefly if another process holds the store lock, then fails with
82/// [`GraphError::Locked`] carrying an actionable message.
83pub async fn open(path: &Path) -> Result<Surreal<Db>, GraphError> {
84    let surreal_path = path.join("surreal");
85    std::fs::create_dir_all(&surreal_path)?;
86
87    let path_str = surreal_path.to_str().ok_or_else(|| {
88        GraphError::Io(std::io::Error::new(
89            std::io::ErrorKind::InvalidData,
90            "graph store path contains non-UTF8 characters",
91        ))
92    })?;
93
94    let endpoint = format!("surrealkv://{path_str}");
95    let mut attempt: u32 = 0;
96    let db: Surreal<Db> = loop {
97        match surrealdb::engine::any::connect(&endpoint).await {
98            Ok(db) => break db,
99            Err(e) if is_lock_error(&e) && attempt < LOCK_RETRY_ATTEMPTS => {
100                attempt += 1;
101                tokio::time::sleep(LOCK_RETRY_BASE * 2u32.pow(attempt - 1)).await;
102            }
103            Err(e) if is_lock_error(&e) => {
104                return Err(GraphError::Locked(format!(
105                    "graph store at {} is locked by another process. The embedded \
106                     backend allows one process at a time — retried {} times. \
107                     Another recall-echo command (or the serve daemon) is using it; \
108                     wait for it to finish, or use server mode to share the store.",
109                    surreal_path.display(),
110                    LOCK_RETRY_ATTEMPTS
111                )));
112            }
113            Err(e) => return Err(e.into()),
114        }
115    };
116    db.use_ns("recall").use_db("graph").await?;
117
118    Ok(db)
119}
120
121/// Accept a server URL with or without a scheme.
122///
123/// Before the runtime-backend change, server mode always used the WebSocket
124/// connector, so configs in the wild carry bare `host:port`. `engine::any`
125/// dispatches on the scheme and rejects those with "Invalid URL", which turns
126/// a working config into a hard failure on upgrade. A schemeless value keeps
127/// meaning what it always meant.
128fn normalize_server_url(url: &str) -> String {
129    const SCHEMES: [&str; 6] = [
130        "ws://",
131        "wss://",
132        "http://",
133        "https://",
134        "surrealkv://",
135        "mem://",
136    ];
137    if SCHEMES.iter().any(|s| url.starts_with(s)) {
138        url.to_string()
139    } else {
140        format!("ws://{url}")
141    }
142}
143
144/// Connect to a SurrealDB server (e.g. `ws://localhost:8787`).
145pub async fn connect(config: &ServerConfig) -> Result<Surreal<Db>, GraphError> {
146    let db = surrealdb::engine::any::connect(normalize_server_url(&config.url)).await?;
147    db.signin(surrealdb::opt::auth::Database {
148        namespace: config.namespace.clone(),
149        database: config.database.clone(),
150        username: config.username.clone(),
151        password: config.password.clone(),
152    })
153    .await?;
154    db.use_ns(&config.namespace)
155        .use_db(&config.database)
156        .await?;
157
158    Ok(db)
159}
160
161/// Initialize the graph schema, then bring the store up to
162/// [`SCHEMA_VERSION`]. Idempotent — safe to call on every open.
163pub async fn init_schema(db: &Surreal<Db>) -> Result<MigrationReport, GraphError> {
164    define_schema(db).await?;
165    migrate(db).await
166}
167
168/// Declare tables, fields and indexes. Every statement is `IF NOT EXISTS`.
169async fn define_schema(db: &Surreal<Db>) -> Result<(), GraphError> {
170    db.query(
171        r#"
172        DEFINE TABLE IF NOT EXISTS entity SCHEMAFULL;
173        DEFINE FIELD IF NOT EXISTS name         ON entity TYPE string;
174        DEFINE FIELD IF NOT EXISTS entity_type  ON entity TYPE string;
175        DEFINE FIELD IF NOT EXISTS abstract     ON entity TYPE string;
176        DEFINE FIELD IF NOT EXISTS overview     ON entity TYPE string;
177        DEFINE FIELD IF NOT EXISTS content      ON entity TYPE option<string>;
178        DEFINE FIELD IF NOT EXISTS attributes ON entity TYPE option<object> FLEXIBLE;
179        DEFINE FIELD IF NOT EXISTS embedding    ON entity TYPE option<array<float>>;
180        DEFINE FIELD IF NOT EXISTS mutable      ON entity TYPE bool DEFAULT true;
181        DEFINE FIELD IF NOT EXISTS access_count ON entity TYPE int DEFAULT 0;
182        DEFINE FIELD IF NOT EXISTS utility_score    ON entity TYPE float DEFAULT 0.5;
183        DEFINE FIELD IF NOT EXISTS utility_updates  ON entity TYPE int DEFAULT 0;
184        DEFINE FIELD IF NOT EXISTS created_at   ON entity TYPE datetime DEFAULT time::now();
185        DEFINE FIELD IF NOT EXISTS updated_at   ON entity TYPE datetime DEFAULT time::now();
186        DEFINE FIELD IF NOT EXISTS source       ON entity TYPE option<string>;
187
188        DEFINE INDEX IF NOT EXISTS entity_name   ON entity FIELDS name;
189        DEFINE INDEX IF NOT EXISTS entity_type   ON entity FIELDS entity_type;
190        DEFINE INDEX IF NOT EXISTS entity_vector ON entity FIELDS embedding HNSW DIMENSION 384 DIST COSINE;
191
192        -- Pipeline attribute indexes
193        DEFINE INDEX IF NOT EXISTS entity_pipeline_stage  ON entity FIELDS attributes.pipeline_stage;
194        DEFINE INDEX IF NOT EXISTS entity_pipeline_status ON entity FIELDS attributes.pipeline_status;
195
196        DEFINE TABLE IF NOT EXISTS relates_to SCHEMAFULL TYPE RELATION;
197        DEFINE FIELD IF NOT EXISTS rel_type    ON relates_to TYPE string;
198        DEFINE FIELD IF NOT EXISTS description ON relates_to TYPE option<string>;
199        DEFINE FIELD IF NOT EXISTS valid_from  ON relates_to TYPE datetime DEFAULT time::now();
200        DEFINE FIELD IF NOT EXISTS valid_until ON relates_to TYPE option<datetime>;
201        DEFINE FIELD IF NOT EXISTS confidence  ON relates_to TYPE float DEFAULT 1.0;
202        -- Persisted Beta evidence. `option` because edges written before
203        -- schema version 1 have none until the backfill reaches them.
204        DEFINE FIELD IF NOT EXISTS alpha       ON relates_to TYPE option<float>;
205        DEFINE FIELD IF NOT EXISTS beta        ON relates_to TYPE option<float>;
206        DEFINE FIELD IF NOT EXISTS self_reinforcements ON relates_to TYPE option<int>;
207        DEFINE FIELD IF NOT EXISTS last_reinforced ON relates_to TYPE option<datetime>;
208        DEFINE FIELD IF NOT EXISTS source      ON relates_to TYPE option<string>;
209
210        DEFINE INDEX IF NOT EXISTS rel_type_idx ON relates_to FIELDS rel_type;
211
212        DEFINE TABLE IF NOT EXISTS episode SCHEMAFULL;
213        DEFINE FIELD IF NOT EXISTS session_id  ON episode TYPE string;
214        DEFINE FIELD IF NOT EXISTS timestamp   ON episode TYPE datetime DEFAULT time::now();
215        DEFINE FIELD IF NOT EXISTS abstract    ON episode TYPE string;
216        DEFINE FIELD IF NOT EXISTS overview    ON episode TYPE option<string>;
217        DEFINE FIELD IF NOT EXISTS content     ON episode TYPE option<string>;
218        DEFINE FIELD IF NOT EXISTS embedding   ON episode TYPE option<array<float>>;
219        DEFINE FIELD IF NOT EXISTS log_number  ON episode TYPE option<int>;
220        DEFINE FIELD IF NOT EXISTS extracted  ON episode TYPE bool DEFAULT false;
221        -- How many times retrieval has returned this episode. Absent on
222        -- episodes written before the field existed; read paths resolve that
223        -- to zero, which is also what it means. No backfill, no version bump.
224        DEFINE FIELD IF NOT EXISTS access_count ON episode TYPE option<int>;
225        -- Authorship class: 'external' | 'user' | 'self'. `option` with no
226        -- default on purpose — an absent value means an episode written
227        -- before provenance existed, and reads resolve that to 'self'. No
228        -- backfill, so no schema version bump: the absent case is already
229        -- the conservative one.
230        DEFINE FIELD IF NOT EXISTS provenance ON episode TYPE option<string>;
231
232        DEFINE INDEX IF NOT EXISTS episode_session ON episode FIELDS session_id;
233        DEFINE INDEX IF NOT EXISTS episode_time    ON episode FIELDS timestamp;
234        DEFINE INDEX IF NOT EXISTS episode_vector  ON episode FIELDS embedding HNSW DIMENSION 384 DIST COSINE;
235
236        DEFINE TABLE IF NOT EXISTS contributed_to SCHEMAFULL TYPE RELATION;
237        DEFINE FIELD IF NOT EXISTS outcome_result ON contributed_to TYPE string;
238        DEFINE FIELD IF NOT EXISTS was_used       ON contributed_to TYPE bool DEFAULT true;
239        DEFINE FIELD IF NOT EXISTS session_id     ON contributed_to TYPE string;
240        DEFINE FIELD IF NOT EXISTS timestamp      ON contributed_to TYPE datetime DEFAULT time::now();
241
242        DEFINE INDEX IF NOT EXISTS ct_session ON contributed_to FIELDS session_id;
243
244        DEFINE TABLE IF NOT EXISTS meta SCHEMAFULL;
245        DEFINE FIELD IF NOT EXISTS schema_version ON meta TYPE int DEFAULT 0;
246        "#,
247    )
248    .await?
249    .check()?;
250
251    Ok(())
252}
253
254/// What one migration pass did. `edges_backfilled` is zero on an already
255/// current store.
256#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
257pub struct MigrationReport {
258    /// Schema version the store was at when the pass started.
259    pub from_version: i64,
260    /// Schema version the store is at now.
261    pub to_version: i64,
262    /// Number of edges that gained evidence counts in this pass.
263    pub edges_backfilled: u64,
264}
265
266impl MigrationReport {
267    /// True when this pass actually moved the store forward.
268    #[must_use]
269    pub fn ran(&self) -> bool {
270        self.from_version < self.to_version
271    }
272}
273
274/// Bring the store up to [`SCHEMA_VERSION`].
275///
276/// Crash-only: the backfill runs *before* the version marker is written, and
277/// only touches edges that still lack evidence (`alpha IS NONE`). An
278/// interrupted pass therefore leaves a store that re-opens, finishes the
279/// remaining edges, and never counts an edge twice.
280async fn migrate(db: &Surreal<Db>) -> Result<MigrationReport, GraphError> {
281    let from_version = read_schema_version(db).await?;
282    if from_version >= SCHEMA_VERSION {
283        return Ok(MigrationReport {
284            from_version,
285            to_version: from_version,
286            edges_backfilled: 0,
287        });
288    }
289
290    let edges_backfilled = backfill_edge_evidence(db).await?;
291    write_schema_version(db, SCHEMA_VERSION).await?;
292
293    Ok(MigrationReport {
294        from_version,
295        to_version: SCHEMA_VERSION,
296        edges_backfilled,
297    })
298}
299
300/// Give every evidence-less edge the Beta counts implied by its stored mean.
301///
302/// `alpha = confidence · C`, `beta = (1 − confidence) · C` with
303/// `C = PRIOR_CONCENTRATION`: the mean is preserved exactly, and the edge
304/// gains the honest low concentration of something never actually counted.
305///
306/// A single re-runnable statement — `WHERE alpha IS NONE` makes re-entry a
307/// no-op for edges that already have evidence.
308async fn backfill_edge_evidence(db: &Surreal<Db>) -> Result<u64, GraphError> {
309    let mut response = db
310        .query(
311            r#"
312            UPDATE relates_to SET
313                alpha = confidence * $concentration,
314                beta = (1 - confidence) * $concentration,
315                self_reinforcements = 0
316            WHERE alpha IS NONE
317            RETURN id
318            "#,
319        )
320        .bind(("concentration", PRIOR_CONCENTRATION))
321        .await?;
322
323    let updated: Vec<serde_json::Value> = super::deserialize_take(&mut response, 0)?;
324    Ok(updated.len() as u64)
325}
326
327/// Read the store's schema version. An absent meta record means version 0 —
328/// a store written before versioning existed.
329async fn read_schema_version(db: &Surreal<Db>) -> Result<i64, GraphError> {
330    let mut response = db
331        .query("SELECT schema_version FROM type::record($id)")
332        .bind(("id", META_RECORD.to_string()))
333        .await?;
334
335    #[derive(serde::Deserialize)]
336    struct VersionRow {
337        schema_version: i64,
338    }
339
340    let rows: Vec<VersionRow> = super::deserialize_take(&mut response, 0)?;
341    Ok(rows.first().map(|r| r.schema_version).unwrap_or(0))
342}
343
344async fn write_schema_version(db: &Surreal<Db>, version: i64) -> Result<(), GraphError> {
345    db.query("UPSERT type::record($id) SET schema_version = $version")
346        .bind(("id", META_RECORD.to_string()))
347        .bind(("version", version))
348        .await?
349        .check()?;
350    Ok(())
351}
352
353#[cfg(test)]
354mod tests {
355    use super::*;
356
357    #[test]
358    fn lock_message_detected() {
359        assert!(is_lock_message(
360            "database: the database at /x/surreal/lock is already locked by another process"
361        ));
362        assert!(is_lock_message("file lock held by another process"));
363    }
364
365    #[test]
366    fn non_lock_messages_pass_through() {
367        assert!(!is_lock_message("connection refused"));
368        assert!(!is_lock_message("lockstep protocol mismatch")); // 'lock' without already/held
369        assert!(!is_lock_message("table entity already exists"));
370    }
371}
372
373#[cfg(test)]
374mod url_compat_tests {
375    use super::normalize_server_url;
376
377    #[test]
378    fn schemeless_urls_keep_meaning_websocket() {
379        assert_eq!(
380            normalize_server_url("127.0.0.1:8787"),
381            "ws://127.0.0.1:8787"
382        );
383        assert_eq!(normalize_server_url("db.local:8000"), "ws://db.local:8000");
384    }
385
386    #[test]
387    fn explicit_schemes_pass_through_untouched() {
388        for url in [
389            "ws://127.0.0.1:8787",
390            "wss://db.example:443",
391            "http://localhost:8000",
392            "surrealkv:///var/lib/store",
393            "mem://",
394        ] {
395            assert_eq!(normalize_server_url(url), url);
396        }
397    }
398}
recall_echo/graph/store.rs

recall_echo/graph/
store.rs
