Struct EmbeddedKernelStore 

pub struct EmbeddedKernelStore { /* private fields */ }

Every kernel persistence port on one local store.

The engine behind it is chosen when the data directory is created and hidden behind the storage seam (ADR-018): redb by default, SQLite when asked for and compiled in. Cloning is cheap (shared engine handle). Commits are fsync-durable on both engines, so each successful port write survives kill -9; a crash mid-transaction loses only the in-flight transaction.

Implementations

impl EmbeddedKernelStore

pub async fn migrate_data_dir<F>( source_dir: &Path, destination_dir: &Path, derive: F, ) -> Result<(Self, StoreMigrationReceipt), PortError>

where F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,

Migrates source_dir into destination_dir and opens the result. The destination is created with the default engine.

derive is the projection derivation the composition root owns (kmp_application::projection_mutations_for_context_event), kept injected so this adapter stays free of the application layer.

pub async fn migrate_data_dir_to<F>( source_dir: &Path, destination_dir: &Path, destination_engine: StorageEngine, derive: F, ) -> Result<(Self, StoreMigrationReceipt), PortError>

where F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,

migrate_data_dir with the destination engine chosen. This is how a store changes engines (ADR-018): the event log is the source of truth and projections are derived, so a redb store becomes a SQLite store by replaying its history into a fresh SQLite directory — the same operation a format bump has always been. The source is not modified; the receipt records both formats.

pub async fn open_or_migrate_data_dir<F>( source_dir: &Path, destination_dir: &Path, derive: F, ) -> Result<(Self, Option<StoreMigrationReceipt>), PortError>

where F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,

Migrate once, reopen afterwards: safe to call on every start.

A destination that already holds a store is opened as it is — the migration is not repeated, and the receipt (when there is one) says where that memory came from.

pub async fn open_or_migrate_data_dir_to<F>( source_dir: &Path, destination_dir: &Path, destination_engine: StorageEngine, derive: F, ) -> Result<(Self, Option<StoreMigrationReceipt>), PortError>

where F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,

open_or_migrate_data_dir with the destination engine chosen. The engine only matters on the call that migrates; a destination that already holds a store opens as whatever it is.

pub async fn migration_receipt( &self, ) -> Result<Option<StoreMigrationReceipt>, PortError>

The receipt of the migration that produced this store, if any.

impl EmbeddedKernelStore

pub async fn export_bundle(&self) -> Result<String, PortError>

Serializes the full event log as a JSON-Lines bundle: one header line followed by one event per line, in sequence order.

pub async fn import_bundle<F>( &self, bundle: &str, derive: F, ) -> Result<ImportReport, PortError>

where F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,

Replays a bundle into this store. Fail-fast rules: the store must be empty (no merge semantics in v1 — ADR-011 rationale applies), the header must match supported formats, and every event must reproduce exactly the revision it was exported with.

impl EmbeddedKernelStore

pub async fn rebuild_projections<F>( &self, derive: F, ) -> Result<ProjectionRebuildReport, PortError>

where F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,

Drops every projection table and rebuilds them by replaying the event log in sequence order — the recovery and migration story in one.

The mutation derivation is injected so this adapter stays free of application-layer dependencies; the composition root passes kmp_application::projection_mutations_for_context_event. The whole rebuild is one transaction: a crash mid-rebuild leaves the previous projections intact.

impl EmbeddedKernelStore

pub fn open(data_dir: &Path) -> Result<Self, PortError>

Opens (or initializes) the store inside data_dir, applying the ADR-012 fail-fast rules before touching the engine. A fresh directory gets the default engine; an existing one opens with the engine it was created with.

pub fn open_with_engine( data_dir: &Path, engine: StorageEngine, ) -> Result<Self, PortError>

open with the engine chosen: a fresh directory is created for engine, and an existing one must already be engine — a store is never reinterpreted as another engine’s.

pub fn engine_of(data_dir: &Path) -> Result<StorageEngine, PortError>

The engine a data directory was created with, without opening it.

pub async fn event_log_stats(&self) -> Result<(u64, u64), PortError>

Number of events in the append-only log and the highest sequence — audit surface used by recovery checks and operational tooling.

pub fn compact_data_dir(data_dir: &Path) -> Result<bool, PortError>

Compacts the store file in place, reclaiming free pages left by past transactions (e.g. after a projection rebuild). Requires exclusive access: call it with no other store handle open on the same data directory.

Trait Implementations

impl Clone for EmbeddedKernelStore

fn clone(&self) -> EmbeddedKernelStore

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl ContextEventStore for EmbeddedKernelStore

async fn append( &self, event: ContextUpdatedEvent, expected_revision: u64, ) -> Result<u64, PortError>

Append an event. Fails with PortError::Conflict if expected_revision does not match the current revision for this (root_node_id, role) aggregate.

async fn current_revision( &self, root_node_id: &str, role: &str, ) -> Result<u64, PortError>

Returns the current revision for the aggregate, or 0 if no events exist.

async fn current_content_hash( &self, root_node_id: &str, role: &str, ) -> Result<Option<String>, PortError>

Returns the content hash of the last accepted event, or None if no events exist.

async fn find_by_idempotency_key( &self, key: &str, ) -> Result<Option<IdempotentOutcome>, PortError>

Checks if an event with this idempotency key was already accepted.

impl Debug for EmbeddedKernelStore

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl GraphNeighborhoodReader for EmbeddedKernelStore

async fn load_neighborhood( &self, root_node_id: &str, depth: u32, ) -> Result<Option<NodeNeighborhood>, PortError>

async fn load_context_path( &self, root_node_id: &str, target_node_id: &str, subtree_depth: u32, ) -> Result<Option<ContextPathNeighborhood>, PortError>

impl MemoryAboutIndexReader for EmbeddedKernelStore

async fn list_memory_abouts(&self) -> Result<Vec<String>, PortError>

async fn list_memory_abouts_by_dimensions( &self, dimension_ids: &[String], ) -> Result<Vec<String>, PortError>

impl NodeDetailReader for EmbeddedKernelStore

async fn load_node_detail( &self, node_id: &str, ) -> Result<Option<NodeDetailProjection>, PortError>

async fn load_node_details_batch( &self, node_ids: Vec<String>, ) -> Result<Vec<Option<NodeDetailProjection>>, PortError>

impl NodeRelationshipReader for EmbeddedKernelStore

async fn load_node_relationships( &self, node_id: &str, ) -> Result<Option<NodeRelationships>, PortError>

impl ProcessedEventStore for EmbeddedKernelStore

async fn has_processed( &self, consumer_name: &str, event_id: &str, ) -> Result<bool, PortError>

async fn record_processed( &self, consumer_name: &str, event_id: &str, ) -> Result<(), PortError>

impl ProjectionCheckpointStore for EmbeddedKernelStore

async fn load_checkpoint( &self, consumer_name: &str, stream_name: &str, ) -> Result<Option<ProjectionCheckpoint>, PortError>

async fn save_checkpoint( &self, checkpoint: ProjectionCheckpoint, ) -> Result<(), PortError>

impl ProjectionWriter for EmbeddedKernelStore

async fn apply_mutations( &self, mutations: Vec<ProjectionMutation>, ) -> Result<(), PortError>

impl SnapshotStore for EmbeddedKernelStore

async fn save_bundle_with_options( &self, bundle: &KmpBundle, options: SnapshotSaveOptions, ) -> Result<(), PortError>

fn save_bundle( &self, bundle: &KmpBundle, ) -> impl Future<Output = Result<(), PortError>> + Send