offline-protocol-data 0.30.0

//! Unit tests for the replicated-document layer.

use crate::doc::{CatchUp, DataDoc, ImportOutcome, RemoteImport};
use crate::error::DataError;
use crate::policy;
use crate::value::DataValue;
use crate::MAX_VALUE_BYTES;

fn doc_with_one_set() -> DataDoc {
    let mut doc = DataDoc::new();
    doc.map_set("profile", "name", DataValue::text("ada"))
        .expect("set");
    doc
}

#[test]
fn a_map_round_trips_through_a_commit() {
    let mut doc = doc_with_one_set();
    let delta = doc.commit().expect("commit").expect("a change");
    assert!(!delta.bytes.is_empty());
    assert_eq!(
        doc.map_get("profile", "name").expect("get"),
        Some(DataValue::text("ada"))
    );
}

#[test]
fn committing_twice_reports_no_second_change() {
    let mut doc = doc_with_one_set();
    doc.commit().expect("commit").expect("a change");
    assert!(doc.commit().expect("commit").is_none());
}

#[test]
fn a_delta_applied_to_a_second_replica_reproduces_the_state() {
    let mut source = doc_with_one_set();
    let delta = source.commit().expect("commit").expect("a change");

    let mut replica = DataDoc::new();
    replica.import(&delta.bytes).expect("import");

    assert_eq!(
        replica.map_get("profile", "name").expect("get"),
        Some(DataValue::text("ada"))
    );
}

#[test]
fn importing_the_same_delta_twice_is_a_no_op() {
    let mut source = DataDoc::new();
    source.list_push("log", DataValue::int(1)).expect("push");
    let delta = source.commit().expect("commit").expect("a change");

    let mut replica = DataDoc::new();
    replica.import(&delta.bytes).expect("first import");
    replica.import(&delta.bytes).expect("second import");

    // At-least-once delivery is the ladder's guarantee, so a duplicate has
    // to be free rather than merely tolerated.
    assert_eq!(replica.list_len("log").expect("len"), 1);
}

#[test]
fn deltas_applied_out_of_order_converge() {
    let mut source = DataDoc::new();
    source.text_insert("body", 0, "hello").expect("insert");
    let first = source.commit().expect("commit").expect("a change");
    source.text_insert("body", 5, " world").expect("insert");
    let second = source.commit().expect("commit").expect("a change");

    let mut replica = DataDoc::new();
    // Deliberately reversed: an unordered transport is exactly what this
    // layer is designed to ride.
    replica.import(&second.bytes).expect("import second");
    replica.import(&first.bytes).expect("import first");

    assert_eq!(replica.text_value("body").expect("text"), "hello world");
}

#[test]
fn concurrent_edits_on_two_replicas_merge_both_ways() {
    let mut left = DataDoc::new();
    left.map_set("settings", "theme", DataValue::text("dark"))
        .expect("set");
    let seed = left.commit().expect("commit").expect("a change");

    let mut right = DataDoc::new();
    right.import(&seed.bytes).expect("seed");

    left.map_set("settings", "font", DataValue::text("mono"))
        .expect("set");
    let from_left = left.commit().expect("commit").expect("a change");

    right
        .counter_increment("visits", 3.0)
        .expect("counter increment");
    let from_right = right.commit().expect("commit").expect("a change");

    left.import(&from_right.bytes).expect("import right");
    right.import(&from_left.bytes).expect("import left");

    for (label, replica) in [("left", &left), ("right", &right)] {
        assert_eq!(
            replica.map_get("settings", "theme").expect("get"),
            Some(DataValue::text("dark")),
            "{label} lost the seed"
        );
        assert_eq!(
            replica.map_get("settings", "font").expect("get"),
            Some(DataValue::text("mono")),
            "{label} lost the left edit"
        );
        assert_eq!(
            replica.counter_value("visits").expect("counter"),
            3.0,
            "{label} lost the right edit"
        );
    }
}

#[test]
fn a_compacted_snapshot_restores_the_document() {
    let mut source = DataDoc::new();
    for index in 0..200 {
        source
            .map_set("items", &format!("k{index}"), DataValue::int(index))
            .expect("set");
        if index % 50 == 0 {
            source.commit().expect("commit");
        }
    }
    source.commit().expect("commit");
    let compacted = source.export_compacted().expect("compact");

    let mut restored = DataDoc::new();
    restored.import(&compacted).expect("import compacted");
    assert_eq!(
        restored.map_get("items", "k199").expect("get"),
        Some(DataValue::int(199))
    );
}

#[test]
fn mark_persisted_stops_history_being_re_emitted() {
    let mut doc = doc_with_one_set();
    doc.commit().expect("commit").expect("a change");
    let compacted = doc.export_compacted().expect("compact");

    // Re-open the way the store does: load the record, then declare it
    // persisted. A commit right after must not hand back the history that
    // was just read off disk.
    let mut reopened = DataDoc::new();
    reopened.import(&compacted).expect("import");
    reopened.mark_persisted();
    assert!(reopened.commit().expect("commit").is_none());

    reopened
        .map_set("profile", "name", DataValue::text("grace"))
        .expect("set");
    let delta = reopened.commit().expect("commit").expect("a change");
    assert!(!delta.bytes.is_empty());
}

#[test]
fn a_reopened_document_continues_the_delta_chain() {
    let mut doc = doc_with_one_set();
    doc.commit().expect("commit");
    let compacted = doc.export_compacted().expect("compact");

    let mut reopened = DataDoc::new();
    reopened.import(&compacted).expect("import");
    reopened.mark_persisted();
    reopened
        .map_set("profile", "city", DataValue::text("london"))
        .expect("set");
    let delta = reopened.commit().expect("commit").expect("a change");

    // A peer holding only the compacted record must be brought up to date
    // by that delta alone.
    let mut peer = DataDoc::new();
    peer.import(&compacted).expect("import compacted");
    peer.import(&delta.bytes).expect("import delta");
    assert_eq!(
        peer.map_get("profile", "city").expect("get"),
        Some(DataValue::text("london"))
    );
}

