Struct NodeDbLite 

pub struct NodeDbLite<S: StorageEngine> { /* private fields */ }

NodeDB-Lite — the embedded edge database.

Fully capable of vector search, graph traversal, and document CRUD entirely offline. Optional sync to Origin via WebSocket.

Implementations

impl<S: StorageEngine> NodeDbLite<S>

pub async fn create_array( &self, name: &str, schema: ArraySchema, ) -> NodeDbResult<()>

Create a new ND sparse array with the given schema.

Returns an error if an array named name already exists.

pub async fn array_put_cell( &self, name: &str, coord: Vec<CoordValue>, attrs: Vec<CellValue>, system_from_ms: i64, valid_from_ms: i64, valid_until_ms: i64, ) -> NodeDbResult<()>

Write a cell into array name at coord.

system_from_ms is the bitemporal system time (typically now()). valid_from_ms / valid_until_ms are the valid-time bounds (OPEN_UPPER = open-ended, no expiry).

pub async fn array_slice( &self, name: &str, ranges: Vec<Option<DimRange>>, as_of_system_ms: Option<i64>, ) -> NodeDbResult<Vec<CellPayload>>

Slice query: return all live cells whose coordinates fall within ranges at or before as_of_system_ms (defaults to i64::MAX for the current live snapshot).

pub async fn array_read_coord( &self, name: &str, coord: &[CoordValue], as_of_system_ms: Option<i64>, ) -> NodeDbResult<Option<CellPayload>>

Read the most recent live payload for coord at or before as_of_system_ms. Returns None if not found, tombstoned, or erased.

pub async fn array_delete_cell( &self, name: &str, coord: Vec<CoordValue>, system_from_ms: i64, ) -> NodeDbResult<()>

Soft-delete a cell by writing a tombstone version at system_from_ms.

The cell is still visible AS-OF any system time < system_from_ms.

pub async fn array_gdpr_erase_cell( &self, name: &str, coord: Vec<CoordValue>, system_from_ms: i64, ) -> NodeDbResult<()>

GDPR erasure: write the 0xFE sentinel and flush to disk.

After this call array_read_coord returns None for the erased coordinate at any system_as_of >= system_from_ms, and the raw payload bytes are not present on disk.

pub async fn array_flush(&self, name: &str) -> NodeDbResult<()>

Flush any pending memtable data for name to a durable segment.

pub fn array_schema_hlc(&self, name: &str) -> Option<Hlc>

Return the current schema HLC for name from the local schema registry.

Used by tests that need to construct ArrayOpHeader::schema_hlc values matching the locally registered schema.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn batch_vector_insert( &self, collection: &str, vectors: &[(&str, &[f32])], ) -> NodeDbResult<()>

Batch insert vectors — O(1) CRDT delta export instead of O(N).

Use this for bulk loading (cold-start hydration, benchmark setup, imports). Each vector is inserted into HNSW and tracked in the ID map. An id repeated in the batch, or already indexed, keeps one node: the latest vector. Each id’s embedding dimension merges into its CRDT row, whose other fields are kept.

pub async fn batch_graph_insert_edges( &self, collection: &str, edges: &[(NodeId, NodeId, &str, Option<Document>)], ) -> NodeDbResult<Vec<EdgeId>>

Batch insert graph edges into a named collection — O(1) CRDT delta export instead of O(N). Edges are isolated to collection.

Mirrors graph_insert_edge’s argument order (from, to, edge_type, properties) and shares its edge-preparation logic (EdgeId allocation, CRDT field list): a batch of one edge stores the same EdgeId key and the same fields a single graph_insert_edge call would, so graph_delete_edge can remove either durably and rebuild never resurrects a deleted batch-inserted edge.

Bitemporal history is recorded per edge when collection is bitemporal, and memory stats are updated once at the end. Returns the allocated EdgeId for each input edge, in order.

pub fn compact_graph(&self) -> NodeDbResult<()>

Compact all per-collection CSR graph indices (merge buffer into dense arrays).

pub async fn evict_collections( &self, max_to_evict: usize, ) -> NodeDbResult<usize>

Evict HNSW collections to reduce memory usage.

Persists each evicted collection to storage first, then drops it from memory. Collections are evicted smallest-first.

pub async fn check_and_evict(&self) -> NodeDbResult<usize>

Check memory pressure and evict if needed.

Matches the old thresholds: nodedb_mem’s Critical (85-95%) is where Lite’s own Warning used to start, and Emergency (>95%) is where Lite’s own Critical used to start. The new Warning tier (70-85%) is below both old thresholds, so it evicts nothing, same as Normal.

impl<S: StorageEngine> NodeDbLite<S>

pub fn bulk_update( &self, collection: &str, filters: &[ScanFilter], updates: &HashMap<String, Value>, ) -> NodeDbResult<u64>

Bulk update documents matching a predicate.

Scans all documents, evaluates ScanFilter predicates, and applies updates to matching documents — all under a single CRDT lock to prevent concurrent writes from being lost between scan and update.

Returns the number of documents updated.

pub fn bulk_delete( &self, collection: &str, filters: &[ScanFilter], ) -> NodeDbResult<u64>

Bulk delete documents matching a predicate.

Same single-lock pattern as bulk_update. Returns the number of documents deleted.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn create_collection( &self, name: &str, fields: &[(String, String)], ) -> NodeDbResult<()>

Create a collection with optional schema.

If the collection already exists, returns Ok (idempotent). Schema is advisory — documents are schemaless by default.

pub async fn create_kv_collection( &self, name: &str, config: &KvConfig, ) -> NodeDbResult<()>

Create a KV collection with typed schema and optional TTL.

Stores the KvConfig as JSON in the collection metadata so that the KV engine can reconstruct the schema on startup.

pub async fn drop_collection(&self, name: &str) -> NodeDbResult<()>

Drop a collection — deletes all documents and metadata.

Uses clear_collection for single-batch deletion (one Loro delta for all document removals). Also removes the secondary and text indexes.

pub async fn list_collections(&self) -> NodeDbResult<Vec<CollectionMeta>>

List all collections.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn copy_from_ndjson( &self, collection: &str, ndjson: &str, ) -> NodeDbResult<u64>

Import documents from NDJSON (newline-delimited JSON) text.

