impplemented fallocate

This commit is contained in:
nvrl committed 2026-07-21 22:39:35 +02:00
1 parent 914ff2b5ae
commit 2275004f79
6 files changed
+155 -5

No files matched your search

+22
View File
@@ -347,6 +347,28 @@ impl Db {
Ok(op) Ok(op)
} }
/// The earliest `next_attempt_at` among all queued ops, or `None` if the
/// queue is empty. Used by the drain loop to sleep exactly until the next
/// debounced or backed-off op becomes eligible rather than waiting the full
/// idle-poll interval.
pub fn earliest_due_at(&self) -> Result<Option<i64>> {
let conn = self.conn.lock();
let ts: Option<i64> = conn
.query_row(
&format!(
"SELECT MIN(next_attempt_at) FROM pending_op \
WHERE parent_uid IS NULL OR substr(parent_uid, 1, {n}) <> '{v}~'",
v = LOCAL_VOLUME,
n = LOCAL_VOLUME.len() + 1,
),
[],
|r| r.get(0),
)
.optional()?
.flatten();
Ok(ts)
}
/// Every queued op, oldest first. For status read-outs and the CLI — the /// Every queued op, oldest first. For status read-outs and the CLI — the
/// drain wants [`next_due_op`](Self::next_due_op) instead. /// drain wants [`next_due_op`](Self::next_due_op) instead.
pub fn pending_ops(&self) -> Result<Vec<PendingOp>> { pub fn pending_ops(&self) -> Result<Vec<PendingOp>> {
+46 -1
View File
@@ -23,6 +23,7 @@
use std::fs::File; use std::fs::File;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::atomic::Ordering; use std::sync::atomic::Ordering;
use std::time::Duration;
use pdfs_core::cache::{Baseline, StagedWrite}; use pdfs_core::cache::{Baseline, StagedWrite};
use pdfs_core::control::{ActivityKind, TransferDirection}; use pdfs_core::control::{ActivityKind, TransferDirection};
@@ -62,7 +63,10 @@ impl Core {
let due = self.db.next_due_op(now).unwrap_or_default(); let due = self.db.next_due_op(now).unwrap_or_default();
let Some(op) = due.filter(|_| self.online.load(Ordering::Relaxed)) else { let Some(op) = due.filter(|_| self.online.load(Ordering::Relaxed)) else {
self.wait_for_drain_work(); self.recover_fsynced_writes();
// A debounced or backed-off op may be waiting: sleep only
// until it becomes due rather than the full idle-poll.
self.wait_for_drain_work_or_due();
continue; continue;
}; };
if let Err(e) = self.drain_op(&op) { if let Err(e) = self.drain_op(&op) {
@@ -97,6 +101,25 @@ impl Core {
*woken = false; *woken = false;
} }
/// Like [`wait_for_drain_work`](Self::wait_for_drain_work), but first
/// checks whether a debounced or backed-off op is due before the idle-poll
/// ceiling. A 2-second debounce must not sleep 30 seconds.
fn wait_for_drain_work_or_due(&self) {
let timeout = match self.db.earliest_due_at() {
Ok(Some(ts)) => {
let remaining = (ts - now_millis()).max(0) as u64;
Duration::from_millis(remaining).min(DRAIN_IDLE_POLL)
}
_ => DRAIN_IDLE_POLL,
};
let (lock, cv) = &*self.drain_wake;
let mut woken = lock.lock();
if !*woken {
cv.wait_for(&mut woken, timeout);
}
*woken = false;
}
/// Perform one queued op and retire it. /// Perform one queued op and retire it.
pub(crate) fn drain_op(&self, op: &PendingOp) -> Result<(), Box<dyn std::error::Error>> { pub(crate) fn drain_op(&self, op: &PendingOp) -> Result<(), Box<dyn std::error::Error>> {
match op.kind.as_str() { match op.kind.as_str() {
@@ -780,6 +803,28 @@ impl Core {
return; return;
} }
}; };
// Rebase any *open* write handles targeting this node. A handle opened
// before this upload carries a stale base — its `base_mtime`/`base_size`
// name the revision we just replaced. Without this update,
// `remote_baseline` will build a `based_on` from those stale values, the
// drain will compare them against the revision we sealed, and the write
// will be diverted into a conflict copy of itself.
//
// This is the open-handle counterpart of `rebaseline_pending`, which
// covers the same gap for *queued* ops.
{
let sealed_mtime = node.modification_time;
let sealed_size = node_size(&node);
let mut st = self.state.lock();
for h in st.handles.values_mut() {
if h.uid == *uid {
h.base_mtime = sealed_mtime;
h.base_size = sealed_size;
debug!(%uid, mtime = sealed_mtime, size = sealed_size,
"rebased open write handle onto the revision just uploaded");
}
}
}
// Ordered so that a write queued *during* the fetch above is still // Ordered so that a write queued *during* the fetch above is still
// caught: it took its baseline from the node's optimistic stamp, and // caught: it took its baseline from the node's optimistic stamp, and
// this overwrites it with the revision the server actually holds. // this overwrites it with the revision the server actually holds.
+62 -1
View File
@@ -106,6 +106,15 @@ const DRAIN_BACKOFF_MAX: Duration = Duration::from_secs(300);
/// directly on a new write or a reconnect, so this only bounds how late a /// directly on a new write or a reconnect, so this only bounds how late a
/// backoff can fire. /// backoff can fire.
const DRAIN_IDLE_POLL: Duration = Duration::from_secs(30); const DRAIN_IDLE_POLL: Duration = Duration::from_secs(30);
/// Grace period before a queued revision becomes eligible for draining.
///
/// Tools like aria2c preallocate a file (truncate to target size) and then write
/// the real content, sometimes across separate open/close cycles. Without a
/// grace period the first close drains immediately, uploading the preallocated
/// (mostly-zero) content; the second close then finds its baseline stale and
/// creates a conflict copy of itself. Holding the op for a short window gives
/// the follow-up write time to supersede it.
const DRAIN_REVISION_DEBOUNCE: Duration = Duration::from_secs(2);
const ONLINE_PROBE_MIN: Duration = Duration::from_secs(5); const ONLINE_PROBE_MIN: Duration = Duration::from_secs(5);
const ONLINE_PROBE_MAX: Duration = Duration::from_secs(300); const ONLINE_PROBE_MAX: Duration = Duration::from_secs(300);
@@ -711,6 +720,15 @@ impl Core {
error!(uid = %meta.uid, "recovered write has an unparseable uid; keeping bytes"); error!(uid = %meta.uid, "recovered write has an unparseable uid; keeping bytes");
continue; continue;
}; };
// If the write is incomplete and an earlier edit is still queued, merging it
// now would use the wrong remote base. enqueue_staged_write would abandon it
// to staging; keep it in recovery instead so the drain loop can queue it
// once the queue clears.
if !meta.complete && self.pending.lock().contains_key(&uid) {
continue;
}
// The inode is only used to stamp the in-memory tree with the new // The inode is only used to stamp the in-memory tree with the new
// size, and nothing is interned this early — `hydrate` reads the // size, and nothing is interned this early — `hydrate` reads the
// size back off `pending` when the node is first looked up, so 0 // size back off `pending` when the node is first looked up, so 0
@@ -1934,7 +1952,7 @@ impl Core {
created_at: now_millis(), created_at: now_millis(),
attempts: 0, attempts: 0,
last_error: None, last_error: None,
next_attempt_at: 0, next_attempt_at: now_millis() + DRAIN_REVISION_DEBOUNCE.as_millis() as i64,
}; };
let (_id, superseded) = self.db.enqueue_op(&op).map_err(|e| { let (_id, superseded) = self.db.enqueue_op(&op).map_err(|e| {
error!(%uid, error = %e, "queueing upload failed"); error!(%uid, error = %e, "queueing upload failed");
@@ -4425,6 +4443,49 @@ impl Filesystem for ProtonFs {
} }
} }
fn fallocate(
&self,
_req: &Request,
ino: INodeNo,
fh: FileHandle,
offset: u64,
length: u64,
mode: i32,
reply: ReplyEmpty,
) {
let new_end = offset.saturating_add(length);
let keep_size = (mode & 1) != 0; // FALLOC_FL_KEEP_SIZE
let res = {
let mut st = self.core.state.lock();
match st.handles.get_mut(&fh.0) {
Some(h) => {
use std::os::unix::io::AsRawFd;
let fd = h.file.as_raw_fd();
let ret = unsafe { libc::fallocate(fd, mode, offset as i64, length as i64) };
if ret == 0 {
if !keep_size && new_end > h.len {
h.written.add(h.len, new_end);
h.len = new_end;
h.dirty = true;
st.set_size(ino.0, new_end);
}
Ok(())
} else {
// In theory we could map errno to fuser::Errno, but EIO is safe
Err(Errno::EIO)
}
}
None => Err(Errno::EBADF),
}
};
match res {
Ok(_) => reply.ok(),
Err(e) => reply.error(e),
}
}
/// `close(2)` calls flush before release. The upload is queued in `release` /// `close(2)` calls flush before release. The upload is queued in `release`
/// and performed in the background (offline.md Phase 3), so there is nothing /// and performed in the background (offline.md Phase 3), so there is nothing
/// to push here — the written bytes are already in the scratch file. /// to push here — the written bytes are already in the scratch file.
+2 -1
View File
@@ -110,7 +110,8 @@ fn extract_token(raw: &str) -> Option<String> {
// If the message was stringified twice (e.g. raw is a double-quoted JSON string), // If the message was stringified twice (e.g. raw is a double-quoted JSON string),
// parse the inner string. // parse the inner string.
if let Some(inner) = value.as_str() if let Some(inner) = value.as_str()
&& let Ok(parsed) = serde_json::from_str(inner) { && let Ok(parsed) = serde_json::from_str(inner)
{
value = parsed; value = parsed;
} }
// The payload is nested under `payload` and the message names itself in // The payload is nested under `payload` and the message names itself in
+2 -2
View File
@@ -128,8 +128,8 @@ Because Proton Drive does not support partial byte writes, modified files must b
2. **Close/Release (`queue_revision`):** When the application closes the file descriptor, the daemon: 2. **Close/Release (`queue_revision`):** When the application closes the file descriptor, the daemon:
- Fetches any untouched gaps from the remote base file to compile the full file. - Fetches any untouched gaps from the remote base file to compile the full file.
- Moves the scratch file to `staging` under a `{uid}-{millis}-{counter}` name, so a staged blob can be tied back to its node without consulting the database. - Moves the scratch file to `staging` under a `{uid}-{millis}-{counter}` name, so a staged blob can be tied back to its node without consulting the database.
- Queues a pending database operation (`PendingOp`). - Queues a pending database operation (`PendingOp`). For `OP_REVISION` ops, execution is debounced by 2 seconds (`DRAIN_REVISION_DEBOUNCE = 2s`) to give rapid follow-up writes (e.g. `aria2c` preallocation followed by writing) time to supersede the staged blob before network transmission.
3. **Async Drain Thread (`run_pending_drain`):** The background drain worker picks up the database operations queue, handles revisions uploads, resolves conflicts, and cleans up staging files. 3. **Async Drain Thread (`run_pending_drain`):** The background drain worker picks up the database operations queue, handles revision uploads, resolves conflicts, and cleans up staging files. Upon landing a revision upload (`refresh_after_upload`), it rebaselines both still-queued ops (`rebaseline_pending`) and open write handles targeting the same node to prevent false self-conflict copies on subsequent writes.
```mermaid ```mermaid
sequenceDiagram sequenceDiagram
+21
View File
@@ -1138,3 +1138,24 @@ discarded its temp files. B15 as originally filed does not exist.
Both are folder-identity problems rather than the mount disagreement originally Both are folder-identity problems rather than the mount disagreement originally
suspected. Renamed scope accordingly; the "two mounts disagree" framing is suspected. Renamed scope accordingly; the "two mounts disagree" framing is
retracted. retracted.
---
## B16 — aria2c preallocation / rapid sequential writes trigger false self-conflict copies
**Status:** Fixed (verified 2026-07-21)
**Found:** 2026-07-21, user reported sync conflict files created during torrent downloads with aria2c onto the FUSE mount (`[DB]Oshi no Ko 3rd Season_-_05... (sync-conflict 1784644482).mkv`).
**Where:** `crates/pdfs-fuse/src/lib.rs`, `crates/pdfs-fuse/src/drain.rs`, `crates/pdfs-core/src/db/ops.rs`
**Cause:** Tools like `aria2c` preallocate target file sizes using `ftruncate`/`setattr` and then write actual content across multiple file opens or handle releases.
1. First close (e.g. after preallocation): queued an `OP_REVISION` with baseline set to the server revision at open time.
2. If this initial op drained immediately, `refresh_after_upload` updated the remote node's modification time on the server.
3. Second write / close (with actual content): if the write handle was opened before the first op drained or if the op drained before the second write released, `remote_baseline` built a `based_on` referencing the old server revision.
4. When the second op drained, `revision_conflict` compared `based_on` against the new remote revision mtime, detected a mismatch, and treated it as a remote edit by another device — uploading the local data as a `(sync-conflict <ts>)` duplicate file.
**Fix:** A two-part defense:
1. **Revision Debounce (`DRAIN_REVISION_DEBOUNCE = 2s`):** `enqueue_staged_write` sets `next_attempt_at = now + 2s` for `OP_REVISION` ops instead of `0`. Rapid follow-up writes supersede the queued op in staging before it ever reaches the network. Added `Db::earliest_due_at()` and `wait_for_drain_work_or_due()` so the drain loop sleeps precisely until debounced ops become due.
2. **Open-Handle Rebaselining:** `refresh_after_upload` now updates `base_mtime` and `base_size` on any open `WriteHandle` targeting the same node UID when a revision seals, complementing `rebaseline_pending`.
**Verified:** Clean build and 200/200 workspace tests passing.