Commit a46bcc611e2e0f0799e9271027594e97e0df9e92 | snowbound

author	clover caruso <git@paperclover.net> 2026-10-12 22:37:02-07:00
committer	clover caruso <git@paperclover.net> 2026-10-12 22:38:18-07:00
log	a46bcc611e2e0f0799e9271027594e97e0df9e92
tree	c1e40cc1f35c7ab2f1bafc2f818d77df93114d79
parent	6badb301e0d6afedf55c92da8041ef24f0a84caa
signature	Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU (~clover)

fix(live): fence handover control frames to current streams

Route credential rotation through authenticated access streams and serialize native welcome snapshots with their enqueue. Release browser membership locks before fallible sends and stop a superseded rotation batch. Preserve stream incarnation checks and terminal retirement.

Fixes #141
Assisted-by: gpt-6.1-sol

8 files changed, 1440 insertions(+), 242 deletions(-)

crates
- notebook
  - src
    live.rs
    live
    relay.rs
    share.rs
    share
    batch.rs
    handover_tests.rs
    membership.rs
    web.rs
  - tests
    live_membership.rs

crates/notebook/src/live.rs+55-24

...	...	@@ -61,7 +61,8 @@ struct Shared {
61	61	secret: Vec<u8>,
62	62	reach: Option<Reach>,
63	63	state: Mutex<State>,
64		events: Box<dyn Fn(Event) + Send + Sync>,
	64	#[allow(clippy::type_complexity)]
	65	events: Box<dyn Fn(&Arc<Shared>, Event) + Send + Sync>,
65	66	stopped: AtomicBool,
66	67	connections: AtomicU64,
67	68	}
...	...	@@ -231,6 +232,16 @@ impl Live {
231	232	reach: Option<Reach>,
232	233	relay: Option<&str>,
233	234	events: impl Fn(Event) + Send + Sync + 'static,
	235	) -> io::Result<Live> {
	236	Self::start_with(me, room, reach, relay, move \|_, event\| events(event))
	237	}
	238
	239	fn start_with(
	240	me: Hello,
	241	room: &Room,
	242	reach: Option<Reach>,
	243	relay: Option<&str>,
	244	events: impl Fn(&Arc<Shared>, Event) + Send + Sync + 'static,
234	245	) -> io::Result<Live> {
235	246	let host = match reach {
236	247	Some(Reach::Network) => IpAddr::V4(Ipv4Addr::UNSPECIFIED),
...	...	@@ -440,6 +451,21 @@ impl Drop for Live {
440	451	}
441	452
442	453	impl Shared {
	454	fn gone(&self, hello: &Arc<Hello>) -> bool {
	455	!self.state.lock().unwrap().peers.contains_key(&hello.peer)
	456	}
	457
	458	fn connected(&self, hello: &Arc<Hello>) -> bool {
	459	!self.stopped.load(Ordering::Acquire)
	460	&& self
	461	.state
	462	.lock()
	463	.unwrap()
	464	.peers
	465	.get(&hello.peer)
	466	.is_some_and(\|link\| Arc::ptr_eq(&link.peer.hello, hello))
	467	}
	468
443	469	/// The peers in the room: those met directly, and the rest as the relay's group or a
444	470	/// stream through the relay last heard of them.
445	471	fn peers(&self) -> Vec<Peer> {
...	...	@@ -547,7 +573,7 @@ impl Shared {
547	573
548	574	/// Counts a meeting that failed on the secret: the end sharing a code burns it after too
549	575	/// many, and an end that typed one gives up at once.
550		fn failed(&self) {
	576	fn failed(self: &Arc<Self>) {
551	577	let mut state = self.state.lock().unwrap();
552	578	state.failed += 1;
553	579	match &self.room {
...	...	@@ -556,7 +582,7 @@ impl Shared {
556	582	_ => {}
557	583	}
558	584	drop(state);
559		(self.events)(Event::Changed);
	585	(self.events)(self, Event::Changed);
560	586	}
561	587
562	588	/// Meets the peer at the other end of `pipe` in room `tag`, then reads from it until it
...	...	@@ -597,7 +623,7 @@ impl Shared {
597	623	if matches!(self.room, Room::Code { owner: false, .. }) {
598	624	self.stopped.store(true, Ordering::Release);
599	625	}
600		(self.events)(Event::Changed);
	626	(self.events)(&self, Event::Changed);
601	627	} else if error.kind() == io::ErrorKind::InvalidData {
602	628	pipe.met(false);
603	629	self.failed();
...	...	@@ -614,17 +640,14 @@ impl Shared {
614	640	let connection = self.connections.fetch_add(1, Ordering::Relaxed);
615	641	let (out, outgoing) = mpsc::channel();
616	642	let line = Line(out);
617		{
	643	let retired = {
618	644	let mut state = self.state.lock().unwrap();
619	645	// A peer met both directly and through a relay keeps the direct connection, as
620	646	// both ends then agree.
621	647	let kept = match state.peers.get(&peer) {
622	648	_ if peer == self.me.peer => false,
623	649	Some(link) if link.pipe.direct() \|\| !pipe.direct() => false,
624		Some(link) => {
625		link.pipe.shutdown();
626		true
627		}
	650	Some(_) => true,
628	651	None => true,
629	652	};
630	653	if !kept {
...	...	@@ -632,9 +655,6 @@ impl Shared {
632	655	pipe.shutdown();
633	656	return false;
634	657	}
635		if self.carries_presence(&*pipe) {
636		let _ = line.0.send(Out::Presence);
637		}
638	658	state.peers.insert(
639	659	peer,
640	660	Link {
...	...	@@ -646,13 +666,19 @@ impl Shared {
646	666	line: line.clone(),
647	667	pipe: Arc::clone(&pipe),
648	668	},
649		);
	669	)
	670	};
	671	if let Some(retired) = retired {
	672	retired.pipe.shutdown();
	673	}
	674	if self.carries_presence(&*pipe) {
	675	let _ = line.0.send(Out::Presence);
650	676	}
651	677	let writing = Arc::clone(&pipe);
652	678	let shared = Arc::clone(&self);
653	679	thread::spawn(move \|\| shared.write(&*writing, send, outgoing));
654		(self.events)(Event::Met(&hello, &line));
655		(self.events)(Event::Changed);
	680	(self.events)(&self, Event::Met(&hello, &line));
	681	(self.events)(&self, Event::Changed);
656	682	let ended = loop {
657	683	let (message, body) = match receive.receive(&mut stream) {
658	684	Ok(frame) => frame,
...	...	@@ -664,17 +690,22 @@ impl Shared {
664	690	let Ok(presence) = minicbor::decode::<Presence>(&body) else {
665	691	break None;
666	692	};
667		if let Some(link) = self.state.lock().unwrap().peers.get_mut(&peer) {
	693	if let Some(link) = self.state.lock().unwrap().peers.get_mut(&peer)
	694	&& link.connection == connection
	695	{
668	696	link.peer.presence = Some(presence);
669	697	}
670		(self.events)(Event::Changed);
	698	(self.events)(&self, Event::Changed);
671	699	}
672	700	kind => {
673		(self.events)(Event::Frame {
674		from: &hello,
675		kind,
676		body: &body,
677		});
	701	(self.events)(
	702	&self,
	703	Event::Frame {
	704	from: &hello,
	705	kind,
	706	body: &body,
	707	},
	708	);
678	709	// The peer hangs up after its bye.
679	710	if kind == kind::BYE {
680	711	break None;
...	...	@@ -695,8 +726,8 @@ impl Shared {
695	726	{
696	727	state.peers.remove(&peer);
697	728	drop(state);
698		(self.events)(Event::Left(&hello));
699		(self.events)(Event::Changed);
	729	(self.events)(&self, Event::Left(&hello));
	730	(self.events)(&self, Event::Changed);
700	731	}
701	732	true
702	733	}

crates/notebook/src/live/relay.rs+16-13

...	...	@@ -44,12 +44,12 @@ pub(super) fn join(shared: &Arc<Shared>, url: &str, port: u16) -> io::Result<()>
44	44	}
45	45
46	46	/// Records how the relay answered, telling `shared`'s events where it changed.
47		fn answered(shared: &Shared, answer: Answer) {
	47	fn answered(shared: &Arc<Shared>, answer: Answer) {
48	48	let mut state = shared.state.lock().unwrap();
49	49	if state.relayed.as_ref() != Some(&answer) {
50	50	state.relayed = Some(answer);
51	51	drop(state);
52		(shared.events)(Event::Changed);
	52	(shared.events)(shared, Event::Changed);
53	53	}
54	54	}
55	55
...	...	@@ -245,7 +245,7 @@ fn session(
245	245	state.code = Some(code);
246	246	drop(state);
247	247	shared.advertise(port);
248		(shared.events)(Event::Changed);
	248	(shared.events)(shared, Event::Changed);
249	249	}
250	250	}
251	251	Ok(Notice::Welcome { you, members }) => {
...	...	@@ -263,14 +263,14 @@ fn session(
263	263	let left = (socket.group.as_ref())
264	264	.and_then(\|group\| group.members.lock().unwrap().remove(&slot));
265	265	if left.is_some() {
266		(shared.events)(Event::Changed);
	266	(shared.events)(shared, Event::Changed);
267	267	}
268	268	}
269	269	Ok(Notice::Burned) => {
270	270	// Coming back, ask for another number.
271	271	*nameplate = None;
272	272	shared.state.lock().unwrap().burned = true;
273		(shared.events)(Event::Changed);
	273	(shared.events)(shared, Event::Changed);
274	274	}
275	275	Err(()) => {}
276	276	},
...	...	@@ -296,7 +296,7 @@ fn session(
296	296	if let Some(group) = &socket.group {
297	297	*group.out.lock().unwrap() = None;
298	298	group.members.lock().unwrap().clear();
299		(shared.events)(Event::Changed);
	299	(shared.events)(shared, Event::Changed);
300	300	}
301	301	}
302	302
...	...	@@ -412,7 +412,7 @@ fn heard(shared: &Arc<Shared>, socket: &Arc<Socket>, tag: Option<&str>, slot: u3
412	412	if serves {
413	413	meet(shared, socket, tag, slot, None);
414	414	}
415		(shared.events)(Event::Changed);
	415	(shared.events)(shared, Event::Changed);
416	416	}
417	417	kind::PRESENCE => {
418	418	let Ok(presence) = minicbor::decode::<Presence>(&body) else {
...	...	@@ -422,17 +422,20 @@ fn heard(shared: &Arc<Shared>, socket: &Arc<Socket>, tag: Option<&str>, slot: u3
422	422	if let Some(peer) = members.get_mut(&slot).and_then(\|m\| m.peer.as_mut()) {
423	423	peer.presence = Some(presence);
424	424	drop(members);
425		(shared.events)(Event::Changed);
	425	(shared.events)(shared, Event::Changed);
426	426	}
427	427	}
428	428	// A peer met directly says the rest there.
429	429	kind if kind < 256 => {
430	430	if let Some(hello) = hello.filter(\|hello\| !shared.direct(&hello.peer)) {
431		(shared.events)(Event::Frame {
432		from: &hello,
433		kind,
434		body: &body,
435		});
	431	(shared.events)(
	432	shared,
	433	Event::Frame {
	434	from: &hello,
	435	kind,
	436	body: &body,
	437	},
	438	);
436	439	}
437	440	}
438	441	_ => {}

crates/notebook/src/live/share.rs+294-132

...	...	@@ -18,7 +18,7 @@ use std::{
18	18	io,
19	19	path::PathBuf,
20	20	sync::{
21		Arc, Condvar, Mutex,
	21	Arc, Condvar, Mutex, Weak,
22	22	atomic::{AtomicBool, AtomicU64, Ordering},
23	23	mpsc,
24	24	},
...	...	@@ -29,6 +29,9 @@ use web_time::Instant;
29	29
30	30	#[path = "share/batch.rs"]
31	31	mod batch;
	32	#[cfg(all(test, not(target_arch = "wasm32")))]
	33	#[path = "share/handover_tests.rs"]
	34	mod handover_tests;
32	35	#[path = "share/membership.rs"]
33	36	mod membership;
34	37	#[cfg(target_arch = "wasm32")]
...	...	@@ -304,6 +307,7 @@ struct Served {
304	307	/// many more it may start now, and the sections it read lately, newest first, as it keeps
305	308	/// their images.
306	309	struct Admitted {
	310	hello: Arc<Hello>,
307	311	line: Line,
308	312	queue: mpsc::SyncSender<(u16, Vec<u8>)>,
309	313	starts: f64,
...	...	@@ -336,9 +340,32 @@ fn refused(kind: io::ErrorKind, message: &str) -> Error {
336	340
337	341	impl Served {
338	342	/// Serves `peer` on `line`, through workers of its own that end as it leaves.
339		fn admit(self: &Arc<Self>, peer: [u8; 16], line: &Line) {
	343	fn admit(
	344	self: &Arc<Self>,
	345	source: &Arc<super::Shared>,
	346	hello: &Arc<Hello>,
	347	line: &Line,
	348	) -> bool {
	349	let peer = hello.peer;
340	350	let (queue, waiting) = mpsc::sync_channel::<(u16, Vec<u8>)>(QUEUED - WORKERS);
341	351	let waiting = Arc::new(Mutex::new(waiting));
	352	let guest = Admitted {
	353	hello: Arc::clone(hello),
	354	line: line.clone(),
	355	queue,
	356	starts: BURST,
	357	counted: Instant::now(),
	358	held: Vec::new(),
	359	};
	360	let mut guests = self.guests.lock().unwrap();
	361	if !source.connected(hello) {
	362	drop(guests);
	363	drop(guest);
	364	return false;
	365	}
	366	let retired = guests.insert(peer, guest);
	367	drop(guests);
	368	drop(retired);
342	369	for _ in 0..WORKERS {
343	370	let (served, waiting, line) =
344	371	(Arc::downgrade(self), Arc::clone(&waiting), line.clone());
...	...	@@ -352,23 +379,20 @@ impl Served {
352	379	}
353	380	});
354	381	}
355		let guest = Admitted {
356		line: line.clone(),
357		queue,
358		starts: BURST,
359		counted: Instant::now(),
360		held: Vec::new(),
361		};
362		self.guests.lock().unwrap().insert(peer, guest);
	382	true
363	383	}
364	384
365	385	/// Hands a request from `peer` to its workers, or hangs up on a guest that has too many
366	386	/// waiting or starts them too fast.
367		fn queue(&self, peer: &[u8; 16], kind: u16, body: &[u8]) {
	387	fn queue(&self, source: &Arc<super::Shared>, hello: &Arc<Hello>, kind: u16, body: &[u8]) {
	388	let peer = &hello.peer;
368	389	let mut guests = self.guests.lock().unwrap();
369	390	let Some(guest) = guests.get_mut(peer) else {
370	391	return;
371	392	};
	393	if !Arc::ptr_eq(&guest.hello, hello) \|\| !source.connected(hello) {
	394	return;
	395	}
372	396	let continued = kind == kind::PUT
373	397	\|\| matches!(kind, kind::READ \| kind::READ_FILE)
374	398	&& minicbor::decode::<Request>(body).is_ok_and(\|request\| request.handle.is_some());
...	...	@@ -380,21 +404,33 @@ impl Served {
380	404	guest.starts -= 1.0;
381	405	}
382	406	if guest.starts < 0.0 \|\| guest.queue.try_send((kind, body.to_vec())).is_err() {
383		guest.line.hang_up(FLOODED);
384		guests.remove(peer);
	407	let guest = guests.remove(peer);
	408	drop(guests);
	409	if let Some(guest) = guest {
	410	guest.line.hang_up(FLOODED);
	411	}
385	412	}
386	413	}
387	414
388		fn forget(&self, peer: &[u8; 16]) {
389		self.guests.lock().unwrap().remove(peer);
390		self.snapshots
391		.lock()
392		.unwrap()
393		.retain(\|(guest, _), _\| guest != peer);
394		self.puts
395		.lock()
396		.unwrap()
397		.retain(\|(guest, _), _\| guest != peer);
	415	fn forget(&self, peer: &[u8; 16], expected: Option<(&Arc<super::Shared>, &Arc<Hello>)>) {
	416	let retired = {
	417	let mut snapshots = self.snapshots.lock().unwrap();
	418	let mut puts = self.puts.lock().unwrap();
	419	let mut guests = self.guests.lock().unwrap();
	420	if let Some((source, hello)) = expected
	421	&& (!source.gone(hello)
	422	\|\| !guests
	423	.get(peer)
	424	.is_some_and(\|guest\| Arc::ptr_eq(&guest.hello, hello)))
	425	{
	426	return;
	427	}
	428	let retired = guests.remove(peer);
	429	snapshots.retain(\|(guest, _), _\| guest != peer);
	430	puts.retain(\|(guest, _), _\| guest != peer);
	431	retired
	432	};
	433	drop(retired);
398	434	}
399	435
400	436	/// Tells every guest, and the host, that a guest changed the files at `paths`.
...	...	@@ -1086,10 +1122,16 @@ impl Guest {
1086	1122	catalog: Mutex::default(),
1087	1123	});
1088	1124	let heard = Arc::clone(&inner);
1089		let live = Live::start(me, &Room::Notebook(secret), reach, relay, move \|event\| {
1090		heard.heard(event);
1091		(heard.events)();
1092		})?;
	1125	let live = Live::start_with(
	1126	me,
	1127	&Room::Notebook(secret),
	1128	reach,
	1129	relay,
	1130	move \|source, event\| {
	1131	heard.heard(source, event);
	1132	(heard.events)();
	1133	},
	1134	)?;
1093	1135	Ok(Arc::new(Self { live, inner }))
1094	1136	}
1095	1137
...	...	@@ -1565,146 +1607,266 @@ impl Inner {
1565	1607	}
1566	1608	}
1567	1609
1568		fn heard(self: &Arc<Self>, event: Event) {
1569		let serves = \|hello: &Hello\| {
1570		self.host
1571		.lock()
1572		.unwrap()
1573		.as_ref()
1574		.is_some_and(\|(host, _)\| host.peer == hello.peer)
1575		};
	1610	fn heard(self: &Arc<Self>, source: &Arc<super::Shared>, event: Event) {
1576	1611	match event {
1577	1612	Event::Met(hello, line) if hello.serves == Some(self.share) => {
1578		*self.host.lock().unwrap() = Some((Arc::clone(hello), line.clone()));
	1613	let mut host = self.host.lock().unwrap();
	1614	let retired = if source.connected(hello) {
	1615	host.replace((Arc::clone(hello), line.clone()))
	1616	} else {
	1617	None
	1618	};
	1619	drop(host);
	1620	drop(retired);
1579	1621	}
1580		Event::Left(hello) if serves(hello) => {
1581		*self.host.lock().unwrap() = None;
1582		drop(self.presence.lock().unwrap().take());
1583		self.current.lock().unwrap().clear();
1584		// Each request waiting hears its answer was lost.
1585		self.pending.lock().unwrap().clear();
1586		if let Some(reports) = self.watch.lock().unwrap().take() {
	1622	Event::Left(hello) => {
	1623	let (retired, presence, pending, reports) = {
	1624	let mut host = self.host.lock().unwrap();
	1625	if !host
	1626	.as_ref()
	1627	.is_some_and(\|(admitted, _)\| Arc::ptr_eq(admitted, hello))
	1628	\|\| !source.gone(hello)
	1629	{
	1630	return;
	1631	}
	1632	let retired = host.take();
	1633	let presence = self.presence.lock().unwrap().take();
	1634	self.current.lock().unwrap().clear();
	1635	let pending = std::mem::take(&mut *self.pending.lock().unwrap());
	1636	let reports = self.watch.lock().unwrap().take();
	1637	(retired, presence, pending, reports)
	1638	};
	1639	drop(retired);
	1640	drop(presence);
	1641	drop(pending);
	1642	if let Some(reports) = reports {
1587	1643	reports.lost();
1588	1644	}
1589	1645	}
1590		Event::Frame {
1591		from, kind, body, ..
1592		} => match kind {
1593		kind::WELCOME if serves(from) => {
1594		if let Ok(welcome) = minicbor::decode::<Welcome>(body) {
1595		self.join_presence(welcome.room);
1596		}
	1646	Event::Frame { from, kind, body } => {
	1647	let host = self.host.lock().unwrap();
	1648	if !host
	1649	.as_ref()
	1650	.is_some_and(\|(admitted, _)\| Arc::ptr_eq(admitted, from))
	1651	\|\| !source.connected(from)
	1652	{
	1653	return;
1597	1654	}
1598		kind::REPLY if serves(from) => {
1599		if let Ok(reply) = minicbor::decode::<Reply>(body)
1600		&& let Some(waiting) = self.pending.lock().unwrap().remove(&reply.id)
1601		{
1602		let _ = waiting.send(reply);
	1655	match kind {
	1656	kind::WELCOME => {
	1657	if let Ok(welcome) = minicbor::decode::<Welcome>(body) {
	1658	drop(host);
	1659	self.join_presence(source, from, welcome.room);
	1660	}
1603	1661	}
1604		}
1605		kind::DELTA if serves(from) => {
1606		if let Ok(delta) = minicbor::decode::<Delta>(body) {
1607		self.apply(&delta);
	1662	kind::REPLY => {
	1663	if let Ok(reply) = minicbor::decode::<Reply>(body) {
	1664	let waiting = self.pending.lock().unwrap().remove(&reply.id);
	1665	drop(host);
	1666	if let Some(waiting) = waiting {
	1667	let _ = waiting.send(reply);
	1668	}
	1669	}
1608	1670	}
1609		}
1610		kind::TOUCHED if serves(from) => {
1611		if let Ok(touched) = minicbor::decode::<Touched>(body) {
1612		self.touched(&touched);
1613		let reports = self.watch.lock().unwrap().clone();
1614		if let Some(reports) = reports {
1615		reports.touched(&touched.paths);
	1671	kind::DELTA => {
	1672	if let Ok(delta) = minicbor::decode::<Delta>(body) {
	1673	self.apply(&delta);
1616	1674	}
1617	1675	}
1618		}
1619		kind::BYE if serves(from) => {
1620		if let Ok(bye) = minicbor::decode::<wire::Bye>(body) {
1621		let ended = match bye.reason.as_str() {
1622		STOPPED => Some(Ended::Stopped),
1623		"removed" => Some(Ended::Removed),
1624		_ => None,
1625		};
1626		if ended.is_some() {
1627		*self.ended.lock().unwrap() = ended;
	1676	kind::TOUCHED => {
	1677	if let Ok(touched) = minicbor::decode::<Touched>(body) {
	1678	self.touched(&touched);
	1679	let reports = self.watch.lock().unwrap().clone();
	1680	drop(host);
	1681	if let Some(reports) = reports {
	1682	reports.touched(&touched.paths);
	1683	}
	1684	}
	1685	}
	1686	kind::BYE => {
	1687	if let Ok(bye) = minicbor::decode::<wire::Bye>(body) {
	1688	let ended = match bye.reason.as_str() {
	1689	STOPPED => Some(Ended::Stopped),
	1690	"removed" => Some(Ended::Removed),
	1691	_ => None,
	1692	};
	1693	if ended.is_some() {
	1694	*self.ended.lock().unwrap() = ended;
	1695	}
1628	1696	}
1629	1697	}
	1698	_ => {}
1630	1699	}
1631		_ => {}
1632		},
	1700	}
1633	1701	_ => {}
1634	1702	}
1635	1703	}
1636	1704
1637		fn join_presence(self: &Arc<Self>, secret: [u8; 16]) {
1638		let mut presence = self.presence.lock().unwrap();
1639		if presence.as_ref().is_some_and(\|(held, _)\| *held == secret) {
1640		return;
	1705	fn join_presence(
	1706	self: &Arc<Self>,
	1707	source: &Arc<super::Shared>,
	1708	from: &Arc<Hello>,
	1709	secret: [u8; 16],
	1710	) {
	1711	{
	1712	let host = self.host.lock().unwrap();
	1713	if !host
	1714	.as_ref()
	1715	.is_some_and(\|(admitted, _)\| Arc::ptr_eq(admitted, from))
	1716	\|\| !source.connected(from)
	1717	\|\| self
	1718	.presence
	1719	.lock()
	1720	.unwrap()
	1721	.as_ref()
	1722	.is_some_and(\|(held, _)\| *held == secret)
	1723	{
	1724	return;
	1725	}
1641	1726	}
1642	1727	let inner = Arc::downgrade(self);
1643	1728	let connected = Mutex::new(None);
1644		let live = Live::start(
	1729	let live = Live::start_with(
1645	1730	self.me.clone(),
1646	1731	&Room::Notebook(secret),
1647	1732	self.reach,
1648	1733	self.relay.as_deref(),
1649		move \|event\| {
1650		if let Some(inner) = inner.upgrade() {
1651		if matches!(event, Event::Changed) {
1652		let mut connected = connected.lock().unwrap();
1653		let host = inner
1654		.host
1655		.lock()
1656		.unwrap()
1657		.as_ref()
1658		.map(\|(host, _)\| host.peer);
1659		let peers = inner
1660		.presence
1661		.lock()
1662		.unwrap()
1663		.as_ref()
1664		.map(\|(_, live)\| live.peers())
1665		.unwrap_or_default();
1666		let present =
1667		host.filter(\|host\| peers.iter().any(\|peer\| peer.hello.peer == *host));
1668		if *connected != present {
1669		*connected = present;
1670		inner.current.lock().unwrap().clear();
1671		drop(connected);
1672		let reports = inner.watch.lock().unwrap().clone();
1673		if present.is_some()
1674		&& let Some(reports) = reports
1675		{
1676		reports.touched(&[String::new()]);
	1734	move \|source, event\| {
	1735	let Some(inner) = inner.upgrade() else {
	1736	return;
	1737	};
	1738	let reports = {
	1739	let host = inner.host.lock().unwrap();
	1740	let presence = inner.presence.lock().unwrap();
	1741	let Some((_, live)) = presence.as_ref() else {
	1742	return;
	1743	};
	1744	#[cfg(not(target_arch = "wasm32"))]
	1745	let selected = Arc::ptr_eq(&live.shared, source);
	1746	#[cfg(target_arch = "wasm32")]
	1747	let selected = Arc::ptr_eq(&live.0, source);
	1748	if !selected {
	1749	return;
	1750	}
	1751	match event {
	1752	Event::Changed => {
	1753	let mut connected = connected.lock().unwrap();
	1754	let host = host.as_ref().map(\|(host, _)\| host.peer);
	1755	let peers = source.peers();
	1756	let present = host
	1757	.filter(\|host\| peers.iter().any(\|peer\| peer.hello.peer == *host));
	1758	if *connected == present {
	1759	None
	1760	} else {
	1761	*connected = present;
	1762	inner.current.lock().unwrap().clear();
	1763	if present.is_some() {
	1764	inner
	1765	.watch
	1766	.lock()
	1767	.unwrap()
	1768	.clone()
	1769	.map(\|reports\| (reports, vec![String::new()]))
	1770	} else {
	1771	None
	1772	}
1677	1773	}
1678	1774	}
1679		}
1680		if matches!(
1681		event,
1682		Event::Frame {
1683		kind: kind::DELTA \| kind::TOUCHED,
1684		..
	1775	Event::Frame { from, kind, body }
	1776	if host
	1777	.as_ref()
	1778	.is_some_and(\|(host, _)\| host.peer == from.peer)
	1779	&& source
	1780	.peers()
	1781	.iter()
	1782	.any(\|peer\| Arc::ptr_eq(&peer.hello, from)) =>
	1783	{
	1784	match kind {
	1785	kind::DELTA => {
	1786	if let Ok(delta) = minicbor::decode::<Delta>(body) {
	1787	inner.apply(&delta);
	1788	}
	1789	None
	1790	}
	1791	kind::TOUCHED => {
	1792	if let Ok(touched) = minicbor::decode::<Touched>(body) {
	1793	inner.touched(&touched);
	1794	inner
	1795	.watch
	1796	.lock()
	1797	.unwrap()
	1798	.clone()
	1799	.map(\|reports\| (reports, touched.paths))
	1800	} else {
	1801	None
	1802	}
	1803	}
	1804	_ => None,
	1805	}
1685	1806	}
1686		) {
1687		inner.heard(event);
	1807	_ => None,
1688	1808	}
1689		(inner.events)();
	1809	};
	1810	if let Some((reports, paths)) = reports {
	1811	reports.touched(&paths);
1690	1812	}
	1813	(inner.events)();
1691	1814	},
1692	1815	);
1693	1816	match live {
1694		Ok(live) => {
1695		live.set_presence(self.here.lock().unwrap().clone());
1696		*presence = Some((secret, live));
1697		drop(presence);
1698		let mut current = self.current.lock().unwrap();
1699		let paths: Vec<String> = current.keys().cloned().collect();
1700		current.clear();
1701		drop(current);
1702		let reports = self.watch.lock().unwrap().clone();
1703		if let Some(reports) = reports {
1704		reports.touched(&paths);
	1817	Ok(live) => self.install_presence(source, from, secret, live),
	1818	Err(error) => eprintln!("Live Share: could not join presence: {error}"),
	1819	}
	1820	}
	1821
	1822	fn install_presence(
	1823	self: &Arc<Self>,
	1824	source: &Arc<super::Shared>,
	1825	from: &Arc<Hello>,
	1826	secret: [u8; 16],
	1827	live: Live,
	1828	) {
	1829	let here = self.here.lock().unwrap();
	1830	live.set_presence(here.clone());
	1831	#[cfg(not(target_arch = "wasm32"))]
	1832	let refresh = Arc::clone(&live.shared);
	1833	#[cfg(target_arch = "wasm32")]
	1834	let refresh = Arc::clone(&live.0);
	1835	let mut candidate = Some(live);
	1836	let (retired, paths, reports) = {
	1837	let host = self.host.lock().unwrap();
	1838	if host
	1839	.as_ref()
	1840	.is_some_and(\|(admitted, _)\| Arc::ptr_eq(admitted, from))
	1841	&& source.connected(from)
	1842	{
	1843	let mut presence = self.presence.lock().unwrap();
	1844	if presence.as_ref().is_some_and(\|(held, _)\| *held == secret) {
	1845	(None, Vec::new(), None)
	1846	} else {
	1847	let retired = presence.replace((secret, candidate.take().unwrap()));
	1848	let paths = self
	1849	.current
	1850	.lock()
	1851	.unwrap()
	1852	.drain()
	1853	.map(\|(path, _)\| path)
	1854	.collect::<Vec<_>>();
	1855	let reports = self.watch.lock().unwrap().clone();
	1856	(retired, paths, reports)
1705	1857	}
	1858	} else {
	1859	(None, Vec::new(), None)
1706	1860	}
1707		Err(error) => eprintln!("Live Share: could not join presence: {error}"),
	1861	};
	1862	drop(here);
	1863	drop(retired);
	1864	if candidate.is_none() {
	1865	(refresh.events)(&refresh, Event::Changed);
	1866	}
	1867	drop(candidate);
	1868	if let Some(reports) = reports {
	1869	reports.touched(&paths);
1708	1870	}
1709	1871	}
1710	1872	}

crates/notebook/src/live/share/batch.rs+4

...	...	@@ -342,6 +342,10 @@ mod tests {
342	342	served.guests.lock().unwrap().insert(
343	343	peer,
344	344	Admitted {
	345	hello: Arc::new(Hello {
	346	peer,
	347	..Hello::new("Guest".into(), None).unwrap()
	348	}),
345	349	line: crate::live::Line(line),
346	350	queue,
347	351	starts: BURST,

crates/notebook/src/live/share/handover_tests.rs created+721

...	...	@@ -0,0 +1,721 @@
	1	use super::super::{GONE, OPENING, Pipe, Shared, State};
	2	use super::*;
	3	use std::{
	4	io::{Read, Write},
	5	net::{Shutdown, TcpListener, TcpStream},
	6	};
	7
	8	struct Barrier {
	9	reached: mpsc::Sender<()>,
	10	resume: mpsc::Receiver<()>,
	11	}
	12
	13	impl Barrier {
	14	fn wait(self) -> io::Result<()> {
	15	self.reached.send(()).map_err(io::Error::other)?;
	16	self.resume.recv_timeout(GONE).map_err(io::Error::other)
	17	}
	18	}
	19
	20	fn barrier() -> (Barrier, mpsc::Receiver<()>, mpsc::Sender<()>) {
	21	let (reached, waiting) = mpsc::channel();
	22	let (resume, continued) = mpsc::channel();
	23	(
	24	Barrier {
	25	reached,
	26	resume: continued,
	27	},
	28	waiting,
	29	resume,
	30	)
	31	}
	32
	33	struct HandoverPipe {
	34	stream: TcpStream,
	35	direct: bool,
	36	body: Mutex<Option<Barrier>>,
	37	retired: mpsc::Sender<()>,
	38	}
	39
	40	impl Pipe for HandoverPipe {
	41	fn read(&self, buffer: &mut [u8]) -> io::Result<usize> {
	42	let gate = if buffer.len() > 4 {
	43	self.body.lock().unwrap().take()
	44	} else {
	45	None
	46	};
	47	if let Some(gate) = gate {
	48	// Preserve a complete copied ciphertext block across the old pipe's shutdown.
	49	Read::read_exact(&mut &self.stream, buffer)?;
	50	gate.wait()?;
	51	Ok(buffer.len())
	52	} else {
	53	Read::read(&mut &self.stream, buffer)
	54	}
	55	}
	56
	57	fn write(&self, bytes: &[u8]) -> io::Result<usize> {
	58	Write::write(&mut &self.stream, bytes)
	59	}
	60
	61	fn set_read_timeout(&self, timeout: Duration) -> io::Result<()> {
	62	self.stream.set_read_timeout(Some(timeout))
	63	}
	64
	65	fn shutdown(&self) {
	66	let _ = self.stream.shutdown(Shutdown::Both);
	67	}
	68
	69	fn direct(&self) -> bool {
	70	self.direct
	71	}
	72	}
	73
	74	impl Drop for HandoverPipe {
	75	fn drop(&mut self) {
	76	let _ = self.retired.send(());
	77	}
	78	}
	79
	80	struct Handover {
	81	shared: Arc<Shared>,
	82	guest: Arc<Inner>,
	83	pipes: Vec<Arc<HandoverPipe>>,
	84	readers: Vec<(mpsc::Receiver<()>, thread::JoinHandle<bool>)>,
	85	retired: (mpsc::Sender<()>, mpsc::Receiver<()>),
	86	observed: mpsc::Receiver<()>,
	87	releases: Vec<mpsc::Sender<()>>,
	88	}
	89
	90	impl Handover {
	91	fn new(met: Option<Barrier>) -> Self {
	92	let me = Hello::new("Guest".into(), None).unwrap();
	93	let guest = Arc::new(Inner {
	94	share: [41; 16],
	95	me: me.clone(),
	96	reach: None,
	97	relay: None,
	98	events: Arc::new(\|\| {}),
	99	presence: Mutex::default(),
	100	here: Mutex::default(),
	101	host: Mutex::default(),
	102	pending: Mutex::default(),
	103	next: AtomicU64::new(1),
	104	watch: Mutex::default(),
	105	ended: Mutex::default(),
	106	asked: Default::default(),
	107	images: Mutex::default(),
	108	current: Mutex::default(),
	109	});
	110	let heard = Arc::clone(&guest);
	111	let met = Mutex::new(met);
	112	let (events, observed) = mpsc::channel();
	113	let room = Room::Notebook([42; 16]);
	114	let shared = Arc::new(Shared {
	115	me,
	116	secret: room.secret(),
	117	room,
	118	reach: None,
	119	state: Mutex::new(State::default()),
	120	events: Box::new(move \|source, event\| {
	121	let meeting = matches!(&event, Event::Met(..));
	122	if meeting {
	123	let gate = met.lock().unwrap().take();
	124	if let Some(gate) = gate {
	125	gate.wait().unwrap();
	126	}
	127	}
	128	heard.heard(source, event);
	129	if meeting {
	130	let _ = events.send(());
	131	}
	132	}),
	133	stopped: AtomicBool::new(false),
	134	connections: AtomicU64::new(0),
	135	});
	136	Self {
	137	shared,
	138	guest,
	139	pipes: Vec::new(),
	140	readers: Vec::new(),
	141	retired: mpsc::channel(),
	142	observed,
	143	releases: Vec::new(),
	144	}
	145	}
	146
	147	fn open(&mut self, host: &Hello, direct: bool) -> (TcpStream, wire::Sealer) {
	148	let listener = TcpListener::bind("127.0.0.1:0").unwrap();
	149	let mut remote =
	150	TcpStream::connect_timeout(&listener.local_addr().unwrap(), OPENING).unwrap();
	151	remote.set_read_timeout(Some(OPENING)).unwrap();
	152	remote.set_write_timeout(Some(OPENING)).unwrap();
	153	let (local, _) = listener.accept().unwrap();
	154	local.set_write_timeout(Some(OPENING)).unwrap();
	155	let pipe = Arc::new(HandoverPipe {
	156	stream: local,
	157	direct,
	158	body: Mutex::default(),
	159	retired: self.retired.0.clone(),
	160	});
	161	self.pipes.push(Arc::clone(&pipe));
	162	let shared = Arc::clone(&self.shared);
	163	let tag = shared.room.tag().unwrap();
	164	let reader_tag = tag.clone();
	165	let (finished, done) = mpsc::channel();
	166	let reader = thread::spawn(move \|\| {
	167	let retry = shared.run(pipe, wire::Side::Responder, &reader_tag);
	168	let _ = finished.send(());
	169	retry
	170	});
	171	self.readers.push((done, reader));
	172	let (mut send, mut receive) = wire::open(
	173	&mut remote,
	174	wire::Side::Initiator,
	175	&tag,
	176	&self.shared.secret,
	177	)
	178	.unwrap();
	179	send.send(&mut remote, kind::HELLO, host).unwrap();
	180	let (message, body) = receive.receive(&mut remote).unwrap();
	181	assert_eq!(message, kind::HELLO);
	182	assert_eq!(minicbor::decode::<Hello>(&body).unwrap(), self.shared.me);
	183	(remote, send)
	184	}
	185
	186	fn heard(&self) {
	187	self.observed.recv_timeout(OPENING).unwrap();
	188	}
	189	}
	190
	191	impl Drop for Handover {
	192	fn drop(&mut self) {
	193	self.shared.stopped.store(true, Ordering::Release);
	194	for release in self.releases.drain(..) {
	195	let _ = release.send(());
	196	}
	197	for pipe in &self.pipes {
	198	pipe.shutdown();
	199	}
	200	self.shared.state.lock().unwrap().peers.clear();
	201	let deadline = std::time::Instant::now() + OPENING;
	202	for (_, reader) in &self.readers {
	203	while !reader.is_finished() && std::time::Instant::now() < deadline {
	204	thread::yield_now();
	205	}
	206	}
	207	let readers_done = self.readers.iter().all(\|(_, reader)\| reader.is_finished());
	208	self.guest.host.lock().unwrap().take();
	209	let count = self.pipes.len();
	210	self.pipes.clear();
	211	let retired = (0..count).all(\|_\| self.retired.1.recv_timeout(OPENING).is_ok());
	212	let mut joined = readers_done && retired;
	213	if readers_done {
	214	for (_, reader) in self.readers.drain(..) {
	215	joined &= reader.join().is_ok();
	216	}
	217	}
	218	if !thread::panicking() {
	219	println!("HANDOVER_READERS_JOINED_AND_PIPE_OWNERS_RETIRED={joined};PIPES={count}");
	220	assert!(
	221	joined,
	222	"handover reader/writer owners did not retire and join"
	223	);
	224	}
	225	}
	226	}
	227
	228	#[test]
	229	fn handover_stale_frame_cannot_end_promoted_guest() {
	230	let mut fixture = Handover::new(None);
	231	let mut host = Hello::new("Old host route".into(), None).unwrap();
	232	host.serves = Some(fixture.guest.share);
	233	let (mut old, mut send) = fixture.open(&host, false);
	234	fixture.heard();
	235	let (gate, copied, resume) = barrier();
	236	fixture.releases.push(resume.clone());
	237	*fixture.pipes[0].body.lock().unwrap() = Some(gate);
	238	send.send(
	239	&mut old,
	240	kind::BYE,
	241	&wire::Bye {
	242	reason: STOPPED.into(),
	243	},
	244	)
	245	.unwrap();
	246	copied.recv_timeout(OPENING).unwrap();
	247	host.name = "Promoted host route".into();
	248	let (_new, _) = fixture.open(&host, true);
	249	fixture.heard();
	250	assert!(
	251	fixture.shared.state.lock().unwrap().peers[&host.peer]
	252	.pipe
	253	.direct()
	254	);
	255	assert!(fixture.guest.ended.lock().unwrap().is_none());
	256	resume.send(()).unwrap();
	257	fixture.readers[0].0.recv_timeout(OPENING).unwrap();
	258	let ended = *fixture.guest.ended.lock().unwrap();
	259	drop(fixture);
	260	println!("HANDOVER_STALE_FRAME_ENDED={ended:?}");
	261	assert_eq!(ended, None, "stale frame ended the promoted guest");
	262	}
	263
	264	#[test]
	265	fn handover_stale_met_cannot_replace_promoted_guest_line() {
	266	let (gate, delayed, resume) = barrier();
	267	let mut fixture = Handover::new(Some(gate));
	268	fixture.releases.push(resume.clone());
	269	let mut host = Hello::new("Old host route".into(), None).unwrap();
	270	host.serves = Some(fixture.guest.share);
	271	let (_old, _) = fixture.open(&host, false);
	272	delayed.recv_timeout(OPENING).unwrap();
	273	host.name = "Promoted host route".into();
	274	let (_new, _) = fixture.open(&host, true);
	275	let overlap = match fixture.observed.recv_timeout(OPENING) {
	276	Ok(()) => {
	277	assert_eq!(
	278	fixture.guest.host.lock().unwrap().as_ref().unwrap().0.name,
	279	host.name
	280	);
	281	resume.send(()).unwrap();
	282	fixture.heard();
	283	true
	284	}
	285	Err(mpsc::RecvTimeoutError::Timeout) => {
	286	resume.send(()).unwrap();
	287	fixture.heard();
	288	fixture.heard();
	289	false
	290	}
	291	Err(error) => panic!("handover event channel closed: {error}"),
	292	};
	293	assert_eq!(
	294	fixture.shared.state.lock().unwrap().peers[&host.peer]
	295	.peer
	296	.hello
	297	.name,
	298	host.name
	299	);
	300	let selected = fixture
	301	.guest
	302	.host
	303	.lock()
	304	.unwrap()
	305	.as_ref()
	306	.unwrap()
	307	.0
	308	.name
	309	.clone();
	310	drop(fixture);
	311	println!("HANDOVER_MET_PROMOTION_OVERLAP={overlap}");
	312	assert_eq!(
	313	selected, host.name,
	314	"stale Met replaced the promoted guest Line"
	315	);
	316	}
	317
	318	impl Handover {
	319	fn intercept(
	320	&mut self,
	321	before: impl Fn(&Arc<Shared>, &Event<'_>) + Send + Sync + 'static,
	322	) -> mpsc::Receiver<u16> {
	323	let guest = Arc::clone(&self.guest);
	324	let (observed, waiting) = mpsc::channel();
	325	Arc::get_mut(&mut self.shared).unwrap().events = Box::new(move \|source, event\| {
	326	let kind = match &event {
	327	Event::Met(..) => kind::HELLO,
	328	Event::Left(..) => kind::BYE,
	329	Event::Frame { kind, .. } => *kind,
	330	Event::Changed => 0,
	331	};
	332	before(source, &event);
	333	guest.heard(source, event);
	334	let _ = observed.send(kind);
	335	});
	336	waiting
	337	}
	338	}
	339
	340	fn observed(events: &mpsc::Receiver<u16>, wanted: u16) {
	341	let end = std::time::Instant::now() + OPENING;
	342	loop {
	343	let kind = events
	344	.recv_timeout(end.saturating_duration_since(std::time::Instant::now()))
	345	.unwrap();
	346	if kind == wanted {
	347	return;
	348	}
	349	}
	350	}
	351
	352	#[test]
	353	fn handover_old_left_preserves_promoted_host_and_pending_reply() {
	354	let mut fixture = Handover::new(None);
	355	let (gate, removed, resume) = barrier();
	356	fixture.releases.push(resume.clone());
	357	let gate = Mutex::new(Some(gate));
	358	let events = fixture.intercept(move \|_, event\| {
	359	if matches!(event, Event::Left(_))
	360	&& let Some(gate) = gate.lock().unwrap().take()
	361	{
	362	gate.wait().unwrap();
	363	}
	364	});
	365	let mut host = Hello::new("Old host route".into(), None).unwrap();
	366	host.serves = Some(fixture.guest.share);
	367	let (old, _) = fixture.open(&host, false);
	368	observed(&events, kind::HELLO);
	369	old.shutdown(Shutdown::Both).unwrap();
	370	removed.recv_timeout(OPENING).unwrap();
	371	host.name = "Promoted host route".into();
	372	let (_new, _) = fixture.open(&host, true);
	373	observed(&events, kind::HELLO);
	374	let (answer, waiting) = crate::task::response::<Reply>();
	375	fixture.guest.pending.lock().unwrap().insert(73, answer);
	376	resume.send(()).unwrap();
	377	fixture.readers[0].0.recv_timeout(OPENING).unwrap();
	378	let host_kept = fixture
	379	.guest
	380	.host
	381	.lock()
	382	.unwrap()
	383	.as_ref()
	384	.is_some_and(\|(hello, _)\| hello.name == host.name);
	385	let pending_kept = fixture.guest.pending.lock().unwrap().contains_key(&73);
	386	let answer_kept = matches!(waiting.try_recv(), Err(mpsc::TryRecvError::Empty));
	387	drop(fixture);
	388	assert!(
	389	host_kept && pending_kept && answer_kept,
	390	"retired Left cleared the promoted host or pending reply"
	391	);
	392	}
	393
	394	#[test]
	395	fn handover_selected_loss_clears_host_and_disconnects_pending_reply() {
	396	let mut fixture = Handover::new(None);
	397	let mut host = Hello::new("Selected host".into(), None).unwrap();
	398	host.serves = Some(fixture.guest.share);
	399	let (remote, _) = fixture.open(&host, true);
	400	fixture.heard();
	401	let (answer, waiting) = crate::task::response::<Reply>();
	402	fixture.guest.pending.lock().unwrap().insert(74, answer);
	403	remote.shutdown(Shutdown::Both).unwrap();
	404	fixture.readers[0].0.recv_timeout(OPENING).unwrap();
	405	let host_cleared = fixture.guest.host.lock().unwrap().is_none();
	406	let pending_cleared = fixture.guest.pending.lock().unwrap().is_empty();
	407	let answer_lost = matches!(waiting.try_recv(), Err(mpsc::TryRecvError::Disconnected));
	408	drop(fixture);
	409	assert!(
	410	host_cleared && pending_cleared && answer_lost,
	411	"selected route loss retained the host or pending reply"
	412	);
	413	}
	414
	415	fn served() -> (tempfile::TempDir, Arc<Served>) {
	416	let directory = tempfile::tempdir().unwrap();
	417	let folder = directory.path().join("notebook");
	418	std::fs::create_dir(&folder).unwrap();
	419	let served = Arc::new(Served {
	420	storage: crate::session::Notebook::open(&folder, directory.path().join("cache"))
	421	.unwrap()
	422	.into_storage(),
	423	images: Mutex::default(),
	424	snapshots: Mutex::default(),
	425	puts: Mutex::default(),
	426	guests: Mutex::default(),
	427	writers: Mutex::default(),
	428	host: Mutex::default(),
	429	room: Mutex::default(),
	430	});
	431	(directory, served)
	432	}
	433
	434	#[test]
	435	fn handover_copied_host_request_cannot_enter_promoted_queue() {
	436	let (_directory, served) = served();
	437	let queues = Arc::new(Mutex::new(Vec::new()));
	438	let mut fixture = Handover::new(None);
	439	let serving = Arc::clone(&served);
	440	let received = Arc::clone(&queues);
	441	let events = fixture.intercept(move \|source, event\| match event {
	442	Event::Met(hello, line) => {
	443	let (queue, waiting) = mpsc::sync_channel(4);
	444	let retired = serving.guests.lock().unwrap().insert(
	445	hello.peer,
	446	Admitted {
	447	hello: Arc::clone(hello),
	448	line: (*line).clone(),
	449	queue,
	450	starts: BURST,
	451	counted: Instant::now(),
	452	held: Vec::new(),
	453	},
	454	);
	455	drop(retired);
	456	received.lock().unwrap().push(waiting);
	457	}
	458	Event::Frame { from, kind, body } => serving.queue(source, from, *kind, body),
	459	Event::Left(hello) => serving.forget(&hello.peer, Some((source, hello))),
	460	_ => {}
	461	});
	462	let mut host = Hello::new("Old route".into(), None).unwrap();
	463	host.serves = Some(fixture.guest.share);
	464	let (mut old, mut send) = fixture.open(&host, false);
	465	observed(&events, kind::HELLO);
	466	let (gate, copied, resume) = barrier();
	467	fixture.releases.push(resume.clone());
	468	*fixture.pipes[0].body.lock().unwrap() = Some(gate);
	469	send.send(
	470	&mut old,
	471	kind::PUT,
	472	&Request {
	473	id: 80,
	474	bytes: Some(vec![1, 2, 3]),
	475	..Request::default()
	476	},
	477	)
	478	.unwrap();
	479	copied.recv_timeout(OPENING).unwrap();
	480	host.name = "Promoted route".into();
	481	let (mut new, mut send) = fixture.open(&host, true);
	482	observed(&events, kind::HELLO);
	483	resume.send(()).unwrap();
	484	fixture.readers[0].0.recv_timeout(OPENING).unwrap();
	485	let no_stale_request = matches!(
	486	queues.lock().unwrap()[1].try_recv(),
	487	Err(mpsc::TryRecvError::Empty)
	488	);
	489	send.send(
	490	&mut new,
	491	kind::PUT,
	492	&Request {
	493	id: 81,
	494	bytes: Some(vec![4, 5, 6]),
	495	..Request::default()
	496	},
	497	)
	498	.unwrap();
	499	let (kind, body) = queues.lock().unwrap()[1].recv_timeout(OPENING).unwrap();
	500	let accepted = kind == kind::PUT && minicbor::decode::<Request>(&body).unwrap().id == 81;
	501	served.guests.lock().unwrap().clear();
	502	drop(fixture);
	503	assert!(
	504	no_stale_request && accepted,
	505	"retired request entered the promoted queue or current request was lost"
	506	);
	507	}
	508
	509	#[test]
	510	fn handover_old_left_preserves_promoted_host_uploads_and_snapshots() {
	511	let (_directory, served) = served();
	512	let mut fixture = Handover::new(None);
	513	let (gate, removed, resume) = barrier();
	514	fixture.releases.push(resume.clone());
	515	let gate = Mutex::new(Some(gate));
	516	let serving = Arc::clone(&served);
	517	let waiting = Arc::new(Mutex::new(Vec::new()));
	518	let keep_queues = Arc::clone(&waiting);
	519	let events = fixture.intercept(move \|source, event\| match event {
	520	Event::Met(hello, line) => {
	521	let (queue, received) = mpsc::sync_channel(1);
	522	let retired = serving.guests.lock().unwrap().insert(
	523	hello.peer,
	524	Admitted {
	525	hello: Arc::clone(hello),
	526	line: (*line).clone(),
	527	queue,
	528	starts: BURST,
	529	counted: Instant::now(),
	530	held: Vec::new(),
	531	},
	532	);
	533	drop(retired);
	534	keep_queues.lock().unwrap().push(received);
	535	}
	536	Event::Left(hello) => {
	537	if let Some(gate) = gate.lock().unwrap().take() {
	538	gate.wait().unwrap();
	539	}
	540	serving.forget(&hello.peer, Some((source, hello)));
	541	}
	542	_ => {}
	543	});
	544	let mut host = Hello::new("Old route".into(), None).unwrap();
	545	host.serves = Some(fixture.guest.share);
	546	let (old, _) = fixture.open(&host, false);
	547	observed(&events, kind::HELLO);
	548	old.shutdown(Shutdown::Both).unwrap();
	549	removed.recv_timeout(OPENING).unwrap();
	550	host.name = "Promoted route".into();
	551	let (_new, _) = fixture.open(&host, true);
	552	observed(&events, kind::HELLO);
	553	served
	554	.puts
	555	.lock()
	556	.unwrap()
	557	.insert((host.peer, 90), vec![7, 8, 9]);
	558	served
	559	.snapshots
	560	.lock()
	561	.unwrap()
	562	.insert((host.peer, 91), (Arc::new(vec![10, 11]), Instant::now()));
	563	resume.send(()).unwrap();
	564	fixture.readers[0].0.recv_timeout(OPENING).unwrap();
	565	let guest_kept = served
	566	.guests
	567	.lock()
	568	.unwrap()
	569	.get(&host.peer)
	570	.is_some_and(\|guest\| guest.hello.name == host.name);
	571	let upload_kept = served.puts.lock().unwrap().get(&(host.peer, 90)) == Some(&vec![7, 8, 9]);
	572	let snapshot_kept = served
	573	.snapshots
	574	.lock()
	575	.unwrap()
	576	.get(&(host.peer, 91))
	577	.is_some_and(\|(bytes, _)\| **bytes == vec![10, 11]);
	578	served.guests.lock().unwrap().clear();
	579	drop(fixture);
	580	assert!(
	581	guest_kept && upload_kept && snapshot_kept,
	582	"retired Left removed promoted host resources"
	583	);
	584	}
	585
	586	#[test]
	587	fn handover_deferred_presence_cannot_replace_promoted_host_room() {
	588	let mut fixture = Handover::new(None);
	589	let mut host = Hello::new("Old host route".into(), None).unwrap();
	590	host.serves = Some(fixture.guest.share);
	591	let (_old, _) = fixture.open(&host, false);
	592	fixture.heard();
	593	let from = Arc::clone(&fixture.guest.host.lock().unwrap().as_ref().unwrap().0);
	594	let secret = [61; 16];
	595	let live = Live::start(
	596	fixture.guest.me.clone(),
	597	&Room::Notebook(secret),
	598	None,
	599	None,
	600	\|_\| {},
	601	)
	602	.unwrap();
	603	let guest = Arc::clone(&fixture.guest);
	604	let source = Arc::clone(&fixture.shared);
	605	let here_owner = Arc::clone(&fixture.guest);
	606	let here = here_owner.here.lock().unwrap();
	607	let (entered, started) = mpsc::channel();
	608	let (finished, done) = mpsc::channel();
	609	let installer = thread::spawn(move \|\| {
	610	entered.send(()).unwrap();
	611	guest.install_presence(&source, &from, secret, live);
	612	let _ = finished.send(());
	613	true
	614	});
	615	fixture.readers.push((done, installer));
	616	started.recv_timeout(OPENING).unwrap();
	617	host.name = "Promoted host route".into();
	618	let (_new, _) = fixture.open(&host, true);
	619	fixture.heard();
	620	drop(here);
	621	fixture.readers[1].0.recv_timeout(OPENING).unwrap();
	622	let refused = fixture.guest.presence.lock().unwrap().is_none();
	623	let promoted = fixture
	624	.guest
	625	.host
	626	.lock()
	627	.unwrap()
	628	.as_ref()
	629	.is_some_and(\|(hello, _)\| hello.name == host.name);
	630	drop(fixture);
	631	assert!(
	632	refused && promoted,
	633	"deferred old WELCOME installed its presence room"
	634	);
	635	}
	636
	637	#[test]
	638	fn handover_presence_install_seeds_latest_here() {
	639	let mut fixture = Handover::new(None);
	640	let mut host = Hello::new("Current host".into(), None).unwrap();
	641	host.serves = Some(fixture.guest.share);
	642	let (_remote, _) = fixture.open(&host, true);
	643	fixture.heard();
	644	let from = Arc::clone(&fixture.guest.host.lock().unwrap().as_ref().unwrap().0);
	645	let secret = [62; 16];
	646	let live = Live::start(
	647	fixture.guest.me.clone(),
	648	&Room::Notebook(secret),
	649	None,
	650	None,
	651	\|_\| {},
	652	)
	653	.unwrap();
	654	let guest = Arc::clone(&fixture.guest);
	655	let source = Arc::clone(&fixture.shared);
	656	let mut here = fixture.guest.here.lock().unwrap();
	657	let latest = Presence {
	658	section: Some([63; 16]),
	659	..Presence::default()
	660	};
	661	let (entered, started) = mpsc::channel();
	662	let (finished, done) = mpsc::channel();
	663	let installer = thread::spawn(move \|\| {
	664	entered.send(()).unwrap();
	665	guest.install_presence(&source, &from, secret, live);
	666	let _ = finished.send(());
	667	true
	668	});
	669	fixture.readers.push((done, installer));
	670	started.recv_timeout(OPENING).unwrap();
	671	*here = latest.clone();
	672	drop(here);
	673	fixture.readers[1].0.recv_timeout(OPENING).unwrap();
	674	let seeded = fixture
	675	.guest
	676	.presence
	677	.lock()
	678	.unwrap()
	679	.as_ref()
	680	.is_some_and(\|(held, live)\| {
	681	*held == secret && live.shared.state.lock().unwrap().presence == latest
	682	});
	683	let installed = fixture.guest.presence.lock().unwrap().take();
	684	drop(installed);
	685	drop(fixture);
	686	assert!(
	687	seeded,
	688	"presence installation overwrote current here with a stale seed"
	689	);
	690	}
	691
	692	pub(super) fn promoted_routes<T>(
	693	room: Room,
	694	before: impl FnOnce(&Arc<Shared>, &Arc<Hello>, &Line),
	695	check: impl FnOnce(&Arc<Shared>, &Arc<Hello>, &Line, &Arc<Hello>, &Line) -> T,
	696	) -> T {
	697	let mut fixture = Handover::new(None);
	698	let shared = Arc::get_mut(&mut fixture.shared).unwrap();
	699	shared.secret = room.secret();
	700	shared.room = room;
	701	let host = Hello::new("Pairing device".into(), None).unwrap();
	702	let (_old, _) = fixture.open(&host, false);
	703	fixture.heard();
	704	let (old, old_line) = {
	705	let state = fixture.shared.state.lock().unwrap();
	706	let link = &state.peers[&host.peer];
	707	(Arc::clone(&link.peer.hello), link.line.clone())
	708	};
	709	before(&fixture.shared, &old, &old_line);
	710	let (_new, _) = fixture.open(&host, true);
	711	fixture.heard();
	712	let (new, new_line) = {
	713	let state = fixture.shared.state.lock().unwrap();
	714	let link = &state.peers[&host.peer];
	715	(Arc::clone(&link.peer.hello), link.line.clone())
	716	};
	717	let result = check(&fixture.shared, &old, &old_line, &new, &new_line);
	718	drop((old_line, new_line));
	719	drop(fixture);
	720	result
	721	}

crates/notebook/src/live/share/membership.rs+265-37

...	...	@@ -13,7 +13,7 @@ struct Members {
13	13	relay: Option<String>,
14	14	sharing: Mutex<Sharing>,
15	15	changes: Mutex<()>,
16		pending: Mutex<BTreeMap<[u8; 16], (Arc<Hello>, Line)>>,
	16	pending: Mutex<BTreeMap<[u8; 16], (Arc<Hello>, Line, Weak<super::super::Shared>)>>,
17	17	access: Mutex<HashMap<[u8; 16], Live>>,
18	18	room: Mutex<Option<Live>>,
19	19	served: Arc<Served>,
...	...	@@ -134,7 +134,7 @@ impl Host {
134	134	.lock()
135	135	.unwrap()
136	136	.values()
137		.map(\|(hello, _)\| Arc::clone(hello))
	137	.map(\|(hello, _, _)\| Arc::clone(hello))
138	138	.collect()
139	139	}
140	140
...	...	@@ -155,14 +155,27 @@ impl Host {
155	155	}
156	156
157	157	pub fn allow(&self, peer: &[u8; 16]) -> io::Result<()> {
158		let pending = self.members.pending.lock().unwrap().remove(peer);
159		if let Some((hello, line)) = pending {
	158	let request = {
	159	let mut pending = self.members.pending.lock().unwrap();
	160	if !pending.get(peer).is_some_and(\|(hello, _, source)\| {
	161	source
	162	.upgrade()
	163	.is_some_and(\|source\| source.connected(hello))
	164	}) {
	165	return Ok(());
	166	}
	167	pending.remove(peer)
	168	};
	169	if let Some((hello, line, source)) = request {
160	170	if let Err(error) = self.members.grant(&hello, &line) {
161		self.members
162		.pending
163		.lock()
164		.unwrap()
165		.insert(*peer, (hello, line));
	171	let mut pending = self.members.pending.lock().unwrap();
	172	if source
	173	.upgrade()
	174	.is_some_and(\|source\| source.connected(&hello))
	175	&& !pending.contains_key(peer)
	176	{
	177	pending.insert(*peer, (hello, line, source));
	178	}
166	179	return Err(error);
167	180	}
168	181	(self.members.events)();
...	...	@@ -171,7 +184,8 @@ impl Host {
171	184	}
172	185
173	186	pub fn decline(&self, peer: &[u8; 16]) {
174		if let Some((_, line)) = self.members.pending.lock().unwrap().remove(peer) {
	187	let request = self.members.pending.lock().unwrap().remove(peer);
	188	if let Some((_, line, _)) = request {
175	189	let _ = line.send(kind::APPROVAL, &wire::Approval::Declined);
176	190	(self.members.events)();
177	191	}
...	...	@@ -198,20 +212,35 @@ impl Host {
198	212	let removed = self.members.access.lock().unwrap().remove(secret);
199	213	if let Some(removed) = removed {
200	214	for peer in removed.peers() {
201		self.members.served.forget(&peer.hello.peer);
	215	self.members.served.forget(&peer.hello.peer, None);
202	216	}
203	217	removed.leave("removed");
204	218	}
205	219	self.members.presence(next.secret)?;
206	220	*self.pairing.lock().unwrap() = Some(self.members.pair(&next)?);
207		let access = self.members.access.lock().unwrap();
208		for device in &next.members {
209		if let Some(live) = access.get(&device.secret) {
210		live.sender().send(
211		kind::WELCOME,
212		&self.members.welcome(&next, device.secret),
213		None,
214		);
	221	let welcomes = {
	222	let access = self.members.access.lock().unwrap();
	223	next.members
	224	.iter()
	225	.filter_map(\|device\| {
	226	let live = access.get(&device.secret)?;
	227	let lines = live
	228	.peers()
	229	.into_iter()
	230	.filter_map(\|peer\| live.line(&peer.hello.peer))
	231	.collect::<Vec<_>>();
	232	Some((self.members.welcome(&next, device.secret), lines))
	233	})
	234	.collect::<Vec<_>>()
	235	};
	236	#[cfg(target_arch = "wasm32")]
	237	drop(_change);
	238	'welcomes: for (welcome, lines) in welcomes {
	239	for line in lines {
	240	if self.members.sharing.lock().unwrap().secret != welcome.room {
	241	break 'welcomes;
	242	}
	243	let _ = line.send(kind::WELCOME, &welcome);
215	244	}
216	245	}
217	246	(self.members.events)();
...	...	@@ -274,7 +303,7 @@ impl Members {
274	303
275	304	fn open(self: &Arc<Self>, secret: [u8; 16]) -> io::Result<()> {
276	305	let members = Arc::downgrade(self);
277		let live = Live::start(
	306	let live = Live::start_with(
278	307	Hello {
279	308	serves: Some(self.sharing.lock().unwrap().share),
280	309	..self.me.clone()
...	...	@@ -282,25 +311,42 @@ impl Members {
282	311	&Room::Notebook(secret),
283	312	self.reach,
284	313	self.relay.as_deref(),
285		move \|event\| {
	314	move \|source, event\| {
286	315	let Some(members) = members.upgrade() else {
287	316	return;
288	317	};
289	318	match event {
290	319	Event::Met(hello, line) => {
291	320	let sharing = members.sharing.lock().unwrap();
292		if !sharing.members.iter().any(\|device\| device.secret == secret) {
293		line.hang_up("removed");
294		return;
	321	let admitted =
	322	if sharing.members.iter().any(\|device\| device.secret == secret) {
	323	Ok(members
	324	.served
	325	.admit(source, hello, line)
	326	.then(\|\| members.welcome(&sharing, secret)))
	327	} else {
	328	Err(())
	329	};
	330	// Browser send failures may reenter membership callbacks.
	331	#[cfg(target_arch = "wasm32")]
	332	drop(sharing);
	333	match admitted {
	334	Ok(Some(welcome)) => {
	335	let _ = line.send(kind::WELCOME, &welcome);
	336	}
	337	Err(()) => {
	338	#[cfg(not(target_arch = "wasm32"))]
	339	drop(sharing);
	340	line.hang_up("removed");
	341	}
	342	Ok(None) => {}
295	343	}
296		members.served.admit(hello.peer, line);
297		let _ = line.send(kind::WELCOME, &members.welcome(&sharing, secret));
298	344	}
299		Event::Left(hello) => members.served.forget(&hello.peer),
	345	Event::Left(hello) => members.served.forget(&hello.peer, Some((source, hello))),
300	346	Event::Frame { from, kind, body }
301	347	if wire::KNOWN.contains(&kind) && kind > 256 && kind != kind::REPLY =>
302	348	{
303		members.served.queue(&from.peer, kind, body)
	349	members.served.queue(source, from, kind, body)
304	350	}
305	351	Event::Changed => (members.events)(),
306	352	_ => {}
...	...	@@ -349,7 +395,7 @@ impl Members {
349	395
350	396	fn pair(self: &Arc<Self>, sharing: &Sharing) -> io::Result<Live> {
351	397	let members = Arc::downgrade(self);
352		Live::start(
	398	Live::start_with(
353	399	Hello {
354	400	serves: None,
355	401	..self.me.clone()
...	...	@@ -357,21 +403,41 @@ impl Members {
357	403	&Room::share(&sharing.code, &sharing.password),
358	404	self.reach,
359	405	self.relay.as_deref(),
360		move \|event\| {
	406	move \|source, event\| {
361	407	let Some(members) = members.upgrade() else {
362	408	return;
363	409	};
364	410	match event {
365	411	Event::Met(hello, line) => {
366		if members.sharing.lock().unwrap().approve {
	412	let approve = members.sharing.lock().unwrap().approve;
	413	let (approval, retired) = {
367	414	let mut pending = members.pending.lock().unwrap();
368		if pending.len() < 32 {
369		pending.insert(hello.peer, (Arc::clone(hello), line.clone()));
370		let _ = line.send(kind::APPROVAL, &wire::Approval::Pending);
	415	if !source.connected(hello) {
	416	return;
	417	}
	418	if approve {
	419	if pending.len() < 32 \|\| pending.contains_key(&hello.peer) {
	420	(
	421	Some(wire::Approval::Pending),
	422	pending.insert(
	423	hello.peer,
	424	(
	425	Arc::clone(hello),
	426	line.clone(),
	427	Arc::downgrade(source),
	428	),
	429	),
	430	)
	431	} else {
	432	(Some(wire::Approval::Failed), None)
	433	}
371	434	} else {
372		let _ = line.send(kind::APPROVAL, &wire::Approval::Failed);
	435	(None, None)
373	436	}
374		drop(pending);
	437	};
	438	drop(retired);
	439	if let Some(approval) = approval {
	440	let _ = line.send(kind::APPROVAL, &approval);
375	441	(members.events)();
376	442	} else if let Err(error) = members.grant(hello, line) {
377	443	eprintln!("Live Share: could not admit a device: {error}");
...	...	@@ -379,7 +445,19 @@ impl Members {
379	445	}
380	446	}
381	447	Event::Left(hello) => {
382		members.pending.lock().unwrap().remove(&hello.peer);
	448	let retired = {
	449	let mut pending = members.pending.lock().unwrap();
	450	if source.gone(hello)
	451	&& pending
	452	.get(&hello.peer)
	453	.is_some_and(\|(admitted, _, _)\| Arc::ptr_eq(admitted, hello))
	454	{
	455	pending.remove(&hello.peer)
	456	} else {
	457	None
	458	}
	459	};
	460	drop(retired);
383	461	(members.events)();
384	462	}
385	463	Event::Changed => (members.events)(),
...	...	@@ -389,3 +467,153 @@ impl Members {
389	467	)
390	468	}
391	469	}
	470
	471	#[cfg(all(test, not(target_arch = "wasm32")))]
	472	mod handover_tests {
	473	use super::*;
	474	use std::sync::atomic::AtomicUsize;
	475
	476	#[test]
	477	fn handover_pair_pending_keeps_promoted_identity_and_rejects_retired_allow() {
	478	let directory = tempfile::tempdir().unwrap();
	479	let folder = directory.path().join("notebook");
	480	std::fs::create_dir(&folder).unwrap();
	481	let sharing = Sharing {
	482	approve: true,
	483	code: super::super::super::code::format(701, "ABCDEF").unwrap(),
	484	..Sharing::new("").unwrap()
	485	};
	486	let room = Room::share(&sharing.code, &sharing.password);
	487	let host = Host::start(
	488	crate::session::Notebook::open(&folder, directory.path().join("cache"))
	489	.unwrap()
	490	.into_storage(),
	491	Hello::new("Notebook owner".into(), None).unwrap(),
	492	sharing,
	493	"Fixture",
	494	None,
	495	None,
	496	\|\| {},
	497	\|_\| Ok(()),
	498	)
	499	.unwrap();
	500	let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared);
	501	let valid = super::super::handover_tests::promoted_routes(
	502	room,
	503	\|source, old, old_line\| {
	504	(callback.events)(source, Event::Met(old, old_line));
	505	},
	506	\|source, old, old_line, new, new_line\| {
	507	host.allow(&old.peer).unwrap();
	508	let no_grant = host.sharing().members.is_empty();
	509	(callback.events)(source, Event::Met(new, new_line));
	510	(callback.events)(source, Event::Met(old, old_line));
	511	(callback.events)(source, Event::Left(old));
	512	let kept = host
	513	.requests()
	514	.first()
	515	.is_some_and(\|request\| Arc::ptr_eq(request, new));
	516	host.stop();
	517	kept && no_grant
	518	},
	519	);
	520	assert!(
	521	valid,
	522	"pairing admitted a retired route or retired Left removed the promoted request"
	523	);
	524	}
	525
	526	#[test]
	527	fn handover_pair_retired_met_cannot_persist_another_credential() {
	528	let directory = tempfile::tempdir().unwrap();
	529	let folder = directory.path().join("notebook");
	530	std::fs::create_dir(&folder).unwrap();
	531	let sharing = Sharing {
	532	approve: false,
	533	code: super::super::super::code::format(702, "ABCDEF").unwrap(),
	534	..Sharing::new("").unwrap()
	535	};
	536	let room = Room::share(&sharing.code, &sharing.password);
	537	let kept = Arc::new(AtomicUsize::new(0));
	538	let count = Arc::clone(&kept);
	539	let host = Host::start(
	540	crate::session::Notebook::open(&folder, directory.path().join("cache"))
	541	.unwrap()
	542	.into_storage(),
	543	Hello::new("Notebook owner".into(), None).unwrap(),
	544	sharing,
	545	"Fixture",
	546	None,
	547	None,
	548	\|\| {},
	549	move \|_\| {
	550	count.fetch_add(1, Ordering::SeqCst);
	551	Ok(())
	552	},
	553	)
	554	.unwrap();
	555	let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared);
	556	let valid = super::super::handover_tests::promoted_routes(
	557	room,
	558	\|_, _, _\| {},
	559	\|source, old, old_line, new, new_line\| {
	560	(callback.events)(source, Event::Met(new, new_line));
	561	let after_current = kept.load(Ordering::SeqCst);
	562	(callback.events)(source, Event::Met(old, old_line));
	563	let refused = kept.load(Ordering::SeqCst) == after_current
	564	&& after_current == 2
	565	&& host.sharing().members.len() == 1;
	566	host.stop();
	567	refused
	568	},
	569	);
	570	assert!(valid, "retired pairing Met persisted another credential");
	571	}
	572
	573	#[test]
	574	fn handover_retired_membership_rejects_deferred_access_met() {
	575	let directory = tempfile::tempdir().unwrap();
	576	let folder = directory.path().join("notebook");
	577	std::fs::create_dir(&folder).unwrap();
	578	let secret = [71; 16];
	579	let sharing = Sharing {
	580	code: super::super::super::code::format(703, "ABCDEF").unwrap(),
	581	members: vec![Device {
	582	secret,
	583	name: "Retired device".into(),
	584	device: None,
	585	}],
	586	..Sharing::new("").unwrap()
	587	};
	588	let host = Host::start(
	589	crate::session::Notebook::open(&folder, directory.path().join("cache"))
	590	.unwrap()
	591	.into_storage(),
	592	Hello::new("Notebook owner".into(), None).unwrap(),
	593	sharing,
	594	"Fixture",
	595	None,
	596	None,
	597	\|\| {},
	598	\|_\| Ok(()),
	599	)
	600	.unwrap();
	601	let callback = Arc::clone(&host.members.access.lock().unwrap()[&secret].shared);
	602	let refused = super::super::handover_tests::promoted_routes(
	603	Room::Notebook(secret),
	604	\|_, _, _\| {},
	605	\|source, _, _, current, line\| {
	606	host.remove(&secret).unwrap();
	607	(callback.events)(source, Event::Met(current, line));
	608	let refused = host.members.served.guests.lock().unwrap().is_empty()
	609	&& host.sharing().members.is_empty();
	610	host.stop();
	611	refused
	612	},
	613	);
	614	assert!(
	615	refused,
	616	"deferred access callback readmitted a removed device"
	617	);
	618	}
	619	}

crates/notebook/src/live/web.rs+80-35

...	...	@@ -42,7 +42,8 @@ struct Shared {
42	42	me: Hello,
43	43	room: Room,
44	44	url: String,
45		events: Box<dyn Fn(Event) + Send + Sync>,
	45	#[allow(clippy::type_complexity)]
	46	events: Box<dyn Fn(&Arc<Shared>, Event) + Send + Sync>,
46	47	stopped: AtomicBool,
47	48	state: Mutex<State>,
48	49	}
...	...	@@ -139,9 +140,19 @@ impl Live {
139	140	pub fn start(
140	141	me: Hello,
141	142	room: &Room,
142		_: Option<Reach>,
	143	reach: Option<Reach>,
143	144	relay: Option<&str>,
144	145	events: impl Fn(Event) + Send + Sync + 'static,
	146	) -> io::Result<Self> {
	147	Self::start_with(me, room, reach, relay, move \|_, event\| events(event))
	148	}
	149
	150	fn start_with(
	151	me: Hello,
	152	room: &Room,
	153	_: Option<Reach>,
	154	relay: Option<&str>,
	155	events: impl Fn(&Arc<Shared>, Event) + Send + Sync + 'static,
145	156	) -> io::Result<Self> {
146	157	let tag = room.tag().ok_or(io::ErrorKind::InvalidInput)?;
147	158	let relay = relay
...	...	@@ -223,17 +234,7 @@ impl Live {
223	234	Sender(Arc::downgrade(&self.0))
224	235	}
225	236	pub fn peers(&self) -> Vec<Peer> {
226		let state = self.0.state.lock().unwrap();
227		let mut peers = BTreeMap::new();
228		for peer in state
229		.members
230		.values()
231		.filter_map(\|m\| m.peer.as_ref())
232		.chain(state.streams.values().filter_map(\|s\| s.peer.as_ref()))
233		{
234		peers.insert(peer.hello.peer, peer.clone());
235		}
236		peers.into_values().collect()
	237	self.0.peers()
237	238	}
238	239	pub fn line(&self, peer: &[u8; 16]) -> Option<Line> {
239	240	self.0
...	...	@@ -288,11 +289,45 @@ impl Live {
288	289	impl Drop for Live {
289	290	fn drop(&mut self) {
290	291	self.0.stopped.store(true, Ordering::Release);
291		close(self.0.state.lock().unwrap().socket);
	292	let socket = self.0.state.lock().unwrap().socket;
	293	close(socket);
292	294	}
293	295	}
294	296
295	297	impl Shared {
	298	fn peers(&self) -> Vec<Peer> {
	299	let state = self.state.lock().unwrap();
	300	let mut peers = BTreeMap::new();
	301	for peer in state
	302	.members
	303	.values()
	304	.filter_map(\|m\| m.peer.as_ref())
	305	.chain(state.streams.values().filter_map(\|s\| s.peer.as_ref()))
	306	{
	307	peers.insert(peer.hello.peer, peer.clone());
	308	}
	309	peers.into_values().collect()
	310	}
	311
	312	fn gone(&self, hello: &Arc<Hello>) -> bool {
	313	!self.state.lock().unwrap().streams.values().any(\|stream\| {
	314	stream
	315	.peer
	316	.as_ref()
	317	.is_some_and(\|peer\| Arc::ptr_eq(&peer.hello, hello))
	318	})
	319	}
	320
	321	fn connected(&self, hello: &Arc<Hello>) -> bool {
	322	!self.stopped.load(Ordering::Acquire)
	323	&& self.state.lock().unwrap().streams.values().any(\|stream\| {
	324	stream
	325	.peer
	326	.as_ref()
	327	.is_some_and(\|peer\| Arc::ptr_eq(&peer.hello, hello))
	328	})
	329	}
	330
296	331	fn connect(self: &Arc<Self>) -> io::Result<()> {
297	332	let weak = Arc::downgrade(self);
298	333	let message = Closure::<dyn FnMut(JsValue)>::new(move \|data: JsValue\| {
...	...	@@ -324,13 +359,12 @@ impl Shared {
324	359	}
325	360
326	361	fn disconnected(self: &Arc<Self>) {
327		let peers = {
	362	let (socket, peers) = {
328	363	let mut state = self.state.lock().unwrap();
329	364	let socket = std::mem::replace(&mut state.socket, 0);
330	365	if socket == 0 {
331	366	return;
332	367	}
333		close(socket);
334	368	let peers: Vec<_> = state
335	369	.streams
336	370	.values()
...	...	@@ -343,12 +377,13 @@ impl Shared {
343	377	state.members.clear();
344	378	state.sealer = None;
345	379	state.relayed = Relayed::Unreachable(Trouble::Other);
346		peers
	380	(socket, peers)
347	381	};
	382	close(socket);
348	383	for hello in peers {
349		(self.events)(Event::Left(&hello));
	384	(self.events)(self, Event::Left(&hello));
350	385	}
351		(self.events)(Event::Changed);
	386	(self.events)(self, Event::Changed);
352	387	let weak = Arc::downgrade(self);
353	388	let _ = crate::task::spawn("live reconnect", move \|\| async move {
354	389	let (_, wait) = crate::task::channel();
...	...	@@ -464,13 +499,13 @@ impl Shared {
464	499	})
465	500	};
466	501	if let Some(hello) = hello {
467		(self.events)(Event::Left(&hello));
	502	(self.events)(self, Event::Left(&hello));
468	503	}
469	504	}
470	505	::relay::Notice::Burned => self.state.lock().unwrap().burned = true,
471	506	_ => {}
472	507	}
473		(self.events)(Event::Changed);
	508	(self.events)(self, Event::Changed);
474	509	return Ok(());
475	510	}
476	511	let bytes = js_sys::Uint8Array::new(&data).to_vec();
...	...	@@ -523,7 +558,7 @@ impl Shared {
523	558	if serves {
524	559	self.meet(slot)?;
525	560	}
526		(self.events)(Event::Changed);
	561	(self.events)(self, Event::Changed);
527	562	}
528	563	kind::PRESENCE => {
529	564	let presence = minicbor::decode(&body).map_err(\|_\| io::ErrorKind::InvalidData)?;
...	...	@@ -537,15 +572,18 @@ impl Shared {
537	572	{
538	573	peer.presence = Some(presence);
539	574	}
540		(self.events)(Event::Changed);
	575	(self.events)(self, Event::Changed);
541	576	}
542	577	kind if kind < 256 => {
543	578	if let Some(hello) = hello {
544		(self.events)(Event::Frame {
545		from: &hello,
546		kind,
547		body: &body,
548		});
	579	(self.events)(
	580	self,
	581	Event::Frame {
	582	from: &hello,
	583	kind,
	584	body: &body,
	585	},
	586	);
549	587	}
550	588	}
551	589	_ => {}
...	...	@@ -568,7 +606,9 @@ impl Shared {
568	606	loop {
569	607	let event = {
570	608	let mut state = self.state.lock().unwrap();
571		let stream = state.streams.get_mut(&slot).unwrap();
	609	let Some(stream) = state.streams.get_mut(&slot) else {
	610	break;
	611	};
572	612	let Some(length) = stream.bytes.get(..4) else {
573	613	break;
574	614	};
...	...	@@ -613,12 +653,17 @@ impl Shared {
613	653	if let Some((line, hello, kind, body)) = event {
614	654	match hello {
615	655	None => line.send(kind::HELLO, &self.me)?,
616		Some(hello) if kind == kind::HELLO => (self.events)(Event::Met(&hello, &line)),
617		Some(hello) => (self.events)(Event::Frame {
618		from: &hello,
619		kind,
620		body: &body,
621		}),
	656	Some(hello) if kind == kind::HELLO => {
	657	(self.events)(self, Event::Met(&hello, &line))
	658	}
	659	Some(hello) => (self.events)(
	660	self,
	661	Event::Frame {
	662	from: &hello,
	663	kind,
	664	body: &body,
	665	},
	666	),
622	667	}
623	668	}
624	669	}

crates/notebook/tests/live_membership.rs+5-1

...	...	@@ -104,6 +104,9 @@ fn removal_retires_one_credential_and_other_devices_reconnect_after_a_restart()
104	104	&directory.path().join("alice"),
105	105	);
106	106	let (bob, _) = guest("Bob", &original_code, &url, &directory.path().join("bob"));
	107	until("the initial presence room never met both devices", \|\| {
	108	host.guests().len() == 2
	109	});
107	110	let before = host.sharing();
108	111	let alice_key = before
109	112	.members
...	...	@@ -122,7 +125,8 @@ fn removal_retires_one_credential_and_other_devices_reconnect_after_a_restart()
122	125	bob.ended() == Some(share::Ended::Removed) && bob.host().is_none()
123	126	});
124	127	until("presence did not move to the new room", \|\| {
125		host.guests().len() == 1
	128	let guests = host.guests();
	129	guests.len() == 1 && guests[0].hello.name == "Alice"
126	130	});
127	131	let current = host.sharing();
128	132	assert_ne!(before.secret, current.secret);