[AUTOMATED] Index cached FID payload rows

crates/fission-signatures/src/fpk.rs

@@ -29,7 +29,7 @@ use std::collections::HashMap;
 use std::fs::File;
 use std::io::Read;
 use std::path::Path;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, OnceLock};
 
 const MAGIC: &[u8; 4] = b"FPK1";
 const HEADER_LEN: usize = 72;
@@ -86,6 +86,41 @@ struct BlockRef {
     raw_len: usize,
 }
 
+/// One immutable decoded payload, with optional random row access. Ordinary
+/// block consumers keep their shared String without building the row index.
+struct CachedPayloadBlock {
+    text: Arc<String>,
+    rows: OnceLock<Vec<std::ops::Range<usize>>>,
+}
+
+impl CachedPayloadBlock {
+    fn new(text: String) -> Self {
+        Self {
+            text: Arc::new(text),
+            rows: OnceLock::new(),
+        }
+    }
+
+    fn line_at(&self, row: usize) -> Option<&str> {
+        let rows = self.rows.get_or_init(|| {
+            let mut offset = 0usize;
+            self.text
+                .split_inclusive('\n')
+                .map(|raw| {
+                    let content = raw
+                        .strip_suffix('\n')
+                        .map(|line| line.strip_suffix('\r').unwrap_or(line))
+                        .unwrap_or(raw);
+                    let range = offset..offset + content.len();
+                    offset += raw.len();
+                    range
+                })
+                .collect()
+        });
+        self.text.get(rows.get(row)?.clone())
+    }
+}
+
 pub struct FpkReader {
     map: Mmap,
     blocks: Vec<BlockRef>,
@@ -95,7 +130,7 @@ pub struct FpkReader {
     /// resolve names that sort near each other, so the same block answers many
     /// of them. Without this a lazy database pays a decompress per lookup
     /// instead of per block.
-    cache: Mutex<HashMap<usize, Arc<String>>>,
+    cache: Mutex<HashMap<usize, Arc<CachedPayloadBlock>>>,
 }
 
 fn u16_at(b: &[u8], at: usize) -> u16 {
@@ -237,16 +272,23 @@ impl FpkReader {
 
     /// Decompressed block `index`, from the cache when it is already there.
     pub fn block_at(&self, index: usize) -> Result<Arc<String>, FpkError> {
+        self.cached_block_at(index)
+            .map(|block| Arc::clone(&block.text))
+    }
+
+    fn cached_block_at(&self, index: usize) -> Result<Arc<CachedPayloadBlock>, FpkError> {
         if let Ok(cache) = self.cache.lock()
             && let Some(hit) = cache.get(&index)
         {
             return Ok(Arc::clone(hit));
         }
-        let text = Arc::new(self.decompress(&self.blocks[index])?);
+        let block = Arc::new(CachedPayloadBlock::new(
+            self.decompress(&self.blocks[index])?,
+        ));
         if let Ok(mut cache) = self.cache.lock() {
-            cache.insert(index, Arc::clone(&text));
+            cache.insert(index, Arc::clone(&block));
         }
-        Ok(text)
+        Ok(block)
     }
 
     /// The index directory, or `None` when the file has none.
@@ -347,8 +389,8 @@ impl FpkReader {
                         self.blocks.len()
                     )));
                 }
