Method save

nodedb/src/event/watermark.rs:100–128

Persist the last-processed LSN for a given core.

(&self, core_id: usize, lsn: Lsn)

    /// Persist the last-processed LSN for a given core.
    pub fn save(&self, core_id: usize, lsn: Lsn) -> crate::Result<()> {
        let key = core_key(core_id);
        let value = lsn.as_u64().to_le_bytes();

        let txn = self.db.begin_write().map_err(|e| crate::Error::Storage {
            engine: "event_plane".into(),
            detail: format!("begin_write: {e}"),
        })?;
        {
            let mut table = txn
                .open_table(WATERMARKS)
                .map_err(|e| crate::Error::Storage {
                    engine: "event_plane".into(),
                    detail: format!("open_table: {e}"),
                })?;
            table
                .insert(key.as_str(), value.as_slice())
                .map_err(|e| crate::Error::Storage {
                    engine: "event_plane".into(),
                    detail: format!("insert watermark core {core_id}: {e}"),
                })?;
        }
        txn.commit().map_err(|e| crate::Error::Storage {
            engine: "event_plane".into(),
            detail: format!("commit watermark core {core_id}: {e}"),
        })?;

        Ok(())
    }

    /// Load watermarks for all cores (0..num_cores).
    pub fn load_all(&self, num_cores: usize) -> crate::Result<Vec<Lsn>> {