#[test]
fn corrupt_bytes_are_refused_rather_than_applied() {
    let mut doc = DataDoc::new();
    let err = doc
        .import(b"this is not an encoded document")
        .expect_err("garbage must be refused");
    assert!(matches!(err, DataError::Corrupt(_)), "got {err:?}");
}

#[test]
fn a_truncated_delta_is_refused() {
    let mut source = doc_with_one_set();
    let delta = source.commit().expect("commit").expect("a change");
    let truncated = &delta.bytes[..delta.bytes.len() / 2];

    let mut replica = DataDoc::new();
    let err = replica
        .import(truncated)
        .expect_err("truncation is corrupt");
    assert!(matches!(err, DataError::Corrupt(_)), "got {err:?}");
}

#[test]
fn a_bit_flip_in_a_delta_is_refused() {
    let mut source = doc_with_one_set();
    let delta = source.commit().expect("commit").expect("a change");
    let mut flipped = delta.bytes.clone();
    let last = flipped.len() - 1;
    flipped[last] ^= 0xff;

    let mut replica = DataDoc::new();
    assert!(replica.import(&flipped).is_err(), "a flipped byte applied");
}

#[test]
fn every_value_kind_survives_a_round_trip() {
    let mut doc = DataDoc::new();
    let cases = [
        ("null", DataValue::Null),
        ("bool", DataValue::bool(true)),
        ("int", DataValue::int(-42)),
        ("float", DataValue::float(1.5)),
        ("text", DataValue::text("hello")),
        ("bytes", DataValue::bytes(vec![0u8, 1, 255])),
        (
            "attachment",
            DataValue::attachment(
                "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
                4096,
            ),
        ),
    ];
    for (key, value) in &cases {
        doc.map_set("values", key, value.clone()).expect("set");
    }
    let delta = doc.commit().expect("commit").expect("a change");

    let mut replica = DataDoc::new();
    replica.import(&delta.bytes).expect("import");
    for (key, value) in &cases {
        assert_eq!(
            replica.map_get("values", key).expect("get").as_ref(),
            Some(value),
            "{key} did not survive"
        );
    }
}

#[test]
fn deleting_a_map_key_removes_it() {
    let mut doc = doc_with_one_set();
    doc.commit().expect("commit");
    doc.map_delete("profile", "name").expect("delete");
    doc.commit().expect("commit");
    assert_eq!(doc.map_get("profile", "name").expect("get"), None);
}

#[test]
fn list_positions_outside_the_collection_are_refused() {
    let mut doc = DataDoc::new();
    doc.list_push("log", DataValue::int(1)).expect("push");

    let err = doc
        .list_insert("log", 5, DataValue::int(2))
        .expect_err("out of range");
    assert!(matches!(err, DataError::OutOfRange { .. }), "got {err:?}");

    let err = doc.list_delete("log", 0, 4).expect_err("out of range");
    assert!(matches!(err, DataError::OutOfRange { .. }), "got {err:?}");
}

#[test]
fn text_positions_outside_the_collection_are_refused() {
    let mut doc = DataDoc::new();
    doc.text_insert("body", 0, "hi").expect("insert");

    let err = doc.text_insert("body", 9, "x").expect_err("out of range");
    assert!(matches!(err, DataError::OutOfRange { .. }), "got {err:?}");

    let err = doc.text_delete("body", 1, 8).expect_err("out of range");
    assert!(matches!(err, DataError::OutOfRange { .. }), "got {err:?}");
}

#[test]
fn names_that_would_break_a_record_key_are_refused() {
    for bad in ["", "a/b", "../etc", "with space", &"x".repeat(129)] {
        assert!(
            crate::validate_name(bad).is_err(),
            "{bad:?} should be refused"
        );
    }
    for good in ["docs", "my-doc", "my_doc.v2", "A1"] {
        crate::validate_name(good).expect("should be accepted");
    }
}

#[test]
fn a_bad_collection_name_is_refused_at_the_operation() {
    let mut doc = DataDoc::new();
    let err = doc
        .map_set("space/doc", "k", DataValue::int(1))
        .expect_err("separator in a collection name");
    assert!(matches!(err, DataError::InvalidName { .. }), "got {err:?}");
}

#[test]
fn oversized_map_keys_are_refused() {
    let mut doc = DataDoc::new();
    let key = "k".repeat(crate::MAX_KEY_LEN + 1);
    assert!(doc.map_set("m", &key, DataValue::int(1)).is_err());
    assert!(doc.map_set("m", "", DataValue::int(1)).is_err());
}

#[test]
fn json_export_shows_the_current_state() {
    let mut doc = DataDoc::new();
    doc.map_set("profile", "name", DataValue::text("ada"))
        .expect("set");
    doc.text_insert("body", 0, "notes").expect("insert");
    doc.commit().expect("commit");

    let json = doc.export_json().expect("json");
    let parsed: serde_json::Value = serde_json::from_str(&json).expect("valid json");
    assert_eq!(parsed["profile"]["name"], serde_json::json!("ada"));
    assert_eq!(parsed["body"], serde_json::json!("notes"));
}

#[test]
fn raw_export_restores_a_document() {
    let mut doc = doc_with_one_set();
    doc.commit().expect("commit");
    let raw = doc.export_raw().expect("raw");

    let mut restored = DataDoc::new();
    restored.import(&raw).expect("import raw");
    assert_eq!(
        restored.map_get("profile", "name").expect("get"),
        Some(DataValue::text("ada"))
    );
}