Each line is a JSON object. The “id” field is used as document ID; if missing, a UUID v7 is generated.

Returns the number of documents imported.

pub async fn copy_from_csv( &self, collection: &str, csv_text: &str, ) -> NodeDbResult<u64>

Import documents from CSV text.

First line is the header (column names). Each subsequent line is a row. An “id” column is used as document ID; if missing, UUIDs are generated.

Two-pass: first pass infers column types (int/float/bool/string) from ALL rows. Second pass coerces every value to the column’s dominant type. This ensures type consistency within each column.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn kv_put( &self, collection: &str, key: &str, value: &[u8], ) -> NodeDbResult<()>

KV PUT: store a key-value pair with no expiry.

Buffered in memory — call kv_flush() to commit, or let the auto-flush threshold handle it.

pub async fn kv_put_with_ttl( &self, collection: &str, key: &str, value: &[u8], ttl_ms: u64, ) -> NodeDbResult<()>

KV PUT WITH TTL: store a key-value pair that expires after ttl_ms ms.

After ttl_ms milliseconds, kv_get will return None for this key and lazy-delete it. The deadline survives a database reopen.

pub async fn kv_get( &self, collection: &str, key: &str, ) -> NodeDbResult<Option<Vec<u8>>>

KV GET: retrieve a value by key.

Returns None for missing or expired keys. Expired keys are lazily deleted from storage on read.

Checks the in-memory write buffer first (for uncommitted writes), then falls through to the KV store.

pub async fn kv_delete(&self, collection: &str, key: &str) -> NodeDbResult<bool>

KV DELETE: remove a key.

Returns true when a live value was present and removed, false when the key was absent (or already expired) — mirroring HashMap::remove.

The delete is recorded for sync even when the key is absent here, because Origin can hold the key when this replica does not.

pub async fn kv_flush(&self) -> NodeDbResult<usize>

Flush buffered KV writes to storage as a single transaction. Returns the number of writes flushed.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn kv_range_scan( &self, collection: &str, start: Option<&[u8]>, end: Option<&[u8]>, limit: Option<usize>, ) -> NodeDbResult<Vec<(Vec<u8>, Vec<u8>)>>

KV RANGE SCAN: ordered key scan with optional bounds and limit.

Returns (key, value) pairs where start <= key < end, ordered by key in lexicographic byte order. Expired keys are skipped and lazily deleted.

• start = None means scan from the beginning of the collection.
• end = None means scan to the end of the collection.
• limit = None means no cap on results.

Flushes the write buffer before scanning so the KV store reflects all pending writes.

pub async fn kv_compact_expired(&self, collection: &str) -> NodeDbResult<usize>

KV COMPACT EXPIRED: eagerly remove all expired keys in a collection.

Flushes the write buffer, then scans all keys in the collection and deletes any whose TTL deadline has passed. Returns the count of keys removed.

pub async fn kv_scan( &self, collection: &str, cursor: &str, count: usize, ) -> NodeDbResult<Vec<(String, Vec<u8>)>>

KV SCAN: iterate keys in sorted order starting from cursor.

Returns up to count key-value pairs where key >= cursor (inclusive). Pass an empty cursor to start from the beginning of the collection.

Flushes the write buffer first to ensure the KV store has all data, then uses the storage’s B-tree range scan — O(log N + count).

pub async fn kv_keys(&self, collection: &str) -> NodeDbResult<Vec<String>>

List all keys in a KV collection.

pub async fn kv_subscribe_shape( &self, collection: &str, key_pattern: &str, ) -> NodeDbResult<Vec<String>>

Subscribe to a subset of KV keys matching a pattern.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn kv_increment( &self, collection: &str, key: &str, delta: i64, ) -> NodeDbResult<i64>

KV INCREMENT: add delta to the counter at key and return the new value.

The counter follows the rules the SQL INCR follows:

• a raw value is decimal text in and decimal text out;
• a typed row moves its first integer column in key order;
• an absent or expired key starts at 0 and stores decimal text;
• the key keeps its expiry.

A value that is not a counter, or a sum outside i64, is an error and writes nothing. The new value syncs to Origin like any put.

impl<S: StorageEngine> NodeDbLite<S>

pub fn transaction(&self, ops: &[TransactionOp]) -> NodeDbResult<u64>

Execute a batch of operations atomically.

Atomicity model: all operations are validated upfront. If any validation fails, none are applied. Once validation passes, all operations are applied under a single CRDT lock hold. Each row still exports its own delta (see CrdtEngine::batch_upsert) — a delta spanning rows or collections is not independently applicable by the receiver.

This is NOT ACID rollback — Loro doesn’t support undo. Instead, we validate-then-apply: no partial commits are possible because we only start writing after all checks pass.

Returns the number of operations applied.

pub fn set_conflict_policy(&self, collection: &str, policy: CollectionPolicy)

Set conflict resolution policy for a collection.

Policies are evaluated on sync when Origin rejects a delta. Available policies from nodedb-crdt::PolicyRegistry:

• LastWriterWins (default)
• RenameSuffix
• CascadeDefer
• EscalateToDlq

impl<S: StorageEngine> NodeDbLite<S>

pub fn start_auto_compact(self: &Arc<Self>, interval_ms: u64)

Start a background task that calls the global compact() every interval_ms milliseconds, reclaiming dead pages and truncating the backing file so on-disk growth stays bounded.

When to use

Compaction is opt-in and unnecessary for most workloads. Enable it when writing one commit per entry, where the pagedb deferred-free list would otherwise accumulate and the file would grow without bound. For engines with nothing to compact (in-memory / test stores) the call is a cheap no-op every tick.

Cost vs. auto-flush

This is heavier than start_auto_flush: compaction repacks the B+ trees and truncates the file, and it no-ops while a reader pins the reclaimable page range. Use a much larger interval than the flush interval — minutes, not seconds.

Usage

The open* constructors already start this task from LiteConfig::auto_compact_ms, which defaults to 0 (disabled). Set that field to enable compaction:

let config = LiteConfig { auto_compact_ms: 300_000, ..LiteConfig::default() };
let db = NodeDbLite::open_with_config(storage, config).await?;

Calling this directly spawns an additional task rather than replacing a running one.

Task lifecycle

