471 lines
16 KiB
Rust
471 lines
16 KiB
Rust
// auto_pair_crdt_acceptance.rs --- auto-pairing over the wire.
|
|
|
|
//! Auto-pairing two-replica acceptance (docs/auto-pairing-framing.md):
|
|
//! a synthetic source replica plus a synthetic observer replica against
|
|
//! a real daemon subprocess.
|
|
//!
|
|
//! Dispatch route (built-in pair chars, Q#AP1): the source sends
|
|
//! round-tripped `Key` events; the daemon pairs/skips and broadcasts
|
|
//! `DaemonKey` ops to both replicas. Undo grain (Q#AP5) is pinned for
|
|
//! both routing models — the TUI's single-key optimistic undo is the
|
|
//! source replica's own peer-bound undo, `C-x u` is a round-tripped
|
|
//! daemon undo — as assertions of the named cross-peer substrate
|
|
//! limit, NOT frontend-equivalence claims.
|
|
//!
|
|
//! Optimistic route (custom pair char via user config, Q#AP1 cost
|
|
//! paragraph): the opener arrives as a `FrontendEvent::CrdtOp`; the
|
|
//! daemon's hook-queued closer is broadcast BEFORE the opener's
|
|
//! rebroadcast (the ordering quirk named in the framing), and the
|
|
//! observer must still converge. The source mirror's undo removes the
|
|
//! opener and leaves the closer — the pinned degraded undo.
|
|
|
|
#![cfg(feature = "crdt")]
|
|
|
|
use std::time::Duration;
|
|
|
|
use pmacs::crdt::CrdtState;
|
|
use pmacs::protocol::{FrontendEvent, FrontendId, Key, KeyEvent, Modifiers};
|
|
use pmacs::rope::CrdtOp as RopeCrdtOp;
|
|
use pmacs::transport::write_message;
|
|
|
|
mod common;
|
|
use common::daemon::{TestDaemon, attach_multi};
|
|
|
|
/// Read the daemon's initial `BufferSnapshot` for a freshly-attached
|
|
/// replica stream (the daemon always emits it first).
|
|
fn read_initial_snapshot(
|
|
stream: &mut std::os::unix::net::UnixStream,
|
|
) -> (pmacs::buffer::BufferId, Vec<u8>) {
|
|
match pmacs::transport::read_message::<pmacs::protocol::InstanceMessage>(stream)
|
|
.expect("read initial BufferSnapshot")
|
|
{
|
|
pmacs::protocol::InstanceMessage::BufferSnapshot {
|
|
buffer_id,
|
|
crdt_snapshot,
|
|
} => (buffer_id, crdt_snapshot),
|
|
other => panic!("expected initial BufferSnapshot, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
/// One attached synthetic replica: stream + mirror + identity.
|
|
struct Replica {
|
|
stream: std::os::unix::net::UnixStream,
|
|
state: CrdtState,
|
|
fid: FrontendId,
|
|
buffer_id: pmacs::buffer::BufferId,
|
|
}
|
|
|
|
fn attach_replica(daemon: &TestDaemon) -> Replica {
|
|
let (hello, mut stream) = attach_multi(daemon);
|
|
let fid = hello.assigned_frontend_id;
|
|
let (buffer_id, snap) = read_initial_snapshot(&mut stream);
|
|
let state = CrdtState::new(fid.0).expect("CrdtState::new");
|
|
state.import_snapshot(&snap).expect("import_snapshot");
|
|
Replica {
|
|
stream,
|
|
state,
|
|
fid,
|
|
buffer_id,
|
|
}
|
|
}
|
|
|
|
/// Mutate the local replica, export the delta, and ship it as an
|
|
/// optimistic `FrontendEvent::CrdtOp` (the `m10_11` idiom).
|
|
fn send_optimistic_op<F>(replica: &mut Replica, mutate: F)
|
|
where
|
|
F: FnOnce(&CrdtState),
|
|
{
|
|
let v = replica.state.version();
|
|
mutate(&replica.state);
|
|
let op_bytes = replica
|
|
.state
|
|
.export_updates_since(&v)
|
|
.expect("export updates after local mutation");
|
|
write_message(
|
|
&mut replica.stream,
|
|
&FrontendEvent::CrdtOp {
|
|
frontend_id: replica.fid,
|
|
buffer_id: replica.buffer_id,
|
|
op: RopeCrdtOp {
|
|
peer_id: replica.fid.0,
|
|
bytes: op_bytes,
|
|
},
|
|
},
|
|
)
|
|
.expect("write CrdtOp");
|
|
}
|
|
|
|
fn send_key(replica: &mut Replica, key: Key, mods: Modifiers) {
|
|
write_message(
|
|
&mut replica.stream,
|
|
&FrontendEvent::Key(KeyEvent {
|
|
frontend_id: replica.fid,
|
|
key,
|
|
mods,
|
|
timestamp_ns: 0,
|
|
}),
|
|
)
|
|
.expect("send Key");
|
|
}
|
|
|
|
/// `C-x u` — the always-dispatched daemon undo.
|
|
fn send_daemon_undo(replica: &mut Replica) {
|
|
send_key(replica, Key::Char('x'), Modifiers::CTRL);
|
|
send_key(replica, Key::Char('u'), Modifiers::NONE);
|
|
}
|
|
|
|
/// `C-x r` — daemon redo.
|
|
fn send_daemon_redo(replica: &mut Replica) {
|
|
send_key(replica, Key::Char('x'), Modifiers::CTRL);
|
|
send_key(replica, Key::Char('r'), Modifiers::NONE);
|
|
}
|
|
|
|
/// What a pump observed so far: materialized text, the daemon's last
|
|
/// `CursorByte` for the shared buffer, ops imported this call.
|
|
struct Observed {
|
|
text: String,
|
|
cursor: Option<u64>,
|
|
imported: usize,
|
|
}
|
|
|
|
/// Pump broadcast messages into the replica until `pred` holds or the
|
|
/// deadline passes. Imports every `CrdtOp` for the shared buffer and
|
|
/// tracks the latest `CursorByte`.
|
|
fn pump_until<P: Fn(&Observed) -> bool>(
|
|
replica: &mut Replica,
|
|
timeout: Duration,
|
|
what: &str,
|
|
pred: P,
|
|
) -> Observed {
|
|
let deadline = std::time::Instant::now() + timeout;
|
|
let mut obs = Observed {
|
|
text: replica.state.materialize_string(),
|
|
cursor: None,
|
|
imported: 0,
|
|
};
|
|
loop {
|
|
if pred(&obs) {
|
|
return obs;
|
|
}
|
|
assert!(
|
|
std::time::Instant::now() < deadline,
|
|
"pump timeout waiting for {what}; text={:?} cursor={:?} imported={}",
|
|
obs.text,
|
|
obs.cursor,
|
|
obs.imported
|
|
);
|
|
let remaining = deadline.saturating_duration_since(std::time::Instant::now());
|
|
replica
|
|
.stream
|
|
.set_read_timeout(Some(remaining.min(Duration::from_millis(100))))
|
|
.ok();
|
|
match pmacs::transport::read_message::<pmacs::protocol::InstanceMessage>(
|
|
&mut replica.stream,
|
|
) {
|
|
Ok(pmacs::protocol::InstanceMessage::CrdtOp { buffer_id: b, op })
|
|
if b == replica.buffer_id =>
|
|
{
|
|
let _ = replica.state.import_updates(&op.bytes);
|
|
obs.imported += 1;
|
|
obs.text = replica.state.materialize_string();
|
|
}
|
|
Ok(pmacs::protocol::InstanceMessage::CursorByte {
|
|
buffer_id: b,
|
|
byte_pos,
|
|
}) if b == replica.buffer_id => {
|
|
obs.cursor = Some(byte_pos);
|
|
}
|
|
Ok(_) | Err(_) => {}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Pump for `window` expecting NO text change — the negative
|
|
/// assertion for "a further daemon undo cannot reach source-peer
|
|
/// history". Ops are still imported (there should be none that change
|
|
/// text); panics if the text leaves `expected`.
|
|
fn assert_text_stays(replica: &mut Replica, expected: &str, window: Duration) {
|
|
let deadline = std::time::Instant::now() + window;
|
|
while std::time::Instant::now() < deadline {
|
|
replica
|
|
.stream
|
|
.set_read_timeout(Some(Duration::from_millis(50)))
|
|
.ok();
|
|
if let Ok(pmacs::protocol::InstanceMessage::CrdtOp { buffer_id: b, op }) =
|
|
pmacs::transport::read_message::<pmacs::protocol::InstanceMessage>(&mut replica.stream)
|
|
&& b == replica.buffer_id
|
|
{
|
|
let _ = replica.state.import_updates(&op.bytes);
|
|
assert_eq!(
|
|
replica.state.materialize_string(),
|
|
expected,
|
|
"text must not change during the negative window"
|
|
);
|
|
}
|
|
}
|
|
assert_eq!(replica.state.materialize_string(), expected);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Dispatch route (built-in chars)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Round-tripped `(` pairs daemon-side and both replicas converge to
|
|
/// `()` with the daemon cursor between the pair; a round-tripped `)`
|
|
/// then skips (insert + swallow-delete, two more ops) and the daemon
|
|
/// cursor steps over the closer.
|
|
#[test]
|
|
fn dispatch_route_pair_and_skip_converge_on_both_replicas() {
|
|
let daemon = TestDaemon::spawn();
|
|
let mut source = attach_replica(&daemon);
|
|
let mut observer = attach_replica(&daemon);
|
|
|
|
send_key(&mut source, Key::Char('('), Modifiers::NONE);
|
|
pump_until(&mut observer, Duration::from_secs(5), "observer ()", |o| {
|
|
o.text == "()"
|
|
});
|
|
pump_until(
|
|
&mut source,
|
|
Duration::from_secs(5),
|
|
"source () with cursor between",
|
|
|o| o.text == "()" && o.cursor == Some(1),
|
|
);
|
|
|
|
// Skip: the typed `)` inserts then swallows the duplicate — two
|
|
// ops that leave the text identical, so convergence is detected
|
|
// by the op count plus the daemon cursor stepping to 2.
|
|
send_key(&mut source, Key::Char(')'), Modifiers::NONE);
|
|
pump_until(
|
|
&mut source,
|
|
Duration::from_secs(5),
|
|
"source skip (two ops, cursor after the closer)",
|
|
|o| o.text == "()" && o.imported >= 2 && o.cursor == Some(2),
|
|
);
|
|
pump_until(
|
|
&mut observer,
|
|
Duration::from_secs(5),
|
|
"observer skip (two ops, text still ())",
|
|
|o| o.text == "()" && o.imported >= 2,
|
|
);
|
|
}
|
|
|
|
/// Q#AP5 undo grain over the wire: the pair is two adjacent
|
|
/// daemon-peer units. Two `C-x u` restore `(` then empty on BOTH
|
|
/// replicas; two `C-x r` restore `(` then `()` in order.
|
|
#[test]
|
|
fn dispatch_route_daemon_undo_redo_walk_the_pair_on_both_replicas() {
|
|
let daemon = TestDaemon::spawn();
|
|
let mut source = attach_replica(&daemon);
|
|
let mut observer = attach_replica(&daemon);
|
|
|
|
send_key(&mut source, Key::Char('('), Modifiers::NONE);
|
|
pump_until(&mut source, Duration::from_secs(5), "source ()", |o| {
|
|
o.text == "()"
|
|
});
|
|
pump_until(&mut observer, Duration::from_secs(5), "observer ()", |o| {
|
|
o.text == "()"
|
|
});
|
|
|
|
send_daemon_undo(&mut source);
|
|
pump_until(&mut source, Duration::from_secs(5), "source (", |o| {
|
|
o.text == "("
|
|
});
|
|
pump_until(&mut observer, Duration::from_secs(5), "observer (", |o| {
|
|
o.text == "("
|
|
});
|
|
|
|
send_daemon_undo(&mut source);
|
|
pump_until(&mut source, Duration::from_secs(5), "source empty", |o| {
|
|
o.text.is_empty()
|
|
});
|
|
pump_until(
|
|
&mut observer,
|
|
Duration::from_secs(5),
|
|
"observer empty",
|
|
|o| o.text.is_empty(),
|
|
);
|
|
|
|
send_daemon_redo(&mut source);
|
|
pump_until(
|
|
&mut source,
|
|
Duration::from_secs(5),
|
|
"source ( redone",
|
|
|o| o.text == "(",
|
|
);
|
|
pump_until(
|
|
&mut observer,
|
|
Duration::from_secs(5),
|
|
"observer ( redone",
|
|
|o| o.text == "(",
|
|
);
|
|
|
|
send_daemon_redo(&mut source);
|
|
pump_until(
|
|
&mut source,
|
|
Duration::from_secs(5),
|
|
"source () redone",
|
|
|o| o.text == "()",
|
|
);
|
|
pump_until(
|
|
&mut observer,
|
|
Duration::from_secs(5),
|
|
"observer () redone",
|
|
|o| o.text == "()",
|
|
);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Mixed source/daemon history (the named substrate limit, pinned)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// TUI routing model: with optimistic `a` already in the source
|
|
/// mirror, the single-key optimistic undo (the mirror's own peer-bound
|
|
/// undo) removes `a` — NOT the daemon-peer closer — leaving `()`.
|
|
#[test]
|
|
fn mixed_history_source_mirror_undo_removes_the_optimistic_char_first() {
|
|
let daemon = TestDaemon::spawn();
|
|
let mut source = attach_replica(&daemon);
|
|
let mut observer = attach_replica(&daemon);
|
|
|
|
send_optimistic_op(&mut source, |r| {
|
|
r.insert(0, "a").expect("insert a");
|
|
});
|
|
send_key(&mut source, Key::Char('('), Modifiers::NONE);
|
|
pump_until(&mut source, Duration::from_secs(5), "source a()", |o| {
|
|
o.text == "a()"
|
|
});
|
|
pump_until(&mut observer, Duration::from_secs(5), "observer a()", |o| {
|
|
o.text == "a()"
|
|
});
|
|
|
|
// The TUI's single-key undo: mirror-local, peer-bound.
|
|
send_optimistic_op(&mut source, |r| {
|
|
r.undo().expect("mirror undo");
|
|
});
|
|
assert_eq!(
|
|
source.state.materialize_string(),
|
|
"()",
|
|
"the mirror undo removed source-peer `a`, not the adjacent daemon closer"
|
|
);
|
|
pump_until(&mut observer, Duration::from_secs(5), "observer ()", |o| {
|
|
o.text == "()"
|
|
});
|
|
}
|
|
|
|
/// `C-x u` routing model (and the GPU model, which reaches the daemon
|
|
/// the same way): daemon undos peel the pair — closer, then opener —
|
|
/// and a FURTHER daemon undo cannot reach the source-peer `a`.
|
|
#[test]
|
|
fn mixed_history_daemon_undo_peels_the_pair_but_cannot_reach_source_history() {
|
|
let daemon = TestDaemon::spawn();
|
|
let mut source = attach_replica(&daemon);
|
|
let mut observer = attach_replica(&daemon);
|
|
|
|
send_optimistic_op(&mut source, |r| {
|
|
r.insert(0, "a").expect("insert a");
|
|
});
|
|
send_key(&mut source, Key::Char('('), Modifiers::NONE);
|
|
pump_until(&mut source, Duration::from_secs(5), "source a()", |o| {
|
|
o.text == "a()"
|
|
});
|
|
|
|
send_daemon_undo(&mut source);
|
|
pump_until(&mut source, Duration::from_secs(5), "source a(", |o| {
|
|
o.text == "a("
|
|
});
|
|
pump_until(&mut observer, Duration::from_secs(5), "observer a(", |o| {
|
|
o.text == "a("
|
|
});
|
|
|
|
send_daemon_undo(&mut source);
|
|
pump_until(&mut source, Duration::from_secs(5), "source a", |o| {
|
|
o.text == "a"
|
|
});
|
|
|
|
// The named limit: daemon undo is peer-bound too — source-peer
|
|
// `a` is beyond its reach. (Cross-peer chronological arbitration
|
|
// is deferred substrate work, not pair.lua's claim.)
|
|
send_daemon_undo(&mut source);
|
|
assert_text_stays(&mut source, "a", Duration::from_millis(800));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Optimistic route (custom pair char from user config)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
const CUSTOM_PAIR_CONFIG: &str = "table.insert(pmacs.pair.sets.default, \"<>\")\n";
|
|
|
|
/// A user-extended pair char still arrives optimistically: the opener
|
|
/// is a source-peer op, the daemon's hook-queued `>` closer is
|
|
/// broadcast BEFORE the opener's rebroadcast (the framing's ordering
|
|
/// quirk — the observer receives the causally dependent closer
|
|
/// first), and both replicas must still converge. The skip route
|
|
/// converges likewise.
|
|
#[test]
|
|
fn optimistic_route_custom_char_pairs_and_skips_despite_closer_first_broadcast() {
|
|
let daemon = TestDaemon::spawn_with_config(CUSTOM_PAIR_CONFIG);
|
|
let mut source = attach_replica(&daemon);
|
|
let mut observer = attach_replica(&daemon);
|
|
|
|
send_optimistic_op(&mut source, |r| {
|
|
r.insert(0, "<").expect("insert <");
|
|
});
|
|
pump_until(&mut observer, Duration::from_secs(5), "observer <>", |o| {
|
|
o.text == "<>"
|
|
});
|
|
pump_until(&mut source, Duration::from_secs(5), "source <>", |o| {
|
|
o.text == "<>"
|
|
});
|
|
|
|
// Skip: the source optimistically types the closer before the
|
|
// existing `>`; the daemon swallows the duplicate. Text returns
|
|
// to `<>`; the extra daemon delete op must reach both replicas.
|
|
send_optimistic_op(&mut source, |r| {
|
|
r.insert(1, ">").expect("insert >");
|
|
});
|
|
pump_until(
|
|
&mut source,
|
|
Duration::from_secs(5),
|
|
"source skip converged",
|
|
|o| o.text == "<>" && o.imported >= 1,
|
|
);
|
|
pump_until(
|
|
&mut observer,
|
|
Duration::from_secs(5),
|
|
"observer skip converged",
|
|
|o| o.text == "<>" && o.imported >= 2,
|
|
);
|
|
}
|
|
|
|
/// The pinned degraded undo for optimistic pair chars: the opener and
|
|
/// closer live on DIFFERENT peers, so the source mirror's undo removes
|
|
/// its own opener and leaves the daemon's closer behind.
|
|
#[test]
|
|
fn optimistic_route_mirror_undo_removes_the_opener_leaving_the_closer() {
|
|
let daemon = TestDaemon::spawn_with_config(CUSTOM_PAIR_CONFIG);
|
|
let mut source = attach_replica(&daemon);
|
|
let mut observer = attach_replica(&daemon);
|
|
|
|
send_optimistic_op(&mut source, |r| {
|
|
r.insert(0, "<").expect("insert <");
|
|
});
|
|
pump_until(&mut source, Duration::from_secs(5), "source <>", |o| {
|
|
o.text == "<>"
|
|
});
|
|
pump_until(&mut observer, Duration::from_secs(5), "observer <>", |o| {
|
|
o.text == "<>"
|
|
});
|
|
|
|
send_optimistic_op(&mut source, |r| {
|
|
r.undo().expect("mirror undo");
|
|
});
|
|
assert_eq!(
|
|
source.state.materialize_string(),
|
|
">",
|
|
"peer-bound mirror undo removes the opener; the daemon-peer closer stays"
|
|
);
|
|
pump_until(&mut observer, Duration::from_secs(5), "observer >", |o| {
|
|
o.text == ">"
|
|
});
|
|
}
|