#[test]
fn a_single_field_delta_stays_inside_one_bluetooth_fragment() {
    // The measured figure the engine choice rests on: one map set encodes
    // to well under the 139 usable bytes of an unnegotiated BLE fragment.
    // If a change here pushes a single-field update past that, the layer
    // stops being usable on the weakest transport and the engine decision
    // has to be re-opened rather than absorbed.
    let mut doc = DataDoc::new();
    doc.map_set("m", "k", DataValue::int(1)).expect("set");
    let delta = doc.commit().expect("commit").expect("a change");
    assert!(
        delta.bytes.len() < 139,
        "single-field delta grew to {} bytes",
        delta.bytes.len()
    );
}

#[test]
fn compaction_keeps_a_busy_document_far_smaller_than_its_history() {
    let mut doc = DataDoc::new();
    for index in 0..2_000 {
        doc.map_set("counters", "hot", DataValue::int(index))
            .expect("set");
        if index % 50 == 0 {
            doc.commit().expect("commit");
        }
    }
    doc.commit().expect("commit");

    let compacted = doc.export_compacted().expect("compact").len();
    let full = doc.export_raw().expect("raw").len();
    assert!(
        compacted * 4 < full,
        "compaction bought little: {compacted} vs {full}"
    );
}

#[test]
fn size_checks_stay_quiet_for_a_small_document() {
    let mut doc = doc_with_one_set();
    doc.commit().expect("commit");
    let (verdict, measured) = doc.check_size().expect("check");
    assert_eq!(verdict, policy::SizeVerdict::Ok);
    assert!(
        measured.is_none(),
        "a small document should not pay for a measurement"
    );
}

#[test]
fn a_rewound_commit_is_handed_out_again() {
    // The failure this pins: a caller whose persistence failed has already
    // had the export marker advanced by commit(), so without a rewind the
    // change is never offered again. The document still shows it, so nothing
    // looks wrong until a restart, when it is simply not there.
    let mut doc = DataDoc::new();
    doc.map_set("m", "k", DataValue::text("first"))
        .expect("set");
    let first = doc.commit().expect("commit").expect("a change");

    doc.rewind_last_commit();

    doc.map_set("m", "k2", DataValue::text("second"))
        .expect("set");
    let second = doc.commit().expect("commit").expect("a change");

    // The second delta must carry BOTH changes: the rewound one and the new
    // one. A replica that only ever saw the second delta has to end up whole.
    let mut replica = DataDoc::new();
    replica.import(&second.bytes).expect("import");
    assert_eq!(
        replica.map_get("m", "k").expect("get"),
        Some(DataValue::text("first")),
        "the rewound change was not re-exported"
    );
    assert_eq!(
        replica.map_get("m", "k2").expect("get"),
        Some(DataValue::text("second"))
    );
    // And the original delta stays valid, so a rewind after a write that
    // actually succeeded costs a duplicate, not a conflict.
    replica.import(&first.bytes).expect("re-import the first");
    assert_eq!(
        replica.map_get("m", "k").expect("get"),
        Some(DataValue::text("first"))
    );
}

#[test]
fn rewinding_without_a_commit_is_harmless() {
    let mut doc = DataDoc::new();
    doc.rewind_last_commit();
    doc.map_set("m", "k", DataValue::int(1)).expect("set");
    let delta = doc.commit().expect("commit").expect("a change");

    let mut replica = DataDoc::new();
    replica.import(&delta.bytes).expect("import");
    assert_eq!(
        replica.map_get("m", "k").expect("get"),
        Some(DataValue::int(1))
    );
}

#[test]
fn a_reopened_document_does_not_forget_how_big_it_is() {
    // The bug this pins: a document loaded from a stored snapshot starts
    // with a zeroed size estimate, so the cheap check answers "no need to
    // measure" no matter how large it actually is. A document already near
    // the cap would then sail past it, and the first symptom would be a
    // record that will not fit.
    let mut doc = DataDoc::new();
    doc.map_set("m", "k", DataValue::int(1)).expect("set");
    doc.commit().expect("commit");
    let compacted = doc.export_compacted().expect("compact");

    let mut reopened = DataDoc::new();
    reopened.import(&compacted).expect("import");
    reopened.mark_persisted();

    // Stand in for a document loaded at three quarters of the cap.
    reopened.restore_bookkeeping(policy::DOC_SIZE_WARN_BYTES, 0, 0);
    let (_, measured) = reopened.check_size().expect("check");
    assert!(
        measured.is_some(),
        "a document restored near the cap must be measured, not assumed small"
    );
}

#[test]
fn restored_bookkeeping_counts_the_delta_log_toward_the_estimate() {
    let mut doc = DataDoc::new();
    doc.map_set("m", "k", DataValue::int(1)).expect("set");
    doc.commit().expect("commit");

    // A small stored snapshot with a large replayed log is still a large
    // document: the estimate has to add them, or the log is invisible until
    // the next compaction.
    doc.restore_bookkeeping(1024, policy::DOC_SIZE_WARN_BYTES, 7);
    let (_, measured) = doc.check_size().expect("check");
    assert!(measured.is_some(), "the replayed log was not counted");
    assert_eq!(
        doc.commits_since_compaction(),
        7,
        "the replayed commit count drives the compaction trigger"
    );
}

#[test]
fn stats_report_the_commit_count_since_compaction() {
    let mut doc = DataDoc::new();
    for index in 0..3 {
        doc.map_set("m", "k", DataValue::int(index)).expect("set");
        doc.commit().expect("commit");
    }
    let stats = doc.stats().expect("stats");
    assert_eq!(stats.commits_since_compaction, 3);
    assert!(stats.compacted_bytes > 0);

    doc.mark_compacted(stats.compacted_bytes);
    assert_eq!(doc.commits_since_compaction(), 0);
}

