Struct Store 

pub struct Store {
    pub path: PathBuf,
    /* private fields */
}

An append-only event stream backed by a directory on disk.

Open one with new. Store is cheaply Cloneable: every clone shares the same underlying database, broadcast channel, and content-addressed store, so clone it freely across tasks and threads instead of wrapping it in an Arc.

See the module docs for the topic and retention model.

Fields

path: PathBuf

Directory backing this store (the path passed to new).

Implementations

impl Store

pub fn new(path: PathBuf) -> Result<Store, StoreError>

Open the store at path, creating the directory layout if it does not exist. Spawns a background worker that garbage-collects expired frames.

Errors

Returns StoreError::Locked if another process already holds the store open, or StoreError::Other for any other database error.

use xs::Store;

let store = Store::new("./clipboard-store".into())?;

pub fn with_base_engine(self, base: EngineState) -> Self

Set a base engine the processor engines clone.

A program that embeds xs prepares an EngineState with its own commands, env, and consts; prepared_base clones it per spawn and adds the store commands, so the actor, service, and action processors get those commands. Shared across Store clones, so set it once, before the processors spawn. No base set: Engine::new().

pub fn base_engine(&self) -> Option<&EngineState>

The base engine an embedder attached via [with_base_engine], if any.

pub async fn wait_for_gc(&self)

Wait until the background garbage-collection worker has processed every task queued so far. Useful in tests to observe TTL eviction deterministically.

pub async fn read(&self, options: ReadOptions) -> Receiver<Frame>

Read frames into an async channel according to options.

By default this replays matching historical frames oldest-first and then closes the channel. With FollowOption::On it instead keeps the channel open and streams new appends as they arrive. When following, a single ephemeral xs.threshold frame is emitted to mark the boundary between replayed history and live events.

The returned Receiver is bounded; dropping it stops the read. For a blocking, non-async caller use read_sync.

use xs::{Store, ReadOptions, FollowOption};

let mut rx = store
    .read(ReadOptions::builder().follow(FollowOption::On).build())
    .await;
while let Some(frame) = rx.recv().await {
    if frame.topic == "xs.threshold" {
        // caught up to live; everything after this is new
        continue;
    }
    println!("{} {}", frame.id, frame.topic);
}

pub fn read_sync( &self, options: ReadOptions, ) -> impl Iterator<Item = Frame> + '_

Replay matching historical frames as a blocking iterator.

This honours the topic, from, after, limit, and last parts of ReadOptions but ignores follow: it never streams live appends. Use read when you need to follow.

use xs::{Store, ReadOptions};

let opts = ReadOptions::builder().topic("clip.*".to_string()).last(10).build();
for frame in store.read_sync(opts) {
    println!("{} {}", frame.id, frame.topic);
}

pub fn nu_modules_at(&self, as_of: &Scru128Id) -> HashMap<String, Integrity>

Returns the current module state as of a given point in the stream.

Scans all frames up to (and including) as_of and returns a mapping of module name to CAS hash for the latest frame on each xs.module.<name> topic. Resolve the set of registered Nushell modules as of a given frame ID.

Scans xs.module.<name> frames up to and including as_of and returns a map from module name to the content hash of its latest definition. Used by the scripting runtime; rarely needed when embedding the store directly.

pub fn get(&self, id: &Scru128Id) -> Option<Frame>

Fetch a single frame by ID, or None if no such frame exists.

pub fn remove(&self, id: &Scru128Id) -> Result<(), Error>

Delete a frame and its topic index entries. Removing a frame that does not exist is a no-op and returns Ok(()).

This removes the stream entry only; any payload bytes in the content-addressed store are left in place.

pub async fn cas_reader(&self, hash: Integrity) -> Result<Reader>

Open a streaming reader for the payload identified by hash.

pub fn cas_reader_sync(&self, hash: Integrity) -> Result<SyncReader>

Blocking variant of cas_reader.

pub async fn cas_writer(&self) -> Result<Writer>

Open a streaming writer; finish it to obtain the payload’s integrity hash.

pub fn cas_writer_sync(&self) -> Result<SyncWriter>

Blocking variant of cas_writer.

pub async fn cas_insert(&self, content: impl AsRef<[u8]>) -> Result<Integrity>

Store content and return its integrity hash, ready to attach to a Frame::hash.

pub fn cas_insert_sync(&self, content: impl AsRef<[u8]>) -> Result<Integrity>

Blocking variant of cas_insert.

pub async fn cas_insert_bytes(&self, bytes: &[u8]) -> Result<Integrity>

Convenience wrapper over cas_insert for a byte slice.

pub fn cas_insert_bytes_sync(&self, bytes: &[u8]) -> Result<Integrity>

Blocking variant of cas_insert_bytes.

pub async fn cas_read(&self, hash: &Integrity) -> Result<Vec<u8>>

Read back the full payload for hash into a Vec<u8>.

pub fn cas_read_sync(&self, hash: &Integrity) -> Result<Vec<u8>>

Blocking variant of cas_read.

pub fn insert_frame(&self, frame: &Frame) -> Result<(), Error>

Persist a frame exactly as given, including its existing id, without broadcasting it to live readers or scheduling TTL garbage collection.

Most callers want append instead, which assigns a fresh ID, handles ephemeral and Last retention, and notifies subscribers. Use insert_frame only when you are reconstructing a stream with predetermined IDs (for example when restoring a backup).

pub fn append(&self, frame: Frame) -> Result<Frame, Error>

Append a frame to the stream and return it with its freshly assigned id.

This is the primary write path. It:

• assigns a new time-sortable ID (overwriting any ID on the input);
• validates the topic (see validate_topic);
• persists the frame, unless its TTL is TTL::Ephemeral, in which case it is only broadcast to live readers;
• schedules garbage collection for TTL::Last retention;
• broadcasts the frame to everyone currently in a following read.

Appends are serialized internally, so frames are assigned IDs and delivered to subscribers in a consistent order.

Errors

Returns an error if the topic is invalid or the underlying write fails.

use xs::{Store, Frame, TTL};

let hash = store.cas_insert("hello clipboard").await?;
let frame = store.append(
    Frame::builder("clip.add").hash(hash).ttl(TTL::Last(100)).build(),
)?;
println!("appended {}", frame.id);

Trait Implementations

impl Clone for Store

fn clone(&self) -> Store

Returns a duplicate of the value.

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

Performs copy-assignment from source.