diff --git a/CHANGELOG.md b/CHANGELOG.md index b2ac960..1a3e704 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,20 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht ## [Unreleased] +### Added + +- `Drasi.create` and `Drasi.from_config` accept + `wal_provider={"path": ...}`. The built-in redb provider stores one WAL file + per durable transient source in that directory; `kind` defaults to `redb`. +- `add_python_source(..., durability=True)` persists changes before acknowledging + them and replays them after a fresh-process restart. The async and blocking + APIs expose the same option. + +### Changed + +- Volatile Python sources now report that they cannot replay changes. Persistent + queries must use a durable Python source with a configured WAL provider. + ## [0.3.0] - 2026-10-01 Built against drasi-lib 0.9. This is a minor release because two behaviour diff --git a/Cargo.lock b/Cargo.lock index bd08b61..dd3e93e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -897,6 +897,7 @@ dependencies = [ "drasi-lib", "drasi-plugin-sdk", "drasi-state-store-redb", + "drasi-wal-redb", "futures-util", "log", "pyo3", @@ -956,6 +957,23 @@ dependencies = [ "tokio", ] +[[package]] +name = "drasi-wal-redb" +version = "0.2.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0a220606c2dc3f9d6857ed14083a01864bec8c16ad62ddfd46af536ef3f93e2" +dependencies = [ + "async-trait", + "bincode", + "drasi-core", + "drasi-lib", + "log", + "ordered-float", + "redb", + "serde", + "tokio", +] + [[package]] name = "dunce" version = "1.0.5" diff --git a/Cargo.toml b/Cargo.toml index 70929fe..34a28b4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -41,6 +41,7 @@ drasi-core = "=0.5.10" drasi-host-sdk = { version = "=0.11.3", features = ["watcher", "registry"] } drasi-plugin-sdk = "=0.11.3" drasi-state-store-redb = "=0.2.8" +drasi-wal-redb = "=0.2.10" drasi-index-rocksdb = { version = "=0.6.4", optional = true, features = ["plugin-descriptor"] } anyhow = "1" diff --git a/README.md b/README.md index 9045066..bb6aa18 100644 --- a/README.md +++ b/README.md @@ -163,8 +163,12 @@ drasi = await Drasi.create( "app", state_store={"kind": "redb", "path": "state.redb"}, index_store={"kind": "rocksdb", "path": "index"}, + wal_provider={"path": "wal"}, ) +await drasi.start() +await drasi.add_python_source("orders", durability=True) + async def handle(event): await write_somewhere(event) # if this raises, the event is retried diff --git a/docs/api-audit.md b/docs/api-audit.md index 421663e..a06d128 100644 --- a/docs/api-audit.md +++ b/docs/api-audit.md @@ -25,7 +25,7 @@ one genuine advantage — streaming as async iterators — has been carried over | Components | **loaded at runtime** from `ghcr.io/drasi-project` | **linked at compile time**, one PyPI package each | | PyPI packages | 1 | 23 + workspace root | | Engine methods (at audit) | 37 | 27 | -| Errors | 8 classes, 26 codes | 1 class, no codes | +| Errors | 8 classes, 30 codes | 1 class, no codes | | Streaming | none | async iterators | | Python | ≥3.10 (abi3) | ≥3.11 | @@ -185,6 +185,7 @@ At audit: 30 of 48 methods (62%). **Now 48 of 48 (100%).** | Secret store | `secrets`, and the resolver plugins call back through | ✅ | | State store | redb | ✅ | | Index store | RocksDB, behind a Cargo feature | ✅ | +| WAL provider | redb, for durable transient Python sources | ✅ | | Identity | password and token | ✅ | **21 methods exist here that Node does not have**: the eight streaming @@ -198,17 +199,17 @@ iterators (`query_results`, `query_events`, `source_events`, ## 4. Gap analysis vs the `drasi-lib` engine -Comparing against `drasi-lib` 0.8.9 `src/lib_core_ops/`, the engine also offers, -unexposed by *either* prototype: +Comparing against `drasi-lib` 0.8.9 `src/lib_core_ops/`, the engine also offers +the following surface. The WAL-provider row is now closed: -| Engine capability | Notes | -| --- | --- | -| `get_query_config` / `get_query_info` | inspect a running query | -| `get_source_info` / `get_reaction_info` | | -| `snapshot_configuration` / `get_current_config` | round-trip config to disk | -| `subscribe_all_component_events` | | -| Query tuning: `with_middleware`, `with_recovery_policy`, `with_storage_backend`, `with_outbox_capacity`, `dispatch_mode` | `Query` builder options; only `joins` and `language` are exposed today | -| WAL providers | | +| Engine capability | Notes | Status | +| --- | --- | --- | +| `get_query_config` / `get_query_info` | inspect a running query | | +| `get_source_info` / `get_reaction_info` | | | +| `snapshot_configuration` / `get_current_config` | round-trip config to disk | | +| `subscribe_all_component_events` | | | +| Query tuning: `with_middleware`, `with_recovery_policy`, `with_storage_backend`, `with_outbox_capacity`, `dispatch_mode` | `Query` builder options; only `joins` and `language` are exposed today | | +| WAL providers | redb provider plus per-Python-source durability opt-in | ✅ | These are lower priority than the Node parity gaps — they are engine surface neither binding has needed yet — but `snapshot_configuration` pairs naturally diff --git a/python/drasi/_drasi.pyi b/python/drasi/_drasi.pyi index ddc2a28..7c02dec 100644 --- a/python/drasi/_drasi.pyi +++ b/python/drasi/_drasi.pyi @@ -44,6 +44,7 @@ from .types import ( SourceSchema, SourceSubscription, StateStore, + WalProvider, ) # Directory arguments reach Rust as a PathBuf, which accepts anything @@ -103,6 +104,7 @@ class Drasi: secrets: Mapping[str, str] | None = None, state_store: StateStore | None = None, index_store: IndexStore | None = None, + wal_provider: WalProvider | None = None, identity: Identity | None = None, plugins_dir: StrPath | None = None, ) -> Awaitable[Drasi]: ... @@ -164,7 +166,9 @@ class Drasi: def wait_for_query(self, id: str, *, timeout: float = 30.0) -> Awaitable[None]: ... # components defined in Python - def add_python_source(self, id: str, *, auto_start: bool = True) -> Awaitable[None]: ... + def add_python_source( + self, id: str, *, auto_start: bool = True, durability: bool = False + ) -> Awaitable[None]: ... def push_change(self, source_id: str, change: SourceChange) -> Awaitable[None]: ... def add_python_reaction( self, diff --git a/python/drasi/sync.py b/python/drasi/sync.py index 0749deb..be3ca8e 100644 --- a/python/drasi/sync.py +++ b/python/drasi/sync.py @@ -71,6 +71,7 @@ SourceSchema, SourceSubscription, StateStore, + WalProvider, ) StrPath = str | PathLike[str] @@ -168,6 +169,7 @@ def create( secrets: Mapping[str, str] | None = None, state_store: StateStore | None = None, index_store: IndexStore | None = None, + wal_provider: WalProvider | None = None, identity: Identity | None = None, plugins_dir: StrPath | None = None, ) -> Drasi: @@ -178,6 +180,7 @@ def create( secrets=secrets, state_store=state_store, index_store=index_store, + wal_provider=wal_provider, identity=identity, plugins_dir=plugins_dir, ) @@ -291,8 +294,12 @@ def list_queries(self) -> list[tuple[str, str]]: # ------------------------------------------- components defined in Python - def add_python_source(self, id: str, *, auto_start: bool = True) -> None: - _Loop.run(lambda: self._inner.add_python_source(id, auto_start=auto_start)) + def add_python_source( + self, id: str, *, auto_start: bool = True, durability: bool = False + ) -> None: + _Loop.run( + lambda: self._inner.add_python_source(id, auto_start=auto_start, durability=durability) + ) def push_change(self, source_id: str, change: SourceChange) -> None: _Loop.run(lambda: self._inner.push_change(source_id, change)) diff --git a/python/drasi/types.py b/python/drasi/types.py index 802d064..7fc36d6 100644 --- a/python/drasi/types.py +++ b/python/drasi/types.py @@ -67,6 +67,7 @@ "SourceSchema", "SourceSubscription", "StateStore", + "WalProvider", ] QueryLanguage = Literal["cypher", "gql"] @@ -103,6 +104,16 @@ class IndexStore(TypedDict, total=False): direct_io: bool +class _WalProviderRequired(TypedDict): + path: str + + +class WalProvider(_WalProviderRequired, total=False): + """Directory containing one redb WAL file per durable transient source.""" + + kind: Literal["redb"] + + class Identity(TypedDict, total=False): """Credentials handed to plugins that ask for them. @@ -446,6 +457,7 @@ class DrasiConfig(TypedDict, total=False): secrets: Mapping[str, str] state_store: StateStore index_store: IndexStore + wal_provider: WalProvider identity: Identity plugins_dir: str sources: Sequence[SourceConfig] diff --git a/scripts/generate_api_reference.py b/scripts/generate_api_reference.py index 73f6abd..66bd1cb 100644 --- a/scripts/generate_api_reference.py +++ b/scripts/generate_api_reference.py @@ -62,8 +62,14 @@ itself and a `Stream` becomes an ordinary iterator — see [the blocking API](../guides/blocking-api/). -The shapes referenced here (`SourceChange`, `QueryResultEvent`, `StateStore` and the -rest) are declared in `drasi.types`, described in the [typing guide](../guides/typing/). +The shapes referenced here (`SourceChange`, `QueryResultEvent`, `StateStore`, +`WalProvider` and the rest) are declared in `drasi.types`, described in the +[typing guide](../guides/typing/). + +`wal_provider={"path": ...}` configures a directory. The redb provider creates one +WAL file per durable transient source inside it. This is Drasi's local +transient-source write-ahead log; it is not a PostgreSQL replication slot and does +not create, manage or resume one. {{% alert title="Generated from the shipped stubs" color="info" %}} The tables below are generated from `python/drasi/_drasi.pyi`, the same file your type diff --git a/src/components.rs b/src/components.rs index 69e4809..a989f47 100644 --- a/src/components.rs +++ b/src/components.rs @@ -20,14 +20,17 @@ use std::any::Any; use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; +use std::time::Duration; use anyhow::Result; use async_trait::async_trait; use drasi_core::models::SourceChange; -use drasi_lib::channels::{QueryResult, SubscriptionResponse}; +use drasi_lib::channels::{QueryResult, SourceEvent, SourceEventWrapper, SubscriptionResponse}; use drasi_lib::config::SourceSubscriptionSettings; use drasi_lib::context::{ReactionRuntimeContext, SourceRuntimeContext}; +use drasi_lib::wal::{WalProvider, WriteAheadLogConfig}; use drasi_lib::{ BootstrapProvider, ComponentStatus, Reaction, ReactionBase, ReactionBaseParams, ReactionRecoveryPolicy, Source, SourceBase, SourceBaseParams, @@ -35,7 +38,7 @@ use drasi_lib::{ use pyo3::prelude::*; use pyo3_async_runtimes::TaskLocals; use serde_json::Value; -use tokio::sync::Mutex as TokioMutex; +use tokio::sync::{Mutex as TokioMutex, RwLock as TokioRwLock}; use crate::conversions::json_to_py; use crate::streams::{StreamItem, StreamSender}; @@ -46,32 +49,81 @@ const PYTHON_COMPONENT_TYPE: &str = "python"; /// A source that emits the changes pushed into it from Python. pub struct PythonSource { base: SourceBase, - /// Serialises dispatch so concurrent pushes cannot be reordered. - /// - /// `dispatch_source_change` assigns a monotonic sequence with `fetch_add` - /// and only then awaits its way to the subscribers, so two overlapping - /// pushes can be delivered in the opposite order to the sequences they - /// took. The query side treats sequence order as delivery order - - /// `SequenceDedup::should_skip` drops anything at or below the highest - /// sequence already seen - so the lower one is silently discarded. - /// `asyncio.gather` over `push_change` hit this for real, losing roughly - /// one change in twenty at a few percent of runs. + durability: bool, + wal: TokioRwLock>>, + prune_task: TokioRwLock>>, + running: AtomicBool, + /// Serialises persistence and dispatch so concurrent pushes cannot reorder. dispatch: TokioMutex<()>, } impl PythonSource { - pub fn new(id: &str, auto_start: bool) -> Result { + pub fn new(id: &str, auto_start: bool, durability: bool) -> Result { let params = SourceBaseParams::new(id).with_auto_start(auto_start); Ok(Self { base: SourceBase::new(params)?, + durability, + wal: TokioRwLock::new(None), + prune_task: TokioRwLock::new(None), + running: AtomicBool::new(false), dispatch: TokioMutex::new(()), }) } + async fn start_pruner(&self, wal: Arc) { + let base = self.base.clone_shared(); + let source_id = self.base.get_id().to_string(); + let handle = tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(30)); + loop { + interval.tick().await; + let Some(confirmed) = base.compute_confirmed_position().await else { + continue; + }; + if confirmed == 0 { + continue; + } + if let Err(err) = wal.prune_up_to(&source_id, confirmed).await { + log::warn!( + "failed to prune WAL for Python source '{source_id}' through \ + sequence {confirmed}: {err}" + ); + } + } + }); + if let Some(previous) = self.prune_task.write().await.replace(handle) { + previous.abort(); + } + } + /// Emits a change to every subscribed query. pub async fn push(&self, change: SourceChange) -> Result<()> { let _ordered = self.dispatch.lock().await; - self.base.dispatch_source_change(change).await + if !self.running.load(Ordering::Acquire) { + anyhow::bail!( + "Python source '{}' must be running before push_change", + self.base.get_id() + ); + } + if !self.durability { + return self.base.dispatch_source_change(change).await; + } + + let wal = self + .wal + .read() + .await + .clone() + .ok_or_else(|| anyhow::anyhow!("durable Python source WAL is not initialized"))?; + let sequence = wal.append(self.base.get_id(), &change).await?; + let event = SourceEventWrapper::new( + self.base.get_id().to_string(), + SourceEvent::Change(change), + chrono::Utc::now(), + sequence, + ) + .with_source_position(sequence.to_be_bytes().to_vec().into()); + self.base.dispatch_event(event).await } } @@ -93,15 +145,41 @@ impl Source for PythonSource { self.base.get_auto_start() } + fn supports_replay(&self) -> bool { + self.durability + } + async fn start(&self) -> Result<()> { - // There is nothing to spawn: changes arrive from the application rather - // than from a connection this source owns. + let _ordered = self.dispatch.lock().await; + if self.durability { + let context = self + .base + .context() + .await + .ok_or_else(|| anyhow::anyhow!("Python source context is not initialized"))?; + let wal = context.wal_provider.clone().ok_or_else(|| { + anyhow::anyhow!("Durability enabled but no WAL provider configured on DrasiLib") + })?; + wal.register(self.base.get_id(), WriteAheadLogConfig::default()) + .await?; + let head = wal.head_sequence(self.base.get_id()).await?; + self.base.set_next_sequence(head); + *self.wal.write().await = Some(Arc::clone(&wal)); + self.start_pruner(wal).await; + } + self.running.store(true, Ordering::Release); self.base.set_status(ComponentStatus::Running, None).await; Ok(()) } async fn stop(&self) -> Result<()> { - self.base.stop_common().await + let _ordered = self.dispatch.lock().await; + self.running.store(false, Ordering::Release); + if let Some(handle) = self.prune_task.write().await.take() { + handle.abort(); + } + self.base.set_status(ComponentStatus::Stopped, None).await; + Ok(()) } async fn status(&self) -> ComponentStatus { @@ -112,6 +190,38 @@ impl Source for PythonSource { &self, settings: SourceSubscriptionSettings, ) -> Result { + if self.durability { + let wal = + self.wal.read().await.clone().ok_or_else(|| { + anyhow::anyhow!("durable Python source WAL is not initialized") + })?; + let resume_sequence = if let Some(resume_from) = settings.resume_from.as_ref() { + if resume_from.len() < 8 { + anyhow::bail!( + "invalid Python source resume position: expected at least 8 bytes, got {}", + resume_from.len() + ); + } + Some(u64::from_be_bytes( + resume_from[..8] + .try_into() + .expect("resume position length checked"), + )) + } else { + settings.resume_sequence + }; + let replay_sequence = match resume_sequence { + Some(sequence) => Some(sequence), + None if settings.enable_bootstrap && !self.base.has_bootstrap_provider() => Some(0), + None => None, + }; + if let Some(sequence) = replay_sequence { + return self + .base + .subscribe_with_replay(&settings, wal.as_ref(), sequence, PYTHON_COMPONENT_TYPE) + .await; + } + } self.base .subscribe_with_bootstrap(&settings, PYTHON_COMPONENT_TYPE) .await @@ -121,9 +231,31 @@ impl Source for PythonSource { self } + async fn deprovision(&self) -> Result<()> { + if let Some(wal) = self.wal.read().await.clone() { + wal.delete_wal(self.base.get_id()).await?; + } + Ok(()) + } + async fn initialize(&self, context: SourceRuntimeContext) { self.base.initialize(context).await; } + + async fn set_bootstrap_provider(&self, provider: Box) { + self.base.set_bootstrap_provider(provider).await; + } + + async fn remove_position_handle(&self, query_id: &str) { + self.base.remove_position_handle(query_id).await; + } + + async fn set_identity_provider( + &self, + provider: Arc, + ) { + self.base.set_identity_provider(provider).await; + } } /// A cloneable handle to a [`PythonSource`]. @@ -147,10 +279,22 @@ impl Source for SharedSource { self.0.properties() } + fn dispatch_mode(&self) -> drasi_lib::DispatchMode { + self.0.dispatch_mode() + } + fn auto_start(&self) -> bool { self.0.auto_start() } + fn supports_replay(&self) -> bool { + self.0.supports_replay() + } + + fn describe_schema(&self) -> Option { + self.0.describe_schema() + } + async fn start(&self) -> Result<()> { self.0.start().await } @@ -174,9 +318,32 @@ impl Source for SharedSource { self } + async fn deprovision(&self) -> Result<()> { + self.0.deprovision().await + } + async fn initialize(&self, context: SourceRuntimeContext) { self.0.initialize(context).await; } + + async fn set_bootstrap_provider(&self, provider: Box) { + self.0.set_bootstrap_provider(provider).await; + } + + async fn remove_position_handle(&self, query_id: &str) { + self.0.remove_position_handle(query_id).await; + } + + async fn on_subscriptions_complete(&self) { + self.0.on_subscriptions_complete().await; + } + + async fn set_identity_provider( + &self, + provider: Arc, + ) { + self.0.set_identity_provider(provider).await; + } } /// Adapts a `Box` produced by a plugin descriptor. @@ -595,3 +762,61 @@ impl Reaction for PythonReaction { self.base.enqueue_query_result(result).await } } + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + use std::sync::Arc; + use std::time::Duration; + + use drasi_core::models::{ + Element, ElementMetadata, ElementPropertyMap, ElementReference, SourceChange, + }; + use drasi_lib::channels::SourceEvent; + use drasi_lib::config::SourceSubscriptionSettings; + use drasi_lib::Source; + + use super::PythonSource; + + fn change() -> SourceChange { + SourceChange::Insert { + element: Element::Node { + metadata: ElementMetadata { + reference: ElementReference::new("orders", "o1"), + labels: Arc::from(vec![Arc::from("Order")]), + effective_from: 1, + }, + properties: ElementPropertyMap::new(), + }, + } + } + + #[tokio::test(flavor = "current_thread")] + async fn push_dispatches_before_returning() { + let source = PythonSource::new("orders", true, false).expect("source"); + source.start().await.expect("start source"); + let mut subscription = source + .subscribe(SourceSubscriptionSettings { + source_id: "orders".to_string(), + enable_bootstrap: false, + query_id: "q".to_string(), + nodes: HashSet::new(), + relations: HashSet::new(), + resume_from: None, + resume_sequence: None, + request_position_handle: false, + }) + .await + .expect("subscribe"); + + let expected = change(); + source.push(expected.clone()).await.expect("push"); + source.stop().await.expect("stop source"); + + let event = tokio::time::timeout(Duration::from_millis(50), subscription.receiver.recv()) + .await + .expect("push returned before dispatch") + .expect("receive"); + assert_eq!(event.event, SourceEvent::Change(expected)); + } +} diff --git a/src/engine.rs b/src/engine.rs index 35ab94b..f69274b 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -293,24 +293,28 @@ impl Drasi { /// /// The optional stores are all in-memory unless configured: `secrets` seeds /// the store plugins resolve `ConfigValue::Secret` against, `state_store` - /// persists plugin state, `index_store` persists query indexes, and - /// `identity` supplies credentials to plugins that ask for them. + /// persists plugin state, `index_store` persists query indexes, + /// `wal_provider` persists durable transient-source changes, and `identity` + /// supplies credentials to plugins that ask for them. #[staticmethod] #[pyo3(signature = ( - id, *, secrets = None, state_store = None, index_store = None, identity = None, - plugins_dir = None, + id, *, secrets = None, state_store = None, index_store = None, wal_provider = None, + identity = None, plugins_dir = None, ))] + #[allow(clippy::too_many_arguments)] fn create<'py>( py: Python<'py>, id: String, secrets: Option>, state_store: Option<&Bound<'py, PyAny>>, index_store: Option<&Bound<'py, PyAny>>, + wal_provider: Option<&Bound<'py, PyAny>>, identity: Option<&Bound<'py, PyAny>>, plugins_dir: Option, ) -> PyResult> { // Parse eagerly so a malformed option raises before the caller awaits. - let options = CreateOptions::parse(secrets, state_store, index_store, identity)?; + let options = + CreateOptions::parse(secrets, state_store, index_store, wal_provider, identity)?; let durable_capable = options.has_state_store(); let persistent_indexes = options.has_index_store(); let wanted_identity = options.identity_plugin(); @@ -427,6 +431,10 @@ impl Drasi { .get_item("index_store")? .filter(|v| !v.is_none()) .as_ref(), + config + .get_item("wal_provider")? + .filter(|v| !v.is_none()) + .as_ref(), config .get_item("identity")? .filter(|v| !v.is_none()) @@ -2123,17 +2131,19 @@ impl Drasi { // ------------------------------------------------- Python-defined sources /// Registers a source that you push changes into with `push_change`. - #[pyo3(signature = (id, *, auto_start = true))] + #[pyo3(signature = (id, *, auto_start = true, durability = false))] fn add_python_source<'py>( &self, py: Python<'py>, id: String, auto_start: bool, + durability: bool, ) -> PyResult> { let inner = self.inner(); inner.ensure_open()?; future_into_py(py, async move { - let source = Arc::new(PythonSource::new(&id, auto_start).map_err(engine_error)?); + let source = + Arc::new(PythonSource::new(&id, auto_start, durability).map_err(engine_error)?); inner .core .add_source(SharedSource(Arc::clone(&source))) diff --git a/src/errors.rs b/src/errors.rs index ef8ee4e..3379d17 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -52,6 +52,8 @@ pub enum DrasiErrorCode { UnknownStateStoreKind, IndexStorePathRequired, UnknownIndexStoreKind, + WalProviderPathRequired, + UnknownWalProviderKind, IdentityKindRequired, UnknownIdentityKind, IdentityConfigInvalid, @@ -85,6 +87,8 @@ impl DrasiErrorCode { Self::UnknownStateStoreKind => "UNKNOWN_STATE_STORE_KIND", Self::IndexStorePathRequired => "INDEX_STORE_PATH_REQUIRED", Self::UnknownIndexStoreKind => "UNKNOWN_INDEX_STORE_KIND", + Self::WalProviderPathRequired => "WAL_PROVIDER_PATH_REQUIRED", + Self::UnknownWalProviderKind => "UNKNOWN_WAL_PROVIDER_KIND", Self::IdentityKindRequired => "IDENTITY_KIND_REQUIRED", Self::UnknownIdentityKind => "UNKNOWN_IDENTITY_KIND", Self::IdentityConfigInvalid => "IDENTITY_CONFIG_INVALID", @@ -120,6 +124,8 @@ impl DrasiErrorCode { UnknownStateStoreKind, IndexStorePathRequired, UnknownIndexStoreKind, + WalProviderPathRequired, + UnknownWalProviderKind, IdentityKindRequired, UnknownIdentityKind, IdentityConfigInvalid, @@ -146,6 +152,7 @@ impl DrasiErrorCode { | UnknownSecretStoreKind | UnknownStateStoreKind | UnknownIndexStoreKind + | UnknownWalProviderKind | UnknownIdentityKind | UnknownQueryLanguage | UnknownChangeOp => py.get_type::().into(), @@ -153,6 +160,7 @@ impl DrasiErrorCode { BootstrapKindRequired | StateStorePathRequired | IndexStorePathRequired + | WalProviderPathRequired | IdentityKindRequired | IdentityConfigInvalid | DurableRequiresStateStore diff --git a/src/stores.rs b/src/stores.rs index d250610..c4b34b9 100644 --- a/src/stores.rs +++ b/src/stores.rs @@ -14,9 +14,10 @@ //! Optional backing stores and credentials supplied when creating an engine. //! -//! All four are optional. Without them the engine keeps secrets, plugin state -//! and query indexes in memory, which is fine for tests and short-lived -//! processes but loses everything on restart. +//! All five are optional. Without them the engine keeps secrets, plugin state +//! and query indexes in memory and has no write-ahead log for transient +//! sources, which is fine for tests and short-lived processes but loses +//! everything on restart. use std::collections::HashMap; use std::sync::Arc; @@ -37,6 +38,7 @@ pub struct CreateOptions { pub secrets: HashMap, pub state_store: Option, pub index_store: Option, + pub wal_provider: Option, pub identity: Option, } @@ -51,6 +53,10 @@ pub struct IndexStore { pub config: Value, } +pub struct WalProvider { + pub path: String, +} + pub enum Identity { Password { username: String, @@ -87,12 +93,14 @@ impl CreateOptions { secrets: Option>, state_store: Option<&Bound<'_, PyAny>>, index_store: Option<&Bound<'_, PyAny>>, + wal_provider: Option<&Bound<'_, PyAny>>, identity: Option<&Bound<'_, PyAny>>, ) -> PyResult { Ok(Self { secrets: secrets.unwrap_or_default(), state_store: state_store.map(parse_state_store).transpose()?, index_store: index_store.map(parse_index_store).transpose()?, + wal_provider: wal_provider.map(parse_wal_provider).transpose()?, identity: identity.map(parse_identity).transpose()?, }) } @@ -160,6 +168,12 @@ impl CreateOptions { builder = apply_index_store(builder, index_store).await?; } + if let Some(wal_provider) = self.wal_provider { + builder = builder.with_wal_provider(Arc::new(drasi_wal_redb::RedbWalProvider::new( + wal_provider.path, + ))); + } + if let Some(identity) = self.identity { let provider: Arc = match identity { Identity::Password { username, password } => { @@ -295,6 +309,40 @@ fn parse_index_store(value: &Bound<'_, PyAny>) -> PyResult { Ok(IndexStore { config }) } +fn parse_wal_provider(value: &Bound<'_, PyAny>) -> PyResult { + let options = value.cast::().map_err(|_| { + error( + DrasiErrorCode::ConfigInvalid, + "'wal_provider' must be a mapping", + ) + })?; + + let kind = match options.get_item("kind")? { + None => "redb".to_string(), + Some(value) => value.extract().map_err(|_| { + error( + DrasiErrorCode::UnknownWalProviderKind, + "'wal_provider.kind' must be a string", + ) + })?, + }; + if kind != "redb" { + return Err(error( + DrasiErrorCode::UnknownWalProviderKind, + format!("unknown WAL provider kind '{kind}', expected 'redb'"), + )); + } + + Ok(WalProvider { + path: required( + options, + "path", + DrasiErrorCode::WalProviderPathRequired, + "a redb WAL provider requires a 'path'", + )?, + }) +} + /// Accepts the snake_case spelling for a backend key that is camelCase. fn rename(fields: &mut serde_json::Map, from: &str, to: &str) { if let Some(value) = fields.remove(from) { diff --git a/tests/e2e/test_concurrency.py b/tests/e2e/test_concurrency.py index b998306..6ac815a 100644 --- a/tests/e2e/test_concurrency.py +++ b/tests/e2e/test_concurrency.py @@ -50,8 +50,8 @@ async def collect() -> list[Any]: return await asyncio.wait_for(collect(), timeout) -async def running_orders_engine(engine: Drasi) -> Drasi: - await engine.add_python_source("orders") +async def running_orders_engine(engine: Drasi, *, durability: bool = False) -> Drasi: + await engine.add_python_source("orders", durability=durability) await engine.add_query("open", OPEN_ORDERS, ["orders"]) await engine.start() await wait_for_query_running(engine, "open") @@ -129,9 +129,10 @@ async def test_a_slow_durable_reaction_does_not_block_query_updates( "t-slow-reaction", state_store={"kind": "redb", "path": str(tmp_path / "state.redb")}, index_store={"kind": "rocksdb", "path": str(tmp_path / "index")}, + wal_provider={"path": str(tmp_path / "wal")}, ) try: - await running_orders_engine(drasi) + await running_orders_engine(drasi, durability=True) started = asyncio.Event() release = asyncio.Event() seen: list[str] = [] diff --git a/tests/e2e/test_durable_and_config.py b/tests/e2e/test_durable_and_config.py index 820dcd9..1ced742 100644 --- a/tests/e2e/test_durable_and_config.py +++ b/tests/e2e/test_durable_and_config.py @@ -58,9 +58,10 @@ async def durable_engine(tmp_path: Path, name: str = "durable") -> Drasi: name, state_store={"kind": "redb", "path": str(tmp_path / "state.redb")}, index_store={"kind": "rocksdb", "path": str(tmp_path / "index")}, + wal_provider={"path": str(tmp_path / "wal")}, ) await engine.start() - await engine.add_python_source("orders") + await engine.add_python_source("orders", durability=True) await engine.add_query("q", ORDERS_QUERY, ["orders"]) await wait_for_query_running(engine, "q") return engine diff --git a/tests/e2e/test_python_components.py b/tests/e2e/test_python_components.py index f868888..58881db 100644 --- a/tests/e2e/test_python_components.py +++ b/tests/e2e/test_python_components.py @@ -19,6 +19,7 @@ from __future__ import annotations +from pathlib import Path from typing import Any import pytest @@ -402,6 +403,64 @@ async def test_source_registered_without_auto_start_can_be_started(engine: Drasi assert await engine.get_source_status("orders") == "Running" +async def test_restarted_source_keeps_existing_query_subscriptions(engine: Drasi) -> None: + await engine.add_python_source("orders") + await engine.add_query("open", OPEN_ORDERS, ["orders"]) + await engine.start() + await wait_for_query_running(engine, "open") + + await engine.push_change( + "orders", + { + "op": "insert", + "id": "o1", + "labels": ["Order"], + "properties": {"id": "o1", "status": "open", "total": 1}, + }, + ) + await wait_for_rows(engine, "open") + + await engine.stop_source("orders") + await engine.start_source("orders") + await engine.push_change( + "orders", + { + "op": "insert", + "id": "o2", + "labels": ["Order"], + "properties": {"id": "o2", "status": "open", "total": 2}, + }, + ) + + rows = await wait_for_rows(engine, "open", count=2) + assert sorted(rows, key=lambda row: row["id"]) == [ + {"id": "o1", "total": 1}, + {"id": "o2", "total": 2}, + ] + + +async def test_durable_source_rejects_pushes_before_start(tmp_path: Path) -> None: + async with await Drasi.create( + "t-durable-not-started", + wal_provider={"path": str(tmp_path / "wal")}, + ) as engine: + await engine.start() + await engine.add_python_source("orders", auto_start=False, durability=True) + + with pytest.raises(DrasiError, match="must be running") as caught: + await engine.push_change( + "orders", + { + "op": "insert", + "id": "o1", + "labels": ["Order"], + "properties": {"id": "o1"}, + }, + ) + + assert caught.value.code == "ENGINE_FAILURE" + + async def test_running_source_with_dependents_is_not_removed(engine: Drasi) -> None: await engine.add_python_source("orders") await engine.add_query("open", OPEN_ORDERS, ["orders"]) diff --git a/tests/e2e/test_stores.py b/tests/e2e/test_stores.py index 8812eac..dfce2a7 100644 --- a/tests/e2e/test_stores.py +++ b/tests/e2e/test_stores.py @@ -72,6 +72,33 @@ async def test_index_store_requires_a_known_kind(tmp_path: Path) -> None: assert caught.value.code == "UNKNOWN_INDEX_STORE_KIND" +async def test_wal_provider_requires_a_path() -> None: + with pytest.raises(ConfigError) as caught: + await Drasi.create("t", wal_provider={"kind": "redb"}) # pyright: ignore[reportArgumentType] # invalid on purpose + assert caught.value.code == "WAL_PROVIDER_PATH_REQUIRED" + + +async def test_wal_provider_path_must_be_a_string() -> None: + with pytest.raises(ConfigError) as caught: + await Drasi.create("t", wal_provider={"path": 1}) # pyright: ignore[reportArgumentType] # invalid on purpose + assert caught.value.code == "WAL_PROVIDER_PATH_REQUIRED" + + +async def test_wal_provider_requires_a_known_kind(tmp_path: Path) -> None: + with pytest.raises(UnknownKindError) as caught: + await Drasi.create("t", wal_provider={"kind": "sqlite", "path": str(tmp_path)}) # pyright: ignore[reportArgumentType] # invalid on purpose + assert caught.value.code == "UNKNOWN_WAL_PROVIDER_KIND" + + +@pytest.mark.parametrize("kind", [None, 1]) +async def test_wal_provider_rejects_an_explicit_non_string_kind( + tmp_path: Path, kind: object +) -> None: + with pytest.raises(UnknownKindError) as caught: + await Drasi.create("t", wal_provider={"kind": kind, "path": str(tmp_path)}) # pyright: ignore[reportArgumentType] # invalid on purpose + assert caught.value.code == "UNKNOWN_WAL_PROVIDER_KIND" + + async def test_identity_requires_a_kind() -> None: with pytest.raises(ConfigError) as caught: await Drasi.create("t", identity={"username": "u", "password": "p"}) @@ -131,10 +158,12 @@ async def test_a_redb_state_store_is_accepted_and_written(tmp_path: Path) -> Non async def test_a_rocksdb_index_store_backs_query_state(tmp_path: Path) -> None: path = tmp_path / "index" async with await Drasi.create( - "t-rocksdb", index_store={"kind": "rocksdb", "path": str(path)} + "t-rocksdb", + index_store={"kind": "rocksdb", "path": str(path)}, + wal_provider={"path": str(tmp_path / "wal")}, ) as drasi: await drasi.start() - await drasi.add_python_source("s") + await drasi.add_python_source("s", durability=True) await drasi.add_query("q", ORDERS_QUERY, ["s"]) await wait_for_query_running(drasi, "q") await drasi.push_change( @@ -145,6 +174,18 @@ async def test_a_rocksdb_index_store_backs_query_state(tmp_path: Path) -> None: assert path.is_dir() and any(path.iterdir()), "the index should be on disk" +@pytest.mark.skipif(not rocksdb_available(), reason="built without the rocksdb feature") +async def test_a_persistent_index_rejects_a_volatile_python_source(tmp_path: Path) -> None: + async with await Drasi.create( + "t-volatile-source", + index_store={"kind": "rocksdb", "path": str(tmp_path / "index")}, + ) as drasi: + await drasi.start() + await drasi.add_python_source("s") + with pytest.raises(DrasiError, match="do not support replay"): + await drasi.add_query("q", ORDERS_QUERY, ["s"]) + + @pytest.mark.skipif(rocksdb_available(), reason="built with the rocksdb feature") async def test_rocksdb_without_the_feature_says_so(tmp_path: Path) -> None: with pytest.raises(UnknownKindError) as caught: diff --git a/tests/e2e/test_sync.py b/tests/e2e/test_sync.py index 284a579..e502518 100644 --- a/tests/e2e/test_sync.py +++ b/tests/e2e/test_sync.py @@ -21,6 +21,7 @@ from __future__ import annotations import inspect +from pathlib import Path import pytest @@ -128,6 +129,25 @@ def test_from_config_builds_a_started_engine() -> None: assert engine.is_running() is True +def test_wal_provider_and_durable_python_source_match_async_api(tmp_path: Path) -> None: + wal_path = tmp_path / "wal" + with SyncDrasi.create("sync-wal", wal_provider={"path": str(wal_path)}) as engine: + engine.start() + engine.add_python_source("orders", durability=True) + engine.add_query("open", OPEN_ORDERS, ["orders"]) + engine.wait_for_query("open") + engine.push_change("orders", order("o1")) + + rows = engine.get_query_results("open") + while rows != [{"id": "o1"}]: + rows = engine.get_query_results("open") + + assert rows == [{"id": "o1"}] + + assert wal_path.is_dir() + assert (wal_path / "orders.redb").is_file() + + def test_repr_identifies_the_engine() -> None: with running() as engine: assert "sync.Drasi(id='sync-test')" == repr(engine) diff --git a/tests/e2e/test_wal_recovery.py b/tests/e2e/test_wal_recovery.py new file mode 100644 index 0000000..c8e4c88 --- /dev/null +++ b/tests/e2e/test_wal_recovery.py @@ -0,0 +1,68 @@ +# Copyright 2026 The Drasi Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Crash recovery for a durable Python source with only a redb WAL.""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parents[2] +WORKER = Path(__file__).with_name("wal_recovery_worker.py") + + +def run_worker(mode: str, wal_path: Path) -> dict[str, Any]: + try: + result = subprocess.run( + [sys.executable, str(WORKER), mode, str(wal_path)], + cwd=ROOT, + capture_output=True, + text=True, + timeout=45, + ) + except subprocess.TimeoutExpired as exc: + raise AssertionError( + f"{mode} timed out\nstdout:\n{exc.stdout or ''}\nstderr:\n{exc.stderr or ''}" + ) from exc + + assert result.returncode == 0, ( + f"{mode} exited {result.returncode}\nstdout:\n{result.stdout}\nstderr:\n{result.stderr}" + ) + try: + result_line = result.stdout.rstrip().splitlines()[-1] + return json.loads(result_line) + except (IndexError, json.JSONDecodeError) as exc: + raise AssertionError( + f"{mode} did not emit exact JSON\nstdout:\n{result.stdout}\nstderr:\n{result.stderr}" + ) from exc + + +def test_durable_python_source_replays_after_a_fresh_process(tmp_path: Path) -> None: + wal_path = tmp_path / "wal" + + assert run_worker("writer", wal_path) == { + "mode": "writer", + "rows": [{"id": "o1"}], + } + assert wal_path.is_dir() + assert (wal_path / "orders.redb").is_file() + + assert run_worker("reader", wal_path) == { + "mode": "reader", + "rows": [{"id": "o1"}], + } diff --git a/tests/e2e/wal_recovery_worker.py b/tests/e2e/wal_recovery_worker.py new file mode 100644 index 0000000..b3535bd --- /dev/null +++ b/tests/e2e/wal_recovery_worker.py @@ -0,0 +1,81 @@ +# Copyright 2026 The Drasi Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Fresh-process worker for the Python-source WAL recovery acceptance test.""" + +from __future__ import annotations + +import asyncio +import json +import sys +from pathlib import Path +from typing import Any + +from drasi import Drasi + +QUERY = "MATCH (o:Order) RETURN o.id AS id" +EXPECTED_ROWS = [{"id": "o1"}] + + +async def wait_for_rows(drasi: Drasi) -> list[dict[str, Any]]: + async def poll() -> list[dict[str, Any]]: + while True: + rows = await drasi.get_query_results("q") + if rows == EXPECTED_ROWS: + return rows + await asyncio.sleep(0.02) + + return await asyncio.wait_for(poll(), timeout=15) + + +async def writer(wal_path: Path) -> dict[str, object]: + drasi = await Drasi.create("wal-recovery", wal_provider={"path": str(wal_path)}) + try: + await drasi.start() + await drasi.add_python_source("orders", durability=True) + await drasi.add_query("q", QUERY, ["orders"]) + await drasi.wait_for_query("q", timeout=15) + await drasi.push_change( + "orders", + { + "op": "insert", + "id": "o1", + "labels": ["Order"], + "properties": {"id": "o1"}, + }, + ) + return {"mode": "writer", "rows": await wait_for_rows(drasi)} + finally: + await drasi.close() + + +async def reader(wal_path: Path) -> dict[str, object]: + drasi = await Drasi.from_config({"id": "wal-recovery", "wal_provider": {"path": str(wal_path)}}) + try: + await drasi.add_python_source("orders", durability=True) + await drasi.add_query("q", QUERY, ["orders"]) + await drasi.wait_for_query("q", timeout=15) + return {"mode": "reader", "rows": await wait_for_rows(drasi)} + finally: + await drasi.close() + + +async def main() -> None: + mode, raw_path = sys.argv[1:] + action = {"writer": writer, "reader": reader}[mode] + print(json.dumps(await action(Path(raw_path)), sort_keys=True, separators=(",", ":"))) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/website/content/docs/api/_index.md b/website/content/docs/api/_index.md index 467f34e..8be7c30 100644 --- a/website/content/docs/api/_index.md +++ b/website/content/docs/api/_index.md @@ -13,8 +13,14 @@ same name on `drasi.sync.Drasi`, where the `Awaitable[...]` return becomes the v itself and a `Stream` becomes an ordinary iterator — see [the blocking API](../guides/blocking-api/). -The shapes referenced here (`SourceChange`, `QueryResultEvent`, `StateStore` and the -rest) are declared in `drasi.types`, described in the [typing guide](../guides/typing/). +The shapes referenced here (`SourceChange`, `QueryResultEvent`, `StateStore`, +`WalProvider` and the rest) are declared in `drasi.types`, described in the +[typing guide](../guides/typing/). + +`wal_provider={"path": ...}` configures a directory. The redb provider creates one +WAL file per durable transient source inside it. This is Drasi's local +transient-source write-ahead log; it is not a PostgreSQL replication slot and does +not create, manage or resume one. {{% alert title="Generated from the shipped stubs" color="info" %}} The tables below are generated from `python/drasi/_drasi.pyi`, the same file your type @@ -29,7 +35,7 @@ from what the package exposes. | --- | --- | | `id` _(property)_ | `str` | | `from_config(config: Mapping[str, Any])` _(static)_ | `Awaitable[Drasi]` | -| `create(id: str, *, secrets: Mapping[str, str] \| None = None, state_store: StateStore \| None = None, index_store: IndexStore \| None = None, identity: Identity \| None = None, plugins_dir: StrPath \| None = None)` _(static)_ | `Awaitable[Drasi]` | +| `create(id: str, *, secrets: Mapping[str, str] \| None = None, state_store: StateStore \| None = None, index_store: IndexStore \| None = None, wal_provider: WalProvider \| None = None, identity: Identity \| None = None, plugins_dir: StrPath \| None = None)` _(static)_ | `Awaitable[Drasi]` | ### Lifecycle @@ -58,7 +64,7 @@ from what the package exposes. | Member | Returns | | --- | --- | -| `add_python_source(id: str, *, auto_start: bool = True)` | `Awaitable[None]` | +| `add_python_source(self, id: str, *, auto_start: bool = True, durability: bool = False )` | `Awaitable[None]` | | `push_change(source_id: str, change: SourceChange)` | `Awaitable[None]` | | `add_python_reaction(self, id: str, query_ids: Sequence[str], callback: Callable[[QueryResultEvent], object])` | `Awaitable[None]` | | `add_durable_python_reaction(self, id: str, query_ids: Sequence[str], callback: Callable[[QueryResultEvent], Awaitable[object]], *, recovery_policy: Literal["strict", "auto_reset", "skip_gap"] = "strict")` | `Awaitable[None]` | diff --git a/website/content/docs/guides/error-handling.md b/website/content/docs/guides/error-handling.md index e6f0571..7144d32 100644 --- a/website/content/docs/guides/error-handling.md +++ b/website/content/docs/guides/error-handling.md @@ -73,6 +73,8 @@ The full set is available at runtime as `drasi.ERROR_CODES`. | `UNKNOWN_STATE_STORE_KIND` | The state store kind is not one this build supports | | `INDEX_STORE_PATH_REQUIRED` | An index store was configured without a path | | `UNKNOWN_INDEX_STORE_KIND` | The index store kind is not compiled into this wheel | +| `WAL_PROVIDER_PATH_REQUIRED` | A WAL provider was configured without its directory path | +| `UNKNOWN_WAL_PROVIDER_KIND` | The WAL provider kind is not supported | | `IDENTITY_KIND_REQUIRED` | An identity block was given without a `kind` | | `UNKNOWN_IDENTITY_KIND` | No identity provider matches that kind | | `IDENTITY_CONFIG_INVALID` | The identity block is missing fields its kind needs | diff --git a/website/content/docs/guides/python-reactions.md b/website/content/docs/guides/python-reactions.md index 1a2965b..efcc192 100644 --- a/website/content/docs/guides/python-reactions.md +++ b/website/content/docs/guides/python-reactions.md @@ -68,9 +68,14 @@ drasi = await Drasi.create( "app", state_store={"kind": "redb", "path": "/var/lib/myapp/drasi/state.redb"}, index_store={"kind": "rocksdb", "path": "/var/lib/myapp/drasi/index"}, + wal_provider={"path": "/var/lib/myapp/drasi/wal"}, ) ``` +If the query reads a Python source, register that source with +`durability=True`. A persistent query rejects a volatile Python source because +it cannot replay changes after a restart. + Then register the reaction with a recovery policy: ```python diff --git a/website/content/docs/guides/python-sources.md b/website/content/docs/guides/python-sources.md index 9f7fcc1..bbde4eb 100644 --- a/website/content/docs/guides/python-sources.md +++ b/website/content/docs/guides/python-sources.md @@ -13,6 +13,22 @@ your application changes something worth querying. await drasi.add_python_source("orders") ``` +By default, a Python source is volatile: changes that arrived before a process +restart cannot be replayed. To make it durable, configure an engine-wide WAL +directory and opt the source in: + +```python +drasi = await Drasi.create( + "app", + wal_provider={"path": "/var/lib/myapp/drasi/wal"}, +) +await drasi.start() +await drasi.add_python_source("orders", durability=True) +``` + +`kind` is optional and defaults to `redb`. The path is a directory; Drasi creates +one redb file per durable source in it. Volatile sources do not create WAL files. + ## Pushing a node ```python diff --git a/website/content/docs/guides/typing.md b/website/content/docs/guides/typing.md index a37a591..25833ae 100644 --- a/website/content/docs/guides/typing.md +++ b/website/content/docs/guides/typing.md @@ -45,7 +45,7 @@ Some of the more useful ones: | `ComponentEvent` | A lifecycle event | | `LogMessage` | A log line from a component | | `HostInfo` | Versions and target of this build | -| `StateStore` / `IndexStore` / `Identity` | Engine creation options | +| `StateStore` / `IndexStore` / `WalProvider` / `Identity` | Engine creation options | | `InstalledPlugin` / `ResolvedPlugin` / `LockedPlugin` | Plugin metadata | ## Literal types catch typos