1use super::*;
2use crate::{PendingEdit, merge};
3use onestore::{Arena, ExGuid, Section, op::Edit};
4use std::collections::BTreeSet;
5
6#[derive(serde::Serialize, serde::Deserialize)]
7#[serde(deny_unknown_fields)]
8pub(super) struct Edits {
9 pub edits: Vec<(String, Edit)>,
10 pub revisions: BTreeMap<ExGuid, ExGuid>,
11}
12
13pub(super) fn encode(edits: &Edits, limit: usize) -> io::Result<Vec<u8>> {
14 struct Buffer {
15 bytes: Vec<u8>,
16 limit: usize,
17 }
18 impl io::Write for Buffer {
19 fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
20 if bytes.len() > self.limit.saturating_sub(self.bytes.len()) {
21 return Err(io::ErrorKind::FileTooLarge.into());
22 }
23 self.bytes.extend_from_slice(bytes);
24 Ok(bytes.len())
25 }
26
27 fn flush(&mut self) -> io::Result<()> {
28 Ok(())
29 }
30 }
31 let mut buffer = Buffer {
32 bytes: Vec::new(),
33 limit,
34 };
35 serde_json::to_writer(&mut buffer, edits).map_err(|error| {
36 io::Error::new(
37 error.io_error_kind().unwrap_or(io::ErrorKind::InvalidData),
38 error,
39 )
40 })?;
41 Ok(buffer.bytes)
42}
43
44pub(super) struct Waiting {
45 peer: [u8; 16],
46 request: Request,
47 reply: mpsc::Sender<Reply>,
48}
49
50fn retry(message: &str) -> Error {
51 Error::Remote(CommitError {
52 state: CommitState::NotCommitted,
53 error: io::Error::new(io::ErrorKind::ResourceBusy, message.to_owned()),
54 })
55}
56
57impl Served {
58 pub(super) fn batch(self: &Arc<Self>, peer: &[u8; 16], request: Request) -> Result<Reply> {
59 let path = request.path.clone();
60 if !self.writers.lock().unwrap().contains_key(&path) {
61 let image = self.image(&path)?;
62 let arena = Arena::default();
63 Section::open(&arena, (*image).clone())?;
64 }
65 let mut writers = self.writers.lock().unwrap();
66 let writer = writers.entry(path.clone()).or_insert_with(|| {
67 let (send, receive) = mpsc::sync_channel::<Waiting>(64);
68 let served = Arc::downgrade(self);
69 thread::spawn(move || {
70 while let Ok(first) = receive.recv() {
71 let mut waiting = vec![first];
72 if let Ok(next) = receive.recv_timeout(Duration::from_millis(5)) {
73 waiting.push(next);
74 }
75 waiting.extend(receive.try_iter().take(62));
76 let Some(served) = served.upgrade() else {
77 return;
78 };
79 served.write_batches(&path, waiting);
80 }
81 });
82 send
83 });
84 let (reply, receive) = mpsc::channel();
85 writer
86 .try_send(Waiting {
87 peer: *peer,
88 request,
89 reply,
90 })
91 .map_err(|_| retry("The section's writer is busy"))?;
92 drop(writers);
93 receive.recv_timeout(TIMEOUT).map_err(|_| {
94 Error::Remote(CommitError {
95 state: CommitState::Unknown,
96 error: io::Error::new(
97 io::ErrorKind::TimedOut,
98 "The section's writer did not answer",
99 ),
100 })
101 })
102 }
103
104 fn write_batches(&self, path: &str, waiting: Vec<Waiting>) {
105 let before = match self.image(path) {
106 Ok(image) => image,
107 Err(error) => {
108 for batch in waiting {
109 let _ = batch.reply.send(failed(0, retry(&error.to_string())));
110 }
111 return;
112 }
113 };
114 let mut image = Arc::clone(&before);
115 let mut combined: Option<Transaction> = None;
116 let mut accepted = Vec::new();
117 for batch in waiting {
118 match self.prepare_batch(path, &image, &batch) {
119 Ok(Some(transaction)) => {
120 let mut next = (*image).clone();
121 let extended =
122 transaction
123 .apply(&mut next)
124 .and_then(|()| match &mut combined {
125 Some(combined) => combined.extend(transaction),
126 None => {
127 combined = Some(transaction);
128 Ok(())
129 }
130 });
131 match extended {
132 Ok(()) => {
133 image = Arc::new(next);
134 accepted.push(batch);
135 }
136 Err(error) => {
137 let _ = batch.reply.send(failed(0, error.into()));
138 }
139 }
140 }
141 Ok(None) => accepted.push(batch),
142 Err(error) => {
143 let _ = batch.reply.send(failed(0, error));
144 }
145 }
146 }
147 if accepted.is_empty() {
148 return;
149 }
150 let result = match &combined {
151 Some(transaction) => self.storage.commit(path, transaction),
152 None => self
153 .storage
154 .confirm(path, &Stamp::of(&image).expect("section stamp"))
155 .map_err(Error::from)
156 .and_then(|()| {
157 image = self.image(path).map_err(|error| CommitError {
158 state: CommitState::Committed,
159 error: match error {
160 Error::Io(error) | Error::RemoteIo(error) => error,
161 Error::Remote(error) => error.error,
162 error => io::Error::other(error),
163 },
164 })?;
165 Ok(())
166 }),
167 };
168 if result.is_ok() {
169 self.tell_delta(path, &before, &image, None);
170 self.keep(path, Arc::clone(&image));
171 self.changed(&[path.to_owned()]);
172 }
173 for batch in accepted {
174 let reply = match &result {
175 Ok(()) => Reply {
176 stamp: Some((&Stamp::of(&image).expect("section stamp")).into()),
177 ..Reply::default()
178 },
179 Err(Error::Remote(error)) => failed(
180 0,
181 Error::Remote(CommitError {
182 state: error.state,
183 error: io::Error::new(error.error.kind(), error.error.to_string()),
184 }),
185 ),
186 Err(error) => failed(0, retry(&error.to_string())),
187 };
188 let _ = batch.reply.send(reply);
189 }
190 }
191
192 fn prepare_batch(
193 &self,
194 path: &str,
195 image: &[u8],
196 batch: &Waiting,
197 ) -> Result<Option<Transaction>> {
198 if !self.guests.lock().unwrap().contains_key(&batch.peer) {
199 return Err(refused(
200 io::ErrorKind::PermissionDenied,
201 "The device is no longer connected",
202 ));
203 }
204 let stamp: Stamp = batch
205 .request
206 .stamp
207 .as_ref()
208 .ok_or_else(|| retry("No batch base"))?
209 .try_into()?;
210 let edits: Edits = serde_json::from_slice(&self.carried(&batch.peer, &batch.request)?)
211 .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
212 if edits.revisions.is_empty()
213 || edits
214 .revisions
215 .values()
216 .any(|revision| revision.guid == [0; 16])
217 {
218 return Err(retry("No batch revisions"));
219 }
220 let store = Store::parse(image)?;
221 let index = RevisionIndex::parse(&store)?;
222 if edits.revisions.iter().all(|(space, revision)| {
223 index
224 .spaces
225 .get(space)
226 .is_some_and(|space| space.revisions.contains_key(revision))
227 }) {
228 return Ok(None);
229 }
230 let names = edits
231 .revisions
232 .iter()
233 .filter(|(space, revision)| {
234 !index
235 .spaces
236 .get(space)
237 .is_some_and(|space| space.revisions.contains_key(revision))
238 })
239 .map(|(space, revision)| (*space, *revision))
240 .collect();
241 let arena = Arena::default();
242 let mut next = Section::open(&arena, image.to_vec())?;
243 let queued: Vec<PendingEdit> = edits
244 .edits
245 .into_iter()
246 .enumerate()
247 .map(|(id, (author, edit))| PendingEdit {
248 id: id as u64,
249 author,
250 edit,
251 })
252 .collect();
253 if Stamp::of(image)? == stamp {
254 for edit in &queued {
255 next.apply(&edit.author, &edit.edit)?;
256 }
257 } else {
258 let base = self
259 .images
260 .lock()
261 .unwrap()
262 .iter()
263 .find_map(|(held, at, image)| {
264 (held == path && *at == stamp).then(|| Arc::clone(image))
265 })
266 .ok_or_else(|| retry("The batch's base is no longer held"))?;
267 let old_arena = Arena::default();
268 let mut old = Section::open(&old_arena, (*base).clone())?;
269 if old.root() != next.root() {
270 return Err(retry("The batch belongs to another section"));
271 }
272 let merged = merge::rebase(&mut old, &mut next, &queued, &BTreeSet::new())?;
273 if !merged.conflicts.is_empty() {
274 let local_arena = Arena::default();
275 let mut local = Section::open(&local_arena, (*base).clone())?;
276 for edit in &queued {
277 local.apply(&edit.author, &edit.edit)?;
278 }
279 for (space, (author, objects)) in &merged.conflicts {
280 let page = local.page(*space)?;
281 if next.page(*space).is_ok_and(|remote| remote == page) {
282 return Err(retry("The batch's edits already landed"));
283 }
284 merge::conflict_page(
285 &mut next,
286 &mut local,
287 &merged.moved,
288 *space,
289 &page,
290 author,
291 objects,
292 crate::now(),
293 None,
294 )?;
295 }
296 }
297 }
298 let transaction = next.seal_as(&names)?;
299 if edits.revisions.iter().any(|(space, revision)| {
300 !next
301 .newest()
302 .any(|(held, at)| held == *space && at == *revision)
303 }) {
304 return Err(refused(
305 io::ErrorKind::Unsupported,
306 "The batch needs a guest's commit",
307 ));
308 }
309 Ok(transaction)
310 }
311}
312
313#[cfg(test)]
314mod tests {
315 use super::*;
316 use onestore::{
317 op::{Op, PageOp},
318 page::{Attachment, PageObject},
319 };
320
321 #[cfg(not(target_arch = "wasm32"))]
322 #[test]
323 fn invalid_sections_do_not_retain_writers_and_can_be_repaired() {
324 let directory = tempfile::tempdir().unwrap();
325 let folder = directory.path().join("notebook");
326 std::fs::create_dir(&folder).unwrap();
327 let served = Arc::new(Served {
328 storage: crate::session::Notebook::open(&folder, directory.path().join("cache"))
329 .unwrap()
330 .into_storage(),
331 images: Mutex::default(),
332 snapshots: Mutex::default(),
333 puts: Mutex::default(),
334 guests: Mutex::default(),
335 writers: Mutex::default(),
336 host: Mutex::default(),
337 room: Mutex::default(),
338 });
339 let peer = [1; 16];
340 let (line, _outbound) = mpsc::channel();
341 let (queue, _requests) = mpsc::sync_channel(1);
342 served.guests.lock().unwrap().insert(
343 peer,
344 Admitted {
345 hello: Arc::new(Hello {
346 peer,
347 ..Hello::new("Guest".into(), None).unwrap()
348 }),
349 line: crate::live::Line(line),
350 queue,
351 starts: BURST,
352 counted: Instant::now(),
353 held: Vec::new(),
354 },
355 );
356 let toc = onestore::create_table_of_contents("Invalid.onetoc2", &[]).unwrap();
357 {
358 let store = Store::parse(&toc).unwrap();
359 RevisionIndex::parse(&store).unwrap();
360 }
361 let arena = Arena::default();
362 assert!(Section::open(&arena, toc.clone()).is_err());
363 for image in [None, Some(vec![0; 1024]), Some(toc)] {
364 for at in 0..128 {
365 let path = format!("Missing{at}.one");
366 if let Some(image) = &image {
367 std::fs::write(folder.join(&path), image).unwrap();
368 }
369 let reply = served.handle(
370 &peer,
371 kind::EDITS,
372 &minicbor::to_vec(Request {
373 path,
374 ..Request::default()
375 })
376 .unwrap(),
377 );
378 assert!(reply.failure.is_some());
379 }
380 assert!(served.writers.lock().unwrap().is_empty());
381 }
382
383 let path = "Missing0.one";
384 let image = onestore::create_section(path, "Original text", "Fixture").unwrap();
385 std::fs::write(folder.join(path), &image).unwrap();
386 let store = Store::parse(&image).unwrap();
387 let index = RevisionIndex::parse(&store).unwrap();
388 let document = onestore::document::Document::parse(&index).unwrap();
389 let (space, text) = document
390 .spaces
391 .iter()
392 .find_map(|(space, object)| {
393 object.revisions[&object.contexts[&ExGuid::default()]]
394 .nodes
395 .iter()
396 .find_map(|(id, node)| {
397 matches!(node.kind, onestore::document::Kind::RichText { .. })
398 .then_some((*space, *id))
399 })
400 })
401 .unwrap();
402 let edit = Edit {
403 at: 134_000_000_000_000_000,
404 ops: vec![Op::Page {
405 space,
406 op: PageOp::Text {
407 text,
408 range: 13..13,
409 with: "X".into(),
410 },
411 }],
412 };
413 let arena = Arena::default();
414 let mut section = Section::open(&arena, image).unwrap();
415 section.apply("Alice", &edit).unwrap();
416 let transaction = section.seal().unwrap().unwrap();
417 let request = Request {
418 path: path.into(),
419 stamp: Some(transaction.base().into()),
420 bytes: Some(
421 encode(
422 &Edits {
423 edits: vec![("Alice".into(), edit)],
424 revisions: section.newest().filter(|(at, _)| *at == space).collect(),
425 },
426 LIMIT,
427 )
428 .unwrap(),
429 ),
430 ..Request::default()
431 };
432 let first = served.batch(&peer, request.clone()).unwrap();
433 assert!(first.failure.is_none(), "{:?}", first.failure);
434 let committed = std::fs::read(folder.join(path)).unwrap();
435 let stamp = Stamp::of(&committed).unwrap();
436 assert_eq!(
437 Stamp::try_from(first.stamp.as_ref().unwrap()).unwrap(),
438 stamp
439 );
440 let read_arena = Arena::default();
441 let reread = Section::open(&read_arena, committed.clone()).unwrap();
442 assert_eq!(reread.page(space).unwrap(), section.page(space).unwrap());
443 let repeated = served.batch(&peer, request).unwrap();
444 assert!(repeated.failure.is_none(), "{:?}", repeated.failure);
445 let replayed = std::fs::read(folder.join(path)).unwrap();
446 let confirmed = Stamp::of(&replayed).unwrap();
447 assert_eq!(
448 Stamp::try_from(repeated.stamp.as_ref().unwrap()).unwrap(),
449 confirmed
450 );
451 {
452 let images = served.images.lock().unwrap();
453 let (_, held, image) = images.iter().find(|(held, _, _)| held == path).unwrap();
454 assert_eq!(*held, confirmed);
455 assert_eq!(**image, replayed);
456 }
457 assert_eq!(&replayed[1024..], &committed[1024..]);
458 let replay_arena = Arena::default();
459 let replay = Section::open(&replay_arena, replayed).unwrap();
460 assert_eq!(replay.page(space).unwrap(), section.page(space).unwrap());
461 assert_eq!(served.writers.lock().unwrap().len(), 1);
462 }
463
464 #[test]
465 fn attachment_encoding_stops_at_the_upload_budget() {
466 let id = ExGuid {
467 guid: [1; 16],
468 n: 1,
469 };
470 let bytes: Arc<[u8]> = (0..4096)
471 .map(|at| (at % 251) as u8)
472 .collect::<Vec<_>>()
473 .into();
474 let edits = Edits {
475 edits: vec![(
476 "Ada".into(),
477 Edit {
478 at: 0,
479 ops: vec![Op::Page {
480 space: id,
481 op: PageOp::Add {
482 object: PageObject::Attachment(Attachment {
483 id,
484 filename: "payload.bin".into(),
485 source_path: None,
486 size: None,
487 layout: Default::default(),
488 bytes: Some(Arc::clone(&bytes)),
489 preview: None,
490 recording: None,
491 tags: Vec::new(),
492 }),
493 before: None,
494 },
495 }],
496 },
497 )],
498 revisions: BTreeMap::new(),
499 };
500 let encoded = serde_json::to_vec(&edits).unwrap();
501 assert_eq!(encode(&edits, encoded.len()).unwrap(), encoded);
502 let decoded: Edits = serde_json::from_slice(&encoded).unwrap();
503 assert_eq!(decoded.edits, edits.edits);
504 assert_eq!(
505 encode(&edits, encoded.len() - 1).unwrap_err().kind(),
506 io::ErrorKind::FileTooLarge
507 );
508 assert_eq!(
509 encode(&edits, 128).unwrap_err().kind(),
510 io::ErrorKind::FileTooLarge
511 );
512 }
513}