hawdb-storage 0.13.0

// Copyright 2026 Nowledge
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#[cfg(not(target_arch = "wasm32"))]
#[path = "pinned_rows_tests.rs"]
mod pinned_rows_tests;

use super::*;
use crate::NodeId;

fn checkpoint_unit_scheduler() -> hawdb_qos::LocalQosScheduler {
    hawdb_qos::LocalQosScheduler::new(hawdb_qos::LocalQosPolicy {
        max_background_operations: Some(1),
        max_total_background_operations: Some(1),
        ..hawdb_qos::LocalQosPolicy::default()
    })
}

fn assert_checkpoint_unit_cancellation(error: &HawDBError) {
    // The sidecar API converts the typed work error to its existing facade
    // storage error, which must remain nonfatal and distinguish cancellation.
    assert!(
        matches!(error, HawDBError::Storage(message) if
        message == "checkpoint build stopped: cancelled" ||
        message == "checkpoint build I/O stopped: runtime I/O wave stopped: cancelled"),
        "expected checkpoint cancellation: {error}"
    );
}

#[test]
fn checkpoint_units_source_scan_reopens_complete_segments_and_exact_candidates() {
    use crate::background::CheckpointWorkProbe;
    use std::sync::atomic::Ordering;
    use std::sync::Arc;

    let directory = TestDir::new();
    let scheduler = checkpoint_unit_scheduler();
    let probe = Arc::new(CheckpointWorkProbe::default());
    let work = probe.context(scheduler.clone());
    let nodes = (0..2500u64)
        .map(|id| {
            let mut node = source(
                id,
                LabelId(if id.is_multiple_of(5) { 8 } else { 7 }),
                "scope",
            );
            node.properties.insert("rank".into(), Value::Int(id as i64));
            node.properties.insert(
                "nested".into(),
                Value::List(vec![Value::Int(id as i64), Value::Binary(vec![0, 255])]),
            );
            if id.is_multiple_of(3) {
                node.properties.insert("optional".into(), Value::Null);
            }
            node
        })
        .collect::<Vec<_>>();
    let expected = rows(&nodes, Some(LabelId(7)));
    assert_eq!(expected.len(), 2000);
    let mut projection = build_with_work_context(
        73,
        Some(LabelId(7)),
        nodes.iter().inspect(|_| {
            assert!(probe.active_units.load(Ordering::SeqCst) > 0);
        }),
        &work,
    )
    .unwrap();
    let publication = write_with_work_context(directory.path(), &mut projection, &work).unwrap();
    let loaded = load(directory.path(), 73, publication.descriptor_checksum())
        .unwrap()
        .unwrap();
    let payloads = fs::read(directory.path().join(SOURCE_SCAN_PAYLOAD_FILE)).unwrap();
    let summaries = expected
        .chunks(128)
        .enumerate()
        .map(|(id, rows)| reference::summary(id as u64, rows))
        .collect::<Vec<_>>();
    assert_eq!(loaded.segments().len(), summaries.len());
    let mut recovered = Vec::new();
    let mut ranges = Vec::new();
    for (segment, summary) in loaded.segments().iter().zip(&summaries) {
        assert_eq!(&segment.summary, summary);
        let range = segment.payload_range;
        let payload =
            &payloads[range.offset as usize..(range.offset + range.length.get()) as usize];
        assert_eq!(reference::crc(payload), range.checksum);
        let segment_rows = decode_payload(payload).unwrap();
        assert_eq!(unpack(payload), reference::payload(&segment_rows));
        recovered.extend(segment_rows);
        ranges.push(range);
    }
    assert_eq!(recovered, expected);
    let descriptor = reference::descriptor(73, &summaries, &ranges);
    let (file, checksum) = reference::descriptor_file(&descriptor);
    assert_eq!(publication.descriptor_checksum(), checksum);
    assert_eq!(
        fs::read_to_string(directory.path().join(SOURCE_SCAN_DESCRIPTOR_FILE)).unwrap(),
        file
    );
    for (index, row) in expected.iter().enumerate() {
        let crate::scan::ScanSegmentAccessPlan::Read(mut plan) = loaded.plan_scan(
            73,
            &ScanPredicate::Eq {
                property: "id".into(),
                value: row.properties["id"].clone(),
            },
        ) else {
            panic!("current epoch must use the sidecar");
        };
        assert_eq!(plan.segments.len(), 1);
        assert_eq!(plan.skipped_segment_count, loaded.segments().len() - 1);
        assert_eq!(plan.segments[0].segment_id, (index / 128) as u64);
        assert_eq!(
            plan.segments[0].candidates.as_mut().unwrap().next_batch(2),
            vec![(index % 128) as u64]
        );
    }
    let empty = build_with_work_context(
        73,
        None,
        std::iter::from_fn(|| -> Option<&NodeRecord> {
            panic!("an absent Source label must not hydrate the graph");
        }),
        &work,
    )
    .unwrap();
    assert!(empty.segments.is_empty());
    assert!(probe.completed.load(Ordering::SeqCst) > 2000);
    assert!(probe.peak_units.load(Ordering::SeqCst) <= 1);
    probe.assert_released(&scheduler);
}

