1// One Win32 call, where std has no safe form: `session`'s hidden attribute.
2#![deny(unsafe_code)]
3#![doc = include_str!("../README.md")]
4
5pub mod discover;
6pub mod fs;
7#[cfg(feature = "live")]
8#[cfg_attr(target_arch = "wasm32", path = "live/web.rs")]
9pub mod live;
10#[cfg(feature = "smb")]
11pub mod smb;
12
13use fs::OpenOptions;
14use onestore::{
15 ExGuid,
16 op::{Edit, OpError},
17 page::Page,
18 protected::Key,
19};
20use rusqlite::{Connection, OpenFlags, TransactionBehavior, params};
21use std::{
22 io,
23 path::{Path, PathBuf},
24 sync::{Arc, Mutex, MutexGuard, mpsc},
25 time::Duration,
26};
27
28mod assets;
29mod background;
30mod base;
31pub mod location;
32mod merge;
33mod migrate;
34pub mod package;
35pub mod provider;
36mod queue;
37mod recovery;
38mod resolve;
39pub use resolve::Version;
40mod schema;
41pub mod session;
42pub use recovery::{Recovery, RecoverySummary};
43pub mod sidecar;
44mod sync;
45pub use sync::{EditStatus, Remote, Synced};
46mod task;
47mod worker;
48mod working;
49#[cfg(feature = "smb")]
50pub use smb::SmbRemote;
51pub use worker::SyncWorker;
52
53/// The largest file a notebook session or Live Share reads or writes whole.
54pub const MAX_FILE_BYTES: usize = 256 << 20;
55
56#[derive(Debug, thiserror::Error)]
57pub enum Error {
58 #[error(transparent)]
59 Database(#[from] rusqlite::Error),
60 #[error(transparent)]
61 Io(#[from] io::Error),
62 #[error(transparent)]
63 Document(#[from] onestore::Error),
64 #[error(transparent)]
65 Remote(#[from] onestore::CommitError),
66 #[error(transparent)]
67 RemoteIo(io::Error),
68 #[error(transparent)]
69 Discovery(#[from] discover::Error),
70 /// The section refused an edit; the pages it names are as they were.
71 #[error(transparent)]
72 Rejected(#[from] OpError),
73 #[error("External payload identity now refers to different bytes")]
74 AssetChanged,
75 /// A password-protected section's password did not match, or its format is unknown.
76 #[error(transparent)]
77 Protected(#[from] onestore::protected::Error),
78}
79
80impl Error {
81 /// Whether the replica is open elsewhere, as in another section session of this process.
82 pub fn busy(&self) -> bool {
83 matches!(self, Self::Database(rusqlite::Error::SqliteFailure(error, _))
84 if error.code == rusqlite::ErrorCode::DatabaseBusy)
85 }
86}
87
88type Result<T> = std::result::Result<T, Error>;
89
90const APPLICATION_ID: u32 = 0x4f4e454f;
91
92/// A page as a move copies it (`Replica::moving`), with its conflict pages and versions, which
93/// stay behind, read together.
94#[derive(Debug, PartialEq)]
95pub struct Moving {
96 space: ExGuid,
97 page: Page,
98 conflicts: Vec<onestore::ConflictPage>,
99 versions: Vec<onestore::PageVersion>,
100}
101
102impl Moving {
103 pub fn page(&self) -> &Page {
104 &self.page
105 }
106}
107
108/// A locally durable edit; its ID remains stable across cache reopen.
109#[derive(Debug, Clone, PartialEq)]
110pub struct PendingEdit {
111 pub id: u64,
112 pub author: String,
113 pub edit: Edit,
114}
115
116/// How an uncertain attempt ends after review: `Mine` publishes it again, `Theirs` drops
117/// the whole unpublished branch.
118#[derive(Debug, Clone, Copy, PartialEq, Eq)]
119pub enum Resolution {
120 Mine,
121 Theirs,
122}
123
124/// Owns one local cache. Share this handle between threads; a second open fails busy.
125/// Edits apply on the cache's section thread, which keeps the section parsed; SQLite's
126/// exclusive connection retains ownership between local transactions.
127pub struct Replica {
128 section: Arc<working::Thread>,
129 synchronization: Mutex<()>,
130 /// The section's root object space, which names the document.
131 root: ExGuid,
132}
133
134/// Edits that sections of this process hold in memory because their cache refused them for now,
135/// a dropped replica's included, which its section thread keeps writing.
136static UNWRITTEN: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
137
138/// How many edits wait in memory for a cache to write them; quitting loses them.
139pub fn unwritten() -> usize {
140 UNWRITTEN.load(std::sync::atomic::Ordering::Relaxed)
141}
142
143/// Runs once the replica is released and its cache closed, as `Background::hold` asks.
144#[derive(Default)]
145struct Released(Mutex<Option<Box<dyn FnOnce() + Send>>>);
146
147impl Drop for Released {
148 fn drop(&mut self) {
149 if let Some(released) = self.0.get_mut().ok().and_then(Option::take) {
150 released();
151 }
152 }
153}
154
155impl Replica {
156 /// Seeds a new cache from a validated section image, refusing any existing path.
157 /// An initialization error preserves the created file for inspection.
158 pub fn create(path: impl AsRef<Path>, source: &[u8]) -> Result<Self> {
159 Self::seed(path.as_ref(), source, None)?;
160 Self::start(cache_connection(path.as_ref())?, None)
161 }
162
163 /// Opens the cache at `path`, a protected section's under `key`, first creating it from
164 /// the image `source` reads where there is none. A cache another thread creates meanwhile,
165 /// as the background makes an offline copy, is opened instead.
166 pub fn open_or_create(
167 path: impl AsRef<Path>,
168 key: Option<&Key>,
169 source: impl FnOnce() -> Result<Vec<u8>>,
170 ) -> Result<Self> {
171 let path = path.as_ref();
172 if fs::metadata(path).is_err() {
173 match Self::seed(path, &source()?, key) {
174 Err(Error::Io(error)) if error.kind() == io::ErrorKind::AlreadyExists => {}
175 seeded => seeded?,
176 }
177 }
178 Self::open_with(path, key.cloned())
179 }
180
181 /// `create` without opening the cache it made, as for an offline copy.
182 pub(crate) fn seed(path: &Path, source: &[u8], key: Option<&Key>) -> Result<()> {
183 validate(source, key)?;
184 // Built under a name of its own and moved into place whole, so that a cache that
185 // exists is complete, and two threads making the same one never share a build.
186 static BUILDS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
187 let build = BUILDS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
188 let mut building = path.as_os_str().to_owned();
189 building.push(format!(".creating-{}-{build}", fs::process_id()));
190 let building = PathBuf::from(building);
191 let built = Self::build(&building, source);
192 let placed = built.and_then(|()| Ok(fs::rename_new(&building, path)?));
193 for suffix in ["", "-wal", "-shm"] {
194 let mut file = building.as_os_str().to_owned();
195 file.push(suffix);
196 let _ = fs::remove_file(file);
197 }
198 placed
199 }
200
201 fn build(path: &Path, source: &[u8]) -> Result<()> {
202 let mut options = OpenOptions::new();
203 options.read(true).write(true).create_new(true);
204 #[cfg(unix)]
205 {
206 use std::os::unix::fs::OpenOptionsExt;
207 options.mode(0o600);
208 }
209 drop(options.open(path)?);
210 let mut connection = cache_connection(path)?;
211 write_ahead(&connection)?;
212 let transaction = connection.transaction_with_behavior(TransactionBehavior::Exclusive)?;
213 let tables: i64 =
214 transaction.query_row("SELECT count(*) FROM sqlite_schema", [], |row| row.get(0))?;
215 let application: u32 =
216 transaction.pragma_query_value(None, "application_id", |row| row.get(0))?;
217 if application != 0 || tables != 0 {
218 return Err(io::Error::new(
219 io::ErrorKind::AlreadyExists,
220 "Cache initialization found an existing database",
221 )
222 .into());
223 }
224 transaction.pragma_update(None, "application_id", APPLICATION_ID)?;
225 transaction.pragma_update(None, "user_version", schema::VERSION)?;
226 schema::create(&transaction)?;
227 base::write(&transaction, base::Image::Base, source)?;
228 transaction.commit()?;
229 Ok(connection.close().map_err(|(_, error)| error)?)
230 }
231
232 /// Reopens an existing cache and its durable pending edits without network access.
233 /// A schema-14 cache is converted once, after exporting it to `<path>.v14-recovery`;
234 /// a conversion that cannot reproduce every queued page leaves it untouched.
235 /// Unrecognized databases and unsupported journal modes are rejected.
236 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
237 Self::open_with(path.as_ref(), None)
238 }
239
240 fn open_with(path: &Path, key: Option<Key>) -> Result<Self> {
241 let mut connection = cache_connection(path)?;
242 let application: u32 =
243 connection.pragma_query_value(None, "application_id", |row| row.get(0))?;
244 if application != APPLICATION_ID {
245 return Err(io::Error::new(io::ErrorKind::InvalidData, "Not a notebook cache").into());
246 }
247 let integrity: String = connection.query_row("PRAGMA quick_check", [], |row| row.get(0))?;
248 if integrity != "ok" {
249 return Err(
250 io::Error::new(io::ErrorKind::InvalidData, "Cache integrity check failed").into(),
251 );
252 }
253 let version: u32 = connection.pragma_query_value(None, "user_version", |row| row.get(0))?;
254 // A schema-14 cache converts in its rollback journal, so a failed conversion leaves
255 // the file as it was.
256 match version {
257 migrate::VERSION => migrate::migrate(&mut connection, path)?,
258 schema::PREVIOUS => schema::upgrade(&mut connection)?,
259 schema::VERSION => {}
260 _ => {
261 return Err(io::Error::new(
262 io::ErrorKind::InvalidData,
263 format!(
264 "Cache schema version {version} is not the supported version {}",
265 schema::VERSION
266 ),
267 )
268 .into());
269 }
270 }
271 write_ahead(&connection)?;
272 Self::start(connection, key)
273 }
274
275 fn start(connection: Connection, key: Option<Key>) -> Result<Self> {
276 let (section, root) = working::spawn(connection, key)?;
277 Ok(Self {
278 section,
279 synchronization: Mutex::new(()),
280 root,
281 })
282 }
283
284 fn lock(&self) -> Result<MutexGuard<'_, Connection>> {
285 lock(&self.section.connection)
286 }
287
288 /// Hands `request` to the section thread.
289 fn send(&self, request: working::Request) -> Result<()> {
290 self.section.send(request)
291 }
292
293 /// Asks the section thread and waits for its answer.
294 fn ask<T: Send + 'static>(
295 &self,
296 request: impl FnOnce(working::Reply<T>) -> working::Request,
297 ) -> Result<T> {
298 let (sender, receiver) = mpsc::sync_channel(1);
299 self.send(request(Box::new(move |result| {
300 let _ = sender.send(result);
301 })))?;
302 receiver
303 .recv()
304 .map_err(|_| io::Error::other("The section thread stopped"))?
305 }
306
307 async fn ask_async<T: Send + 'static>(
308 &self,
309 request: impl FnOnce(working::Reply<T>) -> working::Request + Send,
310 ) -> Result<T> {
311 let (sender, receiver) = task::response();
312 self.send(request(Box::new(move |result| {
313 let _ = sender.send(result);
314 })))?;
315 #[cfg(target_arch = "wasm32")]
316 let answer = task::receive(&receiver).await;
317 #[cfg(not(target_arch = "wasm32"))]
318 let answer = receiver.recv();
319 answer.map_err(|_| io::Error::other("The section thread stopped"))?
320 }
321
322 /// Applies an edit and queues it durably for publication, returning its id once
323 /// written. A refused edit returns `Rejected` and leaves every page as it was.
324 /// Browser callers use `apply_async`; this returns `WouldBlock` without submitting.
325 pub fn apply(&self, author: &str, edit: Edit) -> Result<u64> {
326 task::blocking(self.apply_async(author, edit))?
327 }
328
329 /// Applies an edit, awaiting its durable local queue receipt without blocking browser reads.
330 pub async fn apply_async(&self, author: &str, edit: Edit) -> Result<u64> {
331 self.ask_async(|reply| {
332 working::Request::Apply(working::Accepted {
333 author: author.to_owned(),
334 edit,
335 reply,
336 waiting: None,
337 moving: None,
338 })
339 })
340 .await
341 }
342
343 /// Page `space` as a move copies it, read at once with what the copy leaves behind.
344 pub fn moving(&self, space: ExGuid) -> Result<Moving> {
345 self.ask(|reply| working::Request::Moving { space, reply })
346 }
347
348 /// `retire_async`, blocking.
349 pub fn retire(&self, author: &str, edit: Edit, moving: Moving) -> Result<u64> {
350 task::blocking(self.retire_async(author, edit, moving))?
351 }
352
353 /// Applies `edit`, which deletes the page `moving` read once its copy has landed elsewhere,
354 /// only while the page and what it leaves behind are as read: checked in order with every
355 /// other edit, so none made since is deleted with it. One changed is refused as `Rejected`.
356 pub async fn retire_async(&self, author: &str, edit: Edit, moving: Moving) -> Result<u64> {
357 self.ask_async(|reply| {
358 working::Request::Apply(working::Accepted {
359 author: author.to_owned(),
360 edit,
361 reply,
362 waiting: None,
363 moving: Some(Box::new(moving)),
364 })
365 })
366 .await
367 }
368
369 /// `apply` without waiting: `reply` runs once the edit is durable or refused. While the
370 /// cache refuses to write it, the edit waits in memory and each failed write tells `waiting`.
371 pub(crate) fn submit(
372 &self,
373 author: &str,
374 edit: Edit,
375 reply: working::Reply<u64>,
376 waiting: working::Notice,
377 ) -> Result<()> {
378 self.send(working::Request::Apply(working::Accepted {
379 author: author.to_owned(),
380 edit,
381 reply,
382 waiting: Some(waiting),
383 moving: None,
384 }))
385 }
386
387 /// The page in `space` as the queued edits leave it; O(page), for opening and reloading.
388 pub fn page(&self, space: ExGuid) -> Result<Page> {
389 self.ask(|reply| working::Request::Page { space, reply })
390 }
391
392 /// Page spaces, titles and outline levels (1 at the top) in section order.
393 pub fn pages(&self) -> Result<Vec<(ExGuid, String, u32)>> {
394 self.ask(|reply| working::Request::Pages { reply })
395 }
396
397 /// The conflict pages of each page that has them (`onestore::Section::conflicts`).
398 pub fn conflicts(&self) -> Result<Vec<(ExGuid, Vec<onestore::ConflictPage>)>> {
399 self.ask(|reply| working::Request::Conflicts { reply })
400 }
401
402 /// The versions of each page that has them (`onestore::Section::versions`).
403 pub fn versions(&self) -> Result<Vec<(ExGuid, Vec<onestore::PageVersion>)>> {
404 self.ask(|reply| working::Request::Versions { reply })
405 }
406
407 /// A page as one of its versions holds it; O(section).
408 pub fn version(&self, space: ExGuid, version: ExGuid) -> Result<Page> {
409 self.ask(|reply| working::Request::Version {
410 space,
411 version,
412 reply,
413 })
414 }
415
416 /// The section image the queued edits leave, the unsealed ones sealed as one more
417 /// revision whose identities differ per call: O(section).
418 pub fn snapshot(&self) -> Result<Vec<u8>> {
419 self.ask(|reply| working::Request::Flush {
420 reply,
421 durable: false,
422 })?;
423 working::image(&*self.lock()?, self.section.key.as_ref())
424 }
425
426 /// Waits until the edits applied before it are durable in the local queue.
427 /// Browser callers use `written_async`; this returns `WouldBlock` without submitting.
428 pub fn written(&self) -> Result<()> {
429 task::blocking(self.written_async())?
430 }
431
432 /// Waits for durable local storage of the edits submitted before this request.
433 pub async fn written_async(&self) -> Result<()> {
434 self.ask_async(|reply| working::Request::Flush {
435 reply,
436 durable: true,
437 })
438 .await
439 }
440
441 /// The section file's identity, which internal links name as `section-id`.
442 pub fn identity(&self) -> Result<[u8; 16]> {
443 let stamp = base::base_stamp(&*self.lock()?)?;
444 Ok(onestore::Header::parse(&stamp.header)?.file_id)
445 }
446
447 /// Queued edits, oldest first.
448 pub fn pending(&self) -> Result<Vec<PendingEdit>> {
449 pending(&*self.lock()?, self.section.key.as_ref())
450 }
451
452 /// The edits putting each page the queue changed, as the queue leaves it, into another
453 /// section as a copy under fresh identities, by the author of the first edit changing it;
454 /// content outside the page model stays behind.
455 pub(crate) fn copies(&self) -> Result<Vec<(String, Edit)>> {
456 let arena = onestore::Arena::default();
457 let base = base::base(&*self.lock()?)?;
458 let base = working::open(&arena, base, self.section.key.as_ref())?;
459 let mut seen = std::collections::BTreeSet::from([self.root]);
460 let mut copies = Vec::new();
461 for queued in self.pending()? {
462 for space in queue::spaces(&queued.edit, self.root) {
463 if !seen.insert(space) {
464 continue;
465 }
466 let Ok(mut page) = self.page(space) else {
467 continue;
468 };
469 if base.page(space).is_ok_and(|base| base == page) {
470 continue;
471 }
472 page.objects
473 .retain(|object| !matches!(object, onestore::page::PageObject::Unsupported(_)));
474 let creation =
475 onestore::PageCreation::new(None, Some(&page.title), &queued.author)?;
476 let ops = vec![onestore::op::Op::Section(onestore::op::SectionOp::Import {
477 creation,
478 page: page.copy()?,
479 })];
480 copies.push((queued.author.clone(), Edit { at: now(), ops }));
481 }
482 }
483 Ok(copies)
484 }
485
486 fn wake_sync(&self) {
487 wake(&self.section.worker);
488 }
489}
490
491impl Drop for Replica {
492 fn drop(&mut self) {
493 self.section.stop();
494 }
495}
496
497fn lock(connection: &Mutex<Connection>) -> Result<MutexGuard<'_, Connection>> {
498 connection
499 .lock()
500 .map_err(|_| io::Error::other("Cache owner panicked").into())
501}
502
503/// Tells the worker a burst of local edits is durable.
504fn edited(worker: &Mutex<std::sync::Weak<worker::Signal>>) {
505 if let Ok(worker) = worker.lock()
506 && let Some(worker) = worker.upgrade()
507 {
508 worker.edited();
509 }
510}
511
512fn wake(worker: &Mutex<std::sync::Weak<worker::Signal>>) {
513 if let Ok(worker) = worker.lock()
514 && let Some(worker) = worker.upgrade()
515 {
516 worker.wake();
517 }
518}
519
520/// Fully validates a section image, a protected one under `key`, returning its root object
521/// space.
522fn validate(source: &[u8], key: Option<&Key>) -> Result<ExGuid> {
523 let arena = onestore::Arena::default();
524 Ok(match key {
525 Some(key) => onestore::Section::unlock(&arena, source.to_vec(), key)?.root(),
526 None => onestore::Section::open(&arena, source.to_vec())?.root(),
527 })
528}
529
530/// A closed cache, without opening the replica.
531fn closed(path: &Path) -> Result<Connection> {
532 let connection = cache_connection(path)?;
533 let application: u32 =
534 connection.pragma_query_value(None, "application_id", |row| row.get(0))?;
535 let version: u32 = connection.pragma_query_value(None, "user_version", |row| row.get(0))?;
536 if application != APPLICATION_ID || version != schema::VERSION {
537 return Err(io::Error::from(io::ErrorKind::InvalidData).into());
538 }
539 Ok(connection)
540}
541
542/// A closed cache's base stamp and how many edits wait.
543fn peek(connection: &Connection) -> Result<(onestore::Stamp, u64)> {
544 let base = base::base_stamp(connection)?;
545 let queued: i64 = connection.query_row("SELECT count(*) FROM edits", [], |row| row.get(0))?;
546 Ok((base, unsigned(queued)?))
547}
548
549fn pending(connection: &Connection, key: Option<&Key>) -> Result<Vec<PendingEdit>> {
550 queue::load(connection, key, None)
551}
552
553fn unsigned(value: i64) -> Result<u64> {
554 Ok(u64::try_from(value).map_err(|_| io::Error::from(io::ErrorKind::InvalidData))?)
555}
556
557fn signed(value: u64) -> Result<i64> {
558 Ok(i64::try_from(value).map_err(|_| io::Error::from(io::ErrorKind::InvalidInput))?)
559}
560
561/// Opens a cache without writing to it: exclusive locking, a full sync of every commit
562/// (`F_FULLFSYNC` on macOS, the WAL's syncs included) and foreign keys, each queried back.
563fn cache_connection(path: &Path) -> Result<Connection> {
564 #[cfg(target_arch = "wasm32")]
565 fs::install_sqlite();
566 let connection = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_WRITE)?;
567 connection.busy_timeout(Duration::ZERO)?;
568 connection.execute_batch(
569 "PRAGMA locking_mode=EXCLUSIVE; PRAGMA synchronous=FULL; PRAGMA fullfsync=ON; PRAGMA foreign_keys=ON;",
570 )?;
571 let locking: String = connection.pragma_query_value(None, "locking_mode", |row| row.get(0))?;
572 let journal: String = connection.pragma_query_value(None, "journal_mode", |row| row.get(0))?;
573 if locking != "exclusive" || !["wal", "delete"].contains(&journal.as_str()) {
574 return Err(io::Error::new(
575 io::ErrorKind::InvalidData,
576 "Unsupported cache locking or journal mode",
577 )
578 .into());
579 }
580 for (name, expected) in [("synchronous", 2), ("fullfsync", 1), ("foreign_keys", 1)] {
581 let actual: i64 = connection.pragma_query_value(None, name, |row| row.get(0))?;
582 if actual != expected {
583 return Err(io::Error::new(
584 io::ErrorKind::Unsupported,
585 "Required cache synchronization is unavailable",
586 )
587 .into());
588 }
589 }
590 Ok(connection)
591}
592
593/// Switches a cache to write-ahead logging: a commit appends its pages to `<cache>-wal`
594/// instead of copying the originals to a rollback journal. Under exclusive locking the WAL
595/// index lives in memory, so the WAL is the only file beside the cache; it is checkpointed
596/// and removed on close, and replayed on the next open after a crash.
597fn write_ahead(connection: &Connection) -> Result<()> {
598 let mode: String =
599 connection.pragma_update_and_check(None, "journal_mode", "WAL", |row| row.get(0))?;
600 if mode != "wal" {
601 return Err(io::Error::new(
602 io::ErrorKind::Unsupported,
603 "The cache cannot use write-ahead logging",
604 )
605 .into());
606 }
607 Ok(())
608}
609
610/// FILETIME now, as an edit's `at`: later than any this process took before, as Windows 7's
611/// clock ticks only every 15.6 ms and a later edit must still order after an earlier one.
612pub(crate) fn now() -> u64 {
613 use std::sync::atomic::{AtomicU64, Ordering};
614 static LAST: AtomicU64 = AtomicU64::new(0);
615 let unix = web_time::SystemTime::now()
616 .duration_since(web_time::UNIX_EPOCH)
617 .unwrap_or_default();
618 let clock =
619 (unix.as_secs() + 11_644_473_600) * 10_000_000 + u64::from(unix.subsec_nanos() / 100);
620 let previous = LAST
621 .try_update(Ordering::Relaxed, Ordering::Relaxed, |last| {
622 Some(clock.max(last + 1))
623 })
624 .unwrap_or_default();
625 clock.max(previous + 1)
626}