| 219 | } |
| 220 | |
| 221 | fn compact(&mut self, index: u64, term: u64) -> nodedb_raft::error::Result<()> { |
| 222 | let write_txn = |
| 223 | self.db |
| 224 | .begin_write() |
| 225 | .map_err(|e| nodedb_raft::error::RaftError::Storage { |
| 226 | detail: format!("write txn: {e}"), |
| 227 | })?; |
| 228 | { |
| 229 | // Remove entries <= index. |
| 230 | let mut table = write_txn.open_table(ENTRIES).map_err(|e| { |
| 231 | nodedb_raft::error::RaftError::Storage { |
| 232 | detail: format!("open entries: {e}"), |
| 233 | } |
| 234 | })?; |
| 235 | |
| 236 | let end = index_key(index + 1); |
| 237 | let keys_to_remove: Vec<[u8; 8]> = table |
| 238 | .range(..end.as_slice()) |
| 239 | .map_err(|e| nodedb_raft::error::RaftError::Storage { |
| 240 | detail: format!("range: {e}"), |
| 241 | })? |
| 242 | .filter_map(|r| { |
| 243 | r.ok().map(|(k, _)| { |
| 244 | let mut buf = [0u8; 8]; |
| 245 | buf.copy_from_slice(k.value()); |
| 246 | buf |
| 247 | }) |
| 248 | }) |
| 249 | .collect(); |
| 250 | |
| 251 | for key in &keys_to_remove { |
| 252 | table.remove(key.as_slice()).map_err(|e| { |
| 253 | nodedb_raft::error::RaftError::Storage { |
| 254 | detail: format!("remove: {e}"), |
| 255 | } |
| 256 | })?; |
| 257 | } |
| 258 | |
| 259 | // Save snapshot metadata. |
| 260 | let mut meta = |
| 261 | write_txn |
| 262 | .open_table(META) |
| 263 | .map_err(|e| nodedb_raft::error::RaftError::Storage { |
| 264 | detail: format!("open meta: {e}"), |
| 265 | })?; |
| 266 | |
| 267 | let idx_bytes = zerompk::to_msgpack_vec(&index).map_err(|e| { |
| 268 | nodedb_raft::error::RaftError::Storage { |
| 269 | detail: format!("serialize: {e}"), |
| 270 | } |
| 271 | })?; |
| 272 | let term_bytes = zerompk::to_msgpack_vec(&term).map_err(|e| { |
| 273 | nodedb_raft::error::RaftError::Storage { |
| 274 | detail: format!("serialize: {e}"), |
| 275 | } |
| 276 | })?; |
| 277 | |
| 278 | meta.insert(KEY_SNAPSHOT_INDEX, idx_bytes.as_slice()) |