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

lb_rs/io/
mod.rs
