diff --git a/internal/api/xrpc/mapper.go b/internal/api/xrpc/mapper.go index bef653f..725f72f 100644 --- a/internal/api/xrpc/mapper.go +++ b/internal/api/xrpc/mapper.go @@ -139,12 +139,24 @@ var sharedRules = []Rule{ "Invalid request to PDS"), Sentinel(pds.ErrNotFound, http.StatusNotFound, "NotFound", "Record not found on PDS"), + // ErrSwapConflict before ErrConflict: a 409 InvalidSwap wraps both + // sentinels, and the lost-swap message is the more actionable one. A 400 + // InvalidSwap wraps only ErrSwapConflict, so without this rule it would + // fall through to the generic 500 — a lost race reported as our failure. + Sentinel(pds.ErrSwapConflict, http.StatusConflict, "Conflict", + "Record was modified by another operation, please retry"), Sentinel(pds.ErrConflict, http.StatusConflict, "Conflict", "Record was modified by another operation"), Sentinel(pds.ErrPayloadTooLarge, http.StatusRequestEntityTooLarge, "PayloadTooLarge", "Request payload exceeds size limit"), Sentinel(pds.ErrRateLimited, http.StatusTooManyRequests, "RateLimitExceeded", "Too many requests, please try again later"), + // A PDS 5xx is a classified upstream failure, not our internal error, so it + // answers 502 rather than falling through to internalError — the same call + // the image proxy makes for a PDS it cannot reach. The message is fixed: + // the PDS's own text may carry internal detail we must not forward. + Sentinel(pds.ErrServerError, http.StatusBadGateway, "UpstreamFailure", + "PDS failed to process the request"), // Request lifecycle. A cancellation is the client's own doing, so it is a // 4xx; a deadline we blew is ours to report as a gateway timeout. diff --git a/internal/api/xrpc/mapper_test.go b/internal/api/xrpc/mapper_test.go index 3391cb9..e985505 100644 --- a/internal/api/xrpc/mapper_test.go +++ b/internal/api/xrpc/mapper_test.go @@ -92,6 +92,48 @@ func TestResolve(t *testing.T) { wantCode: "PayloadTooLarge", wantMatch: true, }, + { + // A lost swap arrives from a live PDS as HTTP 400, so without its + // own rule it would be indistinguishable from a malformed request — + // or worse, fall to 500. It must answer 409: a retryable conflict. + name: "pds swap conflict is 409 not 400", + err: pds.ErrSwapConflict, + wantStatus: http.StatusConflict, + wantCode: "Conflict", + wantMatch: true, + }, + { + name: "pds swap conflict wrapped with %w", + err: fmt.Errorf("applyWrites: %w: CID mismatch", pds.ErrSwapConflict), + wantStatus: http.StatusConflict, + wantCode: "Conflict", + wantMatch: true, + }, + { + // A 409 InvalidSwap wraps ErrConflict and ErrSwapConflict at once; + // both mean 409 Conflict, whichever rule wins. + name: "pds 409 swap wraps both conflict sentinels", + err: fmt.Errorf("applyWrites: %w: %w: stale", pds.ErrConflict, pds.ErrSwapConflict), + wantStatus: http.StatusConflict, + wantCode: "Conflict", + wantMatch: true, + }, + { + // A PDS 5xx is a classified upstream failure: 502, not the + // content-free 500 reserved for errors nothing recognized. + name: "pds server error is 502 upstream failure", + err: pds.ErrServerError, + wantStatus: http.StatusBadGateway, + wantCode: "UpstreamFailure", + wantMatch: true, + }, + { + name: "pds server error wrapped with %w", + err: fmt.Errorf("applyWrites: %w: Internal Server Error", pds.ErrServerError), + wantStatus: http.StatusBadGateway, + wantCode: "UpstreamFailure", + wantMatch: true, + }, { name: "shared typed validation error", err: coreerrors.NewValidationError("handle", "is required"), diff --git a/internal/atproto/pds/applywrites.go b/internal/atproto/pds/applywrites.go index 5344afc..f3a47c5 100644 --- a/internal/atproto/pds/applywrites.go +++ b/internal/atproto/pds/applywrites.go @@ -208,6 +208,26 @@ func (c *client) ApplyWrites(ctx context.Context, writes []Write, swapCommit str return nil, fmt.Errorf("applyWrites: PDS returned success without a commit rev (%d writes)", len(writes)) } + // THE RESULTS ARE POSITIONAL — the lexicon returns one per submitted write, + // in order — and callers index into them to find the record their commit + // made stand. A response with fewer (or more) results than writes would + // silently hand a caller the WRONG entry; a create/update result without + // uri+cid would hand it empty identity for a record that committed. Both + // are malformed successes from the PDS or something in front of it, and are + // refused for the same reason recordCommit refuses them on single-record + // writes: the missing fields are precisely what the caller is about to + // persist. + if len(response.Results) != len(writes) { + return nil, fmt.Errorf("applyWrites: PDS returned %d results for %d writes", len(response.Results), len(writes)) + } + for i, entry := range response.Results { + op := writes[i].Op + if (op == WriteOpCreate || op == WriteOpUpdate) && (entry.URI == "" || entry.CID == "") { + return nil, fmt.Errorf("applyWrites: PDS returned success without uri/cid for writes[%d] (%s %s/%s)", + i, op, writes[i].Collection, writes[i].RKey) + } + } + result := &ApplyWritesResult{ CommitRev: response.Commit.Rev, CommitCID: response.Commit.CID, diff --git a/internal/atproto/pds/applywrites_test.go b/internal/atproto/pds/applywrites_test.go index 19106de..c62fdc9 100644 --- a/internal/atproto/pds/applywrites_test.go +++ b/internal/atproto/pds/applywrites_test.go @@ -179,8 +179,12 @@ func TestClient_ApplyWrites_UpdateUsesTheUpdateDiscriminant(t *testing.T) { _ = json.NewDecoder(r.Body).Decode(&payload) w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ - "commit": map[string]any{"cid": testCommitCID, "rev": testCommitRev}, - "results": []any{map[string]any{"$type": "com.atproto.repo.applyWrites#updateResult"}}, + "commit": map[string]any{"cid": testCommitCID, "rev": testCommitRev}, + "results": []any{map[string]any{ + "$type": "com.atproto.repo.applyWrites#updateResult", + "uri": "at://did:plc:test/social.coves.community.removal/rk", + "cid": "bafyreiremovalaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + }}, }) }) defer closeServer() @@ -209,6 +213,81 @@ func TestClient_ApplyWrites_UpdateUsesTheUpdateDiscriminant(t *testing.T) { } } +func TestClient_ApplyWrites_RefusesAShortResultsArray(t *testing.T) { + // The results are POSITIONAL — one per submitted write, in order — and the + // caller indexes into them to find the record its commit made stand + // (standCIDOf in the community writer). A server returning fewer results + // than writes would silently hand the caller the WRONG entry, or an empty + // CID for a record that committed. That is a malformed success and must be + // an error, not a zero value the caller persists. + c, closeServer := newCommitClient(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "commit": map[string]any{"cid": testCommitCID, "rev": testCommitRev}, + // Two writes went up; one result comes back. + "results": []any{ + map[string]any{"$type": "com.atproto.repo.applyWrites#deleteResult"}, + }, + }) + }) + defer closeServer() + + _, err := c.ApplyWrites(context.Background(), []Write{ + {Op: WriteOpDelete, Collection: "social.coves.community.acceptance", RKey: "rk"}, + { + Op: WriteOpCreate, + Collection: "social.coves.community.removal", + RKey: "rk", + Record: map[string]any{"$type": "social.coves.community.removal", "code": "spam"}, + }, + }, testCommitCID) + if err == nil { + t.Fatal("a results array shorter than the batch is a malformed response and must be an error") + } +} + +func TestClient_ApplyWrites_RefusesACreateOrUpdateResultWithoutURIOrCID(t *testing.T) { + // A create or an update committed a record the caller is about to + // reference: its result's uri and cid are exactly what gets persisted onto + // the admission row. A 200 that omits them is the same class of malformed + // body recordCommit refuses for single-record writes. + for name, entry := range map[string]map[string]any{ + "create without cid": { + "$type": "com.atproto.repo.applyWrites#createResult", + "uri": "at://did:plc:test/social.coves.community.removal/rk", + }, + "update without uri": { + "$type": "com.atproto.repo.applyWrites#updateResult", + "cid": "bafyreiremovalaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + }, + } { + t.Run(name, func(t *testing.T) { + c, closeServer := newCommitClient(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "commit": map[string]any{"cid": testCommitCID, "rev": testCommitRev}, + "results": []any{entry}, + }) + }) + defer closeServer() + + op := WriteOpCreate + if name == "update without uri" { + op = WriteOpUpdate + } + _, err := c.ApplyWrites(context.Background(), []Write{{ + Op: op, + Collection: "social.coves.community.removal", + RKey: "rk", + Record: map[string]any{"$type": "social.coves.community.removal", "code": "spam"}, + }}, "") + if err == nil { + t.Fatal("a create/update result without uri+cid is a malformed response and must be an error") + } + }) + } +} + func TestClient_ApplyWrites_MapsInvalidSwapToSwapConflict(t *testing.T) { // VERIFIED AGAINST A LIVE PDS: a failed swap comes back as HTTP 400 with // "error": "InvalidSwap", NOT the 409 the lexicon documents. That is why @@ -377,6 +456,32 @@ func TestWrapAPIError_ServerErrorsAreTheirOwnClass(t *testing.T) { } } +func TestWrapAPIError_NameChecksOutrankTheStatusEvenOn5xx(t *testing.T) { + // The name says what happened; the status says how the server framed it. A + // PDS (or a proxy in front of one) that wraps an InvalidSwap in a 500 is + // still reporting a lost swap, and a caller that saw only ErrServerError + // would resend the same shape instead of re-reading — the exact behaviour + // the sentinel exists to prevent. So the name checks run BEFORE the 5xx + // branch, for every status. + err := wrapAPIError(&atclient.APIError{ + StatusCode: 500, + Name: "InvalidSwap", + Message: "Commit was at bafyreiotheraaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + }, "applyWrites") + if !errors.Is(err, ErrSwapConflict) { + t.Errorf("a 500-carrying InvalidSwap must still map ErrSwapConflict, got %v", err) + } + + err = wrapAPIError(&atclient.APIError{ + StatusCode: 500, + Name: "RecordNotFound", + Message: "Record not found", + }, "getRecord") + if !errors.Is(err, ErrNotFound) { + t.Errorf("a 500-carrying RecordNotFound must still map ErrNotFound, got %v", err) + } +} + func TestWrapAPIError_InvalidSwapIsNotAPlainBadRequest(t *testing.T) { err := wrapAPIError(&atclient.APIError{ StatusCode: 400, diff --git a/internal/atproto/pds/client.go b/internal/atproto/pds/client.go index 11037c8..b8a5cd9 100644 --- a/internal/atproto/pds/client.go +++ b/internal/atproto/pds/client.go @@ -98,23 +98,24 @@ func wrapAPIError(err error, operation string) error { // Check if it's an APIError from atclient var apiErr *atclient.APIError if errors.As(err, &apiErr) { - // THE ERROR NAME IS CONSULTED ON TOP OF THE STATUS, for one case. A lost - // swap comes back from a live PDS as HTTP 400 with "error": - // "InvalidSwap", not the 409 the lexicon documents, so the status alone - // maps it onto ErrBadRequest — indistinguishable from a malformed - // record. It is the one 400 that must be re-read and retried rather than - // reported. + // THE NAME CHECKS RUN FIRST, BEFORE EVERY STATUS BRANCH — including the + // 5xx one. The name says what happened; the status only says how the + // server framed it, and PDS implementations and the proxies in front of + // them disagree on the framing. A lost swap comes back from a live PDS + // as HTTP 400 with "error": "InvalidSwap", not the 409 the lexicon + // documents; a 500 carrying InvalidSwap is STILL a lost swap, and a + // caller that saw only ErrServerError would resend the same shape + // instead of re-reading — the exact behaviour the sentinel exists to + // prevent. // // A 409 InvalidSwap — what the lexicon says, and what some // implementation may yet send — is BOTH sentinels at once, so callers // written against either one behave correctly whichever status arrives. if apiErr.Name == "InvalidSwap" { - switch apiErr.StatusCode { - case 400: - return fmt.Errorf("%s: %w: %s", operation, ErrSwapConflict, apiErr.Message) - case 409: + if apiErr.StatusCode == 409 { return fmt.Errorf("%s: %w: %w: %s", operation, ErrConflict, ErrSwapConflict, apiErr.Message) } + return fmt.Errorf("%s: %w: %s", operation, ErrSwapConflict, apiErr.Message) } // "No such record" is the other name that outranks its status. The @@ -124,7 +125,19 @@ func wrapAPIError(err error, operation string) error { // record a malformed request. A writer that shapes create-vs-update from // a pre-read cannot tell those apart, and every caller already testing // errors.Is(err, ErrNotFound) after a GetRecord is silently never true. - if apiErr.StatusCode == 400 && (apiErr.Name == "RecordNotFound" || apiErr.Name == "NotFound") { + // + // THE NAME IS THE ONLY THING TRUSTED HERE — never the message. The + // reference PDS also spells some misses as InvalidRequest with "could + // not locate record" in the MESSAGE (the getProfile shape; + // internal/core/users/profile_backfill.go matches it deliberately, at + // its own call site, against that one operation). That spelling is NOT + // mapped at this layer: a transport-wide substring match would turn any + // error that merely mentions those words into ErrNotFound for every + // caller of every method. Our PDS answers the record operations this + // client wraps with the RecordNotFound name — pinned by the idempotent + // re-delete in service_writeforward_test.go, which fails if that ever + // stops being true. + if apiErr.Name == "RecordNotFound" || apiErr.Name == "NotFound" { return fmt.Errorf("%s: %w: %s", operation, ErrNotFound, apiErr.Message) } diff --git a/internal/core/posts/community_writer.go b/internal/core/posts/community_writer.go index fbb4ecc..8942c84 100644 --- a/internal/core/posts/community_writer.go +++ b/internal/core/posts/community_writer.go @@ -4,8 +4,11 @@ import ( "context" "errors" "fmt" + "math/rand/v2" "time" + "github.com/bluesky-social/indigo/atproto/syntax" + "Coves/internal/atproto/pds" ) @@ -102,12 +105,20 @@ type CommunityWriteResult struct { RKey string CID string - // Rev is the repo revision the write committed in: the §5.2 watermark. + // Rev is a repo revision the caller may stamp as the §5.2 watermark. // - // It is EMPTY when Skipped is true, and that is not an oversight — nothing - // committed, so there is no revision to report, and getRecord does not - // reveal the revision an existing record was written at. A caller must - // therefore not stamp a watermark from a skipped write; see Skipped. + // For a write that committed, it is the revision the commit landed in. For + // a SKIPPED write it is the repo's HEAD revision, read BEFORE the pre-read + // that found the standing record — the catch-up watermark. That is safe to + // stamp because the standing record proves what the repo says about this + // subject as of the pre-read: a standing acceptance pinning the target CID + // means no subject-scoped community event lies between the acceptance's + // commit and that head (a removal would have deleted the record; a repin + // would pin a different CID), so a row stranded by an earlier failed stamp + // is caught up rather than left pending forever. Reading the head BEFORE + // the records keeps the rev conservative — a removal committing between + // the two reads has a rev strictly greater than the stamp, so its firehose + // event still applies. Rev string // Skipped reports that the repo already held exactly this record, so @@ -168,6 +179,19 @@ type CommunityRecordWriter interface { type communityRecordWriter struct { repos CommunityRepoFactory now Clock + sleep func(ctx context.Context, d time.Duration) error +} + +// WriterOption configures a CommunityRecordWriter. +type WriterOption func(*communityRecordWriter) + +// WithSwapRetrySleeper replaces the pause between swap retries. +// +// Injected for the same reason the clock is: docs/TEST_ARCHITECTURE.md §3.3 +// forbids a test from actually sleeping, so a test hands in a recorder and +// asserts on the durations instead of waiting through them. +func WithSwapRetrySleeper(sleep func(ctx context.Context, d time.Duration) error) WriterOption { + return func(w *communityRecordWriter) { w.sleep = sleep } } // NewCommunityRecordWriter returns the writer that publishes acceptances and @@ -176,10 +200,40 @@ type communityRecordWriter struct { // The clock is injected for the same reason admitPost's is: createdAt is the // one field a test cannot otherwise pin, and docs/TEST_ARCHITECTURE.md §3.3 // forbids sleeping to move time. -func NewCommunityRecordWriter(repos CommunityRepoFactory, now Clock) CommunityRecordWriter { - return &communityRecordWriter{repos: repos, now: now} +func NewCommunityRecordWriter(repos CommunityRepoFactory, now Clock, opts ...WriterOption) CommunityRecordWriter { + w := &communityRecordWriter{repos: repos, now: now, sleep: sleepWithContext} + for _, opt := range opts { + opt(w) + } + return w } +// sleepWithContext is the production pause: a timer that a cancelled context +// cuts short, so a shutting-down worker is not held hostage by a backoff. +func sleepWithContext(ctx context.Context, d time.Duration) error { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +// ErrRemovalStands reports that an acceptance write met a standing removal +// record for its subject. +// +// §5.5 makes removal terminal: the ONLY sanctioned exit is a moderator restore, +// which deletes the removal in the same commit that writes the fresh acceptance +// (RestoreAcceptance). An acceptance created OVER a standing removal would leave +// both records live at once, and every consumer ordering by the §5.2 tuple +// would see the acceptance outrank the older removal — a moderated post +// laundered back into feeds by a retry. The writer therefore refuses, and the +// engine classifies the refusal as a deferral: the row is owed a decision, but +// not this one. +var ErrRemovalStands = errors.New("a removal record stands for this subject") + // swapRetryLimit is how many times a writer re-reads and re-shapes after losing // an optimistic guard before it gives up and lets the caller try again later. // @@ -191,6 +245,27 @@ func NewCommunityRecordWriter(repos CommunityRepoFactory, now Clock) CommunityRe // deferring the subject and moving on. const swapRetryLimit = 2 +// swapRetryBaseDelay is the first retry's backoff ceiling. Each further retry +// doubles it. +const swapRetryBaseDelay = 25 * time.Millisecond + +// backoff pauses before retry number `attempt` (zero-based), for a jittered +// duration in (0, base<= len(result.Results) { return "" @@ -571,6 +695,28 @@ func removalRecord(cmd CommunityRemovalCommand, createdAt string) map[string]any return record } +// removalCodeMaxLength is the removal lexicon's maxLength for `code` +// (internal/atproto/lexicon/social/coves/community/removal.json), in BYTES — +// which is what a lexicon maxLength counts. +const removalCodeMaxLength = 64 + +// validateSubjectURI refuses a subject that is not a parseable at:// URI. +// +// EVERY RECORD THESE WRITERS SEND GOES OUT WITH validate:false — the PDS has +// never been taught Coves lexicons, so it checks nothing. What this process +// sends is exactly what the firehose carries under the community's signature, +// and a subject that is not an AT-URI would be a malformed strongRef every +// conformant consumer is entitled to refuse. The engine only ever hands the +// writers URIs read from its own admission rows, so a violation here is a +// programming error, refused before any network call. +func validateSubjectURI(kind, postURI string) error { + if _, err := syntax.ParseATURI(postURI); err != nil { + return fmt.Errorf("%s: %w", kind, NewValidationError("postURI", + fmt.Sprintf("must be a parseable at:// URI (%v) — the record embeds it in a strongRef and is sent with validate:false, so nothing downstream re-checks it", err))) + } + return nil +} + // validateWriteCommand refuses an acceptance that would pin nothing. // // A strongRef without a CID is the one thing an acceptance may not be: the @@ -586,7 +732,7 @@ func validateWriteCommand(cmd CommunityWriteCommand) error { return fmt.Errorf("acceptance write: %w", NewValidationError("postCID", "is required — an acceptance's subject is a strongRef, and one without a CID pins nothing")) } - return nil + return validateSubjectURI("acceptance write", cmd.PostURI) } // validateRemovalCommand refuses a removal with no reason code. `code` is @@ -603,6 +749,12 @@ func validateRemovalCommand(cmd CommunityRemovalCommand) error { "is required — it records the version present at removal time")) case cmd.Code == "": return fmt.Errorf("removal write: %w", NewValidationError("code", "is required")) + case len(cmd.Code) > removalCodeMaxLength: + // Sent with validate:false, so the PDS would happily commit a longer + // one — and every conformant consumer could then refuse the record this + // community signed. + return fmt.Errorf("removal write: %w", NewValidationError("code", + fmt.Sprintf("is %d bytes; the removal lexicon caps code at %d (maxLength)", len(cmd.Code), removalCodeMaxLength))) } - return nil + return validateSubjectURI("removal write", cmd.PostURI) } diff --git a/internal/core/posts/community_writer_test.go b/internal/core/posts/community_writer_test.go new file mode 100644 index 0000000..42b2e37 --- /dev/null +++ b/internal/core/posts/community_writer_test.go @@ -0,0 +1,493 @@ +package posts + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "Coves/internal/atproto/pds" +) + +// The community-record writers at T0: everything about their behaviour that a +// fake repo can prove — local validation, call ordering, swap-guard plumbing, +// and the repin's refusal to invent an acceptance. The outer contract against a +// real PDS is engine_contract_test.go. + +const ( + writerCommunityDID = "did:plc:cccccccccccccccccccccccc" + writerPostURI = "at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/social.coves.community.postv2/3kjzl5kcb2s2v" + writerPostCID = "bafyreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" +) + +// writerClock is the fixed time every T0 writer test stamps records with. +func writerClock() time.Time { return time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) } + +// refusingFactory fails the test if the writer reaches for a repo at all — the +// assertion behind "violations are refused before any network call". +func refusingFactory(t *testing.T) CommunityRepoFactory { + t.Helper() + return func(_ context.Context, communityDID string) (CommunityRepo, error) { + t.Errorf("the writer opened the repo of %s; a locally invalid command must be refused "+ + "before any network call", communityDID) + return nil, errors.New("must not be reached") + } +} + +// --------------------------------------------------------------------------- +// Local validation (§3.2/§3.3): records are sent with validate:false, so the +// PDS checks nothing — what this process sends is what the firehose carries. +// --------------------------------------------------------------------------- + +func TestWriter_RefusesAMalformedSubjectURIBeforeAnyNetworkCall(t *testing.T) { + t.Parallel() + + // Every record these writers produce embeds the subject as a strongRef, + // and the put goes out with validate:false — the PDS has never been taught + // Coves lexicons, so nothing downstream re-checks the shape. A subject + // that is not an at:// URI is therefore a programming error at THIS + // boundary: let it through and the malformed strongRef is published to the + // firehose under the community's signature. + for name, uri := range map[string]string{ + "https scheme": "https://example.com/not-a-post", + "bare identifier": "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa", + "unparseable": "at://", + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + writer := NewCommunityRecordWriter(refusingFactory(t), writerClock) + ctx := context.Background() + + writeCmd := CommunityWriteCommand{ + CommunityDID: writerCommunityDID, + PostURI: uri, + PostCID: writerPostCID, + } + removalCmd := CommunityRemovalCommand{ + CommunityDID: writerCommunityDID, + PostURI: uri, + PostCID: writerPostCID, + Code: DecisionSpam, + } + + _, err := writer.WriteAcceptance(ctx, writeCmd) + require.Errorf(t, err, "WriteAcceptance accepted subject URI %q", uri) + _, err = writer.RepinAcceptance(ctx, writeCmd) + require.Errorf(t, err, "RepinAcceptance accepted subject URI %q", uri) + _, err = writer.RestoreAcceptance(ctx, writeCmd) + require.Errorf(t, err, "RestoreAcceptance accepted subject URI %q", uri) + _, err = writer.WriteRemoval(ctx, removalCmd) + require.Errorf(t, err, "WriteRemoval accepted subject URI %q", uri) + }) + } +} + +func TestWriter_RefusesARemovalCodeOverTheLexiconMaxLength(t *testing.T) { + t.Parallel() + + // The removal lexicon caps `code` at 64 bytes (maxLength). validate:false + // means the PDS will happily commit a longer one — and every conformant + // consumer on the network is then entitled to refuse the record this + // community signed. A code the engine minted that long is a programming + // error, caught here rather than published. + writer := NewCommunityRecordWriter(refusingFactory(t), writerClock) + + _, err := writer.WriteRemoval(context.Background(), CommunityRemovalCommand{ + CommunityDID: writerCommunityDID, + PostURI: writerPostURI, + PostCID: writerPostCID, + Code: DecisionCode(strings.Repeat("x", 65)), + }) + require.Error(t, err, "a 65-byte code exceeds the lexicon's maxLength of 64 and must be refused") + + // The boundary itself is legal: 64 bytes is the maxLength, not one past it. + repo := newFakeCommunityRepo(writerCommunityDID) + boundedWriter := NewCommunityRecordWriter(fixedFactory(repo), writerClock) + _, err = boundedWriter.WriteRemoval(context.Background(), CommunityRemovalCommand{ + CommunityDID: writerCommunityDID, + PostURI: writerPostURI, + PostCID: writerPostCID, + Code: DecisionCode(strings.Repeat("x", 64)), + }) + assert.NoError(t, err, "a 64-byte code is exactly the lexicon's maxLength and must pass") +} + +// --------------------------------------------------------------------------- +// Swap-retry backoff +// --------------------------------------------------------------------------- + +func TestWriter_BacksOffWithJitterBetweenSwapRetries(t *testing.T) { + t.Parallel() + + // The writers that lose a swap to each other are the three acceptance + // writers of §3.2 converging on the same rkey, so a retry fired + // immediately — or after a FIXED interval — collides again on schedule. + // Each retry therefore waits a jittered duration bounded by a doubling + // ceiling. (Actually serializing the contenders is the per-community + // queue, which is task 5's job; this only decorrelates the collisions.) + // + // The sleeper is injected and recorded: docs/TEST_ARCHITECTURE.md §3.3 + // forbids a test from actually sleeping, so the assertion is on the + // durations, not the wall clock. + repo := newFakeCommunityRepo(writerCommunityDID) + repo.putErrs = []error{pds.ErrSwapConflict, pds.ErrSwapConflict, nil} + + var pauses []time.Duration + writer := NewCommunityRecordWriter(fixedFactory(repo), writerClock, + WithSwapRetrySleeper(func(_ context.Context, d time.Duration) error { + pauses = append(pauses, d) + return nil + })) + + result, err := writer.WriteAcceptance(context.Background(), CommunityWriteCommand{ + CommunityDID: writerCommunityDID, + PostURI: writerPostURI, + PostCID: writerPostCID, + }) + require.NoError(t, err, "two lost swaps are within the retry budget and must converge") + assert.False(t, result.Skipped) + + require.Lenf(t, pauses, 2, "one pause per retry: two conflicts, two pauses") + for i, pause := range pauses { + ceiling := 25 * time.Millisecond << i + assert.Positivef(t, pause, "pause %d must be positive — a zero pause is no backoff at all", i) + assert.LessOrEqualf(t, pause, ceiling, + "pause %d must stay under its doubling ceiling %v", i, ceiling) + } +} + +func TestWriter_ACancelledBackoffAbortsTheRetry(t *testing.T) { + t.Parallel() + + // A worker shutting down mid-backoff must not fire another write: the + // sleeper reports the cancellation and the writer surfaces it instead of + // retrying into a dying process. + repo := newFakeCommunityRepo(writerCommunityDID) + repo.putErrs = []error{pds.ErrSwapConflict} + + writer := NewCommunityRecordWriter(fixedFactory(repo), writerClock, + WithSwapRetrySleeper(func(_ context.Context, _ time.Duration) error { + return context.Canceled + })) + + _, err := writer.WriteAcceptance(context.Background(), CommunityWriteCommand{ + CommunityDID: writerCommunityDID, + PostURI: writerPostURI, + PostCID: writerPostCID, + }) + require.ErrorIs(t, err, context.Canceled) + assert.Lenf(t, repo.puts, 1, "the cancelled backoff must prevent the retry's put") +} + +// --------------------------------------------------------------------------- +// Call ordering and the swapCommit guard +// --------------------------------------------------------------------------- + +func TestWriter_RemovalCommitReadsTheHeadBeforeTheRecordsAndGuardsWithIt(t *testing.T) { + t.Parallel() + + // THE ORDER IS THE GUARD. A swapCommit read AFTER the records could be + // newer than the state the batch was shaped from — it would guard the + // commit against a revision that already contains the change the shape + // assumed absent. Read first, any interleaved write makes the guard stale: + // a detected conflict rather than a silent clobber. No outcome value + // reveals this order, so the fake records it. + repo := newFakeCommunityRepo(writerCommunityDID) + rkey := SubjectRkey(writerPostURI) + repo.setRecord(AcceptanceCollection, rkey, "bafyreiacceptanceaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", map[string]any{ + "$type": AcceptanceCollection, + "subject": map[string]any{"uri": writerPostURI, "cid": writerPostCID}, + "createdAt": "2026-06-30T09:00:00Z", + }) + + writer := NewCommunityRecordWriter(fixedFactory(repo), writerClock) + result, err := writer.WriteRemoval(context.Background(), CommunityRemovalCommand{ + CommunityDID: writerCommunityDID, + PostURI: writerPostURI, + PostCID: writerPostCID, + Code: DecisionSpam, + }) + require.NoError(t, err) + assert.False(t, result.Skipped) + + assert.Equal(t, []string{ + "GetLatestCommit", + "GetRecord:" + RemovalCollection, + "GetRecord:" + AcceptanceCollection, + "ApplyWrites", + }, repo.calls, "the head must be read BEFORE the record pre-reads the batch is shaped from") + + require.Len(t, repo.batches, 1) + assert.Equalf(t, repo.head.CID, repo.batches[0].swapCommit, + "the head CID that was read is the swapCommit the batch must be guarded by") +} + +// --------------------------------------------------------------------------- +// RepinAcceptance +// --------------------------------------------------------------------------- + +func TestWriter_RepinUpdatesTheStandingAcceptanceInPlace(t *testing.T) { + t.Parallel() + + // The bridgedStats exception of §5.5: the acceptance moves onto the new + // content CID as an UPDATE of the same record — guarded by the standing + // record's CID, carrying its createdAt forward — so the acceptance's + // createdAt keeps meaning "when this community accepted this post" and the + // record's identity survives every stats refresh. + repo := newFakeCommunityRepo(writerCommunityDID) + rkey := SubjectRkey(writerPostURI) + standingCID := "bafyreiacceptanceaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + repo.setRecord(AcceptanceCollection, rkey, standingCID, map[string]any{ + "$type": AcceptanceCollection, + "subject": map[string]any{"uri": writerPostURI, "cid": writerPostCID}, + "createdAt": "2026-06-30T09:00:00Z", + }) + + refreshedCID := "bafyreirefreshedaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + writer := NewCommunityRecordWriter(fixedFactory(repo), writerClock) + result, err := writer.RepinAcceptance(context.Background(), CommunityWriteCommand{ + CommunityDID: writerCommunityDID, + PostURI: writerPostURI, + PostCID: refreshedCID, + }) + require.NoError(t, err) + assert.False(t, result.Skipped, "the standing acceptance pins the old CID, so there was work to do") + assert.NotEmpty(t, result.Rev) + + require.Len(t, repo.puts, 1) + put := repo.puts[0] + assert.Equal(t, rkey, put.rkey, "the repin must reuse the subject's deterministic rkey") + assert.Equalf(t, standingCID, put.swapRecord, + "the update must be guarded by the record CID the pre-read found — an unguarded put would "+ + "clobber a concurrent writer in exactly the window the pre-read opened") + subject, _ := put.record["subject"].(map[string]any) + assert.Equal(t, refreshedCID, subject["cid"]) + assert.Equalf(t, "2026-06-30T09:00:00Z", put.record["createdAt"], + "createdAt must be carried forward — restamping it would rewrite when the community "+ + "accepted the post, on every stats refresh") +} + +func TestWriter_RepinRefusesWhenNoAcceptanceStands(t *testing.T) { + t.Parallel() + + // A repin moves a STANDING acceptance and re-decides nothing, so it has no + // authority to create one. An absent record means the AppView's row and + // the community's repo disagree, and silently minting an acceptance nobody + // decided is the one thing a path that skips admission must never do. The + // documented outcome is pds.ErrNotFound, which the caller defers on. + repo := newFakeCommunityRepo(writerCommunityDID) + writer := NewCommunityRecordWriter(fixedFactory(repo), writerClock) + + _, err := writer.RepinAcceptance(context.Background(), CommunityWriteCommand{ + CommunityDID: writerCommunityDID, + PostURI: writerPostURI, + PostCID: writerPostCID, + }) + require.Error(t, err) + assert.ErrorIs(t, err, pds.ErrNotFound) + assert.Emptyf(t, repo.puts, "nothing may be written: a repin never creates") + assert.Emptyf(t, repo.batches, "nothing may be committed") +} + +func TestWriter_AcceptanceRefusesAStandingRemoval(t *testing.T) { + t.Parallel() + + // The T0 face of the §5.5 removal guard (the real-PDS races are + // engine_contract_test.go): a removal standing at the subject's rkey + // refuses both the write and the repin before anything is put. + repo := newFakeCommunityRepo(writerCommunityDID) + rkey := SubjectRkey(writerPostURI) + repo.setRecord(RemovalCollection, rkey, "bafyreiremovalaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", map[string]any{ + "$type": RemovalCollection, + "subject": map[string]any{"uri": writerPostURI, "cid": writerPostCID}, + "code": string(DecisionSpam), + "createdAt": "2026-06-30T09:00:00Z", + }) + + writer := NewCommunityRecordWriter(fixedFactory(repo), writerClock) + cmd := CommunityWriteCommand{ + CommunityDID: writerCommunityDID, + PostURI: writerPostURI, + PostCID: writerPostCID, + } + + _, err := writer.WriteAcceptance(context.Background(), cmd) + require.ErrorIs(t, err, ErrRemovalStands) + _, err = writer.RepinAcceptance(context.Background(), cmd) + require.ErrorIs(t, err, ErrRemovalStands) + + assert.Emptyf(t, repo.puts, "an acceptance over a standing removal must never reach the repo") +} + +// --------------------------------------------------------------------------- +// The factory DID check +// --------------------------------------------------------------------------- + +func TestWriter_RefusesAFactoryThatReturnsTheWrongRepo(t *testing.T) { + t.Parallel() + + // The repo's DID is the AUTHORITY half of every record URI this writer + // produces. A factory bug that handed back another community's session + // would have one community vouching for a post with another community's + // key — an acceptance that looks perfectly valid to every consumer on the + // network. openRepo therefore proves the session is on the DID that was + // asked for, and refuses before anything is read or written. + wrongRepo := newFakeCommunityRepo("did:plc:dddddddddddddddddddddddd") + writer := NewCommunityRecordWriter(fixedFactory(wrongRepo), writerClock) + + _, err := writer.WriteAcceptance(context.Background(), CommunityWriteCommand{ + CommunityDID: writerCommunityDID, + PostURI: writerPostURI, + PostCID: writerPostCID, + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "did:plc:dddddddddddddddddddddddd", + "the refusal must name the DID the factory actually returned") + + _, err = writer.WriteRemoval(context.Background(), CommunityRemovalCommand{ + CommunityDID: writerCommunityDID, + PostURI: writerPostURI, + PostCID: writerPostCID, + Code: DecisionSpam, + }) + require.Error(t, err) + + // Refused up front: the wrong repo is never read, let alone written. + assert.Emptyf(t, wrongRepo.calls, + "openRepo must refuse before touching the repo at all; calls: %v", wrongRepo.calls) +} + +// --------------------------------------------------------------------------- +// The fake repo +// --------------------------------------------------------------------------- + +// fakeCommunityRepo is an in-memory CommunityRepo that records the order of +// every call, so a test can pin sequencing that no outcome value reveals. +type fakeCommunityRepo struct { + did string + calls []string + + // records is keyed by collection+"/"+rkey. + records map[string]*pds.RecordResponse + + head pds.LatestCommit + + // putErrs and applyErrs are consumed one per call, so a test can fail the + // first attempt and let the retry through. A nil entry means success. + putErrs []error + applyErrs []error + + // puts and batches record what was sent. + puts []fakePut + batches []fakeBatch +} + +type fakePut struct { + collection string + rkey string + record map[string]any + swapRecord string +} + +type fakeBatch struct { + writes []pds.Write + swapCommit string +} + +func newFakeCommunityRepo(did string) *fakeCommunityRepo { + return &fakeCommunityRepo{ + did: did, + records: map[string]*pds.RecordResponse{}, + head: pds.LatestCommit{CID: "bafyreiheadaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", Rev: "3kjzl5headaaa"}, + } +} + +func fixedFactory(repo CommunityRepo) CommunityRepoFactory { + return func(_ context.Context, _ string) (CommunityRepo, error) { return repo, nil } +} + +func (r *fakeCommunityRepo) key(collection, rkey string) string { return collection + "/" + rkey } + +func (r *fakeCommunityRepo) setRecord(collection, rkey, cid string, value map[string]any) { + r.records[r.key(collection, rkey)] = &pds.RecordResponse{ + URI: "at://" + r.did + "/" + collection + "/" + rkey, + CID: cid, + Value: value, + } +} + +func (r *fakeCommunityRepo) GetRecord(_ context.Context, collection, rkey string) (*pds.RecordResponse, error) { + r.calls = append(r.calls, "GetRecord:"+collection) + record, ok := r.records[r.key(collection, rkey)] + if !ok { + return nil, pds.ErrNotFound + } + return record, nil +} + +func (r *fakeCommunityRepo) PutRecordWithCommit(_ context.Context, collection, rkey string, record any, swapRecord string) (*pds.RecordCommit, error) { + r.calls = append(r.calls, "PutRecordWithCommit:"+collection) + body, _ := record.(map[string]any) + r.puts = append(r.puts, fakePut{collection: collection, rkey: rkey, record: body, swapRecord: swapRecord}) + + if len(r.putErrs) > 0 { + err := r.putErrs[0] + r.putErrs = r.putErrs[1:] + if err != nil { + return nil, err + } + } + + cid := "bafyreiputaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + r.setRecord(collection, rkey, cid, body) + return &pds.RecordCommit{ + URI: "at://" + r.did + "/" + collection + "/" + rkey, + CID: cid, + CommitRev: "3kjzl5putaaaa", + CommitCID: "bafyreicommitputaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + }, nil +} + +func (r *fakeCommunityRepo) ApplyWrites(_ context.Context, writes []pds.Write, swapCommit string) (*pds.ApplyWritesResult, error) { + r.calls = append(r.calls, "ApplyWrites") + r.batches = append(r.batches, fakeBatch{writes: writes, swapCommit: swapCommit}) + + if len(r.applyErrs) > 0 { + err := r.applyErrs[0] + r.applyErrs = r.applyErrs[1:] + if err != nil { + return nil, err + } + } + + results := make([]pds.WriteResult, len(writes)) + for i, write := range writes { + switch write.Op { + case pds.WriteOpDelete: + delete(r.records, r.key(write.Collection, write.RKey)) + results[i] = pds.WriteResult{Op: write.Op} + default: + body, _ := write.Record.(map[string]any) + cid := "bafyreibatchaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + r.setRecord(write.Collection, write.RKey, cid, body) + results[i] = pds.WriteResult{ + Op: write.Op, + URI: "at://" + r.did + "/" + write.Collection + "/" + write.RKey, + CID: cid, + } + } + } + return &pds.ApplyWritesResult{CommitRev: "3kjzl5batchaa", CommitCID: "bafyreicommitbatchaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", Results: results}, nil +} + +func (r *fakeCommunityRepo) GetLatestCommit(_ context.Context) (*pds.LatestCommit, error) { + r.calls = append(r.calls, "GetLatestCommit") + head := r.head + return &head, nil +} + +func (r *fakeCommunityRepo) DID() string { return r.did } diff --git a/internal/core/posts/engine.go b/internal/core/posts/engine.go index f504558..2cc78da 100644 --- a/internal/core/posts/engine.go +++ b/internal/core/posts/engine.go @@ -17,9 +17,24 @@ import ( // agree with the answer: an acceptance record, a removal commit, or an // AppView-local rejection that writes no record at all. // -// It is the ONLY writer of community-repo records in the post system, which is -// what makes "every write is idempotent" a property of the system rather than a -// convention each call site has to remember. +// CommunityRecordWriter is the single component that writes community-repo +// records in the post system, and this engine is its decision point: every +// verdict that becomes a record flows through here first. That funnel is what +// makes "every write is idempotent" a property of the system rather than a +// convention each call site has to remember — the writers get their +// idempotency from deterministic rkeys and swap guards, and the engine is the +// one place that decides they should fire at all. +// +// THERE IS NO LEASE, AND THAT IS DELIBERATE. Nothing stops two passes — the +// fast path, a firehose redelivery, a notify — from processing the same +// subject at the same moment, and no lock or per-subject claim is taken. +// Safety comes from the layers instead: deterministic rkeys make the racers +// aim at the same record; every put and batch is swap-guarded, so a loser is +// told rather than clobbering; a loser that re-reads and finds the winner +// wrote its exact target converges as a skip; and the repository's watermark +// CAS makes the row's state advance monotonically no matter which pass +// stamps first. Serializing the passes properly (a per-community queue) is +// task 5's job; until then concurrent passes are expected and harmless. // EngineOutcome reports what one pass over one subject DID. // @@ -46,13 +61,27 @@ const ( // EngineRepinned means a standing acceptance moved onto new content with no // re-decision — the bridgedStats exception of §5.5. + // + // NOT PRODUCED BY ANY PATH YET. ProcessAdmission runs full re-admission on + // every edit; the repin path — classifyRecordDiff choosing the exception, + // the bridge-trust gate approving the author, RepinAcceptance moving the + // record — is task 5's consumer wiring. The outcome is declared now so + // that path lands against a named contract instead of minting one. EngineRepinned EngineOutcome = "repinned" - // EngineDeferred means NOTHING was written anywhere and the subject is - // still owed a decision. It covers an undecided policy answer, a row whose - // content CID is not yet known, a row already in a terminal state, and a - // credential failure. In every one of those cases the correct next step is - // to look again later, never to record a verdict. + // EngineDeferred means the subject is still owed a decision and nothing + // NEW was verdicted. It covers an undecided policy answer, a row whose + // content CID is not yet known, a row already in a terminal state, a + // credential failure, and a repo write refused by the §5.5 removal guard. + // In every one of those cases the correct next step is to look again + // later, never to record a verdict. + // + // "Nothing was written" is the local truth, but a REMOTE outcome can be + // ambiguous: a PDS write whose response was lost may have committed + // anyway. That ambiguity is why deferral is always safe to re-fire — the + // next pass's pre-read finds whatever actually stands, a write that + // already landed converges as a skip, and the skip's catch-up stamp + // (see accept) reconciles the row with it. EngineDeferred EngineOutcome = "deferred" ) @@ -219,12 +248,25 @@ func (e *AcceptanceEngine) accept(ctx context.Context, communityDID, postURI, ev return EngineDeferred, err } - // A SKIPPED WRITE STAMPS NOTHING. The repo already held this exact - // acceptance, so nothing committed and there is no revision to report — - // getRecord does not reveal the revision an existing record was written at. - // The repository refuses an empty rev as a fabricated watermark, and - // inventing one to get past that would write a clock value no commit had. - if written.Skipped { + // A SKIPPED WRITE STILL STAMPS — with the catch-up watermark. The repo + // already held this exact acceptance, so nothing committed; the writer + // reports the repo's HEAD rev instead (read before its pre-read — see + // CommunityWriteResult.Rev). Stamping it is what un-strands a row whose + // previous pass committed the acceptance and then failed this very stamp: + // until task 5's reconciler exists, a re-fire of this engine is the only + // thing that revisits the subject, and a skip that stamped nothing would + // leave the row pending forever. + // + // WHY THE HEAD REV IS SAFE: a standing acceptance pinning our CID proves no + // subject-scoped community event lies between the acceptance's commit and + // the head — a removal would have deleted the record, and a repin would pin + // a different CID. The stamp is also conservative: the head was read before + // the record, so any event racing the pre-read carries a strictly greater + // rev and its firehose copy still applies. + if written.Skipped && written.Rev == "" { + // Defensive only: the writer contract reports the head rev on every + // skip. An empty one must not be stamped — the repository refuses it as + // a fabricated watermark, correctly. return EngineAccepted, nil } @@ -254,7 +296,7 @@ func (e *AcceptanceEngine) accept(ctx context.Context, communityDID, postURI, ev // community's repository — spam would otherwise be permanently archived in the // repo of the community that refused it. func (e *AcceptanceEngine) reject(ctx context.Context, communityDID, postURI, evaluatedCID string, code DecisionCode) (EngineOutcome, error) { - if _, err := e.admissions.RecordRejection(ctx, RecordRejectionCommand{ + result, err := e.admissions.RecordRejection(ctx, RecordRejectionCommand{ CommunityDID: communityDID, PostURI: postURI, DecisionCode: string(code), @@ -266,10 +308,23 @@ func (e *AcceptanceEngine) reject(ctx context.Context, communityDID, postURI, ev // A policy refusal is terminal. Leaving this true would have the // dead-letter redrive pass retry a decision that will never change. Redrivable: false, - }); err != nil { + }) + if err != nil { return EngineDeferred, fmt.Errorf("recording the rejection of %s in %s: %w", postURI, communityDID, err) } + // A SKIPPED REJECTION DID NOT LAND, and the outcome must say so. The CAS + // refuses in two ways and both mean "no rejection was recorded": + // skipped_stale, the author edited between the read and the verdict, so + // the judged CID is no longer the row's and the edit will re-drive the + // subject; skipped_terminal, another writer settled the row first. + // Reporting EngineRejected for either would claim a refusal the row does + // not hold — and the caller would answer the author with a verdict that + // was never made. + if result.Outcome != AdmissionApplied { + return EngineDeferred, nil + } + return EngineRejected, nil } @@ -289,7 +344,13 @@ func (e *AcceptanceEngine) remove(ctx context.Context, communityDID, postURI, ev return EngineDeferred, err } - if written.Skipped { + // The removal twin of accept's catch-up stamp: a skipped removal reports + // the head rev, and stamping it un-strands a row whose previous pass + // committed the removal but failed the stamp. Safe for the same shape of + // reason — a standing removal carrying this decision proves no + // subject-scoped community event lies between its commit and the head (a + // restore would have deleted the removal in its own commit). + if written.Skipped && written.Rev == "" { return EngineRemoved, nil } diff --git a/internal/core/posts/engine_contract_test.go b/internal/core/posts/engine_contract_test.go index d710c2a..a5cef13 100644 --- a/internal/core/posts/engine_contract_test.go +++ b/internal/core/posts/engine_contract_test.go @@ -4,6 +4,7 @@ package posts_test import ( "context" + "errors" "testing" "time" @@ -257,9 +258,9 @@ func TestEngine_AcceptanceLandsInTheCommunityRepoAndSurvivesRefiring(t *testing. require.NoError(t, err) assert.Truef(t, result.Skipped, "the repo already held this exact acceptance, so the writer must write NOTHING") - assert.Emptyf(t, result.Rev, - "nothing committed, so there is no revision to report — and a caller that stamped one "+ - "would write a watermark no commit ever had") + assert.NotEmptyf(t, result.Rev, + "a skip reports the repo's head rev — the catch-up watermark that un-strands a row whose "+ + "earlier stamp failed after the acceptance committed (see CommunityWriteResult.Rev)") // THE ASSERTION WITH TEETH. assert.Equalf(t, firstRecordCID, f.acceptanceOf(t, post.URI).CID, @@ -410,6 +411,170 @@ func TestEngine_ApplyAcceptanceTwiceAtTheSameRevIsASkipThatChangesNothing(t *tes "how a replay looks like a fresh decision in the moderation log") } +// stampFailingAdmissions fails ApplyAcceptance a scripted number of times and +// then delegates — the database blip that strikes AFTER the PDS commit landed. +type stampFailingAdmissions struct { + posts.AdmissionRepository + failures int +} + +func (a *stampFailingAdmissions) ApplyAcceptance(ctx context.Context, cmd posts.ApplyAcceptanceCommand) (posts.AdmissionResult, error) { + if a.failures > 0 { + a.failures-- + return posts.AdmissionResult{}, errAppViewDown + } + return a.AdmissionRepository.ApplyAcceptance(ctx, cmd) +} + +// errAppViewDown is the injected stamp failure: the AppView's own database +// refusing the write, after the PDS commit already landed. +var errAppViewDown = errors.New("injected: the admissions store is unreachable") + +func TestEngine_ReFireAfterAFailedStampCatchesUpViaTheSkipPath(t *testing.T) { + t.Parallel() + + // THE STRANDED-ROW SCENARIO. Pass one commits the acceptance on the PDS and + // then fails to stamp the row — a database blip after the write landed. The + // record stands, the row is still pending, and until task 5's reconciler + // exists the ONLY thing that revisits the subject is a re-fire of this same + // engine. On that re-fire the writer skips (the record already pins the + // target), so if the skip path stamps nothing the row is pending forever. + // + // The skip therefore carries the repo's head rev and the engine stamps it. + // Safe, because a standing acceptance pinning our CID proves no + // subject-scoped community event lies between the acceptance's commit and + // the head: a removal would have deleted the record, and a repin would pin + // a different CID. + f := newEngineFixture(t) + ctx := context.Background() + + post := f.publishPost(t, "a post whose first stamp fails") + f.seedPending(t, post.URI, post.CID) + + admissions := &stampFailingAdmissions{AdmissionRepository: f.admissions, failures: 1} + engine := posts.NewAcceptanceEngine(admissions, f.decider, f.writer, f.refreshes) + + outcome, err := engine.ProcessAdmission(ctx, f.community.DID, post.URI) + require.Error(t, err, "the failed stamp must be visible — the repo and the row now disagree") + require.Equal(t, posts.EngineAccepted, outcome, + "the acceptance IS in the community's repo; that is what the outcome reports") + + row, err := f.admissions.Get(ctx, f.community.DID, post.URI) + require.NoError(t, err) + require.Equalf(t, posts.AdmissionStatusPending, row.Status, + "precondition: the stamp failed, so the row must still be pending") + + // The re-fire. The writer finds the acceptance standing and skips; the + // catch-up stamp is what moves the row. + outcome, err = engine.ProcessAdmission(ctx, f.community.DID, post.URI) + require.NoError(t, err) + assert.Equal(t, posts.EngineAccepted, outcome) + + row, err = f.admissions.Get(ctx, f.community.DID, post.URI) + require.NoError(t, err) + assert.Equalf(t, posts.AdmissionStatusAccepted, row.Status, + "the re-fire must catch the row up: the acceptance stands on the PDS and nothing else "+ + "re-drives this subject until task 5 exists") + require.NotNil(t, row.LastCommunityEvent) + assert.NotEmpty(t, row.LastCommunityEvent.Rev) + + // And the catch-up minted nothing: the record's CID is untouched. + acceptance := f.acceptanceOf(t, post.URI) + assertSubject(t, acceptance, post.URI, post.CID) +} + +// --------------------------------------------------------------------------- +// The removal guard +// --------------------------------------------------------------------------- + +func TestEngine_AcceptanceRefusesToWriteOverAStandingRemoval(t *testing.T) { + t.Parallel() + + // §5.5: removal is terminal, and the ONLY sanctioned exit is a moderator + // restore — one commit that deletes the removal AND writes the fresh + // acceptance. An acceptance created over a standing removal leaves both + // records live, and the acceptance's younger watermark outranks the removal + // at every consumer: a moderated post laundered back into feeds by a retry. + // + // Here the removal already stands at the writer's pre-read: the community + // removed the post, and this engine pass is working from a row the firehose + // has not caught up yet. + f := newEngineFixture(t) + ctx := context.Background() + + post := f.publishPost(t, "a post removed before the engine fires") + _, err := f.writer.WriteRemoval(ctx, posts.CommunityRemovalCommand{ + CommunityDID: f.community.DID, + PostURI: post.URI, + PostCID: post.CID, + Code: posts.DecisionSpam, + }) + require.NoError(t, err) + + f.seedPending(t, post.URI, post.CID) + + outcome, err := f.process(t, post.URI) + require.Error(t, err, "an acceptance over a standing removal must be refused, not committed") + assert.ErrorIs(t, err, posts.ErrRemovalStands) + assert.Equalf(t, posts.EngineDeferred, outcome, + "the refusal is a deferral — the subject is owed a decision, but not this one") + + rkey := posts.SubjectRkey(post.URI) + removal := f.communityAt.GetRecord(t, posts.RemovalCollection, rkey) + assert.Equalf(t, string(posts.DecisionSpam), removal.Value["code"], + "the removal must still stand untouched") + f.assertRecordAbsent(t, posts.AcceptanceCollection, rkey, "the acceptance record") +} + +func TestEngine_AcceptanceRefusesARemovalDiscoveredMidConvergence(t *testing.T) { + t.Parallel() + + // The harder shape of the same guard: the removal lands BETWEEN the + // writer's pre-read and its put. The put loses its swapRecord guard (the + // acceptance it aimed at was deleted by the removal commit), and the + // convergence re-read is where the standing removal has to be discovered — + // a re-read that only looked at the acceptance rkey would see "nothing + // there" and create straight over the removal. + f := newEngineFixture(t) + ctx := context.Background() + + post := f.publishPost(t, "a post removed mid-write") + _, err := f.writer.WriteAcceptance(ctx, posts.CommunityWriteCommand{ + CommunityDID: f.community.DID, + PostURI: post.URI, + PostCID: post.CID, + }) + require.NoError(t, err) + + edited := f.editPost(t, post, "an edit whose re-acceptance races a removal") + + // Between our pre-read and our put, a moderation removal commits: the + // acceptance is deleted and the removal created, in one commit. + racing := f.racingWriter(t, func() { + _, removeErr := f.writer.WriteRemoval(ctx, posts.CommunityRemovalCommand{ + CommunityDID: f.community.DID, + PostURI: post.URI, + PostCID: post.CID, + Code: posts.DecisionRuleViolation, + }) + require.NoError(t, removeErr) + }) + + _, err = racing.WriteAcceptance(ctx, posts.CommunityWriteCommand{ + CommunityDID: f.community.DID, + PostURI: post.URI, + PostCID: edited.CID, + }) + require.Error(t, err, + "the convergence re-read met a standing removal and must refuse rather than create over it") + assert.ErrorIs(t, err, posts.ErrRemovalStands) + + rkey := posts.SubjectRkey(post.URI) + removal := f.communityAt.GetRecord(t, posts.RemovalCollection, rkey) + assert.Equal(t, string(posts.DecisionRuleViolation), removal.Value["code"]) + f.assertRecordAbsent(t, posts.AcceptanceCollection, rkey, "the acceptance record") +} + // --------------------------------------------------------------------------- // Swap conflicts // --------------------------------------------------------------------------- @@ -447,6 +612,10 @@ func (f *engineFixture) racingWriter(t *testing.T, fn func()) posts.CommunityRec return &racingRepo{CommunityRepo: repo, race: fn}, nil }, func() time.Time { return time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) }, + // The race is deterministic here, so the retry backoff would only slow + // the suite: docs/TEST_ARCHITECTURE.md §3.3 — waiting is asserted on, + // never performed. + posts.WithSwapRetrySleeper(func(context.Context, time.Duration) error { return nil }), ) } @@ -462,6 +631,99 @@ func (f *engineFixture) writeAcceptanceDirectly(t *testing.T, postURI, pinnedCID }) } +// racingCommitRepo fires its race once, before the FIRST ApplyWrites, and +// records every inner result — so a test can prove the first attempt met a +// REAL InvalidSwap from a real PDS and the writer then converged, rather than +// merely observing a final state that an unguarded batch would also reach. +type racingCommitRepo struct { + posts.CommunityRepo + + race func() + fired bool + errs []error +} + +func (r *racingCommitRepo) ApplyWrites(ctx context.Context, writes []pds.Write, swapCommit string) (*pds.ApplyWritesResult, error) { + if !r.fired { + r.fired = true + r.race() + } + result, err := r.CommunityRepo.ApplyWrites(ctx, writes, swapCommit) + r.errs = append(r.errs, err) + return result, err +} + +func TestEngine_RemovalCommitLosesItsSwapCommitAndConverges(t *testing.T) { + t.Parallel() + + // The commitPair twin of the putRecord swap races below. The removal batch + // is guarded by the head CID read before its pre-reads, so ANY commit + // landing in the community's repo mid-window — here an unrelated + // acceptance for a different post — makes the guard stale. The PDS must + // answer with a real InvalidSwap (not a fake's error value), and the + // writer must re-read, re-shape and converge rather than either failing or + // silently clobbering. + f := newEngineFixture(t) + ctx := context.Background() + + post := f.publishPost(t, "a post whose removal loses the swapCommit race") + _, err := f.writer.WriteAcceptance(ctx, posts.CommunityWriteCommand{ + CommunityDID: f.community.DID, + PostURI: post.URI, + PostCID: post.CID, + }) + require.NoError(t, err) + + otherPost := f.publishPost(t, "an unrelated post whose acceptance advances the head") + + generic, err := pds.NewFromAccessToken(f.pds.URL(), f.communityAt.DID, f.communityAt.AccessToken) + require.NoError(t, err) + commitClient, ok := generic.(pds.CommitClient) + require.True(t, ok) + + racing := &racingCommitRepo{ + CommunityRepo: commitClient, + race: func() { + // A competing write commits between the batch's head read and its + // applyWrites — another engine pass, the fast path, a moderator. + f.writeAcceptanceDirectly(t, otherPost.URI, otherPost.CID) + }, + } + writer := posts.NewCommunityRecordWriter( + func(_ context.Context, _ string) (posts.CommunityRepo, error) { return racing, nil }, + func() time.Time { return time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) }, + posts.WithSwapRetrySleeper(func(context.Context, time.Duration) error { return nil }), + ) + + result, err := writer.WriteRemoval(ctx, posts.CommunityRemovalCommand{ + CommunityDID: f.community.DID, + PostURI: post.URI, + PostCID: post.CID, + Code: posts.DecisionSpam, + }) + require.NoErrorf(t, err, "a lost swapCommit within the retry budget must converge, not surface") + assert.False(t, result.Skipped, "the removal had real work to do") + assert.NotEmpty(t, result.Rev) + + // THE GUARD WAS REALLY ENGAGED. Without this assertion an unguarded batch + // — one that dropped swapCommit — would sail through first try and reach + // the same final state, and the test would prove nothing about the guard. + require.GreaterOrEqualf(t, len(racing.errs), 2, + "the first batch must have been refused and retried; attempts: %d", len(racing.errs)) + assert.ErrorIsf(t, racing.errs[0], pds.ErrSwapConflict, + "the competing commit must surface as a real InvalidSwap from the PDS, mapped to "+ + "ErrSwapConflict — got: %v", racing.errs[0]) + assert.NoError(t, racing.errs[len(racing.errs)-1], "the re-shaped batch must commit cleanly") + + // And the converged state is the moderation action, whole: acceptance + // gone, removal standing, in the community's repo. + rkey := posts.SubjectRkey(post.URI) + f.assertRecordAbsent(t, posts.AcceptanceCollection, rkey, "the acceptance record") + removal := f.communityAt.GetRecord(t, posts.RemovalCollection, rkey) + assert.Equal(t, string(posts.DecisionSpam), removal.Value["code"]) + assertSubject(t, removal, post.URI, post.CID) +} + func TestEngine_LostSwapRaceToTheSameCIDIsAlreadyDone(t *testing.T) { t.Parallel() diff --git a/internal/core/posts/engine_matrix_test.go b/internal/core/posts/engine_matrix_test.go index df21d46..9b81ac4 100644 --- a/internal/core/posts/engine_matrix_test.go +++ b/internal/core/posts/engine_matrix_test.go @@ -85,6 +85,13 @@ type fakeDecider struct { code DecisionCode err error + // cause is the VALUE-shaped undecided answer: Cause set, Code empty, and + // a NIL error. It is a separate field from err so the fake can produce + // each of the two undecided shapes independently — a policy bug (or a + // future refusal path) can hand back exactly this, and the engine must + // treat it as undecided rather than as a verdict. + cause error + lastCommunityDID string lastPostURI string } @@ -99,6 +106,9 @@ func (d *fakeDecider) DecideAdmission(_ context.Context, communityDID, postURI s // sees "not admitted". return AdmissionDecision{Cause: d.err}, d.err } + if d.cause != nil { + return AdmissionDecision{Cause: d.cause}, nil + } return AdmissionDecision{Code: d.code}, nil } @@ -485,6 +495,27 @@ func TestEngine_UndecidedWritesNothingAnywhere(t *testing.T) { } } +func TestEngine_ValueShapedUndecidedWritesNothingAnywhere(t *testing.T) { + t.Parallel() + + // The undecided answer arriving as a VALUE: Cause set, Code empty, error + // nil. Admitted() is false for a refusal and for an undecided answer + // alike, and the empty code is the only thing telling them apart — an + // answer with neither a code nor a clean bill is one nothing may be + // written from. A decider bug that produced this shape must cost a + // deferral, never a verdict. + lookupFailed := errors.New("aggregator authorization: connection refused") + h := newEngineHarness(AdmissionStatusPending, cidPtr(engineIndexedCID)) + h.decider.cause = lookupFailed + + outcome, err := h.process(t) + assert.Equal(t, EngineDeferred, outcome) + require.Error(t, err, "a policy that returned no verdict is a genuine failure and must be visible") + assert.ErrorIs(t, err, lookupFailed) + + assertWroteNothing(t, h.rec) +} + func TestEngine_AdmittedWithNoIndexedCIDDefers(t *testing.T) { t.Parallel() @@ -658,17 +689,15 @@ func TestEngine_TreatsRepositorySkipsAsSuccess(t *testing.T) { } } -func TestEngine_SkippedWriteStampsNothing(t *testing.T) { +func TestEngine_SkippedWriteWithNoRevStampsNothing(t *testing.T) { t.Parallel() - // When the community's repo already holds an acceptance pinning this exact - // CID, the writer writes nothing and reports no commit rev — getRecord does - // not reveal the revision an existing record was written at. - // - // So the engine must NOT stamp the row: ApplyAcceptance refuses an empty - // rev with ErrInvalidWatermark (correctly — an empty rev is a fabricated - // clock value), and inventing one to get past that would write a watermark - // no commit ever had. + // THE DEFENSIVE HALF of the skip contract. The writer reports the repo's + // head rev on every skip (see TestEngine_SkippedWriteStampsTheCatchUpWatermark), + // but if a skip ever arrives WITHOUT one, the engine must not stamp: + // ApplyAcceptance refuses an empty rev with ErrInvalidWatermark (correctly + // — an empty rev is a fabricated clock value), and inventing one to get + // past that would write a watermark no commit ever had. h := newEngineHarness(AdmissionStatusPending, cidPtr(engineIndexedCID)) h.writer.acceptanceResult = CommunityWriteResult{ URI: "at://" + engineCommunityDID + "/" + AcceptanceCollection + "/" + SubjectRkey(enginePostURI), @@ -687,6 +716,95 @@ func TestEngine_SkippedWriteStampsNothing(t *testing.T) { "a skipped write has no commit rev, and the repository refuses an empty one as a fabricated watermark") } +func TestEngine_SkippedWriteStampsTheCatchUpWatermark(t *testing.T) { + t.Parallel() + + // THE RE-FIRE AFTER A LOST STAMP. The first pass wrote the acceptance and + // then failed ApplyAcceptance — a database blip after a successful PDS + // commit. The record stands, the row is still pending, and the next pass's + // write is a skip. If a skip stamps nothing, that row is stranded until a + // human notices: nothing else re-drives it before task 5 exists. + // + // So a skip carries the repo's HEAD rev and the engine stamps it. That is + // safe because a standing acceptance pinning our CID proves no + // subject-scoped community event lies between the acceptance's commit and + // the head: a removal would have deleted the record, and a repin would pin + // a different CID. + h := newEngineHarness(AdmissionStatusPending, cidPtr(engineIndexedCID)) + h.writer.acceptanceResult = CommunityWriteResult{ + URI: "at://" + engineCommunityDID + "/" + AcceptanceCollection + "/" + SubjectRkey(enginePostURI), + RKey: SubjectRkey(enginePostURI), + CID: engineRecordCID, + // The head rev the writer read around its pre-read — the catch-up + // watermark. + Rev: engineCommitRev, + Skipped: true, + } + + outcome, err := h.process(t) + require.NoError(t, err) + assert.Equal(t, EngineAccepted, outcome) + + assert.Equal(t, []string{"Get", "DecideAdmission", "WriteAcceptance", "ApplyAcceptance"}, h.rec.calls, + "a skipped write that reports a head rev must still stamp the row — that is what un-strands "+ + "a row whose first stamp failed after the PDS commit landed") + require.Len(t, h.admissions.acceptanceCmds, 1) + assert.Equal(t, engineCommitRev, h.admissions.acceptanceCmds[0].Watermark.Rev) + assert.Equal(t, engineIndexedCID, h.admissions.acceptanceCmds[0].PinnedCID) +} + +func TestEngine_SkippedRemovalStampsTheCatchUpWatermark(t *testing.T) { + t.Parallel() + + // The removal twin of the acceptance catch-up: a removal commit landed, the + // stamp failed, and the re-fire's write is a skip carrying the head rev. + h := newEngineHarness(AdmissionStatusPendingReacceptance, cidPtr(engineIndexedCID)) + h.decider.code = DecisionSpam + h.writer.removalResult = CommunityWriteResult{ + URI: "at://" + engineCommunityDID + "/" + RemovalCollection + "/" + SubjectRkey(enginePostURI), + RKey: SubjectRkey(enginePostURI), + CID: engineRemovalCID, + Rev: engineCommitRev, + Skipped: true, + } + + outcome, err := h.process(t) + require.NoError(t, err) + assert.Equal(t, EngineRemoved, outcome) + + assert.Equal(t, []string{"Get", "DecideAdmission", "WriteRemoval", "ApplyRemoval"}, h.rec.calls) + require.Len(t, h.admissions.removalCmds, 1) + assert.Equal(t, engineCommitRev, h.admissions.removalCmds[0].Watermark.Rev) +} + +func TestEngine_ARejectionThatDidNotLandIsDeferredNotRejected(t *testing.T) { + t.Parallel() + + // RecordRejection is a pending-only CAS carrying the judged CID, and both + // of its skip outcomes mean THE REJECTION DID NOT LAND: skipped_stale, the + // author edited between the read and the write, so the verdict judged + // content the row no longer holds; skipped_terminal, another writer settled + // the row first. Reporting EngineRejected for either would claim a refusal + // that was never recorded — and the caller would tell the author their post + // was rejected while the row says otherwise. The honest outcome is a + // deferral: nothing landed, and the edit (or the settled state) is what + // drives the subject next. + for _, skip := range []AdmissionOutcome{AdmissionSkippedStale, AdmissionSkippedTerminal} { + t.Run(string(skip), func(t *testing.T) { + t.Parallel() + + h := newEngineHarness(AdmissionStatusPending, cidPtr(engineIndexedCID)) + h.decider.code = DecisionSpam + h.admissions.rejectionResult = AdmissionResult{Outcome: skip, Admission: h.admissions.row} + + outcome, err := h.process(t) + require.NoError(t, err, "a rejection refused by the CAS is the guard working, not a failure") + assert.Equalf(t, EngineDeferred, outcome, + "the rejection did not land, so the pass must not report EngineRejected") + }) + } +} + func TestEngine_ReportsAFailedStamp(t *testing.T) { t.Parallel() diff --git a/internal/core/posts/record_diff.go b/internal/core/posts/record_diff.go index 6036714..86466be 100644 --- a/internal/core/posts/record_diff.go +++ b/internal/core/posts/record_diff.go @@ -32,6 +32,16 @@ const ( // classifyRecordDiff reports whether the change between two versions of a post // record is the bridgedStats refresh of §5.5 or an edit needing re-admission. // +// IT CLASSIFIES THE DIFF ONLY. §5.5 conditions the repin on the bridge-trust +// gate as well — "the record diff touches only bridgedStats AND the author +// passes the bridge-trust gate" — and the gate is the CALLER's to apply. A +// RecordDiffBridgedStatsOnly answer is necessary for a repin, never sufficient. +// +// NOT CALLED BY ANYTHING YET. It ships with the engine so that the repin path +// has a decision procedure to call when it lands (task 5's consumer wiring, +// alongside EngineRepinned and CommunityRecordWriter.RepinAcceptance), and it +// is specified now so that path cannot be written against a guess. +// // IT TAKES DECODED RECORDS, NOT PostRecord VALUES, AND THAT IS THE WHOLE POINT. // A typed struct silently drops every field it does not know about, so an // author who added an unmodelled field — or a bridge running a newer lexicon diff --git a/internal/core/posts/record_diff_test.go b/internal/core/posts/record_diff_test.go index 4016177..86e2c47 100644 --- a/internal/core/posts/record_diff_test.go +++ b/internal/core/posts/record_diff_test.go @@ -24,6 +24,11 @@ import ( // struct discards unknown fields, so an author who added one would produce two // structs comparing equal and an edit classified as "nothing changed". // +// The classification is half of the §5.5 rule, not all of it: §5.5 conditions +// the repin on the bridge-trust gate as well — this function classifies the +// diff only; the caller applies the gate. A stats-only answer is necessary for +// a repin, never sufficient. +// // classifyRecordDiff is not called by anything yet. It ships with the engine so // that the repin path has a decision procedure to call when it lands, and it is // specified here so that path cannot be written against a guess. diff --git a/internal/core/posts/service_writeforward_test.go b/internal/core/posts/service_writeforward_test.go index cac7d11..99f2bf2 100644 --- a/internal/core/posts/service_writeforward_test.go +++ b/internal/core/posts/service_writeforward_test.go @@ -246,6 +246,12 @@ func TestService_DeleteRemovesTheRecordFromTheCommunityRepo(t *testing.T) { assert.NoError(t, f.service.DeletePost(ctx, sessionFor(t, f.author, f.pds.URL()), posts.DeletePostRequest{URI: resp.URI}), "a repeated delete is idempotent — the retried delete after a lost response succeeds") + + // And idempotent means the record STAYED gone: a second delete that + // somehow resurrected or re-wrote the record would also return success, + // so the absence has to be re-asserted, not assumed. + assert.True(t, testkit.IsNotFound(getRecordErr(ctx, community, postCollection, rkey)), + "the record must still be absent after the idempotent re-delete") } func TestService_DeleteRefusesEveryoneButTheAuthor(t *testing.T) {