feat!: Add lockfile to account manager (#4310)

Opening the same account (context) from multiple processes is dangerous, can result in duplicate
downloads of the same message etc. Same for account manager, attempts to modify the same
accounts.toml even if done atomically with may result in corrupted files as atomic replacement
procedure does not expect that multiple processes may write to the same temporary file.

accounts.toml cannot be used as a lockfile because it is replaced during atomic update. Therefore, a
new file next to accounts.toml is needed to prevent starting second account manager in the same
directory.

But iOS needs to be able to open accounts from multiple processes at the same time. This is required
as the "share-to-DC extension" is a separate process by iOS design -- this process may or may not be
started while the main app is running. Accounts are not altered however by this extension, so let's
add to the `Accounts::new()` constructor an `rdwr` parameter which allows to read the accounts
config w/o locking the lockfile.

Cargo.lock

@@ -1132,6 +1132,7 @@ dependencies = [
  "encoded-words",
  "escaper",
  "fast-socks5",
+ "fd-lock",
  "format-flowed",
  "futures",
  "futures-lite",

Cargo.toml

@@ -48,6 +48,7 @@ email = { git = "https://github.com/deltachat/rust-email", branch = "master" }
 encoded-words = { git = "https://github.com/async-email/encoded-words", branch = "master" }
 escaper = "0.1"
 fast-socks5 = "0.8"
+fd-lock = "3.0.11"
 futures = "0.3"
 futures-lite = "1.13.0"
 hex = "0.4.0"

benches/create_account.rs

@@ -8,7 +8,8 @@ async fn create_accounts(n: u32) {
     let dir = tempdir().unwrap();
     let p: PathBuf = dir.path().join("accounts");
 
-    let mut accounts = Accounts::new(p.clone()).await.unwrap();
+    let writable = true;
+    let mut accounts = Accounts::new(p.clone(), writable).await.unwrap();
 
     for expected_id in 2..n {
         let id = accounts.add_account().await.unwrap();

deltachat-ffi/deltachat.h

@@ -2915,12 +2915,15 @@ int dc_receive_backup (dc_context_t* context, const char* qr);
  * @param dir The directory to create the context-databases in.
  *     If the directory does not exist,
  *     dc_accounts_new() will try to create it.
+ * @param writable Whether the returned account manager is writable, i.e. calling these functions on
+ *     it is possible: dc_accounts_add_account(), dc_accounts_add_closed_account(),
+ *     dc_accounts_migrate_account(), dc_accounts_remove_account(), dc_accounts_select_account().
  * @return An account manager object.
  *     The object must be passed to the other account manager functions
  *     and must be freed using dc_accounts_unref() after usage.
  *     On errors, NULL is returned.
  */
-dc_accounts_t* dc_accounts_new                  (const char* os_name, const char* dir);
+dc_accounts_t* dc_accounts_new                  (const char* dir, int writable);
 
 
 /**

deltachat-ffi/src/lib.rs

@@ -4676,17 +4676,17 @@ pub type dc_accounts_t = AccountsWrapper;
 
 #[no_mangle]
 pub unsafe extern "C" fn dc_accounts_new(
-    _os_name: *const libc::c_char,
+    dir: *const libc::c_char,
-    dbfile: *const libc::c_char,
+    writable: libc::c_int,
 ) -> *mut dc_accounts_t {
     setup_panic!();
 
-    if dbfile.is_null() {
+    if dir.is_null() {
         eprintln!("ignoring careless call to dc_accounts_new()");
         return ptr::null_mut();
     }
 
-    let accs = block_on(Accounts::new(as_path(dbfile).into()));
+    let accs = block_on(Accounts::new(as_path(dir).into(), writable != 0));
 
     match accs {
         Ok(accs) => Box::into_raw(Box::new(AccountsWrapper::new(accs))),

deltachat-jsonrpc/src/lib.rs

@@ -13,7 +13,8 @@ mod tests {
     #[tokio::test(flavor = "multi_thread")]
     async fn basic_json_rpc_functionality() -> anyhow::Result<()> {
         let tmp_dir = TempDir::new().unwrap().path().into();
-        let accounts = Accounts::new(tmp_dir).await?;
+        let writable = true;
+        let accounts = Accounts::new(tmp_dir, writable).await?;
         let api = CommandApi::new(accounts);
 
         let (sender, mut receiver) = unbounded::<String>();
@@ -54,7 +55,8 @@ mod tests {
     #[tokio::test(flavor = "multi_thread")]
     async fn test_batch_set_config() -> anyhow::Result<()> {
         let tmp_dir = TempDir::new().unwrap().path().into();
-        let accounts = Accounts::new(tmp_dir).await?;
+        let writable = true;
+        let accounts = Accounts::new(tmp_dir, writable).await?;
         let api = CommandApi::new(accounts);
 
         let (sender, mut receiver) = unbounded::<String>();

deltachat-jsonrpc/src/webserver.rs

@@ -19,7 +19,8 @@ async fn main() -> Result<(), std::io::Error> {
         .map(|port| port.parse::<u16>().expect("DC_PORT must be a number"))
         .unwrap_or(DEFAULT_PORT);
     log::info!("Starting with accounts directory `{path}`.");
-    let accounts = Accounts::new(PathBuf::from(&path)).await.unwrap();
+    let writable = true;
+    let accounts = Accounts::new(PathBuf::from(&path), writable).await.unwrap();
     let state = CommandApi::new(accounts);
 
     let app = Router::new()

deltachat-rpc-server/src/main.rs

@@ -56,7 +56,8 @@ async fn main_impl() -> Result<()> {
 
     let path = std::env::var("DC_ACCOUNTS_PATH").unwrap_or_else(|_| "accounts".to_string());
     log::info!("Starting with accounts directory `{}`.", path);
-    let accounts = Accounts::new(PathBuf::from(&path)).await?;
+    let writable = true;
+    let accounts = Accounts::new(PathBuf::from(&path), writable).await?;
 
     log::info!("Creating JSON-RPC API.");
     let accounts = Arc::new(RwLock::new(accounts));

node/lib/deltachat.ts

@@ -21,12 +21,15 @@ export class AccountManager extends EventEmitter {
   accountDir: string
   jsonRpcStarted = false
 
-  constructor(cwd: string, os = 'deltachat-node') {
+  constructor(cwd: string, writable = true) {
     super()
     debug('DeltaChat constructor')
 
     this.accountDir = cwd
-    this.dcn_accounts = binding.dcn_accounts_new(os, this.accountDir)
+    this.dcn_accounts = binding.dcn_accounts_new(
+      this.accountDir,
+      writable ? 1 : 0
+    )
   }
 
   getAllAccountIds() {

node/src/module.c

@@ -2903,8 +2903,8 @@ NAPI_METHOD(dcn_msg_get_webxdc_blob){
 
 NAPI_METHOD(dcn_accounts_new) {
   NAPI_ARGV(2);
-  NAPI_ARGV_UTF8_MALLOC(os_name, 0);
+  NAPI_ARGV_UTF8_MALLOC(dir, 0);
-  NAPI_ARGV_UTF8_MALLOC(dir, 1);
+  NAPI_ARGV_INT32(writable, 1);
   TRACE("calling..");
 
   dcn_accounts_t* dcn_accounts = calloc(1, sizeof(dcn_accounts_t));
@@ -2913,7 +2913,7 @@ NAPI_METHOD(dcn_accounts_new) {
   }
 
 
-  dcn_accounts->dc_accounts = dc_accounts_new(os_name, dir);
+  dcn_accounts->dc_accounts = dc_accounts_new(dir, writable);
 
   napi_value result;
   NAPI_STATUS_THROWS(napi_create_external(env, dcn_accounts,

python/tests/test_4_lowlevel.py

@@ -221,8 +221,9 @@ def test_logged_ac_process_ffi_failure(acfactory):
 
 def test_jsonrpc_blocking_call(tmp_path):
     accounts_fname = tmp_path / "accounts"
+    writable = True
     accounts = ffi.gc(
-        lib.dc_accounts_new(ffi.NULL, str(accounts_fname).encode("ascii")),
+        lib.dc_accounts_new(str(accounts_fname).encode("ascii"), writable),
         lib.dc_accounts_unref,
     )
     jsonrpc = ffi.gc(lib.dc_jsonrpc_init(accounts), lib.dc_jsonrpc_unref)

src/accounts.rs

@@ -7,6 +7,9 @@ use anyhow::{ensure, Context as _, Result};
 use serde::{Deserialize, Serialize};
 use tokio::fs;
 use tokio::io::AsyncWriteExt;
+use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
+use tokio::time::{sleep, Duration};
 use uuid::Uuid;
 
 use crate::context::Context;
@@ -33,16 +36,16 @@ pub struct Accounts {
 
 impl Accounts {
     /// Loads or creates an accounts folder at the given `dir`.
-    pub async fn new(dir: PathBuf) -> Result<Self> {
+    pub async fn new(dir: PathBuf, writable: bool) -> Result<Self> {
-        if !dir.exists() {
+        if writable && !dir.exists() {
             Accounts::create(&dir).await?;
         }
 
-        Accounts::open(dir).await
+        Accounts::open(dir, writable).await
     }
 
     /// Creates a new default structure.
-    pub async fn create(dir: &Path) -> Result<()> {
+    async fn create(dir: &Path) -> Result<()> {
         fs::create_dir_all(dir)
             .await
             .context("failed to create folder")?;
@@ -54,13 +57,13 @@ impl Accounts {
 
     /// Opens an existing accounts structure. Will error if the folder doesn't exist,
     /// no account exists and no config exists.
-    pub async fn open(dir: PathBuf) -> Result<Self> {
+    async fn open(dir: PathBuf, writable: bool) -> Result<Self> {
         ensure!(dir.exists(), "directory does not exist");
 
         let config_file = dir.join(CONFIG_NAME);
         ensure!(config_file.exists(), "{:?} does not exist", config_file);
 
-        let config = Config::from_file(config_file)
+        let config = Config::from_file(config_file, writable)
             .await
             .context("failed to load accounts config")?;
         let events = Events::new();
@@ -298,14 +301,20 @@ impl Accounts {
 /// Configuration file name.
 pub const CONFIG_NAME: &str = "accounts.toml";
 
+/// Lockfile name.
+pub const LOCKFILE_NAME: &str = "accounts.lock";
+
 /// Database file name.
 pub const DB_NAME: &str = "dc.db";
 
 /// Account manager configuration file.
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug)]
 struct Config {
     file: PathBuf,
     inner: InnerConfig,
+    // We lock the lockfile in the Config constructors to protect also from having multiple Config
+    // objects for the same config file.
+    lock_task: Option<JoinHandle<anyhow::Result<()>>>,
 }
 
 /// Account manager configuration file contents.
@@ -319,17 +328,74 @@ struct InnerConfig {
     pub accounts: Vec<AccountConfig>,
 }
 
+impl Drop for Config {
+    fn drop(&mut self) {
+        if let Some(lock_task) = self.lock_task.take() {
+            lock_task.abort();
+        }
+    }
+}
+
 impl Config {
-    /// Creates a new configuration file in the given account manager directory.
+    /// Creates a new Config for `file`, but doesn't open/sync it.
-    pub async fn new(dir: &Path) -> Result<Self> {
+    async fn new_nosync(file: PathBuf, lock: bool) -> Result<Self> {
+        let dir = file.parent().context("Cannot get config file directory")?;
         let inner = InnerConfig {
             accounts: Vec::new(),
             selected_account: 0,
             next_id: 1,
         };
-        let file = dir.join(CONFIG_NAME);
+        if !lock {
-        let mut cfg = Self { file, inner };
+            let cfg = Self {
+                file,
+                inner,
+                lock_task: None,
+            };
+            return Ok(cfg);
+        }
+        let lockfile = dir.join(LOCKFILE_NAME);
+        let mut lock = fd_lock::RwLock::new(fs::File::create(lockfile).await?);
+        let (locked_tx, locked_rx) = oneshot::channel();
+        let lock_task: JoinHandle<anyhow::Result<()>> = tokio::spawn(async move {
+            let mut timeout = Duration::from_millis(100);
+            let _guard = loop {
+                match lock.try_write() {
+                    Ok(guard) => break Ok(guard),
+                    Err(err) => {
+                        if timeout.as_millis() > 1600 {
+                            break Err(err);
+                        }
+                        // We need to wait for the previous lock_task to be aborted thus unlocking
+                        // the lockfile. We don't open configs for writing often outside of the
+                        // tests, so this adds delays to the tests, but otherwise ok.
+                        sleep(timeout).await;
+                        if err.kind() == std::io::ErrorKind::WouldBlock {
+                            timeout *= 2;
+                        }
+                    }
+                }
+            }?;
+            locked_tx
+                .send(())
+                .ok()
+                .context("Cannot notify about lockfile locking")?;
+            let (_tx, rx) = oneshot::channel();
+            rx.await?;
+            Ok(())
+        });
+        let cfg = Self {
+            file,
+            inner,
+            lock_task: Some(lock_task),
+        };
+        locked_rx.await?;
+        Ok(cfg)
+    }
 
+    /// Creates a new configuration file in the given account manager directory.
+    pub async fn new(dir: &Path) -> Result<Self> {
+        let lock = true;
+        let mut cfg = Self::new_nosync(dir.join(CONFIG_NAME), lock).await?;
         cfg.sync().await?;
 
         Ok(cfg)
@@ -339,6 +405,11 @@ impl Config {
     /// Takes a mutable reference because the saved file is a part of the `Config` state. This
     /// protects from parallel calls resulting to a wrong file contents.
     async fn sync(&mut self) -> Result<()> {
+        ensure!(!self
+            .lock_task
+            .as_ref()
+            .context("Config is read-only")?
+            .is_finished());
         let tmp_path = self.file.with_extension("toml.tmp");
         let mut file = fs::File::create(&tmp_path)
             .await
@@ -357,24 +428,28 @@ impl Config {
     }
 
     /// Read a configuration from the given file into memory.
-    pub async fn from_file(file: PathBuf) -> Result<Self> {
+    pub async fn from_file(file: PathBuf, writable: bool) -> Result<Self> {
-        let dir = file.parent().context("can't get config file directory")?;
+        let dir = file
-        let bytes = fs::read(&file).await.context("failed to read file")?;
+            .parent()
+            .context("Cannot get config file directory")?
+            .to_path_buf();
+        let mut config = Self::new_nosync(file, writable).await?;
+        let bytes = fs::read(&config.file)
+            .await
+            .context("Failed to read file")?;
         let s = std::str::from_utf8(&bytes)?;
-        let mut inner: InnerConfig = toml::from_str(s).context("failed to parse config")?;
+        config.inner = toml::from_str(s).context("Failed to parse config")?;
 
         // Previous versions of the core stored absolute paths in account config.
         // Convert them to relative paths.
         let mut modified = false;
-        for account in &mut inner.accounts {
+        for account in &mut config.inner.accounts {
-            if let Ok(new_dir) = account.dir.strip_prefix(dir) {
+            if let Ok(new_dir) = account.dir.strip_prefix(&dir) {
                 account.dir = new_dir.to_path_buf();
                 modified = true;
             }
         }
-
+        if modified && writable {
-        let mut config = Self { file, inner };
-        if modified {
             config.sync().await?;
         }
 
@@ -518,26 +593,44 @@ mod tests {
         let p: PathBuf = dir.path().join("accounts1");
 
         {
-            let mut accounts = Accounts::new(p.clone()).await.unwrap();
+            let writable = true;
+            let mut accounts = Accounts::new(p.clone(), writable).await.unwrap();
             accounts.add_account().await.unwrap();
 
             assert_eq!(accounts.accounts.len(), 1);
             assert_eq!(accounts.config.get_selected_account(), 1);
         }
-        {
+        for writable in [true, false] {
-            let accounts = Accounts::open(p).await.unwrap();
+            let accounts = Accounts::new(p.clone(), writable).await.unwrap();
 
             assert_eq!(accounts.accounts.len(), 1);
             assert_eq!(accounts.config.get_selected_account(), 1);
         }
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_account_new_open_conflict() {
+        let dir = tempfile::tempdir().unwrap();
+        let p: PathBuf = dir.path().join("accounts");
+        let writable = true;
+        let _accounts = Accounts::new(p.clone(), writable).await.unwrap();
+
+        let writable = true;
+        assert!(Accounts::new(p.clone(), writable).await.is_err());
+
+        let writable = false;
+        let accounts = Accounts::new(p, writable).await.unwrap();
+        assert_eq!(accounts.accounts.len(), 0);
+        assert_eq!(accounts.config.get_selected_account(), 0);
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_account_new_add_remove() {
         let dir = tempfile::tempdir().unwrap();
         let p: PathBuf = dir.path().join("accounts");
 
-        let mut accounts = Accounts::new(p.clone()).await.unwrap();
+        let writable = true;
+        let mut accounts = Accounts::new(p.clone(), writable).await.unwrap();
         assert_eq!(accounts.accounts.len(), 0);
         assert_eq!(accounts.config.get_selected_account(), 0);
 
@@ -564,7 +657,8 @@ mod tests {
         let dir = tempfile::tempdir()?;
         let p: PathBuf = dir.path().join("accounts");
 
-        let mut accounts = Accounts::new(p.clone()).await?;
+        let writable = true;
+        let mut accounts = Accounts::new(p.clone(), writable).await?;
         assert!(accounts.get_selected_account().is_none());
         assert_eq!(accounts.config.get_selected_account(), 0);
 
@@ -585,7 +679,8 @@ mod tests {
         let dir = tempfile::tempdir().unwrap();
         let p: PathBuf = dir.path().join("accounts");
 
-        let mut accounts = Accounts::new(p.clone()).await.unwrap();
+        let writable = true;
+        let mut accounts = Accounts::new(p.clone(), writable).await.unwrap();
         assert_eq!(accounts.accounts.len(), 0);
         assert_eq!(accounts.config.get_selected_account(), 0);
 
@@ -622,7 +717,8 @@ mod tests {
         let dir = tempfile::tempdir().unwrap();
         let p: PathBuf = dir.path().join("accounts");
 
-        let mut accounts = Accounts::new(p.clone()).await.unwrap();
+        let writable = true;
+        let mut accounts = Accounts::new(p.clone(), writable).await.unwrap();
 
         for expected_id in 1..10 {
             let id = accounts.add_account().await.unwrap();
@@ -642,7 +738,8 @@ mod tests {
         let dummy_accounts = 10;
 
         let (id0, id1, id2) = {
-            let mut accounts = Accounts::new(p.clone()).await?;
+            let writable = true;
+            let mut accounts = Accounts::new(p.clone(), writable).await?;
             accounts.add_account().await?;
             let ids = accounts.get_all();
             assert_eq!(ids.len(), 1);
@@ -677,7 +774,8 @@ mod tests {
         assert!(id2 > id1 + dummy_accounts);
 
         let (id0_reopened, id1_reopened, id2_reopened) = {
-            let accounts = Accounts::new(p.clone()).await?;
+            let writable = false;
+            let accounts = Accounts::new(p.clone(), writable).await?;
             let ctx = accounts.get_selected_account().unwrap();
             assert_eq!(
                 ctx.get_config(crate::config::Config::Addr).await?,
@@ -722,7 +820,8 @@ mod tests {
         let dir = tempfile::tempdir().unwrap();
         let p: PathBuf = dir.path().join("accounts");
 
-        let accounts = Accounts::new(p.clone()).await?;
+        let writable = true;
+        let accounts = Accounts::new(p.clone(), writable).await?;
 
         // Make sure there are no accounts.
         assert_eq!(accounts.accounts.len(), 0);
@@ -748,7 +847,8 @@ mod tests {
         let dir = tempfile::tempdir().context("failed to create tempdir")?;
         let p: PathBuf = dir.path().join("accounts");
 
-        let mut accounts = Accounts::new(p.clone())
+        let writable = true;
+        let mut accounts = Accounts::new(p.clone(), writable)
             .await
             .context("failed to create accounts manager")?;
 
@@ -768,7 +868,8 @@ mod tests {
         assert!(passphrase_set_success);
         drop(accounts);
 
-        let accounts = Accounts::new(p.clone())
+        let writable = false;
+        let accounts = Accounts::new(p.clone(), writable)
             .await
             .context("failed to create second accounts manager")?;
         let account = accounts
@@ -792,7 +893,8 @@ mod tests {
         let dir = tempfile::tempdir().unwrap();
         let p: PathBuf = dir.path().join("accounts");
 
-        let mut accounts = Accounts::new(p.clone()).await?;
+        let writable = true;
+        let mut accounts = Accounts::new(p.clone(), writable).await?;
         accounts.add_account().await?;
         accounts.add_account().await?;