fix(core): keep the upload queue when a damaged database is rebuilt

A database SQLite reported as corrupt at open was moved aside and replaced
by an empty one, on the grounds that everything in it is fetched again. The
queue of writes Drive has not received, the mirror baselines and the pins
are not, so they were lost. They are now copied out of the damaged file,
read only and table by table, into the new one (B216).

crates/pdfs-core/src/db/mod.rs

@@ -216,18 +216,34 @@ impl Db {
         let conn = match Self::open_configured(path) {
             Ok(conn) => conn,
             Err(e) if is_corrupt(&e) => {
-                // Every byte in here is derived from the account or from the
-                // content cache directory, so the recovery is to start over —
-                // loudly, and keeping the damaged file for a post-mortem rather
-                // than deleting evidence.
+                // Most of what is in here is derived from the account and is
+                // fetched again. The queue is not: it holds writes accepted
+                // here that Drive does not have yet, and the mirror baselines
+                // and pins are this machine's alone. So the damaged file is
+                // moved aside, kept as evidence, and those tables are copied
+                // out of it into the new one as far as SQLite can still read
+                // them (`docs/BUGS.md` B216).
                 tracing::error!(
                     db = %path.display(),
                     error = %e,
-                    "cache database is corrupt — moving it aside and rebuilding \
-                     (metadata will be re-fetched; no user data is stored here)"
+                    "cache database is corrupt; moving it aside, rebuilding, and \
+                     copying the upload queue, sync baselines and pins out of it"
                 );
-                quarantine_corrupt_db(path)?;
-                Self::open_configured(path)?
+                let damaged = quarantine_corrupt_db(path)?;
+                let conn = Self::open_configured(path)?;
+                for (table, result) in salvage(&conn, &damaged) {
+                    match result {
+                        Ok(rows) => {
+                            tracing::warn!(table, rows, "salvaged from the damaged database")
+                        }
+                        Err(error) => tracing::error!(
+                            table,
+                            %error,
+                            "could not salvage from the damaged database"
+                        ),
+                    }
+                }
+                conn
             }
             Err(e) => return Err(e),
         };
@@ -428,12 +444,96 @@ fn is_corrupt(e: &Error) -> bool {
     )
 }
 
-/// Move a damaged database (and its WAL sidecars) out of the way so the next
-/// open starts from empty.
+/// The tables [`salvage`] copies out of a damaged database, in order, each
+/// with the rows it keeps.
 ///
-/// Renamed rather than deleted: this is a cache, but it is also the only
-/// evidence of what went wrong, and it costs one file the user can remove.
-fn quarantine_corrupt_db(path: &Path) -> Result<()> {
+/// The queue goes with the node rows it names by local id: a create made here
+/// has no other record of its node, and a real node keeps its local id when its
+/// folder is listed again. Those rows come back unlisted, so their folders are
+/// fetched again rather than trusted. Claims are cleared by [`Db::open`] as
+/// after any other crash.
+const SALVAGE: &[(&str, &str)] = &[
+    (
+        "nodes",
+        "lid IN (SELECT lid FROM old.pending_op WHERE lid IS NOT NULL
+                 UNION SELECT parent_lid FROM old.pending_op WHERE parent_lid IS NOT NULL)
+         OR uid LIKE 'local~%'",
+    ),
+    ("pending_op", "1"),
+    ("sync_folder", "1"),
+    ("sync_entry", "1"),
+    ("pins", "1"),
+    ("own_sealed_rev", "1"),
+];
+
+/// Copy what only this machine has out of the damaged database at `damaged`
+/// into the fresh one `conn` just created. Returns how each table went.
+///
+/// Each table is copied in its own transaction, so a table SQLite cannot read
+/// to the end is left out whole rather than half copied, and does not stop
+/// the others. Only the columns both files have are copied, which covers a
+/// file that broke in the middle of an upgrade. The damaged file is opened read
+/// only and left as it was.
+fn salvage(conn: &Connection, damaged: &Path) -> Vec<(&'static str, Result<usize>)> {
+    let uri = format!(
+        "file:{}?mode=ro",
+        damaged
+            .to_string_lossy()
+            .replace('%', "%25")
+            .replace('?', "%3f")
+            .replace('#', "%23")
+    );
+    if let Err(e) = conn.execute("ATTACH DATABASE ?1 AS old", [&uri]) {
+        let e = Error::Db(e);
+        tracing::error!(error = %e, "could not open the damaged database to salvage it");
+        return Vec::new();
+    }
+    let results = SALVAGE
+        .iter()
+        .map(|&(table, rows)| (table, salvage_table(conn, table, rows)))
+        .collect();
+    let _ = conn.execute("UPDATE nodes SET listed = 0", []);
+    if let Err(e) = conn.execute("DETACH DATABASE old", []) {
+        tracing::warn!(error = %e, "could not detach the damaged database");
+    }
+    results
+}
+
+fn salvage_table(conn: &Connection, table: &str, rows: &str) -> Result<usize> {
+    let columns = |schema: &str| -> Result<Vec<String>> {
+        let mut stmt = conn.prepare(&format!("PRAGMA {schema}.table_info({table})"))?;
+        let names = stmt
+            .query_map([], |r| r.get::<_, String>(1))?
+            .collect::<rusqlite::Result<Vec<_>>>()?;
+        Ok(names)
+    };
+    let new = columns("main")?;
+    let shared: Vec<String> = columns("old")?
+        .into_iter()
+        .filter(|c| new.contains(c))
+        .map(|c| format!("\"{c}\""))
+        .collect();
+    if shared.is_empty() {
+        return Ok(0);
+    }
+    let shared = shared.join(", ");
+    let tx = conn.unchecked_transaction()?;
+    let copied = tx.execute(
+        &format!(
+            "INSERT OR IGNORE INTO main.{table} ({shared}) SELECT {shared} FROM old.{table} WHERE {rows}"
+        ),
+        [],
+    )?;
+    tx.commit()?;
+    Ok(copied)
+}
+
+/// Move a damaged database (and its WAL sidecars) out of the way so the next
+/// open starts from empty. Returns where the database itself went.
+///
+/// Renamed rather than deleted: it is the only evidence of what went wrong,
+/// and [`salvage`] reads what only this machine has back out of it.
+fn quarantine_corrupt_db(path: &Path) -> Result<PathBuf> {
     let stamp = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .map(|d| d.as_secs())
@@ -457,7 +557,9 @@ fn quarantine_corrupt_db(path: &Path) -> Result<()> {
             Err(e) => tracing::warn!(file = %from.display(), error = %e, "could not quarantine"),
         }
     }