#[test]
fn a_delta_whose_predecessor_is_missing_is_reported_as_parked() {
    let mut source = DataDoc::new();
    source.map_set("m", "k1", DataValue::int(1)).expect("set");
    let first = source.commit().expect("commit").expect("a delta");
    source.map_set("m", "k2", DataValue::int(2)).expect("set");
    let second = source.commit().expect("commit").expect("a delta");

    // Only the second delta arrives. The engine takes it and answers Ok, so
    // the outcome is the only thing that distinguishes "applied" from "held
    // out of sight" — and an owner that cannot tell them apart will compact
    // this document and delete the record it is waiting for.
    let mut replica = DataDoc::new();
    let outcome = replica.import(&second.bytes).expect("import");
    assert_eq!(
        outcome,
        ImportOutcome::Parked,
        "an orphan delta was reported as applied"
    );
    assert_eq!(
        replica.map_get("m", "k2").expect("get"),
        None,
        "a parked change must not read as present"
    );

    // And it must stay out of a compacted export, which is what makes
    // compacting-then-deleting destructive here.
    let compacted = replica.export_compacted().expect("export");
    let mut round_trip = DataDoc::new();
    round_trip.import(&compacted).expect("import");
    assert_eq!(round_trip.map_get("m", "k2").expect("get"), None);

    // Once the predecessor lands, both apply.
    let outcome = replica.import(&first.bytes).expect("import");
    assert_eq!(outcome, ImportOutcome::Applied);
    assert_eq!(
        replica.map_get("m", "k2").expect("get"),
        Some(DataValue::int(2)),
        "the parked change did not apply once its predecessor arrived"
    );
}

#[test]
fn a_complete_delta_is_reported_as_applied() {
    let mut source = DataDoc::new();
    source.map_set("m", "k", DataValue::int(7)).expect("set");
    let delta = source.commit().expect("commit").expect("a delta");

    let mut replica = DataDoc::new();
    assert_eq!(
        replica.import(&delta.bytes).expect("import"),
        ImportOutcome::Applied,
        "a delta with no missing predecessor was reported as parked"
    );
}

#[test]
fn a_value_larger_than_any_document_is_refused() {
    let mut doc = DataDoc::new();
    let huge = "x".repeat(MAX_VALUE_BYTES + 1);

    assert!(matches!(
        doc.map_set("m", "k", DataValue::text(huge.clone())),
        Err(DataError::ValueTooLarge { .. })
    ));
    assert!(matches!(
        doc.list_push("l", DataValue::bytes(vec![0u8; MAX_VALUE_BYTES + 1])),
        Err(DataError::ValueTooLarge { .. })
    ));
    assert!(matches!(
        doc.list_insert("l", 0, DataValue::text(huge.clone())),
        Err(DataError::ValueTooLarge { .. })
    ));
    assert!(matches!(
        doc.text_insert("t", 0, &huge),
        Err(DataError::ValueTooLarge { .. })
    ));

    // The refusal is at the operation, so nothing reached the document.
    assert_eq!(doc.map_get("m", "k").expect("get"), None);
    assert!(doc.commit().expect("commit").is_none());
}

/// A document holding `commits` separate commits, plus the deltas it made,
/// in commit order.
fn source_with_commits(commits: usize) -> (DataDoc, Vec<Vec<u8>>) {
    let mut doc = DataDoc::new();
    let mut deltas = Vec::new();
    for n in 0..commits {
        doc.map_set("m", &format!("k{n}"), DataValue::int(n as i64))
            .expect("set");
        deltas.push(doc.commit().expect("commit").expect("a delta").bytes);
    }
    (doc, deltas)
}

/// A replica that has applied `deltas` and then compacted, so its history is
/// trimmed at its current frontier: the shape a document has after every
/// ordinary compaction pass, and the shape the engine mishandles.
fn compacted_replica(deltas: &[Vec<u8>]) -> DataDoc {
    let mut staging = DataDoc::new();
    for delta in deltas {
        staging.import(delta).expect("import");
    }
    let compacted = staging.export_compacted().expect("compact");

    let mut replica = DataDoc::new();
    replica.import(&compacted).expect("import compacted");
    replica
}

#[test]
fn a_redelivered_delta_never_reaches_a_compacted_document() {
    let (_source, deltas) = source_with_commits(2);
    let mut replica = compacted_replica(&deltas);

    // The ladder promises at-least-once, so this redelivery is ordinary
    // traffic and not an attack. It is also the exact shape that refers to
    // history the compaction trimmed away, which the engine answers with a
    // panic rather than an error (loro #1068) and which a host built with
    // `panic = "abort"` turns into a dead process.
    let meta = replica.inspect(&deltas[1]).expect("inspect");
    assert!(
        meta.already_applied(),
        "a delta already folded into the snapshot was not recognized"
    );
    assert_eq!(
        replica.import_remote(&deltas[1]).expect("import remote"),
        RemoteImport::AlreadyHave,
        "a redelivered delta was handed to the engine"
    );

    // The document is untouched and still readable, which is the point.
    assert_eq!(
        replica.map_get("m", "k1").expect("get"),
        Some(DataValue::int(1))
    );
    assert!(!replica.is_poisoned());
}

