cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
DioCrafts
2026-06-10 22:03:49 +02:00
parent 45fae4b27e
commit f678ff414e
12 changed files with 789 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/repositories/pg/file_blob_write_repository.rs
+106 -90
View File
@@ -20,6 +20,7 @@ use crate::domain::entities::file::File;
use crate::domain::services::path_service::StoragePath;
use super::folder_db_repository::FolderDbRepository;
use super::transaction_utils::retry_on_deadlock;
use crate::infrastructure::services::dedup_service::DedupService;
/// File write repository backed by PostgreSQL metadata + blob storage.
@@ -157,24 +158,28 @@ impl FileBlobWriteRepository {
modified_at: Option<i64>,
) -> Result<(String, i64), DomainError> {
// Atomic CTE: capture old hash then update in one round-trip, no TOCTOU.
let (old_hash, updated_at) = match sqlx::query_as::<_, (String, i64)>(
r#"
WITH old AS (
SELECT id, blob_hash FROM storage.files WHERE id = $3::uuid FOR UPDATE
// Deadlock victims (40P01) retry before the compensation below runs —
// a successful retry must keep the new blob reference alive.
let (old_hash, updated_at) = match retry_on_deadlock("files.swap_blob_hash", || {
sqlx::query_as::<_, (String, i64)>(
r#"
WITH old AS (
SELECT id, blob_hash FROM storage.files WHERE id = $3::uuid FOR UPDATE
)
UPDATE storage.files f
SET blob_hash = $1, size = $2,
updated_at = COALESCE(to_timestamp($4), NOW())
FROM old
WHERE f.id = old.id
RETURNING old.blob_hash, EXTRACT(EPOCH FROM f.updated_at)::bigint
"#,
)
UPDATE storage.files f
SET blob_hash = $1, size = $2,
updated_at = COALESCE(to_timestamp($4), NOW())
FROM old
WHERE f.id = old.id
RETURNING old.blob_hash, EXTRACT(EPOCH FROM f.updated_at)::bigint
"#,
)
.bind(new_hash)
.bind(new_size)
.bind(file_id)
.bind(modified_at.map(|t| t as f64))
.fetch_optional(self.pool.as_ref())
.bind(new_hash)
.bind(new_size)
.bind(file_id)
.bind(modified_at.map(|t| t as f64))
.fetch_optional(self.pool.as_ref())
})
.await
{
Ok(Some(row)) => row,
@@ -236,23 +241,30 @@ impl FileBlobWriteRepository {
let is_new_blob = !dedup_result.was_deduplicated();
let blob_hash = dedup_result.hash().to_string();
let row = match sqlx::query_as::<_, (String, i64, i64)>(
r#"
INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type, category_order)
VALUES ($1, $2::uuid, $3, $4, $5, $6, $7)
RETURNING id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
)
.bind(&name)
.bind(&folder_id)
.bind(user_id)
.bind(&blob_hash)
.bind(size as i64)
.bind(&content_type)
.bind(category_order_for(&name, &content_type))
.fetch_one(self.pool.as_ref())
// Deadlock victims (40P01) retry before the compensation below runs —
// a successful retry must keep the blob reference alive. The final
// attempt's error falls through untouched so the 23505 mapping holds
// (a retried INSERT can legitimately lose to a concurrent identical
// upload).
let row = match retry_on_deadlock("files.insert", || {
sqlx::query_as::<_, (String, i64, i64)>(
r#"
INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type, category_order)
VALUES ($1, $2::uuid, $3, $4, $5, $6, $7)
RETURNING id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
)
.bind(&name)
.bind(&folder_id)
.bind(user_id)
.bind(&blob_hash)
.bind(size as i64)
.bind(&content_type)
.bind(category_order_for(&name, &content_type))
.fetch_one(self.pool.as_ref())
})
.await
{
Ok(row) => row,
@@ -372,46 +384,48 @@ impl FileWritePort for FileBlobWriteRepository {
// Single round-trip; blob content is NOT copied (dedup makes this zero-copy).
let target_fid = target_folder_id.clone();
let row = sqlx::query_as::<
_,
(
String,
String,
Option<String>,
i64,
String,
i64,
i64,
String,
),
>(
r#"
WITH src AS (
SELECT name, folder_id, user_id, blob_hash, size, mime_type, category_order
FROM storage.files
WHERE id = $1::uuid AND NOT is_trashed
),
new_file AS (
INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type, category_order)
SELECT name,
COALESCE($2::uuid, folder_id),
user_id,
blob_hash,
size,
mime_type,
category_order
FROM src
RETURNING id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
blob_hash
let row = retry_on_deadlock("files.copy", || {
sqlx::query_as::<
_,
(
String,
String,
Option<String>,
i64,
String,
i64,
i64,
String,
),
>(
r#"
WITH src AS (
SELECT name, folder_id, user_id, blob_hash, size, mime_type, category_order
FROM storage.files
WHERE id = $1::uuid AND NOT is_trashed
),
new_file AS (
INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type, category_order)
SELECT name,
COALESCE($2::uuid, folder_id),
user_id,
blob_hash,
size,
mime_type,
category_order
FROM src
RETURNING id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
blob_hash
)
SELECT * FROM new_file
"#,
)
SELECT * FROM new_file
"#,
)
.bind(file_id)
.bind(&target_fid)
.fetch_optional(self.pool.as_ref())
.bind(file_id)
.bind(&target_fid)
.fetch_optional(self.pool.as_ref())
})
.await
.map_err(|e| {
if let sqlx::Error::Database(ref db_err) = e
@@ -557,23 +571,25 @@ impl FileWritePort for FileBlobWriteRepository {
// The write-behind cache will call update_file_content later.
let placeholder_hash = "0000000000000000000000000000000000000000000000000000000000000000";
let row = sqlx::query_as::<_, (String, i64, i64)>(
r#"
INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type, category_order)
VALUES ($1, $2::uuid, $3, $4, $5, $6, $7)
RETURNING id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
)
.bind(&name)
.bind(&folder_id)
.bind(user_id)
.bind(placeholder_hash)
.bind(size as i64)
.bind(&content_type)
.bind(category_order_for(&name, &content_type))
.fetch_one(self.pool.as_ref())
let row = retry_on_deadlock("files.insert_deferred", || {
sqlx::query_as::<_, (String, i64, i64)>(
r#"
INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type, category_order)
VALUES ($1, $2::uuid, $3, $4, $5, $6, $7)
RETURNING id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
)
.bind(&name)
.bind(&folder_id)
.bind(user_id)
.bind(placeholder_hash)
.bind(size as i64)
.bind(&content_type)
.bind(category_order_for(&name, &content_type))
.fetch_one(self.pool.as_ref())
})
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("deferred: {e}")))?;
src/infrastructure/repositories/pg/folder_db_repository.rs
+112 -88
View File
@@ -12,6 +12,7 @@ use sqlx::PgPool;
use std::sync::Arc;
use uuid::Uuid;
use super::transaction_utils::retry_on_deadlock;
use crate::application::dtos::folder_dto::{FolderResourceCursor, FolderResourceRow};
use crate::common::errors::DomainError;
use crate::domain::entities::folder::Folder;
@@ -450,20 +451,25 @@ impl FolderRepository for FolderDbRepository {
// The BEFORE UPDATE trigger recomputes path/lpath for this row;
// the AFTER UPDATE cascade trigger then batch-updates all
// descendants in a single UPDATE using the GiST lpath index.
let row = sqlx::query_as::<_, FolderRow>(
r#"
UPDATE storage.folders
SET name = $1, updated_at = NOW()
WHERE id = $2::uuid AND NOT is_trashed
RETURNING id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
"#,
)
.bind(&new_name)
.bind(id)
.fetch_optional(self.pool())
// That multi-row rewrite can deadlock against the tree-ETag
// flusher's id-ordered ancestor bump — retry instead of failing
// the user's operation (40P01 only; 23505 still maps below).
let row = retry_on_deadlock("folders.rename", || {
sqlx::query_as::<_, FolderRow>(
r#"
UPDATE storage.folders
SET name = $1, updated_at = NOW()
WHERE id = $2::uuid AND NOT is_trashed
RETURNING id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
"#,
)
.bind(&new_name)
.bind(id)
.fetch_optional(self.pool())
})
.await
.map_err(|e| {
if let sqlx::Error::Database(ref db_err) = e
@@ -486,20 +492,23 @@ impl FolderRepository for FolderDbRepository {
// The BEFORE UPDATE trigger recomputes path/lpath for this row;
// the AFTER UPDATE cascade trigger then batch-updates all
// descendants in a single UPDATE using the GiST lpath index.
let row = sqlx::query_as::<_, FolderRow>(
r#"
UPDATE storage.folders
SET parent_id = $1::uuid, updated_at = NOW()
WHERE id = $2::uuid AND NOT is_trashed
RETURNING id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
"#,
)
.bind(new_parent_id)
.bind(id)
.fetch_optional(self.pool())
// Retried on deadlock vs the tree-ETag flusher (see rename_folder).
let row = retry_on_deadlock("folders.move", || {
sqlx::query_as::<_, FolderRow>(
r#"
UPDATE storage.folders
SET parent_id = $1::uuid, updated_at = NOW()
WHERE id = $2::uuid AND NOT is_trashed
RETURNING id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
"#,
)
.bind(new_parent_id)
.bind(id)
.fetch_optional(self.pool())
})
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("move: {e}")))?
.ok_or_else(|| DomainError::not_found("Folder", id))?;
@@ -511,24 +520,30 @@ impl FolderRepository for FolderDbRepository {
// Delete all files whose folder is anywhere in the subtree.
// Uses the GiST-indexed ltree `<@` operator — O(log N) vs the
// O(depth × N) recursive CTE it replaces.
sqlx::query(
"DELETE FROM storage.files \
WHERE folder_id IN ( \
SELECT id FROM storage.folders \
WHERE lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid) \
)",
)
.bind(id)
.execute(self.pool())
// Both statements retried on deadlock vs the tree-ETag flusher's
// id-ordered ancestor bump (multi-row exclusive locks).
retry_on_deadlock("folders.delete_files", || {
sqlx::query(
"DELETE FROM storage.files \
WHERE folder_id IN ( \
SELECT id FROM storage.folders \
WHERE lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid) \
)",
)
.bind(id)
.execute(self.pool())
})
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("delete files: {e}")))?;
// Then delete the folder (CASCADE will remove descendant folders)
let result = sqlx::query("DELETE FROM storage.folders WHERE id = $1::uuid")
.bind(id)
.execute(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("delete: {e}")))?;
let result = retry_on_deadlock("folders.delete", || {
sqlx::query("DELETE FROM storage.folders WHERE id = $1::uuid")
.bind(id)
.execute(self.pool())
})
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("delete: {e}")))?;
if result.rows_affected() == 0 {
return Err(DomainError::not_found("Folder", id));
@@ -570,22 +585,24 @@ impl FolderRepository for FolderDbRepository {
// Child files and sub-folders are implicitly hidden because their
// ancestor is trashed — list queries already filter NOT is_trashed,
// and folder navigation won't reach a trashed folder's children.
let result = sqlx::query_scalar::<_, i64>(
r#"
WITH trash_folder AS (
UPDATE storage.folders
SET is_trashed = TRUE,
trashed_at = NOW(),
original_parent_id = parent_id,
updated_at = NOW()
WHERE id = $1::uuid AND NOT is_trashed
RETURNING 1
let result = retry_on_deadlock("folders.trash", || {
sqlx::query_scalar::<_, i64>(
r#"
WITH trash_folder AS (
UPDATE storage.folders
SET is_trashed = TRUE,
trashed_at = NOW(),
original_parent_id = parent_id,
updated_at = NOW()
WHERE id = $1::uuid AND NOT is_trashed
RETURNING 1
)
SELECT COUNT(*) FROM trash_folder
"#,
)
SELECT COUNT(*) FROM trash_folder
"#,
)
.bind(folder_id)
.fetch_one(self.pool())
.bind(folder_id)
.fetch_one(self.pool())
})
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("trash: {e}")))?;
@@ -607,23 +624,25 @@ impl FolderRepository for FolderDbRepository {
// The BEFORE UPDATE trigger recomputes path/lpath when
// original_parent_id is restored; the cascade trigger
// batch-updates all descendants via the GiST lpath index.
let result = sqlx::query_scalar::<_, i64>(
r#"
WITH restore_folder AS (
UPDATE storage.folders
SET is_trashed = FALSE,
trashed_at = NULL,
parent_id = COALESCE(original_parent_id, parent_id),
original_parent_id = NULL,
updated_at = NOW()
WHERE id = $1::uuid AND is_trashed
RETURNING 1
let result = retry_on_deadlock("folders.restore", || {
sqlx::query_scalar::<_, i64>(
r#"
WITH restore_folder AS (
UPDATE storage.folders
SET is_trashed = FALSE,
trashed_at = NULL,
parent_id = COALESCE(original_parent_id, parent_id),
original_parent_id = NULL,
updated_at = NOW()
WHERE id = $1::uuid AND is_trashed
RETURNING 1
)
SELECT COUNT(*) FROM restore_folder
"#,
)
SELECT COUNT(*) FROM restore_folder
"#,
)
.bind(folder_id)
.fetch_one(self.pool())
.bind(folder_id)
.fetch_one(self.pool())
})
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("restore: {e}")))?;
@@ -636,25 +655,30 @@ impl FolderRepository for FolderDbRepository {
async fn delete_folder_permanently(&self, folder_id: &str) -> Result<(), DomainError> {
// Delete all files whose folder is anywhere in the subtree
// (GiST ltree index, same pattern as delete_folder).
sqlx::query(
"DELETE FROM storage.files \
WHERE folder_id IN ( \
SELECT id FROM storage.folders \
WHERE lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid) \
)",
)
.bind(folder_id)
.execute(self.pool())
// (GiST ltree index, same pattern as delete_folder — both
// statements retried on deadlock vs the tree-ETag flusher).
retry_on_deadlock("folders.perm_delete_files", || {
sqlx::query(
"DELETE FROM storage.files \
WHERE folder_id IN ( \
SELECT id FROM storage.folders \
WHERE lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid) \
)",
)
.bind(folder_id)
.execute(self.pool())
})
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("perm delete files: {e}")))?;
// Then permanently delete folder — CASCADE handles descendant folders
let result = sqlx::query("DELETE FROM storage.folders WHERE id = $1::uuid")
.bind(folder_id)
.execute(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("perm delete: {e}")))?;
let result = retry_on_deadlock("folders.perm_delete", || {
sqlx::query("DELETE FROM storage.folders WHERE id = $1::uuid")
.bind(folder_id)
.execute(self.pool())
})
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("perm delete: {e}")))?;
if result.rows_affected() == 0 {
return Err(DomainError::not_found("Folder", folder_id));
src/infrastructure/repositories/pg/mod.rs
+1 -1
View File
@@ -16,7 +16,7 @@ mod session_pg_repository;
mod settings_pg_repository;
mod share_pg_repository;
mod subject_group_pg_repository;
mod transaction_utils;
pub(crate) mod transaction_utils;
mod user_pg_repository;
// ── Blob-storage repositories ──
src/infrastructure/repositories/pg/transaction_utils.rs
+127 -1
View File
@@ -1,6 +1,7 @@
use sqlx::{Error as SqlxError, PgPool, Postgres, Transaction};
use std::future::Future;
use std::sync::Arc;
use tracing::{debug, error, info};
use tracing::{debug, error, info, warn};
/// Helper function to execute database operations in a transaction
/// Takes a database pool and a closure that will be executed within a transaction
@@ -56,3 +57,128 @@ where
}
}
}
/// True when the error is a PostgreSQL deadlock abort (SQLSTATE `40P01`).
///
/// Deadlock victims are safe to re-run when the statement is a single
/// autocommit round-trip: the aborted implicit transaction left nothing
/// behind, and PostgreSQL chose this session as the victim precisely so
/// the competing transaction could finish — a retry usually succeeds
/// immediately.
pub fn is_deadlock(err: &SqlxError) -> bool {
matches!(err, SqlxError::Database(db) if db.code().as_deref() == Some("40P01"))
}
/// Re-run `op` while it fails with an error matching `should_retry`, up to
/// 3 retries with a short growing backoff. Errors that don't match the
/// predicate — and the final attempt's error — are returned untouched, so
/// callers' existing error mapping (e.g. `23505` → already-exists) still
/// sees exactly what it expects.
pub async fn retry_when<T, E, F, Fut, P>(
operation_name: &str,
should_retry: P,
op: F,
) -> Result<T, E>
where
F: Fn() -> Fut,
Fut: Future<Output = Result<T, E>>,
P: Fn(&E) -> bool,
{
const BACKOFF_MS: [u64; 3] = [10, 50, 150];
let mut attempt = 0;
loop {
match op().await {
Ok(value) => return Ok(value),
Err(e) if attempt < BACKOFF_MS.len() && should_retry(&e) => {
warn!(
"Retryable failure on {} (attempt {}/{}), backing off {}ms",
operation_name,
attempt + 1,
BACKOFF_MS.len() + 1,
BACKOFF_MS[attempt]
);
tokio::time::sleep(std::time::Duration::from_millis(BACKOFF_MS[attempt])).await;
attempt += 1;
}
Err(e) => return Err(e),
}
}
}
/// [`retry_when`] specialised to PostgreSQL deadlocks (`40P01`) — the only
/// transient SQLSTATE our single-statement write paths can hit.
pub async fn retry_on_deadlock<T, F, Fut>(operation_name: &str, op: F) -> Result<T, SqlxError>
where
F: Fn() -> Fut,
Fut: Future<Output = Result<T, SqlxError>>,
{
retry_when(operation_name, is_deadlock, op).await
}
#[cfg(test)]
mod tests {
use super::retry_when;
use std::sync::atomic::{AtomicU32, Ordering};
#[derive(Debug, PartialEq)]
enum FakeError {
Transient,
Fatal,
}
#[tokio::test]
async fn retries_transient_errors_until_success() {
let calls = AtomicU32::new(0);
let result = retry_when(
"test",
|e| *e == FakeError::Transient,
|| {
let n = calls.fetch_add(1, Ordering::SeqCst);
async move {
if n < 2 {
Err(FakeError::Transient)
} else {
Ok(n)
}
}
},
)
.await;
assert_eq!(result, Ok(2));
assert_eq!(calls.load(Ordering::SeqCst), 3);
}
#[tokio::test]
async fn gives_up_after_max_attempts_returning_last_error() {
let calls = AtomicU32::new(0);
let result: Result<(), _> = retry_when(
"test",
|e| *e == FakeError::Transient,
|| {
calls.fetch_add(1, Ordering::SeqCst);
async { Err(FakeError::Transient) }
},
)
.await;
assert_eq!(result, Err(FakeError::Transient));
// 1 initial attempt + 3 retries
assert_eq!(calls.load(Ordering::SeqCst), 4);
}
#[tokio::test]
async fn non_matching_errors_are_not_retried() {
let calls = AtomicU32::new(0);
let result: Result<(), _> = retry_when(
"test",
|e| *e == FakeError::Transient,
|| {
calls.fetch_add(1, Ordering::SeqCst);
async { Err(FakeError::Fatal) }
},
)
.await;
assert_eq!(result, Err(FakeError::Fatal));
assert_eq!(calls.load(Ordering::SeqCst), 1);
}
}
src/infrastructure/services/mod.rs
+1
View File
@@ -29,6 +29,7 @@ pub mod thumbnail_service;
#[cfg(test)]
mod thumbnail_service_test;
pub mod trash_cleanup_service;
pub mod tree_etag_flush_service;
pub mod webdav_lock_service;
pub mod wopi_discovery_service;
pub mod zip_service;
src/infrastructure/services/tree_etag_flush_service.rs
+173
View File
@@ -0,0 +1,173 @@
use sqlx::PgPool;
use std::sync::Arc;
use tracing::{debug, error, info, instrument};
use crate::infrastructure::repositories::pg::transaction_utils::retry_on_deadlock;
/// Background drainer for `storage.tree_etag_dirty` — the asynchronous half
/// of the tree-ETag bump (see migration `20260627000000_async_tree_etag_queue`).
///
/// The statement triggers on `storage.files` / `storage.folders` only append
/// "bump request" lpaths to the queue, taking zero folder-row locks, so any
/// number of concurrent uploads never conflict with each other. This service
/// turns those requests into actual `tree_modified_at` updates: every
/// `interval_ms` it drains the queue and bumps the deduplicated ancestor
/// closure in ONE batched UPDATE with deterministic id-order locking.
///
/// Correctness invariants this relies on:
/// * The drain (DELETE) and the bump (UPDATE) are one statement — one
/// transaction. A crash mid-flush rolls both back and the queue rows
/// survive; the first flush after startup drains any leftovers, so no
/// bump is ever lost (a lost bump means sync clients never discover
/// the change).
/// * Queue rows are dual-keyed (`lpath` captured at mutation time,
/// `folder_id` re-resolved at flush time) and the target set is the
/// UNION of both — so neither a folder deleted/moved away (captured
/// lpath wins) nor a folder re-rooted by a move inside the flush
/// window (resolved lpath wins) can lose its pending bump.
/// * The bump is monotonic per folder:
/// `GREATEST(NOW(), tree_modified_at + interval '1 second')`. Folder
/// ETags have whole-second granularity (`{id}-{epoch_seconds}`, see
/// `Folder::compute_etag`); two flushes inside the same wall-clock
/// second must still produce two distinct ETags or a client polling
/// between them would permanently miss the second change.
/// * The flusher is the ONLY writer of `tree_modified_at`, runs as a
/// single instance, and locks victims in id order — so it cannot
/// deadlock against itself. It can still lose a race against a folder
/// move's descendant lpath cascade, which is why the statement runs
/// under [`retry_on_deadlock`]; a retried flush re-drains the intact
/// queue, invisible to users.
pub struct TreeEtagFlushService {
pool: Arc<PgPool>,
interval_ms: u64,
}
/// Rows drained from the queue per statement. Bounds the lock footprint of
/// one flush under burst load (bulk trash purge, recursive copy); leftovers
/// are picked up by the in-tick drain loop or the next tick.
const DRAIN_BATCH: i64 = 5_000;
/// Max drain statements per tick, so a pathological backlog cannot
/// monopolise the maintenance connection forever within one tick.
const MAX_BATCHES_PER_TICK: u32 = 8;
impl TreeEtagFlushService {
pub fn new(pool: Arc<PgPool>, interval_ms: u64) -> Self {
Self {
pool,
// Floor the cadence so a misconfiguration can't busy-loop the
// maintenance pool.
interval_ms: interval_ms.max(100),
}
}
/// Spawn the flush loop. Fire-and-forget: the loop logs and survives
/// every error (an exited loop would silently freeze all folder ETags),
/// and the first flush runs immediately to drain rows left over from a
/// previous run.
#[instrument(skip(self))]
pub fn start_flush_job(&self) {
let pool = self.pool.clone();
let interval_ms = self.interval_ms;
info!(
"Starting tree-ETag flush job (every {}ms, batch {})",
interval_ms, DRAIN_BATCH
);
tokio::spawn(async move {
let mut ticker = tokio::time::interval(std::time::Duration::from_millis(interval_ms));
// If a flush overruns the interval, fire the next one a full
// interval later instead of bursting to catch up.
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
ticker.tick().await;
match Self::flush(&pool).await {
Ok((0, _)) => {}
Ok((drained, bumped)) => {
debug!(
"Tree-ETag flush: drained {} queue row(s), bumped {} folder(s)",
drained, bumped
);
}
Err(e) => {
error!("Tree-ETag flush failed (queue preserved, will retry): {e}");
}
}
}
});
}
/// Drain the dirty queue and bump the ancestor closure. Returns
/// `(queue rows drained, folder rows bumped)` summed over the batches
/// processed this tick.
async fn flush(pool: &PgPool) -> Result<(i64, i64), sqlx::Error> {
let mut total_drained = 0i64;
let mut total_bumped = 0i64;
for _ in 0..MAX_BATCHES_PER_TICK {
let (drained, bumped) = retry_on_deadlock("tree_etag_flush", || {
sqlx::query_as::<_, (i64, i64)>(
r#"
WITH drained AS (
DELETE FROM storage.tree_etag_dirty
WHERE id IN (SELECT id
FROM storage.tree_etag_dirty
ORDER BY id
LIMIT $1)
RETURNING lpath, folder_id
),
targets AS (
-- Captured chain: covers target folders deleted or
-- moved away since enqueue (the old location's
-- surviving ancestors still get their bump).
SELECT lpath FROM drained
UNION
-- Flush-time resolution: a folder MOVED since
-- enqueue had its subtree's lpaths rewritten, so
-- the captured lpath no longer matches it — its id
-- resolves to the CURRENT chain instead. Without
-- this, a bump queued just before a move would be
-- silently lost and sync clients would never
-- discover the change.
SELECT fo.lpath
FROM storage.folders fo
JOIN drained d ON fo.id = d.folder_id
),
victims AS (
-- `lpath @> target` = the target folder itself plus
-- every ancestor (GiST-indexed). Folder rows deleted
-- since enqueue simply don't match. Lock in id order
-- so overlapping closures cannot deadlock.
SELECT f.id
FROM storage.folders f
WHERE EXISTS (SELECT 1 FROM targets t
WHERE f.lpath @> t.lpath)
ORDER BY f.id
FOR NO KEY UPDATE
),
bumped AS (
UPDATE storage.folders f
SET tree_modified_at =
GREATEST(NOW(), f.tree_modified_at + interval '1 second')
FROM victims v
WHERE f.id = v.id
RETURNING f.id
)
SELECT (SELECT COUNT(*) FROM drained)::bigint,
(SELECT COUNT(*) FROM bumped)::bigint
"#,
)
.bind(DRAIN_BATCH)
.fetch_one(pool)
})
.await?;
total_drained += drained;
total_bumped += bumped;
if drained < DRAIN_BATCH {
break;
}
}
Ok((total_drained, total_bumped))
}
}