From b69cd3591d5b11fc6409f3d9e84d642bffa4b412 Mon Sep 17 00:00:00 2001 From: Bretton Date: Fri, 7 Aug 2026 18:07:46 -0700 Subject: [PATCH] fix(admissions): apply task-2 second-opinion batch (10 streams) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Headline fixes, all red-first: RecordRejection became a real CAS (pending-only + judged CID — was able to stomp accepted rows and concurrent edits); ApplyAcceptance treats NULL evaluated_cid as pin-trusting so restore-vs-absent converges order-independently (codex); RepinAcceptedCID gained its accepted-only guard + FOR UPDATE refusal classification (its plain-UPDATE shape broke the lock-held claim — empirically probed); op-rank derived repo-side + schema CHECK; account deletion sweeps admissions by DID prefix (unindexed subjects were retained — GDPR gap); standalone removal-delete resets redrivable; dead fk_author mapping removed; same-tuple + author-interleave concurrency races; pagination tie-break/clamps pinned; testkit down-migration version guard + clone-prefix guard. PRD rev 2.4. Co-Authored-By: Claude Fable 5 --- docs/PRD_AUTHOR_OWNED_POSTS.md | 20 +- internal/core/posts/admissions.go | 53 +- internal/core/users/interfaces.go | 7 +- internal/core/users/user_integration_test.go | 35 + .../db/migrations/034_author_owned_posts.sql | 8 + internal/db/postgres/admission_repo.go | 278 +++++-- .../admission_repo_concurrency_test.go | 194 +++++ .../postgres/admission_repo_lifecycle_test.go | 3 +- .../db/postgres/admission_repo_matrix_test.go | 736 +++++++++++++++++- .../db/postgres/admission_repo_query_test.go | 117 ++- .../db/postgres/admission_repo_schema_test.go | 48 +- internal/db/postgres/post_repo.go | 15 +- internal/db/postgres/user_repo.go | 15 +- internal/db/postgres/user_repo_test.go | 90 +++ tests/testkit/migrate.go | 58 +- 15 files changed, 1555 insertions(+), 122 deletions(-) diff --git a/docs/PRD_AUTHOR_OWNED_POSTS.md b/docs/PRD_AUTHOR_OWNED_POSTS.md index 9941d39..c4cdf80 100644 --- a/docs/PRD_AUTHOR_OWNED_POSTS.md +++ b/docs/PRD_AUTHOR_OWNED_POSTS.md @@ -25,7 +25,10 @@ special-casing silently dropped moderator restores (the restore commit is symmetric to the removal commit and indistinguishable from ordinary events on the wire); restore is now defined as any community acceptance winning the tuple CAS over a removal. Stale/terminal skips are outcome values, not -errors (033 precedent — sentinels would dead-letter healthy skips).** +errors (033 precedent — sentinels would dead-letter healthy skips). +Rev 2.4 (2026-08-08): rejection narrowed to pending-only CAS with judged CID; +op-rank derived repo-side; NULL-evaluated acceptance treated as pin-trusting +(task-2 second-opinion catches).** **Supersedes** the write-path architecture in `docs/federation-prd.md`: that document solves cross-instance posting by service-auth-forwarding the write to @@ -443,7 +446,11 @@ for communities this AppView hosts (from the fast path, the firehose consumer, or notify). It runs `admitPost` — the extracted §4.1 checks plus the new ban/rate-limit policy — then writes/updates the acceptance, or (rejection) records the decision in the admissions table, or (re-acceptance failure / -moderation) performs the atomic acceptance-delete + removal. It is the only +moderation) performs the atomic acceptance-delete + removal. Rejection is a +pending-only CAS carrying the judged CID: a rejection evaluated against +content the row no longer holds refuses rather than applies, and accepted or +removed rows are never rejected — a failed re-acceptance is a removal per +§5.5, not a rejection. It is the only writer of community-repo records in the post system, and every write is idempotent via deterministic rkeys + swap semantics. @@ -470,9 +477,12 @@ New table `community_post_admissions`: - `redrivable BOOLEAN NOT NULL DEFAULT true` — policy rejections are `false` (terminal; not retried by DLQ redrive); transient evaluation failures stay `true` -- `last_community_rev TEXT COLLATE "C"` + `last_community_op_rank SMALLINT` — - the §5.2 subject-scoped composite watermark (033 pins `COLLATE "C"` for - rev comparison; same rule here) +- `last_community_rev TEXT COLLATE "C"` + `last_community_op_rank SMALLINT` + with a `CHECK` restricting the op-rank to `0 | 1` — the §5.2 subject-scoped + composite watermark (033 pins `COLLATE "C"` for rev comparison; same rule + here). The op-rank is derived repo-side from the event's operation + (delete = 0, put = 1) and is never caller-supplied: an out-of-range rank + would corrupt the tuple comparison silently - Partial index on `status = 'accepted'` for feed queries `posts` changes: drop the `users` FK + CASCADE (§5.3), drop the in-record diff --git a/internal/core/posts/admissions.go b/internal/core/posts/admissions.go index ded632b..93c84da 100644 --- a/internal/core/posts/admissions.go +++ b/internal/core/posts/admissions.go @@ -2,6 +2,7 @@ package posts import ( "context" + "errors" "time" ) @@ -89,13 +90,31 @@ const ( CommunityOpPut CommunityOpRank = 1 ) +// ErrInvalidWatermark reports a community event carrying a watermark that can +// never have come off the wire — today, an empty Rev. It is a genuine error +// rather than a skip outcome: skips are the ordering gate WORKING, while an +// empty rev means the event was decoded wrong upstream, and stamping it would +// write a fabricated clock value onto the row. Routing it to the dead-letter +// queue is exactly right. +var ErrInvalidWatermark = errors.New("invalid community watermark") + // CommunityWatermark is the subject-scoped composite ordering key of §5.2: // the repo revision of the last APPLIED community event about this (community, // post) pair, plus that event's rank within its commit. // // Rev is a base32-sortable atProto TID, so lexicographic comparison of Rev IS // commit order within one repo — the same property migration 033's per-record -// gate relies on, which is why the column carries COLLATE "C". +// gate relies on, which is why the column carries COLLATE "C". Rev MUST be +// non-empty: every Jetstream commit carries one, so an empty rev is an +// upstream decoding bug, and repositories refuse it with a genuine error +// wrapping ErrInvalidWatermark rather than stamping a clock value that never +// existed. +// +// On a COMMAND, only Rev is consumed: OpRank is derived by the repository from +// the operation itself (a put ranks 1, a delete ranks 0), because the rank IS +// the operation's kind and letting a caller assert otherwise would let one +// mislabeled event reorder a commit. On a RESULT (Admission.LastCommunityEvent) +// both halves are meaningful — they report the tuple actually stored. // // The per-record gate cannot do this job: acceptance and removal are DIFFERENT // record URIs describing the SAME subject, so ordering them requires a key @@ -149,11 +168,13 @@ type Admission struct { // nothing, and making it fetch that separately would reintroduce the read-then- // write race the single-statement CAS exists to avoid. // -// Admission is nil in exactly one case: a mutation that may not CREATE a row -// met a subject that has none. Only RepinAcceptedCID can be in that position — -// every other mutation inserts when the subject is absent — and there is -// genuinely no row to describe, so the outcome is skipped_terminal and the -// caller has nothing to reconcile. +// Admission is nil in exactly one case: RepinAcceptedCID meeting a subject +// that has no row. The five event mutations insert when the subject is absent, +// so they always have a row to report; RecordRejection, the other mutation +// that may not create one, treats an absent subject as an ERROR rather than a +// result (the engine rejects rows it read from its own queue). That leaves the +// repin, where there is genuinely no row to describe: the outcome is +// skipped_terminal and the caller has nothing to reconcile. type AdmissionResult struct { Outcome AdmissionOutcome Admission *Admission @@ -221,6 +242,13 @@ type RepinAcceptanceCommand struct { // RecordRejectionCommand records an AppView-LOCAL rejection. // +// JudgedCID is the exact content CID the decision judged, and it is part of +// the guard: a rejection lands only on a `pending` row still holding this CID. +// The engine reads a pending row, evaluates its content, and writes the +// verdict — if the author edited in between, the verdict judged content the +// row no longer holds and must not land on the new content (§5.5: new content +// is judged fresh, and re-acceptance failure is a removal, not a rejection). +// // Redrivable is the caller's classification of WHY: a policy rejection is // terminal and must not be retried by the dead-letter redrive pass, while a // transient evaluation failure has to stay retryable. Getting it backwards @@ -230,6 +258,7 @@ type RecordRejectionCommand struct { CommunityDID string PostURI string DecisionCode string + JudgedCID string Redrivable bool } @@ -271,13 +300,23 @@ type AdmissionRepository interface { ApplyRemovalDelete(ctx context.Context, cmd CommunityDeleteCommand) (AdmissionResult, error) // RepinAcceptedCID moves a standing acceptance onto new content without a - // status transition — the §5.5 bridgedStats exception. + // status transition — the §5.5 bridgedStats exception. It applies only to a + // row that is `accepted` NOW: any other status (or an absent row) refuses + // with skipped_terminal, because a repin re-decides nothing and so has no + // business changing what a non-accepted row would need decided. RepinAcceptedCID(ctx context.Context, cmd RepinAcceptanceCommand) (AdmissionResult, error) // RecordRejection records the AppView's own decision not to admit a post. // It is not a community-repo record, so it must NOT advance the community // watermark: a local decision that could outrank a genuine community event // would suppress the very acceptance that overrules it. + // + // It lands only on a `pending` row still holding cmd.JudgedCID — rejection's + // one legal source state (§5.5: re-acceptance failure is a removal, and a + // rejection overwriting an accepted row would suppress a live acceptance). + // Any other standing row refuses it with a skip outcome; a subject with NO + // row is an ERROR, because the engine rejects rows it read from its own + // queue and their absence is a caller bug, not delivery skew. RecordRejection(ctx context.Context, cmd RecordRejectionCommand) (AdmissionResult, error) // Get returns the admission row for one subject, or ErrNotFound if the diff --git a/internal/core/users/interfaces.go b/internal/core/users/interfaces.go index 70e2408..cf0f849 100644 --- a/internal/core/users/interfaces.go +++ b/internal/core/users/interfaces.go @@ -67,9 +67,10 @@ type UserRepository interface { // 6. user_blocks (explicit DELETE - both directions) // 7. comments (explicit DELETE) // 8. votes (explicit DELETE - FK removed in migration 014) - // 9. community_post_admissions for this author's posts (explicit DELETE - - // no FK to posts by design, migration 034; must precede the posts it - // reads to find its subjects) + // 9. community_post_admissions for this author's posts (explicit DELETE + // by DID-prefix match on post_uri - no FK to posts by design, + // migration 034, and an admission's subject post may never have been + // indexed at all, so the sweep cannot go through the posts table) // 10. posts (explicit DELETE - fk_author CASCADE removed by migration 034) // 11. users // diff --git a/internal/core/users/user_integration_test.go b/internal/core/users/user_integration_test.go index 18d60fe..d488194 100644 --- a/internal/core/users/user_integration_test.go +++ b/internal/core/users/user_integration_test.go @@ -1034,6 +1034,23 @@ func TestAccountDeletion_Integration(t *testing.T) { } } + // Admissions: one about an indexed post, one whose subject post was + // never indexed (acceptance-before-post is an ordinary state, and the + // table deliberately has no FK to posts — migration 034 — so nothing + // but account deletion's own sweep can remove these rows) + for _, postURI := range []string{ + fmt.Sprintf("at://%s/social.coves.post/delete1", testDID), + fmt.Sprintf("at://%s/social.coves.community.postv2/neverindexed%d", testDID, uniqueSuffix), + } { + _, err = db.Exec(` + INSERT INTO community_post_admissions (community_did, post_uri, status) + VALUES ($1, $2, 'pending') + `, testCommunityDID, postURI) + if err != nil { + t.Fatalf("Failed to insert admission for %s: %v", postURI, err) + } + } + // Community subscription _, err = db.Exec(` INSERT INTO community_subscriptions (user_did, community_did, subscribed_at) @@ -1096,6 +1113,13 @@ func TestAccountDeletion_Integration(t *testing.T) { t.Fatalf("Expected 3 posts, got %d (err: %v)", count, err) } + // Check admissions exist (community_did is this test's own community, + // so the count is mechanism-independent) + err = db.QueryRow(`SELECT COUNT(*) FROM community_post_admissions WHERE community_did = $1`, testCommunityDID).Scan(&count) + if err != nil || count != 2 { + t.Fatalf("Expected 2 admissions, got %d (err: %v)", count, err) + } + // Check subscription exists err = db.QueryRow(`SELECT COUNT(*) FROM community_subscriptions WHERE user_did = $1`, testDID).Scan(&count) if err != nil || count != 1 { @@ -1154,6 +1178,17 @@ func TestAccountDeletion_Integration(t *testing.T) { t.Errorf("Expected 0 posts after deletion, got %d", count) } + // Check admissions deleted — BOTH of them, including the row whose + // subject post was never indexed. That row's only tie to the deleted + // account is the DID inside its post_uri. + err = db.QueryRow(`SELECT COUNT(*) FROM community_post_admissions WHERE community_did = $1`, testCommunityDID).Scan(&count) + if err != nil { + t.Fatalf("Error checking admissions: %v", err) + } + if count != 0 { + t.Errorf("Expected 0 admissions after deletion, got %d (an unindexed subject's admission survived the sweep?)", count) + } + // Check subscription deleted err = db.QueryRow(`SELECT COUNT(*) FROM community_subscriptions WHERE user_did = $1`, testDID).Scan(&count) if err != nil { diff --git a/internal/db/migrations/034_author_owned_posts.sql b/internal/db/migrations/034_author_owned_posts.sql index cf18750..f6648d1 100644 --- a/internal/db/migrations/034_author_owned_posts.sql +++ b/internal/db/migrations/034_author_owned_posts.sql @@ -109,6 +109,14 @@ CREATE TABLE community_post_admissions ( -- subject would be skipped forever, silently. CONSTRAINT chk_admission_watermark_complete CHECK ( (last_community_rev IS NULL) = (last_community_op_rank IS NULL) + ), + + -- The rank vocabulary is exactly {0 = delete, 1 = put} (§5.2). SMALLINT + -- admits 32766 other values, every one of which would outrank every + -- genuine put and freeze its subject forever — closed here, where every + -- writer meets it, rather than by repository convention. + CONSTRAINT chk_admission_op_rank CHECK ( + last_community_op_rank IN (0, 1) ) ); diff --git a/internal/db/postgres/admission_repo.go b/internal/db/postgres/admission_repo.go index be288a8..1f754f9 100644 --- a/internal/db/postgres/admission_repo.go +++ b/internal/db/postgres/admission_repo.go @@ -18,18 +18,28 @@ import ( // PostgreSQL storage for per-(community, post) admission decisions // (docs/PRD_AUTHOR_OWNED_POSTS.md §5.2, §5.5, §6.1; migration 034). // -// THE SHAPE EVERY MUTATION TAKES. All five are one INSERT ... ON CONFLICT DO -// UPDATE ... WHERE , and the guard is the whole decision: Postgres -// evaluates it against the conflicting row while holding that row's lock, so -// two consumers draining overlapping Jetstream feeds cannot interleave a read -// and a write. There is no SELECT-then-decide anywhere in this file, which is -// what makes a duplicate delivery a genuine no-op rather than a re-stamped -// decision timestamp. +// THE SHAPE EVERY MUTATION TAKES. All seven are single-statement compare-and- +// swaps whose guard is the whole decision. Five — the author-repo observation +// and the four community events — are one INSERT ... ON CONFLICT DO UPDATE ... +// WHERE , because each may legitimately meet an absent subject and must +// create the row that records the event was seen. The other two are guarded +// UPDATEs, because each must NEVER create a row: a repin moves an acceptance +// that stands, and a rejection lands on the pending row the engine read from +// its own queue. Either way, Postgres evaluates the guard against the current +// row inside the writing statement, so two consumers draining overlapping +// Jetstream feeds cannot interleave a read and a write. There is no +// SELECT-then-decide anywhere in this file, which is what makes a duplicate +// delivery — RecordRejection's included — a genuine no-op rather than a +// re-stamped decision timestamp. // // updated_at is set ONLY inside the guarded SET clause. A refused event must // leave the row byte-identical — the moderation audit trail would otherwise // become a function of how many feeds happened to carry the event. // +// INVARIANT: redrivable is false only while the decision that set it stands; +// every transition that reopens evaluation (new content reopening a rejection, +// a removal withdrawn with nothing in its place) resets it to true. +// // WHAT A REFUSAL RETURNS. A skip is an outcome, never an error (migration 033's // precedent, restated in §5.2): stale cross-feed copies, dead-letter redrives // and author edits of removed posts are the system working, and returning them @@ -38,10 +48,12 @@ import ( // notify or re-emit needs it and fetching it separately would reintroduce the // race the CAS exists to close. // -// The classifying SELECT that follows a refused upsert runs in the SAME -// transaction, so it reads the row under the lock the failed ON CONFLICT DO -// UPDATE already took: the row reported back is exactly the row the guard -// refused, not whatever a concurrent consumer left a moment later. +// The classifying SELECT that follows a refused mutation runs in the SAME +// transaction and takes the row lock itself (FOR UPDATE). The lock has to be +// explicit: a refused ON CONFLICT DO UPDATE holds the conflicting row's lock +// already, but a guarded UPDATE that matched nothing holds no lock at all, and +// without one the row reported back could be whatever a concurrent consumer +// left a moment after the guard refused rather than the row it refused. type postgresAdmissionRepo struct { db *sql.DB @@ -109,6 +121,10 @@ func (r *postgresAdmissionRepo) UpsertPending(ctx context.Context, cmd posts.Ups // function of the incoming CID, so a re-delivery carrying the CID already // recorded cannot change any column — and must therefore write none, // including updated_at. + // Removal terminality needs no arm of its own: a removed row matches none + // of the transitions below, so it keeps its status through the ELSE — the + // matrix tests "an edit while removed records the content and nothing else" + // and "same CID on a removed row" pin exactly that. query := ` INSERT INTO community_post_admissions ( community_did, post_uri, status, evaluated_cid, created_at, updated_at @@ -116,8 +132,6 @@ func (r *postgresAdmissionRepo) UpsertPending(ctx context.Context, cmd posts.Ups ON CONFLICT (community_did, post_uri) DO UPDATE SET evaluated_cid = excluded.evaluated_cid, status = CASE - WHEN community_post_admissions.status = 'removed' - THEN community_post_admissions.status WHEN community_post_admissions.status = 'rejected' THEN 'pending' WHEN community_post_admissions.status = 'accepted' @@ -144,7 +158,7 @@ func (r *postgresAdmissionRepo) UpsertPending(ctx context.Context, cmd posts.Ups WHERE community_post_admissions.evaluated_cid IS DISTINCT FROM excluded.evaluated_cid RETURNING ` + admissionColumns - return r.compareAndSwap(ctx, "UpsertPending", cmd.CommunityDID, cmd.PostURI, nonCommunityEventOutcome, + return r.compareAndSwap(ctx, "UpsertPending", cmd.CommunityDID, cmd.PostURI, upsertPendingOutcome, rowRequired, query, cmd.CommunityDID, cmd.PostURI, cmd.EvaluatedCID) } @@ -165,11 +179,21 @@ func (r *postgresAdmissionRepo) UpsertPending(ctx context.Context, cmd posts.Ups // arrives after the fresh acceptance, that deletion is refused as not-greater, // so the acceptance is the only event left that can clear the removal. func (r *postgresAdmissionRepo) ApplyAcceptance(ctx context.Context, cmd posts.ApplyAcceptanceCommand) (posts.AdmissionResult, error) { - // On a subject this AppView has no row for, the acceptance is the first - // thing known about it, and the CID the community pinned is the only - // content identifier available — so it is recorded as evaluated. A post - // event that later lands a different CID moves the row to - // pending_reacceptance through the ordinary path. + if err := validateWatermark("ApplyAcceptance", cmd.CommunityDID, cmd.PostURI, cmd.Watermark); err != nil { + return posts.AdmissionResult{}, err + } + + // On a subject this AppView holds NO CONTENT for, the CID the community + // pinned is the only content identifier available — so it is recorded as + // evaluated and the row lands accepted. That covers two shapes of row: the + // absent subject (INSERT path) and the NULL-evaluated tombstone a restore + // commit's removal-delete half leaves when IT met the absent subject first + // (conflict path, hence the COALESCE — NULL evaluated_cid means "nothing + // recorded yet", never "content that mismatches", and treating it as a + // mismatch would make the two delivery orders of {removal-delete, + // acceptance} converge on different statuses). A post event that later + // lands a different CID moves the row to pending_reacceptance through the + // ordinary path. query := ` INSERT INTO community_post_admissions ( community_did, post_uri, status, @@ -178,10 +202,11 @@ func (r *postgresAdmissionRepo) ApplyAcceptance(ctx context.Context, cmd posts.A ) VALUES ($1, $2, 'accepted', $3, $4, $5, $5, $6, $7, NOW(), NOW()) ON CONFLICT (community_did, post_uri) DO UPDATE SET status = CASE - WHEN community_post_admissions.evaluated_cid IS NOT DISTINCT FROM excluded.accepted_cid + WHEN COALESCE(community_post_admissions.evaluated_cid, excluded.accepted_cid) = excluded.accepted_cid THEN 'accepted' ELSE 'pending_reacceptance' END, + evaluated_cid = COALESCE(community_post_admissions.evaluated_cid, excluded.accepted_cid), acceptance_uri = excluded.acceptance_uri, acceptance_rkey = excluded.acceptance_rkey, accepted_cid = excluded.accepted_cid, @@ -192,10 +217,10 @@ func (r *postgresAdmissionRepo) ApplyAcceptance(ctx context.Context, cmd posts.A updated_at = NOW()` + communityWatermarkGuard + ` RETURNING ` + admissionColumns - return r.compareAndSwap(ctx, "ApplyAcceptance", cmd.CommunityDID, cmd.PostURI, communityEventOutcome, + return r.compareAndSwap(ctx, "ApplyAcceptance", cmd.CommunityDID, cmd.PostURI, communityEventOutcome, rowRequired, query, cmd.CommunityDID, cmd.PostURI, cmd.AcceptanceURI, cmd.AcceptanceRkey, cmd.PinnedCID, - cmd.Watermark.Rev, int16(cmd.Watermark.OpRank)) + cmd.Watermark.Rev, int16(posts.CommunityOpPut)) } // ApplyAcceptanceDelete applies the deletion of the acceptance record. @@ -207,6 +232,10 @@ func (r *postgresAdmissionRepo) ApplyAcceptance(ctx context.Context, cmd posts.A // columns clear, which is the point: the row must record that this event was // seen, or the stale acceptance-create it superseded would apply on redelivery. func (r *postgresAdmissionRepo) ApplyAcceptanceDelete(ctx context.Context, cmd posts.CommunityDeleteCommand) (posts.AdmissionResult, error) { + if err := validateWatermark("ApplyAcceptanceDelete", cmd.CommunityDID, cmd.PostURI, cmd.Watermark); err != nil { + return posts.AdmissionResult{}, err + } + // A deletion for a subject with no row still inserts one: it is a // tombstone, and without it the acceptance-create this deletion supersedes // would apply the next time a feed replays it. @@ -229,8 +258,8 @@ func (r *postgresAdmissionRepo) ApplyAcceptanceDelete(ctx context.Context, cmd p updated_at = NOW()` + communityWatermarkGuard + ` RETURNING ` + admissionColumns - return r.compareAndSwap(ctx, "ApplyAcceptanceDelete", cmd.CommunityDID, cmd.PostURI, communityEventOutcome, - query, cmd.CommunityDID, cmd.PostURI, cmd.Watermark.Rev, int16(cmd.Watermark.OpRank)) + return r.compareAndSwap(ctx, "ApplyAcceptanceDelete", cmd.CommunityDID, cmd.PostURI, communityEventOutcome, rowRequired, + query, cmd.CommunityDID, cmd.PostURI, cmd.Watermark.Rev, int16(posts.CommunityOpDelete)) } // ApplyRemoval applies a community removal record write under the §5.2 @@ -245,6 +274,10 @@ func (r *postgresAdmissionRepo) ApplyAcceptanceDelete(ctx context.Context, cmd p // A removal with no prior acceptance is valid and indexes normally — // communities may remove pre-emptively — so the absent-row case inserts. func (r *postgresAdmissionRepo) ApplyRemoval(ctx context.Context, cmd posts.ApplyRemovalCommand) (posts.AdmissionResult, error) { + if err := validateWatermark("ApplyRemoval", cmd.CommunityDID, cmd.PostURI, cmd.Watermark); err != nil { + return posts.AdmissionResult{}, err + } + query := ` INSERT INTO community_post_admissions ( community_did, post_uri, status, decision_code, decision_at, @@ -262,19 +295,26 @@ func (r *postgresAdmissionRepo) ApplyRemoval(ctx context.Context, cmd posts.Appl updated_at = NOW()` + communityWatermarkGuard + ` RETURNING ` + admissionColumns - return r.compareAndSwap(ctx, "ApplyRemoval", cmd.CommunityDID, cmd.PostURI, communityEventOutcome, + return r.compareAndSwap(ctx, "ApplyRemoval", cmd.CommunityDID, cmd.PostURI, communityEventOutcome, rowRequired, query, cmd.CommunityDID, cmd.PostURI, cmd.DecisionCode, - cmd.Watermark.Rev, int16(cmd.Watermark.OpRank)) + cmd.Watermark.Rev, int16(posts.CommunityOpPut)) } // ApplyRemovalDelete applies the deletion of the removal record. // // Withdrawing a removal returns the subject to pending and drops the decision // with it — a row that is no longer removed must not keep the code a reader -// would render. It does not restore any acceptance: the moderator's restore -// commit carries a FRESH acceptance alongside this deletion, and that -// acceptance is what makes the post visible again. +// would render. Redrivable resets to true with it: the standing decision is +// what justified refusing to re-evaluate, and a pre-removal terminal rejection +// leaving redrivable=false behind would make the reopened pending row a post +// the redrive pass never judges. It does not restore any acceptance: the +// moderator's restore commit carries a FRESH acceptance alongside this +// deletion, and that acceptance is what makes the post visible again. func (r *postgresAdmissionRepo) ApplyRemovalDelete(ctx context.Context, cmd posts.CommunityDeleteCommand) (posts.AdmissionResult, error) { + if err := validateWatermark("ApplyRemovalDelete", cmd.CommunityDID, cmd.PostURI, cmd.Watermark); err != nil { + return posts.AdmissionResult{}, err + } + query := ` INSERT INTO community_post_admissions ( community_did, post_uri, status, @@ -293,13 +333,17 @@ func (r *postgresAdmissionRepo) ApplyRemovalDelete(ctx context.Context, cmd post WHEN community_post_admissions.status = 'removed' THEN NULL ELSE community_post_admissions.decision_at END, + redrivable = CASE + WHEN community_post_admissions.status = 'removed' THEN true + ELSE community_post_admissions.redrivable + END, last_community_rev = excluded.last_community_rev, last_community_op_rank = excluded.last_community_op_rank, updated_at = NOW()` + communityWatermarkGuard + ` RETURNING ` + admissionColumns - return r.compareAndSwap(ctx, "ApplyRemovalDelete", cmd.CommunityDID, cmd.PostURI, communityEventOutcome, - query, cmd.CommunityDID, cmd.PostURI, cmd.Watermark.Rev, int16(cmd.Watermark.OpRank)) + return r.compareAndSwap(ctx, "ApplyRemovalDelete", cmd.CommunityDID, cmd.PostURI, communityEventOutcome, rowRequired, + query, cmd.CommunityDID, cmd.PostURI, cmd.Watermark.Rev, int16(posts.CommunityOpDelete)) } // RepinAcceptedCID moves a standing acceptance onto new content without @@ -316,16 +360,25 @@ func (r *postgresAdmissionRepo) ApplyRemovalDelete(ctx context.Context, cmd post // // The caller establishes that the diff touches only bridgedStats and that the // author passes the bridge-trust gate. What this enforces is narrower and -// structural: a repin updates an acceptance that STANDS. Without that -// precondition a repin arriving at a removed post — whose removal cleared the -// acceptance — would write a fresh accepted_cid onto it, which is exactly the -// live-acceptance-on-a-removed-post state the removal path takes care to -// prevent. +// structural: a repin updates an acceptance that STANDS on a row that is +// `accepted` NOW. Both halves of that guard earn their place. Without the +// acceptance-columns check, a repin arriving at a removed post — whose removal +// cleared the acceptance — would write a fresh accepted_cid onto it, which is +// exactly the live-acceptance-on-a-removed-post state the removal path takes +// care to prevent. Without the status check, a repin arriving at +// pending_reacceptance — which still CARRIES the acceptance columns — would +// move both CIDs under a status that says the acceptance does not cover the +// current content, silently converting an author's real edit into a +// stats-only refresh. // // It is an UPDATE rather than an upsert for the same reason: a repin must never // be able to CREATE an admission. Manufacturing an accepted row from a bridge's // refresh would be an auto-admission the community never wrote. func (r *postgresAdmissionRepo) RepinAcceptedCID(ctx context.Context, cmd posts.RepinAcceptanceCommand) (posts.AdmissionResult, error) { + if err := validateWatermark("RepinAcceptedCID", cmd.CommunityDID, cmd.PostURI, cmd.Watermark); err != nil { + return posts.AdmissionResult{}, err + } + query := ` UPDATE community_post_admissions SET accepted_cid = $3, @@ -335,14 +388,15 @@ func (r *postgresAdmissionRepo) RepinAcceptedCID(ctx context.Context, cmd posts. updated_at = NOW() WHERE community_did = $1 AND post_uri = $2 + AND status = 'accepted' AND acceptance_uri IS NOT NULL AND (last_community_rev IS NULL OR (last_community_rev, last_community_op_rank) < ($4::text COLLATE "C", $5::smallint)) RETURNING ` + admissionColumns - return r.compareAndSwap(ctx, "RepinAcceptedCID", cmd.CommunityDID, cmd.PostURI, repinOutcome, + return r.compareAndSwap(ctx, "RepinAcceptedCID", cmd.CommunityDID, cmd.PostURI, repinOutcome, rowOptional, query, cmd.CommunityDID, cmd.PostURI, cmd.PinnedCID, - cmd.Watermark.Rev, int16(cmd.Watermark.OpRank)) + cmd.Watermark.Rev, int16(posts.CommunityOpPut)) } // RecordRejection records the AppView's OWN decision not to admit a post. @@ -354,29 +408,39 @@ func (r *postgresAdmissionRepo) RepinAcceptedCID(ctx context.Context, cmd posts. // evaluated_cid either — the rejection judges the content already recorded, it // does not change what was judged. // +// The guard is the engine's read made safe: rejection's ONLY legal source +// state is `pending` (§5.5 — re-acceptance failure is expressed as a removal, +// and a rejection landing on accepted or pending_reacceptance would suppress a +// live community acceptance with a local decision), and the row must still +// hold the exact CID the verdict judged, or an author's edit slipped in +// between the engine's read and its write and the verdict is about content +// that no longer exists. A replay of the decision already recorded meets +// status = 'rejected' and is refused without touching a byte — no re-stamped +// decision_at, however many feeds redrive it. +// +// It is an UPDATE, never an insert: the engine rejects rows it read from its +// own queue, so a subject with NO row is a caller bug that must surface as an +// error, not a decision to be recorded against nothing. +// // Redrivable is the caller's classification of WHY, stored verbatim: a policy // rejection is terminal and must not be retried by the dead-letter pass, while // a transient evaluation failure has to stay retryable. func (r *postgresAdmissionRepo) RecordRejection(ctx context.Context, cmd posts.RecordRejectionCommand) (posts.AdmissionResult, error) { - // Guarded against `removed` because a removal is the moderator's decision - // and a rejection is ours. Overwriting decision_code there would replace - // what a reader renders as #removedPost with a code the community never - // issued. query := ` - INSERT INTO community_post_admissions ( - community_did, post_uri, status, decision_code, decision_at, redrivable, created_at, updated_at - ) VALUES ($1, $2, 'rejected', $3, NOW(), $4, NOW(), NOW()) - ON CONFLICT (community_did, post_uri) DO UPDATE SET + UPDATE community_post_admissions SET status = 'rejected', - decision_code = excluded.decision_code, - decision_at = excluded.decision_at, - redrivable = excluded.redrivable, + decision_code = $3, + decision_at = NOW(), + redrivable = $4, updated_at = NOW() - WHERE community_post_admissions.status <> 'removed' + WHERE community_did = $1 + AND post_uri = $2 + AND status = 'pending' + AND evaluated_cid IS NOT DISTINCT FROM $5 RETURNING ` + admissionColumns - return r.compareAndSwap(ctx, "RecordRejection", cmd.CommunityDID, cmd.PostURI, nonCommunityEventOutcome, - query, cmd.CommunityDID, cmd.PostURI, cmd.DecisionCode, cmd.Redrivable) + return r.compareAndSwap(ctx, "RecordRejection", cmd.CommunityDID, cmd.PostURI, rejectionOutcome, rowRequired, + query, cmd.CommunityDID, cmd.PostURI, cmd.DecisionCode, cmd.Redrivable, cmd.JudgedCID) } // GetByPostURIs returns every community's decision about each of the given @@ -616,50 +680,95 @@ func communityEventOutcome(wrote bool, _ *posts.Admission) posts.AdmissionOutcom return posts.AdmissionSkippedStale } -// nonCommunityEventOutcome classifies the two mutations that carry no -// watermark — the author-repo content observation and the AppView's own -// rejection — where a write and an applied transition are not the same thing. +// upsertPendingOutcome classifies the author-repo content observation, where a +// write and an applied transition are not the same thing. // -// Both are refused by the same thing: removal is terminal against everything -// except a community event that outranks it. UpsertPending still records the -// content a removed row now holds, so it WRITES while the transition it asked -// for was refused; that is skipped_terminal, audit columns notwithstanding. -// Anything else that wrote applied, and anything that did not write met a row -// already holding what it carried. -func nonCommunityEventOutcome(wrote bool, current *posts.Admission) posts.AdmissionOutcome { +// A delivery that wrote NOTHING met a row already holding exactly this content +// — a multi-feed duplicate — and that is skipped_stale whatever the row's +// status, a removed row included: nothing was recorded, so nothing was +// refused. A delivery that wrote but met `removed` recorded audit columns +// while the transition it asked for was refused by removal terminality +// (§5.5); that is skipped_terminal. Anything else that wrote applied. +func upsertPendingOutcome(wrote bool, current *posts.Admission) posts.AdmissionOutcome { + if !wrote { + return posts.AdmissionSkippedStale + } if current.Status == posts.AdmissionStatusRemoved { return posts.AdmissionSkippedTerminal } + return posts.AdmissionApplied +} + +// rejectionOutcome classifies the AppView's own rejection, whose guard refuses +// in two honestly different ways. A row still `pending` refused because it no +// longer holds the judged CID — the verdict is about content that has been +// edited away, which is ordering skew: skipped_stale. A row already `rejected` +// met a replay of the decision it records: skipped_stale too. Any other status +// (accepted, pending_reacceptance, removed) refuses by STATE — a community +// decision or a standing acceptance outranks a local verdict regardless of +// when it arrives: skipped_terminal. +func rejectionOutcome(wrote bool, current *posts.Admission) posts.AdmissionOutcome { if wrote { return posts.AdmissionApplied } - return posts.AdmissionSkippedStale + switch current.Status { + case posts.AdmissionStatusPending, posts.AdmissionStatusRejected: + return posts.AdmissionSkippedStale + default: + return posts.AdmissionSkippedTerminal + } } // repinOutcome classifies a bridge repin, whose guard has two halves and so two -// ways to refuse. No standing acceptance to move is the row's state refusing -// the transition regardless of ordering; anything else is the watermark. +// ways to refuse. A row that is not `accepted` with its acceptance standing is +// the row's state refusing the transition regardless of ordering; anything +// else is the watermark. func repinOutcome(wrote bool, current *posts.Admission) posts.AdmissionOutcome { if wrote { return posts.AdmissionApplied } - if current.AcceptanceURI == nil { + if current.Status != posts.AdmissionStatusAccepted || current.AcceptanceURI == nil { return posts.AdmissionSkippedTerminal } return posts.AdmissionSkippedStale } -// compareAndSwap runs one guarded upsert and reports the row it left behind. +// validateWatermark refuses a community event whose watermark cannot have come +// off the wire. Every Jetstream commit carries a rev, so an empty one is an +// upstream decoding bug — a genuine error bound for the dead-letter queue, not +// a skip — and stamping it would write a clock value that never existed onto +// the row (see posts.ErrInvalidWatermark). +func validateWatermark(operation, communityDID, postURI string, watermark posts.CommunityWatermark) error { + if watermark.Rev == "" { + return fmt.Errorf("%s for %s in %s: %w: empty rev", operation, postURI, communityDID, posts.ErrInvalidWatermark) + } + return nil +} + +// rowExpectation states whether a mutation may legitimately find NO row when +// its guard refuses. The upserts insert on an absent subject, so for them a +// missing row after a refusal is impossible and reads as corruption; the +// UPDATE-shaped mutations differ, and only the repin treats absence as an +// ordinary state (see compareAndSwap). +const ( + rowRequired = false + rowOptional = true +) + +// compareAndSwap runs one guarded mutation and reports the row it left behind. // // The transaction exists for the refusal path. A guard that fails returns no -// rows, so the current state has to be read — and reading it in the same -// transaction means reading it under the row lock the failed ON CONFLICT DO -// UPDATE already holds, which is what makes the returned row provably the row -// the guard was evaluated against. +// rows, so the current state has to be read — and the classifying read locks +// the row (FOR UPDATE) in the same transaction, which is what makes the +// returned row provably the row the guard was evaluated against. A refused ON +// CONFLICT DO UPDATE already holds the conflicting row's lock, but a guarded +// UPDATE that matched nothing holds none, and without the explicit lock a +// concurrent consumer could rewrite the row between the refusal and the read. func (r *postgresAdmissionRepo) compareAndSwap( ctx context.Context, operation, communityDID, postURI string, classify admissionOutcome, + mayLackRow bool, query string, args ...interface{}, ) (posts.AdmissionResult, error) { @@ -687,18 +796,25 @@ func (r *postgresAdmissionRepo) compareAndSwap( admission, err = scanAdmission(tx.QueryRowContext(ctx, `SELECT `+admissionColumns+` FROM community_post_admissions - WHERE community_did = $1 AND post_uri = $2`, + WHERE community_did = $1 AND post_uri = $2 + FOR UPDATE`, communityDID, postURI)) if errors.Is(err, sql.ErrNoRows) { - // Only the UPDATE-shaped mutations reach this. The upserts insert - // when the subject is absent, so their guard can refuse nothing but - // an existing row; a repin, which must never create an admission, - // simply has no row to move. That is a state the row refuses by not - // existing, and it is not an error — the ordering skew that - // delivers a community event early delivers a bridge refresh early - // too, and routing it to the dead-letter queue would bury it among - // genuine failures. - return posts.AdmissionResult{Outcome: posts.AdmissionSkippedTerminal}, nil + // Only the UPDATE-shaped mutations can reach this — the upserts + // insert when the subject is absent, so their guard can refuse + // nothing but an existing row. Whether absence is a state or a bug + // is per-operation. A repin, which must never create an admission, + // simply has no acceptance to move: the ordering skew that delivers + // a community event early delivers a bridge refresh early too, so + // that is an outcome. A rejection, by contrast, is the engine + // writing a verdict for a row it read from its own queue; no row + // means the caller judged nothing, and burying that as a skip + // would hide a genuine bug from the dead-letter queue. + if mayLackRow { + return posts.AdmissionResult{Outcome: posts.AdmissionSkippedTerminal}, nil + } + return posts.AdmissionResult{}, fmt.Errorf( + "%s for %s in %s: no admission row to decide against: %w", operation, postURI, communityDID, posts.ErrNotFound) } } if err != nil { diff --git a/internal/db/postgres/admission_repo_concurrency_test.go b/internal/db/postgres/admission_repo_concurrency_test.go index 60d0d4a..756971a 100644 --- a/internal/db/postgres/admission_repo_concurrency_test.go +++ b/internal/db/postgres/admission_repo_concurrency_test.go @@ -168,3 +168,197 @@ func TestAdmissionRepo_ConcurrentCommunityEventsConverge(t *testing.T) { "eight goroutines raced for the INSERT branch; the primary key has to be what arbitrates that, "+ "not a SELECT that decided the row was absent") } + +func TestAdmissionRepo_SameTupleDuplicateAppliesExactlyOnce(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // The multi-feed overlap case in its purest form: two consumers deliver the + // SAME event — identical tuple, identical payload — simultaneously, to a + // subject with no row yet. Exactly one delivery may apply; the other must be + // the equal-tuple replay, and the row must not betray that it happened twice. + subject := newAdmissionSubject(t, db) + rev := testkit.TID() + const duplicates = 2 + + results := make([]posts.AdmissionResult, duplicates) + failures := make([]error, duplicates) + + var waitGroup sync.WaitGroup + waitGroup.Add(duplicates) + for i := 0; i < duplicates; i++ { + go func(i int) { + defer waitGroup.Done() + results[i], failures[i] = repo.ApplyRemoval(ctx, posts.ApplyRemovalCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + DecisionCode: "duplicate_delivery", + Watermark: posts.CommunityWatermark{Rev: rev, OpRank: posts.CommunityOpPut}, + }) + }(i) + } + waitGroup.Wait() + + applied, skipped := 0, 0 + for i := 0; i < duplicates; i++ { + require.NoErrorf(t, failures[i], "delivery %d: a duplicate is the system working, never an error", i) + switch results[i].Outcome { + case posts.AdmissionApplied: + applied++ + case posts.AdmissionSkippedStale: + skipped++ + default: + assert.Failf(t, "unexpected duplicate outcome", "delivery %d returned %q", i, results[i].Outcome) + } + } + assert.Equal(t, 1, applied, "exactly one of two identical deliveries may apply") + assert.Equal(t, 1, skipped, "the other must be refused as the equal-tuple replay it is") + + final, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + for i := 0; i < duplicates; i++ { + if results[i].Outcome == posts.AdmissionApplied { + assert.Equal(t, results[i].Admission, final, + "the row must be byte-stable: the refused duplicate re-stamped nothing, updated_at included") + } else { + assert.Equal(t, results[i].Admission, final, + "the refused delivery must have been shown the applied row — the same row the guard was evaluated against") + } + } + + var rowCount int + require.NoError(t, db.QueryRowContext(ctx, ` + SELECT count(*) FROM community_post_admissions WHERE community_did = $1 AND post_uri = $2 + `, subject.CommunityDID, subject.PostURI).Scan(&rowCount)) + assert.Equal(t, 1, rowCount, "two goroutines raced the INSERT branch; the primary key arbitrates, one row results") +} + +func TestAdmissionRepo_AuthorEditsRacingCommunityEvents(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // Production's actual mix: the author edits (fast path AND firehose carry + // the same events) while the community's decisions stream in from + // overlapping feeds. Author events carry no watermark, so they slot into + // the community-event race at arbitrary points — and whatever interleaving + // the scheduler picks, the highest community tuple is a removal and removal + // is terminal against author events, so the final STATUS is not negotiable. + subject := newAdmissionSubject(t, db) + + const communityEventCount = 6 + const authorEditCount = 4 + const finalDecisionCode = "highest_tuple_removal" + + revs := increasingRevs(t, communityEventCount) + + type racedEvent struct { + describe string + isAuthor bool + apply func() (posts.AdmissionResult, error) + } + + var events []racedEvent + for i, rev := range revs { + rev := rev + switch { + case i == communityEventCount-1: + events = append(events, racedEvent{"removal (highest tuple)", false, func() (posts.AdmissionResult, error) { + return repo.ApplyRemoval(ctx, posts.ApplyRemovalCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + DecisionCode: finalDecisionCode, + Watermark: posts.CommunityWatermark{Rev: rev, OpRank: posts.CommunityOpPut}, + }) + }}) + case i%2 == 0: + events = append(events, racedEvent{"acceptance", false, func() (posts.AdmissionResult, error) { + acceptanceURI, acceptanceRkey := acceptanceRecord(t, subject.CommunityDID) + return repo.ApplyAcceptance(ctx, posts.ApplyAcceptanceCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + AcceptanceURI: acceptanceURI, + AcceptanceRkey: acceptanceRkey, + PinnedCID: contentCID(t, "race"), + Watermark: posts.CommunityWatermark{Rev: rev, OpRank: posts.CommunityOpPut}, + }) + }}) + default: + events = append(events, racedEvent{"acceptance deletion", false, func() (posts.AdmissionResult, error) { + return repo.ApplyAcceptanceDelete(ctx, posts.CommunityDeleteCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + Watermark: posts.CommunityWatermark{Rev: rev, OpRank: posts.CommunityOpDelete}, + }) + }}) + } + } + for i := 0; i < authorEditCount; i++ { + cid := contentCID(t, "edit") + events = append(events, racedEvent{"author edit", true, func() (posts.AdmissionResult, error) { + return repo.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + EvaluatedCID: cid, + }) + }}) + } + + shuffled := make([]int, len(events)) + for i := range shuffled { + shuffled[i] = i + } + rand.New(rand.NewSource(20260807)).Shuffle(len(shuffled), func(i, j int) { + shuffled[i], shuffled[j] = shuffled[j], shuffled[i] + }) + + outcomes := make([]posts.AdmissionOutcome, len(events)) + failures := make([]error, len(events)) + + var waitGroup sync.WaitGroup + waitGroup.Add(len(events)) + for _, index := range shuffled { + go func(index int) { + defer waitGroup.Done() + result, err := events[index].apply() + outcomes[index], failures[index] = result.Outcome, err + }(index) + } + waitGroup.Wait() + + for i, err := range failures { + require.NoErrorf(t, err, "%s (event %d): contention is not an error condition", events[i].describe, i) + switch outcomes[i] { + case posts.AdmissionApplied, posts.AdmissionSkippedStale: + case posts.AdmissionSkippedTerminal: + assert.Truef(t, events[i].isAuthor, + "%s (event %d): only an author event may be refused as terminal here — every community tuple is distinct", + events[i].describe, i) + default: + assert.Failf(t, "unexpected outcome under contention", "%s (event %d) returned %q", + events[i].describe, i, outcomes[i]) + } + } + + final, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + + assert.Equal(t, posts.AdmissionStatusRemoved, final.Status, + "the highest community tuple is a removal and removal is terminal against author events, whatever the interleaving") + assertNullableString(t, finalDecisionCode, final.DecisionCode, "decision_code") + assertWatermark(t, revs[communityEventCount-1], posts.CommunityOpPut, final.LastCommunityEvent) + assert.Nil(t, final.AcceptanceURI, "the winning removal must have cleared the acceptance columns") + assert.Nil(t, final.AcceptanceRkey) + assert.Nil(t, final.AcceptedCID) + + var rowCount int + require.NoError(t, db.QueryRowContext(ctx, ` + SELECT count(*) FROM community_post_admissions WHERE community_did = $1 AND post_uri = $2 + `, subject.CommunityDID, subject.PostURI).Scan(&rowCount)) + assert.Equal(t, 1, rowCount, "however the race lands, there is exactly one row per subject") +} diff --git a/internal/db/postgres/admission_repo_lifecycle_test.go b/internal/db/postgres/admission_repo_lifecycle_test.go index f3a249c..9a147c0 100644 --- a/internal/db/postgres/admission_repo_lifecycle_test.go +++ b/internal/db/postgres/admission_repo_lifecycle_test.go @@ -42,7 +42,6 @@ import ( type admissionSubject struct { CommunityDID string PostURI string - AuthorDID string } // newAdmissionSubject seeds a real community and a real post row, then returns @@ -65,7 +64,7 @@ func newAdmissionSubject(t *testing.T, db *sql.DB) admissionSubject { authorDID := fixtures.DID(testkit.UniqueID(t)) postURI := fixtures.Post(t, db, communityDID, authorDID, "a post seeking admission", 0, time.Now()) - return admissionSubject{CommunityDID: communityDID, PostURI: postURI, AuthorDID: authorDID} + return admissionSubject{CommunityDID: communityDID, PostURI: postURI} } // increasingRevs returns n real atProto TIDs whose lexicographic order is their diff --git a/internal/db/postgres/admission_repo_matrix_test.go b/internal/db/postgres/admission_repo_matrix_test.go index 421fcfb..4930a20 100644 --- a/internal/db/postgres/admission_repo_matrix_test.go +++ b/internal/db/postgres/admission_repo_matrix_test.go @@ -72,6 +72,7 @@ func rejectedSubject(t *testing.T, db *sql.DB, repo posts.AdmissionRepository, c CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, DecisionCode: decisionCode, + JudgedCID: cid, Redrivable: redrivable, }) require.NoError(t, err, "seeding the rejection") @@ -554,7 +555,8 @@ func TestAdmissionRepo_RecordRejection(t *testing.T) { // A subject that HAS a watermark, so "does not touch it" is // observable rather than vacuous: accepted, then the acceptance is // withdrawn, and the engine re-runs admitPost on what is left. - subject := acceptedSubject(t, db, repo, contentCID(t, "judged"), revs[0]) + judgedCID := contentCID(t, "judged") + subject := acceptedSubject(t, db, repo, judgedCID, revs[0]) _, err := repo.ApplyAcceptanceDelete(ctx, posts.CommunityDeleteCommand{ CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, @@ -570,6 +572,7 @@ func TestAdmissionRepo_RecordRejection(t *testing.T) { CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, DecisionCode: "banned_author", + JudgedCID: judgedCID, Redrivable: redrivable, }) require.NoError(t, err) @@ -589,3 +592,734 @@ func TestAdmissionRepo_RecordRejection(t *testing.T) { }) } } + +// --------------------------------------------------------------------------- +// RecordRejection as a full CAS (review finding 1) +// --------------------------------------------------------------------------- + +func TestAdmissionRepo_RecordRejectionCAS(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + const decisionCode = "rule_violation" + + t.Run("a rejection cannot land on an accepted row", func(t *testing.T) { + // Rejection's ONLY legal source state is pending (§5.5: re-acceptance + // failure is a removal, not a rejection). A rejection overwriting an + // accepted row would suppress a live community acceptance with a local + // decision — the exact inversion of authority RecordRejection's + // no-watermark rule exists to prevent. + cid := contentCID(t, "live") + subject := acceptedSubject(t, db, repo, cid, testkit.TID()) + + before, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + + result, err := repo.RecordRejection(ctx, posts.RecordRejectionCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + DecisionCode: decisionCode, + JudgedCID: cid, + Redrivable: false, + }) + require.NoError(t, err, "a refused rejection is an outcome, not an error") + assert.Equal(t, posts.AdmissionSkippedTerminal, result.Outcome, + "an accepted row refuses a rejection by its state, not by ordering") + + after, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + assert.Equal(t, before, after, + "a refused rejection must leave the accepted row byte-identical, updated_at included") + }) + + t.Run("a rejection cannot land on a row awaiting re-acceptance", func(t *testing.T) { + // pending_reacceptance still carries a live acceptance record; §5.5 + // says re-acceptance failure is expressed as a REMOVAL. A rejection + // here would strand the acceptance columns under a local decision. + subject := acceptedSubject(t, db, repo, contentCID(t, "orig"), testkit.TID()) + editedCID := contentCID(t, "edited") + _, err := repo.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + EvaluatedCID: editedCID, + }) + require.NoError(t, err) + + before, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + require.Equal(t, posts.AdmissionStatusPendingReacceptance, before.Status) + + result, err := repo.RecordRejection(ctx, posts.RecordRejectionCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + DecisionCode: decisionCode, + JudgedCID: editedCID, + Redrivable: false, + }) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionSkippedTerminal, result.Outcome) + + after, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + assert.Equal(t, before, after, "the refused rejection must not disturb the standing acceptance") + }) + + t.Run("a rejection judging a CID the row no longer holds is refused", func(t *testing.T) { + // The engine read the row, judged its content, and wrote the verdict — + // but the author edited in between. The verdict judged content that no + // longer exists and must not land on the new content. + subject := newAdmissionSubject(t, db) + judgedCID := contentCID(t, "judged") + editedCID := contentCID(t, "edited") + + _, err := repo.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + EvaluatedCID: editedCID, + }) + require.NoError(t, err) + + before, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + + result, err := repo.RecordRejection(ctx, posts.RecordRejectionCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + DecisionCode: decisionCode, + JudgedCID: judgedCID, + Redrivable: false, + }) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionSkippedStale, result.Outcome, + "a verdict for content the row does not hold is stale, and the new content awaits its own judgment") + + after, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionStatusPending, after.Status, + "the row stays pending: the edited content has never been judged") + assert.Equal(t, before, after, "the refused rejection must leave the row byte-identical") + }) + + t.Run("a re-delivered rejection is a byte-identical no-op", func(t *testing.T) { + cid := contentCID(t, "judged") + subject := rejectedSubject(t, db, repo, cid, decisionCode, false) + + before, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + + result, err := repo.RecordRejection(ctx, posts.RecordRejectionCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + DecisionCode: decisionCode, + JudgedCID: cid, + Redrivable: false, + }) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionSkippedStale, result.Outcome, + "an exact duplicate of the decision already recorded is a replay") + + after, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + assert.Equal(t, before, after, + "the duplicate must not re-stamp decision_at or updated_at — deep-equal, not merely same status") + }) + + t.Run("a rejection and an author edit converge in either order", func(t *testing.T) { + // Whichever of the two lands first, the end state must be the same: + // pending on the edited content, never `rejected` carrying the edited + // CID — the rejection judged the OLD content only. + judgedCID := contentCID(t, "judged") + editedCID := contentCID(t, "edited") + + assertConverged := func(t *testing.T, subject admissionSubject) { + t.Helper() + final, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionStatusPending, final.Status) + assertNullableString(t, editedCID, final.EvaluatedCID, "evaluated_cid") + assert.Nil(t, final.DecisionCode, "no decision may survive against content it never judged") + assert.Nil(t, final.DecisionAt) + assert.True(t, final.Redrivable, "the edited content must be evaluable") + } + + t.Run("edit first, then the stale rejection", func(t *testing.T) { + subject := newAdmissionSubject(t, db) + _, err := repo.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, EvaluatedCID: judgedCID, + }) + require.NoError(t, err) + _, err = repo.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, EvaluatedCID: editedCID, + }) + require.NoError(t, err) + + result, err := repo.RecordRejection(ctx, posts.RecordRejectionCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + DecisionCode: decisionCode, JudgedCID: judgedCID, Redrivable: false, + }) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionSkippedStale, result.Outcome) + assertConverged(t, subject) + }) + + t.Run("rejection first, then the edit reopens it", func(t *testing.T) { + subject := rejectedSubject(t, db, repo, judgedCID, decisionCode, false) + + result, err := repo.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, EvaluatedCID: editedCID, + }) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionApplied, result.Outcome) + assertConverged(t, subject) + }) + }) +} + +func TestAdmissionRepo_RecordRejectionOnAnUnseenSubjectIsAnError(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + + // The engine rejects rows it read from its own queue. A subject with no row + // cannot have been read from anything, so its absence is a caller bug — a + // genuine error for the dead-letter queue, not delivery skew to skip over. + subject := newAdmissionSubject(t, db) + + _, err := repo.RecordRejection(context.Background(), posts.RecordRejectionCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + DecisionCode: "rule_violation", + JudgedCID: contentCID(t, "unseen"), + Redrivable: false, + }) + require.Error(t, err, "rejecting a subject the AppView has never recorded must be an error") + assert.ErrorIs(t, err, posts.ErrNotFound) + + _, getErr := repo.Get(context.Background(), subject.CommunityDID, subject.PostURI) + assert.ErrorIs(t, getErr, posts.ErrNotFound, "the failed rejection must not have manufactured a row") +} + +// --------------------------------------------------------------------------- +// OpRank is derived from the operation, never taken from the caller (finding 4) +// --------------------------------------------------------------------------- + +func TestAdmissionRepo_OpRankIsDerivedFromTheOperation(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // Every command below carries the WRONG rank on purpose. The rank IS the + // operation's kind — rank(delete)=0 < rank(put)=1 — so it is derived inside + // each method; a caller that could assert otherwise could reorder a commit + // with one mislabeled event. The row's stored watermark is the proof. + + t.Run("ApplyAcceptance stamps a put rank", func(t *testing.T) { + subject := newAdmissionSubject(t, db) + rev := testkit.TID() + acceptanceURI, acceptanceRkey := acceptanceRecord(t, subject.CommunityDID) + result, err := repo.ApplyAcceptance(ctx, posts.ApplyAcceptanceCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + AcceptanceURI: acceptanceURI, AcceptanceRkey: acceptanceRkey, + PinnedCID: contentCID(t, "rank"), + Watermark: posts.CommunityWatermark{Rev: rev, OpRank: posts.CommunityOpDelete}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, result.Outcome) + assertWatermark(t, rev, posts.CommunityOpPut, result.Admission.LastCommunityEvent) + }) + + t.Run("ApplyAcceptanceDelete stamps a delete rank", func(t *testing.T) { + subject := newAdmissionSubject(t, db) + rev := testkit.TID() + result, err := repo.ApplyAcceptanceDelete(ctx, posts.CommunityDeleteCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + Watermark: posts.CommunityWatermark{Rev: rev, OpRank: posts.CommunityOpPut}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, result.Outcome) + assertWatermark(t, rev, posts.CommunityOpDelete, result.Admission.LastCommunityEvent) + }) + + t.Run("ApplyRemoval stamps a put rank", func(t *testing.T) { + subject := newAdmissionSubject(t, db) + rev := testkit.TID() + result, err := repo.ApplyRemoval(ctx, posts.ApplyRemovalCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + DecisionCode: "rule_violation", + Watermark: posts.CommunityWatermark{Rev: rev, OpRank: posts.CommunityOpDelete}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, result.Outcome) + assertWatermark(t, rev, posts.CommunityOpPut, result.Admission.LastCommunityEvent) + }) + + t.Run("ApplyRemovalDelete stamps a delete rank", func(t *testing.T) { + subject := newAdmissionSubject(t, db) + rev := testkit.TID() + result, err := repo.ApplyRemovalDelete(ctx, posts.CommunityDeleteCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + Watermark: posts.CommunityWatermark{Rev: rev, OpRank: posts.CommunityOpPut}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, result.Outcome) + assertWatermark(t, rev, posts.CommunityOpDelete, result.Admission.LastCommunityEvent) + }) + + t.Run("RepinAcceptedCID stamps a put rank", func(t *testing.T) { + revs := increasingRevs(t, 2) + subject := acceptedSubject(t, db, repo, contentCID(t, "bridged"), revs[0]) + result, err := repo.RepinAcceptedCID(ctx, posts.RepinAcceptanceCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + PinnedCID: contentCID(t, "restats"), + Watermark: posts.CommunityWatermark{Rev: revs[1], OpRank: posts.CommunityOpDelete}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, result.Outcome) + assertWatermark(t, revs[1], posts.CommunityOpPut, result.Admission.LastCommunityEvent) + }) +} + +// --------------------------------------------------------------------------- +// Restore-vs-absent convergence for ApplyAcceptance (finding 2) +// --------------------------------------------------------------------------- + +func TestAdmissionRepo_AcceptanceOntoANullEvaluatedTombstoneLandsAccepted(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // A restore commit's removal-delete half arriving at an ABSENT subject + // inserts a tombstone with NO evaluated content. The acceptance half must + // still converge on accepted: the pinned CID is the only content identifier + // anyone has, exactly as on the acceptance-first insert path — NULL + // evaluated_cid means "nothing recorded yet", not "content that mismatches". + subject := newAdmissionSubject(t, db) + revs := increasingRevs(t, 2) + + tombstone, err := repo.ApplyRemovalDelete(ctx, posts.CommunityDeleteCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + Watermark: posts.CommunityWatermark{Rev: revs[0], OpRank: posts.CommunityOpDelete}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, tombstone.Outcome) + require.Nil(t, tombstone.Admission.EvaluatedCID, "the arrangement needs a NULL-evaluated tombstone") + + pinnedCID := contentCID(t, "restored") + acceptanceURI, acceptanceRkey := acceptanceRecord(t, subject.CommunityDID) + result, err := repo.ApplyAcceptance(ctx, posts.ApplyAcceptanceCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + AcceptanceURI: acceptanceURI, + AcceptanceRkey: acceptanceRkey, + PinnedCID: pinnedCID, + Watermark: posts.CommunityWatermark{Rev: revs[1], OpRank: posts.CommunityOpPut}, + }) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionApplied, result.Outcome) + + require.NotNil(t, result.Admission) + assert.Equal(t, posts.AdmissionStatusAccepted, result.Admission.Status, + "an acceptance meeting a row with NO recorded content must land accepted, not pending_reacceptance") + assertNullableString(t, pinnedCID, result.Admission.AcceptedCID, "accepted_cid") + assertNullableString(t, pinnedCID, result.Admission.EvaluatedCID, + "evaluated_cid: the pinned CID is the only content identifier available, same as the insert path records") +} + +func TestAdmissionRepo_RestoreCommitConvergesOnAnAbsentRow(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // The restore commit {removal-delete@(R,0), acceptance@(R,1)} delivered to + // a subject this AppView holds NO row for — a relay coverage gap swallowed + // the whole earlier history. Both delivery orders must leave the same row. + rev := testkit.TID() + + deliver := func(t *testing.T, subject admissionSubject, pinnedCID string, acceptanceFirst bool) *posts.Admission { + t.Helper() + acceptanceURI, acceptanceRkey := acceptanceRecord(t, subject.CommunityDID) + acceptance := func() { + result, err := repo.ApplyAcceptance(ctx, posts.ApplyAcceptanceCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + AcceptanceURI: acceptanceURI, AcceptanceRkey: acceptanceRkey, + PinnedCID: pinnedCID, + Watermark: posts.CommunityWatermark{Rev: rev, OpRank: posts.CommunityOpPut}, + }) + require.NoError(t, err) + require.NotNil(t, result.Admission) + } + removalDelete := func() { + _, err := repo.ApplyRemovalDelete(ctx, posts.CommunityDeleteCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + Watermark: posts.CommunityWatermark{Rev: rev, OpRank: posts.CommunityOpDelete}, + }) + require.NoError(t, err) + } + if acceptanceFirst { + acceptance() + removalDelete() + } else { + removalDelete() + acceptance() + } + final, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + return final + } + + assertRestored := func(t *testing.T, got *posts.Admission, pinnedCID string) { + t.Helper() + assert.Equal(t, posts.AdmissionStatusAccepted, got.Status) + assertNullableString(t, pinnedCID, got.AcceptedCID, "accepted_cid") + assertNullableString(t, pinnedCID, got.EvaluatedCID, "evaluated_cid") + assertWatermark(t, rev, posts.CommunityOpPut, got.LastCommunityEvent) + assert.Nil(t, got.DecisionCode) + assert.Nil(t, got.DecisionAt) + assert.True(t, got.Redrivable) + } + + t.Run("removal-delete first", func(t *testing.T) { + subject := newAdmissionSubject(t, db) + pinnedCID := contentCID(t, "convA") + assertRestored(t, deliver(t, subject, pinnedCID, false), pinnedCID) + }) + + t.Run("acceptance first", func(t *testing.T) { + subject := newAdmissionSubject(t, db) + pinnedCID := contentCID(t, "convB") + assertRestored(t, deliver(t, subject, pinnedCID, true), pinnedCID) + }) +} + +// --------------------------------------------------------------------------- +// RepinAcceptedCID applies to accepted rows ONLY (finding 3) +// --------------------------------------------------------------------------- + +func TestAdmissionRepo_RepinRefusalMatrix(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + const decisionCode = "rule_violation" + + repin := func(t *testing.T, subject admissionSubject, rev string) (posts.AdmissionResult, error) { + t.Helper() + return repo.RepinAcceptedCID(ctx, posts.RepinAcceptanceCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + PinnedCID: contentCID(t, "repin"), + Watermark: posts.CommunityWatermark{Rev: rev, OpRank: posts.CommunityOpPut}, + }) + } + + t.Run("an accepted row applies", func(t *testing.T) { + revs := increasingRevs(t, 2) + subject := acceptedSubject(t, db, repo, contentCID(t, "bridged"), revs[0]) + result, err := repin(t, subject, revs[1]) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionApplied, result.Outcome) + require.NotNil(t, result.Admission) + assert.Equal(t, posts.AdmissionStatusAccepted, result.Admission.Status) + }) + + // Every other status refuses by STATE — a repin re-decides nothing, so it + // has no business at a row that would need a decision — and the refusal + // must leave the row byte-identical. + for _, refusal := range []struct { + name string + arrange func(t *testing.T) admissionSubject + }{ + { + name: "a pending row refuses", + arrange: func(t *testing.T) admissionSubject { + subject := newAdmissionSubject(t, db) + _, err := repo.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + EvaluatedCID: contentCID(t, "pending"), + }) + require.NoError(t, err) + return subject + }, + }, + { + name: "a row awaiting re-acceptance refuses", + arrange: func(t *testing.T) admissionSubject { + // The dangerous one: the row still CARRIES an acceptance URI, so + // a guard on the acceptance columns alone would let the repin + // through — writing a fresh accepted_cid under a status that + // says the acceptance does not cover the current content. + subject := acceptedSubject(t, db, repo, contentCID(t, "orig"), testkit.TID()) + _, err := repo.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + EvaluatedCID: contentCID(t, "edited"), + }) + require.NoError(t, err) + return subject + }, + }, + { + name: "a rejected row refuses", + arrange: func(t *testing.T) admissionSubject { + return rejectedSubject(t, db, repo, contentCID(t, "judged"), decisionCode, false) + }, + }, + { + name: "a removed row refuses", + arrange: func(t *testing.T) admissionSubject { + revs := increasingRevs(t, 2) + return removedSubject(t, db, repo, contentCID(t, "removed"), revs[0], revs[1], decisionCode) + }, + }, + } { + t.Run(refusal.name, func(t *testing.T) { + subject := refusal.arrange(t) + before, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + + result, err := repin(t, subject, testkit.TID()) + require.NoError(t, err, "a refused repin is an outcome, not an error") + assert.Equal(t, posts.AdmissionSkippedTerminal, result.Outcome, + "the row's state refuses the repin regardless of ordering") + require.NotNil(t, result.Admission) + + after, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + assert.Equal(t, before, after, "a refused repin must leave the row byte-identical") + assert.Equal(t, after, result.Admission, "the refused caller must be shown the row that refused it") + }) + } + + t.Run("an absent subject refuses with no row to describe", func(t *testing.T) { + subject := newAdmissionSubject(t, db) + result, err := repin(t, subject, testkit.TID()) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionSkippedTerminal, result.Outcome) + assert.Nil(t, result.Admission, + "the documented nil-Admission shape: a repin may never CREATE a row, and there is genuinely nothing to report") + }) +} + +// --------------------------------------------------------------------------- +// ApplyRemovalDelete standalone, and acceptance-delete against removed (finding 5) +// --------------------------------------------------------------------------- + +func TestAdmissionRepo_StandaloneRemovalDeleteReopensEvaluation(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // A removal withdrawn with NO accompanying acceptance: the subject returns + // to pending with the decision gone — including redrivable, which a + // pre-removal terminal rejection may have left false. A pending row that + // the redrive pass refuses to evaluate is a post nobody will ever judge. + cid := contentCID(t, "judged") + subject := rejectedSubject(t, db, repo, cid, "spam", false) + + revs := increasingRevs(t, 2) + removed, err := repo.ApplyRemoval(ctx, posts.ApplyRemovalCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + DecisionCode: "rule_violation", + Watermark: posts.CommunityWatermark{Rev: revs[0], OpRank: posts.CommunityOpPut}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, removed.Outcome) + require.False(t, removed.Admission.Redrivable, + "the arrangement needs the terminal rejection's redrivable=false to survive into the removal") + + result, err := repo.ApplyRemovalDelete(ctx, posts.CommunityDeleteCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + Watermark: posts.CommunityWatermark{Rev: revs[1], OpRank: posts.CommunityOpDelete}, + }) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionApplied, result.Outcome) + + require.NotNil(t, result.Admission) + assert.Equal(t, posts.AdmissionStatusPending, result.Admission.Status, + "an un-removed post with no accompanying acceptance returns to pending") + assert.Nil(t, result.Admission.DecisionCode, "a row no longer removed must not keep the code a reader would render") + assert.Nil(t, result.Admission.DecisionAt) + assert.True(t, result.Admission.Redrivable, + "reopened evaluation must be redrivable: the standing decision is gone, so nothing justifies refusing to evaluate") + assertNullableString(t, cid, result.Admission.EvaluatedCID, "evaluated_cid: the recorded content survives the un-remove") + assertWatermark(t, revs[1], posts.CommunityOpDelete, result.Admission.LastCommunityEvent) +} + +func TestAdmissionRepo_AcceptanceDeleteAtARemovedRowAdvancesTheWatermarkOnly(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // A later community commit deletes an acceptance while the row is removed — + // e.g. cleanup of a stale acceptance record after the removal. Deleting an + // acceptance cannot un-remove a post, but the event WAS seen, and the + // watermark must say so or a replay of the superseded acceptance would apply. + revs := increasingRevs(t, 3) + subject := removedSubject(t, db, repo, contentCID(t, "removed"), revs[0], revs[1], "rule_violation") + + before, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + + result, err := repo.ApplyAcceptanceDelete(ctx, posts.CommunityDeleteCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + Watermark: posts.CommunityWatermark{Rev: revs[2], OpRank: posts.CommunityOpDelete}, + }) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionApplied, result.Outcome) + + require.NotNil(t, result.Admission) + assert.Equal(t, posts.AdmissionStatusRemoved, result.Admission.Status, + "deleting an acceptance withdraws it; it cannot un-remove a post") + assert.Equal(t, before.DecisionCode, result.Admission.DecisionCode, "the moderator's code must survive") + assert.Equal(t, before.DecisionAt, result.Admission.DecisionAt) + assertWatermark(t, revs[2], posts.CommunityOpDelete, result.Admission.LastCommunityEvent) +} + +// --------------------------------------------------------------------------- +// Hardening (finding 7) +// --------------------------------------------------------------------------- + +func TestAdmissionRepo_EmptyWatermarkRevIsAnError(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // Every Jetstream commit carries a rev, so an empty one is an upstream + // decoding bug — a genuine error for the dead-letter queue, never a skip. + // Stamping it would write a clock value that never existed onto the row. + subject := newAdmissionSubject(t, db) + empty := posts.CommunityWatermark{Rev: "", OpRank: posts.CommunityOpPut} + + acceptanceURI, acceptanceRkey := acceptanceRecord(t, subject.CommunityDID) + for name, call := range map[string]func() (posts.AdmissionResult, error){ + "ApplyAcceptance": func() (posts.AdmissionResult, error) { + return repo.ApplyAcceptance(ctx, posts.ApplyAcceptanceCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + AcceptanceURI: acceptanceURI, AcceptanceRkey: acceptanceRkey, + PinnedCID: contentCID(t, "empty"), Watermark: empty, + }) + }, + "ApplyAcceptanceDelete": func() (posts.AdmissionResult, error) { + return repo.ApplyAcceptanceDelete(ctx, posts.CommunityDeleteCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, Watermark: empty, + }) + }, + "ApplyRemoval": func() (posts.AdmissionResult, error) { + return repo.ApplyRemoval(ctx, posts.ApplyRemovalCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + DecisionCode: "rule_violation", Watermark: empty, + }) + }, + "ApplyRemovalDelete": func() (posts.AdmissionResult, error) { + return repo.ApplyRemovalDelete(ctx, posts.CommunityDeleteCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, Watermark: empty, + }) + }, + "RepinAcceptedCID": func() (posts.AdmissionResult, error) { + return repo.RepinAcceptedCID(ctx, posts.RepinAcceptanceCommand{ + CommunityDID: subject.CommunityDID, PostURI: subject.PostURI, + PinnedCID: contentCID(t, "empty"), Watermark: empty, + }) + }, + } { + t.Run(name, func(t *testing.T) { + _, err := call() + require.Error(t, err, "an empty rev must be refused before it reaches the row") + assert.ErrorIs(t, err, posts.ErrInvalidWatermark) + }) + } + + _, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + assert.ErrorIs(t, err, posts.ErrNotFound, "no refused call may have written a row") +} + +func TestAdmissionRepo_UpsertPendingSameCIDOnARemovedRowIsAStaleNoOp(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // Re-delivery of the exact content a removed row already records: nothing + // is written — not even audit columns — so the honest label is the + // duplicate-delivery one, skipped_stale, exactly as for an accepted row + // meeting its own content again. skipped_terminal is reserved for the case + // where the observation DID record new content and only the removal kept + // the decision standing. + cid := contentCID(t, "removedsame") + revs := increasingRevs(t, 2) + subject := removedSubject(t, db, repo, cid, revs[0], revs[1], "rule_violation") + + before, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + + result, err := repo.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + EvaluatedCID: cid, + }) + require.NoError(t, err) + assert.Equal(t, posts.AdmissionSkippedStale, result.Outcome, + "the row already holds exactly this content; the delivery is a duplicate, not a refused transition") + + after, err := repo.Get(ctx, subject.CommunityDID, subject.PostURI) + require.NoError(t, err) + assert.Equal(t, before, after, "a duplicate must leave the removed row byte-identical, updated_at included") +} + +func TestAdmissionRepo_AcceptanceUpdateFreezesCreatedAt(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // created_at is when the SUBJECT entered the system — the moderation + // queue's keyset orders by it. An applied acceptance updating it would + // silently reorder the queue under every moderator paging through it. + subject := newAdmissionSubject(t, db) + cid := contentCID(t, "frozen") + + seeded, err := repo.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + EvaluatedCID: cid, + }) + require.NoError(t, err) + + acceptanceURI, acceptanceRkey := acceptanceRecord(t, subject.CommunityDID) + result, err := repo.ApplyAcceptance(ctx, posts.ApplyAcceptanceCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + AcceptanceURI: acceptanceURI, + AcceptanceRkey: acceptanceRkey, + PinnedCID: cid, + Watermark: posts.CommunityWatermark{Rev: testkit.TID(), OpRank: posts.CommunityOpPut}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, result.Outcome, "the freeze only means anything on an APPLIED update") + + assert.Equal(t, seeded.Admission.CreatedAt, result.Admission.CreatedAt, + "an applied acceptance must not move created_at: it is the row's queue position, not an audit column") +} diff --git a/internal/db/postgres/admission_repo_query_test.go b/internal/db/postgres/admission_repo_query_test.go index 60334c4..83c6713 100644 --- a/internal/db/postgres/admission_repo_query_test.go +++ b/internal/db/postgres/admission_repo_query_test.go @@ -6,12 +6,13 @@ import ( "context" "database/sql" "testing" + "time" "Coves/internal/core/posts" "Coves/tests/fixtures" "Coves/tests/testkit" - _ "github.com/lib/pq" + "github.com/lib/pq" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -242,3 +243,117 @@ func TestAdmissionRepo_ListByStatusForCommunity(t *testing.T) { assert.ErrorIs(t, err, posts.ErrInvalidCursor) }) } + +// seedPendingAdmissionsAt inserts n pending admission rows for one community in +// a single statement, all sharing exactly the given created_at. +// +// Direct SQL on purpose, twice over: the repo API stamps created_at with NOW() +// per statement, so it cannot manufacture an exact timestamp tie on demand, and +// one multi-row statement is what makes seeding a page-maximum's worth of rows +// cheap enough for the clamp test below. +func seedPendingAdmissionsAt(t *testing.T, db *sql.DB, communityDID string, n int, createdAt time.Time) []string { + t.Helper() + + uris := make([]string, n) + for i := range uris { + uris[i] = authorOwnedPostURI(t) + } + + _, err := db.ExecContext(context.Background(), ` + INSERT INTO community_post_admissions (community_did, post_uri, status, evaluated_cid, created_at, updated_at) + SELECT $1, uri, 'pending', 'bafyreiseeded', $3, $3 + FROM unnest($2::text[]) AS uri + `, communityDID, pq.Array(uris), createdAt) + require.NoErrorf(t, err, "seeding %d pending admissions at %s", n, createdAt) + + return uris +} + +func TestAdmissionRepo_ListByStatusForCommunityLimitClamps(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // The clamp values are pinned as literals, deliberately: asserting against + // the repo's constants would stay green if someone changed the constants to + // something unbounded, which is the exact regression this test exists to + // catch. 50 is admissionQueuePageSize, 200 is admissionQueuePageMaximum. + const ( + defaultPageSize = 50 + maximumPageSize = 200 + ) + + // One row more than the maximum, so both clamps are observable as a page + // size rather than inferable from internals: an unclamped limit would + // return all 201 rows. + queueCommunity := seedCommunities(t, db, 1)[0] + seedPendingAdmissionsAt(t, db, queueCommunity, maximumPageSize+1, time.Now().UTC()) + + t.Run("limit zero means the default page, not an empty or unbounded one", func(t *testing.T) { + page, cursor, err := repo.ListByStatusForCommunity(ctx, queueCommunity, posts.AdmissionStatusPending, 0, nil) + require.NoError(t, err) + assert.Len(t, page, defaultPageSize, + "limit 0 is a caller expressing no preference; it must get the default page size, not the whole queue and not nothing") + assert.NotNil(t, cursor, "the queue holds more than a default page, so the page must offer a cursor") + }) + + t.Run("a limit above the maximum is clamped to the maximum", func(t *testing.T) { + page, cursor, err := repo.ListByStatusForCommunity(ctx, queueCommunity, posts.AdmissionStatusPending, 100000, nil) + require.NoError(t, err) + assert.Len(t, page, maximumPageSize, + "an oversized limit must be clamped: an unbounded listing is a table scan a moderator's browser cannot render") + assert.NotNil(t, cursor, "the row past the maximum proves the clamp; the cursor must point at it") + }) +} + +func TestAdmissionRepo_ListByStatusForCommunityCreatedAtTieBreak(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + repo := NewAdmissionRepository(db) + ctx := context.Background() + + // Two rows sharing created_at to the microsecond — what rows written in one + // transaction genuinely look like — so the (created_at, post_uri) keyset's + // tie-break column is load-bearing: a cursor on created_at alone could not + // separate them and would either repeat the pair or skip its second half. + queueCommunity := seedCommunities(t, db, 1)[0] + sharedCreatedAt := time.Now().UTC().Truncate(time.Microsecond) + tiedURIs := seedPendingAdmissionsAt(t, db, queueCommunity, 2, sharedCreatedAt) + + full, cursor, err := repo.ListByStatusForCommunity(ctx, queueCommunity, posts.AdmissionStatusPending, 10, nil) + require.NoError(t, err) + require.Len(t, full, 2) + assert.Nil(t, cursor) + + // The order within a tie is deterministic — post_uri ascending, per the + // ORDER BY the keyset mirrors — not whatever the storage layer felt like. + wantFirst, wantSecond := tiedURIs[0], tiedURIs[1] + if wantSecond < wantFirst { + wantFirst, wantSecond = wantSecond, wantFirst + } + assert.Equal(t, wantFirst, full[0].PostURI, "equal created_at must order by post_uri ascending") + assert.Equal(t, wantSecond, full[1].PostURI, "equal created_at must order by post_uri ascending") + + // Page size one forces the cursor to land exactly ON the tie: the second + // page's keyset carries the shared created_at and must advance past the + // first row without skipping the second or repeating the first. + var paged []*posts.Admission + var pageCursor *string + for page := 0; page < 4; page++ { + batch, next, err := repo.ListByStatusForCommunity(ctx, queueCommunity, posts.AdmissionStatusPending, 1, pageCursor) + require.NoErrorf(t, err, "page %d", page) + + paged = append(paged, batch...) + pageCursor = next + if pageCursor == nil { + break + } + } + require.Nil(t, pageCursor, "paging across the tie did not terminate") + assert.Equal(t, full, paged, + "paging across an exact created_at tie must reconstruct the unpaged listing — a duplicate means the "+ + "cursor re-delivered the tied row, a missing row means it skipped the tie's second half") +} diff --git a/internal/db/postgres/admission_repo_schema_test.go b/internal/db/postgres/admission_repo_schema_test.go index 373a821..2eb7499 100644 --- a/internal/db/postgres/admission_repo_schema_test.go +++ b/internal/db/postgres/admission_repo_schema_test.go @@ -318,7 +318,10 @@ func TestMigration034_DownRestoresTheAuthorForeignKeyUnvalidated(t *testing.T) { require.NoError(t, err, "with fk_author dropped, a federated author's post must index even though no users row exists for them") - assert.EqualValues(t, 34, testkit.MigrateDownOne(t, db), + // The expected-version parameter is the tripwire: when migration 035 lands, + // this call fails with the remedy in its message instead of silently + // rolling back 035's Down and leaving 034's untested. + assert.EqualValues(t, 34, testkit.MigrateDownOne(t, db, 34), "this test asserts on 034's Down section; rolling back a different migration would prove nothing about it") var surviving int @@ -488,3 +491,46 @@ func normalizePredicate(predicate string) string { predicate = strings.ReplaceAll(predicate, ")", "") return strings.ReplaceAll(predicate, " ", "") } + +func TestAdmissionsTable_OpRankCheck(t *testing.T) { + t.Parallel() + + // The op-rank vocabulary is exactly {0 = delete, 1 = put} (§5.2). The rank + // is half of the ordering tuple, and SMALLINT admits 32766 values that + // would silently outrank every genuine put forever — a CHECK is the only + // place that can close the vocabulary where every writer meets it. + db := testkit.DB(t) + ctx := context.Background() + requireTableExists(t, db, admissionsTable) + + t.Run("the constraint exists in the catalog", func(t *testing.T) { + var matched string + for name, definition := range checkConstraintDefinitions(t, db, admissionsTable) { + if strings.Contains(definition, "last_community_op_rank") && + strings.Contains(definition, "0") && strings.Contains(definition, "1") { + matched = name + } + } + assert.NotEmpty(t, matched, + "no CHECK constraint restricts last_community_op_rank to (0, 1); a rank outside the vocabulary would outrank every genuine event") + }) + + t.Run("the constraint is enforced", func(t *testing.T) { + insertWithRank := func(t *testing.T, rank int16) error { + t.Helper() + subject := newAdmissionSubject(t, db) + _, err := db.ExecContext(ctx, fmt.Sprintf(` + INSERT INTO %s (community_did, post_uri, status, last_community_rev, last_community_op_rank, created_at, updated_at) + VALUES ($1, $2, 'pending', $3, $4, NOW(), NOW()) + `, admissionsTable), subject.CommunityDID, subject.PostURI, testkit.TID(), rank) + return err + } + + assert.NoError(t, insertWithRank(t, 0), "rank 0 (delete) is in the vocabulary") + assert.NoError(t, insertWithRank(t, 1), "rank 1 (put) is in the vocabulary") + assert.Error(t, insertWithRank(t, 2), + "a rank outside {0, 1} would compare greater than every genuine put and freeze the subject forever") + assert.Error(t, insertWithRank(t, -1), + "a negative rank is equally outside the §5.2 vocabulary") + }) +} diff --git a/internal/db/postgres/post_repo.go b/internal/db/postgres/post_repo.go index fd537f8..025b499 100644 --- a/internal/db/postgres/post_repo.go +++ b/internal/db/postgres/post_repo.go @@ -94,14 +94,13 @@ func (r *postgresPostRepo) Create(ctx context.Context, post *posts.Post) error { return fmt.Errorf("post already indexed: %s", post.URI) } - // Check for foreign key violations - if strings.Contains(err.Error(), "violates foreign key constraint") { - if strings.Contains(err.Error(), "fk_author") { - return fmt.Errorf("author DID not found: %s", post.AuthorDID) - } - if strings.Contains(err.Error(), "fk_community") { - return fmt.Errorf("community DID not found: %s", post.CommunityDID) - } + // Check for a foreign key violation. fk_community is the only FK left + // on posts: migration 034 dropped fk_author (author-owned posts — a + // federated author may have no users row, and that must not block + // indexing). + if strings.Contains(err.Error(), "violates foreign key constraint") && + strings.Contains(err.Error(), "fk_community") { + return fmt.Errorf("community DID not found: %s", post.CommunityDID) } return fmt.Errorf("failed to insert post: %w", err) diff --git a/internal/db/postgres/user_repo.go b/internal/db/postgres/user_repo.go index c26b263..a843af1 100644 --- a/internal/db/postgres/user_repo.go +++ b/internal/db/postgres/user_repo.go @@ -296,15 +296,16 @@ func (r *postgresUserRepo) Delete(ctx context.Context, did string) error { // 9. Delete the admission rows for this author's posts. // - // This runs BEFORE the posts themselves because it reads them to find its - // subjects. community_post_admissions deliberately carries no foreign key - // to posts (migration 034: an acceptance can arrive before the post it is - // about, and an FK would turn that ordinary ordering artefact into an - // insert failure), so nothing but this statement removes the rows — left - // behind, they would be admissions for posts that no longer exist. + // Author-owned post URIs live in the author's own repo — at:///... — + // so a prefix match on the DID reaches every subject, INCLUDING admissions + // whose post is not indexed here: an acceptance can arrive before the post + // it is about, which is exactly why community_post_admissions deliberately + // carries no foreign key to posts (migration 034). A subquery against + // posts would miss those rows, and nothing but this statement removes + // them — left behind, they would be admissions about a deleted account. if _, err := tx.ExecContext(ctx, ` DELETE FROM community_post_admissions - WHERE post_uri IN (SELECT uri FROM posts WHERE author_did = $1) + WHERE starts_with(post_uri, 'at://' || $1 || '/') `, did); err != nil { return fmt.Errorf("failed to delete community_post_admissions for did=%s: %w", did, err) } diff --git a/internal/db/postgres/user_repo_test.go b/internal/db/postgres/user_repo_test.go index 56a9992..c4d77f4 100644 --- a/internal/db/postgres/user_repo_test.go +++ b/internal/db/postgres/user_repo_test.go @@ -236,6 +236,96 @@ func TestUserRepo_Delete_WithPosts_CascadeDeletes(t *testing.T) { assert.Equal(t, 0, postCount, "Deleting a user must still delete their posts") } +// seedAdmissionRow inserts a community_post_admissions row directly via SQL. +// The admissions repo lives in this package, but Delete's sweep has to be +// provable against a row whose subject post is NOT indexed (an acceptance that +// arrived before its post — a design-blessed state, which is exactly why +// migration 034 gave the table no FK to posts), and no repo write path can be +// asked to manufacture that mid-test. +func seedAdmissionRow(t *testing.T, db *sql.DB, communityDID, postURI, status string) { + t.Helper() + _, err := db.Exec(` + INSERT INTO community_post_admissions (community_did, post_uri, status) + VALUES ($1, $2, $3) + `, communityDID, postURI, status) + require.NoError(t, err, "Failed to seed admission row for %s", postURI) +} + +func TestUserRepo_Delete_SweepsAdmissionsForUnindexedPosts(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + + id := testkit.UniqueID(t) + testDID := "did:plc:admsweep" + id + otherAuthorDID := "did:plc:admkeep" + id + communityDID := "did:plc:admsweepcomm" + id + + repo := NewUserRepository(db) + ctx := context.Background() + + // Create test user + user := &users.User{ + DID: testDID, + Handle: "admsweep" + id + ".test", + PDSURL: "https://test.pds", + } + _, err := repo.Create(ctx, user) + require.NoError(t, err) + + createTestCommunity(t, db, communityDID, "c.admsweep"+id, testDID) + + // (a) An INDEXED post of the author, with an admission row about it. + indexedURI := "at://" + testDID + "/social.coves.community.postv2/indexed" + id + _, err = db.Exec(` + INSERT INTO posts (uri, cid, rkey, author_did, community_did, title, created_at) + VALUES ($1, 'bafyadmindexed', $2, $3, $4, 'Indexed Post', NOW()) + `, indexedURI, "indexed"+id, testDID, communityDID) + require.NoError(t, err) + seedAdmissionRow(t, db, communityDID, indexedURI, "pending") + + // (b) An UNINDEXED subject of the same author: the acceptance arrived + // before the post, so there is no posts row for this URI at all. The + // author's DID still lives inside post_uri — that is all the sweep has + // to go on. + unindexedURI := "at://" + testDID + "/social.coves.community.postv2/unindexed" + id + seedAdmissionRow(t, db, communityDID, unindexedURI, "pending") + + // Control: another author's admission in the same community must survive + // the deletion untouched. + otherAuthorURI := "at://" + otherAuthorDID + "/social.coves.community.postv2/keep" + id + seedAdmissionRow(t, db, communityDID, otherAuthorURI, "pending") + + // Delete the user + err = repo.Delete(ctx, testDID) + require.NoError(t, err) + + // Verify user is deleted + _, err = repo.GetByDID(ctx, testDID) + assert.ErrorIs(t, err, users.ErrUserNotFound) + + // Verify the posts went with the user + var count int + err = db.QueryRow("SELECT COUNT(*) FROM posts WHERE author_did = $1", testDID).Scan(&count) + require.NoError(t, err) + assert.Equal(t, 0, count, "Deleting a user must still delete their posts") + + // BOTH admission rows about the deleted author's posts must be gone + err = db.QueryRow("SELECT COUNT(*) FROM community_post_admissions WHERE post_uri = $1", indexedURI).Scan(&count) + require.NoError(t, err) + assert.Equal(t, 0, count, "Admission for the author's INDEXED post must be deleted") + + err = db.QueryRow("SELECT COUNT(*) FROM community_post_admissions WHERE post_uri = $1", unindexedURI).Scan(&count) + require.NoError(t, err) + assert.Equal(t, 0, count, + "Admission for the author's UNINDEXED post must be deleted too: acceptance-before-post "+ + "is an ordinary state, and a sweep that reads posts to find its subjects misses it") + + // Another author's admission must NOT be swept + err = db.QueryRow("SELECT COUNT(*) FROM community_post_admissions WHERE post_uri = $1", otherAuthorURI).Scan(&count) + require.NoError(t, err) + assert.Equal(t, 1, count, "Another author's admission row must survive the deletion") +} + func TestUserRepo_Delete_TransactionRollback(t *testing.T) { t.Parallel() // This test verifies that if any part of the deletion fails, diff --git a/tests/testkit/migrate.go b/tests/testkit/migrate.go index 5099afb..11d94e5 100644 --- a/tests/testkit/migrate.go +++ b/tests/testkit/migrate.go @@ -3,6 +3,7 @@ package testkit import ( "context" "database/sql" + "strings" "Coves/internal/db/migrations" @@ -26,21 +27,44 @@ import ( // a private, per-test database that is dropped when the test ends. Pointing // them at a shared database would migrate it out from under every other test, // and pointing them at the TEMPLATE would corrupt the thing clones are made -// from — which is why neither function accepts a database name. +// from — which is why neither function accepts a database name, and why both +// refuse outright any database whose name does not carry ClonePrefix. -// MigrateDownOne rolls this test's database back by exactly one migration and +// MigrateDownOne rolls this test's database back by exactly one migration — +// after proving that the schema is currently AT expectedCurrentVersion — and // returns the version it undid. // -// The returned version is what lets a caller prove it rolled back the migration -// it meant to: a test that asserts on "the Down of 034" and silently gets 035's -// instead is testing nothing, and nothing else in the harness would notice. -func MigrateDownOne(t TestingT, db *sql.DB) int64 { +// The precondition is the point. A rollback test written against "the Down of +// 034" keeps passing after migration 035 lands, silently asserting on 035's +// Down instead — the strongest wrong answer a green test can give, because +// 034's Down (the thing the test exists to prove non-destructive) stops being +// run at all. Failing loudly here turns "a future migration landed" into a +// named remedy at the call site instead of a quietly retargeted assertion. +// The returned version is the same proof after the fact: the caller can pin +// which migration's Down actually ran. +func MigrateDownOne(t TestingT, db *sql.DB, expectedCurrentVersion int64) int64 { t.Helper() + requireClone(t, db, "MigrateDownOne") provider := migrationProvider(t, db) if provider == nil { return 0 } + + current, err := provider.GetDBVersion(context.Background()) + if err != nil { + t.Fatalf("testkit.MigrateDownOne: reading the current migration version: %v", err) + return 0 + } + if current != expectedCurrentVersion { + t.Fatalf("testkit.MigrateDownOne: the database is at migration %d, not %d — a migration has landed on top of the one this test rolls back. "+ + "Rolling back now would exercise %d's Down section and silently stop testing the one the assertions are about. "+ + "Remedy: update the call site to the new current version and re-check what that migration's Down preserves, "+ + "or roll back through the newer migrations explicitly, one asserted step at a time.", + current, expectedCurrentVersion, current) + return 0 + } + result, err := provider.Down(context.Background()) if err != nil { t.Fatalf("testkit.MigrateDownOne: rolling back one migration: %v", err) @@ -61,6 +85,7 @@ func MigrateDownOne(t TestingT, db *sql.DB) int64 { func MigrateUp(t TestingT, db *sql.DB) { t.Helper() + requireClone(t, db, "MigrateUp") provider := migrationProvider(t, db) if provider == nil { return @@ -70,6 +95,27 @@ func MigrateUp(t TestingT, db *sql.DB) { } } +// requireClone refuses to run migrations against anything but a per-test +// clone. The *sql.DB handed in is supposed to be the one DB(t) returned, but +// nothing in the type system says so — a handle to the template or to a shared +// dev database satisfies the signature just as well, and migrating THOSE +// corrupts every other test (or the developer's data) instead of one clone. +// current_database() is authoritative for where this pool actually points, and +// the clone prefix is the same rail every destructive statement in db.go rides. +func requireClone(t TestingT, db *sql.DB, operation string) { + t.Helper() + + var name string + if err := db.QueryRow(`SELECT current_database()`).Scan(&name); err != nil { + t.Fatalf("testkit.%s: identifying the connected database: %v", operation, err) + return + } + if !strings.HasPrefix(name, ClonePrefix) { + t.Fatalf("testkit.%s: refusing to migrate %q: only per-test clones (prefix %q, from testkit.DB) may be rolled forwards and backwards — anything else is the template or shared state", + operation, name, ClonePrefix) + } +} + // migrationProvider builds a goose provider over the embedded migrations for one // database. Fatalf does not return, so callers still check for nil to satisfy // the compiler and any TestingT implementation that is less abrupt than -- 2.51.2