#[test]
fn checkpoint_units_source_scan_cancels_during_segment_summary() {
    use crate::background::CheckpointWorkProbe;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;

    let scheduler = checkpoint_unit_scheduler();
    let probe = Arc::new(CheckpointWorkProbe::default());
    let work = probe.context(scheduler.clone());
    let nodes = (0..256u64)
        .map(|id| source(id, LabelId(7), "scope"))
        .collect::<Vec<_>>();
    let hydrated = AtomicUsize::new(0);
    let records = nodes.iter().inspect(|_| {
        let count = hydrated.fetch_add(1, Ordering::SeqCst) + 1;
        if count == 128 {
            // Finish this row, 384 field-name insertions and part of the
            // first field's 128-row summary before requesting cancellation.
            probe.cancel_after.store(
                probe.completed.load(Ordering::SeqCst) + 400,
                Ordering::SeqCst,
            );
        }
    });
    let error = build_with_work_context(74, Some(LabelId(7)), records, &work).unwrap_err();
    assert_checkpoint_unit_cancellation(&error);
    assert_eq!(hydrated.load(Ordering::SeqCst), 128);
    probe.assert_released(&scheduler);
}

#[test]
fn checkpoint_units_source_scan_precancel_preserves_unowned_temporary_files() {
    use crate::background::CheckpointWorkProbe;
    use std::sync::Arc;

    let directory = TestDir::new();
    let temporary = directory
        .path()
        .join("source_scan_segment_payloads.hawdb.tmp");
    fs::write(&temporary, b"previous writer evidence").unwrap();
    let nodes = [source(0, LabelId(7), "scope")];
    let mut projection = build(76, Some(LabelId(7)), nodes.iter());
    let scheduler = checkpoint_unit_scheduler();
    let probe = Arc::new(CheckpointWorkProbe::default());
    probe.cancellation.cancel();
    let error = write_with_work_context(
        directory.path(),
        &mut projection,
        &probe.context(scheduler.clone()),
    )
    .unwrap_err();
    assert_checkpoint_unit_cancellation(&error);
    probe.assert_released(&scheduler);
    assert_eq!(fs::read(&temporary).unwrap(), b"previous writer evidence");
    assert_eq!(fs::read_dir(directory.path()).unwrap().count(), 1);
}

#[test]
fn checkpoint_units_source_scan_cancels_at_every_io_admission_and_retries() {
    use crate::background::CheckpointWorkProbe;
    use std::sync::atomic::Ordering;
    use std::sync::Arc;

    let nodes = (0..129u64)
        .map(|id| source(id, LabelId(7), "scope"))
        .collect::<Vec<_>>();
    let complete = TestDir::new();
    let scheduler = checkpoint_unit_scheduler();
    let baseline = Arc::new(CheckpointWorkProbe::default());
    let mut projection = build(75, Some(LabelId(7)), nodes.iter());
    write_with_work_context(
        complete.path(),
        &mut projection,
        &baseline.context(scheduler.clone()),
    )
    .unwrap();
    let waves = baseline.io_waves.load(Ordering::SeqCst);
    assert!(waves >= 9);
    baseline.assert_released(&scheduler);
    for wave in 1..=waves {
        let directory = TestDir::new();
        let scheduler = checkpoint_unit_scheduler();
        let probe = Arc::new(CheckpointWorkProbe::default());
        probe.cancel_on_io_wave.store(wave, Ordering::SeqCst);
        let mut projection = build(75, Some(LabelId(7)), nodes.iter());
        let error = write_with_work_context(
            directory.path(),
            &mut projection,
            &probe.context(scheduler.clone()),
        )
        .unwrap_err();
        assert_checkpoint_unit_cancellation(&error);
        assert_eq!(probe.io_waves.load(Ordering::SeqCst), wave);
        probe.assert_released(&scheduler);
        assert!(!directory
            .path()
            .join("source_scan_segments.hawdb.tmp")
            .exists());
        assert!(!directory
            .path()
            .join("source_scan_segment_payloads.hawdb.tmp")
            .exists());
        let retry = Arc::new(CheckpointWorkProbe::default());
        let publication = write_with_work_context(
            directory.path(),
            &mut projection,
            &retry.context(scheduler.clone()),
        )
        .unwrap();
        retry.assert_released(&scheduler);
        let loaded = load(directory.path(), 75, publication.descriptor_checksum())
            .unwrap()
            .unwrap();
        assert_eq!(
            loaded
                .segments()
                .iter()
                .map(|segment| segment.summary.row_count)
                .sum::<u64>(),
            129
        );
        for name in [SOURCE_SCAN_DESCRIPTOR_FILE, SOURCE_SCAN_PAYLOAD_FILE] {
            assert_eq!(
                fs::read(directory.path().join(name)).unwrap(),
                fs::read(complete.path().join(name)).unwrap()
            );
        }
    }
}

