mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-03 16:41:26 +01:00
feat(fuse,core): record namespace changes locally and send them from the queue
A new folder, a new file, a rename or a delete through the mount no longer waits for Drive: it is recorded locally, the call returns at once, and the drain sends it as it does offline. `"local_first": false` in config.json brings back the old path for this release. Fixes found by the simulation runs on the way: - B129: a file deleted while its first upload is on the wire is trashed once the upload lands. - B132, B140: files made in a new folder survive a restart; a landed folder hands its queued children over in the stored node too. - B138: a create waits for the rename or trash that frees its name. - B139: a file closed while its first upload lands keeps its last write. - B141: a rename a worker holds stays queued, and the newer one waits behind it instead of racing it. - B142: a rename of a file whose create lands meanwhile goes to the uid Drive gave it instead of answering EBUSY. - B143: a create whose read-back fails as the link drops is adopted before the next op for it is sent, so its next write is not a conflict copy.
This commit is contained in:
nvrl committed
1 parent
b3e33a7002
commit
5e0f30cc2c
17 files changed
+1271
-111
No files matched your search
@@ -154,6 +154,13 @@ pub struct AppConfig {
|
||||
/// configs predating the field.
|
||||
#[serde(default)]
|
||||
pub online_map: bool,
|
||||
/// Record a new folder, a new file, a rename or a delete through the mount
|
||||
/// locally and send it from the queue, instead of waiting for Drive in the
|
||||
/// syscall. `None` means on. `false` brings back the old path for one
|
||||
/// release (docs/MILESTONE-3.0.0.md §10). Defaulted for configs predating
|
||||
/// the field.
|
||||
#[serde(default)]
|
||||
pub local_first: Option<bool>,
|
||||
}
|
||||
|
||||
/// The My Files page's view choices, remembered between runs.
|
||||
@@ -228,6 +235,7 @@ impl Default for AppConfig {
|
||||
tray_hidden: false,
|
||||
language: None,
|
||||
online_map: false,
|
||||
local_first: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -584,6 +592,7 @@ mod tests {
|
||||
tray_hidden: true,
|
||||
language: Some("de".to_string()),
|
||||
online_map: true,
|
||||
local_first: Some(false),
|
||||
};
|
||||
let json = serde_json::to_string(&config).unwrap();
|
||||
let decoded: AppConfig = serde_json::from_str(&json).unwrap();
|
||||
|
||||
@@ -56,9 +56,9 @@ pub use local::LocalFileHit;
|
||||
pub use maintenance::{DbStats, VacuumOutcome};
|
||||
pub use nodes::{PublishedSharedRoot, SearchHit, StoredNode};
|
||||
pub use ops::{
|
||||
AttachedBlob, CreateLanding, FAILING_ATTEMPTS, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME,
|
||||
OP_REVISION, OP_TRASH, PARK_EXPIRY_MS, PARK_UNTIL, PendingCounts, PendingOp, RenameMeta,
|
||||
op_supersedes,
|
||||
AttachedBlob, CreateLanding, CreateRetired, FAILING_ATTEMPTS, LOCAL_VOLUME, OP_CREATE,
|
||||
OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PARK_EXPIRY_MS, PARK_UNTIL, PendingCounts,
|
||||
PendingOp, RenameMeta, op_supersedes,
|
||||
};
|
||||
pub use photos::{StoredPhoto, StoredPlace, THUMB_HAVE, THUMB_NONE, THUMB_UNKNOWN, TimelineRow};
|
||||
pub use pins::PinRow;
|
||||
|
||||
@@ -6,6 +6,7 @@ use std::collections::HashSet;
|
||||
use rusqlite::{OptionalExtension, Transaction, params};
|
||||
|
||||
use super::Db;
|
||||
use super::ops::{OP_CREATE, OP_MKDIR, OP_RENAME, wake_ops_waiting_for_tx};
|
||||
use crate::{Access, Result};
|
||||
use proton_drive_rs::proton_sdk::ids::{LinkId, NodeUid, VolumeId};
|
||||
use proton_drive_rs::{Node, NodeKind};
|
||||
@@ -266,10 +267,22 @@ impl Db {
|
||||
/// resolve its shared-tree access. Removing the op first could resurrect
|
||||
/// that row after a crash; removing the row first could strand a now
|
||||
/// unauthorizable op. One transaction closes both windows.
|
||||
///
|
||||
/// An op that found the name still held by this trash failed with a
|
||||
/// backoff, and the name is free now. It is made due at once
|
||||
/// ([`Db::wake_ops_waiting_for`]).
|
||||
pub fn complete_trash_op(&self, op_id: i64, uid: &NodeUid) -> Result<()> {
|
||||
let uid = uid.to_string();
|
||||
let mut conn = self.conn.lock();
|
||||
let tx = conn.transaction()?;
|
||||
let name: Option<String> = tx
|
||||
.query_row(
|
||||
"SELECT name FROM pending_op WHERE id = ?1",
|
||||
params![op_id],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.optional()?
|
||||
.flatten();
|
||||
let rowid: Option<i64> = tx
|
||||
.query_row(
|
||||
"SELECT rowid FROM nodes WHERE uid = ?1",
|
||||
@@ -282,6 +295,9 @@ impl Db {
|
||||
tx.execute("DELETE FROM nodes_fts WHERE rowid = ?1", params![rowid])?;
|
||||
}
|
||||
tx.execute("DELETE FROM pending_op WHERE id = ?1", params![op_id])?;
|
||||
if let Some(name) = name {
|
||||
wake_ops_waiting_for_tx(&tx, &name)?;
|
||||
}
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
}
|
||||
@@ -549,6 +565,18 @@ impl Db {
|
||||
}
|
||||
}
|
||||
|
||||
/// The uid of the node with local id `lid`: `None` when it has no row, or
|
||||
/// is not on Drive yet.
|
||||
pub fn uid_of_lid(&self, lid: i64) -> Result<Option<String>> {
|
||||
let conn = self.read();
|
||||
Ok(conn
|
||||
.query_row("SELECT uid FROM nodes WHERE lid = ?1", params![lid], |r| {
|
||||
r.get(0)
|
||||
})
|
||||
.optional()?
|
||||
.flatten())
|
||||
}
|
||||
|
||||
/// The local id of the node stored under `uid`, or `None` when it has no
|
||||
/// row. It stays the same for as long as the row exists, including when a
|
||||
/// create lands and the row takes the uid Drive gave it.
|
||||
@@ -636,6 +664,40 @@ impl Db {
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// What a queued change does to `parent`'s listing, which Drive does not
|
||||
/// show until it lands: the children made or moved in by a queued create,
|
||||
/// mkdir or rename, as the DB holds them, and the nodes a queued rename
|
||||
/// moves elsewhere.
|
||||
///
|
||||
/// A listing read from Drive is laid over with these, or a folder listed
|
||||
/// while a create in it is queued, after a restart say, loses the file
|
||||
/// until the create lands (`docs/BUGS.md` B132).
|
||||
pub fn queued_children(&self, parent: &NodeUid) -> Result<(Vec<Node>, HashSet<String>)> {
|
||||
let conn = self.read();
|
||||
let parent = parent.to_string();
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT n.node_json FROM nodes n
|
||||
WHERE n.parent_uid = ?1 AND n.node_json IS NOT NULL AND n.trashed = 0
|
||||
AND EXISTS (SELECT 1 FROM pending_op p
|
||||
WHERE p.uid = n.uid AND p.kind IN (?2, ?3, ?4))",
|
||||
)?;
|
||||
let rows = stmt.query_map(params![parent, OP_CREATE, OP_MKDIR, OP_RENAME], |r| {
|
||||
r.get::<_, String>(0)
|
||||
})?;
|
||||
let mut here = Vec::new();
|
||||
for json in rows {
|
||||
here.push(serde_json::from_str(&json?)?);
|
||||
}
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT DISTINCT p.uid FROM pending_op p JOIN nodes n ON n.uid = p.uid
|
||||
WHERE p.kind = ?2 AND n.parent_uid IS NOT ?1",
|
||||
)?;
|
||||
let gone = stmt
|
||||
.query_map(params![parent, OP_RENAME], |r| r.get::<_, String>(0))?
|
||||
.collect::<std::result::Result<HashSet<_>, _>>()?;
|
||||
Ok((here, gone))
|
||||
}
|
||||
|
||||
// --- Content-cache LRU index (P4) -------------------------------------
|
||||
//
|
||||
// Replaces the per-eviction `read_dir` scans in `ContentCache`. Each cached
|
||||
@@ -671,9 +733,18 @@ pub(super) fn adopt_placeholder_row_tx(
|
||||
local: &str,
|
||||
real: &str,
|
||||
) -> Result<()> {
|
||||
// The node a row holds names its parent too, and is what a restart builds
|
||||
// the tree from: a folder that landed with no write to its children after
|
||||
// left them under a parent no row has (`docs/BUGS.md` B140).
|
||||
let real_json = parse_node_uid(real)
|
||||
.map(|uid| serde_json::to_string(&uid))
|
||||
.transpose()?;
|
||||
tx.execute(
|
||||
"UPDATE nodes SET parent_uid = ?2 WHERE parent_uid = ?1",
|
||||
params![local, real],
|
||||
"UPDATE nodes SET parent_uid = ?2,
|
||||
node_json = CASE WHEN ?3 IS NULL OR node_json IS NULL THEN node_json
|
||||
ELSE json_set(node_json, '$.parent_uid', json(?3)) END
|
||||
WHERE parent_uid = ?1",
|
||||
params![local, real, real_json],
|
||||
)?;
|
||||
let row: Option<(i64, Option<String>)> = tx
|
||||
.query_row(
|
||||
|
||||
+174
-11
@@ -92,6 +92,13 @@ pub fn op_supersedes(kind: &str) -> bool {
|
||||
/// Other ops are taken from the node tree only for revisions. A queued rename
|
||||
/// that moved a node *into* the folder has to land, or the node would stay
|
||||
/// where it was on the server instead of going to the trash with the folder.
|
||||
///
|
||||
/// A create already on the wire cannot be stopped, so its row is not deleted
|
||||
/// but turned into a trash of the node it is making, still claimed by the
|
||||
/// upload. [`Db::finish_create`] points it at the landed node. Until then it
|
||||
/// holds the name, as any queued trash does ([`Db::has_pending_trash_named`]):
|
||||
/// a rename over the file sent first found the name taken and landed under a
|
||||
/// conflict name (`docs/BUGS.md` B129).
|
||||
fn drop_doomed_ops(tx: &rusqlite::Transaction<'_>, uid: &str) -> Result<Vec<String>> {
|
||||
const DOOMED: &str = "
|
||||
WITH RECURSIVE
|
||||
@@ -124,12 +131,48 @@ fn drop_doomed_ops(tx: &rusqlite::Transaction<'_>, uid: &str) -> Result<Vec<Stri
|
||||
rows.collect::<std::result::Result<Vec<_>, _>>()?
|
||||
};
|
||||
tx.execute(
|
||||
&format!("{DOOMED} DELETE FROM pending_op WHERE id IN (SELECT id FROM doomed)"),
|
||||
&format!(
|
||||
"{DOOMED}
|
||||
UPDATE pending_op
|
||||
SET kind = 'trash', parent_uid = NULL, blob_path = NULL, meta_json = NULL
|
||||
WHERE id IN (SELECT id FROM doomed)
|
||||
AND kind IN ('create', 'mkdir') AND claimed_at <> 0"
|
||||
),
|
||||
params![uid],
|
||||
)?;
|
||||
tx.execute(
|
||||
&format!(
|
||||
"{DOOMED}
|
||||
DELETE FROM pending_op
|
||||
WHERE id IN (SELECT id FROM doomed)
|
||||
AND NOT (kind = 'trash' AND claimed_at <> 0 AND uid LIKE '{LOCAL_VOLUME}~%')"
|
||||
),
|
||||
params![uid],
|
||||
)?;
|
||||
Ok(blobs)
|
||||
}
|
||||
|
||||
/// The `pending_op` rows, `?1` being [`OP_RENAME`], that move a node away from
|
||||
/// the name `?2`.
|
||||
const MOVES_FROM: &str = "kind = ?1
|
||||
AND json_extract(meta_json, '$.original_name') = ?2
|
||||
AND (name <> ?2 OR parent_uid <> json_extract(meta_json, '$.original_parent_uid'))";
|
||||
|
||||
/// [`Db::wake_ops_waiting_for`] within a transaction.
|
||||
pub(super) fn wake_ops_waiting_for_tx(tx: &rusqlite::Transaction<'_>, name: &str) -> Result<()> {
|
||||
tx.execute(
|
||||
&format!(
|
||||
"UPDATE pending_op SET next_attempt_at = 0
|
||||
WHERE name = ?2 AND kind IN (?3, ?4, ?1) AND attempts > 0
|
||||
AND next_attempt_at < ?5
|
||||
AND NOT EXISTS (SELECT 1 FROM pending_op WHERE kind = ?6 AND name = ?2)
|
||||
AND NOT EXISTS (SELECT 1 FROM pending_op WHERE {MOVES_FROM})"
|
||||
),
|
||||
params![OP_RENAME, name, OP_CREATE, OP_MKDIR, PARK_UNTIL, OP_TRASH],
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The volume id given to a node that exists only on this machine, so far. A
|
||||
/// real [`NodeUid`] is `{volume}~{link}`, so a placeholder is `local~<uuid>` and
|
||||
/// round-trips through the same `Display`/parse path as any other uid.
|
||||
@@ -242,6 +285,20 @@ pub struct CreateLanding<'a> {
|
||||
pub landed: (&'a str, &'a str),
|
||||
}
|
||||
|
||||
/// What became of a queued create when it landed, from [`Db::finish_create`].
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum CreateRetired {
|
||||
/// Its row is gone and nothing of it is left queued.
|
||||
Landed,
|
||||
/// A newer write attached to it on the wire and is now queued as a
|
||||
/// revision of the real node: its blob and its rewritten sidecar.
|
||||
Newer { blob: String, meta: String },
|
||||
/// Its row had been dropped while it was on the wire: the node was deleted,
|
||||
/// or a rename replaced it. The node Drive just made is unwanted, and a
|
||||
/// trash of it is queued under this id, claimed by the caller.
|
||||
Withdrawn { trash: i64 },
|
||||
}
|
||||
|
||||
struct CreateRow {
|
||||
blob: Option<String>,
|
||||
meta: Option<String>,
|
||||
@@ -278,6 +335,32 @@ impl Db {
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
/// Whether a queued rename moves a node away from `name`.
|
||||
///
|
||||
/// Like a queued trash ([`Db::has_pending_trash_named`]) it frees the name
|
||||
/// once it lands, so an op that wants the name can wait for it. A create
|
||||
/// renamed while it was on the wire queues one (`Db::finish_create`).
|
||||
pub fn has_pending_move_from(&self, name: &str) -> Result<bool> {
|
||||
let conn = self.read();
|
||||
conn.query_row(
|
||||
&format!("SELECT EXISTS(SELECT 1 FROM pending_op WHERE {MOVES_FROM})"),
|
||||
params![OP_RENAME, name],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
/// Make the ops that failed because `name` was held due at once, now that
|
||||
/// what held it let go. Nothing changes while another queued trash or
|
||||
/// rename still holds it, or for a parked op.
|
||||
pub fn wake_ops_waiting_for(&self, name: &str) -> Result<()> {
|
||||
let mut conn = self.conn.lock();
|
||||
let tx = conn.transaction()?;
|
||||
wake_ops_waiting_for_tx(&tx, name)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Whether a specific desired-state operation is still queued for a node.
|
||||
pub fn has_pending_op(&self, uid: &str, kind: &str) -> Result<bool> {
|
||||
let conn = self.read();
|
||||
@@ -304,21 +387,29 @@ impl Db {
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
/// Queue `op`, returning its id and the staged blob of the op it replaced.
|
||||
///
|
||||
/// A rename a drain worker holds is not replaced but queued behind: the
|
||||
/// claim keeps the newer one from being sent until it retires, so the two
|
||||
/// land in the order they were made (`docs/BUGS.md` B141). Should it fail,
|
||||
/// [`Db::record_op_failure`] drops it.
|
||||
pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option<String>)> {
|
||||
let mut conn = self.conn.lock();
|
||||
let tx = conn.transaction()?;
|
||||
let superseded: Option<String> = if op_supersedes(&op.kind) {
|
||||
let blob: Option<String> = tx
|
||||
.query_row(
|
||||
"SELECT blob_path FROM pending_op WHERE uid = ?1 AND kind = ?2",
|
||||
params![op.uid, op.kind],
|
||||
"SELECT blob_path FROM pending_op
|
||||
WHERE uid = ?1 AND kind = ?2 AND (kind <> ?3 OR claimed_at = 0)",
|
||||
params![op.uid, op.kind, OP_RENAME],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.optional()?
|
||||
.flatten();
|
||||
tx.execute(
|
||||
"DELETE FROM pending_op WHERE uid = ?1 AND kind = ?2",
|
||||
params![op.uid, op.kind],
|
||||
"DELETE FROM pending_op
|
||||
WHERE uid = ?1 AND kind = ?2 AND (kind <> ?3 OR claimed_at = 0)",
|
||||
params![op.uid, op.kind, OP_RENAME],
|
||||
)?;
|
||||
blob
|
||||
} else {
|
||||
@@ -429,21 +520,28 @@ impl Db {
|
||||
/// row names another parent or name than `landing.sent`, a rename of `real`
|
||||
/// to that target is queued in the same transaction, or the move would be
|
||||
/// lost with the row.
|
||||
///
|
||||
/// A delete while the create is on the wire cannot stop the upload. It
|
||||
/// turns the row into a trash ([`Db::delete_ops_for_uid`]), which is
|
||||
/// pointed at `real` here, still claimed by the caller, so the node is not
|
||||
/// left on Drive holding the name (`docs/BUGS.md` B129). A row gone
|
||||
/// altogether is queued as such a trash. The placeholder row is left to
|
||||
/// the delete that withdrew it.
|
||||
pub fn finish_create(
|
||||
&self,
|
||||
id: i64,
|
||||
uploaded: Option<&str>,
|
||||
landing: &CreateLanding<'_>,
|
||||
rewrite: impl FnOnce(&str) -> Option<String>,
|
||||
) -> Result<Option<(String, String)>> {
|
||||
) -> Result<CreateRetired> {
|
||||
let (local, real) = (landing.local, landing.real);
|
||||
let mut conn = self.conn.lock();
|
||||
let tx = conn.transaction()?;
|
||||
let row: Option<CreateRow> = tx
|
||||
.query_row(
|
||||
"SELECT blob_path, meta_json, parent_uid, name, created_at
|
||||
FROM pending_op WHERE id = ?1",
|
||||
params![id],
|
||||
FROM pending_op WHERE id = ?1 AND kind <> ?2",
|
||||
params![id, OP_TRASH],
|
||||
|r| {
|
||||
Ok(CreateRow {
|
||||
blob: r.get(0)?,
|
||||
@@ -455,6 +553,28 @@ impl Db {
|
||||
},
|
||||
)
|
||||
.optional()?;
|
||||
let withdrawn = tx.execute(
|
||||
"UPDATE pending_op SET uid = ?2, name = ?3 WHERE id = ?1 AND kind = ?4",
|
||||
params![id, real, landing.landed.1, OP_TRASH],
|
||||
)?;
|
||||
if withdrawn > 0 {
|
||||
tx.commit()?;
|
||||
return Ok(CreateRetired::Withdrawn { trash: id });
|
||||
}
|
||||
if row.is_none() {
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map_or(1, |d| d.as_millis() as i64);
|
||||
tx.execute(
|
||||
"INSERT INTO pending_op
|
||||
(kind, uid, parent_uid, name, created_at, next_attempt_at, claimed_at)
|
||||
VALUES (?1, ?2, NULL, ?3, ?4, 0, ?4)",
|
||||
params![OP_TRASH, real, landing.landed.1, now],
|
||||
)?;
|
||||
let trash = tx.last_insert_rowid();
|
||||
tx.commit()?;
|
||||
return Ok(CreateRetired::Withdrawn { trash });
|
||||
}
|
||||
let newer = match &row {
|
||||
Some(CreateRow {
|
||||
blob: Some(blob),
|
||||
@@ -505,7 +625,10 @@ impl Db {
|
||||
)?;
|
||||
adopt_placeholder_row_tx(&tx, local, real)?;
|
||||
tx.commit()?;
|
||||
Ok(newer)
|
||||
Ok(match newer {
|
||||
Some((blob, meta)) => CreateRetired::Newer { blob, meta },
|
||||
None => CreateRetired::Landed,
|
||||
})
|
||||
}
|
||||
|
||||
/// Replace the sidecar of a queued op, for a baseline that has moved under
|
||||
@@ -539,12 +662,17 @@ impl Db {
|
||||
/// *is* the rename. Returns false when the create has already drained, in
|
||||
/// which case the node has a real uid and the caller must rename it there
|
||||
/// instead (offline.md Phase 3b).
|
||||
///
|
||||
/// A create backing off is made due: what it failed on may have been the
|
||||
/// name it no longer wants, such as one held by a queued change. A parked
|
||||
/// create stays parked.
|
||||
pub fn rewrite_op_target(&self, uid: &str, parent_uid: &str, name: &str) -> Result<bool> {
|
||||
let conn = self.conn.lock();
|
||||
let n = conn.execute(
|
||||
"UPDATE pending_op SET parent_uid = ?2, name = ?3
|
||||
"UPDATE pending_op SET parent_uid = ?2, name = ?3,
|
||||
next_attempt_at = CASE WHEN next_attempt_at < ?6 THEN 0 ELSE next_attempt_at END
|
||||
WHERE uid = ?1 AND kind IN (?4, ?5)",
|
||||
params![uid, parent_uid, name, OP_CREATE, OP_MKDIR],
|
||||
params![uid, parent_uid, name, OP_CREATE, OP_MKDIR, PARK_UNTIL],
|
||||
)?;
|
||||
Ok(n > 0)
|
||||
}
|
||||
@@ -978,8 +1106,22 @@ impl Db {
|
||||
/// Clears any access-deferral window: whatever the op was waiting for has
|
||||
/// now been reported, so the next uncleared deferral starts its own window
|
||||
/// rather than escalating again on the following recheck.
|
||||
///
|
||||
/// A rename with a newer one queued behind it ([`Db::enqueue_op`]) is
|
||||
/// dropped instead: retried, it could land after the newer one and undo it.
|
||||
pub fn record_op_failure(&self, id: i64, error: &str, next_attempt_at: i64) -> Result<()> {
|
||||
let conn = self.conn.lock();
|
||||
let dropped = conn.execute(
|
||||
"DELETE FROM pending_op
|
||||
WHERE id = ?1 AND kind = ?2
|
||||
AND EXISTS (SELECT 1 FROM pending_op AS newer
|
||||
WHERE newer.uid = pending_op.uid AND newer.kind = ?2
|
||||
AND newer.id > ?1)",
|
||||
params![id, OP_RENAME],
|
||||
)?;
|
||||
if dropped > 0 {
|
||||
return Ok(());
|
||||
}
|
||||
conn.execute(
|
||||
"UPDATE pending_op
|
||||
SET attempts = attempts + 1, last_error = ?2, next_attempt_at = ?3,
|
||||
@@ -990,6 +1132,27 @@ impl Db {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Make a create due again if it was renamed or moved
|
||||
/// ([`Db::rewrite_op_target`]) while the attempt that just failed was on
|
||||
/// the wire, `parent_uid` and `name` being what that attempt sent. What it
|
||||
/// failed on may have been the target it no longer has. Returns whether it
|
||||
/// was.
|
||||
pub fn retry_if_retargeted(
|
||||
&self,
|
||||
id: i64,
|
||||
parent_uid: Option<&str>,
|
||||
name: Option<&str>,
|
||||
) -> Result<bool> {
|
||||
let conn = self.conn.lock();
|
||||
let n = conn.execute(
|
||||
"UPDATE pending_op SET next_attempt_at = 0
|
||||
WHERE id = ?1 AND kind IN (?4, ?5) AND next_attempt_at < ?6
|
||||
AND (parent_uid IS NOT ?2 OR name IS NOT ?3)",
|
||||
params![id, parent_uid, name, OP_CREATE, OP_MKDIR, PARK_UNTIL],
|
||||
)?;
|
||||
Ok(n > 0)
|
||||
}
|
||||
|
||||
/// Defer a due operation without recording a failed remote attempt.
|
||||
///
|
||||
/// Access checks happen before the drain calls Proton, so a locally denied
|
||||
|
||||
@@ -2474,6 +2474,125 @@ fn a_queued_trash_holds_its_name_until_it_lands() {
|
||||
assert!(!db.has_pending_trash_named("report.txt").unwrap());
|
||||
}
|
||||
|
||||
/// A create that found its name held by a queued trash backed off. The trash
|
||||
/// landing frees the name, so the create is due at once instead of after the
|
||||
/// backoff; a parked create stays parked.
|
||||
#[test]
|
||||
fn a_landed_trash_makes_the_ops_waiting_for_its_name_due() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
let old = uid("old");
|
||||
db.upsert_node(&file("old", "root", "report.txt", 1))
|
||||
.unwrap();
|
||||
let (trash_id, _) = db
|
||||
.replace_ops_with_trash(&old.to_string(), "report.txt", 1)
|
||||
.unwrap();
|
||||
let create = |name: &str, next_attempt_at: i64| {
|
||||
let (id, _) = db
|
||||
.enqueue_op(&PendingOp {
|
||||
id: 0,
|
||||
kind: OP_CREATE.to_string(),
|
||||
uid: format!("local~{name}-{next_attempt_at}"),
|
||||
parent_uid: Some(uid("root").to_string()),
|
||||
name: Some(name.to_string()),
|
||||
blob_path: None,
|
||||
meta_json: None,
|
||||
created_at: 2,
|
||||
attempts: 0,
|
||||
last_error: None,
|
||||
next_attempt_at,
|
||||
})
|
||||
.unwrap();
|
||||
db.record_op_failure(id, "report.txt is still held", next_attempt_at)
|
||||
.unwrap();
|
||||
id
|
||||
};
|
||||
let waiting = create("report.txt", 9_000);
|
||||
let parked = create("report.txt", PARK_UNTIL);
|
||||
let other = create("other.txt", 9_000);
|
||||
|
||||
db.complete_trash_op(trash_id, &old).unwrap();
|
||||
|
||||
let due = |id| {
|
||||
db.pending_ops()
|
||||
.unwrap()
|
||||
.into_iter()
|
||||
.find(|op| op.id == id)
|
||||
.unwrap()
|
||||
.next_attempt_at
|
||||
};
|
||||
assert_eq!(due(waiting), 0);
|
||||
assert_eq!(due(parked), PARK_UNTIL);
|
||||
assert_eq!(due(other), 9_000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_queued_rename_away_from_a_name_holds_it_until_it_lands() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
let root = uid("root").to_string();
|
||||
let rename = |node: &str, from: &str, to: &str| {
|
||||
let meta = serde_json::to_string(&RenameMeta {
|
||||
original_parent_uid: root.clone(),
|
||||
original_name: Some(from.to_string()),
|
||||
})
|
||||
.unwrap();
|
||||
db.enqueue_op(&PendingOp {
|
||||
id: 0,
|
||||
kind: OP_RENAME.to_string(),
|
||||
uid: uid(node).to_string(),
|
||||
parent_uid: Some(root.clone()),
|
||||
name: Some(to.to_string()),
|
||||
blob_path: None,
|
||||
meta_json: Some(meta),
|
||||
created_at: 1,
|
||||
attempts: 0,
|
||||
last_error: None,
|
||||
next_attempt_at: 0,
|
||||
})
|
||||
.unwrap()
|
||||
.0
|
||||
};
|
||||
let away = rename("away", "c.bin", "d");
|
||||
rename("back", "e.md", "e.md");
|
||||
let (waiting, _) = db
|
||||
.enqueue_op(&PendingOp {
|
||||
id: 0,
|
||||
kind: OP_CREATE.to_string(),
|
||||
uid: "local~waiting".to_string(),
|
||||
parent_uid: Some(root.clone()),
|
||||
name: Some("c.bin".to_string()),
|
||||
blob_path: None,
|
||||
meta_json: None,
|
||||
created_at: 2,
|
||||
attempts: 0,
|
||||
last_error: None,
|
||||
next_attempt_at: 0,
|
||||
})
|
||||
.unwrap();
|
||||
db.record_op_failure(waiting, "c.bin is still held", 9_000)
|
||||
.unwrap();
|
||||
|
||||
assert!(db.has_pending_move_from("c.bin").unwrap());
|
||||
assert!(!db.has_pending_move_from("d").unwrap());
|
||||
assert!(
|
||||
!db.has_pending_move_from("e.md").unwrap(),
|
||||
"a rename that keeps the name in its folder frees nothing"
|
||||
);
|
||||
db.wake_ops_waiting_for("c.bin").unwrap();
|
||||
let due = || {
|
||||
db.pending_ops()
|
||||
.unwrap()
|
||||
.into_iter()
|
||||
.find(|op| op.id == waiting)
|
||||
.unwrap()
|
||||
.next_attempt_at
|
||||
};
|
||||
assert_eq!(due(), 9_000, "the rename still holds the name");
|
||||
|
||||
db.delete_op(away).unwrap();
|
||||
db.wake_ops_waiting_for("c.bin").unwrap();
|
||||
assert_eq!(due(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn failed_atomic_trash_insert_keeps_prior_revision_and_blob_ownership() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
@@ -2796,6 +2915,106 @@ fn renaming_a_queued_create_rewrites_its_target() {
|
||||
);
|
||||
}
|
||||
|
||||
/// A create that failed on a name held by a queued change waits out a
|
||||
/// backoff. Renamed away from that name, it is due again at once; parked, it
|
||||
/// stays parked.
|
||||
#[test]
|
||||
fn renaming_a_backing_off_create_makes_it_due() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
let create = |local: &str| PendingOp {
|
||||
id: 0,
|
||||
kind: OP_CREATE.to_string(),
|
||||
uid: local.to_string(),
|
||||
parent_uid: Some(uid("dir").to_string()),
|
||||
name: Some("a.txt".to_string()),
|
||||
blob_path: Some(format!("/staging/{local}")),
|
||||
meta_json: Some("{}".to_string()),
|
||||
created_at: 1,
|
||||
attempts: 0,
|
||||
last_error: None,
|
||||
next_attempt_at: 0,
|
||||
};
|
||||
let (failed, _) = db.enqueue_op(&create("local~failed")).unwrap();
|
||||
db.record_op_failure(failed, "a.txt is still held", 10_000)
|
||||
.unwrap();
|
||||
db.enqueue_op(&create("local~parked")).unwrap();
|
||||
db.set_create_hold("local~parked", true).unwrap();
|
||||
|
||||
db.rewrite_op_target("local~failed", &uid("dir").to_string(), "d")
|
||||
.unwrap();
|
||||
db.rewrite_op_target("local~parked", &uid("dir").to_string(), "e.tmp")
|
||||
.unwrap();
|
||||
|
||||
let ops = db.pending_ops().unwrap();
|
||||
let op = |local: &str| ops.iter().find(|o| o.uid == local).unwrap();
|
||||
assert_eq!(op("local~failed").next_attempt_at, 0);
|
||||
assert_eq!(op("local~failed").attempts, 1);
|
||||
assert_eq!(op("local~parked").next_attempt_at, PARK_UNTIL);
|
||||
|
||||
// Renamed while an attempt was on the wire, which then failed: the
|
||||
// failure is recorded after the rename, and the create is still due.
|
||||
let sent = op("local~failed").clone();
|
||||
db.rewrite_op_target("local~failed", &uid("dir").to_string(), "f")
|
||||
.unwrap();
|
||||
db.record_op_failure(sent.id, "d is still held", 10_000)
|
||||
.unwrap();
|
||||
assert!(
|
||||
db.retry_if_retargeted(sent.id, sent.parent_uid.as_deref(), sent.name.as_deref())
|
||||
.unwrap()
|
||||
);
|
||||
let ops = db.pending_ops().unwrap();
|
||||
assert_eq!(
|
||||
ops.iter()
|
||||
.find(|o| o.id == sent.id)
|
||||
.unwrap()
|
||||
.next_attempt_at,
|
||||
0
|
||||
);
|
||||
assert!(
|
||||
!db.retry_if_retargeted(sent.id, sent.parent_uid.as_deref(), Some("f"))
|
||||
.unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
/// A rename queued while an older one is on the wire waits for it. The older
|
||||
/// one, if it fails, is dropped rather than retried after the newer one.
|
||||
#[test]
|
||||
fn a_rename_on_the_wire_lands_before_the_one_that_replaces_it() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
let file = uid("file").to_string();
|
||||
let rename = |parent: &str| PendingOp {
|
||||
id: 0,
|
||||
kind: OP_RENAME.to_string(),
|
||||
uid: file.clone(),
|
||||
parent_uid: Some(uid(parent).to_string()),
|
||||
name: Some("e.md".to_string()),
|
||||
blob_path: None,
|
||||
meta_json: None,
|
||||
created_at: 1,
|
||||
attempts: 0,
|
||||
last_error: None,
|
||||
next_attempt_at: 0,
|
||||
};
|
||||
let (first, _) = db.enqueue_op(&rename("x")).unwrap();
|
||||
assert_eq!(db.claim_next_due_op(1).unwrap().unwrap().id, first);
|
||||
|
||||
let (second, _) = db.enqueue_op(&rename("c0")).unwrap();
|
||||
assert!(db.claim_next_due_op(1).unwrap().is_none());
|
||||
|
||||
db.record_op_failure(first, "original hash is out of date", 10_000)
|
||||
.unwrap();
|
||||
db.release_op_claim(first).unwrap();
|
||||
let ops = db.pending_ops().unwrap();
|
||||
assert_eq!(ops.iter().map(|o| o.id).collect::<Vec<_>>(), [second]);
|
||||
assert_eq!(db.claim_next_due_op(1).unwrap().unwrap().id, second);
|
||||
|
||||
// Unclaimed, a queued rename is replaced as before.
|
||||
db.release_op_claim(second).unwrap();
|
||||
let (third, _) = db.enqueue_op(&rename("x")).unwrap();
|
||||
let ops = db.pending_ops().unwrap();
|
||||
assert_eq!(ops.iter().map(|o| o.id).collect::<Vec<_>>(), [third]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_write_folds_into_a_queued_create_instead_of_superseding_it() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
@@ -2885,10 +3104,10 @@ fn a_write_attached_while_its_create_uploads_becomes_a_revision() {
|
||||
|
||||
assert_eq!(
|
||||
newer,
|
||||
Some((
|
||||
"/staging/written".to_string(),
|
||||
"{\"uid\":\"vol~real\"}".to_string()
|
||||
))
|
||||
CreateRetired::Newer {
|
||||
blob: "/staging/written".to_string(),
|
||||
meta: "{\"uid\":\"vol~real\"}".to_string()
|
||||
}
|
||||
);
|
||||
let ops = db.pending_ops().unwrap();
|
||||
assert_eq!(ops.len(), 1, "the write is still queued");
|
||||
@@ -2943,7 +3162,7 @@ fn a_create_whose_blob_landed_is_retired() {
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(newer, None);
|
||||
assert_eq!(newer, CreateRetired::Landed);
|
||||
let ops = db.pending_ops().unwrap();
|
||||
assert_eq!(ops.len(), 1, "only the child is left");
|
||||
assert_eq!(ops[0].parent_uid.as_deref(), Some(real.as_str()));
|
||||
@@ -3102,6 +3321,60 @@ fn a_landed_folder_keeps_a_row_until_the_server_copy_replaces_it() {
|
||||
assert_eq!(found[0].path, "New folder/inside.txt");
|
||||
}
|
||||
|
||||
/// A file deleted while its create was on the wire still lands: nothing stops
|
||||
/// an upload in flight. The node it made is unwanted, so a trash of it is
|
||||
/// queued in the same transaction, already claimed by the drain that landed it
|
||||
/// (`docs/BUGS.md` B129).
|
||||
#[test]
|
||||
fn a_create_withdrawn_while_it_uploads_queues_a_trash_of_the_landed_node() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
let local = "local~gone";
|
||||
let real = uid("real").to_string();
|
||||
let parent = uid("parent").to_string();
|
||||
let (id, _) = db
|
||||
.enqueue_op(&PendingOp {
|
||||
id: 0,
|
||||
kind: OP_CREATE.to_string(),
|
||||
uid: local.to_string(),
|
||||
parent_uid: Some(parent.clone()),
|
||||
name: Some("d".to_string()),
|
||||
blob_path: Some("/staging/sent".to_string()),
|
||||
meta_json: Some("{}".to_string()),
|
||||
created_at: 1,
|
||||
attempts: 0,
|
||||
last_error: None,
|
||||
next_attempt_at: 0,
|
||||
})
|
||||
.unwrap();
|
||||
db.claim_next_due_op(10)
|
||||
.unwrap()
|
||||
.expect("the drain takes the create");
|
||||
db.delete_ops_for_uid(local).unwrap();
|
||||
|
||||
let retired = db
|
||||
.finish_create(
|
||||
id,
|
||||
Some("/staging/sent"),
|
||||
&landing(local, &real, &parent, "d"),
|
||||
|_| None,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let CreateRetired::Withdrawn { trash } = retired else {
|
||||
panic!("a withdrawn create is not landed: {retired:?}");
|
||||
};
|
||||
let ops = db.pending_ops().unwrap();
|
||||
assert_eq!(ops.len(), 1);
|
||||
assert_eq!(ops[0].id, trash);
|
||||
assert_eq!(ops[0].kind, OP_TRASH);
|
||||
assert_eq!(ops[0].uid, real);
|
||||
assert_eq!(ops[0].name.as_deref(), Some("d"));
|
||||
assert!(
|
||||
db.claim_next_due_op(i64::MAX / 2).unwrap().is_none(),
|
||||
"the trash is the landing drain's to send"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_landed_placeholder_never_overwrites_the_server_copy() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
@@ -3194,6 +3467,7 @@ fn a_landed_placeholder_keeps_its_lid() {
|
||||
|
||||
assert_eq!(db.lid_of(&local.to_string()).unwrap(), None);
|
||||
assert_eq!(db.lid_of(&real.to_string()).unwrap(), Some(lid));
|
||||
assert_eq!(db.uid_of_lid(lid).unwrap(), Some(real.to_string()));
|
||||
let name = if known {
|
||||
"New folder (1)"
|
||||
} else {
|
||||
@@ -3207,6 +3481,85 @@ fn a_landed_placeholder_keeps_its_lid() {
|
||||
}
|
||||
}
|
||||
|
||||
/// Drive lists none of what is queued for a folder, so a listing read from
|
||||
/// it is laid over with what the queue made, moved in and moved away (B132).
|
||||
#[test]
|
||||
fn queued_children_are_what_drive_does_not_list_yet() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
db.upsert_node(&folder("root", None, "My Files")).unwrap();
|
||||
db.upsert_node(&folder("dir", Some("root"), "dir")).unwrap();
|
||||
db.upsert_node(&folder("other", Some("root"), "other"))
|
||||
.unwrap();
|
||||
let op = |kind: &str, uid: &str, parent: &str, name: &str| PendingOp {
|
||||
id: 0,
|
||||
kind: kind.to_string(),
|
||||
uid: uid.to_string(),
|
||||
parent_uid: Some(parent.to_string()),
|
||||
name: Some(name.to_string()),
|
||||
blob_path: None,
|
||||
meta_json: None,
|
||||
created_at: 1,
|
||||
attempts: 0,
|
||||
last_error: None,
|
||||
next_attempt_at: 0,
|
||||
};
|
||||
let dir = uid("dir").to_string();
|
||||
let other = uid("other").to_string();
|
||||
|
||||
let local = NodeUid::new(VolumeId::from("local"), LinkId::from("new"));
|
||||
let mut made = file("x", "dir", "made.txt", 3);
|
||||
made.uid = local.clone();
|
||||
db.upsert_node(&made).unwrap();
|
||||
db.enqueue_op(&op(OP_CREATE, &local.to_string(), &dir, "made.txt"))
|
||||
.unwrap();
|
||||
db.upsert_node(&file("in", "dir", "in.txt", 3)).unwrap();
|
||||
db.enqueue_op(&op(OP_RENAME, &uid("in").to_string(), &dir, "in.txt"))
|
||||
.unwrap();
|
||||
db.upsert_node(&file("out", "other", "out.txt", 3)).unwrap();
|
||||
db.enqueue_op(&op(OP_RENAME, &uid("out").to_string(), &other, "out.txt"))
|
||||
.unwrap();
|
||||
db.upsert_node(&file("still", "dir", "still.txt", 3))
|
||||
.unwrap();
|
||||
|
||||
let (here, gone) = db.queued_children(&uid("dir")).unwrap();
|
||||
let mut names: Vec<_> = here.iter().map(|n| n.name.as_str()).collect();
|
||||
names.sort();
|
||||
assert_eq!(names, ["in.txt", "made.txt"]);
|
||||
assert_eq!(
|
||||
gone,
|
||||
std::collections::HashSet::from([uid("out").to_string()])
|
||||
);
|
||||
}
|
||||
|
||||
/// A restart builds the tree from the node each row holds. A folder that
|
||||
/// landed handed its children to the real uid in the row's column only, so
|
||||
/// they came back under a parent no row has (B140).
|
||||
#[test]
|
||||
fn a_landed_folder_hands_its_children_over_in_the_stored_node_too() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
db.upsert_node(&folder("root", None, "My Files")).unwrap();
|
||||
let local = NodeUid::new(VolumeId::from("local"), LinkId::from("dir"));
|
||||
let mut placeholder = folder("x", Some("root"), "dir");
|
||||
placeholder.uid = local.clone();
|
||||
db.upsert_node(&placeholder).unwrap();
|
||||
let mut child = file("child", "x", "a.txt", 3);
|
||||
child.parent_uid = Some(local.clone());
|
||||
db.upsert_node(&child).unwrap();
|
||||
|
||||
db.remap_local_uid(&local.to_string(), &uid("dir").to_string())
|
||||
.unwrap();
|
||||
|
||||
let stored = db.node_by_uid(&uid("child").to_string()).unwrap().unwrap();
|
||||
assert_eq!(stored.parent_uid, Some(uid("dir")));
|
||||
let names: Vec<_> = db
|
||||
.known_children(&uid("dir"))
|
||||
.unwrap()
|
||||
.into_iter()
|
||||
.map(|n| n.name)
|
||||
.collect();
|
||||
assert_eq!(names, ["a.txt"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn migration_v36_gives_every_node_its_rowid_as_lid() {
|
||||
let path = std::env::temp_dir().join(format!(
|
||||
|
||||
Reference in new issue
Block a user