mirror of
https://github.com/block/buzz.git
synced 2026-11-14 05:12:27 +00:00
feat(deletion): add durable whole-community deletion (#4425)
## Summary Adds a durable, operator-controlled V1 for deleting an entire Buzz community without deleting another tenant's data. The workflow is exposed through `buzz-admin deletions`: - `sweep` records independent fleet storage-taxonomy observations - `submit`, `list`, `inspect`, and `approve` manage a deletion request - `unblock` resumes a fail-closed request after an operator records remediation identity and reason - `run` and `drain` execute bounded work Requests advance through a PostgreSQL-backed state machine and stop at `retention_pending` after logical deletion has been independently verified across PostgreSQL, object storage, and Redis. This PR ships the engine and CLI, not a continuously running worker or Kubernetes packaging. For V1, a cluster/VM administrator invokes `/usr/local/bin/buzz-admin` from the existing relay image, for example with `kubectl exec` or an equivalent container/VM exec path. ## What whole-community V1 removes For the target community, V1 removes: - rows from the allowlisted community-scoped PostgreSQL catalog, including members, profiles, authored events and bodies, DMs, reactions, mentions, memberships, tokens, workflows, moderation, audit, feedback, and rate-limit state - media sidecars and upload-attribution records under `_meta/<community>/` and `_uploads/<community>/` - Git repository pointers under `repos/<community>/` - Redis keys under `buzz:<community>:*` The community row survives as a permanent tombstone, and deletion control-plane records remain as evidence of the request, approval, execution, and result. ## Safety model Deletion is not a broad `DELETE CASCADE` followed by optimistic cleanup. The destructive boundaries are durable and fail closed. ### 1. Inventory and approval - `submit` resolves the target and freezes the schema plus summary-only storage inventory. - Approval is bound to the exact request, community, and frozen inventory digest. - Unsupported manifest versions, malformed keys inside the target's owned prefixes, live scoped-table/write-fence coverage drift, frozen-inventory mismatch, and approval mismatch block execution rather than guessing. Migration and catalog revision numbers are not authorization gates; the executor validates the live safety shape instead. - Storage inventory is server-side prefix scoped to exactly: - `_meta/<community>/` - `_uploads/<community>/` - `repos/<community>/` - The deletion path never lists the whole shared bucket and has no arbitrary per-community object cap. Its listing work is proportional to the target community's bindings, not total fleet storage. - Fleet-wide taxonomy sweeps remain independent observability. They report unknown writer shapes but do not gate deletion submission, fencing, or destructive progress. Maintainers must add deletion taxonomy coverage whenever a new community-owned object-key class is introduced; writer-coverage tests bind the current media and Git writers to that contract. ### 2. Quiesce, fence, and destructive freeze - Writes continue through submission, inventory, and approval. They stop when execution moves the target into `quiescing` and then establishes the durable fence. - Already-admitted external effects finish under heartbeated serving-write leases; the exact admitted lease may renew while the community is quiescing, but new lease acquisition is rejected. The executor drains admitted leases before destructive work. - Invite minting after quiescing begins fails as typed `AccessDenied` (HTTP 503 at the relay boundary) before an invite can be persisted. - Database triggers enforce the community write fence across the complete catalog of community-scoped tables. Startup/readiness and destructive execution validate that catalog so a newly added but unfenced table cannot silently escape. - **Named isolation assumption — fresh write snapshot.** Every writer transaction that can reach a community-fenced relation must use PostgreSQL `READ COMMITTED`; each guarded write therefore observes a statement snapshot no older than acquisition of the community deletion lock. `REPEATABLE READ` and `SERIALIZABLE` can retain a pre-fence snapshot and are unsupported for writers. The writer pool refuses non-`READ COMMITTED` sessions at connection setup, and both SQL fence functions reject an explicit per-transaction isolation override with SQLSTATE `25000`. Configuration-delivered bad isolation can surface through SQLx as a pool-acquire timeout because every `after_connect` attempt is rejected; the precise `community writes require READ COMMITTED isolation` reason remains observable when the SQL guard is reached. Read-only replica transactions are outside this assumption. - Holding the shared advisory lock until the guarded write executes is a separate liveness condition: under `READ COMMITTED`, releasing it early does not permit resurrection because the trigger rechecks the fence, but it can turn a fleet sweep into a statement-wide SQLSTATE `55000` abort. - After the fence closes writers, storage is re-enumerated into chunked side-table rows. Per-prefix counts and digests bind those concrete keys to the destructive manifest. - Manifest chunk insertion, update, and deletion are protected after freeze. This closes the race where an unbound key could otherwise appear after the manifest was committed. ### 3. Checkpointed destruction - Target-owned object bindings are deleted from the frozen destructive manifest in bounded batches with durable progress. - The concrete key list lives in chunked side-table rows rather than one request-row JSON value. It supports large communities, resumable execution, and terminal cleanup. - Missing objects are accepted as idempotent crash-window outcomes; malformed ownership, changed evidence, and unexplained target-prefix drift fail closed. - PostgreSQL purging remains scoped by `community_id`, including the guarded NIP-RS hard-delete path discovered with real Desktop kind `30078` read-state data. - Redis cleanup explicitly scans and `UNLINK`s only `buzz:<community_id>:*`. Natural expiry is insufficient because some keys, including tunnel generation counters used as fencing state, are deliberately persistent. ### 4. Independent verification - PostgreSQL logical absence is checked after purge. - The three target-owned storage prefixes are freshly inventoried again and must be empty. - Redis requires two complete empty namespace scans. - Only after all three stores pass does the request advance through `logically_verified` to `retention_pending`. ## What V1 deliberately does not erase ### Shared content-addressed storage Per-community deletion removes bindings, metadata, attribution records, and Git pointers. It does **not** physically delete fleet-shared CAS bytes that another community may still reference: - media blobs and thumbnails - Git manifests, packs, and indexes (`manifests/`, `packs/`, and `idx/`) Safe reclamation requires a separate fleet-wide reachability and retention GC. Unknown keys elsewhere in the shared bucket do not block one community's deletion; malformed or unrecognized keys inside that community's three owned prefixes still fail closed. ### External retained copies The online logical-deletion proof does not erase object versions/replicas, database backups/WAL, CDN copies, provider retention copies, or observability exports. Those require their own retention and purge controls. ### Member-only erasure This PR erases a whole community. It does not implement the different operation "erase one npub while preserving the community." Removing membership or accepting NIP-09 is not member erasure. A member-only workflow would need to find and selectively remove or redact authored event content and pubkeys, profile data, DMs, reactions, mentions, memberships/roles, tokens, workflows/subscriptions, upload attribution, moderation/audit history, repository attribution, and identity embedded in tags or JSON. It would also need explicit rules for ownership transfer, surviving replies and thread metadata, audit-chain integrity, immutable Git history, and shared-CAS reachability. That requires a pubkey-level fence and selective graph rewrite; it is a separate deletion product, not a safe extension of this whole-tenant worker. ## In scope - migration `0029_community_deletion.sql`: requests, approvals, leases, manifest chunks, checkpoints, tombstones, and the universal write-fence catalog - durable executor leases, generations, heartbeats, retry/block state, and resumable stage transitions - operator-driven `sweep`, `submit`, `list`, `inspect`, `approve`, `unblock`, `run`, and `drain` commands - serving-path fences for database writes and external effects across event ingest, media, Git, workflow, push, invites, mesh/tunnel, and related paths - target-prefix-only storage inventory, summary manifests, post-fence destructive chunks, and bounded batch deletion - exact community Redis namespace purge and two-pass absence verification - cross-community isolation, crash/resume, manifest-integrity, writer-taxonomy, and schema/migration regressions - desired-state `schema/schema.sql` support without requiring a SQLx migration ledger ## Deferred / not covered - dedicated Helm/chart worker Deployment, service account, secrets, probes, resources, and network policy - autonomous `buzz-admin deletions worker` poll loop and worker-only health server - least-privilege separation among migration, relay-serving, and destructive execution roles - fleet-wide shared-CAS physical GC - backup/provider/CDN/observability retention completion - member-only erasure - provider-native conditional-delete improvements - a general force-continue escape hatch; permanent safety failures remain fail closed unless an operator remediates the cause and records an audited `unblock` The removed continuous-worker implementation remains deferred; no remote follow-up branch is claimed by this PR. ## Validation ### Current PR head and repository state Current pushed head: `359d8402ee15f049768f54156f67b953c7a7e2ed`, rebased onto `cc9a2f783375e51a6e8d1f2f9d01d5f7e22813d1` (`origin/main` at push time). The complete PR diff is now 47 files, 9,834 additions, and 517 deletions. The bespoke source-scanner stack was removed to keep this PR scoped to community deletion. Tyler/team requested the underlying fenced-write safety behavior, not `ast-grep`, `crates/buzz-db/tests/community_fenced_writes.rs`, its 27 fixtures, or the new `scripts/lints/community_*.yml` rules. Those scanner-specific files, dependencies, Hermit links, and runner wiring are absent from the current tree. The production database write fence, startup/destructive live-catalog validation, and deletion behavior remain. Source validation on this exact SHA passed: - `cargo fmt --all -- --check` - `bash -n scripts/run-tests.sh` - `cargo nextest run -p buzz-db --all-targets`: 102 passed, 173 skipped, 0 failed - `cargo nextest run -p buzz-deletion --all-targets`: 10 passed, 9 skipped, 0 failed - `cargo nextest run -p buzz-admin --all-targets`: 1 passed, 0 failed - affected-package/all-target Clippy with warnings denied - lockfile consistency - Helm 3.16.4 lint and all 44 chart unit tests - Helm region controls using that fixture: default `BUZZ_S3_REGION=us-east-1`, explicit `eu-west-2` override, and blank-region schema rejection The prior Kubernetes battery below was run against `928992237358a3294621ac0280830b77155abc04`. It remains useful evidence for the patch-equivalent production deletion implementation, but it is **not** claimed as exact-SHA evidence for current head `359d8402ee15f049768f54156f67b953c7a7e2ed`; the current cleanup removes only scanner/test/tooling infrastructure. CI restarted for the new head after the rebase and is pending. Human review remains `CHANGES_REQUESTED`. ### Prior-head live Kubernetes deletion and safety gates The full program used one immutable image, real PostgreSQL, Redis, MinIO, and a three-relay Kubernetes release: - source: `928992237358a3294621ac0280830b77155abc04` (**prior head**) - image: `buzz-e2e:sha-928992237358` - immutable image digest: `sha256:a1a204f4618ac22d9e210be5e5290645a15d79831ae30b0e44379357c8e4a895` - evidence root: `/tmp/buzz-e2e/20260807T033025Z-928992237358-full-gates/` - evidence-manifest digest: `82875c5bc9bea7370b796a7aef3457b3a1c8306c84c59e0f7388bbb5ad30e865` Passed gates at that prior head: - **Chart/operator region:** default `us-east-1`, explicit nondefault propagation, blank-region schema rejection, live in-pod environment, and an in-pod taxonomy sweep over 18 objects with zero unknown. - **Fenced writers and lifecycle:** open-write/fence ordering; 100-attempt anti-starvation; invite, push matcher, and exhausted-reaper bystander isolation; non-`READ-COMMITTED` rejection; manifest/tombstone contracts; eight-failure stage block and audited `unblock`. - **Destructive lifecycle:** submit → approve → run → `retention_pending`; PostgreSQL tombstone and Redis/S3 verification true; zero retries/errors; terminal reruns rejected with exit 5. - **Fresh 10,001-object crash boundary:** exactly two chunks (10,000 + 1). The executor deleted chunk 0 from MinIO while its PostgreSQL stamp was row-lock-blocked, was killed with `SIGKILL`, left one object and both stamps absent, then resumed the same request under generation 2 to zero objects and terminal state. - **Independent dead-owner recovery:** a dedicated executor claimed generation 1, blocked before effects, and was killed through containerd with `SIGKILL` (no TERM cleanup). The request remained owned and unreclaimable before lease expiry; a successor claimed generation 2 after 60 seconds and completed with two attempts and zero retries. - **Three-pod socket isolation:** ordinary NIP-42 and joined huddle-audio target witnesses on every replica received exact `1008 / community deleted`; healthy-tenant witnesses on those pods remained live; deleted-host reconnect returned HTTP 404. - **Health/provenance:** all replicas independently returned ready and retained the exact image digest before/after destructive runs and an audio-enabled rolling restart; PostgreSQL, Redis, and MinIO were healthy at close. Instrument corrections were retained as evidence rather than counted as product failures: a foreground PostgreSQL forward caused an initial `PoolTimedOut`; Kubernetes pod deletion exercised graceful TERM rather than dead-owner recovery; shell-background socket witnesses died with their parent; and the first image build hit the corporate TLS proxy. Detached forwarding/witnesses, containerd `SIGKILL`, and the configured internal CA/Artifactory mirror produced the discriminating runs without weakening product security. ### Prior-head cleanup For the prior-head Kubernetes run, the Helm release was removed, namespace absence was verified, run-owned Screen sessions were absent, and that source worktree remained clean. The evidence manifest was independently recomputed and every indexed artifact passed `shasum -a 256 -c`. The current `359d8402` source worktree is also clean after the scanner-only cleanup and push. --------- Signed-off-by: npub122y0pqkertljmedu303rl0aqrj3w8pvu43t6jxm6875lzg6f2pwqegc3xc <5288f082d91aff2de5bc8be23fbfa01ca2e3859cac57a91b7a3fa9f12349505c@buzz.block.builderlab.xyz> Signed-off-by: npub1dccv64krpcpse5cmkzfeh998cftungyatw3djt8jwdw6g43f7fyqzzmrf7 <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@buzz.block.builderlab.xyz> Signed-off-by: Kalvin Chau <kalvin@block.xyz> Signed-off-by: am <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@buzz.block.builderlab.xyz> Signed-off-by: cid <d9f92a72922bf45c17379a47d64dae84b6020397c2d5a52b5317d512068cd9d3@buzz.block.builderlab.xyz> Co-authored-by: npub122y0pqkertljmedu303rl0aqrj3w8pvu43t6jxm6875lzg6f2pwqegc3xc <5288f082d91aff2de5bc8be23fbfa01ca2e3859cac57a91b7a3fa9f12349505c@buzz.block.builderlab.xyz> Co-authored-by: npub1dccv64krpcpse5cmkzfeh998cftungyatw3djt8jwdw6g43f7fyqzzmrf7 <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@buzz.block.builderlab.xyz> Co-authored-by: cid <d9f92a72922bf45c17379a47d64dae84b6020397c2d5a52b5317d512068cd9d3@buzz.block.builderlab.xyz>
This commit is contained in:
Kalvin C
@@ -474,6 +474,17 @@ mod tests {
|
||||
m.validate().expect("no parent is fine (first push)");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pointer_writer_is_covered_by_deletion_taxonomy() {
|
||||
let community = CommunityId::from_uuid(uuid::Uuid::from_u128(1));
|
||||
let owner = "a".repeat(64);
|
||||
let key = pointer_key(community, &owner, "repo");
|
||||
let prefixes = buzz_media::tenant_prefixes(*community.as_uuid());
|
||||
|
||||
assert!(prefixes.iter().any(|prefix| key.starts_with(prefix)));
|
||||
assert!(buzz_media::is_tenant_owned_key(*community.as_uuid(), &key));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pointer_key_strips_dot_git() {
|
||||
let c = CommunityId::from_uuid(uuid::Uuid::from_u128(1));
|
||||
|
||||
@@ -1783,6 +1783,21 @@ pub(crate) struct PushContext {
|
||||
pub repo_handle: HydratedRepo,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct FinalizePushHooks {
|
||||
#[cfg(test)]
|
||||
post_cas_gate: Option<Arc<PostCasGate>>,
|
||||
#[cfg(test)]
|
||||
fail_ref_state_insert: bool,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[derive(Default)]
|
||||
struct PostCasGate {
|
||||
reached: tokio::sync::Notify,
|
||||
resume: tokio::sync::Notify,
|
||||
}
|
||||
|
||||
/// Finalize a push request: CAS-commit the new state into the object
|
||||
/// store, derive kind:30618 from the committed manifest, and only then
|
||||
/// build the success response.
|
||||
@@ -1793,6 +1808,17 @@ pub(crate) struct PushContext {
|
||||
/// constructor of a push 2xx, so the seam is structural (not by
|
||||
/// convention).
|
||||
async fn finalize_push(state: &Arc<AppState>, ctx: PushContext) -> Response {
|
||||
finalize_push_inner(state, ctx, &FinalizePushHooks::default()).await
|
||||
}
|
||||
|
||||
async fn finalize_push_inner(
|
||||
state: &Arc<AppState>,
|
||||
ctx: PushContext,
|
||||
hooks: &FinalizePushHooks,
|
||||
) -> Response {
|
||||
#[cfg(not(test))]
|
||||
let _ = hooks;
|
||||
|
||||
// The push fence, part 0 — **a rejected push publishes nothing.**
|
||||
//
|
||||
// `ctx.pack.ok` is false when git aborted the ref updates: either the
|
||||
@@ -1823,10 +1849,41 @@ async fn finalize_push(state: &Arc<AppState>, ctx: PushContext) -> Response {
|
||||
return response;
|
||||
}
|
||||
|
||||
// An already-running receive-pack may cross the durable fence after
|
||||
// request admission. Revalidate immediately before object-store CAS; DB
|
||||
// trigger fencing alone cannot roll back an S3 pointer mutation.
|
||||
let serving_write = match buzz_deletion::acquire_serving_write(
|
||||
&state.db,
|
||||
ctx.tenant.community(),
|
||||
"git_publish",
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(guard) => guard,
|
||||
Err(error) => {
|
||||
warn!(owner = %ctx.owner, repo = %ctx.repo, %error, "push rejected by community deletion fence");
|
||||
return (
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"community writes are fenced",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
};
|
||||
|
||||
if let Err(error) = serving_write.verify().await {
|
||||
warn!(owner = %ctx.owner, repo = %ctx.repo, %error, "push lost community serving lease");
|
||||
return (
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"community write lease lost",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
|
||||
// Step 7 (CAS). The PushContext binds `parent_state` (observed at
|
||||
// hydrate) to the CAS predicate here — no re-reading of the pointer
|
||||
// between hydrate and CAS.
|
||||
let success = match cas_publish(
|
||||
// between hydrate and CAS. Observe serving-lease loss throughout the
|
||||
// potentially long upload/CAS operation, not only at its boundaries.
|
||||
let publish = cas_publish(
|
||||
&state.git_store,
|
||||
&ctx.tenant,
|
||||
ctx.repo_handle.path(),
|
||||
@@ -1838,72 +1895,87 @@ async fn finalize_push(state: &Arc<AppState>, ctx: PushContext) -> Response {
|
||||
max_pack_bytes: state.config.git_max_pack_bytes,
|
||||
max_repo_bytes: state.config.git_max_repo_bytes,
|
||||
},
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(s) => s,
|
||||
Err(CasError::Conflict {
|
||||
winner_manifest_key,
|
||||
..
|
||||
}) => {
|
||||
warn!(
|
||||
owner = %ctx.owner,
|
||||
repo = %ctx.repo,
|
||||
winner = %winner_manifest_key,
|
||||
"push lost CAS race; tempdir dropped, returning 409"
|
||||
);
|
||||
);
|
||||
let success = match serving_write.protect(publish).await {
|
||||
Ok(result) => match result {
|
||||
Ok(s) => s,
|
||||
Err(CasError::Conflict {
|
||||
winner_manifest_key,
|
||||
..
|
||||
}) => {
|
||||
warn!(
|
||||
owner = %ctx.owner,
|
||||
repo = %ctx.repo,
|
||||
winner = %winner_manifest_key,
|
||||
"push lost CAS race; tempdir dropped, returning 409"
|
||||
);
|
||||
return (
|
||||
StatusCode::CONFLICT,
|
||||
"push superseded by a concurrent writer; pull and retry",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
Err(CasError::ManifestInvalid(e)) => {
|
||||
// 4xx-class: the workspace produced refs/HEAD/oids the
|
||||
// manifest validator rejects (unsafe refname, malformed oid,
|
||||
// empty head, malformed parent). Pre-CAS — no pointer was
|
||||
// written.
|
||||
warn!(
|
||||
owner = %ctx.owner,
|
||||
repo = %ctx.repo,
|
||||
error = %e,
|
||||
"push rejected: manifest validation failed"
|
||||
);
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
"push produced invalid manifest state",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
Err(CasError::ResourceLimit(e)) => {
|
||||
warn!(
|
||||
owner = %ctx.owner,
|
||||
repo = %ctx.repo,
|
||||
error = %e,
|
||||
"push rejected: repo exceeds relay resource limits"
|
||||
);
|
||||
return (
|
||||
StatusCode::PAYLOAD_TOO_LARGE,
|
||||
"repository exceeds relay resource limits",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
Err(e) => {
|
||||
// 5xx-class: ManifestReadFailed (parent corruption),
|
||||
// Backend, PackCapture. The tempdir drops on scope exit; no
|
||||
// pointer was written (or, on rare ManifestReadFailed during
|
||||
// winner-fetch, the winner is already installed and the
|
||||
// loser's data is unrelated).
|
||||
error!(
|
||||
owner = %ctx.owner,
|
||||
repo = %ctx.repo,
|
||||
error = %e,
|
||||
"push failed pre-response"
|
||||
);
|
||||
return (StatusCode::INTERNAL_SERVER_ERROR, "git backend error").into_response();
|
||||
}
|
||||
},
|
||||
Err(error) => {
|
||||
warn!(owner = %ctx.owner, repo = %ctx.repo, %error, "push lost community serving lease during CAS publish");
|
||||
return (
|
||||
StatusCode::CONFLICT,
|
||||
"push superseded by a concurrent writer; pull and retry",
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"community write lease lost",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
Err(CasError::ManifestInvalid(e)) => {
|
||||
// 4xx-class: the workspace produced refs/HEAD/oids the
|
||||
// manifest validator rejects (unsafe refname, malformed oid,
|
||||
// empty head, malformed parent). Pre-CAS — no pointer was
|
||||
// written.
|
||||
warn!(
|
||||
owner = %ctx.owner,
|
||||
repo = %ctx.repo,
|
||||
error = %e,
|
||||
"push rejected: manifest validation failed"
|
||||
);
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
"push produced invalid manifest state",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
Err(CasError::ResourceLimit(e)) => {
|
||||
warn!(
|
||||
owner = %ctx.owner,
|
||||
repo = %ctx.repo,
|
||||
error = %e,
|
||||
"push rejected: repo exceeds relay resource limits"
|
||||
);
|
||||
return (
|
||||
StatusCode::PAYLOAD_TOO_LARGE,
|
||||
"repository exceeds relay resource limits",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
Err(e) => {
|
||||
// 5xx-class: ManifestReadFailed (parent corruption),
|
||||
// Backend, PackCapture. The tempdir drops on scope exit; no
|
||||
// pointer was written (or, on rare ManifestReadFailed during
|
||||
// winner-fetch, the winner is already installed and the
|
||||
// loser's data is unrelated).
|
||||
error!(
|
||||
owner = %ctx.owner,
|
||||
repo = %ctx.repo,
|
||||
error = %e,
|
||||
"push failed pre-response"
|
||||
);
|
||||
return (StatusCode::INTERNAL_SERVER_ERROR, "git backend error").into_response();
|
||||
}
|
||||
};
|
||||
|
||||
#[cfg(test)]
|
||||
if let Some(gate) = &hooks.post_cas_gate {
|
||||
gate.reached.notify_one();
|
||||
gate.resume.notified().await;
|
||||
}
|
||||
|
||||
// Derived after CAS: kind:30618 ref-state event over the *committed*
|
||||
// manifest's refs/head. Spec §Implementation Correspondence:
|
||||
// "kind:30618 is derived after CAS, never the commit." We emit only
|
||||
@@ -1927,7 +1999,7 @@ async fn finalize_push(state: &Arc<AppState>, ctx: PushContext) -> Response {
|
||||
(Some(before), Some(after)) => before != after,
|
||||
_ => true, // first push (parent None) or impossible-shape after key → publish
|
||||
};
|
||||
if manifest_changed {
|
||||
let publication_result: Result<(), String> = if manifest_changed {
|
||||
let inputs = RefStateInputs {
|
||||
repo_id: &ctx.repo_id,
|
||||
head: &success.manifest.head,
|
||||
@@ -1938,11 +2010,23 @@ async fn finalize_push(state: &Arc<AppState>, ctx: PushContext) -> Response {
|
||||
Ok(event) => {
|
||||
// Relay-signed kind:30618 belongs to the same server-resolved
|
||||
// tenant as the git request that committed the pointer.
|
||||
match state
|
||||
#[cfg(test)]
|
||||
let insert_result = if hooks.fail_ref_state_insert {
|
||||
Err(buzz_db::DbError::InvalidData(
|
||||
"injected kind:30618 insert failure".to_string(),
|
||||
))
|
||||
} else {
|
||||
state
|
||||
.db
|
||||
.insert_event_with_serving_write_guard(serving_write.lease(), &event, None)
|
||||
.await
|
||||
};
|
||||
#[cfg(not(test))]
|
||||
let insert_result = state
|
||||
.db
|
||||
.insert_event(ctx.tenant.community(), &event, None)
|
||||
.await
|
||||
{
|
||||
.insert_event_with_serving_write_guard(serving_write.lease(), &event, None)
|
||||
.await;
|
||||
match insert_result {
|
||||
Ok((stored, true)) => {
|
||||
// Routed through the guarded send path for uniformity;
|
||||
// the access gate no-ops for this globally-scoped
|
||||
@@ -1959,6 +2043,7 @@ async fn finalize_push(state: &Arc<AppState>, ctx: PushContext) -> Response {
|
||||
manifest = %success.manifest_key,
|
||||
"kind:30618 published (derived after CAS)"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
Ok((_, false)) => {
|
||||
info!(
|
||||
@@ -1966,26 +2051,41 @@ async fn finalize_push(state: &Arc<AppState>, ctx: PushContext) -> Response {
|
||||
repo = %ctx.repo_id,
|
||||
"kind:30618 deduplicated by relay db"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(
|
||||
owner = %ctx.owner,
|
||||
repo = %ctx.repo_id,
|
||||
error = %e,
|
||||
"kind:30618 insert failed; push remains durable in object store"
|
||||
);
|
||||
}
|
||||
Err(error) => Err(format!("kind:30618 insert failed: {error}")),
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(
|
||||
owner = %ctx.owner,
|
||||
repo = %ctx.repo_id,
|
||||
error = %e,
|
||||
"kind:30618 build failed; push remains durable in object store"
|
||||
);
|
||||
}
|
||||
Err(error) => Err(format!("kind:30618 build failed: {error}")),
|
||||
}
|
||||
} else {
|
||||
Ok(())
|
||||
};
|
||||
|
||||
// The admitted serving write spans the complete publication attempt. Fence
|
||||
// acquisition cannot overtake the pointer CAS, durable 30618 insert, or
|
||||
// local fan-out attempt; only now may the lease be released.
|
||||
if let Err(error) = serving_write.finish().await {
|
||||
warn!(owner = %ctx.owner, repo = %ctx.repo, %error, "failed to release community serving lease after push publication");
|
||||
return (
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"community write lease lost during publication",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
if let Err(error) = publication_result {
|
||||
error!(
|
||||
owner = %ctx.owner,
|
||||
repo = %ctx.repo_id,
|
||||
manifest = %success.manifest_key,
|
||||
%error,
|
||||
"push pointer committed but kind:30618 publication failed"
|
||||
);
|
||||
return (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
"push committed but ref-state publication failed; retry",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
|
||||
// Only now — after CAS commit and (optional) 30618 emission — build
|
||||
@@ -2014,12 +2114,14 @@ pub fn git_router(state: Arc<AppState>) -> Router {
|
||||
#[cfg(test)]
|
||||
mod track_c_tests {
|
||||
use super::*;
|
||||
use crate::api::git::hydrate::{hydrate_for_write, HydrationOptions};
|
||||
use crate::api::git::manifest::Manifest;
|
||||
use buzz_core::CommunityId;
|
||||
use nostr::{EventBuilder, Keys, Kind, Tag};
|
||||
use std::collections::BTreeMap;
|
||||
use std::io::Write;
|
||||
use std::process::Output;
|
||||
use tempfile::TempDir;
|
||||
|
||||
fn oid_sha1() -> String {
|
||||
"cb09a769da1c01f458fa6959d4e8eded38fac8d3".to_string()
|
||||
@@ -2160,6 +2262,303 @@ mod track_c_tests {
|
||||
assert!(remote.join("refs/heads/master").exists());
|
||||
}
|
||||
|
||||
async fn run_finalize_git(repo: &Path, args: &[&str]) -> std::process::Output {
|
||||
let mut command = Command::new("git");
|
||||
command.current_dir(repo).args(args);
|
||||
harden_git_env(&mut command);
|
||||
let output = command.output().await.expect("spawn git");
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"git {args:?}: {}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
output
|
||||
}
|
||||
|
||||
async fn finalize_test_state() -> (Arc<AppState>, sqlx::PgPool) {
|
||||
const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1
|
||||
let mut config = crate::config::Config::from_env().expect("default config loads");
|
||||
config.require_relay_membership = false;
|
||||
config.redis_url = "redis://127.0.0.1:1".to_string();
|
||||
config.database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
|
||||
.or_else(|_| std::env::var("DATABASE_URL"))
|
||||
.unwrap_or_else(|_| TEST_DB_URL.to_string());
|
||||
let pool = sqlx::PgPool::connect(&config.database_url)
|
||||
.await
|
||||
.expect("connect test DB");
|
||||
let db = buzz_db::Db::from_pool(pool.clone());
|
||||
db.migrate().await.expect("migrate test DB");
|
||||
let redis_pool = deadpool_redis::Config::from_url(&config.redis_url)
|
||||
.create_pool(Some(deadpool_redis::Runtime::Tokio1))
|
||||
.expect("redis pool");
|
||||
let pubsub = Arc::new(
|
||||
buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone())
|
||||
.await
|
||||
.expect("pubsub manager"),
|
||||
);
|
||||
let audit = buzz_audit::AuditService::new(pool.clone());
|
||||
let auth = buzz_auth::AuthService::new(config.auth.clone());
|
||||
let search = buzz_search::SearchService::new(pool.clone());
|
||||
let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new(
|
||||
db.clone(),
|
||||
buzz_workflow::WorkflowConfig::default(),
|
||||
));
|
||||
let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage");
|
||||
let (state, _audit_shutdown) = AppState::new(
|
||||
config,
|
||||
db,
|
||||
redis_pool,
|
||||
audit,
|
||||
pubsub,
|
||||
auth,
|
||||
search,
|
||||
workflow_engine,
|
||||
Keys::generate(),
|
||||
media_storage,
|
||||
);
|
||||
(Arc::new(state), pool)
|
||||
}
|
||||
|
||||
async fn approved_deletion(
|
||||
state: &AppState,
|
||||
host: &str,
|
||||
) -> (
|
||||
buzz_db::deletion::DeletionRequest,
|
||||
buzz_db::deletion::ClaimedDeletion,
|
||||
) {
|
||||
use buzz_db::deletion::{
|
||||
FrozenInventory, KeyStreamDigest, PrefixManifest, StorageManifest,
|
||||
DEFAULT_LEASE_DURATION,
|
||||
};
|
||||
|
||||
let store = state.db.deletion_store();
|
||||
let request = store
|
||||
.submit(host, "git-finalize-test", Some("post-CAS lease regression"))
|
||||
.await
|
||||
.expect("submit deletion");
|
||||
let inventory = FrozenInventory {
|
||||
schema: store
|
||||
.inventory_schema(request.community_id)
|
||||
.await
|
||||
.expect("schema inventory"),
|
||||
storage: StorageManifest {
|
||||
version: 4,
|
||||
prefixes: buzz_media::tenant_prefixes(*request.community_id.as_uuid())
|
||||
.into_iter()
|
||||
.map(|prefix| PrefixManifest {
|
||||
prefix,
|
||||
object_count: 0,
|
||||
total_bytes: 0,
|
||||
keys_digest: KeyStreamDigest::new().finish().0,
|
||||
})
|
||||
.collect(),
|
||||
},
|
||||
};
|
||||
store
|
||||
.freeze_inventory(request.id, &inventory)
|
||||
.await
|
||||
.expect("freeze inventory");
|
||||
store
|
||||
.approve(request.id, "git-finalize-test", None)
|
||||
.await
|
||||
.expect("approve deletion");
|
||||
let claim = store
|
||||
.claim_specific(request.id, "git-finalize-test", DEFAULT_LEASE_DURATION)
|
||||
.await
|
||||
.expect("claim deletion")
|
||||
.expect("won deletion claim");
|
||||
(request, claim)
|
||||
}
|
||||
|
||||
async fn pushed_context(
|
||||
state: &AppState,
|
||||
community: CommunityId,
|
||||
host: &str,
|
||||
owner: String,
|
||||
repo: String,
|
||||
pusher: nostr::PublicKey,
|
||||
scratch: &Path,
|
||||
) -> PushContext {
|
||||
let tenant = TenantContext::resolved(community, host);
|
||||
let (hydrated, parent_state) = hydrate_for_write(
|
||||
&state.git_store,
|
||||
&tenant,
|
||||
&owner,
|
||||
&repo,
|
||||
HydrationOptions {
|
||||
pack_cache: &state.git_pack_cache,
|
||||
scratch_dir: scratch,
|
||||
max_pack_bytes: 1024 * 1024,
|
||||
max_repo_bytes: 2 * 1024 * 1024,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("hydrate empty test repo");
|
||||
let source = scratch.join("source");
|
||||
tokio::fs::create_dir(&source)
|
||||
.await
|
||||
.expect("source directory");
|
||||
run_finalize_git(&source, &["init", "--quiet", "--initial-branch=main"]).await;
|
||||
run_finalize_git(&source, &["config", "user.email", "finalize@test"]).await;
|
||||
run_finalize_git(&source, &["config", "user.name", "finalize"]).await;
|
||||
tokio::fs::write(source.join("file.txt"), b"committed\n")
|
||||
.await
|
||||
.expect("write source file");
|
||||
run_finalize_git(&source, &["add", "file.txt"]).await;
|
||||
run_finalize_git(&source, &["commit", "--quiet", "-m", "committed"]).await;
|
||||
let remote = hydrated.path().to_str().expect("hydrated path utf8");
|
||||
run_finalize_git(&source, &["push", "--quiet", remote, "main"]).await;
|
||||
|
||||
PushContext {
|
||||
pack: PackOutput {
|
||||
stdout: b"push-ok".to_vec(),
|
||||
ok: true,
|
||||
},
|
||||
parent_state,
|
||||
owner,
|
||||
repo: repo.clone(),
|
||||
repo_id: repo,
|
||||
pusher,
|
||||
tenant,
|
||||
repo_handle: hydrated,
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres and MinIO"]
|
||||
async fn finalize_push_holds_serving_lease_through_post_cas_publication() {
|
||||
let (state, pool) = finalize_test_state().await;
|
||||
let host = format!("git-finalize-{}.example", uuid::Uuid::new_v4().simple());
|
||||
let community = state
|
||||
.db
|
||||
.ensure_configured_community(&host)
|
||||
.await
|
||||
.expect("create test community")
|
||||
.id;
|
||||
let (request, claim) = approved_deletion(&state, &host).await;
|
||||
let scratch = TempDir::new().expect("scratch");
|
||||
let owner = format!("owner-{}", uuid::Uuid::new_v4().simple());
|
||||
let repo = format!("repo-{}", uuid::Uuid::new_v4().simple());
|
||||
let ctx = pushed_context(
|
||||
&state,
|
||||
community,
|
||||
&host,
|
||||
owner,
|
||||
repo.clone(),
|
||||
Keys::generate().public_key(),
|
||||
scratch.path(),
|
||||
)
|
||||
.await;
|
||||
let gate = Arc::new(PostCasGate::default());
|
||||
let hooks = FinalizePushHooks {
|
||||
post_cas_gate: Some(Arc::clone(&gate)),
|
||||
fail_ref_state_insert: false,
|
||||
};
|
||||
let finalize_state = Arc::clone(&state);
|
||||
let finalize =
|
||||
tokio::spawn(async move { finalize_push_inner(&finalize_state, ctx, &hooks).await });
|
||||
|
||||
gate.reached.notified().await;
|
||||
state
|
||||
.db
|
||||
.deletion_store()
|
||||
.begin_quiescing(&claim.lease)
|
||||
.await
|
||||
.expect("quiesce after CAS");
|
||||
let error = state
|
||||
.db
|
||||
.deletion_store()
|
||||
.fence(&claim.lease)
|
||||
.await
|
||||
.expect_err("post-CAS serving lease must block fence");
|
||||
assert!(matches!(
|
||||
error,
|
||||
buzz_db::DbError::ServingWritesNotDrained { .. }
|
||||
));
|
||||
assert!(!state
|
||||
.db
|
||||
.deletion_store()
|
||||
.is_serving_active(community)
|
||||
.await
|
||||
.expect("quiescing rejects new serving work"));
|
||||
|
||||
gate.resume.notify_one();
|
||||
let response = finalize.await.expect("finalize task");
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let mut query = buzz_db::event::EventQuery::for_community(community);
|
||||
query.kinds = Some(vec![30_618]);
|
||||
query.d_tag = Some(repo);
|
||||
let events = state.db.query_events(&query).await.expect("query 30618");
|
||||
assert_eq!(events.len(), 1, "kind:30618 must be durable before release");
|
||||
assert!(state
|
||||
.db
|
||||
.deletion_store()
|
||||
.serving_writes_drained(community)
|
||||
.await
|
||||
.expect("serving lease released"));
|
||||
let generation = state
|
||||
.db
|
||||
.deletion_store()
|
||||
.fence(&claim.lease)
|
||||
.await
|
||||
.expect("fence after publication");
|
||||
assert_eq!(generation, 1);
|
||||
assert_eq!(
|
||||
state
|
||||
.db
|
||||
.deletion_store()
|
||||
.get(request.id)
|
||||
.await
|
||||
.expect("fenced request")
|
||||
.stage,
|
||||
buzz_db::deletion::DeletionStage::Fenced
|
||||
);
|
||||
drop(state);
|
||||
pool.close().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres and MinIO"]
|
||||
async fn finalize_push_db_failure_after_cas_is_not_success_and_releases_lease() {
|
||||
let (state, pool) = finalize_test_state().await;
|
||||
let host = format!(
|
||||
"git-finalize-fail-{}.example",
|
||||
uuid::Uuid::new_v4().simple()
|
||||
);
|
||||
let community = state
|
||||
.db
|
||||
.ensure_configured_community(&host)
|
||||
.await
|
||||
.expect("create test community")
|
||||
.id;
|
||||
let scratch = TempDir::new().expect("scratch");
|
||||
let ctx = pushed_context(
|
||||
&state,
|
||||
community,
|
||||
&host,
|
||||
format!("owner-{}", uuid::Uuid::new_v4().simple()),
|
||||
format!("repo-{}", uuid::Uuid::new_v4().simple()),
|
||||
Keys::generate().public_key(),
|
||||
scratch.path(),
|
||||
)
|
||||
.await;
|
||||
let hooks = FinalizePushHooks {
|
||||
post_cas_gate: None,
|
||||
fail_ref_state_insert: true,
|
||||
};
|
||||
|
||||
let response = finalize_push_inner(&state, ctx, &hooks).await;
|
||||
assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
|
||||
assert!(state
|
||||
.db
|
||||
.deletion_store()
|
||||
.serving_writes_drained(community)
|
||||
.await
|
||||
.expect("serving lease released on failure"));
|
||||
drop(state);
|
||||
pool.close().await;
|
||||
}
|
||||
|
||||
/// A gzip-encoded request body is transparently inflated before it
|
||||
/// reaches the git subprocess. Git's smart-HTTP client gzips the
|
||||
/// upload-pack/receive-pack request body past a size threshold (fires
|
||||
|
||||
Reference in New Issue
Block a user