defter 0.5.2

Self-hosted Zotero sync and file storage server, with metadata in SQLite and attachments in S3.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244

//! The libraries' databases, `libraries/<u|g><id>.sqlite3`, opened when first
//! asked for and kept open while used.
//!
//! Each file is open as at most one [`Db`], whose single writer keeps a write's
//! version check and bump atomic. Up to `cap` stay open; past that, the least
//! recently used that no request holds are closed.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};

use crate::control::{self, Control, LibraryRef};
use crate::db::{self, Db};
use crate::error::AppError;
use crate::storage::FileScope;

/// Idle read connections kept per open library.
const IDLE_READERS: usize = 4;

/// An open library.
pub struct Library {
    pub id: LibraryRef,
    pub db: Arc<Db>,
    /// Where its attachment files are.
    pub files: FileScope,
    /// The highest version it is known to have served (see [`Libraries::served`]).
    version_high: AtomicI64,
}

pub struct Libraries {
    dir: PathBuf,
    control: Arc<Control>,
    cap: usize,
    open: tokio::sync::Mutex<HashMap<LibraryRef, (Arc<Library>, u64)>>,
    clock: AtomicU64,
}

/// The database file of `library` in `dir`.
pub fn path(dir: &Path, library: LibraryRef) -> PathBuf {
    dir.join(format!("{library}.sqlite3"))
}

impl Libraries {
    /// The libraries in `dir`, keeping up to `cap` open.
    pub fn new(dir: PathBuf, control: Arc<Control>, cap: usize) -> Self {
        Libraries {
            dir,
            control,
            cap: cap.max(1),
            open: Default::default(),
            clock: AtomicU64::new(0),
        }
    }

    #[cfg(test)]
    pub fn dir(&self) -> &Path {
        &self.dir
    }

    /// The library, opened if it is not. `NotFound` for a library that does
    /// not exist or is disabled.
    pub async fn get(&self, id: LibraryRef) -> Result<Arc<Library>, AppError> {
        let tick = self.clock.fetch_add(1, Ordering::Relaxed);
        // Held while opening, so a file is never opened twice.
        let mut open = self.open.lock().await;
        if let Some((library, used)) = open.get_mut(&id) {
            *used = tick;
            return Ok(library.clone());
        }

        let files = match id {
            LibraryRef::User(user_id) => match self.control.user(user_id).await? {
                Some(user) if !user.disabled => user.files(),
                _ => return Err(AppError::NotFound("no such library".into())),
            },
            LibraryRef::Group(_) => return Err(AppError::NotFound("no such library".into())),
        };
        let version_high = self.control.version_high(id).await?;
        let file = path(&self.dir, id);
        let db = tokio::task::spawn_blocking(move || {
            // A library that has served versions is never started afresh: clients
            // holding them would be told the library went back to nothing.
            if version_high > 0 && !file.exists() {
                return Err(AppError::Internal(format!(
                    "{} is missing, but the library served version {version_high}: restore \
                     it from a backup (defter restore)",
                    file.display()
                )));
            }
            Db::open_with(&file, db::LIBRARY, IDLE_READERS)
        })
        .await
        .map_err(|e| AppError::Internal(format!("blocking task: {e}")))??;
        // What its files take, for the admin page, as it may not have been
        // counted since it last changed.
        match db.read(|c| crate::store::file_usage(c, "")).await {
            Ok((bytes, _)) => self.control.note_usage(id, bytes).await?,
            Err(error) => tracing::warn!(%error, library = %id, "counting storage use"),
        }
        let library = Arc::new(Library {
            id,
            db,
            files,
            version_high: AtomicI64::new(version_high),
        });
        open.insert(id, (library.clone(), tick));

        if open.len() > self.cap {
            let mut idle: Vec<(LibraryRef, u64)> = open
                .iter()
                // Held only by the map: no request is using it.
                .filter(|(_, (library, _))| Arc::strong_count(library) == 1)
                .map(|(id, (_, used))| (*id, *used))
                .collect();
            idle.sort_by_key(|(_, used)| *used);
            let excess = open.len() - self.cap;
            for (id, _) in idle.into_iter().take(excess) {
                open.remove(&id);
            }
        }
        Ok(library)
    }

