diff --git a/FOLLOWUPS.md b/FOLLOWUPS.md index 8863603..675b626 100644 --- a/FOLLOWUPS.md +++ b/FOLLOWUPS.md @@ -63,6 +63,37 @@ task documents and git history rather than this list. subtract Tidepool's written-back tally during subsequent Lemmy re-seeds. - Moderation federation and DMs. +## Outbound delivery (task 15) + +- **outbound_deliveries.ClaimNext lacks a standalone `seq` index.** The + loose-scan CTE builds the head set via the `(ordering_key, seq)` partial + index, but the outer `c.seq = ANY(ARRAY(...)) FOR UPDATE` re-check has no + index on `seq` alone (`seq` is BIGSERIAL, not the PK `(activity_id, + target_inbox)`). Unlike inbox_events (where `id` IS the PK), this is not a + point-fetch. A dedicated `UNIQUE INDEX (seq)` in its OWN migration (021 — + amending the already-applied 020 is a silent no-op under goose) would + restore O(keys × log N); add it if the claim path profiles hot. +- **OutboundDeliveries is a 12-method interface** (go-proverbs SHOULD). It + is one cohesive repository seam but the worker uses only the + claim/mark subset and the admin API only inspect/redrive/cancel. If it + grows, split into a `deliveryClaimer` (worker) + `deliveryAdmin` + (admin) at the consumer packages. +- **DeliveredStateUndone is reserved but never written** — task 15 deletes + the outbound_votes row on Undo-delivery success rather than transitioning + to 'undone'. Kept in the enum + CHECK against a future keep-the-record + policy; nothing consumes it today. +- **Image embeds don't federate outbound yet** — `social.coves.embed.images` + → `attachment [{Image,url}]` and external-embed thumbnails need a + blob→author-PDS-getBlob-URL seam not yet designed (decision 13: native + blobs live on the author's PDS, never Tidepool). Link embeds + text + + nsfw + name + content/source all federate; images are dropped with a + documented gap in apobject/translator. +- **Enqueuer resolves the inbox (a network fetch on cache miss) inside the + consumer's rev-gate tx** — a slow/hanging community Group doc head-of-line + blocks the consumer's hot path (gate row locks held for fetch latency). + Rare (1h TTL), but consider pre-resolving before the gate tx or a short + distinct resolve timeout. + ## Postv2 flip (task 19) - **Legacy-set migration is deferred deliberately** (product decision diff --git a/cmd/tidepool/main.go b/cmd/tidepool/main.go index dea0b67..890a6d0 100644 --- a/cmd/tidepool/main.go +++ b/cmd/tidepool/main.go @@ -20,6 +20,7 @@ import ( "github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5/middleware" + "tidepool/internal/accept" "tidepool/internal/ap" "tidepool/internal/config" "tidepool/internal/consume" @@ -665,11 +666,32 @@ func startConsumer( } } + // The acceptance engine (task 16): a native postv2 targeting a bridged + // community is admitted here, the community-signed acceptance is written, and + // the Create{Page} enqueued atomically with it. Wired whenever the consumer + // runs, so postv2 events are admitted rather than skipped at debug. + engine, err := accept.NewEngine(accept.Options{ + Repos: repoManager, + Enqueuer: enqueuer, + Actors: minter, + Resolver: resolver, + Communities: store.NewCommunities(database), + Objects: store.NewOutboundObjects(database), + Prefs: store.NewFederationPrefs(database), + Admissions: accept.NewAdmissions(database), + UserOrigin: cfg.APUserOrigin, + Logger: logger, + }) + if err != nil { + return nil, fmt.Errorf("consumer: acceptance engine: %w", err) + } + dispatcher, err := consume.NewDispatcher(consume.Options{ DB: database, Actors: minter, Resolver: resolver, Enqueuer: enqueuer, + Engine: engine, // Reads committed records so a subject's community resolves for // mappings written before migration 016 filled community_did. Records: repoManager, diff --git a/internal/accept/admissions.go b/internal/accept/admissions.go new file mode 100644 index 0000000..3c85b7b --- /dev/null +++ b/internal/accept/admissions.go @@ -0,0 +1,110 @@ +package accept + +import ( + "context" + "database/sql" + stderrors "errors" + "fmt" + + "tidepool/internal/errors" +) + +// Admission statuses (migration 021). accepted/rejected/removed are terminal +// for a given evaluated CID; pending/pending_reacceptance are in-flight. +const ( + StatusPending = "pending" + StatusAccepted = "accepted" + StatusPendingReacceptance = "pending_reacceptance" + StatusRejected = "rejected" + StatusRemoved = "removed" +) + +// Admission is one row of the engine's decision ledger: the machine-readable +// WHY and the state a post was left in, per (community, post). It is the admin/ +// debug surface, NOT the correctness path (Coves reads state from the +// firehose-visible acceptance/removal records). +type Admission struct { + CommunityDID string + PostURI string + Status string + DecisionCode string + EvaluatedCID string + AcceptanceRKey string + AcceptedCID string + Redrivable bool +} + +// Admissions persists the decision ledger. +type Admissions struct{ db *sql.DB } + +// NewAdmissions builds the store. +func NewAdmissions(db *sql.DB) *Admissions { return &Admissions{db: db} } + +// Record upserts one admission on its own connection — the path a REJECTION +// takes, which writes no repo record and rides no commit. +func (a *Admissions) Record(ctx context.Context, adm Admission) error { + return a.record(ctx, a.db, adm) +} + +// RecordTx upserts one admission on an existing transaction — the path an +// ACCEPT takes, so the "accepted" ledger row commits with the acceptance record +// (a rolled-back acceptance must not leave an accepted admission behind). A nil +// tx is an error satisfying errors.IsValidation. +func (a *Admissions) RecordTx(ctx context.Context, tx *sql.Tx, adm Admission) error { + if tx == nil { + return errors.NewValidationError("tx", "must not be nil") + } + return a.record(ctx, tx, adm) +} + +type execer interface { + ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) +} + +func (a *Admissions) record(ctx context.Context, ex execer, adm Admission) error { + if adm.CommunityDID == "" || adm.PostURI == "" { + return errors.NewValidationError("admission", "community_did and post_uri are required") + } + if adm.Status == "" { + return errors.NewValidationError("admission.status", "must be set") + } + _, err := ex.ExecContext(ctx, ` + INSERT INTO admissions + (community_did, post_uri, status, decision_code, evaluated_cid, + acceptance_rkey, accepted_cid, redrivable) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + ON CONFLICT (community_did, post_uri) DO UPDATE SET + status = EXCLUDED.status, + decision_code = EXCLUDED.decision_code, + evaluated_cid = EXCLUDED.evaluated_cid, + acceptance_rkey = EXCLUDED.acceptance_rkey, + accepted_cid = EXCLUDED.accepted_cid, + redrivable = EXCLUDED.redrivable, + updated_at = now()`, + adm.CommunityDID, adm.PostURI, adm.Status, adm.DecisionCode, adm.EvaluatedCID, + adm.AcceptanceRKey, adm.AcceptedCID, adm.Redrivable) + if err != nil { + return fmt.Errorf("accept: record admission %s/%s: %w", adm.CommunityDID, adm.PostURI, err) + } + return nil +} + +// Get returns the admission for a (community, post), or an error satisfying +// errors.IsNotFound when the engine has never decided on it. +func (a *Admissions) Get(ctx context.Context, communityDID, postURI string) (*Admission, error) { + var adm Admission + err := a.db.QueryRowContext(ctx, ` + SELECT community_did, post_uri, status, decision_code, evaluated_cid, + acceptance_rkey, accepted_cid, redrivable + FROM admissions WHERE community_did = $1 AND post_uri = $2`, + communityDID, postURI).Scan( + &adm.CommunityDID, &adm.PostURI, &adm.Status, &adm.DecisionCode, &adm.EvaluatedCID, + &adm.AcceptanceRKey, &adm.AcceptedCID, &adm.Redrivable) + if stderrors.Is(err, sql.ErrNoRows) { + return nil, errors.NewNotFoundError("admission", communityDID+"/"+postURI) + } + if err != nil { + return nil, fmt.Errorf("accept: get admission %s/%s: %w", communityDID, postURI, err) + } + return &adm, nil +} diff --git a/internal/accept/engine.go b/internal/accept/engine.go new file mode 100644 index 0000000..3b70e6d --- /dev/null +++ b/internal/accept/engine.go @@ -0,0 +1,300 @@ +// Package accept is the acceptance engine for bridged communities. A native +// user writes a social.coves.community.postv2 into their own repo targeting a +// bridged community; Tidepool — the community's key holder — decides ADMISSION +// and, on admit, writes the community-signed acceptance record while enqueueing +// the Create/Update/Delete{Page} for Lemmy delivery ATOMICALLY with it (both +// ride ONE acceptrec commit via its side effect). Un-accepted posts are +// invisible on both sides by construction. +// +// The engine is the POLICY layer over task 19's mechanics (acceptrec: digest +// rkeys, multi-op commits) and task 15's outbound (the enqueue). It owns the +// native-author lifecycle: lazy actor mint on first accepted post, re-acceptance +// on edits, acceptance-delete on author-delete, and the admissions ledger that +// records every rejection with a machine-readable reason. +package accept + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "log/slog" + "time" + + "tidepool/internal/acceptrec" + "tidepool/internal/consume" + "tidepool/internal/errors" + "tidepool/internal/repo" + "tidepool/internal/store" +) + +// Decision codes recorded on a rejected/removed admission (migration 021's +// decision_code). Distinct codes are what the admin surface needs; the +// firehose acceptance/removal records cannot carry them. +const ( + // DecisionOptedOut: the author has an opt-out federation record, so pushing + // their post outward is exactly what they refused. + DecisionOptedOut = "opted-out" + // DecisionTitleRequired: a postv2 with no title (media-only) — Lemmy + // rejects a titleless post, and no title-derivation product decision exists. + DecisionTitleRequired = "title-required" + // DecisionTitleTooLong: over Lemmy's 200-char title cap. + DecisionTitleTooLong = "title-too-long" +) + +// lemmyTitleCap is Lemmy 0.19.20's post-title length limit. +const lemmyTitleCap = 200 + +// Options wires an Engine. +type Options struct { + // Repos is the community-repo commit surface (acceptrec drives it). A + // *repo.Manager satisfies it. + Repos acceptrec.RepoManager + // Enqueuer is the task 15 outbound seam; the engine hands it the Page intent + // as the acceptance commit's side effect. + Enqueuer consume.OutboundEnqueuer + // Actors lazily mints the native author's AP identity on the first accept. + Actors consume.ActorMinter + // Resolver bidirectionally verifies the author's handle before the first + // mint (the local part is frozen at creation). + Resolver consume.DIDResolver + // Communities resolves a community DID to its AP Group id (for the intent's + // addressing) and confirms it is still bridged. + Communities store.Communities + // Objects is the post's outbound state row (community_did, snapshot) a later + // Delete is rebuilt from — the engine owns outbound_objects for posts. + Objects store.OutboundObjects + // Prefs reads the author's federation preference: the opt-out check MOVED + // here from the consumer, so an opted-out author's post reaches the engine + // and is RECORDED as a rejection rather than silently dropped upstream. + Prefs store.FederationPrefs + // Admissions is the decision ledger (migration 021). + Admissions *Admissions + // UserOrigin is AP_USER_ORIGIN: the origin every deterministic activity id + // is minted under. + UserOrigin string + // Logger receives drop reasons. Nil uses slog.Default(). + Logger *slog.Logger +} + +// Engine admits native posts into bridged communities. +type Engine struct { + repos acceptrec.RepoManager + enqueuer consume.OutboundEnqueuer + actors consume.ActorMinter + resolver consume.DIDResolver + communities store.Communities + objects store.OutboundObjects + prefs store.FederationPrefs + admissions *Admissions + userOrigin string + logger *slog.Logger +} + +// The engine is the task 16 acceptance seam the dispatcher hands postv2 commits +// to. +var _ consume.AcceptanceEngine = (*Engine)(nil) + +// NewEngine wires an Engine. All seams are required except Logger. +func NewEngine(opts Options) (*Engine, error) { + switch { + case opts.Repos == nil: + return nil, errors.NewValidationError("Repos", "must not be nil") + case opts.Enqueuer == nil: + return nil, errors.NewValidationError("Enqueuer", "must not be nil") + case opts.Actors == nil: + return nil, errors.NewValidationError("Actors", "must not be nil") + case opts.Resolver == nil: + return nil, errors.NewValidationError("Resolver", "must not be nil") + case opts.Communities == nil: + return nil, errors.NewValidationError("Communities", "must not be nil") + case opts.Objects == nil: + return nil, errors.NewValidationError("Objects", "must not be nil") + case opts.Prefs == nil: + return nil, errors.NewValidationError("Prefs", "must not be nil") + case opts.Admissions == nil: + return nil, errors.NewValidationError("Admissions", "must not be nil") + case opts.UserOrigin == "": + return nil, errors.NewValidationError("UserOrigin", "must not be empty") + } + logger := opts.Logger + if logger == nil { + logger = slog.Default() + } + return &Engine{ + repos: opts.Repos, + enqueuer: opts.Enqueuer, + actors: opts.Actors, + resolver: opts.Resolver, + communities: opts.Communities, + objects: opts.Objects, + prefs: opts.Prefs, + admissions: opts.Admissions, + userOrigin: opts.UserOrigin, + logger: logger, + }, nil +} + +// AdmitPost decides admission for one postv2 commit and, on admit, writes the +// community acceptance record while enqueueing the Page delivery atomically with +// it. create/update run admission on the event's content; delete takes the +// acceptance down and enqueues Delete{Page} from stored state. +func (e *Engine) AdmitPost(ctx context.Context, did string, commit *consume.CommitEvent) error { + // This cycle handles create/update admission. A delete (author retraction) + // takes the acceptance down and enqueues Delete{Page} from stored state; + // that lifecycle lands next cycle. A delete carries no record body, so there + // is nothing to admit or reject here yet. + if commit.Operation == operationDelete { + return nil + } + + postURI := fmt.Sprintf("at://%s/%s/%s", did, commit.Collection, commit.RKey) + communityDID, _ := commit.Record["community"].(string) + if communityDID == "" { + // The consumer already refuses a postv2 with no community, but the engine + // re-asserts it: WE sign the acceptance, so a missing target is fail-closed. + return errors.NewValidationError("community", "postv2 "+commit.RKey+" names no community") + } + + // The opt-out check MOVED here from the consumer: an opted-out author's post + // REACHES the engine and is RECORDED as a rejection with a distinct + // machine-readable reason — no acceptance written, nothing enqueued. Content + // authored while opted out never federates (decision 11). + federating, err := e.mayFederate(ctx, did) + if err != nil { + return err + } + if !federating { + e.logger.Debug("rejecting postv2 from an opted-out author", + slog.String("did", did), slog.String("post", postURI)) + return e.admissions.Record(ctx, Admission{ + CommunityDID: communityDID, + PostURI: postURI, + Status: StatusRejected, + DecisionCode: DecisionOptedOut, + EvaluatedCID: commit.CID, + }) + } + + // The community's AP Group id is the Page's addressing target; the lookup + // also re-confirms the community is one we federate. + community, err := e.communities.GetByDID(ctx, communityDID) + if err != nil { + return fmt.Errorf("accept: resolve community %s: %w", communityDID, err) + } + + // Lazy-mint the author BEFORE the acceptance tx: the outbound enqueue (the + // acceptance commit's side effect) resolves the author's AP actor by DID on + // the community-repo tx, so the actor row must already be committed and + // visible when that side effect runs. This is the first federating + // interaction, so the mint happens here rather than eagerly. + if err := e.ensureActor(ctx, did); err != nil { + return err + } + + snapshot, err := json.Marshal(map[string]any{ + "atUri": postURI, + "cid": commit.CID, + "rev": commit.Rev, + "collection": commit.Collection, + "record": commit.Record, + "communityApId": community.APGroupID, + }) + if err != nil { + return fmt.Errorf("accept: snapshot %s: %w", postURI, err) + } + + rkey := acceptrec.SubjectRKey(postURI) + apObjectID := e.userOrigin + "/ap/object/" + did + "/" + commit.Collection + "/" + commit.RKey + + // The side effect rides the acceptance commit's transaction: the outbound + // state row, the delivery enqueue, and the accepted ledger row all land WITH + // the acceptance record or not at all. A failing enqueue returns the error, + // which rolls the acceptance back too (ApplyOpsTx side-effect atomicity), and + // AdmitPost propagates it so the event retries. + sideEffect := func(sctx context.Context, tx *sql.Tx, _ *repo.CommitResult) error { + stored, err := e.objects.UpsertTx(sctx, tx, store.OutboundObject{ + ATURI: postURI, + APObjectID: apObjectID, + LastCID: commit.CID, + LastRev: commit.Rev, + CommunityDID: communityDID, + CommunityAPID: community.APGroupID, + TranslatedSnapshot: snapshot, + Depth: 0, + }) + if err != nil { + return fmt.Errorf("accept: write outbound state for %s: %w", postURI, err) + } + intent := consume.PostIntent{ + Op: commit.Operation, + ATURI: postURI, + ID: consume.ActivityID(e.userOrigin, postURI, commit.Operation, stored.LastActivitySeq), + CommunityAPID: community.APGroupID, + Snapshot: snapshot, + } + // A post has no causal parent, so orderingKey is the author DID and there + // is no parentATURI. + if err := e.enqueuer.EnqueueActivity(sctx, tx, did, did, "", intent); err != nil { + return err + } + return e.admissions.RecordTx(sctx, tx, Admission{ + CommunityDID: communityDID, + PostURI: postURI, + Status: StatusAccepted, + EvaluatedCID: commit.CID, + AcceptanceRKey: rkey, + AcceptedCID: commit.CID, + }) + } + + if _, err := acceptrec.AcceptSubject(ctx, e.repos, communityDID, postURI, commit.CID, + publishedAtOf(commit.Record), sideEffect); err != nil { + return fmt.Errorf("accept: admit %s into %s: %w", postURI, communityDID, err) + } + return nil +} + +// operationDelete is the Jetstream commit operation for a record deletion. +const operationDelete = "delete" + +// mayFederate reports whether the author permits outbound federation. A missing +// preference MEANS default-on (decision 11), not unknown. +func (e *Engine) mayFederate(ctx context.Context, did string) (bool, error) { + pref, err := e.prefs.Get(ctx, did) + if errors.IsNotFound(err) { + return true, nil + } + if err != nil { + return false, fmt.Errorf("accept: read federation preference for %s: %w", did, err) + } + return pref.Enabled, nil +} + +// ensureActor lazily mints the author's AP identity. CreateActorForDID is +// get-or-create, so a redelivery (or a second accepted post) reuses the existing +// actor rather than minting a second. The handle is resolved through the +// bidirectional verifier because the local part is frozen at creation. +func (e *Engine) ensureActor(ctx context.Context, did string) error { + handle, err := e.resolver.ResolveDIDHandle(ctx, did) + if err != nil { + return fmt.Errorf("accept: resolve handle for %s: %w", did, err) + } + if _, err := e.actors.CreateActorForDID(ctx, did, handle); err != nil { + return fmt.Errorf("accept: mint actor for %s: %w", did, err) + } + return nil +} + +// publishedAtOf derives the timestamp the acceptance's createdAt is rendered +// from — the post's own createdAt, so a redelivery re-puts byte-identical bytes +// and the repo layer's no-op path absorbs it. An unparseable or absent value +// falls back to the zero time, which is still deterministic. +func publishedAtOf(record map[string]any) time.Time { + if s, ok := record["createdAt"].(string); ok && s != "" { + if t, err := time.Parse(time.RFC3339, s); err == nil { + return t + } + } + return time.Time{} +} diff --git a/internal/accept/outer_acceptance_test.go b/internal/accept/outer_acceptance_test.go new file mode 100644 index 0000000..006adf2 --- /dev/null +++ b/internal/accept/outer_acceptance_test.go @@ -0,0 +1,397 @@ +package accept + +import ( + "context" + "database/sql" + "encoding/json" + stderrors "errors" + "fmt" + "testing" + + "github.com/bluesky-social/indigo/atproto/atcrypto" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/acceptrec" + "tidepool/internal/consume" + "tidepool/internal/identity" + "tidepool/internal/outbound" + "tidepool/internal/personas" + "tidepool/internal/repo" + "tidepool/internal/store" + "tidepool/internal/testutil" +) + +// The world this outer acceptance test builds. +const ( + acUserOrigin = "https://coves.social" + + acCommunityDID = "did:plc:44ybard66vv44zksje25o7dz" + acCommunityAPID = "https://lemmy.world/c/technology" + acCommunityHost = "lemmy.world" + acCommunityName = "technology" + acCommunityInbox = "https://lemmy.world/c/technology/inbox" + + // The author. Unseen: no ap_actors, no repo_state, no anything. Their actor + // must be minted by the act of getting a post accepted. + acAuthorDID = "did:plc:7iza6de2dwap2sbkpav7c6c6" + acAuthorHandle = "author.coves.social" + + acPostRKey = "3lzpostaaaa11" + acPostCID = "bafyreievgu2ty7qbiaaom5zhmkznsnajuzideek3lo7e65dwqlrvrxnmo4" + acPostRev = "3lzpostrev001" + acPostURI = "at://" + acAuthorDID + "/social.coves.community.postv2/" + acPostRKey + + acPostTimeUS = int64(1_775_000_000_000_000) +) + +// acKEK seals minted actors' AP RSA keys (32 bytes, AES-256). +var acKEK = []byte("0123456789abcdef0123456789abcdef") + +// TestEngineAdmitsUnseenNativePost is the OUTER acceptance test for task 16. +// +// GIVEN a bridged community and an UNSEEN native author, WHEN a postv2 create +// event targeting that community arrives through the dispatcher→engine seam, +// THEN: +// +// 1. the author's AP actor is lazily minted (first federating interaction); +// 2. a social.coves.community.acceptance record exists in the COMMUNITY repo at +// the digest rkey, pinning the postv2 uri + CID; +// 3. EXACTLY ONE Create{Page} activity is enqueued (outbound_activities + +// outbound_deliveries) under the deterministic activity id; +// 4. an outbound_objects row for the post exists (community_did + snapshot) — +// the state a later Delete{Page} is rebuilt from; +// 5. an admissions row records status=accepted. +// +// AND a full replay of the same event enqueues nothing new and writes no second +// acceptance (the rev gate + idempotent acceptance/enqueue). +// +// No network: the community inbox is resolved through a stub, and minting is +// local (RSA keygen + a DB row, no PLC). +func TestEngineAdmitsUnseenNativePost(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + + seedBridgedCommunity(t, conn) + requireUnseenAuthor(t, conn) + + repos := newRepos(t, conn) + engine := wireEngine(t, conn, repos, realEnqueuer(t, conn)) + dispatcher := wireDispatcher(t, conn, engine, realEnqueuer(t, conn)) + + require.NoError(t, dispatcher.HandleEvent(ctx, postV2CreateEvent(t)), + "a well-formed postv2 to a bridged community must be admitted, not dead-lettered") + + // 1. Lazy mint. + assert.Equal(t, 1, countWhere(t, conn, "ap_actors", "did", acAuthorDID), + "the author's first accepted post must lazily mint exactly one AP actor") + + // 2. The acceptance record in the community repo. + rkey := acceptrec.SubjectRKey(acPostURI) + record, _, err := repos.GetRecord(ctx, acCommunityDID, acceptrec.CollectionAcceptance, rkey) + require.NoError(t, err, + "a social.coves.community.acceptance must exist in the COMMUNITY repo at SubjectRKey(%s)", acPostURI) + assert.Equal(t, acceptrec.CollectionAcceptance, record["$type"]) + subject, ok := record["subject"].(map[string]any) + require.True(t, ok, "the acceptance pins a strongRef subject, got %v", record["subject"]) + assert.Equal(t, acPostURI, subject["uri"], "the acceptance pins the postv2 at-uri") + assert.Equal(t, acPostCID, subject["cid"], "the acceptance pins the postv2 CID it evaluated") + + // 3. Exactly one Create{Page} activity + one delivery, deterministic id. + require.Equal(t, 1, countRows(t, conn, "outbound_activities"), + "admission enqueues exactly one canonical activity for the post") + require.Equal(t, 1, countRows(t, conn, "outbound_deliveries"), + "...and exactly one per-community delivery") + + wantID := consume.ActivityID(acUserOrigin, acPostURI, "create", 0) + var kind, objType string + require.NoError(t, conn.QueryRowContext(ctx, + `SELECT kind, payload->'object'->>'type' FROM outbound_activities WHERE activity_id = $1`, + wantID).Scan(&kind, &objType), + "the activity is keyed by the deterministic id ActivityID(origin, postURI, create, 0)") + assert.Equal(t, "Create", kind, "a post create federates as Create{Page}") + assert.Equal(t, "Page", objType, "the inner object is a Page (the Note/Page addressing split)") + + // 4. Outbound state for the post. + stored, err := store.NewOutboundObjects(conn).GetByATURI(ctx, acPostURI) + require.NoError(t, err, + "outbound_objects must hold the post's state: a later Delete{Page} carries no body") + assert.Equal(t, acCommunityDID, stored.CommunityDID, "the post's community is recorded") + assert.NotEmpty(t, stored.TranslatedSnapshot, "the snapshot the Delete is rebuilt from is stored") + + // 5. The admissions ledger. + status, code := admissionOf(t, conn, acCommunityDID, acPostURI) + assert.Equal(t, StatusAccepted, status, "the ledger records the post as accepted") + assert.Empty(t, code, "a clean accept carries no rejection code") + + // ------------------------------------------------------------------- + // Replay: the same event enqueues nothing new and writes no second acceptance. + // ------------------------------------------------------------------- + require.NoError(t, dispatcher.HandleEvent(ctx, postV2CreateEvent(t)), + "a replay is a normal skip, so the cursor still advances") + + assert.Equal(t, 1, countRows(t, conn, "outbound_activities"), + "a replay must not enqueue a second activity (rev gate + idempotent id)") + assert.Equal(t, 1, countRows(t, conn, "outbound_deliveries")) + assert.Equal(t, 1, countWhere(t, conn, "ap_actors", "did", acAuthorDID), + "a replay must not mint a second actor") +} + +// TestEngineCrashInjectionIsAtomic is the crash-injection DoD: a fault-injected +// enqueuer that errors inside the acceptance commit's side effect must leave +// NEITHER the acceptance record NOR the outbound rows — the acceptance write and +// the enqueue are one transaction (acceptrec's side-effect commit), so a failing +// enqueue rolls the acceptance back too, and the failure PROPAGATES for retry. +func TestEngineCrashInjectionIsAtomic(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + + seedBridgedCommunity(t, conn) + repos := newRepos(t, conn) + + faulty := &faultyEnqueuer{err: stderrors.New("enqueue exploded mid-commit")} + engine := wireEngine(t, conn, repos, faulty) + dispatcher := wireDispatcher(t, conn, engine, faulty) + + err := dispatcher.HandleEvent(ctx, postV2CreateEvent(t)) + require.Error(t, err, + "an enqueue failure inside the acceptance commit must PROPAGATE so the event retries — "+ + "a swallowed error would strand a post with an acceptance nobody delivered") + + rkey := acceptrec.SubjectRKey(acPostURI) + _, _, gerr := repos.GetRecord(ctx, acCommunityDID, acceptrec.CollectionAcceptance, rkey) + assert.Error(t, gerr, "no acceptance record may survive a failed enqueue (atomic)") + + assert.Zero(t, countRows(t, conn, "outbound_activities"), + "and no outbound activity: the whole acceptance+enqueue transaction rolled back") + assert.Zero(t, countRows(t, conn, "outbound_deliveries")) +} + +// TestEngineRecordsOptedOutRejection pins the opt-out check MOVED into the +// engine: an opted-out author's postv2 now reaches the engine (the consumer no +// longer gates it), and the engine RECORDS a rejection with decision_code +// opted-out — writing no acceptance and enqueueing nothing. +func TestEngineRecordsOptedOutRejection(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + + seedBridgedCommunity(t, conn) + _, err := store.NewFederationPrefs(conn).Upsert(ctx, store.FederationPref{ + DID: acAuthorDID, + Source: store.FederationPrefSourceRecord, + }) + require.NoError(t, err, "seed the author's opt-out record") + + repos := newRepos(t, conn) + engine := wireEngine(t, conn, repos, realEnqueuer(t, conn)) + dispatcher := wireDispatcher(t, conn, engine, realEnqueuer(t, conn)) + + require.NoError(t, dispatcher.HandleEvent(ctx, postV2CreateEvent(t)), + "an opted-out author's post is a recorded rejection, not a failure to retry") + + status, code := admissionOf(t, conn, acCommunityDID, acPostURI) + assert.Equal(t, StatusRejected, status, "the engine records the post as rejected") + assert.Equal(t, DecisionOptedOut, code, + "the rejection carries a distinct machine-readable reason the admin surface can show") + + rkey := acceptrec.SubjectRKey(acPostURI) + _, _, gerr := repos.GetRecord(ctx, acCommunityDID, acceptrec.CollectionAcceptance, rkey) + assert.Error(t, gerr, "a rejected post gets no acceptance record") + assert.Zero(t, countRows(t, conn, "outbound_activities"), + "and nothing is federated for a post authored under an opt-out") +} + +// --------------------------------------------------------------------------- +// Harness +// --------------------------------------------------------------------------- + +func acceptanceDB(t *testing.T) *sql.DB { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, + "ap_actors", "ap_objects", "communities", "repo_state", "blocks", "firehose_events", + "outbound_activities", "outbound_deliveries", "outbound_objects", "outbound_votes", + "federation_prefs", "jetstream_record_revs", "jetstream_dead_letters", "admissions") + return database +} + +// staticKeys signs the community's acceptance commit with one fixed key. +type staticKeys struct{ key *atcrypto.PrivateKeyK256 } + +func (s staticKeys) SigningKey(context.Context, string, repo.KeyUse) (atcrypto.PrivateKey, error) { + return s.key, nil +} + +func newRepos(t *testing.T, conn *sql.DB) *repo.Manager { + t.Helper() + key, err := atcrypto.GeneratePrivateKeyK256() + require.NoError(t, err) + manager, err := repo.NewManager(conn, staticKeys{key: key}, nil) + require.NoError(t, err) + return manager +} + +func newMinter(t *testing.T, conn *sql.DB) *personas.Service { + t.Helper() + custodian, err := identity.NewCustodian(acKEK) + require.NoError(t, err) + svc, err := personas.New(personas.Options{DB: conn, Custodian: custodian, UserOrigin: acUserOrigin}) + require.NoError(t, err) + return svc +} + +func realEnqueuer(t *testing.T, conn *sql.DB) *outbound.Enqueuer { + t.Helper() + enq, err := outbound.NewEnqueuer(outbound.EnqueuerOptions{ + DB: conn, + Translator: outbound.NewTranslator(acUserOrigin), + Inboxes: stubInboxResolver{}, + Actors: store.NewAPActors(conn), + UserOrigin: acUserOrigin, + }) + require.NoError(t, err) + return enq +} + +func wireEngine(t *testing.T, conn *sql.DB, repos *repo.Manager, enqueuer consume.OutboundEnqueuer) *Engine { + t.Helper() + engine, err := NewEngine(Options{ + Repos: repos, + Enqueuer: enqueuer, + Actors: newMinter(t, conn), + Resolver: stubResolver{}, + Communities: store.NewCommunities(conn), + Objects: store.NewOutboundObjects(conn), + Prefs: store.NewFederationPrefs(conn), + Admissions: NewAdmissions(conn), + UserOrigin: acUserOrigin, + }) + require.NoError(t, err) + return engine +} + +func wireDispatcher(t *testing.T, conn *sql.DB, engine consume.AcceptanceEngine, enqueuer consume.OutboundEnqueuer) *consume.Dispatcher { + t.Helper() + dispatcher, err := consume.NewDispatcher(consume.Options{ + DB: conn, + Actors: newMinter(t, conn), + Enqueuer: enqueuer, + Resolver: stubResolver{}, + Engine: engine, + UserOrigin: acUserOrigin, + }) + require.NoError(t, err) + return dispatcher +} + +// seedBridgedCommunity registers the community (bridged = a communities row). It +// deliberately gets NO repo_state row: the engine genesis-commits the community +// repo on the first acceptance, exactly as production does. +func seedBridgedCommunity(t *testing.T, conn *sql.DB) { + t.Helper() + _, err := store.NewCommunities(conn).UpsertCommunity(context.Background(), store.Community{ + APGroupID: acCommunityAPID, + DID: acCommunityDID, + PreferredUsername: acCommunityName, + Instance: acCommunityHost, + FollowState: store.FollowStateAccepted, + }) + require.NoError(t, err, "seed bridged community") +} + +func requireUnseenAuthor(t *testing.T, conn *sql.DB) { + t.Helper() + assert.Zero(t, countWhere(t, conn, "ap_actors", "did", acAuthorDID), + "the author must be unseen so the lazy-mint assertion is real") + assert.Zero(t, countWhere(t, conn, "repo_state", "did", acAuthorDID)) +} + +// postV2CreateEvent builds the postv2 create frame as literal wire JSON (pinning +// the shape) and parses it into the JetstreamEvent the dispatcher consumes. +func postV2CreateEvent(t *testing.T) *consume.JetstreamEvent { + t.Helper() + frame := []byte(fmt.Sprintf(`{ + "did": %q, + "time_us": %d, + "kind": "commit", + "commit": { + "rev": %q, + "operation": "create", + "collection": "social.coves.community.postv2", + "rkey": %q, + "cid": %q, + "record": { + "$type": "social.coves.community.postv2", + "community": %q, + "title": "hello from atproto", + "content": "the body of the post", + "createdAt": "2026-08-12T10:00:00.000Z" + } + } +}`, acAuthorDID, acPostTimeUS, acPostRev, acPostRKey, acPostCID, acCommunityDID)) + + var event consume.JetstreamEvent + require.NoError(t, json.Unmarshal(frame, &event), "the test frame must be valid wire JSON") + return &event +} + +// --------------------------------------------------------------------------- +// Test doubles +// --------------------------------------------------------------------------- + +// stubResolver returns the author's handle without touching the network (the +// handle verification itself has its own tests in consume). +type stubResolver struct{} + +func (stubResolver) ResolveDIDHandle(context.Context, string) (string, error) { + return acAuthorHandle, nil +} + +// stubInboxResolver answers the community's inbox from a fixed map — the outer +// test never dials Lemmy. +type stubInboxResolver struct{} + +func (stubInboxResolver) ResolveInbox(_ context.Context, communityAPID string) (string, error) { + if communityAPID == acCommunityAPID { + return acCommunityInbox, nil + } + return "", fmt.Errorf("no inbox route for %s", communityAPID) +} + +// faultyEnqueuer fails EnqueueActivity to exercise the crash-injection atomicity +// path: it stands in for a delivery seam that errors inside the acceptance +// commit's side effect. +type faultyEnqueuer struct{ err error } + +func (f *faultyEnqueuer) EnqueueActivity(context.Context, *sql.Tx, string, string, string, consume.Intent) error { + return f.err +} + +// --------------------------------------------------------------------------- +// Query helpers +// --------------------------------------------------------------------------- + +func countRows(t *testing.T, conn *sql.DB, table string) int { + t.Helper() + var n int + require.NoError(t, conn.QueryRowContext(context.Background(), `SELECT COUNT(*) FROM `+table).Scan(&n)) + return n +} + +func countWhere(t *testing.T, conn *sql.DB, table, col, val string) int { + t.Helper() + var n int + require.NoError(t, conn.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM `+table+` WHERE `+col+` = $1`, val).Scan(&n)) + return n +} + +func admissionOf(t *testing.T, conn *sql.DB, communityDID, postURI string) (status, code string) { + t.Helper() + err := conn.QueryRowContext(context.Background(), + `SELECT status, decision_code FROM admissions WHERE community_did = $1 AND post_uri = $2`, + communityDID, postURI).Scan(&status, &code) + require.NoError(t, err, + "an admissions row must exist for (%s, %s): the engine records every decision", communityDID, postURI) + return status, code +} diff --git a/internal/acceptrec/acceptrec.go b/internal/acceptrec/acceptrec.go new file mode 100644 index 0000000..b75ff7a --- /dev/null +++ b/internal/acceptrec/acceptrec.go @@ -0,0 +1,236 @@ +// Package acceptrec owns the shared mechanics of a community's acceptance and +// removal records: the digest record key both share (SubjectRKey), and the +// multi-op, side-effect-parameterized commits that write, repin and delete +// them. It is a LEAF: it depends only on internal/repo (the commit primitive) +// and internal/errors, so both the materializer (which drives it with a nil +// side effect, characterizing the existing behavior) and the acceptance engine +// (which drives it with an outbound enqueue as the side effect) build on ONE +// implementation of the acceptance-commit discipline instead of two that drift. +// +// The subject of every record here is the post's atproto at-uri. Two engines +// write into the same community repos — Coves' own for native posts, this +// bridge for bridged ones — so SubjectRKey MUST byte-match Coves' derivation +// or a post acquires two acceptance keys and neither side can see the other's. +package acceptrec + +import ( + "context" + "crypto/sha256" + "encoding/base32" + stderrors "errors" + "fmt" + "strings" + "time" + + "tidepool/internal/errors" + "tidepool/internal/repo" +) + +// maxAcceptAttempts bounds the CAS retry loop: a subject whose acceptance keeps +// moving underneath the read-modify-write (a racing repin or a stats stamp) +// eventually gives up rather than spinning forever. Matches the materializer's +// stats-commit cap it was lifted from. +const maxAcceptAttempts = 4 + +// Record collections written here. They must match Coves' and the +// materializer's exactly (a post's acceptance and removal share ONE digest +// rkey, one per subject). +const ( + // CollectionAcceptance is the community's attestation that it accepts a + // post — what makes a postv2 visible in the community at all. + CollectionAcceptance = "social.coves.community.acceptance" + // CollectionRemoval is the community's record that a post was removed. It + // shares the acceptance's digest rkey and replaces it in one atomic commit. + CollectionRemoval = "social.coves.community.removal" +) + +// ErrRemovalStands is returned by AcceptSubject when a removal record already +// stands at the subject's rkey. A removal is terminal — exited only by an +// explicit restore — so a fresh acceptance is refused, the acceptance record is +// NOT written, and the side effect is NOT run. The caller (the engine) treats +// it as a decided-out post, not a failure to retry. +var ErrRemovalStands = stderrors.New("acceptrec: a removal stands at the subject rkey") + +// RepoManager is the slice of *repo.Manager the acceptance commits need: the +// multi-op, side-effect-parameterized commit and a point read of the current +// acceptance (to carry createdAt forward and derive the CAS precondition). +type RepoManager interface { + ApplyOpsTx(ctx context.Context, did string, ops []repo.RecordOp, sideEffect repo.TxSideEffect) (*repo.CommitResult, error) + GetRecord(ctx context.Context, did, collection, rkey string) (record map[string]any, recordCID string, err error) +} + +// subjectRKeyEncoding is RFC 4648 base32 (STANDARD alphabet), padding dropped +// and lowercased so the key stays inside the atProto record-key charset. This +// must match materialize.SubjectRKey and Coves' posts.SubjectRkey byte for byte. +var subjectRKeyEncoding = base32.StdEncoding.WithPadding(base32.NoPadding) + +// SubjectRKey derives the record key a community's acceptance and removal for +// one post share: the unpadded lowercase base32 of SHA-256 of the post's +// at-uri, a fixed 52 rkey-safe characters. The argument is BYTES, not a parsed +// URI — no normalization — because those bytes are the identity Coves indexes +// under. See the golden vectors in the test beside this file. +func SubjectRKey(subjectATURI string) string { + digest := sha256.Sum256([]byte(subjectATURI)) + return strings.ToLower(subjectRKeyEncoding.EncodeToString(digest[:])) +} + +// AcceptSubject writes the community's acceptance of one post at SubjectRKey, +// with a removal guard (an inert delete of the removal at that rkey with a +// "must not exist" precondition, so the batch fails if a removal stands), and +// runs sideEffect INSIDE the acceptance commit transaction — the outbound +// enqueue rides here, so acceptance and enqueue land together or not at all. +// +// createdAt is derived deterministically (carried forward from any existing +// acceptance, else from publishedAt) so a redelivery re-puts byte-identical +// bytes and the repo layer's NoOp path absorbs it — but the side effect STILL +// runs on that NoOp path (at-least-once enqueue). A standing removal returns +// ErrRemovalStands with no acceptance written and the side effect not run. +func AcceptSubject(ctx context.Context, repos RepoManager, communityDID, subjectURI, subjectCID string, publishedAt time.Time, sideEffect repo.TxSideEffect) (*repo.CommitResult, error) { + rkey := SubjectRKey(subjectURI) + + // Read-modify-write under a CAS precondition: the read happens outside the + // commit serialization, so a racing repin (or a stats-driven CID change on + // the subject) can move the record underneath it. Losing that race means + // re-reading, never overwriting. + for attempt := 0; ; attempt++ { + createdAt := recordDatetime(publishedAt) + expectPrevCID := "" + + stored, storedCID, err := repos.GetRecord(ctx, communityDID, CollectionAcceptance, rkey) + switch { + case err == nil: + expectPrevCID = storedCID + // createdAt is carried forward, never restamped: the community + // accepted this post once, and re-pinning the version it accepts is + // not a new acceptance. Deriving it (rather than stamping a clock) + // is what makes a redelivery re-put byte-identical bytes. + if when, ok := stored["createdAt"].(string); ok && when != "" { + createdAt = when + } + case errors.IsNotFound(err): + // Either the first acceptance or a crash-window heal. The empty + // precondition asserts the record is still absent, so a concurrent + // writer that got there first sends us round the loop instead of + // clobbering its acceptance. + default: + return nil, fmt.Errorf("acceptrec: read acceptance %s/%s/%s: %w", + communityDID, CollectionAcceptance, rkey, err) + } + + record := map[string]any{ + "$type": CollectionAcceptance, + "subject": strongRef(subjectURI, subjectCID), + "createdAt": createdAt, + } + + // The terminality guard is an inert op: deleting the removal at this + // rkey with a precondition of "must not exist" claims nothing and + // changes nothing, but it makes the batch FAIL if a removal has appeared + // since the read above — so the refusal holds at COMMIT time, under the + // commit's own locks, not merely at read time. Without it a RemovePost + // landing in the window would leave an acceptance beside a standing + // removal, a post simultaneously visible and removed. + noRemoval := "" + res, err := repos.ApplyOpsTx(ctx, communityDID, []repo.RecordOp{ + {Action: repo.OpActionDelete, Collection: CollectionRemoval, RKey: rkey, ExpectPrevCID: &noRemoval}, + {Action: repo.OpActionUpdate, Collection: CollectionAcceptance, RKey: rkey, Record: record, ExpectPrevCID: &expectPrevCID}, + }, sideEffect) + if stderrors.Is(err, repo.ErrPreconditionFailed) { + // Either a removal appeared or the acceptance moved. Ask which: a + // removal means the community has since decided this post is out, and + // that decision is terminal — the acceptance is refused and the side + // effect (which the failed batch never reached) does not fire. + removed, rerr := removalStands(ctx, repos, communityDID, rkey) + if rerr != nil { + return nil, rerr + } + if removed { + return nil, ErrRemovalStands + } + if attempt+1 < maxAcceptAttempts { + continue + } + return nil, fmt.Errorf("acceptrec: accept %s: acceptance kept changing across %d attempts: %w", + subjectURI, maxAcceptAttempts, err) + } + if err != nil { + return nil, fmt.Errorf("acceptrec: put acceptance %s/%s/%s: %w", + communityDID, CollectionAcceptance, rkey, err) + } + return res, nil + } +} + +// removalStands reports whether the community currently holds a removal for the +// subject at rkey. Acceptance and removal share the digest key, so this is a +// point lookup rather than a search. +func removalStands(ctx context.Context, repos RepoManager, communityDID, rkey string) (bool, error) { + _, _, err := repos.GetRecord(ctx, communityDID, CollectionRemoval, rkey) + switch { + case err == nil: + return true, nil + case errors.IsNotFound(err): + return false, nil + default: + return false, fmt.Errorf("acceptrec: read removal %s/%s/%s: %w", + communityDID, CollectionRemoval, rkey, err) + } +} + +// recordDatetime renders a timestamp in the atproto datetime format (RFC3339, +// UTC, millisecond precision). It must match materialize.recordDatetime byte for +// byte so a post accepted by either engine yields identical acceptance bytes. +func recordDatetime(t time.Time) string { + return t.UTC().Format("2006-01-02T15:04:05.000Z") +} + +// strongRef builds a com.atproto.repo.strongRef value. +func strongRef(uri, cid string) map[string]any { + return map[string]any{"uri": uri, "cid": cid} +} + +// DeleteAcceptance removes a post's acceptance from its community and runs +// sideEffect inside that commit — the author-delete path, whose side effect is +// the outbound Delete{Page} enqueue. A missing acceptance is inert (the batch +// commits nothing), but the side effect still runs so the retraction still goes +// out. +func DeleteAcceptance(ctx context.Context, repos RepoManager, communityDID, subjectURI string, sideEffect repo.TxSideEffect) (*repo.CommitResult, error) { + rkey := SubjectRKey(subjectURI) + res, err := repos.ApplyOpsTx(ctx, communityDID, []repo.RecordOp{ + {Action: repo.OpActionDelete, Collection: CollectionAcceptance, RKey: rkey}, + }, sideEffect) + if err != nil { + return nil, fmt.Errorf("acceptrec: delete acceptance %s/%s/%s: %w", + communityDID, CollectionAcceptance, rkey, err) + } + return res, nil +} + +// Remove atomically deletes the acceptance and writes a removal at the shared +// rkey (moderation / edit-fail), running sideEffect (the Delete{Page} enqueue) +// inside the commit. +func Remove(ctx context.Context, repos RepoManager, communityDID, subjectURI, subjectCID, code, reason string, at time.Time, sideEffect repo.TxSideEffect) (*repo.CommitResult, error) { + rkey := SubjectRKey(subjectURI) + removal := map[string]any{ + "$type": CollectionRemoval, + "subject": strongRef(subjectURI, subjectCID), + "code": code, + "createdAt": recordDatetime(at), + } + // Omitted rather than written blank: an empty reason renders in a moderation + // log as a blank explanation instead of as none given. + if reason != "" { + removal["reason"] = reason + } + // One commit: the acceptance is withdrawn and the removal written together, + // so the firehose never shows a window where the post is neither accepted + // nor removed. + res, err := repos.ApplyOpsTx(ctx, communityDID, []repo.RecordOp{ + {Action: repo.OpActionDelete, Collection: CollectionAcceptance, RKey: rkey}, + {Action: repo.OpActionUpdate, Collection: CollectionRemoval, RKey: rkey, Record: removal}, + }, sideEffect) + if err != nil { + return nil, fmt.Errorf("acceptrec: remove %s from %s: %w", subjectURI, communityDID, err) + } + return res, nil +} diff --git a/internal/acceptrec/acceptrec_test.go b/internal/acceptrec/acceptrec_test.go new file mode 100644 index 0000000..c02fdd1 --- /dev/null +++ b/internal/acceptrec/acceptrec_test.go @@ -0,0 +1,203 @@ +package acceptrec + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/bluesky-social/indigo/atproto/atcrypto" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/errors" + "tidepool/internal/repo" + "tidepool/internal/testutil" +) + +// The world these pins run in. +const ( + arCommunityDID = "did:plc:44ybard66vv44zksje25o7dz" + arAuthorDID = "did:plc:7iza6de2dwap2sbkpav7c6c6" + arPostRKey = "3lzpostaaaa11" + arPostURI = "at://" + arAuthorDID + "/social.coves.community.postv2/" + arPostRKey + arPostCID = "bafyreievgu2ty7qbiaaom5zhmkznsnajuzideek3lo7e65dwqlrvrxnmo4" +) + +var arPublishedAt = time.Date(2026, 8, 12, 10, 0, 0, 0, time.UTC) + +// staticKeys signs every DID with one fixed key — acceptrec only needs the repo +// layer to be able to sign the community's acceptance commit; key custody has +// its own tests. +type staticKeys struct{ key *atcrypto.PrivateKeyK256 } + +func (s staticKeys) SigningKey(context.Context, string, repo.KeyUse) (atcrypto.PrivateKey, error) { + return s.key, nil +} + +func newRepos(t *testing.T) (*repo.Manager, *sql.DB) { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, "blocks", "repo_state", "firehose_events") + key, err := atcrypto.GeneratePrivateKeyK256() + require.NoError(t, err) + manager, err := repo.NewManager(database, staticKeys{key: key}, nil) + require.NoError(t, err) + return manager, database +} + +// marker is a scratch table a side effect writes into, so a test can prove the +// side effect ran (and committed) or did not. +func newMarker(t *testing.T, database *sql.DB) { + t.Helper() + _, err := database.Exec(`CREATE TABLE IF NOT EXISTS acceptrec_marker (note TEXT NOT NULL)`) + require.NoError(t, err) + _, err = database.Exec(`TRUNCATE acceptrec_marker`) + require.NoError(t, err) + t.Cleanup(func() { _, _ = database.Exec(`DROP TABLE IF EXISTS acceptrec_marker`) }) +} + +func markerCount(t *testing.T, database *sql.DB) int { + t.Helper() + var n int + require.NoError(t, database.QueryRow(`SELECT COUNT(*) FROM acceptrec_marker`).Scan(&n)) + return n +} + +func writeMarker(ctx context.Context) repo.TxSideEffect { + return func(_ context.Context, tx *sql.Tx, _ *repo.CommitResult) error { + _, err := tx.ExecContext(ctx, `INSERT INTO acceptrec_marker (note) VALUES ('enqueued')`) + return err + } +} + +// --------------------------------------------------------------------------- +// Golden: acceptrec.SubjectRKey must byte-match Coves and the materializer. +// --------------------------------------------------------------------------- + +// The golden values are copied verbatim from the materializer's +// subject_rkey_test.go (which copied them from Coves' rkey_test.go, computed +// OUTSIDE Go). If acceptrec's derivation forks from either, a post acquires two +// acceptance keys and neither engine can see the other's record. +func TestSubjectRKey_GoldenVectorMatchesCoves(t *testing.T) { + t.Parallel() + assert.Equal(t, + "xxdmibjaexx43drostplutjbp7g4oaw3uriugf5twafpldfkupca", + SubjectRKey("at://did:plc:abc123/social.coves.community.postv2/3kjzl5kcb2s2v")) + assert.Equal(t, + "iyhgczhg7xsbrayzrrs2qa4fks6amctx7ghyjakqyrlhxztbbl5a", + SubjectRKey("at://did:plc:abc123/social.coves.community.postv2/3kjzl5kcb2s2w")) + // The long-subject vector: a truncating implementation would pass a shape + // check but collide two long subjects. Pinned to a golden value (copied from + // the materializer's longDIDWeb vector). + assert.Equal(t, + "fktiwazbhqfsqjg5e7yypm3ukbalk72bfplcr2d2ukm7iuwqeyqa", + SubjectRKey("at://"+longDIDWeb()+"/social.coves.community.postv2/3kjzl5kcb2s2v")) + assert.Len(t, SubjectRKey(arPostURI), 52, "the key is always a fixed 52 characters") +} + +func longDIDWeb() string { + label := "" + for i := 0; i < 63; i++ { + label += "a" + } + authority := "" + for i := 0; i < 8; i++ { + if i > 0 { + authority += "." + } + authority += label + } + return "did:web:" + authority + ".example.com" +} + +// --------------------------------------------------------------------------- +// AcceptSubject +// --------------------------------------------------------------------------- + +// readAcceptance returns the acceptance record + CID at the subject's rkey, or +// fails the lookup with IsNotFound when none stands. +func readAcceptance(t *testing.T, repos *repo.Manager, communityDID, subjectURI string) (map[string]any, string, error) { + t.Helper() + return repos.GetRecord(context.Background(), communityDID, CollectionAcceptance, SubjectRKey(subjectURI)) +} + +// TestAcceptSubject_WritesRecordAndRunsSideEffectAtomically: a fresh accept +// writes the acceptance record pinning the post's uri+cid at SubjectRKey, and +// runs the side effect inside the same commit. +func TestAcceptSubject_WritesRecordAndRunsSideEffectAtomically(t *testing.T) { + repos, database := newRepos(t) + newMarker(t, database) + ctx := context.Background() + + res, err := AcceptSubject(ctx, repos, arCommunityDID, arPostURI, arPostCID, arPublishedAt, writeMarker(ctx)) + require.NoError(t, err, "a fresh accept with no standing removal must succeed") + require.NotNil(t, res) + assert.False(t, res.NoOp, "the first acceptance is a real commit") + + record, _, err := readAcceptance(t, repos, arCommunityDID, arPostURI) + require.NoError(t, err, + "the acceptance record must exist in the COMMUNITY repo at SubjectRKey(%s)", arPostURI) + assert.Equal(t, CollectionAcceptance, record["$type"]) + subject, ok := record["subject"].(map[string]any) + require.True(t, ok, "the acceptance pins the subject as a strongRef, got %v", record["subject"]) + assert.Equal(t, arPostURI, subject["uri"], "the strongRef pins the post at-uri") + assert.Equal(t, arPostCID, subject["cid"], "the strongRef pins the evaluated post CID") + + assert.Equal(t, 1, markerCount(t, database), + "the side effect (the outbound enqueue) must run in the acceptance commit: "+ + "acceptance and enqueue land together or not at all") +} + +// TestAcceptSubject_RemovalStandsRefusesAndSkipsSideEffect: a removal is +// terminal, so an accept over one must be refused — no acceptance written, and +// the side effect NOT run (a removal means the community decided the post is +// out; re-firing the enqueue would federate a post that must stay hidden). +func TestAcceptSubject_RemovalStandsRefusesAndSkipsSideEffect(t *testing.T) { + repos, database := newRepos(t) + newMarker(t, database) + ctx := context.Background() + + // A standing removal at the shared rkey. + _, err := repos.PutRecord(ctx, arCommunityDID, CollectionRemoval, SubjectRKey(arPostURI), + map[string]any{"$type": CollectionRemoval, "subject": map[string]any{"uri": arPostURI, "cid": arPostCID}, "code": "moderator-discretion", "createdAt": "2026-08-12T10:00:00.000Z"}) + require.NoError(t, err) + + _, err = AcceptSubject(ctx, repos, arCommunityDID, arPostURI, arPostCID, arPublishedAt, writeMarker(ctx)) + require.ErrorIs(t, err, ErrRemovalStands, + "an accept over a standing removal must refuse with ErrRemovalStands (removal terminality)") + + _, _, aerr := readAcceptance(t, repos, arCommunityDID, arPostURI) + assert.True(t, errors.IsNotFound(aerr), "no acceptance may be written while a removal stands") + assert.Zero(t, markerCount(t, database), + "the side effect must NOT run when the accept is refused: the removal-guard fails the "+ + "batch BEFORE the side effect, so no enqueue fires for a decided-out post") +} + +// TestAcceptSubject_RedeliveryRePutsNoOpButStillRunsSideEffect: re-accepting the +// same post produces a byte-identical record (createdAt derived, not stamped), +// so the repo layer takes the NoOp path — but the side effect STILL runs, so +// the at-least-once outbound enqueue re-fires (matching task 15's idempotent +// activity dedup). +func TestAcceptSubject_RedeliveryRePutsNoOpButStillRunsSideEffect(t *testing.T) { + repos, database := newRepos(t) + newMarker(t, database) + ctx := context.Background() + + first, err := AcceptSubject(ctx, repos, arCommunityDID, arPostURI, arPostCID, arPublishedAt, writeMarker(ctx)) + require.NoError(t, err) + require.NotNil(t, first) + assert.False(t, first.NoOp) + require.Equal(t, 1, markerCount(t, database)) + + // The redelivery: same subject, same CID, same publishedAt. + second, err := AcceptSubject(ctx, repos, arCommunityDID, arPostURI, arPostCID, arPublishedAt, writeMarker(ctx)) + require.NoError(t, err) + require.NotNil(t, second) + assert.True(t, second.NoOp, + "an identical re-accept must produce a NoOp record: createdAt is derived, not stamped, "+ + "so the bytes match and the repo layer's no-op path absorbs it") + assert.Equal(t, 2, markerCount(t, database), + "the side effect must run AGAIN on the redelivery even though the record did not change: "+ + "the outbound enqueue is at-least-once and dedupe is the peer's job") +} diff --git a/internal/consume/dispatch.go b/internal/consume/dispatch.go index 9fc7f3d..6f4329a 100644 --- a/internal/consume/dispatch.go +++ b/internal/consume/dispatch.go @@ -476,19 +476,14 @@ func (d *Dispatcher) handlePostV2(ctx context.Context, _ *sql.Tx, did string, co // must reach the engine even from a user who has since opted out. The // engine already knows which posts it accepted and can no-op the rest. if commit.Operation != operationDelete { - // The opt-out gate: a create or an update pushes the author's content - // OUTWARD (an acceptance record plus federation), which is exactly what - // an opted-out author has refused. Same gate comments and votes carry. - federating, err := d.mayFederate(ctx, did) - if err != nil { - return err - } - if !federating { - d.logger.Debug("skipping postv2 from an opted-out author", - slog.String("did", did), slog.String("rkey", commit.RKey)) - return nil - } - + // The opt-out check MOVED into the engine (task 16): an opted-out + // author's post REACHES the engine, which records a distinct rejection + // (decision_code opted-out) in the admissions ledger rather than the + // consumer dropping it silently at debug. post.getStatus and the admin + // surface both need the "why", and only the engine writes it. The + // consumer keeps only the pre-gate an opted-out author's post still fails + // for a DIFFERENT reason: it must name a bridged community for the engine + // to have a repo to reject it INTO. communityDID := stringField(commit.Record, "community") if communityDID == "" { // The lexicon REQUIRES community. A post without one is malformed diff --git a/internal/consume/postv2_test.go b/internal/consume/postv2_test.go index ed9c376..732cbb3 100644 --- a/internal/consume/postv2_test.go +++ b/internal/consume/postv2_test.go @@ -194,17 +194,17 @@ func TestPostV2_DeleteIsNotGatedOnACommunityItCannotSee(t *testing.T) { } // --------------------------------------------------------------------------- -// Second-opinion C3: the opt-out gate applies to postv2 too +// Task 16: the opt-out check MOVED into the engine // --------------------------------------------------------------------------- // -// A postv2 create/update pushes the author's content OUTWARD (into the -// community's repo, via the acceptance engine), so an opted-out author's post -// must not be admitted — the same gate comments and votes already carry. A -// delete is a retraction and stays ungated, for the same reason it does on the -// comment path: removing content is always safe, and it is the only way an -// opted-out user can take down what is already federated. - -func TestPostV2_OptedOutAuthorCreateNeverReachesTheEngine(t *testing.T) { +// Before task 16 the consumer dropped an opted-out author's postv2 create/update +// at debug. Now the check lives in the acceptance engine, which RECORDS the +// rejection (decision_code opted-out) in the admissions ledger — post.getStatus +// and the admin surface both need the "why". So an opted-out author's post must +// REACH the engine; the consumer no longer silently gates it here. A delete +// stays ungated for the same reason it does on the comment path. + +func TestPostV2_OptedOutAuthorCreateReachesTheEngine(t *testing.T) { database := dispatchTestDB(t) seedBridgedCommunity(t, database) ctx := context.Background() @@ -219,13 +219,13 @@ func TestPostV2_OptedOutAuthorCreateNeverReachesTheEngine(t *testing.T) { require.NoError(t, fixture.handle(t, postV2Frame(dispatchNativeDID, dispatchRev, "3lzpostopt01", acceptCommunityDID))) - assert.Zero(t, fixture.engine.Calls(), - "an opted-out author's post must not be admitted: admission writes an "+ - "acceptance record and federates the post, which is exactly the outward "+ - "push the opt-out forbids") + assert.Equal(t, 1, fixture.engine.Calls(), + "an opted-out author's post now REACHES the engine: the engine records a distinct "+ + "rejection (decision_code opted-out) in the admissions ledger rather than the "+ + "consumer dropping it silently — the admin surface needs the reason") } -func TestPostV2_OptedOutAuthorUpdateNeverReachesTheEngine(t *testing.T) { +func TestPostV2_OptedOutAuthorUpdateReachesTheEngine(t *testing.T) { database := dispatchTestDB(t) seedBridgedCommunity(t, database) ctx := context.Background() @@ -246,8 +246,9 @@ func TestPostV2_OptedOutAuthorUpdateNeverReachesTheEngine(t *testing.T) { dispatchNativeDID, dispatchRev, acceptCommunityDID)) require.NoError(t, fixture.handle(t, frame)) - assert.Zero(t, fixture.engine.Calls(), - "an edit is still an outward push, so it is gated exactly like a create") + assert.Equal(t, 1, fixture.engine.Calls(), + "an edit reaches the engine exactly like a create: the engine, not the consumer, "+ + "decides admission and records the reason") } func TestPostV2_OptedOutAuthorDeleteStillReachesTheEngine(t *testing.T) { diff --git a/internal/db/migrations/021_admissions.sql b/internal/db/migrations/021_admissions.sql new file mode 100644 index 0000000..5fa7d76 --- /dev/null +++ b/internal/db/migrations/021_admissions.sql @@ -0,0 +1,54 @@ +-- +goose Up +-- Task 16: the acceptance engine's admission ledger. +-- +-- Coves' post.getStatus reads a post's admission state from the +-- firehose-visible acceptance/removal records the engine writes — that is the +-- authoritative, cross-network answer. This table is the engine's OWN debug and +-- admin surface: it records, for every (community, post) the engine has decided +-- on, the machine-readable WHY of that decision (decision_code) and the state it +-- left the post in, so an operator can list pending/rejected admissions with +-- reasons and force a re-admit. It is NOT a watermark and it is NOT consulted on +-- the correctness path: losing it re-derives from a replay. +-- +-- PK is (community_did, post_uri): one row per post per community. A post edited +-- and re-evaluated updates its row in place (evaluated_cid moves); it is not a +-- log. +CREATE TABLE admissions ( + community_did TEXT NOT NULL, -- the bridged community the post was submitted to + post_uri TEXT NOT NULL, -- the postv2 at-uri (author repo) + status TEXT NOT NULL + CHECK (status IN ('pending', 'accepted', 'pending_reacceptance', 'rejected', 'removed')), + -- decision_code is the machine-readable reason for the current status: '' + -- for a clean accept, else the rejection/removal reason (opted-out, + -- title-required, banned, community-gone, parent-locked, rate-limited, + -- lexicon-invalid, …). Distinct codes are what the admin surface needs and + -- the firehose records cannot carry. + decision_code TEXT NOT NULL DEFAULT '', + -- evaluated_cid is the post CID this decision was made AGAINST (decision 5.5: + -- admission runs against the EVENT's CID). A later event with a different CID + -- supersedes it — the digest rkey converges, and this column is how a replay + -- or a concurrent engine tells "already decided this version" from "new + -- content to re-run admission on". + evaluated_cid TEXT NOT NULL DEFAULT '', + -- acceptance_rkey / accepted_cid pin what the engine actually wrote when it + -- accepted: the community-repo record key (SubjectRKey) and the CID the + -- acceptance pinned. Empty on a rejection. + acceptance_rkey TEXT NOT NULL DEFAULT '', + accepted_cid TEXT NOT NULL DEFAULT '', + -- redrivable marks a rejection an admin (or a transient-cause re-scan) may + -- retry. A hard, permanent rejection (opted-out-history, lexicon-invalid) is + -- NOT redrivable; a soft one (community temporarily gone) is. + redrivable BOOLEAN NOT NULL DEFAULT false, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (community_did, post_uri) +); + +-- The admin "list pending/rejected admissions" query filters by status; the +-- ledger is small (one row per bridged post) but the index keeps that listing +-- from scanning removed/accepted rows. +CREATE INDEX idx_admissions_status ON admissions (status) WHERE status IN ('pending', 'rejected'); + +-- +goose Down +DROP INDEX IF EXISTS idx_admissions_status; +DROP TABLE IF EXISTS admissions; diff --git a/internal/materialize/acceptance.go b/internal/materialize/acceptance.go index 7b3ad45..ccd4785 100644 --- a/internal/materialize/acceptance.go +++ b/internal/materialize/acceptance.go @@ -6,6 +6,7 @@ import ( "fmt" "time" + "tidepool/internal/acceptrec" "tidepool/internal/errors" "tidepool/internal/repo" "tidepool/internal/store" @@ -49,6 +50,10 @@ func (m *Materializer) acceptPost(ctx context.Context, communityDID, postURI, po // restore path does not come through here: it deletes the removal and // writes the acceptance in ONE commit (RestorePost), so it never has to // argue with this guard. + // + // This pre-loop read is an OPTIMIZATION and a test seam (removalCheck): the + // authoritative refusal is the commit-time removal guard inside + // acceptrec.AcceptSubject, which holds even when this read answers stale. removed, err := m.removalCheck(ctx, communityDID, rkey) if err != nil { return err @@ -59,85 +64,23 @@ func (m *Materializer) acceptPost(ctx context.Context, communityDID, postURI, po return nil } - // Read-modify-write under a CAS precondition, bounded like the stats - // stamp: the read happens outside the commit serialization, so a racing - // repin (or a stats-driven CID change on the subject) can move the record - // underneath it. Losing that race means re-reading, never overwriting. - for attempt := 0; ; attempt++ { - createdAt := recordDatetime(publishedAt) - expectPrevCID := "" - - stored, storedCID, err := m.repos.GetRecord(ctx, communityDID, CollectionAcceptance, rkey) - switch { - case err == nil: - expectPrevCID = storedCID - if when, ok := stored["createdAt"].(string); ok && when != "" { - createdAt = when - } - case errors.IsNotFound(err): - // Either the first acceptance or the crash-window heal. The empty - // precondition asserts the record is still absent, so a concurrent - // writer that got there first sends us round the loop instead of - // clobbering its acceptance. - default: - return fmt.Errorf("materialize: read acceptance %s/%s/%s: %w", - communityDID, CollectionAcceptance, rkey, err) - } - - record := map[string]any{ - "$type": CollectionAcceptance, - "subject": strongRef(postURI, postCID), - "createdAt": createdAt, - } - if err := m.validateRecord(record); err != nil { - return err - } - - // The terminality guard is re-asserted HERE, at commit time, as an - // inert op: deleting the removal with a precondition of "must not - // exist" claims nothing and changes nothing, but it makes the batch - // fail if a removal has appeared since the read above. Without it the - // guard is check-then-act, and a RemovePost landing in the window - // leaves an acceptance beside a standing removal — a post - // simultaneously visible and removed, which nothing reconciles because - // each writer believed it saw a consistent world. - // - // Deliberately NOT commitRecord: that path upserts an ap_objects row, - // and an acceptance has no AP object behind it — the bridge mints it as - // the community's own attestation. A mapping would invent an ap_id for - // it, expose it to every spine consumer, and let an announced delete - // aimed at the POST address the acceptance through the same key space. - noRemoval := "" - _, err = m.repos.ApplyOps(ctx, communityDID, []repo.RecordOp{ - {Action: repo.OpActionDelete, Collection: CollectionRemoval, RKey: rkey, ExpectPrevCID: &noRemoval}, - {Action: repo.OpActionUpdate, Collection: CollectionAcceptance, RKey: rkey, Record: record, ExpectPrevCID: &expectPrevCID}, - }) - if stderrors.Is(err, repo.ErrPreconditionFailed) { - // Either a removal appeared or the acceptance moved. Ask which, - // against the repo itself rather than the seam: a removal means the - // community has since decided this post is out, and that decision - // stands — the acceptance is simply not written. - removed, rerr := m.removalStands(ctx, communityDID, rkey) - if rerr != nil { - return rerr - } - if removed { - m.logger.Debug("post was removed from the community while accepting; not re-accepting", - "community_did", communityDID, "post", postURI) - return nil - } - if attempt+1 < maxStatsCommitAttempts { - continue - } - return fmt.Errorf("materialize: accept %s: acceptance kept changing across %d attempts: %w", - postURI, maxStatsCommitAttempts, err) - } - if err != nil { - return fmt.Errorf("materialize: put acceptance %s/%s/%s for %s: %w", - communityDID, CollectionAcceptance, rkey, postURI, err) - } + // The CAS/removal-guard mechanics live in acceptrec now, shared with the + // acceptance engine so both build on ONE implementation. The materializer + // drives it with a nil side effect: it attests what the bridge already + // decided by materializing the post, and owns no outbound enqueue here. + _, err = acceptrec.AcceptSubject(ctx, m.repos, communityDID, postURI, postCID, publishedAt, nil) + if stderrors.Is(err, acceptrec.ErrRemovalStands) { + // A removal appeared under the write: the community decided this post is + // out, and that decision is terminal. A refused acceptance is not an + // error — the removal simply stands. + m.logger.Debug("post was removed from the community while accepting; not re-accepting", + "community_did", communityDID, "post", postURI) return nil } + if err != nil { + return fmt.Errorf("materialize: accept %s into %s: %w", postURI, communityDID, err) + } + return nil } // removalStands reports whether the community currently holds a removal for diff --git a/internal/repo/applyops_tx_test.go b/internal/repo/applyops_tx_test.go new file mode 100644 index 0000000..b5c0337 --- /dev/null +++ b/internal/repo/applyops_tx_test.go @@ -0,0 +1,178 @@ +package repo + +import ( + "context" + "database/sql" + stderrors "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/errors" +) + +// ApplyOpsTx generalizes the single-op PutRecordTx side-effect seam to the +// multi-op commit: acceptance transitions (create-accept, update-repin, +// edit-fail, author-delete) each ride ONE ApplyOpsTx whose side effect is the +// outbound enqueue, so the community-repo commit and the enqueue land together +// or not at all. These pins clone TestPutRecordTxSideEffectAtomicity for the +// batch path — including the subtle trap that the side effect must ALSO run on +// the all-inert NoOp branch (the at-least-once redelivery enqueue), where no +// repo commit happens but the side effect still has to be made durable. + +// applyOpsTxMarker is a scratch table a side effect writes into so a test can +// prove the side effect's OWN write persisted (or rolled back) independently of +// the record ops. It survives across the test binary; each test truncates it. +func applyOpsTxMarker(t *testing.T, database *sql.DB) { + t.Helper() + _, err := database.Exec(`CREATE TABLE IF NOT EXISTS applyops_tx_marker (note TEXT NOT NULL)`) + require.NoError(t, err) + _, err = database.Exec(`TRUNCATE applyops_tx_marker`) + require.NoError(t, err) + t.Cleanup(func() { _, _ = database.Exec(`DROP TABLE IF EXISTS applyops_tx_marker`) }) +} + +func markerCount(t *testing.T, database *sql.DB) int { + t.Helper() + var n int + require.NoError(t, database.QueryRow(`SELECT COUNT(*) FROM applyops_tx_marker`).Scan(&n)) + return n +} + +// writeMarker returns a side effect that inserts one marker row, then returns +// the given error (nil to succeed). +func writeMarker(ctx context.Context, note string, ret error) TxSideEffect { + return func(_ context.Context, tx *sql.Tx, _ *CommitResult) error { + if _, err := tx.ExecContext(ctx, `INSERT INTO applyops_tx_marker (note) VALUES ($1)`, note); err != nil { + return err + } + return ret + } +} + +// TestApplyOpsTx_SideEffectErrorRollsBackBoth: a side effect that errors inside +// ApplyOpsTx rolls back the record ops AND anything the side effect itself +// wrote. Nothing persists: the record the batch would have deleted still +// exists, the record it would have written is absent, no firehose event, and +// the side effect's marker row is gone. +func TestApplyOpsTx_SideEffectErrorRollsBackBoth(t *testing.T) { + manager, database, _ := testManager(t) + applyOpsTxMarker(t, database) + ctx := context.Background() + + seeded, err := manager.PutRecord(ctx, testDID, testOtherCollection, testRKey(1), testRecord("accepted")) + require.NoError(t, err) + headBefore, revBefore, err := manager.Head(ctx, testDID) + require.NoError(t, err) + + sentinel := stderrors.New("side effect refused") + _, err = manager.ApplyOpsTx(ctx, testDID, []RecordOp{ + {Action: OpActionDelete, Collection: testOtherCollection, RKey: testRKey(1)}, + {Action: OpActionCreate, Collection: testCollection, RKey: testRKey(2), Record: testRecord("removed")}, + }, writeMarker(ctx, "phantom", sentinel)) + require.ErrorIs(t, err, sentinel, + "a side effect erroring inside ApplyOpsTx must surface, not be swallowed") + + _, _, err = manager.GetRecord(ctx, testDID, testOtherCollection, testRKey(1)) + assert.NoError(t, err, "the deleted record must be rolled back with the failed side effect") + _, _, err = manager.GetRecord(ctx, testDID, testCollection, testRKey(2)) + assert.True(t, errors.IsNotFound(err), "the written record must not survive a failed side effect") + + head, rev, err := manager.Head(ctx, testDID) + require.NoError(t, err) + assert.Equal(t, headBefore, head, "a rolled-back batch must not advance the head") + assert.Equal(t, revBefore, rev) + assert.Empty(t, eventsSince(t, manager, testDID, seeded.Seq), + "no firehose event may follow the seed commit when the side effect failed") + assert.Zero(t, markerCount(t, database), + "the side effect's OWN write must roll back too: record and bookkeeping are one unit") +} + +// TestApplyOpsTx_AllInertStillRunsSideEffectAndCommits is the trap. A batch +// where every op is inert — a delete of a missing record plus a byte-identical +// re-put — produces NO new commit and NO firehose event (the NoOp branch). But +// the side effect STILL runs and its write STILL commits, because that side +// effect is the at-least-once outbound enqueue a redelivery must re-fire even +// though the acceptance record did not change. +func TestApplyOpsTx_AllInertStillRunsSideEffectAndCommits(t *testing.T) { + manager, database, _ := testManager(t) + applyOpsTxMarker(t, database) + ctx := context.Background() + + seeded, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(1), testRecord("same")) + require.NoError(t, err) + headBefore, revBefore, err := manager.Head(ctx, testDID) + require.NoError(t, err) + eventsBefore := eventCount(t, manager, testDID) + + res, err := manager.ApplyOpsTx(ctx, testDID, []RecordOp{ + // Byte-identical re-put: keeps the batch a WRITE (so it is NOT the + // genesis-delete-only branch) while changing nothing. + {Action: OpActionUpdate, Collection: testCollection, RKey: testRKey(1), Record: testRecord("same")}, + // Delete of a record that is not there: inert. + {Action: OpActionDelete, Collection: testOtherCollection, RKey: testRKey(2)}, + }, writeMarker(ctx, "redelivery-enqueue", nil)) + require.NoError(t, err) + require.NotNil(t, res) + + assert.True(t, res.NoOp, "an all-inert batch reports NoOp") + assert.Zero(t, res.Seq, "a NoOp batch emits no firehose event, so it has no seq") + assert.Equal(t, seeded.CommitCID, res.CommitCID, "the head is unchanged") + + head, rev, err := manager.Head(ctx, testDID) + require.NoError(t, err) + assert.Equal(t, headBefore, head, "an all-inert batch must not advance the head") + assert.Equal(t, revBefore, rev) + assert.Equal(t, eventsBefore, eventCount(t, manager, testDID), + "an all-inert batch must not emit a firehose event") + + assert.Equal(t, 1, markerCount(t, database), + "the side effect must STILL run on the NoOp branch and its write must be durable: "+ + "the outbound enqueue is at-least-once, so a redelivery whose acceptance record is "+ + "byte-identical must re-fire the enqueue even though no commit happened") +} + +// TestApplyOpsTx_SuccessRunsSideEffectInSameCommit: a batch that really changes +// records runs the side effect in the same transaction — the side effect +// observes the commit result, its write lands, and the record ops all land. +func TestApplyOpsTx_SuccessRunsSideEffectInSameCommit(t *testing.T) { + manager, database, _ := testManager(t) + applyOpsTxMarker(t, database) + ctx := context.Background() + + seeded, err := manager.PutRecord(ctx, testDID, testOtherCollection, testRKey(1), testRecord("accepted")) + require.NoError(t, err) + + var hookRes *CommitResult + res, err := manager.ApplyOpsTx(ctx, testDID, []RecordOp{ + {Action: OpActionDelete, Collection: testOtherCollection, RKey: testRKey(1)}, + {Action: OpActionCreate, Collection: testCollection, RKey: testRKey(2), Record: testRecord("removed")}, + }, func(sctx context.Context, tx *sql.Tx, r *CommitResult) error { + hookRes = r + // The firehose row for this commit must already be visible in the tx. + var n int + if err := tx.QueryRowContext(sctx, + `SELECT COUNT(*) FROM firehose_events WHERE seq = $1`, r.Seq).Scan(&n); err != nil { + return err + } + if n != 1 { + return stderrors.New("firehose event not visible to side effect") + } + _, err := tx.ExecContext(sctx, `INSERT INTO applyops_tx_marker (note) VALUES ('committed')`) + return err + }) + require.NoError(t, err) + require.NotNil(t, res) + require.NotNil(t, hookRes, "the side effect must run on the committed branch") + + assert.False(t, res.NoOp, "a batch that changes records is a real commit") + assert.Equal(t, res.Seq, hookRes.Seq, "the side effect observes this commit's result") + assert.Greater(t, res.Seq, seeded.Seq, "a new firehose event was emitted") + + assert.Equal(t, 1, markerCount(t, database), "the side effect's write commits with the record ops") + _, _, err = manager.GetRecord(ctx, testDID, testOtherCollection, testRKey(1)) + assert.True(t, errors.IsNotFound(err), "the deleted record is gone") + _, _, err = manager.GetRecord(ctx, testDID, testCollection, testRKey(2)) + assert.NoError(t, err, "the written record exists") +} diff --git a/internal/repo/repo.go b/internal/repo/repo.go index 1e830a5..66dfd45 100644 --- a/internal/repo/repo.go +++ b/internal/repo/repo.go @@ -270,6 +270,17 @@ type RecordOp struct { // rejection costs no commit work; anything that fails later rolls back with // the transaction, leaving neither a record change nor a firehose event. func (m *Manager) ApplyOps(ctx context.Context, did string, ops []RecordOp) (*CommitResult, error) { + return m.ApplyOpsTx(ctx, did, ops, nil) +} + +// ApplyOpsTx is ApplyOps with a side effect executed inside the commit +// transaction (see TxSideEffect). The side effect runs on BOTH the committed +// branch (after the ops write, before COMMIT) AND the all-inert NoOp branch +// (every op turned out to change nothing, so there is no new commit — but the +// side effect must still run and be made durable, because it carries the +// at-least-once outbound enqueue a redelivery has to re-fire). A nil sideEffect +// is exactly ApplyOps. +func (m *Manager) ApplyOpsTx(ctx context.Context, did string, ops []RecordOp, sideEffect TxSideEffect) (*CommitResult, error) { if len(ops) == 0 { return nil, errors.NewValidationError("ops", "must not be empty") } @@ -419,6 +430,20 @@ func (m *Manager) ApplyOps(ctx context.Context, did string, ops []RecordOp) (*Co // head exactly and can go back in the cache. m.cacheTree(did, state.headCID, *prevData, tree) } + if sideEffect != nil { + // The side effect (the at-least-once outbound enqueue) still runs + // and must still be durable, so this — otherwise write-free — + // transaction commits even though no repo commit happened. A + // redelivery whose acceptance record is byte-identical must re-fire + // the enqueue; dedupe is the peer's job. The head is unchanged + // whatever the side effect does, so the cacheTree above stays valid. + if err := sideEffect(ctx, tx, res); err != nil { + return nil, err + } + if err := tx.Commit(); err != nil { + return nil, fmt.Errorf("repo: commit no-op side effect for %s: %w", did, err) + } + } return res, nil } @@ -427,6 +452,14 @@ func (m *Manager) ApplyOps(ctx context.Context, did string, ops []RecordOp) (*Co if err != nil { return nil, err } + if sideEffect != nil { + // Inside the transaction, after the ops write and before COMMIT: a + // failing side effect rolls the record ops back too — the acceptance + // commit and the outbound enqueue land together or not at all. + if err := sideEffect(ctx, tx, res); err != nil { + return nil, err + } + } if err := tx.Commit(); err != nil { return nil, fmt.Errorf("repo: commit tx for %s: %w", did, err) }