#[test]
fn a_blob_spanning_compacted_history_is_refused_rather_than_imported() {
    let (mut source, deltas) = source_with_commits(2);
    let mut replica = compacted_replica(&deltas);

    // A third change the replica has not seen, encoded from the beginning of
    // history: partly new, but reaching back below the replica's trim point.
    // No run of updates can express that gap, so the only safe answer is to
    // decline and ask for a snapshot.
    source.map_set("m", "k2", DataValue::int(2)).expect("set");
    source.commit().expect("commit").expect("a delta");
    let from_scratch = match source
        .export_since(&DataDoc::new().version())
        .expect("export since nothing")
    {
        CatchUp::Updates(bytes) => bytes,
        other => panic!("expected updates, got {other:?}"),
    };

    let meta = replica.inspect(&from_scratch).expect("inspect");
    assert!(
        !meta.already_applied(),
        "the blob carries a change the replica lacks"
    );
    assert!(
        meta.spans_trimmed_history(),
        "a blob reaching below the trim point was not recognized"
    );
    assert_eq!(
        replica.import_remote(&from_scratch).expect("import remote"),
        RemoteImport::RefusedTrimmedHistory,
        "a blob spanning trimmed history was handed to the engine"
    );
    assert!(!replica.is_poisoned());

    // A snapshot carries its own base, so it is the answer to that refusal.
    let snapshot = source.export_raw().expect("snapshot");
    assert!(replica.inspect(&snapshot).expect("inspect").is_snapshot());
    assert_eq!(
        replica.import_remote(&snapshot).expect("import snapshot"),
        RemoteImport::Applied
    );
    assert_eq!(
        replica.map_get("m", "k2").expect("get"),
        Some(DataValue::int(2)),
        "the snapshot did not close the gap"
    );
}

#[test]
fn a_fresh_remote_delta_applies_and_a_parked_one_is_reported() {
    let (_source, deltas) = source_with_commits(2);

    let mut replica = DataDoc::new();
    assert_eq!(
        replica.import_remote(&deltas[1]).expect("import"),
        RemoteImport::Parked,
        "an orphan remote delta was reported as applied"
    );
    assert_eq!(
        replica.import_remote(&deltas[0]).expect("import"),
        RemoteImport::Applied
    );
    assert_eq!(
        replica.map_get("m", "k1").expect("get"),
        Some(DataValue::int(1)),
        "the parked change did not apply once its predecessor arrived"
    );
}

#[test]
fn corrupt_remote_bytes_are_refused_without_poisoning_the_document() {
    let (_source, deltas) = source_with_commits(1);
    let mut replica = DataDoc::new();

    let mut corrupt = deltas[0].clone();
    let last = corrupt.len() - 1;
    corrupt[last] ^= 0xff;
    assert!(matches!(
        replica.import_remote(&corrupt),
        Err(DataError::Corrupt(_))
    ));
    assert!(matches!(
        replica.inspect(b"not a blob at all"),
        Err(DataError::Corrupt(_))
    ));

    // Refusing junk must leave the handle usable: a peer that can poison a
    // document by sending one bad frame has a denial of service.
    assert!(!replica.is_poisoned());
    assert_eq!(
        replica.import_remote(&deltas[0]).expect("import"),
        RemoteImport::Applied
    );
}

#[test]
fn catch_up_serves_exactly_what_a_peer_is_missing() {
    let (mut source, deltas) = source_with_commits(1);

    let mut peer = DataDoc::new();
    peer.import(&deltas[0]).expect("import");
    let peer_version = peer.version();

    // Nothing new yet.
    assert_eq!(
        source.export_since(&peer_version).expect("export since"),
        CatchUp::UpToDate
    );

    source
        .map_set("m", "later", DataValue::int(9))
        .expect("set");
    source.commit().expect("commit").expect("a delta");

    let updates = match source.export_since(&peer_version).expect("export since") {
        CatchUp::Updates(bytes) => bytes,
        other => panic!("expected updates, got {other:?}"),
    };
    assert_eq!(
        peer.import_remote(&updates).expect("import"),
        RemoteImport::Applied,
        "catch-up updates did not apply cleanly"
    );
    assert_eq!(
        peer.map_get("m", "later").expect("get"),
        Some(DataValue::int(9))
    );

    // And a peer that already has everything is told so rather than being
    // sent bytes: on a mesh the difference is a frame per idle reconnect.
    assert_eq!(
        source.export_since(&peer.version()).expect("export since"),
        CatchUp::UpToDate
    );
}

#[test]
fn catch_up_from_before_a_compaction_asks_for_a_snapshot() {
    let (_source, deltas) = source_with_commits(2);
    let mut replica = compacted_replica(&deltas);
    replica.map_set("m", "own", DataValue::int(5)).expect("set");
    replica.commit().expect("commit").expect("a delta");

    // A peer that has nothing cannot be served from a document whose early
    // history was compacted away. Answering with updates anyway is the
    // sending half of the same defect `import_remote` declines.
    assert_eq!(
        replica
            .export_since(&DataDoc::new().version())
            .expect("export since nothing"),
        CatchUp::NeedsSnapshot
    );
}

#[test]
fn a_malformed_version_token_is_refused_rather_than_answered_with_everything() {
    let (source, _deltas) = source_with_commits(1);
    let garbled = crate::doc::VersionToken::from_bytes(vec![0xff; 16]);

    // Treating an undecodable token as "knows nothing" would make one
    // corrupt byte an amplification lever on the sync path.
    assert!(matches!(
        source.export_since(&garbled),
        Err(DataError::Corrupt(_))
    ));
}