fn source_candidate_request() -> SourceCandidateScanRequest {
    SourceCandidateScanRequest {
        predicate: ScanPredicate::True,
        after: None,
        limit: 1,
        max_payload_bytes: 1024,
        property_names: vec!["source_type".to_string()],
    }
}

fn source_candidate_node(id: u64, created_at: i64, source_id: &str) -> NodeRecord {
    NodeRecord {
        id: NodeId(id),
        labels: BTreeSet::new(),
        properties: BTreeMap::from([
            ("id".to_string(), Value::String(source_id.to_string())),
            ("created_at".to_string(), Value::Int(created_at)),
            ("source_type".to_string(), Value::String("file".to_string())),
            ("excluded".to_string(), Value::String("value".to_string())),
        ]),
    }
}

#[test]
fn source_candidate_page_preserves_cursor_order_and_projection_bounds() {
    let request = source_candidate_request();
    let nodes = vec![
        source_candidate_node(11, 10, "source-old"),
        source_candidate_node(12, 20, "source-new"),
    ];

    let first = render_source_candidate_page(
        7,
        nodes.clone(),
        &request,
        SourceCandidateScanOrigin::Sidecar {
            graph_epoch: 7,
            skipped_segment_count: 0,
        },
        None,
    )
    .unwrap();

    assert_eq!(first.rows.len(), 1);
    assert_eq!(first.rows[0].node_id, 12);
    assert_eq!(first.rows[0].source_id.as_deref(), Some("source-new"));
    assert_eq!(
        first.rows[0].properties,
        BTreeMap::from([("source_type".to_string(), Value::String("file".to_string()))])
    );
    assert_eq!(
        first.next_cursor,
        Some(SourceCandidateCursor {
            created_at: Some(Value::Int(20)),
            node_id: 12,
        })
    );

    let second = render_source_candidate_page(
        7,
        nodes,
        &SourceCandidateScanRequest {
            after: first.next_cursor,
            ..request
        },
        SourceCandidateScanOrigin::CanonicalFallback {
            reason: ScanSegmentFallback::NoManifest,
        },
        None,
    )
    .unwrap();
    assert_eq!(second.rows.len(), 1);
    assert_eq!(second.rows[0].node_id, 11);
    assert!(second.next_cursor.is_none());
}

#[test]
fn source_candidate_request_keeps_resource_bounds() {
    let mut request = source_candidate_request();
    request.limit = 0;
    assert!(validate_source_candidate_scan_request(&request)
        .unwrap_err()
        .to_string()
        .contains("positive limit"));

    request = source_candidate_request();
    request.max_payload_bytes = 0;
    assert!(validate_source_candidate_scan_request(&request)
        .unwrap_err()
        .to_string()
        .contains("positive payload budget"));
}

#[test]
fn historical_decoder_tolerance_is_not_tightened() {
    let decoded = decode_payload(&reference::envelope(
        "row\t1\t\n\nHAWDB_SOURCE_SCAN_SEGMENT_V1\n",
    ))
    .unwrap();
    assert_eq!(
        decoded,
        vec![SourceScanRow {
            node_id: 1,
            properties: BTreeMap::new()
        }]
    );
    let (descriptor, checksum) = reference::descriptor_file("graph_epoch\t1\ngraph_epoch\t2\n");
    assert_eq!(
        decode_descriptor(&descriptor, checksum)
            .unwrap()
            .graph_epoch,
        2
    );
}

#[test]
fn source_scan_differential_smoke() {
    campaign(2, 8);
}

#[test]
#[ignore = "explicit local generated source sidecar campaign"]
fn source_scan_differential_campaign() {
    campaign(32, 64);
}

struct Generator(u64);
impl Generator {
    fn next(&mut self) -> u64 {
        self.0 ^= self.0 << 13;
        self.0 ^= self.0 >> 7;
        self.0 ^= self.0 << 17;
        self.0
    }
}