    /// Record that `library` served `version` (its `Last-Modified-Version`), so
    /// that a restore stamps versions above it.
    pub async fn served(&self, library: &Library, version: i64) -> Result<(), AppError> {
        if library.version_high.load(Ordering::Relaxed) >= version {
            return Ok(());
        }
        // Remembered only once recorded, so a failed write is tried again.
        self.control.note_version(library.id, version).await?;
        library.version_high.fetch_max(version, Ordering::Relaxed);
        Ok(())
    }

    /// How many libraries are open.
    #[cfg(test)]
    pub async fn open_count(&self) -> usize {
        self.open.lock().await.len()
    }
}

/// The owner of a server from before there were several libraries, as
/// `defter migrate` and a new server's first start make them.
pub fn owner(id: u64, username: &str, legacy_files: bool) -> control::User {
    control::User {
        id,
        orcid: None,
        username: username.to_string(),
        display_name: None,
        profile_hidden: false,
        disabled: false,
        legacy_files,
        quota_bytes: None,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    async fn setup(cap: usize, users: &[u64]) -> (tempfile::TempDir, Arc<Control>, Libraries) {
        let dir = tempfile::tempdir().unwrap();
        let control = Arc::new(Control::open(&dir.path().join("control.sqlite3")).unwrap());
        for id in users {
            let user = owner(*id, &format!("user{id}"), false);
            control
                .db
                .write(move |c| control::insert_user(c, &user))
                .await
                .unwrap();
        }
        let libraries = Libraries::new(dir.path().join("libraries"), control.clone(), cap);
        (dir, control, libraries)
    }

    #[tokio::test]
    async fn opens_each_library_once() {
        let (_dir, _control, libraries) = setup(10, &[1, 2]).await;
        let one = libraries.get(LibraryRef::User(1)).await.unwrap();
        let again = libraries.get(LibraryRef::User(1)).await.unwrap();
        assert!(Arc::ptr_eq(&one.db, &again.db));
        assert_eq!(one.files, FileScope::User(1));
        let two = libraries.get(LibraryRef::User(2)).await.unwrap();
        assert!(!Arc::ptr_eq(&one.db, &two.db));
        assert!(path(libraries.dir(), LibraryRef::User(2)).exists());
        for unknown in [LibraryRef::User(3), LibraryRef::Group(1)] {
            assert!(matches!(
                libraries.get(unknown).await,
                Err(AppError::NotFound(_))
            ));
        }
    }

    #[tokio::test]
    async fn closes_the_least_recently_used_that_nobody_holds() {
        let (_dir, _control, libraries) = setup(2, &[1, 2, 3, 4]).await;
        let held = libraries.get(LibraryRef::User(1)).await.unwrap();
        drop(libraries.get(LibraryRef::User(2)).await.unwrap());
        drop(libraries.get(LibraryRef::User(3)).await.unwrap());
        // 2 went; 1 is held, however long ago it was asked for.
        assert_eq!(libraries.open_count().await, 2);
        let again = libraries.get(LibraryRef::User(1)).await.unwrap();
        assert!(Arc::ptr_eq(&held.db, &again.db));
        // With every one held, more stay open than the cap rather than two Dbs.
        let three = libraries.get(LibraryRef::User(3)).await.unwrap();
        let four = libraries.get(LibraryRef::User(4)).await.unwrap();
        assert_eq!(libraries.open_count().await, 3);
        drop((three, four, again, held));
    }

    #[tokio::test]
    async fn never_starts_a_library_afresh_that_served_versions() {
        let (_dir, control, libraries) = setup(10, &[1]).await;
        let library = libraries.get(LibraryRef::User(1)).await.unwrap();
        libraries.served(&library, 7).await.unwrap();
        libraries.served(&library, 5).await.unwrap();
        assert_eq!(control.version_high(LibraryRef::User(1)).await.unwrap(), 7);
        drop(library);

        let fresh = Libraries::new(libraries.dir().to_path_buf(), control.clone(), 10);
        std::fs::remove_file(path(fresh.dir(), LibraryRef::User(1))).unwrap();
        let refused = fresh.get(LibraryRef::User(1)).await.err().unwrap();
        assert!(
            refused.to_string().contains("served version 7"),
            "{refused}"
        );
    }

    #[tokio::test]
    async fn a_disabled_user_has_no_library() {
        let (_dir, control, libraries) = setup(10, &[1]).await;
        control
            .db
            .write(|c| c.execute("UPDATE user SET disabled = 1", []))
            .await
            .unwrap();
        assert!(matches!(
            libraries.get(LibraryRef::User(1)).await,
            Err(AppError::NotFound(_))
        ));
    }
}