Repository navigation
Honest import panics after an import rejected by state validation was rolled back (arena values) #1161
Description
Summary
After an import is rejected by state validation (e.g. a List insert past the end, rejected by ListState::validate_diff) and rolled back, a later honest import can panic:
diff_calc.rs:1914:arena.get_value(..).unwrap()inMovableListDiffCalculator::apply_change, orarena.rs:519: slice out of range.
The panic happens with the doc locks held, so the doc is poisoned and the process aborts in the destructor.
In the round-2 adversarial review of #1125, forged plain-List out-of-bounds changes were injected into a random multi-peer workload, and about 3% of seeds aborted later. main and the PR fail on the same seeds (72, 85, 109, 199, 246, 283, 309, 323, 329, 338, 339; main also 269). Rejections that happen before diff calculation (#1125's movable-list element checks) never triggered it (0/400 seeds). Pre-existing, not fixed in #1125.
Reproduction (main 6e294c87, debug build)
I could not reduce it to a small standalone test yet. It reproduces deterministically with the review harness below:
- Save it as
crates/loro/tests/review_1125_diff.rs. - Run:
R1125_FORGE=1 R1125_TRACE=1 R1125_SEED_START=120 R1125_SEEDS=1 R1125_OUT=/tmp/s120.log \
cargo test -p loro --test review_1125_diff -- --nocaptureWhat seed 120 does:
- Step 81 imports a forged JSON change into peer 0: peer 3's next counter, a
cid:root-pl:Listinsert at pos 99999. It is rejected withlist diff retains 99999 items but state only has 0and rolled back. - Ten steps later, peer 0 imports honest chunks while detached and calls
attach().attach→checkout_to_latest→calc_diff_internal→MovableListDiffCalculator::apply_changepanics atdiff_calc.rs:1914(called Option::unwrap() on a None value). - The release build of the same seed did not panic in my run, and the reviewer's seed list was from their build, so a debug build is the reliable way to reproduce.
Hypothesis (unconfirmed)
OpLog::rollback_import truncates the arena's values back to the checkpoint taken before the import. Anything that allocated values during the scope and survives the rollback then holds slices past the end of the arena. One candidate: change blocks parsed lazily from the KV store during the rejected import's diff calculation. They stay cached in mem_parsed_kv, and their ops' value slices point at the truncated range. A later diff reads them through arena.get_value(..). A targeted probe (snapshot-loaded doc, concurrent local edit, rejected forged List insert, then a checkout replay) did not reproduce it, so the path may be different.
Harness (review_1125_diff.rs, temporary review tool)
//! TEMPORARY randomized main-vs-PR differential for PR #1125 review (round 2). Do not commit.
//! Env: R1125_OUT (log path), R1125_SEED_START, R1125_SEEDS, R1125_STEPS,
//! R1125_FORGE=0|1|2 (1: forged plain-List OOB changes that both builds reject;
//! 2: additionally forged movable-list elem refs via JSON, PR only), R1125_TRACE.
#![allow(dead_code, deprecated)]
use loro::{
Container, ExportMode, Frontiers, IdSpan, LoroDoc, LoroError, LoroMap, LoroMovableList,
LoroText, ToJson, UndoManager, ValueOrContainer, VersionVector,
};
use std::collections::hash_map::DefaultHasher;
use std::fmt::Write as _;
use std::hash::{Hash, Hasher};
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::{Arc, Mutex};
struct Rng(u64);
impl Rng {
fn next(&mut self) -> u64 {
self.0 = self.0.wrapping_add(0x9E3779B97F4A7C15);
let mut z = self.0;
z = (z ^ (z >> 30)).wrapping_mul(0xBF58476D1CE4E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D049BB133111EB);
z ^ (z >> 31)
}
fn below(&mut self, n: usize) -> usize {
if n == 0 { 0 } else { (self.next() % n as u64) as usize }
}
fn chance(&mut self, p: f64) -> bool {
(self.next() % 10_000) as f64 / 10_000.0 < p
}
}
fn h<T: Hash>(t: &T) -> u64 {
let mut s = DefaultHasher::new();
t.hash(&mut s);
s.finish()
}
fn trace() -> bool {
std::env::var("R1125_TRACE").is_ok()
}
struct Peer {
doc: LoroDoc,
undo: UndoManager,
events: Arc<Mutex<u64>>,
event_count: Arc<Mutex<usize>>,
_sub: loro::Subscription,
history: Vec<Frontiers>,
shallow: bool,
}
fn new_peer(doc: LoroDoc) -> Peer {
let events = Arc::new(Mutex::new(0u64));
let event_count = Arc::new(Mutex::new(0usize));
let (e2, c2) = (events.clone(), event_count.clone());
let sub = doc.subscribe_root(Arc::new(move |e| {
let s = format!(
"{:?}|{:?}",
e.triggered_by,
e.events.iter().map(|d| format!("{:?}:{:?}", d.target, d.diff)).collect::<Vec<_>>()
);
if std::env::var("R1125_EVDUMP").is_ok() {
eprintln!("EVENT {s}");
}
let mut g = e2.lock().unwrap();
*g = h(&(*g, s));
*c2.lock().unwrap() += 1;
}));
let mut undo = UndoManager::new(&doc);
undo.set_merge_interval(0);
Peer { doc, undo, events, event_count, _sub: sub, history: vec![], shallow: false }
}
fn value_str(doc: &LoroDoc) -> String {
format!("{:?}", doc.get_deep_value().to_json_value())
}
fn random_value(rng: &mut Rng) -> loro::LoroValue {
match rng.below(3) {
0 => (rng.below(1000) as i64).into(),
1 => format!("s{}", rng.below(100)).into(),
_ => loro::LoroValue::Bool(rng.chance(0.5)),
}
}
fn edit_movable_list(rng: &mut Rng, ml: &LoroMovableList, depth: usize) {
let len = ml.len();
match rng.below(10) {
0 | 1 => {
ml.insert(rng.below(len + 1), random_value(rng)).unwrap();
}
2 if depth < 2 => {
let pos = rng.below(len + 1);
match rng.below(3) {
0 => { ml.insert_container(pos, LoroMap::new()).unwrap(); }
1 => { ml.insert_container(pos, LoroText::new()).unwrap(); }
_ => { ml.insert_container(pos, LoroMovableList::new()).unwrap(); }
}
}
3 if len > 0 => {
let pos = rng.below(len);
let n = 1 + rng.below((len - pos).min(3));
ml.delete(pos, n).unwrap();
}
4 | 5 if len > 1 => {
ml.mov(rng.below(len), rng.below(len)).unwrap();
}
6 if len > 0 => {
let pos = rng.below(len);
if depth < 2 && rng.chance(0.2) {
ml.set_container(pos, LoroMap::new()).unwrap();
} else {
ml.set(pos, random_value(rng)).unwrap();
}
}
7 | 8 if len > 0 => {
let pos = rng.below(len);
if let Some(ValueOrContainer::Container(c)) = ml.get(pos) {
match c {
Container::Map(m) => { m.insert(&format!("k{}", rng.below(4)), random_value(rng)).unwrap(); }
Container::Text(t) => { let l = t.len_unicode(); t.insert(rng.below(l + 1), "xy").unwrap(); }
Container::MovableList(inner) => edit_movable_list(rng, &inner, depth + 1),
_ => {}
}
} else {
ml.insert(rng.below(len + 1), random_value(rng)).unwrap();
}
}
_ => { ml.push(random_value(rng)).unwrap(); }
}
}
fn local_edit(rng: &mut Rng, p: &mut Peer) {
let doc = &p.doc;
let n = 1 + rng.below(4);
for _ in 0..n {
match rng.below(10) {
0 => { let t = doc.get_text("t"); let l = t.len_unicode(); t.insert(rng.below(l + 1), "ab").unwrap(); }
1 => { doc.get_map("m").insert(&format!("k{}", rng.below(5)), random_value(rng)).unwrap(); }
2 => {
let inner = doc.get_map("m").get_or_create_container("inner", LoroMovableList::new()).unwrap();
edit_movable_list(rng, &inner, 1);
}
3 if std::env::var("R1125_PL").is_ok() => { let l = doc.get_list("pl"); let len = l.len(); l.insert(rng.below(len + 1), rng.below(100) as i64).unwrap(); }
_ => edit_movable_list(rng, &doc.get_movable_list("ml"), 0),
}
}
if rng.chance(0.6) {
doc.commit();
p.history.push(doc.oplog_frontiers());
}
}
/// Change-aligned spans (arbitrary sub-change multi-span exports hit #1155).
fn chunks_between(rng: &mut Rng, src: &LoroDoc, from: &VersionVector, to: &VersionVector) -> Vec<IdSpan> {
let mut spans = vec![];
for (peer, end) in to.iter() {
let mut start = from.get(peer).copied().unwrap_or(0);
while start < *end {
let mut e = start;
let n = 1 + rng.below(3);
for _ in 0..n {
if e >= *end { break; }
let Some(meta) = src.get_change(loro::ID::new(*peer, e)) else { e = *end; break; };
e = (meta.id.counter + meta.len as i32).min(*end);
}
spans.push(IdSpan::new(*peer, start, e));
start = e;
}
}
spans
}
fn shuffle<T>(rng: &mut Rng, v: &mut [T]) {
for i in (1..v.len()).rev() {
let j = rng.below(i + 1);
v.swap(i, j);
}
}
fn record_import(log: &mut String, label: &str, r: Result<(), LoroError>) {
match r {
Ok(()) => { let _ = writeln!(log, " {label}: ok"); }
Err(LoroError::ImportUpdatesThatDependsOnOutdatedVersion) => { let _ = writeln!(log, " {label}: outdated"); }
Err(e) => {
if trace() { eprintln!("TRACE UNEXPECTED {label}: {e:?}"); }
let _ = writeln!(log, " {label}: UNEXPECTED_ERR {e:?}");
}
}
}
fn record_forged(log: &mut String, label: &str, r: Result<(), LoroError>) {
if trace() { eprintln!("TRACE FORGED {label}: {:?}", r.as_ref().map_err(|e| format!("{e}").chars().take(80).collect::<String>())); }
match r {
Ok(()) => { let _ = writeln!(log, " {label}: FORGED_ACCEPTED"); }
Err(_) => { let _ = writeln!(log, " {label}: rejected"); }
}
}
/// JSON for one forged change by `src`'s own peer at its next counter, depending on
/// src's frontiers. kind 0: plain List insert out of bounds; 1: movable-list move of an
/// unknown element.
fn forged_json(src: &LoroDoc, kind: usize) -> Option<String> {
let f = src.oplog_frontiers();
let me = src.peer_id();
let next = src.oplog_vv().get(&me).copied().unwrap_or(0);
let mut peers: Vec<String> = vec![me.to_string()];
let mut deps = vec![];
let mut lamport = 0u32;
for id in f.iter() {
let meta = src.get_change(id)?;
lamport = lamport.max(meta.lamport + (id.counter - meta.id.counter) as u32 + 1);
let idx = match peers.iter().position(|p| *p == id.peer.to_string()) {
Some(i) => i,
None => { peers.push(id.peer.to_string()); peers.len() - 1 }
};
deps.push(format!("{}@{}", id.counter, idx));
}
let content = if kind == 0 {
serde_json::json!({"container": "cid:root-pl:List", "content": {"type": "insert", "pos": 99999, "value": [1]}, "counter": next})
} else {
serde_json::json!({"container": "cid:root-ml:MovableList", "content": {"type": "move", "from": 0, "to": 0, "elem_id": "L999999@0"}, "counter": next})
};
Some(serde_json::json!({
"schema_version": 1, "start_version": {}, "peers": peers,
"changes": [{"id": format!("{next}@0"), "timestamp": 0, "deps": deps, "lamport": lamport, "msg": null, "ops": [content]}]
}).to_string())
}
/// Binary of a forged plain-List OOB change (detached imports do not state-validate).
fn forged_list_blob(src: &LoroDoc) -> Option<Vec<u8>> {
let json = forged_json(src, 0)?;
let carrier = LoroDoc::new();
carrier.import(&src.export(ExportMode::all_updates()).ok()?).ok()?;
carrier.detach();
carrier.import_json_updates(json).ok()?;
carrier.export(ExportMode::updates(&src.oplog_vv())).ok()
}
fn sync(rng: &mut Rng, peers: &mut [Peer], a: usize, b: usize, log: &mut String) {
let mode = rng.below(9);
let forge: usize = std::env::var("R1125_FORGE").ok().and_then(|s| s.parse().ok()).unwrap_or(0);
let src = &peers[a].doc;
let dst_vv = peers[b].doc.oplog_vv();
let src_vv = src.oplog_vv();
let _ = writeln!(log, " sync {a}->{b} mode {mode}");
if trace() { eprintln!("TRACE sync {a}->{b} mode {mode}"); }
let dst = &peers[b].doc;
let hist = peers[b].history.clone();
match mode {
0 => {
let Ok(bytes) = src.export(ExportMode::updates(&dst_vv)) else { let _ = writeln!(log, " export failed"); return; };
record_import(log, "import", dst.import(&bytes).map(|_| ()));
}
1 | 2 | 3 => {
let mut spans = chunks_between(rng, src, &dst_vv, &src_vv);
shuffle(rng, &mut spans);
let mut blobs = vec![];
for s in spans {
match src.export(ExportMode::updates_in_range(vec![s])) {
Ok(b) => blobs.push(b),
Err(_) => { let _ = writeln!(log, " export failed"); return; }
}
}
match mode {
1 => { for (i, b) in blobs.iter().enumerate() { record_import(log, &format!("chunk{i}"), dst.import(b).map(|_| ())); } }
2 => {
let k = rng.below(blobs.len() + 1);
record_import(log, "batch1", dst.import_batch(&blobs[..k]).map(|_| ()));
record_import(log, "batch2", dst.import_batch(&blobs[k..]).map(|_| ()));
}
_ => {
dst.detach();
for (i, b) in blobs.iter().enumerate() { record_import(log, &format!("dchunk{i}"), dst.import(b).map(|_| ())); }
dst.attach();
}
}
}
4 => {
if !hist.is_empty() {
let f = hist[rng.below(hist.len())].clone();
if dst.checkout(&f).is_err() { let _ = writeln!(log, " checkout failed"); }
}
let Ok(bytes) = src.export(ExportMode::updates(&dst_vv)) else { dst.checkout_to_latest(); return; };
record_import(log, "import@checkout", dst.import(&bytes).map(|_| ()));
dst.checkout_to_latest();
}
5 => {
let json = src.export_json_updates(&dst_vv, &src_vv);
record_import(log, "json", dst.import_json_updates(json).map(|_| ()));
}
6 => {
let Ok(bytes) = src.export(ExportMode::Snapshot) else { let _ = writeln!(log, " export failed"); return; };
record_import(log, "snapshot-into-nonempty", dst.import(&bytes).map(|_| ()));
}
7 => {
let spans = chunks_between(rng, src, &dst_vv, &src_vv);
let (first, second): (Vec<_>, Vec<_>) = spans.into_iter().partition(|_| rng.chance(0.5));
for (label, part) in [("part2-first", second), ("part1", first)] {
for span in part {
match src.export(ExportMode::updates_in_range(vec![span])) {
Ok(b) => record_import(log, label, dst.import(&b).map(|_| ())),
Err(_) => { let _ = writeln!(log, " export failed"); }
}
}
}
if let Ok(bytes) = src.export(ExportMode::updates(&dst.oplog_vv())) {
record_import(log, "rest", dst.import(&bytes).map(|_| ()));
}
}
_ => {
let mut blobs = vec![];
if let Ok(b) = src.export(ExportMode::updates(&dst_vv)) { blobs.push(b); }
if let Ok(b) = src.export(ExportMode::Snapshot) { blobs.push(b); }
record_import(log, "batch-mixed", dst.import_batch(&blobs).map(|_| ()));
}
}
let p = &mut peers[b];
p.history.push(p.doc.oplog_frontiers());
let (src, dst) = (&peers[a].doc, &peers[b].doc);
if forge >= 1 && !peers[a].shallow && !peers[b].shallow && !dst.is_detached()
&& dst.oplog_vv().includes_vv(&src.oplog_vv()) && rng.chance(0.5)
{
let before = dst.oplog_vv();
let kind = if forge == 3 { 1 } else if forge >= 2 { rng.below(2) } else { 0 };
let how = rng.below(3);
match (kind, how) {
(0, 0) | (1, _) => {
if let Some(json) = forged_json(src, kind) {
record_forged(log, &format!("forged{kind}-json"), dst.import_json_updates(json).map(|_| ()));
}
}
(0, 1) => {
if let Some(blob) = forged_list_blob(src) {
record_forged(log, "forged0-bin", dst.import(&blob).map(|_| ()));
}
}
_ => {
if let (Some(blob), Ok(snap)) = (forged_list_blob(src), src.export(ExportMode::Snapshot)) {
record_forged(log, "forged0-batch", dst.import_batch(&[snap, blob]).map(|_| ()));
}
}
}
let _ = writeln!(log, " after-forge vv_same={} attached={}", dst.oplog_vv() == before, !dst.is_detached());
}
}
fn run_seed(seed: u64, steps: usize, log: &mut String) {
let mut rng = Rng(seed);
let n = 3 + rng.below(3);
let mut peers: Vec<Peer> = (0..n)
.map(|i| { let doc = LoroDoc::new(); doc.set_peer_id(i as u64 + 1).unwrap(); new_peer(doc) })
.collect();
let _ = writeln!(log, "seed {seed} peers {n}");
for step in 0..steps {
let kind = rng.below(100);
let _ = writeln!(log, "step {step} kind {kind}");
if trace() { eprintln!("TRACE seed {seed} step {step} kind {kind}"); }
if kind < 50 {
let i = rng.below(n);
local_edit(&mut rng, &mut peers[i]);
} else if kind < 85 {
let a = rng.below(n);
let mut b = rng.below(n);
if a == b { b = (b + 1) % n; }
sync(&mut rng, &mut peers, a, b, log);
} else if kind < 93 {
let i = rng.below(n);
let p = &mut peers[i];
p.doc.commit();
let r = if rng.chance(0.5) { p.undo.undo() } else { p.undo.redo() };
let _ = writeln!(log, " undo/redo on {i}: {:?}", r.map_err(|e| format!("{e:?}")));
} else if kind < 97 {
let i = rng.below(n);
let p = &peers[i];
p.doc.commit();
let bytes = p.doc.export(ExportMode::Snapshot).unwrap();
let d = LoroDoc::new();
d.import(&bytes).unwrap();
d.set_peer_id(i as u64 + 1).unwrap();
let shallow = p.shallow;
peers[i] = new_peer(d);
peers[i].shallow = shallow;
let _ = writeln!(log, " restart {i} from snapshot");
} else {
let i = rng.below(n);
let p = &peers[i];
p.doc.commit();
let f = p.doc.oplog_frontiers();
let bytes = p.doc.export(ExportMode::shallow_snapshot(&f)).unwrap();
let d = LoroDoc::new();
d.import(&bytes).unwrap();
d.set_peer_id(i as u64 + 1).unwrap();
peers[i] = new_peer(d);
peers[i].shallow = true;
let _ = writeln!(log, " restart {i} from shallow snapshot");
}
for (i, p) in peers.iter().enumerate() {
let _ = writeln!(log, " p{i} v={:x} ev={:x}/{} det={}", h(&value_str(&p.doc)), *p.events.lock().unwrap(), *p.event_count.lock().unwrap(), p.doc.is_detached());
}
}
for _ in 0..2 {
for i in 1..n {
for (a, b) in [(i, 0), (0, i)] {
let dst_vv = peers[b].doc.oplog_vv();
if let Ok(bytes) = peers[a].doc.export(ExportMode::updates(&dst_vv)) {
let r = peers[b].doc.import(&bytes).map(|_| ());
record_import(log, &format!("final {a}->{b}"), r);
}
}
}
}
let v0 = value_str(&peers[0].doc);
for (i, p) in peers.iter().enumerate() {
let v = value_str(&p.doc);
let _ = writeln!(log, "final p{i} v={:x} shallow={} same_as_p0={}", h(&v), p.shallow, v == v0);
}
if let Some(p) = peers.iter().find(|p| !p.shallow) {
let all = p.doc.export(ExportMode::all_updates()).unwrap();
let d = LoroDoc::new();
record_import(log, "full-replay", d.import(&all).map(|_| ()));
let _ = writeln!(log, "full replay matches: {}", value_str(&d) == value_str(&p.doc));
if std::env::var("R1125_REPLAYDUMP").is_ok() && value_str(&d) != value_str(&p.doc) {
eprintln!("IS_SHALLOW {}", p.doc.is_shallow());
eprintln!("LIVE {}", p.doc.get_deep_value().to_json_value());
eprintln!("REPLAY {}", d.get_deep_value().to_json_value());
}
}
}
#[test]
fn review_1125_random_differential() {
let out = std::env::var("R1125_OUT").unwrap_or_else(|_| "/tmp/r1125-diff.log".into());
let start: u64 = std::env::var("R1125_SEED_START").ok().and_then(|s| s.parse().ok()).unwrap_or(0);
let seeds: u64 = std::env::var("R1125_SEEDS").ok().and_then(|s| s.parse().ok()).unwrap_or(200);
let steps: usize = std::env::var("R1125_STEPS").ok().and_then(|s| s.parse().ok()).unwrap_or(120);
let mut all = String::new();
let (mut panics, mut unexpected, mut forged_ok) = (0, 0, 0);
for seed in start..start + seeds {
let mut log = String::new();
let r = catch_unwind(AssertUnwindSafe(|| run_seed(seed, steps, &mut log)));
if let Err(p) = r {
let msg = p.downcast_ref::<String>().cloned().or_else(|| p.downcast_ref::<&str>().map(|s| s.to_string())).unwrap_or_default();
let _ = writeln!(log, "PANIC seed {seed}: {msg}");
panics += 1;
}
unexpected += log.matches("UNEXPECTED_ERR").count();
forged_ok += log.matches("FORGED_ACCEPTED").count();
all.push_str(&log);
}
let _ = writeln!(all, "SUMMARY panics={panics} unexpected_err={unexpected} forged_accepted={forged_ok}");
std::fs::write(&out, &all).unwrap();
eprintln!("wrote {out}: panics={panics} unexpected_err={unexpected} forged_accepted={forged_ok}");
}🤖 Generated with Claude Code
Activity
test: a failed import that loads old movable-list values (#1161)
fix: resolve a container's parent from the change that created it (#1159