-                let text = self.block_at(block)?;
-                match text.lines().nth(row) {
+                let text = self.cached_block_at(block)?;
+                match text.line_at(row) {
                     Some(line) => out.push(line.to_owned()),
                     None => {
                         return Err(FpkError::Corrupt(format!(
@@ -995,6 +1037,105 @@ mod tests {
         std::fs::remove_file(&path).ok();
     }
 
+    #[test]
+    fn cached_payload_rows_match_str_lines() {
+        for text in [
+            "",
+            "\n",
+            "\n\n",
+            "\r",
+            "\r\n",
+            "a",
+            "a\n",
+            "a\r",
+            "a\r\n",
+            "a\n\nb",
+            "a\r\n\r\nb\n",
+            "\nlast",
+            "a\r\rb\n",
+            "값\r\n다음\n끝",
+            "a\u{2028}b\u{0085}c",
+        ] {
+            let block = CachedPayloadBlock::new(text.to_owned());
+            assert!(block.rows.get().is_none());
+            let expected: Vec<_> = text.lines().collect();
+            for row in 0..expected.len() + 3 {
+                assert_eq!(
+                    block.line_at(row),
+                    expected.get(row).copied(),
+                    "{text:?} row={row}"
+                );
+            }
+            assert_eq!(block.line_at(usize::MAX), None);
+        }
+    }
+
+    #[test]
+    fn hash_rows_share_the_existing_payload_and_preserve_order_under_concurrency() {
+        let records: Vec<_> = (0..80).map(|i| format!("key|row{i:04}|payload")).collect();
+        let (mut image, locators) = pack_with_locators(&records, 1, CODEC_ZLIB, 1, 64 * 1024);
+        append_hash_index(
+            &mut image,
+            locators
+                .iter()
+                .map(|l| HashEntry {
+                    key: 42,
+                    block: l.block,
+                    row: l.row,
+                })
+                .collect(),
+        );
+        let path = write_temp(&image);
+        let reader = FpkReader::open(&path).unwrap();
+        assert_eq!(reader.block_count(), 1);
+        let public = reader.block_at(0).unwrap();
+        let cached = reader.cached_block_at(0).unwrap();
+        assert!(
+            cached.rows.get().is_none(),
+            "ordinary reads must not index lines"
+        );
+        std::thread::scope(|scope| {
+            let handles: Vec<_> = (0..8)
+                .map(|_| {
+                    scope.spawn(|| {
+                        for _ in 0..20 {
+                            assert_eq!(reader.records_by_key(42).unwrap(), records);
+                            assert!(reader.records_by_key(43).unwrap().is_empty());
+                        }
+                    })
+                })
+                .collect();
+            for handle in handles {
+                handle.join().unwrap();
+            }
+        });
+        assert_eq!(cached.rows.get().unwrap().len(), records.len());
+        assert!(Arc::ptr_eq(&public, &reader.block_at(0).unwrap()));
+        assert!(Arc::ptr_eq(&cached, &reader.cached_block_at(0).unwrap()));
+        std::fs::remove_file(path).ok();
+    }
+
+    #[test]
+    fn indexed_payload_keeps_invalid_locator_errors() {
+        let records = vec!["key|one".to_owned()];
+        let (mut image, _) = pack_with_locators(&records, 1, CODEC_ZLIB, 1, 4096);
+        append_hash_index(
+            &mut image,
+            vec![HashEntry {
+                key: 7,
+                block: 0,
+                row: 100,
+            }],
+        );
+        let path = write_temp(&image);
+        let reader = FpkReader::open(&path).unwrap();
+        assert_eq!(
+            reader.records_by_key(7).unwrap_err().to_string(),
+            "fpk corrupt: hash index points at row 100 of block 0"
+        );
+        std::fs::remove_file(path).ok();
+    }
+
     #[test]
     fn a_file_without_a_hash_index_says_so() {
         let path = write_temp(&pack(&["a|1", "b|2"]));

docs/proposals/2026-10-15-fid-row-access.md

@@ -0,0 +1,27 @@
+# Indexed access to immutable FPK payload rows
+
+Mechanical performance follow-up to PR #205; no quality, FID-admission, naming, ordering, cache-key, runtime-dependency, or worker-policy change.
+
+## Measured anchor and owner
+
+Pinned qualification revision 07dd95dbe6214b8a78fc1e4a55961ab515986489. PR #205's measured release candidate (SHA256 4c0485b7da200135afbc2e0bfbdbb9de8a7058549a1f9a1d6af3e91c9c6dc320) is the new control: selected small medians 2.224/1.171/1.380 s and large control 103.820 s. A symbolized candidate diagnostic at 65 s shows FID matching in FpkReader::records_by_key line 351, text.lines().nth(row), with 1337 samples at that line in a representative worker stack. The 35 s diagnostic still shows native decode, so this is a distinct later phase. Native samples are diagnostics, not timing or output-parity runs; they are retained under benchmark/artifacts/performance/2026-10-14-kuna/startup-large-profile-*.txt.
+
+The FPK reader already caches immutable decompressed payload blocks. Each hash locator names a row in such a block. Starting String::lines from the beginning for every locator repeats the same prefix scan. The previous rejected index-chunk experiment avoided decompression; it did not avoid this row scan. Keep that experiment excluded.
+
+## Invariant and implementation boundary
+
+Extend the existing per-reader payload cache entry with a lazily initialized line-range index. Construct ranges from newline-delimited chunks, removing exactly the same LF/optional CRLF terminators as str::lines. Empty input, empty interior lines, trailing newline, bare CR, Unicode and unterminated final rows must match str::lines exactly. Read one row by a bounds-checked string slice. Keep public block_at/block_for returning the original shared Arc<String>; direct block consumers need not build the row index.
+
+Keep index-directory searches, decompression/validation, locator checks, matched record order, duplicate handling, full/specific hashes, FID thresholds and error messages unchanged. The row index is built synchronously from the already validated immutable string and does not call Rayon, I/O or the reader cache. Retain it only with that existing cached payload, with two usize offsets per line plus vector capacity overhead; do not add a parallel program-fact map, hidden runtime dependency or global cache. Preserve cold-cache publication behavior and do not cache failures.
+
+## Verification and acceptance
+
+- Compare indexed rows with str::lines across empty/LF/CRLF/bare-CR/Unicode/repeated rows and out-of-range indices.
+- Assert public Arc ownership/warm identity remains unchanged; ordinary block reads leave line indexes uninitialized; repeated/concurrent hash reads return identical ordered records.
+- Retain existing cross-chunk duplicate-key, payload format, malformed input and block-boundary tests. Run focused signatures nextest then signatures/static/decompiler/CLI suites, checks, release CLI and workspace formatting.
+- Freeze a separate candidate binary. Measure five alternating small pairs and a sequential large control against PR #205's immutable measured binary. Fresh full 250-function stripped replay must preserve every non-time output field; no score or ranking change is claimed.
+- Keep only an observed practical gain with output parity. Report the added row-index memory and any timing regression. The remaining Kuna gap is not assumed closed.
+
+## Accepted local evidence
+
+15 focused FPK tests and 398 related tests passed (one skipped), with crate checks, release CLI and workspace formatting passing. Five-pair small controls are unchanged on Linux/macOS and improve Windows 1.1763 -> 0.9614 s. The fresh sequential large control improves 112.924 -> 42.235 s (CPU 1044.333 -> 233.619 s), with unchanged six outputs and similar observed maximum RSS (3.624/3.598 GiB). Fresh 250-function replay changes no non-time records. Details and all limits are recorded in docs/validation/fid-row-access-performance.md and its compact JSON. The 22-function pinned Kuna comparison remains 4.17–15.47 times slower; Kuna-level performance is not achieved.