fn campaign(seeds: u64, cases_per_seed: usize) {
    let values = vec![
        Value::Null,
        Value::Bool(false),
        Value::Bool(true),
        Value::Int(i64::MIN),
        Value::Int(i64::MAX),
        Value::Int(-9_007_199_254_740_993),
        Value::Int(-9_007_199_254_740_992),
        Value::Int(9_007_199_254_740_992),
        Value::Int(9_007_199_254_740_993),
        Value::Int(0),
        Value::Float(-0.0),
        Value::Float(0.0),
        Value::Float(-2.5),
        Value::Float(3.5),
        Value::Float(f64::INFINITY),
        Value::Float(f64::NEG_INFINITY),
        Value::Float(f64::from_bits(0x7ff8_0000_0000_0001)),
        Value::String("".into()),
        Value::String("a\t;=:\n\0\u{65e5}\u{1f980}".into()),
        Value::String("2026-08-01T00:00:00Z".into()),
        Value::String("2026-08-01T08:00:00+08:00".into()),
        Value::String("1969-12-31T23:59:59.999Z".into()),
        Value::String("not-a-date".into()),
        Value::Binary(vec![0, 0xff, 0x80]),
        Value::Uuid(hawdb_core::Uuid::from_u128(7)),
        Value::List(vec![
            Value::Null,
            Value::Int(-1),
            Value::String("nested".into()),
        ]),
        Value::Map(BTreeMap::from([(
            "nested;=".into(),
            Value::Binary(vec![0, 1]),
        )])),
    ];
    let mut state_checks = 0;
    let mut disk_cases = 0;
    for seed in 0..seeds {
        let mut rng = Generator(seed + 1);
        for case in 0..cases_per_seed {
            let count = [0, 1, 127, 128, 129, 255, 256, 257][case % 8];
            let label = (case % 9 != 0).then_some(LabelId(7));
            let start_id = if case % 5 == 0 {
                u64::MAX - count as u64
            } else {
                seed * 1000
            };
            let nodes = (0..count)
                .map(|index| {
                    let item = values[(rng.next() % values.len() as u64) as usize].clone();
                    let mut properties = BTreeMap::from([("value".into(), item.clone())]);
                    if index % 11 != 0 {
                        properties.insert(
                            "id".into(),
                            if index % 3 == 0 {
                                item
                            } else {
                                Value::String(format!("id-{}", index % 13))
                            },
                        );
                    }
                    if index % 7 == 0 {
                        properties.insert("optional".into(), Value::Null);
                    }
                    if index % 5 == 0 {
                        properties.insert("".into(), Value::String("2026-08-02T00:00:00Z".into()));
                    }
                    NodeRecord {
                        id: NodeId(start_id + index as u64),
                        labels: BTreeSet::from([if rng.next().is_multiple_of(4) {
                            LabelId(8)
                        } else {
                            LabelId(7)
                        }]),
                        properties,
                    }
                })
                .collect::<Vec<_>>();
            let persist = case % 16 == (seed % 16) as usize;
            let epoch = if case % 7 == 0 { u64::MAX } else { rng.next() };
            check_projection(&nodes, label, epoch, persist);
            if persist {
                disk_cases += 1;
            }
            state_checks += 1;
        }
        state_checks += corruption_states(seed);
    }
    eprintln!("source-scan-differential-v1 seeds={seeds} generated_cases={} disk_cases={disk_cases} state_checks={state_checks}", seeds as usize * cases_per_seed);
}

fn corruption_states(seed: u64) -> usize {
    let directory = TestDir::new();
    let nodes = [source(seed + 1, LabelId(7), "state")];
    let mut projection = build(seed, Some(LabelId(7)), nodes.iter());
    let publication = write(directory.path(), &mut projection).unwrap();
    let checksum = publication.descriptor_checksum();
    let descriptor_path = directory.path().join(SOURCE_SCAN_DESCRIPTOR_FILE);
    let payload_path = directory.path().join(SOURCE_SCAN_PAYLOAD_FILE);
    let descriptor = fs::read(&descriptor_path).unwrap();
    let payload = fs::read(&payload_path).unwrap();
    assert!(load(directory.path(), seed, checksum).unwrap().is_some());
    assert!(load(directory.path(), seed + 1, checksum)
        .unwrap()
        .is_none());
    let mut changed = payload.clone();
    let index = seed as usize % changed.len();
    changed[index] ^= 1;
    fs::write(&payload_path, changed).unwrap();
    assert_eq!(
        storage_error(load(directory.path(), seed, checksum)),
        "source scan payload checksum mismatch"
    );
    fs::write(&payload_path, &payload).unwrap();
    assert!(load(directory.path(), seed, checksum).unwrap().is_some());
    let mut changed = descriptor.clone();
    changed[seed as usize % 16] ^= 1;
    fs::write(&descriptor_path, changed).unwrap();
    assert_eq!(
        storage_error(load(directory.path(), seed, checksum)),
        "source scan descriptor checksum mismatch"
    );
    fs::write(&descriptor_path, descriptor).unwrap();
    assert!(load(directory.path(), seed, checksum).unwrap().is_some());
    6
}