The task is registered with the database’s task registry, so NodeDbLite::shutdown stops it before the host drops its async runtime. It also holds a Weak reference, so a database dropped without a shutdown still ends the loop at its next tick rather than keeping the database alive.

Disabling

Pass interval_ms = 0 to skip spawning entirely (auto-compaction disabled — the default). Compaction can still be triggered manually via compact().

impl<S: StorageEngine> NodeDbLite<S>

pub fn start_auto_flush(self: &Arc<Self>, interval_ms: u64)

Start a background task that calls the global flush() every interval_ms milliseconds, bounding the data-loss window uniformly across all engines (KV buffer, vector id-map, CRDT deltas, CSR graph, spatial, FTS).

Durability contract

await-ing a write operation (e.g. kv_put, vector_insert) returning Ok does NOT guarantee on-disk durability. Durability is bounded by interval_ms. For guaranteed durability, call flush() explicitly after writes.

Usage

The open* constructors already start this task from LiteConfig::auto_flush_ms, so calling it is only needed to change the interval afterwards:

let db = NodeDbLite::open(storage).await?;
db.start_auto_flush(5_000); // slow the flusher down to five seconds

Each call spawns an additional task rather than replacing the running one, so opening with auto_flush_ms: 0 is the way to take full manual control of when flushes happen.

Task lifecycle

The task is registered with the database’s task registry, so NodeDbLite::shutdown stops it before the host drops its async runtime. It also holds a Weak reference, so a database dropped without a shutdown still ends the loop at its next tick rather than keeping the database alive.

Disabling

Pass interval_ms = 0 to skip spawning entirely (auto-flush disabled).

impl<S: StorageEngine> NodeDbLite<S>

pub fn crdt_snapshot_export_count(&self) -> u64

Number of full CRDT snapshot exports performed since this handle was opened.

A snapshot export costs O(document), so this is the term that decides both flush latency and how much the file grows per tick. It is a counter, not a timing, so a caller can assert on it directly: an idle store must not advance it.

pub fn crdt_delta_write_count(&self) -> u64

Number of unsent-delta queue entries written since this handle was opened.

The queue is append-only, so this advances by what was added, not by the queue’s length. An idle store must not advance it at all.

pub async fn flush(&self) -> NodeDbResult<()>

Persist all in-memory state to storage (call before shutdown).

impl<S: StorageEngine> NodeDbLite<S>

pub async fn open(storage: S) -> NodeDbResult<Arc<Self>>

Open or create a Lite database backed by the given storage engine.

Configuration is resolved from environment variables via LiteConfig::from_env(), falling back to defaults when variables are absent or malformed. The default auto-flush interval is one second. The timer schedules flushes without a hard completion deadline.

The instance’s Loro peer id is not a parameter: it is minted on first open and persisted with the data it authors. A caller-supplied id is the same constant in every install of an application, which hands two live replicas one producer identity and has the CRDT merge discard one of them; and an id that lives only in the caller’s argument cannot be rotated durably when Origin refuses it.

pub async fn open_with_config( storage: S, config: LiteConfig, ) -> NodeDbResult<Arc<Self>>

Open with an explicit LiteConfig.

This is the primary constructor for callers that need control over memory budgets or the background maintenance intervals. The config is validated before any storage work happens, so an incoherent budget is rejected rather than silently over-allocating.

pub async fn open_with_budget( storage: S, memory_budget: usize, ) -> NodeDbResult<Arc<Self>>

Open with a custom memory budget, taking every other setting from LiteConfig::default() — including the default auto-flush interval.

Prefer open_with_config for new callers.

impl NodeDbLite<PagedbStorageDefault>

pub async fn open_at_path( path: impl AsRef<Path>, encryption: Encryption, ) -> NodeDbResult<Arc<Self>>

Open (or create) a Lite database at path.

A store that cannot be read is reported and left untouched — the FailClosed default. Use open_at_path_with_config with corruption_policy set to choose otherwise.

pub async fn open_at_path_with_config( path: impl AsRef<Path>, encryption: Encryption, config: LiteConfig, ) -> NodeDbResult<Arc<Self>>

Like open_at_path, but with an explicit LiteConfig.

config.corruption_policy governs both corruption points. Under DiscardStoreAndRecreate a post-open corruption renames the store aside and retries the open exactly once; the retry is unguarded, so a second corruption surfaces to the caller rather than looping.

impl<S: StorageEngine> NodeDbLite<S>

pub fn set_sync_gate(&self, gate: Arc<dyn SyncGate>)

Install a per-document SyncGate. Documents the gate rejects are kept local-only (excluded from CRDT delta, FTS, and vector sync channels).

pub fn pending_crdt_deltas(&self) -> NodeDbResult<Vec<PendingDelta>>

Access pending CRDT deltas (for sync client).

pub fn acknowledge_deltas(&self, acked_id: u64) -> NodeDbResult<()>

Acknowledge synced deltas (called after Origin ACK).

pub fn set_crdt_pending_delta_seq( &self, mutation_id: u64, seq: u64, ) -> NodeDbResult<()>

Assign a stable stream seq to a pending CRDT delta (first-send only).

No-op if the delta already has a non-zero seq. Called by the sync transport to ensure the same seq is reused on reconnect re-sends.

pub fn import_remote_deltas( &self, collection: &str, data: &[u8], ) -> NodeDbResult<()>

Import remote deltas from Origin into collection’s Loro document.

The collection must be supplied by the caller: each collection has its own document, and the update bytes alone do not identify which one.

pub async fn apply_remote_row( &self, collection: &str, document_id: &str, payload: &[u8], delete: bool, ) -> NodeDbResult<()>

Apply a server-originated row post-image from Origin.

Unlike Self::import_remote_deltas, the payload here is a MessagePack row image rather than Loro update bytes. Origin sends it for writes that have no client-authored CRDT operation to replicate (SQL DML, DDL-managed system rows). An empty payload with delete set removes the row.

A KV collection’s row is the {key, value…} row every KV read returns. It lands in the KV store as the body a local write of the same row stores. Any other collection’s row lands in its Loro document.

Neither write reaches the outbound queue: the write came FROM Origin, so pushing it back would echo it into a loop.

pub fn reject_delta(&self, mutation_id: u64) -> NodeDbResult<()>

