Struct CursorStore 

pub struct CursorStore { /* private fields */ }

Persistent cursor storage backed by SQLite.

Supports debounced writes: updates go to cache immediately, and are flushed to disk periodically via flush_dirty().

Implementations

impl CursorStore

pub async fn new(path: impl AsRef<Path>) -> Result<Self>

Create a new cursor store at the given path.

Creates the database and tables if they don’t exist.

pub async fn get(&self, peer_id: &str) -> Option<String>

Get the cursor for a peer (from cache).

Returns None if no cursor exists (first sync with this peer).

pub async fn get_or_start(&self, peer_id: &str) -> String

Get the cursor, or return “0” for first-time sync.

Redis XREAD interprets “0” as “from the beginning”.

pub async fn set(&self, peer_id: &str, stream_id: &str)

Update the cursor for a peer (debounced).

Updates cache immediately, marks cursor as dirty. Call flush_dirty() periodically to persist to disk.

This is the preferred method for hot path updates where we want to minimize SQLite writes.

pub async fn flush_dirty(&self) -> Result<usize>

Flush all dirty cursors to disk.

Call this periodically (e.g., every 5 seconds) and on shutdown. Returns the number of cursors flushed.

pub async fn has_dirty(&self) -> bool

Check if there are any dirty (unflushed) cursors.

pub async fn dirty_count(&self) -> usize

Get count of dirty cursors pending flush.

pub async fn delete(&self, peer_id: &str) -> Result<()>

Delete cursor for a peer (e.g., when peer is removed from mesh). Retries on SQLITE_BUSY/SQLITE_LOCKED with exponential backoff.

pub async fn get_all(&self) -> HashMap<String, String>

Get all cursors (for metrics/debugging).

pub fn path(&self) -> &str

Get database path (for diagnostics).

pub async fn checkpoint(&self) -> Result<()>

Force flush WAL to main database (for clean shutdown). Retries on SQLITE_BUSY/SQLITE_LOCKED with exponential backoff.

pub async fn close(&self)

Close the connection pool gracefully.

Flushes any dirty cursors and checkpoints WAL before closing.