From 44fae513fb888df370f0c4972d1e64e002ddfff3 Mon Sep 17 00:00:00 2001 From: Bretton Date: Sat, 15 Aug 2026 03:20:26 +0000 Subject: [PATCH] feat(reconcile): report what disagrees, and never repair it (17e) The last of task 17. A sweep comparing atproto state against outbound state, surfaced as GET /admin/divergence and four sweep-set gauges. IT NEVER WRITES — not to remote instances, and not to our own tables (decision 19). That is not a limitation to work around; it is the reason the job is safe to run on a timer. Both sides of every comparison are LOCAL, which is also the line that decided scope: of the eight divergences prior sub-runs routed here, the four whose comparison can be read from our own tables ship, and the four that need a peer's state are recorded in FOLLOWUPS with the reasoning. The outer acceptance test enforces the constraint literally — it snapshots fifteen tables whole-row before and after the sweep and requires byte-identity, because a reconciler that writes has no witness: it is the only thing reading both sides, so a wrong rule silently rewrites the state it was measuring and every later sweep agrees with itself. WHAT IT REPORTS Persona vote events (expected 0, forever): vote_events joined to ap_actors on VOTER IDENTITY. Never on activity ids — Lemmy 0.19 reconstructs an echoed Undo with a freshly generated inner id, typed "Like" even when the live vote is a Dislike, so an id-keyed probe misses every echoed Undo while appearing to work. And never against bridged_actors, who are real Lemmy humans mirrored into atproto: that swap reports every genuine vote on the network as our own echo and takes the vote pipeline dark. Acceptances that never reached the peer, classified cancelled / poisoned / stale-pending, which is the operator's whole triage and subsumes the lapsed-ban residual with no second query. A delivery HELD FOR SETTLEMENT is excluded: it looks exactly like a stuck one and is the opposite — the peer already accepted it, and accepted_at is missing precisely because writing it is the step that failed. Reporting it would fire on every settlement retry, and an operator who learns to ignore this report also ignores the cancelled acceptance beside it. The re-cast divergence: a vote the peer holds that our own state no longer claims. Read from the append-only activity history, with the vote row present only as an exclusion, because the vote row is what this bug erases — voteCallback resolves by activity id and returns nil on NotFound, so an old in-flight delivery landing after a re-cast no-ops silently. The Undo is matched by TIME, not by id: a re-cast mints new ids, so id-chasing would miss an Undo that withdrew an earlier incarnation of the same pair. Poisoned deliveries as explicitly UNKNOWN, partitioned by evidence quality rather than by outcome: a refusal carrying a status is evidence of non-application but not proof, while a transport failure is silent about whether the peer applied it. They are separate sub-counts and neither claims the peer does or does not hold it. The metric names say "unknown" and a test asserts the name strings, because a metric name is a claim and the only true claim here is that we do not know. Nothing proposes a repair: /admin/outbound /redrive already exists and is deliberately refused unscoped, so a report with a button would make that refusal decorative. DETAILS THAT ARE LOAD-BEARING Gauges are sweep-set, never expvar.Func — these are multi-table joins and a scrape must not run them on the endpoint an operator is reading BECAUSE the system is broken. They publish -1 until the first sweep completes, so the window before any sweep cannot masquerade as health, and a FAILED sweep leaves the previous values standing rather than writing a zero that reads as healthy. Every class is registered in one gauge map, so a class added to the report without a gauge fails at the map literal instead of going unwatched. The entry cap is applied at append time rather than by truncating a finished list: a report that must first be built in full is not bounded, and the sweep that hits the cap is running against the database with the stopped queue in it. Counts stay true even when entries are capped, and the response says it truncated. Refused is stored rather than derived, because production spells "no answer came back" as a literal 0, not NULL — deriving it from NULL would file every dial timeout on the network under the count that says the peer replied. No migration. Every column needed already existed. Tests: whole-package -count=2 across store/ingest/votes/outbound/consume, full suite 21/21, e2e 288s. Every control tooth-checked, including the two that were vacuous until the implementation landed. Co-Authored-By: Claude Opus 5 (1M context) --- FOLLOWUPS.md | 107 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ cmd/tidepool/main.go | 17 +++++++++++++++++ internal/config/config.go | 13 +++++++++++++ internal/ingest/divergence.go | 493 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/ingest/divergence_endpoint_test.go | 424 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/ingest/divergence_test.go | 676 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/ingest/follow.go | 35 +++++++++++++++++++++++++++++++++++ internal/store/divergence.go | 440 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/store/divergence_acceptance_test.go | 361 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/store/divergence_test.go | 236 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/store/divergence_unknown_test.go | 239 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/votes/divergence_recast_test.go | 224 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 12 file(s) changed, 3265 insertion(s)(+), 0 deletion(s)(-) diff --git a/FOLLOWUPS.md b/FOLLOWUPS.md --- a/FOLLOWUPS.md +++ b/FOLLOWUPS.md @@ -404,3 +404,110 @@ one that will never be sent.** The purge enumerates what the peer holds (delivered, plus held-for-settlement), but a delivery the worker has claimed and is POSTing right now is neither. Detecting it needs the peer's state, not ours. Belongs to 17e (reconciliation). + +## Deferred by 17e (reconciliation scoped to detect-only) + +17e reports divergence and never repairs it (decision 19). These are the repairs +and the comparisons it deliberately did not build. + +- **The re-cast race, leg 1 — `upsert` clobbers a delivered vote.** + `internal/consume/votes.go` hardcodes `pending` on the vote write and + `internal/store/outbound_votes.go` `Upsert` sets + `delivered_state = EXCLUDED.delivered_state`, so re-casting a DELIVERED vote + resets the row to pending while Lemmy still holds the OLD vote in the OLD + direction: we subtract nothing and keep our stale vote. Transient normally, + PERMANENT if that delivery poisons. THE FIX, and it already has a model in + the tree: make the upsert refuse to write `pending` over `delivered` exactly + the way `SetDeliveredState` now refuses to write over `undone` (17d), so a + re-cast leaves a row that still owes an Undo. Vote-accounting change with its + own RED test — 17e's report is its regression oracle, which is why the report + ships first. + +- **The re-cast race, leg 2 — the settlement silently forgets the old vote. + Recorded nowhere before now.** `internal/outbound/worker.go` `voteCallback` + resolves via `GetByActivityID(activity.ActivityID)` and returns `nil` on + NotFound. After a re-cast the OLD Like's activity id no longer matches + `current_activity_id`, so when that in-flight old delivery lands the callback + no-ops and the delivery is marked delivered. The peer then demonstrably holds + a vote that NO `outbound_votes` column records at all. This is why 17e reads + divergence out of the delivery ledger (`outbound_activities` joined to + `outbound_deliveries`) rather than out of the vote row: the vote row is + exactly what the bug erases. + +- **REJECTED, with reasons, so it is not re-proposed: the "what the peer holds" + vs "what the user wants" column pair.** It changes the write path that feeds + a number users read — 17b's binding ruling pins `delivered_state = 'delivered'` + as a positive equality in `SeedAggregates`, and 17d made that correct only + because `undone` is terminal; a second pair forces the seeder, the purge's + `ListStandingForActor`, `voteCallback` and `CancelOutwardForActorTx` to each + re-decide which column they meant. It also relocates the uncertainty rather + than removing it: the pair's only honest maintainer is the delivery callback, + which is where leg 2 already lives — two things that can be wrong instead of + one, and then reconciliation has to reconcile *them*. Detection needs none of + it; `outbound_activities` is append-only and already durable. + +- **Instances that learned of an actor via search/WebFinger are unreachable by + any local comparison.** `DistinctInboxesForActor` is, in its own words, the + only record of which instances hold a user's content. Building the other side + would mean logging the requesting host of every WebFinger and actor-document + GET — a new surveillance log built to serve a deletion, which is the wrong + trade for an erasure feature. 17e's report states its own coverage bound + instead: the fan-out set is the delivery history. A reconciler reporting zero + here would be asserting completeness it cannot have. + +- **A delivery cancelled while mid-POST cannot be distinguished locally, and + the evidence is erased on purpose.** Every cancel statement writes + `claimed_until = NULL`, so after the cancel commits a row cancelled mid-flight + is byte-identical to one cancelled while idle. Recovering it needs a new + column plus edits to the five most safety-critical cancel statements in the + outbound package — to record a fact that still would not say whether the POST + landed. If ever wanted, the cheap version is `RETURNING` a count of rows with + `claimed_until > now()` at each cancel site, bumping a counter where the + decision is actually taken rather than reconstructing it later. + +- **A ban-caused cancellation is indistinguishable from a consent cancellation** + in `outbound_deliveries` — `cancelForActor`, the community cancel, the ban's + intersection cancel and the consent cancel all write the same `cancelled` + with no reason column. 17e reports the count rather than inventing the column. + `community_bans` is joinable on `(community_did, subject_did)` with + `expires_at IS NULL OR expires_at > now()`, so the reason is expressible + read-side today if an operator needs it. + +- **`bindFetch` own-id belt (noted in 17a, still unimplemented).** A typed + "refusing to fetch our own id" error in `internal/ingest/handler.go` would + cover the ancestor walk, bare-IRI announce, `resolveDelivered`, and + `handleUndoDelete`'s restore in ONE place. Explicitly NOT 17e's: it changes + what the inbound path processes, and a sub-run whose constraint is "report, + never act" should not ship a new drop site — on the highest-volume inbound + path, which already carries a recorded perf concern. + +- **The re-cast divergence sweep drives off a sequential scan of + `outbound_activities`, the highest-volume table in the system.** Confirmed by + `EXPLAIN` on the real schema: every other leg is index-served (the Undo + subquery rides `idx_outbound_activities_actor`, the delivered-delivery joins + ride `idx_outbound_deliveries_activity`, the ledger exclusion rides + `outbound_votes_actor_delivered_idx`), but the outer driver is a full scan + filtered on `kind IN ('Like','Dislike') AND parent_at_uri <> ''`, on every + sweep. Deliberately NOT fixed in 17e — adding an index there is a migration + whose write cost lands on the delivery worker's hot path, and the sweep runs + every 15 minutes against a table that is currently small. The fix when it is + needed: + + CREATE INDEX outbound_activities_vote_subject_idx + ON outbound_activities (actor_did, parent_at_uri, created_at) + WHERE kind IN ('Like','Dislike'); + + It serves the driver and tightens the Undo subquery, which today index-scans + by `actor_did` and applies kind/parent_at_uri/created_at as residuals; a + sibling partial index on `kind = 'Undo'` finishes that half. Row estimates + from a near-empty test database are meaningless — the plan SHAPE is the + finding, so re-profile against production volumes before choosing. + +- **An acceptance with NO delivery row at all is not reported by the + acceptance-undelivered classes**, and this is stated in the query rather than + swallowed. All three classes are delivery states, and a row with no delivery + has none; folding it into the nearest class would misdescribe it. It should + not be reachable going forward — the acceptance record, the + `outbound_objects` row, the delivery enqueue and the `admissions` ledger row + all land in ONE transaction — so a test asserting it today would be vacuous. + If it is ever observed, it wants its own class, not a widened existing one. diff --git a/cmd/tidepool/main.go b/cmd/tidepool/main.go --- a/cmd/tidepool/main.go +++ b/cmd/tidepool/main.go @@ -482,6 +482,23 @@ admin.SetFollowReconciler(reconciler) go reconciler.Run(ctx) } + // The reconciliation sweep (task 17e, decision 19). Unlike the follow + // reconciler there is nothing to gate it on: both sides of every comparison + // are local, so the database is the only dependency, and the sweep writes + // NOTHING — to peers or to our own tables — which is what makes running it + // on a schedule safe. The interval only sets how often the background pass + // refreshes the gauges; GET /admin/divergence runs one on demand. + divergence, err := ingest.NewDivergenceReconciler(ingest.DivergenceOptions{ + DB: database, + Interval: cfg.DivergenceInterval, + Logger: logger, + }) + if err != nil { + return err + } + admin.SetDivergenceReconciler(divergence) + go divergence.Run(ctx) + // The vote-aggregate XRPC (the AppView's side-channel read). votesXRPC, err := votes.NewXRPC(votes.XRPCOptions{DB: database, Logger: logger}) if err != nil { diff --git a/internal/config/config.go b/internal/config/config.go --- a/internal/config/config.go +++ b/internal/config/config.go @@ -246,6 +246,15 @@ // positive). Periodic re-sweeps self-heal subscribes that failed // because the remote was down at startup; pending→accepted retries are // the follow retrier's job, not the reconciler's. FollowListInterval time.Duration + // DivergenceInterval is the reconciliation sweep's cadence + // (DIVERGENCE_INTERVAL, a Go duration, default 15m, must be positive). + // + // The sweep is always wired — both sides of every comparison are local, so + // it has nothing to be configured WITH — and this knob only decides how + // often the background pass refreshes the gauges. GET /admin/divergence + // runs one on demand regardless. It never writes anything (decision 19), + // which is what makes an always-on schedule safe. + DivergenceInterval time.Duration } // Load reads configuration from the environment. logger must not be nil; @@ -579,6 +588,10 @@ // contents are validated at startup by the caller, not here — config // only carries the knob. cfg.FollowListPath = os.Getenv("FOLLOW_LIST_PATH") cfg.FollowListInterval, err = durationVar(logger, "FOLLOW_LIST_INTERVAL", 15*time.Minute) + if err != nil { + return nil, err + } + cfg.DivergenceInterval, err = durationVar(logger, "DIVERGENCE_INTERVAL", 15*time.Minute) if err != nil { return nil, err } diff --git a/internal/ingest/divergence.go b/internal/ingest/divergence.go new file mode 100644 --- /dev/null +++ b/internal/ingest/divergence.go @@ -0,0 +1,493 @@ +package ingest + +import ( + "context" + "database/sql" + "expvar" + "fmt" + "log/slog" + "time" + + "tidepool/internal/errors" + "tidepool/internal/store" +) + +// The reconciliation job (task 17e, PLAN.md decision 19): compare atproto state +// against outbound state, REPORT what disagrees, and change nothing. +// +// The prohibition is the design rather than a caveat on it. This sweep is the +// only thing that reads both sides, so if it healed and its rule were wrong, +// there would be no witness — it would silently rewrite the state it was +// measuring and every later sweep would agree with itself. Half of what it can +// see is genuinely unknowable from here (a poisoned delivery may or may not have +// reached the peer; a held settlement is a success that looks pending), and the +// actions available are the irreversible ones: re-sending, cancelling, deleting +// on an instance that does not un-delete. An operator reading a report can weigh +// that. A loop cannot. +// +// The shape mirrors FollowReconciler (reconcile.go): validated options, an +// exported Sweep the admin endpoint and tests drive synchronously, and a Run +// that loops over it. + +// defaultDivergenceInterval is the sweep cadence when options leave it zero. +const defaultDivergenceInterval = 15 * time.Minute + +// DefaultAcceptanceStaleAfter is how long a delivery may sit pending before its +// absence from the peer is worth an operator's attention. +// +// It is a named constant rather than a literal in a query because it is a +// POLICY — the line between "in flight" and "stuck" — and the person tuning it, +// or deciding whether the report is crying wolf, has to be able to find it. +// +// TWELVE HOURS, derived rather than picked. Two envelopes make a pending +// delivery legitimately old: the retry schedule (DefaultMaxDeliveryAttempts=8 +// steps of DefaultBackoffBase=30s doubling to a one-hour cap, so roughly two to +// three hours before a failing delivery poisons) and the causal wait +// (DefaultCausalWaitBudget=6h, during which a reply sits pending for a parent +// that has not been accepted). A window inside either one reports the system +// working. Twelve hours clears both with room, and stays well inside "noticed +// the same day" — a queue that stopped moving this morning is in the report +// before the day ends. +const DefaultAcceptanceStaleAfter = 12 * time.Hour + +// Divergence classes. The class is what an operator triages on, so it names the +// SHAPE of the disagreement rather than the query that found it. +const ( + // DivergencePersonaVoteEvent: an inbound vote_events row attributed to one + // of our own personas. Expected count zero, forever — 17a's voter probe + // refuses these before they are written — so one existing means the probe + // was bypassed and the subject's tally counts our own vote twice: once as + // this inbound event, once as the delivered outbound row 17b's reseed + // subtracts from the origin's total. + DivergencePersonaVoteEvent = "persona-vote-event" + + // DivergenceVoteRecastUndelivered: a peer is holding a vote this bridge no + // longer claims. Re-casting a delivered vote resets the row to pending under + // a new activity id; when that new delivery poisons, the peer keeps counting + // the OLD vote while our ledger accounts for nothing — permanently, since + // nothing re-drives a poisoned delivery and the reseed subtracts only + // delivered rows. + DivergenceVoteRecastUndelivered = "vote-recast-undelivered" + + // DivergenceDeliveryUnknownRefused / …Unanswered: a delivery we SENT and + // never got confirmation for. Both names carry UNKNOWN because that is the + // only true claim available: refused means the peer answered with a status + // (evidence of non-application, not proof — a peer can apply and then fail + // to answer), and unanswered means a transport failure that is silent about + // everything. Neither may be folded into a class whose name asserts what the + // peer holds. + DivergenceDeliveryUnknownRefused = "delivery-unknown-refused" + DivergenceDeliveryUnknownUnanswered = "delivery-unknown-unanswered" + + // The undelivered-acceptance classes. A community's own repo says it + // accepted the post — Coves shows it there — and the peer was never told. + // + // THREE CLASSES RATHER THAN ONE, because they are three different jobs. One + // "undelivered" bucket would tell an operator how many posts are missing + // from Lemmy and nothing about what to do about any of them, and it could + // only ever be alerted on at the noise level of whichever kind is most + // common. + // + // DivergenceAcceptanceCancelled: a DECISION took the delivery out of the + // queue — a ban, an opt-out. Usually correct to leave exactly as it is; the + // report exists so the resulting Coves-only post is visible rather than + // silent. + DivergenceAcceptanceCancelled = "acceptance-undelivered-cancelled" + // DivergenceAcceptancePoisoned: the delivery FAILED. Redrivable, through an + // operator surface that already exists (POST /admin/outbound/redrive). + DivergenceAcceptancePoisoned = "acceptance-undelivered-poisoned" + // DivergenceAcceptanceStale: still pending long past the window. Nothing is + // wrong with the post — the QUEUE is not moving, which is an investigation + // that starts at the worker rather than at the community. + DivergenceAcceptanceStale = "acceptance-undelivered-stale" +) + +// Divergence gauges. The "tidepool" prefix is load-bearing: scopedMetrics +// (follow.go) serves ONLY keys carrying it, so a gauge named without it is +// published to expvar and then filtered out of the endpoint — invisible in +// exactly the way a check that never runs is invisible. +const ( + MetricDivergencePersonaVoteEvents = "tidepool_divergence_persona_vote_events" + MetricDivergenceAcceptanceCancelled = "tidepool_divergence_acceptance_undelivered_cancelled" + MetricDivergenceAcceptancePoisoned = "tidepool_divergence_acceptance_undelivered_poisoned" + MetricDivergenceAcceptanceStale = "tidepool_divergence_acceptance_undelivered_stale" + MetricDivergenceVoteRecast = "tidepool_divergence_vote_recast_undelivered" + MetricDivergenceUnknownRefused = "tidepool_divergence_delivery_unknown_refused" + MetricDivergenceUnknownUnanswered = "tidepool_divergence_delivery_unknown_unanswered" +) + +// divergenceUnswept is what a gauge reads before any sweep has completed. +// +// It is NEGATIVE for the same reason consume's dead-letter depth is: a count +// cannot be negative, so the value is unmistakable — where a 0 would claim the +// invariant is holding at exactly the moment nobody has checked. That +// distinction is the whole point of publishing these at all. The invariant this +// class reports is expected to hold forever, so the gauge spends its life at 0; +// if "never swept" also read 0, a broken schedule would look exactly like a +// healthy bridge, and the broken schedule is the one an operator needs to see. +const divergenceUnswept = -1 + +// The gauges are package-level and published at init, so the keys exist on +// /admin/metrics from process start rather than appearing only once something +// goes wrong. They are SET from each sweep's result (never added to): the +// number reports how many divergences stand right now, so it returns to zero by +// itself when a problem is resolved. +// +// expvar.Int, deliberately NOT expvar.Func. These are multi-table joins; behind +// a Func they would run on the /admin/metrics scrape — that is, on the endpoint +// an operator reads BECAUSE the system is already struggling. +// +// ONE GAUGE PER CLASS, for the same reason there are three classes: an operator +// alerts on them separately. A stale queue is a page, a cancelled acceptance is +// a note, and a single combined number can only be tuned for whichever of them +// is noisiest. +var ( + metricPersonaVoteEvents = newDivergenceGauge(MetricDivergencePersonaVoteEvents) + metricAcceptanceCancelled = newDivergenceGauge(MetricDivergenceAcceptanceCancelled) + metricAcceptancePoisoned = newDivergenceGauge(MetricDivergenceAcceptancePoisoned) + metricAcceptanceStale = newDivergenceGauge(MetricDivergenceAcceptanceStale) + metricVoteRecast = newDivergenceGauge(MetricDivergenceVoteRecast) + metricUnknownRefused = newDivergenceGauge(MetricDivergenceUnknownRefused) + metricUnknownUnanswered = newDivergenceGauge(MetricDivergenceUnknownUnanswered) +) + +// divergenceGauges maps each class to the gauge that reports it, so publish +// cannot set one and forget another: a class added to the report without a +// gauge here fails to compile at the map literal rather than going unwatched. +var divergenceGauges = map[string]*expvar.Int{ + DivergencePersonaVoteEvent: metricPersonaVoteEvents, + DivergenceAcceptanceCancelled: metricAcceptanceCancelled, + DivergenceAcceptancePoisoned: metricAcceptancePoisoned, + DivergenceAcceptanceStale: metricAcceptanceStale, + DivergenceVoteRecastUndelivered: metricVoteRecast, + DivergenceDeliveryUnknownRefused: metricUnknownRefused, + DivergenceDeliveryUnknownUnanswered: metricUnknownUnanswered, +} + +func newDivergenceGauge(name string) *expvar.Int { + gauge := expvar.NewInt(name) + gauge.Set(divergenceUnswept) + return gauge +} + +// DivergenceOptions configures a DivergenceReconciler. +type DivergenceOptions struct { + // DB is the bridge database. BOTH sides of every comparison are local + // (decision 19): a reconciler that needed a peer's state would be a + // reconciler that talks to peers. + DB *sql.DB + // Interval is the sweep cadence for Run. Zero uses the default. + Interval time.Duration + // AcceptanceStaleAfter is how long a pending delivery may sit before its + // acceptance is reported as stale. Zero uses DefaultAcceptanceStaleAfter. + AcceptanceStaleAfter time.Duration + // Logger receives sweep outcomes. + Logger *slog.Logger + // Divergences overrides the read-only store. It is nil in production and + // constructed from DB; a test wraps it in a double that fails one read, to + // prove a failed sweep publishes NOTHING rather than a zero that reads as + // health. Same seam, same reason, as consume.Options' store overrides. + Divergences store.Divergences +} + +// DivergenceReconciler compares atproto state against outbound state and +// REPORTS what disagrees. It never writes: not to remote instances, and not to +// our own tables (PLAN.md decision 19). +// +// There is NO MUTEX, unlike FollowReconciler. That one serializes because two +// concurrent sweeps could each see a community as absent and subscribe it +// twice, and the racing EnsureCommunity calls can mint a permanent DID twice. +// Nothing here writes anything, so concurrent sweeps can only read the same rows +// and reach the same answer; a lock would buy nothing and would make the admin +// report queue behind a background pass. +type DivergenceReconciler struct { + divergences store.Divergences + interval time.Duration + staleAfter time.Duration + logger *slog.Logger +} + +// NewDivergenceReconciler validates options and builds the reconciler. +func NewDivergenceReconciler(opts DivergenceOptions) (*DivergenceReconciler, error) { + if opts.DB == nil { + return nil, errors.NewValidationError("db", "must not be nil") + } + logger := opts.Logger + if logger == nil { + logger = slog.Default() + } + interval := opts.Interval + if interval <= 0 { + interval = defaultDivergenceInterval + } + staleAfter := opts.AcceptanceStaleAfter + if staleAfter <= 0 { + staleAfter = DefaultAcceptanceStaleAfter + } + return &DivergenceReconciler{ + // The read-only store, constructed here rather than injected: this + // reconciler has exactly one dependency and it must be the one that + // cannot write (store/divergence.go). + divergences: orDefaultDivergences(opts.Divergences, store.NewDivergences(opts.DB)), + interval: interval, + staleAfter: staleAfter, + logger: logger, + }, nil +} + +// DivergenceEntry is one disagreement, named so an operator can act on it. +// +// Subject is what to look at — an at-uri, an activity id — because a class and a +// count alone give an operator a number they cannot investigate. Detail carries +// the why in words. +type DivergenceEntry struct { + Class string `json:"class"` + Subject string `json:"subject"` + Detail string `json:"detail"` +} + +// DivergenceReport is one sweep's findings: every entry, plus a count per class. +// +// Counts carries EVERY class the sweep checks, including the ones that found +// nothing. A report that listed only non-zero classes could not tell an operator +// which comparisons ran, so a class silently dropped from the sweep would read +// as a class with nothing to report. +type DivergenceReport struct { + Entries []DivergenceEntry `json:"entries"` + Counts map[string]int `json:"counts"` + // Truncated reports that Entries was capped at MaxDivergenceEntries. The + // COUNTS stay true when it is set — an operator must be able to see how big + // the problem is even when the list of examples is bounded. + Truncated bool `json:"truncated"` +} + +// MaxDivergenceEntries caps the entries one report carries. +// +// A system that is diverging badly diverges in bulk — one broken community, one +// stopped queue — and an unbounded list is then a second outage: a response +// nobody can load, on the endpoint an operator reaches for BECAUSE something is +// wrong. The counts stay exact; only the examples are bounded. +const MaxDivergenceEntries = 500 + +// addEntry records one example, while there is room for one. +// +// The cap is applied HERE rather than by truncating a finished list, because a +// list that must first be built in full is not bounded at all: the sweep that +// hits this limit is the sweep running against a database with a stopped queue +// in it, and materialising every row of that before cutting it down would put +// the memory spike in exactly the process an operator is trying to keep alive. +// +// COUNTS ARE NOT TOUCHED HERE, deliberately: every caller counts what the +// comparison returned, not what fitted. A cap that also bounded the measurement +// would cap the number an operator escalates on at the size of a page, and +// "500" would then mean both "500" and "a catastrophe". +func (report *DivergenceReport) addEntry(entry DivergenceEntry) { + if len(report.Entries) >= MaxDivergenceEntries { + // SAY SO. A silently truncated list reads as the whole problem, and an + // operator sizes an incident from what they can see. + report.Truncated = true + return + } + report.Entries = append(report.Entries, entry) +} + +// newDivergenceReport is an EMPTY report with every class present at zero — and +// non-nil throughout, so the JSON is `{"entries":[],"counts":{...}}` rather than +// nulls a client has to special-case. +func newDivergenceReport() DivergenceReport { + return DivergenceReport{ + Entries: []DivergenceEntry{}, + Counts: map[string]int{ + DivergencePersonaVoteEvent: 0, + DivergenceAcceptanceCancelled: 0, + DivergenceAcceptancePoisoned: 0, + DivergenceAcceptanceStale: 0, + DivergenceVoteRecastUndelivered: 0, + DivergenceDeliveryUnknownRefused: 0, + DivergenceDeliveryUnknownUnanswered: 0, + }, + } +} + +// Run sweeps once immediately, then on every interval tick until ctx is +// cancelled. A failed sweep is logged, never fatal: the next tick is the retry, +// and there is no state to leave half-applied because there is no state. +func (r *DivergenceReconciler) Run(ctx context.Context) { + r.logger.Info("divergence reconciler started", "interval", r.interval) + if _, err := r.Sweep(ctx); err != nil && ctx.Err() == nil { + r.logger.Error("divergence reconciler: startup sweep failed", "error", err) + } + ticker := time.NewTicker(r.interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if _, err := r.Sweep(ctx); err != nil && ctx.Err() == nil { + r.logger.Error("divergence reconciler: sweep failed", "error", err) + } + } + } +} + +// Sweep runs one comparison pass and returns what it found. Exported so the +// admin endpoint and tests can drive one synchronously, like +// FollowReconciler.Sweep. +// +// A READ FAILURE ABORTS THE WHOLE PASS. A report missing one class looks exactly +// like a report whose class found nothing, so returning the classes that +// happened to succeed would publish a claim about state nobody read — and the +// gauges would then be SET to a number that quietly excludes it. The error goes +// back instead, the gauges keep their previous values (see publish), and the +// operator sees a failed sweep rather than a clean one. +func (r *DivergenceReconciler) Sweep(ctx context.Context) (DivergenceReport, error) { + report := newDivergenceReport() + + personaVotes, err := r.divergences.PersonaVoteEvents(ctx) + if err != nil { + return DivergenceReport{}, fmt.Errorf("divergence: read persona vote events: %w", err) + } + for _, vote := range personaVotes { + report.addEntry(DivergenceEntry{ + Class: DivergencePersonaVoteEvent, + Subject: vote.ActivityID, + Detail: fmt.Sprintf( + "vote on %s is attributed to our own persona %s (%s), so that subject's tally "+ + "counts it twice: once as this inbound event and once as the delivered "+ + "outbound vote the reseed subtracts", + vote.SubjectAPID, vote.VoterAPID, vote.ActorDID), + }) + } + // The COUNT is the comparison's length, never len(Entries): see addEntry. + report.Counts[DivergencePersonaVoteEvent] = len(personaVotes) + + undelivered, err := r.divergences.UndeliveredAcceptances(ctx, r.staleAfter) + if err != nil { + return DivergenceReport{}, fmt.Errorf("divergence: read undelivered acceptances: %w", err) + } + for _, acceptance := range undelivered { + class := acceptanceClass(acceptance.DeliveryState) + if class == "" { + // A delivery state this sweep has no class for. Logged rather than + // filed under the nearest class: putting it in one would describe it + // wrongly, and dropping it silently would make an unclassifiable + // state look like a healthy one. + r.logger.Warn("divergence: undelivered acceptance in an unclassified delivery state", + "post", acceptance.PostURI, "state", acceptance.DeliveryState) + continue + } + report.addEntry(DivergenceEntry{ + Class: class, + Subject: acceptance.PostURI, + Detail: fmt.Sprintf( + "community %s accepted this post and the peer was never told: its delivery is %s", + acceptance.CommunityDID, acceptance.DeliveryState), + }) + report.Counts[class]++ + } + + recasts, err := r.divergences.RecastDivergences(ctx) + if err != nil { + return DivergenceReport{}, fmt.Errorf("divergence: read recast divergences: %w", err) + } + for _, recast := range recasts { + report.addEntry(DivergenceEntry{ + Class: DivergenceVoteRecastUndelivered, + // The SUBJECT is what an operator checks the tally of; the delivered + // activity id is what the peer is actually holding, and the only + // handle a manual Undo could embed. Both are needed: the class and + // the subject alone say a number is wrong without saying which vote + // is making it wrong. + Subject: recast.SubjectATURI, + Detail: fmt.Sprintf( + "the peer still holds vote activity %s cast by %s, which this bridge no longer "+ + "claims: the row was re-cast under a new activity id and that delivery never "+ + "landed, so nothing in the ledger accounts for the vote they are counting", + recast.DeliveredActivityID, recast.ActorDID), + }) + } + report.Counts[DivergenceVoteRecastUndelivered] = len(recasts) + + unknown, err := r.divergences.UnknownDeliveryOutcomes(ctx) + if err != nil { + return DivergenceReport{}, fmt.Errorf("divergence: read unknown delivery outcomes: %w", err) + } + for _, outcome := range unknown { + // TWO SUB-COUNTS, and the split is the only information these rows carry. + // A peer that ANSWERED told us something a silent one did not — so the + // two are never added together here, and the class an entry lands in is + // decided by the stored fact of whether an answer arrived, not by + // re-reading the status code (a 0 means nobody spoke, in both of the + // spellings the delivery table stores that in). + class := DivergenceDeliveryUnknownUnanswered + detail := fmt.Sprintf( + "we sent this %s to %s and got no answer at all (%s): it may never have arrived, or "+ + "it may have been applied and the response lost — %s is the instance to ask, "+ + "because we cannot tell from here", + outcome.Kind, outcome.TargetInbox, outcome.LastErrorClass, outcome.TargetInbox) + if outcome.Refused { + class = DivergenceDeliveryUnknownRefused + detail = fmt.Sprintf( + "we sent this %s to %s and the peer answered %d (%s): that is evidence it was not "+ + "applied and not proof of it, since a peer can apply an activity and then "+ + "fail to respond", + outcome.Kind, outcome.TargetInbox, outcome.LastStatusCode, outcome.LastErrorClass) + } + report.addEntry(DivergenceEntry{ + // The SUBJECT is the activity id: what was sent is the only handle + // that exists here. There is no post at-uri to point at — these are + // every kind of activity, and the acceptance classes are where a post + // the peer never received is reported. + Class: class, + Subject: outcome.ActivityID, + Detail: detail, + }) + report.Counts[class]++ + } + + r.publish(report) + return report, nil +} + +// acceptanceClass maps a delivery state to the class an operator triages on. +// An unknown state yields "" — the caller reports that rather than guessing, +// because a state nobody has a response for is itself worth seeing. +func acceptanceClass(state store.DeliveryState) string { + switch state { + case store.DeliveryStateCancelled: + return DivergenceAcceptanceCancelled + case store.DeliveryStatePoisoned: + return DivergenceAcceptancePoisoned + case store.DeliveryStatePending: + return DivergenceAcceptanceStale + default: + return "" + } +} + +// publish assigns the gauges from a COMPLETED sweep. +// +// It runs only on success, and that is the discipline the whole gauge design +// rests on: a failed sweep must leave the previous values standing rather than +// write a zero, because a zero written by a sweep that read nothing is a claim +// that everything is fine, made at the moment nobody could check. +func (r *DivergenceReconciler) publish(report DivergenceReport) { + for class, gauge := range divergenceGauges { + gauge.Set(int64(report.Counts[class])) + } + if len(report.Entries) > 0 { + r.logger.Warn("divergence sweep found disagreements", + "entries", len(report.Entries), "counts", report.Counts) + return + } + r.logger.Debug("divergence sweep found nothing", "counts", report.Counts) +} + +// orDefaultDivergences returns the override when a test supplies one. +func orDefaultDivergences(override, fallback store.Divergences) store.Divergences { + if override != nil { + return override + } + return fallback +} diff --git a/internal/ingest/divergence_endpoint_test.go b/internal/ingest/divergence_endpoint_test.go new file mode 100644 --- /dev/null +++ b/internal/ingest/divergence_endpoint_test.go @@ -0,0 +1,424 @@ +package ingest + +import ( + "context" + "database/sql" + stderrors "errors" + "net/http" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/store" +) + +// TASK 17e CYCLE 4 — THE HONESTY OF THE REPORT ITSELF. +// +// Everything up to here has been about finding divergences. This is about what +// the report is allowed to CLAIM, and it is where a reconciliation job most +// easily ships dishonestly: by publishing a number when it has none, by naming +// an uncertainty as a fact, or by proposing the repair that decision 19 forbids +// it from making. + +// --------------------------------------------------------------------------- +// (a) Unknown means unknown, in the class names and in the metric names +// --------------------------------------------------------------------------- + +// TestDivergenceMetricNamesForPoisonedDeliveriesSayUnknown looks trivial and is +// not. +// +// The two poisoned classes report deliveries this bridge SENT and never got +// confirmation for. We do not know whether the peer applied them: a transport +// timeout is silent about it, and even a refusal is evidence rather than proof +// — a peer can apply an activity and then fail to answer. A METRIC NAME IS A +// CLAIM, and it is the claim that survives: it ends up on a dashboard, in an +// alert rule, and in the sentence somebody says in an incident review. Named +// `_undelivered` or `_lost`, these counts assert something nobody can know, and +// the next reader who sees two adjacent numbers with confident names will add +// them into one. +// +// The precedent is deliberate: consume's deadLetterDepthUnavailable reports +// UNAVAILABLE rather than a plausible zero, for the same reason and in the same +// direction. +func TestDivergenceMetricNamesForPoisonedDeliveriesSayUnknown(t *testing.T) { + for _, name := range []string{ + MetricDivergenceUnknownRefused, + MetricDivergenceUnknownUnanswered, + DivergenceDeliveryUnknownRefused, + DivergenceDeliveryUnknownUnanswered, + } { + assert.Contains(t, name, "unknown", + "%q must say UNKNOWN. It is the only true claim about a poisoned delivery, and the "+ + "name is what an operator reads at 3am with no context", name) + assert.NotContains(t, name, "undelivered", + "%q must NOT claim non-delivery: a refusal is evidence, not proof, and a transport "+ + "failure is silent — the activity may have been applied and the response lost", name) + assert.NotContains(t, name, "lost", + "%q must NOT claim the activity was lost: that is the same assertion in a friendlier "+ + "word, and it is equally unfounded", name) + } + + assert.NotEqual(t, MetricDivergenceUnknownRefused, MetricDivergenceUnknownUnanswered, + "and the two stay SEPARATE. A peer that answered told us something a silent one did "+ + "not; one number for both throws away the only distinction available and reads as "+ + "a total of things we know") +} + +// TestDivergenceReportSeparatesRefusedFromUnansweredAndClaimsNothingElse drives +// both shapes through the sweep. +func TestDivergenceReportSeparatesRefusedFromUnansweredAndClaimsNothingElse(t *testing.T) { + h := newHarness(t) + world := newModerationWorld(t, h) + h.admin.SetDivergenceReconciler(newDivergenceReconciler(t, h)) + deliverEverythingQueued(t, h.db) + _ = world + + // Two poisoned deliveries that are NOT post acceptances, so what is under + // test is the delivery-outcome question alone. + refused := seedPoisonedComment(t, h.db, "dv-c4-refused", "4xx", 422) + unanswered := seedPoisonedComment(t, h.db, "dv-c4-unanswered", "transport", 0) + + report := fetchDivergence(t, h) + + assert.Equal(t, []string{refused}, subjectsOfClass(report, DivergenceDeliveryUnknownRefused), + "the peer ANSWERED this one — with a rejection — which is evidence of non-application "+ + "and not proof of it") + assert.Equal(t, []string{unanswered}, subjectsOfClass(report, DivergenceDeliveryUnknownUnanswered), + "and this one got no answer at all: it may never have arrived, or it may have been "+ + "applied and the response lost. Two sub-counts, because the two rows do not carry "+ + "the same amount of information") + assert.Equal(t, 1, report.Counts[DivergenceDeliveryUnknownRefused]) + assert.Equal(t, 1, report.Counts[DivergenceDeliveryUnknownUnanswered]) + + // Neither may appear in a class that asserts what the peer holds. + for _, entry := range report.Entries { + if entry.Subject != refused && entry.Subject != unanswered { + continue + } + assert.Contains(t, entry.Class, "unknown", + "an activity whose outcome is unknown may appear ONLY under a class that says so. "+ + "Counting it anywhere that claims the peer does — or does not — hold it turns "+ + "an unanswered question into a number somebody will act on") + } + + assert.Equal(t, 1, gaugeValue(t, h, MetricDivergenceUnknownRefused)) + assert.Equal(t, 1, gaugeValue(t, h, MetricDivergenceUnknownUnanswered)) +} + +// TestDivergenceReportProposesNoRepair pins the omission decision 19 requires. +// +// The reconciler is the one component that reads both sides, so it is the one +// most tempted to fix what it sees — and least able to. It cannot know whether a +// poisoned row is stale relative to newer user intent, whether a cancelled +// delivery was cancelled on purpose, or whether the peer already applied the +// activity it is about to re-send. POST /admin/outbound/redrive exists and is +// deliberately refused unscoped: a HUMAN scoping a redrive is the design, and a +// report that came with a button would make that refusal decorative. +func TestDivergenceReportProposesNoRepair(t *testing.T) { + h := newHarness(t) + world := newModerationWorld(t, h) + h.admin.SetDivergenceReconciler(newDivergenceReconciler(t, h)) + deliverEverythingQueued(t, h.db) + _ = world + seedPoisonedComment(t, h.db, "dv-c4-norepair", "transport", 0) + // A divergence that is ALREADY reported, so the body under test actually + // contains entries: an absence assertion over an empty report is true of a + // report that proposes nothing and of one that found nothing, and only the + // first is the claim being made here. + insertVoteEvent(t, h.db, "https://lemmy.world/activities/like/dv-c4-norepair", + actorIDOf(t, h.db, mtAuthorDID), "up") + + report := fetchDivergence(t, h) + require.NotEmpty(t, report.Entries, + "precondition: the report carries at least one entry, or the assertions below hold "+ + "vacuously over an empty list") + + rec := h.adminRequest(http.MethodGet, "/admin/divergence", nil) + require.Equal(t, http.StatusOK, rec.Code) + body := rec.Body.String() + + for _, forbidden := range []string{`"action"`, `"remedy"`, `"repair"`, `"fix"`, `"suggested`} { + assert.NotContains(t, body, forbidden, + "the report must carry no %s field: a reconciler that recommends is one step from a "+ + "reconciler that acts, and the recommendation would be made without the two "+ + "things only a human has — whether the state is intended, and whether the peer "+ + "already applied what we are about to re-send", forbidden) + } +} + +// seedPoisonedComment writes a poisoned delivery for an activity that is NOT a +// post acceptance, and returns its activity id. status 0 means the peer never +// answered. +func seedPoisonedComment(t *testing.T, db *sql.DB, suffix, errorClass string, status int) string { + t.Helper() + ctx := context.Background() + activityID := mtUserOrigin + "/ap/activity/" + suffix + _, err := db.ExecContext(ctx, ` + INSERT INTO outbound_activities (activity_id, actor_did, kind, payload) + VALUES ($1, $2, 'Create', '{"type":"Create","object":{"type":"Note"}}'::jsonb)`, + activityID, mtAuthorDID) + require.NoError(t, err) + + // NULL, not zero: a delivery that got no answer must not be stored as one + // answered with 0. + var statusArg any + if status != 0 { + statusArg = status + } + _, err = db.ExecContext(ctx, ` + INSERT INTO outbound_deliveries (activity_id, target_inbox, ordering_key, state, + last_error_class, last_status_code, created_at) + VALUES ($1, $2, $3, 'poisoned', $4, $5, now() - interval '2 hours')`, + activityID, "https://lemmy.world/inbox", groupID, errorClass, statusArg) + require.NoError(t, err) + return activityID +} + +// --------------------------------------------------------------------------- +// (b) The endpoint contract +// --------------------------------------------------------------------------- + +// failingDivergences fails ONE read, so a sweep that cannot see everything +// cannot publish anything. +type failingDivergences struct { + store.Divergences + mu sync.Mutex + calls int + err error +} + +func (f *failingDivergences) PersonaVoteEvents(ctx context.Context) ([]store.PersonaVoteEvent, error) { + f.mu.Lock() + f.calls++ + err := f.err + f.mu.Unlock() + if err != nil { + return nil, err + } + return f.Divergences.PersonaVoteEvents(ctx) +} + +func (f *failingDivergences) Calls() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.calls +} + +// TestDivergenceEndpoint_NotConfiguredIsNotImplemented is the nil-dependency +// pattern the rest of /admin already follows: a deployment without the sweep +// says so, rather than 404ing as if the operator mistyped the path. +func TestDivergenceEndpoint_NotConfiguredIsNotImplemented(t *testing.T) { + h := newHarness(t) + + rec := h.adminRequest(http.MethodGet, "/admin/divergence", nil) + assert.Equal(t, http.StatusNotImplemented, rec.Code, + "with no reconciler wired the endpoint reports NOT IMPLEMENTED: an operator who gets a "+ + "404 goes looking for a typo, and one who gets an empty report concludes the bridge "+ + "is healthy") +} + +// TestDivergenceEndpoint_AFailedSweepPublishesNothing is the one that matters. +// +// A sweep that cannot read one class must not publish ANY of them. Two failures +// are on the table and both are silent: +// +// - a PARTIAL REPORT is indistinguishable from a class that found nothing, so +// an operator reads "no persona votes" when the truth is "nobody looked"; +// - a ZERO WRITTEN BY A FAILED SWEEP is a claim of health made at exactly the +// moment nobody could check, and it OVERWRITES the last real number — the +// one an operator was watching. +func TestDivergenceEndpoint_AFailedSweepPublishesNothing(t *testing.T) { + h := newHarness(t) + world := newModerationWorld(t, h) + deliverEverythingQueued(t, h.db) + _ = world + + // A real divergence, so the gauges carry a NON-ZERO value a failure could + // overwrite. A fixture whose healthy value is 0 cannot tell "left standing" + // from "written as zero". + personaActorID := actorIDOf(t, h.db, mtAuthorDID) + insertVoteEvent(t, h.db, "https://lemmy.world/activities/like/dv-c4-fail", personaActorID, "up") + + h.admin.SetDivergenceReconciler(newDivergenceReconciler(t, h)) + require.Equal(t, http.StatusOK, h.adminRequest(http.MethodGet, "/admin/divergence", nil).Code) + require.Equal(t, 1, gaugeValue(t, h, MetricDivergencePersonaVoteEvents), + "precondition: a completed sweep published a real number") + + // Now the store starts failing. + faulty := &failingDivergences{ + Divergences: store.NewDivergences(h.db), + err: stderrors.New("connection reset by peer"), + } + failing, err := NewDivergenceReconciler(DivergenceOptions{DB: h.db, Divergences: faulty}) + require.NoError(t, err) + h.admin.SetDivergenceReconciler(failing) + + rec := h.adminRequest(http.MethodGet, "/admin/divergence", nil) + assert.Equal(t, http.StatusInternalServerError, rec.Code, + "a sweep that could not read every class is a FAILURE, not a report: returning the "+ + "classes that happened to succeed publishes a claim about state nobody read") + assert.NotContains(t, rec.Body.String(), `"entries"`, + "and no partial report rides out with the error — a report missing one class reads "+ + "exactly like a class that found nothing") + require.NotZero(t, faulty.Calls(), "precondition: the sweep really did try to read") + + assert.Equal(t, 1, gaugeValue(t, h, MetricDivergencePersonaVoteEvents), + "THE PREVIOUS VALUE STANDS. A failed sweep that wrote 0 would replace the last real "+ + "measurement with a claim of health, made at the moment nobody could check — and "+ + "the operator watching that gauge would see the problem disappear") +} + +// TestDivergenceReportIsBounded pins the cap and the flag that admits it. +// +// A diverging system diverges in BULK — one broken community, one stopped queue +// — so an unbounded list is a second outage: a response nobody can load, served +// from the endpoint an operator reaches for because something is already wrong. +// The counts stay exact; only the examples are bounded, and the report says so +// rather than letting a truncated list read as the whole problem. +func TestDivergenceReportIsBounded(t *testing.T) { + h := newHarness(t) + world := newModerationWorld(t, h) + h.admin.SetDivergenceReconciler(newDivergenceReconciler(t, h)) + deliverEverythingQueued(t, h.db) + _ = world + + // MaxDivergenceEntries + 1 of ONE class, so the cap is unambiguous. + personaActorID := actorIDOf(t, h.db, mtAuthorDID) + overflowPersonaVotes(t, h.db, personaActorID, MaxDivergenceEntries+1) + + report := fetchDivergence(t, h) + + assert.Len(t, report.Entries, MaxDivergenceEntries, + "the response carries at most MaxDivergenceEntries examples: a report that grows with "+ + "the outage is a report that cannot be read during one") + assert.True(t, report.Truncated, + "and it SAYS SO. A silently truncated list is worse than a long one: an operator sizes "+ + "the problem from what they can see, and %d examples that look complete argue for a "+ + "smaller incident than there is", MaxDivergenceEntries) + assert.Equal(t, MaxDivergenceEntries+1, report.Counts[DivergencePersonaVoteEvent], + "while the COUNT stays true — the cap bounds the examples, never the measurement, or "+ + "the number an operator escalates on would be capped at the size of a page") +} + +// overflowPersonaVotes writes n inbound vote events attributed to one persona. +// One statement, because 501 round trips is a slow way to say "a lot". +func overflowPersonaVotes(t *testing.T, db *sql.DB, voterAPID string, n int) { + t.Helper() + _, err := db.ExecContext(context.Background(), ` + INSERT INTO vote_events (activity_id, voter_ap_id, subject_ap_id, direction) + SELECT 'https://lemmy.world/activities/like/bulk-' || i, $1, $2, 'up' + FROM generate_series(1, $3) AS i`, voterAPID, mtPostAPID, n) + require.NoError(t, err) +} + +// --------------------------------------------------------------------------- +// (c) Run is a loop, not a one-shot that dies +// --------------------------------------------------------------------------- + +// countingDivergences signals every read, so the test can wait for real sweeps +// instead of sleeping. +type countingDivergences struct { + store.Divergences + swept chan struct{} + err error +} + +func (c *countingDivergences) PersonaVoteEvents(ctx context.Context) ([]store.PersonaVoteEvent, error) { + select { + case c.swept <- struct{}{}: + default: + } + if c.err != nil { + return nil, c.err + } + return c.Divergences.PersonaVoteEvents(ctx) +} + +// TestDivergenceRunKeepsSweepingAfterAFailure is the difference between a +// reconciler and a cron job that dies at 3am. +// +// The sweep runs unattended, against a database that will occasionally refuse a +// connection. A failure that ended the loop would leave every gauge frozen at +// its last value — which is precisely the shape of a healthy system, so nothing +// would ever page — and the next real divergence would go unreported until +// somebody restarted the process. +func TestDivergenceRunKeepsSweepingAfterAFailure(t *testing.T) { + h := newHarness(t) + + counting := &countingDivergences{ + Divergences: store.NewDivergences(h.db), + swept: make(chan struct{}, 8), + err: stderrors.New("connection reset by peer"), + } + reconciler, err := NewDivergenceReconciler(DivergenceOptions{ + DB: h.db, + Divergences: counting, + Interval: time.Millisecond, + }) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + reconciler.Run(ctx) + close(done) + }() + + // Every sweep here FAILS. THREE signals is the assertion, not two: the first + // is the startup pass, so a loop that dies on its first failing TICK still + // produces two. Only the third proves it failed, logged, and came back. + for i := 0; i < 3; i++ { + select { + case <-counting.swept: + case <-time.After(5 * time.Second): + t.Fatalf("Run stopped sweeping after %d failed sweeps: a reconciler that dies on a "+ + "transient database error leaves every gauge frozen at its last value, which "+ + "looks exactly like a healthy system and pages nobody", i) + } + } + + // And it stops when it is told to, rather than only when it breaks. + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Run must return when its context is cancelled: a sweep that outlives shutdown " + + "holds the process open and keeps reading a database that is going away") + } +} + +// TestDivergenceRunSweepsImmediately pins the startup pass. +// +// A reconciler whose first sweep waits a full interval publishes nothing at all +// for that interval — and the gauges read their unswept sentinel, so a fresh +// process looks like a broken one for as long as the cadence says. +func TestDivergenceRunSweepsImmediately(t *testing.T) { + h := newHarness(t) + + counting := &countingDivergences{ + Divergences: store.NewDivergences(h.db), + swept: make(chan struct{}, 4), + } + reconciler, err := NewDivergenceReconciler(DivergenceOptions{ + DB: h.db, + Divergences: counting, + Interval: time.Hour, // long enough that only the startup pass can fire + }) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go reconciler.Run(ctx) + + select { + case <-counting.swept: + case <-time.After(5 * time.Second): + t.Fatal("Run must sweep once IMMEDIATELY, before the first tick: otherwise a restarted " + + "process publishes its unswept sentinel for a whole interval, and an operator " + + "watching the gauges cannot tell a fresh start from a stalled sweep") + } +} diff --git a/internal/ingest/divergence_test.go b/internal/ingest/divergence_test.go new file mode 100644 --- /dev/null +++ b/internal/ingest/divergence_test.go @@ -0,0 +1,676 @@ +package ingest + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TASK 17e — THE RECONCILIATION JOB, FROM OUTSIDE. +// +// Decision 19 asks for a sweep that compares atproto state against outbound +// state and reports divergence through metrics and an admin report — and NEVER +// self-heals. That prohibition is the whole design, not a caveat on it: +// +// - A reconciler that WRITES has, by construction, no witness. It is the only +// thing that reads both sides, so when it is wrong there is nothing left to +// notice; a bad rule silently rewrites the state it was measuring, and the +// next sweep agrees with itself. +// - Half of what it finds is genuinely UNKNOWABLE from here. A poisoned +// delivery may or may not have reached the peer; a held settlement is a +// delivery that succeeded and looks pending. "Repairing" those means acting +// on a guess, at instances that do not un-delete. +// - And the actions available to it are the irreversible ones — re-sending, +// cancelling, deleting on a peer. An operator reading a report can decide; +// a loop cannot. +// +// So the outer contract has two halves, and the second is the load-bearing one: +// the report NAMES the divergence, and the sweep CHANGES NOTHING. +// +// THE FIXTURE is the first divergence class, produced through the real paths: +// a native post is admitted into a bridged community — the acceptance record is +// written into the community repo, so Coves shows the post in that community — +// and then the author opts out, which cancels the queued Create before it +// leaves. Lemmy therefore never hears of a post that the community's own repo +// says is in it. Nothing about that state is wrong to have; it is exactly what +// the two tiers of 17d promise. What is wrong is for nobody to be able to SEE +// it, which is what this sweep is for. + +const ( + dvPostRKey = "3lzdvpost00001" + dvPostRev = "3lzdvrev000001" + dvOptOut = "3lzdvrev000002" + + // dvAcceptanceUndelivered is the report class for THIS fixture's divergence: + // an acceptance whose delivery was CANCELLED. The classes are keyed on the + // delivery state — cancelled / poisoned / stale — because the three need + // three different operator responses; see the classification test below. + // This one is the cancelled sibling because a cancellation is what the + // opt-out in this fixture produces. + dvAcceptanceUndelivered = "acceptance-undelivered-cancelled" + + // dvGauge is the sweep-set gauge for that class. The tidepool_ prefix is not + // cosmetic: scopedMetrics (follow.go) filters /admin/metrics to keys with it, + // so a gauge named anything else is invisible to the operator in exactly the + // way a drop site that never fires is invisible. + dvGauge = "tidepool_divergence_acceptance_undelivered_cancelled" +) + +// dvTables is every table the sweep could conceivably touch, plus the repo +// tables underneath them. It is deliberately wider than the reconciler's +// inputs: the assertion is not "it did not write the rows I expected it to +// leave alone", it is "it did not write". +var dvTables = []string{ + // The outbound queue and ledger — the side an over-eager "fix" would + // re-drive, cancel or re-flip. + "outbound_deliveries", "outbound_activities", "outbound_votes", "outbound_objects", + // The atproto side. A reconciler that "restored" an acceptance would commit + // to a community repo, which lands here as new blocks, a moved head and a + // firehose frame — the loudest possible write, and the easiest to make by + // accident when the report has already computed exactly what is missing. + "blocks", "repo_state", "firehose_events", + // The bridge's own records of what it decided. + "admissions", "ap_objects", "ap_actors", "federation_prefs", + "community_bans", "object_moderation", "vote_events", "vote_aggregates", +} + +// TestDivergenceReportNamesAnUndeliveredAcceptanceAndWritesNothing is the OUTER +// acceptance test for 17e. +func TestDivergenceReportNamesAnUndeliveredAcceptanceAndWritesNothing(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + h.admin.SetDivergenceReconciler(newDivergenceReconciler(t, h)) + + // --- GIVEN: everything the world already queued really did reach the + // community. This is the DISCRIMINATION half of the fixture: the world + // opens with an accepted post of its own, so without this the run holds + // two undelivered acceptances and a sweep that simply listed every + // acceptance would be indistinguishable from one that compares. + deliverEverythingQueued(t, h.db) + + // --- AND: a second post accepted into community A, whose delivery is then + // cancelled by the author's own opt-out. + admitPost(t, world, mtAuthorDID, dvPostRKey, world.communityADID, dvPostRev, 1_775_000_060_000_001) + postURI := "at://" + mtAuthorDID + "/social.coves.community.postv2/" + dvPostRKey + require.Equal(t, []string{"pending"}, deliveryStatesForPost(t, h.db, postURI), + "precondition: the acceptance enqueued a Create that has not gone out yet") + + require.NoError(t, world.dispatcher.HandleEvent(ctx, + odFederationEvent(t, mtAuthorDID, dvOptOut, "create", false, false, 1_775_000_061_000_001))) + require.Equal(t, []string{"cancelled"}, deliveryStatesForPost(t, h.db, postURI), + "precondition: and the opt-out cancelled it, so the peer will never receive this post") + require.Equal(t, []string{"delivered"}, deliveryStatesForPost(t, h.db, mtPostATURI), + "precondition: while the world's OTHER accepted post is delivered and stays delivered "+ + "— the two differ in delivery state and in nothing else") + + acceptanceStands(t, h, world.communityADID, postURI) + + // --- The state of the world, in full, immediately before the sweep. + before := snapshotTables(t, h.db, dvTables...) + + // --- WHEN: the operator asks what diverges. + rec := h.adminRequest(http.MethodGet, "/admin/divergence", nil) + require.Equal(t, http.StatusOK, rec.Code, + "GET /admin/divergence must serve the report (body: %s)", rec.Body.String()) + + var report divergenceReport + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &report), + "the report is the operator's whole interface to this sweep, so it is JSON with a "+ + "stable shape (body: %s)", rec.Body.String()) + + // --- THEN: it names THIS post, under the class that says what is wrong. + require.NotEmpty(t, report.Entries, + "a report with no entries is the same output this sweep produces when everything is "+ + "healthy — and this world is not healthy: a community repo says it accepted a post "+ + "that no peer has ever been told about") + var matched int + for _, entry := range report.Entries { + if entry.Class == dvAcceptanceUndelivered && entry.Subject == postURI { + matched++ + } + } + assert.Equal(t, 1, matched, + "the report must name %s under class %q, exactly once. Naming the class without the "+ + "SUBJECT gives an operator a number they cannot act on, and the only action "+ + "available to a human here — re-drive it, or accept that the post is Coves-only — "+ + "needs to know which post. Entries were: %+v", postURI, dvAcceptanceUndelivered, report.Entries) + assert.Equal(t, 1, report.Counts[dvAcceptanceUndelivered], + "and the report's own count agrees with its entries") + + // ...and NOT the post the community actually received. This is what makes + // the class a comparison rather than a listing: the two posts differ in + // nothing an acceptance-side query can see — same author, same community, + // same repo, adjacent rkeys — and only the delivery state tells them apart. + for _, entry := range report.Entries { + assert.NotEqual(t, mtPostATURI, entry.Subject, + "a DELIVERED acceptance is not a divergence: the community repo says the post is "+ + "in the community and the peer has it, which is the healthy state. Reporting it "+ + "turns the report into a list of every post the bridge has ever accepted, and an "+ + "operator who cannot tell the finding from the background stops reading it") + } + + // --- AND: the gauge an operator watches reads 1, through the surface they + // actually watch it through. + assert.Equal(t, 1, gaugeValue(t, h, dvGauge), + "%s must read 1 on /admin/metrics. It is a SWEEP-SET gauge, not a counter: it reports "+ + "how many divergences stand RIGHT NOW, so it must be assigned from each sweep's "+ + "result — a counter that only ever adds would show a growing number for one "+ + "unchanging problem, and would never return to zero when it is resolved", dvGauge) + + // --- AND, THE POINT: the sweep changed NOTHING. + after := snapshotTables(t, h.db, dvTables...) + for _, table := range dvTables { + assert.Equal(t, before[table], after[table], + "REPORTING IS THE WHOLE JOB: %s must be byte-identical across the sweep. A "+ + "reconciler that writes has no witness — it is the only thing that reads both "+ + "sides, so a wrong rule silently rewrites the state it was measuring and every "+ + "later sweep agrees with itself. And the writes available here are the "+ + "irreversible ones: re-driving a cancelled delivery publishes content the user "+ + "withdrew, deleting an acceptance takes a post out of a community on a guess, "+ + "and re-committing one writes to a repo whose author never asked. An operator "+ + "reading this report can decide between those; a loop cannot", table) + } +} + +// TASK 17e CYCLE 1, AT THE SWEEP — the standing persona-vote invariant. +// +// The join and its two lethal-confusion controls live at the store, beside the +// table name. What only this layer can pin is the OPERATOR CONTRACT: the +// invariant is reported as a named class with the offending row in it, and the +// gauge an operator watches is SET by every sweep — including to zero. +// +// Zero is the interesting value here, and it is why this assertion is not +// redundant with the store's. This invariant is expected to hold forever, so +// the gauge spends its whole life at 0; if a sweep that finds nothing simply +// never touches the Var, the key is ABSENT from /admin/metrics, and "healthy" +// is then indistinguishable from "the sweep never ran" — which is exactly the +// state a broken schedule produces, and the state an operator would most want +// to notice. +const ( + // dvPersonaVoteClass is the report class for an inbound vote attributed to + // one of our own personas. + dvPersonaVoteClass = "persona-vote-event" + // dvPersonaVoteGauge is its sweep-set gauge. tidepool-prefixed or invisible. + dvPersonaVoteGauge = "tidepool_divergence_persona_vote_events" +) + +func TestDivergenceReportCountsPersonaVoteEventsAndPublishesZeroWhenHealthy(t *testing.T) { + h := newHarness(t) + world := newModerationWorld(t, h) + h.admin.SetDivergenceReconciler(newDivergenceReconciler(t, h)) + _ = world + + // --- GIVEN: genuine Lemmy voters, and none of ours. + insertVoteEvent(t, h.db, "https://lemmy.world/activities/like/dv-1", + "https://lemmy.world/u/genuine", "up") + insertVoteEvent(t, h.db, "https://lemmy.ml/activities/dislike/dv-2", + "https://lemmy.ml/u/another", "down") + + // --- THEN: the sweep publishes the invariant as HOLDING. + require.Equal(t, http.StatusOK, h.adminRequest(http.MethodGet, "/admin/divergence", nil).Code, + "GET /admin/divergence must serve the report") + assert.Equal(t, 0, gaugeValue(t, h, dvPersonaVoteGauge), + "%s must be PUBLISHED AND ZERO while the invariant holds. A gauge only written when "+ + "something is found is absent exactly when everything is fine — so a healthy bridge "+ + "and a sweep that never ran look identical on /admin/metrics, and the second one is "+ + "the failure an operator most needs to see", dvPersonaVoteGauge) + for _, entry := range fetchDivergence(t, h).Entries { + assert.NotEqual(t, dvPersonaVoteClass, entry.Class, + "and no entry: every vote in this world was cast by a real person on another "+ + "instance") + } + + // --- GIVEN: one vote_events row attributed to one of OUR personas. + personaActorID := actorIDOf(t, h.db, mtAuthorDID) + echoedActivity := "https://lemmy.world/activities/like/dv-echo" + insertVoteEvent(t, h.db, echoedActivity, personaActorID, "up") + + // --- THEN: the sweep names it. + report := fetchDivergence(t, h) + var matched int + for _, entry := range report.Entries { + if entry.Class == dvPersonaVoteClass && entry.Subject == echoedActivity { + matched++ + } + } + assert.Equal(t, 1, matched, + "the report must name the offending vote_events row under class %q. This invariant is "+ + "'zero forever', which is the kind of claim that stops being true silently: the "+ + "subject's tally now counts one vote twice — once as this inbound event, once as "+ + "the delivered outbound row the 17b reseed subtracts — and no other check in the "+ + "system will ever mention it. Entries were: %+v", dvPersonaVoteClass, report.Entries) + assert.Equal(t, 1, report.Counts[dvPersonaVoteClass]) + assert.Equal(t, 1, gaugeValue(t, h, dvPersonaVoteGauge), + "and the gauge moves to 1 with it: SET from this sweep's result, so it returns to zero "+ + "by itself once the row is cleaned up rather than reporting a problem that no longer "+ + "exists") +} + +// insertVoteEvent writes one inbound vote row, as the aggregator would. +func insertVoteEvent(t *testing.T, db *sql.DB, activityID, voterAPID, direction string) { + t.Helper() + _, err := db.ExecContext(context.Background(), ` + INSERT INTO vote_events (activity_id, voter_ap_id, subject_ap_id, direction) + VALUES ($1, $2, $3, $4)`, activityID, voterAPID, mtPostAPID, direction) + require.NoError(t, err) +} + +// actorIDOf reads a persona's AP actor id from storage rather than composing it +// from the origin and the DID: the id a vote must match is the one that was +// MINTED, and a test that rebuilds the string would keep passing if minting +// changed its shape. +func actorIDOf(t *testing.T, db *sql.DB, did string) string { + t.Helper() + var actorID string + require.NoError(t, db.QueryRowContext(context.Background(), + `SELECT actor_id FROM ap_actors WHERE did = $1`, did).Scan(&actorID), + "the world must have minted a persona for %s", did) + return actorID +} + +// divergenceReport is the wire shape of GET /admin/divergence. +type divergenceReport struct { + Entries []struct { + Class string `json:"class"` + Subject string `json:"subject"` + Detail string `json:"detail"` + } `json:"entries"` + Counts map[string]int `json:"counts"` + // Truncated says the entry list was capped. It is read from the WIRE + // because that is where an operator reads it: a flag the sweep sets and the + // endpoint drops would leave a truncated list looking complete. + Truncated bool `json:"truncated"` +} + +func fetchDivergence(t *testing.T, h *harness) divergenceReport { + t.Helper() + rec := h.adminRequest(http.MethodGet, "/admin/divergence", nil) + require.Equal(t, http.StatusOK, rec.Code, + "GET /admin/divergence must serve the report (body: %s)", rec.Body.String()) + var report divergenceReport + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &report), + "the report is the operator's whole interface to this sweep (body: %s)", rec.Body.String()) + return report +} + +// newDivergenceReconciler builds the reconciler over the harness's database, +// exactly as production would: both sides of every comparison are LOCAL, so the +// database is the only dependency there is. A reconciler that needed the AP +// client would already have failed decision 19. +func newDivergenceReconciler(t *testing.T, h *harness) *DivergenceReconciler { + t.Helper() + reconciler, err := NewDivergenceReconciler(DivergenceOptions{ + DB: h.db, + Logger: slog.New(slog.NewTextHandler(h.logs, &slog.HandlerOptions{Level: slog.LevelDebug})), + }) + require.NoError(t, err) + require.NotNil(t, reconciler) + return reconciler +} + +// deliveryStatesForPost lists the delivery states of the activities that +// carry ONE post, joined through the outbound state that records which AP id a +// record federates as. Two posts by one author into one community share an +// actor and an ordering key, so the per-actor helpers cannot separate them — +// and separating them is the whole fixture. +func deliveryStatesForPost(t *testing.T, db *sql.DB, postURI string) []string { + t.Helper() + return queryStrings(t, db, ` + SELECT d.state + FROM outbound_deliveries d + JOIN outbound_activities a ON a.activity_id = d.activity_id + JOIN outbound_objects o ON position(o.ap_object_id in a.payload::text) > 0 + WHERE o.at_uri = $1 + ORDER BY d.target_inbox`, postURI) +} + +// deliverEverythingQueued marks every delivery standing in the queue as +// delivered — "the peer received all of this" — so the divergence the test +// then creates is the only one in the world. +// +// It is written as a fixture rather than by running a worker because the wire +// is not what is under test here, and it REQUIRES rows: if the world stops +// enqueueing on admission, the premise dies loudly here instead of quietly +// making the count assertion below pass for the wrong reason. +func deliverEverythingQueued(t *testing.T, db *sql.DB) { + t.Helper() + result, err := db.ExecContext(context.Background(), ` + UPDATE outbound_deliveries + SET state = 'delivered', claimed_until = NULL, last_status_code = 202, updated_at = now() + WHERE state = 'pending'`) + require.NoError(t, err) + affected, err := result.RowsAffected() + require.NoError(t, err) + require.NotZero(t, affected, + "the world must have queued something to deliver: with an empty queue this fixture "+ + "proves nothing about telling delivered acceptances from undelivered ones") +} + +// acceptanceStands asserts the community repo really does carry the acceptance +// for this post — the atproto half of the divergence. Without it, "the peer +// never got the post" would be true of a post that was never accepted either, +// which is not a divergence at all but ordinary rejection. +func acceptanceStands(t *testing.T, h *harness, communityDID, postURI string) { + t.Helper() + _, _, err := h.manager.GetRecord(context.Background(), + communityDID, "social.coves.community.acceptance", testDigestRKey(postURI)) + require.NoError(t, err, + "precondition: the community repo carries the acceptance, so Coves shows this post in "+ + "the community while Lemmy has never heard of it — that gap IS the divergence (%v)", err) +} + +// gaugeValue reads one tidepool_ gauge from /admin/metrics. +// +// It goes through the ENDPOINT rather than through expvar directly, for two +// reasons. A test that reads the Var it just watched proves the sweep can call +// Set, not that an operator can see the result — and the prefix filter in +// scopedMetrics is exactly the kind of silent drop that a direct read cannot +// catch: a gauge named "divergence_…" behaves perfectly and appears nowhere. +func gaugeValue(t *testing.T, h *harness, name string) int { + t.Helper() + rec := h.adminRequest(http.MethodGet, "/admin/metrics", nil) + require.Equal(t, http.StatusOK, rec.Code) + var metrics map[string]json.RawMessage + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &metrics), + "/admin/metrics must stay parseable JSON (body: %s)", rec.Body.String()) + raw, ok := metrics[name] + require.True(t, ok, + "%s must be published on /admin/metrics. A gauge whose name does not start with "+ + "'tidepool' is filtered out by scopedMetrics and is indistinguishable from a "+ + "divergence that never fired. Published keys: %v", name, keysOf(metrics)) + var value int + require.NoError(t, json.Unmarshal(raw, &value), "%s must be a number, got %s", name, raw) + return value +} + +func keysOf(m map[string]json.RawMessage) []string { + keys := make([]string, 0, len(m)) + for key := range m { + keys = append(keys, key) + } + return keys +} + +// snapshotTables captures every row of every named table as full-row JSON. +// +// WHOLE ROWS, not a column or a count. The write this guards against is not a +// hypothetical column-level bug: it is a well-meaning "while we are here, fix +// it" — a redrive that flips one state, a repin that moves a CID, a stamp on +// updated_at. Any of those changes some column of some row, and only a +// comparison that carries every column can say so. row_to_json also survives a +// migration adding a column, which a hand-listed projection would silently stop +// covering on the day the schema grows. +func snapshotTables(t *testing.T, db *sql.DB, tables ...string) map[string][]string { + t.Helper() + snapshot := make(map[string][]string, len(tables)) + for _, table := range tables { + rows, err := db.QueryContext(context.Background(), + `SELECT row_to_json(t)::text FROM `+table+` t ORDER BY 1`) + require.NoError(t, err, "snapshot %s", table) + var out []string + for rows.Next() { + var row string + require.NoError(t, rows.Scan(&row)) + out = append(out, row) + } + require.NoError(t, rows.Err()) + require.NoError(t, rows.Close()) + snapshot[table] = out + } + return snapshot +} + +// TASK 17e CYCLE 2, AT THE SWEEP — the three classes, and the two posts that +// must not appear in any of them. +// +// The join and its false-positive control live at the store, beside the query. +// What only this layer can pin is what an operator ends up looking at: three +// SEPARATE classes with three SEPARATE counts and three SEPARATE gauges, so the +// report can be triaged. A single "undelivered" bucket would tell an operator +// how many posts are missing from Lemmy and nothing about what to do, and the +// three answers are genuinely different work: a cancelled delivery is usually +// correct and wants no action, a poisoned one is redrivable through a surface +// that already exists, and a stale one is a queue that has stopped moving. +const ( + dvUndeliveredCancelled = "acceptance-undelivered-cancelled" + dvUndeliveredPoisoned = "acceptance-undelivered-poisoned" + dvUndeliveredStale = "acceptance-undelivered-stale" + + dvGaugeCancelled = "tidepool_divergence_acceptance_undelivered_cancelled" + dvGaugePoisoned = "tidepool_divergence_acceptance_undelivered_poisoned" + dvGaugeStale = "tidepool_divergence_acceptance_undelivered_stale" +) + +func TestDivergenceReportClassifiesUndeliveredAcceptancesByDeliveryState(t *testing.T) { + h := newHarness(t) + world := newModerationWorld(t, h) + h.admin.SetDivergenceReconciler(newDivergenceReconciler(t, h)) + + // The world's own accepted post really reached the community. + deliverEverythingQueued(t, h.db) + + // Four more posts by the same author into the same community, differing + // ONLY in what happened to their delivery. + cancelled := admitDivergencePost(t, h, world, "3lzdvcls00001", "3lzdvcls00011", 1_775_000_070_000_001) + poisoned := admitDivergencePost(t, h, world, "3lzdvcls00002", "3lzdvcls00012", 1_775_000_070_000_002) + stale := admitDivergencePost(t, h, world, "3lzdvcls00003", "3lzdvcls00013", 1_775_000_070_000_003) + // The IN-FLIGHT post: accepted moments ago, delivery still queued. This is + // the ordinary state of every post between acceptance and delivery, and it + // is the false positive that would put every healthy post in the report. + inFlight := admitDivergencePost(t, h, world, "3lzdvcls00004", "3lzdvcls00014", 1_775_000_070_000_004) + + setDeliveryState(t, h.db, cancelled, "cancelled", "", 2*time.Hour) + setDeliveryState(t, h.db, poisoned, "poisoned", "4xx", 2*time.Hour) + // The age is expressed INDEPENDENTLY of the constant — longer than the + // widest window the bounds test permits — so this fixture keeps meaning + // "unambiguously stale" whatever value GREEN chooses, and cannot be made to + // pass by moving the threshold. + setDeliveryState(t, h.db, stale, "pending", "", 25*time.Hour) + + report := fetchDivergence(t, h) + + assert.Equal(t, []string{cancelled}, subjectsOfClass(report, dvUndeliveredCancelled), + "a CANCELLED delivery is its own class: the post is on Coves and will never be on "+ + "Lemmy, and that was a decision — a ban or an opt-out — so the operator's job is "+ + "to know it happened, not to fix it") + assert.Equal(t, []string{poisoned}, subjectsOfClass(report, dvUndeliveredPoisoned), + "a POISONED delivery is its own class: this one failed and is redrivable through "+ + "POST /admin/outbound/redrive, which is a different action entirely") + assert.Equal(t, []string{stale}, subjectsOfClass(report, dvUndeliveredStale), + "and a STALE pending delivery is its own class: nothing is wrong with the post, the "+ + "QUEUE has stopped moving, and the investigation starts at the worker rather than "+ + "at the community") + + assert.Equal(t, 1, report.Counts[dvUndeliveredCancelled]) + assert.Equal(t, 1, report.Counts[dvUndeliveredPoisoned]) + assert.Equal(t, 1, report.Counts[dvUndeliveredStale], + "each class carries its own count: one number for 'undelivered' hides which of three "+ + "unrelated problems the bridge has") + + // --- The two that must appear in NO class. + for _, entry := range report.Entries { + assert.NotEqual(t, inFlight, entry.Subject, + "a post accepted moments ago whose delivery is still queued is the system WORKING. "+ + "Reporting it makes every healthy post a finding, and the report degrades into "+ + "a list of everything the bridge has ever accepted — which is the failure mode "+ + "that looks like thoroughness") + assert.NotEqual(t, mtPostATURI, entry.Subject, + "and neither is the post the peer actually received: accepted_at is stamped, which "+ + "is the one signal that says the delivery landed") + } + + // --- The gauges an operator watches, one per class. + assert.Equal(t, 1, gaugeValue(t, h, dvGaugeCancelled)) + assert.Equal(t, 1, gaugeValue(t, h, dvGaugePoisoned)) + assert.Equal(t, 1, gaugeValue(t, h, dvGaugeStale), + "three gauges, because an operator alerts on them separately: a stale queue is a page, "+ + "a cancelled acceptance is a note, and one combined number can only ever be tuned "+ + "for whichever of them is noisiest") +} + +// TestDivergenceStalenessThresholdIsANamedConstant pins the dial itself. +// +// A threshold written as a literal inside a query cannot be found by the person +// who has to change it, and cannot be reasoned about by the person deciding +// whether the report is crying wolf. The value is a policy — how long a delivery +// may sit before its absence from the peer is worth an operator's attention — +// and policies belong beside the sweep that applies them. +func TestDivergenceStalenessThresholdIsANamedConstant(t *testing.T) { + assert.Greater(t, DefaultAcceptanceStaleAfter, time.Minute, + "the window must be comfortably longer than a delivery takes, or every post in flight "+ + "is a finding") + assert.LessOrEqual(t, DefaultAcceptanceStaleAfter, 24*time.Hour, + "and short enough that a queue which stopped moving is noticed the same day: a window "+ + "measured in days is a report nobody can act on while it still matters") +} + +// admitDivergencePost admits one more post by the fixture author into community +// A and returns its at-uri. +func admitDivergencePost(t *testing.T, h *harness, world moderationWorld, rkey, rev string, timeUS int64) string { + t.Helper() + admitPost(t, world, mtAuthorDID, rkey, world.communityADID, rev, timeUS) + uri := "at://" + mtAuthorDID + "/social.coves.community.postv2/" + rkey + require.Equal(t, []string{"pending"}, deliveryStatesForPost(t, h.db, uri), + "precondition: %s was accepted and enqueued", rkey) + return uri +} + +// setDeliveryState drives one post's delivery into a terminal or aged state. +// +// The states are written directly because this test is about CLASSIFICATION, +// not about how a row reaches each state — the real cancellation path is driven +// end to end by the outer acceptance test, and a poisoned delivery costs eight +// failed HTTP attempts to reach honestly. +func setDeliveryState(t *testing.T, db *sql.DB, postURI, state, errorClass string, age time.Duration) { + t.Helper() + result, err := db.ExecContext(context.Background(), ` + UPDATE outbound_deliveries d + SET state = $2, last_error_class = $3, + created_at = now() - $4::interval, updated_at = now() - $4::interval + FROM outbound_activities a, outbound_objects o + WHERE a.activity_id = d.activity_id + AND position(o.ap_object_id in a.payload::text) > 0 + AND o.at_uri = $1`, + postURI, state, errorClass, fmt.Sprintf("%d seconds", int(age.Seconds()))) + require.NoError(t, err) + affected, err := result.RowsAffected() + require.NoError(t, err) + require.EqualValues(t, 1, affected, "exactly one delivery carries %s", postURI) +} + +// subjectsOfClass lists the subjects reported under one class. +func subjectsOfClass(report divergenceReport, class string) []string { + subjects := []string{} + for _, entry := range report.Entries { + if entry.Class == class { + subjects = append(subjects, entry.Subject) + } + } + return subjects +} + +// TASK 17e — THE RESIDUAL A LAPSED BAN LEAVES BEHIND. +// +// 17c-3 recorded it as a known residual and left it there: a temporary ban +// cancels the banned author's queued deliveries, and when the ban expires those +// cancellations do not come back. `cancelled` is terminal, and nothing re-drives +// it. So the community repo keeps acceptance records for posts the peer will +// never receive, for an author who is no longer banned — a state nobody decided +// on and nobody is told about. +// +// It needs NO NEW COMPARISON. The rows are exactly cycle 2's: an accepted +// admission, no accepted_at stamp, a cancelled delivery. A ban-specific sweep +// would be a second join over the same tables answering the same question, and +// it would drift from this one — the two would disagree the first time either +// changed, and an operator would have two numbers for one problem. What the ban +// contributes is the REASON, which belongs in the entry's detail, not in a class +// of its own. +func TestALapsedBansCancellationsAreReportedByTheExistingCancelledClass(t *testing.T) { + h := newHarness(t) + world := newModerationWorld(t, h) + h.admin.SetDivergenceReconciler(newDivergenceReconciler(t, h)) + deliverEverythingQueued(t, h.db) + + // --- GIVEN: a post accepted into community A... + banned := admitDivergencePost(t, h, world, "3lzdvban00001", "3lzdvban00011", 1_775_000_080_000_001) + + // ...and a TEMPORARY ban that cancels its delivery. Driven through the real + // announced Block, so the cancellation is the production one. + h.announceBlock(world.groupA, "https://lemmy.world/activities/block/dv-1", mtAuthorDID, groupID, + map[string]any{"expires": time.Now().Add(time.Hour).UTC().Format(time.RFC3339)}) + require.Equal(t, []string{"cancelled"}, deliveryStatesForPost(t, h.db, banned), + "precondition: the ban cancelled the queued delivery") + + // --- AND: the ban lapses. Lemmy sends NO Undo when a temporary ban expires + // (17c-3), so this is exactly how the state arrives: the row stays, its + // expiry passes, and the author is federating again. + lapseBan(t, h.db, world.communityADID, mtAuthorDID) + + report := fetchDivergence(t, h) + + // --- THEN: the post is reported, under the class it already belongs to. + assert.Equal(t, []string{banned}, subjectsOfClass(report, dvUndeliveredCancelled), + "a lapsed ban's residual IS an undelivered acceptance: the acceptance stands in the "+ + "community repo, the delivery is terminally cancelled, and the author is no longer "+ + "banned — so nothing will ever carry that post to the peer and nobody decided that") + + // --- AND: the sweep grew no ban-specific comparison. + // + // THE LIST IS EXHAUSTIVE ON PURPOSE, and it is the sweep's whole class + // vocabulary in one place: every comparison 17e performs, and nothing else. + // It fails in BOTH directions — a ban-specific class appearing here is the + // case this test was written for, and a class disappearing is a comparison + // that silently stopped running. Adding a genuinely new class means editing + // this line, deliberately, which is the point. + assert.ElementsMatch(t, + []string{ + DivergencePersonaVoteEvent, + DivergenceAcceptanceCancelled, + DivergenceAcceptancePoisoned, + DivergenceAcceptanceStale, + DivergenceVoteRecastUndelivered, + DivergenceDeliveryUnknownRefused, + DivergenceDeliveryUnknownUnanswered, + }, + classesIn(report), + "the report's CLASS SET must not have grown a BAN-SPECIFIC class: this residual is "+ + "cycle 2's rows read by cycle 2's query, and a ban-specific class would mean a "+ + "second join over the same tables answering the same question. Two queries for one "+ + "condition drift the first time either is touched, and then an operator has two "+ + "numbers and no way to tell which is stale") +} + +// lapseBan moves a standing ban's expiry into the past — the way time does. +// Lemmy sends no Undo when a temporary ban runs out, so there is no activity to +// deliver here and nothing else to change: the ban row simply stops applying. +func lapseBan(t *testing.T, db *sql.DB, communityDID, subjectDID string) { + t.Helper() + result, err := db.ExecContext(context.Background(), ` + UPDATE community_bans SET expires_at = now() - interval '1 minute' + WHERE community_did = $1 AND subject_did = $2`, communityDID, subjectDID) + require.NoError(t, err) + affected, err := result.RowsAffected() + require.NoError(t, err) + require.EqualValues(t, 1, affected, + "there must BE a ban to lapse: without one this test would be asserting about an "+ + "ordinary cancellation and would say nothing about bans at all") +} + +// classesIn lists the classes the report accounts for — every key of Counts, +// including the ones that found nothing. +func classesIn(report divergenceReport) []string { + classes := make([]string, 0, len(report.Counts)) + for class := range report.Counts { + classes = append(classes, class) + } + return classes +} diff --git a/internal/ingest/follow.go b/internal/ingest/follow.go --- a/internal/ingest/follow.go +++ b/internal/ingest/follow.go @@ -69,6 +69,7 @@ // POST /admin/communities/backfill {"community":"!tech@lemmy.world"} // POST /admin/communities/reconcile (follow list configured only) // POST /admin/reemit {"did":"did:plc:..."} (or {} for all) // POST /admin/objects/sweep-deleted {"ap_ids":["https://..."]} +// GET /admin/divergence (reconciliation report; READ-ONLY) // GET /admin/metrics (tidepool's own expvar counters) // // All endpoints require "Authorization: Bearer $ADMIN_TOKEN". @@ -87,12 +88,20 @@ // reconciler serves POST /admin/communities/reconcile; nil (the // endpoint answers 501) unless a follow list is configured. Set once // during startup via SetFollowReconciler, before the server listens. reconciler *FollowReconciler + // divergence serves GET /admin/divergence; nil (the endpoint answers + // 501) unless the reconciliation sweep is wired. Set once during + // startup via SetDivergenceReconciler, before the server listens. + divergence *DivergenceReconciler } // SetFollowReconciler wires the optional follow-list reconciler in after // construction (the reconciler itself needs the Admin's subscribe cores, so // it is necessarily built second). func (a *Admin) SetFollowReconciler(r *FollowReconciler) { a.reconciler = r } + +// SetDivergenceReconciler wires the optional reconciliation sweep in after +// construction, as SetFollowReconciler does for the follow list. +func (a *Admin) SetDivergenceReconciler(r *DivergenceReconciler) { a.divergence = r } // NewAdmin validates options and builds the Admin API. func NewAdmin(opts AdminOptions) (*Admin, error) { @@ -145,6 +154,7 @@ r.Post("/communities/backfill", a.handleBackfill) r.Post("/communities/reconcile", a.handleReconcile) r.Post("/reemit", a.handleReemit) r.Post("/objects/sweep-deleted", a.handleSweepDeleted) + r.Get("/divergence", a.handleDivergence) r.Get("/outbound", a.handleOutboundInspect) r.Post("/outbound/redrive", a.handleOutboundRedrive) r.Post("/outbound/cancel", a.handleOutboundCancel) @@ -529,6 +539,31 @@ // handleReconcile runs one synchronous follow-list sweep on demand, so an // operator can converge right after editing the file instead of waiting for // the next tick. 501 when no follow list is configured (the handleBackfill // nil-dependency pattern). +// handleDivergence serves the reconciliation report (task 17e, decision 19). +// +// GET, not POST, and that is a statement rather than a convention: this sweep +// compares local state against local state and CHANGES NOTHING, so it is safe +// to repeat, safe to cache-bust, and safe for an operator to hit while they are +// still working out what is wrong. The moment it needed POST it would have +// stopped being a report. +// +// A sweep failure is a 500 carrying the reason: a partial report would be a lie +// about the classes it never reached, and this endpoint is operator-facing. +func (a *Admin) handleDivergence(w http.ResponseWriter, r *http.Request) { + if a.divergence == nil { + http.Error(w, "divergence reconciliation is not configured", http.StatusNotImplemented) + return + } + report, err := a.divergence.Sweep(r.Context()) + if err != nil { + a.logger.Error("divergence sweep failed", "error", err) + http.Error(w, "divergence sweep failed: "+err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json; charset=utf-8") + _ = json.NewEncoder(w).Encode(report) +} + func (a *Admin) handleReconcile(w http.ResponseWriter, r *http.Request) { if a.reconciler == nil { http.Error(w, "follow list reconciliation is not configured", http.StatusNotImplemented) diff --git a/internal/store/divergence.go b/internal/store/divergence.go new file mode 100644 --- /dev/null +++ b/internal/store/divergence.go @@ -0,0 +1,440 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "time" +) + +// The reconciliation sweep's reads (task 17e, decision 19). +// +// READ-ONLY BY CONSTRUCTION, and that is the point of giving the sweep its own +// store file rather than adding methods to the existing repositories: decision +// 19 forbids self-healing writes, and a reconciler whose store cannot write is +// a reconciler that cannot heal by accident. Nothing in this file may ever grow +// an Exec — there is none below, and adding one would make this the only +// component that both reads a divergence and can act on it, which is exactly +// the shape decision 19 forbids. + +// PersonaVoteEvent is one INBOUND vote_events row whose voter resolves to one +// of our own native personas — a vote the bridge cast on a user's behalf, +// arriving back as if a stranger had cast it. +// +// Expected count: ZERO, forever. Task 17a's probe refuses such a row before it +// is written, so one existing means the probe was bypassed and the subject's +// tally is double-counting our own echo: once as an inbound event, once as the +// delivered outbound vote the 17b reseed subtracts. +type PersonaVoteEvent struct { + // ActivityID is the inbound activity the row was written under. + ActivityID string + // VoterAPID is the AP actor id the vote was attributed to. + VoterAPID string + // SubjectAPID is the object voted on — what an operator needs to know is + // mis-counted. + SubjectAPID string + // ActorDID is the persona that id belongs to. + ActorDID string +} + +// UndeliveredAcceptance is a post the community's own repo says it accepted — +// Coves renders it in that community — whose federation never reached the peer. +// +// The two sides are the acceptance ledger (admissions, status accepted) and +// outbound_objects.accepted_at, which is stamped ONLY by delivery success and is +// therefore the honest "did it land" signal. DeliveryState carries WHY, because +// the three reasons need different operator responses: a cancelled delivery was +// a decision (a ban, an opt-out) and is usually correct to leave alone; a +// poisoned one is a delivery that failed and may be redrivable; a pending one +// stuck past the staleness window is a queue that is not moving. +type UndeliveredAcceptance struct { + // PostURI is the postv2 at-uri in the author's repo — the subject an + // operator investigates. + PostURI string + // CommunityDID is the bridged community whose repo carries the acceptance. + CommunityDID string + // ActivityID is the outbound activity that was to carry it, "" when the + // acceptance never enqueued one. + ActivityID string + // DeliveryState is the state of that activity's delivery, and is what the + // sweep classifies on. + DeliveryState DeliveryState + // LastErrorClass is the delivery's outcome class, which is how a delivery + // HELD FOR SETTLEMENT is told from a stuck one. + LastErrorClass string +} + +// Divergences reads the comparisons the reconciliation sweep reports on. Every +// method is a READ; both sides of every comparison are local. +type Divergences interface { + // PersonaVoteEvents lists inbound vote events attributed to our own + // personas, joining vote_events.voter_ap_id to ap_actors.actor_id. + PersonaVoteEvents(ctx context.Context) ([]PersonaVoteEvent, error) + + // UndeliveredAcceptances lists accepted posts whose federation never + // reached the peer. A pending delivery counts only once it is older than + // staleAfter: until then it is in flight, which is the ordinary state of + // every post between acceptance and delivery. + UndeliveredAcceptances(ctx context.Context, staleAfter time.Duration) ([]UndeliveredAcceptance, error) + + // RecastDivergences lists (actor, subject) pairs where a peer is holding a + // vote this bridge no longer claims — see RecastDivergence. + RecastDivergences(ctx context.Context) ([]RecastDivergence, error) + + // UnknownDeliveryOutcomes lists poisoned deliveries: activities we SENT and + // never got confirmation for — see UnknownDeliveryOutcome. + UnknownDeliveryOutcomes(ctx context.Context) ([]UnknownDeliveryOutcome, error) +} + +// UnknownDeliveryOutcome is a delivery whose result this bridge DOES NOT KNOW. +// +// A poisoned delivery is one we sent and never got confirmation for, and the +// two ways that happens are not equally informative: +// +// REFUSED — the peer answered with a status code. That is EVIDENCE OF +// non-application, not proof: a peer can apply an activity and +// then fail to respond, and several implementations do exactly +// that under load. +// UNANSWERED — a transport failure with no status at all. Silent about +// everything: the request may never have arrived, or may have +// been applied and the response lost. +// +// Neither is a fact about what the peer holds, which is why this is its own +// read rather than a column on the undelivered-acceptance comparison. The +// separation is kept all the way to the metric names: the only true claim about +// these rows is that we do not know. +type UnknownDeliveryOutcome struct { + // ActivityID is what was sent. + ActivityID string + // TargetInbox is who it was sent to — the instance an operator would have + // to ask, since we cannot. + TargetInbox string + // Kind is the activity kind, so an operator can tell a lost vote from a + // lost post without a second query. + Kind string + // LastErrorClass is the delivery's recorded failure class. + LastErrorClass string + // LastStatusCode is the status the peer answered with, or 0 when it never + // answered. + LastStatusCode int + // Refused reports whether a status code came back at all. It is the whole + // distinction between the two sub-counts, and it is stored rather than + // derived so that "0" cannot be read as "the peer said 0". + Refused bool +} + +// RecastDivergence is a vote a peer HOLDS that our own state does not claim. +// +// It is produced by the re-cast race 17b recorded and deferred: re-casting a +// delivered vote re-upserts the SAME row back to pending under a new activity +// id, while the peer still holds the old vote in the old direction. Transient +// while the new delivery is in flight — and PERMANENT the moment it poisons. +// +// IT CANNOT BE READ FROM THE VOTE ROW, which is what makes it a reconciliation +// item rather than a query. worker.voteCallback resolves its row through +// GetByActivityID and returns nil on NotFound, so when a delivery that was +// already in flight lands AFTER a re-cast, its id no longer matches +// current_activity_id and the settlement silently no-ops. The row is precisely +// the evidence the bug erases. outbound_activities is append-only and its +// parent_at_uri carries the subject at-uri from both vote enqueue sites, so the +// activity/delivery history is the durable record of what each peer was +// actually told. +type RecastDivergence struct { + // ActorDID is the persona whose vote it is. + ActorDID string + // SubjectATURI is the thing voted on — the pair (actor, subject) is the + // identity of a vote, since only one may be live at a time. + SubjectATURI string + // DeliveredActivityID is the vote activity the PEER ACCEPTED: what they are + // counting right now, and the only thing an operator can reconcile against. + DeliveredActivityID string +} + +type postgresDivergences struct{ db *sql.DB } + +// NewDivergences creates the postgres-backed read-only divergence reader. +func NewDivergences(db *sql.DB) Divergences { return &postgresDivergences{db: db} } + +func (r *postgresDivergences) PersonaVoteEvents(ctx context.Context) ([]PersonaVoteEvent, error) { + // THE JOIN IS ON IDENTITY, NEVER ON ACTIVITY IDS. Decision 16 originally + // proposed matching inbound votes against outbound_activities by activity + // id; the 17a plan review overturned it on measured behaviour — Lemmy 0.19 + // reconstructs Announce{Undo{Like}} with a FRESHLY GENERATED inner activity + // id and types it "Like" even when the live vote is a Dislike. An id-keyed + // probe therefore matches none of the echoes it exists to find, reports + // zero, and is indistinguishable from a healthy bridge. The voter's identity + // is the only stable handle. + // + // ap_actors — NEVER bridged_actors. ap_actors holds OUR native personas, the + // Coves users this bridge federates FOR. bridged_actors holds REAL LEMMY + // HUMANS mirrored INTO atproto, whose DIDs we minted and whose votes are the + // entire inbound vote stream. Probing bridged_actors here would report every + // genuine vote on the network as our own echo — every community's tally + // reading as double-counted — and the two tables are confusable precisely + // because both carry DIDs we minted and AP ids we can spell. + // + // Exact string equality, deliberately narrow: migration 023's one-time + // cleanup used the same equality, so this reports exactly the population + // that cleanup would have removed. A zero therefore means "no + // exactly-spelled persona vote", not "no persona vote" — see the KNOWNNARROW + // case in divergence_test.go, which pins that limit rather than papering it. + query := ` + SELECT v.activity_id, v.voter_ap_id, v.subject_ap_id, a.did + FROM vote_events v + JOIN ap_actors a ON a.actor_id = v.voter_ap_id + ORDER BY v.activity_id` + + rows, err := r.db.QueryContext(ctx, query) + if err != nil { + return nil, fmt.Errorf("list persona vote events: %w", err) + } + defer func() { _ = rows.Close() }() + + // Non-nil, so a caller rendering this straight to JSON emits [] rather than + // null: an operator reading "null" cannot tell an empty result from a field + // the sweep never filled. + found := make([]PersonaVoteEvent, 0) + for rows.Next() { + var event PersonaVoteEvent + if err := rows.Scan(&event.ActivityID, &event.VoterAPID, &event.SubjectAPID, &event.ActorDID); err != nil { + return nil, fmt.Errorf("scan persona vote event: %w", err) + } + found = append(found, event) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("list persona vote events: %w", err) + } + return found, nil +} + +func (r *postgresDivergences) UndeliveredAcceptances(ctx context.Context, staleAfter time.Duration) ([]UndeliveredAcceptance, error) { + // BOTH SIDES ARE LOCAL, and the second one is the whole comparison. + // + // admissions.status = 'accepted' — the community's repo carries an + // acceptance record, so Coves renders the post in that community. + // outbound_objects.accepted_at IS NULL — the peer never confirmed it. That + // column is stamped ONLY by delivery success (worker.stampAccepted), so + // it is the one signal in the schema that means "it landed" rather than + // "we tried". + // + // The activity join is the SAME correspondence the worker uses to stamp + // that column: a Create/Update payload names the object it federates in + // object.id, and that id is outbound_objects.ap_object_id. Read as a JSONB + // path rather than a substring search, so an id that merely appears + // somewhere in another activity's payload cannot masquerade as this one's. + // + // LATEST DELIVERY FIRST, then classify. Picking the newest attempt (by seq) + // and asking what state IT is in describes the situation now; filtering + // first and taking the newest survivor would report a post whose older + // attempt was cancelled while a fresh one is still in flight. + // + // KNOWN GAP, stated rather than silently swallowed: an acceptance with NO + // delivery row at all (the LEFT JOINs yield NULL) is not reported, because + // the three classes here are all delivery states and it has none. It is a + // real divergence — nothing will ever carry that post — and it needs its own + // class rather than being folded into one that would misdescribe it. + query := ` + WITH latest AS ( + SELECT DISTINCT ON (o.at_uri) + o.at_uri AS post_uri, + o.community_did AS community_did, + d.activity_id AS activity_id, + d.state AS delivery_state, + d.last_error_class AS last_error_class, + d.created_at AS delivery_created_at + FROM admissions adm + JOIN outbound_objects o ON o.at_uri = adm.post_uri + LEFT JOIN outbound_activities a + ON a.payload -> 'object' ->> 'id' = o.ap_object_id + LEFT JOIN outbound_deliveries d ON d.activity_id = a.activity_id + WHERE adm.status = 'accepted' + AND o.accepted_at IS NULL + ORDER BY o.at_uri, d.seq DESC NULLS LAST + ) + SELECT post_uri, + community_did, + COALESCE(activity_id, ''), + COALESCE(delivery_state, ''), + COALESCE(last_error_class, '') + FROM latest + WHERE delivery_state IN ($1, $2) + OR (delivery_state = $3 + AND last_error_class <> $4 + AND delivery_created_at < now() - $5::interval) + ORDER BY post_uri` + + // A HELD SETTLEMENT IS NOT A DIVERGENCE, and it is the one case this report + // would otherwise cry wolf on. It is pending, old, and unstamped — identical + // to a stuck delivery on every column above except this one — and it means + // the opposite: the peer ALREADY ACCEPTED the activity, and accepted_at is + // missing precisely because writing it is the step that failed. The worker + // resumes it on its next claim. Reporting it fires the sweep on every + // settlement retry, and an operator who learns to ignore this report also + // ignores the cancelled acceptance beside it, which is the finding that + // never heals itself. + rows, err := r.db.QueryContext(ctx, query, + string(DeliveryStateCancelled), string(DeliveryStatePoisoned), + string(DeliveryStatePending), DeliveryHeldForSettlement, + fmt.Sprintf("%d seconds", int(staleAfter.Seconds()))) + if err != nil { + return nil, fmt.Errorf("list undelivered acceptances: %w", err) + } + defer func() { _ = rows.Close() }() + + found := make([]UndeliveredAcceptance, 0) + for rows.Next() { + var entry UndeliveredAcceptance + var state string + if err := rows.Scan(&entry.PostURI, &entry.CommunityDID, &entry.ActivityID, + &state, &entry.LastErrorClass); err != nil { + return nil, fmt.Errorf("scan undelivered acceptance: %w", err) + } + entry.DeliveryState = DeliveryState(state) + found = append(found, entry) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("list undelivered acceptances: %w", err) + } + return found, nil +} + +func (r *postgresDivergences) RecastDivergences(ctx context.Context) ([]RecastDivergence, error) { + // THE EVIDENCE IS THE HISTORY, NEVER THE VOTE ROW. outbound_activities is + // append-only and its parent_at_uri carries the subject at-uri from both + // vote enqueue sites, so a Like/Dislike with a DELIVERED delivery is durable + // proof of what a peer was told — which is exactly what the vote row stops + // being the moment a re-cast rewrites it. + // + // The row appears below only as an EXCLUSION, and the direction matters: it + // can suppress a finding, never create one. If our ledger still names this + // exact activity as the live vote AND still calls it delivered, then we + // account for what the peer holds and there is nothing to reconcile. When + // the row has been reset, retracted or deleted — every shape this bug takes + // — the row cannot answer, and the history stands on its own. + // + // TWO INDEPENDENT EXCLUSIONS, because they answer different questions and + // each is the whole defence against a different way of ruining this report: + // + // the ledger still claims it — without this, EVERY cleanly delivered vote + // is a finding. That is most of the highest-volume table in the system, + // and a class that names all of it hides the one case that matters. + // a delivered Undo followed it — without this, a withdrawal that WORKED is + // reported forever. The history is append-only, so the delivered + // Like/Dislike never goes away; only the Undo beside it says the peer + // holds nothing now. + // + // The Undo comparison is by time rather than by id on purpose: an Undo names + // the activity it withdraws in its payload, but a re-cast mints new ids, so + // id-chasing would miss an Undo that withdrew an earlier incarnation of the + // same (actor, subject) vote. Only one vote per pair may be live at a time, + // which is what makes "any delivered Undo at or after this delivery" the + // right question. + // + // DISTINCT because one activity may have several deliveries (the fan-out + // schema); one delivered copy is one thing the peer holds. + query := ` + SELECT DISTINCT a.actor_did, a.parent_at_uri, a.activity_id + FROM outbound_activities a + JOIN outbound_deliveries d + ON d.activity_id = a.activity_id AND d.state = $1 + WHERE a.kind IN ($2, $3) + AND a.parent_at_uri <> '' + AND NOT EXISTS ( + SELECT 1 + FROM outbound_votes v + WHERE v.actor_did = a.actor_did + AND v.subject_at_uri = a.parent_at_uri + AND v.current_activity_id = a.activity_id + AND v.delivered_state = $4) + AND NOT EXISTS ( + SELECT 1 + FROM outbound_activities u + JOIN outbound_deliveries ud + ON ud.activity_id = u.activity_id AND ud.state = $1 + WHERE u.kind = $5 + AND u.actor_did = a.actor_did + AND u.parent_at_uri = a.parent_at_uri + AND u.created_at >= a.created_at) + ORDER BY a.actor_did, a.parent_at_uri, a.activity_id` + + rows, err := r.db.QueryContext(ctx, query, + string(DeliveryStateDelivered), "Like", "Dislike", + string(DeliveredStateDelivered), "Undo") + if err != nil { + return nil, fmt.Errorf("list recast divergences: %w", err) + } + defer func() { _ = rows.Close() }() + + found := make([]RecastDivergence, 0) + for rows.Next() { + var entry RecastDivergence + if err := rows.Scan(&entry.ActorDID, &entry.SubjectATURI, &entry.DeliveredActivityID); err != nil { + return nil, fmt.Errorf("scan recast divergence: %w", err) + } + found = append(found, entry) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("list recast divergences: %w", err) + } + return found, nil +} + +func (r *postgresDivergences) UnknownDeliveryOutcomes(ctx context.Context) ([]UnknownDeliveryOutcome, error) { + // POISONED ONLY, and every other state is excluded for a reason that is not + // symmetry: + // + // delivered — the peer confirmed it. The one outcome here we DO know. + // cancelled — it never reached the wire: a ban or an opt-out took it out of + // the queue, so the peer does not have it and that is a fact, not a + // question. Cycle 2's cancelled class already names it. Folding it in + // would put a KNOWN non-delivery into the bucket whose entire meaning is + // that the answer is unavailable, and these are the two numbers an + // operator has to be able to trust as small. + // pending — still in flight, or held for settlement. Nothing is unresolved + // about a delivery that has not finished trying. + // + // AGE IS NOT PART OF THE DEFINITION, unlike the stale-acceptance class: a + // poisoned delivery is terminal the moment it poisons — nothing re-drives it + // — so a recent one and an old one are the same state and the same unknown. + // + // REFUSED IS COMPUTED FROM WHETHER A PEER ANSWERED, NOT FROM THE NUMBER, and + // this is the one subtle thing in the query. The column is nullable, but the + // worker does not use the NULL: classify() passes a literal 0 for a transport + // failure and for an unresolvable signer (worker.go), and MarkPoisoned writes + // that 0. So "the peer never answered" arrives in this table in TWO + // spellings, NULL and 0, and only a predicate that treats both as silence + // keeps a dial timeout out of the sub-count that says the peer spoke. It is + // selected as a boolean rather than left to the caller to infer, so that no + // reader downstream can rediscover the wrong rule from LastStatusCode. + query := ` + SELECT d.activity_id, + d.target_inbox, + a.kind, + d.last_error_class, + COALESCE(d.last_status_code, 0), + COALESCE(d.last_status_code, 0) > 0 + FROM outbound_deliveries d + JOIN outbound_activities a ON a.activity_id = d.activity_id + WHERE d.state = $1 + ORDER BY d.activity_id, d.target_inbox` + + rows, err := r.db.QueryContext(ctx, query, string(DeliveryStatePoisoned)) + if err != nil { + return nil, fmt.Errorf("list unknown delivery outcomes: %w", err) + } + defer func() { _ = rows.Close() }() + + found := make([]UnknownDeliveryOutcome, 0) + for rows.Next() { + var entry UnknownDeliveryOutcome + if err := rows.Scan(&entry.ActivityID, &entry.TargetInbox, &entry.Kind, + &entry.LastErrorClass, &entry.LastStatusCode, &entry.Refused); err != nil { + return nil, fmt.Errorf("scan unknown delivery outcome: %w", err) + } + found = append(found, entry) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("list unknown delivery outcomes: %w", err) + } + return found, nil +} diff --git a/internal/store/divergence_acceptance_test.go b/internal/store/divergence_acceptance_test.go new file mode 100644 --- /dev/null +++ b/internal/store/divergence_acceptance_test.go @@ -0,0 +1,361 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/testutil" +) + +// TASK 17e CYCLE 2 — AN ACCEPTANCE THAT NEVER REACHED THE PEER. +// +// This is the divergence a Coves user can SEE. The community's own repo carries +// an acceptance record, so Coves renders the post in that community; Lemmy was +// never told. Both sides are local — the acceptance ledger, and +// outbound_objects.accepted_at, which is stamped ONLY by delivery success and is +// therefore the one honest "did it land" signal in the schema. +// +// THE CLASSIFICATION IS THE POINT, because the three reasons need three +// different operator responses, and lumping them makes the report unusable: +// +// cancelled — a DECISION took it out of the queue (a ban, an opt-out). Usually +// correct to leave exactly as it is; the report exists so the +// resulting Coves-only post is visible rather than silent. +// poisoned — a delivery that FAILED. Redrivable, and the operator surface for +// that already exists (POST /admin/outbound/redrive). +// stale — still pending long after it should have gone. Nothing is wrong +// with the post; the QUEUE is not moving, which is a different +// investigation entirely. +// +// AND THE FALSE POSITIVE IS THE WHOLE TEST. A query that reported every +// acceptance would satisfy every "is it found?" assertion here and be worse than +// no report at all: an operator who cannot tell the finding from the background +// stops reading it. So every case below carries a delivered sibling that must be +// absent, and the siblings differ from the findings in delivery state and in +// nothing else. + +const ( + dvCommunityDID = "did:plc:dvcommunity00000001" + dvCommunityAPID = "https://lemmy.world/c/divergence" + dvCommunityInbox = "https://lemmy.world/c/divergence/inbox" + dvAuthorDID = "did:plc:dvauthor000000000001" + dvUserOrigin = "https://coves.social" + + // dvStaleAfter is the window the tests pass explicitly. The production + // default is a named constant beside the sweep; what is pinned here is the + // BOUNDARY behaviour, at one minute either side of whatever window it is. + dvStaleAfter = time.Hour +) + +// dvPost describes one accepted post to seed, and the delivery it is waiting on. +type dvPost struct { + rkey string + // state / errorClass / age describe the delivery. age is how long ago the + // delivery row was created. + state DeliveryState + errorClass string + age time.Duration + // accepted stamps outbound_objects.accepted_at — "the peer took it". + accepted bool + // noDelivery omits the delivery row entirely. + noDelivery bool +} + +func (p dvPost) uri() string { + return "at://" + dvAuthorDID + "/social.coves.community.postv2/" + p.rkey +} + +func (p dvPost) apObjectID() string { + return dvUserOrigin + "/ap/object/" + dvAuthorDID + "/social.coves.community.postv2/" + p.rkey +} + +func (p dvPost) activityID() string { + return dvUserOrigin + "/ap/activity/" + p.rkey +} + +// seedAcceptedPost writes the whole local world for one accepted post: the +// acceptance ledger row, the outbound object it federates as, and (unless the +// case says otherwise) the activity and delivery carrying it. +// +// It is written through raw SQL on purpose. The engine that produces this state +// lives two packages up, and the sweep must read the state as it IS on disk — +// including states no current writer produces, which is exactly what a +// divergence report is for. +func seedAcceptedPost(t *testing.T, database *sql.DB, post dvPost) { + t.Helper() + ctx := context.Background() + + acceptedAt := "NULL" + if post.accepted { + acceptedAt = "now()" + } + _, err := database.ExecContext(ctx, fmt.Sprintf(` + INSERT INTO outbound_objects (at_uri, ap_object_id, community_did, community_ap_id, + translated_snapshot, accepted_at) + VALUES ($1, $2, $3, $4, '{"type":"Page"}'::jsonb, %s)`, acceptedAt), + post.uri(), post.apObjectID(), dvCommunityDID, dvCommunityAPID) + require.NoError(t, err, "seed outbound_objects for %s", post.rkey) + + _, err = database.ExecContext(ctx, ` + INSERT INTO admissions (community_did, post_uri, author_did, status, decision_code) + VALUES ($1, $2, $3, 'accepted', '')`, + dvCommunityDID, post.uri(), dvAuthorDID) + require.NoError(t, err, "seed admissions for %s", post.rkey) + + if post.noDelivery { + return + } + + // The payload carries the object id, exactly as the translator writes it — + // it is the same correspondence the delivery worker uses to stamp + // accepted_at when the Create succeeds. + _, err = database.ExecContext(ctx, ` + INSERT INTO outbound_activities (activity_id, actor_did, kind, payload) + VALUES ($1, $2, 'Create', $3::jsonb)`, + post.activityID(), dvAuthorDID, + fmt.Sprintf(`{"id":%q,"type":"Create","actor":%q,"object":{"type":"Page","id":%q}}`, + post.activityID(), dvUserOrigin+"/ap/actor/"+dvAuthorDID, post.apObjectID())) + require.NoError(t, err, "seed outbound_activities for %s", post.rkey) + + _, err = database.ExecContext(ctx, ` + INSERT INTO outbound_deliveries (activity_id, target_inbox, ordering_key, state, + last_error_class, created_at, updated_at) + VALUES ($1, $2, $3, $4, $5, now() - $6::interval, now() - $6::interval)`, + post.activityID(), dvCommunityInbox, dvCommunityAPID, string(post.state), post.errorClass, + fmt.Sprintf("%d seconds", int(post.age.Seconds()))) + require.NoError(t, err, "seed outbound_deliveries for %s", post.rkey) +} + +// acceptanceTestDB truncates every table this comparison reads. It starts from +// the cycle-1 list rather than a second one of its own: this sweep READS every +// table any other test writes, so a divergence fixture is the worst possible +// place for two truncate lists to drift apart. +func acceptanceTestDB(t *testing.T) *sql.DB { + t.Helper() + database := divergenceTestDB(t) + testutil.Truncate(t, database, "admissions", "outbound_objects") + return database +} + +// urisOf lists the reported post uris, so assertions read as sets. +func urisOf(found []UndeliveredAcceptance) []string { + uris := make([]string, 0, len(found)) + for _, entry := range found { + uris = append(uris, entry.PostURI) + } + return uris +} + +// --------------------------------------------------------------------------- +// The three classes, and the sibling that must not appear +// --------------------------------------------------------------------------- + +// TestUndeliveredAcceptances_ClassifiesByDeliveryState is the cycle's contract. +func TestUndeliveredAcceptances_ClassifiesByDeliveryState(t *testing.T) { + database := acceptanceTestDB(t) + + cancelled := dvPost{rkey: "3lzdvacc00001", state: DeliveryStateCancelled, age: 2 * time.Hour} + poisoned := dvPost{rkey: "3lzdvacc00002", state: DeliveryStatePoisoned, errorClass: "4xx", age: 2 * time.Hour} + stale := dvPost{rkey: "3lzdvacc00003", state: DeliveryStatePending, age: 2 * time.Hour} + // THE FALSE-POSITIVE CONTROL. Same author, same community, same repo, + // adjacent rkey — it differs from the three above in delivery state and in + // nothing else that any acceptance-side query can see. + landed := dvPost{rkey: "3lzdvacc00004", state: DeliveryStateDelivered, age: 2 * time.Hour, accepted: true} + + for _, post := range []dvPost{cancelled, poisoned, stale, landed} { + seedAcceptedPost(t, database, post) + } + + found, err := NewDivergences(database).UndeliveredAcceptances(context.Background(), dvStaleAfter) + require.NoError(t, err) + + assert.ElementsMatch(t, []string{cancelled.uri(), poisoned.uri(), stale.uri()}, urisOf(found), + "exactly the three undelivered acceptances, and NOT the one that landed. A query that "+ + "reported every acceptance would pass every 'is it found' assertion in this file "+ + "while being worse than no report: an operator who cannot tell the finding from the "+ + "background stops reading it, and the one real divergence goes out with the noise") + + byURI := map[string]UndeliveredAcceptance{} + for _, entry := range found { + byURI[entry.PostURI] = entry + } + require.Len(t, byURI, 3, "one row per post") + + assert.Equal(t, DeliveryStateCancelled, byURI[cancelled.uri()].DeliveryState, + "the delivery state rides along, because the three reasons are three different "+ + "investigations: a cancelled delivery was a DECISION (a ban, an opt-out) and is "+ + "usually correct to leave alone") + assert.Equal(t, DeliveryStatePoisoned, byURI[poisoned.uri()].DeliveryState, + "a poisoned one FAILED and is redrivable through an operator surface that already exists") + assert.Equal(t, DeliveryStatePending, byURI[stale.uri()].DeliveryState, + "and a stale pending one says nothing is wrong with the POST — the queue is not moving, "+ + "which is a different problem with a different fix") + + for _, entry := range found { + assert.Equal(t, dvCommunityDID, entry.CommunityDID, + "every entry names the community whose repo carries the acceptance: that repo is "+ + "where an operator looks, and one post can be accepted into more than one") + } +} + +// --------------------------------------------------------------------------- +// The staleness boundary +// --------------------------------------------------------------------------- + +// TestUndeliveredAcceptances_PendingIsStaleOnlyPastTheThreshold pins the EDGE +// rather than a value deep inside the window. +// +// A pending delivery is the ordinary state of every post between acceptance and +// delivery, so the threshold is the only thing separating "in flight" from +// "stuck". Tested at one minute either side, a threshold that drifts — or one +// applied to the wrong column, or with the comparison inverted — fails here +// instead of quietly reporting every post the bridge accepts. +func TestUndeliveredAcceptances_PendingIsStaleOnlyPastTheThreshold(t *testing.T) { + database := acceptanceTestDB(t) + + justInside := dvPost{rkey: "3lzdvage00001", state: DeliveryStatePending, age: dvStaleAfter - time.Minute} + justOutside := dvPost{rkey: "3lzdvage00002", state: DeliveryStatePending, age: dvStaleAfter + time.Minute} + seedAcceptedPost(t, database, justInside) + seedAcceptedPost(t, database, justOutside) + + found, err := NewDivergences(database).UndeliveredAcceptances(context.Background(), dvStaleAfter) + require.NoError(t, err) + + assert.Equal(t, []string{justOutside.uri()}, urisOf(found), + "only the delivery PAST the window is a divergence. A post accepted a minute ago and "+ + "not yet delivered is the system working — reporting it makes every healthy post a "+ + "finding, and the report becomes a list of everything the bridge has ever accepted") + + // And the window is a real dial, not a constant baked into the SQL: widening + // it past both ages must empty the result. + found, err = NewDivergences(database).UndeliveredAcceptances(context.Background(), 3*time.Hour) + require.NoError(t, err) + assert.Empty(t, found, + "with a wider window neither delivery is stale yet: the threshold has to be the "+ + "parameter, or an operator tuning it changes nothing") +} + +// --------------------------------------------------------------------------- +// The one that looks exactly like the problem and is the opposite +// --------------------------------------------------------------------------- + +// TestUndeliveredAcceptances_AHeldSettlementIsNotADivergence is the case this +// report would otherwise cry wolf on. +// +// A delivery HELD FOR SETTLEMENT is pending, old, and carries an unstamped +// accepted_at — pixel-identical to a stuck delivery on every column this query +// reads. It is the opposite: the peer ALREADY ACCEPTED the activity, and the +// only thing outstanding is our own bookkeeping, which the worker resumes on its +// next claim. The stamp is missing precisely BECAUSE the settlement that writes +// it is the step that failed. +// +// Counting it means the report fires on every settlement retry — a routine, +// self-healing event — and a report that cries wolf is worse than no report: +// the operator who learns to ignore it also ignores the cancelled acceptance +// sitting next to it, which is the finding that never self-heals. +func TestUndeliveredAcceptances_AHeldSettlementIsNotADivergence(t *testing.T) { + database := acceptanceTestDB(t) + + held := dvPost{ + rkey: "3lzdvheld00001", + state: DeliveryStatePending, + errorClass: DeliveryHeldForSettlement, + age: 2 * time.Hour, + } + // A genuinely stuck delivery beside it, so "found nothing" cannot pass by + // the query being broken: the two differ ONLY in last_error_class. + stuck := dvPost{rkey: "3lzdvheld00002", state: DeliveryStatePending, age: 2 * time.Hour} + seedAcceptedPost(t, database, held) + seedAcceptedPost(t, database, stuck) + + found, err := NewDivergences(database).UndeliveredAcceptances(context.Background(), dvStaleAfter) + require.NoError(t, err) + + assert.Equal(t, []string{stuck.uri()}, urisOf(found), + "the held delivery is NOT a divergence and the stuck one is. They are identical on "+ + "every column this query reads except last_error_class = %q, and that one column "+ + "inverts the meaning: held means the peer ALREADY ACCEPTED it and our own "+ + "bookkeeping is outstanding — the accepted_at stamp is missing precisely because "+ + "writing it is the step that failed. Reporting it fires on every settlement retry, "+ + "and an operator who learns to ignore this report also ignores the cancelled "+ + "acceptance beside it, which is the one that never heals itself", + DeliveryHeldForSettlement) +} + +// --------------------------------------------------------------------------- +// The OTHER exclusion, pinned on its own +// --------------------------------------------------------------------------- + +// TestUndeliveredAcceptances_AnEditThatFailedIsNotAPostThatNeverLanded pins +// `accepted_at IS NULL` as an INDEPENDENTLY NECESSARY term. +// +// Every other case in this file is excluded twice over — the delivered sibling +// has both a stamped accepted_at and a delivered delivery — so either term +// alone keeps the whole suite green and neither is actually pinned. This is the +// state that separates them, and it is ordinary: a post that federated fine, and +// an EDIT that did not. +// +// The classification reads the LATEST delivery, which here is the cancelled +// Update. On the delivery state alone this post is "an acceptance that never +// reached the peer" — and that is false in the way that matters most to an +// operator: the peer HAS this post. Only the edit is missing. Reporting it +// sends someone to investigate a community view that is correct, and worse, it +// is the most common shape in the report once editing sees real use, so the +// class fills up with posts that are on Lemmy right now. +func TestUndeliveredAcceptances_AnEditThatFailedIsNotAPostThatNeverLanded(t *testing.T) { + database := acceptanceTestDB(t) + + // The post itself: Create delivered, and the stamp to prove the peer took it. + edited := dvPost{rkey: "3lzdvedit00001", state: DeliveryStateDelivered, age: 3 * time.Hour, accepted: true} + seedAcceptedPost(t, database, edited) + // The edit, later and cancelled — the newest delivery for this object. + seedFollowUpDelivery(t, database, edited, "Update", DeliveryStateCancelled, "", time.Hour) + + // A post that genuinely never landed, so "found nothing" cannot pass by the + // query being broken. The two differ ONLY in accepted_at. + never := dvPost{rkey: "3lzdvedit00002", state: DeliveryStateCancelled, age: 3 * time.Hour} + seedAcceptedPost(t, database, never) + + found, err := NewDivergences(database).UndeliveredAcceptances(context.Background(), dvStaleAfter) + require.NoError(t, err) + + assert.Equal(t, []string{never.uri()}, urisOf(found), + "a post whose CREATE landed is not an undelivered acceptance, however its later edits "+ + "fared: accepted_at is stamped, the peer has the post, and the community view is "+ + "correct. The delivery-state predicate cannot see that difference — the newest "+ + "delivery here is cancelled either way — so accepted_at IS NULL is doing this work "+ + "alone. Without it the class fills with posts that are on Lemmy right now, and the "+ + "one post that never arrived is lost among them") +} + +// seedFollowUpDelivery adds a LATER activity and delivery for a post that +// already has one — an edit, a re-delivery — so the fixture can express a post +// whose history has more than one step. +func seedFollowUpDelivery(t *testing.T, database *sql.DB, post dvPost, kind string, + state DeliveryState, errorClass string, age time.Duration) { + t.Helper() + ctx := context.Background() + activityID := post.activityID() + "-" + kind + + _, err := database.ExecContext(ctx, ` + INSERT INTO outbound_activities (activity_id, actor_did, kind, payload) + VALUES ($1, $2, $3, $4::jsonb)`, + activityID, dvAuthorDID, kind, + fmt.Sprintf(`{"id":%q,"type":%q,"actor":%q,"object":{"type":"Page","id":%q}}`, + activityID, kind, dvUserOrigin+"/ap/actor/"+dvAuthorDID, post.apObjectID())) + require.NoError(t, err, "seed follow-up %s for %s", kind, post.rkey) + + _, err = database.ExecContext(ctx, ` + INSERT INTO outbound_deliveries (activity_id, target_inbox, ordering_key, state, + last_error_class, created_at, updated_at) + VALUES ($1, $2, $3, $4, $5, now() - $6::interval, now() - $6::interval)`, + activityID, dvCommunityInbox, dvCommunityAPID, string(state), errorClass, + fmt.Sprintf("%d seconds", int(age.Seconds()))) + require.NoError(t, err, "seed follow-up delivery for %s", post.rkey) +} diff --git a/internal/store/divergence_test.go b/internal/store/divergence_test.go new file mode 100644 --- /dev/null +++ b/internal/store/divergence_test.go @@ -0,0 +1,236 @@ +package store + +import ( + "context" + "database/sql" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/testutil" +) + +// TASK 17e CYCLE 1 — THE STANDING PERSONA-VOTE INVARIANT, AT THE JOIN. +// +// The invariant is one sentence: a vote this bridge cast on a native user's +// behalf must never appear in vote_events. It is where our own echo would land +// if the 17a voter probe were ever bypassed, and the cost is exact and +// permanent — the subject's served tally counts that one vote twice, once as a +// live inbound event and once as the delivered outbound row 17b's reseed +// subtracts from the origin's total. +// +// Expected count: ZERO, forever. A reconciler is the right shape for it because +// zero-forever is precisely the kind of claim that quietly stops being true: +// nothing else re-checks it, and the damage is a number nobody can trace back. +// +// THE TABLE NAME IS THE WHOLE TEST. This file exists here, beside the query, +// because the lethal mistake is one word: +// +// ap_actors — OUR native personas (Coves users we federate FOR) +// bridged_actors — REAL LEMMY HUMANS mirrored INTO atproto (we hold their DIDs) +// +// Both tables hold DIDs we minted and AP ids we can spell, so the confusion is +// easy and the assertion "the probe finds our own actors" cannot tell them +// apart. Probing bridged_actors matches EVERY genuine inbound vote — every vote +// on the network arrives from a Lemmy human, and every Lemmy human who has ever +// voted has a bridged_actors row — so the reconciler would report every real +// vote as our own echo, and any later code trusting that signal takes the vote +// pipeline dark: every community's tally goes to zero. + +const ( + dvPersonaDID = "did:plc:dvpersona0000000001" + dvPersonaActorID = "https://coves.social/ap/actor/" + dvPersonaDID + dvSubjectAPID = "https://lemmy.world/post/9001" + + // A real Lemmy human, mirrored into atproto. Their AP id is a LEMMY url and + // their DID is one we minted — the exact shape that makes the two tables + // confusable. + dvHumanAPID = "https://lemmy.world/u/genuine" + dvHumanDID = "did:plc:dvbridgedhuman00001" +) + +func divergenceTestDB(t *testing.T) *sql.DB { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, + "vote_events", "vote_aggregates", "ap_actors", "bridged_actors", + "outbound_activities", "outbound_deliveries", "outbound_votes") + return database +} + +// seedPersona writes one native persona: the identity the bridge federates FOR. +func seedPersona(t *testing.T, database *sql.DB, did, actorID string) { + t.Helper() + _, err := database.ExecContext(context.Background(), ` + INSERT INTO ap_actors (did, kind, actor_id, normalized_origin, local_part, + rsa_key_sealed, rsa_key_version, public_key_pem) + VALUES ($1, 'person', $2, 'coves.social', $3, '\x00'::bytea, 1, 'pem')`, + did, actorID, "p"+did[len(did)-6:]) + require.NoError(t, err) +} + +// seedBridgedHuman writes one real Lemmy human mirrored into atproto. +func seedBridgedHuman(t *testing.T, database *sql.DB, apID, did string) { + t.Helper() + _, err := database.ExecContext(context.Background(), ` + INSERT INTO bridged_actors (ap_actor_id, actor_type, did) + VALUES ($1, 'person', $2)`, apID, did) + require.NoError(t, err) +} + +// seedVoteEvent writes one inbound vote exactly as the aggregator would. +func seedVoteEvent(t *testing.T, database *sql.DB, activityID, voterAPID, direction string) { + t.Helper() + _, err := database.ExecContext(context.Background(), ` + INSERT INTO vote_events (activity_id, voter_ap_id, subject_ap_id, direction) + VALUES ($1, $2, $3, $4)`, activityID, voterAPID, dvSubjectAPID, direction) + require.NoError(t, err) +} + +// --------------------------------------------------------------------------- +// The invariant +// --------------------------------------------------------------------------- + +// TestPersonaVoteEvents_GenuineVotersAreNotOurOwn is the healthy world: real +// Lemmy people voting on a bridged post, and a persona that has never had an +// inbound vote attributed to it. +func TestPersonaVoteEvents_GenuineVotersAreNotOurOwn(t *testing.T) { + database := divergenceTestDB(t) + seedPersona(t, database, dvPersonaDID, dvPersonaActorID) + seedBridgedHuman(t, database, dvHumanAPID, dvHumanDID) + seedVoteEvent(t, database, "https://lemmy.world/activities/like/1", dvHumanAPID, "up") + seedVoteEvent(t, database, "https://lemmy.world/activities/dislike/2", "https://lemmy.ml/u/other", "down") + + found, err := NewDivergences(database).PersonaVoteEvents(context.Background()) + require.NoError(t, err) + assert.Empty(t, found, + "the ordinary state of the world is ZERO: every vote here was cast by a real person on "+ + "another instance, and reporting any of them would make the healthy case "+ + "indistinguishable from the broken one") +} + +// TestPersonaVoteEvents_APersonasOwnVoteIsFound is the divergence. +// +// THE FIXTURE IS BUILT SO AN ID-KEYED PROBE CANNOT PASS IT. Decision 16 +// originally proposed matching inbound votes against outbound_activities by +// ACTIVITY ID; the 17a plan review overturned it on measured behaviour — +// Lemmy 0.19 reconstructs Announce{Undo{Like}} with a FRESHLY GENERATED inner +// activity id, and types it "Like" even when the live vote is a Dislike. So the +// row below carries an id that appears in no outbound_activities row, while the +// persona's real Like sits there under a different one. An implementation that +// probed ids finds nothing, reports zero, and looks like it is working. +func TestPersonaVoteEvents_APersonasOwnVoteIsFound(t *testing.T) { + database := divergenceTestDB(t) + ctx := context.Background() + seedPersona(t, database, dvPersonaDID, dvPersonaActorID) + seedBridgedHuman(t, database, dvHumanAPID, dvHumanDID) + + // The activity the bridge really sent for this persona: a DISLIKE, under an + // id of our own minting. + ourActivityID := "https://coves.social/ap/activity/" + repeatHex('d') + _, err := database.ExecContext(ctx, ` + INSERT INTO outbound_activities (activity_id, actor_did, kind, payload) + VALUES ($1, $2, 'Dislike', '{"type":"Dislike"}'::jsonb)`, ourActivityID, dvPersonaDID) + require.NoError(t, err) + + // What came back: our own persona's vote, echoed in with an id Lemmy + // generated and a direction that does not match what we sent. + echoedID := "https://lemmy.world/activities/like/reconstructed-99" + seedVoteEvent(t, database, echoedID, dvPersonaActorID, "up") + // ...beside a genuine vote, so "found something" is not the same as "found + // everything". + seedVoteEvent(t, database, "https://lemmy.world/activities/like/3", dvHumanAPID, "up") + + found, err := NewDivergences(database).PersonaVoteEvents(ctx) + require.NoError(t, err) + require.Len(t, found, 1, + "exactly the persona's row is a divergence: the join is voter_ap_id = "+ + "ap_actors.actor_id, an IDENTITY match. An activity-id probe against "+ + "outbound_activities misses this row — Lemmy reconstructs the echo with a fresh id "+ + "(and the wrong type) — while still finding nothing to report, which is the failure "+ + "mode that looks exactly like success") + assert.Equal(t, echoedID, found[0].ActivityID) + assert.Equal(t, dvPersonaActorID, found[0].VoterAPID) + assert.Equal(t, dvSubjectAPID, found[0].SubjectAPID, + "the SUBJECT rides along: it is the tally that is wrong, and an operator cannot check "+ + "a count they cannot name") + assert.Equal(t, dvPersonaDID, found[0].ActorDID, + "and the persona it belongs to, which is where the investigation starts") +} + +// --------------------------------------------------------------------------- +// CONTROL 1 — the table name +// --------------------------------------------------------------------------- + +// TestPersonaVoteEvents_ABridgedHumansVoteIsNeverOurOwn is the control for the +// one-word mistake. +// +// The fixture makes the swap MEASURABLE rather than hypothetical: a real Lemmy +// human with a bridged_actors row, voting, exactly as the whole network does all +// day. Under the correct table this is invisible. Under bridged_actors it is a +// reported divergence — and so is every other vote the bridge has ever received. +func TestPersonaVoteEvents_ABridgedHumansVoteIsNeverOurOwn(t *testing.T) { + database := divergenceTestDB(t) + seedPersona(t, database, dvPersonaDID, dvPersonaActorID) + seedBridgedHuman(t, database, dvHumanAPID, dvHumanDID) + + // Three genuine humans, one of them mirrored, all voting normally. + seedVoteEvent(t, database, "https://lemmy.world/activities/like/10", dvHumanAPID, "up") + seedVoteEvent(t, database, "https://lemmy.world/activities/like/11", "https://lemmy.world/u/second", "up") + seedVoteEvent(t, database, "https://lemmy.ml/activities/dislike/12", "https://lemmy.ml/u/third", "down") + + found, err := NewDivergences(database).PersonaVoteEvents(context.Background()) + require.NoError(t, err) + assert.Empty(t, found, + "a BRIDGED ACTOR is a real Lemmy human we mirrored INTO atproto — not one of our "+ + "personas — and their votes are the entire inbound vote stream. Probing "+ + "bridged_actors instead of ap_actors reports every genuine vote on the network as "+ + "our own echo: every community's tally reads as double-counted, and anything that "+ + "acts on this signal takes the vote pipeline dark. The two tables are confusable "+ + "because both hold DIDs we minted, and no assertion about 'finding our actors' can "+ + "tell them apart — only this one can") +} + +// --------------------------------------------------------------------------- +// The known-narrow edge, documented rather than claimed +// --------------------------------------------------------------------------- + +// TestPersonaVoteEvents_ANonCanonicalSpellingIsNotMatched_KNOWNNARROW records a +// limitation of the invariant, and it deliberately asserts what the code DOES +// rather than what the invariant's name promises. +// +// The probe that PREVENTS these rows (echo.identifyActor) normalizes: it +// compares a NORMALIZED HOST plus scheme against ap_actors.normalized_origin, so +// "https://Coves.social:443/ap/actor/{did}" is ours to the probe. The join here +// is exact string equality, and so was migration 023's one-time cleanup — which +// means a legacy row spelled non-canonically survived the cleanup AND is missed +// by this sweep. +// +// The consequence, stated plainly so the next reader is not misled by a green +// gauge: THIS GAUGE READING ZERO IS NOT PROOF THAT NO PERSONA VOTE EXISTS. It +// proves no persona vote exists SPELLED EXACTLY AS THE PERSONA'S actor_id. +// Production is empty of such rows (write-back has never been deployed and no +// code path here writes a non-canonical voter id), which is why the narrow form +// is accepted rather than fixed; closing it means re-implementing the +// normalization in SQL, and that is a followup, not a silent assumption. +func TestPersonaVoteEvents_ANonCanonicalSpellingIsNotMatched_KNOWNNARROW(t *testing.T) { + database := divergenceTestDB(t) + seedPersona(t, database, dvPersonaDID, dvPersonaActorID) + + // The same identity, spelled the way a legacy row might carry it: explicit + // default port and mixed case. echo.identifyActor calls this ours. + seedVoteEvent(t, database, + "https://lemmy.world/activities/like/legacy-1", + "https://Coves.social:443/ap/actor/"+dvPersonaDID, "up") + + found, err := NewDivergences(database).PersonaVoteEvents(context.Background()) + require.NoError(t, err) + assert.Empty(t, found, + "DOCUMENTED NARROWNESS, not a claim of correctness: the join is exact equality, so a "+ + "non-canonically spelled persona id is invisible to it — and was equally invisible "+ + "to migration 023's cleanup, which used the same exact equality. A zero here "+ + "therefore means 'no exactly-spelled persona vote', NOT 'no persona vote'. If this "+ + "test ever fails because the join learned to normalize, that is an improvement: "+ + "delete this test and say so in the report") +} diff --git a/internal/store/divergence_unknown_test.go b/internal/store/divergence_unknown_test.go new file mode 100644 --- /dev/null +++ b/internal/store/divergence_unknown_test.go @@ -0,0 +1,239 @@ +package store + +import ( + "context" + "database/sql" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TASK 17e CYCLE 4 — WHAT WE DO NOT KNOW, KEPT AS NOT KNOWN. +// +// A poisoned delivery is one this bridge SENT and never got confirmation for. +// Whether the peer applied it is genuinely unknowable from here, and the two +// ways a delivery poisons are not equally informative: +// +// REFUSED — the peer answered with a status code. Evidence of +// non-application; NOT proof. A peer can apply an activity and +// then fail to respond, which is ordinary under load. +// UNANSWERED — a transport failure, no status at all. Silent about +// everything: it may never have arrived, or it may have been +// applied and the response lost. +// +// So the read keeps them apart, and keeps both out of any answer about what the +// peer holds. 17b already relies on this: a poisoned vote's row deliberately +// KEEPS its activity id and error class so the uncertainty stays queryable — +// erasing it would turn "we do not know" into "it never happened". +// +// AGE IS DELIBERATELY NOT PART OF THE DEFINITION, unlike the stale-acceptance +// class: a poisoned delivery is terminal the moment it poisons — nothing +// re-drives it on its own — so a recent one and an old one are the same state. + +const ( + dvUnknownInboxA = "https://lemmy.world/c/divergence/inbox" + dvUnknownInboxB = "https://lemmy.zip/c/elsewhere/inbox" +) + +// seedPoisonedDelivery writes one activity and its poisoned delivery. status 0 +// means the peer never answered at all. +func seedPoisonedDelivery(t *testing.T, database *sql.DB, activityID, kind, inbox, errorClass string, status int) { + t.Helper() + ctx := context.Background() + _, err := database.ExecContext(ctx, ` + INSERT INTO outbound_activities (activity_id, actor_did, kind, payload) + VALUES ($1, $2, $3, '{"type":"Create"}'::jsonb)`, activityID, dvAuthorDID, kind) + require.NoError(t, err) + + // The status is written EXACTLY as the caller says, including a literal 0 — + // see seedUnansweredAsProductionWrites below for why that matters. + statusArg := any(status) + if status < 0 { + // A negative status is this fixture's spelling of "the column is NULL", + // which production does not write but the schema permits. + statusArg = nil + } + _, err = database.ExecContext(ctx, ` + INSERT INTO outbound_deliveries (activity_id, target_inbox, ordering_key, state, + last_error_class, last_status_code, created_at) + VALUES ($1, $2, $3, 'poisoned', $4, $5, now() - interval '2 hours')`, + activityID, inbox, dvCommunityAPID, errorClass, statusArg) + require.NoError(t, err) +} + +// TestUnknownDeliveryOutcomes_SeparatesRefusedFromUnanswered is the contract. +func TestUnknownDeliveryOutcomes_SeparatesRefusedFromUnanswered(t *testing.T) { + database := acceptanceTestDB(t) + + refusedID := "https://coves.social/ap/activity/dv-unknown-refused" + unansweredID := "https://coves.social/ap/activity/dv-unknown-unanswered" + seedPoisonedDelivery(t, database, refusedID, "Create", dvUnknownInboxA, "4xx", 422) + seedPoisonedDelivery(t, database, unansweredID, "Create", dvUnknownInboxB, "transport", 0) + + // A delivery that DID land, so "found two" is not "found everything". + seedPoisonedSibling(t, database) + + found, err := NewDivergences(database).UnknownDeliveryOutcomes(context.Background()) + require.NoError(t, err) + require.Len(t, found, 2, + "both poisoned deliveries are unknown outcomes, and the delivered one is not: a "+ + "delivery that was confirmed is the one case here we DO know") + + byID := map[string]UnknownDeliveryOutcome{} + for _, outcome := range found { + byID[outcome.ActivityID] = outcome + } + + refused := byID[refusedID] + assert.True(t, refused.Refused, + "a peer that ANSWERED — even with a rejection — told us something, and that is the "+ + "only distinction available between these two rows") + assert.Equal(t, 422, refused.LastStatusCode, + "with the status it answered, because 422 and 503 send an operator to different places") + assert.Equal(t, dvUnknownInboxA, refused.TargetInbox, + "and the inbox: we cannot know whether they applied it, so the instance to ASK is the "+ + "most actionable thing this row carries") + + unanswered := byID[unansweredID] + assert.False(t, unanswered.Refused, + "a transport failure is silent: the request may never have arrived, or may have been "+ + "applied and the response lost. Reading that as 'the peer said no' invents an "+ + "answer nobody gave") + assert.Zero(t, unanswered.LastStatusCode, + "and there is no status to report — which is why Refused is stored rather than derived "+ + "from the code, so a 0 can never be read as 'the peer answered 0'") + assert.Equal(t, "transport", unanswered.LastErrorClass, + "the recorded class rides along: it is what an operator triages on and what 17b "+ + "deliberately preserved so the uncertainty stays queryable") +} + +// seedPoisonedSibling writes a DELIVERED delivery beside the poisoned ones — +// the case we genuinely know the answer to, which must never appear among the +// unknowns. +func seedPoisonedSibling(t *testing.T, database *sql.DB) { + t.Helper() + ctx := context.Background() + activityID := "https://coves.social/ap/activity/dv-unknown-delivered" + _, err := database.ExecContext(ctx, ` + INSERT INTO outbound_activities (activity_id, actor_did, kind, payload) + VALUES ($1, $2, 'Create', '{"type":"Create"}'::jsonb)`, activityID, dvAuthorDID) + require.NoError(t, err) + _, err = database.ExecContext(ctx, ` + INSERT INTO outbound_deliveries (activity_id, target_inbox, ordering_key, state, + last_status_code, delivered_at, created_at) + VALUES ($1, $2, $3, 'delivered', 202, now(), now() - interval '2 hours')`, + activityID, dvUnknownInboxA, dvCommunityAPID) + require.NoError(t, err) +} + +// TestUnknownDeliveryOutcomes_ATransportFailureIsUnansweredAsProductionWritesIt +// is the case the rest of this file could not see. +// +// PRODUCTION NEVER WRITES NULL HERE. worker.go passes a literal 0 for a +// transport failure (dial, TLS, timeout) and 0 again for an unresolvable +// signer, so "the peer never answered" reaches this table as a ZERO. Every +// other fixture in this file spelled it NULL, which made a predicate of +// `last_status_code IS NOT NULL` pass every test while filing every real dial +// timeout under REFUSED — the sub-count that asserts the peer answered. That is +// exactly the invented answer this file's header forbids, and it would have +// shipped behind green tests. +// +// The distinction has to be "did a status actually come back", and the only +// spelling of that which survives contact with the writer is a value test. +func TestUnknownDeliveryOutcomes_ATransportFailureIsUnansweredAsProductionWritesIt(t *testing.T) { + database := acceptanceTestDB(t) + + // The two shapes the worker really produces. + timeoutID := "https://coves.social/ap/activity/dv-unknown-timeout" + signerID := "https://coves.social/ap/activity/dv-unknown-signer" + seedPoisonedDelivery(t, database, timeoutID, "Create", dvUnknownInboxA, "transport", 0) + seedPoisonedDelivery(t, database, signerID, "Create", dvUnknownInboxB, "signer", 0) + // And a genuine refusal beside them, so "everything is unanswered" cannot + // pass either: the two must come apart. + refusedID := "https://coves.social/ap/activity/dv-unknown-503" + seedPoisonedDelivery(t, database, refusedID, "Create", dvUnknownInboxA, "5xx", 503) + + found, err := NewDivergences(database).UnknownDeliveryOutcomes(context.Background()) + require.NoError(t, err) + require.Len(t, found, 3) + + byID := map[string]UnknownDeliveryOutcome{} + for _, outcome := range found { + byID[outcome.ActivityID] = outcome + } + + assert.False(t, byID[timeoutID].Refused, + "a transport failure is stored with last_status_code = 0 — the literal the worker "+ + "passes, not NULL — and it means NO ANSWER CAME BACK. A predicate that tests for "+ + "NULL instead of for a real status files every dial timeout on the network under "+ + "the count that says the peer replied, which is precisely the answer nobody gave") + assert.False(t, byID[signerID].Refused, + "and so is a delivery that never reached the wire at all: an unresolvable signer is "+ + "written with the same 0, and the peer certainly did not answer a request we never "+ + "sent") + assert.True(t, byID[refusedID].Refused, + "while a real 503 IS an answer: the two must come apart on the value, or the "+ + "distinction the sub-counts exist for does not exist") + assert.Equal(t, 503, byID[refusedID].LastStatusCode) +} + +// TestUnknownDeliveryOutcomes_ANullStatusIsAlsoUnanswered keeps the defensive +// case. No writer produces a NULL here today — the column is nullable and every +// caller passes an int — so this pins the schema's remaining freedom rather than +// a live path, and it exists so that a future writer which does leave it NULL +// cannot land the row in the "the peer answered" bucket by default. +func TestUnknownDeliveryOutcomes_ANullStatusIsAlsoUnanswered(t *testing.T) { + database := acceptanceTestDB(t) + + nullID := "https://coves.social/ap/activity/dv-unknown-null" + seedPoisonedDelivery(t, database, nullID, "Create", dvUnknownInboxA, "transport", -1) + + found, err := NewDivergences(database).UnknownDeliveryOutcomes(context.Background()) + require.NoError(t, err) + require.Len(t, found, 1) + assert.False(t, found[0].Refused, + "an absent status is not an answer either: NULL and 0 must land in the same bucket, or "+ + "the meaning of the sub-count depends on which writer produced the row") + assert.Zero(t, found[0].LastStatusCode, + "and it is reported as 0 rather than as a NULL a JSON reader would see as null: the "+ + "field is a status, and there was none") +} + +// TestUnknownDeliveryOutcomes_ACancelledDeliveryIsNotUnknown separates "we do +// not know" from "it was never sent". +// +// A cancelled delivery never reached the wire — a ban or an opt-out took it out +// of the queue — so there is nothing uncertain about it: the peer does not have +// it, and cycle 2's classes say so. Folding it in here would put a KNOWN +// non-delivery into the bucket whose whole meaning is that the answer is +// unavailable, and the unknown counts are exactly the numbers an operator must +// be able to trust as small. +func TestUnknownDeliveryOutcomes_ACancelledDeliveryIsNotUnknown(t *testing.T) { + database := acceptanceTestDB(t) + ctx := context.Background() + + activityID := "https://coves.social/ap/activity/dv-unknown-cancelled" + _, err := database.ExecContext(ctx, ` + INSERT INTO outbound_activities (activity_id, actor_did, kind, payload) + VALUES ($1, $2, 'Create', '{"type":"Create"}'::jsonb)`, activityID, dvAuthorDID) + require.NoError(t, err) + _, err = database.ExecContext(ctx, ` + INSERT INTO outbound_deliveries (activity_id, target_inbox, ordering_key, state, created_at) + VALUES ($1, $2, $3, 'cancelled', now() - interval '2 hours')`, + activityID, dvUnknownInboxA, dvCommunityAPID) + require.NoError(t, err) + + // One genuinely unknown row beside it, so an empty result cannot pass for + // the right answer. + seedPoisonedDelivery(t, database, "https://coves.social/ap/activity/dv-unknown-live", + "Create", dvUnknownInboxA, "transport", 0) + + found, err := NewDivergences(database).UnknownDeliveryOutcomes(ctx) + require.NoError(t, err) + require.Len(t, found, 1, "only the poisoned delivery is an unknown outcome") + assert.Equal(t, "https://coves.social/ap/activity/dv-unknown-live", found[0].ActivityID, + "a CANCELLED delivery never reached the wire, so its outcome is known: the peer does "+ + "not have it. Counting it as unknown inflates the one number whose value is that "+ + "it is small and honest") +} diff --git a/internal/votes/divergence_recast_test.go b/internal/votes/divergence_recast_test.go new file mode 100644 --- /dev/null +++ b/internal/votes/divergence_recast_test.go @@ -0,0 +1,224 @@ +package votes + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ingest" + "tidepool/internal/store" +) + +// TASK 17e — THE RE-CAST DIVERGENCE: THE PEER HOLDS A VOTE WE DO NOT CLAIM. +// +// 17b found this and deferred it here. Re-casting a delivered vote re-upserts +// the SAME outbound_votes row back to 'pending' under a new activity id, while +// the peer goes on holding the old vote in the old direction. Transient while +// the new delivery is in flight; PERMANENT the moment it poisons — nothing +// re-drives a poisoned delivery on its own, and the reseed subtracts only +// 'delivered' rows, so the community's score keeps counting a vote we have +// stopped accounting for and will never correct. +// +// THE FIXTURE IS THE ENTIRE TEST, and it must be DRIVEN, not assembled. This is +// 17b's own blind spot by name: "the fixture nobody writes is the one where the +// row's history has more than one step." Every state below is produced by the +// real path — +// +// consume.Dispatcher.HandleEvent (vote commit) → outbound_votes + a Dislike +// outbound.Worker.DeliverNext → delivered, ledger flipped +// consume.Dispatcher.HandleEvent (the re-cast) → SAME row reset to pending, +// new activity id, a Like +// outbound.Worker.DeliverNext (failing sender) → that Like POISONS +// +// — because a hand-inserted row would be some steady state a fixture author +// chose, and the whole condition here is a row whose history has two steps that +// disagree. +// +// AND IT MUST NOT BE READ FROM THE VOTE ROW. worker.voteCallback resolves via +// GetByActivityID and returns nil on NotFound, so a delivery already in flight +// when the re-cast lands settles into silence: its id no longer matches +// current_activity_id and the callback no-ops. The row is what the bug erases. +// outbound_activities is append-only and carries the subject in parent_at_uri +// from both vote enqueue sites, so the activity/delivery history is the only +// durable record of what the peer was actually told. + +// recastWorld drives one persona's vote through a history and hands back the +// pieces the assertions need. +type recastWorld struct { + *lifecycle + subject string +} + +func newRecastWorld(t *testing.T, maxAttempts int) *recastWorld { + t.Helper() + l := newLifecycle(t, maxAttempts) + return &recastWorld{lifecycle: l, subject: l.subjectURI} +} + +// deliveredVoteActivity is the activity id of the vote the peer accepted, read +// from the delivery history rather than from the vote row — the same evidence +// the sweep has to use, so the assertion cannot pass through a path the sweep +// cannot see. +func deliveredVoteActivity(t *testing.T, w *recastWorld) string { + t.Helper() + var id string + require.NoError(t, w.db.QueryRow(` + SELECT a.activity_id + FROM outbound_activities a + JOIN outbound_deliveries d ON d.activity_id = a.activity_id + WHERE a.kind IN ('Like', 'Dislike') AND d.state = 'delivered' + ORDER BY d.seq DESC + LIMIT 1`).Scan(&id), + "precondition: a vote really was delivered to the peer") + return id +} + +// sweep runs the real reconciler over this world and returns its report. +func sweep(t *testing.T, w *recastWorld) ingest.DivergenceReport { + t.Helper() + reconciler, err := ingest.NewDivergenceReconciler(ingest.DivergenceOptions{DB: w.db}) + require.NoError(t, err) + report, err := reconciler.Sweep(context.Background()) + require.NoError(t, err) + return report +} + +func recastEntries(report ingest.DivergenceReport) []ingest.DivergenceEntry { + var found []ingest.DivergenceEntry + for _, entry := range report.Entries { + if entry.Class == ingest.DivergenceVoteRecastUndelivered { + found = append(found, entry) + } + } + return found +} + +// --------------------------------------------------------------------------- +// The divergence +// --------------------------------------------------------------------------- + +// TestRecastDivergence_APoisonedRecastLeavesThePeerHoldingTheOldVote is the +// cycle's contract. +func TestRecastDivergence_APoisonedRecastLeavesThePeerHoldingTheOldVote(t *testing.T) { + w := newRecastWorld(t, 1) // one attempt, so the re-cast's delivery poisons + ctx := context.Background() + + // --- STEP 1: a down-vote, delivered for real. + w.castVote(t, "3lztprev00001", directionDown) + w.deliver(t) + require.Equal(t, []string{"Dislike"}, w.sender.kinds(), + "precondition: the peer received a DISLIKE") + require.Equal(t, string(store.DeliveredStateDelivered), w.state(t)) + held := deliveredVoteActivity(t, w) + + // --- STEP 2: the user changes their mind. The SAME record is rewritten, so + // the row resets to pending under a new activity id while the peer's + // copy of the old vote is untouched. + w.sender.fail(fmt.Errorf("lemmy is unreachable")) + w.castVote(t, "3lztprev00002", directionUp) + require.Equal(t, string(store.DeliveredStatePending), w.state(t), + "precondition: the re-cast reset the row — this is the step that erases our record "+ + "of what the peer holds") + + // --- STEP 3: and the new vote never lands. + w.deliver(t) + var poisoned int + require.NoError(t, w.db.QueryRow( + `SELECT COUNT(*) FROM outbound_deliveries WHERE state = 'poisoned'`).Scan(&poisoned)) + require.Equal(t, 1, poisoned, + "precondition: the re-cast's delivery POISONED, which is what makes this permanent "+ + "rather than a moment in flight") + + // --- THEN: the store names the pair, citing what the peer is holding. + found, err := store.NewDivergences(w.db).RecastDivergences(ctx) + require.NoError(t, err) + require.Len(t, found, 1, + "exactly one divergence: the peer is counting a Dislike this bridge no longer claims. "+ + "Our vote row says 'pending' — it was reset by the re-cast — so nothing in the "+ + "ledger records that a vote of ours stands on that instance, and the reseed "+ + "subtracts only 'delivered' rows. Read from the vote row this condition is "+ + "invisible by construction; only the append-only activity history still knows") + assert.Equal(t, tpNativeDID, found[0].ActorDID) + assert.Equal(t, w.subject, found[0].SubjectATURI, + "the pair (actor, subject) IS the identity of a vote — only one may be live at a time — "+ + "and the subject is what an operator checks the tally of") + assert.Equal(t, held, found[0].DeliveredActivityID, + "citing the DELIVERED activity: the id the peer accepted is the only handle anyone has "+ + "on what they are actually counting, and it is what a manual Undo would have to "+ + "embed. Citing the pending one would name the vote nobody received") + + // --- AND the operator surface says the same thing. + report := sweep(t, w) + entries := recastEntries(report) + require.Len(t, entries, 1, "the sweep reports it under its own class") + assert.Equal(t, w.subject, entries[0].Subject) + assert.Contains(t, entries[0].Detail, held, + "the entry's detail carries the delivered activity id, because a class and a subject "+ + "alone do not tell an operator what the peer is holding") + assert.Contains(t, entries[0].Detail, tpNativeDID, "and the actor it belongs to") + assert.Equal(t, 1, report.Counts[ingest.DivergenceVoteRecastUndelivered]) +} + +// --------------------------------------------------------------------------- +// The two controls — both must be silent +// --------------------------------------------------------------------------- + +// TestRecastDivergence_ACleanlyDeliveredVoteIsNotADivergence is the ordinary +// case, and it is most of the table: a vote cast once and delivered once. +// +// The peer holds exactly what our state claims. If this reported, the class +// would name every vote the bridge has ever cast — the failure mode that looks +// like thoroughness, on the highest-volume table in the system. +func TestRecastDivergence_ACleanlyDeliveredVoteIsNotADivergence(t *testing.T) { + w := newRecastWorld(t, 5) + + w.castVote(t, "3lztprev00001", directionDown) + w.deliver(t) + require.Equal(t, string(store.DeliveredStateDelivered), w.state(t), + "precondition: cast once, delivered once — the state the whole system is usually in") + + found, err := store.NewDivergences(w.db).RecastDivergences(context.Background()) + require.NoError(t, err) + assert.Empty(t, found, + "a delivered vote nobody re-cast is not a divergence: the peer holds exactly what we "+ + "claim. Reporting it would put every vote in the system in this class, and the one "+ + "real case would be indistinguishable from the background") + assert.Empty(t, recastEntries(sweep(t, w))) +} + +// TestRecastDivergence_ADeliveredThenUndoneVoteIsNotADivergence is the case +// that separates "the peer holds something" from "the peer once held +// something". +// +// The Undo delivered, so the peer holds NOTHING — and the vote row is gone, +// deleted by the Undo's own callback. An implementation reasoning from the +// activity history alone, without weighing the Undo, sees a delivered Dislike +// with no live row and reports it forever: a permanent finding about a vote that +// was correctly withdrawn. +func TestRecastDivergence_ADeliveredThenUndoneVoteIsNotADivergence(t *testing.T) { + w := newRecastWorld(t, 5) + + w.castVote(t, "3lztprev00001", directionDown) + w.deliver(t) + require.Equal(t, string(store.DeliveredStateDelivered), w.state(t)) + + w.deleteVote(t, "3lztprev00002") + w.deliver(t) + require.Equal(t, []string{"Dislike", "Undo"}, w.sender.kinds(), + "precondition: the withdrawal reached the peer") + require.Equal(t, "", w.state(t), + "precondition: and the row is gone, cleared by the Undo's delivery callback — so the "+ + "ONLY remaining evidence is the activity history this sweep reads") + + found, err := store.NewDivergences(w.db).RecastDivergences(context.Background()) + require.NoError(t, err) + assert.Empty(t, found, + "a vote whose Undo was DELIVERED leaves the peer holding nothing, so there is nothing "+ + "to reconcile. The delivered Dislike is still in the activity history and always "+ + "will be — the history is append-only — so a sweep that looks only for 'a delivered "+ + "vote with no live row' reports this pair forever, about a withdrawal that worked") + assert.Empty(t, recastEntries(sweep(t, w))) +} -- tangled.sh