diff --git a/FOLLOWUPS.md b/FOLLOWUPS.md index a3933c3..4571bd3 100644 --- a/FOLLOWUPS.md +++ b/FOLLOWUPS.md @@ -113,6 +113,124 @@ harness, **(relay)** by task 09's relay pipeline. translation is records→Like. Symmetric records would let frontends drop the `getVoteAggregates` XRPC for everything except historical baselines. + - **FINAL direction (2026-07-10, superseding the community-repo-records + lean above): votes never become records — locked decision 7 stands.** + Bridged counts ride the CONTENT record instead: an optional + `bridgedStats {upvotes, downvotes, asOf}` field on post (and + comment) records, written at materialization and refreshed by a + debounced sweeper as `vote_aggregates` rows change; Coves folds the + counts into its denormalized counters and score. Rationale: every + per-vote record would be authored and signed by the bridge anyway, + so itemizing votes adds ZERO verifiability over a bridge-asserted + aggregate — same authority either way. The field needs no new + collection and no voter-identity/DID questions (voter identity now + never leaves the bridge, mooting the privacy question), keeps the + four-collection e2e whitelist intact, and turns firehose load from + one commit per vote into one debounced record update per subject. + `getVoteAggregates` demotes to legacy/debug once Coves consumes the + field. **Correction (was wrongly called "no Coves trust-model + change"): `bridgedStats` IS a new trust surface** — a bridge-asserted + vote count Coves folds into its own score — so Coves is gaining a + PROVENANCE GATE in a parallel fix (only records from the bridge's DID + may carry it; a self-authored `bridgedStats` must be ignored). That + gate is the trust-model change this design does require; what it + avoids is a NEW COLLECTION and per-voter records. + - **Coves-side facts (2026-07-10 investigation) that shaped this:** + Coves' post firehose consumer silently DROPS update operations + (update handler marked future work) — so a bridged Lemmy post EDIT + never re-indexes today (title/content/cid go stale in Coves); the + update handler this design needs closes a gap Coves had regardless. + Nothing in Coves keys or validates on `posts.cid` for POSTS (votes + and viewer-state match by URI), so stats-driven CID churn on a post + is safe; unknown record fields parse harmlessly (no lexicon + validation on ingest). **Correction (the earlier blanket "CID churn + is safe" was wrong for COMMENTS): Coves' comment consumer DOES + validate reply strongRef CIDs** — it treats a comment's reply + root/parent CIDs as immutable across updates and rejects a changed + ref as thread hijacking. A stats stamp on a PARENT post churns that + parent's CID, so a later comment edit that re-resolved its reply refs + would be rejected permanently. Fixed on the bridge side: a comment + rebuild now carries the STORED reply refs forward verbatim (exactly + like `bridgedStats`) instead of re-resolving them + (`TestCommentEditCarriesReplyRefsForward`); the create path still + resolves fresh. `score` is a plain column recomputed imperatively at + each vote-consumer write site — making it INCLUSIVE (native + + bridged) at every recompute site leaves the hot/top sort expressions, + the top-cursor keyset, and `idx_posts_community_score` untouched. The + comment consumer already has an update path to ride. + - **Baseline/backfill is subsumed by the field**: `bridgedStats` is + the unified baseline+live surface (tidepool's `seeded_* + live` + recompute feeds it), so external consumers never fold two sources + and the re-seed double-count hazard collapses into tidepool's own + aggregates. ~~(Flag, still open and independent of this design: the + CURRENT seeded+live recompute may already double-count on backfill + re-seed — a re-seeded baseline includes federated votes that are + also live `vote_events` rows. Verify.)~~ **Verified real and fixed + (2026-07-10)**: `SeedAggregates` now stores the baseline NET of the + subject's live `vote_events` counts (per direction, clamped at + zero, computed under the aggregate row lock), so served totals + equal the origin's counts at seed time and every re-seed converges + to the origin's truth instead of compounding + (`TestReseedDoesNotDoubleCountLiveVotes`). Residual, documented on + the method: a vote federating between the origin API fetch and the + seed tx under-counts by one until the next re-seed (self-healing; + the old code's over-count was permanent and compounding). + - **Write-back interaction**: echo suppression is still required — + once write-back ships, Lemmy announces the bridge's own + written-back Likes back to it, and the aggregator must skip + bridge-managed voters or the counts inflate; post-write-back + re-seeds must additionally subtract the bridge's own written-back + tally (Lemmy's totals will include it). Both are entirely + tidepool-side now. Per-voter provenance for write-back comes from + native Coves vote records, which is where it always lived. + - Declined alternatives, for the record: per-voter repos (a permanent + public did:plc per drive-by voter, and every first-time voter's + vote queuing behind the MintGate; claiming stays consistent — + claimed users' posts already live in community repos, so "claiming + doesn't relocate your content" is existing precedent) and + community-repo vote records with a `voter` field (a real Coves + trust-model change: its vote consumer hard-codes voter = repo DID, + `votes.voter_did` has a `^did:` check constraint a raw AP IRI + fails, and its create/delete-only consumer would silently ignore a + deterministic-rkey flip-as-update — all to buy nothing the + bridge-signed aggregate doesn't already provide). Privacy note kept + for posterity: Lemmy votes are public enough to bridge (Bridgy + Fed's stance, snarfed/bridgy-fed#372; Lemmy federates voter + identity to every peer) — but the chosen design publishes no voter + identity at all. + - **LANDED (emission side): `bridgedStats` rides post/comment records.** + Coves' `post.json`/`comment.json` gained an optional `bridgedStats + {upvotes, downvotes, asOf}` (`#bridgedStats` def, required all three, + ints min 0); synced into tidepool's embedded `lexicons/`. Migration 014 + adds `vote_aggregates.stats_emitted_at` (nullable per-row watermark). A + debounced sweeper (`internal/votes/refresher.go`, `STATS_REFRESH_INTERVAL` + 30s / `STATS_REFRESH_BATCH` 200) selects rows where `updated_at > + stats_emitted_at OR stats_emitted_at IS NULL`, oldest-first, bounded, and + for each folds the counts onto the materialized record via + `materialize.Materializer.SetBridgedStats` (lexicon-validated, + mapping-CID kept in sync, one commit tx). Watermark is set to the + `updated_at` READ during the sweep (never now()), so a vote landing + mid-sweep re-dirties the row — no lost update. Counts-unchanged emits + NOTHING (only `asOf` would churn — not worth a firehose event); a + permanent skip (missing/soft-deleted mapping, deleted record → + `IsNotFound`, consent-frozen repo → `IsTombstoned`) advances the + watermark, a transient error leaves the row dirty for retry. + `commitRecord` carries an existing `bridgedStats` forward across a Lemmy + EDIT (the rebuild path never emits it), which also keeps an unchanged + re-ingest an idempotent no-op. Verified edges: `recomputeAggregate` + (hence `SeedAggregates` re-seed AND `ScrubVoter`) sets `updated_at = + CURRENT_TIMESTAMP`, so re-seeds and scrubs naturally re-emit corrected + counts; seeded-only subjects (no live events) still get a row and still + emit. ~~STILL OPEN and untouched here: the seeded+live re-seed + double-count flag above (Lemmy's `counts` includes federated votes that + are also live `vote_events` rows) — independent of emission.~~ **Closed + — see the verified-and-fixed note above (net-of-live seeding).** e2e: + `docker-compose.e2e.yml` sets `STATS_REFRESH_INTERVAL=2s`; the refresher + now adds one post/comment UPDATE per seeded/voted subject to the + firehose, so the restart backfill-redo dedup and the post-unsubscribe + negative window exclude `isBridgedStatsUpdate` events deliberately (they + are the feature, not a re-commit or a bridging violation), and the + zz-sweep lexicon-validates every one. ## Federation & interop @@ -393,7 +511,11 @@ harness, **(relay)** by task 09's relay pipeline. - Baseline-voter drift: a voter counted only in the seeded baseline who later flips federates a bare Dislike (no Undo), leaving the baseline upvote next to the new live downvote; a clear sends an Undo the live-vote - fallback cannot act on (no live row). Counts stale until re-seed. + fallback cannot act on (no live row). Counts stale until re-seed — and + since the net-of-live seeding fix (2026-07-10), a re-seed genuinely heals + the drift (`TestReseedHealsBaselineVoterDrift`); before it, a re-seed + re-imported the flipped vote AND kept the live row, making things worse, + not better. ## Materializer (task 05 notes) diff --git a/README.md b/README.md index 4ddf660..3ad70bc 100644 --- a/README.md +++ b/README.md @@ -208,6 +208,8 @@ production**: | `MINT_RATE_PER_MINUTE` / `MINT_BURST` | `60` / `120` | rate gate on inbound DID minting (PLC registrations are forever; unseen authors in delivered content trigger mints) | | `INGEST_WORKERS` | `4` | inbox queue worker-pool size | | `SEED_COUNTS_FROM_API` | on | seed backfilled posts' vote aggregates from the origin instance's public API (`/api/v3/post` `counts`); set `0` to disable | +| `STATS_REFRESH_INTERVAL` | `30s` | how often the bridged-vote-stats refresher sweeps `vote_aggregates` and folds changed counts onto each subject's post/comment record (`bridgedStats` field); a debounce, so a hot subject's votes coalesce into one record update per sweep — longer is staler counts + fewer firehose events, shorter is fresher + more commit-lock traffic | +| `STATS_REFRESH_BATCH` | `200` | max aggregates one refresher sweep processes (emits, or skips permanently, per row); commits are globally serialized, so the batch keeps a sweep from flooding the commit lock (the remainder waits for the next sweep) | | `TOMBSTONE_RETENTION` | `720h` | how long `ap_tombstones` markers (the delete-before-create guard) are kept before the hourly pruner reclaims them | | `VOTE_EVENT_RETENTION` | `2160h` | how long **undone** (superseded/retracted) `vote_events` rows are kept; live rows are the counts and are never pruned | | `BLOCKS_GC_RETENTION` | `72h` | how long superseded (head-unreachable) repo blocks are kept before the GC sweep (every 6h) reclaims them; the window doubles as the sweep's race guard, so keep it far above sweep duration — and comfortably above any app↔DB clock skew plus the sweep's compute→delete gap, since the cutoff comes from the app clock while `created_at` refreshes use the DB clock (see `internal/repo/gc.go`) | @@ -343,8 +345,17 @@ degenerates from per-IP into a GLOBAL cap. ## Vote aggregates (the AppView integration point) Votes never become records (nothing may strongRef a vote): Lemmy -`Like`/`Dislike`/`Undo` activities maintain bridge-side aggregate counts, -served over **one sanctioned side-channel XRPC** the Coves AppView polls: +`Like`/`Dislike`/`Undo` activities maintain bridge-side aggregate counts. +Those counts reach the AppView two ways. The primary path (locked decision 7 +final direction) folds them onto the CONTENT record: post and comment records +carry an optional `bridgedStats {upvotes, downvotes, asOf}` field, written by +a debounced sweeper (`STATS_REFRESH_INTERVAL`/`STATS_REFRESH_BATCH`) as the +aggregates change, so the counts ride the firehose the AppView already +consumes — a debounced update whose true bound is at most one record update +per subject per sweep, collapsing a burst of votes rather than emitting one +event per vote. +The legacy/debug path is **one sanctioned side-channel XRPC** the Coves +AppView can poll: ``` GET /xrpc/social.coves.bridge.getVoteAggregates?uris=at://…&uris=at://… @@ -372,7 +383,16 @@ backfilled posts would start near zero. `SEED_COUNTS_FROM_API` (default on) compensates by seeding a baseline from the origin's public API during backfill; live votes stack on top, and an undo of a vote that only exists in the baseline is a no-op (accepted drift, refreshed on re-seed). Comment -scores are not seeded in v1 — comments accumulate live votes only. +scores are not seeded in v1 — comments accumulate live votes only (a comment +with live votes still gets a `bridgedStats` field once the refresher folds +them onto its record). + +Both surfaces read the same `vote_aggregates` totals, so `bridgedStats` and +`getVoteAggregates` never disagree beyond the sweep's debounce lag (the field +is `asOf`-stamped with the aggregate's `updated_at`, which the AppView can use +to discard a stale update). Every `bridgedStats` write goes through the same +lexicon validation and mapping bookkeeping as any other record commit, and a +Lemmy edit that rebuilds a record carries an existing `bridgedStats` forward. ## Verifying with Jetstream diff --git a/cmd/tidepool/main.go b/cmd/tidepool/main.go index 50d1985..c7d058e 100644 --- a/cmd/tidepool/main.go +++ b/cmd/tidepool/main.go @@ -245,6 +245,20 @@ func run(logger *slog.Logger) error { // vote_events rows age out like firehose events do. go prune.Run(ctx, "ap_tombstones", cfg.TombstoneRetention, 0, tombstones.Prune, logger) go prune.Run(ctx, "vote_events(undone)", cfg.VoteEventRetention, 0, voteAggregator.PruneUndoneEvents, logger) + + // Bridged-vote-stats refresher (FOLLOWUPS locked decision 7 final + // direction): fold changed vote_aggregates counts onto each subject's + // materialized post/comment record as an optional bridgedStats field, + // debounced so a hot post's votes coalesce into one record update per + // sweep instead of one firehose event per vote. Emits through the + // materializer (lexicon validation + mapping-CID bookkeeping) exactly + // like every other record write. + statsRefresher, err := votes.NewRefresher(database, objects, materializer, + cfg.StatsRefreshInterval, cfg.StatsRefreshBatch, logger) + if err != nil { + return err + } + go statsRefresher.Run(ctx) // Seeding imports historical scores for backfilled posts from the origin // instance's public API (AP alone cannot provide them). var seeder ingest.CountSeeder diff --git a/docker-compose.e2e.yml b/docker-compose.e2e.yml index 582b60d..7b61dad 100644 --- a/docker-compose.e2e.yml +++ b/docker-compose.e2e.yml @@ -121,6 +121,17 @@ services: PLC_DIRECTORY_URL: http://plc:3000 ADMIN_TOKEN: e2e-admin-token SEED_COUNTS_FROM_API: "1" + # Short stats-refresh interval (default 30s) so the vote scenarios + # actually exercise the bridged-vote-stats refresher: within ~2s of a + # vote landing, the subject's post/comment record gains (or updates) its + # bridgedStats field via a real UPDATE commit on the firehose. The + # suite's zz-sweep lexicon-validates every create AND update, so the + # synced #bridgedStats def must make those pass. Every emitted update is + # a post/comment in the four-collection whitelist, so no negative + # assertion is tripped (see votes_hammer_test.go: it awaits the vote XRPC + # counts and never asserts "no further commits"). + STATS_REFRESH_INTERVAL: 2s + STATS_REFRESH_BATCH: "50" # Announce ourselves to the local relay on startup via the REAL # RequestCrawlAll path. ALLOW_DEV_REQUEST_CRAWL exists precisely for # this stack (dev is otherwise log-only, and production refuses the diff --git a/internal/config/config.go b/internal/config/config.go index c519c89..00dc120 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -163,6 +163,18 @@ type Config struct { SyncRatePerSecond int SyncRateBurst int SyncMaxSubscribers int + // StatsRefreshInterval is how often the bridged-vote-stats refresher + // sweeps vote_aggregates and folds changed counts onto the materialized + // post/comment records' bridgedStats field (STATS_REFRESH_INTERVAL, a Go + // duration, default 30s, must be positive). A debounce, not a deadline: a + // hot subject's votes coalesce into at most one record update per sweep. + // Longer quiesces emission (staler bridged counts, fewer firehose events); + // shorter tightens freshness at more commit-lock traffic. + StatsRefreshInterval time.Duration + // StatsRefreshBatch bounds how many due aggregates one sweep emits + // (STATS_REFRESH_BATCH, default 200, must be positive). Commits are + // globally serialized, so the batch keeps one sweep from flooding the lock. + StatsRefreshBatch int } // Load reads configuration from the environment. logger must not be nil; @@ -392,6 +404,17 @@ func Load(logger *slog.Logger) (*Config, error) { return nil, err } + // Bridged-vote-stats refresher: tuning knobs with real defaults in every + // environment (positive-enforced, like the other interval/size knobs). + cfg.StatsRefreshInterval, err = durationVar(logger, "STATS_REFRESH_INTERVAL", 30*time.Second) + if err != nil { + return nil, err + } + cfg.StatsRefreshBatch, err = intVar(logger, "STATS_REFRESH_BATCH", 200) + if err != nil { + return nil, err + } + defaultUserAgent := fmt.Sprintf("tidepool/0.1 (+https://%s)", cfg.BridgeHostname) cfg.UserAgent = os.Getenv("USER_AGENT") if cfg.UserAgent == "" { diff --git a/internal/db/migrations/014_vote_stats_watermark.sql b/internal/db/migrations/014_vote_stats_watermark.sql new file mode 100644 index 0000000..e7527ee --- /dev/null +++ b/internal/db/migrations/014_vote_stats_watermark.sql @@ -0,0 +1,29 @@ +-- +goose Up +-- Bridged-vote-stats emission (FOLLOWUPS "Votes-as-records", locked decision +-- 7 final direction): bridged vote counts ride the CONTENT record as an +-- optional bridgedStats field, refreshed by a debounced sweeper as +-- vote_aggregates rows change. stats_emitted_at is that sweeper's per-row +-- watermark: the vote_aggregates.updated_at value the refresher last emitted +-- onto the materialized record. A row is DUE when updated_at > stats_emitted_at +-- (a vote landed since the last emit) or stats_emitted_at IS NULL (never +-- emitted). Nullable — every existing aggregate is born un-emitted, so the +-- first sweep after this migration emits every row's current counts. +-- +-- The watermark is advanced to the updated_at READ during the sweep (never +-- now()) on EMIT or on a deliberate PERMANENT SKIP (missing/soft-deleted +-- mapping, record gone, consent-frozen repo, persistently invalid record) — so +-- a subject that can never be stamped stops being reconsidered every sweep, +-- while a transient failure leaves the row dirty for retry. A vote landing +-- mid-sweep bumps updated_at past the watermark (via clock_timestamp() in the +-- recompute) and re-dirties the row for the next sweep — no lost update. +ALTER TABLE vote_aggregates ADD COLUMN stats_emitted_at TIMESTAMPTZ; + +-- The sweep selects due rows ordered by updated_at; a partial index over the +-- un-emitted watermark keeps that scan bounded to the actually-dirty rows +-- rather than the whole table once steady state is reached. +CREATE INDEX idx_vote_aggregates_stats_due ON vote_aggregates (updated_at) + WHERE stats_emitted_at IS NULL OR updated_at > stats_emitted_at; + +-- +goose Down +DROP INDEX IF EXISTS idx_vote_aggregates_stats_due; +ALTER TABLE vote_aggregates DROP COLUMN IF EXISTS stats_emitted_at; diff --git a/internal/errors/errors.go b/internal/errors/errors.go index 7c44288..1f2dd60 100644 --- a/internal/errors/errors.go +++ b/internal/errors/errors.go @@ -20,6 +20,15 @@ var ( // a missing object may be fetched and materialized, a tombstoned one // must not be. IsNotFound(tombstoned) is false. ErrTombstoned = errors.New("resource tombstoned") + // ErrRecordGone marks a bridged-stats emission that cannot proceed + // because the TARGET RECORD itself is gone — deleted out from under its + // vote aggregate, or its mapping soft-deleted inside the commit + // transaction. Deliberately distinct from ErrNotFound: the vote-stats + // refresher advances its watermark on a real record-gone but NOT on an + // unrelated NotFound surfaced from deeper in a commit (a missing + // bridged_actor row or signing key — a key-escrow inconsistency to + // retry, never to skip). IsNotFound(recordGone) is false. + ErrRecordGone = errors.New("bridged record gone") ) // ValidationError reports a rejected field value. @@ -102,3 +111,6 @@ func IsValidation(err error) bool { return errors.Is(err, ErrInvalidInput) } // IsTombstoned reports whether err is, wraps, or unwraps to ErrTombstoned. func IsTombstoned(err error) bool { return errors.Is(err, ErrTombstoned) } + +// IsRecordGone reports whether err is, wraps, or unwraps to ErrRecordGone. +func IsRecordGone(err error) bool { return errors.Is(err, ErrRecordGone) } diff --git a/internal/materialize/bridgedstats.go b/internal/materialize/bridgedstats.go new file mode 100644 index 0000000..9b5c86d --- /dev/null +++ b/internal/materialize/bridgedstats.go @@ -0,0 +1,177 @@ +package materialize + +import ( + "context" + "database/sql" + stderrors "errors" + "fmt" + "time" + + "tidepool/internal/errors" + "tidepool/internal/repo" + "tidepool/internal/store" +) + +// bridgedStatsField is the optional post/comment record field the bridge uses +// to assert the origin platform's aggregate vote counts (the Coves +// #bridgedStats def: {upvotes, downvotes, asOf}). Records are BORN without it +// (the create path never sets it); the vote-stats refresher adds it and keeps +// it fresh, and commitRecord carries it forward across Lemmy edits. +const bridgedStatsField = "bridgedStats" + +// maxStatsCommitAttempts bounds the compare-and-swap retry loop below. A +// mismatch means the record changed between the read and the commit (a Lemmy +// edit, or a concurrent stats stamp); a handful of retries absorbs realistic +// churn, and exhausting them is reported as a transient error so the refresher +// leaves the row dirty and retries next sweep. +const maxStatsCommitAttempts = 4 + +// EmitBridgedStats folds an aggregate's counts onto its materialized record +// and reports whether a real commit happened (a NoOp — counts unchanged — +// reports committed=false). It is the narrow seam votes.StatsEmitter is defined +// over, so internal/votes needs neither *materialize.Result nor an import of +// this package to drive the sweep. SetBridgedStats keeps the rich Result for +// this package's own tests. +func (m *Materializer) EmitBridgedStats(ctx context.Context, mapping *store.APObjectMapping, upvotes, downvotes int, asOf time.Time) (committed bool, err error) { + res, err := m.SetBridgedStats(ctx, mapping, upvotes, downvotes, asOf) + if err != nil { + return false, err + } + return res != nil && !res.NoOp, nil +} + +// SetBridgedStats writes upvotes/downvotes (sampled as of asOf) onto the +// bridgedStats field of the record named by mapping, keeping the ap_objects +// mapping CID in sync — record + mapping land in ONE commit transaction, the +// same PutRecord+PutMappingTx discipline commitRecord uses (task 11). +// +// The read (GetRecord) happens OUTSIDE the commit serialization, so the write +// goes through PutRecordCAS with the read's CID as an optimistic-concurrency +// precondition: if a Lemmy edit or another stamp changed the record in +// between, the commit fails ErrPreconditionFailed and this re-reads and +// retries rather than silently reverting the edit or resurrecting a +// just-deleted record. Inside the commit the mapping's soft-delete is +// re-checked too, so a refresher racing a Delete cannot un-tombstone the +// mapping via PutMappingTx's unconditional deleted_at = NULL. +// +// Contract the refresher relies on: +// - Counts already equal to what the record carries (a re-emit where only +// asOf would move) → returns a NoOp Result WITHOUT committing. Re-putting +// would mint a fresh CID and a firehose event just to bump a timestamp; +// that churn is pure noise for relays and the AppView, so it is skipped. +// - Record gone (deleted between the vote landing and this call — AP +// delivery is unordered — or its mapping soft-deleted inside the commit) +// → errors.IsRecordGone, a sentinel DISTINCT from a bare NotFound so the +// refresher advances only on a genuine record-gone, never on a NotFound +// raised deeper in the commit (a missing bridged_actor row or signing +// key). +// - Repo frozen by a consent revocation → errors.IsTombstoned (the commit's +// signing-key consent gate refuses writes to a tombstoned actor). Both +// record-gone and tombstoned are permanent; the refresher advances its +// watermark past them. +func (m *Materializer) SetBridgedStats(ctx context.Context, mapping *store.APObjectMapping, upvotes, downvotes int, asOf time.Time) (*Result, error) { + if mapping == nil { + return nil, errors.NewValidationError("mapping", "must not be nil") + } + if upvotes < 0 || downvotes < 0 { + return nil, errors.NewValidationError("counts", "must not be negative") + } + + for attempt := 0; ; attempt++ { + record, prevCID, err := m.repos.GetRecord(ctx, mapping.DID, mapping.Collection, mapping.RKey) + if err != nil { + if errors.IsNotFound(err) { + // The record was deleted out from under its aggregate. Translate + // only THIS read's NotFound into the record-gone sentinel — the + // refresher's permanent skip — so a NotFound from elsewhere in the + // commit stays transient. + return nil, fmt.Errorf("materialize: stats target %s gone: %w", mapping.ATURI, errors.ErrRecordGone) + } + return nil, fmt.Errorf("materialize: read record for stats %s: %w", mapping.ATURI, err) + } + + // Counts unchanged? Re-emitting would only move asOf, minting a new CID + // and a firehose event for nothing. Skip the commit entirely (the caller + // still advances its watermark). + if up, down, ok := bridgedStatsCounts(record); ok && up == upvotes && down == downvotes { + return &Result{DID: mapping.DID, ATURI: mapping.ATURI, CID: mapping.CID, NoOp: true}, nil + } + + // asOf renders at microsecond precision (recordDatetimeMicros), the + // resolution vote_aggregates.updated_at actually carries — so two + // versions in the same millisecond stay distinguishable to a consumer's + // asOf guard. + record[bridgedStatsField] = map[string]any{ + "upvotes": int64(upvotes), + "downvotes": int64(downvotes), + "asOf": recordDatetimeMicros(asOf), + } + if err := m.validateRecord(record); err != nil { + return nil, err + } + + updated := *mapping + res, err := m.repos.PutRecordCAS(ctx, mapping.DID, mapping.Collection, mapping.RKey, record, prevCID, + func(ctx context.Context, tx *sql.Tx, res *repo.CommitResult) error { + // Resurrection guard: re-check the mapping's soft-delete INSIDE + // the commit tx. The CAS precondition already refuses to re-create + // a record deleted from the repo; this additionally refuses to + // un-tombstone the MAPPING (PutMappingTx clears deleted_at + // unconditionally). The legitimate restore path clears deleted_at + // via objects.Restore BEFORE its rebuild commit, so it passes here. + deleted, derr := m.objects.DeletedInTx(ctx, tx, mapping.APID) + switch { + case derr == nil: + if deleted { + return fmt.Errorf("materialize: stats target %s soft-deleted mid-commit: %w", mapping.ATURI, errors.ErrRecordGone) + } + case errors.IsNotFound(derr): + return fmt.Errorf("materialize: stats target %s mapping vanished mid-commit: %w", mapping.ATURI, errors.ErrRecordGone) + default: + return derr + } + updated.CID = res.RecordCID + _, mapErr := m.objects.PutMappingTx(ctx, tx, updated) + return mapErr + }) + if stderrors.Is(err, repo.ErrPreconditionFailed) { + if attempt+1 < maxStatsCommitAttempts { + continue // the record changed under us; re-read and retry + } + // Persistent churn: leave it for the next sweep (transient, not a + // record-gone — the record is very much alive, just moving). + return nil, fmt.Errorf("materialize: stats for %s: record kept changing across %d attempts: %w", mapping.ATURI, maxStatsCommitAttempts, err) + } + if err != nil { + return nil, fmt.Errorf("materialize: put stats for %s: %w", mapping.ATURI, err) + } + return &Result{DID: mapping.DID, ATURI: mapping.ATURI, CID: res.RecordCID, NoOp: res.NoOp}, nil + } +} + +// bridgedStatsCounts reads the upvotes/downvotes a record's bridgedStats field +// currently asserts. ok is false when the field is absent or malformed (the +// record has never been stats-stamped). Integers arrive as int64 through the +// CBOR read path; the other numeric arms tolerate a JSON-decoded record. +func bridgedStatsCounts(record map[string]any) (up, down int, ok bool) { + stats, ok := record[bridgedStatsField].(map[string]any) + if !ok { + return 0, 0, false + } + up, upOK := asInt(stats["upvotes"]) + down, downOK := asInt(stats["downvotes"]) + return up, down, upOK && downOK +} + +func asInt(v any) (int, bool) { + switch n := v.(type) { + case int64: + return int(n), true + case int: + return n, true + case float64: + return int(n), true + default: + return 0, false + } +} diff --git a/internal/materialize/bridgedstats_test.go b/internal/materialize/bridgedstats_test.go new file mode 100644 index 0000000..4dea7e5 --- /dev/null +++ b/internal/materialize/bridgedstats_test.go @@ -0,0 +1,319 @@ +package materialize + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/errors" +) + +// asOf is a fixed sample time for the bridgedStats tests. +var statsAsOf = time.Date(2026, 7, 10, 9, 0, 0, 0, time.UTC) + +// testValidCID is a well-formed CIDv1 for the lexicon-validation strongRefs +// (the validator checks CID syntax, not reachability). +const testValidCID = "bafyreib2rxk3rybk3aobmv5cjuql3bm2twh4jo5uxgf5kpqrsqxi3jgxte" + +// TestSetBridgedStatsStampsRecord: SetBridgedStats writes the counts onto the +// record, keeps the ap_objects mapping CID in sync, and the stamped record +// still lexicon-validates (the harness runs StrictValidation). +func TestSetBridgedStatsStampsRecord(t *testing.T) { + h := newHarness(t) + h.serveLemmyWorldFixtures() + ctx := context.Background() + + created, err := h.m.MaterializePost(ctx, loadFixtureObject(t, "page_lemmy_world.json")) + require.NoError(t, err) + require.False(t, created.NoOp) + + mapping, err := h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + + res, err := h.m.SetBridgedStats(ctx, mapping, 42, 3, statsAsOf) + require.NoError(t, err) + assert.False(t, res.NoOp, "the first stamp is a real commit") + assert.NotEqual(t, created.CID, res.CID, "adding bridgedStats mints a new CID") + + record, _, err := h.manager.GetRecord(ctx, mapping.DID, mapping.Collection, mapping.RKey) + require.NoError(t, err) + stats, ok := record["bridgedStats"].(map[string]any) + require.True(t, ok, "the record carries bridgedStats") + assert.EqualValues(t, 42, stats["upvotes"]) + assert.EqualValues(t, 3, stats["downvotes"]) + assert.Equal(t, recordDatetimeMicros(statsAsOf), stats["asOf"]) + + // The mapping CID tracks the stamped version (new comments strongRef it). + refreshed, err := h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + assert.Equal(t, res.CID, refreshed.CID, "the mapping CID follows the stamped record") +} + +// TestSetBridgedStatsUnchangedCountsNoOp: re-stamping the same counts with a +// LATER asOf is a no-op (no commit, no firehose event) — asOf-only churn must +// not hit the firehose. +func TestSetBridgedStatsUnchangedCountsNoOp(t *testing.T) { + h := newHarness(t) + h.serveLemmyWorldFixtures() + ctx := context.Background() + + _, err := h.m.MaterializePost(ctx, loadFixtureObject(t, "page_lemmy_world.json")) + require.NoError(t, err) + mapping, err := h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + + first, err := h.m.SetBridgedStats(ctx, mapping, 10, 1, statsAsOf) + require.NoError(t, err) + require.False(t, first.NoOp) + eventsAfterFirst := len(h.firehoseEvents()) + + // Same counts, a later asOf: no commit, no new event. + mapping, err = h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + second, err := h.m.SetBridgedStats(ctx, mapping, 10, 1, statsAsOf.Add(time.Hour)) + require.NoError(t, err) + assert.True(t, second.NoOp, "unchanged counts must not commit") + assert.Equal(t, eventsAfterFirst, len(h.firehoseEvents()), "no firehose event for an asOf-only bump") + + record, _, err := h.manager.GetRecord(ctx, mapping.DID, mapping.Collection, mapping.RKey) + require.NoError(t, err) + stats := record["bridgedStats"].(map[string]any) + assert.Equal(t, recordDatetimeMicros(statsAsOf), stats["asOf"], "asOf did not churn") +} + +// TestSetBridgedStatsChangedCountsCommits: different counts DO commit and move +// asOf forward. +func TestSetBridgedStatsChangedCountsCommits(t *testing.T) { + h := newHarness(t) + h.serveLemmyWorldFixtures() + ctx := context.Background() + + _, err := h.m.MaterializePost(ctx, loadFixtureObject(t, "page_lemmy_world.json")) + require.NoError(t, err) + mapping, err := h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + + _, err = h.m.SetBridgedStats(ctx, mapping, 10, 1, statsAsOf) + require.NoError(t, err) + mapping, err = h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + + later := statsAsOf.Add(time.Hour) + res, err := h.m.SetBridgedStats(ctx, mapping, 11, 1, later) + require.NoError(t, err) + assert.False(t, res.NoOp, "a changed upvote count commits") + + record, _, err := h.manager.GetRecord(ctx, mapping.DID, mapping.Collection, mapping.RKey) + require.NoError(t, err) + stats := record["bridgedStats"].(map[string]any) + assert.EqualValues(t, 11, stats["upvotes"]) + assert.Equal(t, recordDatetimeMicros(later), stats["asOf"]) +} + +// TestSetBridgedStatsRecordDeleted: a stamp for a record deleted out from +// under its aggregate returns the record-gone sentinel (the refresher's +// permanent skip) — distinct from a bare NotFound so an unrelated NotFound +// deeper in the commit stays transient. +func TestSetBridgedStatsRecordDeleted(t *testing.T) { + h := newHarness(t) + h.serveLemmyWorldFixtures() + ctx := context.Background() + + _, err := h.m.MaterializePost(ctx, loadFixtureObject(t, "page_lemmy_world.json")) + require.NoError(t, err) + mapping, err := h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + + require.NoError(t, h.m.HandleDelete(ctx, pageID)) + + _, err = h.m.SetBridgedStats(ctx, mapping, 5, 0, statsAsOf) + assert.True(t, errors.IsRecordGone(err), "a stamp for a deleted record is record-gone, got %v", err) + assert.False(t, errors.IsNotFound(err), "record-gone must NOT satisfy IsNotFound") +} + +// TestEditCarriesBridgedStatsForward is the core carry-forward guarantee: a +// Lemmy EDIT rebuilds the record from AP data (which never carries stats), and +// the rebuild must NOT drop an existing bridgedStats field. +func TestEditCarriesBridgedStatsForward(t *testing.T) { + h := newHarness(t) + h.serveLemmyWorldFixtures() + ctx := context.Background() + + page := loadFixtureObject(t, "page_lemmy_world.json") + _, err := h.m.MaterializePost(ctx, page) + require.NoError(t, err) + mapping, err := h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + + // The refresher stamps the counts. + _, err = h.m.SetBridgedStats(ctx, mapping, 40, 2, statsAsOf) + require.NoError(t, err) + + // A later edit rebuilds from AP data with no stats field. + edited := loadFixtureObject(t, "page_lemmy_world.json") + edited.Source = &ap.Source{Content: "edited body text", MediaType: "text/markdown"} + res, err := h.m.HandleUpdate(ctx, edited) + require.NoError(t, err) + require.False(t, res.NoOp, "an edited body is a real commit") + + record, _, err := h.manager.GetRecord(ctx, res.DID, CollectionPost, mapping.RKey) + require.NoError(t, err) + assert.Equal(t, "edited body text", record["content"], "the edit applied") + stats, ok := record["bridgedStats"].(map[string]any) + require.True(t, ok, "the edit must not drop bridgedStats") + assert.EqualValues(t, 40, stats["upvotes"]) + assert.EqualValues(t, 2, stats["downvotes"]) + assert.Equal(t, recordDatetimeMicros(statsAsOf), stats["asOf"]) +} + +// TestUnchangedReingestAfterStatsIsNoOp: once stats are stamped, re-ingesting +// the IDENTICAL post (deterministic rkey → re-put) must stay an idempotent +// no-op — carry-forward keeps the rebuilt record byte-identical, so no +// spurious firehose event churns the counts away. +func TestUnchangedReingestAfterStatsIsNoOp(t *testing.T) { + h := newHarness(t) + h.serveLemmyWorldFixtures() + ctx := context.Background() + + page := loadFixtureObject(t, "page_lemmy_world.json") + _, err := h.m.MaterializePost(ctx, page) + require.NoError(t, err) + mapping, err := h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + stamped, err := h.m.SetBridgedStats(ctx, mapping, 7, 1, statsAsOf) + require.NoError(t, err) + + eventsAfterStamp := len(h.firehoseEvents()) + + // Re-ingest the identical post (a re-delivery). + again, err := h.m.MaterializePost(ctx, loadFixtureObject(t, "page_lemmy_world.json")) + require.NoError(t, err) + assert.True(t, again.NoOp, "an unchanged re-ingest after stamping must be a no-op") + assert.Equal(t, stamped.CID, again.CID, "carry-forward keeps the CID identical") + assert.Equal(t, eventsAfterStamp, len(h.firehoseEvents()), "no extra firehose event") +} + +// TestBridgedStatsRecordsLexiconValidate pins that a bridgedStats-bearing post +// and comment validate against the vendored Coves lexicons, and that a +// malformed one (missing required asOf) is rejected — the synced #bridgedStats +// def is what the emission path relies on. +func TestBridgedStatsRecordsLexiconValidate(t *testing.T) { + h := newHarness(t) + stats := map[string]any{ + "upvotes": int64(12), + "downvotes": int64(4), + "asOf": recordDatetime(statsAsOf), + } + + post := map[string]any{ + "$type": CollectionPost, + "community": testServiceDID, + "author": testServiceDID, + "createdAt": recordDatetime(statsAsOf), + "title": "a bridged post", + "bridgedStats": stats, + } + require.NoError(t, h.m.validateRecord(post), "a bridgedStats post must validate") + + comment := map[string]any{ + "$type": CollectionComment, + "reply": map[string]any{ + "root": strongRef("at://"+testServiceDID+"/"+CollectionPost+"/3jzfcijpj2z2a", testValidCID), + "parent": strongRef("at://"+testServiceDID+"/"+CollectionPost+"/3jzfcijpj2z2a", testValidCID), + }, + "content": "a bridged comment", + "createdAt": recordDatetime(statsAsOf), + "bridgedStats": stats, + } + require.NoError(t, h.m.validateRecord(comment), "a bridgedStats comment must validate") + + // Missing the required asOf is a validation error. + bad := map[string]any{ + "$type": CollectionPost, + "community": testServiceDID, + "author": testServiceDID, + "createdAt": recordDatetime(statsAsOf), + "bridgedStats": map[string]any{ + "upvotes": int64(1), + "downvotes": int64(0), + }, + } + require.Error(t, h.m.validateRecord(bad), "bridgedStats without asOf must fail validation") +} + +// TestBridgedStatsAsOfMicrosecondPrecision pins that asOf renders at +// microsecond precision: two aggregate versions in the SAME millisecond +// (concurrent voters under clock_timestamp) must produce DISTINCT asOf strings, +// or a consumer's newer-or-equal guard (and the watermark bookkeeping) would +// conflate them. The old millisecond dialect is the regression it guards. +func TestBridgedStatsAsOfMicrosecondPrecision(t *testing.T) { + v1 := time.Date(2026, 7, 10, 9, 0, 0, 123456*1000, time.UTC) // …09:00:00.123456Z + v2 := v1.Add(3 * time.Microsecond) // …09:00:00.123459Z + require.Equal(t, v1.UnixMilli(), v2.UnixMilli(), "the two versions are in the same millisecond") + + assert.NotEqual(t, recordDatetimeMicros(v1), recordDatetimeMicros(v2), + "two versions in the same ms must render distinct asOf strings at microsecond precision") + assert.Equal(t, recordDatetime(v1), recordDatetime(v2), + "millisecond precision would conflate them — this is the regression the microsecond asOf avoids") +} + +// TestCommentEditCarriesReplyRefsForward is the reply-ref carry-forward +// guarantee (Coves treats reply root/parent CIDs as immutable across updates): +// after the parent post is stats-stamped — churning its CID — a Lemmy comment +// EDIT must reuse the ORIGINAL stored reply refs, not re-resolve them to the +// churned CID, or Coves rejects the rebuilt comment as thread hijacking. +func TestCommentEditCarriesReplyRefsForward(t *testing.T) { + h := newHarness(t) + h.serveLemmyWorldFixtures() + h.serveObject("/u/alice", person("https://lemmy.world/u/alice", "alice", nil)) + ctx := context.Background() + + _, err := h.m.MaterializePost(ctx, loadFixtureObject(t, "page_lemmy_world.json")) + require.NoError(t, err) + + c1 := note("https://lemmy.world/comment/1001", "https://lemmy.world/u/alice", + pageID, "original comment", "2026-07-07T04:00:00.000000Z") + h.serveObject("/comment/1001", c1) + _, err = h.m.MaterializeComment(ctx, objectFromMap(t, c1)) + require.NoError(t, err) + + c1Mapping, err := h.objects.GetByAPID(ctx, "https://lemmy.world/comment/1001") + require.NoError(t, err) + origRecord, _, err := h.manager.GetRecord(ctx, c1Mapping.DID, c1Mapping.Collection, c1Mapping.RKey) + require.NoError(t, err) + origRoot, ok := extractStrongRef(origRecord, "reply", "root") + require.True(t, ok) + origParent, ok := extractStrongRef(origRecord, "reply", "parent") + require.True(t, ok) + + // Stamp the parent post: bridgedStats mints a new version, so the post's + // mapping CID (what a re-resolve would return) changes. + pageMapping, err := h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + stamped, err := h.m.SetBridgedStats(ctx, pageMapping, 9, 1, statsAsOf) + require.NoError(t, err) + require.NotEqual(t, pageMapping.CID, stamped.CID, "stamping the post changes its CID") + + // A Lemmy edit rebuilds the comment from AP data. The object is passed + // directly (its parent is already mapped), so no re-fetch/re-serve is needed. + edited := note("https://lemmy.world/comment/1001", "https://lemmy.world/u/alice", + pageID, "edited comment body", "2026-07-07T04:00:00.000000Z") + res, err := h.m.HandleUpdate(ctx, objectFromMap(t, edited)) + require.NoError(t, err) + require.False(t, res.NoOp, "an edited body is a real commit") + + rebuilt, _, err := h.manager.GetRecord(ctx, c1Mapping.DID, c1Mapping.Collection, c1Mapping.RKey) + require.NoError(t, err) + assert.Equal(t, "edited comment body", rebuilt["content"], "the edit applied") + gotRoot, ok := extractStrongRef(rebuilt, "reply", "root") + require.True(t, ok) + gotParent, ok := extractStrongRef(rebuilt, "reply", "parent") + require.True(t, ok) + assert.Equal(t, origRoot, gotRoot, + "reply.root must be carried forward verbatim, NOT re-resolved to the churned post CID") + assert.Equal(t, origParent, gotParent, "reply.parent must be carried forward verbatim") +} diff --git a/internal/materialize/materializer.go b/internal/materialize/materializer.go index f746bf1..c9a8fea 100644 --- a/internal/materialize/materializer.go +++ b/internal/materialize/materializer.go @@ -247,18 +247,21 @@ func (m *Materializer) commitRecord(ctx context.Context, did, collection, rkey s // layer clears the tombstone explicitly on an Undo(Delete)/restore, and // its ap_tombstones marker covers the delete-before-create ordering // (a Delete for a never-materialized object). + // + // carryForward marks the update path for a live post/comment mapping: only + // there does the rebuild carry fields it cannot reconstruct forward + // (bridgedStats, and comments' reply refs), and only there does the commit + // need the optimistic-concurrency guard against a racing stats stamp. + carryForward := false if existing, err := m.objects.GetByAPID(ctx, obj.ID); err == nil { if existing.IsDeleted() { return nil, skip(obj.ID, "object was deleted upstream; not resurrecting") } + carryForward = collection == CollectionPost || collection == CollectionComment } else if !errors.IsNotFound(err) { return nil, fmt.Errorf("materialize: check mapping for %s: %w", obj.ID, err) } - if err := m.validateRecord(record); err != nil { - return nil, err - } - mapping := store.APObjectMapping{ APID: obj.ID, APType: obj.Type, @@ -275,29 +278,108 @@ func (m *Materializer) commitRecord(ctx context.Context, did, collection, rkey s } var stored *store.APObjectMapping - res, err := m.repos.PutRecordTx(ctx, did, collection, rkey, record, - func(ctx context.Context, tx *sql.Tx, res *repo.CommitResult) error { - mapping.CID = res.RecordCID - var mapErr error - stored, mapErr = m.objects.PutMappingTx(ctx, tx, mapping) - if mapErr != nil { - if errors.IsAlreadyExists(mapErr) { - // A different AP id already claimed this at-uri: a - // deterministic TID collision (near-impossible after the - // hash-filled-micros change; see repo.DeterministicTID). - // Loud by design — this is a bug signal, and failing here - // now rolls the record write back with it. - m.logger.Error("deterministic rkey collision: different ap_id claimed the same at-uri", - "ap_id", obj.ID, "did", did, "collection", collection, "rkey", rkey) - } - return fmt.Errorf("materialize: map %s: %w", obj.ID, mapErr) + putMapping := func(ctx context.Context, tx *sql.Tx, res *repo.CommitResult) error { + mapping.CID = res.RecordCID + var mapErr error + stored, mapErr = m.objects.PutMappingTx(ctx, tx, mapping) + if mapErr != nil { + if errors.IsAlreadyExists(mapErr) { + // A different AP id already claimed this at-uri: a + // deterministic TID collision (near-impossible after the + // hash-filled-micros change; see repo.DeterministicTID). + // Loud by design — this is a bug signal, and failing here + // now rolls the record write back with it. + m.logger.Error("deterministic rkey collision: different ap_id claimed the same at-uri", + "ap_id", obj.ID, "did", did, "collection", collection, "rkey", rkey) } - return nil - }) - if err != nil { - return nil, fmt.Errorf("materialize: put %s/%s/%s for %s: %w", did, collection, rkey, obj.ID, err) + return fmt.Errorf("materialize: map %s: %w", obj.ID, mapErr) + } + return nil + } + + if !carryForward { + // Create, or a profile update: no read-modify-write, so no precondition + // — an idempotent re-put must still reach the repo's NoOp path. + if err := m.validateRecord(record); err != nil { + return nil, err + } + res, err := m.repos.PutRecordTx(ctx, did, collection, rkey, record, putMapping) + if err != nil { + return nil, fmt.Errorf("materialize: put %s/%s/%s for %s: %w", did, collection, rkey, obj.ID, err) + } + return &Result{DID: did, ATURI: stored.ATURI, CID: res.RecordCID, NoOp: res.NoOp}, nil + } + + // Update path: carry forward the fields the AP rebuild cannot reconstruct, + // under a CAS precondition on the stored record's CID. If a concurrent + // stats stamp (or another edit) changed the record between the read and the + // commit, the commit fails ErrPreconditionFailed and we re-read and retry — + // so an edit can neither drop a just-committed stamp nor churn reply refs a + // stats stamp already moved. + for attempt := 0; ; attempt++ { + expectPrevCID, cerr := m.carryForwardFields(ctx, did, collection, rkey, record, obj, attempt) + if cerr != nil { + return nil, cerr + } + if err := m.validateRecord(record); err != nil { + return nil, err + } + res, err := m.repos.PutRecordCAS(ctx, did, collection, rkey, record, expectPrevCID, putMapping) + if stderrors.Is(err, repo.ErrPreconditionFailed) { + if attempt+1 < maxStatsCommitAttempts { + continue + } + return nil, fmt.Errorf("materialize: put %s: record kept changing across %d attempts: %w", obj.ID, maxStatsCommitAttempts, err) + } + if err != nil { + return nil, fmt.Errorf("materialize: put %s/%s/%s for %s: %w", did, collection, rkey, obj.ID, err) + } + return &Result{DID: did, ATURI: stored.ATURI, CID: res.RecordCID, NoOp: res.NoOp}, nil + } +} + +// carryForwardFields folds the fields a Lemmy rebuild cannot reconstruct out of +// the currently-stored record onto the freshly-built one, and returns the +// stored record's CID for use as the commit's CAS precondition: +// +// - bridgedStats: the vote-stats refresher writes it; the rebuild never +// does. Dropping it would lose the counts until the next sweep and mint a +// needless firehose event (breaking idempotent re-ingest). +// - reply (comments only): Coves' comment consumer treats reply root/parent +// strongRef CIDs as IMMUTABLE across updates, but a stats stamp on the +// parent churns its CID — so re-resolving would hand Coves changed refs it +// rejects as thread hijacking. Carrying the stored refs verbatim keeps the +// thread anchoring stable across edits. The CREATE path still resolves +// fresh refs (this runs only for a live existing mapping). +// +// A record absent on the FIRST attempt is the crash window between a delete +// commit and its soft-delete: let the rebuild stand as a guarded create +// (expectCID ""). Absent on a RETRY means it was deleted concurrently — skip +// rather than resurrect it. +func (m *Materializer) carryForwardFields(ctx context.Context, did, collection, rkey string, record map[string]any, obj *ap.Object, attempt int) (expectPrevCID string, err error) { + stored, storedCID, rerr := m.repos.GetRecord(ctx, did, collection, rkey) + switch { + case rerr == nil: + if stats, ok := stored[bridgedStatsField]; ok { + record[bridgedStatsField] = stats + } else { + delete(record, bridgedStatsField) // clear any carried by a prior attempt + } + if collection == CollectionComment { + if reply, ok := stored["reply"]; ok { + record["reply"] = reply + } + } + return storedCID, nil + case errors.IsNotFound(rerr): + if attempt > 0 { + return "", skip(obj.ID, "record deleted concurrently during rebuild; not resurrecting") + } + delete(record, bridgedStatsField) + return "", nil + default: + return "", fmt.Errorf("materialize: read record for carry-forward %s: %w", obj.ID, rerr) } - return &Result{DID: did, ATURI: stored.ATURI, CID: res.RecordCID, NoOp: res.NoOp}, nil } // validateRecord checks the record against the vendored Coves lexicons — @@ -347,6 +429,20 @@ func recordDatetime(t time.Time) string { return t.UTC().Format("2006-01-02T15:04:05.000Z") } +// recordDatetimeMicros renders a timestamp at MICROSECOND precision — the +// resolution vote_aggregates.updated_at (a postgres timestamptz) actually +// carries. bridgedStats.asOf uses this, not recordDatetime's millisecond +// dialect: two distinct aggregate versions landing in the same millisecond +// (concurrent voters under clock_timestamp()) would otherwise serialize to +// EQUAL asOf strings, letting a newer-or-equal consumer guard and the +// watermark bookkeeping conflate them. atproto's datetime format permits +// fractional-second digits, so six is as valid as three; createdAt keeps its +// existing millisecond dialect (its source `published` time is only that +// precise, and Coves already parses it). +func recordDatetimeMicros(t time.Time) string { + return t.UTC().Format("2006-01-02T15:04:05.000000Z") +} + // recordRKey derives the deterministic record key for an AP object, // failing closed (as a skip) when the object has no usable published time. func recordRKey(obj *ap.Object) (string, error) { diff --git a/internal/repo/repo.go b/internal/repo/repo.go index 07e0db4..7f0df67 100644 --- a/internal/repo/repo.go +++ b/internal/repo/repo.go @@ -29,6 +29,24 @@ import ( "tidepool/internal/errors" ) +// ErrPreconditionFailed is returned by PutRecordCAS when the record currently +// at the target path does not match the caller's expected previous CID — an +// optimistic-concurrency mismatch. It is the signal for the caller's +// read-modify-write retry: re-read the current record, re-derive the write, and +// try again. It exists because the vote-stats and carry-forward paths read the +// record OUTSIDE the commit serialization (per-DID mutex + global advisory +// lock); the precondition, checked INSIDE it, turns that stale read into a +// harmless retry instead of a silent lost update or resurrection. +var ErrPreconditionFailed = stderrors.New("repo: record precondition failed") + +// casPrecondition, when non-nil, requires the record currently at the write +// path to match expectCID before the commit proceeds. expectCID == "" means +// the record must NOT currently exist (a guarded create). A nil precondition +// disables the check entirely (the ordinary PutRecord/DeleteRecord path). +type casPrecondition struct { + expectCID string +} + // KeyUse says what a signing key is being requested for, so key custody // can apply consent policy per operation kind: tombstoned (consent-revoked) // actors are frozen for new writes but their records must remain deletable. @@ -177,6 +195,22 @@ func (m *Manager) PutRecord(ctx context.Context, did, collection, rkey string, r // PutRecordTx is PutRecord with a side effect executed inside the commit // transaction (see TxSideEffect). A nil sideEffect is exactly PutRecord. func (m *Manager) PutRecordTx(ctx context.Context, did, collection, rkey string, record map[string]any, sideEffect TxSideEffect) (*CommitResult, error) { + return m.putRecord(ctx, did, collection, rkey, record, nil, sideEffect) +} + +// PutRecordCAS is PutRecordTx with an optimistic-concurrency precondition: the +// record currently at (did, collection, rkey) must have CID expectedPrevCID +// before the write commits — the empty string requiring the record to NOT +// currently exist. On mismatch it returns ErrPreconditionFailed WITHOUT +// committing (the side effect never runs), so a caller whose read happened +// outside the commit serialization can re-read and retry. The precondition is +// evaluated under the same per-DID mutex and global advisory lock the commit +// takes, so it cannot itself race a concurrent commit to the same record. +func (m *Manager) PutRecordCAS(ctx context.Context, did, collection, rkey string, record map[string]any, expectedPrevCID string, sideEffect TxSideEffect) (*CommitResult, error) { + return m.putRecord(ctx, did, collection, rkey, record, &casPrecondition{expectCID: expectedPrevCID}, sideEffect) +} + +func (m *Manager) putRecord(ctx context.Context, did, collection, rkey string, record map[string]any, pre *casPrecondition, sideEffect TxSideEffect) (*CommitResult, error) { if err := validateRecord(record); err != nil { return nil, err } @@ -188,14 +222,14 @@ func (m *Manager) PutRecordTx(ctx context.Context, did, collection, rkey string, if err != nil { return nil, err } - return m.commitWrite(ctx, did, collection, rkey, &c, recordBytes, sideEffect) + return m.commitWrite(ctx, did, collection, rkey, &c, recordBytes, pre, sideEffect) } // DeleteRecord removes a record and commits the change. A missing record — // or a repo that does not exist yet — is an error satisfying // errors.IsNotFound. The result's RecordCID is empty. func (m *Manager) DeleteRecord(ctx context.Context, did, collection, rkey string) (*CommitResult, error) { - return m.commitWrite(ctx, did, collection, rkey, nil, nil, nil) + return m.commitWrite(ctx, did, collection, rkey, nil, nil, nil, nil) } // GetRecord reads the current version of a record. Missing repo or record @@ -281,7 +315,7 @@ func (m *Manager) readState(ctx context.Context, did string) (*repoState, error) // and the global commit advisory lock, applies the mutation to the MST, // signs a new commit, and persists blocks, head, and the firehose event in // one transaction. -func (m *Manager) commitWrite(ctx context.Context, did, collection, rkey string, newCID *cid.Cid, recordBytes []byte, sideEffect TxSideEffect) (*CommitResult, error) { +func (m *Manager) commitWrite(ctx context.Context, did, collection, rkey string, newCID *cid.Cid, recordBytes []byte, pre *casPrecondition, sideEffect TxSideEffect) (*CommitResult, error) { path, parsedDID, err := validatePath(did, collection, rkey) if err != nil { return nil, err @@ -366,6 +400,25 @@ func (m *Manager) commitWrite(ctx context.Context, did, collection, rkey string, if newCID == nil && op.Prev == nil { return nil, errors.NewNotFoundError("record", fmt.Sprintf("at://%s/%s", did, path)) } + + // Optimistic-concurrency precondition (see casPrecondition): the record + // that was at this path before this op (op.Prev) must match what the caller + // based its read-modify-write on. Checked here, under the commit locks, so a + // stale read outside them cannot revert a concurrent edit, drop a + // just-committed stamp, or resurrect a deleted record. op.Prev == nil means + // no record is present (deleted or never existed); expectCID "" is the + // caller asserting exactly that. + if pre != nil { + var current string + if op.Prev != nil { + current = op.Prev.String() + } + if current != pre.expectCID { + return nil, fmt.Errorf("repo: precondition for %s/%s: expected prev cid %q, have %q: %w", + did, path, pre.expectCID, current, ErrPreconditionFailed) + } + } + if newCID != nil && op.Prev != nil && op.Prev.Equals(*newCID) { // Identical re-put: idempotent no-op, keep the existing commit. // op.Prev != nil implies the repo exists, so state is non-nil here. diff --git a/internal/store/ap_objects.go b/internal/store/ap_objects.go index 72732c5..d4ff1e8 100644 --- a/internal/store/ap_objects.go +++ b/internal/store/ap_objects.go @@ -96,6 +96,22 @@ func (r *postgresAPObjects) GetByAPID(ctx context.Context, apID string) (*APObje return mapping, nil } +func (r *postgresAPObjects) DeletedInTx(ctx context.Context, tx *sql.Tx, apID string) (bool, error) { + if tx == nil { + return false, errors.NewValidationError("tx", "must not be nil") + } + var deletedAt sql.NullTime + err := tx.QueryRowContext(ctx, + `SELECT deleted_at FROM ap_objects WHERE ap_id = $1`, apID).Scan(&deletedAt) + if stderrors.Is(err, sql.ErrNoRows) { + return false, errors.NewNotFoundError("ap_object", apID) + } + if err != nil { + return false, fmt.Errorf("read deleted_at for ap_object %q: %w", apID, err) + } + return deletedAt.Valid, nil +} + func (r *postgresAPObjects) GetByATURI(ctx context.Context, atURI string) (*APObjectMapping, error) { query := `SELECT` + apObjectColumns + ` FROM ap_objects WHERE at_uri = $1` mapping, err := scanAPObject(r.db.QueryRowContext(ctx, query, atURI)) diff --git a/internal/store/interfaces.go b/internal/store/interfaces.go index e6064c0..aa02b59 100644 --- a/internal/store/interfaces.go +++ b/internal/store/interfaces.go @@ -36,6 +36,14 @@ type APObjects interface { // soft-deleted rows (callers can check IsDeleted to detect tombstones). GetByAPID(ctx context.Context, apID string) (*APObjectMapping, error) + // DeletedInTx reports whether the mapping for apID is currently + // soft-deleted, read ON THE GIVEN TRANSACTION so a commit's side effect + // can re-check consent state atomically with its write — the guard that + // stops the vote-stats refresher from resurrecting a soft-deleted mapping + // (PutMappingTx unconditionally clears deleted_at). A missing mapping is an + // error satisfying errors.IsNotFound. + DeletedInTx(ctx context.Context, tx *sql.Tx, apID string) (bool, error) + // GetByATURI returns the mapping for an at-uri, including soft-deleted // rows. GetByATURI(ctx context.Context, atURI string) (*APObjectMapping, error) diff --git a/internal/votes/aggregator.go b/internal/votes/aggregator.go index f4c3a25..906db26 100644 --- a/internal/votes/aggregator.go +++ b/internal/votes/aggregator.go @@ -336,13 +336,30 @@ func (a *Aggregator) RetractVote(ctx context.Context, vote *ap.Object, community // SeedAggregates imports a baseline (upvotes, downvotes) for a bridged // 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; re-seeding -// (backfill redo) overwrites the baseline idempotently. Known drift: a voter -// counted only in the baseline who later flips federates a bare Dislike -// (Lemmy sends no Undo on flips), so the retired upvote stays in the -// baseline next to the new live downvote; a clear sends an Undo the -// fallback cannot act on (no live vote row). Accepted for v1; the baseline -// refreshes on re-seed. +// 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 +// 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): +// - 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. +// // Subjects not present in ap_objects are dropped and logged at debug, like // ApplyVote. func (a *Aggregator) SeedAggregates(ctx context.Context, subjectAPID string, upvotes, downvotes int) error { @@ -368,19 +385,28 @@ func (a *Aggregator) SeedAggregates(ctx context.Context, subjectAPID string, upv atURI := mapping.ATURI return 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 + // on the same subject. + if err := lockAggregate(ctx, tx, subjectAPID, atURI); err != nil { + return err + } if _, err := tx.ExecContext(ctx, ` - INSERT INTO vote_aggregates ( - subject_ap_id, subject_at_uri, - seeded_upvotes, seeded_downvotes, upvotes, downvotes - ) VALUES ($1, $2, $3, $4, $3, $4) - ON CONFLICT (subject_ap_id) DO UPDATE SET - subject_at_uri = EXCLUDED.subject_at_uri, - seeded_upvotes = EXCLUDED.seeded_upvotes, - seeded_downvotes = EXCLUDED.seeded_downvotes`, - subjectAPID, atURI, upvotes, downvotes); err != nil { + UPDATE vote_aggregates a + SET seeded_upvotes = GREATEST(0, $2 - live.up), + seeded_downvotes = GREATEST(0, $3 - live.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 { return fmt.Errorf("seed vote aggregate for %q: %w", subjectAPID, err) } - // Fold any live events that landed before the seed into the totals. return recomputeAggregate(ctx, tx, subjectAPID) }) } @@ -653,12 +679,23 @@ func lockAggregate(ctx context.Context, tx *sql.Tx, subject, atURI string) error // baseline plus the live (non-undone) events. Recompute-per-subject over // incremental arithmetic: a Lemmy post sees at most a few thousand votes, // and recomputing inside the locking transaction cannot drift. +// +// updated_at uses clock_timestamp() (the real wall time at statement +// execution), NOT CURRENT_TIMESTAMP (which is transaction-START time). Under +// concurrent voters (INGEST_WORKERS > 1) two commits' transaction-start times +// can order oppositely to their aggregate-row-lock acquisition, so +// CURRENT_TIMESTAMP could stamp a later-committed vote with an EARLIER +// updated_at — landing it below the stats refresher's already-advanced +// watermark, never to be emitted (permanent staleness on a then-quiet +// subject). clock_timestamp() is taken while THIS transaction holds the +// per-subject aggregate row lock, so per-subject updated_at is monotonic and +// every committed vote strictly advances it past any prior watermark. func recomputeAggregate(ctx context.Context, tx *sql.Tx, subject string) error { if _, err := tx.ExecContext(ctx, ` UPDATE vote_aggregates a SET upvotes = a.seeded_upvotes + live.up, downvotes = a.seeded_downvotes + live.down, - updated_at = CURRENT_TIMESTAMP + updated_at = clock_timestamp() FROM ( SELECT COUNT(*) FILTER (WHERE direction = 'up') AS up, diff --git a/internal/votes/aggregator_test.go b/internal/votes/aggregator_test.go index 097c3dd..a99cf11 100644 --- a/internal/votes/aggregator_test.go +++ b/internal/votes/aggregator_test.go @@ -325,10 +325,12 @@ func TestSeedAggregates(t *testing.T) { assert.Equal(t, 41, up) assert.Equal(t, 3, down) - // Re-seeding (backfill redo) replaces the baseline and keeps live votes. + // Re-seeding (backfill redo) replaces the baseline. The origin's counts + // are a TOTAL that already includes Alice's live vote, so the served + // totals must equal them — not add her again. require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 50, 5)) up, down, _ = counts(t, database, subjectPost) - assert.Equal(t, 51, up) + assert.Equal(t, 50, up, "a re-seed must not double-count live votes") assert.Equal(t, 5, down) // The served at-uri is the mapping's. @@ -346,13 +348,110 @@ func TestSeedAggregatesBeforeLiveVotesFoldsThem(t *testing.T) { ctx := context.Background() // A live vote lands BEFORE the seed (announce raced the backfill): the - // seed must fold it in, not clobber it. + // seed must not clobber the live row — and must not count it twice + // either. Lemmy counted Alice's vote before announcing it, so the + // fetched total (10 up) already includes her. require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 10, 1)) up, down, _ := counts(t, database, subjectPost) - assert.Equal(t, 11, up) + assert.Equal(t, 10, up, "the origin total already includes the live vote") assert.Equal(t, 1, down) + + // Alice clears her vote: the live row is undone, and her upvote must + // leave the served totals exactly once (it is no longer subtracted from + // the baseline once it is not live). + require.NoError(t, agg.RetractVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + up, down, _ = counts(t, database, subjectPost) + assert.Equal(t, 9, up) + assert.Equal(t, 1, down) +} + +// TestReseedDoesNotDoubleCountLiveVotes pins the backfill re-seed contract: +// votes that federated live between two seeds appear in BOTH the origin's +// fetched totals and vote_events, and must be served exactly once. Before +// the net-of-live subtraction, every re-seed re-imported the live voters +// into the baseline and the recompute added them again — compounding on +// each subsequent redo. +func TestReseedDoesNotDoubleCountLiveVotes(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + // Initial backfill: the post has 10 up / 0 down of pre-subscribe history. + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 10, 0)) + + // Live federation since: two upvotes and a downvote. Lemmy's own counts + // are now 12 up / 1 down. + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 2), voterBob, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, dislike(activityID(t, 3), voterCarol, subjectPost), "")) + up, down, _ := counts(t, database, subjectPost) + require.Equal(t, 12, up) + require.Equal(t, 1, down) + + // Backfill redo re-seeds with the origin's current totals. Served counts + // must match them exactly, not 14/2. + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 12, 1)) + up, down, _ = counts(t, database, subjectPost) + assert.Equal(t, 12, up, "re-seed must not re-import live voters into the baseline") + assert.Equal(t, 1, down) + + // And it must be idempotent: another redo with unchanged origin totals + // changes nothing (no compounding). + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 12, 1)) + up, down, _ = counts(t, database, subjectPost) + assert.Equal(t, 12, up) + assert.Equal(t, 1, down) +} + +// TestReseedHealsBaselineVoterDrift: a voter counted only in the seeded +// baseline who flips federates a bare Dislike (Lemmy sends no Undo on +// flips), leaving the retired upvote in the baseline next to the new live +// downvote. Accepted drift while it lasts — but a re-seed must converge the +// served totals back to the origin's truth, which the raw-overwrite seed +// never did (it re-imported the flipped vote AND kept the live row). +func TestReseedHealsBaselineVoterDrift(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + // Alice's pre-subscribe upvote is part of the 10/0 baseline. + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 10, 0)) + + // She flips: a bare Dislike, no Undo. Known drift — her baseline upvote + // lingers next to the live downvote (Lemmy's truth is 9/1). + require.NoError(t, agg.ApplyVote(ctx, dislike(activityID(t, 1), voterAlice, subjectPost), "")) + up, down, _ := counts(t, database, subjectPost) + require.Equal(t, 10, up) + require.Equal(t, 1, down) + + // Re-seed with the origin's current totals: served counts converge. + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 9, 1)) + up, down, _ = counts(t, database, subjectPost) + assert.Equal(t, 9, up, "re-seed must heal baseline-voter drift") + assert.Equal(t, 1, down) +} + +// TestSeedAggregatesClampsBaselineAtZero: an origin reporting totals LOWER +// than the bridge's live counts (author auto-upvote asymmetry, an origin +// that lost votes, a hostile API) must clamp the derived baseline at zero +// per direction — never go negative — and the live counts still serve. +func TestSeedAggregatesClampsBaselineAtZero(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 2), voterBob, subjectPost), "")) + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 1, 0)) + + up, down, _ := counts(t, database, subjectPost) + assert.Equal(t, 2, up, "live votes serve even when the origin under-reports") + assert.Equal(t, 0, down) } func TestSeedAggregatesUnbridgedSubjectDropped(t *testing.T) { diff --git a/internal/votes/refresher.go b/internal/votes/refresher.go new file mode 100644 index 0000000..80b9790 --- /dev/null +++ b/internal/votes/refresher.go @@ -0,0 +1,275 @@ +package votes + +import ( + "context" + "database/sql" + "log/slog" + "time" + + "tidepool/internal/errors" + "tidepool/internal/store" +) + +// Refresher is the debounced sweeper that keeps each bridged post/comment +// record's bridgedStats field in step with its vote_aggregates row (FOLLOWUPS +// "Votes-as-records", locked decision 7 final direction: bridged counts ride +// the CONTENT record, not per-vote records). Votes maintain aggregate counts +// synchronously (Aggregator); this asynchronously folds those counts onto the +// materialized records on an interval, coalescing a burst of votes on one +// subject into at most one record update per sweep. +// +// Watermark model (the stats_emitted_at column, migration 014): a row is DUE +// when a vote has landed since it was last emitted — updated_at > +// stats_emitted_at, or stats_emitted_at IS NULL (never emitted). After +// emitting, the watermark is set to the updated_at value READ during the +// sweep, never now(): a vote landing mid-sweep bumps updated_at past that +// value and re-dirties the row for the next sweep, so no update is lost. +// +// Why a sweeper and not emit-on-vote: commits are globally serialized (the +// repo advisory lock, ~300/s ceiling shared by ALL writes) and each commit is +// one firehose event. A hot post taking a hundred votes a minute must not mint +// a hundred record updates; the sweep collapses them to one, bounded per run. +type Refresher struct { + db *sql.DB + objects store.APObjects + emitter StatsEmitter + interval time.Duration + batch int + logger *slog.Logger +} + +// StatsEmitter writes an aggregate's counts onto its materialized record and +// reports whether a real commit happened (counts-unchanged is committed=false). +// *materialize.Materializer implements it via EmitBridgedStats. The seam is +// deliberately narrow — a bool, not *materialize.Result — so internal/votes +// need not import internal/materialize just to drive the sweep, and so the +// sweep/watermark logic here is testable against a fake emitter without a full +// repo stack. +type StatsEmitter interface { + EmitBridgedStats(ctx context.Context, mapping *store.APObjectMapping, upvotes, downvotes int, asOf time.Time) (committed bool, err error) +} + +// DefaultRefreshInterval is how often the sweep runs when the caller passes a +// non-positive interval. +const DefaultRefreshInterval = 30 * time.Second + +// DefaultRefreshBatch bounds how many due aggregates one sweep emits when the +// caller passes a non-positive batch. Sized so a sweep cannot flood the global +// commit lock: at the benchmarked ~3 ms steady-state commit, 200 updates is +// well under a second of the shared lock even in the worst case where every +// due row actually commits. +const DefaultRefreshBatch = 200 + +// NewRefresher validates dependencies and builds a Refresher. A non-positive +// interval or batch falls back to the package default (config parsing already +// rejects non-positive values; this is the library-level guard). +func NewRefresher(db *sql.DB, objects store.APObjects, emitter StatsEmitter, interval time.Duration, batch int, logger *slog.Logger) (*Refresher, error) { + if db == nil { + return nil, errors.NewValidationError("db", "must not be nil") + } + if objects == nil { + return nil, errors.NewValidationError("objects", "must not be nil") + } + if emitter == nil { + return nil, errors.NewValidationError("emitter", "must not be nil") + } + if interval <= 0 { + interval = DefaultRefreshInterval + } + if batch <= 0 { + batch = DefaultRefreshBatch + } + if logger == nil { + logger = slog.Default() + } + return &Refresher{db: db, objects: objects, emitter: emitter, interval: interval, batch: batch, logger: logger}, nil +} + +// Run sweeps once immediately and then every interval until ctx is cancelled +// (the background-runner shape the pruners and FollowRetrier use). +func (r *Refresher) Run(ctx context.Context) { + r.Sweep(ctx) + ticker := time.NewTicker(r.interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + r.Sweep(ctx) + } + } +} + +// dueRow is one aggregate that needs (re-)emitting: a consistent snapshot of +// its counts and the updated_at those counts correspond to (which becomes both +// the record's asOf and, on success/permanent-skip, the new watermark). +type dueRow struct { + subject string + upvotes int + downvotes int + updatedAt time.Time +} + +// Sweep emits one bounded batch of due aggregates. Exported for the tests to +// drive one deterministic pass without the ticker. +func (r *Refresher) Sweep(ctx context.Context) { + // Read the whole batch into memory FIRST, then emit. Each emit runs its own + // commit transaction taking the global advisory lock; holding this SELECT's + // cursor (and its connection) open across those commits would pin a + // connection for the whole sweep. + // + // Ordering caveat (oldest-dirty first, ORDER BY updated_at ASC): a + // TRANSIENT-failing row keeps its old updated_at, so it sorts to the HEAD of + // every batch and is retried FIRST, not behind fresh work — and ≥batch + // persistently-failing rows would monopolize the batch and starve emission. + // That is tolerable ONLY because the two permanent-failure classes advance + // the watermark out of the due set instead of failing forever: a genuinely + // gone/frozen subject (record-gone, tombstoned) and a persistently invalid + // record (lexicon-validation failure — see refreshOne's validation arm). + // What remains at the head is a truly transient fault (a DB blip), which is + // the right thing to retry first. + rows, err := r.db.QueryContext(ctx, ` + SELECT subject_ap_id, upvotes, downvotes, updated_at + FROM vote_aggregates + WHERE stats_emitted_at IS NULL OR updated_at > stats_emitted_at + ORDER BY updated_at ASC + LIMIT $1`, r.batch) + if err != nil { + if ctx.Err() == nil { + r.logger.Error("stats refresh: select due aggregates failed", "error", err) + } + return + } + var due []dueRow + for rows.Next() { + var d dueRow + if err := rows.Scan(&d.subject, &d.upvotes, &d.downvotes, &d.updatedAt); err != nil { + _ = rows.Close() + r.logger.Error("stats refresh: scan due aggregate failed", "error", err) + return + } + due = append(due, d) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + if ctx.Err() == nil { + r.logger.Error("stats refresh: iterate due aggregates failed", "error", err) + } + return + } + _ = rows.Close() + + var emitted, failed int + for _, d := range due { + if ctx.Err() != nil { + return + } + committed, transientFail := r.refreshOne(ctx, d) + if committed { + emitted++ + } + if transientFail { + failed++ + } + } + // Summarize whenever anything committed OR anything failed transiently — the + // failed count is how a wedged sweep (a batch head jammed by failing rows) + // becomes visible instead of silently emitting nothing every interval. + if emitted > 0 || failed > 0 { + r.logger.Info("bridged vote stats sweep", "emitted", emitted, "failed", failed, "due", len(due)) + } +} + +// refreshOne emits one due row. It returns committed (a record commit actually +// happened, for the emitted count) and transientFail (a non-permanent error +// that left the row dirty, for the sweep's failed count and wedge visibility). +// +// Watermark discipline: advance on success AND on every PERMANENT skip +// (missing/soft-deleted mapping, record gone, consent-frozen repo, +// persistently invalid record) so those rows stop being reconsidered every +// sweep; do NOT advance on a transient error (a DB hiccup) so the next sweep +// retries with the row still dirty. Every advance-on-skip logs its reason at +// Debug (precedent: the aggregator's vote-drop debug logs) so a silently +// skipped subject is still traceable. +func (r *Refresher) refreshOne(ctx context.Context, d dueRow) (committed, transientFail bool) { + mapping, err := r.objects.GetByAPID(ctx, d.subject) + switch { + case err == nil: + case errors.IsNotFound(err): + // An aggregate implies its subject was mapped once; a missing mapping + // means the ap_objects row was hard-removed. Nothing to stamp, and it + // will never come back under this id — advance so we stop looking. + r.logger.Debug("stats refresh: skip, mapping hard-removed; advancing watermark", "subject", d.subject) + r.advance(ctx, d.subject, d.updatedAt) + return false, false + default: + r.logger.Error("stats refresh: resolve subject failed", "subject", d.subject, "error", err) + return false, true // transient: leave dirty, retry next sweep + } + if mapping.IsDeleted() { + // Content deleted after the vote landed (unordered AP delivery). The + // record is gone; its asserted counts are moot. Permanent: advance. + r.logger.Debug("stats refresh: skip, mapping soft-deleted; advancing watermark", "subject", d.subject, "at_uri", mapping.ATURI) + r.advance(ctx, d.subject, d.updatedAt) + return false, false + } + + committed, err = r.emitter.EmitBridgedStats(ctx, mapping, d.upvotes, d.downvotes, d.updatedAt) + switch { + case err == nil: + r.advance(ctx, d.subject, d.updatedAt) + return committed, false + case errors.IsRecordGone(err): + // The record was deleted out from under its aggregate (or its mapping + // soft-deleted inside the commit). A DISTINCT sentinel, not a bare + // NotFound: a NotFound raised deeper in the commit (missing + // bridged_actor row or signing key) is a key-escrow inconsistency to + // retry, and must NOT advance every watermark it touches. + r.logger.Debug("stats refresh: skip, record gone; advancing watermark", "subject", d.subject, "at_uri", mapping.ATURI) + r.advance(ctx, d.subject, d.updatedAt) + return false, false + case errors.IsTombstoned(err): + // Repo frozen by a consent revocation: the commit's signing-key gate + // refuses writes to a tombstoned actor, and always will. Permanent: + // advance so a frozen actor's stale row stops being swept forever. + r.logger.Debug("stats refresh: skip, repo consent-frozen; advancing watermark", "subject", d.subject, "at_uri", mapping.ATURI) + r.advance(ctx, d.subject, d.updatedAt) + return false, false + case errors.IsValidation(err): + // The record fails lexicon validation (strict mode). It keeps its old + // updated_at and would sort to the HEAD of every batch, wedging the + // sweep for every row behind it — so advance past it, loudly. A + // persistently invalid record is a bug to investigate, not a reason to + // stop emitting everyone else's counts. + r.logger.Warn("stats refresh: record fails lexicon validation; advancing watermark to avoid wedging the sweep (investigate)", + "subject", d.subject, "at_uri", mapping.ATURI, "error", err) + r.advance(ctx, d.subject, d.updatedAt) + return false, false + default: + // Transient (a DB blip, a lock timeout, persistent CAS churn): keep the + // row dirty so the next sweep retries it. + r.logger.Error("stats refresh: emit failed", "subject", d.subject, "at_uri", mapping.ATURI, "error", err) + return false, true + } +} + +// advance stamps the watermark with the updated_at READ during the sweep +// (never now()). The `AND updated_at = $2` guard is load-bearing: if a +// concurrent recompute bumped updated_at after this sweep read it — which +// clock_timestamp() in recomputeAggregate makes strictly newer, per subject — +// the UPDATE matches no row, the watermark does NOT advance, and +// `updated_at > stats_emitted_at` keeps the row due for the next sweep. Without +// the guard, advancing to the stale value could still leave a just-committed +// vote below the (older, transaction-start CURRENT_TIMESTAMP) watermark and +// never re-emit it. The row is debounced, never dropped. +func (r *Refresher) advance(ctx context.Context, subject string, watermark time.Time) { + if _, err := r.db.ExecContext(ctx, + `UPDATE vote_aggregates SET stats_emitted_at = $2 WHERE subject_ap_id = $1 AND updated_at = $2`, + subject, watermark); err != nil && ctx.Err() == nil { + // A failed watermark advance is self-healing: the row stays due and the + // next sweep re-emits (an idempotent no-op if the record already carries + // the counts) and re-advances. Log, don't wedge the sweep. + r.logger.Error("stats refresh: advance watermark failed", "subject", subject, "error", err) + } +} diff --git a/internal/votes/refresher_test.go b/internal/votes/refresher_test.go new file mode 100644 index 0000000..e8268c7 --- /dev/null +++ b/internal/votes/refresher_test.go @@ -0,0 +1,420 @@ +package votes + +import ( + "context" + "database/sql" + stderrors "errors" + "log/slog" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/errors" + "tidepool/internal/materialize" + "tidepool/internal/store" +) + +// The real materializer is the production StatsEmitter; pin the interface so a +// signature drift is a compile error, not an e2e surprise. +var _ StatsEmitter = (*materialize.Materializer)(nil) + +// emitCall records one SetBridgedStats invocation. +type emitCall struct { + subject string + up, down int + asOf time.Time +} + +// fakeEmitter stands in for the materializer: it records calls and returns a +// configurable committed/error so the watermark logic can be exercised in +// isolation (the real record write is covered in internal/materialize). +type fakeEmitter struct { + mu sync.Mutex + calls []emitCall + fn func(mapping *store.APObjectMapping, up, down int) (committed bool, err error) +} + +func (f *fakeEmitter) EmitBridgedStats(_ context.Context, mapping *store.APObjectMapping, up, down int, asOf time.Time) (bool, error) { + f.mu.Lock() + f.calls = append(f.calls, emitCall{subject: mapping.APID, up: up, down: down, asOf: asOf}) + fn := f.fn + f.mu.Unlock() + if fn != nil { + return fn(mapping, up, down) + } + return true, nil +} + +func (f *fakeEmitter) snapshot() []emitCall { + f.mu.Lock() + defer f.mu.Unlock() + return append([]emitCall(nil), f.calls...) +} + +// testRefresher builds a Refresher over the test database with a fake emitter. +func testRefresher(t *testing.T, database *sql.DB, emitter StatsEmitter) *Refresher { + t.Helper() + r, err := NewRefresher(database, store.NewAPObjects(database), emitter, 0, 0, slog.Default()) + require.NoError(t, err) + return r +} + +// aggregateTimes reads a subject's updated_at and stats_emitted_at watermark. +func aggregateTimes(t *testing.T, database *sql.DB, subject string) (updated time.Time, emitted sql.NullTime, found bool) { + t.Helper() + err := database.QueryRow(` + SELECT updated_at, stats_emitted_at FROM vote_aggregates WHERE subject_ap_id = $1`, + subject).Scan(&updated, &emitted) + if stderrors.Is(err, sql.ErrNoRows) { + return time.Time{}, sql.NullTime{}, false + } + require.NoError(t, err) + return updated, emitted, true +} + +// TestRefresherEmitsDueRowOnce: a row with votes is emitted once, its +// watermark is stamped with the row's updated_at (the counts' asOf), and a +// second sweep with no new vote does not re-emit. +func TestRefresherEmitsDueRowOnce(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, agg.ApplyVote(ctx, dislike(activityID(t, 2), voterBob, subjectPost), "")) + + emitter := &fakeEmitter{} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + + calls := emitter.snapshot() + require.Len(t, calls, 1, "the dirty row is emitted exactly once") + assert.Equal(t, subjectPost, calls[0].subject) + assert.Equal(t, 1, calls[0].up) + assert.Equal(t, 1, calls[0].down) + + updated, emitted, found := aggregateTimes(t, database, subjectPost) + require.True(t, found) + require.True(t, emitted.Valid, "the watermark was stamped") + assert.True(t, emitted.Time.Equal(updated), "watermark is the row's updated_at, not now()") + assert.True(t, calls[0].asOf.Equal(updated), "asOf is the counts' updated_at") + + // Nothing changed: a second sweep must not re-emit. + r.Sweep(ctx) + assert.Len(t, emitter.snapshot(), 1, "an un-dirtied row is not re-emitted") +} + +// TestRefresherConcurrentUpdateRedirties: a vote landing after the watermark +// was stamped bumps updated_at past it, so the next sweep re-emits with the +// fresh counts (the debounce never drops an update). +func TestRefresherConcurrentUpdateRedirties(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + emitter := &fakeEmitter{} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + require.Len(t, emitter.snapshot(), 1) + + // A new vote lands (updated_at moves past the watermark). + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 2), voterBob, subjectPost), "")) + r.Sweep(ctx) + + calls := emitter.snapshot() + require.Len(t, calls, 2, "the re-dirtied row is emitted again") + assert.Equal(t, 2, calls[1].up, "the re-emit carries the fresh count") +} + +// TestRefresherGuardedAdvanceKeepsMidSweepVoteDue is the watermark-strand +// regression: a vote that commits BETWEEN the sweep's read of a due row and the +// sweep's watermark advance must keep the row due. The emitter fn injects that +// vote — a real ApplyVote whose recompute stamps updated_at via clock_timestamp +// (strictly newer than the value the sweep read). The advance is guarded +// (`AND updated_at = $2`), so it matches no row and does NOT stamp the stale +// watermark; the row stays due and the next sweep re-emits the fresh counts. +// Without the guard the advance would stamp the stale updated_at — and with the +// pre-fix CURRENT_TIMESTAMP recompute (transaction-start time) the concurrent +// vote could even regress updated_at below that watermark and be lost forever. +func TestRefresherGuardedAdvanceKeepsMidSweepVoteDue(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + before, _, found := aggregateTimes(t, database, subjectPost) + require.True(t, found) + + injected := false + emitter := &fakeEmitter{fn: func(_ *store.APObjectMapping, _, _ int) (bool, error) { + // Simulate a vote landing mid-sweep: after the sweep read this row but + // before it advances the watermark. Exactly once, or the second sweep's + // emit would inject again. + if !injected { + injected = true + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 2), voterBob, subjectPost), "")) + } + return true, nil + }} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + + after, emitted, _ := aggregateTimes(t, database, subjectPost) + assert.True(t, after.After(before), "clock_timestamp recompute must strictly advance updated_at per subject") + assert.False(t, emitted.Valid, + "the guarded advance must not stamp a watermark the mid-sweep vote already superseded — the row stays due") + + // The next sweep re-emits with the fresh counts (the concurrent vote is not lost). + r.Sweep(ctx) + calls := emitter.snapshot() + require.Len(t, calls, 2, "the still-due row is swept again") + assert.Equal(t, 2, calls[1].up, "the re-sweep carries the mid-sweep vote's count") + _, emitted2, _ := aggregateTimes(t, database, subjectPost) + assert.True(t, emitted2.Valid, "with no further votes, the second sweep advances the watermark") +} + +// TestRefresherUnchangedCountsAdvancesWithoutCommit: when the emitter reports +// NoOp (the record already carried these counts, only asOf would move), the +// refresher still advances the watermark — the row must not be re-swept every +// interval forever. +func TestRefresherUnchangedCountsAdvancesWithoutCommit(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + + emitter := &fakeEmitter{fn: func(_ *store.APObjectMapping, _, _ int) (bool, error) { + return false, nil // NoOp: no commit happened + }} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + require.Len(t, emitter.snapshot(), 1) + + _, emitted, _ := aggregateTimes(t, database, subjectPost) + require.True(t, emitted.Valid, "a NoOp emit still advances the watermark") + + r.Sweep(ctx) + assert.Len(t, emitter.snapshot(), 1, "a NoOp-emitted row is not re-swept") +} + +// TestRefresherDeletedMappingSkippedAndAdvanced: a subject whose mapping is +// soft-deleted (content deleted after the vote) is never handed to the +// emitter, and its watermark is advanced so it stops being reconsidered. +func TestRefresherDeletedMappingSkippedAndAdvanced(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + require.NoError(t, objects.SoftDelete(ctx, subjectPost)) + + emitter := &fakeEmitter{} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + + assert.Empty(t, emitter.snapshot(), "a soft-deleted subject is never emitted") + _, emitted, _ := aggregateTimes(t, database, subjectPost) + assert.True(t, emitted.Valid, "the deleted-mapping row is advanced past") +} + +// TestRefresherMissingMappingAdvanced: an aggregate whose ap_objects row was +// hard-removed (no mapping at all) is advanced past, not retried forever. +func TestRefresherMissingMappingAdvanced(t *testing.T) { + database := testDB(t) + testAggregator(t, database) // migrate + truncate + ctx := context.Background() + + // A bare aggregate with no backing mapping. + orphan := "https://lemmy.world/post/orphan" + _, err := database.ExecContext(ctx, ` + INSERT INTO vote_aggregates (subject_ap_id, subject_at_uri, upvotes, downvotes) + VALUES ($1, $2, 3, 1)`, orphan, "at://"+testDID+"/"+testCollection+"/3jzorphanzzza") + require.NoError(t, err) + + emitter := &fakeEmitter{} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + + assert.Empty(t, emitter.snapshot(), "no mapping means nothing to stamp") + _, emitted, _ := aggregateTimes(t, database, orphan) + assert.True(t, emitted.Valid, "an orphan aggregate is advanced past") +} + +// TestRefresherTransientErrorLeavesDirty: a transient emit error does NOT +// advance the watermark, so the next sweep retries the still-dirty row. +func TestRefresherTransientErrorLeavesDirty(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + + fail := true + emitter := &fakeEmitter{fn: func(_ *store.APObjectMapping, _, _ int) (bool, error) { + if fail { + return false, stderrors.New("db connection reset") + } + return true, nil + }} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + + _, emitted, _ := aggregateTimes(t, database, subjectPost) + assert.False(t, emitted.Valid, "a transient failure leaves the watermark unset (still dirty)") + + // The next sweep retries; this time it succeeds and advances. + fail = false + r.Sweep(ctx) + assert.Len(t, emitter.snapshot(), 2, "the still-dirty row is retried") + _, emitted, _ = aggregateTimes(t, database, subjectPost) + assert.True(t, emitted.Valid, "the successful retry advances the watermark") +} + +// TestRefresherTombstonedRepoAdvances: a consent-frozen repo rejects the +// commit with errors.IsTombstoned — a PERMANENT failure the refresher advances +// past (never retries forever). +func TestRefresherTombstonedRepoAdvances(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + + emitter := &fakeEmitter{fn: func(_ *store.APObjectMapping, _, _ int) (bool, error) { + return false, errors.NewTombstonedError("repo", subjectPost) + }} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + + _, emitted, _ := aggregateTimes(t, database, subjectPost) + assert.True(t, emitted.Valid, "a tombstoned repo is a permanent skip: advance the watermark") + r.Sweep(ctx) + assert.Len(t, emitter.snapshot(), 1, "a tombstoned row is not re-swept") +} + +// TestRefresherRecordGoneDuringEmitAdvances: EmitBridgedStats returning the +// record-gone sentinel (record deleted between the mapping read and the commit) +// is a permanent skip. +func TestRefresherRecordGoneDuringEmitAdvances(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + + emitter := &fakeEmitter{fn: func(_ *store.APObjectMapping, _, _ int) (bool, error) { + return false, errors.ErrRecordGone + }} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + + _, emitted, _ := aggregateTimes(t, database, subjectPost) + assert.True(t, emitted.Valid, "a vanished record is a permanent skip") +} + +// TestRefresherPlainNotFoundStaysTransient: a bare NotFound from the emit path +// (NOT the record-gone sentinel) — e.g. a missing bridged_actor row or signing +// key surfaced from deep in the commit — must NOT advance the watermark. Such a +// key-escrow inconsistency is transient: silently advancing every watermark it +// touched would drop those subjects' counts. +func TestRefresherPlainNotFoundStaysTransient(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + + emitter := &fakeEmitter{fn: func(_ *store.APObjectMapping, _, _ int) (bool, error) { + return false, errors.NewNotFoundError("signing_key", subjectPost) + }} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + + _, emitted, _ := aggregateTimes(t, database, subjectPost) + assert.False(t, emitted.Valid, "a non-record-gone NotFound must leave the row dirty for retry") +} + +// TestRefresherValidationErrorAdvances: a persistently invalid record (lexicon +// validation failure in strict mode) must not wedge the sweep at the head of +// the batch — the refresher advances past it loudly. +func TestRefresherValidationErrorAdvances(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, 1), voterAlice, subjectPost), "")) + + emitter := &fakeEmitter{fn: func(_ *store.APObjectMapping, _, _ int) (bool, error) { + return false, errors.NewValidationError("record", "fails lexicon validation") + }} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + + _, emitted, _ := aggregateTimes(t, database, subjectPost) + assert.True(t, emitted.Valid, "a persistently invalid record is advanced past, not retried forever") + r.Sweep(ctx) + assert.Len(t, emitter.snapshot(), 1, "a validation-wedged row is not re-swept") +} + +// TestRefresherSeededOnlySubjectEmits: a subject with only a seeded baseline +// (no live vote events) is still due and still emitted — the counts came from +// the origin's public API, not the firehose, but they must still ride the +// record. +func TestRefresherSeededOnlySubjectEmits(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + bridgeSubject(t, objects, subjectPost, "3jzfcijpj2z2a") + ctx := context.Background() + + require.NoError(t, agg.SeedAggregates(ctx, subjectPost, 25, 4)) + + emitter := &fakeEmitter{} + r := testRefresher(t, database, emitter) + r.Sweep(ctx) + + calls := emitter.snapshot() + require.Len(t, calls, 1, "a seeded-only subject is emitted") + assert.Equal(t, 25, calls[0].up) + assert.Equal(t, 4, calls[0].down) +} + +// TestRefresherBatchBounded: a sweep emits at most `batch` rows; the rest wait +// for the next sweep (commits are globally serialized — a sweep must not flood +// the lock). +func TestRefresherBatchBounded(t *testing.T) { + database := testDB(t) + agg, objects := testAggregator(t, database) + ctx := context.Background() + + for i := 0; i < 5; i++ { + subject := subjectPost + "/" + string(rune('a'+i)) + bridgeSubject(t, objects, subject, "3jzfcijpj2z2"+string(rune('a'+i))) + require.NoError(t, agg.ApplyVote(ctx, like(activityID(t, i), voterAlice, subject), "")) + } + + emitter := &fakeEmitter{} + r, err := NewRefresher(database, objects, emitter, time.Second, 2, slog.Default()) + require.NoError(t, err) + + r.Sweep(ctx) + assert.Len(t, emitter.snapshot(), 2, "one sweep emits at most `batch` rows") + r.Sweep(ctx) + assert.Len(t, emitter.snapshot(), 4, "the next sweep picks up the next batch") + r.Sweep(ctx) + assert.Len(t, emitter.snapshot(), 5, "and the remainder") +} diff --git a/lexicons/MANIFEST.sha256 b/lexicons/MANIFEST.sha256 index 395aad6..14ba091 100644 --- a/lexicons/MANIFEST.sha256 +++ b/lexicons/MANIFEST.sha256 @@ -23,7 +23,7 @@ d3e077e34c9b9ccd8a8892148fe4ce7ab1de65873250a8906aec510d040bfa4c social/coves/a d688327a711491aeb58b75187f468f213d270adad0c0ca59b96fbef0229cd3f4 social/coves/aggregator/updateConfig.json 020b4a33837455e17e1b0e258304f11241931d14929be099f33ca67a88fc2f49 social/coves/bridge/getVoteAggregates.json 88fb6259698d0097995200ed5d3a7887165f9cd29907c1c60880150c6568aebf social/coves/community/block.json -b8ae0bf6561a53785970b3da1db9007a07ecb8a1f4db527c5461f8ca80da8124 social/coves/community/comment.json +b7e6f675003d29f6eca49f9728ac7d028050e136cfc1e3aae830484c65a25269 social/coves/community/comment.json 4ff4e1b35004b4757f25778922ec948fb7abaff87c7c46b46e1dadfdc7ab5217 social/coves/community/comment/create.json 153c147d2d307182e91e880c5ce8209090247de1ecefed6b3b776502a23991fe social/coves/community/comment/defs.json 50e2528f308e045b003e6c7d05bdadb9b04d6f4b43b249db8539af4a842b4447 social/coves/community/comment/delete.json @@ -36,7 +36,7 @@ ea5cac9a27aca28f818d87a90a01e9564920de9e96e6f3606d71daa0ea6294a8 social/coves/c c51d12359017474850b5b9bb682154f3c58ebd96d5975ba5103853bc9fa640d2 social/coves/community/getSubscribers.json 4f5fc7cf0aa4b8f17205a28664acbc6b03d398850b336f64434ba113939ae47c social/coves/community/list.json 78608976d210bdd66eb7005e60e99de3394d2c73129cd9a8d914a1bd798b80e4 social/coves/community/moderator.json -f412d89df7ad9cd3afc3cbd35b23b6a77abc857f713e0e56545b543c2e5794c8 social/coves/community/post.json +7e35ec35611fc0f9083abe1a95ad99d3ade65e4e710d3d02330e246f7b4a0809 social/coves/community/post.json a93fda24dd3c895283a2ddf5b4b4021061271a81ce2d338cba615bd77d89c04b social/coves/community/post/create.json e233f344054fb75b2d28744ff8975c560aa29210c652cc4b59de635eb637695b social/coves/community/post/delete.json 02811d7e45ce19ea52b774dcfeba2475d14266a0386780e8d98cc3064d2a1770 social/coves/community/post/get.json diff --git a/lexicons/social/coves/community/comment.json b/lexicons/social/coves/community/comment.json index d774fcf..a68ba59 100644 --- a/lexicons/social/coves/community/comment.json +++ b/lexicons/social/coves/community/comment.json @@ -55,10 +55,37 @@ "type": "string", "format": "datetime", "description": "Timestamp of comment creation" + }, + "bridgedStats": { + "type": "ref", + "ref": "#bridgedStats", + "description": "Bridge-asserted aggregate of origin-platform votes for federated/bridged content. Set by the bridge that materialized this record; absent for natively-authored comments." } } } }, + "bridgedStats": { + "type": "object", + "description": "Aggregate vote counts asserted by the bridge for content federated from an origin platform (e.g. Lemmy). These supplement, and are kept separate from, native atproto votes.", + "required": ["upvotes", "downvotes", "asOf"], + "properties": { + "upvotes": { + "type": "integer", + "minimum": 0, + "description": "Number of upvotes on the origin platform as of asOf" + }, + "downvotes": { + "type": "integer", + "minimum": 0, + "description": "Number of downvotes on the origin platform as of asOf" + }, + "asOf": { + "type": "string", + "format": "datetime", + "description": "Timestamp the origin-platform counts were sampled; used to discard stale updates" + } + } + }, "replyRef": { "type": "object", "description": "References for maintaining thread structure. Root always points to the original post, parent points to the immediate parent (post or comment).", diff --git a/lexicons/social/coves/community/post.json b/lexicons/social/coves/community/post.json index 78b4947..a2041ce 100644 --- a/lexicons/social/coves/community/post.json +++ b/lexicons/social/coves/community/post.json @@ -92,9 +92,36 @@ "type": "string", "format": "datetime", "description": "Timestamp of post creation" + }, + "bridgedStats": { + "type": "ref", + "ref": "#bridgedStats", + "description": "Bridge-asserted aggregate of origin-platform votes for federated/bridged content. Set by the bridge that materialized this record; absent for natively-authored posts." } } } + }, + "bridgedStats": { + "type": "object", + "description": "Aggregate vote counts asserted by the bridge for content federated from an origin platform (e.g. Lemmy). These supplement, and are kept separate from, native atproto votes.", + "required": ["upvotes", "downvotes", "asOf"], + "properties": { + "upvotes": { + "type": "integer", + "minimum": 0, + "description": "Number of upvotes on the origin platform as of asOf" + }, + "downvotes": { + "type": "integer", + "minimum": 0, + "description": "Number of downvotes on the origin platform as of asOf" + }, + "asOf": { + "type": "string", + "format": "datetime", + "description": "Timestamp the origin-platform counts were sampled; used to discard stale updates" + } + } } } } diff --git a/tests/e2e/bridge_test.go b/tests/e2e/bridge_test.go index 6be8d56..d98fe26 100644 --- a/tests/e2e/bridge_test.go +++ b/tests/e2e/bridge_test.go @@ -287,6 +287,21 @@ func TestVotes_SideChannelOnly(t *testing.T) { voter.likeComment(t, comment.ID, 1) awaitAggregates(t, h, commentEv.atURI(), 1, 0) + // POSITIVE emission assertion (the other half of locked decision 7): the + // vote-stats refresher folds the comment's live upvote onto its record as a + // bridgedStats UPDATE (STATS_REFRESH_INTERVAL=2s in this stack). Await that + // update on the comment's rkey carrying upvotes=1 with a parseable asOf — + // end-to-end proof the counts reach the AppView on the firehose, not only + // the side-channel XRPC. The comment's final vote state is a stable 1/0 (no + // further votes), so this target does not race a later count change. + l.await("comment bridgedStats update (upvotes=1)", func(e *jsEvent) bool { + if e.Commit.Collection != colComment || e.Commit.Operation != opUpdate || e.atURI() != commentEv.atURI() { + return false + } + up, ok := bridgedStatsUpvotes(e.Commit.Record) + return ok && up == 1 && bridgedStatsAsOfParses(e.Commit.Record) + }) + // The whole time, nothing vote-shaped may have hit the firehose. The // listener's centralized vetting already fails fast on any unexpected // collection; this explicit unfiltered sweep is the belt to that @@ -480,16 +495,37 @@ func TestRestart_ReplayIsIdempotent(t *testing.T) { // Replay everything from the ORIGINAL cursor. Key the accounting by // (did, collection/rkey) — NOT operation — so a create+update pair for - // one record counts as the duplicate it is. + // one record counts as the duplicate it is. EXCEPTION: the vote-stats + // refresher legitimately emits one bridgedStats UPDATE per seeded/backfilled + // post (SEED_COUNTS_FROM_API is on, so the backfill redo re-seeds these + // posts and the sweeper folds the counts onto each record) — those are the + // debounced feature, not a re-commit. They are excluded via statsDedup, + // which recognises a stats emission as an update equal to the prior record + // MODULO bridgedStats. A blanket isBridgedStatsUpdate would be too loose + // here: once stamped, EVERY update carries the field via carry-forward, so a + // non-idempotent rebuild that duplicated a record's CONTENT would also carry + // it and slip past — the modulo comparison still catches that (its content + // differs), so this assertion keeps its teeth. l2 := h.newListener(t, cursor, colCommunityProfile, colActorProfile, colPost) title2 := "Post-restart " + h.suffix author.createPost(t, community.ID, title2, "after the bounce") seen := map[string]int{} + dedup := newStatsDedup() keyOf := func(ev *jsEvent) string { return fmt.Sprintf("%s %s/%s", ev.Did, ev.Commit.Collection, ev.Commit.RKey) } + account := func(ev *jsEvent) { + if ev.Kind != kindCommit || ev.Commit == nil { + return + } + key := keyOf(ev) + if dedup.isPureStatsEmission(ev, key) { + return // the debounced bridgedStats emission, not a re-commit + } + seen[key]++ + } gapKey := "" sawNew, sawGap := false, false deadline := time.Now().Add(eventTimeout) @@ -498,8 +534,11 @@ func TestRestart_ReplayIsIdempotent(t *testing.T) { if ev.Kind != kindCommit || ev.Commit == nil { continue } - seen[keyOf(ev)]++ - if ev.Did == sub.DID && ev.Commit.Collection == colPost { + pureStats := dedup.isPureStatsEmission(ev, keyOf(ev)) + if !pureStats { + seen[keyOf(ev)]++ + } + if ev.Did == sub.DID && ev.Commit.Collection == colPost && !pureStats { switch got, _ := fieldOf(ev.Commit.Record, "title"); got { case gapTitle: sawGap, gapKey = true, keyOf(ev) @@ -518,9 +557,7 @@ func TestRestart_ReplayIsIdempotent(t *testing.T) { // The backfill redo finished before l2 dialed; a short trailing drain // still catches any late duplicate emission in flight. for _, ev := range l2.drain(8 * time.Second) { - if ev.Kind == kindCommit && ev.Commit != nil { - seen[keyOf(ev)]++ - } + account(ev) } // The replayed pre-restart post must appear EXACTLY once: zero would diff --git a/tests/e2e/helpers.go b/tests/e2e/helpers.go index 899f7b8..099ac9f 100644 --- a/tests/e2e/helpers.go +++ b/tests/e2e/helpers.go @@ -37,6 +37,7 @@ import ( comatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/atproto/atdata" "github.com/bluesky-social/indigo/atproto/lexicon" + "github.com/bluesky-social/indigo/atproto/syntax" indigoevents "github.com/bluesky-social/indigo/events" "github.com/gorilla/websocket" @@ -1415,6 +1416,123 @@ func (l *jsListener) vetEvent(ev *jsEvent) { } } +// isBridgedStatsUpdate reports whether a commit is the vote-stats refresher's +// debounced bridgedStats emission: an UPDATE to a post/comment record carrying +// a bridgedStats object (FOLLOWUPS "Votes-as-records" locked decision 7 final +// direction — bridged vote counts ride the CONTENT record). With +// SEED_COUNTS_FROM_API on, EVERY backfilled/seeded subject gets one of these +// shortly after materialization, plus more as its origin vote counts change — +// so the content-dedup ("exactly once per rkey") and post-unsubscribe negative +// windows must not mistake a legitimate stats update for a duplicate emission +// or a bridging violation. +// +// Caveat, deliberate: a genuine CONTENT edit ALSO carries bridgedStats +// (commitRecord carries the field forward). The scenarios that use this helper +// perform no content edits on the affected records, so an update-with-stats +// there is unambiguously a stats emission — the helper is not a general +// "is this only a stats change" classifier. +func isBridgedStatsUpdate(ev *jsEvent) bool { + if ev.Kind != kindCommit || ev.Commit == nil || ev.Commit.Operation != opUpdate { + return false + } + if ev.Commit.Collection != colPost && ev.Commit.Collection != colComment { + return false + } + if len(ev.Commit.Record) == 0 { + return false + } + var m map[string]any + if err := json.Unmarshal(ev.Commit.Record, &m); err != nil { + return false + } + _, ok := m["bridgedStats"].(map[string]any) + return ok +} + +// statsDedup tracks the last record seen per (did, collection, rkey) so a +// scenario can tell a PURE vote-stats emission (safe to exclude from a +// content-dedup count) from a genuine re-commit. isBridgedStatsUpdate alone can +// no longer do this: once a record is stamped, EVERY later update carries the +// field via carry-forward, so a non-idempotent rebuild that duplicated the +// record would also look like a stats update and slip past the dedup. The +// distinguisher is equality MODULO bridgedStats — a stats emission changes only +// that field; a real duplicate changes content too. +type statsDedup struct { + last map[string]json.RawMessage +} + +func newStatsDedup() *statsDedup { return &statsDedup{last: map[string]json.RawMessage{}} } + +// isPureStatsEmission reports whether ev is an update that differs from the +// previously-seen record for its key ONLY in the bridgedStats field, then +// records ev as the new latest for that key. A create, a first sighting, or an +// update that also changed content returns false (and is the thing a dedup +// count must catch). +func (s *statsDedup) isPureStatsEmission(ev *jsEvent, key string) bool { + if ev.Kind != kindCommit || ev.Commit == nil { + return false + } + prev, seen := s.last[key] + s.last[key] = ev.Commit.Record + if !seen || ev.Commit.Operation != opUpdate { + return false + } + if ev.Commit.Collection != colPost && ev.Commit.Collection != colComment { + return false + } + return recordEqualsModuloBridgedStats(prev, ev.Commit.Record) +} + +// recordEqualsModuloBridgedStats reports whether two firehose records are +// identical after dropping bridgedStats from both — i.e. the only thing that +// changed (if anything) is the vote-stats stamp. +func recordEqualsModuloBridgedStats(a, b json.RawMessage) bool { + return canonicalWithoutBridgedStats(a) == canonicalWithoutBridgedStats(b) +} + +func canonicalWithoutBridgedStats(raw json.RawMessage) string { + var m map[string]any + if len(raw) == 0 || json.Unmarshal(raw, &m) != nil { + return string(raw) + } + delete(m, "bridgedStats") + canon, err := json.Marshal(m) // encoding/json sorts map keys → canonical + if err != nil { + return string(raw) + } + return string(canon) +} + +// bridgedStatsUpvotes reads bridgedStats.upvotes off a record; ok is false when +// the field is absent or malformed. +func bridgedStatsUpvotes(record json.RawMessage) (int, bool) { + var m struct { + BridgedStats *struct { + Upvotes *int `json:"upvotes"` + AsOf string `json:"asOf"` + } `json:"bridgedStats"` + } + if err := json.Unmarshal(record, &m); err != nil || m.BridgedStats == nil || m.BridgedStats.Upvotes == nil { + return 0, false + } + return *m.BridgedStats.Upvotes, true +} + +// bridgedStatsAsOfParses reports whether bridgedStats.asOf is present and a +// valid atproto datetime (the field the AppView uses to discard stale updates). +func bridgedStatsAsOfParses(record json.RawMessage) bool { + var m struct { + BridgedStats *struct { + AsOf string `json:"asOf"` + } `json:"bridgedStats"` + } + if err := json.Unmarshal(record, &m); err != nil || m.BridgedStats == nil || m.BridgedStats.AsOf == "" { + return false + } + _, err := syntax.ParseDatetime(m.BridgedStats.AsOf) + return err == nil +} + // await returns the first commit event matching pred — scanning events an // earlier await consumed-but-buffered FIRST (see pending: the relay does // not preserve cross-repo ordering), then the live stream — failing the diff --git a/tests/e2e/lifecycle_test.go b/tests/e2e/lifecycle_test.go index 9859f05..f35ba4d 100644 --- a/tests/e2e/lifecycle_test.go +++ b/tests/e2e/lifecycle_test.go @@ -414,7 +414,26 @@ func TestUnsubscribe_StopsBridging(t *testing.T) { continue } if ev.Did == unsSub.DID && ev.TimeUs > preEv.TimeUs { - t.Errorf("unsubscribed community repo %s emitted after Undo{Follow}: %s", unsSub.DID, ev) + // A trailing bridgedStats UPDATE on an ALREADY-bridged post is the + // vote-stats refresher settling seeded counts (SEED_COUNTS_FROM_API + // is on), not new content bridged after the Undo{Follow} — + // unsubscribe stops NEW content, it does not roll back stats the + // aggregates already hold. Tolerated, but ONLY as a pure stats + // settle: carry-forward ships the whole record, so a genuine content + // change could hide inside a stats-shaped update. Pin the pre-post's + // title AND content unchanged so a real edit cannot pass as one. + if isBridgedStatsUpdate(ev) { + if got, _ := fieldOf(ev.Commit.Record, "title"); got != preTitle { + t.Errorf("stats-shaped update on unsubscribed repo %s changed the title to %q (want the pre-post %q): %s", + unsSub.DID, got, preTitle, ev) + } + if got, _ := fieldOf(ev.Commit.Record, "content"); got != "flows while subscribed" { + t.Errorf("stats-shaped update on unsubscribed repo %s changed the content to %q (want the pre-post body): %s", + unsSub.DID, got, ev) + } + } else { + t.Errorf("unsubscribed community repo %s emitted after Undo{Follow}: %s", unsSub.DID, ev) + } } if got, _ := fieldOf(ev.Commit.Record, "title"); got == deadTitle { t.Errorf("post in unsubscribed community reached the firehose: %s", ev)