Reject a specific delta (rollback optimistic local state).

pub fn start_sync( self: &Arc<Self>, config: SyncConfig, ) -> Option<Arc<SyncClient>>

Start background sync to Origin.

Spawns a task that connects to the Origin WebSocket endpoint, pushes pending deltas, and receives shape updates, reconnecting on its own.

Returns immediately — the sync runs in the background. Returns None when sync is already running on this database: a second loop would push the same deltas twice, and the first would no longer be the one stop_sync stops.

The task is registered with the database, so stop_sync and shutdown both stop it.

Requires the sync feature.

pub async fn stop_sync(&self) -> bool

Stop background sync, leaving the database open and usable.

Returns true when sync was running. Sync can be restarted afterwards with a fresh start_sync — the reconnect-with-a- new-token path.

Requires the sync feature.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn strict_insert( &self, collection: &str, values: &[Value], ) -> NodeDbResult<()>

Insert a row into a strict collection and update secondary indexes.

Combines StrictEngine.insert() with index_row() for geometry, vector, and text columns.

pub async fn strict_delete( &self, collection: &str, pk: &Value, ) -> NodeDbResult<bool>

Delete a row from a strict collection and clean up text indexes.

pub async fn columnar_insert( &self, collection: &str, values: &[Value], ) -> NodeDbResult<()>

Insert a row into a columnar collection and update secondary indexes.

pub async fn strict_crdt_patch( &self, collection: &str, pk: &Value, field_updates: &HashMap<String, Value>, ) -> NodeDbResult<()>

Apply a CRDT field-level update to a strict collection row.

Used during sync: a remote delta specifies field changes for a row. This reads the current tuple, patches the fields, and writes back.

impl<S: StorageEngine> NodeDbLite<S>

pub fn update_memory_stats(&self)

Update memory governor with current engine usage.

pub fn loaded_collections(&self) -> NodeDbResult<Vec<String>>

List currently loaded HNSW collections.

pub fn governor(&self) -> &Arc<MemoryGovernor>

Access the memory governor.

pub fn sequences(&self) -> &Arc<LiteSequenceRegistry>

Access the sequence registry that nextval / currval / setval in a SELECT list read. Register a LiteSequenceDef here before querying it.

pub fn strict_engine(&self) -> &Arc<StrictEngine<S>>

Access the strict document engine (for direct Binary Tuple CRUD).

pub fn columnar_engine(&self) -> &Arc<ColumnarEngine<S>>

Access the columnar analytics engine (for direct segment operations).

pub fn htap_bridge(&self) -> &Arc<HtapBridge>

Access the HTAP bridge (for materialized view inspection).

pub fn timeseries_engine(&self) -> &Arc<Mutex<TimeseriesEngine>>

Access the timeseries engine (continuous aggregates, ingest, flush).

impl<S: StorageEngine> NodeDbLite<S>

pub fn peer_id(&self) -> u64

The Loro peer id every local operation is authored under.

pub async fn rotate_peer_id(&self) -> NodeDbResult<u64>

Adopt a new Loro peer id and re-author every document under it.

Called when Origin refuses a delta because its peer id belongs to another replica. Continuing under the refused id is not an option: the server refuses every subsequent write, and were it not to, the CRDT merge would discard them as replays of the owning replica’s operations.

The new id is persisted before the documents adopt it, so a crash mid-rotation cannot leave the store authoring under an id its own identity record no longer claims. The rebuilt documents are then flushed, because they exist only in memory until they are — and a restart that reloaded the pre-rotation snapshots would resurrect the operations the rotation exists to abandon.

Every row is queued for re-push: the rebuilt documents share no history with what Origin holds, so the rotation is a resync, not a resume.

pub async fn regenerate_identity(&self) -> NodeDbResult<()>

Replace the whole producer identity and re-author every document.

Called when Origin reports this instance as forked: the history it holds under this lite_id diverged from the one here, so the instance cannot resume the producer stream. A new lite_id and epoch make it a distinct producer, and the peer id goes with them — keeping it would collide with the history the fork was detected against, which is the same refusal one step later.

impl<S: StorageEngine> NodeDbLite<S>

pub fn spatial_insert( &self, collection: &str, field: &str, doc_id: &str, geometry: &Geometry, )

Index a geometry in a collection’s spatial index.

field identifies which geometry field is being indexed (allows a collection to carry multiple spatial fields). If the document was previously indexed under the same (collection, doc_id), the old entry is replaced (upsert semantics).

pub fn spatial_delete(&self, collection: &str, field: &str, doc_id: &str)

Remove a document’s geometry from the spatial index.

pub fn spatial_search_bbox( &self, collection: &str, field: &str, query: &BoundingBox, ) -> Vec<RTreeEntry>

Bounding-box range search: returns all doc entry IDs whose bbox intersects the query rectangle.

pub fn spatial_nearest( &self, collection: &str, field: &str, lng: f64, lat: f64, k: usize, ) -> Vec<NnResult>

Nearest-neighbor search: returns the k closest spatial entries to the given (lng, lat) point.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn shutdown(&self)

Stop every background task this database started, and wait for them.

Call this before dropping the async runtime the database was opened on. Auto-flush, auto-compact and the sync loop each outlive the call that started them; a runtime torn down while one is still polling takes the task down mid-poll, which on a native runtime has crashed the process.

Call it also before reopening the same store in-process. A background flush in flight holds the database, and so the store, open past the drop of the last handle, and the reopen fails with AlreadyOpen.

Each task is signalled first and leaves its loop at a point it chose. One that ignores the signal past TASK_STOP_TIMEOUT is aborted. Admitted search DDL completes before this returns.

Idempotent, and safe on a database that started no tasks. It does not flush: what is in memory at shutdown is still in memory, so call flush first when the data must be durable.

impl NodeDbLite<PagedbStorageDefault>

pub async fn snapshot_to(&self, path: impl AsRef<Path>) -> NodeDbResult<()>

Flush pending state and export a native snapshot to a new directory.

Quiesce concurrent writes for one application-level point in time. PageDB pins its published state and blocks storage writes during export. This API is unavailable on WASM.

pub async fn restore_from( source: impl AsRef<Path>, destination: impl AsRef<Path>, encryption: Encryption, config: LiteConfig, ) -> NodeDbResult<Arc<Self>>

