Merge pull request #688 from EdouardVanbelle/fix/copy_folder_ref_count_issue

This commit is contained in:
Dionisio Pozo
2026-08-24 14:42:35 +02:00
committed by GitHub
28 changed files with 4541 additions and 247 deletions
+1091 -71
View File
File diff suppressed because it is too large Load Diff
+246
View File
@@ -0,0 +1,246 @@
# Plan — Hidden system drive for user-owned objects
**Status:** design captured 2026-08-21. Not implemented. Sibling to
`docs/plan/derived-blobs.md`, which answers "where does *derived*
content live"; this one answers "where do *user-owned binaries* live".
The two share one rule — **point at a file, never at a blob** — and
that rule is the reason neither needs new blob-referencing tables.
## Problem — binaries in the users row
`auth.users.image TEXT` (migration `20260526000000_add_user_image.sql`)
holds the avatar inline, up to 512 KiB. It is the wrong home, and the
cost is already measured rather than theoretical:
- **It TOASTs, and every wide read pays.** The repository comment on
`get_users_by_ids` records a group fan-out that "detoasted + shipped
+ parsed M avatars purely to discard them", fixed by adding a narrow
projection (`benches/ROUND12.md §Q1`, `ROUND13.md §Q1`). That
workaround exists *because* the column is in the wrong place; the
rule it leaves behind — "add a wide sibling rather than widening
this one back" — is a permanent tax on every future query.
- **Base64 inflation.** A ~384 KB image becomes ~512 KB of TEXT.
- **No dedup.** N users sharing a default or IdP-supplied avatar cost
N copies.
- **None of the storage stack applies** — no `EncryptedBlobBackend`,
no backend migration, no key rotation, no local cache, no
consistency coverage.
- **Backups and replication carry it.** Binary weight lands in the
logical dump and on every replica, forever.
The same pressure is coming for the UI background and a signature
image, so this needs a general answer, not another column.
## The rule — point at a file, never at a blob
```sql
ALTER TABLE auth.users
ADD COLUMN avatar_file_id UUID REFERENCES storage.files(id) ON DELETE SET NULL,
ADD COLUMN background_file_id UUID REFERENCES storage.files(id) ON DELETE SET NULL,
ADD COLUMN signature_file_id UUID REFERENCES storage.files(id) ON DELETE SET NULL;
```
`storage.files` is already a `BlobReferenceSource`, already covered by
every consistency edge, already GC-integrated, already copy- and
version-aware. A pointer to a *file* therefore adds **zero new
reference sources and zero new consistency edges** — `auth.users`
holds no blob reference at all, only a pointer to a row that does.
Deleting the file decrements the blob refcount through the existing
file-deletion path.
Point at `id`, never at a name or path: a rename must not break a
profile.
For a small fixed set, **columns beat a table.** A table becomes right
only when the object set is open-ended, and it would cost exactly what
the pointer avoids.
### Rejected alternatives
| Option | Why not |
|---|---|
| **New table → blobs/chunks** | Needs a new `BlobReferenceSource`, a new fragment in the manifest sweep, and a new dangling check. Precisely the complexity the pointer removes. |
| **Direct backend paths** (`profile/{uuid}/avatar.png`) | `backend_migration` enumerates *blobs*, so a Local→S3 cutover **silently drops every avatar**. `EncryptedBlobBackend` is hash-keyed, so writes either bypass encryption or need a parallel path. `backend_consistency` raises `unknown_backend_file` (severity `anomaly`, "non-canonical file in blob namespace") per object on every sweep. And a fixed key overwritten in place breaks the immutability everything else rests on, killing `Cache-Control: immutable`. |
| **Reserved folder in the user's own drive** (`.profile/`, `.oxiprofile/`) | The user can write to it, so there are **two write paths and only one is validated** — the upload endpoint's format/size checks are bypassable over WebDAV and sync. It is in the sync tree, so "hidden" is not hidden in the protocols that matter. Existence-by-path drags back the extension problem. And any reserved name squats a namespace users own — `.profile` is a POSIX shell file, so it collides by default for anyone syncing a Linux home. |
| **One drive per user** | Per-user creation at signup, backfill for existing users, per-user quota exemption, and a cascade on account deletion. All avoidable — see below. |
Underneath all of it: these objects are owned by the **application on
the user's behalf**, not by the user as documents. Putting them in a
document tree conflates the two, and every problem above follows.
## The hidden system drive
**One shared drive, not one per user.** `kind = 'system'`, alongside
today's `CHECK (kind IN ('personal', 'shared'))`. Files inside are
owned by their respective users via the normal `created_by` /
ownership columns; the drive is only a container.
Sharing one drive drops per-user creation at signup, backfill for
existing users, and per-user quota exemption. Deleting a user becomes
a query over `storage.files` rather than a drive cascade.
Properties it needs:
- **Hidden at drive enumeration.** This is the single filter point,
and it is why the drive beats a folder: a folder must be filtered in
directory listings, search results, recent items, trash, photo
indexing and sync deltas, whereas a drive is filtered once where
drives are listed. Every surface must honour it — REST, WebDAV,
NextCloud, search, quota reporting. **A missed filter is the
characteristic bug of this design**, so it deserves a test per
surface rather than per call site.
- **Trash disabled.** Otherwise every replaced avatar lands in a trash
nobody can see, holding a blob reference that GC cannot reclaim
while retention keeps it alive — invisible storage growth with no
signal. Deletion here is immediate.
- **Exempt from the user quota envelope.** Nobody should pay quota for
their own avatar.
- **Created at install, fail-fast at boot.** If the drive is missing,
panic rather than silently disabling profile objects — a silently
absent avatar surface is worse than a refusal to start.
- **Visible to admins.** Ops need to see it for storage accounting
even though it is hidden from users.
## Visibility — per kind, in code
Reads go through a service method carrying an explicit policy, audited
like any other authorization decision. Because the column set is fixed
and small, the policy is a `match`, not stored data — there is nothing
to misconfigure, and adding a column forces adding an arm:
| Object | Who may read | Why |
|---|---|---|
| `avatar` | **the same rule as profile visibility** | Not "any authenticated user". `AGENTS.md` has `user_profile.rejected` return **404, never 403**, for an external caller with no relationship, specifically so existence cannot be confirmed. An avatar endpoint answering 200 for any caller is an oracle around that control. |
| `background` | owner only | Nobody else has a reason to fetch it. |
| `signature` | owner only, plus the document render path | A handwritten signature is forgery material. It is "public" only in the sense that it appears on documents you may already read — which argues for rendering it into those documents, not exposing it as a directly-readable object. |
Note the consequence: the drive's own permission model is **not** what
governs these reads. The object lives in a drive and is read through a
different door. That is a deliberate choice, not an oversight — record
it so nobody later "fixes" it by granting cross-user drive access.
## What must NOT live here
> **If losing control of it is a security incident rather than a
> cosmetic bug, it stays in the database.** The system drive is for
> user-facing binaries.
So E2E/Vault key material — public key bundle, passphrase-wrapped
private key, recovery kit — stays in `auth.users` columns. Four
reasons, the last decisive:
1. **Failure-mode asymmetry.** The characteristic bug here is a missed
listing filter. For a wallpaper that is cosmetic; for key material
it is disclosure.
2. **Atomicity.** Rotating a passphrase rewraps the private key
*together with* the credential change. A DB column makes that one
transaction; a file write plus a column update cannot be atomic.
3. **Size.** A few KB — blob storage buys nothing.
4. **`EncryptedBlobBackend` encrypts under a key the server holds.**
For E2E material the whole premise is that the server *cannot*
decrypt. Routing a wrapped private key through the blob layer
encrypts it twice, once under a key the operator controls, adding
no protection while creating the impression of it.
Users who want to store genuinely private *files* already have the
personal drive, with the full AuthZ engine behind it. There is no gap.
## Migrating the avatar off `auth.users.image`
Volume is one row per user, so unlike the thumbnail migration this
needs **no read-through phase** — a single batch job is enough.
**Phase 1.** Add the pointer columns and the system drive. Write path
switches to files; read path prefers `avatar_file_id` and falls back
to `image` when null.
**Phase 2.** `profile_image_import`, a registered `JobRegistry` job
(subject-first naming, per convention). For each user with a non-null
`image`:
1. Decode the data URI; skip and log if it does not parse, rather than
failing the batch.
2. `store_from_stream` the decoded bytes → derived blob + manifest.
3. Insert a `storage.files` row in the system drive, owned by that
user.
4. Set `avatar_file_id`.
Idempotent (`WHERE avatar_file_id IS NULL`), resumable via a user-id
cursor, and reports imported / skipped-unparseable / failed counts.
**Phase 3.** Drop `auth.users.image` and the fallback, gated on the
job reporting zero remaining. Dropping the column is what actually
reclaims the TOAST weight and retires the narrow-projection rule in
`get_users_by_ids`.
**IdP-sourced avatars.** OIDC login already refreshes the avatar
("same IdP avatar, already verified" — `user_pg_repository.rs:1435`).
That path must be converted at Phase 1, not Phase 3, or it keeps
writing to a column the migration is draining.
## Object catalogue
**Now:** avatar, UI background, signature image.
**Strong future candidates** — these are what justify a drive rather
than three columns and a corner:
| Object | Why it fits |
|---|---|
| **Data exports** (GDPR takeout, drive-export zip) | Generated async, large, downloadable, should expire. Today there is nowhere to put them. |
| **Staged imports** (Google Takeout, NextCloud export) | Multi-step ingestion needs durability beyond a temp file. |
| **Share-page branding / logo** | Per-user or per-org, served on public share pages. |
**Same problem, different owner — this drive does not help:**
`carddav.contacts.photo_url TEXT` (contact photos, today a URL or an
inlined data URI) and CalDAV `ATTACH` event attachments. They are keyed
by contact and by event, not by user. But the *pointer* generalises:
`contacts.photo_file_id UUID REFERENCES storage.files(id)` solves them
with no new blob-referencing table either — they simply live in the
address book's or calendar's own drive rather than here. Own plan.
## Operational details
- **Replace must delete.** Write new file → update pointer →
hard-delete the previous file. `ON DELETE SET NULL` protects the
pointer when a file vanishes, but nothing deletes the old file
because the pointer moved.
- **Validation lives at the endpoint** and is now the only write path,
which is the point of not using a user-writable location. Enforce
format, dimensions and size there.
- **Account deletion** deletes the user's files in the system drive
explicitly; the pointer columns are on the row being deleted anyway.
## Non-goals
- **Secrets of any kind.** See the discriminator above.
- **Per-user system drives.** One shared drive; revisit only if
per-user quota or trash semantics ever become necessary.
- **Contact photos and event attachments.** Same pointer pattern,
different owner, different drive — separate plan.
- **A generic "user objects" API.** The column set is fixed and small
on purpose. Reach for a table only when it demonstrably is not.
## Open questions
- Who owns the system drive row itself, and what does
`drives_consistency` expect of a drive with no human owner?
- Does the signature object survive the "owner only" rule, or does
document rendering need a broader read path than expected?
- Should exports live here or in a short-lived namespace with its own
expiry, given they are the only candidate with a natural TTL?
## References
- `docs/plan/derived-blobs.md` — the sibling plan; shares the
point-at-a-file rule and documents the consistency coverage matrix
these objects inherit for free.
- `migrations/20260526000000_add_user_image.sql` — the column being
retired.
- `migrations/20260802100000_drives_schema_additive.sql` — the
`kind IN ('personal','shared')` constraint this extends.
- `src/AGENTS.md` — the backend-abstraction rules, and the
anti-enumeration pattern the avatar visibility rule follows.
- Memory `project_drive_naming_and_vault_reservation` — "Vault"
reserved for the future E2E kind whose key material this plan
explicitly keeps out of the drive.
+17 -6
View File
@@ -60,11 +60,21 @@ export function listJobs(): Promise<JobSummary[]> {
}
/**
* `POST /api/admin/jobs/{name}/trigger?force=X&deep=X` — dispatch a job
* on-demand. `force` bypasses per-tenant idempotency checks (e.g.
* `trash_cleanup` skipping when nothing is due). `deep` opts into slow
* variants (currently only `storage_consistency`, propagated by
* `consistency_batch` to every child).
* `POST /api/admin/jobs/{name}/trigger?force=X&deep=X&repair=X` —
* dispatch a job on-demand.
*
* - `force` bypasses per-tenant idempotency checks (e.g. `trash_cleanup`
* skipping when nothing is due).
* - `deep` opts into slow variants (currently only `storage_consistency`,
* propagated by `consistency_batch` to every child).
* - `repair` opts into corrective action on the refcount consistency
* tenants (`blobs_consistency`, `manifests_consistency`, and
* `consistency_batch` which fans out to both). Content-safe: only the
* stored counter changes to match the auditor's computed value. Race-
* safe: the corrective UPDATE recomputes the auditor formula in the
* same statement, so a concurrent write can't leave a stale value.
* Default `false` preserves discovery-only behaviour — surface a
* confirm-first flow when calling with `repair: true`.
*
* Throws on 4xx / 5xx with the backend's error message when present.
* A 404 means the job name isn't registered — surface that specifically
@@ -72,11 +82,12 @@ export function listJobs(): Promise<JobSummary[]> {
*/
export async function triggerJob(
name: string,
opts: { force?: boolean; deep?: boolean; storage?: string } = {}
opts: { force?: boolean; deep?: boolean; storage?: string; repair?: boolean } = {}
): Promise<TriggerResponse> {
const params = new URLSearchParams();
if (opts.force) params.set('force', 'true');
if (opts.deep) params.set('deep', 'true');
if (opts.repair) params.set('repair', 'true');
// `storage` scopes tenants that respect JobRunArgs.storage —
// currently blobs_consistency / backend_consistency (probes the
// named entry instead of the live backend). See
+297 -18
View File
@@ -21,6 +21,7 @@
import { SvelteSet } from 'svelte/reactivity';
import Icon from '$lib/icons/Icon.svelte';
import Modal from '$lib/components/Modal.svelte';
import { confirmDialog } from '$lib/stores/dialogs.svelte';
import { t } from '$lib/i18n/index.svelte';
import { errorMessage } from '$lib/utils/errors';
import { ui } from '$lib/stores/ui.svelte';
@@ -66,6 +67,43 @@
else busyKeys.delete(key);
}
// Per-job "Run" split-button menu state. Keyed by job name so
// two rows can open their menus independently (though the
// outside-click handler below closes all on any click outside
// any menu — matching the /files upload dropdown pattern). Only
// rows with `supportsDeep` OR `supportsRepair` render a chevron;
// the plain-Run rows (drives/folders/files/backend/… consistency,
// trash_cleanup, dedup_gc, …) show a bare "Run" button with no
// menu, keeping the common case one-click.
let runMenuOpen = $state<Record<string, boolean>>({});
function toggleRunMenu(name: string) {
runMenuOpen = { ...runMenuOpen, [name]: !runMenuOpen[name] };
}
function closeAllRunMenus() {
runMenuOpen = {};
}
// Global outside-click + Escape dismiss. Only registered while at
// least one menu is open — a background admin tab doesn't hold
// listeners.
$effect(() => {
const anyOpen = Object.values(runMenuOpen).some((v) => v);
if (!anyOpen) return;
const onDown = (e: MouseEvent) => {
if (!(e.target as HTMLElement).closest('.jobs-panel__split')) {
closeAllRunMenus();
}
};
const onKey = (e: KeyboardEvent) => {
if (e.key === 'Escape') closeAllRunMenus();
};
window.addEventListener('pointerdown', onDown);
window.addEventListener('keydown', onKey);
return () => {
window.removeEventListener('pointerdown', onDown);
window.removeEventListener('keydown', onKey);
};
});
// Purge-modal state. Null = closed; otherwise carries the
// draft retention days the operator's picking. Kept separate
// from the top-bar action state so mouse-away doesn't lose
@@ -234,8 +272,10 @@
// ─── Actions ───────────────────────────────────────────────────────
async function onTrigger(name: string, opts: { deep?: boolean } = {}) {
const key = `trigger:${name}${opts.deep ? ':deep' : ''}`;
async function onTrigger(name: string, opts: { deep?: boolean; repair?: boolean } = {}) {
// Key suffix has to keep every dispatched variant distinct so the
// button-disabled state of one doesn't lock out another mid-flight.
const key = `trigger:${name}${opts.deep ? ':deep' : ''}${opts.repair ? ':repair' : ''}`;
markBusy(key, true);
try {
// Fire the trigger + a follow-up loadJobs after a short delay
@@ -265,10 +305,49 @@
if (!res.outcome) {
// dispatched (detached) — no outcome to render
} else if (res.outcome.outcome === 'ok') {
ui.notify(
t('admin.jobs.triggered_ok', { name }, '{{name}} triggered successfully'),
'success'
);
// Repair runs surface a rollup so the operator sees
// whether corrective UPDATEs actually fired. `extra`
// carries `repaired_count` on the two refcount tenants
// directly, and nested under `per_check[*].extra` when
// dispatched via `consistency_batch`. Sum across the
// per_check dict if present, else read the top-level.
let repairedTotal = 0;
let sawRepair = false;
const extra = (res.outcome.extra ?? {}) as {
repair_requested?: boolean;
repaired_count?: number;
per_check?: Record<
string,
{ extra?: { repair_requested?: boolean; repaired_count?: number } }
>;
};
if (extra.repair_requested) {
sawRepair = true;
repairedTotal += extra.repaired_count ?? 0;
}
if (extra.per_check) {
for (const child of Object.values(extra.per_check)) {
if (child?.extra?.repair_requested) {
sawRepair = true;
repairedTotal += child.extra.repaired_count ?? 0;
}
}
}
if (sawRepair) {
ui.notify(
t(
'admin.jobs.triggered_ok_repair',
{ name, n: repairedTotal },
'{{name}}: {{n}} counter(s) repaired'
),
'success'
);
} else {
ui.notify(
t('admin.jobs.triggered_ok', { name }, '{{name}} triggered successfully'),
'success'
);
}
} else {
ui.notify(
t(
@@ -557,6 +636,31 @@
return name === 'consistency_batch' || name === 'blobs_consistency';
}
// Jobs whose handler consults `args.repair` and applies a
// corrective UPDATE against the finding it just emitted. Only the
// two ref_count tenants today; `consistency_batch` also accepts
// the flag (fans out to both) and is surfaced separately as the
// top-bar "Repair ref_counts" button. Keep this list narrow —
// adding a job here without a matching backend handler produces a
// silently no-op button that confuses operators.
function supportsRepair(name: string): boolean {
return name === 'blobs_consistency' || name === 'manifests_consistency';
}
async function onTriggerWithRepairConfirm(name: string) {
const ok = await confirmDialog({
title: t('admin.jobs.run_repair_confirm_title', 'Repair drifted ref_counts?'),
message: t(
'admin.jobs.run_repair_confirm_body_scoped',
{ name },
'Runs {{name}} and applies a corrective UPDATE to any counter that disagrees with its live reference count. Content-safe: only counters change; blob content and file rows are untouched.'
),
confirmText: t('admin.jobs.run_repair_confirm', 'Repair'),
danger: true
});
if (ok) await onTrigger(name, { repair: true });
}
function isRunning(job: JobSummary): boolean {
return job.running;
}
@@ -614,6 +718,39 @@
<Icon name="play" />
{t('admin.jobs.run_deep', 'Run deep')}
</button>
<!-- Repair goes behind a confirm because it issues corrective
UPDATEs on `storage.blobs.ref_count` and
`storage.chunk_manifests.ref_count`. Content-safe (only
counters change, matching the auditor's computed truth)
and race-safe (each UPDATE recomputes inside the same
statement), but writing-a-lot is still writing-a-lot.
One click, one confirm, one batch dispatched to both
refcount tenants via consistency_batch's arg
propagation. See `?repair=true` on
`POST /api/admin/jobs/{name}/trigger`. -->
<button
class="jobs-panel__btn jobs-panel__btn--warn"
disabled={busyKeys.has('trigger:consistency_batch:repair')}
title={t(
'admin.jobs.run_repair_hint',
'Corrects any drifted ref_counts (blobs + manifests) found by the audit. Content-safe — only counters change, not data.'
)}
onclick={async () => {
const ok = await confirmDialog({
title: t('admin.jobs.run_repair_confirm_title', 'Repair drifted ref_counts?'),
message: t(
'admin.jobs.run_repair_confirm_body',
'Runs the audit against every blob + manifest and applies a corrective UPDATE to any counter that disagrees with its live reference count. Content-safe: only the counter changes; blob content and file rows are untouched. Safe to run at any time; a discovery-only run happens first so you can see the drift before this repair overwrites it.'
),
confirmText: t('admin.jobs.run_repair_confirm', 'Repair'),
danger: true
});
if (ok) await onTrigger('consistency_batch', { repair: true });
}}
>
<Icon name="cog" />
{t('admin.jobs.run_repair', 'Repair ref_counts')}
</button>
{/if}
<!-- Purge is orthogonal to consistency — it works even
when the batch coordinator isn't registered, so it
@@ -754,22 +891,80 @@
{t('admin.jobs.cancel', 'Cancel')}
</button>
{:else}
<button
class="jobs-panel__btn jobs-panel__btn--small"
disabled={busyKeys.has(`trigger:${job.name}`)}
onclick={() => onTrigger(job.name)}
>
{t('admin.jobs.run', 'Run')}
</button>
{#if supportsDeep(job.name)}
<!-- Split-button: primary "Run" fires the default
trigger; the chevron opens a menu with the
tenant-specific variants (Run deep / Repair).
Rows without any variant render a bare Run
button — no chevron, no menu, no extra
width. Preserves one-click discovery for
the common case. -->
{@const hasRunVariants = supportsDeep(job.name) || supportsRepair(job.name)}
<span class="jobs-panel__split">
<button
class="jobs-panel__btn jobs-panel__btn--small"
disabled={busyKeys.has(`trigger:${job.name}:deep`)}
onclick={() => onTrigger(job.name, { deep: true })}
class:jobs-panel__split-main={hasRunVariants}
disabled={busyKeys.has(`trigger:${job.name}`)}
onclick={() => {
closeAllRunMenus();
void onTrigger(job.name);
}}
>
{t('admin.jobs.run_deep', 'Run deep')}
{t('admin.jobs.run', 'Run')}
</button>
{/if}
{#if hasRunVariants}
<button
class="jobs-panel__btn jobs-panel__btn--small jobs-panel__split-toggle"
aria-haspopup="menu"
aria-expanded={runMenuOpen[job.name] ?? false}
aria-label={t('admin.jobs.run_variants_menu', 'Run variants menu')}
onclick={() => toggleRunMenu(job.name)}
>
<Icon name="caret-down" />
</button>
{#if runMenuOpen[job.name]}
<div class="jobs-panel__run-menu" role="menu">
{#if supportsDeep(job.name)}
<button
type="button"
class="jobs-panel__run-menu-item"
role="menuitem"
disabled={busyKeys.has(`trigger:${job.name}:deep`)}
title={t(
'admin.jobs.run_deep_hint',
'Also runs slow variants (blob re-hash, bitrot detection).'
)}
onclick={() => {
closeAllRunMenus();
void onTrigger(job.name, { deep: true });
}}
>
<Icon name="search" />
<span>{t('admin.jobs.run_deep', 'Run deep')}</span>
</button>
{/if}
{#if supportsRepair(job.name)}
<button
type="button"
class="jobs-panel__run-menu-item jobs-panel__run-menu-item--warn"
role="menuitem"
disabled={busyKeys.has(`trigger:${job.name}:repair`)}
title={t(
'admin.jobs.run_repair_hint',
'Corrects any drifted ref_counts (blobs + manifests) found by the audit. Content-safe — only counters change, not data.'
)}
onclick={() => {
closeAllRunMenus();
void onTriggerWithRepairConfirm(job.name);
}}
>
<Icon name="cog" />
<span>{t('admin.jobs.run_repair', 'Repair')}</span>
</button>
{/if}
</div>
{/if}
{/if}
</span>
{/if}
{#if isRunning(job) && canExpand}
{#if isRecoverable(job)}
@@ -1304,6 +1499,90 @@
color: var(--color-danger-text-alt);
}
/* Warn variant — used for actions that mutate data but are content-
safe / reversible-in-outcome (e.g. Repair ref_counts). Signals
"read the tooltip and the confirm before clicking" without the
danger red reserved for destructive delete-style buttons. */
.jobs-panel__btn--warn {
border-color: var(--color-warning-border);
color: var(--color-warning-text);
}
/* Split-button — inline flex holding a primary "Run" (fires default
action) and a chevron (opens the variants menu). `position:
relative` anchors the menu below the toggle. Only rendered on
rows whose job supports at least one variant; plain-Run rows
sidestep this whole structure. */
.jobs-panel__split {
display: inline-flex;
position: relative;
}
/* Attached-button trick: main loses its right border-radius, toggle
loses its left. Toggle also loses its left border so the two
don't render a double-thick divider. */
.jobs-panel__split-main {
border-top-right-radius: 0;
border-bottom-right-radius: 0;
}
.jobs-panel__split-toggle {
border-top-left-radius: 0;
border-bottom-left-radius: 0;
border-left: none;
padding-left: 0.35rem;
padding-right: 0.35rem;
}
/* The variants menu — dropdown below the toggle, right-aligned so
it doesn't overflow the Actions column edge into the next row's
badge cell. Shadow + surface bg mirror the /files upload
dropdown (`upload-dropdown-menu`); using local CSS here rather
than the ported class so the jobs-panel keeps its scoped styling. */
.jobs-panel__run-menu {
position: absolute;
top: calc(100% + 2px);
right: 0;
z-index: 30;
min-width: 10rem;
background: var(--color-bg-surface);
border: 1px solid var(--color-border);
border-radius: var(--radius-md, 6px);
box-shadow: var(--shadow-md);
padding: 0.25rem 0;
}
.jobs-panel__run-menu-item {
display: flex;
align-items: center;
gap: 0.5rem;
width: 100%;
padding: 0.4rem 0.75rem;
background: transparent;
border: none;
text-align: left;
font: inherit;
color: var(--color-text);
cursor: pointer;
white-space: nowrap;
}
.jobs-panel__run-menu-item:hover:not(:disabled) {
background: var(--color-bg-hover);
}
.jobs-panel__run-menu-item:disabled {
opacity: 0.5;
cursor: not-allowed;
}
/* Warn colour on the menu item mirrors the button variant so the
Repair option carries the same "attention-worthy but not
destructive" visual weight as its top-bar counterpart. */
.jobs-panel__run-menu-item--warn {
color: var(--color-warning-text);
}
.jobs-panel__pill {
display: inline-block;
padding: 0.1rem 0.5rem;
+8
View File
@@ -1275,6 +1275,14 @@
"run_all_consistency": "Run all consistency checks",
"run_deep": "Run deep",
"run_deep_hint": "Also runs slow variants (blob re-hash, bitrot detection).",
"run_repair": "Repair ref_counts",
"run_repair_hint": "Corrects any drifted ref_counts (blobs + manifests) found by the audit. Content-safe — only counters change, not data.",
"run_repair_confirm_title": "Repair drifted ref_counts?",
"run_repair_confirm_body": "Runs the audit against every blob + manifest and applies a corrective UPDATE to any counter that disagrees with its live reference count. Content-safe: only the counter changes; blob content and file rows are untouched. Safe to run at any time; a discovery-only run happens first so you can see the drift before this repair overwrites it.",
"run_repair_confirm_body_scoped": "Runs {{name}} and applies a corrective UPDATE to any counter that disagrees with its live reference count. Content-safe: only counters change; blob content and file rows are untouched.",
"run_variants_menu": "Run variants menu",
"run_repair_confirm": "Repair",
"triggered_ok_repair": "{{name}}: {{n}} counter(s) repaired",
"col_name": "Name",
"col_cadence": "Cadence",
"col_last_run": "Last run",
+8
View File
@@ -1197,6 +1197,14 @@
"run_all_consistency": "Exécuter tous les contrôles de cohérence",
"run_deep": "Analyse approfondie",
"run_deep_hint": "Exécute également des variantes lentes (re-hachage de blob, détection bitrot).",
"run_repair": "Réparer les compteurs",
"run_repair_hint": "Corrige les compteurs de références (blobs + manifestes) désynchronisés détectés par l'audit. Sûr pour les données — seuls les compteurs changent, pas le contenu.",
"run_repair_confirm_title": "Réparer les compteurs de références ?",
"run_repair_confirm_body": "Lance l'audit sur chaque blob et manifeste, puis applique un UPDATE correctif à chaque compteur qui ne correspond pas au nombre réel de références. Sans risque pour les données : seuls les compteurs changent ; le contenu des blobs et les enregistrements de fichiers ne sont pas touchés. Vous pouvez exécuter cela à tout moment ; un passage en lecture seule s'exécute d'abord pour visualiser l'écart avant que la réparation ne l'écrase.",
"run_repair_confirm_body_scoped": "Lance {{name}} et applique un UPDATE correctif à chaque compteur qui ne correspond pas au nombre réel de références. Sans risque pour les données : seuls les compteurs changent ; le contenu et les enregistrements de fichiers ne sont pas touchés.",
"run_variants_menu": "Menu des variantes d'exécution",
"run_repair_confirm": "Réparer",
"triggered_ok_repair": "{{name}} : {{n}} compteur(s) réparé(s)",
"col_name": "Nom",
"col_cadence": "Cadence",
"col_last_run": "Dernière exécution",
@@ -0,0 +1,211 @@
-- Fix: `storage.copy_folder_tree` never incremented `chunk_manifests.ref_count`.
--
-- The function bumped only `storage.blobs`:
--
-- UPDATE storage.blobs b SET ref_count = ref_count + hc.cnt
-- FROM (...) hc WHERE b.hash = hc.blob_hash;
--
-- but a CDC file's `blob_hash` names a MANIFEST, not a chunk. For any
-- multi-chunk file that predicate matches nothing, so a folder copy took
-- NO reference. Delete the original afterwards and `remove_reference`
-- walks the manifest to 0, `dedup_gc` reaps the manifest and every chunk
-- behind it — and the copy is unreadable. Silent data loss on an ordinary
-- UI operation.
--
-- Single-chunk files escaped by accident: their whole-file hash equals
-- their lone chunk's hash, so the UPDATE did match — bumping the wrong
-- counter, which shows up as a manifest under-count plus a blob
-- over-count rather than as loss.
--
-- Reproduced on a 5 MiB / 18-chunk file copied through the UI:
-- `chunk_manifests.ref_count` stayed at 1 while two `storage.files` rows
-- referenced it; `manifests_consistency` reported
-- `manifest_refcount_mismatch` with `delta: 1, reap_risk: true`.
--
-- This migration only rewrites the reference-counting block; everything
-- else is `20260902000001_copy_folder_tree_drop_user_id.sql` verbatim.
--
-- NOTE: existing drift is NOT repaired here. Run `manifests_consistency`
-- to find it — a data fix belongs with the recovery framework, not in a
-- schema migration that cannot know which counter is authoritative.
CREATE OR REPLACE FUNCTION storage.copy_folder_tree(
p_source_id UUID,
p_target_parent_id UUID, -- NULL = copy to root (keeps source drive)
p_dest_name TEXT DEFAULT NULL -- NULL = keep source folder name
) RETURNS TABLE(new_root_id TEXT, folders_copied BIGINT, files_copied BIGINT) AS $$
DECLARE
v_root_lpath ltree;
v_root_depth INT;
v_max_depth INT;
v_level INT;
v_folders BIGINT := 0;
v_files BIGINT := 0;
v_inserted BIGINT;
v_new_root UUID;
v_dest_drive_id UUID;
BEGIN
-- Validate source exists.
SELECT fo.lpath, nlevel(fo.lpath)
INTO v_root_lpath, v_root_depth
FROM storage.folders fo
WHERE fo.id = p_source_id AND NOT fo.is_trashed;
IF v_root_lpath IS NULL THEN
RAISE EXCEPTION 'Source folder not found: %', p_source_id
USING ERRCODE = 'P0002'; -- no_data_found
END IF;
-- Resolve destination drive_id once up front (cross-drive copy path).
IF p_target_parent_id IS NULL THEN
SELECT fo.drive_id INTO v_dest_drive_id
FROM storage.folders fo
WHERE fo.id = p_source_id;
ELSE
SELECT fo.drive_id INTO v_dest_drive_id
FROM storage.folders fo
WHERE fo.id = p_target_parent_id AND NOT fo.is_trashed;
IF v_dest_drive_id IS NULL THEN
RAISE EXCEPTION 'Target parent folder not found: %', p_target_parent_id
USING ERRCODE = 'P0002';
END IF;
END IF;
-- Temp mapping: every folder in the subtree → new UUID.
CREATE TEMP TABLE IF NOT EXISTS _copy_map(
old_id UUID PRIMARY KEY,
new_id UUID NOT NULL DEFAULT gen_random_uuid()
) ON COMMIT DROP;
TRUNCATE _copy_map;
INSERT INTO _copy_map(old_id)
SELECT fo.id
FROM storage.folders fo
WHERE NOT fo.is_trashed
AND fo.lpath <@ v_root_lpath;
SELECT cm.new_id INTO v_new_root
FROM _copy_map cm WHERE cm.old_id = p_source_id;
SELECT MAX(nlevel(fo.lpath))
INTO v_max_depth
FROM storage.folders fo
JOIN _copy_map cm ON fo.id = cm.old_id;
-- ── Insert folders level by level ──
-- Post-D7: `user_id` intentionally omitted from the column list so
-- copied rows leave the (now-nullable) column NULL. Provenance is
-- carried by `created_by` / `updated_by` (§14 columns) — preserved
-- from source so authorship survives the copy.
FOR v_level IN v_root_depth .. v_max_depth LOOP
INSERT INTO storage.folders(
id, name, parent_id,
drive_id, created_by, updated_by
)
SELECT cm.new_id,
CASE WHEN fo.id = p_source_id AND p_dest_name IS NOT NULL
THEN p_dest_name ELSE fo.name END,
CASE WHEN fo.id = p_source_id THEN p_target_parent_id
ELSE pm.new_id END,
v_dest_drive_id,
fo.created_by,
fo.updated_by
FROM storage.folders fo
JOIN _copy_map cm ON fo.id = cm.old_id
LEFT JOIN _copy_map pm ON fo.parent_id = pm.old_id
WHERE NOT fo.is_trashed
AND nlevel(fo.lpath) = v_level;
GET DIAGNOSTICS v_inserted = ROW_COUNT;
v_folders := v_folders + v_inserted;
END LOOP;
-- Temp mapping for files src→dst (dst ids pre-allocated so we can
-- reference them in the dead-property duplication below).
CREATE TEMP TABLE IF NOT EXISTS _copy_file_map(
old_id UUID PRIMARY KEY,
new_id UUID NOT NULL DEFAULT gen_random_uuid()
) ON COMMIT DROP;
TRUNCATE _copy_file_map;
INSERT INTO _copy_file_map(old_id)
SELECT f.id
FROM storage.files f
JOIN _copy_map cm ON f.folder_id = cm.old_id
WHERE NOT f.is_trashed;
-- ── Batch copy all files (zero-copy: same blob_hash) ──
-- Post-D7: `user_id` omitted. Provenance via `created_by`/`updated_by`.
INSERT INTO storage.files(
id, name, folder_id, blob_hash, size, mime_type,
media_sort_date, drive_id, created_by, updated_by
)
SELECT fm.new_id, f.name, cm.new_id, f.blob_hash, f.size,
f.mime_type, f.media_sort_date, v_dest_drive_id, f.created_by,
f.updated_by
FROM storage.files f
JOIN _copy_map cm ON f.folder_id = cm.old_id
JOIN _copy_file_map fm ON fm.old_id = f.id
WHERE NOT f.is_trashed;
GET DIAGNOSTICS v_files = ROW_COUNT;
-- Batch increment reference counts — MANIFEST FIRST, blobs only as
-- fallback. This mirrors `DedupService::add_reference`, and the order
-- is the whole point:
--
-- A CDC file's `blob_hash` names a MANIFEST (`chunk_manifests.file_hash`),
-- not a chunk. The previous version of this block updated only
-- `storage.blobs`, so for a multi-chunk file the predicate
-- `b.hash = hc.blob_hash` matched ZERO rows and the copy took no
-- reference at all. Deleting the original then walked the manifest's
-- ref_count to 0, dedup_gc reaped the manifest and every chunk behind
-- it, and the copy became unreadable. Reproduced via the UI folder
-- copy on a 5 MiB (18-chunk) file: ref_count stayed 1 with two files
-- referencing it.
--
-- The `NOT EXISTS (bumped)` guard on the blobs branch is load-bearing.
-- For a SINGLE-chunk file the whole-file hash equals its lone chunk's
-- hash, so without it the copy would be counted at both levels and
-- turn an under-count into an over-count.
IF v_files > 0 THEN
WITH hc AS (
SELECT f.blob_hash, COUNT(*)::int AS cnt
FROM storage.files f
JOIN _copy_map cm ON f.folder_id = cm.new_id
WHERE NOT f.is_trashed
GROUP BY f.blob_hash
),
bumped AS (
UPDATE storage.chunk_manifests m
SET ref_count = m.ref_count + hc.cnt
FROM hc
WHERE m.file_hash = hc.blob_hash
RETURNING m.file_hash
)
UPDATE storage.blobs b
SET ref_count = b.ref_count + hc.cnt,
-- Matches add_reference: a blob resurrected inside its GC
-- grace window must lose its orphan stamp.
orphaned_at = NULL
FROM hc
WHERE b.hash = hc.blob_hash
AND NOT EXISTS (SELECT 1 FROM bumped WHERE file_hash = hc.blob_hash);
END IF;
-- Duplicate dead properties per RFC 4918 §8.8 — id-keyed store.
INSERT INTO storage.webdav_dead_properties
(folder_id, namespace, local_name, value)
SELECT cm.new_id, dp.namespace, dp.local_name, dp.value
FROM storage.webdav_dead_properties dp
JOIN _copy_map cm ON dp.folder_id = cm.old_id;
INSERT INTO storage.webdav_dead_properties
(file_id, namespace, local_name, value)
SELECT fm.new_id, dp.namespace, dp.local_name, dp.value
FROM storage.webdav_dead_properties dp
JOIN _copy_file_map fm ON dp.file_id = fm.old_id;
RETURN QUERY SELECT v_new_root::text, v_folders, v_files;
END;
$$ LANGUAGE plpgsql;
@@ -0,0 +1,108 @@
-- Fix: `trg_files_decrement_blob_ref` decremented the wrong counter for
-- CDC files.
--
-- The original trigger (2026-03-07 initial schema) unconditionally ran:
--
-- UPDATE storage.blobs
-- SET ref_count = GREATEST(ref_count - 1, 0)
-- WHERE hash = OLD.blob_hash;
--
-- That's correct for a legacy whole-file blob, where `OLD.blob_hash`
-- names a `storage.blobs` row directly. For a CDC file, `OLD.blob_hash`
-- names a `storage.chunk_manifests.file_hash` — the blob table row (if
-- one exists at all) holds a DIFFERENT counter, incremented by the
-- MANIFEST's presence in its own `chunk_hashes[]`, not by the file.
--
-- Consequences before this fix:
-- 1. `storage.chunk_manifests.ref_count` never decremented on file
-- DELETE → over-count grows unboundedly across delete/purge
-- cycles.
-- 2. `storage.blobs.ref_count` decremented for hashes it shouldn't
-- (CDC whole-file hashes) → the counter drops toward 0 while the
-- manifest still legitimately references the chunk. GC then reaps
-- a live blob → downloadable-then-404 data loss.
--
-- Both bugs surfaced by `tests/api/refcount_cascade.hurl` on the
-- 135-byte fixture (single-chunk CDC file, worst case for confusion
-- because the whole-file hash equals its lone chunk's hash). The
-- 2026-08-22 sandbox drift (`storage.blobs.ref_count = 0`,
-- `actual_auditor = 1`) is the same bug at rest.
--
-- Sibling fix: `20261016000000_copy_folder_tree_manifest_refcount.sql`
-- fixed the mirror-image INCREMENT bug in `storage.copy_folder_tree`.
-- This migration closes the decrement half.
--
-- Cross-references:
-- - `DedupService::add_reference` (dedup_service.rs:1703) — app-layer
-- twin for the increment direction: manifest first, blob fallback.
-- - `manifests_consistency` tenant (2026-08-23) — surfaces any
-- residual drift after this fix lands.
--
-- ── DESIGN NOTE — decrement only, no manifest reap here ──
--
-- The trigger DELIBERATELY does not delete manifests or walk chunks on
-- a last-ref decrement. Both actions used to live inside
-- `DedupService::cleanup_if_orphaned` and its callee
-- `remove_manifest_reference`, and both fire `fire_blob_hooks` —
-- the Rust callback that reaps disk artefacts keyed by the whole-file
-- content hash (thumbnails, face embeddings, audio tags, media
-- metadata). SQL triggers can't invoke Rust callbacks, so if this
-- trigger reaped the manifest itself, dedup_gc Phase 1
-- (`dedup_service.rs:2660-2772`) — the ONLY code path that knows to
-- fire `fire_blob_hooks` for a reaped manifest's `file_hash` — would
-- find nothing to do on its next sweep, and every derived artefact
-- would leak on disk. `storage_cleanup_check.sh`'s "N thumbnail
-- file(s) remain on disk" gate catches this class immediately.
--
-- Contract: trigger decrements the correct counter atomically inside
-- the DELETE txn. GC (`dedup_gc`) is responsible for:
-- • finding manifests whose ref_count hit 0 (or that no reference
-- source references, covering bulk-delete paths),
-- • deleting them,
-- • decrementing each chunk in `chunk_hashes[]`,
-- • firing `fire_blob_hooks(file_hash)` so Rust callbacks reap
-- derived disk artefacts,
-- • the corresponding legacy-blob path for ref_count = 0 blobs.
--
-- NOTE: pre-existing drift is NOT repaired here. Run `manifests_
-- consistency` + `blobs_consistency` after deploy; feed the findings
-- into the recovery framework.
CREATE OR REPLACE FUNCTION storage.decrement_blob_ref()
RETURNS trigger AS $$
BEGIN
-- Manifest-first, mirroring the increment side. We touch ONE
-- counter and return — the manifest reap + chunk walk + hook
-- firing lives in `dedup_gc` where Rust callbacks can run.
IF EXISTS (
SELECT 1 FROM storage.chunk_manifests
WHERE file_hash = OLD.blob_hash
) THEN
UPDATE storage.chunk_manifests
SET ref_count = GREATEST(ref_count - 1, 0)
WHERE file_hash = OLD.blob_hash;
ELSE
-- Legacy whole-file blob path: no manifest, blob is referenced
-- directly by this file row. Preserves the original behaviour
-- verbatim for the pre-CDC path.
UPDATE storage.blobs
SET ref_count = GREATEST(ref_count - 1, 0),
orphaned_at = CASE
WHEN GREATEST(ref_count - 1, 0) = 0
THEN now()
ELSE orphaned_at
END
WHERE hash = OLD.blob_hash;
END IF;
RETURN OLD;
END;
$$ LANGUAGE plpgsql;
COMMENT ON FUNCTION storage.decrement_blob_ref() IS
'Decrement the correct ref_count when a file is deleted. '
'Manifest-aware (2026-10-17): dispatches to chunk_manifests.ref_count '
'when the file''s blob_hash names a manifest, else to '
'storage.blobs.ref_count for legacy whole-file blobs. Decrement-only: '
'physical cleanup + Rust lifecycle hooks fire from dedup_gc, which '
'can invoke callbacks a SQL trigger cannot.';
@@ -0,0 +1,94 @@
-- One-time repair of ref_count drift accumulated under the pre-fix
-- copy/delete code paths.
--
-- Why this is atomic with the upgrade rather than a manual admin action
-- ─────────────────────────────────────────────────────────────────────
-- The two prior migrations on this branch:
-- * `20261016000000_copy_folder_tree_manifest_refcount.sql`
-- (fix the INCREMENT path — copy was bumping the wrong counter for
-- CDC files)
-- * `20261017000000_file_delete_trigger_manifest_aware.sql`
-- (fix the DECREMENT path — trigger was decrementing the wrong
-- counter for CDC files; folder-cascade + trash-empty paths
-- inherited that drift silently)
-- both close the bugs going forward, but production DBs upgrading
-- through this branch may carry accumulated drift from every prior
-- copy → delete cycle a CDC file went through. Under-count is the
-- dangerous direction: the next `dedup_gc` pass would reap a live
-- blob → user-facing 404 → silent data loss.
--
-- Waiting for an operator to open the admin panel and click "Repair
-- ref_counts" is the wrong default for a data-loss-preventing fix.
-- Ed's rule (`[[feedback_no_silent_auto_repair]]`): consistency
-- tenants must default to discovery-only so future bugs surface — but
-- fixing KNOWN pre-existing drift on the upgrade itself is the
-- bounded exception, because at that specific moment the source of
-- drift is known + closed, and there is no upstream mystery to
-- preserve.
--
-- Content-safety guarantees:
-- * Only counter columns change (`storage.chunk_manifests.ref_count`,
-- `storage.blobs.ref_count`). No file rows, no blob rows, no
-- manifest rows, no chunk arrays, no backend files.
-- * The corrective UPDATE sets `stored = actual` where `actual` is
-- computed from the SAME auditor formulas that
-- `manifests_consistency` / `blobs_consistency` use, so this
-- migration and those tenants agree by construction.
-- * Race-safe against concurrent writes (migrations run
-- single-connection at startup before the server serves any
-- traffic; nobody else is writing).
-- * Idempotent — fresh installs and already-clean DBs no-op (both
-- `stored` and `actual` are equal, the `WHERE <>` filters
-- everything out).
--
-- The panel button + `?repair=true` on the trigger endpoints stay for
-- FUTURE drift (regression detector; not for repeat use on this
-- accumulated set).
DO $$
DECLARE
v_m_fixed int;
v_b_fixed int;
BEGIN
-- Manifest counter: `actual` = # files whose blob_hash names this
-- manifest's file_hash. Same formula as
-- `manifests_consistency_service::manifest_page_sql` (via the
-- BlobReferenceRegistry at RefLevel::Manifest) — inline here
-- because migrations can't call Rust.
UPDATE storage.chunk_manifests m
SET ref_count = (SELECT COUNT(*) FROM storage.files
WHERE blob_hash = m.file_hash)
WHERE m.ref_count <> (SELECT COUNT(*) FROM storage.files
WHERE blob_hash = m.file_hash);
GET DIAGNOSTICS v_m_fixed = ROW_COUNT;
-- Blob counter: two-term formula mirroring
-- `blobs_consistency_service.rs:395-408`:
-- (files pointing at this blob AND having NO manifest for their
-- blob_hash — legacy whole-file path)
-- + (manifests including this hash as a chunk in chunk_hashes[])
UPDATE storage.blobs b
SET ref_count = (
(SELECT COUNT(*) FROM storage.files f
WHERE f.blob_hash = b.hash
AND NOT EXISTS (SELECT 1 FROM storage.chunk_manifests m
WHERE m.file_hash = f.blob_hash))
+ (SELECT COUNT(*) FROM storage.chunk_manifests m
WHERE b.hash = ANY(m.chunk_hashes))
)
WHERE b.ref_count <> (
(SELECT COUNT(*) FROM storage.files f
WHERE f.blob_hash = b.hash
AND NOT EXISTS (SELECT 1 FROM storage.chunk_manifests m
WHERE m.file_hash = f.blob_hash))
+ (SELECT COUNT(*) FROM storage.chunk_manifests m
WHERE b.hash = ANY(m.chunk_hashes))
);
GET DIAGNOSTICS v_b_fixed = ROW_COUNT;
-- Landed in the deploy log so an operator upgrading a huge instance
-- can see the migration did work — silent no-op on fresh installs.
RAISE NOTICE '[refcount_repair] fixed % manifest(s), % blob(s)',
v_m_fixed, v_b_fixed;
END;
$$;
@@ -0,0 +1,313 @@
//! `BlobReferenceSource` — the extension point that teaches ref-counting
//! and the consistency jobs about a table holding blob references.
//!
//! Before this port, "who references this hash" was hardcoded SQL in two
//! places (`dedup_gc`'s reap predicate and `blobs_consistency`'s refcount
//! recompute), both naming `storage.files` and `storage.chunk_manifests`
//! directly. Any new blob-owning table therefore risked silent orphaning:
//! `dedup_gc` sees `ref_count = 0`, or a manifest with no `storage.files`
//! row behind it, and reaps live content.
//!
//! See `docs/plan/derived-blobs.md` for the design and the coverage matrix.
//!
//! # Two levels, and why a source may span both
//!
//! [`DedupService::add_reference`] bumps `chunk_manifests.ref_count` first
//! and only falls back to `storage.blobs.ref_count`. So a reference lands
//! on whichever counter its hash names, and the two must be recomputed
//! separately — mixing them double-counts, systematically:
//!
//! * A **Blob** (`chunk_manifests.file_hash`) is "the content of a file".
//! * A **Chunk** (`storage.blobs.hash`) is a physical byte payload.
//! * For a single-chunk Blob the two hashes are **equal**, because both are
//! BLAKE3 over the same bytes. That aliasing is why today's chunk-level
//! recompute carries a `NOT EXISTS` clause, and why every fragment here
//! must be level-correct rather than merely plausible.
//!
//! A source is not confined to one level: [`RefLevel::Chunk`] and
//! [`RefLevel::Manifest`] fragments are requested independently, and
//! `storage.files` legitimately contributes to both — a manifest-less
//! legacy row references a chunk, a CDC row references a Blob.
//!
//! # Why SQL fragments rather than a per-hash count
//!
//! `blobs_consistency` recomputes refcounts with **one query per page**,
//! the expected count inlined as correlated subqueries. Asking each source
//! for a count per hash would turn that into `sources × rows` round-trips —
//! a catastrophic regression on a table with millions of rows. So sources
//! contribute a *fragment* that the registry sums into the existing page
//! query, and [`BlobReferenceSource::count_references`] exists only for the
//! on-demand path (`dedup_gc` checking a single reap candidate, where the
//! candidate set is already filtered to `ref_count = 0`).
use std::sync::Arc;
use async_trait::async_trait;
use crate::domain::errors::DomainError;
/// Which counter a source's references land on.
///
/// Not a property of the source — see the module docs; the same source may
/// contribute at both levels.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum RefLevel {
/// References a physical chunk. Feeds `storage.blobs.ref_count`.
Chunk,
/// References a Blob via its manifest. Feeds
/// `chunk_manifests.ref_count`.
Manifest,
}
impl RefLevel {
/// Both levels, for callers that sweep each in turn.
pub const ALL: [RefLevel; 2] = [RefLevel::Chunk, RefLevel::Manifest];
/// Stable name for logs and consistency-finding fields.
pub fn as_str(self) -> &'static str {
match self {
RefLevel::Chunk => "chunk",
RefLevel::Manifest => "manifest",
}
}
}
/// One table that holds references to blob hashes.
///
/// Implementors are registered on [`BlobReferenceRegistry`] during DI.
/// Adding a blob-owning table **without** registering it is the failure
/// this port exists to prevent.
#[async_trait]
pub trait BlobReferenceSource: Send + Sync {
/// Short stable identifier for logs and consistency-finding `source`
/// fields — `"files"`, `"chunks"`, `"content_derived"`, …
///
/// Stable across releases: log aggregators key off it.
fn source_name(&self) -> &'static str;
/// A correlated-subquery fragment counting this source's references
/// **at `level`** to `outer_hash_expr`, or `None` when this source
/// holds no references at that level.
///
/// `outer_hash_expr` is the SQL expression naming the hash of the row
/// being recomputed — `"b.hash"` when sweeping `storage.blobs`,
/// `"m.file_hash"` when sweeping `storage.chunk_manifests`. The
/// fragment must be a parenthesised scalar subquery so the registry can
/// join fragments with `+`.
///
/// **Identifiers only.** `outer_hash_expr` is supplied by the sweep, never
/// by a request; no fragment may interpolate caller input.
fn ref_count_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String>;
/// Existence form of [`Self::ref_count_sql`] — a boolean fragment, true
/// when this source holds at least one reference at `level`.
///
/// Defaults to `(<count>) > 0`. Override when the source can express a
/// short-circuiting `EXISTS`, which the planner can stop at the first
/// matching row: `dedup_gc`'s reap predicate runs this per candidate
/// manifest, and a heavily-deduplicated blob has many referrers, so
/// counting all of them where existence would do is a real regression.
fn ref_exists_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String> {
self.ref_count_sql(level, outer_hash_expr)
.map(|fragment| format!("{fragment} > 0"))
}
/// Count of references this source holds on `blob_hash`, across both
/// levels.
///
/// **On-demand path only** — `dedup_gc` checking a single reap
/// candidate. The consistency sweeps must use [`Self::ref_count_sql`];
/// calling this per row would turn one query per page into
/// `sources × rows` round-trips.
async fn count_references(&self, blob_hash: &str) -> Result<u64, DomainError>;
/// Iterate the hashes this source references, paged by the
/// implementation's natural cursor (typically a primary key).
///
/// Used by `backend_consistency` to walk the backend against the union
/// of all sources. Returns the page plus the cursor to resume from,
/// `None` when exhausted.
async fn list_referenced_blobs(
&self,
cursor: Option<Vec<u8>>,
limit: usize,
) -> Result<(Vec<String>, Option<Vec<u8>>), DomainError>;
/// Notification that `dedup_gc` reaped this blob.
///
/// Sources maintaining a denormalised refcount can clean up here. Most
/// leave the default noop — the mapping row is normally deleted by the
/// owning service's `on_blob_deleted` hook instead.
fn on_blob_reaped(&self, _blob_hash: &str) {}
}
/// The set of registered [`BlobReferenceSource`]s.
///
/// Assembled once during DI and shared (`Arc`) by `dedup_gc` and the
/// consistency jobs, so all three agree on what "referenced" means.
#[derive(Default)]
pub struct BlobReferenceRegistry {
sources: Vec<Arc<dyn BlobReferenceSource>>,
}
impl BlobReferenceRegistry {
pub fn new() -> Self {
Self::default()
}
/// Register a source. Order is irrelevant — fragments are summed and
/// counts added.
pub fn register(&mut self, source: Arc<dyn BlobReferenceSource>) {
self.sources.push(source);
}
pub fn sources(&self) -> &[Arc<dyn BlobReferenceSource>] {
&self.sources
}
/// The summed SQL expression counting every source's references at
/// `level` to `outer_hash_expr`.
///
/// Returns `"0"` when no source contributes at this level, which keeps
/// the caller's query valid without a special case.
pub fn ref_count_expr(&self, level: RefLevel, outer_hash_expr: &str) -> String {
let fragments: Vec<String> = self
.sources
.iter()
.filter_map(|s| s.ref_count_sql(level, outer_hash_expr))
.collect();
if fragments.is_empty() {
"0".to_string()
} else {
fragments.join("\n + ")
}
}
/// Predicate selecting rows that **no** registered source references at
/// `level` — i.e. reap candidates.
///
/// Returns `None` when no source contributes at this level, and callers
/// **must** treat that as "refuse to act" rather than substituting a
/// default. The natural default would be the sum-equals-zero form, which
/// on an empty registry reduces to `0 = 0` — vacuously true for every
/// row, i.e. "delete everything". Returning `None` makes that
/// unrepresentable at the call site instead of merely discouraged.
pub fn no_reference_predicate(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String> {
let fragments: Vec<String> = self
.sources
.iter()
.filter_map(|s| s.ref_exists_sql(level, outer_hash_expr))
.collect();
if fragments.is_empty() {
return None;
}
Some(format!("NOT ({})", fragments.join("\n OR ")))
}
/// Total references held on `hash` across every source.
///
/// On-demand path only — see [`BlobReferenceSource::count_references`].
pub async fn total_references(&self, hash: &str) -> Result<u64, DomainError> {
let mut total = 0u64;
for source in &self.sources {
total = total.saturating_add(source.count_references(hash).await?);
}
Ok(total)
}
/// Fan out a reap notification to every source.
pub fn notify_reaped(&self, blob_hash: &str) {
for source in &self.sources {
source.on_blob_reaped(blob_hash);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
struct Stub {
name: &'static str,
chunk: Option<&'static str>,
manifest: Option<&'static str>,
count: u64,
}
#[async_trait]
impl BlobReferenceSource for Stub {
fn source_name(&self) -> &'static str {
self.name
}
fn ref_count_sql(&self, level: RefLevel, outer: &str) -> Option<String> {
let tmpl = match level {
RefLevel::Chunk => self.chunk?,
RefLevel::Manifest => self.manifest?,
};
Some(tmpl.replace("{outer}", outer))
}
async fn count_references(&self, _blob_hash: &str) -> Result<u64, DomainError> {
Ok(self.count)
}
async fn list_referenced_blobs(
&self,
_cursor: Option<Vec<u8>>,
_limit: usize,
) -> Result<(Vec<String>, Option<Vec<u8>>), DomainError> {
Ok((Vec::new(), None))
}
}
fn registry() -> BlobReferenceRegistry {
let mut r = BlobReferenceRegistry::new();
r.register(Arc::new(Stub {
name: "a",
chunk: Some("(SELECT 1 WHERE {outer} = 'x')"),
manifest: None,
count: 2,
}));
r.register(Arc::new(Stub {
name: "b",
chunk: Some("(SELECT 2 WHERE {outer} = 'y')"),
manifest: Some("(SELECT 3 WHERE {outer} = 'z')"),
count: 5,
}));
r
}
#[test]
fn chunk_level_sums_every_contributing_source() {
let expr = registry().ref_count_expr(RefLevel::Chunk, "b.hash");
assert!(expr.contains("b.hash = 'x'"), "{expr}");
assert!(expr.contains("b.hash = 'y'"), "{expr}");
assert!(expr.contains('+'), "fragments must be summed: {expr}");
}
/// A source returning `None` for a level must contribute nothing there —
/// this is what keeps manifest-only tables out of the chunk recompute,
/// where they would double-count against the single-chunk hash alias.
#[test]
fn manifest_level_skips_non_contributing_sources() {
let expr = registry().ref_count_expr(RefLevel::Manifest, "m.file_hash");
assert!(expr.contains("m.file_hash = 'z'"), "{expr}");
assert!(!expr.contains('+'), "only one source contributes: {expr}");
}
/// An empty level must still yield a valid scalar expression, so callers
/// need no special case before a registry is fully populated.
#[test]
fn empty_level_yields_zero_literal() {
let r = BlobReferenceRegistry::new();
assert_eq!(r.ref_count_expr(RefLevel::Chunk, "b.hash"), "0");
}
#[tokio::test]
async fn total_references_adds_across_sources() {
assert_eq!(registry().total_references("deadbeef").await.unwrap(), 7);
}
}
+1
View File
@@ -1,6 +1,7 @@
pub mod auth_ports;
pub mod authorization_ports;
pub mod blob_lifecycle;
pub mod blob_reference_ports;
pub mod blob_storage_ports;
pub mod cache_ports;
pub mod calendar_ports;
+37 -1
View File
@@ -440,6 +440,22 @@ impl AppServiceFactory {
// `blob_backend` into DedupService.
let blob_backend_for_consistency = blob_backend.clone();
// Every table holding blob references. Built ONCE and shared by the
// GC reap predicate and the consistency recompute so the two cannot
// disagree about what "referenced" means — a disagreement reaps live
// content. New blob-owning tables register here.
// See docs/plan/derived-blobs.md.
let blob_reference_registry = {
use crate::infrastructure::repositories::pg::blob_reference_sources::{
ChunksReferenceSource, FilesReferenceSource,
};
let mut registry =
crate::application::ports::blob_reference_ports::BlobReferenceRegistry::new();
registry.register(Arc::new(FilesReferenceSource::new(db_pool.clone())));
registry.register(Arc::new(ChunksReferenceSource::new(db_pool.clone())));
Arc::new(registry)
};
// Deduplication service — PRIMARY blob storage engine (PostgreSQL-backed index)
let dedup_service = Arc::new(
crate::infrastructure::services::dedup_service::DedupService::new(
@@ -447,7 +463,8 @@ impl AppServiceFactory {
db_pool.clone(),
maintenance_pool.clone(),
)
.with_blob_lifecycle(blob_lifecycle),
.with_blob_lifecycle(blob_lifecycle)
.with_reference_registry(blob_reference_registry.clone()),
);
dedup_service.initialize().await?;
@@ -1460,6 +1477,22 @@ impl AppServiceFactory {
.register_recoverable_job(&core.job_registry, &job_store_provider_dyn)
.await;
// Reconciles `chunk_manifests.ref_count` — the SECOND reference
// counter, and the one nothing verified before. add_reference bumps
// it first and only falls back to storage.blobs.ref_count, so every
// CDC file (and every derived artifact, once those land) counts here
// rather than at the chunk level. Uses the same registry dedup_gc
// reaps from, so the two cannot disagree.
// See docs/plan/derived-blobs.md.
let _ = Arc::new(
crate::infrastructure::services::manifests_consistency_service::ManifestsConsistencyCheck::new(
maintenance_pool.clone(),
core.dedup_service.reference_registry(),
),
)
.register_recoverable_job(&core.job_registry, &job_store_provider_dyn)
.await;
// Third recoverable-run tenant. Iterates `storage.files`
// and reports parent-folder-trashed cascade misses,
// `missing_blob` (data-loss indicator — file references
@@ -1493,6 +1526,9 @@ impl AppServiceFactory {
core.blob_backend.clone(),
core.config.storage_entries.clone(),
self.storage_path.clone(),
// Same registry instance GC reaps from — see
// DedupService::reference_registry.
core.dedup_service.reference_registry(),
),
)
.register_recoverable_job(&core.job_registry, &job_store_provider_dyn)
@@ -0,0 +1,353 @@
//! The two implicit blob-reference sources, made explicit.
//!
//! Before this module, "who references this hash" lived as hardcoded SQL
//! inside `blobs_consistency`'s refcount recompute and `dedup_gc`'s reap
//! predicate. These two implementations reproduce that SQL **exactly** —
//! the fragments below sum to today's `actual_ref_count` expression — so
//! the registry can be wired in without changing any observed count.
//!
//! See `docs/plan/derived-blobs.md` and
//! [`crate::application::ports::blob_reference_ports`].
use std::sync::Arc;
use async_trait::async_trait;
use sqlx::{PgPool, Row};
use uuid::Uuid;
use crate::application::ports::blob_reference_ports::{BlobReferenceSource, RefLevel};
use crate::domain::errors::DomainError;
/// Aliases used inside the emitted fragments.
///
/// Deliberately distinct from the aliases the sweeps use for their outer
/// row (`b` for `storage.blobs`, `m` for `storage.chunk_manifests`): a
/// fragment reusing `m` would shadow the outer alias in the manifest-level
/// sweep and silently correlate against itself.
const FILES_ALIAS: &str = "cnt_f";
const MANIFEST_ALIAS: &str = "cnt_m";
/// Fragment for [`FilesReferenceSource`], as a free function so the SQL
/// shape can be tested without constructing a pool — it is a property of
/// the module, not of an instance.
fn files_ref_sql(level: RefLevel, outer_hash_expr: &str) -> Option<String> {
let f = FILES_ALIAS;
match level {
// Legacy whole-file blobs only — CDC files are counted at the
// manifest level, and counting them here too would double up on the
// single-chunk hash alias.
RefLevel::Chunk => Some(format!(
"(SELECT COUNT(*) FROM storage.files {f}
WHERE {f}.blob_hash = {outer_hash_expr}
AND NOT EXISTS (
SELECT 1 FROM storage.chunk_manifests {MANIFEST_ALIAS}
WHERE {MANIFEST_ALIAS}.file_hash = {f}.blob_hash
))"
)),
RefLevel::Manifest => Some(format!(
"(SELECT COUNT(*) FROM storage.files {f}
WHERE {f}.blob_hash = {outer_hash_expr})"
)),
}
}
/// Short-circuiting existence form of [`files_ref_sql`].
///
/// `dedup_gc` evaluates this per candidate manifest, so counting every
/// referrer where existence would do is a real cost on a heavily-deduplicated
/// blob. This is also the exact shape the reap predicate used before the
/// registry existed, so wiring it in changes no plan.
fn files_exists_sql(level: RefLevel, outer_hash_expr: &str) -> Option<String> {
let f = FILES_ALIAS;
match level {
RefLevel::Chunk => Some(format!(
"EXISTS (SELECT 1 FROM storage.files {f} \
WHERE {f}.blob_hash = {outer_hash_expr} \
AND NOT EXISTS (SELECT 1 FROM storage.chunk_manifests {MANIFEST_ALIAS} \
WHERE {MANIFEST_ALIAS}.file_hash = {f}.blob_hash))"
)),
RefLevel::Manifest => Some(format!(
"EXISTS (SELECT 1 FROM storage.files {f} WHERE {f}.blob_hash = {outer_hash_expr})"
)),
}
}
/// Fragment for [`ChunksReferenceSource`]. See [`files_ref_sql`].
fn chunks_ref_sql(level: RefLevel, outer_hash_expr: &str) -> Option<String> {
match level {
RefLevel::Chunk => {
let m = MANIFEST_ALIAS;
Some(format!(
"(SELECT COUNT(*) FROM storage.chunk_manifests {m}
WHERE {outer_hash_expr} = ANY({m}.chunk_hashes))"
))
}
// A manifest is never referenced by another manifest.
RefLevel::Manifest => None,
}
}
// ─── storage.files ───────────────────────────────────────────────────────
/// References held by `storage.files.blob_hash`.
///
/// Contributes at **both** levels, which is why `RefLevel` is a parameter
/// rather than a property of the source:
///
/// * [`RefLevel::Manifest`] — a CDC file's `blob_hash` names a manifest.
/// * [`RefLevel::Chunk`] — a pre-CDC legacy file, whose `blob_hash` names a
/// whole-file blob with no manifest behind it. The `NOT EXISTS` guard is
/// load-bearing: for a single-chunk file the whole-file hash *equals* its
/// lone chunk's hash, so without it the row would be counted at both
/// levels.
pub struct FilesReferenceSource {
pool: Arc<PgPool>,
}
impl FilesReferenceSource {
pub fn new(pool: Arc<PgPool>) -> Self {
Self { pool }
}
}
#[async_trait]
impl BlobReferenceSource for FilesReferenceSource {
fn source_name(&self) -> &'static str {
"files"
}
fn ref_count_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String> {
files_ref_sql(level, outer_hash_expr)
}
fn ref_exists_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String> {
files_exists_sql(level, outer_hash_expr)
}
async fn count_references(&self, blob_hash: &str) -> Result<u64, DomainError> {
// No level split here: the question is "how many file rows name this
// exact hash", and a hash names either a manifest or a legacy blob,
// never both at once from the caller's point of view.
let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM storage.files WHERE blob_hash = $1")
.bind(blob_hash)
.fetch_one(self.pool.as_ref())
.await
.map_err(|e| {
DomainError::internal_error("BlobRefSource", format!("files count: {e}"))
})?;
Ok(n.max(0) as u64)
}
async fn list_referenced_blobs(
&self,
cursor: Option<Vec<u8>>,
limit: usize,
) -> Result<(Vec<String>, Option<Vec<u8>>), DomainError> {
// Paged by the file's own PK so the cursor is stable under concurrent
// inserts; `blob_hash` is not unique and would skip or repeat rows.
let after: Option<Uuid> = match cursor {
Some(bytes) => Some(decode_uuid_cursor(&bytes)?),
None => None,
};
let rows = sqlx::query(
"SELECT id, blob_hash FROM storage.files
WHERE ($1::uuid IS NULL OR id > $1)
ORDER BY id
LIMIT $2",
)
.bind(after)
.bind(limit as i64)
.fetch_all(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("BlobRefSource", format!("files page: {e}")))?;
let next = rows
.last()
.map(|r| r.get::<Uuid, _>("id").as_bytes().to_vec())
.filter(|_| rows.len() == limit);
let hashes = rows
.iter()
.map(|r| r.get::<String, _>("blob_hash"))
.collect();
Ok((hashes, next))
}
}
// ─── storage.chunk_manifests ─────────────────────────────────────────────
/// References held by `storage.chunk_manifests.chunk_hashes[]`.
///
/// Chunk level only — a manifest never references another manifest, so
/// [`RefLevel::Manifest`] yields `None` and this source contributes nothing
/// to the manifest recompute.
pub struct ChunksReferenceSource {
pool: Arc<PgPool>,
}
impl ChunksReferenceSource {
pub fn new(pool: Arc<PgPool>) -> Self {
Self { pool }
}
}
#[async_trait]
impl BlobReferenceSource for ChunksReferenceSource {
fn source_name(&self) -> &'static str {
"chunks"
}
fn ref_count_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String> {
chunks_ref_sql(level, outer_hash_expr)
}
async fn count_references(&self, blob_hash: &str) -> Result<u64, DomainError> {
let n: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM storage.chunk_manifests WHERE $1 = ANY(chunk_hashes)",
)
.bind(blob_hash)
.fetch_one(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("BlobRefSource", format!("chunks count: {e}")))?;
Ok(n.max(0) as u64)
}
async fn list_referenced_blobs(
&self,
cursor: Option<Vec<u8>>,
limit: usize,
) -> Result<(Vec<String>, Option<Vec<u8>>), DomainError> {
// Paged by the manifest PK, not by the unnested chunk hash: a single
// manifest expands to many hashes, so the page boundary has to fall
// between manifests or the cursor cannot be resumed unambiguously.
let after: Option<String> = match cursor {
Some(bytes) => Some(String::from_utf8(bytes).map_err(|e| {
DomainError::internal_error("BlobRefSource", format!("bad chunk cursor: {e}"))
})?),
None => None,
};
let rows = sqlx::query(
"SELECT file_hash, chunk_hashes FROM storage.chunk_manifests
WHERE ($1::text IS NULL OR file_hash > $1)
ORDER BY file_hash
LIMIT $2",
)
.bind(after)
.bind(limit as i64)
.fetch_all(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("BlobRefSource", format!("chunks page: {e}")))?;
let next = rows
.last()
.map(|r| r.get::<String, _>("file_hash").into_bytes())
.filter(|_| rows.len() == limit);
let hashes = rows
.iter()
.flat_map(|r| r.get::<Vec<String>, _>("chunk_hashes"))
.collect();
Ok((hashes, next))
}
}
fn decode_uuid_cursor(bytes: &[u8]) -> Result<Uuid, DomainError> {
let raw: [u8; 16] = bytes.try_into().map_err(|_| {
DomainError::internal_error(
"BlobRefSource",
format!("bad uuid cursor: expected 16 bytes, got {}", bytes.len()),
)
})?;
Ok(Uuid::from_bytes(raw))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::application::ports::blob_reference_ports::BlobReferenceRegistry;
/// The registry sums whatever the sources emit; these helpers exercise the
/// same code path without needing a pool, since `ref_count_sql` is pure.
fn summed(level: RefLevel, outer: &str) -> String {
let frags: Vec<String> = [files_ref_sql(level, outer), chunks_ref_sql(level, outer)]
.into_iter()
.flatten()
.collect();
if frags.is_empty() {
"0".to_string()
} else {
frags.join("\n + ")
}
}
/// The chunk-level expression must reproduce the two terms
/// `blobs_consistency` inlines today: legacy-only files (guarded by
/// NOT EXISTS) plus manifests citing the chunk.
#[test]
fn chunk_level_reproduces_todays_two_terms() {
let expr = summed(RefLevel::Chunk, "b.hash");
assert!(expr.contains("storage.files"), "{expr}");
assert!(
expr.contains("NOT EXISTS"),
"legacy term must keep the CDC guard: {expr}"
);
assert!(
expr.contains("= ANY(cnt_m.chunk_hashes)"),
"chunk term missing: {expr}"
);
assert!(expr.contains("b.hash"), "must correlate on the outer row");
assert!(expr.contains('+'), "both terms must be summed: {expr}");
}
/// Only `storage.files` references a manifest, so the manifest-level
/// expression is the single files term with no `NOT EXISTS` guard — the
/// guard exists to keep CDC rows *out* of the chunk level, and applying
/// it here would count nothing at all.
#[test]
fn manifest_level_is_files_only_and_unguarded() {
let expr = summed(RefLevel::Manifest, "m.file_hash");
assert!(expr.contains("storage.files"), "{expr}");
assert!(!expr.contains("NOT EXISTS"), "{expr}");
assert!(
!expr.contains("chunk_hashes"),
"chunks must not contribute at manifest level: {expr}"
);
assert!(!expr.contains('+'), "only one source contributes: {expr}");
assert!(expr.contains("m.file_hash"));
}
/// Fragments must not use the aliases the sweeps use for their outer row
/// (`b` for storage.blobs, `m` for chunk_manifests), or the manifest sweep
/// would shadow its own alias and silently correlate against itself.
#[test]
fn fragments_avoid_outer_row_aliases() {
for level in RefLevel::ALL {
let expr = summed(level, "m.file_hash");
for bad in [
"storage.files f",
"storage.files b",
"chunk_manifests m ",
"chunk_manifests b",
] {
assert!(!expr.contains(bad), "alias collision at {level:?}: {expr}");
}
}
}
/// A source declining a level must drop out of the sum entirely, which is
/// what keeps manifest-only tables out of the chunk recompute where the
/// single-chunk hash alias would double-count them.
#[test]
fn chunks_source_declines_manifest_level() {
assert!(chunks_ref_sql(RefLevel::Manifest, "m.file_hash").is_none());
assert!(chunks_ref_sql(RefLevel::Chunk, "b.hash").is_some());
}
/// Guards the registry contract the sweeps rely on: an empty level still
/// yields a valid scalar expression.
#[test]
fn empty_registry_yields_zero_literal() {
let r = BlobReferenceRegistry::new();
assert_eq!(r.ref_count_expr(RefLevel::Manifest, "m.file_hash"), "0");
}
}
@@ -1024,10 +1024,21 @@ impl FileWritePort for FileBlobWriteRepository {
DomainError::internal_error("FileBlobWrite", format!("fetch blob_hash: {e}"))
})?;
// DELETE fires trg_files_decrement_blob_ref → storage.blobs.ref_count--
// DELETE fires `trg_files_decrement_blob_ref` — post-2026-08-23
// it dispatches manifest-first (see migration
// `20261017000000_file_delete_trigger_manifest_aware.sql`):
// decrements `chunk_manifests.ref_count` if the hash names a
// manifest (walking chunks on last-ref), else falls back to
// `storage.blobs.ref_count`. Counter state after this call is
// already correct.
self.delete_file(file_id).await?;
// If the blob is now unreferenced, remove disk file + thumbnails.
// Physical cleanup only. `cleanup_if_orphaned` was previously
// manifest-aware and did counter compensation for the old
// trigger's over-decrement; after the trigger rewrite it's a
// legacy-blob-eager-reap helper — safe to keep calling
// unconditionally (no-op for CDC hashes; reaps legacy blobs
// that reached ref_count = 0).
if let Some(hash) = blob_hash {
self.dedup.cleanup_if_orphaned(&hash).await;
}
@@ -1,5 +1,6 @@
mod address_book_pg_repository;
mod app_password_pg_repository;
pub mod blob_reference_sources;
mod calendar_event_pg_repository;
mod calendar_pg_repository;
mod contact_group_pg_repository;
+14
View File
@@ -45,11 +45,25 @@ use serde::{Deserialize, Serialize};
/// of the entry to probe instead of the currently-active backend.
/// `None` falls through to the live backend (today's behaviour).
/// - Others — ignored.
///
/// Semantics of `repair` (added 2026-10-17 for the refcount fix):
/// - `blobs_consistency` / `manifests_consistency` — when `true`,
/// after each `refcount_mismatch` / `manifest_refcount_mismatch`
/// finding is recorded, apply the corrective UPDATE that sets the
/// stored counter to the auditor's computed `actual_ref_count`.
/// Content-safe: the row itself is fine, only the counter is
/// wrong. Race-safe: each UPDATE recomputes the auditor formula
/// in the same statement, so a concurrent write can't leave a
/// stale value. Default `false` preserves discovery-only
/// behaviour. Also propagates through `consistency_batch` to
/// both tenants — one `?repair=true` call fixes both counters.
/// - Others — ignored.
#[derive(Debug, Clone, Default)]
pub struct JobRunArgs {
pub force: bool,
pub deep: bool,
pub storage: Option<String>,
pub repair: bool,
}
/// Uniform outcome the supervisor logs and stores for every job dispatch.
@@ -63,6 +63,7 @@ use async_trait::async_trait;
use chrono::{DateTime, Duration, Utc};
use sqlx::PgPool;
use crate::application::ports::blob_reference_ports::{BlobReferenceRegistry, RefLevel};
use crate::application::ports::blob_storage_ports::BlobStorageBackend;
use crate::common::config::NamedStorageEntry;
use crate::infrastructure::scheduler::{
@@ -114,6 +115,64 @@ pub struct BlobsConsistencyCheck {
/// fallback for a Local target entry with no `_ROOT_DIR`. Same
/// fallback rule the boot path uses.
storage_path_fallback: PathBuf,
/// The chunk-level page query, assembled once from the blob-reference
/// registry so this recompute and `dedup_gc` agree on what "referenced"
/// means. Built at construction rather than per page so the sweep runs a
/// fixed statement — same reasoning as `DedupService::manifest_reap_sql`.
/// See `docs/plan/derived-blobs.md`.
chunk_page_sql: String,
}
/// The chunk-level page query, with `actual_ref_count` summed from the
/// registered reference sources.
///
/// `storage.blobs.ref_count` semantics — the invariant `dedup_service`
/// actually maintains:
///
/// ```text
/// ref_count = (number of chunk_manifests whose chunk_hashes[] contains
/// this hash)
/// + (number of files.blob_hash pointing at this hash on the
/// LEGACY whole-file path — files with NO manifest for their
/// blob_hash)
/// ```
///
/// Naively `COUNT(files) + COUNT(manifests referring)` double-counts
/// single-chunk CDC files: where a file's whole-file hash equals its lone
/// chunk's hash (anything under one CDC chunk), the file appears BOTH in
/// `files.blob_hash` and in the manifest's `chunk_hashes[]`. The
/// `NOT EXISTS` guard inside `FilesReferenceSource`'s chunk-level fragment
/// excludes CDC-path files from the legacy term so the two don't overlap.
///
/// The GIN index on `chunk_hashes` (migration
/// `20260628000000_delta_upload_gin_index`) keeps the `= ANY(chunk_hashes)`
/// probe cheap.
///
/// # Panics
///
/// If no source contributes at [`RefLevel::Chunk`] — a wiring bug that
/// would make every blob look unreferenced and flag the whole table as
/// `refcount_mismatch`.
fn chunk_page_sql(registry: &BlobReferenceRegistry) -> String {
let expected = registry.ref_count_expr(RefLevel::Chunk, "b.hash");
assert!(
expected != "0",
"no chunk-level blob reference source registered: every blob would \
appear unreferenced"
);
format!(
"SELECT
b.hash AS hash,
b.size AS size,
b.ref_count AS ref_count,
b.created_at AS created_at,
({expected})::bigint AS actual_ref_count
FROM storage.blobs b
WHERE ($1::text IS NULL OR b.hash > $1)
ORDER BY b.hash
LIMIT $2"
)
}
impl BlobsConsistencyCheck {
@@ -122,12 +181,14 @@ impl BlobsConsistencyCheck {
backend: Arc<dyn BlobStorageBackend>,
storage_entries: Vec<NamedStorageEntry>,
storage_path_fallback: PathBuf,
reference_registry: Arc<BlobReferenceRegistry>,
) -> Self {
Self {
pool,
backend,
storage_entries,
storage_path_fallback,
chunk_page_sql: chunk_page_sql(&reference_registry),
}
}
@@ -286,6 +347,10 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
// stats.finding_count — actual persistence happens in
// `record_finding` on each emission).
let mut finding_count = 0u64;
// Only touched when `args.repair == true`. Symmetric with
// `manifests_consistency`; reported in completion log +
// `extra_stats` so operators see "found N, fixed M" in one line.
let mut repaired_count = 0u64;
// Deep mode is a per-run flag with two consumers:
// 1. This handler — decides whether to re-hash bytes.
@@ -338,6 +403,41 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
);
}
// Repair mode: same shape as `deep` above so the admin run-
// detail view can display `params.repair = "true"` alongside
// `params.deep`. Fresh persists what the trigger asked for;
// Resume reads back so a paused repair scan stays a repair
// scan (a mid-scan crash mustn't silently downgrade to
// discovery-only for the remaining rows).
let repair = if is_fresh {
let v = if args.repair { "true" } else { "false" };
if let Err(e) = store.set_string_param("repair", v).await {
return RunOutcome::Failed {
message: format!("failed to persist repair flag to params: {e}"),
};
}
args.repair
} else {
match store.get_string_param("repair").await {
Ok(Some(v)) => v == "true",
Ok(None) => false,
Err(e) => {
return RunOutcome::Failed {
message: format!("read `repair` from params: {e}"),
};
}
}
};
if repair {
tracing::info!(
target: "oxicloud::consistency",
event = "blobs_consistency.repair_mode_active",
run_id = %store.run_id(),
"repair mode: refcount_mismatch findings will trigger corrective UPDATE"
);
}
loop {
// Cooperative cancel poll between batches.
match store.status().await {
@@ -364,58 +464,14 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
}
}
// Fetch the next batch. Per-row `actual_ref_count`
// computed inline via correlated subqueries — one for
// legacy whole-file references (`files.blob_hash`), one
// for CDC chunk references (`chunk_manifests.chunk_hashes`).
// GIN index on `chunk_hashes` (migration
// 20260628000000_delta_upload_gin_index) makes the
// `= ANY(chunk_hashes)` probe cheap.
// `storage.blobs.ref_count` semantics — what the invariant
// dedup_service maintains actually is:
//
// ref_count = (number of chunk_manifests whose
// chunk_hashes[] contains this hash)
// + (number of files.blob_hash pointing at
// this hash on the LEGACY whole-file path
// — i.e. files with NO manifest for their
// blob_hash)
//
// Naively `COUNT(files) + COUNT(manifests referring)`
// double-counts single-chunk CDC files: for a file whose
// whole-file hash == its single chunk's hash (any file
// small enough to fit in one CDC chunk — under ~256 KB
// average), the file appears BOTH in `files.blob_hash`
// AND in the manifest's `chunk_hashes[]`. The `NOT
// EXISTS` clause below excludes CDC-path files from the
// legacy count so the two terms don't overlap.
let rows: Vec<BlobRow> = match sqlx::query_as(
r#"
SELECT
b.hash AS hash,
b.size AS size,
b.ref_count AS ref_count,
b.created_at AS created_at,
(
(SELECT COUNT(*) FROM storage.files f
WHERE f.blob_hash = b.hash
AND NOT EXISTS (
SELECT 1 FROM storage.chunk_manifests m
WHERE m.file_hash = f.blob_hash
))
+ (SELECT COUNT(*) FROM storage.chunk_manifests m
WHERE b.hash = ANY(m.chunk_hashes))
)::bigint AS actual_ref_count
FROM storage.blobs b
WHERE ($1::text IS NULL OR b.hash > $1)
ORDER BY b.hash
LIMIT $2
"#,
)
.bind(cursor.as_deref())
.bind(BATCH_SIZE)
.fetch_all(self.pool.as_ref())
.await
// Fetch the next batch. `actual_ref_count` is summed from the
// registered reference sources — see `chunk_page_sql`, which
// documents the invariant and the single-chunk double-count trap.
let rows: Vec<BlobRow> = match sqlx::query_as(&self.chunk_page_sql)
.bind(cursor.as_deref())
.bind(BATCH_SIZE)
.fetch_all(self.pool.as_ref())
.await
{
Ok(r) => r,
Err(e) => {
@@ -431,11 +487,17 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
event = "blobs_consistency.completed",
run_id = %store.run_id(),
finding_count = finding_count,
repaired_count = repaired_count,
repair_requested = repair,
deep = deep,
"blobs_consistency completed with {} finding(s)",
finding_count
"blobs_consistency completed with {} finding(s), {} repaired",
finding_count,
repaired_count
);
return RunOutcome::completed();
return RunOutcome::completed_with(serde_json::json!({
"repair_requested": repair,
"repaired_count": repaired_count,
}));
}
let grace_cutoff = Utc::now() - CREATE_GRACE;
@@ -463,6 +525,67 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
}),
)
.await;
// Repair pass — content-safe corrective UPDATE. Sets
// `stored` to the value the auditor's two-term formula
// would compute at UPDATE time (subquery mirrors
// `chunk_page_sql`'s `actual_ref_count`), so a
// concurrent write between our page fetch and this
// UPDATE can't leave a stale value — the subquery
// re-reads inside the same statement. The
// `<> (subquery)` guard makes the UPDATE a no-op if
// the drift has healed, making this idempotent under
// retry.
if repair {
let expected = "( \
(SELECT COUNT(*) FROM storage.files f \
WHERE f.blob_hash = b.hash \
AND NOT EXISTS ( \
SELECT 1 FROM storage.chunk_manifests m \
WHERE m.file_hash = f.blob_hash \
)) \
+ (SELECT COUNT(*) FROM storage.chunk_manifests m \
WHERE b.hash = ANY(m.chunk_hashes)) \
)";
let update_sql = format!(
"UPDATE storage.blobs b \
SET ref_count = {expected} \
WHERE b.hash = $1 \
AND b.ref_count <> {expected}",
);
match sqlx::query(&update_sql)
.bind(&row.hash)
.execute(self.pool.as_ref())
.await
{
Ok(res) if res.rows_affected() > 0 => {
repaired_count += 1;
tracing::info!(
target: "audit",
event = "blobs_consistency.repaired",
run_id = %store.run_id(),
hash = %row.hash,
stored_was = row.ref_count,
actual = row.actual_ref_count,
"🩹 blob ref_count repaired"
);
}
Ok(_) => {
// No row touched — concurrent repair or
// self-healing drift. Silent no-op.
}
Err(e) => {
tracing::warn!(
target: "oxicloud::consistency",
event = "blobs_consistency.repair_failed",
run_id = %store.run_id(),
hash = %row.hash,
error = %e,
"blob ref_count repair UPDATE failed — finding stays"
);
}
}
}
}
// Skip physical probes for rows within the write
@@ -611,11 +734,17 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
event = "blobs_consistency.completed",
run_id = %store.run_id(),
finding_count = finding_count,
repaired_count = repaired_count,
repair_requested = repair,
deep = deep,
"blobs_consistency completed with {} finding(s)",
finding_count
"blobs_consistency completed with {} finding(s), {} repaired",
finding_count,
repaired_count
);
return RunOutcome::completed();
return RunOutcome::completed_with(serde_json::json!({
"repair_requested": repair,
"repaired_count": repaired_count,
}));
}
}
}
@@ -683,3 +812,64 @@ async fn recompute_hash(
Ok(hasher.finalize().to_hex().to_string())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::infrastructure::repositories::pg::blob_reference_sources::{
ChunksReferenceSource, FilesReferenceSource,
};
fn default_registry() -> BlobReferenceRegistry {
let pool = Arc::new(
sqlx::pool::PoolOptions::<sqlx::Postgres>::new()
.connect_lazy("postgres://invalid/invalid")
.expect("lazy pool never connects"),
);
let mut registry = BlobReferenceRegistry::new();
registry.register(Arc::new(FilesReferenceSource::new(pool.clone())));
registry.register(Arc::new(ChunksReferenceSource::new(pool)));
registry
}
/// Golden test for the chunk-level recompute. Pins the statement
/// byte-for-byte because it is assembled from the registry rather than
/// written as a literal — the reviewer should read the SQL here.
///
/// This expression must stay equal to what the query computed before the
/// registry existed: the legacy-files term guarded by `NOT EXISTS`, plus
/// the manifests-citing-this-chunk term. If a change makes those two
/// overlap, every single-chunk CDC file is counted twice and the whole
/// table reports `refcount_mismatch`.
#[tokio::test]
async fn chunk_page_statement_is_stable() {
let sql = chunk_page_sql(&default_registry());
let expected = r#"SELECT
b.hash AS hash,
b.size AS size,
b.ref_count AS ref_count,
b.created_at AS created_at,
((SELECT COUNT(*) FROM storage.files cnt_f
WHERE cnt_f.blob_hash = b.hash
AND NOT EXISTS (
SELECT 1 FROM storage.chunk_manifests cnt_m
WHERE cnt_m.file_hash = cnt_f.blob_hash
))
+ (SELECT COUNT(*) FROM storage.chunk_manifests cnt_m
WHERE b.hash = ANY(cnt_m.chunk_hashes)))::bigint AS actual_ref_count
FROM storage.blobs b
WHERE ($1::text IS NULL OR b.hash > $1)
ORDER BY b.hash
LIMIT $2"#;
assert_eq!(sql, expected, "chunk page statement changed:\n{sql}");
}
/// With no chunk-level source every blob would look unreferenced and the
/// sweep would report the entire table as `refcount_mismatch`. Refuse to
/// build the statement instead.
#[test]
#[should_panic(expected = "no chunk-level blob reference source")]
fn empty_registry_refuses_to_build_page_statement() {
let _ = chunk_page_sql(&BlobReferenceRegistry::new());
}
}
@@ -178,6 +178,7 @@ impl JobHandler for ConsistencyBatch {
"per_check": per_check,
"deep": args.deep,
"force": args.force,
"repair": args.repair,
"ok": ok_count,
"err": err_count,
}),
+216 -79
View File
@@ -55,6 +55,7 @@ use std::sync::Arc;
use tokio_util::io::StreamReader;
use crate::application::ports::blob_lifecycle::BlobLifecycleHook;
use crate::application::ports::blob_reference_ports::{BlobReferenceRegistry, RefLevel};
use crate::application::ports::blob_storage_ports::BlobStorageBackend;
use crate::application::ports::dedup_ports::{
BlobMetadataDto, DedupPort, DedupResultDto, DedupStatsDto,
@@ -424,6 +425,51 @@ async fn populate_integrity_blob_sizes<'a>(
IntegrityBlobSizes { hashes, sizes }
}
/// Build the manifest reap statement from the registered reference sources.
///
/// A manifest is collectible when either:
/// * `ref_count` reached 0 via `cleanup_if_orphaned` on the single-file
/// delete path, **or**
/// * nothing references it any more — the bulk-delete path (user cascade,
/// `empty_trash`), where the PG trigger only touches `storage.blobs` and
/// the per-file `cleanup_if_orphaned` call is skipped, so `ref_count` is
/// never decremented and the second clause is the only thing that reaps.
///
/// The second clause used to name `storage.files` directly, which hardcoded
/// "files is the only thing that can reference a manifest". Any new referring
/// table — thumbnails via `storage.content_derived_blobs`, previews via
/// `storage.file_attached_blobs` — would then have its manifests reaped on the
/// next sweep *despite a correct `ref_count`*: clause one false, clause two
/// true, `OR` fires, bytes gone. See `docs/plan/derived-blobs.md`.
///
/// # Panics
///
/// If no source contributes at [`RefLevel::Manifest`]. That is a wiring bug,
/// and it must be loud: with no source, "nothing references it" is vacuously
/// true for every row and this statement would delete every manifest in the
/// database. `DedupService::new` always registers `FilesReferenceSource`, so
/// the only way to reach this is to pass a deliberately empty registry.
fn manifest_reap_sql(registry: &BlobReferenceRegistry) -> String {
let orphaned = registry
.no_reference_predicate(RefLevel::Manifest, "m.file_hash")
.expect(
"no manifest-level blob reference source registered: the reap \
predicate would match every manifest",
);
format!(
"DELETE FROM storage.chunk_manifests
WHERE ctid = ANY(
SELECT ctid
FROM storage.chunk_manifests m
WHERE m.ref_count <= 0
OR {orphaned}
LIMIT $1
)
RETURNING file_hash, chunk_hashes, total_size"
)
}
pub struct DedupService {
/// Pluggable blob storage backend (local FS, S3, …).
backend: Arc<dyn BlobStorageBackend>,
@@ -442,6 +488,16 @@ pub struct DedupService {
/// seen immediately), weight-bounded (a manifest is ~72 B per chunk),
/// short TTL so GC'd manifests age out fast (benches/MANIFEST-CACHE.md).
manifest_cache: moka::future::Cache<String, Arc<ChunkManifest>>,
/// Every table that holds blob references, so GC agrees with the
/// consistency jobs on what "referenced" means. Defaults to the two
/// built-in sources; DI replaces it once more tables exist. Never
/// optional — an empty registry would make "nothing references it"
/// vacuously true and the manifest sweep would reap everything.
reference_registry: Arc<BlobReferenceRegistry>,
/// The manifest reap statement, built once from `reference_registry`.
/// Kept as a field so `garbage_collect` runs a fixed statement rather
/// than assembling SQL inside a delete loop — see `manifest_reap_sql`.
manifest_reap_sql: String,
}
impl DedupService {
@@ -455,15 +511,32 @@ impl DedupService {
pool: Arc<PgPool>,
maintenance_pool: Arc<PgPool>,
) -> Self {
let registry = Arc::new(Self::default_reference_registry(pool.clone()));
Self {
backend,
pool,
maintenance_pool,
blob_lifecycle: None,
manifest_cache: Self::build_manifest_cache(),
reference_registry: registry.clone(),
manifest_reap_sql: manifest_reap_sql(&registry),
}
}
/// The two sources that were implicit before the registry existed.
/// Keeping this as the default means every construction path — including
/// tests — has a manifest-level source, so the reap predicate can never
/// degenerate to "nothing references anything".
fn default_reference_registry(pool: Arc<PgPool>) -> BlobReferenceRegistry {
use crate::infrastructure::repositories::pg::blob_reference_sources::{
ChunksReferenceSource, FilesReferenceSource,
};
let mut registry = BlobReferenceRegistry::new();
registry.register(Arc::new(FilesReferenceSource::new(pool.clone())));
registry.register(Arc::new(ChunksReferenceSource::new(pool)));
registry
}
/// See the `manifest_cache` field docs. Weight ≈ real heap bytes of one
/// entry; 32 MiB cap ≈ tens of thousands of typical (sub-1 GB) files.
fn build_manifest_cache() -> moka::future::Cache<String, Arc<ChunkManifest>> {
@@ -476,6 +549,25 @@ impl DedupService {
.build()
}
/// Registers the blob-reference registry used by the manifest reap
/// predicate. Without it `garbage_collect` skips manifest collection
/// entirely — see `docs/plan/derived-blobs.md`.
pub fn with_reference_registry(mut self, registry: Arc<BlobReferenceRegistry>) -> Self {
self.manifest_reap_sql = manifest_reap_sql(&registry);
self.reference_registry = registry;
self
}
/// The registry backing the reap predicate.
///
/// Exposed so `blobs_consistency` recomputes refcounts from the *same*
/// source set GC reaps from. If the two ever diverged, the sweep would
/// bless counts the collector disagrees with — and the collector wins,
/// destructively.
pub fn reference_registry(&self) -> Arc<BlobReferenceRegistry> {
self.reference_registry.clone()
}
/// Registers the blob lifecycle dispatcher (thumbnail cleanup, …).
pub fn with_blob_lifecycle(mut self, lifecycle: Arc<BlobLifecycleService>) -> Self {
self.blob_lifecycle = Some(lifecycle);
@@ -510,12 +602,15 @@ impl DedupService {
.connect_lazy("postgres://invalid:5432/none")
.unwrap(),
);
let stub_registry = Arc::new(Self::default_reference_registry(stub_pool.clone()));
Self {
backend: Arc::new(LocalBlobBackend::new(Path::new("/tmp/oxicloud_stub_blobs"))),
pool: stub_pool.clone(),
maintenance_pool: stub_pool,
maintenance_pool: stub_pool.clone(),
blob_lifecycle: None,
manifest_cache: Self::build_manifest_cache(),
reference_registry: stub_registry.clone(),
manifest_reap_sql: manifest_reap_sql(&stub_registry),
}
}
@@ -523,6 +618,32 @@ impl DedupService {
pub async fn initialize(&self) -> Result<(), DomainError> {
self.backend.initialize().await?;
// The reap statement is assembled from the registered reference
// sources, so it is not greppable in the source tree. It DELETES
// manifests, so log it unconditionally at info rather than hiding it
// behind a filter an operator has to know to enable — if what GC
// considers "referenced" ever changes, that must be visible on the
// next boot without anyone going looking.
//
// Whitespace-collapsed to a single field so a multi-line query does
// not sprawl across the boot log; expand it with
// `sed 's/ AND / AND\n /g'` or just paste it into psql.
tracing::info!(
target: "oxicloud::dedup",
sources = ?self
.reference_registry
.sources()
.iter()
.map(|s| s.source_name())
.collect::<Vec<_>>(),
statement = %self
.manifest_reap_sql
.split_whitespace()
.collect::<Vec<_>>()
.join(" "),
"🧹 manifest reap predicate registered"
);
let blob_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM storage.blobs")
.fetch_one(self.pool.as_ref())
.await
@@ -1838,66 +1959,48 @@ impl DedupService {
pub async fn cleanup_if_orphaned(&self, hash: &str) {
let short = &hash[..hash.len().min(12)];
// ── CDC manifest path (must run FIRST) ───────────────────
// For single-chunk CDC files file_hash == chunk_hash, so the PG
// trigger on storage.files already decremented storage.blobs.ref_count
// when this function is called. try_dedup_hit increments
// chunk_manifests.ref_count but NOT storage.blobs.ref_count, so
// blobs.ref_count can reach 0 while the manifest still has ref_count > 1
// (other files sharing the same blob). Checking the manifest first
// prevents premature blob + manifest deletion.
let manifest = sqlx::query_as::<_, (i32, Vec<String>)>(
"SELECT ref_count, chunk_hashes \
FROM storage.chunk_manifests WHERE file_hash = $1",
)
.bind(hash)
.fetch_optional(self.pool.as_ref())
.await
.unwrap_or(None);
if let Some((ref_count, chunk_hashes)) = manifest {
if ref_count <= 1 {
// Last reference — remove manifest and all its chunks.
if let Err(e) = self
.remove_manifest_reference(hash, ref_count, &chunk_hashes)
.await
{
tracing::warn!("cleanup_if_orphaned: manifest cleanup failed for {short}: {e}");
}
} else {
// Other files still share this blob: just decrement the manifest
// counter and undo the PG trigger's premature chunk ref_count
// decrement (blobs.ref_count is chunk-level; the manifest is the
// authoritative file-level counter).
sqlx::query(
"UPDATE storage.chunk_manifests \
SET ref_count = ref_count - 1 WHERE file_hash = $1",
)
.bind(hash)
.execute(self.pool.as_ref())
.await
.ok();
// Undo the PG trigger's decrement of storage.blobs.ref_count.
// The trigger fired with blob_hash = file_hash, so only the row
// WHERE hash = file_hash is affected. For single-chunk files
// file_hash == chunk_hash and that row exists; for multi-chunk
// files file_hash is not in storage.blobs, making this a no-op.
sqlx::query("UPDATE storage.blobs SET ref_count = ref_count + 1 WHERE hash = $1")
.bind(hash)
.execute(self.pool.as_ref())
.await
.ok();
tracing::debug!(
"cleanup_if_orphaned: manifest {short} ref_count {ref_count}→{}",
ref_count - 1
);
}
return;
}
// ── Legacy blob path (no manifest) ───────────────────────
// 2026-08-23 refactor: this function used to compensate for the
// OLD PG trigger `trg_files_decrement_blob_ref` unconditionally
// decrementing `storage.blobs.ref_count`, which was wrong for
// CDC files (their `blob_hash` names a `chunk_manifests.file_hash`,
// not a chunk-in-a-manifest). The compensation branches would:
// * Decrement `chunk_manifests.ref_count` a SECOND time (the
// trigger having wrongly touched blobs, not the manifest);
// * Undo the trigger's blob decrement (rc > 1 branch);
// * Call `remove_manifest_reference` (rc <= 1 branch), which
// deletes manifest + dereferences chunks — again duplicating
// work the trigger should own.
//
// Migration `20261017000000_file_delete_trigger_manifest_aware.sql`
// rewrote the trigger to be manifest-aware, so it now correctly
// decrements EITHER the manifest OR the blob depending on which
// one the hash names, walks chunks on last-ref manifest delete,
// and leaves the counters in a consistent state without any
// compensation call. Running the old compensation ON TOP of the
// new trigger causes double-decrement / double-delete and is
// exactly what broke `dedup_blob_cleanup.hurl` step 7
// (`ref_count == 1` observed 0 after purging one of two dedup
// uploads).
//
// What remains here: **physical cleanup only**. If the trigger
// brought a LEGACY whole-file blob to ref_count = 0 and no
// manifest still references it (either directly via file_hash or
// indirectly as a chunk in another manifest's chunk_hashes[]),
// reap the DB row and the backend file eagerly. For CDC chunks
// whose ref_count reached 0 via the trigger's last-ref manifest
// path, `dedup_gc` handles physical reap with a grace window
// against re-upload races.
//
// Callers can keep invoking `cleanup_if_orphaned` unconditionally
// — for CDC paths it's a cheap no-op (manifest still exists OR
// the hash never had a blob row), for legacy paths it reaps.
let deleted_blob = sqlx::query_scalar::<_, String>(
"DELETE FROM storage.blobs WHERE hash = $1 AND ref_count <= 0 RETURNING hash",
"DELETE FROM storage.blobs \
WHERE hash = $1 \
AND ref_count <= 0 \
AND NOT EXISTS (SELECT 1 FROM storage.chunk_manifests \
WHERE $1 = ANY(chunk_hashes)) \
RETURNING hash",
)
.bind(hash)
.fetch_optional(self.pool.as_ref())
@@ -1909,7 +2012,7 @@ impl DedupService {
tracing::warn!("cleanup_if_orphaned: disk delete failed for {short}: {e}");
}
self.fire_blob_hooks(hash);
tracing::info!("cleanup_if_orphaned: removed orphaned blob {short}");
tracing::info!("cleanup_if_orphaned: removed orphaned legacy blob {short}");
}
}
@@ -2558,10 +2661,18 @@ impl DedupService {
// A manifest is collectible when:
// • ref_count has been decremented to 0 by cleanup_if_orphaned
// on the single-file-delete service path, OR
// • no `storage.files.blob_hash` references its file_hash
// • NO registered reference source references its file_hash
// (covers bulk-delete paths: user cascade, empty_trash —
// where the PG trigger only touches storage.blobs and the
// per-file cleanup_if_orphaned call is skipped).
//
// The second clause used to name `storage.files` directly. That
// hardcoded "files is the only thing that can reference a manifest",
// so any new referring table (thumbnails via
// storage.content_derived_blobs, …) would see its manifests reaped
// on the next sweep despite a correct ref_count — the first clause
// is false, the second true, and the OR fires. It is now the union
// of every registered source; see docs/plan/derived-blobs.md.
loop {
// Keep the historically cheap DELETE-only shape for the dominant
// no-work sweep. Embedding it in the delete/aggregate/update CTE
@@ -2570,23 +2681,14 @@ impl DedupService {
// update. From two onward, aggregate in-process and issue one UPDATE:
// the measured crossover is already positive at two, while 500 and
// 1,000 manifests improve by 60.03x and 51.16x respectively.
let batch: Vec<(String, Vec<String>, i64)> = sqlx::query_as(
"DELETE FROM storage.chunk_manifests
WHERE ctid = ANY(
SELECT ctid FROM storage.chunk_manifests m
WHERE m.ref_count <= 0
OR NOT EXISTS (
SELECT 1 FROM storage.files f
WHERE f.blob_hash = m.file_hash
)
LIMIT $1
)
RETURNING file_hash, chunk_hashes, total_size",
)
.bind(BATCH_SIZE)
.fetch_all(self.maintenance_pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("Dedup", format!("GC manifests: {e}")))?;
// Assembled once at construction (see `manifest_reap_sql`), not
// per sweep: no string work in the hot path, a stable statement for
// prepared-statement caching, and a byte-for-byte golden test.
let batch: Vec<(String, Vec<String>, i64)> = sqlx::query_as(&self.manifest_reap_sql)
.bind(BATCH_SIZE)
.fetch_all(self.maintenance_pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("Dedup", format!("GC manifests: {e}")))?;
if batch.is_empty() {
break;
@@ -3267,6 +3369,41 @@ impl crate::infrastructure::scheduler::JobHandler for DedupService {
#[cfg(test)]
mod tests {
use super::*;
/// Golden test for the statement `garbage_collect` runs against production
/// data. It is assembled from the registered reference sources rather than
/// written as a literal, so this pins the whole thing byte-for-byte — the
/// point being that a reviewer reads the SQL *here* instead of mentally
/// evaluating the registry.
///
/// If this fails after adding a source, read the diff carefully: the new
/// branch must appear inside the `NOT (...)` group, ORed with the others.
/// A branch landing outside that group inverts the predicate for every
/// other source and reaps live manifests.
#[tokio::test]
async fn manifest_reap_statement_is_stable() {
let sql = DedupService::new_stub().manifest_reap_sql;
let expected = r#"DELETE FROM storage.chunk_manifests
WHERE ctid = ANY(
SELECT ctid
FROM storage.chunk_manifests m
WHERE m.ref_count <= 0
OR NOT (EXISTS (SELECT 1 FROM storage.files cnt_f WHERE cnt_f.blob_hash = m.file_hash))
LIMIT $1
)
RETURNING file_hash, chunk_hashes, total_size"#;
assert_eq!(sql, expected, "reap statement changed:\n{sql}");
}
/// The reap predicate must never match a manifest that some source still
/// references. With an empty registry `NOT (...)` would have no operands,
/// so the builder refuses rather than emitting a statement that deletes
/// every manifest in the database.
#[test]
#[should_panic(expected = "no manifest-level blob reference source")]
fn empty_registry_refuses_to_build_reap_statement() {
let _ = manifest_reap_sql(&BlobReferenceRegistry::new());
}
use std::collections::HashSet;
use tempfile::NamedTempFile;
@@ -0,0 +1,462 @@
//! Reconciles `storage.chunk_manifests.ref_count` against its actual
//! referrers.
//!
//! ### Why this exists
//!
//! There are **two** reference counters, and only one of them was ever
//! verified. `DedupService::add_reference` bumps
//! `chunk_manifests.ref_count` first and only falls back to
//! `storage.blobs.ref_count`, so a reference lands on whichever counter
//! its hash names:
//!
//! * a **chunk** reference → `storage.blobs.ref_count`, reconciled by
//! `blobs_consistency::refcount_mismatch`;
//! * a **Blob** reference (a CDC file, and now every derived artifact) →
//! `chunk_manifests.ref_count`, reconciled by **nothing** before this
//! job existed.
//!
//! That gap was survivable only because `dedup_gc`'s reap predicate had a
//! second clause — "no `storage.files` row references this manifest" —
//! which quietly compensated for drift on the bulk-delete paths where
//! `ref_count` is never decremented. Generalising that clause to the
//! reference registry (so thumbnails stop being reaped) removes the
//! compensation, which is exactly why the manifest counter now has to be
//! checked directly. See `docs/plan/derived-blobs.md`.
//!
//! ### The check
//!
//! * `manifest_refcount_mismatch` (severity `inconsistent`) —
//! `chunk_manifests.ref_count` disagrees with the number of registered
//! referrers. An **under**-count is the dangerous direction: GC reaps a
//! manifest whose content is still reachable, taking its chunks with it.
//! An over-count merely pins storage. Content-safe to report either way
//! — the manifest row and its chunks are intact, the counter is wrong.
//!
//! ### Why a separate job rather than a phase of `blobs_consistency`
//!
//! One subject per job, per the subject-iteration principle the other five
//! consistency tenants follow. It also avoids changing the cursor format of
//! an existing *recoverable* job, which would strand any run paused across
//! the deploy.
use std::sync::Arc;
use async_trait::async_trait;
use sqlx::PgPool;
use crate::application::ports::blob_reference_ports::{BlobReferenceRegistry, RefLevel};
use crate::infrastructure::scheduler::{
JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome,
RunStatus, record_or_log,
};
pub const MANIFESTS_CONSISTENCY_JOB_NAME: &str = "manifests_consistency";
/// Rows per batch. Each row costs one indexed subquery per registered
/// source; 200 matches `blobs_consistency` so the cancel-poll cadence is
/// the same for an operator watching either job.
const BATCH_SIZE: i64 = 200;
/// The page query, with `actual_ref_count` summed from the registered
/// reference sources at [`RefLevel::Manifest`].
///
/// Only sources that reference a **Blob** contribute — `storage.files`
/// today, plus `storage.content_derived_blobs` and
/// `storage.file_attached_blobs` once they exist.
/// `ChunksReferenceSource` returns `None` here: a manifest is never
/// referenced by another manifest, and including it would count this
/// manifest's own chunks as referrers of itself.
///
/// # Panics
///
/// If no source contributes at [`RefLevel::Manifest`] — a wiring bug that
/// would report every manifest as mismatched.
fn manifest_page_sql(registry: &BlobReferenceRegistry) -> String {
let expected = registry.ref_count_expr(RefLevel::Manifest, "m.file_hash");
assert!(
expected != "0",
"no manifest-level blob reference source registered: every manifest \
would appear unreferenced"
);
format!(
"SELECT
m.file_hash AS file_hash,
m.ref_count AS ref_count,
m.total_size AS total_size,
m.chunk_count AS chunk_count,
({expected})::bigint AS actual_ref_count
FROM storage.chunk_manifests m
WHERE ($1::text IS NULL OR m.file_hash > $1)
ORDER BY m.file_hash
LIMIT $2"
)
}
pub struct ManifestsConsistencyCheck {
pool: Arc<PgPool>,
/// Built once from the blob-reference registry so this recompute and
/// `dedup_gc`'s reap predicate answer "what references this manifest"
/// identically. Assembled at construction rather than per page so the
/// sweep runs a fixed statement.
page_sql: String,
}
impl ManifestsConsistencyCheck {
pub fn new(pool: Arc<PgPool>, reference_registry: Arc<BlobReferenceRegistry>) -> Self {
Self {
pool,
page_sql: manifest_page_sql(&reference_registry),
}
}
/// Chainable self-registration. On-demand only — operators fire it
/// from `POST /api/admin/jobs/manifests_consistency/trigger`.
pub async fn register_recoverable_job(
self: Arc<Self>,
registry: &JobRegistry,
provider: &Arc<dyn JobStoreProvider>,
) -> Arc<Self> {
registry
.register_recoverable_job(self.clone(), provider.clone(), None)
.await;
self
}
}
#[derive(Debug, sqlx::FromRow)]
struct ManifestRow {
file_hash: String,
ref_count: i32,
total_size: i64,
chunk_count: i32,
actual_ref_count: i64,
}
#[async_trait]
impl RecoverableJobHandler for ManifestsConsistencyCheck {
fn name(&self) -> &str {
MANIFESTS_CONSISTENCY_JOB_NAME
}
async fn count_total(&self) -> Option<u64> {
let row: Result<(i64,), sqlx::Error> =
sqlx::query_as("SELECT COUNT(*) FROM storage.chunk_manifests")
.fetch_one(self.pool.as_ref())
.await;
match row {
Ok((n,)) => Some(n.max(0) as u64),
Err(e) => {
tracing::debug!(
target: "oxicloud::consistency",
event = "manifests_consistency.count_total_failed",
error = %e,
"count_total failed — run will not surface a progress bar"
);
None
}
}
}
async fn run_resumable(
&self,
store: &dyn JobStore,
args: &JobRunArgs,
resume_cursor: Option<Vec<u8>>,
) -> RunOutcome {
let is_fresh = resume_cursor.is_none();
// Cursor: the last `file_hash` as UTF-8. Same convention as
// `blobs_consistency`, which also pages a hash-keyed table.
let mut cursor: Option<String> = match resume_cursor {
None => None,
Some(bytes) if bytes.is_empty() => None,
Some(bytes) => match String::from_utf8(bytes) {
Ok(s) => Some(s),
Err(e) => {
return RunOutcome::Failed {
message: format!("invalid cursor: not valid UTF-8: {e}"),
};
}
},
};
// Persist the repair flag into `params.repair` so the admin
// run-detail view can display whether the run was a discovery
// scan or an active repair. Fresh takes it from args; Resume
// reads back so a paused repair scan stays a repair scan (a
// mid-scan crash mustn't silently downgrade the remaining
// rows to discovery-only). Same shape as
// `blobs_consistency_service.rs`'s `deep` handling — see the
// reasoning documented there.
let repair = if is_fresh {
let v = if args.repair { "true" } else { "false" };
if let Err(e) = store.set_string_param("repair", v).await {
return RunOutcome::Failed {
message: format!("failed to persist repair flag to params: {e}"),
};
}
args.repair
} else {
match store.get_string_param("repair").await {
Ok(Some(v)) => v == "true",
Ok(None) => false,
Err(e) => {
return RunOutcome::Failed {
message: format!("read `repair` from params: {e}"),
};
}
}
};
if repair {
tracing::info!(
target: "oxicloud::consistency",
event = "manifests_consistency.repair_mode_active",
run_id = %store.run_id(),
"repair mode: manifest_refcount_mismatch findings will trigger corrective UPDATE"
);
}
let mut finding_count = 0u64;
// Only relevant when `repair == true`. Reported inline in
// the completion log + the `extra_stats` payload so operators
// can see "we found N and fixed M" in one line.
let mut repaired_count = 0u64;
loop {
// Cooperative cancel poll between batches.
match store.status().await {
Ok(RunStatus::CancelRequested) => {
tracing::info!(
target: "oxicloud::consistency",
event = "manifests_consistency.cancelled",
run_id = %store.run_id(),
finding_count = finding_count,
"manifests_consistency cancelled cooperatively, pausing"
);
return RunOutcome::Paused {
cursor: cursor
.as_ref()
.map(|s| s.as_bytes().to_vec())
.unwrap_or_default(),
};
}
Ok(_) => {}
Err(e) => {
return RunOutcome::Failed {
message: format!("status poll: {e}"),
};
}
}
let rows: Vec<ManifestRow> = match sqlx::query_as(&self.page_sql)
.bind(cursor.as_deref())
.bind(BATCH_SIZE)
.fetch_all(self.pool.as_ref())
.await
{
Ok(r) => r,
Err(e) => {
return RunOutcome::Failed {
message: format!("batch fetch: {e}"),
};
}
};
if rows.is_empty() {
tracing::info!(
target: "oxicloud::consistency",
event = "manifests_consistency.completed",
run_id = %store.run_id(),
finding_count = finding_count,
repaired_count = repaired_count,
repair_requested = repair,
"manifests_consistency completed with {} finding(s), {} repaired",
finding_count,
repaired_count
);
return RunOutcome::completed_with(serde_json::json!({
"repair_requested": repair,
"repaired_count": repaired_count,
}));
}
for row in &rows {
if row.ref_count as i64 == row.actual_ref_count {
continue;
}
finding_count += 1;
let delta = row.actual_ref_count - row.ref_count as i64;
record_or_log(
store,
MANIFESTS_CONSISTENCY_JOB_NAME,
"manifest_refcount_mismatch",
"inconsistent",
None, // a hash isn't a UUID; the identifier lives in detail
serde_json::json!({
"file_hash": row.file_hash,
"stored": row.ref_count,
"actual": row.actual_ref_count,
"delta": delta,
"total_size": row.total_size,
"chunk_count": row.chunk_count,
// Under-count is the dangerous direction: GC reaps a
// manifest whose content is still reachable.
"reap_risk": delta > 0,
}),
)
.await;
// Repair pass — content-safe corrective UPDATE. The
// stored counter is set to what the auditor formula
// would compute at UPDATE time (subquery matches
// `manifest_page_sql`'s `actual_ref_count` predicate),
// so a concurrent file insert/delete between our page
// fetch and this UPDATE can't leave a stale value —
// the subquery re-reads inside the same statement.
// The `<> (subquery)` guard makes the UPDATE a no-op
// if the value is already correct, so this is
// idempotent under retry.
if repair {
match sqlx::query(
"UPDATE storage.chunk_manifests m \
SET ref_count = ( \
SELECT COUNT(*) FROM storage.files \
WHERE blob_hash = m.file_hash \
) \
WHERE m.file_hash = $1 \
AND m.ref_count <> ( \
SELECT COUNT(*) FROM storage.files \
WHERE blob_hash = m.file_hash \
)",
)
.bind(&row.file_hash)
.execute(self.pool.as_ref())
.await
{
Ok(res) if res.rows_affected() > 0 => {
repaired_count += 1;
tracing::info!(
target: "audit",
event = "manifests_consistency.repaired",
run_id = %store.run_id(),
file_hash = %row.file_hash,
stored_was = row.ref_count,
actual = row.actual_ref_count,
"🩹 manifest ref_count repaired"
);
}
Ok(_) => {
// Row not touched — either another concurrent
// repair fixed it first, or the drift healed
// itself between page fetch and UPDATE.
// Silent no-op.
}
Err(e) => {
tracing::warn!(
target: "oxicloud::consistency",
event = "manifests_consistency.repair_failed",
run_id = %store.run_id(),
file_hash = %row.file_hash,
error = %e,
"manifest ref_count repair UPDATE failed — finding stays"
);
}
}
}
}
// Advance cursor + checkpoint.
let last_hash = rows
.last()
.map(|r| r.file_hash.clone())
.expect("non-empty rows");
cursor = Some(last_hash.clone());
let batch_len = rows.len() as u64;
if let Err(e) = store.checkpoint(last_hash.into_bytes(), batch_len).await {
return RunOutcome::Failed {
message: format!("checkpoint: {e}"),
};
}
if (rows.len() as i64) < BATCH_SIZE {
tracing::info!(
target: "oxicloud::consistency",
event = "manifests_consistency.completed",
run_id = %store.run_id(),
finding_count = finding_count,
repaired_count = repaired_count,
repair_requested = repair,
"manifests_consistency completed with {} finding(s), {} repaired",
finding_count,
repaired_count
);
return RunOutcome::completed_with(serde_json::json!({
"repair_requested": repair,
"repaired_count": repaired_count,
}));
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::infrastructure::repositories::pg::blob_reference_sources::{
ChunksReferenceSource, FilesReferenceSource,
};
fn default_registry() -> BlobReferenceRegistry {
let pool = Arc::new(
sqlx::pool::PoolOptions::<sqlx::Postgres>::new()
.connect_lazy("postgres://invalid/invalid")
.expect("lazy pool never connects"),
);
let mut registry = BlobReferenceRegistry::new();
registry.register(Arc::new(FilesReferenceSource::new(pool.clone())));
registry.register(Arc::new(ChunksReferenceSource::new(pool)));
registry
}
/// Golden test — the statement is assembled from the registry, so pin it
/// byte-for-byte and read the SQL here rather than deriving it mentally.
///
/// Two invariants a future source must not break: the files term carries
/// **no** `NOT EXISTS` guard (that guard exists to keep CDC rows out of
/// the *chunk* level; applying it here would count nothing), and
/// `chunk_hashes` appears nowhere — a manifest citing its own chunks is
/// not a referrer of itself.
#[tokio::test]
async fn manifest_page_statement_is_stable() {
let sql = manifest_page_sql(&default_registry());
let expected = r#"SELECT
m.file_hash AS file_hash,
m.ref_count AS ref_count,
m.total_size AS total_size,
m.chunk_count AS chunk_count,
((SELECT COUNT(*) FROM storage.files cnt_f
WHERE cnt_f.blob_hash = m.file_hash))::bigint AS actual_ref_count
FROM storage.chunk_manifests m
WHERE ($1::text IS NULL OR m.file_hash > $1)
ORDER BY m.file_hash
LIMIT $2"#;
assert_eq!(sql, expected, "manifest page statement changed:\n{sql}");
}
#[tokio::test]
async fn chunks_source_contributes_nothing_at_manifest_level() {
let sql = manifest_page_sql(&default_registry());
assert!(
!sql.contains("chunk_hashes"),
"a manifest must not count its own chunks as referrers: {sql}"
);
}
#[test]
#[should_panic(expected = "no manifest-level blob reference source")]
fn empty_registry_refuses_to_build_page_statement() {
let _ = manifest_page_sql(&BlobReferenceRegistry::new());
}
}
+1
View File
@@ -31,6 +31,7 @@ pub mod last_seen_tracker;
pub mod local_blob_backend;
pub mod local_fs_mount_provider;
pub mod login_lockout_service;
pub mod manifests_consistency_service;
pub mod media_metadata_service;
pub mod mock_email_sender;
pub mod mount_provider_factory;
+12 -1
View File
@@ -2575,6 +2575,12 @@ pub async fn list_jobs(State(state): State<Arc<AppState>>) -> impl IntoResponse
/// `deep=true` opts into slow variants — `consistency_batch` fans it
/// out to sub-jobs; `storage_consistency` (when implemented) will
/// re-BLAKE3 each blob for bitrot detection. See `JobRunArgs.deep`.
///
/// `repair=true` opts into corrective action on the refcount
/// consistency tenants (`blobs_consistency`, `manifests_consistency`,
/// and `consistency_batch` which fans out to both). Default `false`
/// preserves discovery-only. See `JobRunArgs.repair` for the
/// content-safety and race-safety guarantees.
#[derive(serde::Deserialize)]
pub struct TriggerJobQuery {
#[serde(default)]
@@ -2591,6 +2597,8 @@ pub struct TriggerJobQuery {
/// `AppConfig.storage_entries`.
#[serde(default)]
pub storage: Option<String>,
#[serde(default)]
pub repair: bool,
}
/// `POST /api/admin/jobs/{name}/trigger` — dispatch one run off-schedule.
@@ -2629,15 +2637,18 @@ pub async fn trigger_job(
job = %name,
force = query.force,
deep = query.deep,
"👮🏻‍♂️ Admin triggered job {} (force={}, deep={})",
repair = query.repair,
"👮🏻‍♂️ Admin triggered job {} (force={}, deep={}, repair={})",
name,
query.force,
query.deep,
query.repair,
);
let args = JobRunArgs {
force: query.force,
deep: query.deep,
storage: query.storage.clone(),
repair: query.repair,
};
// Jobs that can run for hours (backend_migration, future
+17 -11
View File
@@ -225,9 +225,10 @@ jsonpath "$.outcome.count" exists
# Step 4c — Trigger `consistency_batch`. Coordinator (plain
# JobHandler) — snapshots the registry, filters names
# ending `_consistency`, sequentially triggers each.
# `outcome.count` = number of children dispatched (5 as
# of Slice 10: drives + folders + files + blobs +
# backend). `extra.per_check` carries a per-child outcome
# `outcome.count` = number of children dispatched (6 as
# of the refcount_cascade fix: drives + folders +
# files + blobs + manifests + backend). `extra.per_check`
# carries a per-child outcome
# map. Batch itself always returns ok — child failures
# live inside per_check. `?deep=true` propagates as
# `extra.deep`.
@@ -239,16 +240,21 @@ HTTP 200
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
jsonpath "$.outcome.count" == 5
jsonpath "$.outcome.count" == 6
jsonpath "$.outcome.extra.deep" == true
jsonpath "$.outcome.extra.ok" == 5
jsonpath "$.outcome.extra.ok" == 6
jsonpath "$.outcome.extra.err" == 0
# per_check is keyed by child job name.
jsonpath "$.outcome.extra.per_check.drives_consistency.outcome" == "ok"
jsonpath "$.outcome.extra.per_check.folders_consistency.outcome" == "ok"
jsonpath "$.outcome.extra.per_check.files_consistency.outcome" == "ok"
jsonpath "$.outcome.extra.per_check.blobs_consistency.outcome" == "ok"
jsonpath "$.outcome.extra.per_check.backend_consistency.outcome" == "ok"
# per_check is keyed by child job name. `manifests_consistency` was added
# by the refcount_cascade fix — see docs/plan/derived-blobs.md and
# `[[bug_dual_refcount_divergence]]` for why the second counter needed
# its own tenant. Auto-picked by `consistency_batch` via `.ends_with(
# "_consistency")` (no explicit list in the batch service).
jsonpath "$.outcome.extra.per_check.drives_consistency.outcome" == "ok"
jsonpath "$.outcome.extra.per_check.folders_consistency.outcome" == "ok"
jsonpath "$.outcome.extra.per_check.files_consistency.outcome" == "ok"
jsonpath "$.outcome.extra.per_check.blobs_consistency.outcome" == "ok"
jsonpath "$.outcome.extra.per_check.manifests_consistency.outcome" == "ok"
jsonpath "$.outcome.extra.per_check.backend_consistency.outcome" == "ok"
# ─────────────────────────────────────────────────────────────
+614
View File
@@ -0,0 +1,614 @@
# =============================================================
# OxiCloud – Copy-folder ref_count regression
# =============================================================
# Regression test for the ref_count drift found on Ed's sandbox
# (2026-08-22) and traced to the copy-folder path. When a folder
# is copied, every file inside gets duplicated as a NEW file row
# pointing at the SAME blob(s) — dedup wins bytes-on-disk, but
# `storage.blobs.ref_count` MUST bump by the number of new refs.
# If it doesn't, dedup GC will reap a blob that a live file row
# still references → dangling reference → user gets 404 on
# download of the copied file.
#
# Covers two paths so the CDC boundary can't hide a regression:
#
# 1. Small file (`refcount-cascade-small.txt`) → single legacy whole-
# file blob. `storage.blobs.ref_count` counted directly on
# the file's content hash.
# 2. 2 MB file (`refcount-cascade-cdc.bin`) → FastCDC produces multiple
# distinct chunks. Whole-file `content_hash` still resolves
# through the dedup API.
#
# Both fixtures are DEDICATED — unique content so ref_count
# assertions are absolute (== 1, == 2). Do NOT reuse these
# fixtures in other hurl files or absolute assertions here will
# flake.
#
# Deletion is TWO STEPS in OxiCloud:
# `DELETE /api/folders/{id}` → moves to trash (ref_count
# unchanged; children still
# reference the blob).
# `DELETE /api/trash/{id}` → permanent purge; NOW
# ref_count decrements. If it
# hits 0, blob row is deleted
# synchronously (`exists=false`).
# So the test purges trash after every folder-delete step —
# skipping that would make the assertions wrong regardless of
# whether the copy-side bug is present.
#
# `/api/dedup/check/{hash}` returns `ref_count` only for admin
# callers (regular users get `null` for anti-enumeration). This
# suite requires the admin token; setup.hurl seeds it.
#
# Run:
# hurl --variables-file tests/api/test.env --test \
# tests/api/refcount_cascade.hurl
# =============================================================
# ─────────────────────────────────────────────────────────────
# Step 1 — Login (admin, required for ref_count in dedup API)
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/auth/login
Content-Type: application/json
{
"username": "{{username}}",
"password": "{{password}}"
}
HTTP 200
[Captures]
token: jsonpath "$.access_token"
# ─────────────────────────────────────────────────────────────
# Step 2 — Baseline sweeps: run BOTH consistency tenants that
# check ref_count invariants and capture their finding
# counts. Later checkpoints assert equality with these
# baselines instead of `== 0` — so stale findings from
# previous tests don't flunk this one; only NEW drift
# introduced by our copy/delete does.
#
# Two tenants because there are two counters (see
# `[[bug_dual_refcount_divergence]]`):
#
# - `blobs_consistency` — reconciles
# `storage.blobs.ref_count` against its auditor
# formula. Catches chunk-level drift.
# - `manifests_consistency` — reconciles
# `storage.chunk_manifests.ref_count` against its
# auditor formula. Catches whole-file drift (the
# path the FE + `/api/dedup/check` surface reads).
#
# Trigger returns `outcome.count = stats.finding_count`
# for recoverable tenants (see
# `scheduler/recoverable.rs::JobOutcome::ok_with` in
# the Completed branch). `outcome.outcome == "ok"`
# means the run walked the whole subject; it does NOT
# mean zero findings.
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/admin/jobs/blobs_consistency/trigger
Authorization: Bearer {{token}}
HTTP 200
[Captures]
baseline_blobs_findings: jsonpath "$.outcome.count"
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
POST {{base_url}}/api/admin/jobs/manifests_consistency/trigger
Authorization: Bearer {{token}}
HTTP 200
[Captures]
baseline_manifests_findings: jsonpath "$.outcome.count"
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
# =============================================================
# Scenario A — small file (single legacy whole-file blob)
# =============================================================
# ─────────────────────────────────────────────────────────────
# A1 — Create source folder
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/folders
Authorization: Bearer {{token}}
Content-Type: application/json
{
"name": "hurl-ref-source-small"
}
HTTP 201
[Captures]
src_small_id: jsonpath "$.id"
# ─────────────────────────────────────────────────────────────
# A2 — Upload the small fixture. Content is unique to this
# test, so ref_count starts at exactly 1 (no dedup
# collision with any other fixture in the suite).
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/files/upload
Authorization: Bearer {{token}}
[MultipartFormData]
folder_id: {{src_small_id}}
file: file,fixtures/refcount-cascade-small.txt; text/plain
HTTP 201
[Captures]
small_file_id: jsonpath "$.id"
[Asserts]
jsonpath "$.content_hash" == "2d8eb13178cff0036a22e0c3c42446061e86579f9d59ea73c8343bccc2df0fd3"
# ─────────────────────────────────────────────────────────────
# A3 — Baseline: exactly one live reference.
# ─────────────────────────────────────────────────────────────
GET {{base_url}}/api/dedup/check/2d8eb13178cff0036a22e0c3c42446061e86579f9d59ea73c8343bccc2df0fd3
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.exists" == true
jsonpath "$.ref_count" == 1
# ─────────────────────────────────────────────────────────────
# A4 — Create target folder + copy source into it. The batch
# endpoint is the single entry point every FE / WebDAV
# code path funnels through.
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/folders
Authorization: Bearer {{token}}
Content-Type: application/json
{
"name": "hurl-ref-target-small"
}
HTTP 201
[Captures]
tgt_small_id: jsonpath "$.id"
POST {{base_url}}/api/batch/folders/copy
Authorization: Bearer {{token}}
Content-Type: application/json
{
"folder_ids": ["{{src_small_id}}"],
"target_folder_id": "{{tgt_small_id}}"
}
HTTP 200
[Captures]
copy_small_root: jsonpath "$.successful[0].new_root_folder_id"
[Asserts]
jsonpath "$.stats.successful" == 1
jsonpath "$.stats.failed" == 0
jsonpath "$.successful[0].files_copied" == 1
# ─────────────────────────────────────────────────────────────
# A5 — COPY-SIDE ASSERTION: ref_count must be exactly 2. The
# pre-fix bug left this at 1 — a subsequent GC would then
# reap the blob out from under the copied file.
# ─────────────────────────────────────────────────────────────
GET {{base_url}}/api/dedup/check/2d8eb13178cff0036a22e0c3c42446061e86579f9d59ea73c8343bccc2df0fd3
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.exists" == true
jsonpath "$.ref_count" == 2
# ─────────────────────────────────────────────────────────────
# A5b — Sweep both consistency tenants that check ref_count.
# Complements the single-hash probe above: if the copy
# path miscounted some OTHER blob or manifest, the single-
# hash probe wouldn't catch it. Delta vs the baselines
# isolates NEW drift from ambient.
#
# Both tenants required — one counter each; see
# [[bug_dual_refcount_divergence]] for why the copy path
# must maintain both symmetrically.
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/admin/jobs/blobs_consistency/trigger
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
jsonpath "$.outcome.count" == {{baseline_blobs_findings}}
POST {{base_url}}/api/admin/jobs/manifests_consistency/trigger
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
jsonpath "$.outcome.count" == {{baseline_manifests_findings}}
# ─────────────────────────────────────────────────────────────
# A6 — Soft-delete the copied folder tree (moves to trash).
# ref_count is EXPECTED to stay at 2 — trashed files
# still reference the blob per the `NOT is_trashed` gap
# the auditor deliberately closed (see
# `[[project_by_hash_drop_is_trashed_filter]]`). Asserting
# == 2 here makes the trash-vs-permanent boundary explicit
# so a future refactor that changed the semantics would
# surface at THIS line, not several steps downstream.
# ─────────────────────────────────────────────────────────────
DELETE {{base_url}}/api/folders/{{copy_small_root}}
Authorization: Bearer {{token}}
HTTP 204
GET {{base_url}}/api/dedup/check/2d8eb13178cff0036a22e0c3c42446061e86579f9d59ea73c8343bccc2df0fd3
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.exists" == true
jsonpath "$.ref_count" == 2
# ─────────────────────────────────────────────────────────────
# A7 — Purge the copy folder permanently.
# `DELETE /api/trash/{id}` accepts the ORIGINAL resource
# id as the path param (verified in `trash_handler.rs::
# delete_permanently`) — no need to GET+filter the trash
# listing to translate. If soft-delete silently failed,
# the ref_count assertion two lines below catches it.
# ref_count must drop to exactly 1 (the source folder
# still holds its file). Guards the DECREMENT half of
# the invariant: a double-decrement here would go to 0
# and the next GC wipes the still-live original's blob.
# ─────────────────────────────────────────────────────────────
DELETE {{base_url}}/api/trash/{{copy_small_root}}
Authorization: Bearer {{token}}
HTTP 200
GET {{base_url}}/api/dedup/check/2d8eb13178cff0036a22e0c3c42446061e86579f9d59ea73c8343bccc2df0fd3
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.exists" == true
jsonpath "$.ref_count" == 1
# ─────────────────────────────────────────────────────────────
# A8 — Soft-delete + purge the SOURCE folder (last holder).
# ref_count hits 0 → blob row deleted synchronously →
# `exists == false` on the next probe. Mirror pattern to
# `dedup_blob_cleanup.hurl` step 10, applied to folder-
# scoped delete.
# ─────────────────────────────────────────────────────────────
DELETE {{base_url}}/api/folders/{{src_small_id}}
Authorization: Bearer {{token}}
HTTP 204
DELETE {{base_url}}/api/trash/{{src_small_id}}
Authorization: Bearer {{token}}
HTTP 200
GET {{base_url}}/api/dedup/check/2d8eb13178cff0036a22e0c3c42446061e86579f9d59ea73c8343bccc2df0fd3
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.exists" == false
# ─────────────────────────────────────────────────────────────
# A9 — End-of-Scenario-A sweep on BOTH tenants. Scenario A
# introduced two file rows (source + copy), then deleted
# both. Net effect on the DB is zero — so the drift count
# on each counter must be exactly the baseline.
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/admin/jobs/blobs_consistency/trigger
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
jsonpath "$.outcome.count" == {{baseline_blobs_findings}}
POST {{base_url}}/api/admin/jobs/manifests_consistency/trigger
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
jsonpath "$.outcome.count" == {{baseline_manifests_findings}}
# =============================================================
# Scenario B — 2 MB multi-chunk file (CDC manifest path)
#
# Coverage: this scenario now sweeps BOTH `blobs_consistency`
# (chunk-level ref_counts on `storage.blobs.ref_count`) and
# `manifests_consistency` (whole-file ref_counts on
# `storage.chunk_manifests.ref_count`). Same-shape assertions
# as Scenario A — see the baseline capture block near the top
# of this file and [[bug_dual_refcount_divergence]] for why
# both are needed.
# =============================================================
# ─────────────────────────────────────────────────────────────
# B1 — Source folder for the multi-chunk scenario. Isolated
# from Scenario A so deletion order can't mask a bug (e.g.
# a shared blob whose counter goes negative).
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/folders
Authorization: Bearer {{token}}
Content-Type: application/json
{
"name": "hurl-ref-source-cdc"
}
HTTP 201
[Captures]
src_cdc_id: jsonpath "$.id"
# ─────────────────────────────────────────────────────────────
# B2 — Upload the 2 MB dedicated fixture. Deterministic per-
# position content → FastCDC produces multiple distinct
# chunks (no chunk-level dedup with any other fixture).
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/files/upload
Authorization: Bearer {{token}}
[MultipartFormData]
folder_id: {{src_cdc_id}}
file: file,fixtures/refcount-cascade-cdc.bin; application/octet-stream
HTTP 201
[Captures]
cdc_file_id: jsonpath "$.id"
[Asserts]
jsonpath "$.content_hash" == "fb1e63c28bb792e0f69cd16cd7595989f83c218cf70894e07d1f811ab1dc6f83"
# ─────────────────────────────────────────────────────────────
# B3 — Baseline. If exists=false here, the whole-file hash
# isn't registered in `storage.blobs` for CDC uploads on
# this build — swap to a chunk-hash probe or a
# blobs_consistency-driven assertion. See
# `docs/plan/recovery.md`.
# ─────────────────────────────────────────────────────────────
GET {{base_url}}/api/dedup/check/fb1e63c28bb792e0f69cd16cd7595989f83c218cf70894e07d1f811ab1dc6f83
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.exists" == true
jsonpath "$.ref_count" == 1
# ─────────────────────────────────────────────────────────────
# B4 — Target folder + folder copy.
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/folders
Authorization: Bearer {{token}}
Content-Type: application/json
{
"name": "hurl-ref-target-cdc"
}
HTTP 201
[Captures]
tgt_cdc_id: jsonpath "$.id"
POST {{base_url}}/api/batch/folders/copy
Authorization: Bearer {{token}}
Content-Type: application/json
{
"folder_ids": ["{{src_cdc_id}}"],
"target_folder_id": "{{tgt_cdc_id}}"
}
HTTP 200
[Captures]
copy_cdc_root: jsonpath "$.successful[0].new_root_folder_id"
[Asserts]
jsonpath "$.stats.successful" == 1
jsonpath "$.successful[0].files_copied" == 1
# ─────────────────────────────────────────────────────────────
# B5 — CDC copy-side assertion: ref_count == 2.
# ─────────────────────────────────────────────────────────────
GET {{base_url}}/api/dedup/check/fb1e63c28bb792e0f69cd16cd7595989f83c218cf70894e07d1f811ab1dc6f83
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.exists" == true
jsonpath "$.ref_count" == 2
# ─────────────────────────────────────────────────────────────
# B5b — Full-DB sweep on BOTH tenants after CDC copy. Multi-
# chunk path exercises the manifest side of the invariant
# — a bug that skips one chunk out of N would leak that
# chunk without touching the whole-file assertion above.
# `blobs_consistency` catches chunk-level drift;
# `manifests_consistency` catches whole-file drift.
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/admin/jobs/blobs_consistency/trigger
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
jsonpath "$.outcome.count" == {{baseline_blobs_findings}}
POST {{base_url}}/api/admin/jobs/manifests_consistency/trigger
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
jsonpath "$.outcome.count" == {{baseline_manifests_findings}}
# ─────────────────────────────────────────────────────────────
# B6 — Soft-delete copy: ref_count stays at 2.
# ─────────────────────────────────────────────────────────────
DELETE {{base_url}}/api/folders/{{copy_cdc_root}}
Authorization: Bearer {{token}}
HTTP 204
GET {{base_url}}/api/dedup/check/fb1e63c28bb792e0f69cd16cd7595989f83c218cf70894e07d1f811ab1dc6f83
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.exists" == true
jsonpath "$.ref_count" == 2
# ─────────────────────────────────────────────────────────────
# B7 — Purge copy from trash directly by folder id: ref_count → 1.
# ─────────────────────────────────────────────────────────────
DELETE {{base_url}}/api/trash/{{copy_cdc_root}}
Authorization: Bearer {{token}}
HTTP 200
GET {{base_url}}/api/dedup/check/fb1e63c28bb792e0f69cd16cd7595989f83c218cf70894e07d1f811ab1dc6f83
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.exists" == true
jsonpath "$.ref_count" == 1
# ─────────────────────────────────────────────────────────────
# B8 — Soft-delete + purge SOURCE: ref_count → 0, blob purged.
# ─────────────────────────────────────────────────────────────
DELETE {{base_url}}/api/folders/{{src_cdc_id}}
Authorization: Bearer {{token}}
HTTP 204
DELETE {{base_url}}/api/trash/{{src_cdc_id}}
Authorization: Bearer {{token}}
HTTP 200
GET {{base_url}}/api/dedup/check/fb1e63c28bb792e0f69cd16cd7595989f83c218cf70894e07d1f811ab1dc6f83
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.exists" == false
# ─────────────────────────────────────────────────────────────
# B9 — End-of-Scenario-B sweep on BOTH tenants. All Scenario B
# rows gone; both counter drift counts must be back at
# baseline.
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/admin/jobs/blobs_consistency/trigger
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
jsonpath "$.outcome.count" == {{baseline_blobs_findings}}
POST {{base_url}}/api/admin/jobs/manifests_consistency/trigger
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
jsonpath "$.outcome.count" == {{baseline_manifests_findings}}
# =============================================================
# Cleanup — soft-delete + purge the two empty target folders so
# the run leaves nothing behind. Same direct-by-id pattern as
# above; no trash-listing filter needed.
# =============================================================
DELETE {{base_url}}/api/folders/{{tgt_small_id}}
Authorization: Bearer {{token}}
HTTP 204
DELETE {{base_url}}/api/trash/{{tgt_small_id}}
Authorization: Bearer {{token}}
HTTP 200
DELETE {{base_url}}/api/folders/{{tgt_cdc_id}}
Authorization: Bearer {{token}}
HTTP 204
DELETE {{base_url}}/api/trash/{{tgt_cdc_id}}
Authorization: Bearer {{token}}
HTTP 200
# ─────────────────────────────────────────────────────────────
# Final — force `dedup_gc` synchronously so orphaned manifests +
# blobs actually get reaped and their Rust blob-lifecycle hooks
# fire (which is what deletes disk thumbnails / face embeddings
# / audio tags keyed by the whole-file hash).
#
# The trigger `trg_files_decrement_blob_ref` deliberately only
# adjusts counters (see migration `20261017000000_file_delete_
# trigger_manifest_aware.sql`) — a SQL trigger can't invoke Rust
# callbacks. Physical cleanup + hook firing lives in `dedup_gc`
# Phase 1 (see `dedup_service.rs:2660-2772`), which picks up
# manifests at ref_count <= 0 and calls
# `fire_blob_hooks(file_hash)` per reap.
#
# Without this trigger the test would technically pass (the
# ref_count assertions all hold; the `exists == false` checks
# are user-scoped and don't need the DB row gone), but
# `storage_cleanup_check.sh` running after us would then find
# 6 orphan thumbnails on disk and fail the whole api-test run.
# Making the test self-contained keeps the diagnostic tight —
# if orphans remain after this trigger, the bug is in GC or
# hooks, not in our cleanup order.
#
# `?force=true` bypasses the orphan-grace window (safe: this
# test has no concurrent uploader that could race the reap).
# ─────────────────────────────────────────────────────────────
POST {{base_url}}/api/admin/jobs/dedup_gc/trigger?force=true
Authorization: Bearer {{token}}
HTTP 200
[Asserts]
jsonpath "$.ok" == true
jsonpath "$.outcome.outcome" == "ok"
+141
View File
@@ -0,0 +1,141 @@
#!/usr/bin/env bash
# =============================================================
# refcount_cascade.hurl — post-failure diagnostic
# =============================================================
# When `refcount_cascade.hurl` asserts a specific
# `ref_count` value and the API returns something else, this
# script inspects the two DB tables the API surface consults
# to distinguish which side is broken:
#
# 1. `storage.chunk_manifests.ref_count` — queried FIRST by
# `dedup_service::get_blob_metadata` (`dedup_service.rs`
# :1543-1552). If a manifest row exists for the hash,
# the API returns THIS ref_count.
# 2. `storage.blobs.ref_count` — legacy whole-file fallback,
# returned only when NO manifest row exists.
#
# Small files (< CDC min chunk size) still get a manifest row
# — one degenerate chunk with `chunk_hashes = [file_hash]` —
# so BOTH tables carry a ref_count for the same hash. When the
# copy or purge path only updates one of the two, the counters
# diverge.
#
# The `actual_auditor` column is the source of truth: the
# `blobs_consistency` auditor formula counting live references
# from `storage.files` + `chunk_manifests.chunk_hashes[]`
# (`blobs_consistency_service.rs:395-408`). Both stored values
# should equal this.
#
# Bug interpretation matrix (S=stored, M=manifest, A=auditor):
#
# S == M == A → consistent (test bug, unlikely)
# S < A and M == A → blob decrement over-fires
# S == A and M > A → manifest decrement missed
# S > A and M > A → decrement missed both sides
# S < A and M < A → double-decrement both sides
# S > A and M == A → increment missed on blob side
# S == A and M < A → increment missed on manifest
# (S=0, M=2, A=1) [seen 8/23]→ blob double-decrement +
# manifest never decremented
#
# The 2026-08-22 sandbox drift showed the raw shape
# (`stored=0, actual=1`) that this diagnostic now separates
# per-table.
#
# Called from `run.sh` immediately on hurl failure — see the
# dedicated `if ! hurl …; then bash …_diag.sh; fi` block.
# =============================================================
set -uo pipefail
REPO_ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
# Same connection string every other test-time script uses.
export PGPASSWORD=oxicloud_test
PSQL=(psql -h 127.0.0.1 -p 5433 -U oxicloud_test -d oxicloud_test
--set ON_ERROR_STOP=1 --pset pager=off)
SMALL_HASH='2d8eb13178cff0036a22e0c3c42446061e86579f9d59ea73c8343bccc2df0fd3'
CDC_HASH='fb1e63c28bb792e0f69cd16cd7595989f83c218cf70894e07d1f811ab1dc6f83'
log() { echo "[ref_count-diag] $*"; }
log "─────────────────────────────────────────────────────────"
log "refcount_cascade.hurl failed — running diagnostic."
log "Shows blob.ref_count AND manifest.ref_count side by side —"
log "the API queries manifest first (dedup_service.rs:1543), so"
log "if the two diverge, the API surface + auditor + on-disk"
log "state all report different numbers. See script header."
log "─────────────────────────────────────────────────────────"
# Full picture for each fixture hash — one row per hash.
# LEFT JOINs so a hash present in only one table still surfaces
# (the other column comes back NULL, which is itself diagnostic).
"${PSQL[@]}" <<SQL
SELECT
h.hash,
b.ref_count AS blob_stored,
m.ref_count AS manifest_stored,
(
(SELECT COUNT(*) FROM storage.files f
WHERE f.blob_hash = h.hash
AND NOT EXISTS (
SELECT 1 FROM storage.chunk_manifests mm
WHERE mm.file_hash = f.blob_hash
))
+ (SELECT COUNT(*) FROM storage.chunk_manifests mm
WHERE h.hash = ANY(mm.chunk_hashes))
) AS actual_auditor,
(SELECT COUNT(*) FROM storage.files f
WHERE f.blob_hash = h.hash) AS all_files_rows,
(SELECT COUNT(*) FROM storage.files f
WHERE f.blob_hash = h.hash AND f.is_trashed = TRUE)
AS trashed_files,
b.size AS blob_size,
m.total_size AS manifest_size
FROM (VALUES ('$SMALL_HASH'), ('$CDC_HASH')) AS h(hash)
LEFT JOIN storage.blobs b ON b.hash = h.hash
LEFT JOIN storage.chunk_manifests m ON m.file_hash = h.hash;
SQL
psql_status=$?
if [[ $psql_status -ne 0 ]]; then
log "psql query failed (exit $psql_status) — DB may already be torn down"
exit 0 # don't mask the hurl failure with a psql error
fi
# Show the offending file rows too — helps confirm whether the
# copy's file row exists and where (source folder vs copy).
log "─────────────────────────────────────────────────────────"
log "File rows referencing the fixture hashes (name → folder id)"
log "─────────────────────────────────────────────────────────"
"${PSQL[@]}" <<SQL
SELECT
f.blob_hash,
f.name,
f.folder_id,
f.is_trashed,
f.trashed_at,
f.created_at
FROM storage.files f
WHERE f.blob_hash IN ('$SMALL_HASH', '$CDC_HASH')
ORDER BY f.blob_hash, f.created_at;
SQL
log "─────────────────────────────────────────────────────────"
log "Interpretation (compare blob_stored, manifest_stored, actual_auditor):"
log " all three equal → consistent (test bug, unlikely)"
log " blob=A, manifest>A → manifest decrement missed"
log " blob<A, manifest=A → blob decrement over-fires"
log " blob<A, manifest>A → double-decrement on blob +"
log " no-decrement on manifest"
log " (matches the 2026-08-23 case:"
log " blob=0, manifest=2, actual=1)"
log " blob>A, manifest=A → blob increment missed"
log " blob=A, manifest<A → manifest increment missed"
log " Either column NULL → row absent from that table"
log ""
log "The API surface (GET /api/dedup/check/{hash}) queries manifest"
log "FIRST — that's why hurl saw manifest_stored while auditor +"
log "on-disk state ran off blob_stored."
log "─────────────────────────────────────────────────────────"
+16
View File
@@ -224,6 +224,22 @@ hurl --variables-file "$API_DIR/test.env" --file-root "$REPO_ROOT/tests" --test
#bash "$API_DIR/dedup_bulk_upload.sh"
# ── 4b. copy-folder ref_count regression — dedicated block ──────────────
# Ref_count mismatches surface as "expected 1 got 2" at hurl-assert
# level, which doesn't tell you WHICH half of the invariant broke
# (cascade delete missed rows vs decrement hook didn't fire). The
# `_diag.sh` script inspects `storage.blobs.ref_count` and the
# auditor's `actual_ref_count` formula on the two fixture hashes to
# pin the mode. Extracted from the main hurl array so `set -e`
# doesn't skip the diagnostic on failure — the `if !` guard runs
# the diag first, THEN exits with hurl's failure code so CI still
# reports the regression.
if ! hurl --variables-file "$API_DIR/test.env" --file-root "$REPO_ROOT/tests" --test --jobs 1 \
"$API_DIR/refcount_cascade.hurl"; then
bash "$API_DIR/refcount_cascade_diag.sh" || true
exit 1
fi
bash "$API_DIR/storage_cleanup_check.sh"
# ── 5. OPAQUE crypto handshake — the parts Hurl can't drive ─────────────
Binary file not shown.
+1
View File
@@ -0,0 +1 @@
copy-folder-ref-count regression fixture — do not reuse in other tests (hash must stay unique so ref_count assertions are absolute).