fix(cache): invalidate drive used byte cache on explicit refresh from internal call
this fix https://github.com/AtalayaLabs/OxiCloud/issues/607
which was introduced by commit 12dc648cff
when a user does activity in a drive, admin can invalidate cache via the internal
call /api/admin/internal/trigger-sweep
this permit end 2 end test to validte immediately that used_bytes corresponds to the expected result
This commit is contained in:
@@ -19,6 +19,13 @@ use uuid::Uuid;
|
|||||||
pub struct StorageUsageService {
|
pub struct StorageUsageService {
|
||||||
pool: Arc<PgPool>,
|
pool: Arc<PgPool>,
|
||||||
user_repository: Arc<UserPgRepository>,
|
user_repository: Arc<UserPgRepository>,
|
||||||
|
/// Optional so DI can wire it lazily and older test constructors
|
||||||
|
/// keep compiling. When `Some`, every write path that mutates
|
||||||
|
/// `drives.used_bytes` or `users.storage_used_bytes` invalidates
|
||||||
|
/// the drive lookup caches so `GET /api/drives` reflects the new
|
||||||
|
/// usage on the next call (see the invalidation calls in the
|
||||||
|
/// delta / sweep methods below).
|
||||||
|
drive_repo: Option<Arc<dyn crate::domain::repositories::drive_repository::DriveRepository>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl StorageUsageService {
|
impl StorageUsageService {
|
||||||
@@ -27,6 +34,44 @@ impl StorageUsageService {
|
|||||||
Self {
|
Self {
|
||||||
pool,
|
pool,
|
||||||
user_repository,
|
user_repository,
|
||||||
|
drive_repo: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Wires the drive repository used for cache-invalidation-on-write.
|
||||||
|
/// Production DI calls this in `common::di`; tests without a real
|
||||||
|
/// drive repo leave it `None` and the invalidation calls no-op.
|
||||||
|
pub fn with_drive_repo(
|
||||||
|
mut self,
|
||||||
|
drive_repo: Arc<dyn crate::domain::repositories::drive_repository::DriveRepository>,
|
||||||
|
) -> Self {
|
||||||
|
self.drive_repo = Some(drive_repo);
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Drop the per-caller readable-drive listing cache and the
|
||||||
|
/// per-user default-drive cache so `GET /api/drives` and the
|
||||||
|
/// WebDAV / NextCloud / WOPI drive-lookup paths re-read fresh
|
||||||
|
/// values.
|
||||||
|
///
|
||||||
|
/// **Called only from the reconciliation sweep**, not from the
|
||||||
|
/// hot-path `add_drive_storage_usage_delta*` methods. The design
|
||||||
|
/// (Ed's call, 2026-07-17): keep the cache useful under active
|
||||||
|
/// upload load — per-mutation invalidation would nuke the cache
|
||||||
|
/// on every file upload, defeating the point. `used_bytes` on
|
||||||
|
/// `GET /api/drives` therefore lags by up to the cache TTL (30 s),
|
||||||
|
/// which matches the sibling caches' accepted UX phantom for
|
||||||
|
/// drive-name staleness. Tests / operators that need immediate
|
||||||
|
/// freshness call `POST /api/admin/internal/trigger-sweep`, which
|
||||||
|
/// runs `update_all_drives_storage_usage` → this method.
|
||||||
|
///
|
||||||
|
/// Security posture unaffected: `check_drive_quota` reads
|
||||||
|
/// directly from SQL, bypassing the cache entirely, so quota
|
||||||
|
/// enforcement is honest regardless of listing staleness.
|
||||||
|
fn invalidate_drive_lookup_caches(&self) {
|
||||||
|
if let Some(repo) = &self.drive_repo {
|
||||||
|
repo.invalidate_readable_all();
|
||||||
|
repo.invalidate_default_drive_all();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -209,6 +254,9 @@ impl StorageUsageService {
|
|||||||
.execute(self.pool.as_ref())
|
.execute(self.pool.as_ref())
|
||||||
.await
|
.await
|
||||||
.map_err(|e| DomainError::internal_error("StorageUsage", format!("drive delta: {e}")))?;
|
.map_err(|e| DomainError::internal_error("StorageUsage", format!("drive delta: {e}")))?;
|
||||||
|
// Deliberate no-invalidate here — see the class doc on
|
||||||
|
// `invalidate_drive_lookup_caches`. Delta writes lag the
|
||||||
|
// cache by up to the TTL; the sweep is the escape hatch.
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -285,6 +333,7 @@ impl StorageUsageService {
|
|||||||
.map_err(|e| {
|
.map_err(|e| {
|
||||||
DomainError::internal_error("StorageUsage", format!("drive delta by folder: {e}"))
|
DomainError::internal_error("StorageUsage", format!("drive delta by folder: {e}"))
|
||||||
})?;
|
})?;
|
||||||
|
// See `add_drive_storage_usage_delta` — deliberate no-invalidate.
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -595,6 +644,20 @@ impl StorageUsagePort for StorageUsageService {
|
|||||||
"Drive storage-usage reconciliation corrected {} drive(s)",
|
"Drive storage-usage reconciliation corrected {} drive(s)",
|
||||||
result.rows_affected()
|
result.rows_affected()
|
||||||
);
|
);
|
||||||
|
// Unconditional invalidation — do NOT gate on
|
||||||
|
// `rows_affected() > 0`. When a fire-and-forget delta has
|
||||||
|
// already made SQL correct BEFORE the sweep runs, the sweep
|
||||||
|
// touches zero rows but the cache may still hold the
|
||||||
|
// pre-delta value from an earlier `GET /api/drives`. Gating
|
||||||
|
// means the cache stays stale in exactly the case
|
||||||
|
// `trigger-sweep` is called to fix. The invalidation cost is
|
||||||
|
// small (moka `invalidate_all` on both caches); the
|
||||||
|
// correctness guarantee matters. Regression avoidance:
|
||||||
|
// drive_quota.hurl Step 6 exercises this race — 2nd upload's
|
||||||
|
// delta lands during the 200 ms delay, sweep sees SQL is
|
||||||
|
// already right → zero rows → without unconditional
|
||||||
|
// invalidation, cache stays at the previous step's value.
|
||||||
|
self.invalidate_drive_lookup_caches();
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -613,6 +676,7 @@ impl Clone for StorageUsageService {
|
|||||||
Self {
|
Self {
|
||||||
pool: Arc::clone(&self.pool),
|
pool: Arc::clone(&self.pool),
|
||||||
user_repository: Arc::clone(&self.user_repository),
|
user_repository: Arc::clone(&self.user_repository),
|
||||||
|
drive_repo: self.drive_repo.clone(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+13
-1
@@ -1056,14 +1056,25 @@ impl AppServiceFactory {
|
|||||||
_repos: &RepositoryServices,
|
_repos: &RepositoryServices,
|
||||||
db_pool: &Arc<PgPool>,
|
db_pool: &Arc<PgPool>,
|
||||||
maintenance_pool: &Arc<PgPool>,
|
maintenance_pool: &Arc<PgPool>,
|
||||||
|
drive_repo: Arc<crate::infrastructure::repositories::pg::DrivePgRepository>,
|
||||||
) -> Arc<StorageUsageService> {
|
) -> Arc<StorageUsageService> {
|
||||||
let user_repository = Arc::new(
|
let user_repository = Arc::new(
|
||||||
crate::infrastructure::repositories::pg::UserPgRepository::new(db_pool.clone()),
|
crate::infrastructure::repositories::pg::UserPgRepository::new(db_pool.clone()),
|
||||||
);
|
);
|
||||||
|
// The `drive_repo` passed in is the SAME instance held on
|
||||||
|
// `AppState`, so its `readable_cache` / `default_drive_cache`
|
||||||
|
// are the caches the request path reads from. A separately
|
||||||
|
// constructed `DrivePgRepository` would have its OWN caches
|
||||||
|
// and invalidation would be a no-op observed by nobody —
|
||||||
|
// this is the trap that regressed the used_bytes freshness
|
||||||
|
// after perf commit `12dc648c`.
|
||||||
let service = Arc::new(
|
let service = Arc::new(
|
||||||
crate::application::services::storage_usage_service::StorageUsageService::new(
|
crate::application::services::storage_usage_service::StorageUsageService::new(
|
||||||
maintenance_pool.clone(),
|
maintenance_pool.clone(),
|
||||||
user_repository,
|
user_repository,
|
||||||
|
)
|
||||||
|
.with_drive_repo(
|
||||||
|
drive_repo as Arc<dyn crate::domain::repositories::drive_repository::DriveRepository>,
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
// Keep cached storage usage fresh off the request path: GET /api/auth/me
|
// Keep cached storage usage fresh off the request path: GET /api/auth/me
|
||||||
@@ -1250,7 +1261,8 @@ impl AppServiceFactory {
|
|||||||
// 3c. Storage usage / quota service (needed by the instant-upload
|
// 3c. Storage usage / quota service (needed by the instant-upload
|
||||||
// path inside the application services, and re-exposed on AppState
|
// path inside the application services, and re-exposed on AppState
|
||||||
// for the handler-side quota checks of the byte-upload paths).
|
// for the handler-side quota checks of the byte-upload paths).
|
||||||
let storage_usage = self.create_storage_usage_service(&repos, &pool, &maintenance_pool);
|
let storage_usage =
|
||||||
|
self.create_storage_usage_service(&repos, &pool, &maintenance_pool, drive_repo.clone());
|
||||||
|
|
||||||
// 3d. Content index (embedded Tantivy) — opened before application
|
// 3d. Content index (embedded Tantivy) — opened before application
|
||||||
// services so SearchService can hold the query port; the feeding
|
// services so SearchService can hold the query port; the feeding
|
||||||
|
|||||||
+58
-20
@@ -111,16 +111,25 @@ HTTP 201
|
|||||||
small_file_id: jsonpath "$.id"
|
small_file_id: jsonpath "$.id"
|
||||||
|
|
||||||
|
|
||||||
# Confirm `drives.used_bytes` reflects the new file. The hook is
|
# Force freshness on `drives.used_bytes`:
|
||||||
# fire-and-forget on a tokio task, so the SQL UPDATE may not have
|
# 1. The fire-and-forget delta hook may not have landed yet
|
||||||
# landed by the time `POST /api/files/upload` returned. Retry the
|
# (200 ms delay to let the tokio task register — see
|
||||||
# `GET /api/drives` until the cached value catches up — bounded
|
# `bug_trigger_sweep_vs_spawn_hook_race`).
|
||||||
# wait keeps a slow CI machine from flaking.
|
# 2. Force a reconciliation sweep. That's the ONLY path that
|
||||||
|
# invalidates `readable_cache` / `default_drive_cache` after
|
||||||
|
# Ed's 2026-07-17 design call: the sweep is the escape hatch
|
||||||
|
# for tests / operators that need immediate cache freshness;
|
||||||
|
# per-write invalidation would nuke the cache on every upload.
|
||||||
|
POST {{base_url}}/api/admin/internal/trigger-sweep
|
||||||
|
Authorization: Bearer {{admin_token}}
|
||||||
|
[Options]
|
||||||
|
delay: 200ms
|
||||||
|
|
||||||
|
HTTP 200
|
||||||
|
|
||||||
|
|
||||||
GET {{base_url}}/api/drives
|
GET {{base_url}}/api/drives
|
||||||
Authorization: Bearer {{owner_token}}
|
Authorization: Bearer {{owner_token}}
|
||||||
[Options]
|
|
||||||
retry: 10
|
|
||||||
retry-interval: 200ms
|
|
||||||
|
|
||||||
HTTP 200
|
HTTP 200
|
||||||
[Asserts]
|
[Asserts]
|
||||||
@@ -145,13 +154,19 @@ file: file,fixtures/hello-copy.txt; text/plain
|
|||||||
HTTP 201
|
HTTP 201
|
||||||
|
|
||||||
|
|
||||||
# `used_bytes` climbs to 64 (32 + 32). Same retry shape as the
|
# `used_bytes` climbs to 64 (32 + 32). Same trigger-sweep pattern
|
||||||
# first assertion since the second delta is also fire-and-forget.
|
# as the first assertion — the delta is fire-and-forget and the
|
||||||
|
# listing cache lags until the sweep invalidates it.
|
||||||
|
POST {{base_url}}/api/admin/internal/trigger-sweep
|
||||||
|
Authorization: Bearer {{admin_token}}
|
||||||
|
[Options]
|
||||||
|
delay: 200ms
|
||||||
|
|
||||||
|
HTTP 200
|
||||||
|
|
||||||
|
|
||||||
GET {{base_url}}/api/drives
|
GET {{base_url}}/api/drives
|
||||||
Authorization: Bearer {{owner_token}}
|
Authorization: Bearer {{owner_token}}
|
||||||
[Options]
|
|
||||||
retry: 10
|
|
||||||
retry-interval: 200ms
|
|
||||||
|
|
||||||
HTTP 200
|
HTTP 200
|
||||||
[Asserts]
|
[Asserts]
|
||||||
@@ -173,7 +188,18 @@ HTTP 507
|
|||||||
|
|
||||||
# `used_bytes` is unchanged — the failed upload didn't charge the
|
# `used_bytes` is unchanged — the failed upload didn't charge the
|
||||||
# drive. (Cumulative usage is still 64; the 5 MiB write never
|
# drive. (Cumulative usage is still 64; the 5 MiB write never
|
||||||
# registered a row.)
|
# registered a row.) Trigger the sweep again to guarantee cache
|
||||||
|
# freshness — the 5 MiB attempt was refused pre-write so no
|
||||||
|
# delta was queued, but the previous sweep's invalidation was
|
||||||
|
# consumed by the intervening GET which re-populated the cache
|
||||||
|
# with the pre-refused-write value. Sweep + re-check for
|
||||||
|
# determinism.
|
||||||
|
POST {{base_url}}/api/admin/internal/trigger-sweep
|
||||||
|
Authorization: Bearer {{admin_token}}
|
||||||
|
|
||||||
|
HTTP 200
|
||||||
|
|
||||||
|
|
||||||
GET {{base_url}}/api/drives
|
GET {{base_url}}/api/drives
|
||||||
Authorization: Bearer {{owner_token}}
|
Authorization: Bearer {{owner_token}}
|
||||||
|
|
||||||
@@ -211,13 +237,17 @@ HTTP 201
|
|||||||
|
|
||||||
|
|
||||||
# Unlimited drive's `used_bytes` climbs to the file's exact size
|
# Unlimited drive's `used_bytes` climbs to the file's exact size
|
||||||
# (5 MiB = 5_242_880 bytes). Same retry block because the delta
|
# (5 MiB = 5_242_880 bytes). Trigger-sweep pattern (see above).
|
||||||
# hook is fire-and-forget here too.
|
POST {{base_url}}/api/admin/internal/trigger-sweep
|
||||||
|
Authorization: Bearer {{admin_token}}
|
||||||
|
[Options]
|
||||||
|
delay: 200ms
|
||||||
|
|
||||||
|
HTTP 200
|
||||||
|
|
||||||
|
|
||||||
GET {{base_url}}/api/drives
|
GET {{base_url}}/api/drives
|
||||||
Authorization: Bearer {{owner_token}}
|
Authorization: Bearer {{owner_token}}
|
||||||
[Options]
|
|
||||||
retry: 10
|
|
||||||
retry-interval: 200ms
|
|
||||||
|
|
||||||
HTTP 200
|
HTTP 200
|
||||||
[Asserts]
|
[Asserts]
|
||||||
@@ -384,7 +414,15 @@ HTTP 200
|
|||||||
|
|
||||||
|
|
||||||
# `used_bytes` on the tight drive is unchanged — the two refused
|
# `used_bytes` on the tight drive is unchanged — the two refused
|
||||||
# operations above never wrote anything.
|
# operations above never wrote anything. Trigger-sweep so the
|
||||||
|
# check reads live SQL (see the class doc on the earlier
|
||||||
|
# sweep + GET pair for the design rationale).
|
||||||
|
POST {{base_url}}/api/admin/internal/trigger-sweep
|
||||||
|
Authorization: Bearer {{admin_token}}
|
||||||
|
|
||||||
|
HTTP 200
|
||||||
|
|
||||||
|
|
||||||
GET {{base_url}}/api/drives
|
GET {{base_url}}/api/drives
|
||||||
Authorization: Bearer {{owner_token}}
|
Authorization: Bearer {{owner_token}}
|
||||||
|
|
||||||
|
|||||||
@@ -130,15 +130,27 @@ file: file,fixtures/hello.txt; text/plain
|
|||||||
HTTP 201
|
HTTP 201
|
||||||
|
|
||||||
|
|
||||||
# Wait for the drive-side fire-and-forget delta to settle.
|
# Force freshness on `drives.used_bytes`:
|
||||||
# Acts as the synchronisation point: by the time `drives.used_bytes`
|
# 1. 200 ms delay to let the fire-and-forget tokio task from the
|
||||||
# reflects the upload, the sibling user-side delta task spawned in
|
# upload above land its SQL write (see
|
||||||
# the same call has had its chance to run too.
|
# `bug_trigger_sweep_vs_spawn_hook_race`).
|
||||||
|
# 2. Trigger the reconciliation sweep — the ONLY path that
|
||||||
|
# invalidates `readable_cache` / `default_drive_cache` after
|
||||||
|
# Ed's 2026-07-17 design call (per-write invalidation would
|
||||||
|
# nuke the cache on every upload, defeating the point). Also
|
||||||
|
# acts as the synchronisation point for the user-envelope
|
||||||
|
# assertion below — the sweep is the authoritative
|
||||||
|
# ground-truth for both drive- and user-side counters.
|
||||||
|
POST {{base_url}}/api/admin/internal/trigger-sweep
|
||||||
|
Authorization: Bearer {{admin_token}}
|
||||||
|
[Options]
|
||||||
|
delay: 200ms
|
||||||
|
|
||||||
|
HTTP 200
|
||||||
|
|
||||||
|
|
||||||
GET {{base_url}}/api/drives
|
GET {{base_url}}/api/drives
|
||||||
Authorization: Bearer {{owner_token}}
|
Authorization: Bearer {{owner_token}}
|
||||||
[Options]
|
|
||||||
retry: 10
|
|
||||||
retry-interval: 200ms
|
|
||||||
|
|
||||||
HTTP 200
|
HTTP 200
|
||||||
[Asserts]
|
[Asserts]
|
||||||
|
|||||||
Reference in New Issue
Block a user