clover's git

snowbound

1	//! Live Share: a notebook one Snowbound holds, opened on others through a short code. The
2	//! host serves its notebook's storage verbs (`session::Storage`) and batches guests' ops into
3	//! guarded publications. A guest runs the same replica and durable queue as on an SMB share;
4	//! protected sections use transactions. A guest first meets the host in the code's room,
5	//! where approval grants a device's own access credential. Presence has a separate room
6	//! whose key changes when a device is removed. Large bodies travel a chunk at a time, each answered before
7	//! the next, so a relay never holds much for a slow peer.
8
9	use super::{
10	Event, Hello, Line, Live, Peer, Presence, Reach, Relayed, Room, Sender,
11	wire::{self, Delta, Failure, Reply, Request, Touched, Welcome, WireEntry, Written, kind},
12	};
13	use crate::{Error, Result, background::Reports, discover, session::Storage};
14	use onestore::{CommitError, CommitState, RevisionIndex, Stamp, Store, Transaction};
15	use sha2::{Digest, Sha256};
16	use std::{
17	collections::{BTreeMap, HashMap},
18	io,
19	path::PathBuf,
20	sync::{
21	Arc, Condvar, Mutex, Weak,
22	atomic::{AtomicBool, AtomicU64, Ordering},
23	mpsc,
24	},
25	thread,
26	time::Duration,
27	};
28	use web_time::Instant;
29
30	#[path = "share/batch.rs"]
31	mod batch;
32	#[path = "share/membership.rs"]
33	mod membership;
34	#[cfg(target_arch = "wasm32")]
35	#[path = "share/web.rs"]
36	mod web;
37	pub use membership::Host;
38
39	/// The most bytes one message of a read or an upload carries.
40	const CHUNK: usize = 128 << 10;
41	/// The most bytes of chunks a guest has asked for and not yet been given.
42	const WINDOW: usize = 512 << 10;
43	/// How long a request waits for its reply.
44	const TIMEOUT: Duration = Duration::from_secs(60);
45	use crate::MAX_FILE_BYTES as LIMIT;
46	/// Snapshots of files being read, per guest, and how long one is kept unread.
47	const SNAPSHOTS: usize = 8;
48	const SNAPSHOT_AGE: Duration = Duration::from_secs(120);
49	/// Sections whose image a host keeps to check guests' commits on and tell them what changed,
50	/// and a guest keeps to take those changes on without reading.
51	const IMAGES: usize = 4;
52	/// Earlier images of those a host keeps besides, to tell a guest that missed a change what
53	/// it was, and the most bytes they hold.
54	const VERSIONS: usize = 16;
55	const VERSIONS_BYTES: usize = 64 << 20;
56	/// How long a report of a change settles in a guest's background: its host reports each
57	/// commit once, at once.
58	pub const SETTLE: Duration = Duration::from_millis(20);
59	/// The longest a relay may ask a guest joining to wait before it is told so.
60	const PATIENT: Duration = Duration::from_secs(10);
61	/// What a host says leaving as it stops sharing.
62	const STOPPED: &str = "stopped";
63	/// What a host says hanging up on a guest that asks too much too fast.
64	const FLOODED: &str = "flooded";
65	/// A guest's requests a host holds at once, waiting and in hand, past which it hangs up.
66	const QUEUED: usize = 32;
67	/// The threads working through each guest's requests.
68	const WORKERS: usize = 2;
69	/// Requests a guest may start each second, and at once: every request but a read's later
70	/// chunks and an upload's, which come from memory or go to it.
71	const STARTS: f64 = 100.0;
72	const BURST: f64 = 200.0;
73
74	/// A code as typed, `7kq 4mz 9xr`, as it is shown, `7KQ-4MZ-9XR`; none where it is not a
75	/// code (`super::code`).
76	pub fn code(typed: &str) -> Option<String> {
77	let (number, secret) = super::code::parse(typed)?;
78	super::code::format(number, &secret)
79	}
80
81	/// What a host keeps of a share to take it up again after a relaunch.
82	#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
83	pub struct Sharing {
84	pub share: [u8; 16],
85	/// The presence key, rotated when a device is removed.
86	pub secret: [u8; 16],
87	/// The code, or its secret alone until it has a number (`super::code`).
88	pub code: String,
89	pub password: String,
90	#[serde(default)]
91	pub approve: bool,
92	#[serde(default)]
93	pub members: Vec<Device>,
94	}
95
96	#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
97	pub struct Device {
98	pub secret: [u8; 16],
99	pub name: String,
100	pub device: Option<String>,
101	}
102
103	impl Sharing {
104	/// A new share: its own id, room secret and code.
105	pub fn new(password: &str) -> io::Result<Self> {
106	let mut random = [0; 32];
107	getrandom::fill(&mut random)
108	.map_err(|_| io::Error::other("System random source failed"))?;
109	Ok(Self {
110	share: random[..16].try_into().expect("16 bytes"),
111	secret: random[16..].try_into().expect("16 bytes"),
112	code: super::code::secret()?,
113	password: password.to_owned(),
114	approve: false,
115	members: Vec::new(),
116	})
117	}
118	}
119
120	/// Where a guest keeps a share's notebook: `live://` and the share's id.
121	pub fn location(share: &[u8; 16]) -> String {
122	format!("live://{}", super::hex(share))
123	}
124
125	/// Why joining failed.
126	#[derive(Clone, Debug, PartialEq, Eq)]
127	pub enum Refusal {
128	/// Not a code, or one mistyped, which spends none of the relay's tries.
129	Malformed,
130	/// The person sharing runs a Snowbound of another Live Share version: the newer one's
131	/// `true` where it is theirs, so this one should update.
132	Version {
133	theirs_newer: bool,
134	},
135	/// The code's secret or password is wrong.
136	Wrong,
137	/// No one shares with the code's number now.
138	NoOne,
139	/// The code had too many wrong tries.
140	Expired,
141	/// Too many wrong codes from this network; try again after the wait, where known.
142	TooMany(Option<Duration>),
143	/// The relay is full.
144	Busy,
145	/// The relay couldn't be reached, and why, and no one answered on this network.
146	Unreachable(super::Trouble),
147	/// The relay let this end in, but no one answered.
148	TimedOut,
149	Declined,
150	Cancelled,
151	NotAdmitted,
152	}
153
154	/// Meets the host sharing `code` (and `password`) as `me`, on the networks `reach` names and
155	/// through `relay`: the share it welcomes this end to.
156	pub fn join(
157	me: Hello,
158	code: &str,
159	password: &str,
160	reach: Option<Reach>,
161	relay: Option<&str>,
162	) -> std::result::Result<Welcome, Refusal> {
163	join_while(me, code, password, reach, relay, |_| true)
164	}
165
166	/// Joins while `waiting` returns true, reporting whether the host is deciding approval.
167	pub fn join_while(
168	me: Hello,
169	code: &str,
170	password: &str,
171	reach: Option<Reach>,
172	relay: Option<&str>,
173	continue_joining: impl Fn(bool) -> bool,
174	) -> std::result::Result<Welcome, Refusal> {
175	crate::task::ready(join_while_async(
176	me,
177	code,
178	password,
179	reach,
180	relay,
181	continue_joining,
182	))
183	.map_err(|_| Refusal::Busy)?
184	}
185
186	pub async fn join_while_async(
187	me: Hello,
188	code: &str,
189	password: &str,
190	reach: Option<Reach>,
191	relay: Option<&str>,
192	continue_joining: impl Fn(bool) -> bool,
193	) -> std::result::Result<Welcome, Refusal> {
194	let code = self::code(code).ok_or(Refusal::Malformed)?;
195	let (welcomed, welcome) = mpsc::channel();
196	let approving = Arc::new(AtomicBool::new(false));
197	let approval = Arc::clone(&approving);
198	let (changed, waiting) = crate::task::channel();
199	let live = Live::start(
200	me,
201	&Room::join(&code, password),
202	reach,
203	relay,
204	move |event| match event {
205	Event::Frame {
206	kind: kind::WELCOME,
207	body,
208	..
209	} => {
210	if let Ok(body) = minicbor::decode::<Welcome>(body) {
211	let _ = welcomed.send(Ok(body));
212	}
213	}
214	Event::Frame {
215	kind: kind::APPROVAL,
216	body,
217	..
218	} => {
219	if let Ok(body) = minicbor::decode::<wire::Approval>(body) {
220	match body {
221	wire::Approval::Pending => approval.store(true, Ordering::Release),
222	wire::Approval::Declined => {
223	let _ = welcomed.send(Err(Refusal::Declined));
224	}
225	wire::Approval::Failed => {
226	let _ = welcomed.send(Err(Refusal::NotAdmitted));
227	}
228	}
229	}
230	}
231	_ => {
232	let _ = changed.try_send(());
233	}
234	},
235	)
236	.map_err(|_| Refusal::Unreachable(super::Trouble::Other))?;
237	let start = Instant::now();
238	loop {
239	if !continue_joining(approving.load(Ordering::Acquire)) {
240	return Err(Refusal::Cancelled);
241	}
242	if let Ok(welcome) = welcome.try_recv() {
243	return welcome;
244	}
245	if let Some(version) = live.other_version() {
246	return Err(Refusal::Version {
247	theirs_newer: version > wire::VERSION,
248	});
249	}
250	if live.failed() > 0 {
251	return Err(Refusal::Wrong);
252	}
253	// A peer on this network may yet answer what the relay refused.
254	let waited = start.elapsed();
255	let settled = waited > Duration::from_secs(3) || reach.is_none();
256	match live.relayed() {
257	Relayed::Refused(404, _) if settled => return Err(Refusal::NoOne),
258	Relayed::Refused(410, _) => return Err(Refusal::Expired),
259	// A short wait, as a crowd joining at once meets, passes as the relay is asked
260	// again.
261	Relayed::Refused(429, wait) if wait.is_none_or(|wait| wait > PATIENT) => {
262	return Err(Refusal::TooMany(wait));
263	}
264	Relayed::Refused(503, _) if settled => return Err(Refusal::Busy),
265	Relayed::Unreachable(trouble) if waited > Duration::from_secs(10) => {
266	return Err(Refusal::Unreachable(trouble));
267	}
268	Relayed::Unknown if relay.is_none() && waited > Duration::from_secs(10) => {
269	return Err(Refusal::NoOne);
270	}
271	_ if waited
272	> Duration::from_secs(if approving.load(Ordering::Acquire) {
273	300
274	} else {
275	20
276	}) =>
277	{
278	return Err(Refusal::TimedOut);
279	}
280	_ => {}
281	}
282	crate::task::wait(&waiting, Some(Duration::from_millis(100))).await;
283	}
284	}
285
286	/// A host's side of its guests' storage requests.
287	struct Served {
288	storage: Box<dyn Storage>,
289	/// Sections' images by path, with their stamps, to check commits on.
290	images: Mutex<Vec<Image>>,
291	/// Files being read a chunk at a time, by guest and the read's first request.
292	snapshots: Mutex<ByGuest<(Arc<Vec<u8>>, Instant)>>,
293	/// Bytes a later request carries, by guest and upload.
294	puts: Mutex<ByGuest<Vec<u8>>>,
295	guests: Mutex<BTreeMap<[u8; 16], Admitted>>,
296	writers: Mutex<HashMap<String, mpsc::SyncSender<batch::Waiting>>>,
297	/// Hears the paths guests changed, as the host's own notebook should.
298	host: Mutex<Option<crate::session::Listener>>,
299	/// The share's room, to tell guests what changed.
300	room: Mutex<Option<Sender>>,
301	}
302
303	/// A guest as its host serves it: the line to it, its requests waiting for its workers, how
304	/// many more it may start now, and the sections it read lately, newest first, as it keeps
305	/// their images.
306	struct Admitted {
307	hello: Arc<Hello>,
308	line: Line,
309	queue: mpsc::SyncSender<(u16, Vec<u8>)>,
310	starts: f64,
311	counted: Instant,
312	held: Vec<String>,
313	}
314
315	/// A section's path, stamp and image.
316	type Image = (String, Stamp, Arc<Vec<u8>>);
317	/// What a guest's requests left, by guest and request.
318	type ByGuest<T> = HashMap<([u8; 16], u64), T>;
319
320	/// Whether a guest may name `path`: a catalog path inside the notebook, and not presence's
321	/// own secret, which guests meet in the share's room instead.
322	fn allowed(path: &str) -> bool {
323	path.split('/').all(|part| {
324	!part.is_empty() && part != "." && part != ".." && !part.contains(['\\', '\0', ':'])
325	}) && !path.to_ascii_lowercase().starts_with(".snowbound/live")
326	}
327
328	/// The folder holding catalog path `path`.
329	fn folder(path: &str) -> String {
330	path.rsplit_once('/')
331	.map_or(String::new(), |(folder, _)| folder.to_owned())
332	}
333
334	fn refused(kind: io::ErrorKind, message: &str) -> Error {
335	io::Error::new(kind, message.to_owned()).into()
336	}
337
338	impl Served {
339	/// Serves `peer` on `line`, through workers of its own that end as it leaves.
340	fn admit(
341	self: &Arc<Self>,
342	source: &Arc<super::Shared>,
343	hello: &Arc<Hello>,
344	line: &Line,
345	) -> bool {
346	let peer = hello.peer;
347	let (queue, waiting) = mpsc::sync_channel::<(u16, Vec<u8>)>(QUEUED - WORKERS);
348	let waiting = Arc::new(Mutex::new(waiting));
349	let guest = Admitted {
350	hello: Arc::clone(hello),
351	line: line.clone(),
352	queue,
353	starts: BURST,
354	counted: Instant::now(),
355	held: Vec::new(),
356	};
357	let mut guests = self.guests.lock().unwrap();
358	if !source.connected(hello) {
359	drop(guests);
360	drop(guest);
361	return false;
362	}
363	let retired = guests.insert(peer, guest);
364	drop(guests);
365	drop(retired);
366	for _ in 0..WORKERS {
367	let (served, waiting, line) =
368	(Arc::downgrade(self), Arc::clone(&waiting), line.clone());
369	thread::spawn(move || {
370	loop {
371	let next = waiting.lock().unwrap().recv();
372	let (Ok((kind, body)), Some(served)) = (next, served.upgrade()) else {
373	return;
374	};
375	let _ = line.send(kind::REPLY, &served.handle(&peer, kind, &body));
376	}
377	});
378	}
379	true
380	}
381
382	/// Hands a request from `peer` to its workers, or hangs up on a guest that has too many
383	/// waiting or starts them too fast.
384	fn queue(&self, source: &Arc<super::Shared>, hello: &Arc<Hello>, kind: u16, body: &[u8]) {
385	let peer = &hello.peer;
386	let mut guests = self.guests.lock().unwrap();
387	let Some(guest) = guests.get_mut(peer) else {
388	return;
389	};
390	if !Arc::ptr_eq(&guest.hello, hello) || !source.connected(hello) {
391	return;
392	}
393	let continued = kind == kind::PUT
394	|| matches!(kind, kind::READ | kind::READ_FILE)
395	&& minicbor::decode::<Request>(body).is_ok_and(|request| request.handle.is_some());
396	let now = Instant::now();
397	guest.starts =
398	(guest.starts + now.duration_since(guest.counted).as_secs_f64() * STARTS).min(BURST);
399	guest.counted = now;
400	if !continued {
401	guest.starts -= 1.0;
402	}
403	if guest.starts < 0.0 || guest.queue.try_send((kind, body.to_vec())).is_err() {
404	let guest = guests.remove(peer);
405	drop(guests);
406	if let Some(guest) = guest {
407	guest.line.hang_up(FLOODED);
408	}
409	}
410	}
411
412	fn forget(&self, peer: &[u8; 16], expected: Option<(&Arc<super::Shared>, &Arc<Hello>)>) {
413	let retired = {
414	let mut snapshots = self.snapshots.lock().unwrap();
415	let mut puts = self.puts.lock().unwrap();
416	let mut guests = self.guests.lock().unwrap();
417	if let Some((source, hello)) = expected
418	&& (!source.gone(hello)
419	|| !guests
420	.get(peer)
421	.is_some_and(|guest| Arc::ptr_eq(&guest.hello, hello)))
422	{
423	return;
424	}
425	let retired = guests.remove(peer);
426	snapshots.retain(|(guest, _), _| guest != peer);
427	puts.retain(|(guest, _), _| guest != peer);
428	retired
429	};
430	drop(retired);
431	}
432
433	/// Tells every guest, and the host, that a guest changed the files at `paths`.
434	fn changed(&self, paths: &[String]) {
435	self.tell(paths);
436	if let Some(listener) = &*self.host.lock().unwrap() {
437	listener(paths);
438	}
439	}
440
441	/// Tells every guest the files at `paths` changed, and what they are now.
442	fn tell(&self, paths: &[String]) {
443	let touched = Touched {
444	paths: paths.to_vec(),
445	stamps: (paths.iter())
446	.map(|path| self.storage.stamp(path).ok().map(|stamp| digest(&stamp)))
447	.collect(),
448	};
449	if let Some(room) = self.room.lock().unwrap().as_ref() {
450	room.send(kind::TOUCHED, &touched, None);
451	}
452	}
453
454	/// Notes that `guest` holds the section at `path`, as it now keeps its image.
455	fn holds(guest: &mut Admitted, path: &str) {
456	guest.held.retain(|held| held != path);
457	guest.held.insert(0, path.to_owned());
458	guest.held.truncate(IMAGES);
459	}
460
461	fn handle(self: &Arc<Self>, peer: &[u8; 16], kind: u16, body: &[u8]) -> Reply {
462	let request = match minicbor::decode::<Request>(body) {
463	Ok(request) => request,
464	Err(_) => {
465	return failed(
466	0,
467	refused(io::ErrorKind::InvalidData, "A malformed request"),
468	);
469	}
470	};
471	let id = request.id;
472	if !self.guests.lock().unwrap().contains_key(peer) {
473	return failed(
474	id,
475	refused(
476	io::ErrorKind::PermissionDenied,
477	"The device is no longer connected",
478	),
479	);
480	}
481	match self.answer(peer, kind, request) {
482	Ok(reply) => Reply { id, ..reply },
483	Err(error) => failed(id, error),
484	}
485	}
486
487	fn answer(self: &Arc<Self>, peer: &[u8; 16], kind: u16, request: Request) -> Result<Reply> {
488	let path = request.path.as_str();
489	if !(path.is_empty() && matches!(kind, kind::LIST | kind::PUT) || allowed(path))
490	|| request.to.as_deref().is_some_and(|to| !allowed(to))
491	{
492	return Err(refused(
493	io::ErrorKind::PermissionDenied,
494	"Outside the notebook",
495	));
496	}
497	let to = || {
498	request
499	.to
500	.as_deref()
501	.ok_or_else(|| refused(io::ErrorKind::InvalidInput, "No target"))
502	};
503	let stamp = || -> Result<Stamp> {
504	Ok(request
505	.stamp
506	.as_ref()
507	.ok_or_else(|| refused(io::ErrorKind::InvalidInput, "No stamp"))?
508	.try_into()?)
509	};
510	let done = Reply::default();
511	Ok(match kind {
512	kind::LIST => Reply {
513	entries: Some(
514	self.storage
515	.entries(path)?
516	.into_iter()
517	.filter(|entry| allowed(&within(path, &entry.name)))
518	.map(|entry| wire_entry(&entry))
519	.collect(),
520	),
521	..done
522	},
523	kind::STAMP => Reply {
524	stamp: Some((&self.storage.stamp(path)?).into()),
525	..done
526	},
527	kind::EXISTS => Reply {
528	exists: Some(self.storage.exists(path)),
529	..done
530	},
531	kind::READ | kind::READ_FILE => self.read(peer, kind, &request)?,
532	kind::PUT => {
533	let handle = request.handle.unwrap_or_default();
534	let bytes = request.bytes.unwrap_or_default();
535	let mut puts = self.puts.lock().unwrap();
536	let key = (*peer, handle);
537	let length = puts.get(&key).map_or(0, Vec::len);
538	if request.offset != Some(length as u64) || bytes.is_empty() {
539	puts.remove(&key);
540	return Err(refused(
541	io::ErrorKind::InvalidInput,
542	"An upload out of order",
543	));
544	}
545	let held: usize = puts
546	.iter()
547	.filter(|((guest, _), _)| guest == peer)
548	.map(|(_, bytes)| bytes.len())
549	.sum();
550	if bytes.len() > LIMIT.saturating_sub(held) {
551	puts.remove(&key);
552	return Err(refused(
553	io::ErrorKind::FileTooLarge,
554	"An upload is too large",
555	));
556	}
557	puts.entry(key).or_default().extend_from_slice(&bytes);
558	done
559	}
560	kind::COMMIT => {
561	let transaction = Transaction::from_bytes(&self.carried(peer, &request)?)?;
562	self.commit(peer, path, &transaction)?;
563	self.changed(&[path.to_owned()]);
564	done
565	}
566	kind::EDITS => self.batch(peer, request)?,
567	kind::CONFIRM => {
568	self.storage.confirm(path, &stamp()?)?;
569	done
570	}
571	kind::CREATE => {
572	self.storage.create(path, &self.carried(peer, &request)?)?;
573	self.changed(&[folder(path)]);
574	done
575	}
576	kind::CREATE_DIRECTORY => {
577	self.storage.create_directory(path)?;
578	self.changed(&[folder(path)]);
579	done
580	}
581	kind::HIDE => {
582	self.storage.hide(path)?;
583	done
584	}
585	kind::RENAME | kind::REPLACE => {
586	let to = to()?;
587	if kind == kind::RENAME {
588	self.storage.rename(path, to)?;
589	} else {
590	self.storage.replace(path, to)?;
591	}
592	self.changed(&[folder(path), folder(to)]);
593	done
594	}
595	kind::DELETE => {
596	self.storage.delete(path)?;
597	self.changed(&[folder(path)]);
598	done
599	}
600	kind::PLACE => {
601	let ancestor = request
602	.ancestor
603	.ok_or_else(|| refused(io::ErrorKind::InvalidInput, "No ancestor"))?;
604	let name = request.name.as_deref().unwrap_or_default();
605	self.storage.place(path, ancestor, name)?;
606	self.changed(&[path.to_owned()]);
607	done
608	}
609	kind::SUPERSEDE => {
610	self.storage.supersede(path, &stamp()?, to()?)?;
611	self.changed(&[folder(path)]);
612	done
613	}
614	_ => return Err(refused(io::ErrorKind::Unsupported, "An unknown request")),
615	})
616	}
617
618	/// The bytes a request carries, itself or in its uploads.
619	fn carried(&self, peer: &[u8; 16], request: &Request) -> Result<Vec<u8>> {
620	match (&request.bytes, request.handle) {
621	(Some(bytes), _) => Ok(bytes.clone()),
622	(None, Some(handle)) => self
623	.puts
624	.lock()
625	.unwrap()
626	.remove(&(*peer, handle))
627	.ok_or_else(|| refused(io::ErrorKind::InvalidInput, "No such upload")),
628	(None, None) => Ok(Vec::new()),
629	}
630	}
631
632	/// A chunk of a file as it stood when its read began.
633	fn read(&self, peer: &[u8; 16], kind: u16, request: &Request) -> Result<Reply> {
634	let offset = request.offset.unwrap_or_default() as usize;
635	let mut snapshots = self.snapshots.lock().unwrap();
636	snapshots.retain(|_, (_, read)| read.elapsed() < SNAPSHOT_AGE);
637	if let (kind::READ, None, Some(base)) = (kind, request.handle, &request.stamp)
638	&& let Ok(base) = Stamp::try_from(base)
639	&& let Some(changes) = self.changes(&request.path, &base)?
640	{
641	if let Some(guest) = self.guests.lock().unwrap().get_mut(peer) {
642	Self::holds(guest, &request.path);
643	}
644	return Ok(changes);
645	}
646	let (image, handle) = match request.handle {
647	Some(handle) => {
648	let (image, read) = snapshots
649	.get_mut(&(*peer, handle))
650	.ok_or_else(|| refused(io::ErrorKind::TimedOut, "The read went stale"))?;
651	*read = Instant::now();
652	(Arc::clone(image), handle)
653	}
654	None => {
655	drop(snapshots);
656	if kind == kind::READ
657	&& let Some(guest) = self.guests.lock().unwrap().get_mut(peer)
658	{
659	Self::holds(guest, &request.path);
660	}
661	let limit = (request.limit.unwrap_or(LIMIT as u64) as usize).min(LIMIT);
662	let image = match kind {
663	kind::READ => self.image(&request.path)?,
664	_ => Arc::new(self.storage.read_file(&request.path, limit)?),
665	};
666	if image.len() > limit {
667	return Err(io::Error::from(io::ErrorKind::FileTooLarge).into());
668	}
669	snapshots = self.snapshots.lock().unwrap();
670	if snapshots.keys().filter(|(guest, _)| guest == peer).count() >= SNAPSHOTS {
671	snapshots.retain(|(guest, _), _| guest != peer);
672	}
673	(image, request.id)
674	}
675	};
676	let end = image.len().min(offset.saturating_add(CHUNK));
677	let bytes = image.get(offset..end).unwrap_or_default().to_vec();
678	if end < image.len() {
679	snapshots.insert((*peer, handle), (Arc::clone(&image), Instant::now()));
680	} else {
681	snapshots.remove(&(*peer, handle));
682	}
683	Ok(Reply {
684	bytes: Some(bytes),
685	length: Some(image.len() as u64),
686	handle: Some(handle),
687	stamp: (kind == kind::READ)
688	.then(|| Stamp::of(&image).ok())
689	.flatten()
690	.map(|stamp| (&stamp).into()),
691	..Reply::default()
692	})
693	}
694
695	/// What changed in the section at `path` since the image with stamp `base`, where that
696	/// image is kept and the changes are much smaller than the section.
697	fn changes(&self, path: &str, base: &Stamp) -> Result<Option<Reply>> {
698	let before = self
699	.images
700	.lock()
701	.unwrap()
702	.iter()
703	.find_map(|(held, at, image)| (held == path && at == base).then(|| Arc::clone(image)));
704	let Some(before) = before else {
705	return Ok(None);
706	};
707	let after = self.image(path)?;
708	let writes = match delta(&before, &after) {
709	Some(writes) => writes,
710	None if before == after => Vec::new(),
711	None => return Ok(None),
712	};
713	Ok(Some(Reply {
714	writes: Some(writes),
715	length: Some(after.len() as u64),
716	stamp: Some((&Stamp::of(&after)?).into()),
717	..Reply::default()
718	}))
719	}
720
721	/// The section or TOC at `path` as it stands: the image kept for it while its stamp holds.
722	fn image(&self, path: &str) -> Result<Arc<Vec<u8>>> {
723	let stamp = self.storage.stamp(path)?;
724	let kept = self
725	.images
726	.lock()
727	.unwrap()
728	.iter()
729	.find_map(|(held, at, image)| {
730	(held == path && *at == stamp).then(|| Arc::clone(image))
731	});
732	if let Some(image) = kept {
733	return Ok(image);
734	}
735	let image = Arc::new(self.storage.read(path)?);
736	self.keep(path, Arc::clone(&image));
737	Ok(image)
738	}
739
740	/// Keeps `image` as the section at `path` now, with the newest of each of `IMAGES`
741	/// sections and earlier images up to `VERSIONS` and `VERSIONS_BYTES`.
742	fn keep(&self, path: &str, image: Arc<Vec<u8>>) {
743	let Ok(stamp) = Stamp::of(&image) else {
744	return;
745	};
746	let mut images = self.images.lock().unwrap();
747	images.retain(|(held, at, _)| held != path || *at != stamp);
748	images.insert(0, (path.to_owned(), stamp, image));
749	let (mut newest, mut versions, mut bytes) = (Vec::new(), 0, 0);
750	images.retain(|(held, _, image)| {
751	if !newest.contains(held) {
752	newest.push(held.clone());
753	return newest.len() <= IMAGES;
754	}
755	versions += 1;
756	bytes += image.len();
757	newest.iter().take(IMAGES).any(|kept| kept == held)
758	&& versions <= VERSIONS
759	&& bytes <= VERSIONS_BYTES
760	});
761	}
762
763	/// Commits `guest`'s transaction once the section it makes parses.
764	fn commit(&self, guest: &[u8; 16], path: &str, transaction: &Transaction) -> Result<()> {
765	let not_committed = |error: io::Error| {
766	Error::Remote(CommitError {
767	state: CommitState::NotCommitted,
768	error,
769	})
770	};
771	let image = self.image(path).map_err(|error| match error {
772	Error::Io(error) => not_committed(error),
773	error => error,
774	})?;
775	if Stamp::of(&image).ok().as_ref() != Some(transaction.base()) {
776	return Err(not_committed(io::Error::new(
777	io::ErrorKind::ResourceBusy,
778	"The section changed since",
779	)));
780	}
781	let mut next = (*image).clone();
782	let checked = transaction.apply(&mut next).and_then(|()| {
783	let store = Store::parse(&next)?;
784	RevisionIndex::parse(&store).map(drop)
785	});
786	if let Err(error) = checked {
787	return Err(not_committed(io::Error::new(
788	io::ErrorKind::InvalidData,
789	error.to_string(),
790	)));
791	}
792	self.storage.commit(path, transaction)?;
793	self.tell_delta(path, &image, &next, Some(guest));
794	self.keep(path, Arc::new(next));
795	Ok(())
796	}
797
798	/// Tells the guests holding the section at `path` what a commit changed in it, from
799	/// `before` to `after`, where that is small enough to send: all but the guest whose
800	/// commit it was, which has it already.
801	fn tell_delta(&self, path: &str, before: &[u8], after: &[u8], committed: Option<&[u8; 16]>) {
802	let (Ok(base), Some(writes)) = (Stamp::of(before), delta(before, after)) else {
803	return;
804	};
805	let delta = Delta {
806	path: path.to_owned(),
807	base: (&base).into(),
808	length: after.len() as u64,
809	writes,
810	};
811	let mut guests = self.guests.lock().unwrap();
812	let holding: Vec<[u8; 16]> = guests
813	.iter_mut()
814	.filter(|(id, guest)| Some(*id) != committed && guest.held.iter().any(|h| h == path))
815	.map(|(id, guest)| {
816	Self::holds(guest, path);
817	*id
818	})
819	.collect();
820	drop(guests);
821	if let (Some(room), false) = (self.room.lock().unwrap().as_ref(), holding.is_empty()) {
822	room.send(kind::DELTA, &delta, Some(&holding));
823	}
824	}
825
826	/// The host's own change to the section at `path`, which guests hear as a delta where
827	/// the host kept the image before it.
828	fn changed_here(&self, path: &str) {
829	let kept = self
830	.images
831	.lock()
832	.unwrap()
833	.iter()
834	.find_map(|(held, at, image)| (held == path).then(|| (at.clone(), Arc::clone(image))));
835	let Some((at, before)) = kept else {
836	return;
837	};
838	if self.storage.stamp(path).is_ok_and(|now| now == at) {
839	return;
840	}
841	if let Ok(after) = self.storage.read(path) {
842	self.tell_delta(path, &before, &after, None);
843	self.keep(path, Arc::new(after));
844	}
845	}
846	}
847
848	/// A stamp as `Touched` names it.
849	fn digest(stamp: &Stamp) -> u64 {
850	let hash = Sha256::new()
851	.chain_update(stamp.header)
852	.chain_update(stamp.length.to_le_bytes())
853	.finalize();
854	u64::from_le_bytes(hash[..8].try_into().expect("eight bytes"))
855	}
856
857	/// The writes that make `after` of `before`, a commit's appended bytes, patches and header,
858	/// where their encoding fits a chunk and they are much less than `after` itself.
859	fn delta(before: &[u8], after: &[u8]) -> Option<Vec<Written>> {
860	const BLOCK: usize = 4096;
861	if before.len() < 1024 || after.len() < before.len() {
862	return None;
863	}
864	let mut writes = vec![Written {
865	offset: 0,
866	bytes: after[..1024].to_vec(),
867	}];
868	let mut at = 1024;
869	while at < before.len() {
870	let end = (at + BLOCK).min(before.len());
871	if before[at..end] != after[at..end] {
872	let first = (at..end).find(|&i| before[i] != after[i]).unwrap_or(at);
873	let last = (at..end).rfind(|&i| before[i] != after[i]).unwrap_or(first) + 1;
874	match writes.last_mut() {
875	Some(write) if write.offset as usize + write.bytes.len() + 64 >= first => {
876	let from = write.offset as usize;
877	write.bytes = after[from..last].to_vec();
878	}
879	_ => writes.push(Written {
880	offset: first as u64,
881	bytes: after[first..last].to_vec(),
882	}),
883	}
884	}
885	at = end;
886	}
887	if after.len() > before.len() {
888	writes.push(Written {
889	offset: before.len() as u64,
890	bytes: after[before.len()..].to_vec(),
891	});
892	}
893	let sent: usize = writes.iter().map(|write| write.bytes.len()).sum();
894	(sent <= after.len() / 2 && sent <= CHUNK && minicbor::to_vec(&writes).ok()?.len() <= CHUNK)
895	.then_some(writes)
896	}
897
898	/// `image` with `writes`, `length` long: none where a write falls outside it.
899	fn written(image: &[u8], length: u64, writes: &[Written]) -> Option<Vec<u8>> {
900	let length = usize::try_from(length)
901	.ok()
902	.filter(|length| *length >= image.len())?;
903	let mut next = image.to_vec();
904	next.resize(length, 0);
905	for write in writes {
906	let offset = usize::try_from(write.offset).ok()?;
907	next.get_mut(offset..offset.checked_add(write.bytes.len())?)?
908	.copy_from_slice(&write.bytes);
909	}
910	Some(next)
911	}
912
913	fn within(folder: &str, name: &str) -> String {
914	if folder.is_empty() {
915	name.to_owned()
916	} else {
917	format!("{folder}/{name}")
918	}
919	}
920
921	fn wire_entry(entry: &discover::Entry) -> WireEntry {
922	WireEntry {
923	name: entry.name.clone(),
924	kind: match entry.kind {
925	discover::EntryKind::File => 0,
926	discover::EntryKind::Directory => 1,
927	discover::EntryKind::Other => 2,
928	discover::EntryKind::Evicted => 3,
929	},
930	size: entry.listed.size,
931	modified: entry.listed.modified,
932	}
933	}
934
935	fn entry(entry: &WireEntry) -> discover::Entry {
936	discover::Entry {
937	name: entry.name.clone(),
938	kind: match entry.kind {
939	0 => discover::EntryKind::File,
940	1 => discover::EntryKind::Directory,
941	3 => discover::EntryKind::Evicted,
942	_ => discover::EntryKind::Other,
943	},
944	listed: discover::Listed {
945	size: entry.size,
946	modified: entry.modified,
947	},
948	}
949	}
950
951	fn state_number(state: CommitState) -> u8 {
952	match state {
953	CommitState::NotCommitted => 0,
954	CommitState::Unknown => 1,
955	CommitState::Committed => 2,
956	}
957	}
958
959	fn failed(id: u64, error: Error) -> Reply {
960	let (kind, state) = match &error {
961	Error::Io(error) | Error::RemoteIo(error) => (error.kind(), None),
962	Error::Remote(error) => (error.error.kind(), Some(state_number(error.state))),
963	Error::Document(_) => (io::ErrorKind::InvalidData, None),
964	_ => (io::ErrorKind::Other, None),
965	};
966	let message = match &error {
967	Error::Remote(error) => error.error.to_string(),
968	error => error.to_string(),
969	};
970	Reply {
971	id,
972	failure: Some(Failure {
973	kind: wire::error_number(kind),
974	message,
975	state,
976	}),
977	..Reply::default()
978	}
979	}
980
981	/// How a request failed.
982	enum Failed {
983	/// It never left this end.
984	Unsent(io::Error),
985	/// Its answer was lost: whatever it asked may have happened.
986	Lost(io::Error),
987	/// The host refused it.
988	Refused(Failure),
989	}
990
991	impl Failed {
992	fn io(self) -> io::Error {
993	match self {
994	Failed::Unsent(error) | Failed::Lost(error) => error,
995	Failed::Refused(failure) => {
996	io::Error::new(wire::error_kind(failure.kind), failure.message)
997	}
998	}
999	}
1000
1001	/// As a commit's failure: one never sent was not committed, one whose answer was lost may
1002	/// have been.
1003	fn commit(self) -> CommitError {
1004	let state = match &self {
1005	Failed::Unsent(_) => CommitState::NotCommitted,
1006	Failed::Lost(_) => CommitState::Unknown,
1007	Failed::Refused(failure) => match failure.state {
1008	Some(1) => CommitState::Unknown,
1009	Some(2) => CommitState::Committed,
1010	_ => CommitState::NotCommitted,
1011	},
1012	};
1013	CommitError {
1014	state,
1015	error: self.io(),
1016	}
1017	}
1018	}
1019
1020	/// A guest's way to a share: the share's room, and the requests waiting on its host.
1021	pub struct Guest {
1022	live: Live,
1023	inner: Arc<Inner>,
1024	}
1025
1026	struct Inner {
1027	share: [u8; 16],
1028	me: Hello,
1029	reach: Option<Reach>,
1030	relay: Option<String>,
1031	events: Arc<dyn Fn() + Send + Sync>,
1032	presence: Mutex<Option<([u8; 16], Live)>>,
1033	here: Mutex<Presence>,
1034	/// The host and the line to it, while connected.
1035	host: Mutex<Option<(Arc<Hello>, Line)>>,
1036	pending: Mutex<HashMap<u64, crate::task::Answer<Reply>>>,
1037	next: AtomicU64,
1038	/// Where the host's reports of changed files go, while a background watches.
1039	watch: Mutex<Option<Arc<Reports>>>,
1040	/// Why this device can no longer reach the share.
1041	ended: Mutex<Option<Ended>>,
1042	/// Bytes of chunks asked for and not yet given.
1043	asked: (Mutex<Chunks>, Condvar),
1044	/// The sections read lately, kept as the host's deltas change them, newest first.
1045	images: Mutex<Vec<(String, Arc<Vec<u8>>)>>,
1046	/// The stamps of sections whose image held is the host's now, as its last report of
1047	/// them said.
1048	current: Mutex<HashMap<String, Stamp>>,
1049	#[cfg(target_arch = "wasm32")]
1050	catalog: Mutex<Option<PathBuf>>,
1051	}
1052
1053	#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1054	pub enum Ended {
1055	Stopped,
1056	Removed,
1057	}
1058
1059	#[derive(Default)]
1060	struct Chunks {
1061	bytes: usize,
1062	#[cfg(target_arch = "wasm32")]
1063	waiting: Vec<std::task::Waker>,
1064	}
1065
1066	struct Requested<'a> {
1067	inner: &'a Inner,
1068	id: u64,
1069	chunked: bool,
1070	}
1071
1072	impl Drop for Requested<'_> {
1073	fn drop(&mut self) {
1074	self.inner.pending.lock().unwrap().remove(&self.id);
1075	if self.chunked {
1076	let (asked, room) = &self.inner.asked;
1077	let mut asked = asked.lock().unwrap();
1078	asked.bytes -= CHUNK;
1079	#[cfg(target_arch = "wasm32")]
1080	let waiting = std::mem::take(&mut asked.waiting);
1081	drop(asked);
1082	room.notify_all();
1083	#[cfg(target_arch = "wasm32")]
1084	for waker in waiting {
1085	waker.wake();
1086	}
1087	}
1088	}
1089	}
1090
1091	impl Guest {
1092	/// Joins share `share` through its `secret` as `me`, where `reach` and `relay` say.
1093	/// `events` runs on a network thread whenever the host comes or goes, or the peers change.
1094	pub fn start(
1095	me: Hello,
1096	share: [u8; 16],
1097	secret: [u8; 16],
1098	reach: Option<Reach>,
1099	relay: Option<&str>,
1100	events: impl Fn() + Send + Sync + 'static,
1101	) -> io::Result<Arc<Self>> {
1102	let inner = Arc::new(Inner {
1103	share,
1104	me: me.clone(),
1105	reach,
1106	relay: relay.map(str::to_owned),
1107	events: Arc::new(events),
1108	presence: Mutex::default(),
1109	here: Mutex::default(),
1110	host: Mutex::default(),
1111	pending: Mutex::default(),
1112	next: AtomicU64::new(1),
1113	watch: Mutex::default(),
1114	ended: Mutex::default(),
1115	asked: Default::default(),
1116	images: Mutex::default(),
1117	current: Mutex::default(),
1118	#[cfg(target_arch = "wasm32")]
1119	catalog: Mutex::default(),
1120	});
1121	let heard = Arc::clone(&inner);
1122	let live = Live::start_with(
1123	me,
1124	&Room::Notebook(secret),
1125	reach,
1126	relay,
1127	move |source, event| {
1128	heard.heard(source, event);
1129	(heard.events)();
1130	},
1131	)?;
1132	Ok(Arc::new(Self { live, inner }))
1133	}
1134
1135	/// Where the notebook is kept: `location(share)`.
1136	pub fn location(&self) -> String {
1137	location(&self.inner.share)
1138	}
1139
1140	/// The host, while connected.
1141	pub fn host(&self) -> Option<Arc<Hello>> {
1142	let host = self.inner.host.lock().unwrap();
1143	host.as_ref().map(|(hello, _)| Arc::clone(hello))
1144	}
1145
1146	/// Why the host ended this device's access.
1147	pub fn ended(&self) -> Option<Ended> {
1148	*self.inner.ended.lock().unwrap()
1149	}
1150
1151	/// Everyone in the share's room: the host and the other guests.
1152	pub fn peers(&self) -> Vec<Peer> {
1153	self.inner
1154	.presence
1155	.lock()
1156	.unwrap()
1157	.as_ref()
1158	.map(|(_, live)| live.peers())
1159	.unwrap_or_default()
1160	}
1161
1162	pub fn set_presence(&self, presence: Presence) {
1163	*self.inner.here.lock().unwrap() = presence.clone();
1164	if let Some((_, live)) = &*self.inner.presence.lock().unwrap() {
1165	live.set_presence(presence);
1166	}
1167	}
1168
1169	/// Sends the host's reports of changed files to `reports` while it stays connected.
1170	pub(crate) fn watch(&self, reports: Reports) -> io::Result<()> {
1171	if self.host().is_none() {
1172	return Err(self.offline());
1173	}
1174	*self.inner.watch.lock().unwrap() = Some(Arc::new(reports));
1175	Ok(())
1176	}
1177
1178	fn offline(&self) -> io::Error {
1179	if let Some(version) = self.live.other_version() {
1180	return io::Error::new(io::ErrorKind::NotConnected, wire::Version(version));
1181	}
1182	let message = match &*self.inner.ended.lock().unwrap() {
1183	Some(Ended::Stopped) => "The host stopped sharing this notebook",
1184	Some(Ended::Removed) => {
1185	"This device was removed. Ask the person sharing for a new link or code."
1186	}
1187	None => "The computer sharing this notebook can’t be reached",
1188	};
1189	io::Error::new(io::ErrorKind::NotConnected, message)
1190	}
1191
1192	/// Asks the host `kind` of `request`, waiting for its reply.
1193	#[cfg(not(target_arch = "wasm32"))]
1194	fn request(&self, kind: u16, request: Request) -> std::result::Result<Reply, Failed> {
1195	crate::task::ready(self.request_async(kind, request)).map_err(Failed::Unsent)?
1196	}
1197
1198	async fn request_async(
1199	&self,
1200	kind: u16,
1201	mut request: Request,
1202	) -> std::result::Result<Reply, Failed> {
1203	let Some((_, line)) = self.inner.host.lock().unwrap().clone() else {
1204	return Err(Failed::Unsent(self.offline()));
1205	};
1206	let chunked = matches!(kind, kind::READ | kind::READ_FILE | kind::PUT);
1207	if chunked {
1208	#[cfg(not(target_arch = "wasm32"))]
1209	{
1210	let (asked, _room) = &self.inner.asked;
1211	let mut asked = asked.lock().unwrap();
1212	while asked.bytes + CHUNK > WINDOW {
1213	asked = _room.wait(asked).unwrap();
1214	}
1215	asked.bytes += CHUNK;
1216	}
1217	#[cfg(target_arch = "wasm32")]
1218	std::future::poll_fn(|context| {
1219	let mut asked = self.inner.asked.0.lock().unwrap();
1220	if asked.bytes + CHUNK > WINDOW {
1221	if !asked
1222	.waiting
1223	.iter()
1224	.any(|waker| waker.will_wake(context.waker()))
1225	{
1226	asked.waiting.push(context.waker().clone());
1227	}
1228	std::task::Poll::Pending
1229	} else {
1230	asked.bytes += CHUNK;
1231	std::task::Poll::Ready(())
1232	}
1233	})
1234	.await;
1235	}
1236	let id = self.inner.next.fetch_add(1, Ordering::Relaxed);
1237	request.id = id;
1238	let (answer, answered) = crate::task::response();
1239	self.inner.pending.lock().unwrap().insert(id, answer);
1240	let _requested = Requested {
1241	inner: &self.inner,
1242	id,
1243	chunked,
1244	};
1245	let sent = line.send(kind, &request);
1246	let reply = match sent {
1247	Err(error) => Err(Failed::Unsent(error)),
1248	Ok(()) => match crate::task::answered(&answered, TIMEOUT).await {
1249	Ok(reply) => Ok(reply),
1250	Err(mpsc::RecvTimeoutError::Timeout) => Err(Failed::Lost(io::Error::new(
1251	io::ErrorKind::TimedOut,
1252	"The computer sharing this notebook didn’t answer",
1253	))),
1254	Err(mpsc::RecvTimeoutError::Disconnected) => Err(Failed::Lost(self.offline())),
1255	},
1256	};
1257	let reply = reply?;
1258	match reply.failure {
1259	Some(failure) => Err(Failed::Refused(failure)),
1260	None => Ok(reply),
1261	}
1262	}
1263
1264	#[cfg(not(target_arch = "wasm32"))]
1265	fn ask(&self, kind: u16, request: Request) -> io::Result<Reply> {
1266	self.request(kind, request).map_err(Failed::io)
1267	}
1268
1269	async fn ask_async(&self, kind: u16, request: Request) -> io::Result<Reply> {
1270	self.request_async(kind, request).await.map_err(Failed::io)
1271	}
1272
1273	/// Puts `bytes` in `request`, or uploads them first where they are large.
1274	#[cfg(not(target_arch = "wasm32"))]
1275	fn carry(&self, request: &mut Request, bytes: Vec<u8>) -> std::result::Result<(), Failed> {
1276	crate::task::ready(self.carry_async(request, bytes)).map_err(Failed::Unsent)?
1277	}
1278
1279	async fn carry_async(
1280	&self,
1281	request: &mut Request,
1282	bytes: Vec<u8>,
1283	) -> std::result::Result<(), Failed> {
1284	if bytes.len() > LIMIT {
1285	return Err(Failed::Unsent(io::ErrorKind::FileTooLarge.into()));
1286	}
1287	if bytes.len() <= CHUNK {
1288	request.bytes = Some(bytes);
1289	return Ok(());
1290	}
1291	let handle = self.inner.next.fetch_add(1, Ordering::Relaxed);
1292	for (index, chunk) in bytes.chunks(CHUNK).enumerate() {
1293	let put = Request {
1294	handle: Some(handle),
1295	offset: Some((index * CHUNK) as u64),
1296	bytes: Some(chunk.to_vec()),
1297	..Request::default()
1298	};
1299	// Nothing the upload carries happens before the request that uses it.
1300	self.request_async(kind::PUT, put)
1301	.await
1302	.map_err(|failed| match failed {
1303	Failed::Lost(error) => Failed::Unsent(error),
1304	failed => failed,
1305	})?;
1306	}
1307	request.handle = Some(handle);
1308	Ok(())
1309	}
1310
1311	/// Reads the file at `path` a chunk at a time, as `kind` reads it; a section held, as
1312	/// the changes to it.
1313	#[cfg(not(target_arch = "wasm32"))]
1314	fn read(&self, kind: u16, path: &str, limit: usize) -> io::Result<Vec<u8>> {
1315	crate::task::ready(self.read_async(kind, path, limit))?
1316	}
1317
1318	async fn read_async(&self, kind: u16, path: &str, limit: usize) -> io::Result<Vec<u8>> {
1319	let held = (kind == kind::READ)
1320	.then(|| self.inner.image(path))
1321	.flatten();
1322	let first = self
1323	.ask_async(
1324	kind,
1325	Request {
1326	path: path.to_owned(),
1327	offset: Some(0),
1328	limit: Some(limit as u64),
1329	stamp: (held.as_deref())
1330	.and_then(|image| Stamp::of(image).ok())
1331	.map(|stamp| (&stamp).into()),
1332	..Request::default()
1333	},
1334	)
1335	.await?;
1336	if let (Some(writes), Some(held)) = (&first.writes, held) {
1337	let stamp: Option<Stamp> = first.stamp.as_ref().and_then(|s| s.try_into().ok());
1338	let image = written(&held, first.length.unwrap_or_default(), writes)
1339	.filter(|image| stamp.is_some() && Stamp::of(image).ok() == stamp)
1340	.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Changes that miss"))?;
1341	self.inner.hold(path, Arc::new(image.clone()));
1342	return Ok(image);
1343	}
1344	let length = first.length.unwrap_or_default() as usize;
1345	if length > limit {
1346	return Err(io::ErrorKind::FileTooLarge.into());
1347	}
1348	let mut image = first.bytes.unwrap_or_default();
1349	while image.len() < length {
1350	let chunk = self
1351	.ask_async(
1352	kind,
1353	Request {
1354	path: path.to_owned(),
1355	offset: Some(image.len() as u64),
1356	handle: first.handle,
1357	..Request::default()
1358	},
1359	)
1360	.await?
1361	.bytes
1362	.unwrap_or_default();
1363	if chunk.is_empty() {
1364	return Err(io::ErrorKind::UnexpectedEof.into());
1365	}
1366	image.extend_from_slice(&chunk);
1367	}
1368	image.truncate(length);
1369	if kind == kind::READ {
1370	self.inner.hold(path, Arc::new(image.clone()));
1371	}
1372	Ok(image)
1373	}
1374
1375	/// The image of the section at `path` with stamp `stamp`, where this guest holds it.
1376	fn held(&self, path: &str, stamp: &Stamp) -> Option<Vec<u8>> {
1377	let image = self.inner.image(path)?;
1378	(Stamp::of(&image).ok().as_ref() == Some(stamp)).then(|| image.to_vec())
1379	}
1380
1381	#[cfg(not(target_arch = "wasm32"))]
1382	pub(crate) fn entries(&self, folder: &str) -> io::Result<Vec<discover::Entry>> {
1383	crate::task::ready(self.entries_async(folder))?
1384	}
1385
1386	pub(crate) async fn entries_async(&self, folder: &str) -> io::Result<Vec<discover::Entry>> {
1387	let reply = self
1388	.ask_async(
1389	kind::LIST,
1390	Request {
1391	path: folder.to_owned(),
1392	..Request::default()
1393	},
1394	)
1395	.await?;
1396	let entries: Vec<_> = reply
1397	.entries
1398	.unwrap_or_default()
1399	.iter()
1400	.map(entry)
1401	.collect();
1402	#[cfg(target_arch = "wasm32")]
1403	self.cache_entries(folder, &entries).await?;
1404	Ok(entries)
1405	}
1406
1407	/// The stamp of the file at `path`: the host's last report of it where this guest holds
1408	/// that image, else the host's answer.
1409	#[cfg(not(target_arch = "wasm32"))]
1410	fn stamp(&self, path: &str) -> io::Result<Stamp> {
1411	crate::task::ready(self.stamp_async(path))?
1412	}
1413
1414	async fn stamp_async(&self, path: &str) -> io::Result<Stamp> {
1415	if let Some(stamp) = self.inner.current.lock().unwrap().get(path) {
1416	return Ok(stamp.clone());
1417	}
1418	let reply = self
1419	.ask_async(
1420	kind::STAMP,
1421	Request {
1422	path: path.to_owned(),
1423	..Request::default()
1424	},
1425	)
1426	.await?;
1427	reply
1428	.stamp
1429	.as_ref()
1430	.ok_or_else(|| io::Error::from(io::ErrorKind::InvalidData))?
1431	.try_into()
1432	}
1433
1434	#[cfg(not(target_arch = "wasm32"))]
1435	fn commit(
1436	&self,
1437	path: &str,
1438	transaction: &Transaction,
1439	) -> std::result::Result<(), CommitError> {
1440	crate::task::ready(self.commit_async(path, transaction)).map_err(|error| CommitError {
1441	state: CommitState::NotCommitted,
1442	error,
1443	})?
1444	}
1445
1446	async fn commit_async(
1447	&self,
1448	path: &str,
1449	transaction: &Transaction,
1450	) -> std::result::Result<(), CommitError> {
1451	let mut request = Request {
1452	path: path.to_owned(),
1453	..Request::default()
1454	};
1455	self.carry_async(&mut request, transaction.to_bytes())
1456	.await
1457	.map_err(Failed::commit)?;
1458	self.request_async(kind::COMMIT, request)
1459	.await
1460	.map(drop)
1461	.map_err(Failed::commit)
1462	}
1463
1464	#[cfg(not(target_arch = "wasm32"))]
1465	fn confirm(&self, path: &str, base: &Stamp) -> std::result::Result<(), CommitError> {
1466	crate::task::ready(self.confirm_async(path, base)).map_err(|error| CommitError {
1467	state: CommitState::NotCommitted,
1468	error,
1469	})?
1470	}
1471
1472	async fn confirm_async(
1473	&self,
1474	path: &str,
1475	base: &Stamp,
1476	) -> std::result::Result<(), CommitError> {
1477	let request = Request {
1478	path: path.to_owned(),
1479	stamp: Some(base.into()),
1480	..Request::default()
1481	};
1482	self.request_async(kind::CONFIRM, request)
1483	.await
1484	.map(drop)
1485	.map_err(Failed::commit)
1486	}
1487
1488	async fn edits_async(
1489	&self,
1490	path: &str,
1491	transaction: &Transaction,
1492	edits: &[crate::PendingEdit],
1493	revisions: &BTreeMap<onestore::ExGuid, onestore::ExGuid>,
1494	) -> std::result::Result<(), CommitError> {
1495	let bytes = batch::encode(
1496	&batch::Edits {
1497	edits: edits
1498	.iter()
1499	.map(|edit| (edit.author.clone(), edit.edit.clone()))
1500	.collect(),
1501	revisions: revisions.clone(),
1502	},
1503	LIMIT,
1504	)
1505	.map_err(|error| CommitError {
1506	state: CommitState::NotCommitted,
1507	error,
1508	})?;
1509	let mut request = Request {
1510	path: path.to_owned(),
1511	stamp: Some(transaction.base().into()),
1512	..Request::default()
1513	};
1514	self.carry_async(&mut request, bytes)
1515	.await
1516	.map_err(Failed::commit)?;
1517	match self.request_async(kind::EDITS, request).await {
1518	Ok(reply) => {
1519	if let Some(stamp) = reply.stamp {
1520	let stamp = Stamp::try_from(&stamp).map_err(|error| CommitError {
1521	state: CommitState::Unknown,
1522	error,
1523	})?;
1524	self.inner
1525	.current
1526	.lock()
1527	.unwrap()
1528	.insert(path.to_owned(), stamp);
1529	}
1530	Ok(())
1531	}
1532	Err(Failed::Refused(failure))
1533	if wire::error_kind(failure.kind) == io::ErrorKind::Unsupported =>
1534	{
1535	self.commit_async(path, transaction).await?;
1536	self.inner.published(path, transaction);
1537	Ok(())
1538	}
1539	Err(error) => Err(error.commit()),
1540	}
1541	}
1542
1543	/// A request on `path` that answers nothing but whether it happened.
1544	#[cfg(not(target_arch = "wasm32"))]
1545	fn verb(&self, kind: u16, path: &str, request: Request) -> Result<()> {
1546	self.ask(
1547	kind,
1548	Request {
1549	path: path.to_owned(),
1550	..request
1551	},
1552	)?;
1553	Ok(())
1554	}
1555	}
1556
1557	impl Inner {
1558	fn image(&self, path: &str) -> Option<Arc<Vec<u8>>> {
1559	let images = self.images.lock().unwrap();
1560	images
1561	.iter()
1562	.find_map(|(held, image)| (held == path).then(|| Arc::clone(image)))
1563	}
1564
1565	fn hold(&self, path: &str, image: Arc<Vec<u8>>) {
1566	let mut images = self.images.lock().unwrap();
1567	images.retain(|(held, _)| held != path);
1568	images.insert(0, (path.to_owned(), image));
1569	images.truncate(IMAGES);
1570	}
1571
1572	/// Takes on a delta to an image held.
1573	fn apply(&self, delta: &Delta) {
1574	let held = self.image(&delta.path);
1575	let base: Option<Stamp> = (&delta.base).try_into().ok();
1576	if let Some(image) = held
1577	&& Stamp::of(&image).ok() == base
1578	&& let Some(next) = written(&image, delta.length, &delta.writes)
1579	{
1580	self.hold(&delta.path, Arc::new(next));
1581	}
1582	}
1583
1584	/// Takes on this guest's own commit to an image held; the host's report, not this, says
1585	/// whether the image is the host's, as another's commit may have followed.
1586	fn published(&self, path: &str, transaction: &Transaction) {
1587	if let Some(image) = self.image(path) {
1588	let mut next = (*image).clone();
1589	if transaction.apply(&mut next).is_ok() {
1590	self.hold(path, Arc::new(next));
1591	}
1592	}
1593	}
1594
1595	/// Takes the host's report of what the files at `paths` are now.
1596	fn touched(&self, touched: &Touched) {
1597	let mut current = self.current.lock().unwrap();
1598	for (at, path) in touched.paths.iter().enumerate() {
1599	let stamp = self.image(path).and_then(|image| Stamp::of(&image).ok());
1600	match stamp.filter(|stamp| touched.stamps.get(at) == Some(&Some(digest(stamp)))) {
1601	Some(stamp) => current.insert(path.clone(), stamp),
1602	None => current.remove(path),
1603	};
1604	}
1605	}
1606
1607	fn heard(self: &Arc<Self>, source: &Arc<super::Shared>, event: Event) {
1608	match event {
1609	Event::Met(hello, line) if hello.serves == Some(self.share) => {
1610	let mut host = self.host.lock().unwrap();
1611	let retired = if source.connected(hello) {
1612	host.replace((Arc::clone(hello), line.clone()))
1613	} else {
1614	None
1615	};
1616	drop(host);
1617	drop(retired);
1618	}
1619	Event::Left(hello) => {
1620	let (retired, presence, pending, reports) = {
1621	let mut host = self.host.lock().unwrap();
1622	if !host
1623	.as_ref()
1624	.is_some_and(|(admitted, _)| Arc::ptr_eq(admitted, hello))
1625	|| !source.gone(hello)
1626	{
1627	return;
1628	}
1629	let retired = host.take();
1630	let presence = self.presence.lock().unwrap().take();
1631	self.current.lock().unwrap().clear();
1632	let pending = std::mem::take(&mut *self.pending.lock().unwrap());
1633	let reports = self.watch.lock().unwrap().take();
1634	(retired, presence, pending, reports)
1635	};
1636	drop(retired);
1637	drop(presence);
1638	drop(pending);
1639	if let Some(reports) = reports {
1640	reports.lost();
1641	}
1642	}
1643	// A reply only reaches its waiter, so one read off a retired route still counts.
1644	Event::Frame {
1645	from,
1646	kind: kind::REPLY,
1647	body,
1648	} => {
1649	let serves = (self.host.lock().unwrap().as_ref())
1650	.is_some_and(|(host, _)| host.peer == from.peer);
1651	if serves && let Ok(reply) = minicbor::decode::<Reply>(body) {
1652	let waiting = self.pending.lock().unwrap().remove(&reply.id);
1653	if let Some(waiting) = waiting {
1654	let _ = waiting.send(reply);
1655	}
1656	}
1657	}
1658	Event::Frame { from, kind, body } => {
1659	let host = self.host.lock().unwrap();
1660	if !host
1661	.as_ref()
1662	.is_some_and(|(admitted, _)| Arc::ptr_eq(admitted, from))
1663	|| !source.connected(from)
1664	{
1665	return;
1666	}
1667	match kind {
1668	kind::WELCOME => {
1669	if let Ok(welcome) = minicbor::decode::<Welcome>(body)
1670	&& !(self.presence.lock().unwrap().as_ref())
1671	.is_some_and(|(held, _)| *held == welcome.room)
1672	{
1673	drop(host);
1674	self.join_presence(source, from, welcome.room);
1675	}
1676	}
1677	kind::DELTA => {
1678	if let Ok(delta) = minicbor::decode::<Delta>(body) {
1679	self.apply(&delta);
1680	}
1681	}
1682	kind::TOUCHED => {
1683	if let Ok(touched) = minicbor::decode::<Touched>(body) {
1684	self.touched(&touched);
1685	let reports = self.watch.lock().unwrap().clone();
1686	drop(host);
1687	if let Some(reports) = reports {
1688	reports.touched(&touched.paths);
1689	}
1690	}
1691	}
1692	kind::BYE => {
1693	if let Ok(bye) = minicbor::decode::<wire::Bye>(body) {
1694	let ended = match bye.reason.as_str() {
1695	STOPPED => Some(Ended::Stopped),
1696	"removed" => Some(Ended::Removed),
1697	_ => None,
1698	};
1699	if ended.is_some() {
1700	*self.ended.lock().unwrap() = ended;
1701	}
1702	}
1703	}
1704	_ => {}
1705	}
1706	}
1707	_ => {}
1708	}
1709	}
1710
1711	fn join_presence(
1712	self: &Arc<Self>,
1713	source: &Arc<super::Shared>,
1714	from: &Arc<Hello>,
1715	secret: [u8; 16],
1716	) {
1717	let inner = Arc::downgrade(self);
1718	let connected = Mutex::new(None);
1719	let live = Live::start_with(
1720	self.me.clone(),
1721	&Room::Notebook(secret),
1722	self.reach,
1723	self.relay.as_deref(),
1724	move |source, event| {
1725	let Some(inner) = inner.upgrade() else {
1726	return;
1727	};
1728	let reports = {
1729	let host = inner.host.lock().unwrap();
1730	let presence = inner.presence.lock().unwrap();
1731	let Some((_, live)) = presence.as_ref() else {
1732	return;
1733	};
1734	if !Arc::ptr_eq(&live.shared, source) {
1735	return;
1736	}
1737	match event {
1738	Event::Changed => {
1739	let mut connected = connected.lock().unwrap();
1740	let host = host.as_ref().map(|(host, _)| host.peer);
1741	let peers = source.peers();
1742	let present = host
1743	.filter(|host| peers.iter().any(|peer| peer.hello.peer == *host));
1744	if *connected == present {
1745	None
1746	} else {
1747	*connected = present;
1748	inner.current.lock().unwrap().clear();
1749	if present.is_some() {
1750	inner
1751	.watch
1752	.lock()
1753	.unwrap()
1754	.clone()
1755	.map(|reports| (reports, vec![String::new()]))
1756	} else {
1757	None
1758	}
1759	}
1760	}
1761	Event::Frame { from, kind, body }
1762	if host
1763	.as_ref()
1764	.is_some_and(|(host, _)| host.peer == from.peer)
1765	&& source.connected(from) =>
1766	{
1767	match kind {
1768	kind::DELTA => {
1769	if let Ok(delta) = minicbor::decode::<Delta>(body) {
1770	inner.apply(&delta);
1771	}
1772	None
1773	}
1774	kind::TOUCHED => {
1775	if let Ok(touched) = minicbor::decode::<Touched>(body) {
1776	inner.touched(&touched);
1777	inner
1778	.watch
1779	.lock()
1780	.unwrap()
1781	.clone()
1782	.map(|reports| (reports, touched.paths))
1783	} else {
1784	None
1785	}
1786	}
1787	_ => None,
1788	}
1789	}
1790	_ => None,
1791	}
1792	};
1793	if let Some((reports, paths)) = reports {
1794	reports.touched(&paths);
1795	}
1796	(inner.events)();
1797	},
1798	);
1799	let live = match live {
1800	Ok(live) => live,
1801	Err(error) => return eprintln!("Live Share: could not join presence: {error}"),
1802	};
1803	let here = self.here.lock().unwrap();
1804	live.set_presence(here.clone());
1805	let refresh = Arc::clone(&live.shared);
1806	let mut candidate = Some(live);
1807	let (retired, paths, reports) = {
1808	let host = self.host.lock().unwrap();
1809	if host
1810	.as_ref()
1811	.is_some_and(|(admitted, _)| Arc::ptr_eq(admitted, from))
1812	&& source.connected(from)
1813	{
1814	let mut presence = self.presence.lock().unwrap();
1815	if presence.as_ref().is_some_and(|(held, _)| *held == secret) {
1816	(None, Vec::new(), None)
1817	} else {
1818	let retired = presence.replace((secret, candidate.take().unwrap()));
1819	let paths = self
1820	.current
1821	.lock()
1822	.unwrap()
1823	.drain()
1824	.map(|(path, _)| path)
1825	.collect::<Vec<_>>();
1826	let reports = self.watch.lock().unwrap().clone();
1827	(retired, paths, reports)
1828	}
1829	} else {
1830	(None, Vec::new(), None)
1831	}
1832	};
1833	drop(here);
1834	drop(retired);
1835	if candidate.is_none() {
1836	(refresh.events)(&refresh, Event::Changed);
1837	}
1838	drop(candidate);
1839	if let Some(reports) = reports {
1840	reports.touched(&paths);
1841	}
1842	}
1843	}
1844
1845	/// A section a Live Share host serves, as a guest's replica publishes to it.
1846	pub struct HostedRemote {
1847	guest: Arc<Guest>,
1848	path: String,
1849	/// The stamp last asked for, which an image the guest holds may already have.
1850	seen: Option<Stamp>,
1851	rejected: bool,
1852	#[cfg(target_arch = "wasm32")]
1853	pending: Option<web::Pending>,
1854	}
1855
1856	impl HostedRemote {
1857	pub fn new(guest: &Arc<Guest>, path: &str) -> Self {
1858	Self {
1859	guest: Arc::clone(guest),
1860	path: path.to_owned(),
1861	seen: None,
1862	rejected: false,
1863	#[cfg(target_arch = "wasm32")]
1864	pending: None,
1865	}
1866	}
1867	}
1868
1869	#[cfg(not(target_arch = "wasm32"))]
1870	impl crate::Remote for HostedRemote {
1871	fn accepts_edits(&self) -> bool {
1872	!self.rejected
1873	&& self
1874	.guest
1875	.host()
1876	.is_some_and(|host| host.ops == Some(1) && host.kinds.contains(&kind::EDITS))
1877	}
1878
1879	fn publish_edits(
1880	&mut self,
1881	transaction: &Transaction,
1882	edits: &[crate::PendingEdit],
1883	revisions: &BTreeMap<onestore::ExGuid, onestore::ExGuid>,
1884	) -> std::result::Result<(), CommitError> {
1885	let result =
1886	crate::task::ready(
1887	self.guest
1888	.edits_async(&self.path, transaction, edits, revisions),
1889	)
1890	.map_err(|error| CommitError {
1891	state: CommitState::NotCommitted,
1892	error,
1893	})?;
1894	if result.is_ok() {
1895	self.seen = None;
1896	}
1897	if result
1898	.as_ref()
1899	.is_err_and(|error| error.state == CommitState::NotCommitted)
1900	{
1901	self.rejected = true;
1902	self.guest.inner.current.lock().unwrap().remove(&self.path);
1903	}
1904	result
1905	}
1906
1907	fn read(&mut self) -> io::Result<Vec<u8>> {
1908	self.rejected = false;
1909	if let Some(image) = (self.seen.as_ref()).and_then(|seen| self.guest.held(&self.path, seen))
1910	{
1911	return Ok(image);
1912	}
1913	self.guest.read(kind::READ, &self.path, LIMIT)
1914	}
1915
1916	fn stamp(&mut self) -> io::Result<Stamp> {
1917	let stamp = self.guest.stamp(&self.path)?;
1918	self.seen = Some(stamp.clone());
1919	Ok(stamp)
1920	}
1921
1922	fn publish(&mut self, transaction: &Transaction) -> std::result::Result<(), CommitError> {
1923	if let Err(error) = self.guest.commit(&self.path, transaction) {
1924	// The host's image moved on, and its report may not have come yet.
1925	self.guest.inner.current.lock().unwrap().remove(&self.path);
1926	return Err(error);
1927	}
1928	self.guest.inner.published(&self.path, transaction);
1929	Ok(())
1930	}
1931
1932	fn confirm(&mut self, base: &Stamp) -> std::result::Result<(), CommitError> {
1933	self.guest.confirm(&self.path, base)
1934	}
1935	}
1936
1937	/// A notebook a Live Share host serves, as a guest's `session::Notebook` reaches it. Its
1938	/// folders as last listed are kept at `listed`, so the notebook opens while the host can't
1939	/// be reached.
1940	pub struct Hosted {
1941	guest: Arc<Guest>,
1942	listed: PathBuf,
1943	}
1944
1945	/// Each folder's entries as last listed: name, `WireEntry::kind`, size and modified time.
1946	type Listings = BTreeMap<String, Vec<(String, u8, u64, u64)>>;
1947
1948	impl Hosted {
1949	pub(crate) fn new(guest: Arc<Guest>, listed: PathBuf) -> Self {
1950	#[cfg(target_arch = "wasm32")]
1951	{
1952	*guest.inner.catalog.lock().unwrap() = Some(listed.clone());
1953	}
1954	Self { guest, listed }
1955	}
1956	}
1957
1958	/// The host's folders, or as they were last listed while it can't be reached.
1959	#[cfg(not(target_arch = "wasm32"))]
1960	struct Source<'a> {
1961	guest: &'a Guest,
1962	kept: Listings,
1963	listed: Listings,
1964	}
1965
1966	#[cfg(not(target_arch = "wasm32"))]
1967	impl discover::Source for Source<'_> {
1968	fn entries(&mut self, path: &str, limit: usize) -> io::Result<Vec<discover::Entry>> {
1969	let entries = match self.guest.entries(path) {
1970	Ok(entries) => entries,
1971	Err(error) if error.kind() == io::ErrorKind::NotConnected => self
1972	.kept
1973	.get(path)
1974	.ok_or(error)?
1975	.iter()
1976	.map(|(name, kind, size, modified)| {
1977	entry(&WireEntry {
1978	name: name.clone(),
1979	kind: *kind,
1980	size: *size,
1981	modified: *modified,
1982	})
1983	})
1984	.collect(),
1985	Err(error) => return Err(error),
1986	};
1987	if entries.len() > limit {
1988	return Err(io::ErrorKind::FileTooLarge.into());
1989	}
1990	self.listed.insert(
1991	path.to_owned(),
1992	entries
1993	.iter()
1994	.map(|listed| {
1995	let wire = wire_entry(listed);
1996	(wire.name, wire.kind, wire.size, wire.modified)
1997	})
1998	.collect(),
1999	);
2000	Ok(entries)
2001	}
2002
2003	fn read(&mut self, path: &str, limit: usize) -> io::Result<Vec<u8>> {
2004	self.guest.read(kind::READ, path, limit)
2005	}
2006
2007	fn read_asset(&mut self, path: &str, limit: usize) -> io::Result<Vec<u8>> {
2008	self.guest.read(kind::READ_FILE, path, limit)
2009	}
2010	}
2011
2012	#[cfg(not(target_arch = "wasm32"))]
2013	impl Storage for Hosted {
2014	fn discover(
2015	&self,
2016	cache: &mut discover::Cache,
2017	limits: discover::Limits,
2018	) -> Result<discover::Folder> {
2019	let kept = crate::fs::read(&self.listed)
2020	.ok()
2021	.and_then(|bytes| serde_json::from_slice(&bytes).ok())
2022	.unwrap_or_default();
2023	let mut source = Source {
2024	guest: &self.guest,
2025	kept,
2026	listed: Listings::new(),
2027	};
2028	let folder = cache.discover(&mut source, limits)?;
2029	if let Ok(json) = serde_json::to_vec(&source.listed) {
2030	let _ = crate::fs::write(&self.listed, json);
2031	}
2032	Ok(folder)
2033	}
2034
2035	fn location(&self) -> String {
2036	self.guest.location()
2037	}
2038
2039	fn entries(&self, folder: &str) -> io::Result<Vec<discover::Entry>> {
2040	self.guest.entries(folder)
2041	}
2042
2043	fn exists(&self, path: &str) -> bool {
2044	let request = Request {
2045	path: path.to_owned(),
2046	..Request::default()
2047	};
2048	self.guest
2049	.ask(kind::EXISTS, request)
2050	.is_ok_and(|reply| reply.exists == Some(true))
2051	}
2052
2053	fn stamp(&self, path: &str) -> io::Result<Stamp> {
2054	self.guest.stamp(path)
2055	}
2056
2057	fn read(&self, path: &str) -> Result<Vec<u8>> {
2058	Ok(self.guest.read(kind::READ, path, LIMIT)?)
2059	}
2060
2061	fn read_file(&self, path: &str, limit: usize) -> Result<Vec<u8>> {
2062	Ok(self.guest.read(kind::READ_FILE, path, limit)?)
2063	}
2064
2065	fn create(&self, path: &str, bytes: &[u8]) -> Result<()> {
2066	if bytes.len() > LIMIT {
2067	return Err(io::Error::from(io::ErrorKind::FileTooLarge).into());
2068	}
2069	let mut request = Request {
2070	path: path.to_owned(),
2071	..Request::default()
2072	};
2073	self.guest
2074	.carry(&mut request, bytes.to_vec())
2075	.map_err(Failed::io)?;
2076	self.guest.verb(kind::CREATE, path, request)
2077	}
2078
2079	fn create_directory(&self, path: &str) -> Result<()> {
2080	self.guest
2081	.verb(kind::CREATE_DIRECTORY, path, Request::default())
2082	}
2083
2084	fn hide(&self, path: &str) -> Result<()> {
2085	self.guest.verb(kind::HIDE, path, Request::default())
2086	}
2087
2088	fn rename(&self, from: &str, to: &str) -> Result<()> {
2089	let request = Request {
2090	to: Some(to.to_owned()),
2091	..Request::default()
2092	};
2093	self.guest.verb(kind::RENAME, from, request)
2094	}
2095
2096	fn rename_root(&self, _: &str, _: &[String]) -> Result<String> {
2097	Err(refused(
2098	io::ErrorKind::Unsupported,
2099	"Only the computer sharing this notebook can rename its folder",
2100	))
2101	}
2102
2103	fn replace(&self, from: &str, to: &str) -> Result<()> {
2104	let request = Request {
2105	to: Some(to.to_owned()),
2106	..Request::default()
2107	};
2108	self.guest.verb(kind::REPLACE, from, request)
2109	}
2110
2111	fn delete(&self, path: &str) -> Result<()> {
2112	self.guest.verb(kind::DELETE, path, Request::default())
2113	}
2114
2115	fn place(&self, path: &str, ancestor: [u8; 16], name: &str) -> Result<()> {
2116	let request = Request {
2117	ancestor: Some(ancestor),
2118	name: Some(name.to_owned()),
2119	..Request::default()
2120	};
2121	self.guest.verb(kind::PLACE, path, request)
2122	}
2123
2124	fn commit(&self, path: &str, transaction: &Transaction) -> Result<()> {
2125	Ok(self.guest.commit(path, transaction)?)
2126	}
2127
2128	fn confirm(&self, path: &str, base: &Stamp) -> std::result::Result<(), CommitError> {
2129	self.guest.confirm(path, base)
2130	}
2131
2132	fn supersede(&self, path: &str, base: &Stamp, with: &str) -> Result<()> {
2133	let request = Request {
2134	to: Some(with.to_owned()),
2135	stamp: Some(wire::WireStamp::from(base)),
2136	..Request::default()
2137	};
2138	self.guest.verb(kind::SUPERSEDE, path, request)
2139	}
2140	}
2141
2142	#[cfg(test)]
2143	mod tests {
2144	use super::*;
2145
2146	#[test]
2147	fn refused_uploads_release_their_bytes_and_share_one_guest_budget() {
2148	let directory = tempfile::tempdir().unwrap();
2149	std::fs::create_dir(directory.path().join("notebook")).unwrap();
2150	let served = Arc::new(Served {
2151	storage: crate::session::Notebook::open(
2152	directory.path().join("notebook"),
2153	directory.path().join("cache"),
2154	)
2155	.unwrap()
2156	.into_storage(),
2157	images: Mutex::default(),
2158	snapshots: Mutex::default(),
2159	puts: Mutex::default(),
2160	guests: Mutex::default(),
2161	writers: Mutex::default(),
2162	host: Mutex::default(),
2163	room: Mutex::default(),
2164	});
2165	let peer = [1; 16];
2166	let put = |handle, offset, bytes| {
2167	served.answer(
2168	&peer,
2169	kind::PUT,
2170	Request {
2171	handle: Some(handle),
2172	offset: Some(offset),
2173	bytes: Some(bytes),
2174	..Request::default()
2175	},
2176	)
2177	};
2178	put(1, 0, vec![1; 3]).unwrap();
2179	put(2, 0, vec![2; 2]).unwrap();
2180	assert!(put(1, 2, vec![3]).is_err());
2181	assert_eq!(
2182	served.puts.lock().unwrap().get(&(peer, 2)).unwrap(),
2183	&[2; 2]
2184	);
2185	assert!(!served.puts.lock().unwrap().contains_key(&(peer, 1)));
2186	put(1, 0, vec![4]).unwrap();
2187	assert!(put(1, 1, Vec::new()).is_err());
2188	served
2189	.puts
2190	.lock()
2191	.unwrap()
2192	.insert((peer, 3), vec![0; LIMIT - 2]);
2193	assert!(put(2, 2, vec![5]).is_err());
2194	assert!(!served.puts.lock().unwrap().contains_key(&(peer, 2)));
2195	assert!(put(3, (LIMIT - 2) as u64, vec![6; 3]).is_err());
2196	assert!(served.puts.lock().unwrap().is_empty());
2197	put(4, 0, vec![7]).unwrap();
2198	}
2199
2200	/// A commit's writes, found by comparing images, rebuild the image after it from the one
2201	/// before, and a change touching most of the file is left to be read whole.
2202	#[test]
2203	fn deltas_rebuild_the_image_after_a_commit() {
2204	let before: Vec<u8> = (0..20_000u32).map(|at| (at % 251) as u8).collect();
2205	let mut after = before.clone();
2206	after[3] ^= 1;
2207	after[5000..5010].fill(9);
2208	after[5050] ^= 1;
2209	after[17_000] ^= 1;
2210	after.extend_from_slice(&[7; 3000]);
2211	let writes = delta(&before, &after).unwrap();
2212	assert_eq!(
2213	writes.len(),
2214	4,
2215	"the header, two runs merged as one, one more, the tail"
2216	);
2217	assert_eq!(
2218	written(&before, after.len() as u64, &writes).unwrap(),
2219	after
2220	);
2221	let rewritten: Vec<u8> = before.iter().map(|byte| byte ^ 1).collect();
2222	assert!(delta(&before, &rewritten).is_none());
2223	assert!(delta(&before, &before[..10_000]).is_none());
2224	let outside = [Written {
2225	offset: after.len() as u64,
2226	bytes: vec![1],
2227	}];
2228	assert!(written(&before, after.len() as u64, &outside).is_none());
2229	}
2230
2231	#[test]
2232	fn differential_reads_include_metadata_in_the_chunk_budget() {
2233	let before = vec![0; CHUNK * 4];
2234	let mut after = before.clone();
2235	after.resize(before.len() + CHUNK - 1024, 1);
2236	assert!(delta(&before, &after).is_none());
2237	after.truncate(before.len() + CHUNK / 2);
2238	let writes = delta(&before, &after).unwrap();
2239	assert!(minicbor::to_vec(&writes).unwrap().len() <= CHUNK);
2240	assert_eq!(
2241	written(&before, after.len() as u64, &writes).unwrap(),
2242	after
2243	);
2244	}
2245	}