Restore a native snapshot into a new writable destination and retain Lite replica identity.

Supply the source encryption mode and KDF costs. Passphrase restores create a fresh destination salt. PageDB forks its storage identity through rekeying. Restore always uses fail-closed corruption handling. This API is unavailable on WASM.

impl<S: StorageEngine> NodeDbLite<S>

pub fn sparse_insert( &self, collection: &str, field: &str, doc_id: &str, entries: &[(u32, f32)], ) -> NodeDbResult<()>

Index a sparse vector under (collection, field).

entries are (dimension, weight) pairs; they are sorted, deduplicated and validated by SparseVector. Re-indexing the same doc_id replaces its postings rather than adding to them.

pub fn sparse_delete(&self, collection: &str, field: &str, doc_id: &str) -> bool

Remove a document from one (collection, field) sparse index.

Returns true when the document was indexed.

pub fn sparse_search( &self, collection: &str, field: &str, query_entries: &[(u32, f32)], top_k: usize, ) -> NodeDbResult<Vec<(String, f32)>>

Top-k documents in (collection, field) by sparse dot product.

Results are (doc_id, score) ordered by score descending, ties broken by ascending doc_id. Documents sharing no dimension with the query score zero and are excluded. A collection with no sparse index yields no hits rather than an error.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn put_function(&self, func: &LiteStoredFunction) -> NodeDbResult<()>

Store a function definition in the local catalog.

pub async fn get_function( &self, name: &str, ) -> NodeDbResult<Option<LiteStoredFunction>>

Get a function definition by name.

pub async fn list_functions(&self) -> NodeDbResult<Vec<LiteStoredFunction>>

List all function definitions.

pub async fn delete_function(&self, name: &str) -> NodeDbResult<()>

Delete a function definition.

pub async fn put_trigger(&self, trigger: &LiteStoredTrigger) -> NodeDbResult<()>

Store a trigger definition in the local catalog.

pub async fn get_trigger( &self, name: &str, ) -> NodeDbResult<Option<LiteStoredTrigger>>

Get a trigger definition by name.

pub async fn list_triggers(&self) -> NodeDbResult<Vec<LiteStoredTrigger>>

List all trigger definitions.

pub async fn list_triggers_for_collection( &self, collection: &str, ) -> NodeDbResult<Vec<LiteStoredTrigger>>

List triggers for a specific collection.

pub async fn delete_trigger(&self, name: &str) -> NodeDbResult<()>

Delete a trigger definition.

pub async fn put_procedure( &self, proc: &LiteStoredProcedure, ) -> NodeDbResult<()>

Store a procedure definition in the local catalog.

pub async fn get_procedure( &self, name: &str, ) -> NodeDbResult<Option<LiteStoredProcedure>>

Get a procedure definition by name.

pub async fn list_procedures(&self) -> NodeDbResult<Vec<LiteStoredProcedure>>

List all procedure definitions.

pub async fn delete_procedure(&self, name: &str) -> NodeDbResult<()>

Delete a procedure definition.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn diagnostic_dump(&self) -> DiagnosticDump

Generate a diagnostic dump suitable for bug reports.

