Merge pull request #620 from AtalayaLabs/claude/performance-optimization-analysis-0zdjcl
This commit is contained in:
Dionisio Pozo
+25
@@ -354,6 +354,31 @@ name = "bench_micro_allocs"
|
||||
path = "examples/bench_micro_allocs.rs"
|
||||
required-features = ["bench"]
|
||||
|
||||
# Round-12 battery ────────────────────────────────────────────────────────────
|
||||
|
||||
# Round-12 query-shape pack — sharee narrow read + trgm, login/email stamp
|
||||
# narrowing, session-rotation fused txn, WOPI triple join!, fused quota pair
|
||||
# (needs the dev Postgres up).
|
||||
[[example]]
|
||||
name = "bench_round12_queries"
|
||||
path = "examples/bench_round12_queries.rs"
|
||||
required-features = ["bench"]
|
||||
|
||||
# Round-12 CPU/alloc micro-pack — sized listing JSON, single-pass compression
|
||||
# predicate, fused security-header middleware, media single-read extraction,
|
||||
# chunked-session fused lookups. No Postgres.
|
||||
[[example]]
|
||||
name = "bench_round12_micro"
|
||||
path = "examples/bench_round12_micro.rs"
|
||||
required-features = ["bench"]
|
||||
|
||||
# Blob-cache index — Mutex<LruCache> vs moka byte-weigher (index scaling,
|
||||
# warm-hit reads, eviction-unlink + single-flight safety gates). No Postgres.
|
||||
[[example]]
|
||||
name = "bench_blob_cache_index"
|
||||
path = "examples/bench_blob_cache_index.rs"
|
||||
required-features = ["bench"]
|
||||
|
||||
# Round-11 battery ────────────────────────────────────────────────────────────
|
||||
|
||||
# Round-11 CPU/alloc micro-pack — download DTO hand-off, Last-Modified stack
|
||||
|
||||
@@ -0,0 +1,291 @@
|
||||
# Round 12 — auth write-path narrowing, fused quota gate, moka blob-cache index, media single-read, sized listing JSON
|
||||
|
||||
Benchmark-gated, same rule as ROUND2-11: every change ships with a
|
||||
BEFORE/AFTER benchmark and an equivalence/safety gate; an AFTER that doesn't
|
||||
beat its BEFORE gets rolled back or redesigned. One candidate went through
|
||||
exactly that loop this round (§Rejected): the single-pass compression
|
||||
predicate — the profiler-plausible "28 redundant Content-Type reads" turned
|
||||
out to cost ~4.6 ns TOTAL once monomorphized, and the fused replacement
|
||||
measured within noise, so the declarative chain stays.
|
||||
|
||||
Measured on 4 cores / 15 GiB, local PostgreSQL 16 (fsync off), release
|
||||
profile; frontend on Node 22 / vitest 4. Reproduce any row with the command
|
||||
in its section.
|
||||
|
||||
## Summary
|
||||
|
||||
| # | change | key metric | before → after |
|
||||
|--:|---|---|---|
|
||||
| Q1 | NC sharee search: username-only projection (was 21 wide columns incl. the ≤512 KiB avatar per match) | 26-row page, 3 000 users, all matches avatared | 11.77 → 2.37 ms (**4.98x**) |
|
||||
| Q1b | + `gin_trgm_ops` indexes on `auth.users` (migration 20260719000000) | same page, leading-wildcard ILIKE | → 0.215 ms (**54.7x** total) |
|
||||
| Q2 | Password login: redundant full-row `update_user` deleted (`create_session` already stamps `last_login_at`) | ms/login, 256 KiB avatar | 2.96 → 0.67 (**4.45x**) · −1 txn, −17-column rewrite, −512 KiB clone |
|
||||
| Q3 | Email-verified stamp → narrow conditional UPDATE (magic-link) | ms/stamp | 2.20 → 0.25 (**8.9x**) |
|
||||
| Q3b | OIDC repeat login → in-memory compare, sync only on change | queries per repeat login | full-row rewrite (2.37 ms) → **0 queries** |
|
||||
| Q4 | Refresh-token rotation: 2 transactions → 1 (`rotate_session`) | ms/rotation | 1.135 → 0.959 (**1.18x**) |
|
||||
| Q5 | WOPI CheckFileInfo triple → `tokio::join!` (real `PgAclEngine`) | ms/call | cold 0.485 → 0.363 (**1.34x**) · warm 0.228 → 0.209 |
|
||||
| Q6 | Upload quota pair → ONE fused read (user envelope + drive cap) — NC chunk PUT pays it per chunk | ms/check | 0.350 → 0.193 (**1.81x**) · 2 → 1 queries/chunk |
|
||||
| M1 | Listing JSON: pre-sized buffer (`sized_json`) vs axum `Json`'s 128 B seed | 500-row page | 282.4 → 201.0 µs (**1.40x**) · 13 → 2 allocs |
|
||||
| M3 | Security headers: 4 `SetResponseHeaderLayer` + CSP middleware → 1 fused pass | per request (incl. router) | 5.35 → 3.74 µs (**1.43x**) · −26 allocs |
|
||||
| M4 | Media capture-metadata: single-read (images were read 2-3×, videos opened 2×) | warm geomean / cold cache | **1.44x** warm · **1.6-3.2x** cold · opens 2-3 → 1 |
|
||||
| M5 | Chunked-upload session ops: 5 → 3 map lookups + stack-encoded uuid compare | ns per chunk (prepare+commit) | 469 → 366 (**1.28x**) · −2 allocs |
|
||||
| B1 | Blob-cache index: `Mutex<LruCache>` → moka byte-weigher | pure index probes, K readers | K=2 **2.17x**, K=4 1.61x, K=8 1.46x (mutex scaled NEGATIVELY: 2.08 → 1.07 Mops/s from 1 → 2 readers) |
|
||||
| B2 | `put_blob` populates the cache BEFORE the inner backend consumes the source (was: after → failed 100%) | first read after whole-file put | full remote re-download → local hit |
|
||||
| F1 | SPA list view: 150 px `icon` thumbnails (was 400 px `preview` into a 40 px slot) | pixels per list thumbnail | **~7.1x fewer** (≈4-5x fewer bytes) |
|
||||
|
||||
## [Q1] NC sharee search — the 512 KiB-per-row autocomplete
|
||||
|
||||
```
|
||||
cargo run --release --features bench --example bench_round12_queries # §1
|
||||
```
|
||||
|
||||
`handle_sharees_search` fired `search_users` per keystroke — the full
|
||||
21-column row (incl. the ≤512 KiB avatar data-URI `image`, TOAST-detoasted
|
||||
per match) hydrated into `User` → `UserDto`, of which the handler read ONLY
|
||||
`username`. And the leading-wildcard `ILIKE '%q%'` had no trigram index, so
|
||||
every keystroke seq-scanned `auth.users` (contacts/files/folders all have
|
||||
`gin_trgm_ops`; users was the gap). Now: `search_usernames` port method
|
||||
(same WHERE/ORDER/LIMIT, username-only projection; NULL usernames filtered
|
||||
app-side exactly like the wide flow's post-limit filter) + the two trgm
|
||||
indexes. Gates: identical username lists, with and without the indexes.
|
||||
The wide method stays for the admin table (which serializes `image`).
|
||||
|
||||
## [Q2][Q3][Q3b] Auth write-path narrowing
|
||||
|
||||
```
|
||||
cargo run --release --features bench --example bench_round12_queries # §2-3
|
||||
```
|
||||
|
||||
- **Login** ran `update_user(user.clone())` — a transaction rewriting all
|
||||
17 columns (incl. the avatar, plus a 512 KiB deep clone to feed it) —
|
||||
purely to persist `last_login_at`… which `create_session` overwrites in
|
||||
its own transaction three lines later. Nothing reads the row in between
|
||||
(verified). The call is deleted; the in-memory `register_login()` stays
|
||||
so the response DTO carries the timestamp.
|
||||
- **Magic-link redemption** kept its `update_user` for the email-verified
|
||||
stamp only (last-login again covered by `create_session`) — now a narrow
|
||||
`WHERE … AND email_verified_at IS NULL` single-column UPDATE, idempotency
|
||||
moved into SQL (gated: second stamp is a 0-row no-op, first timestamp
|
||||
preserved).
|
||||
- **OIDC repeat login** additionally syncs the IdP avatar. The row fetched
|
||||
by `get_user_by_oidc_subject` already carries the stored avatar +
|
||||
verification stamp, so the service now compares IN MEMORY and issues NO
|
||||
query at all on the repeat-login common case (same picture, already
|
||||
verified) — the bench's §3b arm is the reason: even a guarded
|
||||
`IS DISTINCT FROM` no-op UPDATE ships the ≤512 KiB avatar parameter over
|
||||
the wire just to compare it (1.20 ms vs the 2.37 ms full-row rewrite;
|
||||
the in-memory skip makes it 0). When something DID change,
|
||||
`sync_oidc_login_profile` runs the guarded narrow UPDATE (image +
|
||||
conditional stamp, `update_storage_usage` pattern) instead of the
|
||||
17-column rewrite.
|
||||
|
||||
## [Q4] Refresh rotation — one transaction
|
||||
|
||||
`refresh_token` paid two full BEGIN/COMMIT pairs per rotation
|
||||
(`revoke_session` then `create_session`), and DAV clients rotate
|
||||
constantly. New `rotate_session(old_id, new_session)` port method: revoke +
|
||||
insert + last-login stamp in one `with_transaction`. Gates: old session
|
||||
revoked, new session live, reuse-detection semantics untouched (family
|
||||
revocation still fires on replay). The per-rotation "Session … revoked"
|
||||
info-line is gone with the old method call (routine rotation is not a
|
||||
security event; explicit logout/family revocation still log).
|
||||
|
||||
## [Q5] WOPI CheckFileInfo — three independent lookups overlapped
|
||||
|
||||
The handler ran require(Read) → get_file → check(Update) serially; all
|
||||
three key off `(caller, file)` alone. Now `tokio::join!` with results
|
||||
evaluated in the original precedence (Read gate first, then 404, then the
|
||||
can_write hint — deny responses byte-identical; the Update probe still
|
||||
skips its query when the token has no write claim). Same fusion applied to
|
||||
`authorize_wopi_access` (host page / editor-url). Cold is the shape that
|
||||
matters: office editors poll CheckFileInfo through a session, but each
|
||||
(file × TTL-window) pays the cold chain once.
|
||||
|
||||
## [Q6] Fused upload-quota gate
|
||||
|
||||
`refuse_if_over_quota` (NC chunked PUT — runs on EVERY chunk) issued the
|
||||
user-envelope read and the drive-cap read serially. One `LEFT JOIN` row
|
||||
now carries both counter pairs; the verdict evaluators were extracted
|
||||
(`eval_user_envelope` / `eval_drive_cap`) and are shared by the old point
|
||||
methods and the fused one, so every error string is identical by
|
||||
construction. Gates: verdict identity across ok / drive-over / user-over
|
||||
(precedence) / unlimited / missing-drive. A `check_upload_quotas_by_folder`
|
||||
twin exists for folder-keyed callers; the three REST once-per-upload pair
|
||||
sites were left as-is (their two checks carry different rejection logs, and
|
||||
one query per whole upload isn't worth entangling that — see §Skipped).
|
||||
|
||||
## [M1] `sized_json` — the 128-byte seed on every listing
|
||||
|
||||
```
|
||||
cargo run --release --features bench --example bench_round12_micro # §1
|
||||
```
|
||||
|
||||
axum's `Json` serializes into `BytesMut::with_capacity(128)`; a 500-row
|
||||
listing (~190 KB) grows it through ~11 doubling reallocs, memcpy-ing ~1.3×
|
||||
the payload. `interfaces::api::sized_json` pre-sizes from the row count
|
||||
(FileDto ≈ 380 B serialized; estimate 384) and serves byte-identical output
|
||||
(gated). Applied to the four hot listing responses: `list_files` (which is
|
||||
UNBOUNDED — no page cap), folder resources, photos timeline, search (both
|
||||
verbs).
|
||||
|
||||
## [M3] Security-header stack 5 → 1
|
||||
|
||||
The CSP middleware already post-processed every response; the four static
|
||||
headers (`x-content-type-options`, `x-frame-options`, `referrer-policy`,
|
||||
`permissions-policy`) each rode their own `SetResponseHeaderLayer` on top.
|
||||
Folded into the same pass — inserted before the 304 early-return because
|
||||
the standalone layers stamped 304s too. Gate: status + full sorted header
|
||||
set byte-identical for json / html / 304 through real axum routers.
|
||||
|
||||
## [M4] Media capture-metadata single-read (the ROUND11 deferred lead)
|
||||
|
||||
```
|
||||
cargo run --release --features bench --example bench_round12_micro # §4
|
||||
```
|
||||
|
||||
`extract_blocking` read each image once wholesale for kamadak, then
|
||||
nom-exif re-opened the SAME file (`read_exif(path)`), and date-less images
|
||||
paid a third open (`read_track(path)` fallback). Videos opened twice (a
|
||||
doomed `read_exif` sniff, then `read_track`). Now: nom-exif parses from the
|
||||
kamadak buffer zero-copy (`MediaSource::from_memory` over the same `Bytes`
|
||||
allocation, API verified on the pinned 3.6.1), one reused `MediaParser`,
|
||||
and videos open once with a `kind()` dispatch. The track fallback for
|
||||
images SURVIVES (fed from the same bytes) — it covers MIME-mislabeled rows,
|
||||
the only case where it ever produced a date; behaviour is
|
||||
observable-identical (gated over dated/undated JPEG, PNG, crafted MP4 —
|
||||
corpus asserted non-vacuous: the crafted EXIF date and mvhd creation time
|
||||
must actually extract). Warm: 1.44x geomean. Cold cache (`drop_caches`
|
||||
arms): dated JPEG 0.81 → 0.34 ms, undated 0.97 → 0.30, PNG 0.12 → 0.06,
|
||||
MP4 0.050 → 0.032. Per-image opens 2-3 → 1; the backfill sweeps multiply
|
||||
this by the library size.
|
||||
|
||||
## [M5] Chunked-upload session ops
|
||||
|
||||
`prepare_chunk` ran `verify_session_owner` (own DashMap lookup + a
|
||||
`Uuid::to_string`) then re-fetched the same entry; `commit_chunk` did the
|
||||
same plus its `get_mut` (3 lookups + allocation per chunk). The owner gate
|
||||
now rides the operation's own lookup (same anti-enum not-found for unknown
|
||||
and foreign sessions — gated), and the uuid compares against a
|
||||
stack-encoded hyphenated form. 5 → 3 shard-lock round-trips and −2 allocs
|
||||
per chunk cycle.
|
||||
|
||||
## [B1][B2] Blob-cache: moka byte-weigher index + the put_blob ordering fix
|
||||
|
||||
```
|
||||
cargo run --release --features bench --example bench_blob_cache_index
|
||||
cargo run --release --features bench --example bench_blob_cache # regression guard
|
||||
```
|
||||
|
||||
The ROUND11 deferred headline. The cache index was a
|
||||
`tokio::sync::Mutex<LruCache>`: every cached chunk read took the one global
|
||||
async mutex to probe+promote (LRU `get` needs `&mut`), so a 100-chunk video
|
||||
playback was 100 serialized critical sections and concurrent readers
|
||||
contended process-wide — measured NEGATIVE scaling (2.08 → 1.07 Mops/s
|
||||
going from 1 to 2 readers). `moka::sync::Cache` with a byte weigher makes
|
||||
the probe lock-free (K=2 **2.17x**, K=8 1.46x; end-to-end warm reads with
|
||||
real files 1.00-1.15x on this 4-core box — the gap is the index share of
|
||||
the path and widens with cores/readers). moka also absorbs the byte budget:
|
||||
the manual `current_size` counter + `collect_evictions` sweep are gone; an
|
||||
eviction listener unlinks size-evicted `.blob` files. Safety gates: budget
|
||||
enforced (100 × 1 MiB into a 10 MiB cap → ≥88 files unlinked, survivors
|
||||
readable), a Replaced entry does NOT unlink its file, Explicit
|
||||
invalidations unlink at their call sites, and the per-hash single-flight
|
||||
still collapses 16 concurrent misses to 1 fetch. The `CachedRef` clone
|
||||
bundle (incl. a `cache_dir` PathBuf clone paid on every HIT for a miss-only
|
||||
struct) is gone — internals now borrow `self`.
|
||||
|
||||
Two behavioural notes, both strict improvements: the write-through PUT
|
||||
paths now respect the byte budget (the old index deliberately skipped
|
||||
eviction there, letting write bursts overshoot until the next read-miss);
|
||||
and a restored over-budget cache trims at startup instead of on the next
|
||||
insert.
|
||||
|
||||
**B2 (the ROUND11 correctness note):** `put_blob` populated the cache AFTER
|
||||
`inner.put_blob` — but every inner backend consumes the source file (local
|
||||
renames it, S3/Azure delete it post-upload), so the `fs::copy` failed 100%
|
||||
of the time, silently (`let _`), and the first read after a whole-file put
|
||||
(the backend-migration copier) re-downloaded the blob from the remote.
|
||||
Cache-first now, with invalidate+unlink if the inner put fails so a
|
||||
rejected blob can never be served. The round-3 stampede guard re-run passes
|
||||
against the migrated backend (16 → 1 remote fetches, cache file verified).
|
||||
|
||||
## [F1] SPA list-view thumbnails (vitest gate)
|
||||
|
||||
```
|
||||
cd frontend && npx vitest run src/lib/api/endpoints/round12.bench.test.ts
|
||||
```
|
||||
|
||||
Both views requested the 400 px `preview` rendition; the list row draws it
|
||||
in a 40×40 slot (the 150 px `icon` rendition is already ≥2× retina density
|
||||
there). `thumbSizeForView` switches list rows to `icon` — ~7.1x fewer
|
||||
pixels per thumbnail, roughly 4-8 KB vs 20-40 KB encoded WebP each, across
|
||||
files/recent/favorites/trash/shared list views. Grid keeps `preview`
|
||||
(100×70 slot at 2x DPR genuinely needs it).
|
||||
|
||||
## Rejected / reworked this round (the discipline working)
|
||||
|
||||
- **Single-pass compression predicate**: the sweep flagged "~28 redundant
|
||||
Content-Type header reads per compressible response" in `main.rs`'s
|
||||
`And`-chain. The bench says otherwise: the monomorphized chain runs in
|
||||
**4.6 ns / 0 allocs** total (straight-line inlined probes), and the
|
||||
hand-fused single-pass node measured 5.2 ns on the compressible hot case
|
||||
— within noise, sometimes slower. Not shipped; the declarative chain
|
||||
stays. `bench_round12_micro` §2 keeps the reproducible evidence.
|
||||
|
||||
## Considered and skipped (cost/benefit, not measurement)
|
||||
|
||||
- **REST per-upload quota pair fusion** (multipart / native-chunked /
|
||||
delta): the two checks sit in separate `if` blocks with distinct
|
||||
rejection logs and folder-id guards; fusing saves ONE query per whole
|
||||
upload (not per chunk) and would entangle that flow. The NC per-chunk
|
||||
site — the hot one — is fused (Q6).
|
||||
- **NC per-session quota budget cache** (0 queries per chunk instead of 1):
|
||||
needs a staleness/invalidation story vs concurrent sessions; the fused
|
||||
read already halves the per-chunk cost with bit-identical semantics.
|
||||
Flagged for a future round.
|
||||
- **`lto = "fat"` on the release profile**: the bench profile already uses
|
||||
it; flipping release trades a large link-time regression for every
|
||||
contributor and CI/Docker build against a low-single-digit runtime gain.
|
||||
That's a project-level call for maintainers, not a bench-gated code
|
||||
change — flagged, not shipped.
|
||||
|
||||
## Deferred / flagged (not shipped this round)
|
||||
|
||||
- **Grouped file/grid views are still unvirtualized** (files route
|
||||
`groupBy != ''` mounts EVERY row in both view modes; ResourceList's
|
||||
grouped GRID branch too — trash is grouped-by-default). Design prepared
|
||||
this round: flatten groups into the existing `VirtualRows`
|
||||
(photos-timeline pattern — headers as first-class rows, grid rows as
|
||||
fixed-height strips of `gridColumns(width)` tiles), which also collapses
|
||||
the per-section `VirtualList` scroll listeners the grouped LIST path
|
||||
pays today (one `getBoundingClientRect` per section per scroll tick).
|
||||
This is the next round's headline; it wants its own pass with UI gates.
|
||||
- **Duplicate `TraceLayer` on `/api`** (`routes.rs` layers it again under
|
||||
the global `ClientIpMakeSpan` layer) and the **per-request `client_ip`
|
||||
String** in the span factory — small, want their own measured arms.
|
||||
- **`CachedBlobBackend::local_blob_path` sync `stat`** (ROUND10/11 flag
|
||||
stands — background-extraction paths only; needs an async port variant).
|
||||
- **Media hooks read the same blob up to 3×** per upload (thumbnail +
|
||||
capture-metadata + faces each pull it independently; the latter two read
|
||||
the RAW blob path directly, bypassing the content cache — and, on
|
||||
encrypted deployments, reading ciphertext: correctness note for
|
||||
maintainers, same class as the ROUND11 put_blob note).
|
||||
- **`mp3_duration::from_path` full-file frame scan** runs even when the
|
||||
ID3 `TLEN` tag is present (ingest-path only). Preferring TLEN is a
|
||||
speed/accuracy tradeoff on VBR files — maintainer call.
|
||||
- **Thumbnail orientation re-parses EXIF** that capture-metadata also
|
||||
parses; reusing the persisted `orientation` is ordering-dependent
|
||||
(hooks run concurrently) — needs a small sequencing decision.
|
||||
|
||||
## Environment / methodology
|
||||
|
||||
- `cargo run --release --features bench --example bench_round12_queries`
|
||||
— needs Postgres; seeds and sweeps its own fixtures (BENCH_PASSES,
|
||||
BENCH_SHR_USERS, BENCH_WOPI_FILES, BENCH_WARM_ITERS).
|
||||
- `cargo run --release --features bench --example bench_round12_micro`
|
||||
— counting allocator; §4's cold arms drop the page cache (root; set
|
||||
BENCH_COLD_ITERS=0 to skip).
|
||||
- `cargo run --release --features bench --example bench_blob_cache_index`
|
||||
— index scaling + eviction/single-flight safety gates.
|
||||
- `cargo run --release --features bench --example bench_blob_cache`
|
||||
— round-3 cross-round regression guard (passes against the moka index).
|
||||
- `cd frontend && npx vitest run src/lib/api/endpoints/round12.bench.test.ts`.
|
||||
@@ -0,0 +1,422 @@
|
||||
//! Blob-cache index benchmark — `Mutex<LruCache>` vs moka byte-weigher
|
||||
//! (the ROUND11 deferred lead; no Postgres).
|
||||
//!
|
||||
//! `CachedBlobBackend` keeps its cache index in a
|
||||
//! `tokio::sync::Mutex<LruCache<String, CacheEntry>>`: EVERY cached chunk
|
||||
//! read acquires the one global async mutex to probe + LRU-promote (the
|
||||
//! promote needs `&mut`), so N-core read concurrency collapses onto a
|
||||
//! single serialization domain — and an N-chunk CDC file read is N
|
||||
//! acquisitions, with every other concurrent reader contending.
|
||||
//!
|
||||
//! AFTER: a `moka::sync::Cache` with a byte weigher — lock-free sharded
|
||||
//! reads with striped recency, byte-budget eviction handled by moka
|
||||
//! (replacing the manual `current_size` + `collect_evictions` machinery),
|
||||
//! and an eviction listener that unlinks the evicted `.blob` file (only on
|
||||
//! size-eviction — Replaced/Explicit must NOT unlink, gated below).
|
||||
//!
|
||||
//! Arms:
|
||||
//! [1] pure index ops, K tasks × M hit-probes (the scaling ceiling)
|
||||
//! [2] end-to-end warm-hit read (index probe + open + 64 KiB read),
|
||||
//! K = 1/2/4/8 readers over a shared corpus
|
||||
//! [3] safety gates: byte budget enforced + evicted files unlinked +
|
||||
//! replaced entries keep their file + single-flight still coalesces
|
||||
//! K concurrent misses onto 1 inner fetch
|
||||
//!
|
||||
//! Run:
|
||||
//! cargo run --release --features bench --example bench_blob_cache_index
|
||||
//! Tunables (env): BENCH_OPS (200000), BENCH_FILES (256), BENCH_READERS (8)
|
||||
|
||||
use std::num::NonZeroUsize;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::time::Instant;
|
||||
|
||||
use lru::LruCache;
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
fn env_or<T: std::str::FromStr>(key: &str, default: T) -> T {
|
||||
std::env::var(key)
|
||||
.ok()
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(default)
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct CacheEntry {
|
||||
size: u64,
|
||||
}
|
||||
|
||||
/// BEFORE, verbatim: the shipped index shape + the per-hit prologue
|
||||
/// allocations of `get_blob_stream` (hash `to_string`, `cached_path`
|
||||
/// build, unconditional `cache_dir.clone()`).
|
||||
struct BeforeIndex {
|
||||
cache_dir: PathBuf,
|
||||
index: Arc<Mutex<LruCache<String, CacheEntry>>>,
|
||||
}
|
||||
|
||||
impl BeforeIndex {
|
||||
fn cached_path(&self, hash: &str) -> PathBuf {
|
||||
let prefix = &hash[..2.min(hash.len())];
|
||||
self.cache_dir.join(prefix).join(format!("{hash}.blob"))
|
||||
}
|
||||
|
||||
/// The exact hit-path prologue of `get_blob_stream`.
|
||||
async fn hit_probe(&self, hash: &str) -> Option<PathBuf> {
|
||||
let hash = hash.to_string();
|
||||
let cached = self.cached_path(&hash);
|
||||
let _cache_dir = self.cache_dir.clone(); // paid on hits, used on misses
|
||||
if self.index.lock().await.get(&hash).is_some() {
|
||||
return Some(cached);
|
||||
}
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// AFTER: moka byte-weigher index + borrow-only hit prologue.
|
||||
struct AfterIndex {
|
||||
cache_dir: PathBuf,
|
||||
index: moka::sync::Cache<String, CacheEntry>,
|
||||
}
|
||||
|
||||
impl AfterIndex {
|
||||
fn new(cache_dir: PathBuf, max_bytes: u64) -> Self {
|
||||
Self {
|
||||
cache_dir,
|
||||
index: moka::sync::Cache::builder()
|
||||
.weigher(|_k: &String, e: &CacheEntry| e.size.clamp(1, u32::MAX as u64) as u32)
|
||||
.max_capacity(max_bytes)
|
||||
.build(),
|
||||
}
|
||||
}
|
||||
|
||||
fn cached_path(&self, hash: &str) -> PathBuf {
|
||||
let prefix = &hash[..2.min(hash.len())];
|
||||
self.cache_dir.join(prefix).join(format!("{hash}.blob"))
|
||||
}
|
||||
|
||||
fn hit_probe(&self, hash: &str) -> Option<PathBuf> {
|
||||
if self.index.get(hash).is_some() {
|
||||
return Some(self.cached_path(hash));
|
||||
}
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
// ────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
async fn section_index_ops(hashes: Arc<Vec<String>>) {
|
||||
let ops: usize = env_or("BENCH_OPS", 200_000);
|
||||
let readers_max: usize = env_or("BENCH_READERS", 8);
|
||||
|
||||
let before = Arc::new(BeforeIndex {
|
||||
cache_dir: PathBuf::from("/tmp/bench-blob-idx"),
|
||||
index: Arc::new(Mutex::new(LruCache::new(
|
||||
NonZeroUsize::new(1_000_000).unwrap(),
|
||||
))),
|
||||
});
|
||||
let after = Arc::new(AfterIndex::new(
|
||||
PathBuf::from("/tmp/bench-blob-idx"),
|
||||
u64::MAX,
|
||||
));
|
||||
for h in hashes.iter() {
|
||||
before
|
||||
.index
|
||||
.lock()
|
||||
.await
|
||||
.put(h.clone(), CacheEntry { size: 1024 });
|
||||
after.index.insert(h.clone(), CacheEntry { size: 1024 });
|
||||
}
|
||||
|
||||
println!("\n## [1] Pure index hit-probes (ops total = {ops}, split across K tasks)");
|
||||
println!("| K | BEFORE Mutex<LruCache> Mops/s | AFTER moka Mops/s | speedup |");
|
||||
for k in [1usize, 2, 4, 8].into_iter().filter(|k| *k <= readers_max) {
|
||||
let per_task = ops / k;
|
||||
|
||||
let t = Instant::now();
|
||||
let mut handles = Vec::new();
|
||||
for t_id in 0..k {
|
||||
let idx = before.clone();
|
||||
let hs = hashes.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
for i in 0..per_task {
|
||||
let h = &hs[(i * 31 + t_id * 7) % hs.len()];
|
||||
std::hint::black_box(idx.hit_probe(h).await);
|
||||
}
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
let before_mops = (per_task * k) as f64 / t.elapsed().as_secs_f64() / 1e6;
|
||||
|
||||
let t = Instant::now();
|
||||
let mut handles = Vec::new();
|
||||
for t_id in 0..k {
|
||||
let idx = after.clone();
|
||||
let hs = hashes.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
for i in 0..per_task {
|
||||
let h = &hs[(i * 31 + t_id * 7) % hs.len()];
|
||||
std::hint::black_box(idx.hit_probe(h));
|
||||
}
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
let after_mops = (per_task * k) as f64 / t.elapsed().as_secs_f64() / 1e6;
|
||||
|
||||
println!(
|
||||
"| {k} | {before_mops:>10.2} | {after_mops:>10.2} | {:>6.2}x |",
|
||||
after_mops / before_mops
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async fn section_warm_reads(hashes: Arc<Vec<String>>) {
|
||||
let readers_max: usize = env_or("BENCH_READERS", 8);
|
||||
let reads: usize = 20_000;
|
||||
|
||||
// Real cached files on disk (64 KiB each).
|
||||
let dir = PathBuf::from("/tmp/bench-blob-idx");
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
let payload = vec![0xA5u8; 64 * 1024];
|
||||
let before = Arc::new(BeforeIndex {
|
||||
cache_dir: dir.clone(),
|
||||
index: Arc::new(Mutex::new(LruCache::new(
|
||||
NonZeroUsize::new(1_000_000).unwrap(),
|
||||
))),
|
||||
});
|
||||
let after = Arc::new(AfterIndex::new(dir.clone(), u64::MAX));
|
||||
for h in hashes.iter() {
|
||||
let p = before.cached_path(h);
|
||||
std::fs::create_dir_all(p.parent().unwrap()).unwrap();
|
||||
std::fs::write(&p, &payload).unwrap();
|
||||
before.index.lock().await.put(
|
||||
h.clone(),
|
||||
CacheEntry {
|
||||
size: payload.len() as u64,
|
||||
},
|
||||
);
|
||||
after.index.insert(
|
||||
h.clone(),
|
||||
CacheEntry {
|
||||
size: payload.len() as u64,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
async fn read_file(path: &PathBuf) -> u64 {
|
||||
use tokio::io::AsyncReadExt;
|
||||
let mut f = tokio::fs::File::open(path).await.unwrap();
|
||||
let mut buf = vec![0u8; 64 * 1024];
|
||||
let mut total = 0u64;
|
||||
loop {
|
||||
let n = f.read(&mut buf).await.unwrap();
|
||||
if n == 0 {
|
||||
break;
|
||||
}
|
||||
total += n as u64;
|
||||
}
|
||||
total
|
||||
}
|
||||
|
||||
println!("\n## [2] Warm-hit read (probe + open + 64 KiB read), {reads} reads split across K");
|
||||
println!("| K | BEFORE Kops/s | AFTER Kops/s | speedup |");
|
||||
for k in [1usize, 2, 4, 8].into_iter().filter(|k| *k <= readers_max) {
|
||||
let per_task = reads / k;
|
||||
|
||||
let t = Instant::now();
|
||||
let mut handles = Vec::new();
|
||||
for t_id in 0..k {
|
||||
let idx = before.clone();
|
||||
let hs = hashes.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
for i in 0..per_task {
|
||||
let h = &hs[(i * 31 + t_id * 7) % hs.len()];
|
||||
let p = idx.hit_probe(h).await.expect("hit");
|
||||
std::hint::black_box(read_file(&p).await);
|
||||
}
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
let before_kops = (per_task * k) as f64 / t.elapsed().as_secs_f64() / 1e3;
|
||||
|
||||
let t = Instant::now();
|
||||
let mut handles = Vec::new();
|
||||
for t_id in 0..k {
|
||||
let idx = after.clone();
|
||||
let hs = hashes.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
for i in 0..per_task {
|
||||
let h = &hs[(i * 31 + t_id * 7) % hs.len()];
|
||||
let p = idx.hit_probe(h).expect("hit");
|
||||
std::hint::black_box(read_file(&p).await);
|
||||
}
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
let after_kops = (per_task * k) as f64 / t.elapsed().as_secs_f64() / 1e3;
|
||||
|
||||
println!(
|
||||
"| {k} | {before_kops:>9.1} | {after_kops:>9.1} | {:>6.2}x |",
|
||||
after_kops / before_kops
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// ────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
async fn section_safety_gates() {
|
||||
use dashmap::DashMap;
|
||||
|
||||
println!("\n## [3] Safety gates");
|
||||
|
||||
// (a) Byte budget + eviction-unlink + replaced-keeps-file, on the moka
|
||||
// shape the production migration ships.
|
||||
let dir = PathBuf::from("/tmp/bench-blob-idx-gate");
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
let unlinked = Arc::new(AtomicU64::new(0));
|
||||
|
||||
let cache_dir = dir.clone();
|
||||
let unlinked_l = unlinked.clone();
|
||||
let cache: moka::sync::Cache<String, CacheEntry> = moka::sync::Cache::builder()
|
||||
.weigher(|_k: &String, e: &CacheEntry| e.size.clamp(1, u32::MAX as u64) as u32)
|
||||
.max_capacity(10 * 1024 * 1024) // 10 MiB budget
|
||||
.eviction_listener(move |hash: Arc<String>, _entry, cause| {
|
||||
// Unlink ONLY blobs moka pushed out for size; a Replaced entry
|
||||
// refers to the same path as its replacement, and Explicit
|
||||
// removals (delete_blob) unlink at the call site.
|
||||
if cause == moka::notification::RemovalCause::Size {
|
||||
let prefix = &hash[..2.min(hash.len())];
|
||||
let p = cache_dir.join(prefix).join(format!("{hash}.blob"));
|
||||
let _ = std::fs::remove_file(&p);
|
||||
unlinked_l.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
})
|
||||
.build();
|
||||
|
||||
let payload = vec![0x5Au8; 1024 * 1024]; // 1 MiB blobs
|
||||
for i in 0..100 {
|
||||
let hash = format!("{i:02x}gatehash{i:04}");
|
||||
let prefix = &hash[..2];
|
||||
let p = dir.join(prefix).join(format!("{hash}.blob"));
|
||||
std::fs::create_dir_all(p.parent().unwrap()).unwrap();
|
||||
std::fs::write(&p, &payload).unwrap();
|
||||
cache.insert(
|
||||
hash,
|
||||
CacheEntry {
|
||||
size: payload.len() as u64,
|
||||
},
|
||||
);
|
||||
}
|
||||
cache.run_pending_tasks();
|
||||
let weighted = cache.weighted_size();
|
||||
assert!(weighted <= 10 * 1024 * 1024, "budget exceeded: {weighted}");
|
||||
// Every surviving entry's file exists; evicted files unlinked.
|
||||
let mut on_disk = 0u64;
|
||||
for i in 0..100 {
|
||||
let hash = format!("{i:02x}gatehash{i:04}");
|
||||
let prefix = &hash[..2];
|
||||
let p = dir.join(prefix).join(format!("{hash}.blob"));
|
||||
let exists = p.exists();
|
||||
if cache.get(&hash).is_some() {
|
||||
assert!(exists, "surviving entry lost its file: {hash}");
|
||||
}
|
||||
if exists {
|
||||
on_disk += 1;
|
||||
}
|
||||
}
|
||||
assert!(
|
||||
on_disk <= 12,
|
||||
"disk not trimmed to budget: {on_disk} files remain"
|
||||
);
|
||||
assert!(unlinked.load(Ordering::Relaxed) >= 88);
|
||||
println!(
|
||||
"# gate (a) OK — weighted {:.1} MiB ≤ 10 MiB budget, {} files on disk, {} unlinked",
|
||||
weighted as f64 / (1024.0 * 1024.0),
|
||||
on_disk,
|
||||
unlinked.load(Ordering::Relaxed)
|
||||
);
|
||||
|
||||
// (b) Replacing an entry must NOT unlink the shared path.
|
||||
let u0 = unlinked.load(Ordering::Relaxed);
|
||||
let some_hash = cache
|
||||
.iter()
|
||||
.next()
|
||||
.map(|(k, _)| (*k).clone())
|
||||
.expect("nonempty");
|
||||
let some_path = {
|
||||
let prefix = &some_hash[..2.min(some_hash.len())];
|
||||
dir.join(prefix).join(format!("{some_hash}.blob"))
|
||||
};
|
||||
cache.insert(some_hash.clone(), CacheEntry { size: 1024 * 1024 });
|
||||
cache.run_pending_tasks();
|
||||
assert!(some_path.exists(), "replace unlinked the live file");
|
||||
assert_eq!(
|
||||
unlinked.load(Ordering::Relaxed),
|
||||
u0,
|
||||
"replace must not count as size-eviction unlink"
|
||||
);
|
||||
println!("# gate (b) OK — replaced entry keeps its file");
|
||||
|
||||
// (c) Single-flight (DashMap gate, unchanged by the migration) still
|
||||
// coalesces K concurrent misses to one inner fetch.
|
||||
let fetches = Arc::new(AtomicU64::new(0));
|
||||
let inflight: Arc<DashMap<String, Arc<Mutex<()>>>> = Arc::new(DashMap::new());
|
||||
let done: Arc<moka::sync::Cache<String, CacheEntry>> =
|
||||
Arc::new(moka::sync::Cache::builder().max_capacity(1_000_000).build());
|
||||
let mut handles = Vec::new();
|
||||
for _ in 0..16 {
|
||||
let fetches = fetches.clone();
|
||||
let inflight = inflight.clone();
|
||||
let done = done.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
let hash = "sf-hash".to_string();
|
||||
let gate = inflight
|
||||
.entry(hash.clone())
|
||||
.or_insert_with(|| Arc::new(Mutex::new(())))
|
||||
.clone();
|
||||
let _guard = gate.lock().await;
|
||||
if done.get(&hash).is_some() {
|
||||
return;
|
||||
}
|
||||
// simulate the remote fetch
|
||||
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
|
||||
fetches.fetch_add(1, Ordering::Relaxed);
|
||||
done.insert(hash.clone(), CacheEntry { size: 1 });
|
||||
inflight.remove(&hash);
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
assert_eq!(fetches.load(Ordering::Relaxed), 1, "single-flight broken");
|
||||
println!("# gate (c) OK — 16 concurrent misses → 1 fetch");
|
||||
}
|
||||
|
||||
#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
|
||||
async fn main() {
|
||||
let n_files: usize = env_or("BENCH_FILES", 256);
|
||||
let hashes: Arc<Vec<String>> = Arc::new(
|
||||
(0..n_files)
|
||||
.map(|i| format!("{:02x}benchhash{i:06}", i % 256))
|
||||
.collect(),
|
||||
);
|
||||
|
||||
println!("#################################################################");
|
||||
println!("# Blob-cache index — Mutex<LruCache> vs moka byte-weigher");
|
||||
println!("#################################################################");
|
||||
|
||||
section_index_ops(hashes.clone()).await;
|
||||
section_warm_reads(hashes.clone()).await;
|
||||
section_safety_gates().await;
|
||||
|
||||
println!("\nGATE PASS (safety gates all hold — adopt if [1]/[2] favour moka)");
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -163,3 +163,13 @@ export function fileThumbnailUrl(
|
||||
): string {
|
||||
return `/api/files/${fileId}/thumbnail/${size}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Thumbnail size matched to the rendering slot. List rows draw thumbnails in
|
||||
* a 40×40 box, so the 150px `icon` rendition is already ≥2× retina density —
|
||||
* fetching the 400px `preview` there moved ~7× more pixels than the slot can
|
||||
* show (benches/ROUND12.md §F1). Grid cards (100×70 slot) keep `preview`.
|
||||
*/
|
||||
export function thumbSizeForView(view: 'grid' | 'list'): 'icon' | 'preview' {
|
||||
return view === 'list' ? 'icon' : 'preview';
|
||||
}
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
// Round-12 §F1 — list-view thumbnail rendition (benches/ROUND12.md).
|
||||
//
|
||||
// The list rows draw file thumbnails in a 40×40 CSS-px slot (100×70 in
|
||||
// grid), but both views requested the 400px `preview` rendition. The list
|
||||
// view now requests the 150px `icon` rendition: still ≥2× device-pixel
|
||||
// density for the 40px slot, at ~1/7th of the decoded pixels (and roughly
|
||||
// icon ≈ 4-8 KB vs preview ≈ 20-40 KB encoded WebP per thumbnail).
|
||||
//
|
||||
// Gates: the URL actually switches per view; grid keeps `preview`; the
|
||||
// pixel-area saving is the documented ~7x.
|
||||
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { fileThumbnailUrl, thumbSizeForView } from './files';
|
||||
|
||||
describe('round12 §F1 — thumbnail rendition per view', () => {
|
||||
it('list view requests the icon rendition, grid keeps preview', () => {
|
||||
expect(thumbSizeForView('list')).toBe('icon');
|
||||
expect(thumbSizeForView('grid')).toBe('preview');
|
||||
expect(fileThumbnailUrl('abc', thumbSizeForView('list'))).toBe('/api/files/abc/thumbnail/icon');
|
||||
expect(fileThumbnailUrl('abc', thumbSizeForView('grid'))).toBe(
|
||||
'/api/files/abc/thumbnail/preview'
|
||||
);
|
||||
});
|
||||
|
||||
it('icon rendition moves ~7x fewer pixels than preview for the 40px slot', () => {
|
||||
// Server renditions: icon = 150px, preview = 400px (see the photos
|
||||
// srcset: `icon 150w, preview 400w, large 800w`).
|
||||
const areaRatio = (400 * 400) / (150 * 150);
|
||||
expect(areaRatio).toBeGreaterThan(7);
|
||||
// The 40×40 slot at 2x DPR needs 80px — icon's 150px still
|
||||
// oversamples it; preview was pure waste.
|
||||
expect(150).toBeGreaterThanOrEqual(80);
|
||||
});
|
||||
});
|
||||
@@ -6,7 +6,10 @@ vi.mock('$lib/api/endpoints/people', () => ({
|
||||
fetchPersonPhotos: vi.fn(),
|
||||
renamePerson: vi.fn()
|
||||
}));
|
||||
vi.mock('$lib/api/endpoints/files', () => ({ fileThumbnailUrl: () => '/thumb.png' }));
|
||||
vi.mock('$lib/api/endpoints/files', () => ({
|
||||
fileThumbnailUrl: () => '/thumb.png',
|
||||
thumbSizeForView: () => 'preview' as const
|
||||
}));
|
||||
vi.mock('$lib/stores/dialogs.svelte', () => ({ promptDialog: vi.fn() }));
|
||||
|
||||
import { fetchPeople, fetchPersonPhotos, renamePerson } from '$lib/api/endpoints/people';
|
||||
|
||||
@@ -4,7 +4,8 @@ vi.mock('$lib/api/endpoints/files', () => ({
|
||||
deleteFile: vi.fn(),
|
||||
fileDownloadUrl: () => '/d',
|
||||
fileInlineUrl: () => '/i',
|
||||
fileThumbnailUrl: () => '/t'
|
||||
fileThumbnailUrl: () => '/t',
|
||||
thumbSizeForView: () => 'preview' as const
|
||||
}));
|
||||
vi.mock('$lib/api/endpoints/favorites', () => ({ addFavorite: vi.fn() }));
|
||||
vi.mock('$lib/api/endpoints/photos', () => ({ fetchFileMetadata: vi.fn() }));
|
||||
|
||||
@@ -78,7 +78,7 @@
|
||||
import { formatBytes } from '$lib/utils/format';
|
||||
import { formatDate, iconNameFromClass, fileIconKindClass } from '$lib/utils/display';
|
||||
import { gridColumns } from '$lib/utils/grid';
|
||||
import { fileThumbnailUrl } from '$lib/api/endpoints/files';
|
||||
import { fileThumbnailUrl, thumbSizeForView } from '$lib/api/endpoints/files';
|
||||
import {
|
||||
canThumbnailClientSide,
|
||||
preloadPdf,
|
||||
@@ -621,7 +621,7 @@
|
||||
{#if enableThumbnails && kind === 'file' && mimeVal && canThumbnailClientSide( { id: item.id, name: item.name, mime_type: mimeVal } )}
|
||||
<img
|
||||
class="file-thumb"
|
||||
src={fileThumbnailUrl(item.id)}
|
||||
src={fileThumbnailUrl(item.id, thumbSizeForView(filesStore.viewMode))}
|
||||
alt=""
|
||||
loading="lazy"
|
||||
onerror={(e) => {
|
||||
|
||||
@@ -18,6 +18,7 @@ vi.mock('$lib/api/endpoints/files', () => ({
|
||||
// src for the fallback path; tests don't render actual thumbnails
|
||||
// but the module import needs to succeed.
|
||||
fileThumbnailUrl: () => '/thumb.png',
|
||||
thumbSizeForView: () => 'preview' as const,
|
||||
renameFile: vi.fn(),
|
||||
deleteFile: vi.fn()
|
||||
}));
|
||||
|
||||
@@ -28,6 +28,7 @@
|
||||
fileThumbnailUrl,
|
||||
moveFile,
|
||||
renameFile,
|
||||
thumbSizeForView,
|
||||
uploadFileWithProgress
|
||||
} from '$lib/api/endpoints/files';
|
||||
import { folderZipUrl } from '$lib/api/endpoints/folders';
|
||||
@@ -2137,7 +2138,7 @@
|
||||
{#if canThumbnail(file)}
|
||||
<img
|
||||
class="file-thumb"
|
||||
src={fileThumbnailUrl(file.id)}
|
||||
src={fileThumbnailUrl(file.id, thumbSizeForView(filesStore.viewMode))}
|
||||
alt=""
|
||||
loading="lazy"
|
||||
onerror={(e) => {
|
||||
|
||||
@@ -39,6 +39,7 @@ vi.mock('$lib/api/endpoints/files', () => ({
|
||||
deleteFile: vi.fn(),
|
||||
fileDownloadUrl: () => '/dl',
|
||||
fileThumbnailUrl: () => '/thumb',
|
||||
thumbSizeForView: () => 'preview' as const,
|
||||
moveFile: vi.fn(),
|
||||
renameFile: vi.fn(),
|
||||
uploadFile: vi.fn(),
|
||||
|
||||
@@ -15,7 +15,8 @@ vi.mock('$lib/api/endpoints/photos', () => ({
|
||||
vi.mock('$lib/api/endpoints/people', () => ({ peopleEnabled: vi.fn() }));
|
||||
vi.mock('$lib/api/endpoints/files', () => ({
|
||||
fileDownloadUrl: () => '/dl',
|
||||
fileThumbnailUrl: () => '/thumb'
|
||||
fileThumbnailUrl: () => '/thumb',
|
||||
thumbSizeForView: () => 'preview' as const
|
||||
}));
|
||||
|
||||
import { fetchPhotos } from '$lib/api/endpoints/photos';
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
-- Trigram indexes for the user search path (NC sharee autocomplete + admin
|
||||
-- user search), which filters with a leading-wildcard `ILIKE '%q%'` that no
|
||||
-- btree can serve — every keystroke was a full `auth.users` seq scan.
|
||||
--
|
||||
-- Mirrors the existing `gin_trgm_ops` indexes on contacts / files / folders
|
||||
-- (pg_trgm is a hard startup requirement, see 20260307000000). Measured in
|
||||
-- benches/ROUND12.md §1: 26-row sharee page over 3 000 users drops from
|
||||
-- 2.37 ms (narrow read, seq scan) to 0.22 ms; the gap widens with user count.
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_users_username_trgm
|
||||
ON auth.users USING gin (username gin_trgm_ops);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_users_email_trgm
|
||||
ON auth.users USING gin (email gin_trgm_ops);
|
||||
@@ -131,6 +131,37 @@ pub trait UserStoragePort: Send + Sync + 'static {
|
||||
include_external: bool,
|
||||
) -> Result<Vec<User>, DomainError>;
|
||||
|
||||
/// Username-only projection of [`search_users`] — same WHERE / ORDER /
|
||||
/// LIMIT semantics, but skips hydrating the 21-column row (incl. the
|
||||
/// up-to-512 KiB avatar `image`) when the caller only needs handles.
|
||||
/// Rows whose username is NULL are returned as `None` so callers can
|
||||
/// keep the wide flow's post-limit filtering semantics.
|
||||
async fn search_usernames(
|
||||
&self,
|
||||
query: &str,
|
||||
limit: i64,
|
||||
include_external: bool,
|
||||
) -> Result<Vec<Option<String>>, DomainError>;
|
||||
|
||||
/// Stamps `email_verified_at = NOW()` iff it is still NULL (idempotent,
|
||||
/// preserves the first timestamp — the SQL twin of
|
||||
/// `User::mark_email_verified`). Narrow single-column write; avoids the
|
||||
/// full-row [`update_user`] (incl. the avatar `image`) on the
|
||||
/// magic-link redemption path.
|
||||
async fn mark_email_verified(&self, user_id: Uuid) -> Result<(), DomainError>;
|
||||
|
||||
/// OIDC repeat-login profile sync: persists the IdP-provided avatar and
|
||||
/// stamps `email_verified_at` (guarded, idempotent) in ONE narrow
|
||||
/// statement. The `IS DISTINCT FROM` guard makes the common case (same
|
||||
/// avatar, already verified) a zero-write no-op — vs the full 17-column
|
||||
/// row rewrite this path used to pay per login. `last_login_at` is NOT
|
||||
/// touched here: session creation stamps it, as on every login path.
|
||||
async fn sync_oidc_login_profile(
|
||||
&self,
|
||||
user_id: Uuid,
|
||||
image: Option<&str>,
|
||||
) -> Result<(), DomainError>;
|
||||
|
||||
/// Lists users by role (e.g., "admin" or "user")
|
||||
async fn list_users_by_role(&self, role: &str) -> Result<Vec<User>, DomainError>;
|
||||
|
||||
@@ -239,6 +270,16 @@ pub trait SessionStoragePort: Send + Sync + 'static {
|
||||
/// Creates a new session
|
||||
async fn create_session(&self, session: Session) -> Result<Session, DomainError>;
|
||||
|
||||
/// Refresh-token rotation: revokes `old_session_id` and creates
|
||||
/// `new_session` in ONE transaction (the refresh path used to pay two
|
||||
/// full BEGIN/COMMIT round-trip pairs per rotation). Also stamps the
|
||||
/// user's `last_login_at` exactly like [`create_session`] does.
|
||||
async fn rotate_session(
|
||||
&self,
|
||||
old_session_id: Uuid,
|
||||
new_session: Session,
|
||||
) -> Result<Session, DomainError>;
|
||||
|
||||
/// Gets a session by refresh token
|
||||
async fn get_session_by_refresh_token(
|
||||
&self,
|
||||
|
||||
@@ -786,9 +786,14 @@ impl AuthApplicationService {
|
||||
lc.dispatch_login(&user).await;
|
||||
}
|
||||
|
||||
// Update last login
|
||||
// Update last login (in-memory only — the DTO below carries it).
|
||||
// The full-row `update_user` this path used to issue was 100%
|
||||
// redundant: `create_session` stamps `last_login_at`/`updated_at`
|
||||
// in its own transaction right below, and nothing re-reads the row
|
||||
// in between. Dropping it removes one transaction + a 17-column
|
||||
// rewrite (incl. the up-to-512 KiB avatar) per password login
|
||||
// (benches/ROUND12.md §2, 4.45x).
|
||||
user.register_login();
|
||||
self.user_storage.update_user(user.clone()).await?;
|
||||
|
||||
// Generate tokens using the injected token service
|
||||
let access_token = self.token_service.generate_access_token(&user)?;
|
||||
@@ -1017,9 +1022,12 @@ impl AuthApplicationService {
|
||||
// PR 23: clicking the magic-link IS proof of email control —
|
||||
// stamp the verification (idempotent, preserves the first
|
||||
// timestamp). Applies to both invitation and login-via-email
|
||||
// tokens.
|
||||
// tokens. Narrow single-column write: `last_login_at` is stamped
|
||||
// by `create_session` below, so the full-row `update_user` this
|
||||
// path used to issue only ever contributed the verification
|
||||
// timestamp (benches/ROUND12.md §3, 8.9x).
|
||||
user.mark_email_verified();
|
||||
self.user_storage.update_user(user.clone()).await?;
|
||||
self.user_storage.mark_email_verified(user.id()).await?;
|
||||
|
||||
let access_token = self.token_service.generate_access_token(&user)?;
|
||||
let refresh_token = self.token_service.generate_refresh_token();
|
||||
@@ -1163,15 +1171,15 @@ impl AuthApplicationService {
|
||||
));
|
||||
}
|
||||
|
||||
// Revoke current session before issuing the next token in the family
|
||||
self.session_storage.revoke_session(session.id()).await?;
|
||||
|
||||
// Generate new tokens
|
||||
let access_token = self.token_service.generate_access_token(&user)?;
|
||||
let new_refresh_token = self.token_service.generate_refresh_token();
|
||||
|
||||
// New session inherits the family_id so reuse of any ancestor triggers
|
||||
// full-family revocation
|
||||
// full-family revocation. Revoking the old session and inserting the
|
||||
// new one happen in ONE transaction (`rotate_session`) — this path
|
||||
// used to pay two BEGIN/COMMIT pairs per refresh, and DAV clients
|
||||
// rotate constantly (benches/ROUND12.md §4).
|
||||
let new_session = Session::new(
|
||||
user.id(),
|
||||
new_refresh_token.clone(),
|
||||
@@ -1181,7 +1189,9 @@ impl AuthApplicationService {
|
||||
session.family_id(),
|
||||
);
|
||||
|
||||
self.session_storage.create_session(new_session).await?;
|
||||
self.session_storage
|
||||
.rotate_session(session.id(), new_session)
|
||||
.await?;
|
||||
|
||||
Ok(AuthResponseDto {
|
||||
user: UserDto::from(user),
|
||||
@@ -2030,6 +2040,24 @@ impl AuthApplicationService {
|
||||
Ok(users.into_iter().map(UserDto::from).collect())
|
||||
}
|
||||
|
||||
/// Username-only search for the NC sharee autocomplete: identical
|
||||
/// predicate / order / limit to [`search_users`], but the repository
|
||||
/// projects just `username` — no 21-column hydration (incl. the
|
||||
/// up-to-512 KiB avatar `image`) per matched row, per keystroke
|
||||
/// (benches/ROUND12.md §1). NULL usernames (email-only signups) are
|
||||
/// filtered app-side, exactly like the wide flow's post-limit filter.
|
||||
pub async fn search_sharee_usernames(
|
||||
&self,
|
||||
query: &str,
|
||||
limit: i64,
|
||||
) -> Result<Vec<String>, DomainError> {
|
||||
let names = self
|
||||
.user_storage
|
||||
.search_usernames(query, limit, false)
|
||||
.await?;
|
||||
Ok(names.into_iter().flatten().collect())
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Admin User Management Methods
|
||||
// ========================================================================
|
||||
@@ -2607,6 +2635,15 @@ impl AuthApplicationService {
|
||||
if let Some(lc) = &self.user_lifecycle {
|
||||
lc.dispatch_login(&existing_user).await;
|
||||
}
|
||||
// Decide BEFORE mutating: the row just fetched already
|
||||
// carries the stored avatar + verification stamp, so the
|
||||
// repeat-login common case (same IdP picture, already
|
||||
// verified) skips the DB entirely — the old shape rewrote
|
||||
// all 17 columns per login, and even a guarded UPDATE
|
||||
// would ship the avatar over the wire just to compare it
|
||||
// (benches/ROUND12.md §3b).
|
||||
let needs_profile_sync = existing_user.email_verified_at().is_none()
|
||||
|| existing_user.image() != claims.picture.as_deref();
|
||||
existing_user.register_login();
|
||||
existing_user.set_image(claims.picture.clone());
|
||||
// PR 23: retroactive email verification for OIDC users
|
||||
@@ -2615,7 +2652,16 @@ impl AuthApplicationService {
|
||||
// any user reaching this branch has a verified email
|
||||
// by the IdP's word; stamping is safe and idempotent.
|
||||
existing_user.mark_email_verified();
|
||||
self.user_storage.update_user(existing_user.clone()).await?;
|
||||
// Narrow guarded sync instead of the 17-column row rewrite:
|
||||
// persists the IdP avatar + the verification stamp only
|
||||
// when either actually changed; `last_login_at` is stamped
|
||||
// by `create_session` at the end of this flow
|
||||
// (benches/ROUND12.md §3).
|
||||
if needs_profile_sync {
|
||||
self.user_storage
|
||||
.sync_oidc_login_profile(existing_user.id(), claims.picture.as_deref())
|
||||
.await?;
|
||||
}
|
||||
existing_user
|
||||
}
|
||||
Err(_) => {
|
||||
|
||||
@@ -16,6 +16,10 @@ use uuid::Uuid;
|
||||
* Storage usage is calculated directly from the `storage.files` table
|
||||
* by summing file sizes for each user (using the `user_id` column).
|
||||
*/
|
||||
/// Fused quota-gate row: `(user_used, user_quota, drive_used, drive_quota,
|
||||
/// drive_found)` — see [`StorageUsageService::check_upload_quotas`].
|
||||
type QuotaPairRow = (i64, i64, Option<i64>, Option<i64>, bool);
|
||||
|
||||
pub struct StorageUsageService {
|
||||
pool: Arc<PgPool>,
|
||||
user_repository: Arc<UserPgRepository>,
|
||||
@@ -372,6 +376,18 @@ impl StorageUsageService {
|
||||
// first, so this branch fires only on a deleted-drive race.
|
||||
return Err(DomainError::not_found("Drive", drive_id.to_string()));
|
||||
};
|
||||
Self::eval_drive_cap(used, quota, additional_bytes)
|
||||
}
|
||||
|
||||
/// Drive-cap verdict over already-fetched counters. Shared by
|
||||
/// [`Self::check_drive_quota`] and the fused
|
||||
/// [`Self::check_upload_quotas`] pair so both produce byte-identical
|
||||
/// errors.
|
||||
fn eval_drive_cap(
|
||||
used: i64,
|
||||
quota: Option<i64>,
|
||||
additional_bytes: u64,
|
||||
) -> Result<(), DomainError> {
|
||||
let Some(quota) = quota else {
|
||||
return Ok(()); // unlimited
|
||||
};
|
||||
@@ -391,6 +407,123 @@ impl StorageUsageService {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// User-envelope verdict over already-fetched counters. Shared by
|
||||
/// `check_storage_quota` and the fused [`Self::check_upload_quotas`]
|
||||
/// pair so both produce byte-identical errors.
|
||||
fn eval_user_envelope(used: i64, quota: i64, additional_bytes: u64) -> Result<(), DomainError> {
|
||||
// Quota of 0 means unlimited
|
||||
if quota <= 0 {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let additional = additional_bytes as i64;
|
||||
|
||||
// Case 1: the single file alone exceeds the entire quota
|
||||
if additional > quota {
|
||||
let quota_fmt = format_bytes(quota);
|
||||
let file_fmt = format_bytes(additional);
|
||||
return Err(DomainError::quota_exceeded(format!(
|
||||
"File size ({}) exceeds your total storage quota ({})",
|
||||
file_fmt, quota_fmt
|
||||
)));
|
||||
}
|
||||
|
||||
// Case 2: the upload would push usage over the quota
|
||||
if used + additional > quota {
|
||||
let available = (quota - used).max(0);
|
||||
let avail_fmt = format_bytes(available);
|
||||
let file_fmt = format_bytes(additional);
|
||||
return Err(DomainError::quota_exceeded(format!(
|
||||
"Not enough storage space. File size: {}, available: {}",
|
||||
file_fmt, avail_fmt
|
||||
)));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Fused pre-upload gate: user envelope + drive cap in ONE round-trip.
|
||||
///
|
||||
/// Upload entry points used to run `check_storage_quota` then
|
||||
/// `check_drive_quota` as two serial point reads — and the NC chunked
|
||||
/// PUT pays that pair on EVERY chunk. One `LEFT JOIN` row carries both
|
||||
/// counter pairs; verdict precedence (user envelope first, then drive
|
||||
/// existence, then drive cap) and every error shape are identical to
|
||||
/// the two-call sequence (benches/ROUND12.md §6, 1.81x).
|
||||
///
|
||||
/// Row shape shared with [`Self::check_upload_quotas_by_folder`]:
|
||||
/// `(user_used, user_quota, drive_used, drive_quota, drive_found)`.
|
||||
pub async fn check_upload_quotas(
|
||||
&self,
|
||||
user_id: Uuid,
|
||||
drive_id: Uuid,
|
||||
additional_bytes: u64,
|
||||
) -> Result<(), DomainError> {
|
||||
let row: Option<QuotaPairRow> = sqlx::query_as(
|
||||
r#"
|
||||
SELECT u.storage_used_bytes, u.storage_quota_bytes,
|
||||
d.used_bytes, d.quota_bytes, (d.id IS NOT NULL)
|
||||
FROM auth.users u
|
||||
LEFT JOIN storage.drives d ON d.id = $2
|
||||
WHERE u.id = $1
|
||||
"#,
|
||||
)
|
||||
.bind(user_id)
|
||||
.bind(drive_id)
|
||||
.fetch_optional(self.pool.as_ref())
|
||||
.await
|
||||
.map_err(|e| {
|
||||
DomainError::internal_error("StorageUsage", format!("upload quota lookup: {e}"))
|
||||
})?;
|
||||
|
||||
let Some((uused, uquota, dused, dquota, drive_found)) = row else {
|
||||
return Err(DomainError::not_found("User", user_id.to_string()));
|
||||
};
|
||||
Self::eval_user_envelope(uused, uquota, additional_bytes)?;
|
||||
if !drive_found {
|
||||
return Err(DomainError::not_found("Drive", drive_id.to_string()));
|
||||
}
|
||||
Self::eval_drive_cap(dused.unwrap_or(0), dquota, additional_bytes)
|
||||
}
|
||||
|
||||
/// [`Self::check_upload_quotas`] with the drive resolved from a parent
|
||||
/// folder id — for the REST upload paths, which hold `folder_id`.
|
||||
/// A missing folder (or a folder whose drive vanished mid-race) maps to
|
||||
/// `not_found("Folder")`, exactly like `check_drive_quota_by_folder`.
|
||||
pub async fn check_upload_quotas_by_folder(
|
||||
&self,
|
||||
user_id: Uuid,
|
||||
folder_id: Uuid,
|
||||
additional_bytes: u64,
|
||||
) -> Result<(), DomainError> {
|
||||
let row: Option<QuotaPairRow> = sqlx::query_as(
|
||||
r#"
|
||||
SELECT u.storage_used_bytes, u.storage_quota_bytes,
|
||||
d.used_bytes, d.quota_bytes, (d.id IS NOT NULL)
|
||||
FROM auth.users u
|
||||
LEFT JOIN storage.folders f ON f.id = $2
|
||||
LEFT JOIN storage.drives d ON d.id = f.drive_id
|
||||
WHERE u.id = $1
|
||||
"#,
|
||||
)
|
||||
.bind(user_id)
|
||||
.bind(folder_id)
|
||||
.fetch_optional(self.pool.as_ref())
|
||||
.await
|
||||
.map_err(|e| {
|
||||
DomainError::internal_error("StorageUsage", format!("upload quota lookup: {e}"))
|
||||
})?;
|
||||
|
||||
let Some((uused, uquota, dused, dquota, drive_found)) = row else {
|
||||
return Err(DomainError::not_found("User", user_id.to_string()));
|
||||
};
|
||||
Self::eval_user_envelope(uused, uquota, additional_bytes)?;
|
||||
if !drive_found {
|
||||
return Err(DomainError::not_found("Folder", folder_id.to_string()));
|
||||
}
|
||||
Self::eval_drive_cap(dused.unwrap_or(0), dquota, additional_bytes)
|
||||
}
|
||||
|
||||
/// Same as [`Self::check_drive_quota`] but resolves the drive id
|
||||
/// from a parent folder id. Mirrors
|
||||
/// [`Self::add_drive_storage_usage_delta_by_folder`] so the upload
|
||||
@@ -563,36 +696,7 @@ impl StorageUsagePort for StorageUsageService {
|
||||
// Narrow 2-column read — the full user row carries the up-to-512 KiB
|
||||
// avatar `image` column, paid on every upload quota check otherwise.
|
||||
let (used, quota) = self.user_repository.get_storage_usage(user_id).await?;
|
||||
|
||||
// Quota of 0 means unlimited
|
||||
if quota <= 0 {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let additional = additional_bytes as i64;
|
||||
|
||||
// Case 1: the single file alone exceeds the entire quota
|
||||
if additional > quota {
|
||||
let quota_fmt = format_bytes(quota);
|
||||
let file_fmt = format_bytes(additional);
|
||||
return Err(DomainError::quota_exceeded(format!(
|
||||
"File size ({}) exceeds your total storage quota ({})",
|
||||
file_fmt, quota_fmt
|
||||
)));
|
||||
}
|
||||
|
||||
// Case 2: the upload would push usage over the quota
|
||||
if used + additional > quota {
|
||||
let available = (quota - used).max(0);
|
||||
let avail_fmt = format_bytes(available);
|
||||
let file_fmt = format_bytes(additional);
|
||||
return Err(DomainError::quota_exceeded(format!(
|
||||
"Not enough storage space. File size: {}, available: {}",
|
||||
file_fmt, avail_fmt
|
||||
)));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
Self::eval_user_envelope(used, quota, additional_bytes)
|
||||
}
|
||||
|
||||
async fn get_user_storage_info(&self, user_id: Uuid) -> Result<(i64, i64), DomainError> {
|
||||
|
||||
@@ -327,6 +327,77 @@ impl SessionStoragePort for SessionPgRepository {
|
||||
.map_err(DomainError::from)
|
||||
}
|
||||
|
||||
/// Revoke + insert + last-login stamp in ONE transaction — the refresh
|
||||
/// rotation used to pay two full BEGIN/COMMIT round-trip pairs
|
||||
/// (`revoke_session` then `create_session`) per token refresh.
|
||||
async fn rotate_session(
|
||||
&self,
|
||||
old_session_id: Uuid,
|
||||
new_session: Session,
|
||||
) -> Result<Session, DomainError> {
|
||||
let session_clone = new_session.clone();
|
||||
with_transaction(&self.pool, "rotate_session", |tx| {
|
||||
Box::pin(async move {
|
||||
sqlx::query("UPDATE auth.sessions SET revoked = true WHERE id = $1")
|
||||
.bind(old_session_id)
|
||||
.execute(&mut **tx)
|
||||
.await
|
||||
.map_err(Self::map_sqlx_error)?;
|
||||
|
||||
sqlx::query(
|
||||
r#"
|
||||
INSERT INTO auth.sessions (
|
||||
id, user_id, refresh_token, expires_at,
|
||||
ip_address, user_agent, created_at, revoked, family_id
|
||||
) VALUES (
|
||||
$1, $2, $3, $4, $5, $6, $7, $8, $9
|
||||
)
|
||||
"#,
|
||||
)
|
||||
.bind(session_clone.id())
|
||||
.bind(session_clone.user_id())
|
||||
.bind(session_clone.refresh_token())
|
||||
.bind(session_clone.expires_at())
|
||||
.bind(session_clone.ip_address())
|
||||
.bind(session_clone.user_agent())
|
||||
.bind(session_clone.created_at())
|
||||
.bind(session_clone.is_revoked())
|
||||
.bind(session_clone.family_id())
|
||||
.execute(&mut **tx)
|
||||
.await
|
||||
.map_err(Self::map_sqlx_error)?;
|
||||
|
||||
sqlx::query(
|
||||
r#"
|
||||
UPDATE auth.users
|
||||
SET last_login_at = NOW(), updated_at = NOW()
|
||||
WHERE id = $1
|
||||
"#,
|
||||
)
|
||||
.bind(session_clone.user_id())
|
||||
.execute(&mut **tx)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
tracing::warn!(
|
||||
"Could not update last_login_at for user {}: {}",
|
||||
session_clone.user_id(),
|
||||
e
|
||||
);
|
||||
SessionRepositoryError::DatabaseError(format!(
|
||||
"Session rotated but could not update last_login_at: {}",
|
||||
e
|
||||
))
|
||||
})?;
|
||||
|
||||
Ok(session_clone)
|
||||
}) as BoxFuture<'_, SessionRepositoryResult<Session>>
|
||||
})
|
||||
.await
|
||||
.map_err(DomainError::from)?;
|
||||
|
||||
Ok(new_session)
|
||||
}
|
||||
|
||||
async fn get_session_by_refresh_token(
|
||||
&self,
|
||||
refresh_token: &str,
|
||||
|
||||
@@ -1067,6 +1067,81 @@ impl UserStoragePort for UserPgRepository {
|
||||
.map_err(DomainError::from)
|
||||
}
|
||||
|
||||
async fn search_usernames(
|
||||
&self,
|
||||
query: &str,
|
||||
limit: i64,
|
||||
include_external: bool,
|
||||
) -> Result<Vec<Option<String>>, DomainError> {
|
||||
// Same predicate / order / limit as `search_users`, username-only
|
||||
// projection — the sharee autocomplete path reads nothing else, and
|
||||
// the wide row drags the avatar `image` per matched user.
|
||||
let pattern = format!("%{}%", query);
|
||||
let rows = sqlx::query(
|
||||
r#"
|
||||
SELECT username
|
||||
FROM auth.users
|
||||
WHERE (username ILIKE $1 OR email ILIKE $1)
|
||||
AND ($3 OR is_external = FALSE)
|
||||
ORDER BY username
|
||||
LIMIT $2
|
||||
"#,
|
||||
)
|
||||
.bind(&pattern)
|
||||
.bind(limit)
|
||||
.bind(include_external)
|
||||
.fetch_all(&*self.pool)
|
||||
.await
|
||||
.map_err(Self::map_sqlx_error)
|
||||
.map_err(DomainError::from)?;
|
||||
Ok(rows.into_iter().map(|row| row.get("username")).collect())
|
||||
}
|
||||
|
||||
async fn mark_email_verified(&self, user_id: Uuid) -> Result<(), DomainError> {
|
||||
// SQL twin of `User::mark_email_verified` — stamps once, keeps the
|
||||
// first timestamp, and touches only the two columns involved.
|
||||
sqlx::query(
|
||||
r#"
|
||||
UPDATE auth.users
|
||||
SET email_verified_at = NOW(), updated_at = NOW()
|
||||
WHERE id = $1 AND email_verified_at IS NULL
|
||||
"#,
|
||||
)
|
||||
.bind(user_id)
|
||||
.execute(&*self.pool)
|
||||
.await
|
||||
.map_err(Self::map_sqlx_error)
|
||||
.map_err(DomainError::from)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn sync_oidc_login_profile(
|
||||
&self,
|
||||
user_id: Uuid,
|
||||
image: Option<&str>,
|
||||
) -> Result<(), DomainError> {
|
||||
// `IS DISTINCT FROM` guard (the `update_storage_usage` pattern): the
|
||||
// common repeat-login case — same IdP avatar, already verified —
|
||||
// writes nothing at all (no dead tuple, no WAL).
|
||||
sqlx::query(
|
||||
r#"
|
||||
UPDATE auth.users
|
||||
SET image = $2,
|
||||
email_verified_at = COALESCE(email_verified_at, NOW()),
|
||||
updated_at = NOW()
|
||||
WHERE id = $1
|
||||
AND (image IS DISTINCT FROM $2 OR email_verified_at IS NULL)
|
||||
"#,
|
||||
)
|
||||
.bind(user_id)
|
||||
.bind(image)
|
||||
.execute(&*self.pool)
|
||||
.await
|
||||
.map_err(Self::map_sqlx_error)
|
||||
.map_err(DomainError::from)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_users_by_role(&self, role: &str) -> Result<Vec<User>, DomainError> {
|
||||
UserRepository::list_users_by_role(self, role)
|
||||
.await
|
||||
|
||||
@@ -10,12 +10,9 @@
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
|
||||
use bytes::Bytes;
|
||||
use dashmap::DashMap;
|
||||
use lru::LruCache;
|
||||
use std::num::NonZeroUsize;
|
||||
use tokio::fs;
|
||||
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
|
||||
use tokio::sync::Mutex;
|
||||
@@ -52,12 +49,22 @@ struct CacheEntry {
|
||||
|
||||
/// A `BlobStorageBackend` decorator that adds an LRU disk cache in front of
|
||||
/// a remote backend.
|
||||
///
|
||||
/// The index is a `moka::sync::Cache` with a byte weigher: cached reads
|
||||
/// probe it lock-free (sharded, striped recency) where the previous
|
||||
/// `tokio::sync::Mutex<LruCache>` serialized EVERY cached chunk read on one
|
||||
/// global async mutex — negative scaling under concurrent readers
|
||||
/// (benches/ROUND12.md §B: 2.08 → 1.07 Mops/s going 1 → 2 readers on the
|
||||
/// mutex; moka holds 1.7-2.4). moka also owns the byte budget: eviction by
|
||||
/// weighted size replaces the manual `current_size` counter +
|
||||
/// `collect_evictions` sweep, and the eviction listener unlinks the evicted
|
||||
/// `.blob` (only on size-eviction — a Replaced entry shares its file with
|
||||
/// the replacement, and Explicit invalidations unlink at their call site).
|
||||
pub struct CachedBlobBackend {
|
||||
inner: Arc<dyn BlobStorageBackend>,
|
||||
cache_dir: PathBuf,
|
||||
max_cache_bytes: u64,
|
||||
index: Arc<Mutex<LruCache<String, CacheEntry>>>,
|
||||
current_size: Arc<AtomicU64>,
|
||||
index: moka::sync::Cache<String, CacheEntry>,
|
||||
/// Per-hash single-flight gates for cache misses. K concurrent cold
|
||||
/// readers of one blob (e.g. a video player's parallel Range probes)
|
||||
/// used to each download the FULL blob from the remote backend — and
|
||||
@@ -67,26 +74,38 @@ pub struct CachedBlobBackend {
|
||||
inflight: Arc<DashMap<String, Arc<Mutex<()>>>>,
|
||||
}
|
||||
|
||||
fn cached_path_in(cache_dir: &Path, hash: &str) -> PathBuf {
|
||||
let prefix = &hash[..2.min(hash.len())];
|
||||
cache_dir.join(prefix).join(format!("{hash}.blob"))
|
||||
}
|
||||
|
||||
impl CachedBlobBackend {
|
||||
/// Create a new cached backend wrapping `inner`.
|
||||
pub fn new(inner: Arc<dyn BlobStorageBackend>, config: &BlobCacheConfig) -> Self {
|
||||
let listener_dir = config.cache_dir.clone();
|
||||
Self {
|
||||
inner,
|
||||
cache_dir: config.cache_dir.clone(),
|
||||
max_cache_bytes: config.max_cache_bytes,
|
||||
// Capacity is essentially unbounded — eviction is by byte budget, not count.
|
||||
index: Arc::new(Mutex::new(LruCache::new(
|
||||
NonZeroUsize::new(1_000_000).unwrap(),
|
||||
))),
|
||||
current_size: Arc::new(AtomicU64::new(0)),
|
||||
index: moka::sync::Cache::builder()
|
||||
.weigher(|_k: &String, e: &CacheEntry| e.size.clamp(1, u32::MAX as u64) as u32)
|
||||
.max_capacity(config.max_cache_bytes)
|
||||
.eviction_listener(move |hash: Arc<String>, _entry, cause| {
|
||||
// Size-evicted blobs lose their on-disk file here (the
|
||||
// sweep `collect_evictions` used to do). A quick unlink
|
||||
// on the inserting task's thread, off the hot get path.
|
||||
if cause == moka::notification::RemovalCause::Size {
|
||||
let _ = std::fs::remove_file(cached_path_in(&listener_dir, &hash));
|
||||
}
|
||||
})
|
||||
.build(),
|
||||
inflight: Arc::new(DashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Path where a blob is cached locally.
|
||||
fn cached_path(&self, hash: &str) -> PathBuf {
|
||||
let prefix = &hash[..2.min(hash.len())];
|
||||
self.cache_dir.join(prefix).join(format!("{hash}.blob"))
|
||||
cached_path_in(&self.cache_dir, hash)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -97,7 +116,6 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
let inner = self.inner.clone();
|
||||
let cache_dir = self.cache_dir.clone();
|
||||
let index = self.index.clone();
|
||||
let current_size = self.current_size.clone();
|
||||
Box::pin(async move {
|
||||
inner.initialize().await?;
|
||||
|
||||
@@ -130,14 +148,13 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
}
|
||||
}
|
||||
}
|
||||
// Bulk-insert the rebuilt index under a single brief lock.
|
||||
{
|
||||
let mut idx = index.lock().await;
|
||||
for (stem, size) in entries {
|
||||
idx.put(stem, CacheEntry { size });
|
||||
}
|
||||
// Rebuild the index; if the restored set exceeds the byte
|
||||
// budget, moka trims it (and the eviction listener unlinks the
|
||||
// trimmed files) — the old index carried the excess until the
|
||||
// next insert.
|
||||
for (stem, size) in entries {
|
||||
index.insert(stem, CacheEntry { size });
|
||||
}
|
||||
current_size.store(total_bytes, Ordering::Relaxed);
|
||||
tracing::info!(
|
||||
"Blob cache initialized: {} bytes in cache at {}",
|
||||
total_bytes,
|
||||
@@ -152,22 +169,28 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
hash: &str,
|
||||
source_path: &Path,
|
||||
) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
|
||||
let inner = self.inner.clone();
|
||||
let hash = hash.to_string();
|
||||
let source = source_path.to_path_buf();
|
||||
let self_ref = CachedRef {
|
||||
cache_dir: self.cache_dir.clone(),
|
||||
max_cache_bytes: self.max_cache_bytes,
|
||||
index: self.index.clone(),
|
||||
current_size: self.current_size.clone(),
|
||||
inflight: self.inflight.clone(),
|
||||
};
|
||||
Box::pin(async move {
|
||||
// Write to inner backend
|
||||
let bytes = inner.put_blob(&hash, &source).await?;
|
||||
// Also cache locally (best-effort)
|
||||
let _ = self_ref.insert_into_cache_static(&hash, &source).await;
|
||||
Ok(bytes)
|
||||
// Cache FIRST: every inner backend consumes the source file
|
||||
// (local renames it, S3/Azure delete it after upload), so the
|
||||
// old populate-after-put ordering failed 100% of the time and
|
||||
// the first read after a whole-file put paid a full remote
|
||||
// re-download (the ROUND11 deferred correctness note; fix
|
||||
// gated in benches/ROUND12.md §B).
|
||||
let cached = self.insert_into_cache(&hash, &source).await.is_ok();
|
||||
match self.inner.put_blob(&hash, &source).await {
|
||||
Ok(bytes) => Ok(bytes),
|
||||
Err(e) => {
|
||||
// Never serve a blob the backend rejected: drop the
|
||||
// just-inserted cache entry + file.
|
||||
if cached {
|
||||
self.index.invalidate(&hash);
|
||||
let _ = fs::remove_file(self.cached_path(&hash)).await;
|
||||
}
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -176,18 +199,10 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
hash: &str,
|
||||
data: Bytes,
|
||||
) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
|
||||
let inner = self.inner.clone();
|
||||
let hash = hash.to_string();
|
||||
let self_ref = CachedRef {
|
||||
cache_dir: self.cache_dir.clone(),
|
||||
max_cache_bytes: self.max_cache_bytes,
|
||||
index: self.index.clone(),
|
||||
current_size: self.current_size.clone(),
|
||||
inflight: self.inflight.clone(),
|
||||
};
|
||||
Box::pin(async move {
|
||||
let size = inner.put_blob_from_bytes(&hash, data.clone()).await?;
|
||||
self_ref.cache_bytes_write_through(hash, &data).await;
|
||||
let size = self.inner.put_blob_from_bytes(&hash, data.clone()).await?;
|
||||
self.cache_bytes_write_through(hash, &data).await;
|
||||
Ok(size)
|
||||
})
|
||||
}
|
||||
@@ -202,20 +217,13 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
hash: &str,
|
||||
data: Bytes,
|
||||
) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
|
||||
let inner = self.inner.clone();
|
||||
let hash = hash.to_string();
|
||||
let self_ref = CachedRef {
|
||||
cache_dir: self.cache_dir.clone(),
|
||||
max_cache_bytes: self.max_cache_bytes,
|
||||
index: self.index.clone(),
|
||||
current_size: self.current_size.clone(),
|
||||
inflight: self.inflight.clone(),
|
||||
};
|
||||
Box::pin(async move {
|
||||
let size = inner
|
||||
let size = self
|
||||
.inner
|
||||
.put_blob_from_bytes_unsynced(&hash, data.clone())
|
||||
.await?;
|
||||
self_ref.cache_bytes_write_through(hash, &data).await;
|
||||
self.cache_bytes_write_through(hash, &data).await;
|
||||
Ok(size)
|
||||
})
|
||||
}
|
||||
@@ -235,40 +243,24 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
) -> Pin<Box<dyn std::future::Future<Output = Result<BlobStream, DomainError>> + Send + '_>>
|
||||
{
|
||||
let hash = hash.to_string();
|
||||
let cached = self.cached_path(&hash);
|
||||
let index = self.index.clone();
|
||||
let inner = self.inner.clone();
|
||||
let cache_dir = self.cache_dir.clone();
|
||||
let max_cache_bytes = self.max_cache_bytes;
|
||||
let current_size = self.current_size.clone();
|
||||
let inflight = self.inflight.clone();
|
||||
Box::pin(async move {
|
||||
// Check cache presence (and bump LRU recency) under a brief lock,
|
||||
// then release it BEFORE touching the filesystem so concurrent
|
||||
// readers don't serialize behind a single open() syscall.
|
||||
if index.lock().await.get(&hash).is_some() {
|
||||
// Lock-free cache probe (bumps moka recency) — the old shape
|
||||
// took the one global async mutex here on EVERY cached chunk
|
||||
// read, and cloned `cache_dir` per hit for a miss-only struct.
|
||||
if self.index.get(&hash).is_some() {
|
||||
let cached = self.cached_path(&hash);
|
||||
if let Ok(file) = fs::File::open(&cached).await {
|
||||
let stream: BlobStream =
|
||||
Box::pin(ReaderStream::with_capacity(file, STREAM_CHUNK_SIZE));
|
||||
return Ok(stream);
|
||||
}
|
||||
// Cache entry stale (file vanished) — drop it from the index.
|
||||
if let Some(entry) = index.lock().await.pop(&hash) {
|
||||
current_size.fetch_sub(entry.size, Ordering::Relaxed);
|
||||
}
|
||||
self.index.invalidate(&hash);
|
||||
}
|
||||
|
||||
// Cache miss — fetch from inner (single-flight), spool to cache
|
||||
let self_ref = CachedRef {
|
||||
cache_dir,
|
||||
max_cache_bytes,
|
||||
index: index.clone(),
|
||||
current_size: current_size.clone(),
|
||||
inflight,
|
||||
};
|
||||
let dest = self_ref
|
||||
.fetch_and_cache_singleflight(&hash, &*inner, &cached)
|
||||
.await?;
|
||||
let cached = self.cached_path(&hash);
|
||||
let dest = self.fetch_and_cache_singleflight(&hash, &cached).await?;
|
||||
let file = fs::File::open(&dest).await.map_err(|e| {
|
||||
DomainError::internal_error("BlobCache", format!("re-open cached: {e}"))
|
||||
})?;
|
||||
@@ -285,18 +277,11 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
) -> Pin<Box<dyn std::future::Future<Output = Result<BlobStream, DomainError>> + Send + '_>>
|
||||
{
|
||||
let hash = hash.to_string();
|
||||
let cached = self.cached_path(&hash);
|
||||
let index = self.index.clone();
|
||||
let inner = self.inner.clone();
|
||||
let cache_dir = self.cache_dir.clone();
|
||||
let max_cache_bytes = self.max_cache_bytes;
|
||||
let current_size = self.current_size.clone();
|
||||
let inflight = self.inflight.clone();
|
||||
Box::pin(async move {
|
||||
// Check cache presence (and bump LRU recency) under a brief lock,
|
||||
// then release it BEFORE the open()/seek() syscalls so concurrent
|
||||
// range readers don't serialize behind the index mutex.
|
||||
if index.lock().await.get(&hash).is_some() {
|
||||
// Lock-free cache probe (bumps moka recency); the filesystem is
|
||||
// only touched after the probe, as before.
|
||||
if self.index.get(&hash).is_some() {
|
||||
let cached = self.cached_path(&hash);
|
||||
if let Ok(mut file) = fs::File::open(&cached).await {
|
||||
file.seek(std::io::SeekFrom::Start(start))
|
||||
.await
|
||||
@@ -309,24 +294,14 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
Box::pin(ReaderStream::with_capacity(limited, STREAM_CHUNK_SIZE));
|
||||
return Ok(stream);
|
||||
}
|
||||
if let Some(entry) = index.lock().await.pop(&hash) {
|
||||
current_size.fetch_sub(entry.size, Ordering::Relaxed);
|
||||
}
|
||||
self.index.invalidate(&hash);
|
||||
}
|
||||
|
||||
// Cache miss — fetch full blob into cache (single-flight: a
|
||||
// player's parallel cold Range probes coalesce onto ONE remote
|
||||
// download), then serve the range locally.
|
||||
let self_ref = CachedRef {
|
||||
cache_dir,
|
||||
max_cache_bytes,
|
||||
index: index.clone(),
|
||||
current_size: current_size.clone(),
|
||||
inflight,
|
||||
};
|
||||
let dest = self_ref
|
||||
.fetch_and_cache_singleflight(&hash, &*inner, &cached)
|
||||
.await?;
|
||||
let cached = self.cached_path(&hash);
|
||||
let dest = self.fetch_and_cache_singleflight(&hash, &cached).await?;
|
||||
let mut file = fs::File::open(&dest)
|
||||
.await
|
||||
.map_err(|e| DomainError::internal_error("BlobCache", format!("re-open: {e}")))?;
|
||||
@@ -345,19 +320,13 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
&self,
|
||||
hash: &str,
|
||||
) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>> {
|
||||
let inner = self.inner.clone();
|
||||
let hash = hash.to_string();
|
||||
let cached = self.cached_path(&hash);
|
||||
let index = self.index.clone();
|
||||
let current_size = self.current_size.clone();
|
||||
Box::pin(async move {
|
||||
inner.delete_blob(&hash).await?;
|
||||
// Remove from cache — drop the index lock before the unlink()
|
||||
// syscall so deletes don't serialize concurrent cache lookups.
|
||||
if let Some(entry) = index.lock().await.pop(&hash) {
|
||||
current_size.fetch_sub(entry.size, Ordering::Relaxed);
|
||||
}
|
||||
let _ = fs::remove_file(&cached).await;
|
||||
self.inner.delete_blob(&hash).await?;
|
||||
// Explicit invalidation unlinks here (the eviction listener
|
||||
// only unlinks size-evictions).
|
||||
self.index.invalidate(&hash);
|
||||
let _ = fs::remove_file(self.cached_path(&hash)).await;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
@@ -366,18 +335,13 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
&self,
|
||||
hash: &str,
|
||||
) -> Pin<Box<dyn std::future::Future<Output = Result<bool, DomainError>> + Send + '_>> {
|
||||
let inner = self.inner.clone();
|
||||
let hash = hash.to_string();
|
||||
let index = self.index.clone();
|
||||
Box::pin(async move {
|
||||
// Check cache first (fast)
|
||||
{
|
||||
let mut idx = index.lock().await;
|
||||
if idx.get(&hash).is_some() {
|
||||
return Ok(true);
|
||||
}
|
||||
// Check cache first (fast, lock-free)
|
||||
if self.index.get(&hash).is_some() {
|
||||
return Ok(true);
|
||||
}
|
||||
inner.blob_exists(&hash).await
|
||||
self.inner.blob_exists(&hash).await
|
||||
})
|
||||
}
|
||||
|
||||
@@ -385,23 +349,17 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
&self,
|
||||
hash: &str,
|
||||
) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
|
||||
let inner = self.inner.clone();
|
||||
let hash = hash.to_string();
|
||||
let index = self.index.clone();
|
||||
let cached = self.cached_path(&hash);
|
||||
Box::pin(async move {
|
||||
// Check cache
|
||||
{
|
||||
let mut idx = index.lock().await;
|
||||
if let Some(entry) = idx.get(&hash) {
|
||||
return Ok(entry.size);
|
||||
}
|
||||
// Check cache (lock-free)
|
||||
if let Some(entry) = self.index.get(&hash) {
|
||||
return Ok(entry.size);
|
||||
}
|
||||
// Fallback to cached file on disk (in case index was lost)
|
||||
if let Ok(meta) = fs::metadata(&cached).await {
|
||||
if let Ok(meta) = fs::metadata(self.cached_path(&hash)).await {
|
||||
return Ok(meta.len());
|
||||
}
|
||||
inner.blob_size(&hash).await
|
||||
self.inner.blob_size(&hash).await
|
||||
})
|
||||
}
|
||||
|
||||
@@ -410,19 +368,18 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
) -> Pin<
|
||||
Box<dyn std::future::Future<Output = Result<StorageHealthStatus, DomainError>> + Send + '_>,
|
||||
> {
|
||||
let inner = self.inner.clone();
|
||||
let cache_dir = self.cache_dir.clone();
|
||||
let current_size = self.current_size.clone();
|
||||
let max_bytes = self.max_cache_bytes;
|
||||
Box::pin(async move {
|
||||
let mut status = inner.health_check().await?;
|
||||
let used = current_size.load(Ordering::Relaxed);
|
||||
let mut status = self.inner.health_check().await?;
|
||||
// Flush moka's pending maintenance so the reported byte count
|
||||
// is current (rare admin path — the cost is fine here).
|
||||
self.index.run_pending_tasks();
|
||||
let used = self.index.weighted_size();
|
||||
status.message = format!(
|
||||
"{} | Cache: {}/{} bytes used at {}",
|
||||
status.message,
|
||||
used,
|
||||
max_bytes,
|
||||
cache_dir.display()
|
||||
self.max_cache_bytes,
|
||||
self.cache_dir.display()
|
||||
);
|
||||
status.backend_type = format!("cached({})", status.backend_type);
|
||||
Ok(status)
|
||||
@@ -446,27 +403,13 @@ impl BlobStorageBackend for CachedBlobBackend {
|
||||
}
|
||||
}
|
||||
|
||||
// ── Helper struct for owned references in async closures ───────────
|
||||
|
||||
/// Cloneable set of cache internals — avoids borrow issues in boxed futures.
|
||||
struct CachedRef {
|
||||
cache_dir: PathBuf,
|
||||
max_cache_bytes: u64,
|
||||
index: Arc<Mutex<LruCache<String, CacheEntry>>>,
|
||||
current_size: Arc<AtomicU64>,
|
||||
inflight: Arc<DashMap<String, Arc<Mutex<()>>>>,
|
||||
}
|
||||
|
||||
impl CachedRef {
|
||||
fn cached_path(&self, hash: &str) -> PathBuf {
|
||||
let prefix = &hash[..2.min(hash.len())];
|
||||
self.cache_dir.join(prefix).join(format!("{hash}.blob"))
|
||||
}
|
||||
// ── Cache internals (miss path + population) ───────────────────────
|
||||
|
||||
impl CachedBlobBackend {
|
||||
/// Best-effort write-through cache population shared by both blob-bytes
|
||||
/// PUT paths. Deliberately no eviction sweep here — the byte budget is
|
||||
/// enforced on read-miss inserts (`insert_into_cache_static`), matching
|
||||
/// the historical write-path behavior.
|
||||
/// PUT paths. moka enforces the byte budget on every insert (the old
|
||||
/// index deliberately skipped the eviction sweep on this path, letting
|
||||
/// write bursts overshoot the budget until the next read-miss insert).
|
||||
async fn cache_bytes_write_through(&self, hash: String, data: &Bytes) {
|
||||
let dest = self.cached_path(&hash);
|
||||
if let Some(parent) = dest.parent() {
|
||||
@@ -474,22 +417,17 @@ impl CachedRef {
|
||||
}
|
||||
let _ = fs::write(&dest, data).await;
|
||||
let data_len = data.len() as u64;
|
||||
let mut idx = self.index.lock().await;
|
||||
if let Some(old) = idx.put(hash, CacheEntry { size: data_len }) {
|
||||
self.current_size.fetch_sub(old.size, Ordering::Relaxed);
|
||||
}
|
||||
self.current_size.fetch_add(data_len, Ordering::Relaxed);
|
||||
self.index.insert(hash, CacheEntry { size: data_len });
|
||||
}
|
||||
|
||||
/// Single-flight wrapper around [`Self::fetch_and_cache_static`]: the
|
||||
/// first caller for a hash becomes the leader and downloads; concurrent
|
||||
/// Single-flight wrapper around [`Self::fetch_and_cache`]: the first
|
||||
/// caller for a hash becomes the leader and downloads; concurrent
|
||||
/// callers queue on the per-hash gate, then re-check the cache and serve
|
||||
/// the leader's file without touching the remote backend. Errors are not
|
||||
/// cached — the gate entry is dropped, so the next caller retries.
|
||||
async fn fetch_and_cache_singleflight(
|
||||
&self,
|
||||
hash: &str,
|
||||
inner: &dyn BlobStorageBackend,
|
||||
cached: &Path,
|
||||
) -> Result<PathBuf, DomainError> {
|
||||
let gate = self
|
||||
@@ -501,42 +439,18 @@ impl CachedRef {
|
||||
|
||||
// Re-check under the gate: if we queued behind the leader, the blob
|
||||
// is on disk now and this turns into a local open.
|
||||
if self.index.lock().await.get(hash).is_some() && fs::metadata(cached).await.is_ok() {
|
||||
if self.index.get(hash).is_some() && fs::metadata(cached).await.is_ok() {
|
||||
return Ok(cached.to_path_buf());
|
||||
}
|
||||
|
||||
let result = self.fetch_and_cache_static(hash, inner).await;
|
||||
let result = self.fetch_and_cache(hash).await;
|
||||
// Drop the gate whether we succeeded or failed; a late-arriving
|
||||
// caller after an error creates a fresh gate and retries the fetch.
|
||||
self.inflight.remove(hash);
|
||||
result
|
||||
}
|
||||
|
||||
/// Pop LRU entries until the cache is back within its byte budget,
|
||||
/// returning the on-disk paths of the evicted blobs.
|
||||
///
|
||||
/// Only the in-memory index is touched here (atomic counter + LRU map);
|
||||
/// the caller MUST unlink the returned paths AFTER releasing the index
|
||||
/// lock so the `remove_file` syscalls never run while the mutex is held.
|
||||
fn collect_evictions(&self, idx: &mut LruCache<String, CacheEntry>) -> Vec<PathBuf> {
|
||||
let mut victims = Vec::new();
|
||||
while self.current_size.load(Ordering::Relaxed) > self.max_cache_bytes {
|
||||
if let Some((evicted_hash, evicted_entry)) = idx.pop_lru() {
|
||||
self.current_size
|
||||
.fetch_sub(evicted_entry.size, Ordering::Relaxed);
|
||||
victims.push(self.cached_path(&evicted_hash));
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
victims
|
||||
}
|
||||
|
||||
async fn insert_into_cache_static(
|
||||
&self,
|
||||
hash: &str,
|
||||
source_path: &Path,
|
||||
) -> Result<(), DomainError> {
|
||||
async fn insert_into_cache(&self, hash: &str, source_path: &Path) -> Result<(), DomainError> {
|
||||
let dest = self.cached_path(hash);
|
||||
if let Some(parent) = dest.parent() {
|
||||
fs::create_dir_all(parent).await.map_err(|e| {
|
||||
@@ -553,29 +467,14 @@ impl CachedRef {
|
||||
DomainError::internal_error("BlobCache", format!("cache copy failed: {e}"))
|
||||
})?;
|
||||
|
||||
// Update the index and pick eviction victims under a single brief
|
||||
// lock, then unlink the evicted files AFTER releasing it — file
|
||||
// removal must not run while the index mutex is held.
|
||||
let to_evict = {
|
||||
let mut idx = self.index.lock().await;
|
||||
if let Some(old) = idx.put(hash.to_string(), CacheEntry { size }) {
|
||||
self.current_size.fetch_sub(old.size, Ordering::Relaxed);
|
||||
}
|
||||
self.current_size.fetch_add(size, Ordering::Relaxed);
|
||||
self.collect_evictions(&mut idx)
|
||||
};
|
||||
for path in to_evict {
|
||||
let _ = fs::remove_file(&path).await;
|
||||
}
|
||||
// moka enforces the byte budget; size-evicted victims are unlinked
|
||||
// by the eviction listener.
|
||||
self.index.insert(hash.to_string(), CacheEntry { size });
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn fetch_and_cache_static(
|
||||
&self,
|
||||
hash: &str,
|
||||
inner: &dyn BlobStorageBackend,
|
||||
) -> Result<PathBuf, DomainError> {
|
||||
let stream = inner.get_blob_stream(hash).await?;
|
||||
async fn fetch_and_cache(&self, hash: &str) -> Result<PathBuf, DomainError> {
|
||||
let stream = self.inner.get_blob_stream(hash).await?;
|
||||
|
||||
let dest = self.cached_path(hash);
|
||||
if let Some(parent) = dest.parent() {
|
||||
@@ -630,17 +529,10 @@ impl CachedRef {
|
||||
));
|
||||
}
|
||||
|
||||
let to_evict = {
|
||||
let mut idx = self.index.lock().await;
|
||||
if let Some(old) = idx.put(hash.to_string(), CacheEntry { size: total }) {
|
||||
self.current_size.fetch_sub(old.size, Ordering::Relaxed);
|
||||
}
|
||||
self.current_size.fetch_add(total, Ordering::Relaxed);
|
||||
self.collect_evictions(&mut idx)
|
||||
};
|
||||
for path in to_evict {
|
||||
let _ = fs::remove_file(&path).await;
|
||||
}
|
||||
// moka enforces the byte budget; size-evicted victims are unlinked
|
||||
// by the eviction listener.
|
||||
self.index
|
||||
.insert(hash.to_string(), CacheEntry { size: total });
|
||||
|
||||
Ok(dest)
|
||||
}
|
||||
|
||||
@@ -514,6 +514,17 @@ impl ChunkedUploadService {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Alloc-free owner compare for the per-chunk hot path: the caller's
|
||||
/// `Uuid` is stack-encoded (hyphenated, the format sessions store) —
|
||||
/// `prepare_chunk`/`commit_chunk` used to pay a `Uuid::to_string` each
|
||||
/// plus a dedicated `verify_session_owner` map lookup per chunk
|
||||
/// (benches/ROUND12.md §M5, 1.28x / −2 allocs per chunk).
|
||||
#[inline]
|
||||
fn owner_matches(session_user_id: &str, user_id: Uuid) -> bool {
|
||||
let mut buf = [0u8; 36];
|
||||
session_user_id == user_id.hyphenated().encode_lower(&mut buf) as &str
|
||||
}
|
||||
|
||||
/// Create a new upload session (persists `session.json` + empty `progress.bin`)
|
||||
async fn create_session_inner(
|
||||
&self,
|
||||
@@ -617,9 +628,8 @@ impl ChunkedUploadService {
|
||||
user_id: Uuid,
|
||||
chunk_index: usize,
|
||||
) -> Result<(PathBuf, usize), DomainError> {
|
||||
self.verify_session_owner(upload_id, &user_id.to_string())
|
||||
.map_err(|e| DomainError::new(ErrorKind::NotFound, "ChunkedUpload", e))?;
|
||||
|
||||
// Single map lookup: the owner gate rides the same guard (same
|
||||
// anti-enum not-found for unknown session and foreign session).
|
||||
let session = self.sessions.get(upload_id).ok_or_else(|| {
|
||||
DomainError::new(
|
||||
ErrorKind::NotFound,
|
||||
@@ -627,6 +637,13 @@ impl ChunkedUploadService {
|
||||
format!("Upload session not found: {}", upload_id),
|
||||
)
|
||||
})?;
|
||||
if !Self::owner_matches(&session.user_id, user_id) {
|
||||
return Err(DomainError::new(
|
||||
ErrorKind::NotFound,
|
||||
"ChunkedUpload",
|
||||
format!("Upload session not found: {}", upload_id),
|
||||
));
|
||||
}
|
||||
|
||||
if chunk_index >= session.chunks.len() {
|
||||
return Err(DomainError::new(
|
||||
@@ -678,20 +695,23 @@ impl ChunkedUploadService {
|
||||
computed_checksum: Option<String>,
|
||||
expected_checksum: Option<String>,
|
||||
) -> Result<ChunkUploadResponseDto, DomainError> {
|
||||
self.verify_session_owner(upload_id, &user_id.to_string())
|
||||
.map_err(|e| DomainError::new(ErrorKind::NotFound, "ChunkedUpload", e))?;
|
||||
|
||||
// Re-fetch chunk metadata under fresh lock — guards against the
|
||||
// (vanishingly unlikely) case of a session expiry / cancellation
|
||||
// racing with the write.
|
||||
// Owner gate folded into the metadata read below — one lookup
|
||||
// instead of two, same anti-enum not-found semantics.
|
||||
let (chunk_path, expected_size, persist_path) = {
|
||||
let session = self.sessions.get(upload_id).ok_or_else(|| {
|
||||
DomainError::new(
|
||||
ErrorKind::NotFound,
|
||||
"ChunkedUpload",
|
||||
"Session disappeared".to_string(),
|
||||
format!("Upload session not found: {}", upload_id),
|
||||
)
|
||||
})?;
|
||||
if !Self::owner_matches(&session.user_id, user_id) {
|
||||
return Err(DomainError::new(
|
||||
ErrorKind::NotFound,
|
||||
"ChunkedUpload",
|
||||
format!("Upload session not found: {}", upload_id),
|
||||
));
|
||||
}
|
||||
if chunk_index >= session.chunks.len() {
|
||||
return Err(DomainError::new(
|
||||
ErrorKind::InvalidInput,
|
||||
|
||||
@@ -96,22 +96,33 @@ impl MediaMetadataService {
|
||||
}
|
||||
|
||||
if Self::is_image_file(mime_type) {
|
||||
// ONE disk read: kamadak needs the full buffer anyway, and
|
||||
// nom-exif 3.6+ parses from in-RAM bytes zero-copy
|
||||
// (`MediaSource::from_memory` over the same allocation). This
|
||||
// path used to re-open the file 1-2 more times — nom-exif's
|
||||
// `read_exif(path)` plus a `read_track(path)` fallback for
|
||||
// date-less images (2-3 opens per image, benches/ROUND12.md §M4:
|
||||
// 1.44x warm geomean, 2-3x cold-cache).
|
||||
let buf = std::fs::read(path).ok()?;
|
||||
// Rich EXIF (GPS / camera / orientation / dimensions + naive date)
|
||||
// from the proven kamadak extractor.
|
||||
let kamadak = std::fs::read(path)
|
||||
.ok()
|
||||
.and_then(|b| ExifService::extract(&b));
|
||||
let kamadak = ExifService::extract(&buf);
|
||||
// nom-exif complements kamadak: a timezone-correct capture date and,
|
||||
// crucially, the date + GPS for files kamadak rejects outright
|
||||
// ("Unexpected next IFD"), where `kamadak` is None and the GPS would
|
||||
// otherwise be lost. See `merge_image_metadata`.
|
||||
merge_image_metadata(kamadak, read_nom_exif(path))
|
||||
let bytes = bytes::Bytes::from(buf);
|
||||
merge_image_metadata(kamadak, read_nom_exif_from_bytes(&bytes))
|
||||
} else if Self::is_video_file(mime_type) {
|
||||
// Videos carry no EXIF — pull the container creation time only.
|
||||
read_nom_exif(path).captured_at.map(|dt| ExifMetadata {
|
||||
captured_at: Some(dt),
|
||||
..Default::default()
|
||||
})
|
||||
// Single open + header sniff; the old shape opened twice (a
|
||||
// doomed `read_exif` sniff, then `read_track`).
|
||||
read_nom_exif_video(path)
|
||||
.captured_at
|
||||
.map(|dt| ExifMetadata {
|
||||
captured_at: Some(dt),
|
||||
..Default::default()
|
||||
})
|
||||
} else {
|
||||
None
|
||||
}
|
||||
@@ -375,34 +386,49 @@ struct NomExif {
|
||||
/// carries `OffsetTimeOriginal` (or a tz-aware container time); otherwise the
|
||||
/// naive wall-clock is interpreted as UTC. Either way it is converted to a true
|
||||
/// UTC instant. GPS is returned as signed decimal degrees.
|
||||
fn read_nom_exif(path: &Path) -> NomExif {
|
||||
use nom_exif::{EntryValue, ExifTag, TrackInfoTag, read_exif, read_track};
|
||||
fn nom_to_utc(ev: &nom_exif::EntryValue) -> Option<DateTime<Utc>> {
|
||||
let edt = ev.as_datetime()?;
|
||||
let utc0 = FixedOffset::east_opt(0)?;
|
||||
Some(edt.or_offset(utc0).with_timezone(&Utc))
|
||||
}
|
||||
|
||||
// Captures nothing → `Copy`, so it can be reused across the calls below.
|
||||
let to_utc = |ev: &EntryValue| -> Option<DateTime<Utc>> {
|
||||
let edt = ev.as_datetime()?;
|
||||
let utc0 = FixedOffset::east_opt(0)?;
|
||||
Some(edt.or_offset(utc0).with_timezone(&Utc))
|
||||
};
|
||||
fn nom_fill_from_exif(exif: &nom_exif::Exif, out: &mut NomExif) {
|
||||
use nom_exif::ExifTag;
|
||||
out.captured_at = exif
|
||||
.get(ExifTag::DateTimeOriginal)
|
||||
.and_then(nom_to_utc)
|
||||
.or_else(|| exif.get(ExifTag::CreateDate).and_then(nom_to_utc));
|
||||
if let Some(gps) = exif.gps_info() {
|
||||
out.latitude = gps.latitude_decimal();
|
||||
out.longitude = gps.longitude_decimal();
|
||||
}
|
||||
}
|
||||
|
||||
/// Image arm: nom-exif fed from the buffer the kamadak pass already read —
|
||||
/// `MediaSource::from_memory` shares the `Bytes` refcount, so this re-parses
|
||||
/// without touching the disk again (the old shape re-opened the file once,
|
||||
/// plus a second time for date-less images). The track fallback stays (fed
|
||||
/// from the same bytes): it covers MIME-mislabeled rows whose actual
|
||||
/// container is a video — the only case where it ever produced a date.
|
||||
fn read_nom_exif_from_bytes(bytes: &bytes::Bytes) -> NomExif {
|
||||
use nom_exif::{MediaParser, MediaSource, TrackInfoTag};
|
||||
|
||||
let mut out = NomExif::default();
|
||||
let mut parser = MediaParser::new();
|
||||
|
||||
// Images: EXIF DateTimeOriginal → DateTimeDigitized (CreateDate), plus GPS.
|
||||
if let Ok(exif) = read_exif(path) {
|
||||
out.captured_at = exif
|
||||
.get(ExifTag::DateTimeOriginal)
|
||||
.and_then(to_utc)
|
||||
.or_else(|| exif.get(ExifTag::CreateDate).and_then(to_utc));
|
||||
if let Some(gps) = exif.gps_info() {
|
||||
out.latitude = gps.latitude_decimal();
|
||||
out.longitude = gps.longitude_decimal();
|
||||
}
|
||||
if let Ok(ms) = MediaSource::from_memory(bytes.clone())
|
||||
&& let Ok(iter) = parser.parse_exif(ms)
|
||||
{
|
||||
let exif: nom_exif::Exif = iter.into();
|
||||
nom_fill_from_exif(&exif, &mut out);
|
||||
}
|
||||
|
||||
// Videos / audio containers (mov/mp4/mkv): track creation time.
|
||||
if out.captured_at.is_none()
|
||||
&& let Ok(track) = read_track(path)
|
||||
&& let Some(dt) = track.get(TrackInfoTag::CreateDate).and_then(to_utc)
|
||||
&& let Ok(ms) = MediaSource::from_memory(bytes.clone())
|
||||
&& let Ok(track) = parser.parse_track(ms)
|
||||
&& let Some(dt) = track.get(TrackInfoTag::CreateDate).and_then(nom_to_utc)
|
||||
{
|
||||
out.captured_at = Some(dt);
|
||||
}
|
||||
@@ -410,6 +436,42 @@ fn read_nom_exif(path: &Path) -> NomExif {
|
||||
out
|
||||
}
|
||||
|
||||
/// Video arm: ONE open, dispatched on the sniffed container kind. Matches
|
||||
/// the old `read_exif(path)`-then-`read_track(path)` observable behaviour
|
||||
/// exactly — a Track container never parsed as EXIF (the old first open was
|
||||
/// pure waste) and an Image container never parsed as a track, so the
|
||||
/// two-open sequence always reduced to a single effective parse.
|
||||
fn read_nom_exif_video(path: &Path) -> NomExif {
|
||||
use nom_exif::{MediaKind, MediaParser, MediaSource, TrackInfoTag};
|
||||
|
||||
let mut out = NomExif::default();
|
||||
let Ok(file) = std::fs::File::open(path) else {
|
||||
return out;
|
||||
};
|
||||
let Ok(ms) = MediaSource::seekable(file) else {
|
||||
return out;
|
||||
};
|
||||
let mut parser = MediaParser::new();
|
||||
match ms.kind() {
|
||||
MediaKind::Image => {
|
||||
// MIME said video, bytes say image (mislabeled row): same EXIF
|
||||
// extraction the old `read_exif(path)` performed.
|
||||
if let Ok(iter) = parser.parse_exif(ms) {
|
||||
let exif: nom_exif::Exif = iter.into();
|
||||
nom_fill_from_exif(&exif, &mut out);
|
||||
}
|
||||
}
|
||||
MediaKind::Track => {
|
||||
if let Ok(track) = parser.parse_track(ms)
|
||||
&& let Some(dt) = track.get(TrackInfoTag::CreateDate).and_then(nom_to_utc)
|
||||
{
|
||||
out.captured_at = Some(dt);
|
||||
}
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Combine kamadak's rich EXIF with nom-exif's date + GPS.
|
||||
///
|
||||
/// nom-exif's tz-correct date wins whenever present; its GPS only fills gaps
|
||||
|
||||
@@ -864,7 +864,13 @@ impl FileHandler {
|
||||
}
|
||||
|
||||
tracing::info!("Found {} files", files.len());
|
||||
let mut resp = (StatusCode::OK, Json(files)).into_response();
|
||||
// Pre-sized serialization — this listing is unbounded (no
|
||||
// page cap), the axum Json 128-byte seed reallocs ~11 times
|
||||
// on a big folder (benches/ROUND12.md §M1).
|
||||
let mut resp = crate::interfaces::api::sized_json::sized_json(
|
||||
64 + files.len() * crate::interfaces::api::sized_json::EST_ROW_BYTES,
|
||||
&files,
|
||||
);
|
||||
resp.headers_mut()
|
||||
.insert(header::ETAG, header::HeaderValue::from_str(&etag).unwrap());
|
||||
resp
|
||||
|
||||
@@ -551,11 +551,15 @@ pub async fn list_folder_resources(
|
||||
})
|
||||
.collect();
|
||||
|
||||
(
|
||||
StatusCode::OK,
|
||||
Json(FolderResourcesDto::with_cursor(items, next_cursor)),
|
||||
)
|
||||
.into_response()
|
||||
{
|
||||
// Pre-sized serialization (benches/ROUND12.md §M1).
|
||||
let body = FolderResourcesDto::with_cursor(items, next_cursor);
|
||||
crate::interfaces::api::sized_json::sized_json(
|
||||
128 + body.items.len()
|
||||
* crate::interfaces::api::sized_json::EST_WRAPPED_ROW_BYTES,
|
||||
&body,
|
||||
)
|
||||
}
|
||||
}
|
||||
Err(e) => AppError::from(e).into_response(),
|
||||
}
|
||||
|
||||
@@ -119,7 +119,11 @@ pub async fn list_photos(
|
||||
})
|
||||
.collect();
|
||||
|
||||
let mut response = Json(&dtos).into_response();
|
||||
// Pre-sized serialization (benches/ROUND12.md §M1).
|
||||
let mut response = crate::interfaces::api::sized_json::sized_json(
|
||||
64 + dtos.len() * crate::interfaces::api::sized_json::EST_WRAPPED_ROW_BYTES,
|
||||
&dtos,
|
||||
);
|
||||
{
|
||||
let h = response.headers_mut();
|
||||
h.insert(header::ETAG, header::HeaderValue::from_str(&etag).unwrap());
|
||||
|
||||
@@ -84,7 +84,14 @@ impl SearchHandler {
|
||||
results.files.len(),
|
||||
results.folders.len()
|
||||
);
|
||||
(StatusCode::OK, Json(&*results)).into_response()
|
||||
{
|
||||
// Pre-sized serialization (benches/ROUND12.md §M1).
|
||||
let rows = results.files.len() + results.folders.len();
|
||||
crate::interfaces::api::sized_json::sized_json(
|
||||
256 + rows * crate::interfaces::api::sized_json::EST_WRAPPED_ROW_BYTES,
|
||||
&*results,
|
||||
)
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
error!("Search error: {}", err);
|
||||
@@ -125,7 +132,14 @@ impl SearchHandler {
|
||||
results.files.len(),
|
||||
results.folders.len()
|
||||
);
|
||||
(StatusCode::OK, Json(&*results)).into_response()
|
||||
{
|
||||
// Pre-sized serialization (benches/ROUND12.md §M1).
|
||||
let rows = results.files.len() + results.folders.len();
|
||||
crate::interfaces::api::sized_json::sized_json(
|
||||
256 + rows * crate::interfaces::api::sized_json::EST_WRAPPED_ROW_BYTES,
|
||||
&*results,
|
||||
)
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
error!("Search error: {}", err);
|
||||
|
||||
@@ -83,14 +83,21 @@ pub struct CheckFileInfoResponse {
|
||||
/// structured `audit` line on denial internally, so ops sees the real
|
||||
/// reason without the attacker being able to distinguish "gone" from
|
||||
/// "revoked".
|
||||
/// Shared id parsing for the WOPI authz paths: a malformed caller sub is a
|
||||
/// bad token (401), a malformed file id can't exist (404, anti-enum).
|
||||
fn parse_wopi_ids(caller_sub: &str, file_id: &str) -> Result<(uuid::Uuid, uuid::Uuid), StatusCode> {
|
||||
let caller_uuid = uuid::Uuid::parse_str(caller_sub).map_err(|_| StatusCode::UNAUTHORIZED)?;
|
||||
let file_uuid = uuid::Uuid::parse_str(file_id).map_err(|_| StatusCode::NOT_FOUND)?;
|
||||
Ok((caller_uuid, file_uuid))
|
||||
}
|
||||
|
||||
async fn require_wopi_perm(
|
||||
authz: &PgAclEngine,
|
||||
caller_sub: &str,
|
||||
file_id: &str,
|
||||
perm: Permission,
|
||||
) -> Result<(uuid::Uuid, uuid::Uuid), StatusCode> {
|
||||
let caller_uuid = uuid::Uuid::parse_str(caller_sub).map_err(|_| StatusCode::UNAUTHORIZED)?;
|
||||
let file_uuid = uuid::Uuid::parse_str(file_id).map_err(|_| StatusCode::NOT_FOUND)?;
|
||||
let (caller_uuid, file_uuid) = parse_wopi_ids(caller_sub, file_id)?;
|
||||
authz
|
||||
.require(Subject::User(caller_uuid), perm, Resource::File(file_uuid))
|
||||
.await
|
||||
@@ -118,25 +125,52 @@ async fn check_file_info(
|
||||
|
||||
// Redemption-time authz: even with a valid token, the caller must
|
||||
// still hold Read on this file. Catches revoked-grant-mid-session.
|
||||
if let Err(status) = require_wopi_perm(
|
||||
state.app_state.authorization.as_ref(),
|
||||
&claims.sub,
|
||||
&file_id,
|
||||
Permission::Read,
|
||||
)
|
||||
.await
|
||||
{
|
||||
return status.into_response();
|
||||
//
|
||||
// The Read gate, the metadata fetch and the Update probe are three
|
||||
// independent lookups keyed only off (caller, file) — overlapped with
|
||||
// `tokio::join!` (benches/ROUND12.md §5). Results are evaluated in the
|
||||
// original precedence: Read gate first, then file existence.
|
||||
let (caller_uuid, file_uuid) = match parse_wopi_ids(&claims.sub, &file_id) {
|
||||
Ok(ids) => ids,
|
||||
Err(status) => return status.into_response(),
|
||||
};
|
||||
let authz = state.app_state.authorization.as_ref();
|
||||
let (read_gate, file, can_write_now) = tokio::join!(
|
||||
authz.require(
|
||||
Subject::User(caller_uuid),
|
||||
Permission::Read,
|
||||
Resource::File(file_uuid)
|
||||
),
|
||||
state
|
||||
.app_state
|
||||
.applications
|
||||
.file_retrieval_service
|
||||
.get_file(&file_id),
|
||||
// `user_can_write` = actual current Update permission ∧ token's
|
||||
// can_write flag. If the caller's Update was revoked since the
|
||||
// token was minted (e.g. their grant was downgraded from Editor
|
||||
// to Viewer), the editor sees the file as read-only and won't
|
||||
// even attempt PutFile. The stricter `require_wopi_perm(Update)`
|
||||
// in put_file is the actual gate; this field is a UI hint.
|
||||
async {
|
||||
if claims.can_write {
|
||||
authz
|
||||
.check(
|
||||
Subject::User(caller_uuid),
|
||||
Permission::Update,
|
||||
Resource::File(file_uuid),
|
||||
)
|
||||
.await
|
||||
.unwrap_or(false)
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
);
|
||||
if read_gate.is_err() {
|
||||
return StatusCode::NOT_FOUND.into_response();
|
||||
}
|
||||
|
||||
// Fetch file metadata
|
||||
let file = match state
|
||||
.app_state
|
||||
.applications
|
||||
.file_retrieval_service
|
||||
.get_file(&file_id)
|
||||
.await
|
||||
{
|
||||
let file = match file {
|
||||
Ok(f) => f,
|
||||
Err(_) => return StatusCode::NOT_FOUND.into_response(),
|
||||
};
|
||||
@@ -146,24 +180,6 @@ async fn check_file_info(
|
||||
.map(|dt| dt.to_rfc3339())
|
||||
.unwrap_or_default();
|
||||
|
||||
// `user_can_write` = actual current Update permission ∧ token's
|
||||
// can_write flag. If the caller's Update was revoked since the
|
||||
// token was minted (e.g. their grant was downgraded from Editor
|
||||
// to Viewer), the editor sees the file as read-only and won't
|
||||
// even attempt PutFile. The stricter `require_wopi_perm(Update)`
|
||||
// in put_file is the actual gate; this field is a UI hint.
|
||||
let can_write_now = claims.can_write
|
||||
&& state
|
||||
.app_state
|
||||
.authorization
|
||||
.check(
|
||||
Subject::User(uuid::Uuid::parse_str(&claims.sub).unwrap_or(uuid::Uuid::nil())),
|
||||
Permission::Update,
|
||||
Resource::File(uuid::Uuid::parse_str(&file_id).unwrap_or(uuid::Uuid::nil())),
|
||||
)
|
||||
.await
|
||||
.unwrap_or(false);
|
||||
|
||||
let response = CheckFileInfoResponse {
|
||||
base_file_name: file.name.clone(),
|
||||
// WOPI's `OwnerId` field is required. Post-D7 the DTO no
|
||||
@@ -550,34 +566,31 @@ async fn authorize_wopi_access<S: FileRetrievalUseCase>(
|
||||
) -> Result<(crate::application::dtos::file_dto::FileDto, bool), StatusCode> {
|
||||
let file_uuid = uuid::Uuid::parse_str(file_id).map_err(|_| StatusCode::NOT_FOUND)?;
|
||||
|
||||
// Step 1 — Read is required to even open the file.
|
||||
authz
|
||||
.require(
|
||||
// The Read gate (step 1), the metadata fetch and the Update probe
|
||||
// (step 2) are independent — overlapped with `tokio::join!`
|
||||
// (benches/ROUND12.md §5); results evaluated in the original order.
|
||||
//
|
||||
// Step 2 rationale — can_write reflects real Update, not the client's
|
||||
// action-string. `check` returns bool without throwing; failure
|
||||
// just means the caller lacks Update, so we degrade the token to
|
||||
// read-only. Deliberately no `require` there — a Viewer opening
|
||||
// the file is legitimate; only the write claim is suppressed.
|
||||
let (read_gate, file, has_update) = tokio::join!(
|
||||
authz.require(
|
||||
Subject::User(caller_id),
|
||||
Permission::Read,
|
||||
Resource::File(file_uuid),
|
||||
)
|
||||
.await
|
||||
.map_err(|_| StatusCode::NOT_FOUND)?;
|
||||
|
||||
let file = file_retrieval
|
||||
.get_file(file_id)
|
||||
.await
|
||||
.map_err(|_| StatusCode::NOT_FOUND)?;
|
||||
|
||||
// Step 2 — can_write reflects real Update, not the client's
|
||||
// action-string. `check` returns bool without throwing; failure
|
||||
// just means the caller lacks Update, so we degrade the token to
|
||||
// read-only. Deliberately no `require` here — a Viewer opening
|
||||
// the file is legitimate; only the write claim is suppressed.
|
||||
let has_update = authz
|
||||
.check(
|
||||
),
|
||||
file_retrieval.get_file(file_id),
|
||||
authz.check(
|
||||
Subject::User(caller_id),
|
||||
Permission::Update,
|
||||
Resource::File(file_uuid),
|
||||
)
|
||||
.await
|
||||
.unwrap_or(false);
|
||||
);
|
||||
read_gate.map_err(|_| StatusCode::NOT_FOUND)?;
|
||||
let file = file.map_err(|_| StatusCode::NOT_FOUND)?;
|
||||
let has_update = has_update.unwrap_or(false);
|
||||
|
||||
// Step 3 — allow explicit view-mode downgrade for Editors.
|
||||
let can_write = has_update && requested_action != "view";
|
||||
|
||||
@@ -2,6 +2,7 @@ pub mod cookie_auth;
|
||||
pub mod deserializer;
|
||||
pub mod handlers;
|
||||
pub mod routes;
|
||||
pub mod sized_json;
|
||||
|
||||
pub use routes::create_api_routes;
|
||||
pub use routes::create_health_routes;
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
//! Pre-sized JSON responses for listing endpoints.
|
||||
//!
|
||||
//! `axum::Json` serializes into a `BytesMut::with_capacity(128)` — a 500-row
|
||||
//! listing grows that seed through ~11 doubling reallocations, memcpy-ing
|
||||
//! ~1.3× the payload on every hot listing response (files, folder
|
||||
//! resources, photos timeline, search). `sized_json` serializes into one
|
||||
//! right-sized `Vec` instead: 2 allocations total and no copy chain
|
||||
//! (benches/ROUND12.md §M1, 1.40x / −11 allocs on a 500-row page).
|
||||
//!
|
||||
//! The per-row estimates are calibrated against the serialized DTOs (a
|
||||
//! realistic `FileDto` row measures ~380 B). Underestimates cost one extra
|
||||
//! doubling — still far better than the 128-byte seed; overestimates waste
|
||||
//! transient capacity only (the buffer is freed after the response).
|
||||
|
||||
use axum::http::{HeaderValue, StatusCode, header};
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use bytes::Bytes;
|
||||
use serde::Serialize;
|
||||
|
||||
/// Serialized size estimate for one file/folder row (FileDto ≈ 380 B).
|
||||
pub const EST_ROW_BYTES: usize = 384;
|
||||
|
||||
/// Serialized size estimate for one wrapped resource row (PhotoDto /
|
||||
/// FolderResourcesDto items carry a FileDto plus wrapper fields).
|
||||
pub const EST_WRAPPED_ROW_BYTES: usize = 448;
|
||||
|
||||
/// Serialize `value` into a single pre-sized buffer and wrap it as an
|
||||
/// `application/json` response — drop-in for `Json(value).into_response()`
|
||||
/// (byte-identical body, gated in `bench_round12_micro` §1), minus the
|
||||
/// doubling-realloc chain.
|
||||
pub fn sized_json<T: Serialize>(estimated_bytes: usize, value: &T) -> Response {
|
||||
let mut buf = Vec::with_capacity(estimated_bytes.max(128));
|
||||
match serde_json::to_writer(&mut buf, value) {
|
||||
Ok(()) => (
|
||||
StatusCode::OK,
|
||||
[(
|
||||
header::CONTENT_TYPE,
|
||||
HeaderValue::from_static("application/json"),
|
||||
)],
|
||||
Bytes::from(buf),
|
||||
)
|
||||
.into_response(),
|
||||
// Mirror axum's Json error arm: 500 + plain-text serializer error.
|
||||
Err(err) => (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
[(
|
||||
header::CONTENT_TYPE,
|
||||
HeaderValue::from_static("text/plain; charset=utf-8"),
|
||||
)],
|
||||
err.to_string(),
|
||||
)
|
||||
.into_response(),
|
||||
}
|
||||
}
|
||||
@@ -342,20 +342,21 @@ pub async fn handle_sharees_search(
|
||||
None => return sharees_response(vec![]).into_response(),
|
||||
};
|
||||
|
||||
// SQL-level ILIKE search with limit — avoids loading all users into memory.
|
||||
let users = auth_service
|
||||
.search_users(&search, 26)
|
||||
// SQL-level ILIKE search with limit — avoids loading all users into
|
||||
// memory. Username-only projection: the wide `search_users` row drags
|
||||
// the up-to-512 KiB avatar `image` per matched user, per keystroke
|
||||
// (benches/ROUND12.md §1). NULL-username (email-only signup) rows are
|
||||
// already filtered by the service, preserving the old post-limit
|
||||
// filtering semantics.
|
||||
let usernames = auth_service
|
||||
.search_sharee_usernames(&search, 26)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
|
||||
// Skip users with no claimed username — NC sharees autocomplete relies
|
||||
// on a username being typeable; users still on the email-only signup
|
||||
// path can't be addressed here. Also skip self (don't suggest sharing
|
||||
// with yourself).
|
||||
let matches: Vec<serde_json::Value> = users
|
||||
// Skip self (don't suggest sharing with yourself).
|
||||
let matches: Vec<serde_json::Value> = usernames
|
||||
.into_iter()
|
||||
.filter_map(|u| {
|
||||
let handle = u.username.clone()?;
|
||||
.filter_map(|handle| {
|
||||
if handle.as_str() == &*user.username {
|
||||
return None;
|
||||
}
|
||||
|
||||
@@ -7,7 +7,6 @@ use std::sync::Arc;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::application::ports::file_ports::FileUploadUseCase;
|
||||
use crate::application::ports::storage_ports::StorageUsagePort;
|
||||
use crate::common::di::AppState;
|
||||
use crate::common::mime_detect::filename_from_path;
|
||||
use crate::interfaces::errors::AppError;
|
||||
@@ -52,10 +51,10 @@ async fn refuse_if_over_quota(
|
||||
// remains authoritative.
|
||||
return Ok(());
|
||||
};
|
||||
svc.check_storage_quota(user_id, additional)
|
||||
.await
|
||||
.map_err(AppError::from)?;
|
||||
svc.check_drive_quota(drive_id, additional)
|
||||
// Fused single round-trip (user envelope + drive cap) — this gate runs
|
||||
// on EVERY chunk PUT, and the serial pair cost two point reads per
|
||||
// chunk (benches/ROUND12.md §6). Verdict precedence unchanged.
|
||||
svc.check_upload_quotas(user_id, drive_id, additional)
|
||||
.await
|
||||
.map_err(AppError::from)
|
||||
}
|
||||
|
||||
+27
-19
@@ -24,7 +24,6 @@ use oxicloud::access_log;
|
||||
use oxicloud::interfaces::middleware::trace_span::{ClientIpMakeSpan, UuidRequestId};
|
||||
use tower_http::limit::RequestBodyLimitLayer;
|
||||
use tower_http::request_id::{PropagateRequestIdLayer, SetRequestIdLayer};
|
||||
use tower_http::set_header::SetResponseHeaderLayer;
|
||||
use tower_http::trace::TraceLayer;
|
||||
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
|
||||
|
||||
@@ -918,11 +917,37 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
// • form-action 'https:': the WOPI office editor is launched by POSTing a
|
||||
// token form to a cross-origin, admin-configured Collabora/OnlyOffice
|
||||
// host. Mirrors the SPA meta policy in frontend/svelte.config.js.
|
||||
// The four static security headers ride in the same response pass —
|
||||
// they used to be four separate `SetResponseHeaderLayer`s stacked on
|
||||
// top of this middleware (5 tower layers per response). Folding them
|
||||
// here measured 1.43x per request / −26 allocs with a byte-identical
|
||||
// header set, including on 304s (benches/ROUND12.md §M3). They are
|
||||
// inserted BEFORE the 304 early-return below because the standalone
|
||||
// layers stamped 304s too.
|
||||
async fn content_security_policy(
|
||||
req: axum::extract::Request,
|
||||
next: axum::middleware::Next,
|
||||
) -> axum::response::Response {
|
||||
let mut res = next.run(req).await;
|
||||
{
|
||||
let h = res.headers_mut();
|
||||
h.insert(
|
||||
HeaderName::from_static("x-content-type-options"),
|
||||
HeaderValue::from_static("nosniff"),
|
||||
);
|
||||
h.insert(
|
||||
HeaderName::from_static("x-frame-options"),
|
||||
HeaderValue::from_static("DENY"),
|
||||
);
|
||||
h.insert(
|
||||
HeaderName::from_static("referrer-policy"),
|
||||
HeaderValue::from_static("strict-origin-when-cross-origin"),
|
||||
);
|
||||
h.insert(
|
||||
HeaderName::from_static("permissions-policy"),
|
||||
HeaderValue::from_static("camera=(), microphone=(), geolocation=()"),
|
||||
);
|
||||
}
|
||||
// A 304 Not Modified carries no entity headers (no Content-Type) since
|
||||
// there's no body — `is_html` would read `None` and misclassify it as
|
||||
// "not html", attaching the strict headerless CSP below. Browsers merge
|
||||
@@ -982,24 +1007,7 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
res
|
||||
}
|
||||
|
||||
app = app
|
||||
.layer(axum::middleware::from_fn(content_security_policy))
|
||||
.layer(SetResponseHeaderLayer::overriding(
|
||||
HeaderName::from_static("x-content-type-options"),
|
||||
HeaderValue::from_static("nosniff"),
|
||||
))
|
||||
.layer(SetResponseHeaderLayer::overriding(
|
||||
HeaderName::from_static("x-frame-options"),
|
||||
HeaderValue::from_static("DENY"),
|
||||
))
|
||||
.layer(SetResponseHeaderLayer::overriding(
|
||||
HeaderName::from_static("referrer-policy"),
|
||||
HeaderValue::from_static("strict-origin-when-cross-origin"),
|
||||
))
|
||||
.layer(SetResponseHeaderLayer::overriding(
|
||||
HeaderName::from_static("permissions-policy"),
|
||||
HeaderValue::from_static("camera=(), microphone=(), geolocation=()"),
|
||||
));
|
||||
app = app.layer(axum::middleware::from_fn(content_security_policy));
|
||||
|
||||
// Warn once at startup if auth cookies are not Secure.
|
||||
// HttpOnly + SameSite protection is nullified over plain HTTP because tokens
|
||||
|
||||
Reference in New Issue
Block a user