use core::fmt;
use crate::error::{Error, Result};
use crate::sync::{self, MergedRecord, Payload, PayloadKind, SeenSyncIds, SyncId};
use crate::vault::{MasterKey, Vault};
use qualia_dik::DeviceId;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Roster {
TwoParty,
MultiParty,
}
impl Roster {
#[must_use]
pub const fn needs_start_signal(self) -> bool {
matches!(self, Self::MultiParty)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Recipient {
One(DeviceId),
All,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TransportCapabilities {
pub roster: Roster,
pub max_message_len: Option<usize>,
pub max_qr_version: Option<u8>,
pub single_use: bool,
pub can_carry_payload: bool,
}
impl TransportCapabilities {
#[must_use]
pub const fn stream(roster: Roster) -> Self {
Self {
roster,
max_message_len: None,
max_qr_version: None,
single_use: false,
can_carry_payload: true,
}
}
#[must_use]
pub const fn qr(max_qr_version: u8, max_message_len: usize) -> Self {
Self {
roster: Roster::TwoParty,
max_message_len: Some(max_message_len),
max_qr_version: Some(max_qr_version),
single_use: true,
can_carry_payload: true,
}
}
pub fn check_fits(&self, len: usize) -> Result<()> {
match self.max_message_len {
Some(limit) if len > limit => {
Err(Error::TransportMessageTooLarge { needed: len, limit })
}
_ => Ok(()),
}
}
#[must_use]
pub const fn needs_start_signal(&self) -> bool {
self.roster.needs_start_signal()
}
}
pub trait Transport {
fn capabilities(&self) -> TransportCapabilities;
fn local_device(&self) -> DeviceId;
fn send(&self, to: Recipient, message: &[u8]) -> Result<()>;
fn receive(&self) -> Result<(DeviceId, Vec<u8>)>;
fn peers(&self) -> Result<Vec<DeviceId>>;
}
#[derive(Debug)]
pub struct SyncOutcome {
pub sync_id: SyncId,
pub merge: sync::MergeOutcome,
}
impl SyncOutcome {
#[must_use]
pub fn made_no_change(&self) -> bool {
self.merge.made_no_change()
}
#[must_use]
pub fn needs_adjudication(&self) -> bool {
self.merge.needs_adjudication()
}
}
impl fmt::Display for SyncOutcome {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"新增 {}、墓碑 {}、应用删除 {}、未变 {}、冲突 {}",
self.merge.added,
self.merge.tombstones_added,
self.merge.deletions_applied,
self.merge.unchanged,
self.merge.conflicts.len()
)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SharePlan {
FullPayload {
len: usize,
},
LinkEstablishment {
needed: usize,
limit: usize,
},
}
impl SharePlan {
#[must_use]
pub const fn needs_link(self) -> bool {
matches!(self, Self::LinkEstablishment { .. })
}
}
impl fmt::Display for SharePlan {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match *self {
Self::FullPayload { len } => write!(f, "整份载荷({len} 字节)"),
Self::LinkEstablishment { needed, limit } => {
write!(f, "装不下:需要 {needed} 字节,上限 {limit} 字节")
}
}
}
}
pub struct SyncSession<'a> {
master: &'a MasterKey,
vault: &'a mut Vault,
seen: &'a mut SeenSyncIds,
allowed_senders: Vec<DeviceId>,
}
impl<'a> SyncSession<'a> {
#[must_use]
pub fn restricted(
master: &'a MasterKey,
vault: &'a mut Vault,
seen: &'a mut SeenSyncIds,
allowed_senders: Vec<DeviceId>,
) -> Self {
Self {
master,
vault,
seen,
allowed_senders,
}
}
#[must_use]
pub fn open(master: &'a MasterKey, vault: &'a mut Vault, seen: &'a mut SeenSyncIds) -> Self {
Self {
master,
vault,
seen,
allowed_senders: Vec::new(),
}
}
pub fn send(
&self,
transport: &dyn Transport,
to: Recipient,
kind: PayloadKind,
) -> Result<SyncId> {
let payload = match kind {
PayloadKind::Vault => {
Payload::from_vault(self.master, transport.local_device(), self.vault)?
}
PayloadKind::Request => Payload::request(transport.local_device())?,
};
let sealed = payload.seal(self.master)?;
transport.capabilities().check_fits(sealed.len())?;
transport.send(to, &sealed)?;
Ok(payload.sync_id)
}
pub fn plan_share(&self, transport: &dyn Transport) -> Result<SharePlan> {
let payload = Payload::from_vault(self.master, transport.local_device(), self.vault)?;
let sealed = payload.seal(self.master)?;
Ok(match transport.capabilities().max_message_len {
Some(limit) if sealed.len() > limit => SharePlan::LinkEstablishment {
needed: sealed.len(),
limit,
},
_ => SharePlan::FullPayload { len: sealed.len() },
})
}
pub fn receive(&mut self, transport: &dyn Transport) -> Result<SyncOutcome> {
let (claimed_sender, message) = transport.receive()?;
if !self.allowed_senders.is_empty() && !self.allowed_senders.contains(&claimed_sender) {
return Err(Error::SyncSenderMismatch);
}
let header = Payload::peek(&message)?;
if self.seen.contains(&header.sync_id) {
return Err(Error::SyncAlreadyApplied);
}
let payload = Payload::open(self.master, &message)?;
if payload.sender != claimed_sender {
return Err(Error::SyncSenderMismatch);
}
let opened = payload.open_records(self.master)?;
let records: Vec<MergedRecord> = opened
.into_iter()
.map(sync::OpenedRecord::into_merged)
.collect();
let mut set = sync::RecordSet::new();
for record in records {
set.insert(record);
}
let merge = sync::merge_into(self.vault, &set)?;
self.seen.record(payload.sync_id)?;
Ok(SyncOutcome {
sync_id: payload.sync_id,
merge,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::otp::{Algorithm, Digits, OtpSecret};
use crate::sync::{RecordSet, merge_into};
use crate::vault::{Entry, EntryId, Timestamp, Vault as VaultAlias};
struct MemoryTransport {
me: DeviceId,
outbox: std::sync::Mutex<Vec<(Recipient, DeviceId, Vec<u8>)>>,
inbox: std::sync::Mutex<Vec<(DeviceId, Vec<u8>)>>,
caps: TransportCapabilities,
peers: Vec<DeviceId>,
}
impl MemoryTransport {
fn new(me: DeviceId, caps: TransportCapabilities) -> Self {
Self {
me,
outbox: std::sync::Mutex::new(Vec::new()),
inbox: std::sync::Mutex::new(Vec::new()),
caps,
peers: Vec::new(),
}
}
fn with_peers(mut self, peers: Vec<DeviceId>) -> Self {
self.peers = peers;
self
}
fn deliver_to(&self, target: &MemoryTransport, honest: bool) {
let outbox = self.outbox.lock().expect("锁未中毒");
let mut inbox = target.inbox.lock().expect("锁未中毒");
for (recipient, sender, bytes) in outbox.iter() {
let addressed = match recipient {
Recipient::All => true,
Recipient::One(id) => *id == target.me,
};
if !addressed {
continue;
}
let claimed = if honest { *sender } else { self.me };
inbox.push((claimed, bytes.clone()));
}
}
fn clear_outbox(&self) {
self.outbox.lock().expect("锁未中毒").clear();
}
fn inject(&self, claimed_sender: DeviceId, bytes: Vec<u8>) {
self.inbox
.lock()
.expect("锁未中毒")
.push((claimed_sender, bytes));
}
}
impl Transport for MemoryTransport {
fn capabilities(&self) -> TransportCapabilities {
self.caps
}
fn local_device(&self) -> DeviceId {
self.me
}
fn send(&self, to: Recipient, message: &[u8]) -> Result<()> {
self.outbox
.lock()
.expect("锁未中毒")
.push((to, self.me, message.to_vec()));
Ok(())
}
fn receive(&self) -> Result<(DeviceId, Vec<u8>)> {
let mut inbox = self.inbox.lock().expect("锁未中毒");
if inbox.is_empty() {
return Err(Error::SyncNoSessionFound);
}
Ok(inbox.remove(0))
}
fn peers(&self) -> Result<Vec<DeviceId>> {
Ok(self.peers.clone())
}
}
fn a_device(seed: u8) -> (DeviceId, MasterKey, Vault) {
let master = MasterKey::generate().expect("OS 随机源可用");
(DeviceId::from_bytes([seed; 32]), master, Vault::new())
}
fn sample_entry(name: &str, updated_at: u64) -> Entry {
Entry::new(
name.into(),
Some("组织".into()),
"user@example.com".into(),
OtpSecret::from_base32("JBSWY3DPEHPK3PXP").expect("合法 Base32"),
Algorithm::Sha1,
Digits::Six,
30,
None,
String::new(),
Timestamp::from(1_700_000_000_u64),
Timestamp::from(updated_at),
)
.expect("字段合法")
}
#[test]
fn a_two_party_exchange_moves_entries() {
let (id_a, master, mut vault_a) = a_device(1);
let (id_b, _, mut vault_b) = a_device(2);
vault_a
.insert(sample_entry("A 的条目", 1_700_000_000))
.expect("随机源可用");
let caps = TransportCapabilities::stream(Roster::TwoParty);
let ta = MemoryTransport::new(id_a, caps).with_peers(vec![id_b]);
let tb = MemoryTransport::new(id_b, caps).with_peers(vec![id_a]);
let mut seen_a = SeenSyncIds::new();
let mut seen_b = SeenSyncIds::new();
let session_a = SyncSession::open(&master, &mut vault_a, &mut seen_a);
session_a
.send(&ta, Recipient::One(id_b), PayloadKind::Vault)
.expect("发送");
ta.deliver_to(&tb, true);
let outcome = {
let mut session_b = SyncSession::open(&master, &mut vault_b, &mut seen_b);
session_b.receive(&tb).expect("接收")
};
assert_eq!(outcome.merge.added, 1, "应当并进来一条:{outcome}");
assert_eq!(vault_b.len(), 1);
assert!(!outcome.needs_adjudication(), "内容相同不该报冲突");
assert_eq!(seen_b.len(), 1, "收到之后应当记一笔");
assert_eq!(seen_a.len(), 0, "发出载荷不该记进重放日志");
ta.deliver_to(&tb, true);
let again = {
let mut session_b = SyncSession::open(&master, &mut vault_b, &mut seen_b);
session_b.receive(&tb)
};
assert!(
matches!(again, Err(Error::SyncAlreadyApplied)),
"重复的载荷未被拒绝,实际 {again:?}"
);
}
#[test]
fn a_replayed_payload_from_a_third_party_is_rejected() {
let (id_a, master, mut vault_a) = a_device(1);
let (id_b, _, mut vault_b) = a_device(2);
let (id_c, _, mut vault_c) = a_device(3);
vault_b
.insert(sample_entry("B 的条目", 1_700_000_000))
.expect("随机源可用");
let caps = TransportCapabilities::stream(Roster::MultiParty);
let tb = MemoryTransport::new(id_b, caps).with_peers(vec![id_a, id_c]);
let ta = MemoryTransport::new(id_a, caps).with_peers(vec![id_b, id_c]);
let tc = MemoryTransport::new(id_c, caps).with_peers(vec![id_a]);
let mut seen_b = SeenSyncIds::new();
let session_b = SyncSession::open(&master, &mut vault_b, &mut seen_b);
session_b
.send(&tb, Recipient::All, PayloadKind::Vault)
.expect("发送");
let b_payload = tb
.outbox
.lock()
.expect("锁未中毒")
.first()
.map(|(_, _, bytes)| bytes.clone())
.expect("B 的出站里应当有那份载荷");
tb.deliver_to(&ta, true);
let mut seen_a = SeenSyncIds::new();
let outcome_a = {
let mut session_a = SyncSession::open(&master, &mut vault_a, &mut seen_a);
session_a.receive(&ta).expect("A 正常收到")
};
assert_eq!(outcome_a.merge.added, 1, "A 应当收下 B 的那条");
ta.send(Recipient::One(id_c), &b_payload).expect("转发");
ta.deliver_to(&tc, true);
let mut seen_c = SeenSyncIds::new();
let mut session_c = SyncSession::restricted(&master, &mut vault_c, &mut seen_c, vec![id_a]);
let result = session_c.receive(&tc);
assert!(
matches!(result, Err(Error::SyncSenderMismatch)),
"被转发的载荷未被拒绝,实际 {result:?}"
);
assert_eq!(vault_c.len(), 0, "被拒绝的载荷不得改动保险库");
assert_eq!(seen_c.len(), 0, "被拒绝的载荷不得记账");
}
#[test]
fn a_payload_from_an_unlisted_peer_is_rejected() {
let (id_a, master, _) = a_device(1);
let (id_b, _, mut vault_b) = a_device(2);
let (id_c, _, mut vault_c) = a_device(3);
vault_b
.insert(sample_entry("B 的条目", 1_700_000_000))
.expect("随机源可用");
let caps = TransportCapabilities::stream(Roster::MultiParty);
let tb = MemoryTransport::new(id_b, caps).with_peers(vec![id_c]);
let tc = MemoryTransport::new(id_c, caps).with_peers(vec![id_b]);
let mut seen_b = SeenSyncIds::new();
SyncSession::open(&master, &mut vault_b, &mut seen_b)
.send(&tb, Recipient::All, PayloadKind::Vault)
.expect("发送");
tb.deliver_to(&tc, true);
let mut seen_c = SeenSyncIds::new();
let result = {
let mut session_c =
SyncSession::restricted(&master, &mut vault_c, &mut seen_c, vec![id_a]);
session_c.receive(&tc)
};
assert!(
matches!(result, Err(Error::SyncSenderMismatch)),
"名单外的对端未被拒绝,实际 {result:?}"
);
assert_eq!(vault_c.len(), 0);
}
#[test]
fn a_garbage_payload_does_not_poison_the_seen_log() {
let (id_a, master, mut vault_a) = a_device(1);
let (id_b, _, _) = a_device(2);
let caps = TransportCapabilities::stream(Roster::TwoParty);
let mut source_vault = Vault::new();
source_vault
.insert(sample_entry("条目", 1_700_000_000))
.expect("随机源可用");
let payload = Payload::from_vault(&master, id_b, &source_vault).expect("打包");
let sealed = payload.seal(&master).expect("封装");
let good_id = payload.sync_id;
let mut broken = sealed.clone();
let last = broken.len() - 1;
broken[last] ^= 0xFF;
let ta = MemoryTransport::new(id_a, caps);
ta.inject(id_b, broken);
let mut seen_a = SeenSyncIds::new();
let result = {
let mut session_a = SyncSession::open(&master, &mut vault_a, &mut seen_a);
session_a.receive(&ta)
};
assert!(result.is_err(), "被篡改的载荷应当被拒绝");
assert_eq!(seen_a.len(), 0, "被拒绝的载荷不得记账");
assert!(
!seen_a.contains(&good_id),
"垃圾载荷把合法载荷的标识污染了——真正的载荷将永远被拒绝"
);
ta.inject(id_b, sealed);
let outcome = {
let mut session_a = SyncSession::open(&master, &mut vault_a, &mut seen_a);
session_a.receive(&ta).expect("合法载荷应当被接受")
};
assert_eq!(outcome.merge.added, 1);
assert_eq!(seen_a.len(), 1);
}
#[test]
fn a_payload_larger_than_the_channel_is_reported_before_sending() {
let (id_a, master, mut vault_a) = a_device(1);
let (id_b, _, _) = a_device(2);
for i in 0..40_u64 {
vault_a
.insert(sample_entry(&format!("条目 {i}"), 1_700_000_000 + i))
.expect("随机源可用");
}
let caps = TransportCapabilities::qr(4, 64);
let ta = MemoryTransport::new(id_a, caps).with_peers(vec![id_b]);
let mut seen = SeenSyncIds::new();
let session = SyncSession::open(&master, &mut vault_a, &mut seen);
let result = session.send(&ta, Recipient::One(id_b), PayloadKind::Vault);
match result {
Err(Error::TransportMessageTooLarge { needed, limit }) => {
assert_eq!(limit, 64);
assert!(needed > 64, "报出来的需求量 {needed} 应当大于上限");
}
other => panic!("应当报「装不下」,实际 {other:?}"),
}
assert!(
ta.outbox.lock().expect("锁未中毒").is_empty(),
"装不下时不该已经往通道里写了东西"
);
}
#[test]
fn only_multi_party_channels_need_a_start_signal() {
assert!(!TransportCapabilities::stream(Roster::TwoParty).needs_start_signal());
assert!(TransportCapabilities::stream(Roster::MultiParty).needs_start_signal());
assert!(!TransportCapabilities::qr(40, 2900).needs_start_signal());
assert!(TransportCapabilities::qr(40, 2900).single_use);
assert!(!TransportCapabilities::stream(Roster::TwoParty).single_use);
}
#[test]
fn a_multi_party_broadcast_reaches_every_peer() {
let (id_a, master, mut vault_a) = a_device(1);
let (id_b, _, mut vault_b) = a_device(2);
let (id_c, _, mut vault_c) = a_device(3);
vault_a
.insert(sample_entry("A 的条目", 1_700_000_000))
.expect("随机源可用");
let caps = TransportCapabilities::stream(Roster::MultiParty);
let ta = MemoryTransport::new(id_a, caps).with_peers(vec![id_b, id_c]);
let tb = MemoryTransport::new(id_b, caps).with_peers(vec![id_a, id_c]);
let tc = MemoryTransport::new(id_c, caps).with_peers(vec![id_a, id_b]);
let mut seen_a = SeenSyncIds::new();
SyncSession::open(&master, &mut vault_a, &mut seen_a)
.send(&ta, Recipient::All, PayloadKind::Vault)
.expect("广播");
ta.deliver_to(&tb, true);
ta.deliver_to(&tc, true);
let mut seen_b = SeenSyncIds::new();
let outcome_b = SyncSession::open(&master, &mut vault_b, &mut seen_b)
.receive(&tb)
.expect("B 接收");
assert_eq!(outcome_b.merge.added, 1);
let mut seen_c = SeenSyncIds::new();
let outcome_c = SyncSession::open(&master, &mut vault_c, &mut seen_c)
.receive(&tc)
.expect("C 接收");
assert_eq!(outcome_c.merge.added, 1);
assert_eq!(vault_b.len(), 1);
assert_eq!(vault_c.len(), 1);
}
#[test]
fn a_targeted_send_does_not_reach_other_peers() {
let (id_a, master, mut vault_a) = a_device(1);
let (id_b, _, mut vault_b) = a_device(2);
let (id_c, _, mut vault_c) = a_device(3);
vault_a
.insert(sample_entry("只给 B", 1_700_000_000))
.expect("随机源可用");
let caps = TransportCapabilities::stream(Roster::MultiParty);
let ta = MemoryTransport::new(id_a, caps).with_peers(vec![id_b, id_c]);
let tb = MemoryTransport::new(id_b, caps);
let tc = MemoryTransport::new(id_c, caps);
let mut seen_a = SeenSyncIds::new();
SyncSession::open(&master, &mut vault_a, &mut seen_a)
.send(&ta, Recipient::One(id_b), PayloadKind::Vault)
.expect("定向发送");
ta.deliver_to(&tc, true);
assert!(
tc.inbox.lock().expect("锁未中毒").is_empty(),
"定向发给 B 的消息送到了 C"
);
ta.deliver_to(&tb, true);
assert!(!tb.inbox.lock().expect("锁未中毒").is_empty(), "应当送到 B");
let _ = &mut vault_b;
let _ = &mut vault_c;
}
#[test]
fn syncing_with_nothing_new_reports_no_change() {
let (id_a, master, mut vault_a) = a_device(1);
let (id_b, _, mut vault_b) = a_device(2);
vault_a
.insert(sample_entry("同一条", 1_700_000_000))
.expect("随机源可用");
let caps = TransportCapabilities::stream(Roster::TwoParty);
let ta = MemoryTransport::new(id_a, caps).with_peers(vec![id_b]);
let tb = MemoryTransport::new(id_b, caps).with_peers(vec![id_a]);
let mut seen_a = SeenSyncIds::new();
SyncSession::open(&master, &mut vault_a, &mut seen_a)
.send(&ta, Recipient::One(id_b), PayloadKind::Vault)
.expect("发送");
ta.deliver_to(&tb, true);
let mut seen_b = SeenSyncIds::new();
let first = SyncSession::open(&master, &mut vault_b, &mut seen_b)
.receive(&tb)
.expect("接收");
assert_eq!(first.merge.added, 1);
assert!(!first.made_no_change(), "第一轮确实改了东西");
ta.clear_outbox();
SyncSession::open(&master, &mut vault_a, &mut seen_a)
.send(&ta, Recipient::One(id_b), PayloadKind::Vault)
.expect("发送");
ta.deliver_to(&tb, true);
let second = SyncSession::open(&master, &mut vault_b, &mut seen_b)
.receive(&tb)
.expect("接收");
assert!(
second.made_no_change(),
"内容相同时应当报「无改动」,实际 {second}"
);
assert!(!second.needs_adjudication());
}
#[test]
fn three_devices_converge_over_the_session() {
let (id_a, master, mut vault_a) = a_device(1);
let (id_b, _, mut vault_b) = a_device(2);
let (id_c, _, mut vault_c) = a_device(3);
vault_a
.insert(sample_entry("A", 1_700_000_000))
.expect("随机源可用");
vault_b
.insert(sample_entry("B", 1_700_000_100))
.expect("随机源可用");
vault_c
.insert(sample_entry("C", 1_700_000_200))
.expect("随机源可用");
let caps = TransportCapabilities::stream(Roster::MultiParty);
let ta = MemoryTransport::new(id_a, caps).with_peers(vec![id_b, id_c]);
let tb = MemoryTransport::new(id_b, caps).with_peers(vec![id_a, id_c]);
let tc = MemoryTransport::new(id_c, caps).with_peers(vec![id_a, id_b]);
let mut seen_a = SeenSyncIds::new();
let mut seen_b = SeenSyncIds::new();
let mut seen_c = SeenSyncIds::new();
for _ in 0..2 {
ta.clear_outbox();
tb.clear_outbox();
tc.clear_outbox();
SyncSession::open(&master, &mut vault_a, &mut seen_a)
.send(&ta, Recipient::All, PayloadKind::Vault)
.expect("A 广播");
SyncSession::open(&master, &mut vault_b, &mut seen_b)
.send(&tb, Recipient::All, PayloadKind::Vault)
.expect("B 广播");
SyncSession::open(&master, &mut vault_c, &mut seen_c)
.send(&tc, Recipient::All, PayloadKind::Vault)
.expect("C 广播");
ta.deliver_to(&tb, true);
ta.deliver_to(&tc, true);
tb.deliver_to(&ta, true);
tb.deliver_to(&tc, true);
tc.deliver_to(&ta, true);
tc.deliver_to(&tb, true);
for transport in [&ta, &tb, &tc] {
loop {
let result = match transport.local_device() {
id if id == id_a => {
SyncSession::open(&master, &mut vault_a, &mut seen_a).receive(transport)
}
id if id == id_b => {
SyncSession::open(&master, &mut vault_b, &mut seen_b).receive(transport)
}
_ => {
SyncSession::open(&master, &mut vault_c, &mut seen_c).receive(transport)
}
};
match result {
Ok(_) => {}
Err(Error::SyncNoSessionFound) => break,
Err(other) => panic!("不该出现的错误:{other:?}"),
}
}
}
}
assert_eq!(vault_a.len(), 3, "A 应当有三条");
assert_eq!(vault_b.len(), 3, "B 应当有三条");
assert_eq!(vault_c.len(), 3, "C 应当有三条");
}
#[test]
fn planning_and_sending_agree() {
let (id_a, master, mut vault_a) = a_device(1);
let (id_b, _, _) = a_device(2);
vault_a
.insert(sample_entry("条目", 1_700_000_000))
.expect("随机源可用");
let roomy = MemoryTransport::new(id_a, TransportCapabilities::stream(Roster::TwoParty));
let tiny = MemoryTransport::new(id_a, TransportCapabilities::qr(1, 8));
let mut seen = SeenSyncIds::new();
let session = SyncSession::open(&master, &mut vault_a, &mut seen);
let len = match session.plan_share(&roomy).expect("规划") {
SharePlan::FullPayload { len } => len,
other => panic!("宽裕通道不该要求建链:{other}"),
};
assert!(len > 0, "一条条目的载荷不可能是空的");
session
.send(&roomy, Recipient::One(id_b), PayloadKind::Vault)
.expect("规划说装得下,发送就必须成功");
let plan = session.plan_share(&tiny).expect("规划");
match plan {
SharePlan::LinkEstablishment { needed, limit } => {
assert_eq!(limit, 8, "上限应当来自通道的能力声明");
assert_eq!(
needed, len,
"两次规划给出的载荷大小不一致——那说明封装不是确定性的长度"
);
}
other => panic!("装不下时应当要求建链:{other}"),
}
let result = session.send(&tiny, Recipient::One(id_b), PayloadKind::Vault);
let is_too_large = matches!(result, Err(Error::TransportMessageTooLarge { .. }));
assert!(
is_too_large,
"规划说装不下,发送也必须报「装不下」:{result:?}"
);
}
#[test]
fn the_share_plan_depends_only_on_the_declared_byte_limit() {
let (id_a, master, mut vault_a) = a_device(1);
vault_a
.insert(sample_entry("条目", 1_700_000_000))
.expect("随机源可用");
let framed = MemoryTransport::new(id_a, TransportCapabilities::qr(40, 4096));
let equivalent = MemoryTransport::new(
id_a,
TransportCapabilities {
roster: Roster::MultiParty,
max_message_len: Some(4096),
max_qr_version: None,
single_use: false,
can_carry_payload: true,
},
);
let tiny_framed = MemoryTransport::new(id_a, TransportCapabilities::qr(1, 8));
let tiny_equivalent = MemoryTransport::new(
id_a,
TransportCapabilities {
roster: Roster::TwoParty,
max_message_len: Some(8),
max_qr_version: None,
single_use: false,
can_carry_payload: true,
},
);
let mut seen = SeenSyncIds::new();
let session = SyncSession::open(&master, &mut vault_a, &mut seen);
for (left, right, what) in [
(&framed, &equivalent, "宽裕"),
(&tiny_framed, &tiny_equivalent, "极小"),
] {
let a = session.plan_share(left).expect("规划");
let b = session.plan_share(right).expect("规划");
assert_eq!(
a, b,
"{what}通道:同样的字节上限必须给出同样的答复(帧的有无不该影响这个判断)"
);
}
}
#[test]
fn a_stream_channel_never_needs_a_link() {
let (id_a, master, mut vault_a) = a_device(1);
for i in 0..64_u64 {
vault_a
.insert(sample_entry(&format!("条目 {i}"), 1_700_000_000 + i))
.expect("随机源可用");
}
let transport =
MemoryTransport::new(id_a, TransportCapabilities::stream(Roster::MultiParty));
assert!(
transport.capabilities().max_message_len.is_none(),
"字节流通道不该有单条上限"
);
let mut seen = SeenSyncIds::new();
let plan = SyncSession::open(&master, &mut vault_a, &mut seen)
.plan_share(&transport)
.expect("规划");
assert!(!plan.needs_link(), "字节流通道不该要求建链:{plan}");
}
#[test]
fn a_request_fits_where_the_full_state_does_not() {
let (id_a, master, mut vault_a) = a_device(1);
let (id_b, _, _) = a_device(2);
for i in 0..8_u64 {
vault_a
.insert(sample_entry(&format!("条目 {i}"), 1_700_000_000 + i))
.expect("随机源可用");
}
let transport = MemoryTransport::new(id_a, TransportCapabilities::qr(10, 200));
let mut seen = SeenSyncIds::new();
let session = SyncSession::open(&master, &mut vault_a, &mut seen);
let plan = session.plan_share(&transport).expect("规划");
assert!(plan.needs_link(), "全量状态应当装不下:{plan}");
session
.send(&transport, Recipient::One(id_b), PayloadKind::Request)
.expect("请求必须装得下——否则「建链」这条路的第一步就不成立");
}
#[test]
fn a_divergent_entry_is_reported_for_adjudication() {
let (id_a, master, mut vault_a) = a_device(1);
let (id_b, _, mut vault_b) = a_device(2);
let shared = EntryId::from_bytes([9; EntryId::LEN]);
vault_a
.insert_with_id(shared, sample_entry("本机版本", 1_700_000_000))
.expect("插入");
vault_b
.insert_with_id(shared, sample_entry("对端版本", 1_700_000_500))
.expect("插入");
let caps = TransportCapabilities::stream(Roster::TwoParty);
let ta = MemoryTransport::new(id_a, caps).with_peers(vec![id_b]);
let tb = MemoryTransport::new(id_b, caps).with_peers(vec![id_a]);
let mut seen_a = SeenSyncIds::new();
SyncSession::open(&master, &mut vault_a, &mut seen_a)
.send(&ta, Recipient::One(id_b), PayloadKind::Vault)
.expect("发送");
ta.deliver_to(&tb, true);
let mut seen_b = SeenSyncIds::new();
let outcome = SyncSession::open(&master, &mut vault_b, &mut seen_b)
.receive(&tb)
.expect("接收");
assert!(
outcome.needs_adjudication(),
"两端内容不同时应当报冲突,实际 {outcome}"
);
assert_eq!(outcome.merge.conflicts.len(), 1);
assert_eq!(
vault_b.get(&shared).expect("条目还在").name,
"对端版本",
"冲突不该自动改掉本地内容"
);
}
#[test]
fn the_merge_helpers_stay_usable_without_a_transport() {
let (_, master, mut vault) = a_device(1);
let mut source = VaultAlias::new();
source
.insert(sample_entry("来自别处", 1_700_000_000))
.expect("随机源可用");
let payload =
Payload::from_vault(&master, DeviceId::from_bytes([2; 32]), &source).expect("打包");
let mut set = RecordSet::new();
for record in payload.open_records(&master).expect("解密") {
set.insert(record.into_merged());
}
let outcome = merge_into(&mut vault, &set).expect("合并");
assert_eq!(outcome.added, 1);
assert_eq!(vault.len(), 1);
}
}