fn rows(nodes: &[NodeRecord], label: Option<LabelId>) -> Vec<SourceScanRow> {
    nodes
        .iter()
        .filter(|node| label.is_some_and(|label| node.labels.contains(&label)))
        .map(|node| SourceScanRow {
            node_id: node.id.0,
            properties: node.properties.clone(),
        })
        .collect()
}

fn unpack(bytes: &[u8]) -> String {
    let boundary = bytes.windows(2).position(|pair| pair == b"\n\n").unwrap();
    let header = std::str::from_utf8(&bytes[..boundary]).unwrap();
    let mut lines = header.lines();
    assert_eq!(lines.next(), Some("HAWDB_COMPRESSED_V1"));
    let fields = lines
        .map(|line| line.split_once('\t').unwrap())
        .collect::<BTreeMap<_, _>>();
    assert_eq!(fields.len(), 5);
    assert_eq!(fields["codec"], "zstd");
    let compressed = &bytes[boundary + 2..];
    let raw = zstd::stream::decode_all(compressed).unwrap();
    assert_eq!(
        fields["compressed_checksum"],
        reference::crc(compressed).to_string()
    );
    assert_eq!(
        fields["uncompressed_checksum"],
        reference::crc(&raw).to_string()
    );
    assert_eq!(fields["compressed_len"], compressed.len().to_string());
    assert_eq!(fields["uncompressed_len"], raw.len().to_string());
    String::from_utf8(raw).unwrap()
}

fn check_projection(nodes: &[NodeRecord], label: Option<LabelId>, epoch: u64, persist: bool) {
    let expected = rows(nodes, label);
    let mut projection = build(epoch, label, nodes.iter());
    assert_eq!(projection.graph_epoch, epoch);
    assert_eq!(projection.segments.len(), expected.len().div_ceil(128));
    let mut summaries = expected
        .chunks(128)
        .enumerate()
        .map(|(id, rows)| reference::summary(id as u64, rows))
        .collect::<Vec<_>>();
    let mut offset = 0;
    let mut ranges = Vec::new();
    let mut payloads = Vec::new();
    for (index, segment) in projection
        .segments
        .make_mut(&CheckpointWorkContext::default())
        .unwrap()
        .iter_mut()
        .enumerate()
    {
        let expected_rows = &expected[index * 128..expected.len().min((index + 1) * 128)];
        assert_eq!(segment.rows, expected_rows);
        reference::validate_signed_zero_ties(
            &mut summaries[index],
            &segment.summary,
            expected_rows,
        );
        let raw = reference::payload(expected_rows);
        let payload = encode_segment_payload(expected_rows).unwrap();
        assert_eq!(unpack(&payload), raw);
        assert_eq!(
            decode_payload(&reference::envelope(&raw)).unwrap(),
            expected_rows
        );
        assert_eq!(decode_payload(&payload).unwrap(), expected_rows);
        let range = SegmentPayloadRange {
            artifact_id: 1,
            offset,
            length: NonZeroU64::new(payload.len() as u64).unwrap(),
            checksum: reference::crc(&payload),
        };
        offset += payload.len() as u64;
        segment.payload_range = Some(range);
        ranges.push(range);
        payloads.extend(payload);
    }
    let body = reference::descriptor(epoch, &summaries, &ranges);
    assert_eq!(encode_descriptor(&projection).unwrap(), body);
    let (file, checksum) = reference::descriptor_file(&body);
    let decoded = decode_descriptor(&file, checksum).unwrap();
    assert_eq!(decoded.graph_epoch, epoch);
    assert_eq!(encode_descriptor(&decoded).unwrap(), body);
    for (segment, expected) in decoded.segments.iter().zip(projection.segments.iter()) {
        assert_eq!(segment.summary, expected.summary);
        assert_eq!(segment.payload_range, expected.payload_range);
        assert!(segment.rows.is_empty());
    }
    if persist {
        let directory = TestDir::new();
        let publication = write(directory.path(), &mut projection).unwrap();
        assert_eq!(publication.graph_epoch(), epoch);
        assert_eq!(publication.descriptor_checksum(), checksum);
        assert_eq!(
            fs::read_to_string(directory.path().join(SOURCE_SCAN_DESCRIPTOR_FILE)).unwrap(),
            file
        );
        assert_eq!(
            fs::read(directory.path().join(SOURCE_SCAN_PAYLOAD_FILE)).unwrap(),
            payloads
        );
        let loaded = load(directory.path(), epoch, checksum).unwrap().unwrap();
        assert_eq!(loaded.graph_epoch(), epoch);
        assert_eq!(
            loaded
                .segments()
                .iter()
                .map(|segment| segment.summary.clone())
                .collect::<Vec<_>>(),
            summaries
        );
        assert_eq!(
            loaded
                .segments()
                .iter()
                .map(|segment| segment.payload_range)
                .collect::<Vec<_>>(),
            ranges
        );
        assert!(!directory
            .path()
            .join("source_scan_segments.hawdb.tmp")
            .exists());
        assert!(!directory
            .path()
            .join("source_scan_segment_payloads.hawdb.tmp")
            .exists());
    }
}

