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

recall_echo/graph/
store.rs
