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 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.