#[test]
fn signed_zero_bounds_preserve_supported_encodings_and_wire_bits() {
    let range = SegmentPayloadRange {
        artifact_id: 1,
        offset: 0,
        length: NonZeroU64::new(1).unwrap(),
        checksum: 0,
    };
    for values in [
        vec![Value::Float(0.0), Value::Float(-0.0)],
        vec![Value::Float(-0.0), Value::Int(0)],
        vec![Value::Float(-0.0), Value::Float(0.0), Value::Float(-0.0)],
        vec![Value::Float(-1.0), Value::Float(0.0), Value::Float(-0.0)],
        vec![Value::Float(-0.0), Value::Float(0.0), Value::Float(1.0)],
    ] {
        let nodes = values
            .iter()
            .enumerate()
            .map(|(index, value)| NodeRecord {
                id: NodeId(index as u64),
                labels: BTreeSet::from([LabelId(7)]),
                properties: BTreeMap::from([("id".into(), value.clone())]),
            })
            .collect::<Vec<_>>();
        check_projection(&nodes, Some(LabelId(7)), 42, true);
        let rows = rows(&nodes, Some(LabelId(7)));
        for min in [-0.0_f64, 0.0] {
            for max in [-0.0_f64, 0.0] {
                let mut expected = reference::summary(0, &rows);
                let mut actual = expected.clone();
                let mut bounds = actual.fields["id"].numeric_min_max.unwrap();
                if bounds.min == 0.0 {
                    bounds.min = min;
                }
                if bounds.max == 0.0 {
                    bounds.max = max;
                }
                actual.fields.get_mut("id").unwrap().numeric_min_max = Some(bounds);
                reference::validate_signed_zero_ties(&mut expected, &actual, &rows);
                let body = reference::descriptor(42, &[expected], &[range]);
                let (file, checksum) = reference::descriptor_file(&body);
                let decoded = decode_descriptor(&file, checksum).unwrap();
                assert_eq!(encode_descriptor(&decoded).unwrap(), body);
                let decoded_bounds = decoded.segments[0].summary.fields["id"]
                    .numeric_min_max
                    .unwrap();
                assert_eq!(decoded_bounds.min.to_bits(), bounds.min.to_bits());
                assert_eq!(decoded_bounds.max.to_bits(), bounds.max.to_bits());
            }
        }
    }
}

#[test]
fn signed_zero_oracle_rejects_unobserved_signs_and_wrong_bounds() {
    for (values, wrong) in [
        (vec![Value::Float(0.0)], -0.0),
        (vec![Value::Int(0)], -0.0),
        (vec![Value::Float(-0.0)], 0.0),
        (
            vec![Value::Float(0.0), Value::Float(-0.0)],
            f64::from_bits(1),
        ),
        (vec![Value::Float(0.0), Value::Float(-0.0)], f64::INFINITY),
        (vec![Value::Float(0.0), Value::Float(-0.0)], f64::NAN),
        (vec![Value::Float(-1.0), Value::Float(1.0)], 0.0),
    ] {
        let rows = values
            .into_iter()
            .enumerate()
            .map(|(index, value)| SourceScanRow {
                node_id: index as u64,
                properties: BTreeMap::from([("id".into(), value)]),
            })
            .collect::<Vec<_>>();
        for change_min in [false, true] {
            let expected = reference::summary(0, &rows);
            let mut actual = expected.clone();
            let bounds = actual
                .fields
                .get_mut("id")
                .unwrap()
                .numeric_min_max
                .as_mut()
                .unwrap();
            if change_min {
                bounds.min = wrong;
            } else {
                bounds.max = wrong;
            }
            assert!(
                std::panic::catch_unwind(|| {
                    let mut expected = expected.clone();
                    reference::validate_signed_zero_ties(&mut expected, &actual, &rows);
                })
                .is_err(),
                "oracle accepted wrong numeric bits {}",
                wrong.to_bits()
            );
        }
    }
}

#[test]
fn segment_boundaries_and_label_selection_match_independent_model() {
    for count in [0, 1, 127, 128, 129, 255, 256, 257] {
        let nodes = (1..=count)
            .map(|id| source(id, LabelId(7), "space"))
            .collect::<Vec<_>>();
        check_projection(&nodes, Some(LabelId(7)), 42, true);
        check_projection(&nodes, Some(LabelId(8)), 42, false);
        check_projection(&nodes, None, 42, true);
    }
}