#[test]
fn a_branch_forked_below_the_trim_point_never_reaches_the_engine() {
    // The shape loro #1068 names, and the reason this gate is structural
    // rather than defensive. Verified on 1.13.9 by removing the refusal:
    // the engine panics in `pending_changes.rs` and the retry panics again
    // on the mutex that first panic poisoned. Under `panic = "abort"`
    // there is no catch to reach at all.
    let (_source, deltas) = source_with_commits(2);

    // A replica that saw only the first change, then edited concurrently.
    // Its change depends on history the compacted replica has trimmed and
    // is not an ancestor of anything that replica holds.
    let mut fork = DataDoc::new();
    fork.import(&deltas[0]).expect("seed the fork");
    fork.map_set("m", "fork", DataValue::int(42)).expect("set");
    let fork_delta = fork.commit().expect("commit").expect("a delta").bytes;

    let mut replica = compacted_replica(&deltas);
    assert!(
        replica
            .inspect(&fork_delta)
            .expect("inspect")
            .spans_trimmed_history(),
        "a branch forked below the trim point was not recognized"
    );
    assert_eq!(
        replica.import_remote(&fork_delta).expect("import remote"),
        RemoteImport::RefusedTrimmedHistory,
        "a branch forked below the trim point was handed to the engine"
    );

    // The document survives, which is the whole point: refusing costs one
    // extra round trip, and importing costs the process.
    assert!(!replica.is_poisoned());
    assert_eq!(
        replica.map_get("m", "k1").expect("get"),
        Some(DataValue::int(1))
    );
}

#[test]
fn an_ordinary_delta_after_a_compaction_still_applies() {
    // The other half of the trim gate: it must refuse the forked branch
    // above without refusing steady-state traffic. A gate that answered
    // "send a snapshot" to every delta a synced peer produces after the
    // first compaction would be correct and useless.
    let mut alice = DataDoc::new();
    alice.map_set("m", "a", DataValue::int(0)).expect("set");
    let from_alice = alice.commit().expect("commit").expect("a delta").bytes;

    let mut bob = DataDoc::new();
    bob.import(&from_alice).expect("import");
    bob.map_set("m", "b", DataValue::int(1)).expect("set");
    let from_bob = bob.commit().expect("commit").expect("a delta").bytes;
    alice.import(&from_bob).expect("import");

    // Alice compacts, as every flush eventually does.
    let compacted = alice.export_compacted().expect("compact");
    let mut alice = DataDoc::new();
    alice.import(&compacted).expect("reopen compacted");

    bob.map_set("m", "b2", DataValue::int(2)).expect("set");
    let next = bob.commit().expect("commit").expect("a delta").bytes;

    assert_eq!(
        alice.import_remote(&next).expect("import remote"),
        RemoteImport::Applied,
        "an ordinary delta was refused because the receiver had compacted"
    );
    assert_eq!(
        alice.map_get("m", "b2").expect("get"),
        Some(DataValue::int(2))
    );
}

#[test]
fn a_snapshot_closes_a_trim_gap_only_when_the_sender_holds_what_we_kept() {
    // The two halves of what refusing a blob can and cannot buy, pinned
    // together because the difference is invisible from the frame and
    // decides whether the peers ever agree again.

    // Healable: the sender's history contains everything the compacted
    // replica retained, so its snapshot connects to what is left.
    let (mut source, deltas) = source_with_commits(2);
    let mut replica = compacted_replica(&deltas);
    source.map_set("m", "k2", DataValue::int(2)).expect("set");
    source.commit().expect("commit").expect("a delta");
    let snapshot = source.export_raw().expect("snapshot");
    assert_eq!(
        replica.import_remote(&snapshot).expect("import"),
        RemoteImport::Applied,
        "a snapshot from a sender that is strictly ahead must close the gap"
    );

    // Unhealable: a replica that forked *below* the trim point holds
    // changes whose ancestors this one deleted. No blob can carry them,
    // including a whole document, because the dependencies are gone from
    // the receiver rather than missing from the message. The refusal is
    // still right — the alternative is aborting the process — but it buys
    // termination, not convergence, and a caller must not report otherwise.
    let mut fork = DataDoc::new();
    fork.import(&deltas[0]).expect("seed the fork");
    fork.map_set("m", "fork", DataValue::int(42)).expect("set");
    fork.commit().expect("commit").expect("a delta");

    let mut compacted = compacted_replica(&deltas);
    assert_eq!(
        compacted
            .import_remote(&fork.export_raw().expect("snapshot"))
            .expect("import remote"),
        RemoteImport::RefusedTrimmedHistory,
        "a snapshot of a branch forked below the trim point was handed to \
         the engine, which aborts on it exactly as it does on a delta"
    );
    assert!(!compacted.is_poisoned());
    assert_eq!(
        compacted.map_get("m", "k1").expect("get"),
        Some(DataValue::int(1)),
        "refusing must leave the document readable"
    );
}

// ---- attachments ------------------------------------------------------

const HASH_A: &str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
const HASH_B: &str = "5891b5b522d5df086d0ff0b110fbd9d21bb4fc7163af34d08286a2e846f6be03";

#[test]
fn an_attachment_reference_round_trips_through_a_commit() {
    let mut doc = DataDoc::new();
    let reference = DataValue::Attachment {
        hash: HASH_A.to_string(),
        size: 4096,
        name: Some("plan.pdf".to_string()),
        mime: Some("application/pdf".to_string()),
    };
    doc.map_set("files", "plan", reference.clone())
        .expect("set");
    doc.commit().expect("commit").expect("a change");

    let mut replica = DataDoc::new();
    replica
        .import(&doc.export_compacted().expect("export"))
        .expect("import");
    assert_eq!(
        replica.map_get("files", "plan").expect("get"),
        Some(reference)
    );
}

#[test]
fn an_attachment_with_no_name_or_type_round_trips_as_absent_fields() {
    let mut doc = DataDoc::new();
    doc.map_set("files", "blob", DataValue::attachment(HASH_A, 12))
        .expect("set");
    assert_eq!(
        doc.map_get("files", "blob").expect("get"),
        Some(DataValue::attachment(HASH_A, 12))
    );
}

