Method load

nodedb/src/event/watermark.rs:65–97

Load the last-processed LSN for a given core. Returns `Lsn::ZERO` if no watermark has been persisted yet (first startup).

(&self, core_id: usize)

    /// Load the last-processed LSN for a given core. Returns `Lsn::ZERO` if
    /// no watermark has been persisted yet (first startup).
    pub fn load(&self, core_id: usize) -> crate::Result<Lsn> {
        let key = core_key(core_id);
        let txn = self.db.begin_read().map_err(|e| crate::Error::Storage {
            engine: "event_plane".into(),
            detail: format!("begin_read: {e}"),
        })?;
        let table = txn
            .open_table(WATERMARKS)
            .map_err(|e| crate::Error::Storage {
                engine: "event_plane".into(),
                detail: format!("open_table: {e}"),
            })?;

        match table.get(key.as_str()) {
            Ok(Some(guard)) => {
                let bytes = guard.value();
                if bytes.len() == 8 {
                    let arr: [u8; 8] = [
                        bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6],
                        bytes[7],
                    ];
                    Ok(Lsn::new(u64::from_le_bytes(arr)))
                } else {
                    Ok(Lsn::ZERO)
                }
            }
            Ok(None) => Ok(Lsn::ZERO),
            Err(e) => Err(crate::Error::Storage {
                engine: "event_plane".into(),
                detail: format!("get watermark core {core_id}: {e}"),
            }),
        }
    }

    /// Persist the last-processed LSN for a given core.
    pub fn save(&self, core_id: usize, lsn: Lsn) -> crate::Result<()> {