fix(file_attached): doe not increment ref_count if new attachement has same hash
- do not increment ref_count if new attachement to a file with same data - add audit log to help identifying other future issue in ref_count - prevent race condition while attaching a blob
This commit is contained in:
@@ -72,6 +72,86 @@ pub const CDC_AVG_CHUNK: usize = 262_144;
|
|||||||
/// Maximum CDC chunk size (1 MB).
|
/// Maximum CDC chunk size (1 MB).
|
||||||
pub const CDC_MAX_CHUNK: usize = 1_048_576;
|
pub const CDC_MAX_CHUNK: usize = 1_048_576;
|
||||||
|
|
||||||
|
// ════════════════════════════════════════════════════════════════════════════
|
||||||
|
// ref_count audit log — attribution trail for ref_count changes
|
||||||
|
//
|
||||||
|
// Emitted at every semantic ref-count mutation site so an unexplained drift
|
||||||
|
// (`manifests_consistency` / `blobs_consistency` finding) can be traced back
|
||||||
|
// to its calling function within one log query. See
|
||||||
|
// `docs/plan/refcount-audit.md` for design rationale, retention model, and
|
||||||
|
// upgrade path to a DB-backed table if log retention proves insufficient.
|
||||||
|
//
|
||||||
|
// Enable at runtime with `RUST_LOG=oxicloud::refcount=info`. Off by default;
|
||||||
|
// info-level so a healthy prod deployment doesn't spam the log stream.
|
||||||
|
// ════════════════════════════════════════════════════════════════════════════
|
||||||
|
|
||||||
|
/// Table names the audit stream uses. Constants (not free strings) so grep
|
||||||
|
/// across a log stream matches a fixed vocabulary and a typo in a call site
|
||||||
|
/// fails to compile instead of silently drifting.
|
||||||
|
mod refcount_audit_table {
|
||||||
|
pub const CHUNK_MANIFESTS: &str = "chunk_manifests";
|
||||||
|
pub const BLOBS: &str = "blobs";
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Source labels for the audit stream — one stable string per Rust function
|
||||||
|
/// that mutates a ref_count. Kept as a module so `grep source=` on the log
|
||||||
|
/// stream shows a fixed enumeration; a rename here is intentional, a rename
|
||||||
|
/// at a call site alone doesn't compile.
|
||||||
|
///
|
||||||
|
/// See `docs/plan/refcount-audit.md § Callsites to instrument` for the full
|
||||||
|
/// list. New callers add a new constant here; adding one string at the call
|
||||||
|
/// site alone is discouraged (breaks the "closed vocabulary" property).
|
||||||
|
mod refcount_audit_source {
|
||||||
|
pub const STORE_FROM_STREAM_NEW_MANIFEST: &str = "store_from_stream.new_manifest";
|
||||||
|
pub const BUMP_MANIFEST_IF_EXISTS: &str = "bump_manifest_if_exists";
|
||||||
|
pub const ADD_REFERENCE_MANIFEST: &str = "add_reference.manifest";
|
||||||
|
pub const ADD_REFERENCE_LEGACY: &str = "add_reference.legacy";
|
||||||
|
pub const REMOVE_MANIFEST_REFERENCE_DECREMENT: &str = "remove_manifest_reference.decrement";
|
||||||
|
pub const REMOVE_MANIFEST_REFERENCE_DELETE: &str = "remove_manifest_reference.delete";
|
||||||
|
pub const REMOVE_LEGACY_REFERENCE: &str = "remove_legacy_reference";
|
||||||
|
pub const STORE_ATTACHED_BLOB_SAME_CONTENT_BALANCE: &str =
|
||||||
|
"store_attached_blob.same_content_balance";
|
||||||
|
pub const STORE_ATTACHED_BLOB_REPLACE_RELEASE: &str = "store_attached_blob.replace_release";
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Outcome of [`DedupService::store_attached_blob_if_absent`]. Split
|
||||||
|
/// so the caller (`thumb_attached_import_service`) can bump its
|
||||||
|
/// `imported` vs `already` counters without a second query.
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub enum AttachedBlobInsertOutcome {
|
||||||
|
/// We won the atomic INSERT — the row now points at `hash`.
|
||||||
|
Inserted { hash: String },
|
||||||
|
/// A row already existed when the atomic INSERT ran (someone else
|
||||||
|
/// won, or the migration was re-triggered). `existing_hash` is
|
||||||
|
/// the row's current `blob_hash` as read moments after the DO
|
||||||
|
/// NOTHING resolved — useful for the import service's
|
||||||
|
/// verify-and-unlink readback.
|
||||||
|
AlreadyPresent { existing_hash: String },
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Emit a single audit line for a ref_count change. Called AFTER the SQL
|
||||||
|
/// UPDATE / INSERT / DELETE returns Ok, so a rolled-back transaction won't
|
||||||
|
/// leave a phantom log line (the SQL error path returns before this call).
|
||||||
|
///
|
||||||
|
/// `delta` is the signed change (`+1` on increment, `-1` on decrement,
|
||||||
|
/// `-old_count` when the row is deleted at its last reference — the
|
||||||
|
/// convention is "resulting ref_count is 0 for reads afterward").
|
||||||
|
///
|
||||||
|
/// The tracing span inherits request-scope context (request_id, caller_id
|
||||||
|
/// from auth middleware, job run id from scheduler) automatically, so
|
||||||
|
/// no explicit correlation-id plumbing is needed here.
|
||||||
|
#[inline]
|
||||||
|
fn audit_ref_count(table: &'static str, hash: &str, delta: i32, source: &'static str) {
|
||||||
|
tracing::info!(
|
||||||
|
target: "oxicloud::refcount",
|
||||||
|
table,
|
||||||
|
hash = %hash,
|
||||||
|
delta,
|
||||||
|
source,
|
||||||
|
"ref_count {}", if delta >= 0 { "+" } else { "-" }
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
// ── CDC helper types ─────────────────────────────────────────────────────────
|
// ── CDC helper types ─────────────────────────────────────────────────────────
|
||||||
|
|
||||||
/// Everything a streaming chunk ingest learned about its byte stream.
|
/// Everything a streaming chunk ingest learned about its byte stream.
|
||||||
@@ -751,24 +831,146 @@ impl DedupService {
|
|||||||
.await
|
.await
|
||||||
.map_err(|e| DomainError::internal_error("Dedup", format!("record attached blob: {e}")))?;
|
.map_err(|e| DomainError::internal_error("Dedup", format!("record attached blob: {e}")))?;
|
||||||
|
|
||||||
// A replaced row's old blob loses its only reference from here. Not
|
// Two shapes to balance depending on whether the UPSERT was a
|
||||||
// releasing it would pin those bytes forever — nothing else points at
|
// real content replacement or a same-content re-store:
|
||||||
// a superseded preview.
|
//
|
||||||
if let Some((old_hash,)) = previous
|
// - `previous == Some(old) && old != attached_hash` — different
|
||||||
&& old_hash != attached_hash
|
// content overwritten in the row. Release the old blob's ref
|
||||||
&& let Err(e) = self.remove_reference(&old_hash).await
|
// (its file_attached row is gone; would leak forever otherwise).
|
||||||
{
|
//
|
||||||
|
// - `previous == Some(old) && old == attached_hash` — SAME-content
|
||||||
|
// re-store. `store_from_stream` above incremented the manifest
|
||||||
|
// unconditionally, but the row's blob_hash didn't change so no
|
||||||
|
// logical reference was added. Cancel the phantom increment
|
||||||
|
// here, or it accumulates one +1 leak per same-content call.
|
||||||
|
// Mirrors the pattern `store_derived_blob` uses on its
|
||||||
|
// `ON CONFLICT DO NOTHING` `inserted == 0` branch.
|
||||||
|
// See `docs/plan/refcount-audit.md` for how the audit stream
|
||||||
|
// would surface this class of drift if it reappears.
|
||||||
|
//
|
||||||
|
// - `previous == None` — brand new (file_id, kind, variant) row.
|
||||||
|
// `store_from_stream`'s +1 pairs with the new row's implicit
|
||||||
|
// reference; nothing to release.
|
||||||
|
if let Some((old_hash,)) = previous {
|
||||||
|
let source = if old_hash == attached_hash {
|
||||||
|
refcount_audit_source::STORE_ATTACHED_BLOB_SAME_CONTENT_BALANCE
|
||||||
|
} else {
|
||||||
|
refcount_audit_source::STORE_ATTACHED_BLOB_REPLACE_RELEASE
|
||||||
|
};
|
||||||
|
if let Err(e) = self.remove_reference(&old_hash).await {
|
||||||
tracing::warn!(
|
tracing::warn!(
|
||||||
target: "oxicloud::dedup",
|
target: "oxicloud::dedup",
|
||||||
error = %e,
|
error = %e,
|
||||||
"failed to release replaced attached-blob reference for {}",
|
kind = source,
|
||||||
|
"failed to balance attached-blob reference for {}",
|
||||||
&old_hash[..old_hash.len().min(12)],
|
&old_hash[..old_hash.len().min(12)],
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
Ok(attached_hash)
|
Ok(attached_hash)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Atomic never-overwrite variant of [`Self::store_attached_blob`],
|
||||||
|
/// for migration/import paths whose semantic is "write if absent,
|
||||||
|
/// leave alone if present" — mirrors the shape
|
||||||
|
/// [`Self::store_derived_blob`] already uses.
|
||||||
|
///
|
||||||
|
/// The plain `store_attached_blob` reads `previous`, then upserts,
|
||||||
|
/// then decrements — non-transactional. That's correct for the
|
||||||
|
/// user-driven PUT thumbnail path (a real replacement should
|
||||||
|
/// release the superseded blob), but it opens a check-then-act
|
||||||
|
/// race with concurrent writers when the caller's intent is
|
||||||
|
/// "only import if this file hasn't already got a preview".
|
||||||
|
/// `thumb_attached_import_service` is exactly that caller.
|
||||||
|
///
|
||||||
|
/// This variant uses a single-statement `INSERT ... ON CONFLICT
|
||||||
|
/// DO NOTHING` — race-free by construction. If the row already
|
||||||
|
/// exists (any content), the atomic INSERT is a no-op and we
|
||||||
|
/// release the reference `store_from_stream` just took. If we
|
||||||
|
/// won the insert, the reference is legitimately held by our
|
||||||
|
/// new row.
|
||||||
|
///
|
||||||
|
/// Return value discriminates the two cases so the caller can
|
||||||
|
/// track its own `imported` vs `already` counters:
|
||||||
|
/// - [`AttachedBlobInsertOutcome::Inserted`] — we wrote the row.
|
||||||
|
/// - [`AttachedBlobInsertOutcome::AlreadyPresent`] — a row was
|
||||||
|
/// there when we arrived; we made no change and released our
|
||||||
|
/// ref. `existing_hash` is the concurrent winner's blob hash,
|
||||||
|
/// returned via a follow-up SELECT (so it's not strictly
|
||||||
|
/// atomic with the INSERT, but that's fine — the row's shape
|
||||||
|
/// is stable now that a concurrent writer can no longer
|
||||||
|
/// collide with us here; any later change goes through the
|
||||||
|
/// full `store_attached_blob` UPSERT path, which is out of
|
||||||
|
/// scope for this method's "if absent" contract).
|
||||||
|
pub async fn store_attached_blob_if_absent(
|
||||||
|
&self,
|
||||||
|
file_id: &str,
|
||||||
|
kind: &str,
|
||||||
|
variant: &str,
|
||||||
|
content_type: &str,
|
||||||
|
bytes: Bytes,
|
||||||
|
uploaded_by: uuid::Uuid,
|
||||||
|
) -> Result<AttachedBlobInsertOutcome, DomainError> {
|
||||||
|
let stored = self
|
||||||
|
.store_from_stream(
|
||||||
|
stream::once(async move { Ok::<Bytes, std::io::Error>(bytes) }),
|
||||||
|
Some(content_type.to_string()),
|
||||||
|
)
|
||||||
|
.await?;
|
||||||
|
let attached_hash = stored.hash().to_string();
|
||||||
|
|
||||||
|
// Single-statement atomic INSERT. `ON CONFLICT DO NOTHING`
|
||||||
|
// means: if another writer got there first, we silently
|
||||||
|
// yield. Same primitive `store_derived_blob` uses.
|
||||||
|
let inserted = sqlx::query(
|
||||||
|
"INSERT INTO storage.file_attached_blobs
|
||||||
|
(file_id, kind, variant, blob_hash, content_type, uploaded_by)
|
||||||
|
VALUES ($1::uuid, $2, $3, $4, $5, $6)
|
||||||
|
ON CONFLICT (file_id, kind, variant) DO NOTHING",
|
||||||
|
)
|
||||||
|
.bind(file_id)
|
||||||
|
.bind(kind)
|
||||||
|
.bind(variant)
|
||||||
|
.bind(&attached_hash)
|
||||||
|
.bind(content_type)
|
||||||
|
.bind(uploaded_by)
|
||||||
|
.execute(self.pool.as_ref())
|
||||||
|
.await
|
||||||
|
.map_err(|e| DomainError::internal_error("Dedup", format!("record attached blob: {e}")))?
|
||||||
|
.rows_affected();
|
||||||
|
|
||||||
|
if inserted == 0 {
|
||||||
|
// Row already existed when we arrived. Release the ref
|
||||||
|
// `store_from_stream` above took — the row that would
|
||||||
|
// justify it is not ours. Best-effort: leaving a
|
||||||
|
// dangling ref is worse than a warn log line.
|
||||||
|
if let Err(e) = self.remove_reference(&attached_hash).await {
|
||||||
|
tracing::warn!(
|
||||||
|
target: "oxicloud::dedup",
|
||||||
|
error = %e,
|
||||||
|
"failed to release duplicate attached-blob reference for {}",
|
||||||
|
&attached_hash[..attached_hash.len().min(12)],
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Fetch the concurrent winner's hash so the import
|
||||||
|
// service can readback-verify against its sidecar. The
|
||||||
|
// window between DO NOTHING and this SELECT is narrow;
|
||||||
|
// if the row gets updated in it, the sidecar delete
|
||||||
|
// path fails its verify and keeps the sidecar — the
|
||||||
|
// conservative fallback.
|
||||||
|
let existing = self.find_attached_blob(file_id, kind, variant).await;
|
||||||
|
return Ok(AttachedBlobInsertOutcome::AlreadyPresent {
|
||||||
|
existing_hash: existing.map(|r| r.blob_hash).unwrap_or_default(),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(AttachedBlobInsertOutcome::Inserted {
|
||||||
|
hash: attached_hash,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
/// Look up bytes attached to a file. File-keyed counterpart of
|
/// Look up bytes attached to a file. File-keyed counterpart of
|
||||||
/// [`Self::find_derived_blob`].
|
/// [`Self::find_derived_blob`].
|
||||||
pub async fn find_attached_blob(
|
pub async fn find_attached_blob(
|
||||||
@@ -1316,6 +1518,17 @@ impl DedupService {
|
|||||||
total_size,
|
total_size,
|
||||||
chunk_hashes.len(),
|
chunk_hashes.len(),
|
||||||
);
|
);
|
||||||
|
// The manifest INSERT above set `ref_count = 1` — that's
|
||||||
|
// the initial reference held by whatever callsite drove
|
||||||
|
// this ingest (a file's body, a preview attachment, a
|
||||||
|
// derivation). Audit-log the +1 so drift investigations
|
||||||
|
// can find where a manifest first came into existence.
|
||||||
|
audit_ref_count(
|
||||||
|
refcount_audit_table::CHUNK_MANIFESTS,
|
||||||
|
file_hash,
|
||||||
|
1,
|
||||||
|
refcount_audit_source::STORE_FROM_STREAM_NEW_MANIFEST,
|
||||||
|
);
|
||||||
self.fire_blob_creation_hooks(file_hash, content_type.as_deref());
|
self.fire_blob_creation_hooks(file_hash, content_type.as_deref());
|
||||||
return Ok(DedupResultDto::NewBlob {
|
return Ok(DedupResultDto::NewBlob {
|
||||||
hash: file_hash.to_string(),
|
hash: file_hash.to_string(),
|
||||||
@@ -1747,7 +1960,7 @@ impl DedupService {
|
|||||||
/// Bump a manifest's ref_count if it exists; returns its total_size.
|
/// Bump a manifest's ref_count if it exists; returns its total_size.
|
||||||
/// Single statement — no window between the existence check and the bump.
|
/// Single statement — no window between the existence check and the bump.
|
||||||
async fn bump_manifest_if_exists(&self, file_hash: &str) -> Result<Option<i64>, DomainError> {
|
async fn bump_manifest_if_exists(&self, file_hash: &str) -> Result<Option<i64>, DomainError> {
|
||||||
sqlx::query_scalar::<_, i64>(
|
let bumped = sqlx::query_scalar::<_, i64>(
|
||||||
"UPDATE storage.chunk_manifests SET ref_count = ref_count + 1
|
"UPDATE storage.chunk_manifests SET ref_count = ref_count + 1
|
||||||
WHERE file_hash = $1
|
WHERE file_hash = $1
|
||||||
RETURNING total_size",
|
RETURNING total_size",
|
||||||
@@ -1757,7 +1970,17 @@ impl DedupService {
|
|||||||
.await
|
.await
|
||||||
.map_err(|e| {
|
.map_err(|e| {
|
||||||
DomainError::internal_error("Dedup", format!("Failed to bump manifest ref_count: {e}"))
|
DomainError::internal_error("Dedup", format!("Failed to bump manifest ref_count: {e}"))
|
||||||
})
|
})?;
|
||||||
|
|
||||||
|
if bumped.is_some() {
|
||||||
|
audit_ref_count(
|
||||||
|
refcount_audit_table::CHUNK_MANIFESTS,
|
||||||
|
file_hash,
|
||||||
|
1,
|
||||||
|
refcount_audit_source::BUMP_MANIFEST_IF_EXISTS,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Ok(bumped)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Stream → chunk store, WITHOUT creating a manifest.
|
/// Stream → chunk store, WITHOUT creating a manifest.
|
||||||
@@ -2219,6 +2442,12 @@ impl DedupService {
|
|||||||
.rows_affected();
|
.rows_affected();
|
||||||
|
|
||||||
if manifest_affected > 0 {
|
if manifest_affected > 0 {
|
||||||
|
audit_ref_count(
|
||||||
|
refcount_audit_table::CHUNK_MANIFESTS,
|
||||||
|
hash,
|
||||||
|
1,
|
||||||
|
refcount_audit_source::ADD_REFERENCE_MANIFEST,
|
||||||
|
);
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2246,6 +2475,12 @@ impl DedupService {
|
|||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
audit_ref_count(
|
||||||
|
refcount_audit_table::BLOBS,
|
||||||
|
hash,
|
||||||
|
1,
|
||||||
|
refcount_audit_source::ADD_REFERENCE_LEGACY,
|
||||||
|
);
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2355,6 +2590,17 @@ impl DedupService {
|
|||||||
&file_hash[..12],
|
&file_hash[..12],
|
||||||
chunk_hashes.len()
|
chunk_hashes.len()
|
||||||
);
|
);
|
||||||
|
// Emit AFTER the commit so a rolled-back TX doesn't leave a
|
||||||
|
// phantom audit line — the "delta = -current_rc" reflects
|
||||||
|
// "the manifest is gone, effective ref_count is 0". Convention
|
||||||
|
// for the audit stream: use the delta that would produce a
|
||||||
|
// read-back of 0.
|
||||||
|
audit_ref_count(
|
||||||
|
refcount_audit_table::CHUNK_MANIFESTS,
|
||||||
|
file_hash,
|
||||||
|
-current_rc,
|
||||||
|
refcount_audit_source::REMOVE_MANIFEST_REFERENCE_DELETE,
|
||||||
|
);
|
||||||
Ok(true)
|
Ok(true)
|
||||||
} else {
|
} else {
|
||||||
// Still has references — just decrement
|
// Still has references — just decrement
|
||||||
@@ -2373,6 +2619,12 @@ impl DedupService {
|
|||||||
.map_err(|e| DomainError::internal_error("Dedup", format!("Commit: {}", e)))?;
|
.map_err(|e| DomainError::internal_error("Dedup", format!("Commit: {}", e)))?;
|
||||||
|
|
||||||
tracing::debug!("Reference removed from manifest {}", &file_hash[..12]);
|
tracing::debug!("Reference removed from manifest {}", &file_hash[..12]);
|
||||||
|
audit_ref_count(
|
||||||
|
refcount_audit_table::CHUNK_MANIFESTS,
|
||||||
|
file_hash,
|
||||||
|
-1,
|
||||||
|
refcount_audit_source::REMOVE_MANIFEST_REFERENCE_DECREMENT,
|
||||||
|
);
|
||||||
Ok(false)
|
Ok(false)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -2430,6 +2682,12 @@ impl DedupService {
|
|||||||
self.reap_blob(hash).await;
|
self.reap_blob(hash).await;
|
||||||
|
|
||||||
tracing::info!("BLOB DELETED: {} (no more references)", &hash[..12]);
|
tracing::info!("BLOB DELETED: {} (no more references)", &hash[..12]);
|
||||||
|
audit_ref_count(
|
||||||
|
refcount_audit_table::BLOBS,
|
||||||
|
hash,
|
||||||
|
-ref_count,
|
||||||
|
refcount_audit_source::REMOVE_LEGACY_REFERENCE,
|
||||||
|
);
|
||||||
Ok(true)
|
Ok(true)
|
||||||
} else {
|
} else {
|
||||||
// Still has references — just decrement
|
// Still has references — just decrement
|
||||||
@@ -2450,6 +2708,12 @@ impl DedupService {
|
|||||||
})?;
|
})?;
|
||||||
|
|
||||||
tracing::debug!("Reference removed from blob {}", &hash[..12]);
|
tracing::debug!("Reference removed from blob {}", &hash[..12]);
|
||||||
|
audit_ref_count(
|
||||||
|
refcount_audit_table::BLOBS,
|
||||||
|
hash,
|
||||||
|
-1,
|
||||||
|
refcount_audit_source::REMOVE_LEGACY_REFERENCE,
|
||||||
|
);
|
||||||
Ok(false)
|
Ok(false)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -274,49 +274,31 @@ impl RecoverableJobHandler for ThumbAttachedImport {
|
|||||||
};
|
};
|
||||||
let file_id_str = file_id.to_string();
|
let file_id_str = file_id.to_string();
|
||||||
|
|
||||||
// Already mapped. Checked BEFORE storing, because
|
// Orphan check first — no atomic-insert exists for a
|
||||||
// `store_attached_blob` is ON CONFLICT DO UPDATE and would
|
// file_id whose FK would reject. Same rationale as before;
|
||||||
// release then retake the reference on every run.
|
// the race window between this check and the INSERT is
|
||||||
if let Some(existing) = self
|
// narrow AND covered by the FK constraint if the file is
|
||||||
.dedup
|
// deleted after we look — the atomic INSERT would then
|
||||||
.find_attached_blob(&file_id_str, "preview", &dir_name)
|
// fail loudly instead of silently drift.
|
||||||
.await
|
//
|
||||||
{
|
// Everything else — "row present" and "row absent" —
|
||||||
already += 1;
|
// used to be split across two branches with a
|
||||||
// Drains on a later run too: importing first and enabling
|
// non-transactional `find_attached_blob` between the
|
||||||
// deletion afterwards is the expected operator sequence,
|
// check and the write. That opened a check-then-act
|
||||||
// so reaching here is the common path rather than an edge
|
// race window: a concurrent thumbnail writer could
|
||||||
// case.
|
// INSERT the row after the check returned None, and the
|
||||||
if delete_imported {
|
// subsequent `store_attached_blob` UPSERT-UPDATE would
|
||||||
let path = self.thumbnails_root.join(&dir_name).join(&name);
|
// fire with same-or-different content. In the
|
||||||
if ThumbDerivedImport::verify_and_unlink(
|
// same-content case that leaked +1 on the manifest ref
|
||||||
&self.dedup,
|
// (pre-fix; guard branch now cancels).
|
||||||
THUMB_ATTACHED_IMPORT_JOB_NAME,
|
//
|
||||||
&file_id_str,
|
// Merged into ONE atomic call
|
||||||
&existing.blob_hash,
|
// `store_attached_blob_if_absent`: single-statement
|
||||||
&path,
|
// `INSERT ... ON CONFLICT DO NOTHING`, race-free by
|
||||||
)
|
// construction. The outcome enum distinguishes the two
|
||||||
.await
|
// paths so `imported` and `already` counters stay
|
||||||
{
|
// accurate.
|
||||||
deleted += 1;
|
if !self.file_exists(file_id).await {
|
||||||
} else {
|
|
||||||
unverified += 1;
|
|
||||||
record_or_log(
|
|
||||||
store,
|
|
||||||
THUMB_ATTACHED_IMPORT_JOB_NAME,
|
|
||||||
"sidecar_delete_unverified",
|
|
||||||
"anomaly",
|
|
||||||
None,
|
|
||||||
serde_json::json!({
|
|
||||||
"path": position,
|
|
||||||
"file_id": file_id_str,
|
|
||||||
"note": "attached blob did not read back; sidecar kept",
|
|
||||||
}),
|
|
||||||
)
|
|
||||||
.await;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} else if !self.file_exists(file_id).await {
|
|
||||||
// The file is gone, so this sidecar is unimportable: the
|
// The file is gone, so this sidecar is unimportable: the
|
||||||
// FK on `file_id` would reject the row. Mirrors the
|
// FK on `file_id` would reject the row. Mirrors the
|
||||||
// dead-source case in thumb_derived_import.
|
// dead-source case in thumb_derived_import.
|
||||||
@@ -389,9 +371,10 @@ impl RecoverableJobHandler for ThumbAttachedImport {
|
|||||||
let path = self.thumbnails_root.join(&dir_name).join(&name);
|
let path = self.thumbnails_root.join(&dir_name).join(&name);
|
||||||
match fs::read(&path).await {
|
match fs::read(&path).await {
|
||||||
Ok(data) => {
|
Ok(data) => {
|
||||||
|
use crate::infrastructure::services::dedup_service::AttachedBlobInsertOutcome;
|
||||||
match self
|
match self
|
||||||
.dedup
|
.dedup
|
||||||
.store_attached_blob(
|
.store_attached_blob_if_absent(
|
||||||
&file_id_str,
|
&file_id_str,
|
||||||
"preview",
|
"preview",
|
||||||
&dir_name,
|
&dir_name,
|
||||||
@@ -404,14 +387,39 @@ impl RecoverableJobHandler for ThumbAttachedImport {
|
|||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
Ok(attached_hash) => {
|
Ok(outcome) => {
|
||||||
|
// Bump the right counter AND pick the
|
||||||
|
// hash we'll verify-and-unlink against:
|
||||||
|
// Inserted — our new blob
|
||||||
|
// AlreadyPresent — the concurrent
|
||||||
|
// winner's blob
|
||||||
|
// Both drain the sidecar identically
|
||||||
|
// (verify-then-unlink on repair mode).
|
||||||
|
let verify_hash = match outcome {
|
||||||
|
AttachedBlobInsertOutcome::Inserted { hash } => {
|
||||||
imported += 1;
|
imported += 1;
|
||||||
if delete_imported {
|
hash
|
||||||
|
}
|
||||||
|
AttachedBlobInsertOutcome::AlreadyPresent {
|
||||||
|
existing_hash,
|
||||||
|
} => {
|
||||||
|
already += 1;
|
||||||
|
existing_hash
|
||||||
|
}
|
||||||
|
};
|
||||||
|
// Empty existing_hash only happens if
|
||||||
|
// the AlreadyPresent path's follow-up
|
||||||
|
// SELECT was overtaken by another
|
||||||
|
// writer. verify_and_unlink would
|
||||||
|
// refuse the sidecar delete in that
|
||||||
|
// case anyway, but skipping the call
|
||||||
|
// saves the pointless readback.
|
||||||
|
if delete_imported && !verify_hash.is_empty() {
|
||||||
if ThumbDerivedImport::verify_and_unlink(
|
if ThumbDerivedImport::verify_and_unlink(
|
||||||
&self.dedup,
|
&self.dedup,
|
||||||
THUMB_ATTACHED_IMPORT_JOB_NAME,
|
THUMB_ATTACHED_IMPORT_JOB_NAME,
|
||||||
&file_id_str,
|
&file_id_str,
|
||||||
&attached_hash,
|
&verify_hash,
|
||||||
&path,
|
&path,
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|||||||
@@ -254,6 +254,71 @@ ATTACHED_2=$(sql "SELECT count(*) FROM storage.file_attached_blobs WHERE file_id
|
|||||||
[[ "$ATTACHED_2" == "$ATTACHED_AFTER" ]] || fail "re-run duplicated attached rows"
|
[[ "$ATTACHED_2" == "$ATTACHED_AFTER" ]] || fail "re-run duplicated attached rows"
|
||||||
log "re-run is a no-op: rows and refcounts unchanged."
|
log "re-run is a no-op: rows and refcounts unchanged."
|
||||||
|
|
||||||
|
# ── 5c. store_attached_blob same-content guard (regression test) ─────────────
|
||||||
|
#
|
||||||
|
# The PUT /api/files/{id}/thumbnail/{size} endpoint calls
|
||||||
|
# `dedup_service::store_attached_blob` directly — no pre-check like
|
||||||
|
# `thumb_attached_import_service` does. A same-content re-PUT is the
|
||||||
|
# ONLY current public surface that exercises the ref-balance branch
|
||||||
|
# added to `store_attached_blob` after the Sept-2026 manifest-drift
|
||||||
|
# investigation:
|
||||||
|
#
|
||||||
|
# if let Some((old_hash,)) = previous {
|
||||||
|
# // same-content: cancel `store_from_stream`'s spurious +1
|
||||||
|
# // different-hash: release the superseded blob's ref
|
||||||
|
# remove_reference(&old_hash)
|
||||||
|
# }
|
||||||
|
#
|
||||||
|
# Before the fix, same-content re-PUT would leak +1 on the manifest's
|
||||||
|
# ref_count every time (store_from_stream incremented, guard skipped
|
||||||
|
# the decrement when old == new). This test PUTs the same thumbnail
|
||||||
|
# bytes twice and asserts the manifest's ref_count is unchanged.
|
||||||
|
#
|
||||||
|
# Uses variant "icon" so we don't collide with the row the import
|
||||||
|
# job populated above (variant "preview"), keeping the two flows
|
||||||
|
# independent. Server re-encodes to JPEG deterministically, so both
|
||||||
|
# PUTs produce byte-identical manifest content.
|
||||||
|
|
||||||
|
log "5c. Same-content re-PUT via API does not churn refcount"
|
||||||
|
|
||||||
|
# Round 1: first PUT populates the row (INSERT — previous=None, guard
|
||||||
|
# doesn't fire). This is the fresh-ingest path; ref_count becomes 1.
|
||||||
|
curl -sf -X PUT \
|
||||||
|
-H "$AUTH" \
|
||||||
|
-H "Content-Type: image/jpeg" \
|
||||||
|
--data-binary "@$UPLOADED_THUMB" \
|
||||||
|
"$base_url/api/files/$FILE_ID/thumbnail/icon" \
|
||||||
|
>/dev/null || fail "5c: first PUT of thumbnail (variant=icon) failed"
|
||||||
|
|
||||||
|
GUARD_HASH=$(sql "SELECT blob_hash FROM storage.file_attached_blobs \
|
||||||
|
WHERE file_id='$FILE_ID' AND kind='preview' AND variant='icon' \
|
||||||
|
LIMIT 1;")
|
||||||
|
[[ -n "$GUARD_HASH" ]] || fail "5c: first PUT did not land a file_attached_blobs row"
|
||||||
|
|
||||||
|
GUARD_REFS_BEFORE=$(sql "SELECT ref_count FROM storage.chunk_manifests WHERE file_hash='$GUARD_HASH';")
|
||||||
|
# Legacy blobs (pre-CDC) don't have a chunk_manifests row — skip
|
||||||
|
# the test in that case rather than fail on an unrelated path.
|
||||||
|
if [[ -z "$GUARD_REFS_BEFORE" ]]; then
|
||||||
|
log "5c: attached blob is on the legacy path (no manifest) — guard test skipped (targets CDC path)"
|
||||||
|
else
|
||||||
|
# Round 2: second PUT with the SAME bytes. UPSERT-UPDATE fires,
|
||||||
|
# previous.blob_hash == new attached_hash, and the guard MUST
|
||||||
|
# cancel `store_from_stream`'s +1. Without the fix, refcount
|
||||||
|
# would go from 1 → 2 here.
|
||||||
|
curl -sf -X PUT \
|
||||||
|
-H "$AUTH" \
|
||||||
|
-H "Content-Type: image/jpeg" \
|
||||||
|
--data-binary "@$UPLOADED_THUMB" \
|
||||||
|
"$base_url/api/files/$FILE_ID/thumbnail/icon" \
|
||||||
|
>/dev/null || fail "5c: same-content re-PUT of thumbnail failed"
|
||||||
|
|
||||||
|
GUARD_REFS_AFTER=$(sql "SELECT ref_count FROM storage.chunk_manifests WHERE file_hash='$GUARD_HASH';")
|
||||||
|
[[ "$GUARD_REFS_AFTER" == "$GUARD_REFS_BEFORE" ]] \
|
||||||
|
|| fail "5c: same-content re-PUT churned refcount: $GUARD_REFS_BEFORE → $GUARD_REFS_AFTER (guard regressed?)"
|
||||||
|
|
||||||
|
log "5c: same-content re-PUT stable, refcount=$GUARD_REFS_AFTER"
|
||||||
|
fi
|
||||||
|
|
||||||
# ── 5b. Deletion: the destructive half, and the only one that can lose data
|
# ── 5b. Deletion: the destructive half, and the only one that can lose data
|
||||||
#
|
#
|
||||||
# Everything above is additive and recoverable. This unlinks files after a
|
# Everything above is additive and recoverable. This unlinks files after a
|
||||||
|
|||||||
Reference in New Issue
Block a user