diff --git a/crates/clawhdf5-agent/src/lib.rs b/crates/clawhdf5-agent/src/lib.rs index b1a5253..151e784 100644 --- a/crates/clawhdf5-agent/src/lib.rs +++ b/crates/clawhdf5-agent/src/lib.rs @@ -629,7 +629,7 @@ impl HDF5Memory { if let Some(existing_idx) = self.cache.find_by_tags(&entry.tags) { if let Some(ref mut w) = self.wal { let wal_entry = wal::WalEntry { - entry_type: wal::WalEntryType::Save, + entry_type: wal::WalEntryType::Update, timestamp: entry.timestamp, chunk: entry.chunk.clone(), embedding: entry.embedding.clone(), @@ -637,6 +637,7 @@ impl HDF5Memory { session_id: entry.session_id.clone(), tags: entry.tags.clone(), tombstone_index: None, + update_index: Some(existing_idx), }; w.append_save(&wal_entry)?; } @@ -668,9 +669,6 @@ impl HDF5Memory { .is_none_or(|w| w.pending_count() as usize > self.config.wal_max_entries); if needs_flush { self.flush()?; - if let Some(ref mut w) = self.wal { - w.truncate()?; - } } return Ok(existing_idx); } @@ -691,6 +689,7 @@ impl AgentMemory for HDF5Memory { session_id: entry.session_id.clone(), tags: entry.tags.clone(), tombstone_index: None, + update_index: None, }; w.append_save(&wal_entry)?; } @@ -716,9 +715,6 @@ impl AgentMemory for HDF5Memory { .is_none_or(|w| w.pending_count() as usize > self.config.wal_max_entries); if needs_flush { self.flush()?; - if let Some(ref mut w) = self.wal { - w.truncate()?; - } } Ok(idx) } @@ -892,9 +888,6 @@ impl HDF5Memory { *w *= d; } self.flush()?; - if let Some(ref mut w) = self.wal { - w.truncate()?; - } Ok(()) } @@ -906,9 +899,6 @@ impl HDF5Memory { /// Explicit WAL merge: flush .h5, truncate WAL. pub fn flush_wal(&mut self) -> Result<()> { self.flush()?; - if let Some(ref mut w) = self.wal { - w.truncate()?; - } Ok(()) } } diff --git a/crates/clawhdf5-agent/src/wal.rs b/crates/clawhdf5-agent/src/wal.rs index aed89e3..dcf2c3b 100644 --- a/crates/clawhdf5-agent/src/wal.rs +++ b/crates/clawhdf5-agent/src/wal.rs @@ -28,7 +28,18 @@ const WAL_HEADER_LEN: u64 = WAL_MAGIC.len() as u64 + 1 + 4; /// its target Save) — the moved/inserted entry's stored CRC was computed /// against a different predecessor than the one now in front of it on disk, /// so the chain breaks at that point and replay stops there. -const WAL_VERSION: u8 = 3; +const WAL_VERSION: u8 = 4; + +/// The chained-CRC format before [`WalEntryType::Update`] records existed. +/// Byte-for-byte the same framing as [`WAL_VERSION`], so it is read by the +/// same code, and `WalFile::open` upgrades it in place by rewriting the +/// header's version byte (the header is not covered by the CRC chain). +/// +/// The bump exists for *older binaries*: they don't know record type 0x04, +/// would treat it as a torn tail, and would truncate it — and everything +/// after it — away. An unknown header version makes them refuse the file +/// with a clear error instead. +const WAL_VERSION_CHAINED_NO_UPDATE: u8 = 3; /// The previous WAL format version: still a CRC32 per entry (so a bit-flip /// within one entry is caught), but not chained to the previous entry's CRC @@ -67,6 +78,10 @@ pub enum WalEntryType { Save = 0x01, Tombstone = 0x02, ActivationUpdate = 0x03, + /// Replace the record at `update_index` in place (`save_or_update` hit). + /// Logged as a plain `Save` before this existed, so replay appended a + /// duplicate instead of updating. + Update = 0x04, } impl WalEntryType { @@ -75,6 +90,7 @@ impl WalEntryType { 0x01 => Some(Self::Save), 0x02 => Some(Self::Tombstone), 0x03 => Some(Self::ActivationUpdate), + 0x04 => Some(Self::Update), _ => None, } } @@ -91,6 +107,8 @@ pub struct WalEntry { pub tags: String, /// For tombstone entries: the index of the entry to delete. pub tombstone_index: Option, + /// For update entries: the index of the record to replace. + pub update_index: Option, } /// How many entries to accumulate before updating the header entry_count. @@ -155,7 +173,15 @@ impl WalFile { let mut ver = [0u8; 1]; f.read_exact(&mut ver)?; match ver[0] { - WAL_VERSION => { + WAL_VERSION | WAL_VERSION_CHAINED_NO_UPDATE => { + if ver[0] == WAL_VERSION_CHAINED_NO_UPDATE { + // Same framing; stamp the current version so an older + // binary refuses this file rather than truncating an + // Update record it can't parse. See the constant. + f.seek(SeekFrom::Start(4))?; + f.write_all(&[WAL_VERSION])?; + f.seek(SeekFrom::Start(5))?; + } let mut count_buf = [0u8; 4]; f.read_exact(&mut count_buf)?; let header_count = u32::from_le_bytes(count_buf); @@ -257,8 +283,20 @@ impl WalFile { 4 + entry.session_id.len() + 4 + entry.tags.len(), ); - buf.push(WalEntryType::Save as u8); - buf.extend_from_slice(&entry.timestamp.to_le_bytes()); + match entry.update_index { + Some(index) => { + let index = u32::try_from(index).map_err(|_| { + MemoryError::Schema(format!("WAL update index {index} exceeds u32")) + })?; + buf.push(WalEntryType::Update as u8); + buf.extend_from_slice(&entry.timestamp.to_le_bytes()); + buf.extend_from_slice(&index.to_le_bytes()); + } + None => { + buf.push(WalEntryType::Save as u8); + buf.extend_from_slice(&entry.timestamp.to_le_bytes()); + } + } serialize_str(&mut buf, &entry.chunk); buf.extend_from_slice(&(emb_len as u32).to_le_bytes()); for &val in &entry.embedding { @@ -375,7 +413,7 @@ impl WalFile { let entry_count_hint = u32::from_le_bytes([header[5], header[6], header[7], header[8]]); match header[4] { - WAL_VERSION => { + WAL_VERSION | WAL_VERSION_CHAINED_NO_UPDATE => { let (entries, _final_crc, _verified_bytes) = read_chained_entries(&mut f, 0, applied); Ok(entries) @@ -492,6 +530,28 @@ pub fn replay_into_cache(entries: &[WalEntry], cache: &mut crate::cache::MemoryC entry.tags.clone(), ); } + WalEntryType::Update => match entry.update_index { + // The index was valid when the record was written; if the + // store no longer has it, keep the data rather than drop it. + Some(idx) if idx < cache.len() => cache.update( + idx, + entry.chunk.clone(), + entry.embedding.clone(), + entry.source_channel.clone(), + entry.timestamp, + entry.session_id.clone(), + ), + _ => { + cache.push( + entry.chunk.clone(), + entry.embedding.clone(), + entry.source_channel.clone(), + entry.timestamp, + entry.session_id.clone(), + entry.tags.clone(), + ); + } + }, WalEntryType::Tombstone => { if let Some(idx) = entry.tombstone_index { cache.mark_deleted(idx); @@ -676,7 +736,14 @@ fn read_one_entry(r: &mut R) -> Result, ()> { let timestamp = f64::from_le_bytes(ts_buf); match entry_type { - WalEntryType::Save => { + WalEntryType::Save | WalEntryType::Update => { + let update_index = if entry_type == WalEntryType::Update { + let mut idx_buf = [0u8; 4]; + r.read_exact(&mut idx_buf).map_err(|_| ())?; + Some(u32::from_le_bytes(idx_buf) as usize) + } else { + None + }; let chunk = read_len_prefixed_str(r).map_err(|_| ())?; let embedding = read_embedding(r).map_err(|_| ())?; let source_channel = read_len_prefixed_str(r).map_err(|_| ())?; @@ -691,6 +758,7 @@ fn read_one_entry(r: &mut R) -> Result, ()> { session_id, tags, tombstone_index: None, + update_index, })) } WalEntryType::Tombstone => { @@ -706,6 +774,7 @@ fn read_one_entry(r: &mut R) -> Result, ()> { session_id: String::new(), tags: String::new(), tombstone_index: Some(idx), + update_index: None, })) } WalEntryType::ActivationUpdate => Ok(None), @@ -729,6 +798,7 @@ mod tests { session_id: "sess-001".to_string(), tags: "tag1,tag2".to_string(), tombstone_index: None, + update_index: None, } } @@ -859,6 +929,7 @@ mod tests { session_id: "sess-öö-123".to_string(), tags: "α,β,γ".to_string(), tombstone_index: None, + update_index: None, }; wal.append_save(&entry).unwrap(); } @@ -1052,6 +1123,58 @@ mod tests { assert_eq!(reopen_chunks(&h5_path), ["a", "b", "c", "d"]); } + #[test] + fn save_or_update_replays_as_update_not_duplicate() { + let dir = TempDir::new().unwrap(); + let config = make_config(&dir); + let h5_path = config.path.clone(); + { + let mut mem = HDF5Memory::create(config).unwrap(); + let mut first = make_entry("v1", &[1.0, 0.0, 0.0, 0.0]); + first.tags = "key".into(); + let mut second = make_entry("v2", &[0.0, 1.0, 0.0, 0.0]); + second.tags = "key".into(); + let a = mem.save_or_update(first).unwrap(); + mem.save(make_entry("other", &[0.0, 0.0, 1.0, 0.0])) + .unwrap(); + let b = mem.save_or_update(second).unwrap(); + assert_eq!(a, b); + assert_eq!(mem.cache.chunks, ["v2", "other"]); + // Dropped without a checkpoint: all three records live in the WAL. + } + let mem = HDF5Memory::open(&h5_path).unwrap(); + assert_eq!(mem.cache.chunks, ["v2", "other"]); + assert_eq!(mem.cache.embeddings[0], [0.0, 1.0, 0.0, 0.0]); + } + + #[test] + fn v3_wal_is_read_and_upgraded_in_place() { + let dir = TempDir::new().unwrap(); + let wal_path = dir.path().join("old.wal"); + { + let mut wal = WalFile::open(&wal_path).unwrap(); + wal.append_save(&make_wal_entry("kept", &[1.0])).unwrap(); + } + // Rewrite the header as the pre-Update chained format. + let mut bytes = std::fs::read(&wal_path).unwrap(); + bytes[4] = WAL_VERSION_CHAINED_NO_UPDATE; + std::fs::write(&wal_path, &bytes).unwrap(); + + assert_eq!(WalFile::read_entries(&wal_path).unwrap().len(), 1); + { + let mut wal = WalFile::open(&wal_path).unwrap(); + assert_eq!(wal.pending_count(), 1); + wal.append_save(&make_wal_entry("new", &[2.0])).unwrap(); + } + assert_eq!(std::fs::read(&wal_path).unwrap()[4], WAL_VERSION); + let chunks: Vec<_> = WalFile::read_entries(&wal_path) + .unwrap() + .into_iter() + .map(|e| e.chunk) + .collect(); + assert_eq!(chunks, ["kept", "new"]); + } + #[test] fn mark_matching_is_exact() { let dir = TempDir::new().unwrap();