Contains NO user data, NO document contents, NO embeddings. Safe to send to support channels.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn graph_rag_search( &self, params: &GraphRagParams<'_>, ) -> NodeDbResult<Vec<SearchResult>>

Fuse vector retrieval with bounded, outgoing graph expansion.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn hybrid_search( &self, params: &HybridSearchParams<'_>, ) -> NodeDbResult<Vec<SearchResult>>

Hybrid search: vector similarity + BM25 text relevance fused via RRF.

1. Vector search returns vector_k candidates by embedding similarity.
2. Text search returns text_k candidates by BM25 relevance.
3. RRF fuses both rankings into a single score per document.
4. Top top_k results returned.

Documents found by both searches get boosted (RRF scores are additive).

impl<S: StorageEngine> NodeDbLite<S>

pub fn storage(&self) -> &S

Borrow the underlying storage engine.

Public so benchmark code can call backend-specific methods like backend-specific methods (e.g. size reporting) for compression-ratio measurement.

pub async fn compact(&self) -> NodeDbResult<CompactionOutcome>

Compact the backing storage engine, reclaiming dead pages and truncating the file to bound on-disk growth.

Forwards to StorageEngine::compact. For the pagedb-backed engine this drains the deferred-free list and truncates main.db; for in-memory or test engines it is a no-op returning a zero CompactionOutcome.

pub fn health(&self) -> HealthStatus

Get a structured health report.

This is a cheap, non-blocking call — reads atomic counters and lock-free state. Safe to call frequently from health check endpoints.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn document_put_with_vector_batch_impl( &self, items: &[BatchItem<'_>], ) -> NodeDbResult<Vec<String>>

Batch upsert of documents with optional embeddings.

Acquires the CRDT lock once and runs every per-item Loro mutation under that single hold. Per CrdtEngine::batch_write, each CRDT row still exports its own delta — one per document, plus one per item with a non-empty embedding for the vector-metadata row — because a delta covering several rows or collections is not independently applicable by the receiver.

FTS indexing, bitemporal history, and HNSW inserts happen after the lock is released, in the same relative order as the single-item path.

Returns the list of document IDs written, in input order.

impl<S: StorageEngine> NodeDbLite<S>

pub async fn vector_delete_field( &self, collection: &str, field_name: &str, id: &str, ) -> NodeDbResult<()>

Delete the named vector field_name of id in collection, keeping the row’s document fields and any other vector attached to it. An empty field_name names the base vector, as vector_delete does.

Trait Implementations

impl<S: StorageEngine> NodeDb for NodeDbLite<S>

fn batch_graph_insert_edges<'life0, 'life1, 'life2, 'life3, 'life4, 'life5, 'async_trait>( &'life0 self, collection: &'life1 str, edges: &'life2 [(&'life3 str, &'life4 str, &'life5 str)], ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, 'life5: 'async_trait,

Overrides the trait’s default (which loops graph_insert_edge) with the batched CSR-lock-once, single-CRDT-write path. edge_type and properties are not part of this trait signature, so every edge is inserted with no properties, matching what looping graph_insert_edge with properties: None would have written.

fn vector_search<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, collection: &'life1 str, query: &'life2 [f32], k: usize, filter: Option<&'life3 MetadataFilter>, allowed_ids: Option<&'life4 HashSet<String>>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<Vec<SearchResult>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait,

Search for the k nearest vectors to query in collection.

fn vector_insert<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, collection: &'life1 str, id: &'life2 str, embedding: &'life3 [f32], metadata: Option<Document>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Insert a vector with optional metadata into collection.

fn vector_delete<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, collection: &'life1 str, id: &'life2 str, ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Delete a vector by ID from collection.

fn vector_insert_field<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, collection: &'life1 str, field_name: &'life2 str, id: &'life3 str, embedding: &'life4 [f32], metadata: Option<Document>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait,

Insert a vector into the named field field_name of collection.

fn vector_search_field<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, collection: &'life1 str, field_name: &'life2 str, query: &'life3 [f32], k: usize, filter: Option<&'life4 MetadataFilter>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<Vec<SearchResult>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait,

Search the named vector field field_name.

fn graph_traverse<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, collection: &'life1 str, start: &'life2 NodeId, depth: u8, direction: Direction, edge_filter: Option<&'life3 EdgeFilter>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<SubGraph>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Traverse the graph from start up to depth hops within collection.

fn graph_insert_edge<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, collection: &'life1 str, from: &'life2 NodeId, to: &'life3 NodeId, edge_type: &'life4 str, properties: Option<Document>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<EdgeId>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait,

Insert a directed edge from → to with label edge_type into collection. Returns the generated edge ID.

fn graph_delete_edge<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, collection: &'life1 str, edge_id: &'life2 EdgeId, ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Delete a graph edge by ID from collection.

fn graph_stats<'life0, 'life1, 'async_trait>( &'life0 self, collection: Option<&'life1 str>, as_of: Option<i64>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<Vec<GraphStats>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Read aggregated graph statistics from the persisted edge store.

fn graph_pagerank<'life0, 'life1, 'async_trait>( &'life0 self, collection: &'life1 str, personalization: Option<HashMap<String, f64>>, damping: Option<f64>, max_iterations: Option<u32>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<Vec<(String, f64)>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Run PageRank on a graph collection’s edges.

fn graph_shortest_path<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, collection: &'life1 str, from: &'life2 NodeId, to: &'life3 NodeId, max_depth: u8, edge_filter: Option<&'life4 EdgeFilter>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<Option<Vec<NodeId>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait,

Find the shortest path between two nodes.

fn list_insert<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, collection: &'life1 str, document_id: &'life2 str, list_path: &'life3 str, index: usize, fields: &'life4 Value, ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait,

Insert a new block into a document’s movable list at index.

fn list_delete<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, collection: &'life1 str, document_id: &'life2 str, list_path: &'life3 str, index: usize, ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Delete the block at index from a document’s movable list.

fn list_move<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, collection: &'life1 str, document_id: &'life2 str, list_path: &'life3 str, from_index: usize, to_index: usize, ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Move the block at from_index to to_index in a document’s movable list.

fn document_get<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, collection: &'life1 str, id: &'life2 str, ) -> Pin<Box<dyn Future<Output = NodeDbResult<Option<Document>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Get a document by ID from collection.

fn document_put<'life0, 'life1, 'async_trait>( &'life0 self, collection: &'life1 str, doc: Document, ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Put (insert or replace) a document into collection.

fn document_delete<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, collection: &'life1 str, id: &'life2 str, ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Delete a document by ID from collection.

fn document_put_with_vector<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, doc_collection: &'life1 str, doc: Document, vector_collection: &'life2 str, id: &'life3 str, embedding: &'life4 [f32], ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait,

Put a document and insert its embedding under one CRDT lock acquisition.

fn document_get_as_of<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, collection: &'life1 str, id: &'life2 str, as_of_ms: Option<i64>, valid_time_ms: Option<i64>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<Option<Document>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Read a document as of a system time, optionally at a valid time.

fn document_put_with_valid_time<'life0, 'life1, 'async_trait>( &'life0 self, collection: &'life1 str, doc: Document, valid_from_ms: Option<i64>, valid_until_ms: Option<i64>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Put a document with explicit valid-time bounds.

fn execute_sql<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, query: &'life1 str, params: &'life2 [Value], ) -> Pin<Box<dyn Future<Output = NodeDbResult<QueryResult>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Execute a raw SQL query with parameters.

fn text_search<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, collection: &'life1 str, field: &'life2 str, query: &'life3 str, top_k: usize, params: TextSearchParams, allowed_ids: Option<&'life4 HashSet<String>>, ) -> Pin<Box<dyn Future<Output = NodeDbResult<Vec<SearchResult>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait,

Full-text BM25 search of the indexed field of collection.

fn list_dropped_collections<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = NodeDbResult<Vec<DroppedCollection>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

List the current tenant’s soft-deleted collections still within their retention window. Empty when there are none.

fn batch_vector_insert<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, collection: &'life1 str, vectors: &'life2 [(&'life3 str, &'life4 [f32])], ) -> Pin<Box<dyn Future<Output = Result<(), NodeDbError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, Self: Sync + 'async_trait,

Batch insert vectors — amortizes CRDT delta export to O(1) per batch.

fn proto_version(&self) -> u16

The protocol version negotiated during the connection handshake. 0 for an implementation that performs no handshake.

fn capabilities(&self) -> u64

The raw capability bitfield advertised by the server. 0 when no handshake ran. Capabilities::from_raw gives named predicates.

fn server_version(&self) -> String

The server version string from HelloAckFrame (e.g. "0.1.0-dev"). Empty when no handshake ran.

fn limits(&self) -> Limits

Per-operation limits announced by the server. Every field is None when no handshake ran. None means no server-side cap.

fn undrop_collection<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), NodeDbError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: Sync + 'async_trait,

Restore a soft-deleted collection within its retention window.

fn drop_collection_purge<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), NodeDbError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: Sync + 'async_trait,

Hard-delete a collection, skipping soft-delete and retention.

