Log: integration with slatedb segments (opendata-oss#450)

This patch integrates LogDb with SlateDb segmentation. Updates the key structure for prefix-based segments. Also updates the common key prefix RFC to remove the record tag field, which is incompatible with prefix-based segmentation.

Author: hachikuji

log/rfcs/0002-logical-segmentation.md

@@ -46,11 +46,22 @@ Key properties of segments:
 
 ### Key Encoding with Segments
 
-The log entry key format is extended to include the segment ID:
+Every log storage key — log entries, listings, segment metadata, and the
+global sequence block — shares a 7-byte segmented prefix:
+
+```
+| subsystem (u8) | version (u8) | segment_id (u32 BE) | record_type (u8) | ... record-specific fields ... |
+```
+
+Placing `segment_id` before `record_type` keeps every record for a given
+segment contiguous in storage regardless of record type. This is the
+property a SlateDB segment extractor relies on: with a fixed 6-byte prefix
+`[subsystem, version, segment_id]`, all records for a segment land in the
+same SlateDB segment.
 
 ```
 Log Entry:
-  | version (u8) | type (u8) | segment_id (u32 BE) | key (TerminatedBytes) | relative_seq (varint u64) |
+  | sub | ver | segment_id (u32 BE) | record_type=0x10 | key (TerminatedBytes) | relative_seq (varint u64) |
 ```
 
 The `segment_id` is a 32-bit identifier, allowing up to ~4 billion segments. Future versions can expand this to 64 bits if needed.
@@ -64,13 +75,33 @@ This encoding ensures:
 - Within a key, entries are ordered by sequence number
 - Prefix scans can be constrained to specific segments
 
+### The system segment
+
+Segment id `0` is reserved as the **system segment**: it never holds user
+log entries. The records that aren't bound to a specific user segment live
+there, sharing the same 7-byte prefix layout (with `segment_id = 0`):
+
+- **`SeqBlock`** (`record_type = 0x20`): the global block-based sequence
+  allocator's persisted state. There is exactly one record.
+- **`SegmentMeta`** (`record_type = 0x30`): per-segment metadata describing
+  user segments. The described segment id is encoded as a 4-byte suffix
+  after `record_type`, so all `SegmentMeta` records sit in a contiguous
+  range and can be scanned with a single prefix scan.
+
+User log segments are numbered from `1`. A custom SlateDB compactor can
+recognize the system segment by its `segment_id = 0` prefix and exempt it
+from any retention-driven drains, while still allowing regular user
+segments to be retired via `CompactionSpec::DrainSegment` and pruned from
+the manifest.
+
 ### Segment Metadata
 
-Each segment has associated metadata stored in a separate record:
+Each user segment has associated metadata stored as a single record in the
+system segment. The described segment's id is the suffix:
 
 ```
 SegmentMeta Record:
-  Key:   | version (u8) | type (u8=0x03) | segment_id (u32 BE) |
+  Key:   | sub | ver | segment_id=0 (u32 BE) | record_type=0x30 | described_segment_id (u32 BE) |
   Value: | start_seq (u64 BE) | start_time_ms (i64 BE) |
 ```
 
@@ -80,6 +111,14 @@ The metadata tracks:
 
 End boundaries (end sequence, end time) are derived from the next segment's start values, or from the current log state for the active segment.
 
+Living in the system segment decouples metadata from the per-segment
+SlateDB lifecycle: when a user segment is drained and pruned from the
+SlateDB manifest, the `SegmentMeta` record for it is not automatically
+removed and must be deleted by the writer as it observes the manifest
+update. The exact cleanup mechanism is out of scope for this RFC and is
+called out under [Segment-Based Deletion](#segment-based-deletion) as
+future work.
+
 #### Metadata Lifecycle
 
 When a new segment is created, a `SegmentMeta` record is written with `start_seq` and `start_time_ms`. This ensures the segment is immediately discoverable.
@@ -196,3 +235,4 @@ This would enable use cases such as:
 |------------|-------------|
 | 2026-01-07 | Initial draft |
 | 2026-01-12 | Added link to varint implementation |
+| 2026-05-19 | Key format v2: `segment_id` precedes `record_type`; system segment (id 0) reserved for `SeqBlock` and `SegmentMeta` records; user segments start at id 1. Enables a fixed 6-byte SlateDB segment-extractor prefix that routes every per-segment record (entries + listings) to the same SlateDB segment. |

log/src/lib.rs

@@ -55,6 +55,7 @@ mod model;
 mod range;
 mod reader;
 mod segment;
+mod segment_extractor;
 mod serde;
 #[cfg(feature = "http-server")]
 pub mod server;

log/src/listing.rs

@@ -153,19 +153,10 @@ mod tests {
                 .unwrap();
         }
 
-        #[storage_test]
-        async fn should_return_empty_for_empty_range(storage: Arc<dyn Storage>) {
-            // when
-            let keys = storage.list_keys(0..0).await.unwrap();
-
-            // then
-            assert!(keys.is_empty());
-        }
-
         #[storage_test]
         async fn should_return_empty_when_no_listing_entries(storage: Arc<dyn Storage>) {
             // when
-            let keys = storage.list_keys(0..10).await.unwrap();
+            let keys = storage.list_keys_in_segment(1).await.unwrap();
 
             // then
             assert!(keys.is_empty());
@@ -174,12 +165,17 @@ mod tests {
         #[storage_test]
         async fn should_iterate_keys_in_single_segment(storage: Arc<dyn Storage>) {
             // given
-            write_listing_entry(&*storage, 0, b"key-a").await;
-            write_listing_entry(&*storage, 0, b"key-b").await;
-            write_listing_entry(&*storage, 0, b"key-c").await;
+            write_listing_entry(&*storage, 1, b"key-a").await;
+            write_listing_entry(&*storage, 1, b"key-b").await;
+            write_listing_entry(&*storage, 1, b"key-c").await;
 
             // when
-            let keys: Vec<Bytes> = storage.list_keys(0..1).await.unwrap().into_iter().collect();
+            let keys: Vec<Bytes> = storage
+                .list_keys_in_segment(1)
+                .await
+                .unwrap()
+                .into_iter()
+                .collect();
 
             // then - keys returned in lexicographic order
             assert_eq!(keys.len(), 3);
@@ -189,60 +185,66 @@ mod tests {
         }
 
         #[storage_test]
-        async fn should_iterate_keys_across_multiple_segments(storage: Arc<dyn Storage>) {
-            // given
-            write_listing_entry(&*storage, 0, b"key-a").await;
-            write_listing_entry(&*storage, 1, b"key-b").await;
-            write_listing_entry(&*storage, 2, b"key-c").await;
+        async fn should_isolate_listings_to_their_segment(storage: Arc<dyn Storage>) {
+            // given — listing entries in multiple segments
+            write_listing_entry(&*storage, 1, b"in-segment-1").await;
+            write_listing_entry(&*storage, 2, b"in-segment-2").await;
 
             // when
-            let keys = storage.list_keys(0..3).await.unwrap();
-
-            // then
-            assert_eq!(keys.len(), 3);
+            let seg1 = storage.list_keys_in_segment(1).await.unwrap();
+            let seg2 = storage.list_keys_in_segment(2).await.unwrap();
+
+            // then — each segment scan returns only its own listing entries
+            assert_eq!(seg1.len(), 1);
+            assert!(seg1.contains(&Bytes::from("in-segment-1")));
+            assert_eq!(seg2.len(), 1);
+            assert!(seg2.contains(&Bytes::from("in-segment-2")));
         }
 
         #[storage_test]
-        async fn should_deduplicate_keys_across_segments(storage: Arc<dyn Storage>) {
-            // given - same key in multiple segments
-            write_listing_entry(&*storage, 0, b"shared-key").await;
-            write_listing_entry(&*storage, 1, b"shared-key").await;
-            write_listing_entry(&*storage, 2, b"shared-key").await;
+        async fn should_skip_log_entries_when_scanning_listings(storage: Arc<dyn Storage>) {
+            use crate::segment::LogSegment;
+            use crate::serde::LogEntryKey;
+            use crate::serde::SegmentMeta;
+
+            // given — listing records alongside log entries in the same segment
+            let segment_id = 1u32;
+            let segment = LogSegment::new(segment_id, SegmentMeta::new(0, 1000));
+            write_listing_entry(&*storage, segment_id, b"alpha").await;
+            // Write a log entry that would lex-sort before listings (tag 0x10 < 0x40).
+            let entry_key = LogEntryKey::new(segment_id, Bytes::from("alpha"), 1).serialize(0);
+            storage
+                .put_with_options(
+                    vec![common::Record::new(entry_key, Bytes::from("v")).into()],
+                    common::WriteOptions::default(),
+                )
+                .await
+                .unwrap();
+            // Anchor used for the segment metadata so deserialization is sane.
+            let _ = segment;
 
             // when
-            let keys = storage.list_keys(0..3).await.unwrap();
+            let keys = storage.list_keys_in_segment(segment_id).await.unwrap();
 
-            // then - only one instance of the key
+            // then — listing scan only sees listing records, not log entries
             assert_eq!(keys.len(), 1);
-            assert!(keys.contains(&Bytes::from("shared-key")));
-        }
-
-        #[storage_test]
-        async fn should_respect_segment_range(storage: Arc<dyn Storage>) {
-            // given
-            write_listing_entry(&*storage, 0, b"key-0").await;
-            write_listing_entry(&*storage, 1, b"key-1").await;
-            write_listing_entry(&*storage, 2, b"key-2").await;
-            write_listing_entry(&*storage, 3, b"key-3").await;
-
-            // when - only query segments 1..3
-            let keys: Vec<Bytes> = storage.list_keys(1..3).await.unwrap().into_iter().collect();
-
-            // then - only keys from segments 1 and 2
-            assert_eq!(keys.len(), 2);
-            assert_eq!(keys[0], Bytes::from("key-1"));
-            assert_eq!(keys[1], Bytes::from("key-2"));
+            assert!(keys.contains(&Bytes::from("alpha")));
         }
 
         #[storage_test]
         async fn should_return_keys_in_lexicographic_order(storage: Arc<dyn Storage>) {
             // given - keys inserted out of order
-            write_listing_entry(&*storage, 0, b"zebra").await;
-            write_listing_entry(&*storage, 0, b"apple").await;
-            write_listing_entry(&*storage, 0, b"mango").await;
+            write_listing_entry(&*storage, 1, b"zebra").await;
+            write_listing_entry(&*storage, 1, b"apple").await;
+            write_listing_entry(&*storage, 1, b"mango").await;
 
             // when
-            let keys: Vec<Bytes> = storage.list_keys(0..1).await.unwrap().into_iter().collect();
+            let keys: Vec<Bytes> = storage
+                .list_keys_in_segment(1)
+                .await
+                .unwrap()
+                .into_iter()
+                .collect();
 
             // then - keys returned in lexicographic order
             assert_eq!(keys[0], Bytes::from("apple"));