diff --git a/FOLLOWUPS.md b/FOLLOWUPS.md --- a/FOLLOWUPS.md +++ b/FOLLOWUPS.md @@ -59,8 +59,6 @@ - Outbound Coves-to-Lemmy federation and ActivityPub actors for Coves users. - Key claiming/migration for bridged users. Handle-collision recovery should eventually reuse an orphaned minted DID via a PLC `updateHandle` operation rather than minting another DID. -- If vote write-back is added, suppress echoes of Tidepool-managed voters and - subtract Tidepool's written-back tally during subsequent Lemmy re-seeds. - Moderation federation and DMs. ## Outbound delivery (task 15) @@ -120,6 +118,34 @@ driven by an event on the post (redelivery, edit, stats sweep). A quiet community's crash-window acceptance gap persists until any such event. The decision-19 reconciliation job (task 18) is the natural home for a periodic acceptance-vs-record pin audit. + +## Vote accounting (task 17b) + +- **The re-seed baseline clamp is a FLOOR BREACH, not a discard + detector.** `GREATEST(0, …)` fires only where the deficit exceeds a + subject's entire fediverse tally, so on any post with a real score a + Lemmy `FederationMode` silently discarding our written-back votes + understates the served total with the raw baseline still positive and + the counter at zero. `tidepool_vote_seed_ours_subtracted` is the + signal that advances on healthy subjects; comparing it against a + Lemmy-side sample is the actual audit. +- **Nothing re-seeds periodically.** `SeedPostCounts` has one caller, + behind `SEED_COUNTS_FROM_API`, skipped inside a 1h window unless an + admin forces a backfill. Any claim that drift "heals on the next + re-seed" may mean never for a quiet community. +- **Migration 023's cleanup DELETE is narrower than the runtime guard + it backfills.** It matches `voter_ap_id` by exact string equality + while `echo.identifyActor` matches normalized host + scheme, so a + legacy row carrying a non-canonical spelling of one of our actor ids + would survive and then be counted in BOTH subtrahends — the one + reachable way they double-subtract the same human. Accepted because + the set is empty in production (write-back never shipped) and no code + path here ever wrote non-canonical ids; a normalizing DELETE would + mean re-implementing the probe in SQL. +- **A `Down` of migration 023 leaves baselines understated + indefinitely.** Anything seeded while 023 was applied is net of our + delivered votes, the pre-023 code does not re-derive it, and with no + periodic re-seed only a forced backfill corrects it. ## Echo suppression (task 17a) diff --git a/internal/db/migrations/023_vote_seed_netting.sql b/internal/db/migrations/023_vote_seed_netting.sql new file mode 100644 --- /dev/null +++ b/internal/db/migrations/023_vote_seed_netting.sql @@ -0,0 +1,84 @@ +-- +goose Up +-- Task 17b: the served vote aggregate is the FEDIVERSE-ONLY tally. +-- +-- Coves keeps native and bridged counts in separate columns, so the votes +-- Tidepool wrote back on behalf of native users must come OUT of what the +-- bridge serves — otherwise one person's single vote is counted twice in the +-- UI, once in each column. SeedAggregates now nets delivered outbound_votes out +-- of the baseline alongside live vote_events. +-- +-- The index is PARTIAL on the predicate the seeder uses. Not +-- (subject_ap_id, delivered_state): task 17d wants (actor_did) WHERE +-- delivered_state = 'delivered', a different index, and a composite here would +-- serve neither well while costing writes on the delivery worker's hot update. +CREATE INDEX outbound_votes_subject_delivered_idx + ON outbound_votes (subject_ap_id) + WHERE delivered_state = 'delivered'; + +-- One-time cleanup: vote_events rows cast by OUR OWN personas, plus a recompute +-- of ONLY the subjects that cleanup touches. +-- +-- Task 17a's voter probe makes such a row unwritable — Aggregator.ApplyVote +-- refuses a voter that resolves to an ap_actors persona before it touches any +-- table — so any row matching this predicate predates the guard and is +-- unconditionally garbage. Write-back has never been deployed, so production +-- has zero of them; this exists so a development database cannot carry one into +-- the new accounting. +-- +-- It is deliberately NOT a filter in the seeder's live subquery. Any such +-- filter would have to be MIRRORED in recomputeAggregate, which reads the same +-- rows on every inbound vote: filtering in one place only yields +-- seeded = api − inbound_filtered and then served = seeded + inbound_unfiltered, +-- wrong by exactly the rows the filter excluded, on every subject those rows +-- touch. +-- +-- The predicate is EXACT-ID equality, which is deliberately NARROWER than the +-- runtime probe it backfills: echo.identifyActor matches an actor route on a +-- normalized host plus scheme, so a legacy row spelled non-canonically +-- (explicit :443, trailing dot, differing case) is "ours" to the probe and +-- survives this DELETE — and would then be subtracted twice, once as a live +-- inbound row and once as a delivered outbound one. Accepted because the set is +-- empty in production and non-canonical ids were never written by any code path +-- here; a normalizing DELETE would have to re-implement the probe in SQL. +-- +-- The recompute is SCOPED to the affected subjects, for two reasons beyond +-- cost: an unqualified UPDATE row-locks every aggregate (blocking vote +-- ingestion for its duration), and it stamps updated_at on every row — which +-- migration 014's stats watermark reads as "due" (updated_at > stats_emitted_at) +-- and turns into a full re-emit sweep of record rewrites and firehose traffic +-- for every voted-on subject in the system. Scoped, this is a true no-op when +-- there is nothing to clean, and re-running it re-triggers nothing. +-- +-- The anti-join inside the counts is NOT redundant with the DELETE: every +-- sub-statement of a WITH sees the SAME snapshot, so the recompute cannot +-- observe the deletion above and must exclude the same rows itself. +WITH scrubbed AS ( + DELETE FROM vote_events + WHERE voter_ap_id IN (SELECT actor_id FROM ap_actors) + RETURNING subject_ap_id +) +UPDATE vote_aggregates a +SET upvotes = a.seeded_upvotes + ( + SELECT COUNT(*) FROM vote_events e + WHERE e.subject_ap_id = a.subject_ap_id AND NOT e.undone AND e.direction = 'up' + AND e.voter_ap_id NOT IN (SELECT actor_id FROM ap_actors)), + downvotes = a.seeded_downvotes + ( + SELECT COUNT(*) FROM vote_events e + WHERE e.subject_ap_id = a.subject_ap_id AND NOT e.undone AND e.direction = 'down' + AND e.voter_ap_id NOT IN (SELECT actor_id FROM ap_actors)), + updated_at = clock_timestamp() +WHERE a.subject_ap_id IN (SELECT subject_ap_id FROM scrubbed); + +-- +goose Down +-- The deleted persona rows are not restorable (they were garbage), and the +-- scoped recompute is idempotent, so the index is all there is to drop. +-- +-- But this Down is NOT a full reversal, and the residue is silent: every +-- baseline seeded while 023 was applied was stored NET of our delivered +-- outbound votes, and the pre-023 code does not re-derive baselines. Nothing +-- re-seeds periodically either (SeedAggregates' only caller is the backfill's +-- post walk, behind SEED_COUNTS_FROM_API and a freshness window), so after a +-- Down those subjects serve totals understated by our own votes INDEFINITELY — +-- not until the next backfill. Re-seeding the affected subjects with a forced +-- backfill is the only correction. +DROP INDEX IF EXISTS outbound_votes_subject_delivered_idx; diff --git a/internal/echo/echo.go b/internal/echo/echo.go --- a/internal/echo/echo.go +++ b/internal/echo/echo.go @@ -169,13 +169,15 @@ return Identity{Class: ClassNone}, nil } // identifyObject mirrors personas.handleObject: rest is did/collection/rkey, -// and the body is served from OUTBOUND_OBJECTS (serving.go:137) — so that table -// is the route's real oracle, and the ap_objects read is defence in depth. +// and the body is served from OUTBOUND_OBJECTS (personas' handleObject reads +// GetByATURI) — so that table is the route's real oracle, and the ap_objects +// read is defence in depth. // // Both are consulted because the two rows are written by different halves of // the bridge and neither implies the other: the enqueuer records a -// bridge-origin ap_objects mapping (enqueuer.go:136-144, 196-205) alongside the -// outbound row, but a legacy v1 write has only the mapping, and a state where +// bridge-origin ap_objects mapping (outbound's Enqueuer.EnqueueActivity, via +// objectMapping) alongside the outbound row, but a legacy v1 write has only the +// mapping, and a state where // the outbound row is missing must not make our own object answer "remote". // // The ap_objects read is keyed by the requested id, but a row alone is NOT diff --git a/internal/outbound/worker.go b/internal/outbound/worker.go --- a/internal/outbound/worker.go +++ b/internal/outbound/worker.go @@ -221,6 +221,16 @@ if err != nil { return fmt.Errorf("load activity %s: %w", delivery.ActivityID, err) } + // A delivery the peer ALREADY ACCEPTED, held because its local settlement + // failed, resumes at the settlement — never at the wire. Re-POSTing would + // re-send an activity the peer holds, and every gate below decides whether + // to SEND: the kill switch, the causal wait and the consent recheck are all + // answers to "should this go out?", asked after it already has. Cancelling + // it here would strand the ledger it was held to settle. + if delivery.LastErrorClass == deliveryLedgerUnsettled { + return w.deliverSuccess(ctx, delivery, activity, statusOf(delivery)) + } + // Kill switch (decision 19): an operator block PARKS the delivery — it stays // pending and resumes when the switch clears, never poisoned or cancelled. scope := DeliveryScope{ @@ -364,27 +374,73 @@ } return w.poison(ctx, delivery, "inbox_gone", "inbox still unreachable after re-resolution", status) } -// deliverSuccess opens the causal gate and marks the delivery delivered, in -// that order so the two are effectively atomic: a parent must NEVER be observed -// delivered while its accepted_at is unset (that strands every child forever). -// The accepted_at stamp is written FIRST; only if it commits is the delivered -// mark applied. If the stamp genuinely fails, we return before marking and the -// retry re-runs both (Lemmy dedupes the re-POST). The only reachable states are -// (¬accepted,¬delivered), (accepted,¬delivered), (accepted,delivered) — never -// the forbidden (¬accepted,delivered). +// deliverSuccess settles everything a successful POST implies, and marks the +// delivery delivered LAST. +// +// The order is the invariant. Marking delivered is TERMINAL — the queue never +// re-claims that row — so every write that happens after it is a write nothing +// will ever retry. Two of them matter: +// +// - accepted_at (the causal gate): a parent observed delivered with its stamp +// unset strands every child forever; +// - the vote ledger: since task 17b, outbound_votes is an INPUT to the number +// users read, so a delivery left terminal beside a 'pending' row over-counts +// that subject permanently, and beside a stale 'delivered' row under-counts +// it permanently. Nothing reconciles either — SeedAggregates only runs +// behind a backfill. +// +// So the reachable states are (¬settled,¬delivered) and (settled,delivered), +// never the forbidden (¬settled,delivered). Both settlements are idempotent, so +// a retry that repeats them costs nothing. +// +// When a settlement fails, the delivery is HELD FOR SETTLEMENT rather than +// failed: see settleLater. Failing it would re-POST an activity the peer has +// already accepted; poisoning it would make the disagreement permanent, which +// is the whole bug. func (w *Worker) deliverSuccess(ctx context.Context, delivery *store.OutboundDelivery, activity *store.OutboundActivity, status int) error { if err := w.stampAccepted(ctx, activity); err != nil { - return fmt.Errorf("stamp accepted for %s: %w", delivery.ActivityID, err) + return w.settleLater(ctx, delivery, status, + fmt.Errorf("stamp accepted for %s: %w", delivery.ActivityID, err)) + } + if err := w.voteCallback(ctx, activity); err != nil { + return w.settleLater(ctx, delivery, status, err) } _, applied, err := w.deliveries.MarkDelivered(ctx, delivery.ActivityID, delivery.TargetInbox, status, *delivery.ClaimedUntil) if err != nil { return fmt.Errorf("mark delivered %s: %w", delivery.ActivityID, err) } if !applied { - return nil // a stale claim: another worker already recorded the outcome + // A stale claim: our lease lapsed mid-POST and another worker owns the + // row now. The settlements above already ran — they are keyed on the + // ACTIVITY, not on the claim, so the ledger is correct whichever worker + // loses the fencing race, and the winner's redelivery repeats them + // idempotently. + return nil } metricDelivered.Add(1) - return w.voteCallback(ctx, activity) + return nil +} + +// settleLater holds a delivery whose POST SUCCEEDED but whose settlement did +// not. The row stays PENDING — non-terminal, so the queue will come back to it +// — carrying deliveryLedgerUnsettled as its outcome class, which is what tells +// the next claim to resume at the settlement instead of at the wire. +// +// It never poisons. A poisoned delivery is terminal, and terminal-with-unsettled +// is exactly the permanent disagreement this exists to prevent; the failure here +// is a LOCAL write, so retrying is both safe and the only thing that can help. +// The backoff still spaces the retries (production's base is 30s), so a +// persistently broken local write does not spin the worker. +func (w *Worker) settleLater(ctx context.Context, delivery *store.OutboundDelivery, status int, cause error) error { + w.logger.Warn("delivery accepted by the peer but not yet settled locally; holding for settlement", + "activity", delivery.ActivityID, "inbox", delivery.TargetInbox, + "attempts", delivery.Attempts, "error", cause) + next := time.Now().Add(w.backoff(delivery.Attempts)) + if _, _, err := w.deliveries.Release(ctx, delivery.ActivityID, delivery.TargetInbox, + deliveryLedgerUnsettled, cause.Error(), status, next, *delivery.ClaimedUntil); err != nil { + return fmt.Errorf("hold %s for settlement: %w", delivery.ActivityID, err) + } + return nil } // stampAccepted opens the causal gate for this object's children: on a @@ -465,6 +521,22 @@ return fmt.Errorf("poison delivery %s: %w", delivery.ActivityID, err) } metricPoisoned.Add(1) return nil +} + +// deliveryLedgerUnsettled labels a delivery the peer accepted whose LOCAL +// settlement (the causal stamp, the vote ledger) has not committed yet. It is +// an outcome class, not an error class — park and parkCausal already use the +// same column for held-not-failed states — and it is the durable fact that lets +// a retry finish the job without repeating the POST. +const deliveryLedgerUnsettled = "ledger_unsettled" + +// statusOf is the status a held delivery was accepted with, so its settlement +// records the same outcome the wire actually produced. +func statusOf(delivery *store.OutboundDelivery) int { + if delivery.LastStatusCode == nil { + return http.StatusAccepted + } + return *delivery.LastStatusCode } // parkDelay is how long a kill-switched or dry-run delivery waits before it can diff --git a/internal/votes/aggregator.go b/internal/votes/aggregator.go --- a/internal/votes/aggregator.go +++ b/internal/votes/aggregator.go @@ -18,6 +18,7 @@ import ( "context" "database/sql" stderrors "errors" + "expvar" "fmt" "log/slog" "sort" @@ -31,11 +32,46 @@ "tidepool/internal/ratelimit" "tidepool/internal/store" ) -// echoVoteLogInterval throttles the suppressed-voter log. Suppression is rare -// by construction, so the sampler costs nothing in steady state — but the -// failure mode this log exists to expose is a probe that has started matching -// GENUINE voters, and that one arrives at full vote volume. -const echoVoteLogInterval = time.Second +// voteWarnInterval throttles each of the aggregator's rare-but-loud notices: a +// suppressed voter, and a clamped seed baseline. Both are rare by construction, +// so a sampler costs nothing in steady state — but each announces itself at +// volume in exactly the failure it exists to expose (a probe that has started +// matching GENUINE voters; an origin whose totals no longer contain the votes +// we wrote back), and one line per vote or per backfilled post would bury it. +// +// The interval is shared; the SAMPLERS are not. Their causes are correlated — +// switching write-back on produces echoed votes AND clamped baselines — so one +// sampler would let each signal suppress the other in every window of precisely +// the incident both lines exist to describe. +const voteWarnInterval = time.Second + +// SeedBaselineClamped counts seeds whose computed baseline came out NEGATIVE +// and was clamped to zero: the origin's total for that DIRECTION was smaller +// than the votes we can already account for on that subject. +// +// Read it for what it is — a floor breach, not a discard detector. It can only +// fire where api_total < live + ours, i.e. on subjects whose entire fediverse +// tally is smaller than the deficit; on any post with a real score, an origin +// silently discarding the votes we write back (a restrictive Lemmy +// FederationMode, decision 16's named risk) understates the served tally with +// the raw baseline still comfortably positive, and this counter stays at zero. +// SeedOursSubtracted is the signal that covers those subjects. +// +// The "tidepool_" prefix is load-bearing: the admin metrics surface serves ONLY +// that prefix, so a counter named without it is published to expvar and then +// filtered straight back out — indistinguishable from a counter that never +// fires. +var SeedBaselineClamped = expvar.NewInt("tidepool_vote_seed_baseline_clamped") + +// SeedOursSubtracted totals the delivered outbound votes netted out of seeded +// baselines, across directions and subjects. +// +// It is the volume half of the clamp's signal, and unlike the clamp it advances +// on ordinary healthy subjects: it says how many votes the bridge BELIEVES the +// origin is holding for our personas. Compared against a Lemmy-side sample of +// the same posts, a persistent gap is a discard being absorbed silently — +// which the clamp only ever catches on near-zero-score subjects. +var SeedOursSubtracted = expvar.NewInt("tidepool_vote_seed_ours_subtracted") // Vote directions (vote_events.direction). const ( @@ -90,6 +126,7 @@ communities store.Communities records RecordReader voters VoterProbe echoLog *ratelimit.Sampler + clampLog *ratelimit.Sampler logger *slog.Logger } @@ -119,7 +156,10 @@ if logger == nil { logger = slog.Default() } return &Aggregator{db: db, objects: objects, communities: communities, records: records, - voters: voters, echoLog: ratelimit.NewSampler(echoVoteLogInterval), logger: logger}, nil + voters: voters, + echoLog: ratelimit.NewSampler(voteWarnInterval), + clampLog: ratelimit.NewSampler(voteWarnInterval), + logger: logger}, nil } // ApplyVote records one Like or Dislike: insert the activity (duplicate @@ -401,27 +441,62 @@ // subject from its origin's public API — history whose individual Like // activities the bridge never saw (Lemmy outboxes announce historical votes // only sparsely). Live vote_events stack on top of the baseline. // -// The origin's counts are a TOTAL: they include every vote that ALSO -// federated live and sits in vote_events as a live row (any vote cast after -// the community was subscribed). Storing them raw would count those voters -// twice — once in the baseline, once in the recompute's live term — so the -// baseline is stored NET of the subject's live counts, per direction, -// clamped at zero. Served totals therefore equal the origin's counts at -// seed time, and live events stack on top from there. This also makes a -// re-seed (backfill redo) the drift healer: a voter counted only in the -// baseline who later flips federates a bare Dislike (Lemmy sends no Undo on +// WHAT THE SERVED AGGREGATE MEANS (task 17b). The Coves appview (a separate +// repo) keeps native and bridged tallies in SEPARATE columns on its post +// record — bridged_upvote_count beside the native count — so what Tidepool +// serves is the FEDIVERSE-ONLY tally: +// +// served(subject) = api_total(subject) − { our personas' votes Lemmy currently holds } +// +// The origin's counts are a TOTAL, and two populations inside it are already +// accounted for elsewhere: +// +// - votes that ALSO federated live and sit in vote_events as live rows (any +// vote cast after the community was subscribed). Keeping them counts those +// voters twice — once in the baseline, once in the recompute's live term; +// - votes TIDEPOOL ITSELF wrote back for native users, which Lemmy is holding +// and reporting. Coves counts those in its NATIVE column, so keeping them +// counts one person's single vote twice across the two columns in the UI. +// +// So the baseline is stored NET of both, per direction, clamped at zero. See +// reportSeed for what the seed publishes about that: the volume it subtracted, +// and — on the rare subject whose whole tally is smaller than the deficit — the +// clamp that would otherwise absorb the difference in silence. +// +// The subtraction lives in the BASELINE rather than the served columns because +// recomputeAggregate rewrites the served columns on every single inbound vote: +// a correction applied there would be undone minutes later by the next voter, +// with nothing connecting the drift back to the seed. +// +// A re-seed (backfill redo) is also the drift healer: a voter counted only in +// the baseline who later flips federates a bare Dislike (Lemmy sends no Undo on // flips), leaving the retired upvote in the baseline next to the new live -// downvote — until the next re-seed, whose subtraction converges the served -// totals back to the origin's truth. Two symmetric residual races span the -// origin API fetch and this transaction, both transient and self-healing on -// the next re-seed (the pre-fix over-count race was PERMANENT and compounding): +// downvote — until the next re-seed converges the served totals back. But +// nothing re-seeds PERIODICALLY: the one caller is ingest's Backfill.seedCounts +// on its post walk, behind SEED_COUNTS_FROM_API, and an un-forced trigger skips +// the whole run inside the freshness window — so for a quiet community "heals +// on the next re-seed" can mean "heals when an admin forces a backfill", and +// may mean never. +// +// Three residual races span the origin API fetch and this transaction. All are +// transient and self-healing on the next re-seed, with the caveat above (the +// pre-fix over-count race was PERMANENT and compounding): // - under-count by one: a vote federates AFTER the fetch but is live here, so // it is net-subtracted from the baseline yet not present in the fetched // total; // - over-count by one (the mirror): a vote already IN the fetched total whose // federated activity arrives AFTER this seed tx — the net-of-live // subtraction cannot yet see it as a live row, so the baseline keeps it AND -// the later live event adds it again, until the next re-seed reconciles. +// the later live event adds it again, until the next re-seed reconciles; +// - over-count by one, outbound side: a native user RE-CASTS a vote they had +// already delivered. consume's applyVoteWrite re-upserts the row and +// OutboundVotes.Upsert resets delivered_state to 'pending', while Lemmy +// still holds the OLD vote in the OLD direction — so this seed subtracts +// nothing for it and the stale vote stays in the served tally. Transient +// (the redelivery flips the row back to 'delivered') but PERMANENT if that +// delivery poisons. Fixing it needs a second column pair modelling "what +// the peer holds" against "what the user wants", which is task 17e's, not +// this one's. // // Subjects not present in ap_objects are dropped and logged at debug, like // ApplyVote. @@ -447,7 +522,12 @@ return nil } atURI := mapping.ATURI - return a.inTx(ctx, func(tx *sql.Tx) error { + // Seed facts are collected inside the transaction and reported AFTER it + // commits: recomputeAggregate can still fail and roll the whole seed back, + // and a counter advanced for a seed that never happened — then advanced + // again by the caller's retry — is worse than no counter. + var seed seedOutcome + if err := a.inTx(ctx, func(tx *sql.Tx) error { // Upsert-and-lock the aggregate row first — the per-subject // serialization point every mutation goes through — so the live-count // read below cannot interleave with a concurrent ApplyVote/RetractVote @@ -455,23 +535,125 @@ // on the same subject. if err := lockAggregate(ctx, tx, subjectAPID, atURI); err != nil { return err } - if _, err := tx.ExecContext(ctx, ` + // ONE statement for both subtrahends, deliberately: a second statement + // would read on its own READ COMMITTED snapshot (inTx takes the default + // isolation), so the two counts could come from different moments — + // creating exactly the torn read the aggregate lock exists to prevent. + // + // ours.* is the votes LEMMY CURRENTLY HOLDS for our personas, and the + // predicate is the POSITIVE EQUALITY delivered_state = 'delivered': + // + // pending (first try, retrying, poisoned) → the peer does not hold it + // delivered → it does: subtract + // delivered, Undo in flight → still subtract; consume's + // applyVoteDelete re-upserts + // 'delivered' precisely because the + // peer has not yet processed the + // withdrawal + // row gone (Undo delivered) → nothing to subtract + // + // A negation ("NOT undone", "<> 'pending'") reads identically TODAY only + // because nothing writes 'undone'. If a policy ever does, it will mean + // the peer ACCEPTED the withdrawal — not live — and every negation + // silently inverts while this equality stays correct. A poisoned Undo + // leaves a delivered row subtracting forever, which is the same hazard + // decision 16 cites for banning queue-history arithmetic: "delivered + // Likes minus delivered Undos" gets that row permanently wrong. + // + // The read takes no row locks. The seed holds the aggregate lock and + // reads outbound_votes lock-free; the delivery worker locks + // outbound_votes and never touches vote_aggregates — so there is no + // cycle, and FOR UPDATE would both create one and park a backfill's + // seeding behind in-flight HTTP deliveries. + if err := tx.QueryRowContext(ctx, ` UPDATE vote_aggregates a - SET seeded_upvotes = GREATEST(0, $2 - live.up), - seeded_downvotes = GREATEST(0, $3 - live.down) + SET seeded_upvotes = GREATEST(0, $2 - live.up - ours.up), + seeded_downvotes = GREATEST(0, $3 - live.down - ours.down) FROM ( SELECT COUNT(*) FILTER (WHERE direction = 'up') AS up, COUNT(*) FILTER (WHERE direction = 'down') AS down FROM vote_events WHERE subject_ap_id = $1 AND NOT undone - ) live - WHERE a.subject_ap_id = $1`, - subjectAPID, upvotes, downvotes); err != nil { + ) live, ( + SELECT + COUNT(*) FILTER (WHERE direction = 'up') AS up, + COUNT(*) FILTER (WHERE direction = 'down') AS down + FROM outbound_votes + WHERE subject_ap_id = $1 AND delivered_state = 'delivered' + ) ours + WHERE a.subject_ap_id = $1 + RETURNING $2 - live.up - ours.up, $3 - live.down - ours.down, + live.up, live.down, ours.up, ours.down`, + subjectAPID, upvotes, downvotes).Scan( + &seed.rawUp, &seed.rawDown, &seed.liveUp, &seed.liveDown, + &seed.oursUp, &seed.oursDown); err != nil { return fmt.Errorf("seed vote aggregate for %q: %w", subjectAPID, err) } return recomputeAggregate(ctx, tx, subjectAPID) - }) + }); err != nil { + return err + } + a.reportSeed(subjectAPID, upvotes, downvotes, seed) + return nil +} + +// seedOutcome is what one committed seed observed: the raw (pre-clamp) signed +// baselines, and the two subtrahends they were computed from. +type seedOutcome struct { + rawUp, rawDown int64 + liveUp, liveDown int64 + oursUp, oursDown int64 +} + +// reportSeed publishes what a COMMITTED seed observed. +// +// SeedOursSubtracted is the routine half: how many of our personas' votes this +// seed believes the origin is holding. It advances on healthy subjects, which +// is the point — a persistent gap against a Lemmy-side sample is how a silent +// discard shows up on posts with real scores. +// +// The clamp is the exceptional half, and it makes GREATEST(0, …) visible: a +// negative raw baseline means the origin's total for that direction is smaller +// than what we can already account for, and the clamp then absorbs the deficit +// silently. Counters always advance (a sampled counter counts nothing); only +// the line is sampled, because a backfill seeds one subject per post and a +// systemic cause produces one line per post for the whole run. +func (a *Aggregator) reportSeed(subject string, apiUp, apiDown int, seed seedOutcome) { + if ours := seed.oursUp + seed.oursDown; ours > 0 { + SeedOursSubtracted.Add(ours) + } + if seed.rawUp >= 0 && seed.rawDown >= 0 { + return + } + SeedBaselineClamped.Add(1) + // Both directions are reported, always: when both breach, naming one hides + // the other, and the pair is what says whether the cause is directional. + direction := directionUp + switch { + case seed.rawUp < 0 && seed.rawDown < 0: + direction = directionUp + "+" + directionDown + case seed.rawDown < 0: + direction = directionDown + } + if a.clampLog.Allow(time.Now()) { + a.logger.Warn("vote seed baseline clamped: the origin's total is short of the votes we can account for", + "subject", subject, "direction", direction, + "deficit_up", min64(seed.rawUp, 0), "deficit_down", min64(seed.rawDown, 0), + "api_up", apiUp, "api_down", apiDown, + "live_up", seed.liveUp, "live_down", seed.liveDown, + "ours_up", seed.oursUp, "ours_down", seed.oursDown) + } +} + +// min64 reports the smaller of two int64s (the deficit fields log 0 for a +// direction that did not breach, rather than a positive baseline that would +// read as one). +func min64(a, b int64) int64 { + if a < b { + return a + } + return b } // ScrubVoter erases every vote_events row a voter ever produced — the vote diff --git a/internal/votes/echo_probe_test.go b/internal/votes/echo_probe_test.go --- a/internal/votes/echo_probe_test.go +++ b/internal/votes/echo_probe_test.go @@ -13,7 +13,6 @@ "tidepool/internal/echo" "tidepool/internal/errors" "tidepool/internal/store" - "tidepool/internal/testutil" ) // The aggregator-level echo guard (task 17a, decision-16 AMENDMENT). @@ -62,11 +61,7 @@ // probeWorld is testDB plus the two actor tables the probe distinguishes. func probeWorld(t *testing.T) (*sql.DB, store.APObjects) { t.Helper() - database := testutil.DB(t) - testutil.Truncate(t, database, - "vote_events", "vote_aggregates", "ap_objects", "communities", - "ap_actors", "bridged_actors", "outbound_deliveries", "outbound_activities", - "outbound_objects") + database := testDB(t) ctx := context.Background() _, err := store.NewAPActors(database).Create(ctx, store.APActor{ diff --git a/internal/votes/reseed_clamp_test.go b/internal/votes/reseed_clamp_test.go new file mode 100644 --- /dev/null +++ b/internal/votes/reseed_clamp_test.go @@ -0,0 +1,215 @@ +package votes + +import ( + "context" + "expvar" + "fmt" + "log/slog" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/echo" + "tidepool/internal/store" +) + +// THE CLAMP IS THE ONE SIGNAL THIS SUBSYSTEM PRODUCES. +// +// GREATEST(0, …) floors a baseline that came out negative — the origin's total +// is smaller than the votes we can already account for on that subject. That is +// not arithmetic noise: the shape it has is a peer accepting our votes, +// answering 200, and counting none of them (Lemmy's FederationMode discard). +// After 17b this is the ONLY place in the entire system where that becomes +// observable, so a clamp that fires silently is the difference between a +// diagnosable outage and a number that is quietly wrong. +// +// Which is why the negative control below matters as much as the positive one: +// a counter that advances on EVERY seed is a seed-rate meter wearing an +// incident's name, and it would read as "healthy, no clamps" exactly never. + +// clampWorld is a fresh database, aggregator and log sink. Each case gets its +// own aggregator on purpose: the Warn is sampled per-aggregator, so two clamps +// through one instance would suppress the second line and the assertions would +// be measuring the sampler. +// +// The outbound rows here are written through the store rather than the full +// consumer→worker path (which reseed_temporal_test.go drives end to end): the +// clamp is arithmetic over a state that file already proves is reachable. +func clampWorld(t *testing.T) (*Aggregator, *tpLogBuffer, store.APObjects) { + t.Helper() + database := testDB(t) + objects := store.NewAPObjects(database) + logs := &tpLogBuffer{} + probe, err := echo.New(echo.Options{ + Objects: objects, + OutboundObjects: store.NewOutboundObjects(database), + Activities: store.NewOutboundActivities(database), + Actors: store.NewAPActors(database), + }) + require.NoError(t, err) + agg, err := NewAggregator(database, objects, store.NewCommunities(database), + &fakeRecords{records: map[string]map[string]any{}}, probe, + slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))) + require.NoError(t, err) + return agg, logs, objects +} + +// deliverOutbound writes one DELIVERED outbound vote for a distinct persona. +func deliverOutbound(t *testing.T, agg *Aggregator, subjectAPID, subjectATURI, direction string, n int) { + t.Helper() + ctx := context.Background() + votes := store.NewOutboundVotes(agg.db) + did := fmt.Sprintf("did:plc:clamppersona%04d", n) + voteATURI := fmt.Sprintf("at://%s/social.coves.interaction.vote/3lzclampvote%d", did, n) + _, err := votes.Upsert(ctx, store.OutboundVote{ + VoteATURI: voteATURI, + ActorDID: did, + SubjectATURI: subjectATURI, + SubjectAPID: subjectAPID, + CommunityDID: rsCommunityDID, + Direction: direction, + CurrentActivityID: fmt.Sprintf("https://coves.social/ap/activity/clamp-%d", n), + DeliveredState: store.DeliveredStatePending, + }) + require.NoError(t, err) + require.NoError(t, votes.SetDeliveredState(ctx, voteATURI, store.DeliveredStateDelivered)) +} + +// TestClampIsObservableAndOnlyWhenItFires carries its own negative control. +// +// Without the control, moving SeedBaselineClamped.Add(1) above reportSeed's +// `if rawUp >= 0 && rawDown >= 0 { return }` guard — or deleting the guard — +// still produces exactly +1 for the clamping seed and still logs. The guard is +// what makes the counter mean something, and nothing else here protects it. +func TestClampIsObservableAndOnlyWhenItFires(t *testing.T) { + agg, logs, objects := clampWorld(t) + ctx := context.Background() + subjectATURI := bridgeSubject(t, objects, subjectPost, "3lzclampsubj01") + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + deliverOutbound(t, agg, subjectPost, subjectATURI, directionDown, 1) + deliverOutbound(t, agg, subjectPost, subjectATURI, directionDown, 2) + + // --- NEGATIVE CONTROL: a seed whose baselines are both non-negative. + // 3 − 1 live = 2 up, 2 − 0 live − 2 ours = 0 down. Nothing is floored. + before := SeedBaselineClamped.Value() + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 3, 2)) + assert.Equal(t, before, SeedBaselineClamped.Value(), + "a HEALTHY seed must not touch the counter: one that advances on every seed is a "+ + "seed-rate meter, and it would read 'healthy' exactly never") + assert.NotContains(t, logs.String(), "clamped", + "and it must not log an incident either") + up, down, found := counts(t, agg.db, subjectPost) + require.True(t, found) + require.Equal(t, 3, up) + require.Equal(t, 0, down, "precondition: 2 origin − 2 delivered ours = 0, exactly at the floor") + + // --- THE CLAMP: the origin now reports FEWER votes than we can account for. + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 1, 0)) + + seededUp, seededDown := seededCounts(t, agg.db, subjectPost) + assert.Equal(t, 0, seededUp, "1 origin − 1 live inbound = 0") + assert.Equal(t, 0, seededDown, + "0 origin − 2 delivered ours = −2, floored: a negative baseline would serve as a "+ + "negative score") + + up, down, _ = counts(t, agg.db, subjectPost) + assert.Equal(t, 1, up, "the served total is the live inbound vote and nothing else") + assert.Equal(t, 0, down) + + assert.Equal(t, before+1, SeedBaselineClamped.Value(), + "the clamp must COUNT: it is the only signal that the origin's total cannot cover "+ + "what we believe we delivered") + + // The line has to carry what makes the counter actionable — WHICH subject, + // HOW short, and how many of the votes were ours. reportSeed threads those + // parameters through for this; a bare "clamped" is a number nobody can act + // on. + line := logs.String() + assert.Contains(t, line, "clamped", "the clamp logs at Warn") + assert.Contains(t, line, subjectPost, "with the subject, or nobody can find the post") + assert.Contains(t, line, "deficit_down=-2", "with the size of the shortfall") + assert.Contains(t, line, "ours_down=2", + "and with how much of it we are responsible for — the difference between 'the origin "+ + "lost our votes' and 'the origin lost everyone's'") +} + +// TestClampNamesBothBreachedDirections covers reportSeed's label branching, +// which only ever runs its down-only arm in the tests above. +// +// The composite case is the one that matters: when both directions breach, +// naming one hides the other, and whether the cause is directional is the first +// question anyone asks of this signal. +func TestClampNamesBothBreachedDirections(t *testing.T) { + t.Run("up only", func(t *testing.T) { + agg, logs, objects := clampWorld(t) + ctx := context.Background() + subjectATURI := bridgeSubject(t, objects, subjectPost, "3lzclampsubj02") + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + deliverOutbound(t, agg, subjectPost, subjectATURI, directionUp, 1) + + // 0 origin up − 1 live − 1 ours = −2; down is 0 − 0 − 0 = 0. + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 0, 0)) + + line := logs.String() + assert.Contains(t, line, "direction=up", "only the up direction breached") + assert.Contains(t, line, "deficit_up=-2") + assert.Contains(t, line, "deficit_down=0", + "a direction that did not breach reports 0, not its positive baseline — which "+ + "would read as a deficit") + }) + + t.Run("both directions", func(t *testing.T) { + agg, logs, objects := clampWorld(t) + ctx := context.Background() + subjectATURI := bridgeSubject(t, objects, subjectPost, "3lzclampsubj03") + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + deliverOutbound(t, agg, subjectPost, subjectATURI, directionDown, 1) + + // up: 0 − 1 live − 0 = −1. down: 0 − 0 − 1 ours = −1. + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 0, 0)) + + line := logs.String() + assert.Contains(t, line, "direction=up+down", + "both breached: naming one direction hides the other, and the pair is what says "+ + "whether the cause is directional") + assert.Contains(t, line, "deficit_up=-1") + assert.Contains(t, line, "deficit_down=-1", + "and BOTH deficits are reported — a composite label with one number is worse "+ + "than either alone, because it looks complete") + }) +} + +// TestSeedMetricsAreServedByTheAdminEndpoint pins the PUBLISHED NAME, not the Go +// variable. The prefix is load-bearing: ingest.scopedMetrics serves only +// tidepool_*, so a counter registered without it is published to expvar and then +// filtered straight out of /admin/metrics — indistinguishable from a counter +// that never fires, which is the exact failure this signal exists to avoid. +// +// Reading SeedBaselineClamped.Value() (as every other test here does) cannot see +// that: renaming the expvar key changes nothing about the variable. +func TestSeedMetricsAreServedByTheAdminEndpoint(t *testing.T) { + for _, counter := range []struct { + what string + v *expvar.Int + }{ + {"the clamp counter", SeedBaselineClamped}, + {"the ours-subtracted counter", SeedOursSubtracted}, + } { + var published string + expvar.Do(func(kv expvar.KeyValue) { + if kv.Value == expvar.Var(counter.v) { + published = kv.Key + } + }) + require.NotEmpty(t, published, + "%s must be REGISTERED with expvar; an unpublished counter is a local variable", + counter.what) + assert.True(t, strings.HasPrefix(published, "tidepool_"), + "%s is published as %q: /admin/metrics serves only the tidepool_ prefix, so a "+ + "counter named without it is silently filtered out and reads as one that "+ + "never fires", counter.what, published) + } +} diff --git a/internal/votes/reseed_faults_test.go b/internal/votes/reseed_faults_test.go new file mode 100644 --- /dev/null +++ b/internal/votes/reseed_faults_test.go @@ -0,0 +1,241 @@ +package votes + +import ( + "context" + stderrors "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/outbound" + "tidepool/internal/store" +) + +// THE LEDGER AND THE QUEUE MUST NEVER DISAGREE PERMANENTLY. +// +// Worker.deliverSuccess marks the delivery delivered — which makes the queue +// entry TERMINAL, never re-claimed — and only then runs voteCallback, as a +// separate autocommit. Everything between those two writes is a window in which +// a crash, a cancelled context or a database blip leaves the two records saying +// different things forever, because nothing ever revisits a terminal delivery. +// +// Before 17b that was cosmetic: outbound_votes was write-side bookkeeping. +// 17b made it an INPUT to a number users read, so each of these windows now has +// a permanent, user-visible price: +// +// (a) Like delivered, SetDeliveredState fails → row 'pending' forever, the +// vote sits at Lemmy uncounted-for → served total OVER-counts by one +// (b) Undo delivered, Delete fails → row 'delivered' forever, the +// peer dropped the vote → served total UNDER-counts by one +// (c) the stale-claim branch returns BEFORE the callback → a POST that +// succeeded leaves the ledger untouched +// +// Nothing self-heals them: SeedAggregates' only caller is the backfill's post +// walk, behind a config flag and a freshness window, so a wrong tally can +// outlive the incident by weeks. +// +// These tests pin the PROPERTY, not a mechanism. GREEN may make the pair atomic +// in one transaction or run an idempotent fenced callback before the terminal +// transition; either satisfies the same two assertions — a terminal delivery +// implies a settled ledger row, and the tally the user sees is right once the +// worker stops having anything to do. + +// faultyVotes fails exactly one of the two ledger writes, on demand, so a fault +// can be injected AFTER the delivery has been marked — the window under test. +type faultyVotes struct { + store.OutboundVotes + failSet bool + failDelete bool + err error +} + +func (f *faultyVotes) SetDeliveredState(ctx context.Context, voteATURI string, state store.DeliveredState) error { + if f.failSet { + return f.err + } + return f.OutboundVotes.SetDeliveredState(ctx, voteATURI, state) +} + +func (f *faultyVotes) Delete(ctx context.Context, voteATURI string) error { + if f.failDelete { + return f.err + } + return f.OutboundVotes.Delete(ctx, voteATURI) +} + +// deliveryStates returns every delivery row's state, newest last. +func deliveryStates(t *testing.T, l *lifecycle) []string { + t.Helper() + rows, err := l.db.Query(`SELECT state FROM outbound_deliveries ORDER BY seq`) + require.NoError(t, err) + defer func() { _ = rows.Close() }() + var states []string + for rows.Next() { + var state string + require.NoError(t, rows.Scan(&state)) + states = append(states, state) + } + require.NoError(t, rows.Err()) + return states +} + +// deliverTolerating runs the worker until it drains, allowing errors (a fault +// injected into the callback surfaces as one). +func deliverTolerating(t *testing.T, l *lifecycle) { + t.Helper() + for i := 0; i < 6; i++ { + worked, err := l.worker.DeliverNext(context.Background()) + if err != nil { + continue // the fault under test; the queue's own state is the assertion + } + if !worked { + return + } + } +} + +// TestDeliveredLikeNeverStrandsThePendingLedgerRow is (a). +// +// The POST succeeded — Lemmy holds the vote — and the delivery is recorded as +// delivered. If the ledger row is still 'pending' and the delivery is terminal, +// nothing will ever reconcile them: the seed will keep handing the community a +// total that includes a vote we cast on their behalf. +func TestDeliveredLikeNeverStrandsThePendingLedgerRow(t *testing.T) { + faulty := &faultyVotes{err: stderrors.New("ledger write failed")} + l := newLifecycle(t, 5, func(o *outbound.WorkerOptions) { + faulty.OutboundVotes = store.NewOutboundVotes(o.DB) + faulty.failSet = true + o.Votes = faulty + }) + ctx := context.Background() + require.NoError(t, l.agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, tpSubject), "")) + + l.castVote(t, "3lztprev00001", directionDown) + deliverTolerating(t, l) + + require.Equal(t, []string{"Dislike"}, l.sender.kinds(), + "precondition: the vote really did reach the peer — this is a fault AFTER the wire") + + states := deliveryStates(t, l) + require.Len(t, states, 1) + if states[0] == string(store.DeliveryStateDelivered) { + assert.Equal(t, string(store.DeliveredStateDelivered), l.state(t), + "a TERMINAL delivery with a 'pending' ledger row is a permanent disagreement: "+ + "the queue will never revisit it, so the vote sits at Lemmy while we account "+ + "for nothing — the served total over-counts by one, forever") + } + + // Whatever the mechanism, once the fault clears the worker must be able to + // finish the job — a delivery it can no longer claim cannot be finished. + faulty.failSet = false + deliverTolerating(t, l) + assert.Equal(t, string(store.DeliveredStateDelivered), l.state(t), + "after the blip passes the ledger must catch up: the vote IS at Lemmy") + + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 2)) + up, down, found := counts(t, l.db, tpSubject) + require.True(t, found) + assert.Equal(t, 3, up) + assert.Equal(t, 1, down, + "and the number the community reads is right: our delivered vote is netted out") +} + +// TestDeliveredUndoNeverStrandsTheDeliveredLedgerRow is (b), the mirror. +// +// The Undo reached Lemmy, so the peer no longer holds the vote. A ledger row +// left saying 'delivered' subtracts it from every future seed — the community's +// score is one lower than the truth, permanently, and in the direction nobody +// investigates because a missing vote looks like a vote never cast. +func TestDeliveredUndoNeverStrandsTheDeliveredLedgerRow(t *testing.T) { + faulty := &faultyVotes{err: stderrors.New("ledger delete failed")} + l := newLifecycle(t, 5, func(o *outbound.WorkerOptions) { + faulty.OutboundVotes = store.NewOutboundVotes(o.DB) + o.Votes = faulty + }) + ctx := context.Background() + require.NoError(t, l.agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, tpSubject), "")) + + // The vote is cast and delivered cleanly. + l.castVote(t, "3lztprev00001", directionDown) + deliverTolerating(t, l) + require.Equal(t, string(store.DeliveredStateDelivered), l.state(t)) + + // The user withdraws it. The Undo reaches Lemmy — and the ledger delete + // fails right after the delivery is marked. + faulty.failDelete = true + l.deleteVote(t, "3lztprev00002") + deliverTolerating(t, l) + require.Equal(t, []string{"Dislike", "Undo"}, l.sender.kinds(), + "precondition: the withdrawal really did reach the peer") + + states := deliveryStates(t, l) + require.Len(t, states, 2) + if states[1] == string(store.DeliveryStateDelivered) { + assert.Equal(t, "", l.state(t), + "a TERMINAL Undo delivery with the row still 'delivered' is the mirror "+ + "disagreement: the peer dropped the vote, we subtract it forever, and the "+ + "community's score is permanently one short") + } + + faulty.failDelete = false + deliverTolerating(t, l) + assert.Equal(t, "", l.state(t), + "once the blip passes the row must go: the peer is not holding this vote") + + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 1)) + up, down, _ := counts(t, l.db, tpSubject) + assert.Equal(t, 3, up) + assert.Equal(t, 1, down, "nothing left of ours to subtract") +} + +// TestStaleClaimStillReachesTheLedger is (c). +// +// A worker whose lease expires while its POST is in flight loses the fencing +// race: MarkDelivered no-ops and deliverSuccess returns BEFORE voteCallback. The +// vote is at Lemmy and the ledger was never touched. +// +// The lease expiry is injected where it actually happens — during the send — +// rather than by editing rows around the worker, so the interleaving is the real +// one. Whether the recovery is a redelivery or a fenced callback is GREEN's +// choice; that it recovers is not. +func TestStaleClaimStillReachesTheLedger(t *testing.T) { + l := newLifecycle(t, 5) + ctx := context.Background() + require.NoError(t, l.agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, tpSubject), "")) + l.castVote(t, "3lztprev00001", directionDown) + + // While this worker is on the wire, another one re-claims the delivery: + // claimed_until IS the fencing token, so the sender's own claim is now + // stale and its MarkDelivered will not apply. + var stolen bool + l.sender.onSend = func() { + if stolen { + return + } + stolen = true + _, err := l.db.ExecContext(ctx, + `UPDATE outbound_deliveries SET claimed_until = now() + interval '2 seconds'`) + require.NoError(t, err) + } + deliverTolerating(t, l) + require.Equal(t, []string{"Dislike"}, l.sender.kinds(), + "precondition: the losing worker's POST SUCCEEDED — the vote is at Lemmy") + + // The interloper's claim lapses; the queue is free to make progress again. + l.sender.onSend = nil + _, err := l.db.ExecContext(ctx, + `UPDATE outbound_deliveries SET claimed_until = NULL, next_attempt_at = now() + WHERE state = 'pending'`) + require.NoError(t, err) + deliverTolerating(t, l) + + assert.Equal(t, string(store.DeliveredStateDelivered), l.state(t), + "a POST that succeeded must reach the ledger EVENTUALLY, whichever worker lost the "+ + "fencing race: the alternative is a vote standing at Lemmy that we never account "+ + "for, and no re-seed corrects it because the ledger is what the seed reads") + + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 2)) + _, down, _ := counts(t, l.db, tpSubject) + assert.Equal(t, 1, down, "and the served total reflects it") +} diff --git a/internal/votes/reseed_migration_test.go b/internal/votes/reseed_migration_test.go new file mode 100644 --- /dev/null +++ b/internal/votes/reseed_migration_test.go @@ -0,0 +1,167 @@ +package votes + +import ( + "context" + "database/sql" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/store" +) + +// MIGRATION 023's DATA STATEMENT. +// +// testutil.DB runs MigrateUp against an already-migrated database, so the +// migration's DELETE and recompute execute against nothing: they are +// "exercised" only in the sense that they parse. That is half of 17b's +// behaviour, and it is the half that touches rows production already has. +// +// The statement is READ FROM THE MIGRATION FILE rather than restated here. A +// copy would pass forever while the migration drifted underneath it — which is +// the failure mode this test exists to prevent, not one to reproduce. +const migrationPath = "../db/migrations/023_vote_seed_netting.sql" + +// migrationCleanupStatement extracts the WITH-scrubbed statement from the Up +// section. It deliberately does NOT run the whole Up: CREATE INDEX would fail +// against an already-migrated database, and the index is not what is untested. +func migrationCleanupStatement(t *testing.T) string { + t.Helper() + raw, err := os.ReadFile(filepath.Clean(migrationPath)) + require.NoError(t, err, "migration 023 must be readable; this test tracks the file, not a copy") + body := string(raw) + up := body + if down := strings.Index(body, "-- +goose Down"); down >= 0 { + up = body[:down] + } + start := strings.Index(up, "WITH scrubbed") + require.GreaterOrEqual(t, start, 0, + "migration 023 must still carry a WITH-scrubbed cleanup statement; if it was renamed "+ + "or removed, this test is the place to find out") + rest := up[start:] + end := strings.Index(rest, ";") + require.Greater(t, end, 0, "the cleanup statement must terminate") + return rest[:end+1] +} + +// TestMigration023RemovesOnlyOurOwnLegacyVoteRows drives the statement against +// rows of both kinds. +// +// A persona-authored vote_events row is unwritable since 17a's voter probe, so +// any surviving row predates the guard and is unconditionally garbage — but the +// recompute that follows it must re-derive the affected subjects' totals from +// what is LEFT, per direction, and must not touch any other subject: an +// unqualified recompute row-locks every aggregate and restamps updated_at, +// which migration 014's stats watermark reads as "due" and turns into a +// full re-emit sweep of record rewrites and firehose traffic. +func TestMigration023RemovesOnlyOurOwnLegacyVoteRows(t *testing.T) { + database := testDB(t) + ctx := context.Background() + objects := store.NewAPObjects(database) + statement := migrationCleanupStatement(t) + + // One of our personas, and a genuine Lemmy human. + const personaActor = "https://coves.social/ap/actor/did:plc:legacypersona0001" + _, err := store.NewAPActors(database).Create(ctx, store.APActor{ + DID: "did:plc:legacypersona0001", + Kind: store.ActorTypePerson, + ActorID: personaActor, + NormalizedOrigin: "coves.social", + LocalPart: "legacypersona", + RSAKeySealed: []byte{0x01, 0x02, 0x03}, + RSAKeyVersion: 1, + PublicKeyPEM: "-----BEGIN PUBLIC KEY-----\nMIIB\n-----END PUBLIC KEY-----\n", + }) + require.NoError(t, err) + + // The AFFECTED subject: a genuine inbound up-vote, plus a legacy persona + // DOWN-vote from before the guard existed. + agg, _ := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3lzmigsubject01") + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + insertLegacyVote(t, database, personaActor, subjectPost, directionDown) + // The pre-guard world counted it, so the served totals include it. + _, err = database.ExecContext(ctx, ` + UPDATE vote_aggregates SET seeded_upvotes = 4, seeded_downvotes = 2, + upvotes = 4 + 1, downvotes = 2 + 1 WHERE subject_ap_id = $1`, subjectPost) + require.NoError(t, err) + + // The UNAFFECTED subject: genuine votes only. Nothing here may move. + other := "https://lemmy.world/post/900" + bridgeSubject(t, objects, other, "3lzmigsubject02") + require.NoError(t, agg.ApplyVote(ctx, dislike(activityID(t, 2), voterBob, other), "")) + otherStamp := updatedAt(t, database, other) + otherUp, otherDown, _ := counts(t, database, other) + + // --- Run the migration's statement. + _, err = database.ExecContext(ctx, statement) + require.NoError(t, err, "the cleanup statement must run against real rows") + + assert.Zero(t, voteRowCount(t, database, personaActor), + "a persona-authored vote row is unwritable today, so any that exists is garbage from "+ + "before the guard — it must not survive into the new accounting, where it would "+ + "be counted once as a live inbound vote AND once as a delivered outbound one") + assert.Equal(t, 1, voteRowCount(t, database, voterAlice), + "the genuine voter's row must be untouched: this DELETE is scoped to ap_actors, and "+ + "a wider predicate would erase the community's real votes") + + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 5, up, "served = seeded 4 + the ONE surviving live up-vote") + assert.Equal(t, 2, down, + "and the down column drops to its baseline: the only live down-vote was ours, and "+ + "the recompute re-derives per direction rather than subtracting a total") + + nowUp, nowDown, _ := counts(t, database, other) + assert.Equal(t, otherUp, nowUp, "an unaffected subject's totals must not move") + assert.Equal(t, otherDown, nowDown) + assert.Equal(t, otherStamp, updatedAt(t, database, other), + "nor its updated_at: the recompute is SCOPED to the subjects the DELETE touched, "+ + "because restamping every aggregate is what migration 014's stats watermark "+ + "reads as 'due' — a full re-emit sweep of the whole corpus") + + // --- Idempotence: a re-run (a redeployed migration, a manual re-apply) + // must be a true no-op, not a second subtraction. + beforeUp, beforeDown, _ := counts(t, database, subjectPost) + affectedStamp := updatedAt(t, database, subjectPost) + _, err = database.ExecContext(ctx, statement) + require.NoError(t, err) + + againUp, againDown, _ := counts(t, database, subjectPost) + assert.Equal(t, beforeUp, againUp, "a second run changes nothing: there is nothing left to scrub") + assert.Equal(t, beforeDown, againDown) + assert.Equal(t, affectedStamp, updatedAt(t, database, subjectPost), + "and it restamps nothing — an empty DELETE returns no subjects, so the recompute "+ + "matches no rows at all") +} + +// insertLegacyVote writes a vote_events row directly: the voter probe refuses +// to create one for a persona, which is precisely why the migration exists. +func insertLegacyVote(t *testing.T, database *sql.DB, voter, subject, 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)`, + "https://coves.social/ap/activity/legacy-"+direction, voter, subject, direction) + require.NoError(t, err) +} + +func voteRowCount(t *testing.T, database *sql.DB, voter string) int { + t.Helper() + var n int + require.NoError(t, database.QueryRow( + `SELECT COUNT(*) FROM vote_events WHERE voter_ap_id = $1`, voter).Scan(&n)) + return n +} + +func updatedAt(t *testing.T, database *sql.DB, subject string) time.Time { + t.Helper() + var stamp time.Time + require.NoError(t, database.QueryRow( + `SELECT updated_at FROM vote_aggregates WHERE subject_ap_id = $1`, subject).Scan(&stamp)) + return stamp +} diff --git a/internal/votes/reseed_temporal_test.go b/internal/votes/reseed_temporal_test.go new file mode 100644 --- /dev/null +++ b/internal/votes/reseed_temporal_test.go @@ -0,0 +1,720 @@ +package votes + +import ( + "context" + "crypto/rsa" + "database/sql" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/consume" + "tidepool/internal/echo" + "tidepool/internal/outbound" + "tidepool/internal/store" +) + +// TASK 17b, THE TEMPORAL CASES. +// +// The states that break this arithmetic are not the ones a fixture author +// chooses; they are the ones the WRITE PATH produces transiently. A hand-written +// row is always in some steady state somebody decided on — a clean 'delivered', +// a clean 'pending' — and asserting against it proves only that the SQL matches +// the string that was typed. The fixture nobody writes is the one where the +// vote's history has MORE THAN ONE STEP. +// +// So the lifecycle below is driven through the REAL path end to end: +// +// consume.Dispatcher.HandleEvent (a vote commit off the firehose) +// → applyVoteWrite → outbound_votes 'pending' + one activity +// outbound.Worker.DeliverNext → voteCallback → SetDeliveredState('delivered') +// consume.Dispatcher.HandleEvent (the vote record's DELETE) +// → applyVoteDelete → re-upsert PRESERVING 'delivered' + Undo enqueued +// outbound.Worker.DeliverNext → voteCallback → the row is DELETED +// +// Every state these tests assert against is a state that path actually reached. + +const ( + tpUserOrigin = "https://coves.social" + tpNativeDID = "did:plc:temporalnative001" + tpNativeHandle = "temporal.coves.social" + tpCommunityDID = "did:plc:temporalcommunity" + tpCommunityAP = "https://lemmy.world/c/technology" + tpSubject = "https://lemmy.world/post/700" + tpSubjectRKey = "3lztemporalpost" + tpVoteRKey = "3lztemporalvote" +) + +// tpRSAKey is generated once: the signing key is not under test and RSA keygen +// dominates the runtime of every test in this file otherwise. +var ( + tpRSAKey *rsa.PrivateKey + tpRSAKeyOnce sync.Once +) + +// --------------------------------------------------------------------------- +// The real write path, wired +// --------------------------------------------------------------------------- + +// tpMinter is the personas seam: it really writes the ap_actors row, because +// the delivery worker's consent recheck reads it. +type tpMinter struct{ db *sql.DB } + +func (m tpMinter) CreateActorForDID(ctx context.Context, did, _ string) (*store.APActor, error) { + actors := store.NewAPActors(m.db) + if existing, err := actors.GetByDID(ctx, did); err == nil { + return existing, nil + } + tpRSAKeyOnce.Do(func() { + key, err := ap.GenerateRSAKey() + if err != nil { + panic(err) + } + tpRSAKey = key + }) + return actors.Create(ctx, store.APActor{ + DID: did, + Kind: store.ActorTypePerson, + ActorID: tpUserOrigin + "/ap/actor/" + did, + NormalizedOrigin: "coves.social", + LocalPart: "temporal", + RSAKeySealed: []byte{0x01, 0x02, 0x03}, + RSAKeyVersion: 1, + PublicKeyPEM: "-----BEGIN PUBLIC KEY-----\nMIIB\n-----END PUBLIC KEY-----\n", + }) +} + +type tpResolver struct{} + +func (tpResolver) ResolveDIDHandle(context.Context, string) (string, error) { + return tpNativeHandle, nil +} + +type tpSigners struct{} + +func (tpSigners) SignerFor(context.Context, string) (*ap.Signer, error) { + tpRSAKeyOnce.Do(func() { + key, err := ap.GenerateRSAKey() + if err != nil { + panic(err) + } + tpRSAKey = key + }) + return ap.NewSigner(tpUserOrigin+"/ap/actor/"+tpNativeDID+"#main-key", tpRSAKey), nil +} + +type tpInbox struct{} + +func (tpInbox) ResolveInbox(context.Context, string) (string, error) { + return "https://lemmy.world/inbox", nil +} + +// tpSender is the wire. It records what went out and can be told to fail, which +// is how a delivery is driven to POISON — the state no happy-path fixture ever +// produces and the one that makes a subtraction permanent. +type tpSender struct { + mu sync.Mutex + sent []string + err error + onSend func() +} + +func (s *tpSender) SendActivityAs(_ context.Context, _ *ap.Signer, _ string, activity any) error { + s.mu.Lock() + hook, err := s.onSend, s.err + s.mu.Unlock() + // The hook runs WHILE the request is on the wire — the only window in which + // a lease can expire under a worker that is about to succeed. + if hook != nil { + hook() + } + s.mu.Lock() + defer s.mu.Unlock() + if err != nil { + return err + } + raw, _ := json.Marshal(activity) + var doc map[string]any + _ = json.Unmarshal(raw, &doc) + kind, _ := doc["type"].(string) + s.sent = append(s.sent, kind) + return nil +} + +func (s *tpSender) fail(err error) { + s.mu.Lock() + defer s.mu.Unlock() + s.err = err +} + +func (s *tpSender) succeed() { + s.mu.Lock() + defer s.mu.Unlock() + s.err = nil +} + +func (s *tpSender) kinds() []string { + s.mu.Lock() + defer s.mu.Unlock() + return append([]string(nil), s.sent...) +} + +// lifecycle is one native persona's vote, drivable through its whole life. +type lifecycle struct { + db *sql.DB + agg *Aggregator + dispatcher *consume.Dispatcher + worker *outbound.Worker + sender *tpSender + votes store.OutboundVotes + subjectURI string + logs *tpLogBuffer +} + +// tpLogBuffer is a concurrency-safe log sink (the clamp Warn is asserted). +type tpLogBuffer struct { + mu sync.Mutex + buf []byte +} + +func (b *tpLogBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + b.buf = append(b.buf, p...) + return len(p), nil +} + +func (b *tpLogBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return string(b.buf) +} + +// newLifecycle wires the real consumer, the real enqueuer and the real delivery +// worker over one test database. maxAttempts is the delivery cap: 1 makes the +// first failure poison, which is how the permanent states are reached. +func newLifecycle(t *testing.T, maxAttempts int, workerOpts ...func(*outbound.WorkerOptions)) *lifecycle { + t.Helper() + ctx := context.Background() + database := testDB(t) + + objects := store.NewAPObjects(database) + // The Lemmy post being voted on, bound to the community the consumer + // resolves the vote's target through. + _, err := store.NewCommunities(database).UpsertCommunity(ctx, store.Community{ + APGroupID: tpCommunityAP, + DID: tpCommunityDID, + PreferredUsername: "technology", + Instance: testInstance, + FollowState: store.FollowStateAccepted, + }) + require.NoError(t, err) + mapping, err := objects.PutMapping(ctx, store.APObjectMapping{ + APID: tpSubject, + APType: "Page", + OriginInstance: testInstance, + Origin: store.OriginFediverse, + DID: tpCommunityDID, + CommunityDID: tpCommunityDID, + Collection: testCollection, + RKey: tpSubjectRKey, + CID: testCID, + }) + require.NoError(t, err) + + logs := &tpLogBuffer{} + logger := slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug})) + + probe, err := echo.New(echo.Options{ + Objects: objects, + OutboundObjects: store.NewOutboundObjects(database), + Activities: store.NewOutboundActivities(database), + Actors: store.NewAPActors(database), + }) + require.NoError(t, err) + agg, err := NewAggregator(database, objects, store.NewCommunities(database), + &fakeRecords{records: map[string]map[string]any{}}, probe, logger) + require.NoError(t, err) + + enqueuer, err := outbound.NewEnqueuer(outbound.EnqueuerOptions{ + DB: database, + Translator: outbound.NewTranslator(tpUserOrigin), + Inboxes: tpInbox{}, + Actors: store.NewAPActors(database), + UserOrigin: tpUserOrigin, + Logger: logger, + }) + require.NoError(t, err) + + dispatcher, err := consume.NewDispatcher(consume.Options{ + DB: database, + Actors: tpMinter{db: database}, + Enqueuer: enqueuer, + Resolver: tpResolver{}, + UserOrigin: tpUserOrigin, + Logger: logger, + }) + require.NoError(t, err) + + sender := &tpSender{} + workerOptions := outbound.WorkerOptions{ + DB: database, + Actors: store.NewAPActors(database), + Objects: store.NewOutboundObjects(database), + Votes: store.NewOutboundVotes(database), + Signers: tpSigners{}, + Inboxes: tpInbox{}, + Sender: sender, + Lease: time.Minute, + MaxAttempts: maxAttempts, + BackoffBase: time.Nanosecond, + Logger: logger, + } + for _, apply := range workerOpts { + apply(&workerOptions) + } + worker, err := outbound.NewWorker(workerOptions) + require.NoError(t, err) + + return &lifecycle{ + db: database, agg: agg, dispatcher: dispatcher, worker: worker, + sender: sender, votes: store.NewOutboundVotes(database), + subjectURI: mapping.ATURI, logs: logs, + } +} + +// castVote drives a vote COMMIT through the real consumer. +func (l *lifecycle) castVote(t *testing.T, rev, direction string) { + t.Helper() + frame := fmt.Sprintf( + `{"did":%q,"time_us":9500,"kind":"commit","commit":{"rev":%q,"operation":"create",`+ + `"collection":"social.coves.feed.vote","rkey":%q,"cid":%q,`+ + `"record":{"$type":"social.coves.feed.vote","subject":{"uri":%q,"cid":%q},`+ + `"direction":%q,"createdAt":"2026-08-13T10:00:00.000Z"}}}`, + tpNativeDID, rev, tpVoteRKey, testCID, l.subjectURI, testCID, direction) + l.handle(t, frame) +} + +// deleteVote drives the vote record's DELETE through the real consumer — the +// step that enqueues the Undo while the row keeps its delivered state. +func (l *lifecycle) deleteVote(t *testing.T, rev string) { + t.Helper() + frame := fmt.Sprintf( + `{"did":%q,"time_us":9600,"kind":"commit","commit":{"rev":%q,"operation":"delete",`+ + `"collection":"social.coves.feed.vote","rkey":%q}}`, + tpNativeDID, rev, tpVoteRKey) + l.handle(t, frame) +} + +func (l *lifecycle) handle(t *testing.T, frame string) { + t.Helper() + var event consume.JetstreamEvent + require.NoError(t, json.Unmarshal([]byte(frame), &event)) + require.NoError(t, l.dispatcher.HandleEvent(context.Background(), &event)) +} + +// deliver runs the delivery worker until its queue drains. +func (l *lifecycle) deliver(t *testing.T) { + t.Helper() + for i := 0; i < 20; i++ { + worked, err := l.worker.DeliverNext(context.Background()) + require.NoError(t, err, "DeliverNext must not error") + if !worked { + return + } + } + t.Fatal("the delivery worker did not drain") +} + +// state reports the vote row's delivered_state, or "" when the row is gone. +func (l *lifecycle) state(t *testing.T) string { + t.Helper() + var s string + err := l.db.QueryRow( + `SELECT delivered_state FROM outbound_votes WHERE vote_at_uri = $1`, + "at://"+tpNativeDID+"/social.coves.feed.vote/"+tpVoteRKey).Scan(&s) + if err == sql.ErrNoRows { + return "" + } + require.NoError(t, err) + return s +} + +// --------------------------------------------------------------------------- +// T1 + T2 — the same vote at two points in its life +// --------------------------------------------------------------------------- + +// TestDeliveredVoteIsSubtractedUntilItsUndoIsDelivered pins the pair that only +// makes sense together: while the withdrawal is IN FLIGHT the origin still +// counts the vote, and once it lands the origin stops — and the served total +// must move by exactly one, at exactly that moment. +func TestDeliveredVoteIsSubtractedUntilItsUndoIsDelivered(t *testing.T) { + l := newLifecycle(t, 5) + ctx := context.Background() + + // A Lemmy human's live inbound up-vote, so the fixture is mixed and the two + // directions are unequal. + require.NoError(t, l.agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, tpSubject), "")) + + // Our persona casts a DOWN-vote, and it is delivered for real. + l.castVote(t, "3lztprev00001", directionDown) + require.Equal(t, string(store.DeliveredStatePending), l.state(t), + "the consumer records INTENT; only the wire makes it delivered") + l.deliver(t) + require.Equal(t, string(store.DeliveredStateDelivered), l.state(t)) + require.Equal(t, []string{"Dislike"}, l.sender.kinds()) + + // T1: the user deletes their vote record. An Undo is enqueued, and the row + // KEEPS delivered — applyVoteDelete re-upserts *stored on purpose. + l.deleteVote(t, "3lztprev00002") + require.Equal(t, string(store.DeliveredStateDelivered), l.state(t), + "the withdrawal is in flight: Lemmy has not processed it, so the row must still "+ + "say the peer holds this vote") + + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 2)) + up, down, found := counts(t, l.db, tpSubject) + require.True(t, found) + assert.Equal(t, 3, up, "the inbound up-vote nets out of the baseline and back into the total") + assert.Equal(t, 1, down, + "T1: an Undo IN FLIGHT changes nothing about what the origin counts — its 2 still "+ + "includes our vote, so it must still be subtracted") + + // T2: the Undo delivers. voteCallback deletes the row, and the origin's + // own count drops by one at the same moment. + l.deliver(t) + require.Equal(t, []string{"Dislike", "Undo"}, l.sender.kinds()) + require.Equal(t, "", l.state(t), "a withdrawn vote leaves no row to subtract") + + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 1)) + up, down, _ = counts(t, l.db, tpSubject) + assert.Equal(t, 3, up) + assert.Equal(t, 1, down, + "T2: the origin's own count dropped by one (2→1) when it processed the withdrawal, "+ + "and our subtrahend dropped with it — so the SERVED total holds at 1. Subtracting "+ + "a row that is already gone would move it to 0 instead") +} + +// --------------------------------------------------------------------------- +// T3 — pending is never subtracted, however it got there +// --------------------------------------------------------------------------- + +// TestPendingVotesAreNeverSubtracted covers both flavours of pending: a first +// delivery still in flight, and a POISONED one that will never advance +// (voteCallback is the only writer of 'delivered', so a poisoned delivery +// leaves the row pending forever). +// +// Subtracting either would UNDERCOUNT: Lemmy cannot be holding a vote it never +// received. +func TestPendingVotesAreNeverSubtracted(t *testing.T) { + ctx := context.Background() + + t.Run("first delivery still in flight", func(t *testing.T) { + l := newLifecycle(t, 5) + require.NoError(t, l.agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, tpSubject), "")) + l.castVote(t, "3lztprev00001", directionDown) + require.Equal(t, string(store.DeliveredStatePending), l.state(t)) + + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 2)) + up, down, found := counts(t, l.db, tpSubject) + require.True(t, found) + assert.Equal(t, 3, up) + assert.Equal(t, 2, down, + "a vote the peer has not received is not in the peer's count: subtracting it "+ + "would show the community one fewer downvote than it has") + }) + + t.Run("poisoned delivery, pending forever", func(t *testing.T) { + l := newLifecycle(t, 1) // one attempt, so the first failure poisons + require.NoError(t, l.agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, tpSubject), "")) + l.castVote(t, "3lztprev00001", directionDown) + l.sender.fail(fmt.Errorf("lemmy is unreachable")) + l.deliver(t) + + assert.Equal(t, string(store.DeliveredStatePending), l.state(t), + "nothing but delivery success writes 'delivered', so a poisoned delivery leaves "+ + "the row pending — permanently") + var poisoned int + require.NoError(t, l.db.QueryRow( + `SELECT COUNT(*) FROM outbound_deliveries WHERE state = 'poisoned'`).Scan(&poisoned)) + require.Equal(t, 1, poisoned, "precondition: the delivery really did poison") + + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 2)) + _, down, _ := counts(t, l.db, tpSubject) + assert.Equal(t, 2, down, + "permanence cuts the other way here: this vote will NEVER reach Lemmy, so it "+ + "must never be subtracted from Lemmy's count") + + // KNOWN AMBIGUITY, recorded rather than resolved: a poisoned delivery + // does not prove what the peer holds. A timeout AFTER Lemmy applied the + // vote leaves us treating a present vote as absent, and no arithmetic + // here can tell the two apart — only reconciliation against the origin + // can (17e). What this pins is that the uncertainty stays QUERYABLE: + // the poisoned row keeps its activity id and error class, so a + // reconciler has something to walk. Erasing it would turn "we do not + // know" into "it never happened". + var activityID, errorClass string + require.NoError(t, l.db.QueryRow( + `SELECT activity_id, last_error_class FROM outbound_deliveries WHERE state = 'poisoned'`). + Scan(&activityID, &errorClass)) + assert.NotEmpty(t, activityID, + "the poisoned delivery must remain identifiable: this row is the only record that "+ + "we do not know whether the peer holds this vote") + assert.NotEmpty(t, errorClass, "with why it died, so a reconciler can triage it") + }) +} + +// --------------------------------------------------------------------------- +// T4 — a delivered vote whose UNDO poisoned is subtracted FOREVER +// --------------------------------------------------------------------------- + +// TestDeliveredVoteWithPoisonedUndoSubtractsForever is the case that decides the +// predicate's SHAPE. Nobody ever advances this row: the vote is delivered, the +// withdrawal is dead, and Lemmy holds the vote permanently. +// +// It is why decision 16 bans queue-history arithmetic. "Delivered Likes minus +// delivered Undos" gets this row wrong for as long as the post exists, and the +// error is invisible — the tally is simply one off, forever, with nothing to +// point at. +func TestDeliveredVoteWithPoisonedUndoSubtractsForever(t *testing.T) { + l := newLifecycle(t, 1) + ctx := context.Background() + require.NoError(t, l.agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, tpSubject), "")) + + l.castVote(t, "3lztprev00001", directionDown) + l.deliver(t) + require.Equal(t, string(store.DeliveredStateDelivered), l.state(t)) + + // The withdrawal is enqueued and then dies on the wire. + l.deleteVote(t, "3lztprev00002") + l.sender.fail(fmt.Errorf("lemmy rejected the undo")) + l.deliver(t) + + require.Equal(t, string(store.DeliveredStateDelivered), l.state(t), + "the row is untouched by a failed Undo: the peer still holds the vote") + var poisoned int + require.NoError(t, l.db.QueryRow( + `SELECT COUNT(*) FROM outbound_deliveries WHERE state = 'poisoned'`).Scan(&poisoned)) + require.Equal(t, 1, poisoned, "precondition: the Undo really did poison") + + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 2)) + up, down, _ := counts(t, l.db, tpSubject) + assert.Equal(t, 3, up) + assert.Equal(t, 1, down, + "the vote stands on the origin forever, so we subtract it forever — an implementation "+ + "that reasons from the QUEUE (a delivered Like whose Undo was also delivered) "+ + "has no way to see that the Undo died, and is permanently wrong here") + + // And a re-seed keeps saying so: this is a standing state, not an event. + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 2)) + _, down, _ = counts(t, l.db, tpSubject) + assert.Equal(t, 1, down, "still subtracted on the next backfill, and the one after that") +} + +// --------------------------------------------------------------------------- +// T5 — re-seed idempotence with an outbound vote present +// --------------------------------------------------------------------------- + +// TestReseedWithOutboundVoteIsIdempotent is the outbound mirror of +// TestReseedDoesNotDoubleCountLiveVotes. A backfill re-run with unchanged origin +// totals must leave the served total exactly where it was — the baseline is +// REPLACED by each seed, never accumulated. +func TestReseedWithOutboundVoteIsIdempotent(t *testing.T) { + l := newLifecycle(t, 5) + ctx := context.Background() + require.NoError(t, l.agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, tpSubject), "")) + l.castVote(t, "3lztprev00001", directionDown) + l.deliver(t) + require.Equal(t, string(store.DeliveredStateDelivered), l.state(t)) + + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 2)) + firstUp, firstDown, found := counts(t, l.db, tpSubject) + require.True(t, found) + require.Equal(t, 3, firstUp) + require.Equal(t, 1, firstDown) + + for i := 0; i < 3; i++ { + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 2)) + } + up, down, _ := counts(t, l.db, tpSubject) + assert.Equal(t, firstUp, up, "a re-seed with unchanged origin totals must not move the total") + assert.Equal(t, firstDown, down, + "and the subtraction must not compound: three re-seeds subtracting our one vote each "+ + "time would walk the community's score down to zero over a few backfills") +} + +// --------------------------------------------------------------------------- +// T8 — the predicate is a POSITIVE EQUALITY +// --------------------------------------------------------------------------- + +// TestUndoneRowsAreNotSubtracted pins the shape of the predicate itself. +// +// No code path writes 'undone' today, so this row is written directly — and +// that is exactly why the test exists. A negation ('not undone', '!= pending') +// reads identically to the equality on every state the system currently +// produces, so nothing else here can tell them apart. If a future policy starts +// writing 'undone' it will mean THE PEER ACCEPTED THE WITHDRAWAL — not live — +// and every negation silently inverts on the day it appears, in the direction +// that removes real votes from the community's score. +func TestUndoneRowsAreNotSubtracted(t *testing.T) { + l := newLifecycle(t, 5) + ctx := context.Background() + require.NoError(t, l.agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, tpSubject), "")) + l.castVote(t, "3lztprev00001", directionDown) + l.deliver(t) + require.Equal(t, string(store.DeliveredStateDelivered), l.state(t)) + + _, err := l.db.ExecContext(ctx, + `UPDATE outbound_votes SET delivered_state = 'undone' WHERE vote_at_uri = $1`, + "at://"+tpNativeDID+"/social.coves.feed.vote/"+tpVoteRKey) + require.NoError(t, err, "a state no writer produces today — the point is what happens WHEN one does") + + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 3, 2)) + up, down, _ := counts(t, l.db, tpSubject) + assert.Equal(t, 3, up) + assert.Equal(t, 2, down, + "'undone' will mean the peer accepted the withdrawal, so the origin's count no longer "+ + "includes our vote and there is nothing to subtract. Only a POSITIVE equality on "+ + "'delivered' gets this right without being rewritten") +} + +// --------------------------------------------------------------------------- +// T6 — the subtraction is UNREACHABLE for the subjects it would be wrong for +// --------------------------------------------------------------------------- + +// refusingTransport fails every request: no test here may reach a network. +type refusingTransport struct{} + +func (refusingTransport) RoundTrip(req *http.Request) (*http.Response, error) { + return nil, fmt.Errorf("refusing outbound request to %s: this test never touches a network", req.URL) +} + +// recordingSeedStore captures whether the seeder reached the arithmetic at all. +type recordingSeedStore struct { + mu sync.Mutex + subjects []string +} + +func (r *recordingSeedStore) SeedAggregates(_ context.Context, subjectAPID string, _, _ int) error { + r.mu.Lock() + defer r.mu.Unlock() + r.subjects = append(r.subjects, subjectAPID) + return nil +} + +func (r *recordingSeedStore) count() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.subjects) +} + +// TestSeedingNeverInvokesTheArithmeticForRefusedSubjects is the BEHAVIOURAL +// half of the guard whose shapes TestLemmyPostAPIURL pins: a refused id must +// stop the seeder before SeedAggregates is reached, and before any request is +// made. +// +// It matters because the refusal is the only thing standing between a native +// post and a subtraction that would corrupt a total nobody seeded — there is no +// check downstream. +func TestSeedingNeverInvokesTheArithmeticForRefusedSubjects(t *testing.T) { + refused := []string{ + "https://coves.social/ap/object/did:plc:ewvi7nxzyoun6zhxrhs64oiz/social.coves.community.postv2/3lznative0001", + "https://lemmy.world/comment/27485395", + "https://lemmy.world/post/49131386/replies", + } + // The transport REFUSES everything, so this stays offline even under a + // deliberately relaxed parser: the assertion is that the guard holds BEFORE + // any request is made. + recorder := &recordingSeedStore{} + seeder, err := NewLemmySeeder(recorder, &http.Client{Transport: refusingTransport{}}, "tidepool-test/0", nil) + require.NoError(t, err) + for _, apID := range refused { + _ = seeder.SeedPostCounts(context.Background(), apID) + } + assert.Zero(t, recorder.count(), + "SeedAggregates must never be invoked for a subject with no external total: the "+ + "guard is the URL shape, and it has to hold before the first packet") +} + +// --------------------------------------------------------------------------- +// R1 — the symmetry trap, mirrored +// --------------------------------------------------------------------------- + +// TestOutboundUpVoteIsSubtractedFromUpOnly closes the gap every other fixture +// in this file leaves open: they all cast DOWN votes, so an implementation that +// buckets the subtrahend into `down` unconditionally passes all of them. That is +// the same trap reseed_test.go's header warns about, reproduced in the other +// direction — a suite is only as directional as its least directional fixture. +func TestOutboundUpVoteIsSubtractedFromUpOnly(t *testing.T) { + l := newLifecycle(t, 5) + ctx := context.Background() + + // A Lemmy human's live inbound DOWN-vote, so the two directions carry + // different populations and cannot be confused for each other. + require.NoError(t, l.agg.ApplyVote(ctx, dislike(activityID(t, 1), voterAlice, tpSubject), "")) + + // Our persona's UP-vote, delivered for real. + l.castVote(t, "3lztprev00001", directionUp) + l.deliver(t) + require.Equal(t, string(store.DeliveredStateDelivered), l.state(t)) + require.Equal(t, []string{"Like"}, l.sender.kinds()) + + require.NoError(t, l.agg.SeedAggregates(ctx, tpSubject, 4, 2)) + up, down, found := counts(t, l.db, tpSubject) + require.True(t, found) + assert.Equal(t, 3, up, + "UP: the origin's 4 includes our persona's up-vote, so the bridged tally is 3") + assert.Equal(t, 2, down, + "DOWN: our vote was cast UP. Subtracting it here — or subtracting the outbound "+ + "TOTAL from both columns — moves a number no native user touched") + + seededUp, seededDown := seededCounts(t, l.db, tpSubject) + assert.Equal(t, 3, seededUp, "4 origin − 0 live inbound up − 1 delivered outbound up = 3") + assert.Equal(t, 1, seededDown, "2 origin − 1 live inbound down − 0 outbound down = 1") +} + +// TestSubtractionIsScopedToTheSeededSubject pins the `subject_ap_id = $1` +// predicate on the outbound subquery. Without it every delivered vote in the +// TABLE is subtracted from whatever subject happens to be seeded — a bridge +// with a thousand native voters would walk every backfilled post's score toward +// zero, and no fixture with one subject in the database could ever see it. +func TestSubtractionIsScopedToTheSeededSubject(t *testing.T) { + l := newLifecycle(t, 5) + ctx := context.Background() + + // Our persona's delivered DOWN-vote — on a DIFFERENT post. + l.castVote(t, "3lztprev00001", directionDown) + l.deliver(t) + require.Equal(t, string(store.DeliveredStateDelivered), l.state(t)) + + // An unrelated bridged post, with no outbound vote of ours on it at all. + other := "https://lemmy.world/post/701" + _, err := store.NewAPObjects(l.db).PutMapping(ctx, store.APObjectMapping{ + APID: other, + APType: "Page", + OriginInstance: testInstance, + Origin: store.OriginFediverse, + DID: tpCommunityDID, + CommunityDID: tpCommunityDID, + Collection: testCollection, + RKey: "3lzotherpost01", + CID: testCID, + }) + require.NoError(t, err) + require.NoError(t, l.agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, other), "")) + + require.NoError(t, l.agg.SeedAggregates(ctx, other, 5, 3)) + up, down, found := counts(t, l.db, other) + require.True(t, found) + assert.Equal(t, 5, up, "the unrelated subject keeps the origin's total") + assert.Equal(t, 3, down, + "our vote lives on ANOTHER post: subtracting it here is a table-wide subtraction "+ + "masquerading as a per-subject one, and it grows with every native voter") +} diff --git a/internal/votes/reseed_test.go b/internal/votes/reseed_test.go new file mode 100644 --- /dev/null +++ b/internal/votes/reseed_test.go @@ -0,0 +1,163 @@ +package votes + +import ( + "context" + "database/sql" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/store" +) + +// TASK 17b — WHAT THE SERVED AGGREGATE MEANS. +// +// Coves keeps native and bridged tallies in SEPARATE columns +// (Coves' bridged_upvote_count column), so Tidepool's aggregate is the +// FEDIVERSE-ONLY tally: +// +// served(subject) = api_total(subject) − { our personas' votes Lemmy currently holds } +// +// The origin's API total includes the votes TIDEPOOL ITSELF wrote back on +// behalf of native users. SeedAggregates nets out live INBOUND events (so a +// federated vote is not counted in both the baseline and the live term) but +// knows nothing about our own outbound votes — so a native user's vote is +// counted twice: once inside Coves' native column, once inside the bridged one. +// +// THE PREDICATE IS delivered_state = 'delivered', a POSITIVE EQUALITY: +// +// pending (first delivery, retrying, or poisoned) → Lemmy does not hold it +// delivered → Lemmy holds it: subtract +// delivered + Undo in flight → still subtract; applyVoteDelete +// re-upserts 'delivered' precisely +// because the peer has not processed +// the withdrawal yet +// row deleted (Undo delivered) → nothing to subtract +// delivered + POISONED Undo → subtract forever, which is why +// decision 16 bans queue-history +// arithmetic ("delivered Likes minus +// delivered Undos" gets this row +// permanently wrong) +// +// A negation ("not undone", "!= pending") reads the same today only because +// 'undone' is never written; if a future policy starts writing it, it will mean +// THE PEER ACCEPTED THE WITHDRAWAL — not live — and every negation silently +// inverts while the equality stays correct. +const ( + rsPersonaDID = "did:plc:reseedpersona0001" + rsPersonaActor = "https://coves.social/ap/actor/" + rsPersonaDID + rsVoteATURI = "at://" + rsPersonaDID + "/social.coves.interaction.vote/3lzreseedvote1" + rsCommunityDID = "did:plc:reseedcommunity01" + rsSubjectRKey = "3lzreseedpost1" +) + +// reseedDB is testDB under a name that says what these fixtures are about. +func reseedDB(t *testing.T) *sql.DB { + t.Helper() + return testDB(t) +} + +// seededCounts reads the stored BASELINE (as opposed to the served total). +func seededCounts(t *testing.T, database *sql.DB, subject string) (up, down int) { + t.Helper() + require.NoError(t, database.QueryRow(` + SELECT seeded_upvotes, seeded_downvotes FROM vote_aggregates WHERE subject_ap_id = $1`, + subject).Scan(&up, &down)) + return up, down +} + +// TestSeedNetsOurDeliveredVotesPerDirection is the OUTER CONTRACT for 17b. +// +// GIVEN a bridged Lemmy post carrying BOTH subtrahends at once — one live +// INBOUND up-vote from a Lemmy human, and one DELIVERED outbound DOWN-vote from +// one of our native personas — WHEN the origin API reports 3 up / 2 down and the +// seed runs, THEN the SERVED total is 3 up / 1 down. +// +// Every number is distinct and the two directions are unequal ON PURPOSE. The +// symmetric fixture (one inbound up, one outbound up, api 2/0) passes under at +// least three WRONG implementations — subtracting totals instead of per +// direction, bucketing the outbound subtrahend into `up` unconditionally, and +// double-subtracting then clamping at zero — so it proves nothing. This is the +// MIXED-POPULATION subject: inbound and outbound, both directions, unequal +// counts, which nobody writes because each half already has its own passing +// test. +func TestSeedNetsOurDeliveredVotesPerDirection(t *testing.T) { + database := reseedDB(t) + agg, objects := testAggregator(t, database) + ctx := context.Background() + subjectATURI := bridgeSubject(t, objects, subjectPost, rsSubjectRKey) + + // --- Subtrahend 1: a Lemmy human's live inbound UP-vote. It is in the + // origin's total AND in vote_events, so the baseline must net it out or + // the recompute counts it twice. (This half already works.) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + + // --- Subtrahend 2: our own persona's DOWN-vote, delivered. It is in the + // origin's total and nowhere else — Coves counts it in its NATIVE + // column, so leaving it in the bridged tally counts one user's single + // vote twice across the two columns. + _, err := store.NewAPActors(database).Create(ctx, store.APActor{ + DID: rsPersonaDID, + Kind: store.ActorTypePerson, + ActorID: rsPersonaActor, + NormalizedOrigin: "coves.social", + LocalPart: "reseedpersona", + RSAKeySealed: []byte{0x01, 0x02, 0x03}, + RSAKeyVersion: 1, + PublicKeyPEM: "-----BEGIN PUBLIC KEY-----\nMIIB\n-----END PUBLIC KEY-----\n", + }) + require.NoError(t, err, "mint the persona whose vote Lemmy is holding") + + outboundVotes := store.NewOutboundVotes(database) + // Written the way the write path writes it: the consumer records INTENT as + // pending, and only delivery success flips the state — a row that starts + // life 'delivered' is a state no code path produces. + _, err = outboundVotes.Upsert(ctx, store.OutboundVote{ + VoteATURI: rsVoteATURI, + ActorDID: rsPersonaDID, + SubjectATURI: subjectATURI, + SubjectAPID: subjectPost, + CommunityDID: rsCommunityDID, + Direction: directionDown, + CurrentActivityID: "https://coves.social/ap/activity/reseed-dislike", + DeliveredState: store.DeliveredStatePending, + }) + require.NoError(t, err) + require.NoError(t, outboundVotes.SetDeliveredState(ctx, rsVoteATURI, store.DeliveredStateDelivered), + "delivery success is what makes Lemmy the holder of this vote") + + // --- The origin's public API, which counts BOTH of the above. + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 3, 2)) + + // --- THE CONTRACT: the served total is the fediverse-only tally. + up, down, found := counts(t, database, subjectPost) + require.True(t, found, "the seed must create the aggregate") + assert.Equal(t, 3, up, + "UP: the origin's 3 includes the Lemmy human's live vote, which the baseline nets "+ + "out and the recompute adds back — 3 stays 3, and no outbound vote may be "+ + "subtracted from a direction it was never cast in") + assert.Equal(t, 1, down, + "DOWN: the origin's 2 includes OUR persona's delivered down-vote. Coves already "+ + "counts that vote in its native column, so the bridged tally must be 1 — "+ + "leaving it at 2 counts one person's one vote twice in the UI") + + // --- Secondary (the contract is the served number above): where the + // subtraction lands. The baseline is its natural home — it is computed + // once per seed, while the served total is recomputed on every single + // inbound vote. + seededUp, seededDown := seededCounts(t, database, subjectPost) + assert.Equal(t, 2, seededUp, "3 origin − 1 live inbound = 2 baseline up") + assert.Equal(t, 1, seededDown, "2 origin − 0 live inbound − 1 delivered outbound = 1 baseline down") + + // --- And it SURVIVES the next recompute. recomputeAggregate runs on every + // ApplyVote, so a fix that patched the served columns after the seed + // instead of the baseline would be silently undone by the next vote to + // arrive — minutes later, with nothing to connect the drift to the seed. + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 2), voterBob, subjectPost), "")) + up, down, _ = counts(t, database, subjectPost) + assert.Equal(t, 4, up, "a second Lemmy human's up-vote stacks on the baseline") + assert.Equal(t, 1, down, + "and our persona's vote stays subtracted: the subtraction must live somewhere a "+ + "recompute preserves, not in the served columns it overwrites") +} diff --git a/internal/votes/seed_test.go b/internal/votes/seed_test.go --- a/internal/votes/seed_test.go +++ b/internal/votes/seed_test.go @@ -26,6 +26,18 @@ "https://lemmy.world/api/v3/post?id=49131386", true}, {"http with port (tests)", "http://127.0.0.1:8080/post/7", "http://127.0.0.1:8080/api/v3/post?id=7", true}, {"comment", "https://lemmy.world/comment/123", "", false}, + // 17b: the subtraction of our own delivered votes is only correct for a + // subject whose ORIGIN total includes them — a Lemmy post. For a NATIVE + // post the origin is us: Coves holds the native tally in its own column, + // there is no external total to net against, and subtracting would + // corrupt a number nobody seeded. Nothing enforces that with a check; + // it is enforced HERE, by the shapes this parser refuses. + {"a native post's AP id", + "https://coves.social/ap/object/did:plc:ewvi7nxzyoun6zhxrhs64oiz/social.coves.community.postv2/3lznative0001", + "", false}, + {"a native object on a vanity origin", + "https://vanity.example/ap/object/did:plc:ewvi7nxzyoun6zhxrhs64oiz/social.coves.community.postv2/3lznative0002", + "", false}, {"non-numeric id", "https://lemmy.world/post/abc", "", false}, {"trailing path", "https://lemmy.world/post/1/extra", "", false}, {"empty id", "https://lemmy.world/post/", "", false}, diff --git a/internal/votes/votes_test.go b/internal/votes/votes_test.go --- a/internal/votes/votes_test.go +++ b/internal/votes/votes_test.go @@ -38,13 +38,41 @@ voterBob = "https://lemmy.zip/u/bob" voterCarol = "https://sh.itjust.works/u/carol" ) -// testDB returns a migrated connection with the vote tables (and the -// ap_objects spine the aggregator resolves subjects through) truncated. +// voteTablesToTruncate is the ONE list every helper in this package starts +// from. There were four divergent lists before, and the omission that mattered +// was invisible: 17b made outbound_votes an INPUT to the seed, so a leftover +// delivered row silently subtracted one from every later test that seeded that +// subject — green on the first run, red on the second. A helper that starts +// from this list gains the next shared-state input automatically instead of +// waiting for someone to notice a discrepancy four places at once. +// +// Truncating more than a test needs is free; truncating less is a bug that +// surfaces somewhere else, later, as an off-by-one. +var voteTablesToTruncate = []string{ + // The aggregate spine. + "vote_events", "vote_aggregates", "ap_objects", "communities", + // The outbound side: an input to the seed since 17b. + "outbound_votes", "outbound_deliveries", "outbound_activities", "outbound_objects", + // Identity: ap_actors is what the voter probe reads, bridged_actors is what + // it must NOT read. + "ap_actors", "bridged_actors", + // Consumer state, for the tests that drive the real write path. + "federation_prefs", "jetstream_record_revs", "jetstream_dead_letters", + "consumer_cursors", "repo_state", +} + +// truncateVoteTables clears everything in voteTablesToTruncate. +func truncateVoteTables(t *testing.T, database *sql.DB) { + t.Helper() + testutil.Truncate(t, database, voteTablesToTruncate...) +} + +// testDB returns a migrated connection with every table this package's +// fixtures touch truncated. func testDB(t *testing.T) *sql.DB { t.Helper() database := testutil.DB(t) - testutil.Truncate(t, database, - "vote_events", "vote_aggregates", "ap_objects", "communities") + truncateVoteTables(t, database) return database } diff --git a/tests/e2e/bridge_test.go b/tests/e2e/bridge_test.go --- a/tests/e2e/bridge_test.go +++ b/tests/e2e/bridge_test.go @@ -313,9 +313,9 @@ } // Scenario 5: votes — on posts AND comments — update the getVoteAggregates // side channel and NEVER appear as records on the firehose (PLAN.md locked -// decision 7). Covers the full lifecycle: upvote, flip to downvote -// (Undo{Like} + Dislike), retract (bare Undo), and a comment vote (task -// 07's reply.root community-binding path). +// decision 7). Covers the full lifecycle: upvote, flip to downvote (a BARE +// Dislike), retract (an Undo carrying a reconstructed inner vote), and a +// comment vote (task 07's reply.root community-binding path). func TestVotes_SideChannelOnly(t *testing.T) { h := newHarness(t) community, sub := setupSubscribedCommunity(t, h, "vote") @@ -343,12 +343,16 @@ // Upvote → aggregates show it. voter.likePost(t, post.ID, 1) awaitAggregates(t, h, postURI, 1, 0) - // Flip to downvote → Lemmy sends Undo{Like} + Dislike; the aggregator - // must retract the upvote and apply the downvote. + // Flip to downvote → Lemmy sends a BARE Dislike with no Undo at all + // (measured against the pinned 0.19; see votes/aggregator.go). ApplyVote's + // supersede marks the prior live upvote undone and applies the downvote. voter.likePost(t, post.ID, -1) awaitAggregates(t, h, postURI, 0, 1) - // Retract → a bare Undo{Dislike}; back to zero on both sides. + // Retract → an Undo whose inner vote is RECONSTRUCTED: a freshly generated + // activity id, and typed "Like" even though the live vote is a Dislike. + // That is why the aggregator keys retraction on the voter, not the id. + // Back to zero on both sides. voter.likePost(t, post.ID, 0) awaitAggregates(t, h, postURI, 0, 0)