From 29c6d5d4119117c1314045afb3bf142f97173cad Mon Sep 17 00:00:00 2001 From: Bretton Date: Fri, 14 Aug 2026 12:58:57 -0700 Subject: [PATCH] fix(moderation): close the legacy lock paths and split the moderation seam (17c-2 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two-stream review of 17c-2. The lock closed the bypasses it targeted, but the path protecting content that already exists was both incomplete and untested. WALK. walkThreadRoot stopped at the outbound/fediverse boundary without consulting ap_objects.thread_root_at_uri on the very row it was reading past, so a native comment written before the root column, hanging under a Lemmy comment materialized after it, resolved to that comment and missed the post's lock. It now reads the mapping at that boundary; a mapping that names no thread still yields the object itself, but a mapping that cannot be READ is an error, because guessing a boundary off a failed lookup is how a locked thread quietly becomes an unlocked one. COVERAGE. The migration path had none: every fixture created comments through the current path, so walkThreadRoot, both forks of the dead end, and the re-materialization invariant were unreachable from any test. Legacy rows are now built by SUBTRACTION from real ones, and the helper refuses to strip a field that is not present — so if the current path ever stops writing the root, those tests fail loudly instead of asserting nothing against modern rows. The climb is exercised over more than one hop, since a one-hop fixture passes against an implementation that only ever reads the parent. SEAM. ObjectModeration was embedded in store.APObjects, which handed SetLock and SetRemoval to the vote refresher, the echo classifier, the materializer and the outbound enqueuer — interfaces they hold only to resolve strongRefs. It is standalone now, with HandlerOptions.Moderation defaulting from a type assertion on opts.Objects, so no call site changed and no test file was touched. main.go passes it explicitly rather than relying on the assertion for the store every moderation decision lands in, and a failed assertion is a retryable error, never a skip: a lock we cannot record reads downstream as "no lock". An Undo of a native comment removal now clears the tombstone and the soft delete as the post restore already did. Without it a comment suppressed by the pre-17c-1 path could never be restored, and moderateAnnouncedDelete would decline forever on IsDeleted() — permanently unmoderatable by the community's own legitimate restore, the same failure RESTORE-1 fixed for posts. Honesty: the summary-less counter and reason no longer narrate a motive the bridge cannot verify (an Announce's inner actor is unauthenticated, and a truthful author self-delete is dropped as an echo before this branch); the lifted counter only counts an actual reversal; the dead-end message carries which of its two causes fired, since a looping chain is corrupt state rather than an old row; and the retry comment states the real recovery path. Co-Authored-By: Claude Opus 5 (1M context) --- cmd/tidepool/main.go | 24 +- internal/consume/comments.go | 67 +-- internal/consume/dispatch.go | 8 + internal/consume/subjects.go | 79 +++- .../db/migrations/025_object_moderation.sql | 18 + internal/ingest/consent.go | 23 +- internal/ingest/handler.go | 23 ++ internal/ingest/moderation.go | 88 +++- internal/ingest/moderation_legacy_test.go | 385 ++++++++++++++++++ internal/ingest/moderation_lock_test.go | 44 +- internal/store/ap_objects.go | 9 +- internal/store/interfaces.go | 28 +- internal/store/object_moderation.go | 69 ++-- 13 files changed, 736 insertions(+), 129 deletions(-) create mode 100644 internal/ingest/moderation_legacy_test.go diff --git a/cmd/tidepool/main.go b/cmd/tidepool/main.go index 4a41b0e..a0b086c 100644 --- a/cmd/tidepool/main.go +++ b/cmd/tidepool/main.go @@ -377,16 +377,20 @@ func run(logger *slog.Logger) error { return err } handler, err := ingest.NewHandler(ingest.HandlerOptions{ - Materializer: materializer, - Fetcher: apClient, - Objects: objects, - Actors: actors, - Communities: communities, - Tombstones: tombstones, - Records: repoManager, - Votes: voteAggregator, - Backfill: backfill, - Echo: echoClassifier, + Materializer: materializer, + Fetcher: apClient, + Objects: objects, + Actors: actors, + Communities: communities, + Tombstones: tombstones, + Records: repoManager, + Votes: voteAggregator, + Backfill: backfill, + Echo: echoClassifier, + // Passed explicitly rather than left to NewHandler's default: this is the + // store every inbound moderation decision is RECORDED in, and production + // should not depend on a type assertion to have one. + Moderation: store.NewObjectModeration(database), ServiceActorID: serviceActor.ID, Logger: logger, }) diff --git a/internal/consume/comments.go b/internal/consume/comments.go index 4c37cc3..78dd19c 100644 --- a/internal/consume/comments.go +++ b/internal/consume/comments.go @@ -167,7 +167,7 @@ func (d *Dispatcher) applyCommentWrite(ctx context.Context, tx *sql.Tx, did stri // read the lock" must never be answered with "there is no lock", which is // exactly the reply the lock exists to stop. func (d *Dispatcher) refuseInLockedThread(ctx context.Context, atURI string, thread *resolvedThread) error { - locked, err := d.objectMappings.LockedAmong(ctx, thread.ParentATURI, thread.RootATURI) + locked, err := d.moderation.LockedAmong(ctx, thread.ParentATURI, thread.RootATURI) if err != nil { return fmt.Errorf("read lock state of the thread above %s: %w", atURI, err) } @@ -187,8 +187,7 @@ func (d *Dispatcher) refuseInLockedThread(ctx context.Context, atURI string, thr // strand replies forever under content nobody has moderated. So the question // narrows to the only one that can still matter: does this community hold // ANY lock the unknown root might be? If it holds none, there is provably - // nothing to miss. If it holds one, we cannot tell, and a retryable failure - // is the honest answer — the operator sees it, and a lifted lock resolves it. + // nothing to miss. // // This is NOT a community-scoped refusal, and it is not reachable from // fediverse content: a mapped subject's thread is answered from state the @@ -196,17 +195,26 @@ func (d *Dispatcher) refuseInLockedThread(ctx context.Context, atURI string, thr // empty answer. A read that could FAIL here would turn one locked post into // a community-wide park of ordinary replies, which is why there is no read // on that path at all. - held, err := d.objectMappings.CommunityHoldsAnyLock(ctx, thread.CommunityDID) + held, err := d.moderation.CommunityHoldsAnyLock(ctx, thread.CommunityDID) if err != nil { return fmt.Errorf("read standing locks of %s: %w", thread.CommunityDID, err) } if !held { return nil } + // RETRYABLE, deliberately — and the recovery is not instant, so it is worth + // stating plainly: this comment first burns its in-line attempts, blocking + // the consumer behind it for that schedule, and then DEAD-LETTERS. What + // makes that acceptable rather than a loss is the redrive window: the + // dead-lettered frame keeps this message, and a redrive after the lock is + // lifted (or the row repaired) re-runs it against an answerable question. + // Nothing about the comment is wrong, so it must not be discarded as + // permanent; nothing about the state improves on its own, so it must not be + // retried as though it would. return fmt.Errorf( "cannot tell whether comment %s is in a locked thread: the chain above it cannot be followed past %s, "+ - "whose outbound state names no parent, and %s holds standing locks", - atURI, thread.RootDeadEnd, thread.CommunityDID) + "and %s holds standing locks — redrive this once the lock is lifted or the row is repaired", + atURI, thread.RootUnresolved, thread.CommunityDID) } // applyCommentDelete withdraws a comment, using ONLY state. @@ -286,12 +294,12 @@ type resolvedThread struct { // the parent itself when the parent IS a root), never from the record's // reply.root, which the author writes and could point anywhere. RootATURI string - // RootDeadEnd names the object the thread resolution stopped at when - // RootATURI could not be established. It is DIAGNOSTIC ONLY — never written - // to the snapshot, never inherited — and exists so the one error that holds - // a comment for an undeterminable thread names something an operator can - // open, instead of a thread nobody can look up. - RootDeadEnd string + // RootUnresolved names the object the thread resolution stopped at, and why, + // when RootATURI could not be established. It is DIAGNOSTIC ONLY — never + // written to the snapshot, never inherited — and exists so the one error that + // holds a comment for an undeterminable thread names something an operator + // can open, instead of a thread nobody can look up. + RootUnresolved *unresolvedThread } // commentThread resolves the thread context for a create or an update. @@ -318,23 +326,24 @@ func (d *Dispatcher) commentThread(ctx context.Context, atURI string, commit *Co return nil, fmt.Errorf("read outbound state for %s: %w", atURI, err) } parent := d.parentFromSnapshot(stored.TranslatedSnapshot) - rootATURI, deadEnd := rootFromSnapshot(stored.TranslatedSnapshot), "" + rootATURI := rootFromSnapshot(stored.TranslatedSnapshot) + var unresolved *unresolvedThread if rootATURI == "" { // Written before the root was recorded: climb this comment's own // parent chain rather than assuming anything. The successful edit // re-writes the snapshot below, so the walk happens once per row. - if rootATURI, deadEnd, err = d.walkThreadRoot(ctx, atURI); err != nil { + if rootATURI, unresolved, err = d.walkThreadRoot(ctx, atURI); err != nil { return nil, err } } return &resolvedThread{ - ParentATURI: parent.ATURI, - ParentAPID: parent.APID, - CommunityDID: stored.CommunityDID, - CommunityAPID: stored.CommunityAPID, - Depth: stored.Depth, - RootATURI: rootATURI, - RootDeadEnd: deadEnd, + ParentATURI: parent.ATURI, + ParentAPID: parent.APID, + CommunityDID: stored.CommunityDID, + CommunityAPID: stored.CommunityAPID, + Depth: stored.Depth, + RootATURI: rootATURI, + RootUnresolved: unresolved, }, nil } return d.resolveParent(ctx, commit) @@ -361,18 +370,18 @@ func (d *Dispatcher) resolveParent(ctx context.Context, commit *CommitEvent) (*r // itself when the parent is a root. Same shape as the depth above: inherited // from the parent's own state, which is what keeps both O(1) instead of // walking the thread on every comment. - rootATURI, deadEnd, err := d.threadRootOf(ctx, parent) + rootATURI, unresolved, err := d.threadRootOf(ctx, parent) if err != nil { return nil, err } return &resolvedThread{ - ParentATURI: parent.ATURI, - ParentAPID: parent.APID, - CommunityDID: parent.CommunityDID, - CommunityAPID: parent.CommunityAPID, - Depth: parent.Depth + 1, - RootATURI: rootATURI, - RootDeadEnd: deadEnd, + ParentATURI: parent.ATURI, + ParentAPID: parent.APID, + CommunityDID: parent.CommunityDID, + CommunityAPID: parent.CommunityAPID, + Depth: parent.Depth + 1, + RootATURI: rootATURI, + RootUnresolved: unresolved, }, nil } diff --git a/internal/consume/dispatch.go b/internal/consume/dispatch.go index 619710b..2b85f55 100644 --- a/internal/consume/dispatch.go +++ b/internal/consume/dispatch.go @@ -195,6 +195,12 @@ type Options struct { Votes store.OutboundVotes Communities store.Communities ObjectMappings store.APObjects + // Moderation is the bridge-owned moderation state the comment path reads to + // refuse a reply in a locked thread. It is a SEPARATE store from + // ObjectMappings on purpose: this consumer only ever reads it, while the + // interface carries the mutators the ingest side writes with, and the + // mapping store is held by half the bridge to resolve strongRefs. + Moderation store.ObjectModeration // UserOrigin is AP_USER_ORIGIN: the origin every deterministic activity // id is minted under. UserOrigin string @@ -222,6 +228,7 @@ type Dispatcher struct { objects store.OutboundObjects votes store.OutboundVotes communities store.Communities + moderation store.ObjectModeration records materialize.RecordGetter hosted *hostedRepos gate *RevGate @@ -291,6 +298,7 @@ func NewDispatcher(opts Options) (*Dispatcher, error) { objects: orDefault[store.OutboundObjects](opts.Objects, store.NewOutboundObjects(opts.DB)), votes: orDefault[store.OutboundVotes](opts.Votes, store.NewOutboundVotes(opts.DB)), communities: orDefault[store.Communities](opts.Communities, store.NewCommunities(opts.DB)), + moderation: orDefault[store.ObjectModeration](opts.Moderation, store.NewObjectModeration(opts.DB)), records: opts.Records, hosted: newHostedRepos(opts.DB), gate: NewRevGate(opts.DB), diff --git a/internal/consume/subjects.go b/internal/consume/subjects.go index 92d91dc..9b2fb68 100644 --- a/internal/consume/subjects.go +++ b/internal/consume/subjects.go @@ -137,6 +137,30 @@ func (d *Dispatcher) resolveSubject(ctx context.Context, atURI string) (*resolve }, nil } +// mappedThreadRoot answers the thread of an object the bridge holds no outbound +// state for, from the mapping it holds instead: the materializer records a +// comment's thread root there (migration 026), and for a post there is nothing +// to record because a post IS its own thread root. +// +// A mapping that names no thread — one written before 026, or a post — leaves +// the object itself as the answer, which is the boundary this walk would have +// drawn anyway. A mapping that cannot be READ is an error, because guessing the +// boundary off a failed lookup is how a locked thread quietly becomes an +// unlocked one. +func (d *Dispatcher) mappedThreadRoot(ctx context.Context, atURI string) (root string, unresolved *unresolvedThread, err error) { + mapping, err := d.objectMappings.GetByATURI(ctx, atURI) + if errors.IsNotFound(err) { + return atURI, nil, nil + } + if err != nil { + return "", nil, fmt.Errorf("read mapped thread of %s: %w", atURI, err) + } + if mapping.ThreadRootATURI != "" { + return mapping.ThreadRootATURI, nil, nil + } + return atURI, nil, nil +} + // threadRootOf answers which THREAD a new child of subject belongs to — the // at-uri a lock on the whole conversation is recorded against. // @@ -155,12 +179,12 @@ func (d *Dispatcher) resolveSubject(ctx context.Context, atURI string) (*resolve // The second result names the object the answer stopped at when the root could // not be established — the only thing an operator can look up when a comment is // held for an undeterminable thread. -func (d *Dispatcher) threadRootOf(ctx context.Context, subject *resolvedSubject) (root, deadEnd string, err error) { +func (d *Dispatcher) threadRootOf(ctx context.Context, subject *resolvedSubject) (root string, unresolved *unresolvedThread, err error) { if subject.RootATURI != "" { - return subject.RootATURI, "", nil + return subject.RootATURI, nil, nil } if subject.Depth == 0 { - return subject.ATURI, "", nil + return subject.ATURI, nil, nil } return d.walkThreadRoot(ctx, subject.ATURI) } @@ -190,23 +214,32 @@ func (d *Dispatcher) threadRootOf(ctx context.Context, subject *resolvedSubject) // object it names is the one an operator has to open to see why. A store failure // IS an error: retrying is the only honest response to a question that was never // answered. -func (d *Dispatcher) walkThreadRoot(ctx context.Context, atURI string) (root, deadEnd string, err error) { +func (d *Dispatcher) walkThreadRoot(ctx context.Context, atURI string) (root string, unresolved *unresolvedThread, err error) { climbed := atURI // Bounded by Lemmy's nesting cap plus the root itself: every hop is one // level up, so a chain longer than that is a cycle, not a conversation. for hop := 0; hop <= maxCommentDepth+1; hop++ { state, err := d.objects.GetByATURI(ctx, climbed) if errors.IsNotFound(err) { - return climbed, "", nil + // The chain has left the bridge's own OUTBOUND state — but that is + // not the same as leaving its state altogether. Content the bridge + // materialized IN (a Lemmy comment, which by construction has no + // outbound row) carries its thread on its MAPPING, and this is the + // row the walk was about to read past. Without this, every legacy + // native reply under a Lemmy comment resolves its thread to that + // comment and sits outside its own locked thread — most of Lemmy, + // and unlike the legacy climb it does not heal when the PARENT is + // re-materialized, because the child is what lacks the root. + return d.mappedThreadRoot(ctx, climbed) } if err != nil { - return "", "", fmt.Errorf("read thread state for %s: %w", climbed, err) + return "", nil, fmt.Errorf("read thread state for %s: %w", climbed, err) } if root := rootFromSnapshot(state.TranslatedSnapshot); root != "" { - return root, "", nil + return root, nil, nil } if state.Depth == 0 { - return climbed, "", nil + return climbed, nil, nil } parent := d.parentFromSnapshot(state.TranslatedSnapshot) if parent.ATURI == "" { @@ -214,13 +247,39 @@ func (d *Dispatcher) walkThreadRoot(ctx context.Context, atURI string) (root, de // ever written names its parent, and a post is at depth 0. d.logger.Warn("thread root is undeterminable: nested outbound state names no parent", slog.String("at_uri", atURI), slog.String("dead_end", climbed)) - return "", climbed, nil + return "", &unresolvedThread{ATURI: climbed, Cause: causeNamesNoParent}, nil } climbed = parent.ATURI } d.logger.Warn("thread root is undeterminable: the parent chain is longer than a thread can be", slog.String("at_uri", atURI), slog.String("dead_end", climbed)) - return "", climbed, nil + return "", &unresolvedThread{ATURI: climbed, Cause: causeChainLoops}, nil +} + +// unresolvedThread names the object a thread walk stopped at and WHY it stopped. +// +// The cause travels because the two are different findings that want different +// responses, and an operator reading one message must not be told the other. A +// row that never recorded its parent is old and repairable — the expected shape +// on the migration path. A chain longer than a thread can be is CYCLIC: the +// state is corrupt rather than merely old, waiting fixes nothing, and that is +// precisely the case where a hard-coded "names no parent" would be false. +type unresolvedThread struct { + ATURI string + Cause string +} + +const ( + causeNamesNoParent = "its outbound state names no parent" + causeChainLoops = "the chain above it loops" +) + +// String renders the pair for the one error that reports it. +func (u *unresolvedThread) String() string { + if u == nil { + return "" + } + return u.ATURI + " (" + u.Cause + ")" } // subjectCommunityDID answers which community a mapped record belongs to. diff --git a/internal/db/migrations/025_object_moderation.sql b/internal/db/migrations/025_object_moderation.sql index 8c15e0b..09bb5d9 100644 --- a/internal/db/migrations/025_object_moderation.sql +++ b/internal/db/migrations/025_object_moderation.sql @@ -28,6 +28,13 @@ CREATE TABLE object_moderation ( -- ap_id is the same object's fediverse id, denormalized so the row can be -- read back from either side of the bridge without a join through -- ap_objects (whose row is the very thing this table must not depend on). + -- + -- FORWARD-LOOKING, and honestly so: every write supplies it and NOTHING + -- reads it today. It is kept because the readers this table is heading + -- towards arrive holding an AP id and not an at-uri — the admin moderation + -- surface, and 17d's delivery-side ban checks, whose OrderingKey is an AP + -- group id — and because a column backfilled later can only be backfilled + -- from the mapping row this table exists to be independent of. ap_id TEXT NOT NULL, -- community_did BINDS the decision to the community that made it. Unbound, -- any co-hosted community's Undo{Lock} could clear a decision it did not @@ -49,5 +56,16 @@ CREATE TABLE object_moderation ( updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); +-- NO INDEX, deliberately, and here is the one query that will eventually want +-- one: CommunityHoldsAnyLock filters (community_did) WHERE locked_at IS NOT +-- NULL. It runs ONLY where a comment's thread could not be determined at all — +-- a cold branch over a table holding one row per moderated object — so a partial +-- index today would cost every lock write to serve a scan of a handful of rows. +-- If that branch ever warms (a backlog of pre-026 rows, a bulk repair), the +-- index to add is exactly: +-- CREATE INDEX object_moderation_locked_community_idx +-- ON object_moderation (community_did) WHERE locked_at IS NOT NULL; +-- Every other read is by at_uri, which the primary key already serves. + -- +goose Down DROP TABLE IF EXISTS object_moderation; diff --git a/internal/ingest/consent.go b/internal/ingest/consent.go index d027528..9523e66 100644 --- a/internal/ingest/consent.go +++ b/internal/ingest/consent.go @@ -187,11 +187,19 @@ var ( NativeCommentRemoved = expvar.NewInt("tidepool_moderation_native_comment_removed") // NativeCommentRemovalLifted counts the Undos that cleared one. NativeCommentRemovalLifted = expvar.NewInt("tidepool_moderation_native_comment_removal_lifted") - // NativeCommentSelfDeleted counts summary-less announced deletes of native - // comments: the author's own delete coming home, taken and deliberately - // recorded nowhere. A DECIDED non-action, so it is counted — a flat zero - // must not be readable as "this never happens". - NativeCommentSelfDeleted = expvar.NewInt("tidepool_moderation_native_comment_self_deleted") + // NativeCommentSummarylessDelete counts summary-less announced deletes of + // native comments: taken, and deliberately recorded nowhere. A DECIDED + // non-action, so it is counted — a flat zero must not be readable as "this + // never happens". + // + // It is named for the SHAPE, not for a motive. The obvious name + // (…_self_deleted) would claim the author deleted their own comment, and + // nothing here establishes that: an Announce's inner actor is not + // authenticated, and a TRUTHFUL author self-delete is dropped as a + // local-actor echo before this branch is reached. So what this counter + // actually sees is summary-less announces carrying foreign or unverifiable + // attribution, and it must not be read as a census of author deletions. + NativeCommentSummarylessDelete = expvar.NewInt("tidepool_moderation_native_comment_summaryless_delete") ) func (h *Handler) moderateAnnouncedDelete(ctx context.Context, del *ap.Object, targetID string, announcer *store.Community) (bool, error) { @@ -546,8 +554,9 @@ func (h *Handler) restoreNativeContent(ctx context.Context, undo *ap.Object, if mapping.Collection != materialize.CollectionPostV2 { // A COMMENT: its removal lives in the bridge's own moderation state, not // in the community repo, so lifting it is a state clear rather than the - // acceptance transition below. - return h.liftNativeCommentRemoval(ctx, undo, mapping, announcer) + // acceptance transition below. It takes the SAME legacy repair with it — + // see liftNativeCommentRemoval, which does both. + return h.liftNativeCommentRemoval(ctx, undo, mapping, announcer, scope) } if err := h.tombstones.Remove(ctx, mapping.APID, scope); err != nil { return fmt.Errorf("ingest: clear tombstone for %s: %w", mapping.APID, err) diff --git a/internal/ingest/handler.go b/internal/ingest/handler.go index 8645de7..51ff8c9 100644 --- a/internal/ingest/handler.go +++ b/internal/ingest/handler.go @@ -99,6 +99,13 @@ type HandlerOptions struct { Records RecordGetter Votes VoteAggregator Backfill Backfiller + // Moderation is the bridge-owned moderation state announced Locks and + // native-comment removals are recorded in (task 17c-2). Optional ONLY in the + // wiring sense: when it is nil, NewHandler takes the moderation view of + // Objects, which the postgres mapping store provides. A dispatcher that ends + // up with neither refuses to moderate rather than silently dropping the + // decision — see moderationState. + Moderation store.ObjectModeration // Echo classifies inbound ids against the bridge's own serving surface so // an activity we sent never re-enters as content (task 17a). Echo EchoClassifier @@ -122,6 +129,7 @@ type Handler struct { records RecordGetter votes VoteAggregator backfill Backfiller + moderation store.ObjectModeration classifier EchoClassifier echoLog *ratelimit.Sampler serviceID string @@ -170,6 +178,20 @@ func NewHandler(opts HandlerOptions) (*Handler, error) { if logger == nil { logger = slog.Default() } + // The moderation state and the mapping store are two repositories over two + // tables, and only the moderation paths may hold the first — so it is a + // separate option rather than methods on APObjects, which the echo + // classifier, the vote aggregator, the stats refresher, the materializer and + // the enqueuer all hold to resolve strongRefs. The default keeps every + // existing call site working: the postgres mapping store IS also that + // repository, so callers that pass one and no moderation store get the + // matching view of the same database rather than a nil. + moderation := opts.Moderation + if moderation == nil { + if fromObjects, ok := opts.Objects.(store.ObjectModeration); ok { + moderation = fromObjects + } + } return &Handler{ mat: opts.Materializer, fetcher: opts.Fetcher, @@ -180,6 +202,7 @@ func NewHandler(opts HandlerOptions) (*Handler, error) { records: opts.Records, votes: opts.Votes, backfill: opts.Backfill, + moderation: moderation, classifier: opts.Echo, echoLog: ratelimit.NewSampler(echoDropLogInterval), serviceID: opts.ServiceActorID, diff --git a/internal/ingest/moderation.go b/internal/ingest/moderation.go index ef922de..0e4fef5 100644 --- a/internal/ingest/moderation.go +++ b/internal/ingest/moderation.go @@ -10,6 +10,23 @@ import ( "tidepool/internal/store" ) +// moderationState is the bridge-owned moderation store, or an error naming the +// gap. Every moderation path asks for it before deciding anything. +// +// The error is RETRYABLE and never a skip. A dispatcher whose mapping store is +// not also the moderation repository (a substitute view, a future alternate +// backend) can still serve every read path — but a lock it cannot record reads +// downstream as "no lock", and a removal it cannot record reads as "nobody +// moderated this". A decision that cannot be stored has to stay on the queue +// where an operator sees it, not be marked processed as though it were handled. +func (h *Handler) moderationState() (store.ObjectModeration, error) { + if h.moderation == nil { + return nil, fmt.Errorf( + "ingest: no moderation state store is wired, so this decision cannot be recorded") + } + return h.moderation, nil +} + // moderateNativeComment decides an announced Delete of a NATIVE comment — a // record in the AUTHOR's own repo that this bridge federated on their behalf. // @@ -21,10 +38,19 @@ import ( // - WITH a summary: a moderator removed the comment. The decision is recorded // bridge-side, bound to the community that made it, under the SAME code a // post removal writes. -// - WITHOUT one: the author deleted their own comment. NOTHING is recorded. -// Writing moderator-discretion here would assert that a moderator acted when -// none did — a removal naming a moderator team that took no action — and it -// would stand until somebody sent an Undo for something that never happened. +// - WITHOUT one: NOTHING is recorded. Writing moderator-discretion here would +// assert that a moderator acted when none did — a removal naming a moderator +// team that took no action — and it would stand until somebody sent an Undo +// for something that never happened. +// +// Lemmy's convention says the summary-less shape IS the author's own delete, and +// that is why it records nothing. But the bridge cannot VERIFY that, and must +// not claim it: an Announce's inner actor is unauthenticated, and a truthful +// author self-delete never even reaches here — the echo classifier takes it by +// the inner actor first (a native comment's author is one of our personas). So +// what actually arrives on this branch is a summary-less announce whose +// attribution is foreign or unverifiable, and the counter and the reason say +// exactly that rather than narrating a motive. // // Both are TAKEN, and that is the point of the branch: falling through runs the // v1 destructive path against the author's record, soft-deleting our own mapping @@ -49,16 +75,20 @@ func (h *Handler) moderateNativeComment(ctx context.Context, del *ap.Object, mapping *store.APObjectMapping, announcer *store.Community) error { if !del.HasSummary() { - NativeCommentSelfDeleted.Add(1) + NativeCommentSummarylessDelete.Add(1) return skip(mapping.APID, - "announced delete of a native comment carries no summary: the author's own delete, "+ - "not a moderator's removal — taken so it cannot destroy their record, and "+ - "recorded nowhere because nobody moderated anything") + "announced delete of a native comment carries no summary, so it is not a moderator "+ + "removal: nothing is recorded, and the activity is taken rather than declined so "+ + "it cannot reach the path that destroys the author's record") } + moderation, err := h.moderationState() + if err != nil { + return err + } // announcer.DID is the community authorizeDelete just proved owns this // mapping, so the row is bound to the community that made the decision. - if err := h.objects.SetRemoval(ctx, store.ModeratedObject{ + if err := moderation.SetRemoval(ctx, store.ModeratedObject{ ATURI: mapping.ATURI, APID: mapping.APID, CommunityDID: announcer.DID, @@ -86,14 +116,40 @@ func (h *Handler) moderateNativeComment(ctx context.Context, del *ap.Object, // key would only create a way for a real restore to be dropped, leaving a // removal the moderators lifted standing forever. // -// Clearing an object nobody removed is a no-op success, which is what a -// re-delivered Undo is. +// IT ALSO CLEARS THE LEGACY DELETE STATE, exactly as the post restore does, and +// for exactly the reason RESTORE-1 exists. A native comment mapping CAN be +// soft-deleted and tombstoned — the pre-17c-1 v1 path did both, and the +// origin-verified delete sweep still can — and if this Undo returned without +// clearing them, moderateAnnouncedDelete would decline forever on IsDeleted() +// and the comment would be permanently unmoderatable, by the community's own +// legitimate restore. Both are idempotent single statements and cost nothing in +// the ordinary case, where there is nothing to clear. +// +// Clearing a removal nobody made is a no-op success — a re-delivered Undo, or +// one from a community that never removed this comment — and it is reported as +// such rather than counted as another reversal. func (h *Handler) liftNativeCommentRemoval(ctx context.Context, undo *ap.Object, - mapping *store.APObjectMapping, announcer *store.Community) error { + mapping *store.APObjectMapping, announcer *store.Community, scope string) error { - if err := h.objects.ClearRemoval(ctx, mapping.ATURI, announcer.DID); err != nil { + moderation, err := h.moderationState() + if err != nil { + return err + } + if err := h.tombstones.Remove(ctx, mapping.APID, scope); err != nil { + return fmt.Errorf("ingest: clear tombstone for %s: %w", mapping.APID, err) + } + if err := h.objects.Restore(ctx, mapping.APID); err != nil && !errors.IsNotFound(err) { + return fmt.Errorf("ingest: restore mapping for %s: %w", mapping.APID, err) + } + cleared, err := moderation.ClearRemoval(ctx, mapping.ATURI, announcer.DID) + if err != nil { return fmt.Errorf("ingest: clear removal of %s: %w", mapping.ATURI, err) } + if !cleared { + return skip(mapping.APID, + "announced restore of a native comment that this community had not removed: "+ + "nothing to lift (a re-delivered undo, or an undo of a delete that recorded nothing)") + } NativeCommentRemovalLifted.Add(1) h.logger.Info("community lifted its removal of a native comment", "ap_id", mapping.APID, "at_uri", mapping.ATURI, @@ -162,10 +218,14 @@ func (h *Handler) handleLock(ctx context.Context, lock *ap.Object, announcer *st return err } + moderation, err := h.moderationState() + if err != nil { + return err + } // announcer.DID is the community the authorization above just proved OWNS // this mapping — the same DID CommunityDIDOf answered with — so the row is // bound to the community that made the decision, not to whoever announced. - if err := h.objects.SetLock(ctx, store.ModeratedObject{ + if err := moderation.SetLock(ctx, store.ModeratedObject{ ATURI: mapping.ATURI, APID: mapping.APID, CommunityDID: announcer.DID, diff --git a/internal/ingest/moderation_legacy_test.go b/internal/ingest/moderation_legacy_test.go new file mode 100644 index 0000000..d7ccb53 --- /dev/null +++ b/internal/ingest/moderation_legacy_test.go @@ -0,0 +1,385 @@ +package ingest + +import ( + "context" + "database/sql" + stderrors "errors" + "net/http" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/consume" +) + +// TASK 17c-2 — THE MIGRATION PATH. +// +// The lock tests all build their threads through the CURRENT code, so every +// snapshot they write names its thread root and the resolution never has to +// work for it. That leaves the half of the code that exists ONLY for content +// already in production — walkThreadRoot, its fediverse boundary, its dead end, +// and the fork that weighs an unknown thread against the locks that exist — +// asserted in prose alone. +// +// It is the wrong half to leave untested. New content is written by the code +// under review and can be re-derived if it is wrong; the rows already out there +// are the ones nobody can rewrite, they are the majority on the day the change +// ships, and they are exactly the comments most likely to be sitting in an old +// thread a moderator is about to close. +// +// FIXTURE-FORGING, DELIBERATE AND NARROW: these tests build a comment through +// the real path and then REMOVE the fields the previous version did not write. +// That is not inventing a state — it is the state of every row written before +// this change, reproduced by subtraction, which is the only honest way to hold +// one. makeLegacySnapshot refuses to run if the field it strips was not there, +// so it can never quietly forge nothing. + +// mtDeepReply hangs under mtNestedReply — three levels below the post, so a +// climb to the thread root has to take more than one hop. +var mtDeepReply = nativeComment{ + did: mtCommenterDID, rkey: "3lzmtcomment07", + root: nativeRef{mtPostATURI, mtPostCID}, + parent: nativeRef{mtNestedReply.atURI(), mtNestedReply.createCID}, + createRev: "3lzmtrev000040", createCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf7a", + editRev: "3lzmtrev000041", editCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf7b", + timeUS: 1_775_000_000_000_400, +} + +// TestALegacyThreadIsClimbedToItsLockedRoot is the CLIMB, and it PASSES TODAY. +// +// A reply arriving under comments written before the thread root was recorded +// has to be placed by following what those rows DO name — their parents — up to +// something that answers. Every hop is a row the bridge wrote in an earlier +// version, and the answer decides whether a moderator's lock reaches the reply. +// +// Getting this wrong is invisible in exactly one direction: the reply federates, +// Lemmy rejects it, and the poisoned delivery names a thread nobody connected to +// the lock. So the climb is pinned on a chain more than one hop long — a +// one-hop fixture would pass against an implementation that only ever looked at +// the parent. +func TestALegacyThreadIsClimbedToItsLockedRoot(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + seedNestedThread(t, world) + + // Both comments become PRE-CHANGE rows: neither names its thread, so the + // only way to the post is up the parent chain, two hops. + makeLegacySnapshot(t, h.db, mtDirectReply.atURI(), "rootAtUri") + makeLegacySnapshot(t, h.db, mtNestedReply.atURI(), "rootAtUri") + + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + h.announceLock(world.groupA, mtLockActivity, mtPostAPID) + + err := world.dispatcher.HandleEvent(ctx, mtDeepReply.create(t)) + require.Error(t, err, + "a reply under legacy rows is still a reply in the locked thread: the rows predate "+ + "the recorded root, and the lock is on the post two levels above them") + assert.True(t, stderrors.Is(err, consume.ErrPermanentEvent), + "refused with the same permanence as a reply under a modern row — an author must not "+ + "discover that whether their comment posts depends on when the comment above it "+ + "was written (err=%v)", err) + assert.Contains(t, err.Error(), "parent-locked", "and with the same reason") + assert.Zero(t, outboundRowsFor(t, h.db, mtDeepReply.atURI())) + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries")) + + // The EDIT path climbs too, from the edited comment's own at-uri: an update + // never re-resolves its thread, so a legacy row being edited has to be + // placed the same way. + editErr := world.dispatcher.HandleEvent(ctx, mtNestedReply.edit(t)) + require.Error(t, editErr, + "editing a legacy comment inside a locked thread is refused for the same reason: an "+ + "Update{Note} is a delivery, and the community that closed the thread has stopped "+ + "accepting them") + assert.Contains(t, editErr.Error(), "parent-locked") + + // --- And the climb resolves the other way once the thread reopens. + h.announceUndoLock(world.groupA, mtUnlockActivity, mtLockActivity+"/lock", mtPostAPID) + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtDeepReply.create(t)), + "with the lock lifted the same climb ends at an open thread and the reply federates") + assert.Equal(t, 1, outboundRowsFor(t, h.db, mtDeepReply.atURI())) +} + +// TestADeadEndedThreadHoldsTheReplyRatherThanGuessing pins BOTH forks of the +// undeterminable case, and it PASSES TODAY. +// +// A chain that stops on state this consumer never wrote — a nested row naming no +// parent — cannot be resolved, and "we do not know which thread this is" is not +// "this thread is not locked". But it must not become a permanent refusal +// either: the reply is fine, the state is what is missing. +// +// So the question narrows to the only one still answerable. If the community +// holds NO lock, nothing could have been missed and the reply goes. If it holds +// one, the honest answer is a retryable failure naming the row the chain stopped +// at — retryable because a lifted lock makes it answerable, and named because +// "undeterminable thread" with nothing to look up is not a report an operator +// can act on. +func TestADeadEndedThreadHoldsTheReplyRatherThanGuessing(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + seedNestedThread(t, world) + + // mtDirectReply names neither its thread nor its parent: the chain has + // nowhere to go from here. mtNestedReply above it is an ordinary legacy row, + // so a reply beneath THAT has to climb into the dead end rather than start + // in it — the walk must be what discovers the break, not the fixture. + makeLegacySnapshot(t, h.db, mtDirectReply.atURI(), "rootAtUri", "parentAtUri") + makeLegacySnapshot(t, h.db, mtNestedReply.atURI(), "rootAtUri") + + // FORK ONE: no lock stands anywhere in the community, so an unresolvable + // thread cannot be a locked one. + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtLateNestedReply.create(t)), + "with no lock in the community there is provably nothing to miss, and holding the "+ + "reply would strand ordinary conversation under rows whose only sin is being old") + require.Equal(t, 1, outboundRowsFor(t, h.db, mtLateNestedReply.atURI())) + + // FORK TWO: a lock now stands. The same unresolvable thread might be it. + h.announceLock(world.groupA, mtLockActivity, mtPostAPID) + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + + err := world.dispatcher.HandleEvent(ctx, mtDeepReply.create(t)) + require.Error(t, err, + "an unresolvable thread beside a standing lock must NOT be read as unlocked: that is "+ + "the fail-open the empty root exists to prevent, and it would let exactly the "+ + "oldest threads leak replies past a lock") + assert.False(t, stderrors.Is(err, consume.ErrPermanentEvent), + "and it must be RETRYABLE, not permanent: nothing about this comment is wrong — the "+ + "answer is missing, and it becomes knowable the moment the lock is lifted, so "+ + "dead-lettering it discards a reply that was always going to be fine (err=%v)", err) + assert.Contains(t, err.Error(), mtDirectReply.atURI(), + "and it must NAME the row the chain stopped at: an operator holding 'thread "+ + "undeterminable' with no object to open cannot tell a data bug from a lock") + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries"), + "nothing is delivered while the question is open") + + // And it resolves itself when the lock lifts — the whole reason it is held + // rather than dropped. + h.announceUndoLock(world.groupA, mtUnlockActivity, mtLockActivity+"/lock", mtPostAPID) + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtDeepReply.create(t)), + "the held reply federates on retry once no lock stands") +} + +// TestARematerializationCannotMoveACommentOutOfItsLockedThread PASSES TODAY and +// pins an invariant that is otherwise only claimed in a comment. +// +// The thread a comment hangs in is read back on the moderation path, and the +// delivery that re-materializes it is EDITABLE by the instance sending it. If a +// re-delivery could re-derive the thread from its own inReplyTo, then editing a +// comment out from under a lock would be one Update away — an unprivileged +// reopening of a closed thread, performed by anyone who can get an Update +// announced. +func TestARematerializationCannotMoveACommentOutOfItsLockedThread(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + thread := seedFediverseThread(t, h, world) + + h.announceLock(world.groupA, mtLockActivity, thread.postAPID) + require.Error(t, world.dispatcher.HandleEvent(ctx, thread.reply.create(t)), + "precondition: a reply under the Lemmy comment is refused while the post is locked") + + // A second, UNLOCKED post in the same community, and an Update of the Lemmy + // comment that re-parents it there. + otherPost := h.bridgePost(world.groupA, "77777") + moved := note(mlLemmyComment, mlReplier, otherPost, "re-parented out of the locked thread", + "2026-08-13T10:00:00.000000Z") + require.Equal(t, http.StatusAccepted, h.deliver(world.groupA, + echoAnnounce("https://lemmy.world/activities/announce/update/ml-9101", map[string]any{ + "id": "https://lemmy.world/activities/update/ml-9101", + "type": "Update", + "actor": mlReplier, + "audience": groupID, + "object": moved, + }))) + h.drain() + + mapping, err := h.objects.GetByAPID(ctx, mlLemmyComment) + require.NoError(t, err) + assert.Equal(t, thread.postATURI, mapping.ThreadRootATURI, + "the recorded thread must SURVIVE the re-materialization: a comment cannot change "+ + "threads, and a binding re-derived from an edited delivery is a binding the sender "+ + "chooses") + + err = world.dispatcher.HandleEvent(ctx, thread.reply.create(t)) + require.Error(t, err, + "so the reply is still refused: if an Update could move the comment, reopening a "+ + "locked thread would cost one announced edit and leave no moderation trace") + assert.Contains(t, err.Error(), "parent-locked") +} + +// TestALegacyReplyUnderAFediverseCommentResolvesThroughTheMapping is the bypass +// the climb left open. +// +// walkThreadRoot stops when it reaches something with no outbound state and +// calls that the top of the thread. For a NATIVE parent that is right — the +// bridge wrote every row above it. For a FEDIVERSE parent it is wrong: a Lemmy +// comment has no outbound row by construction, and the thread above it is +// recorded on its MAPPING, in the very row the walk just read past. +// +// The shape is ordinary: a native comment written before the root was recorded, +// hanging under a Lemmy comment. Its replies resolve their thread to that Lemmy +// comment, so a lock on the post above never reaches them — and unlike the +// legacy climb, this does not heal when the parent is re-materialized, because +// the child is what is missing the root. +func TestALegacyReplyUnderAFediverseCommentResolvesThroughTheMapping(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + thread := seedFediverseThread(t, h, world) + + // The pre-change native comment: federated for real, then reduced to the + // row the previous version would have written. + require.NoError(t, world.dispatcher.HandleEvent(ctx, thread.reply.create(t)), + "precondition: a native reply under the Lemmy comment federates while the thread is open") + makeLegacySnapshot(t, h.db, thread.reply.atURI(), "rootAtUri") + + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + h.announceLock(world.groupA, mtLockActivity, thread.postAPID) + + // A reply to that legacy comment. The climb reaches the LEMMY comment, which + // has no outbound row — and stops there unless it asks the mapping. + child := legacyChildOf(thread.reply) + err := world.dispatcher.HandleEvent(ctx, child.create(t)) + require.Error(t, err, + "the thread above a fediverse comment is recorded on its mapping, and the walk reads "+ + "that very row on its way past: stopping there puts every legacy reply under a "+ + "Lemmy comment outside its own locked thread, which is most of Lemmy") + assert.True(t, stderrors.Is(err, consume.ErrPermanentEvent), + "refused permanently, like every other reply in a locked thread (err=%v)", err) + assert.Contains(t, err.Error(), "parent-locked") + assert.Zero(t, outboundRowsFor(t, h.db, child.atURI())) + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries")) +} + +// TestALegacyEditUnderAFediverseCommentResolvesThroughTheMapping is the same +// bypass on the update path, where it is worse: the comment being edited IS the +// legacy row, so the walk starts at it and dead-reckons off its fediverse parent +// every time — an edit surface that stays live inside a closed thread, for as +// long as the row is never re-created. +func TestALegacyEditUnderAFediverseCommentResolvesThroughTheMapping(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + thread := seedFediverseThread(t, h, world) + + require.NoError(t, world.dispatcher.HandleEvent(ctx, thread.reply.create(t)), + "precondition: the native reply federated while the thread was open") + makeLegacySnapshot(t, h.db, thread.reply.atURI(), "rootAtUri") + + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + h.announceLock(world.groupA, mtLockActivity, thread.postAPID) + + err := world.dispatcher.HandleEvent(ctx, thread.reply.edit(t)) + require.Error(t, err, + "an edit inside a locked thread must be refused even when the row predates the "+ + "recorded root: the community stopped accepting deliveries on this thread, and an "+ + "Update is a delivery") + assert.Contains(t, err.Error(), "parent-locked") + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries"), + "and nothing goes out") +} + +// TestAnUndoRestoresALegacySuppressedNativeComment is the permanent-brick case. +// +// Undoing a comment removal clears the bridge-side removal state — and NOTHING +// else. But a native comment's mapping can be soft-deleted and its AP id +// tombstoned by paths that predate (or sit beside) the comment-removal branch: +// the v1 delete path did exactly that before 17c-1 taught moderateAnnouncedDelete +// to take native comments, and the origin-verified delete sweep still can. +// +// A comment in that state is unreachable from both directions at once: +// moderateAnnouncedDelete declines forever on IsDeleted(), so no removal can be +// recorded, and the Undo that would repair it skips straight to the state clear +// without touching either marker. The community owns the content and has said, +// on the wire, that it wants it back — and nothing the bridge does can produce +// that outcome. +// +// The post path already treats both clears as legacy repair for exactly this +// reason; the comment path routes past them. +func TestAnUndoRestoresALegacySuppressedNativeComment(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + seedNativeComment(t, world) + commentAPID := mtDirectReply.apID() + + // The state a pre-17c announced delete left behind: our own mapping soft + // deleted, our own AP id tombstoned under the announcing community. + require.NoError(t, h.objects.SoftDelete(ctx, commentAPID)) + require.NoError(t, h.tombstones.Record(ctx, commentAPID, groupID)) + suppressed, err := h.objects.GetByAPID(ctx, commentAPID) + require.NoError(t, err) + require.True(t, suppressed.IsDeleted(), "precondition: the comment is legacy-suppressed") + + // The community lifts it, honestly signed and announced. + reason := "restored after appeal" + h.announceUndoDelete(world.groupA, + "https://lemmy.world/activities/announce/undo/ml-suppressed", + "https://lemmy.world/activities/announce/delete/ml-suppressed/delete", + commentAPID, &reason) + + mapping, err := h.objects.GetByAPID(ctx, commentAPID) + require.NoError(t, err) + assert.False(t, mapping.IsDeleted(), + "the restore must clear the SOFT DELETE too: leaving it makes the comment "+ + "permanently unmoderatable — moderateAnnouncedDelete declines on IsDeleted(), so "+ + "the community can never remove it again either, and the Undo it just sent is the "+ + "only repair anyone was going to attempt") + + tombstoned, err := h.tombstones.ExistsFor(ctx, commentAPID, groupID) + require.NoError(t, err) + assert.False(t, tombstoned, + "and the marker with it: a standing tombstone against our own AP id suppresses this "+ + "comment's later activities and drops the Lemmy replies beneath it") + + // The proof that the repair is real rather than cosmetic: the community can + // moderate the comment again. + h.announceDeleteWithSummary(world.groupA, + "https://lemmy.world/activities/announce/delete/ml-again", commentAPID, &reason) + state, found := removalStateFor(t, h.db, mtDirectReply.atURI()) + assert.True(t, found && state.removed, + "a restored comment is moderatable again: that round trip — removed, restored, "+ + "removable — is the whole difference between state and a brick") +} + +// legacyChildOf is a new reply hanging under an existing comment. +func legacyChildOf(parent nativeComment) nativeComment { + return nativeComment{ + did: mtAuthorDID, rkey: "3lzmtcomment08", + root: nativeRef{mtPostATURI, mtPostCID}, + parent: nativeRef{parent.atURI(), parent.createCID}, + createRev: "3lzmtrev000050", createCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf7c", + editRev: "3lzmtrev000051", editCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf7d", + timeUS: 1_775_000_000_000_500, + } +} + +// makeLegacySnapshot reduces a stored snapshot to the shape the PREVIOUS +// version wrote, by removing fields it did not have. +// +// It REFUSES to strip a field that is not there. That check is the whole +// integrity of this fixture: if the current path ever stops writing one of these +// keys — renamed, moved, dropped — a silent no-op here would leave every legacy +// test passing against ordinary modern rows, asserting nothing about the +// migration path they exist to cover. +func makeLegacySnapshot(t *testing.T, db *sql.DB, atURI string, fields ...string) { + t.Helper() + ctx := context.Background() + for _, field := range fields { + var present bool + require.NoError(t, db.QueryRowContext(ctx, + `SELECT jsonb_exists(translated_snapshot, $2) FROM outbound_objects WHERE at_uri = $1`, + atURI, field).Scan(&present), "read the snapshot of %s", atURI) + require.True(t, present, + "precondition: the CURRENT path writes %q into %s's snapshot, so removing it "+ + "produces the row the previous version wrote — a field that is already absent "+ + "means this fixture is forging nothing", field, atURI) + + _, err := db.ExecContext(ctx, + `UPDATE outbound_objects SET translated_snapshot = translated_snapshot - $2::text WHERE at_uri = $1`, + atURI, field) + require.NoError(t, err, "strip %q from %s", field, atURI) + } +} diff --git a/internal/ingest/moderation_lock_test.go b/internal/ingest/moderation_lock_test.go index 0b9ee52..6a265c6 100644 --- a/internal/ingest/moderation_lock_test.go +++ b/internal/ingest/moderation_lock_test.go @@ -354,17 +354,17 @@ func TestAnEditBeneathALockedThreadIsRefused(t *testing.T) { "and the edit went out") } -// TestALockReachesOnlyItsOwnThread is the CONTROL, and it PASSES TODAY — nothing -// refuses a nested reply at all, so it can only fail once a root-aware refusal -// exists to over-reach. It is here as the negative half of the two tests above, -// and it has been tooth-checked (an unconditional refusal in -// refuseUnderLockedParent turns it red). +// TestALockReachesOnlyItsOwnThread pins the SCOPE of the two refusals above: a +// lock reaches its own thread and no further. // // The scope it pins is per-OBJECT. A second thread in a DIFFERENT community // could not pin it: an implementation that refused every comment in a community // holding any locked post would pass that test and fail this one. Same // community, same author, same nesting — the only difference is which post the // thread hangs from, which is exactly the difference the lock is keyed on. +// +// Tooth-checked when it was written: pointing the lock lookup at a fixed at-uri +// — the per-community over-reach — turns it red on its own assertion. func TestALockReachesOnlyItsOwnThread(t *testing.T) { h := newHarness(t) ctx := context.Background() @@ -732,8 +732,8 @@ func TestAReplyBeneathALockedFediverseThreadIsRefused(t *testing.T) { "and it federates") } -// TestAReplyBeneathAnUnlockedFediverseThreadFederates is the CONTROL, and it -// PASSES TODAY — nothing refuses these replies at all. +// TestAReplyBeneathAnUnlockedFediverseThreadFederates pins the ordinary case +// the thread read must never break. // // It is the half that costs something to get wrong in the other direction: // reading a thread root out of a materialized record is a read that can fail, @@ -758,8 +758,8 @@ func TestAReplyBeneathAnUnlockedFediverseThreadFederates(t *testing.T) { assert.Greater(t, rowCount(t, h.db, "outbound_deliveries"), deliveriesBefore) } -// TestAFediverseThreadIsUnaffectedByALockElsewhere is the SCOPE control for the -// same read, and it also PASSES TODAY. +// TestAFediverseThreadIsUnaffectedByALockElsewhere is the SCOPE half of the +// same read. // // The community holds a real, standing lock — on the NATIVE post — while the // Lemmy thread beside it is open. This is the case a conservative fallback @@ -767,6 +767,10 @@ func TestAReplyBeneathAnUnlockedFediverseThreadFederates(t *testing.T) { // holds locks, so refuse" would park every reply to every fediverse comment in // any community that has ever locked one post. The refusal must be about THIS // thread or it is not about a thread at all. +// +// Tooth-checked with its sibling above: making a fediverse parent dead-end and +// refusing on the dead end turns both red, which is what a conservative +// fallback would do to every open thread in a community that locked one post. func TestAFediverseThreadIsUnaffectedByALockElsewhere(t *testing.T) { h := newHarness(t) ctx := context.Background() @@ -913,8 +917,11 @@ func TestACommunitysRemovalOfANativeCommentIsRecorded(t *testing.T) { "and the restore enqueues nothing either") } -// TestCrossCommunityRemovalOfANativeCommentIsRefused is the RELATIONAL control, -// and it PASSES TODAY — no removal is recorded for anyone yet. +// TestCrossCommunityRemovalOfANativeCommentIsRefused is the RELATIONAL half of +// the removal above: community B, co-hosted with A and followed like A, may not +// remove A's comment. Decision 18's conjunction collapses in a one-community +// fixture, so this is the only shape that can tell "the signer IS the community" +// from "the target is IN the community". func TestCrossCommunityRemovalOfANativeCommentIsRefused(t *testing.T) { h := newHarness(t) world := newModerationWorld(t, h) @@ -936,7 +943,7 @@ func TestCrossCommunityRemovalOfANativeCommentIsRefused(t *testing.T) { } // TestAPostRemovalWritesNoBridgeSideRemovalState is the SCOPE control for ruling -// C, and it PASSES TODAY. +// C: the bridge-side removal state is COMMENTS ONLY. // // removed_at is COMMENTS ONLY. A post's removal is a record in the community's // own repo, written by acceptrec in ONE commit with the withdrawal of the @@ -1088,17 +1095,18 @@ func TestASummarylessDeleteOfANativeCommentRecordsNothing(t *testing.T) { assert.NotEqual(t, skipReasonFor(t, h, removalActivity, mtDirectReply.apID()), skipReasonFor(t, h, selfDeleteActivity, mtNestedReply.apID()), - "and the bridge must not tell the SAME story about both: today a moderator's removal "+ - "of a comment and an author's own delete are logged with one reason and counted in "+ - "one counter, so the operator asking 'did a moderator remove this?' reads an answer "+ - "that cannot distinguish yes from no") + "and the bridge must not tell the SAME story about both: one reason and one counter "+ + "for a moderator's removal and an author's own delete leaves the operator asking "+ + "'did a moderator remove this?' reading an answer that cannot distinguish yes from "+ + "no — and the skip reason is the only place either outcome is written down") } // TestASummarylessDeleteOfANativeCommentByOurPersonaIsDroppedAsAnEcho records // which mechanism is actually load-bearing when the attribution is TRUTHFUL. // -// CHARACTERIZATION: this passes today, and it is the comment-shaped twin of the -// post case 17c-1 pinned. A native comment's author IS one of our personas, so a +// CHARACTERIZATION of behaviour 17a already established — the comment-shaped +// twin of the post case 17c-1 pinned, and tooth-checked by removing the +// classifier's actor probe, which drops the count to zero. A native comment's author IS one of our personas, so a // truthful self-delete announced back at us is indistinguishable from our own // Delete coming home — and the echo classifier takes it by the inner ACTOR, // before any authorization or moderation branch runs. diff --git a/internal/store/ap_objects.go b/internal/store/ap_objects.go index 523786f..6fac602 100644 --- a/internal/store/ap_objects.go +++ b/internal/store/ap_objects.go @@ -14,11 +14,18 @@ import ( type postgresAPObjects struct { db *sql.DB + // The object_moderation repository is EMBEDDED, and only so that a holder of + // the concrete mapping store can be type-asserted to store.ObjectModeration + // (ingest.NewHandler defaults its moderation store that way). It is + // deliberately NOT part of the APObjects interface: five callers hold that + // interface to resolve strongRefs, and none of them may reach a moderation + // mutator. + postgresObjectModeration } // NewAPObjects creates the postgres-backed ap_objects repository. func NewAPObjects(db *sql.DB) APObjects { - return &postgresAPObjects{db: db} + return &postgresAPObjects{db: db, postgresObjectModeration: postgresObjectModeration{db: db}} } const apObjectColumns = ` diff --git a/internal/store/interfaces.go b/internal/store/interfaces.go index d3cc91a..1fb4642 100644 --- a/internal/store/interfaces.go +++ b/internal/store/interfaces.go @@ -15,15 +15,18 @@ import ( ) // ObjectModeration is the moderation state the BRIDGE owns for one bridged -// object — today a community's thread LOCK, which has no home in either repo -// (see migration 025). +// object: a community's thread LOCK, and its removal of a native COMMENT — +// decisions with no home in either repo (see migration 025). // -// It rides the APObjects interface, and the two are deliberately different -// things at different levels: the TABLE is separate, because putMapping -// rewrites a whole ap_objects row and a re-pin would clear a lock; the -// ACCESSOR sits here because every holder of a mapping is exactly the caller -// that needs to ask, and both readers (the announced-moderation dispatch and -// the native comment consumer) already hold one. +// It is its OWN interface, held only by the two places that moderate: the +// announced-moderation dispatch and the native comment consumer. It was briefly +// embedded in APObjects, which every strongRef resolution holds — the echo +// classifier, the vote aggregator, the stats refresher, the materializer, the +// outbound enqueuer — and that put SetLock/SetRemoval/ClearRemoval within reach +// of five callers that have no business moderating anything. The TABLE is +// separate from ap_objects for its own reason (putMapping rewrites a whole +// mapping row, so a re-pin would clear a lock); this separation is the other +// one, and they are not the same argument. type ObjectModeration interface { // SetLock records or clears a community's lock on an object. Locking an // already-locked object preserves the ORIGINAL locked_at — a re-announced @@ -57,7 +60,12 @@ type ObjectModeration interface { // made it: clearing is a no-op for anyone else, and a no-op success for an // object nobody removed. The row survives — a lock on the same object is a // separate decision and is not lifted with it. - ClearRemoval(ctx context.Context, atURI, communityDID string) error + // + // cleared reports whether a removal was actually standing, so a caller can + // count and log what HAPPENED rather than what was attempted: a re-delivered + // Undo, or one for a comment this community never removed, is a no-op and + // must not read in the metrics as another moderator reversal. + ClearRemoval(ctx context.Context, atURI, communityDID string) (cleared bool, err error) // CommunityHoldsAnyLock reports whether a community currently holds a lock // on anything at all. It answers the ONE question left when a comment's @@ -72,8 +80,6 @@ type ObjectModeration interface { // as, and back. Every materialization writes a mapping; every strongRef // resolution reads one. type APObjects interface { - ObjectModeration - // PutMapping idempotently upserts a mapping keyed on APID. It validates // DID, Collection, RKey, and CID, derives ATURI from the first three, // and returns the stored row. An empty Origin defaults to diff --git a/internal/store/object_moderation.go b/internal/store/object_moderation.go index 664890b..fb2500d 100644 --- a/internal/store/object_moderation.go +++ b/internal/store/object_moderation.go @@ -5,19 +5,32 @@ import ( "database/sql" stderrors "errors" "fmt" + "slices" "github.com/lib/pq" "tidepool/internal/errors" ) -// object_moderation is the bridge-owned moderation state described on the -// ObjectModeration interface and in migration 025. The methods hang off the -// ap_objects repository — the callers that need them all hold one — but the -// ROW is separate, because putMapping rewrites a mapping wholesale and a -// re-pin must never clear a lock. +// postgresObjectModeration is the object_moderation repository: the moderation +// state the bridge itself owns, described on the ObjectModeration interface and +// in migration 025. +// +// It is its own repository over its own table, and postgresAPObjects EMBEDS it +// so a holder of the concrete mapping store can be type-asserted to +// ObjectModeration (see ingest.NewHandler's default) — a wiring convenience that +// deliberately does NOT widen the APObjects interface, because a store held for +// strongRef resolution must not carry moderation mutators. +type postgresObjectModeration struct { + db *sql.DB +} + +// NewObjectModeration creates the postgres-backed object_moderation repository. +func NewObjectModeration(db *sql.DB) ObjectModeration { + return &postgresObjectModeration{db: db} +} -func (r *postgresAPObjects) SetLock(ctx context.Context, object ModeratedObject, locked bool) error { +func (r *postgresObjectModeration) SetLock(ctx context.Context, object ModeratedObject, locked bool) error { if object.ATURI == "" { return errors.NewValidationError("at_uri", "must not be empty") } @@ -63,7 +76,7 @@ func (r *postgresAPObjects) SetLock(ctx context.Context, object ModeratedObject, return nil } -func (r *postgresAPObjects) SetRemoval(ctx context.Context, object ModeratedObject, code, reason string) error { +func (r *postgresObjectModeration) SetRemoval(ctx context.Context, object ModeratedObject, code, reason string) error { if object.ATURI == "" { return errors.NewValidationError("at_uri", "must not be empty") } @@ -100,33 +113,42 @@ func (r *postgresAPObjects) SetRemoval(ctx context.Context, object ModeratedObje return nil } -func (r *postgresAPObjects) ClearRemoval(ctx context.Context, atURI, communityDID string) error { +func (r *postgresObjectModeration) ClearRemoval(ctx context.Context, atURI, communityDID string) (bool, error) { if atURI == "" { - return errors.NewValidationError("at_uri", "must not be empty") + return false, errors.NewValidationError("at_uri", "must not be empty") } if communityDID == "" { - return errors.NewValidationError("community_did", "must not be empty") + return false, errors.NewValidationError("community_did", "must not be empty") } // The code and the reason go with it: they describe a decision that no // longer stands, and leaving them behind would let an admin surface read a // live reason off a lifted removal. - if _, err := r.db.ExecContext(ctx, ` + // + // `removed_at IS NOT NULL` is what makes the row count meaningful: without + // it, an UPDATE that blanked already-blank columns would report one row + // affected and the caller would log a reversal that reversed nothing. + result, err := r.db.ExecContext(ctx, ` UPDATE object_moderation SET removed_at = NULL, removal_code = '', removal_reason = '', updated_at = now() - WHERE at_uri = $1 AND community_did = $2`, - atURI, communityDID); err != nil { - return fmt.Errorf("clear removal of %q: %w", atURI, err) + WHERE at_uri = $1 AND community_did = $2 AND removed_at IS NOT NULL`, + atURI, communityDID) + if err != nil { + return false, fmt.Errorf("clear removal of %q: %w", atURI, err) } - return nil + affected, err := result.RowsAffected() + if err != nil { + return false, fmt.Errorf("clear removal of %q: rows affected: %w", atURI, err) + } + return affected > 0, nil } -func (r *postgresAPObjects) LockedAmong(ctx context.Context, atURIs ...string) (string, error) { +func (r *postgresObjectModeration) LockedAmong(ctx context.Context, atURIs ...string) (string, error) { // The caller names a thread — a parent and a root, sometimes the same object // twice — so the empties and duplicates it may hold are filtered here rather // than at every call site. candidates := make([]string, 0, len(atURIs)) for _, atURI := range atURIs { - if atURI != "" && !containsString(candidates, atURI) { + if atURI != "" && !slices.Contains(candidates, atURI) { candidates = append(candidates, atURI) } } @@ -150,7 +172,7 @@ func (r *postgresAPObjects) LockedAmong(ctx context.Context, atURIs ...string) ( return locked, nil } -func (r *postgresAPObjects) CommunityHoldsAnyLock(ctx context.Context, communityDID string) (bool, error) { +func (r *postgresObjectModeration) CommunityHoldsAnyLock(ctx context.Context, communityDID string) (bool, error) { if communityDID == "" { return false, errors.NewValidationError("community_did", "must not be empty") } @@ -166,14 +188,3 @@ func (r *postgresAPObjects) CommunityHoldsAnyLock(ctx context.Context, community } return held, nil } - -// containsString reports whether the slice already holds s. The candidate sets -// here are two or three elements, so a scan beats building a map. -func containsString(values []string, s string) bool { - for _, v := range values { - if v == s { - return true - } - } - return false -} -- 2.51.2