fn on_collection_purged<'life0, 'async_trait>( &'life0 self, _handler: Arc<dyn Fn(CollectionPurgedEvent) + Send + Sync>, ) -> Pin<Box<dyn Future<Output = Result<(), NodeDbError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: Sync + 'async_trait,

Register a handler fired when a collection the caller synced is purged on Origin and the local copy is removed.

impl<S: StorageEngine> SyncDelegate for NodeDbLite<S>

fn sync_identity(&self) -> LiteIdentity

This instance’s durable identity: lite_id, epoch, and the Loro peer id local operations are authored under.

fn regenerate_identity<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Replace this instance’s producer identity after Origin reports a fork, re-authoring local state under it.

fn rotate_peer_id<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Adopt a new Loro peer id after Origin refuses the current one as another replica’s, re-authoring local state under it.

fn pending_deltas(&self) -> Vec<PendingDelta>

Get all pending CRDT deltas to push to Origin.

fn set_pending_delta_seq<'life0, 'async_trait>( &'life0 self, mutation_id: u64, seq: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Assign a stable stream seq to a pending CRDT delta (first-send only).

fn acknowledge(&self, mutation_id: u64)

Acknowledge deltas up to the given mutation_id.

fn reject(&self, mutation_id: u64)

Reject a specific delta (rollback optimistic state).

fn reject_with_policy(&self, mutation_id: u64, hint: &CompensationHint)

Reject a delta with policy-aware resolution. Consults the PolicyRegistry before deciding how to handle the rejection.

fn import_remote(&self, collection: &str, data: &[u8])

Import remote deltas from Origin into collection’s local CRDT state.

fn apply_remote_row<'life0, 'life1, 'async_trait>( &'life0 self, msg: &'life1 RowPushMsg, ) -> Pin<Box<dyn Future<Output = Result<(), NodeDbError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Apply a server-originated row post-image (opcode 0x15).

fn handle_array_delta<'life0, 'life1, 'async_trait>( &'life0 self, msg: &'life1 ArrayDeltaMsg, ) -> Pin<Box<dyn Future<Output = Option<ArrayAckMsg>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Apply a single ArrayDelta frame from Origin.

fn handle_array_delta_batch<'life0, 'life1, 'async_trait>( &'life0 self, msg: &'life1 ArrayDeltaBatchMsg, ) -> Pin<Box<dyn Future<Output = Option<ArrayAckMsg>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Apply a batch of ArrayDelta frames from Origin.

fn handle_array_reject<'life0, 'life1, 'async_trait>( &'life0 self, msg: &'life1 ArrayRejectMsg, ) -> Pin<Box<dyn Future<Output = Option<ArrayCatchupRequestMsg>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Process an ArrayReject from Origin.

fn handle_array_snapshot<'life0, 'life1, 'async_trait>( &'life0 self, msg: &'life1 ArraySnapshotMsg, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Buffer an ArraySnapshot header from Origin. Errors are logged.

fn handle_array_snapshot_chunk<'life0, 'life1, 'async_trait>( &'life0 self, msg: &'life1 ArraySnapshotChunkMsg, ) -> Pin<Box<dyn Future<Output = Option<ArrayAckMsg>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Buffer an ArraySnapshotChunk from Origin.

fn begin_array_catchup(&self) -> Vec<ArrayCatchupRequestMsg>

Start the array catch-up round for a newly connected session.

fn pending_array_ops<'life0, 'async_trait>( &'life0 self, after: Option<Hlc>, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<PendingArrayOp>, LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Up to limit queued local array ops with an HLC strictly greater than after, in HLC order. None starts at the head of the queue.

fn persist_array_op_seq<'life0, 'life1, 'async_trait>( &'life0 self, op: &'life1 PendingArrayOp, ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Persist op.seq into the queued op before its first send.

fn retire_array_ops<'life0, 'life1, 'async_trait>( &'life0 self, hlcs: &'life1 [Hlc], ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Remove the queued array ops named in hlcs: ops Origin applied, and ops the push loop cannot encode. Ops no longer queued are skipped.

fn array_schema_announce( &self, array: &str, ) -> Result<Option<ArraySchemaSyncMsg>, LiteError>

The ArraySchema body announcing array’s current schema to Origin: the schema HLC and Loro snapshot, read together. Returns None when array has no local schema.

fn pending_columnar_batches<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Vec<(Vec<u8>, PendingColumnarBatch)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Drain up to PUSH_DRAIN_LIMIT pending batches from durable storage, skipping any currently in-flight entries.

fn mark_columnar_batch_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Record that a columnar batch has been sent and is awaiting Origin ack.

fn ack_columnar_batch_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

On Origin ack: remove in-flight record and delete the durable entry.

fn acknowledge_columnar_batch<'life0, 'async_trait>( &'life0 self, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Delete the durable entry for a confirmed batch directly (used for un-encodable entries that must be discarded at send time).

fn pending_vector_inserts<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Vec<(Vec<u8>, PendingVectorInsert)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Drain up to PUSH_DRAIN_LIMIT pending insert entries, skipping in-flight.

fn mark_vector_insert_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Record that a vector insert has been sent and is awaiting Origin ack.

fn ack_vector_insert_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

On Origin ack: remove in-flight record and delete the durable entry.

fn acknowledge_vector_insert<'life0, 'async_trait>( &'life0 self, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Delete the durable insert entry directly (for un-encodable entries).

fn pending_vector_deletes<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Vec<(Vec<u8>, PendingVectorDelete)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Drain up to PUSH_DRAIN_LIMIT pending delete entries, skipping in-flight.

fn mark_vector_delete_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Record that a vector delete has been sent and is awaiting Origin ack.

fn ack_vector_delete_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

On Origin ack: remove in-flight record and delete the durable entry.

fn acknowledge_vector_delete<'life0, 'async_trait>( &'life0 self, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Delete the durable delete entry directly (for un-encodable entries).

fn pending_fts_indexes<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<(Vec<u8>, PendingFtsIndex)>, LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Drain up to PUSH_DRAIN_LIMIT pending index entries, skipping in-flight.

fn mark_fts_index_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Record that an FTS index entry has been sent and is awaiting Origin ack.

fn ack_fts_index_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

On Origin ack: remove in-flight record and delete the durable entry.

fn acknowledge_fts_index<'life0, 'async_trait>( &'life0 self, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Delete the durable index entry directly (for un-encodable entries).

fn pending_fts_deletes<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<(Vec<u8>, PendingFtsDelete)>, LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Drain up to PUSH_DRAIN_LIMIT pending FTS delete entries, skipping in-flight.

fn mark_fts_delete_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Record that an FTS delete entry has been sent and is awaiting Origin ack.

fn ack_fts_delete_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

On Origin ack: remove in-flight record and delete the durable entry.

fn acknowledge_fts_delete<'life0, 'async_trait>( &'life0 self, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Delete the durable delete entry directly (for un-encodable entries).

fn pending_spatial_inserts<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Vec<(Vec<u8>, PendingSpatialInsert)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Drain up to PUSH_DRAIN_LIMIT pending insert entries, skipping in-flight.

fn mark_spatial_insert_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Record that a spatial insert has been sent and is awaiting Origin ack.

fn ack_spatial_insert_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

On Origin ack: remove in-flight record and delete the durable entry.

fn acknowledge_spatial_insert<'life0, 'async_trait>( &'life0 self, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Delete the durable insert entry directly (for un-encodable entries).

fn pending_spatial_deletes<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Vec<(Vec<u8>, PendingSpatialDelete)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Drain up to PUSH_DRAIN_LIMIT pending spatial delete entries, skipping in-flight.

fn mark_spatial_delete_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Record that a spatial delete has been sent and is awaiting Origin ack.

fn ack_spatial_delete_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

On Origin ack: remove in-flight record and delete the durable entry.

fn acknowledge_spatial_delete<'life0, 'async_trait>( &'life0 self, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Delete the durable delete entry directly (for un-encodable entries).

fn pending_timeseries_batches<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Vec<(Vec<u8>, PendingTimeseriesBatch)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Drain up to PUSH_DRAIN_LIMIT pending batches, skipping in-flight entries.

fn mark_timeseries_batch_in_flight<'life0, 'async_trait>( &'life0 self, stream_seq: u64, batch_id: u64, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Record that a timeseries batch has been sent and is awaiting Origin ack.

fn ack_timeseries_batches_through_seq<'life0, 'async_trait>( &'life0 self, applied_seq: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

On Origin ack: delete all durable entries whose seq ≤ applied_seq.

fn ack_timeseries_batch_by_id<'life0, 'async_trait>( &'life0 self, batch_id: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

On terminal rejection: delete the durable entry for exactly batch_id.

fn acknowledge_timeseries_batch<'life0, 'async_trait>( &'life0 self, durable_key: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Delete the durable entry directly (for empty/un-encodable batches).

fn pending_kv_writes<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<(Vec<u8>, PendingKvWrite)>, LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Up to PUSH_DRAIN_LIMIT pending KV writes in queue order, skipping the ones in flight. Returns (durable_key, write) pairs.

fn persist_kv_write_seq<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, key: &'life1 [u8], write: &'life2 PendingKvWrite, ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Persist the seq assigned to the KV write stored under key.

fn mark_kv_write_in_flight<'life0, 'async_trait>( &'life0 self, batch_id: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Record that the KV write with batch_id was sent and awaits its ack.

fn retire_kv_write<'life0, 'async_trait>( &'life0 self, batch_id: u64, ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Retire the KV write with batch_id after Origin’s final answer: forget it is in flight and delete its durable entry.

fn clear_engine_in_flight<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Clear all engine in-flight maps on reconnect.

fn next_stream_seq<'life0, 'async_trait>( &'life0 self, stream_id: u64, ) -> Pin<Box<dyn Future<Output = u64> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Assign the next monotonic sequence number for the given stream_id.

fn record_stream_ack<'life0, 'async_trait>( &'life0 self, stream_id: u64, applied_seq: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Record that Origin has applied applied_seq for stream_id.

fn persist_producer_state<'life0, 'async_trait>( &'life0 self, producer_id: u64, accepted_epoch: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Durably persist the server-assigned producer_id and accepted_epoch so they survive process restart and can be reloaded on reconnect.

fn load_producer_state<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = (u64, u64)> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Load the last-persisted producer_id and accepted_epoch.

fn import_definition<'life0, 'life1, 'async_trait>( &'life0 self, msg: &'life1 DefinitionSyncMsg, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Import a definition sync message (function/trigger/procedure) from Origin. Async because persisting the definition to storage involves KV store writes through spawn_blocking.

fn import_collection_schema<'life0, 'life1, 'async_trait>( &'life0 self, msg: &'life1 CollectionSchemaSyncMsg, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Materialize a collection announced by a sync peer (opcode 0x13).

fn get_collection_meta<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Option<CollectionMeta>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Look up a collection’s persisted metadata by name, for building the outbound CollectionSchema (opcode 0x13) announce sent before a collection’s first CRDT delta in a sync session.

fn persist_columnar_seq<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, key: &'life1 [u8], batch: &'life2 PendingColumnarBatch, ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Persist an assigned stream seq into the durable columnar entry at key.

fn persist_timeseries_seq<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, key: &'life1 [u8], batch: &'life2 PendingTimeseriesBatch, ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Persist an assigned stream seq into the durable timeseries entry at key.

fn persist_vector_insert_seq<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, key: &'life1 [u8], insert: &'life2 PendingVectorInsert, ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Persist an assigned stream seq into the durable vector insert entry at key.

fn persist_vector_delete_seq<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, key: &'life1 [u8], delete: &'life2 PendingVectorDelete, ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Persist an assigned stream seq into the durable vector delete entry at key.

fn persist_fts_index_seq<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, key: &'life1 [u8], entry: &'life2 PendingFtsIndex, ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Persist an assigned stream seq into the durable FTS index entry at key.

fn persist_fts_delete_seq<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, key: &'life1 [u8], entry: &'life2 PendingFtsDelete, ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Persist an assigned stream seq into the durable FTS delete entry at key.

fn persist_spatial_insert_seq<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, key: &'life1 [u8], insert: &'life2 PendingSpatialInsert, ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Persist an assigned stream seq into the durable spatial insert entry at key.

fn persist_spatial_delete_seq<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, key: &'life1 [u8], delete: &'life2 PendingSpatialDelete, ) -> Pin<Box<dyn Future<Output = Result<(), LiteError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Persist an assigned stream seq into the durable spatial delete entry at key.