lb_rs/io/
mod.rs

1//! Members of this model are concerned with the details of IO, generally
2//! disk and network. This is the module any on-disk migrations will live
3//! and ideas around network, disk, and memory caches will be expressed.
4//! Code here should not be platform dependent, and should strive to be
5//! suitable for a range of devices: iPhones with flaky networks to servers
6//! and workstations with excellent networks.
7
8pub mod docs;
9pub mod legacy;
10pub mod network;
11
12use crate::Lb;
13use crate::model::account::Account;
14use crate::model::file_metadata::Owner;
15use crate::model::signed_meta::SignedMeta;
16use crate::service::activity::DocEvent;
17use crate::service::lb_id::LbID;
18#[cfg(all(unix, not(target_os = "ios")))]
19use crate::subscribers::ipc::Ipc;
20use db_rs::View;
21use db_rs::config::Config;
22use db_rs::errors::Result as DbResult;
23use db_rs::guard::WriteTx;
24use db_rs::views::{
25    composite_view::{Composite, Schema},
26    hashmap::DbHashMap,
27    option::DbOption,
28    vec::DbVec,
29};
30use db_rs_old::hasher::UuidIdentityHasherBuilder;
31use db_rs_old::{Config as OldConfig, Db};
32use legacy::CoreV4;
33use std::error::Error;
34use std::fs;
35use std::path::Path;
36use std::sync::Arc;
37use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
38use uuid::Uuid;
39use web_time::{Duration, Instant};
40
41pub(crate) type LbDb = Arc<RwLock<CoreDb>>;
42// todo: limit visibility
43pub type CoreDb = Composite<CoreV5>;
44pub type MigrationResult<T> = Result<T, Box<dyn Error + Send + Sync>>;
45
46#[derive(Default)]
47pub struct CoreV5 {
48    pub account: DbOption<Account>,
49    pub last_synced: DbOption<i64>,
50    pub root: DbOption<Uuid>,
51    pub local_metadata: DbHashMap<Uuid, SignedMeta, UuidIdentityHasherBuilder>,
52    pub base_metadata: DbHashMap<Uuid, SignedMeta, UuidIdentityHasherBuilder>,
53
54    /// map from pub key to username
55    pub pub_key_lookup: DbHashMap<Owner, String>,
56
57    pub doc_events: DbVec<DocEvent>,
58    pub id: DbOption<LbID>,
59    pub pinned_files: DbVec<Uuid>,
60
61    /// Sentinel for `send_debug_info` throttling: the millisecond timestamp of the
62    /// most recent panic file we've already uploaded. `None` means we have never
63    /// sent debug info; `Some(0)` means we've sent before but no panic file existed.
64    pub last_extracted_panic: DbOption<i64>,
65}
66
67impl Schema for CoreV5 {
68    fn views_mut(&mut self) -> impl AsMut<[&mut dyn View]> {
69        let views: [&mut dyn View; 10] = [
70            &mut self.account,
71            &mut self.last_synced,
72            &mut self.root,
73            &mut self.local_metadata,
74            &mut self.base_metadata,
75            &mut self.pub_key_lookup,
76            &mut self.doc_events,
77            &mut self.id,
78            &mut self.pinned_files,
79            &mut self.last_extracted_panic,
80        ];
81        views
82    }
83}
84
85pub fn init_with_migration(path: &Path) -> MigrationResult<CoreDb> {
86    let directory = path.join("CoreV5");
87    fs::create_dir_all(&directory)?;
88    let mut db = CoreDb::init(&Config::default().log_location(directory))?;
89    let old_path = path.join("CoreV4.db");
90
91    if db.schema.account.is_none() {
92        if !old_path.try_exists()? && !path.join("CoreV4").try_exists()? {
93            return Ok(db);
94        }
95        let source = CoreV4::init(OldConfig {
96            create_db: false,
97            create_path: false,
98            ..OldConfig::in_folder(path)
99        })?;
100        let tx = db.write_tx()?;
101        let dest = &mut db.schema;
102        if let Some(value) = source.account.get() {
103            dest.account.replace(value.clone())?;
104        }
105        if let Some(value) = source.last_synced.get() {
106            dest.last_synced.replace(*value)?;
107        }
108        if let Some(value) = source.root.get() {
109            dest.root.replace(*value)?;
110        }
111        for (key, value) in source.local_metadata.get() {
112            dest.local_metadata.insert(*key, value.clone())?;
113        }
114        for (key, value) in source.base_metadata.get() {
115            dest.base_metadata.insert(*key, value.clone())?;
116        }
117        for (key, value) in source.pub_key_lookup.get() {
118            dest.pub_key_lookup.insert(*key, value.clone())?;
119        }
120        for value in source.doc_events.get() {
121            dest.doc_events.push(*value)?;
122        }
123        if let Some(value) = source.id.get() {
124            dest.id.replace(*value)?;
125        }
126        for value in source.pinned_files.get() {
127            dest.pinned_files.push(*value)?;
128        }
129        if let Some(value) = source.last_extracted_panic.get() {
130            dest.last_extracted_panic.replace(*value)?;
131        }
132        tx.end_tx(&mut db)?;
133        drop(source);
134        fs::remove_file(old_path)?;
135    }
136    Ok(db)
137}
138
139pub struct LbRO<'a> {
140    guard: RwLockReadGuard<'a, CoreDb>,
141}
142
143impl LbRO<'_> {
144    pub fn db(&self) -> &CoreV5 {
145        &self.guard.schema
146    }
147}
148
149pub struct LbTx<'a> {
150    guard: RwLockWriteGuard<'a, CoreDb>,
151    tx: Option<WriteTx>,
152    #[cfg(all(unix, not(target_os = "ios")))]
153    ipc: &'a Ipc,
154}
155
156impl LbTx<'_> {
157    pub fn db(&mut self) -> &mut CoreV5 {
158        &mut self.guard.schema
159    }
160
161    pub fn end(mut self) {
162        self.flush().unwrap();
163    }
164
165    fn flush(&mut self) -> DbResult<()> {
166        let tx = self.tx.take().unwrap();
167        let seq_no = tx.seq_no;
168        let committed_seq = tx.end_tx(&mut *self.guard)?;
169        // Empty transactions retain the previous sequence, including after catch-up.
170        if committed_seq == seq_no {
171            #[cfg(all(unix, not(target_os = "ios")))]
172            self.ipc.notify_peers();
173        }
174        Ok(())
175    }
176}
177
178impl Drop for LbTx<'_> {
179    fn drop(&mut self) {
180        if self.tx.is_some() {
181            if let Err(error) = self.flush() {
182                error!(?error, "failed to flush database transaction on drop");
183            }
184        }
185    }
186}
187
188impl Lb {
189    pub async fn ro_tx(&self) -> LbRO<'_> {
190        let start = Instant::now();
191
192        let guard = self.db.read().await;
193
194        if start.elapsed() > Duration::from_millis(100) {
195            warn!("readonly transaction lock acquisition took {:?}", start.elapsed());
196        }
197
198        LbRO { guard }
199    }
200
201    /// note: you cannot call init for the same config, from the same tokio runtime, or you risk
202    /// deadlock. See the test `contended_transactions_do_not_block_the_runtime` for more info.
203    /// several workarounds exist for this situation that shouldn't really be required:
204    /// Lb is cheap to clone within the same runtime
205    /// Different data directories are fine
206    ///
207    /// If this situation actually becomes required:
208    ///
209    /// a config could be added to turn off file locks which is what results in tokio deadlocks, but
210    /// then what are you doing with multiple log writers writing to the same location?
211    ///
212    /// we could incur a small perf penalty and put the begin_tx in a tokio::blocking wrapper. This
213    /// is how tokio wants us to handle this, but I do not want the overhead.
214    pub async fn begin_tx(&self) -> LbTx<'_> {
215        let start = Instant::now();
216
217        let mut guard = self.db.write().await;
218
219        if start.elapsed() > Duration::from_millis(100) {
220            warn!("readwrite transaction lock acquisition took {:?}", start.elapsed());
221        }
222
223        let previous_seq = guard.last_modified();
224        let tx = guard.write_tx().unwrap();
225        self.notify_catch_up(&guard.schema, previous_seq).await;
226
227        LbTx {
228            guard,
229            tx: Some(tx),
230            #[cfg(all(unix, not(target_os = "ios")))]
231            ipc: &self.ipc,
232        }
233    }
234}
lb_rs/io/mod.rs

lb_rs/io/
mod.rs