fn write_reference(directory: &Path, body: &str, payload: &[u8]) -> u64 {
    let (descriptor, checksum) = reference::descriptor_file(body);
    fs::write(directory.join(SOURCE_SCAN_DESCRIPTOR_FILE), descriptor).unwrap();
    fs::write(directory.join(SOURCE_SCAN_PAYLOAD_FILE), payload).unwrap();
    checksum
}

fn storage_error<T: std::fmt::Debug>(result: Result<T>) -> String {
    match result {
        Err(HawDBError::Storage(message)) => message,
        other => panic!("expected storage error, got {other:?}"),
    }
}

#[test]
fn missing_sidecar_and_stale_epoch_preserve_fallback_and_validation_order() {
    let directory = TestDir::new();
    let path = directory.path();
    fs::write(path.join("source_scan_segments.hawdb.tmp"), b"unfinished").unwrap();
    assert!(load(path, 7, 0).unwrap().is_none());
    let (descriptor, checksum) =
        reference::descriptor_file("HAWDB_SOURCE_SCAN_SEGMENTS_V1\ngraph_epoch\t7\n");
    fs::write(path.join(SOURCE_SCAN_DESCRIPTOR_FILE), descriptor).unwrap();
    // A stale epoch must not try to open an absent payload; checksum admission still precedes it.
    assert!(load(path, 8, checksum).unwrap().is_none());
    assert_eq!(
        storage_error(load(path, 8, checksum ^ 1)),
        "source scan descriptor checksum mismatch"
    );
    assert!(load(path, 7, checksum).is_err());
    fs::write(path.join(SOURCE_SCAN_PAYLOAD_FILE), b"").unwrap();
    assert!(load(path, 7, checksum)
        .unwrap()
        .unwrap()
        .segments()
        .is_empty());
}

#[test]
fn descriptor_rejects_malformed_fields_with_stable_errors() {
    let cases = [
        ("", "source scan descriptor missing graph epoch"),
        (
            "graph_epoch\t7\nfield\t6964\t1\t0\t0\t\t\t\t\t\n",
            "source scan field appears before a segment",
        ),
        (
            "graph_epoch\t7\nexact\t6964\t733631\t0\n",
            "source scan exact cursor appears before a segment",
        ),
        (
            "graph_epoch\t7\nsegment\t0\t1\t0\t0\t0\t1\n",
            "source scan payload length is zero",
        ),
        (
            "graph_epoch\t7\nsegment\t0\t1\t0\t1\t0\t1\nfield\t6964\t2\t0\t0\t\t\t\t\t\n",
            "invalid source scan field counts",
        ),
        (
            "graph_epoch\t7\nsegment\t0\t1\t0\t1\t0\t1\nexact\t6964\t733631\t0\n",
            "source scan exact cursor references unknown field",
        ),
        (
            "unknown\tvalue\n",
            "invalid source scan descriptor line: unknown\tvalue",
        ),
    ];
    for (body, expected) in cases {
        let (file, checksum) = reference::descriptor_file(body);
        assert_eq!(storage_error(decode_descriptor(&file, checksum)), expected);
    }
    assert_eq!(
        storage_error(decode_descriptor("graph_epoch\t7\n", 0)),
        "source scan descriptor missing checksum footer"
    );
    assert_eq!(
        storage_error(decode_row_ids("2,1")),
        "source scan exact row ids are not ordered"
    );
    assert_eq!(
        storage_error(decode_row_ids("1,1")),
        "source scan exact row ids are not ordered"
    );
}

#[test]
fn payload_preserves_order_and_rejects_malformed_rows() {
    for (raw, expected) in [
        (
            "HAWDB_SOURCE_SCAN_SEGMENT_V1\nrow\t2\t\nrow\t1\t\n",
            "source scan segment rows are not strictly ordered",
        ),
        (
            "HAWDB_SOURCE_SCAN_SEGMENT_V1\nrow\t1\t\nrow\t1\t\n",
            "source scan segment rows are not strictly ordered",
        ),
        (
            "HAWDB_SOURCE_SCAN_SEGMENT_V1\nunknown\t1\n",
            "invalid source scan segment line: unknown\t1",
        ),
    ] {
        assert_eq!(
            storage_error(decode_payload(&reference::envelope(raw))),
            expected
        );
    }
    assert_eq!(
        storage_error(decode_payload(b"row\t1\t\n")),
        "source scan segment is missing the V1 compressed envelope"
    );
    // The move does not silently sort a caller's unordered node iterator.
    let nodes = [
        source(2, LabelId(7), "space"),
        source(1, LabelId(7), "space"),
    ];
    let projection = build(7, Some(LabelId(7)), nodes.iter());
    assert_eq!(projection.segments[0].rows[0].node_id, 2);
    assert!(
        decode_payload(&encode_segment_payload(&projection.segments[0].rows).unwrap()).is_err()
    );
}