#[test]
fn attaching_over_a_key_replaces_the_whole_reference() {
    // The single property that makes "no multi-writer attachment mutation"
    // true by construction: a reference is one value, so the second write
    // cannot leave a hash from one blob beside a size from another.
    let mut doc = DataDoc::new();
    doc.map_set("files", "blob", DataValue::attachment(HASH_A, 10))
        .expect("set");
    doc.map_set(
        "files",
        "blob",
        DataValue::Attachment {
            hash: HASH_B.to_string(),
            size: 20,
            name: Some("second".to_string()),
            mime: None,
        },
    )
    .expect("set");

    let Some(DataValue::Attachment {
        hash,
        size,
        name,
        mime,
    }) = doc.map_get("files", "blob").expect("get")
    else {
        panic!("expected an attachment");
    };
    assert_eq!(hash, HASH_B);
    assert_eq!(size, 20);
    assert_eq!(name.as_deref(), Some("second"));
    assert_eq!(mime, None);
}

#[test]
fn the_escape_hatch_describes_an_attachment_in_the_words_the_api_uses() {
    // `export_json` is what an application leaving this SDK reads, so a
    // reference has to describe itself there without a decoder ring. The
    // matching claim about older builds reading the value as absent is
    // pinned against the conversion itself, in `doc.rs`.
    let mut doc = DataDoc::new();
    doc.map_set("files", "blob", DataValue::attachment(HASH_A, 10))
        .expect("set");

    let exported = doc.export_json().expect("json");
    // Against the API's own spelling of the same value rather than against
    // a substring of it. `contains("attachment")` is true of "attachment2"
    // as well, so it pins none of the key names the claim is about: rename
    // any of them and the export diverges from the API while a substring
    // check stays green.
    let exported: serde_json::Value = serde_json::from_str(&exported).expect("the export is JSON");
    let reference = &exported["files"]["blob"];
    assert_eq!(
        reference,
        &serde_json::to_value(DataValue::attachment(HASH_A, 10)).expect("encode"),
        "the escape hatch must describe the reference in the API's own words: {exported}"
    );
}

#[test]
fn a_hash_that_is_not_a_hash_is_refused_at_the_operation() {
    let mut doc = DataDoc::new();
    for bad in [
        "",
        "abc",
        // Uppercase is refused rather than folded: two spellings of one
        // address would fetch twice and compare unequal.
        &HASH_A.to_uppercase(),
        // 64 characters, but not hex.
        &"z".repeat(64),
    ] {
        let err = doc
            .map_set("files", "blob", DataValue::attachment(bad, 1))
            .expect_err("a malformed hash must be refused");
        assert!(
            matches!(err, DataError::InvalidAttachment { .. }),
            "unexpected error for {bad:?}: {err}"
        );
    }
}

#[test]
fn an_attachment_with_no_bytes_is_refused() {
    let mut doc = DataDoc::new();
    let err = doc
        .map_set("files", "blob", DataValue::attachment(HASH_A, 0))
        .expect_err("a zero-length attachment must be refused");
    assert!(matches!(err, DataError::InvalidAttachment { .. }), "{err}");
}

#[test]
fn an_over_long_name_or_type_is_refused() {
    let mut doc = DataDoc::new();
    let err = doc
        .map_set(
            "files",
            "blob",
            DataValue::Attachment {
                hash: HASH_A.to_string(),
                size: 1,
                name: Some("n".repeat(crate::MAX_ATTACHMENT_NAME_LEN + 1)),
                mime: None,
            },
        )
        .expect_err("an over-long name must be refused");
    assert!(matches!(err, DataError::InvalidAttachment { .. }), "{err}");

    let err = doc
        .map_set(
            "files",
            "blob",
            DataValue::Attachment {
                hash: HASH_A.to_string(),
                size: 1,
                name: None,
                mime: Some("m".repeat(crate::MAX_ATTACHMENT_MIME_LEN + 1)),
            },
        )
        .expect_err("an over-long media type must be refused");
    assert!(matches!(err, DataError::InvalidAttachment { .. }), "{err}");
}

#[test]
fn a_size_the_engine_cannot_hold_is_refused_rather_than_clamped() {
    // The engine's integer is signed and this crate's is not, so a size past
    // `MAX_ATTACHMENT_SIZE` cannot survive the conversion. Refused at the
    // operation rather than clamped inside it: a clamped size replicates to
    // the whole space as a number its writer never wrote, and every member
    // then decides whether to fetch against it.
    let mut doc = DataDoc::new();
    let err = doc
        .map_set(
            "files",
            "blob",
            DataValue::attachment(HASH_A, crate::MAX_ATTACHMENT_SIZE + 1),
        )
        .expect_err("a size past the engine's integer must be refused");
    assert!(matches!(err, DataError::InvalidAttachment { .. }), "{err}");

    // The control: the largest size that DOES fit is accepted and reads back
    // unchanged. Without it the assertion above passes for a bound that
    // refuses everything.
    doc.map_set(
        "files",
        "blob",
        DataValue::attachment(HASH_A, crate::MAX_ATTACHMENT_SIZE),
    )
    .expect("the largest representable size must be accepted");
    assert_eq!(
        doc.map_get("files", "blob").expect("get"),
        Some(DataValue::attachment(HASH_A, crate::MAX_ATTACHMENT_SIZE))
    );
}

