mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-04 16:56:59 +01:00
fixed rm before online del, mirror bug
This commit is contained in:
1 parent
1cd6a26be3
commit
bd17d506f8
9 files changed
+288
-144
No files matched your search
@@ -807,9 +807,13 @@ impl ContentCache {
|
||||
.collect();
|
||||
let content_dir = self.content_dir.clone();
|
||||
// Force pool enforcement ignoring current cap check to free disk space immediately
|
||||
let Ok(list) = self.db.cache_eviction_candidates(KIND_BLOB, 10) else { return; };
|
||||
let Ok(list) = self.db.cache_eviction_candidates(KIND_BLOB, 10) else {
|
||||
return;
|
||||
};
|
||||
for (key, bytes) in list {
|
||||
if pinned.contains(&key) { continue; }
|
||||
if pinned.contains(&key) {
|
||||
continue;
|
||||
}
|
||||
let p = content_dir.join(&key);
|
||||
let m = content_dir.join(format!("{key}.meta"));
|
||||
let _ = std::fs::remove_file(&p);
|
||||
|
||||
@@ -683,26 +683,22 @@ impl Core {
|
||||
// fails, since eviction may already have deleted part of the tree.
|
||||
// Recovery is the user's explicit switch back to `mirror`, whose arm
|
||||
// clears the baseline and re-downloads.
|
||||
// Try mounting the secondary FUSE filesystem FIRST before touching local files.
|
||||
// If mount fails, abort safely so local files are preserved intact.
|
||||
let session = match self.spawn_ondemand_mount(&local, root) {
|
||||
Ok(session) => session,
|
||||
Err(e) => {
|
||||
return Err(SwitchBlocked::Failed(format!(
|
||||
"mount ondemand failed; local files preserved: {e}"
|
||||
)));
|
||||
}
|
||||
};
|
||||
|
||||
if let Err(e) = self.db.sync_folder_set_mode(id, "ondemand") {
|
||||
let _ = umount_session_unblocked(session, fuse_connection_id(&local));
|
||||
return Err(SwitchBlocked::Failed(format!("db: {e:?}")));
|
||||
}
|
||||
|
||||
// Reclaim the disk: empty the local dir now that the mount handles filesystem requests.
|
||||
// Reclaim the disk before mounting. Once FUSE covers `local`, a
|
||||
// directory walk addresses the remote namespace, so evicting at
|
||||
// that point sends unlink/rmdir requests to Drive and leaves the
|
||||
// underlying mirror hidden and untouched.
|
||||
if let Err(e) = evict_dir_contents(&local) {
|
||||
warn!(id, path = %local.display(), error = %e, "evict local dir contents failed");
|
||||
}
|
||||
let session = self.spawn_ondemand_mount(&local, root).map_err(|e| {
|
||||
SwitchBlocked::Failed(format!(
|
||||
"mount ondemand failed after local cache eviction; switch back to mirror to restore it: {e}"
|
||||
))
|
||||
})?;
|
||||
self.mounts
|
||||
.lock()
|
||||
.insert(id, (session, fuse_connection_id(&local)));
|
||||
|
||||
@@ -72,29 +72,23 @@ impl Core {
|
||||
if let Err(e) = self.drain_op(&op) {
|
||||
let attempts = op.attempts + 1;
|
||||
let err_str = e.to_string();
|
||||
let is_unrecoverable = attempts >= 5 || err_str.contains("403") || err_str.contains("402") || err_str.contains("413");
|
||||
if is_unrecoverable {
|
||||
warn!(uid = %op.uid, attempts, error = %err_str, "pending upload failed permanently; quarantining op");
|
||||
let _ = self.db.delete_ops_for_uid(&op.uid);
|
||||
self.log_activity(
|
||||
ActivityKind::Upload,
|
||||
&op.uid,
|
||||
format!("quarantined after failure: {err_str}"),
|
||||
false,
|
||||
);
|
||||
} else {
|
||||
let backoff = DRAIN_BACKOFF_MIN
|
||||
.saturating_mul(1u32 << attempts.min(6))
|
||||
.min(DRAIN_BACKOFF_MAX);
|
||||
warn!(uid = %op.uid, attempts, error = %err_str, "pending upload failed; will retry");
|
||||
if let Err(e) = self.db.record_op_failure(
|
||||
op.id,
|
||||
&err_str,
|
||||
now_millis() + backoff.as_millis() as i64,
|
||||
) {
|
||||
error!(uid = %op.uid, error = %e, "recording a drain failure failed");
|
||||
self.wait_for_drain_work();
|
||||
}
|
||||
// Never infer that user data is disposable from an error string
|
||||
// or an arbitrary retry count. The staged blob and this row are
|
||||
// the only durable description of an accepted write. Even a
|
||||
// permission/quota error can become recoverable after the user
|
||||
// changes account state, and deleting the row here made the blob
|
||||
// unreachable after five ordinary network failures.
|
||||
let backoff = DRAIN_BACKOFF_MIN
|
||||
.saturating_mul(1u32 << attempts.min(6))
|
||||
.min(DRAIN_BACKOFF_MAX);
|
||||
warn!(uid = %op.uid, attempts, error = %err_str, "pending upload failed; will retry");
|
||||
if let Err(e) = self.db.record_op_failure(
|
||||
op.id,
|
||||
&err_str,
|
||||
now_millis() + backoff.as_millis() as i64,
|
||||
) {
|
||||
error!(uid = %op.uid, error = %e, "recording a drain failure failed");
|
||||
self.wait_for_drain_work();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+69
-31
@@ -3486,11 +3486,13 @@ impl ProtonFs {
|
||||
return;
|
||||
}
|
||||
};
|
||||
let open_now = {
|
||||
let mut st = self.core.state.lock();
|
||||
st.forget_or_unlink(&uid);
|
||||
st.entries.get(&ino).map(|e| e.open_count).unwrap_or(0) > 0
|
||||
};
|
||||
let open_now = self
|
||||
.core
|
||||
.state
|
||||
.lock()
|
||||
.entries
|
||||
.get(&ino)
|
||||
.is_some_and(|e| e.open_count > 0);
|
||||
// A node the server has never heard of cannot be trashed there; deleting
|
||||
// it just means its queued creation is no longer wanted. This works
|
||||
// offline, which the remote path below cannot (offline.md Phase 3b).
|
||||
@@ -3498,6 +3500,7 @@ impl ProtonFs {
|
||||
if !open_now {
|
||||
self.core.discard_queued_ops(&uid);
|
||||
}
|
||||
self.core.state.lock().forget_or_unlink(&uid);
|
||||
debug!(%uid, name, "deleted a node that had not been created remotely yet");
|
||||
reply.ok();
|
||||
return;
|
||||
@@ -3507,7 +3510,10 @@ impl ProtonFs {
|
||||
// command returns (offline.md Phase 3b).
|
||||
if !self.core.online.load(Ordering::Relaxed) {
|
||||
match self.core.queue_trash(&uid, name) {
|
||||
Ok(()) => reply.ok(),
|
||||
Ok(()) => {
|
||||
self.core.state.lock().forget_or_unlink(&uid);
|
||||
reply.ok();
|
||||
}
|
||||
Err(e) => reply.error(e),
|
||||
}
|
||||
return;
|
||||
@@ -3523,6 +3529,9 @@ impl ProtonFs {
|
||||
reply.error(Errno::EIO);
|
||||
return;
|
||||
}
|
||||
// Only remove the namespace entry after Drive accepted the mutation.
|
||||
// Otherwise an EIO response still made the path disappear locally.
|
||||
self.core.state.lock().forget_or_unlink(&uid);
|
||||
if !open_now {
|
||||
self.core.discard_queued_ops(&uid);
|
||||
self.core.cache.evict(&uid);
|
||||
@@ -4099,6 +4108,9 @@ impl Filesystem for ProtonFs {
|
||||
let (file, path) = match self.core.cache.create_scratch() {
|
||||
Ok(x) => x,
|
||||
Err(e) => {
|
||||
if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) {
|
||||
entry.open_count = entry.open_count.saturating_sub(1);
|
||||
}
|
||||
error!(%uid, error = %e, "create scratch file failed");
|
||||
reply.error(Errno::EIO);
|
||||
return;
|
||||
@@ -4193,12 +4205,7 @@ impl Filesystem for ProtonFs {
|
||||
.get(&e.uid)
|
||||
.map(|p| p.meta.len)
|
||||
.unwrap_or_else(|| node_size(&e.node));
|
||||
(
|
||||
e.uid.clone(),
|
||||
e.node.modification_time,
|
||||
fsize,
|
||||
is_video,
|
||||
)
|
||||
(e.uid.clone(), e.node.modification_time, fsize, is_video)
|
||||
}
|
||||
Some(_) => {
|
||||
reply.error(Errno::EISDIR);
|
||||
@@ -4312,6 +4319,9 @@ impl Filesystem for ProtonFs {
|
||||
};
|
||||
let mut st = self.core.state.lock();
|
||||
let ino = st.intern(parent, node);
|
||||
if let Some(entry) = st.entries.get_mut(&ino) {
|
||||
entry.open_count = entry.open_count.saturating_add(1);
|
||||
}
|
||||
if let Some(kids) = st.children.get_mut(&parent)
|
||||
&& !kids.contains(&ino)
|
||||
{
|
||||
@@ -4503,7 +4513,10 @@ impl Filesystem for ProtonFs {
|
||||
let ret = unsafe { libc::fallocate(fd, mode, offset as i64, length as i64) };
|
||||
if ret == 0 {
|
||||
if is_punch_hole {
|
||||
aw.written.punch_hole(offset, new_end);
|
||||
// A punched range reads as zero. Mark it authored so
|
||||
// queue_revision does not refill it from the remote
|
||||
// baseline and silently undo the hole at commit.
|
||||
aw.written.add(offset, new_end.min(aw.len));
|
||||
aw.dirty = true;
|
||||
} else if !keep_size && new_end > aw.len {
|
||||
aw.written.add(aw.len, new_end);
|
||||
@@ -4661,6 +4674,7 @@ impl Filesystem for ProtonFs {
|
||||
});
|
||||
(h, unlinked_uid)
|
||||
};
|
||||
let was_unlinked = unlinked_uid.is_some();
|
||||
if let Some(uid) = unlinked_uid {
|
||||
self.core.discard_queued_ops(&uid);
|
||||
self.core.cache.evict(&uid);
|
||||
@@ -4670,12 +4684,22 @@ impl Filesystem for ProtonFs {
|
||||
// scratch file is the only copy of what was just written, and blocking
|
||||
// the caller on the network is what made a copy into the mount run at
|
||||
// upload speed (and fail outright offline).
|
||||
match handle {
|
||||
Some(h) => match self.core.queue_revision(&h) {
|
||||
match (handle, was_unlinked) {
|
||||
(Some(h), true) => {
|
||||
// POSIX keeps an unlinked file usable until the last close, but
|
||||
// closing it must not resurrect those bytes as a new revision.
|
||||
if let Err(e) = std::fs::remove_file(&h.path)
|
||||
&& e.kind() != std::io::ErrorKind::NotFound
|
||||
{
|
||||
warn!(path = %h.path.display(), error = %e, "discarding unlinked scratch file failed");
|
||||
}
|
||||
reply.ok();
|
||||
}
|
||||
(Some(h), false) => match self.core.queue_revision(&h) {
|
||||
Ok(()) => reply.ok(),
|
||||
Err(e) => reply.error(e),
|
||||
},
|
||||
None => reply.ok(),
|
||||
(None, _) => reply.ok(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4684,10 +4708,11 @@ impl Filesystem for ProtonFs {
|
||||
if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) {
|
||||
let st = self.core.state.lock();
|
||||
if let Some(entry) = st.entries.get(&ino)
|
||||
&& entry.node.is_folder() {
|
||||
reply.error(Errno::EISDIR);
|
||||
return;
|
||||
}
|
||||
&& entry.node.is_folder()
|
||||
{
|
||||
reply.error(Errno::EISDIR);
|
||||
return;
|
||||
}
|
||||
}
|
||||
self.trash_child(parent.0, &name_str, reply);
|
||||
}
|
||||
@@ -4695,16 +4720,27 @@ impl Filesystem for ProtonFs {
|
||||
fn rmdir(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) {
|
||||
let name_str = name.to_string_lossy();
|
||||
if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) {
|
||||
let st = self.core.state.lock();
|
||||
if let Some(entry) = st.entries.get(&ino) {
|
||||
if !entry.node.is_folder() {
|
||||
{
|
||||
let st = self.core.state.lock();
|
||||
if st
|
||||
.entries
|
||||
.get(&ino)
|
||||
.is_some_and(|entry| !entry.node.is_folder())
|
||||
{
|
||||
reply.error(Errno::ENOTDIR);
|
||||
return;
|
||||
}
|
||||
if st.has_children(ino) {
|
||||
reply.error(Errno::ENOTEMPTY);
|
||||
return;
|
||||
}
|
||||
}
|
||||
// A missing cached listing says "unknown", not "empty". Proton's
|
||||
// trash call is recursive, so enumerate before applying rmdir's
|
||||
// POSIX emptiness rule.
|
||||
if let Err(e) = self.core.ensure_children(ino) {
|
||||
reply.error(e);
|
||||
return;
|
||||
}
|
||||
if self.core.state.lock().has_children(ino) {
|
||||
reply.error(Errno::ENOTEMPTY);
|
||||
return;
|
||||
}
|
||||
}
|
||||
self.trash_child(parent.0, &name_str, reply);
|
||||
@@ -4819,10 +4855,12 @@ impl Filesystem for ProtonFs {
|
||||
{
|
||||
let st = self.core.state.lock();
|
||||
if let Some(entry) = st.entries.get(&ino)
|
||||
&& entry.node.is_folder() && st.is_ancestor_of(ino, newparent) {
|
||||
reply.error(Errno::EINVAL);
|
||||
return;
|
||||
}
|
||||
&& entry.node.is_folder()
|
||||
&& st.is_ancestor_of(ino, newparent)
|
||||
{
|
||||
reply.error(Errno::EINVAL);
|
||||
return;
|
||||
}
|
||||
}
|
||||
// The destination has to be listed either way: the queued path pushes
|
||||
// the node into that listing, and the online one drops it to force a
|
||||
|
||||
@@ -56,27 +56,6 @@ impl Intervals {
|
||||
}
|
||||
}
|
||||
|
||||
/// Punch a hole in `[start, end)`, removing authored status for this range.
|
||||
pub(crate) fn punch_hole(&mut self, start: u64, end: u64) {
|
||||
if start >= end {
|
||||
return;
|
||||
}
|
||||
let mut next = Vec::new();
|
||||
for &(s, e) in &self.0 {
|
||||
if e <= start || s >= end {
|
||||
next.push((s, e));
|
||||
} else {
|
||||
if s < start {
|
||||
next.push((s, start));
|
||||
}
|
||||
if e > end {
|
||||
next.push((end, e));
|
||||
}
|
||||
}
|
||||
}
|
||||
self.0 = next;
|
||||
}
|
||||
|
||||
/// Split `[start, end)` into contiguous `(s, e, authored)` segments, in
|
||||
/// order. `authored == true` means the bytes live in the scratch file;
|
||||
/// `false` means they must come from the remote base (or are a hole).
|
||||
@@ -241,13 +220,15 @@ impl State {
|
||||
/// Check if a directory inode has any child nodes in memory or in the database.
|
||||
pub(crate) fn has_children(&self, parent: u64) -> bool {
|
||||
if let Some(kids) = self.children.get(&parent)
|
||||
&& !kids.is_empty() {
|
||||
return true;
|
||||
}
|
||||
&& !kids.is_empty()
|
||||
{
|
||||
return true;
|
||||
}
|
||||
if let Some(entry) = self.entries.get(&parent)
|
||||
&& let Ok(has_kids) = self.db.has_children(&entry.uid) {
|
||||
return has_kids;
|
||||
}
|
||||
&& let Ok(has_kids) = self.db.has_children(&entry.uid)
|
||||
{
|
||||
return has_kids;
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
@@ -586,33 +567,6 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_intervals_punch_hole_middle() {
|
||||
let mut iv = Intervals::default();
|
||||
iv.add(0, 100);
|
||||
iv.punch_hole(30, 50);
|
||||
assert_eq!(iv.0, vec![(0, 30), (50, 100)]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_intervals_punch_hole_edges() {
|
||||
let mut iv = Intervals::default();
|
||||
iv.add(0, 100);
|
||||
iv.punch_hole(0, 20);
|
||||
assert_eq!(iv.0, vec![(20, 100)]);
|
||||
|
||||
iv.punch_hole(80, 100);
|
||||
assert_eq!(iv.0, vec![(20, 80)]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_intervals_punch_hole_full() {
|
||||
let mut iv = Intervals::default();
|
||||
iv.add(0, 100);
|
||||
iv.punch_hole(0, 100);
|
||||
assert!(iv.0.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_is_ancestor_of_hierarchy() {
|
||||
let (mut st, _dir) = state();
|
||||
|
||||
@@ -1681,14 +1681,20 @@ impl Core {
|
||||
});
|
||||
}
|
||||
let mut out = Vec::new();
|
||||
let mut join_errors = 0usize;
|
||||
while let Some(joined) = set.join_next().await {
|
||||
match joined {
|
||||
Ok(result) => out.push(result),
|
||||
Err(e) => warn!(error = %e, "sync: task panicked"),
|
||||
Err(e) => {
|
||||
join_errors += 1;
|
||||
warn!(error = %e, "sync: task panicked");
|
||||
}
|
||||
}
|
||||
}
|
||||
out
|
||||
(out, join_errors)
|
||||
});
|
||||
let (results, join_errors) = results;
|
||||
outcome.errors += join_errors;
|
||||
for result in results {
|
||||
match result {
|
||||
Ok(applied) => {
|
||||
|
||||
Reference in new issue
Block a user