Open or create the watermark store at `{data_dir}/event_plane/watermarks.redb`.
(data_dir: &Path)
| 27 | impl WatermarkStore { |
| 28 | /// Open or create the watermark store at `{data_dir}/event_plane/watermarks.redb`. |
| 29 | pub fn open(data_dir: &Path) -> crate::Result<Self> { |
| 30 | let dir = data_dir.join("event_plane"); |
| 31 | std::fs::create_dir_all(&dir).map_err(|e| crate::Error::Storage { |
| 32 | engine: "event_plane".into(), |
| 33 | detail: format!("create dir {}: {e}", dir.display()), |
| 34 | })?; |
| 35 | |
| 36 | let path = dir.join("watermarks.redb"); |
| 37 | let db = Database::create(&path).map_err(|e| crate::Error::Storage { |
| 38 | engine: "event_plane".into(), |
| 39 | detail: format!("open watermark db {}: {e}", path.display()), |
| 40 | })?; |
| 41 | |
| 42 | // Ensure table exists. |
| 43 | { |
| 44 | let txn = db.begin_write().map_err(|e| crate::Error::Storage { |
| 45 | engine: "event_plane".into(), |
| 46 | detail: format!("begin_write: {e}"), |
| 47 | })?; |
| 48 | txn.open_table(WATERMARKS) |
| 49 | .map_err(|e| crate::Error::Storage { |
| 50 | engine: "event_plane".into(), |
| 51 | detail: format!("open_table: {e}"), |
| 52 | })?; |
| 53 | txn.commit().map_err(|e| crate::Error::Storage { |
| 54 | engine: "event_plane".into(), |
| 55 | detail: format!("commit: {e}"), |
| 56 | })?; |
| 57 | } |
| 58 | |
| 59 | debug!(path = %path.display(), "watermark store opened"); |
| 60 | Ok(Self { db, path }) |
| 61 | } |
| 62 | |
| 63 | /// Load the last-processed LSN for a given core. Returns `Lsn::ZERO` if |
| 64 | /// no watermark has been persisted yet (first startup). |
no test coverage detected