mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-04 16:56:59 +01:00
implemented real offline structure, with online drain
This commit is contained in:
1 parent
dcc62ba420
commit
50c96cbbf4
7 files changed
+1333
-195
No files matched your search
@@ -12,6 +12,7 @@ use pdfs_core::cache::ContentCache;
|
|||||||
use pdfs_core::config::AppDirs;
|
use pdfs_core::config::AppDirs;
|
||||||
use pdfs_core::control::{
|
use pdfs_core::control::{
|
||||||
RefreshScope, Request as CtlRequest, Response as CtlResponse, ShareEntryKind, SyncPhase,
|
RefreshScope, Request as CtlRequest, Response as CtlResponse, ShareEntryKind, SyncPhase,
|
||||||
|
pending_summary,
|
||||||
};
|
};
|
||||||
use pdfs_core::db::Db;
|
use pdfs_core::db::Db;
|
||||||
|
|
||||||
@@ -832,13 +833,13 @@ fn cmd_status() -> Result<()> {
|
|||||||
pinned,
|
pinned,
|
||||||
online,
|
online,
|
||||||
pending_uploads,
|
pending_uploads,
|
||||||
|
pending_changes,
|
||||||
..
|
..
|
||||||
}) => {
|
}) => {
|
||||||
let state = if online { "" } else { ", offline" };
|
let state = if online { "" } else { ", offline" };
|
||||||
let queued = match pending_uploads {
|
let queued = match pending_summary(pending_uploads, pending_changes) {
|
||||||
0 => String::new(),
|
Some(s) => format!(", {s}"),
|
||||||
1 => ", 1 upload queued".to_string(),
|
None => String::new(),
|
||||||
n => format!(", {n} uploads queued"),
|
|
||||||
};
|
};
|
||||||
println!("Mounted at {mountpoint} ({pinned} pinned{state}{queued})");
|
println!("Mounted at {mountpoint} ({pinned} pinned{state}{queued})");
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -98,8 +98,8 @@ pub struct StagedWrite {
|
|||||||
pub uid: String,
|
pub uid: String,
|
||||||
/// Length of the intended new content.
|
/// Length of the intended new content.
|
||||||
pub len: u64,
|
pub len: u64,
|
||||||
/// Size and mtime of the base revision the write was made against, so a
|
/// Size and mtime of the revision the file's *untouched* ranges came from,
|
||||||
/// retry can tell whether the remote moved on in the meantime.
|
/// i.e. what a gap-fill reads. Meaningless once `complete` is true.
|
||||||
pub base_size: u64,
|
pub base_size: u64,
|
||||||
pub base_mtime: i64,
|
pub base_mtime: i64,
|
||||||
/// Locally authored `[start, end)` ranges. Everything else in the file is a
|
/// Locally authored `[start, end)` ranges. Everything else in the file is a
|
||||||
@@ -108,6 +108,28 @@ pub struct StagedWrite {
|
|||||||
/// True when `authored` covers the whole file, i.e. the staged bytes are the
|
/// True when `authored` covers the whole file, i.e. the staged bytes are the
|
||||||
/// complete new content and can be uploaded directly.
|
/// complete new content and can be uploaded directly.
|
||||||
pub complete: bool,
|
pub complete: bool,
|
||||||
|
/// The remote revision this change was made against, if it is known.
|
||||||
|
///
|
||||||
|
/// Distinct from `base_size`/`base_mtime`, which describe wherever the
|
||||||
|
/// untouched bytes are to be read from — for a write that supersedes an
|
||||||
|
/// earlier queued one, that is the *previous staged blob*, not the remote.
|
||||||
|
/// This instead always names the server's revision, carried across
|
||||||
|
/// supersedes, so the drain can tell whether the remote moved on under a
|
||||||
|
/// queued change and keep a conflict copy rather than clobber it.
|
||||||
|
///
|
||||||
|
/// `None` on a sidecar written before this field existed, and for a node
|
||||||
|
/// that has never existed remotely; both mean "do not conflict-check".
|
||||||
|
#[serde(default)]
|
||||||
|
pub based_on: Option<Baseline>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The identity of a remote revision, as far as we can observe it: a file whose
|
||||||
|
/// size and mtime both still match the ones a queued change was made against has
|
||||||
|
/// not been rewritten by anyone else in the meantime.
|
||||||
|
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
|
pub struct Baseline {
|
||||||
|
pub mtime: i64,
|
||||||
|
pub size: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Content cache rooted at a directory, with a sibling pin-registry file.
|
/// Content cache rooted at a directory, with a sibling pin-registry file.
|
||||||
@@ -1098,6 +1120,10 @@ mod tests {
|
|||||||
base_mtime: 100,
|
base_mtime: 100,
|
||||||
authored: vec![(0, 11)],
|
authored: vec![(0, 11)],
|
||||||
complete: false,
|
complete: false,
|
||||||
|
based_on: Some(Baseline {
|
||||||
|
mtime: 100,
|
||||||
|
size: 40,
|
||||||
|
}),
|
||||||
};
|
};
|
||||||
let staged = c.stage_write(&meta, &scratch).unwrap();
|
let staged = c.stage_write(&meta, &scratch).unwrap();
|
||||||
assert!(!scratch.exists(), "scratch is moved, not copied");
|
assert!(!scratch.exists(), "scratch is moved, not copied");
|
||||||
|
|||||||
@@ -625,6 +625,31 @@ fn default_online() -> bool {
|
|||||||
true
|
true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Phrase the queue depth of a [`Response::Status`] for a human, or `None` when
|
||||||
|
/// there is nothing queued and the caller should say nothing at all.
|
||||||
|
///
|
||||||
|
/// Lives here, next to the counts it describes, because the tray, the CLI and
|
||||||
|
/// the manager window all have to draw the same distinction between bytes that
|
||||||
|
/// have not reached the remote and metadata that has not.
|
||||||
|
pub fn pending_summary(uploads: u64, changes: u64) -> Option<String> {
|
||||||
|
let part = |n: u64, one: &str, many: &str| match n {
|
||||||
|
0 => None,
|
||||||
|
1 => Some(format!("1 {one}")),
|
||||||
|
n => Some(format!("{n} {many}")),
|
||||||
|
};
|
||||||
|
let parts: Vec<String> = [
|
||||||
|
part(uploads, "upload", "uploads"),
|
||||||
|
part(changes, "change", "changes"),
|
||||||
|
]
|
||||||
|
.into_iter()
|
||||||
|
.flatten()
|
||||||
|
.collect();
|
||||||
|
match parts.is_empty() {
|
||||||
|
true => None,
|
||||||
|
false => Some(format!("{} queued", parts.join(", "))),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// The daemon's reply to a [`Request`].
|
/// The daemon's reply to a [`Request`].
|
||||||
#[derive(Serialize, Deserialize, Debug)]
|
#[derive(Serialize, Deserialize, Debug)]
|
||||||
pub enum Response {
|
pub enum Response {
|
||||||
@@ -651,6 +676,11 @@ pub enum Response {
|
|||||||
/// still draining, or it cannot drain because we are offline.
|
/// still draining, or it cannot drain because we are offline.
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pending_uploads: u64,
|
pending_uploads: u64,
|
||||||
|
/// Queued mutations that carry no bytes: `mkdir`, `rename`, `trash`
|
||||||
|
/// (offline.md Phase 3b). Counted apart from `pending_uploads` because
|
||||||
|
/// calling a queued `mkdir` an upload is a lie.
|
||||||
|
#[serde(default)]
|
||||||
|
pending_changes: u64,
|
||||||
},
|
},
|
||||||
/// A human-readable success message.
|
/// A human-readable success message.
|
||||||
Ok { message: String },
|
Ok { message: String },
|
||||||
@@ -732,6 +762,20 @@ pub fn send(socket: &Path, req: &Request) -> Result<Response> {
|
|||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
|
/// The whole point of the split counts: a queued `mkdir` is work, but it is
|
||||||
|
/// not an upload and must never be reported as one.
|
||||||
|
#[test]
|
||||||
|
fn pending_summary_separates_uploads_from_other_changes() {
|
||||||
|
assert_eq!(pending_summary(0, 0), None);
|
||||||
|
assert_eq!(pending_summary(1, 0).as_deref(), Some("1 upload queued"));
|
||||||
|
assert_eq!(pending_summary(3, 0).as_deref(), Some("3 uploads queued"));
|
||||||
|
assert_eq!(pending_summary(0, 1).as_deref(), Some("1 change queued"));
|
||||||
|
assert_eq!(
|
||||||
|
pending_summary(2, 4).as_deref(),
|
||||||
|
Some("2 uploads, 4 changes queued")
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
/// The mutation requests must survive a line-delimited JSON round-trip, since
|
/// The mutation requests must survive a line-delimited JSON round-trip, since
|
||||||
/// that is exactly how they cross the control socket.
|
/// that is exactly how they cross the control socket.
|
||||||
#[test]
|
#[test]
|
||||||
|
|||||||
+335
-27
@@ -46,6 +46,37 @@ pub const OP_CREATE: &str = "create";
|
|||||||
/// time the child drains.
|
/// time the child drains.
|
||||||
pub const OP_MKDIR: &str = "mkdir";
|
pub const OP_MKDIR: &str = "mkdir";
|
||||||
|
|
||||||
|
/// The `kind` of a [`PendingOp`] that gives a node a new name, a new parent, or
|
||||||
|
/// both — the queued form of `mv` (offline.md Phase 3b).
|
||||||
|
///
|
||||||
|
/// `parent_uid` and `name` hold the node's *desired end state*, not a delta, so
|
||||||
|
/// a second rename simply replaces the row (see [`Db::enqueue_op`]) and the
|
||||||
|
/// drain can compare them against the remote and skip whichever half already
|
||||||
|
/// matches. The parent may be a `local~` placeholder — moving a file into a
|
||||||
|
/// folder that was itself created offline — in which case it is rewritten by
|
||||||
|
/// that folder's own drain, exactly as for [`OP_CREATE`].
|
||||||
|
pub const OP_RENAME: &str = "rename";
|
||||||
|
|
||||||
|
/// The `kind` of a [`PendingOp`] that trashes a node the server knows about
|
||||||
|
/// (offline.md Phase 3b).
|
||||||
|
///
|
||||||
|
/// A node that only ever existed locally never gets one of these: there is
|
||||||
|
/// nothing to trash remotely, so deleting it just drops its queued ops
|
||||||
|
/// ([`Db::delete_ops_for_uid`]).
|
||||||
|
pub const OP_TRASH: &str = "trash";
|
||||||
|
|
||||||
|
/// Whether at most one op of this `kind` may be queued per node, so that a newer
|
||||||
|
/// one replaces the older rather than queueing behind it.
|
||||||
|
///
|
||||||
|
/// True exactly of the kinds that describe a node's desired *end state* rather
|
||||||
|
/// than a step towards it: the newest revision already contains every earlier
|
||||||
|
/// one's bytes, the newest name is the only name wanted, and a trash subsumes
|
||||||
|
/// anything queued before it. A `create`/`mkdir` is the opposite — it is the one
|
||||||
|
/// thing that will ever make the node exist, so it must never be replaced.
|
||||||
|
pub fn op_supersedes(kind: &str) -> bool {
|
||||||
|
matches!(kind, OP_REVISION | OP_RENAME | OP_TRASH)
|
||||||
|
}
|
||||||
|
|
||||||
/// The volume id given to a node that exists only on this machine, so far. A
|
/// 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
|
/// real [`NodeUid`] is `{volume}~{link}`, so a placeholder is `local~<uuid>` and
|
||||||
/// round-trips through the same `Display`/parse path as any other uid.
|
/// round-trips through the same `Display`/parse path as any other uid.
|
||||||
@@ -88,6 +119,16 @@ pub struct PendingOp {
|
|||||||
pub next_attempt_at: i64,
|
pub next_attempt_at: i64,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// How much work the queue owes the server, by whether it carries bytes.
|
||||||
|
#[derive(Debug, Clone, Copy, Default)]
|
||||||
|
pub struct PendingCounts {
|
||||||
|
/// Queued `create`/`revision` ops: files whose content is not on the remote.
|
||||||
|
pub uploads: i64,
|
||||||
|
/// Queued `mkdir`/`rename`/`trash` ops: metadata the remote has not been
|
||||||
|
/// told about.
|
||||||
|
pub changes: i64,
|
||||||
|
}
|
||||||
|
|
||||||
/// The outcome of folding freshly written bytes into a queued create.
|
/// The outcome of folding freshly written bytes into a queued create.
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct AttachedBlob {
|
pub struct AttachedBlob {
|
||||||
@@ -368,6 +409,51 @@ impl Db {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Write-through multiple nodes in a single database transaction for performance.
|
||||||
|
pub fn upsert_nodes(&self, nodes: &[Node]) -> Result<()> {
|
||||||
|
if nodes.is_empty() {
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
let mut conn = self.conn.lock().unwrap();
|
||||||
|
let tx = conn.transaction()?;
|
||||||
|
for node in nodes {
|
||||||
|
let json = serde_json::to_string(node)?;
|
||||||
|
let uid = node.uid.to_string();
|
||||||
|
tx.execute(
|
||||||
|
"INSERT INTO nodes
|
||||||
|
(uid, parent_uid, name, is_dir, size, mtime, trashed, node_json)
|
||||||
|
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
|
||||||
|
ON CONFLICT(uid) DO UPDATE SET
|
||||||
|
parent_uid = excluded.parent_uid,
|
||||||
|
name = excluded.name,
|
||||||
|
is_dir = excluded.is_dir,
|
||||||
|
size = excluded.size,
|
||||||
|
mtime = excluded.mtime,
|
||||||
|
trashed = excluded.trashed,
|
||||||
|
node_json = excluded.node_json",
|
||||||
|
params![
|
||||||
|
uid,
|
||||||
|
node.parent_uid.as_ref().map(|u| u.to_string()),
|
||||||
|
node.name,
|
||||||
|
node.is_folder() as i64,
|
||||||
|
node_size(node),
|
||||||
|
node.modification_time,
|
||||||
|
node.trashed as i64,
|
||||||
|
json,
|
||||||
|
],
|
||||||
|
)?;
|
||||||
|
tx.execute("DELETE FROM nodes_fts WHERE uid = ?1", params![uid])?;
|
||||||
|
if !node.trashed {
|
||||||
|
tx.execute(
|
||||||
|
"INSERT INTO nodes_fts (uid, name) VALUES (?1, ?2)",
|
||||||
|
params![uid, node.name],
|
||||||
|
)?;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
tx.commit()?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
/// Drop a node row (delete or trash from the hot cache). Children rows are
|
/// Drop a node row (delete or trash from the hot cache). Children rows are
|
||||||
/// not cascaded here; the daemon forgets a whole subtree node-by-node.
|
/// not cascaded here; the daemon forgets a whole subtree node-by-node.
|
||||||
pub fn delete_node(&self, uid: &NodeUid) -> Result<()> {
|
pub fn delete_node(&self, uid: &NodeUid) -> Result<()> {
|
||||||
@@ -531,27 +617,30 @@ impl Db {
|
|||||||
/// everything the older one did. The superseded blob's path is returned so
|
/// everything the older one did. The superseded blob's path is returned so
|
||||||
/// the caller can delete it.
|
/// the caller can delete it.
|
||||||
///
|
///
|
||||||
/// Only revision rows supersede, and only each other. A create op for the
|
/// Rows supersede only their own kind, and only the kinds
|
||||||
/// same uid must survive — dropping it would leave a file that exists
|
/// [`op_supersedes`] names. A create op for the same uid must survive —
|
||||||
/// nowhere but this machine with nothing left to create it. (Writes to a
|
/// dropping it would leave a file that exists nowhere but this machine with
|
||||||
/// node that is itself still queued fold into the create row instead; see
|
/// nothing left to create it. (Writes to a node that is itself still queued
|
||||||
/// [`Db::attach_blob_to_create`].)
|
/// fold into the create row instead; see [`Db::attach_blob_to_create`].)
|
||||||
pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option<String>)> {
|
pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option<String>)> {
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
let superseded: Option<String> = conn
|
let superseded: Option<String> = if op_supersedes(&op.kind) {
|
||||||
.query_row(
|
let blob: Option<String> = conn
|
||||||
"SELECT blob_path FROM pending_op WHERE uid = ?1 AND kind = ?2",
|
.query_row(
|
||||||
params![op.uid, OP_REVISION],
|
"SELECT blob_path FROM pending_op WHERE uid = ?1 AND kind = ?2",
|
||||||
|r| r.get(0),
|
params![op.uid, op.kind],
|
||||||
)
|
|r| r.get(0),
|
||||||
.optional()?
|
)
|
||||||
.flatten();
|
.optional()?
|
||||||
if op.kind == OP_REVISION {
|
.flatten();
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"DELETE FROM pending_op WHERE uid = ?1 AND kind = ?2",
|
"DELETE FROM pending_op WHERE uid = ?1 AND kind = ?2",
|
||||||
params![op.uid, OP_REVISION],
|
params![op.uid, op.kind],
|
||||||
)?;
|
)?;
|
||||||
}
|
blob
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
};
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"INSERT INTO pending_op
|
"INSERT INTO pending_op
|
||||||
(kind, uid, parent_uid, name, blob_path, meta_json, created_at)
|
(kind, uid, parent_uid, name, blob_path, meta_json, created_at)
|
||||||
@@ -604,23 +693,62 @@ impl Db {
|
|||||||
Ok(Some(AttachedBlob { id, superseded }))
|
Ok(Some(AttachedBlob { id, superseded }))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Drop every op targeting a node, returning the staged blobs they held so
|
/// Point a queued create at a new parent and name, for a node renamed or
|
||||||
/// the caller can delete them.
|
/// moved before it ever reached the server.
|
||||||
|
///
|
||||||
|
/// A `local~` uid means nothing to the API, so there is no rename call to
|
||||||
|
/// make — the node is still only a queued intent, and rewriting that intent
|
||||||
|
/// *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).
|
||||||
|
pub fn rewrite_op_target(&self, uid: &str, parent_uid: &str, name: &str) -> Result<bool> {
|
||||||
|
let conn = self.conn.lock().unwrap();
|
||||||
|
let n = conn.execute(
|
||||||
|
"UPDATE pending_op SET parent_uid = ?2, name = ?3
|
||||||
|
WHERE uid = ?1 AND kind IN (?4, ?5)",
|
||||||
|
params![uid, parent_uid, name, OP_CREATE, OP_MKDIR],
|
||||||
|
)?;
|
||||||
|
Ok(n > 0)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Drop every op targeting a node **or anything queued beneath it**,
|
||||||
|
/// returning the staged blobs they held so the caller can delete them.
|
||||||
///
|
///
|
||||||
/// Used when a node that only ever existed locally is deleted: there is
|
/// Used when a node that only ever existed locally is deleted: there is
|
||||||
/// nothing on the server to trash, so the queued work simply stops being
|
/// nothing on the server to trash, so the queued work simply stops being
|
||||||
/// wanted.
|
/// wanted.
|
||||||
|
///
|
||||||
|
/// The descent is what keeps the queue alive. Deleting a folder created
|
||||||
|
/// offline drops the `mkdir` that would have given it a real uid, and any
|
||||||
|
/// op still queued under that placeholder is then unreachable forever: it
|
||||||
|
/// can never be attempted (its parent is a `local~` uid, so `op_is_ready`
|
||||||
|
/// refuses it) and nothing is left to remap it. It would sit in the queue,
|
||||||
|
/// and in the user's pending count, for the life of the database.
|
||||||
|
///
|
||||||
|
/// Only `create`/`mkdir` ops carry a `parent_uid`, so for a file this
|
||||||
|
/// recursion finds nothing and costs one query.
|
||||||
pub fn delete_ops_for_uid(&self, uid: &str) -> Result<Vec<String>> {
|
pub fn delete_ops_for_uid(&self, uid: &str) -> Result<Vec<String>> {
|
||||||
|
const SUBTREE: &str = "
|
||||||
|
WITH RECURSIVE doomed(uid) AS (
|
||||||
|
SELECT ?1
|
||||||
|
UNION
|
||||||
|
SELECT p.uid FROM pending_op p JOIN doomed d ON p.parent_uid = d.uid
|
||||||
|
)";
|
||||||
let mut conn = self.conn.lock().unwrap();
|
let mut conn = self.conn.lock().unwrap();
|
||||||
let tx = conn.transaction()?;
|
let tx = conn.transaction()?;
|
||||||
let blobs: Vec<String> = {
|
let blobs: Vec<String> = {
|
||||||
let mut stmt = tx.prepare(
|
let mut stmt = tx.prepare(&format!(
|
||||||
"SELECT blob_path FROM pending_op WHERE uid = ?1 AND blob_path IS NOT NULL",
|
"{SUBTREE}
|
||||||
)?;
|
SELECT blob_path FROM pending_op
|
||||||
|
WHERE uid IN (SELECT uid FROM doomed) AND blob_path IS NOT NULL"
|
||||||
|
))?;
|
||||||
let rows = stmt.query_map(params![uid], |r| r.get::<_, String>(0))?;
|
let rows = stmt.query_map(params![uid], |r| r.get::<_, String>(0))?;
|
||||||
rows.collect::<std::result::Result<Vec<_>, _>>()?
|
rows.collect::<std::result::Result<Vec<_>, _>>()?
|
||||||
};
|
};
|
||||||
tx.execute("DELETE FROM pending_op WHERE uid = ?1", params![uid])?;
|
tx.execute(
|
||||||
|
&format!("{SUBTREE} DELETE FROM pending_op WHERE uid IN (SELECT uid FROM doomed)"),
|
||||||
|
params![uid],
|
||||||
|
)?;
|
||||||
tx.commit()?;
|
tx.commit()?;
|
||||||
Ok(blobs)
|
Ok(blobs)
|
||||||
}
|
}
|
||||||
@@ -673,10 +801,25 @@ impl Db {
|
|||||||
Ok(rows)
|
Ok(rows)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// How many uploads are still waiting, for `Response::Status`.
|
/// How many queued ops there are of each kind, for `Response::Status`.
|
||||||
pub fn pending_op_count(&self) -> Result<i64> {
|
///
|
||||||
|
/// Split because "3 uploads queued" has to mean three files whose bytes are
|
||||||
|
/// not on the remote yet. A queued `mkdir`/`rename`/`trash` is also work the
|
||||||
|
/// mount owes the server, but it carries no bytes and reporting it as an
|
||||||
|
/// upload is simply untrue.
|
||||||
|
pub fn pending_op_counts(&self) -> Result<PendingCounts> {
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
Ok(conn.query_row("SELECT COUNT(*) FROM pending_op", [], |r| r.get(0))?)
|
let uploads = conn.query_row(
|
||||||
|
"SELECT COUNT(*) FROM pending_op WHERE kind IN (?1, ?2)",
|
||||||
|
params![OP_REVISION, OP_CREATE],
|
||||||
|
|r| r.get(0),
|
||||||
|
)?;
|
||||||
|
let changes = conn.query_row(
|
||||||
|
"SELECT COUNT(*) FROM pending_op WHERE kind NOT IN (?1, ?2)",
|
||||||
|
params![OP_REVISION, OP_CREATE],
|
||||||
|
|r| r.get(0),
|
||||||
|
)?;
|
||||||
|
Ok(PendingCounts { uploads, changes })
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Drop a queued op, once its upload has actually landed.
|
/// Drop a queued op, once its upload has actually landed.
|
||||||
@@ -2073,6 +2216,22 @@ mod tests {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn upsert_nodes_and_load_all_roundtrip() {
|
||||||
|
let db = Db::open_in_memory().unwrap();
|
||||||
|
let root = folder("root", None, "My Files");
|
||||||
|
let child1 = file("f1", "root", "hello.txt", 42);
|
||||||
|
let child2 = file("f2", "root", "world.txt", 100);
|
||||||
|
db.upsert_nodes(&[root, child1, child2]).unwrap();
|
||||||
|
|
||||||
|
let loaded = db.load_all().unwrap();
|
||||||
|
assert_eq!(loaded.len(), 3);
|
||||||
|
let f1 = loaded.iter().find(|s| s.node.uid == uid("f1")).unwrap();
|
||||||
|
assert_eq!(f1.node.name, "hello.txt");
|
||||||
|
let f2 = loaded.iter().find(|s| s.node.uid == uid("f2")).unwrap();
|
||||||
|
assert_eq!(f2.node.name, "world.txt");
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn upsert_is_idempotent_update() {
|
fn upsert_is_idempotent_update() {
|
||||||
let db = Db::open_in_memory().unwrap();
|
let db = Db::open_in_memory().unwrap();
|
||||||
@@ -2366,7 +2525,156 @@ mod tests {
|
|||||||
assert_eq!(ops.len(), 1, "one queued upload per node");
|
assert_eq!(ops.len(), 1, "one queued upload per node");
|
||||||
assert_eq!(ops[0].id, id2);
|
assert_eq!(ops[0].id, id2);
|
||||||
assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/second"));
|
assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/second"));
|
||||||
assert_eq!(db.pending_op_count().unwrap(), 1);
|
assert_eq!(db.pending_op_counts().unwrap().uploads, 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Deleting a folder that was created offline must take the ops queued
|
||||||
|
/// underneath it with it: they name a placeholder parent that will now never
|
||||||
|
/// become real, so nothing could ever drain them and nothing is left to
|
||||||
|
/// rewrite them.
|
||||||
|
#[test]
|
||||||
|
fn deleting_a_queued_folder_takes_its_queued_children_with_it() {
|
||||||
|
let db = Db::open_in_memory().unwrap();
|
||||||
|
let op = |kind: &str, uid: &str, parent: &str, blob: Option<&str>| PendingOp {
|
||||||
|
id: 0,
|
||||||
|
kind: kind.to_string(),
|
||||||
|
uid: uid.to_string(),
|
||||||
|
parent_uid: Some(parent.to_string()),
|
||||||
|
name: Some("n".to_string()),
|
||||||
|
blob_path: blob.map(str::to_string),
|
||||||
|
meta_json: None,
|
||||||
|
created_at: 1,
|
||||||
|
attempts: 0,
|
||||||
|
last_error: None,
|
||||||
|
next_attempt_at: 0,
|
||||||
|
};
|
||||||
|
let root = uid("root").to_string();
|
||||||
|
db.enqueue_op(&op(OP_MKDIR, "local~dir", &root, None))
|
||||||
|
.unwrap();
|
||||||
|
db.enqueue_op(&op(OP_MKDIR, "local~sub", "local~dir", None))
|
||||||
|
.unwrap();
|
||||||
|
db.enqueue_op(&op(
|
||||||
|
OP_CREATE,
|
||||||
|
"local~deep",
|
||||||
|
"local~sub",
|
||||||
|
Some("/staging/deep"),
|
||||||
|
))
|
||||||
|
.unwrap();
|
||||||
|
// A sibling outside the doomed subtree must survive.
|
||||||
|
db.enqueue_op(&op(OP_CREATE, "local~other", &root, Some("/staging/other")))
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let blobs = db.delete_ops_for_uid("local~dir").unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
blobs,
|
||||||
|
vec!["/staging/deep"],
|
||||||
|
"the subtree's bytes come back"
|
||||||
|
);
|
||||||
|
|
||||||
|
let left: Vec<String> = db
|
||||||
|
.pending_ops()
|
||||||
|
.unwrap()
|
||||||
|
.into_iter()
|
||||||
|
.map(|o| o.uid)
|
||||||
|
.collect();
|
||||||
|
assert_eq!(left, vec!["local~other"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A rename is the node's desired end state, so the newest one is the only
|
||||||
|
/// one worth performing — but it must not disturb the queued *upload* of the
|
||||||
|
/// same node, which is unrelated work.
|
||||||
|
#[test]
|
||||||
|
fn a_second_rename_supersedes_the_first_but_leaves_the_upload_alone() {
|
||||||
|
let db = Db::open_in_memory().unwrap();
|
||||||
|
let rename = |name: &str| PendingOp {
|
||||||
|
id: 0,
|
||||||
|
kind: OP_RENAME.to_string(),
|
||||||
|
uid: uid("a").to_string(),
|
||||||
|
parent_uid: Some(uid("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,
|
||||||
|
};
|
||||||
|
db.enqueue_op(&PendingOp {
|
||||||
|
id: 0,
|
||||||
|
kind: OP_REVISION.to_string(),
|
||||||
|
uid: uid("a").to_string(),
|
||||||
|
parent_uid: None,
|
||||||
|
name: None,
|
||||||
|
blob_path: Some("/staging/blob".to_string()),
|
||||||
|
meta_json: Some("{}".to_string()),
|
||||||
|
created_at: 1,
|
||||||
|
attempts: 0,
|
||||||
|
last_error: None,
|
||||||
|
next_attempt_at: 0,
|
||||||
|
})
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
db.enqueue_op(&rename("first")).unwrap();
|
||||||
|
let (_, superseded) = db.enqueue_op(&rename("second")).unwrap();
|
||||||
|
assert_eq!(superseded, None, "a rename owns no blob to clean up");
|
||||||
|
|
||||||
|
let ops = db.pending_ops().unwrap();
|
||||||
|
assert_eq!(ops.len(), 2, "the queued upload survives the rename");
|
||||||
|
let renames: Vec<_> = ops.iter().filter(|o| o.kind == OP_RENAME).collect();
|
||||||
|
assert_eq!(renames.len(), 1, "one rename per node");
|
||||||
|
assert_eq!(renames[0].name.as_deref(), Some("second"));
|
||||||
|
|
||||||
|
let counts = db.pending_op_counts().unwrap();
|
||||||
|
assert_eq!(counts.uploads, 1, "the revision is the only upload");
|
||||||
|
assert_eq!(counts.changes, 1, "the rename carries no bytes");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Renaming a node whose create has not drained rewrites the intent rather
|
||||||
|
/// than queueing a rename against a uid the server has never issued.
|
||||||
|
#[test]
|
||||||
|
fn renaming_a_queued_create_rewrites_its_target() {
|
||||||
|
let db = Db::open_in_memory().unwrap();
|
||||||
|
let local = "local~abc";
|
||||||
|
db.enqueue_op(&PendingOp {
|
||||||
|
id: 0,
|
||||||
|
kind: OP_CREATE.to_string(),
|
||||||
|
uid: local.to_string(),
|
||||||
|
parent_uid: Some(uid("old").to_string()),
|
||||||
|
name: Some("draft.txt".to_string()),
|
||||||
|
blob_path: Some("/staging/blob".to_string()),
|
||||||
|
meta_json: Some("{}".to_string()),
|
||||||
|
created_at: 1,
|
||||||
|
attempts: 0,
|
||||||
|
last_error: None,
|
||||||
|
next_attempt_at: 0,
|
||||||
|
})
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let rewritten = db
|
||||||
|
.rewrite_op_target(local, &uid("new").to_string(), "final.txt")
|
||||||
|
.unwrap();
|
||||||
|
assert!(rewritten);
|
||||||
|
|
||||||
|
let ops = db.pending_ops().unwrap();
|
||||||
|
assert_eq!(ops.len(), 1, "a rewrite is not a second op");
|
||||||
|
assert_eq!(ops[0].name.as_deref(), Some("final.txt"));
|
||||||
|
assert_eq!(
|
||||||
|
ops[0].parent_uid.as_deref(),
|
||||||
|
Some(uid("new").to_string()).as_deref()
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
ops[0].blob_path.as_deref(),
|
||||||
|
Some("/staging/blob"),
|
||||||
|
"the bytes riding on the create are untouched"
|
||||||
|
);
|
||||||
|
|
||||||
|
// Once the create has drained there is no intent left to rewrite, and the
|
||||||
|
// caller has to rename the real node instead.
|
||||||
|
db.delete_op(ops[0].id).unwrap();
|
||||||
|
assert!(
|
||||||
|
!db.rewrite_op_target(local, &uid("new").to_string(), "final.txt")
|
||||||
|
.unwrap()
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@@ -2484,7 +2792,7 @@ mod tests {
|
|||||||
assert_eq!(db.pending_ops().unwrap()[0].attempts, 2);
|
assert_eq!(db.pending_ops().unwrap()[0].attempts, 2);
|
||||||
|
|
||||||
db.delete_op(id).unwrap();
|
db.delete_op(id).unwrap();
|
||||||
assert_eq!(db.pending_op_count().unwrap(), 0);
|
assert_eq!(db.pending_op_counts().unwrap().uploads, 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|||||||
+900
-146
File diff suppressed because it is too large.
Load diff
+12
-10
@@ -36,7 +36,7 @@ use pdfs_core::control::{
|
|||||||
ActivityEntry, ActivityKind, BookmarkInfo, DeviceInfo, DirEntry, InvitationInfo, JobItem,
|
ActivityEntry, ActivityKind, BookmarkInfo, DeviceInfo, DirEntry, InvitationInfo, JobItem,
|
||||||
PhotoItem, PublicLinkInfo, RefreshScope, Request, Response, SearchHit, ShareEntry,
|
PhotoItem, PublicLinkInfo, RefreshScope, Request, Response, SearchHit, ShareEntry,
|
||||||
ShareEntryKind, SharedItem, SyncFolderInfo, SyncPhase, SyncProgress, TransferDirection,
|
ShareEntryKind, SharedItem, SyncFolderInfo, SyncPhase, SyncProgress, TransferDirection,
|
||||||
TransferItem, send,
|
TransferItem, pending_summary, send,
|
||||||
};
|
};
|
||||||
use pdfs_core::service;
|
use pdfs_core::service;
|
||||||
|
|
||||||
@@ -1965,19 +1965,21 @@ fn refresh_status(ui: &Rc<Ui>) {
|
|||||||
pins,
|
pins,
|
||||||
online,
|
online,
|
||||||
pending_uploads,
|
pending_uploads,
|
||||||
|
pending_changes,
|
||||||
..
|
..
|
||||||
})) => {
|
})) => {
|
||||||
set_mounted(&ui, true);
|
set_mounted(&ui, true);
|
||||||
// Queued uploads are the more useful thing to say when there are
|
// The queue is the more useful thing to say when it has anything
|
||||||
// any: they are why a file that looks saved is not on the remote
|
// in it: it is why a file that looks saved is not on the remote
|
||||||
// yet, and offline is usually the reason they are still queued.
|
// yet, and offline is usually the reason it is still queued.
|
||||||
ui.mount_row.set_subtitle(&match (online, pending_uploads) {
|
let queued = pending_summary(pending_uploads, pending_changes);
|
||||||
(true, 0) => format!("Mounted at {mountpoint}"),
|
ui.mount_row.set_subtitle(&match (online, queued) {
|
||||||
(true, n) => format!("Mounted at {mountpoint} — uploading {n} file(s)"),
|
(true, None) => format!("Mounted at {mountpoint}"),
|
||||||
(false, 0) => format!("Mounted at {mountpoint} — offline, cached files only"),
|
(true, Some(q)) => format!("Mounted at {mountpoint} — {q}"),
|
||||||
(false, n) => {
|
(false, None) => {
|
||||||
format!("Mounted at {mountpoint} — offline, {n} file(s) waiting to upload")
|
format!("Mounted at {mountpoint} — offline, cached files only")
|
||||||
}
|
}
|
||||||
|
(false, Some(q)) => format!("Mounted at {mountpoint} — offline, {q}"),
|
||||||
});
|
});
|
||||||
let fraction = if budget == 0 {
|
let fraction = if budget == 0 {
|
||||||
0.0
|
0.0
|
||||||
|
|||||||
@@ -18,7 +18,9 @@ use ksni::menu::StandardItem;
|
|||||||
use ksni::{MenuItem, Tray, TrayService};
|
use ksni::{MenuItem, Tray, TrayService};
|
||||||
use pdfs_core::auth;
|
use pdfs_core::auth;
|
||||||
use pdfs_core::config::AppDirs;
|
use pdfs_core::config::AppDirs;
|
||||||
use pdfs_core::control::{JobItem, Request, Response, TransferDirection, TransferItem, send};
|
use pdfs_core::control::{
|
||||||
|
JobItem, Request, Response, TransferDirection, TransferItem, pending_summary, send,
|
||||||
|
};
|
||||||
use pdfs_core::service;
|
use pdfs_core::service;
|
||||||
|
|
||||||
/// How often the tray re-polls the daemon to refresh its menu.
|
/// How often the tray re-polls the daemon to refresh its menu.
|
||||||
@@ -90,13 +92,14 @@ fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState {
|
|||||||
pinned,
|
pinned,
|
||||||
online,
|
online,
|
||||||
pending_uploads,
|
pending_uploads,
|
||||||
|
pending_changes,
|
||||||
..
|
..
|
||||||
}) => DriveState {
|
}) => DriveState {
|
||||||
line: match (online, pending_uploads) {
|
line: match (online, pending_summary(pending_uploads, pending_changes)) {
|
||||||
(true, 0) => format!("Mounted at {mountpoint} ({pinned} pinned)"),
|
(true, None) => format!("Mounted at {mountpoint} ({pinned} pinned)"),
|
||||||
(true, n) => format!("Uploading {n} file(s) ({pinned} pinned)"),
|
(true, Some(q)) => format!("Syncing — {q} ({pinned} pinned)"),
|
||||||
(false, 0) => format!("Offline — cached files only ({pinned} pinned)"),
|
(false, None) => format!("Offline — cached files only ({pinned} pinned)"),
|
||||||
(false, n) => format!("Offline — {n} file(s) waiting to upload"),
|
(false, Some(q)) => format!("Offline — {q}"),
|
||||||
},
|
},
|
||||||
mounted: true,
|
mounted: true,
|
||||||
mountpoint: PathBuf::from(mountpoint),
|
mountpoint: PathBuf::from(mountpoint),
|
||||||
|
|||||||
Reference in new issue
Block a user