#[test]
fn an_attachment_reference_serializes_in_the_shape_the_ffi_carries() {
    // The canonical FFI and wire spelling. Absent name and type are absent,
    // not null: a golden vector pins this same shape, and a reader that has
    // to tell `null` from missing has been given two spellings of one value.
    let json = serde_json::to_string(&DataValue::attachment(HASH_A, 7)).expect("json");
    assert_eq!(
        json,
        format!(r#"{{"kind":"attachment","hash":"{HASH_A}","size":7}}"#)
    );

    let round_tripped: DataValue = serde_json::from_str(&json).expect("parse");
    assert_eq!(round_tripped, DataValue::attachment(HASH_A, 7));
}

#[test]
fn an_attachment_reference_is_a_list_value_too() {
    let mut doc = DataDoc::new();
    doc.list_push("gallery", DataValue::attachment(HASH_A, 1))
        .expect("push");
    doc.list_push("gallery", DataValue::attachment(HASH_B, 2))
        .expect("push");
    assert_eq!(doc.list_len("gallery").expect("len"), 2);
}

#[test]
fn a_name_or_type_at_the_bound_is_accepted() {
    // The control the refusal test needs. A check written `>=` refuses the
    // longest legal name, which the spec says MUST be accepted, and the
    // refusal test alone cannot tell the two spellings apart.
    let mut doc = DataDoc::new();
    doc.map_set(
        "files",
        "blob",
        DataValue::Attachment {
            hash: HASH_A.to_string(),
            size: 1,
            name: Some("n".repeat(crate::MAX_ATTACHMENT_NAME_LEN)),
            mime: Some("m".repeat(crate::MAX_ATTACHMENT_MIME_LEN)),
        },
    )
    .expect("a name and type exactly at the bound must be accepted");

    let Some(DataValue::Attachment { name, mime, .. }) = doc.map_get("files", "blob").expect("get")
    else {
        panic!("expected an attachment");
    };
    assert_eq!(
        name.map(|name| name.len()),
        Some(crate::MAX_ATTACHMENT_NAME_LEN)
    );
    assert_eq!(
        mime.map(|mime| mime.len()),
        Some(crate::MAX_ATTACHMENT_MIME_LEN)
    );
}

// A removal floor is a version vector, and the three questions a deletion
// asks of one are all `includes`: is this blob content the removal already
// covered, is this offered version, and is the copy we hold. The tests
// below pin each answer against the engine rather than against a mental
// model of version vectors, because the whole feature rests on them.

#[test]
fn a_version_covers_the_changes_it_was_taken_after() {
    let (source, deltas) = source_with_commits(2);
    let floor = source.version();

    // Every delta that went into the document is content the floor covers.
    for delta in &deltas {
        let end = crate::blob_end_version(delta).expect("end version");
        assert!(
            floor.includes(&end).expect("includes"),
            "a delta taken before the floor read as beyond it"
        );
    }
}

#[test]
fn a_change_made_after_a_floor_reads_as_beyond_it() {
    let (mut source, _deltas) = source_with_commits(1);
    let floor = source.version();

    source
        .map_set("m", "later", DataValue::text("after the removal"))
        .expect("set");
    let delta = source.commit().expect("commit").expect("a delta");
    let end = crate::blob_end_version(&delta.bytes).expect("end version");

    assert!(
        !floor.includes(&end).expect("includes"),
        "an edit made after the floor read as covered by it"
    );
    // And the floor is not beyond the edit either: the edit supersedes it.
    assert!(end.includes(&floor).expect("includes"));
}

#[test]
fn a_snapshot_of_a_fresh_document_is_beyond_every_earlier_floor() {
    // Re-creating a removed name starts a document whose history is
    // disjoint from the floor, which is what makes it alive by the same
    // rule that keeps a concurrent edit alive: nothing about it is covered.
    let (source, _deltas) = source_with_commits(1);
    let floor = source.version();

    let mut fresh = DataDoc::new();
    fresh
        .map_set("m", "k", DataValue::text("second incarnation"))
        .expect("set");
    fresh.commit().expect("commit").expect("a delta");
    let snapshot = fresh.export_compacted().expect("compact");

    let end = crate::blob_end_version(&snapshot).expect("end version");
    assert!(
        !floor.includes(&end).expect("includes"),
        "a document created after a removal read as covered by the removal"
    );
}

#[test]
fn an_empty_version_covers_nothing_and_round_trips() {
    let empty = DataDoc::new().version();
    let (source, _deltas) = source_with_commits(1);

    assert!(
        empty.includes(&empty).expect("includes"),
        "a version did not cover itself"
    );
    assert!(!empty.includes(&source.version()).expect("includes"));
    // Every version covers the empty one, so a floor taken before any
    // change never deletes anything on its own.
    assert!(source.version().includes(&empty).expect("includes"));
}

#[test]
fn two_independent_removals_merge_into_one_floor() {
    // Two replicas edit apart and each removes its own copy. Neither floor
    // covers the other's change; the union covers both, which is what
    // stops the order the offers arrive in from deciding the outcome.
    let (mut left, deltas) = source_with_commits(1);
    let mut right = DataDoc::new();
    right.import(&deltas[0]).expect("import");

    left.map_set("m", "l", DataValue::text("left"))
        .expect("set");
    let left_delta = left.commit().expect("commit").expect("a delta");
    right
        .map_set("m", "r", DataValue::text("right"))
        .expect("set");
    let right_delta = right.commit().expect("commit").expect("a delta");

    let left_floor = left.version();
    let right_floor = right.version();
    assert!(!left_floor.includes(&right_floor).expect("includes"));
    assert!(!right_floor.includes(&left_floor).expect("includes"));

    let merged = left_floor.union(&right_floor).expect("union");
    for delta in [&left_delta.bytes, &right_delta.bytes] {
        let end = crate::blob_end_version(delta).expect("end version");
        assert!(
            merged.includes(&end).expect("includes"),
            "the merged floor did not cover a change one of the removals saw"
        );
    }
}

#[test]
fn a_floor_question_about_unreadable_bytes_is_an_error_not_an_answer() {
    let floor = DataDoc::new().version();
    let garbage = crate::VersionToken::from_bytes(vec![0xff, 0xfe, 0xfd, 0xfc]);

    assert!(matches!(
        floor.includes(&garbage),
        Err(DataError::Corrupt(_))
    ));
    assert!(matches!(
        crate::blob_end_version(&[0xff, 0xfe, 0xfd]),
        Err(DataError::Corrupt(_))
    ));
}