index: add NODES_BY_PROVENANCE with timestamp-sorted values

- Store [negated_timestamp:8][key] as value for descending sort
- recent_by_provenance uses index directly, no capnp reads
- Eliminates 24k×5 capnp reads from subconscious snapshots

Co-Authored-By: Proof of Concept <poc@bcachefs.org>

src/hippocampus/store/index.rs

@@ -24,7 +24,8 @@ pub const NODES: TableDefinition<&str, u64> = TableDefinition::new("nodes");
 // KEY_TO_UUID: key → [uuid:16][node_type:1][timestamp:8] = 25 bytes
 pub const KEY_TO_UUID: TableDefinition<&str, &[u8]> = TableDefinition::new("key_to_uuid");
 pub const UUID_OFFSETS: MultimapTableDefinition<&[u8], u64> = MultimapTableDefinition::new("uuid_offsets");
-pub const NODES_BY_PROVENANCE: MultimapTableDefinition<&str, &str> = MultimapTableDefinition::new("nodes_by_provenance");
+// NODES_BY_PROVENANCE: provenance → [timestamp:8 BE][key] (sorted by timestamp desc via negated ts)
+pub const NODES_BY_PROVENANCE: MultimapTableDefinition<&str, &[u8]> = MultimapTableDefinition::new("nodes_by_provenance");
 // Composite key: [node_type: u8][timestamp: i64 BE] for range queries
 pub const NODES_BY_TYPE: TableDefinition<&[u8], &str> = TableDefinition::new("nodes_by_type");
 
@@ -81,19 +82,62 @@ pub fn unpack_node_meta(data: &[u8]) -> ([u8; 16], u8, i64) {
     }
 }
 
+/// Pack provenance value: [negated_timestamp:8][key] for descending sort
+fn pack_provenance_value(timestamp: i64, key: &str) -> Vec<u8> {
+    let neg_ts = (!timestamp).to_be_bytes(); // negate for descending order
+    let mut buf = Vec::with_capacity(8 + key.len());
+    buf.extend_from_slice(&neg_ts);
+    buf.extend_from_slice(key.as_bytes());
+    buf
+}
+
+/// Unpack provenance value: returns (timestamp, key)
+fn unpack_provenance_value(data: &[u8]) -> (i64, String) {
+    let neg_ts = i64::from_be_bytes([data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7]]);
+    let timestamp = !neg_ts;
+    let key = String::from_utf8_lossy(&data[8..]).to_string();
+    (timestamp, key)
+}
+
 /// Record a node's location in the index.
-pub fn index_node(txn: &WriteTransaction, key: &str, offset: u64, uuid: &[u8; 16], node_type: u8, timestamp: i64) -> Result<()> {
+pub fn index_node(txn: &WriteTransaction, key: &str, offset: u64, uuid: &[u8; 16], node_type: u8, timestamp: i64, provenance: &str) -> Result<()> {
     let mut nodes_table = txn.open_table(NODES)?;
     let mut key_uuid_table = txn.open_table(KEY_TO_UUID)?;
     let mut uuid_offsets = txn.open_multimap_table(UUID_OFFSETS)?;
+    let mut by_provenance = txn.open_multimap_table(NODES_BY_PROVENANCE)?;
 
     nodes_table.insert(key, offset)?;
     let packed = pack_node_meta(uuid, node_type, timestamp);
     key_uuid_table.insert(key, packed.as_slice())?;
     uuid_offsets.insert(uuid.as_slice(), offset)?;
+    let prov_val = pack_provenance_value(timestamp, key);
+    by_provenance.insert(provenance, prov_val.as_slice())?;
     Ok(())
 }
 
+/// Get recent keys for a given provenance, sorted by timestamp descending.
+pub fn recent_by_provenance(db: &Database, provenance: &str, limit: usize) -> Result<Vec<(String, i64)>> {
+    let txn = db.begin_read()?;
+    let table = txn.open_multimap_table(NODES_BY_PROVENANCE)?;
+    let mut results = Vec::new();
+    for entry in table.get(provenance)? {
+        if results.len() >= limit { break; }
+        let (timestamp, key) = unpack_provenance_value(entry?.value());
+        results.push((key, timestamp));
+    }
+    Ok(results)
+}
+
+/// Get node metadata (uuid, node_type, timestamp) from KEY_TO_UUID.
+pub fn get_node_meta(db: &Database, key: &str) -> Result<Option<([u8; 16], u8, i64)>> {
+    let txn = db.begin_read()?;
+    let table = txn.open_table(KEY_TO_UUID)?;
+    match table.get(key)? {
+        Some(data) => Ok(Some(unpack_node_meta(data.value()))),
+        None => Ok(None),
+    }
+}
+
 /// Get offset for a node by key.
 pub fn get_offset(db: &Database, key: &str) -> Result<Option<u64>> {
     let txn = db.begin_read()?;