#[test]
fn load_validates_artifact_ranges_checksums_rows_and_manifest_order() {
    let directory = TestDir::new();
    let path = directory.path();
    let raw = "HAWDB_SOURCE_SCAN_SEGMENT_V1\nrow\t1\t\n";
    let payload = reference::envelope(raw);
    let len = payload.len() as u64;
    let crc = reference::crc(&payload);
    for (id, artifact, offset, length, checksum, count, expected) in [
        (
            0,
            2,
            0,
            len,
            crc,
            1,
            "source scan descriptor has unsupported artifact",
        ),
        (
            0,
            1,
            u64::MAX,
            len,
            crc,
            1,
            "source scan payload range overflow",
        ),
        (
            0,
            1,
            1,
            len,
            crc,
            1,
            "source scan payload range exceeds artifact",
        ),
        (
            0,
            1,
            0,
            len,
            crc ^ 1,
            1,
            "source scan payload checksum mismatch",
        ),
        (
            0,
            1,
            0,
            len,
            crc,
            2,
            "source scan payload row count mismatch",
        ),
        (
            1,
            1,
            0,
            len,
            crc,
            1,
            "scan segment manifest expected id 0, got 1",
        ),
    ] {
        let body = format!("HAWDB_SOURCE_SCAN_SEGMENTS_V1\ngraph_epoch\t7\nsegment\t{id}\t{artifact}\t{offset}\t{length}\t{checksum}\t{count}\n");
        let binding = write_reference(path, &body, &payload);
        assert_eq!(storage_error(load(path, 7, binding)), expected);
    }
    let body = format!("HAWDB_SOURCE_SCAN_SEGMENTS_V1\ngraph_epoch\t7\nsegment\t0\t1\t0\t{len}\t{crc}\t1\nsegment\t1\t1\t0\t{len}\t{crc}\t1\n");
    let binding = write_reference(path, &body, &payload);
    assert_eq!(
        storage_error(load(path, 7, binding)),
        "scan segment manifest has overlapping artifact 1"
    );
    let body = format!(
        "HAWDB_SOURCE_SCAN_SEGMENTS_V1\ngraph_epoch\t7\nsegment\t0\t1\t0\t{len}\t{crc}\t1\n"
    );
    let binding = write_reference(path, &body, &payload);
    assert!(load(path, 7, binding).unwrap().is_some());
    for length in [0, 1, payload.len() / 2, payload.len() - 1] {
        fs::write(path.join(SOURCE_SCAN_PAYLOAD_FILE), &payload[..length]).unwrap();
        assert_eq!(
            storage_error(load(path, 7, binding)),
            "source scan payload range exceeds artifact"
        );
    }
}

fn source(id: u64, source_label: LabelId, scope: &str) -> NodeRecord {
    NodeRecord {
        id: NodeId(id),
        labels: BTreeSet::from([source_label]),
        properties: BTreeMap::from([
            ("id".to_string(), Value::String(format!("source-{id}"))),
            ("space_id".to_string(), Value::String(scope.to_string())),
            (
                "created_at".to_string(),
                Value::String("2026-08-01T00:00:00Z".to_string()),
            ),
        ]),
    }
}

#[test]
fn sidecar_round_trips_payload_ranges_and_exact_cursors() {
    let directory = TestDir::new();
    let directory = directory.path();
    let source_label = LabelId(7);
    let nodes = [
        source(1, source_label, "alpha"),
        source(2, source_label, "beta"),
    ];
    let mut projection = build(4, Some(source_label), nodes.iter());
    let publication = write(directory, &mut projection).unwrap();

    let manifest = load(directory, 4, publication.descriptor_checksum())
        .unwrap()
        .unwrap();
    assert_eq!(manifest.graph_epoch(), 4);
    assert_eq!(manifest.segments().len(), 1);
    let plan = manifest.plan_scan(
        4,
        &crate::scan::ScanPredicate::Eq {
            property: "id".to_string(),
            value: Value::String("source-1".to_string()),
        },
    );
    let crate::scan::ScanSegmentAccessPlan::Read(plan) = plan else {
        panic!("expected scan");
    };
    assert_eq!(plan.segments.len(), 1);
    assert_eq!(plan.segments[0].candidates.as_ref().unwrap().remaining(), 1);
}

struct TestDir(std::path::PathBuf);
impl TestDir {
    fn new() -> Self {
        static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let serial = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let path = std::env::temp_dir().join(format!(
            "hawdb-source-scan-{}-{nonce}-{serial}",
            std::process::id()
        ));
        fs::create_dir(&path).unwrap();
        Self(path)
    }
    fn path(&self) -> &Path {
        &self.0
    }
}
impl Drop for TestDir {
    fn drop(&mut self) {
        let _ = fs::remove_dir_all(&self.0);
    }
}