-    Ok(())
+    let mut kept = path.as_os_str().to_owned();
+    kept.push(format!(".corrupt-{stamp}"));
+    Ok(PathBuf::from(kept))
 }
 
 #[cfg(test)]

crates/pdfs-core/src/db/tests.rs

@@ -1283,6 +1283,89 @@ fn a_corrupt_database_is_moved_aside_and_rebuilt() {
     let _ = std::fs::remove_file(path.with_extension("db.lock"));
 }
 
+/// A rebuilt database must not lose what only this machine has: writes Drive
+/// has not received, the mirror baselines and the pins (`docs/BUGS.md` B216).
+#[test]
+fn a_rebuild_keeps_the_queue_baselines_and_pins_of_the_damaged_file() {
+    let id = now_test_id();
+    let damaged =
+        std::env::temp_dir().join(format!("pdfs-salvage-old-{}-{id}.db", std::process::id()));
+    let fresh =
+        std::env::temp_dir().join(format!("pdfs-salvage-new-{}-{id}.db", std::process::id()));
+
+    let db = Db::open(&damaged).unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    db.upsert_node(&file("listed", "root", "listed.txt", 1))
+        .unwrap();
+    db.upsert_node(&file("queued", "root", "queued.txt", 2))
+        .unwrap();
+    db.set_listed(&uid("root"), true).unwrap();
+    db.enqueue_op(&PendingOp {
+        id: 0,
+        kind: OP_REVISION.to_string(),
+        uid: uid("queued").to_string(),
+        parent_uid: None,
+        name: None,
+        blob_path: Some("/staging/queued".into()),
+        meta_json: None,
+        created_at: 1,
+        attempts: 3,
+        last_error: None,
+        next_attempt_at: 0,
+    })
+    .unwrap();
+    let folder_id = db
+        .sync_folder_add("/home/me/Docs", "vol~docs", "share")
+        .unwrap();
+    db.sync_entry_upsert(
+        folder_id,
+        &StoredSyncEntry {
+            rel_path: "a.txt".into(),
+            remote_uid: Some("vol~a".into()),
+            local_mtime: 5,
+            local_mtime_ns: None,
+            local_size: 6,
+            remote_rev: Some("7".into()),
+            remote_hash: Some("8".into()),
+        },
+    )
+    .unwrap();
+    db.pin_add("vol~pinned", "/pinned", true).unwrap();
+    drop(db);
+
+    let db = Db::open(&fresh).unwrap();
+    let results = salvage(&db.conn.lock(), &damaged);
+    assert!(results.iter().all(|(_, r)| r.is_ok()), "{results:?}");
+
+    let ops = db.pending_ops().unwrap();
+    assert_eq!(ops.len(), 1);
+    assert_eq!(ops[0].uid, uid("queued").to_string());
+    assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/queued"));
+    // The op's node came with it; the rest of the tree is fetched again.
+    assert!(
+        db.node_by_uid(&uid("queued").to_string())
+            .unwrap()
+            .is_some()
+    );
+    assert!(
+        db.node_by_uid(&uid("listed").to_string())
+            .unwrap()
+            .is_none()
+    );
+    assert_eq!(db.sync_entries(folder_id).unwrap().len(), 1);
+    assert!(db.is_pinned("vol~pinned").unwrap());
+    drop(db);
+
+    for path in [&damaged, &fresh] {
+        for suffix in ["", "-wal", "-shm"] {
+            let mut file = path.as_os_str().to_owned();
+            file.push(suffix);
+            let _ = std::fs::remove_file(std::path::PathBuf::from(file));
+        }
+        let _ = std::fs::remove_file(path.with_extension("db.lock"));
+    }
+}
+
 fn local(path: &str, name: &str, is_dir: bool) -> LocalEntry {
     LocalEntry {
         path: path.into(),