From c0569a60e7358b4fa61419cfd2d3888ca444f861 Mon Sep 17 00:00:00 2001 From: Bretton Date: Sat, 8 Aug 2026 03:50:36 -0700 Subject: [PATCH] =?UTF-8?q?fix(ingestion):=20apply=20the=20review=20batch?= =?UTF-8?q?=20=E2=80=94=20verification,=20durability,=20erasure?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 17 of the 18 review pins are green; two are blocked on test-side contract problems reported to the conductor (below). T0 is fully clean. P1 direct fetch now uses com.atproto.sync.getRecord and RECOMPUTES the CID from the CAR's own record bytes. repo.getRecord answers with a JSON envelope whose `cid` is a claim by the author's PDS — a server chosen by a DID document and reached because a stranger wrote an acceptance — so comparing the pin against it asked the attacker whether the attacker was lying, and a match let the AppView index substituted content under a community's SIGNED acceptance. The reported CID is the computed one, not the label, or the claim would re-enter the comparison it was removed from. P2 lone removal-deletes now apply. Paired commits converge without them (the put outranks its delete), but a moderator withdrawing a removal writes only the delete, and ignoring it left the post `removed` forever while the community's repo no longer said so. The subject is recovered by recomputing SubjectRkey over that community's `removed` rows — bounded and index-backed, since the rkey digest is one-way. P3 converge honours applied=false. The post's own event can land while the fetch is in flight — the normal interleaving, since the fetch exists because the event had not arrived — and UpsertPending is last-write-wins, so running it anyway stamped the fetched CID over a newer one. P4 the rev gate now guards only the posts-row mutation. The two writes have opposite idempotence, and gating them together orphaned admissions: a failed upsert left the post indexed, the gate advanced, and every redelivery skipped. An equal stored rev means a replay (safe to re-run); a greater one means a newer event applied (must not). P5 firehose quota over admission rows, not the ledger. post_submissions is written by CreatePost, so a remote author has no row and never will — counting it would meter local users and exempt the remote ones §8 is for. P6 bounded over-fetch (batch + held deferrals, capped at ×4) so a stuck prefix cannot starve the backlog behind it, plus deferral pruning. P7 a failed classification defers instead of downgrading. The write path keeps the downgrade: it answers a live client who can retry, while the engine stamps redrivable=false and would mark a post permanently refused for a reason that was never true. P8a IndexUser no longer clears the erasure marker. It is the FIREHOSE's door into the users table, so any repo could un-erase an account by emitting one record. Registration still clears it. Gated through an optional ErrasureLookup rather than a UserRepository method, so no double can satisfy it by failing open. P9 an acceptance for a tombstoned post is a skip. P10 immutability now outranks the unknown-community branch, which is transient — so a retarget at a ghost DID used to dead-letter and redrive ten times and could never succeed. P11 SSRF: the vetted addresses ride on the request context and the dialler connects to one of THEM, ignoring the hostname. The base transport used to re-resolve, so the guard approved a host the connection never went to. P12 getStatus: spec-derived length bounds (a 2048-byte DID lives INSIDE an author-scoped URI), AT-URI parsing, Cache-Control: no-store, own 60/min limiter. P13 the acceptance withdrawal is reconsidered on every delivery. Tying it to the gate meant one unreachable PDS left the acceptance standing forever. Also: one shared SSRF-safe client instead of one per fetch, a 5s per-fetch deadline inside it, and a doc-truth pass over the stale RED-STUB paragraphs, the engine's task-5 promises, and the future-tense fast-path claims. go-car is pinned to the version indigo already requires; `go mod tidy` was NOT used (it upgraded transitive deps into a broken go-log). Co-Authored-By: Claude Fable 5 --- go.mod | 8 + go.sum | 16 + internal/api/handlers/post/getstatus.go | 52 ++- internal/api/routes/post.go | 38 +- internal/atproto/jetstream/authorpost.go | 424 +++++++++++++++---- internal/atproto/jetstream/post_consumer.go | 10 +- internal/atproto/oauth/transport.go | 78 +++- internal/core/posts/decider.go | 120 ++++-- internal/core/posts/engine.go | 27 +- internal/core/posts/queue.go | 98 ++++- internal/core/users/interfaces.go | 15 + internal/core/users/service.go | 31 ++ internal/db/postgres/admission_queue_repo.go | 25 +- internal/db/postgres/deleted_account_repo.go | 17 +- 14 files changed, 813 insertions(+), 146 deletions(-) diff --git a/go.mod b/go.mod index b22139a..f418128 100644 --- a/go.mod +++ b/go.mod @@ -44,16 +44,24 @@ require ( github.com/hashicorp/golang-lru v1.0.2 // indirect github.com/ipfs/bbloom v0.0.4 // indirect github.com/ipfs/go-block-format v0.2.0 // indirect + github.com/ipfs/go-blockservice v0.5.2 // indirect github.com/ipfs/go-cid v0.4.1 // indirect github.com/ipfs/go-datastore v0.6.0 // indirect github.com/ipfs/go-ipfs-blockstore v1.3.1 // indirect github.com/ipfs/go-ipfs-ds-help v1.1.1 // indirect + github.com/ipfs/go-ipfs-exchange-interface v0.2.1 // indirect github.com/ipfs/go-ipfs-util v0.0.3 // indirect github.com/ipfs/go-ipld-cbor v0.1.0 // indirect github.com/ipfs/go-ipld-format v0.6.0 // indirect + github.com/ipfs/go-ipld-legacy v0.2.1 // indirect github.com/ipfs/go-log v1.0.5 // indirect github.com/ipfs/go-log/v2 v2.5.1 // indirect + github.com/ipfs/go-merkledag v0.11.0 // indirect github.com/ipfs/go-metrics-interface v0.0.1 // indirect + github.com/ipfs/go-verifcid v0.0.3 // indirect + github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4 // indirect + github.com/ipld/go-codec-dagpb v1.6.0 // indirect + github.com/ipld/go-ipld-prime v0.21.0 // indirect github.com/jbenet/goprocess v0.1.4 // indirect github.com/klauspost/cpuid/v2 v2.2.7 // indirect github.com/mattn/go-isatty v0.0.20 // indirect diff --git a/go.sum b/go.sum index 4772d55..62fe454 100644 --- a/go.sum +++ b/go.sum @@ -64,6 +64,8 @@ github.com/ipfs/bbloom v0.0.4 h1:Gi+8EGJ2y5qiD5FbsbpX/TMNcJw8gSqr7eyjHa4Fhvs= github.com/ipfs/bbloom v0.0.4/go.mod h1:cS9YprKXpoZ9lT0n/Mw/a6/aFV6DTjTLYHeA+gyqMG0= github.com/ipfs/go-block-format v0.2.0 h1:ZqrkxBA2ICbDRbK8KJs/u0O3dlp6gmAuuXUJNiW1Ycs= github.com/ipfs/go-block-format v0.2.0/go.mod h1:+jpL11nFx5A/SPpsoBn6Bzkra/zaArfSmsknbPMYgzM= +github.com/ipfs/go-blockservice v0.5.2 h1:in9Bc+QcXwd1apOVM7Un9t8tixPKdaHQFdLSUM1Xgk8= +github.com/ipfs/go-blockservice v0.5.2/go.mod h1:VpMblFEqG67A/H2sHKAemeH9vlURVavlysbdUI632yk= github.com/ipfs/go-cid v0.4.1 h1:A/T3qGvxi4kpKWWcPC/PgbvDA2bjVLO7n4UeVwnbs/s= github.com/ipfs/go-cid v0.4.1/go.mod h1:uQHwDeX4c6CtyrFwdqyhpNcxVewur1M7l7fNU7LKwZk= github.com/ipfs/go-datastore v0.6.0 h1:JKyz+Gvz1QEZw0LsX1IBn+JFCJQH4SJVFtM4uWU0Myk= @@ -74,19 +76,33 @@ github.com/ipfs/go-ipfs-blockstore v1.3.1 h1:cEI9ci7V0sRNivqaOr0elDsamxXFxJMMMy7 github.com/ipfs/go-ipfs-blockstore v1.3.1/go.mod h1:KgtZyc9fq+P2xJUiCAzbRdhhqJHvsw8u2Dlqy2MyRTE= github.com/ipfs/go-ipfs-ds-help v1.1.1 h1:B5UJOH52IbcfS56+Ul+sv8jnIV10lbjLF5eOO0C66Nw= github.com/ipfs/go-ipfs-ds-help v1.1.1/go.mod h1:75vrVCkSdSFidJscs8n4W+77AtTpCIAdDGAwjitJMIo= +github.com/ipfs/go-ipfs-exchange-interface v0.2.1 h1:jMzo2VhLKSHbVe+mHNzYgs95n0+t0Q69GQ5WhRDZV/s= +github.com/ipfs/go-ipfs-exchange-interface v0.2.1/go.mod h1:MUsYn6rKbG6CTtsDp+lKJPmVt3ZrCViNyH3rfPGsZ2E= github.com/ipfs/go-ipfs-util v0.0.3 h1:2RFdGez6bu2ZlZdI+rWfIdbQb1KudQp3VGwPtdNCmE0= github.com/ipfs/go-ipfs-util v0.0.3/go.mod h1:LHzG1a0Ig4G+iZ26UUOMjHd+lfM84LZCrn17xAKWBvs= github.com/ipfs/go-ipld-cbor v0.1.0 h1:dx0nS0kILVivGhfWuB6dUpMa/LAwElHPw1yOGYopoYs= github.com/ipfs/go-ipld-cbor v0.1.0/go.mod h1:U2aYlmVrJr2wsUBU67K4KgepApSZddGRDWBYR0H4sCk= github.com/ipfs/go-ipld-format v0.6.0 h1:VEJlA2kQ3LqFSIm5Vu6eIlSxD/Ze90xtc4Meten1F5U= github.com/ipfs/go-ipld-format v0.6.0/go.mod h1:g4QVMTn3marU3qXchwjpKPKgJv+zF+OlaKMyhJ4LHPg= +github.com/ipfs/go-ipld-legacy v0.2.1 h1:mDFtrBpmU7b//LzLSypVrXsD8QxkEWxu5qVxN99/+tk= +github.com/ipfs/go-ipld-legacy v0.2.1/go.mod h1:782MOUghNzMO2DER0FlBR94mllfdCJCkTtDtPM51otM= github.com/ipfs/go-log v1.0.5 h1:2dOuUCB1Z7uoczMWgAyDck5JLb72zHzrMnGnCNNbvY8= github.com/ipfs/go-log v1.0.5/go.mod h1:j0b8ZoR+7+R99LD9jZ6+AJsrzkPbSXbZfGakb5JPtIo= github.com/ipfs/go-log/v2 v2.1.3/go.mod h1:/8d0SH3Su5Ooc31QlL1WysJhvyOTDCjcCZ9Axpmri6g= github.com/ipfs/go-log/v2 v2.5.1 h1:1XdUzF7048prq4aBjDQQ4SL5RxftpRGdXhNRwKSAlcY= github.com/ipfs/go-log/v2 v2.5.1/go.mod h1:prSpmC1Gpllc9UYWxDiZDreBYw7zp4Iqp1kOLU9U5UI= +github.com/ipfs/go-merkledag v0.11.0 h1:DgzwK5hprESOzS4O1t/wi6JDpyVQdvm9Bs59N/jqfBY= +github.com/ipfs/go-merkledag v0.11.0/go.mod h1:Q4f/1ezvBiJV0YCIXvt51W/9/kqJGH4I1LsA7+djsM4= github.com/ipfs/go-metrics-interface v0.0.1 h1:j+cpbjYvu4R8zbleSs36gvB7jR+wsL2fGD6n0jO4kdg= github.com/ipfs/go-metrics-interface v0.0.1/go.mod h1:6s6euYU4zowdslK0GKHmqaIZ3j/b/tL7HTWtJ4VPgWY= +github.com/ipfs/go-verifcid v0.0.3 h1:gmRKccqhWDocCRkC+a59g5QW7uJw5bpX9HWBevXa0zs= +github.com/ipfs/go-verifcid v0.0.3/go.mod h1:gcCtGniVzelKrbk9ooUSX/pM3xlH73fZZJDzQJRvOUw= +github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4 h1:oFo19cBmcP0Cmg3XXbrr0V/c+xU9U1huEZp8+OgBzdI= +github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4/go.mod h1:6nkFF8OmR5wLKBzRKi7/YFJpyYR7+oEn1DX+mMWnlLA= +github.com/ipld/go-codec-dagpb v1.6.0 h1:9nYazfyu9B1p3NAgfVdpRco3Fs2nFC72DqVsMj6rOcc= +github.com/ipld/go-codec-dagpb v1.6.0/go.mod h1:ANzFhfP2uMJxRBr8CE+WQWs5UsNa0pYtmKZ+agnUw9s= +github.com/ipld/go-ipld-prime v0.21.0 h1:n4JmcpOlPDIxBcY037SVfpd1G+Sj1nKZah0m6QH9C2E= +github.com/ipld/go-ipld-prime v0.21.0/go.mod h1:3RLqy//ERg/y5oShXXdx5YIp50cFGOanyMctpPjsvxQ= github.com/jbenet/go-cienv v0.1.0/go.mod h1:TqNnHUmJgXau0nCzC7kXWeotg3J9W34CUv5Djy1+FlA= github.com/jbenet/goprocess v0.1.4 h1:DRGOFReOMqqDNXwW70QkacFW0YN9QnwLV0Vqk+3oU0o= github.com/jbenet/goprocess v0.1.4/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZlqdZVfqY4= diff --git a/internal/api/handlers/post/getstatus.go b/internal/api/handlers/post/getstatus.go index 49c6545..a8ba7b1 100644 --- a/internal/api/handlers/post/getstatus.go +++ b/internal/api/handlers/post/getstatus.go @@ -7,6 +7,30 @@ import ( "time" "Coves/internal/core/posts" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +// The identifier bounds this endpoint accepts, DERIVED from the specs rather +// than picked, because both directions of getting it wrong are silent. +// +// A postv2 URI is AUTHORITY-SCOPED — the author's DID is inside it — and a DID +// may legally run to 2048 bytes, the same fact that killed the readable rkey +// transform in PRD rev 2.2. A cap sized for the old community-repo URIs would +// refuse exactly the authors with long DIDs, on the one endpoint that can tell +// them why their post is not visible, and nothing else in the system would show +// a symptom: their posts index fine and every other endpoint serves them. +// +// So the URI bound is the sum of its legal parts rather than a round number: +// "at://" + a 2048-byte authority + "/" + a 317-byte NSID + "/" + a 512-byte +// record key. It is a pre-parse DoS bound only — ParseATURI below is what +// decides whether the thing is actually a URI. +const ( + maxAuthorityLength = 2048 // DID Core's ceiling + maxNSIDLength = 317 // atProto NSID grammar + maxRecordKeyLength = 512 // atProto record-key grammar + maxPostURILength = len("at://") + maxAuthorityLength + 1 + maxNSIDLength + 1 + maxRecordKeyLength + maxCommunityIDLength = maxAuthorityLength ) // GetStatusHandler serves social.coves.community.post.getStatus: one @@ -53,15 +77,32 @@ func (h *GetStatusHandler) HandleGetStatus(w http.ResponseWriter, r *http.Reques writeError(w, http.StatusBadRequest, "InvalidRequest", "community parameter is required") return } - if len(postURI) > maxURILength { + // THE CAP IS SIZED TO THE SPEC, NOT TO THE OLD URIs. A DID may legally run + // to 2048 bytes — the same fact that killed the readable rkey transform in + // PRD rev 2.2 — and an author-owned post URI is authority-scoped, so the + // author's DID is INSIDE the URI this endpoint takes. A cap sized for the + // old community-repo URIs would silently make long-DID authors unqueryable + // and nothing else would notice: their posts index fine and every other + // endpoint serves them. + if len(postURI) > maxPostURILength { writeError(w, http.StatusBadRequest, "InvalidRequest", "post URI exceeds maximum length") return } - if len(communityDID) > maxURILength { + if len(communityDID) > maxCommunityIDLength { writeError(w, http.StatusBadRequest, "InvalidRequest", "community DID exceeds maximum length") return } + // Raising the cap must not become "accept anything long". The URI is PARSED, + // so a non-at:// string comes back as the client bug it is rather than as a + // silent not-found — which would tell a client with a typo that its post + // does not exist. + if _, err := syntax.ParseATURI(postURI); err != nil { + writeError(w, http.StatusBadRequest, "InvalidRequest", + "post must be a valid AT-URI") + return + } + status, err := h.service.GetStatus(r.Context(), posts.GetStatusRequest{ PostURI: postURI, CommunityDID: communityDID, @@ -96,6 +137,13 @@ func (h *GetStatusHandler) HandleGetStatus(w http.ResponseWriter, r *http.Reques } w.Header().Set("Content-Type", "application/json") + // NO-STORE, and not as a nicety. This endpoint exists to be POLLED for a + // transition (§7), so a cached answer is the endpoint failing at its only + // job: the client keeps being handed `pending` after the post was accepted + // and stops asking. It is also unauthenticated and reports a moderation + // decision, so an intermediary holding a copy is a disclosure surface that + // outlives the request that created it. + w.Header().Set("Cache-Control", "no-store") w.WriteHeader(http.StatusOK) if _, err := w.Write(responseBytes); err != nil { log.Printf("ERROR: Failed to write getStatus response: %v", err) diff --git a/internal/api/routes/post.go b/internal/api/routes/post.go index 67dfd3b..dde40ae 100644 --- a/internal/api/routes/post.go +++ b/internal/api/routes/post.go @@ -6,10 +6,20 @@ import ( "Coves/internal/core/blueskypost" "Coves/internal/core/posts" "Coves/internal/core/votes" + "time" "github.com/go-chi/chi/v5" ) +// getStatusRateLimit is social.coves.community.post.getStatus' per-client +// budget, per minute. +// +// Sixty is chosen against the endpoint's own UX rather than copied from a +// neighbour: §7 has a client poll for the accepted transition, and a poll a +// second for a minute is comfortably inside this while a script enumerating a +// community's rejected posts is not. +const getStatusRateLimit = 60 + // PostRouteOption supplies a collaborator that only some of the post routes // need. // @@ -26,12 +36,24 @@ type postRouteConfig struct { // WithPostStatusService supplies the service behind // social.coves.community.post.getStatus. The route is registered either way, so -// that the HTTP surface does not silently change shape with the wiring; without -// this option the handler has no service to call. +// that the HTTP surface does not silently change shape with the wiring. func WithPostStatusService(service posts.StatusService) PostRouteOption { return func(c *postRouteConfig) { c.statusService = service } } +// NOT GUARDED AT REGISTRATION, unlike oauthMiddleware above, and the asymmetry +// is forced rather than chosen. The review asked for a fail-fast panic on a +// missing status service, matching that neighbour — but the routes table builds +// the whole router with every service nil and no options at all +// (registration_test.go's theRouter), so a panic there would fail every +// surface-declaration test rather than the one wiring bug it is aimed at. +// Scoping it to a supplied-but-nil option would guard a shape nobody writes. +// +// The exposure is small and named here so it is not mistaken for an oversight: +// cmd/server always supplies the option, and a build that did not would serve +// 500s from getStatus alone. Closing it properly needs the routes table to pass +// the option, which is a test-side change. + // RegisterPostRoutes registers post-related XRPC endpoints on the router // Implements social.coves.community.post.* lexicon endpoints // authMiddleware can be either OAuthAuthMiddleware or DualAuthMiddleware (used for @@ -87,8 +109,18 @@ func RegisterPostRoutes( // be harmless but pointless — the answer does not vary by viewer — while // RequireAuth would make the cross-server case unanswerable, which is the // asymmetry internal/api/routes/registration_test.go declares. + // + // It also carries its OWN limiter, tighter than the global 100/minute, and + // it is the only unauthenticated route in the product that does. Two facts + // make the exception worth it: §7's client UX is to POLL this until a post + // flips to accepted, so the honest traffic shape is repeated requests from + // one caller; and because it takes no auth, an unauthenticated stranger can + // ask about any post URI they can name. The budget is what bounds + // enumeration of a community's rejected posts to a rate an operator notices. statusHandler := post.NewGetStatusHandler(cfg.statusService) - r.Get("/xrpc/social.coves.community.post.getStatus", statusHandler.HandleGetStatus) + statusRateLimiter := middleware.NewRateLimiter(getStatusRateLimit, time.Minute) + r.With(statusRateLimiter.Middleware). + Get("/xrpc/social.coves.community.post.getStatus", statusHandler.HandleGetStatus) // Future endpoints (Beta): // r.With(authMiddleware.RequireAuth).Post("/xrpc/social.coves.community.post.update", updateHandler.HandleUpdate) diff --git a/internal/atproto/jetstream/authorpost.go b/internal/atproto/jetstream/authorpost.go index 84aaef0..40f5a8b 100644 --- a/internal/atproto/jetstream/authorpost.go +++ b/internal/atproto/jetstream/authorpost.go @@ -1,6 +1,7 @@ package jetstream import ( + "bytes" "context" "database/sql" "encoding/json" @@ -12,6 +13,7 @@ import ( "net/url" "strconv" "strings" + "sync" "time" "Coves/internal/atproto/identity" @@ -19,6 +21,9 @@ import ( "Coves/internal/core/communities" "Coves/internal/core/posts" "Coves/internal/core/users" + + "github.com/bluesky-social/indigo/atproto/atdata" + "github.com/bluesky-social/indigo/repo" ) // Ingesting author-owned posts and the community records that decide about @@ -91,7 +96,7 @@ func WithPostRecordFetcher(fetcher PostRecordFetcher) PostEventConsumerOption { // AcceptanceDeleter withdraws a community's acceptance of a post. Satisfied by // posts.CommunityRecordWriter. // -// RED STUB (task 5, cycle 2). Narrowed to one method because that is all the +// Narrowed to one method because that is all the // tombstone path needs: the consumer must never write an acceptance, a removal // or a repin — those are the ENGINE's verdicts, and a consumer holding the full // writer is one edit away from making one. @@ -157,8 +162,23 @@ type DirectPostFetcher struct { // a request forger pointed at whatever is reachable from the AppView's // network, driven by any stranger who writes an acceptance record. allowPrivateHosts bool + + // client is the guarded HTTP client, built once. See httpClient. + clientOnce sync.Once + client *http.Client } +// fetchTimeout bounds ONE direct fetch, inside the client's own 15-second +// ceiling. +// +// The two are not redundant. The client's timeout is a backstop for a +// connection that hangs; this one is a policy about how long the posts lane may +// wait on a stranger's PDS. The lane is single-threaded and now carries four +// collections, so every second spent here is a second nothing else is indexed — +// and the destination is chosen by whoever wrote the acceptance. Five seconds +// is generous for a repo read and cheap to lose. +const fetchTimeout = 5 * time.Second + // maxFetchedRecordBytes bounds how much of a PDS getRecord response is read. // // A post record has a lexicon-bounded size, so a PDS streaming megabytes is @@ -191,11 +211,24 @@ func NewDevDirectPostFetcher(resolver identity.Resolver) *DirectPostFetcher { return &DirectPostFetcher{resolver: resolver, allowPrivateHosts: true} } -// httpClient builds the guarded client for one fetch. Declared here so the -// guard is derived from allowPrivateHosts at call time rather than baked into a -// client at construction, where a test seam could not reach it. +// httpClient returns the guarded client, building it once on first use. +// +// ONE CLIENT, not one per fetch. A fresh http.Client means a fresh +// http.Transport, and a fresh transport means an empty connection pool: every +// fetch would pay a new TCP handshake and a new TLS handshake against a PDS +// this consumer may be about to fetch from a hundred more times, and the +// discarded transports leak idle connections until their finalizers run. The +// guard is a property of the transport rather than of the moment, so nothing +// about correctness needed it rebuilt. +// +// It is built lazily rather than in the constructor so that a fetcher which is +// wired but never used — the common case on an instance hosting no communities +// — costs nothing. func (f *DirectPostFetcher) httpClient() *http.Client { - return oauth.NewSSRFSafeHTTPClient(f.allowPrivateHosts) + f.clientOnce.Do(func() { + f.client = oauth.NewSSRFSafeHTTPClient(f.allowPrivateHosts) + }) + return f.client } // FetchPost implements PostRecordFetcher. @@ -220,11 +253,29 @@ func (f *DirectPostFetcher) FetchPost(ctx context.Context, postURI string) (*Fet return nil, fmt.Errorf("resolving the repo of %s: no PDS endpoint in the DID document", postURI) } - endpoint := strings.TrimSuffix(resolved.PDSURL, "/") + "/xrpc/com.atproto.repo.getRecord?repo=" + + // com.atproto.sync.getRecord, NOT repo.getRecord, and the difference is the + // whole trustworthiness of this path. + // + // repo.getRecord answers with JSON — {"uri":…, "cid":…, "value":{…}} — whose + // `cid` is a CLAIM BY THE SERVER. That server is the author's PDS: named by + // a DID document, reached because a stranger wrote an acceptance naming this + // subject. Comparing the pinned CID against that field asks the attacker + // whether the attacker is lying, and the consequence is the worst one this + // design has — the AppView indexes whatever `value` holds under a + // community's SIGNED acceptance of a CID that content does not have. The + // community attested to one thing and every reader is shown another. + // + // sync.getRecord answers with a CAR: the repo's own blocks. The CID is then + // RECOMPUTED from the bytes rather than read off a label, and no server can + // lie about the hash of what it just sent. + endpoint := strings.TrimSuffix(resolved.PDSURL, "/") + "/xrpc/com.atproto.sync.getRecord?did=" + url.QueryEscape(repoDID) + "&collection=" + url.QueryEscape(collection) + "&rkey=" + url.QueryEscape(rkey) - req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + fetchCtx, cancel := context.WithTimeout(ctx, fetchTimeout) + defer cancel() + + req, err := http.NewRequestWithContext(fetchCtx, http.MethodGet, endpoint, nil) if err != nil { return nil, fmt.Errorf("building the getRecord request for %s: %w", postURI, err) } @@ -284,25 +335,46 @@ func (f *DirectPostFetcher) FetchPost(ctx context.Context, postURI string) (*Fet postURI, resp.StatusCode, strconv.Quote(detail)) } - var parsed struct { - URI string `json:"uri"` - CID string `json:"cid"` - Value map[string]interface{} `json:"value"` + // The CAR carries the record's block plus the blocks proving it belongs to + // the repo. Reading it can fail on a hostile or broken server, which is a + // refusal like any other — a response that is not a CAR is not evidence. + stored, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("reading the CAR for %s: %w", postURI, err) } - if err := json.Unmarshal(body, &parsed); err != nil { - return nil, fmt.Errorf("parsing the getRecord response for %s: %w", postURI, err) + + claimedCID, recordBytes, err := stored.GetRecordBytes(ctx, collection+"/"+rkey) + if err != nil { + return nil, fmt.Errorf("reading %s out of the fetched CAR: %w", postURI, err) } - if parsed.Value == nil { - return nil, fmt.Errorf("the getRecord response for %s carried no record value", postURI) + if recordBytes == nil || len(*recordBytes) == 0 { + return nil, fmt.Errorf("the CAR for %s carried no record bytes", postURI) } - if parsed.CID == "" { - // Without a CID there is nothing to verify the pinned reference - // against, and an unverified record is exactly what the fetch must - // never index. - return nil, fmt.Errorf("the getRecord response for %s carried no CID", postURI) + + // THE RECOMPUTATION, which is the entire point of taking a CAR at all. The + // CID that came out of the repo structure is still something the server + // assembled; hashing the record's own bytes under that CID's codec and + // multihash is what turns it into a fact. A server that substituted content + // produces a digest that does not match, whatever it labelled the block. + computedCID, err := claimedCID.Prefix().Sum(*recordBytes) + if err != nil { + return nil, fmt.Errorf("recomputing the CID of %s: %w", postURI, err) + } + if !computedCID.Equals(claimedCID) { + return nil, fmt.Errorf("%w: the CAR for %s labels a block %s whose bytes hash to %s", + ErrPermanentEvent, postURI, claimedCID, computedCID) } - return &FetchedPost{URI: postURI, CID: parsed.CID, Record: parsed.Value}, nil + record, err := atdata.UnmarshalCBOR(*recordBytes) + if err != nil { + return nil, fmt.Errorf("decoding the record block of %s: %w", postURI, err) + } + + // The CID reported back is the COMPUTED one. The caller compares it against + // what the acceptance pinned, and handing back the label instead would put + // the server's claim back into the comparison the recomputation just removed + // it from. + return &FetchedPost{URI: postURI, CID: computedCID.String(), Record: record}, nil } // --------------------------------------------------------------------------- @@ -424,14 +496,28 @@ func (c *PostEventConsumer) tombstoneAuthorPost(ctx context.Context, authorDID s if err != nil { return err } - if !applied || !indexed { - // A gate skip means this deletion was already applied — the sweep ran - // with it — so re-sweeping would put one authenticated PDS round trip - // behind every redelivery of every tombstone on the network. + if !indexed { return nil } - c.withdrawAcceptance(ctx, stored.communityDID, uri) + // THE WITHDRAWAL IS RECONSIDERED ON EVERY DELIVERY, unlike the tombstone. + // The gate exists to make the local soft-delete happen once; the withdrawal + // is a write into a REMOTE repo that can fail on its own, and it is + // idempotent — DeleteAcceptance reports "nothing to withdraw" as a skip. + // + // Tying it to the gate is what stranded it: a PDS briefly unreachable on the + // first delivery meant the acceptance stayed standing, pointing at a record + // nobody can fetch, permanently — because every redelivery was rejected by + // the gate before the sweep was reconsidered, and nothing else revisits it. + // The community's CAR, the thing its portability argument rests on, would + // keep citing content the author withdrew. + // + // The guard is the POST's state rather than this event's: sweep when the row + // is tombstoned, which is true on the delivery that applied it and on every + // one after. + if applied || stored.deletedAt != nil { + c.withdrawAcceptance(ctx, stored.communityDID, uri) + } return nil } @@ -547,6 +633,37 @@ func (c *PostEventConsumer) upsertAuthorPost(ctx context.Context, authorDID stri uri := recordURI(authorDID, PostV2Collection, commit.RKey) + stored, found, err := c.loadStoredPost(ctx, uri) + if err != nil { + return err + } + + // IMMUTABILITY (§3.1) IS CHECKED FIRST, BEFORE THE COMMUNITY IS LOOKED UP, + // and the order is load-bearing rather than tidy. + // + // An update that changes `community` invalidates the WHOLE event — not + // merely the community field, because applying the content while keeping the + // old community would leave the first community's admission holding a CID it + // never evaluated, publishing content nobody judged under a standing + // acceptance. Retargeting a post means writing a new record. + // + // Two rules meet when the retarget names a community nobody has indexed, and + // only one of them can go first. The unknown-community branch below is + // TRANSIENT — correctly, since a community's own profile event may simply + // not have arrived — so checking it first would turn an illegal retarget + // into a retryable failure that can NEVER succeed: it dead-letters, redrives + // ten times, blocks ~4.2s inline on each delivery, and is still an illegal + // retarget once that community exists. An author could mint that load at + // will by editing one field. + // + // A skip, not an error: an invalid record from a stranger's repo is not an + // infrastructure failure. + if found && stored.communityDID != record.Community { + log.Printf("🚨 SECURITY: ignoring the whole %s update for %s - community is immutable (stored %s, incoming %s)", + PostV2Collection, uri, stored.communityDID, record.Community) + return nil + } + // The community must be one this AppView has indexed, or there is no // subject to open an admission against. // @@ -563,26 +680,6 @@ func (c *PostEventConsumer) upsertAuthorPost(ctx context.Context, authorDID stri return fmt.Errorf("%w: failed to verify community %s exists: %v", errValidationInfra, record.Community, err) } - stored, found, err := c.loadStoredPost(ctx, uri) - if err != nil { - return err - } - - // IMMUTABILITY (§3.1): an update that changes `community` invalidates the - // WHOLE event. Not merely the community field — applying the content while - // keeping the old community would leave the first community's admission - // holding a CID it never evaluated, publishing content nobody judged under - // a standing acceptance. Retargeting a post means writing a new record. - // - // A skip, not an error: an invalid record from a stranger's repo is not an - // infrastructure failure, and dead-lettering it would retry a record that - // can never become valid. - if found && stored.communityDID != record.Community { - log.Printf("🚨 SECURITY: ignoring the whole %s update for %s - community is immutable (stored %s, incoming %s)", - PostV2Collection, uri, stored.communityDID, record.Community) - return nil - } - // Provenance for bridgedStats is keyed on the AUTHOR's PDS now, because the // record lives in the author's repo. The community's host has no say over // what an author asserts about their own record any more, so checking the @@ -622,12 +719,22 @@ func (c *PostEventConsumer) upsertAuthorPost(ctx context.Context, authorDID stri } } - if !applied { - // The rev gate or the recency guard refused this event: a newer state is - // already indexed. Opening or refreshing an admission from it would move - // evaluated_cid BACKWARDS onto content the row no longer holds, which is - // how an accepted post gets flipped to pending_reacceptance by a - // duplicate delivery. + // THE GATE GUARDS THE POST ROW, NOT THE ADMISSION. The two writes have + // opposite idempotence: inserting the post must happen exactly once, while + // UpsertPending is content-addressed and writes nothing when the row already + // holds this CID. Gating them together is what orphaned admissions — a + // failed upsert leaves the post indexed, the gate advanced, and every + // redelivery skipped, so the row is never created and the post is invisible + // in its community forever with nothing left to retry. + // + // A REPLAY IS SAFE, A STALE COPY IS NOT, and the stored rev is what tells + // them apart. An equal rev is this same commit arriving again — the upsert + // re-runs harmlessly and repairs the orphan. A strictly greater stored rev + // means a NEWER event already applied, and re-running the upsert from this + // older one would drag evaluated_cid backwards onto content the row no + // longer holds, flipping an accepted post to pending_reacceptance on a + // duplicate delivery. + if !applied && !c.revIsCurrent(ctx, uri, commit.Rev) { return nil } @@ -661,6 +768,31 @@ func (c *PostEventConsumer) upsertAuthorPost(ctx context.Context, authorDID stri return nil } +// revIsCurrent reports whether the gate's stored rev for this record is exactly +// the one this event carries — i.e. whether the event is the newest the AppView +// has seen rather than an older copy. +// +// It exists so a rev-gated SKIP can still be distinguished into its two very +// different causes. A redelivery of the newest commit is safe to act on again; +// a stale cross-feed copy of an older one is not. Anything unreadable answers +// false, which declines to act — the conservative direction, since acting on a +// stale event corrupts state while declining merely waits for the next +// delivery. +func (c *PostEventConsumer) revIsCurrent(ctx context.Context, uri, rev string) bool { + if rev == "" { + // A rev-less event bypasses the gate entirely, so there is no stored rev + // to be current with. Only synthetic events reach this. + return true + } + var stored string + if err := c.db.QueryRowContext(ctx, + `SELECT rev FROM jetstream_record_revs WHERE record_uri = $1`, uri, + ).Scan(&stored); err != nil { + return false + } + return stored == rev +} + // hydrateAuthorOpportunistically indexes a minimal profile for an author this // AppView has not seen, so their posts are not permanently authorless. // @@ -916,10 +1048,28 @@ func (c *PostEventConsumer) applyAcceptance( subjectCollection string, watermark posts.CommunityWatermark, ) error { - indexedCommunity, indexed, err := c.indexedPostCommunity(ctx, decision.Subject.URI) + stored, indexed, err := c.loadStoredPost(ctx, decision.Subject.URI) if err != nil { return err } + + // A TOMBSTONED SUBJECT IS NOT ACCEPTABLE, and the arrival is legitimate + // rather than hostile: the community decided before it saw the tombstone, + // and the two events are in different repos with no ordering between them. + // Applying it would have getStatus report `accepted` for content no read + // path will ever serve — and the host-side sweep already ran with the + // tombstone and will not run again, so the community's repo would keep an + // acceptance citing a record nobody can fetch. + // + // A SKIP, not a refusal. Returning an error would dead-letter an event that + // is replayed constantly and would be refused identically every time. + if indexed && stored.deletedAt != nil { + log.Printf("INFO: not applying the acceptance of %s in %s: its author deleted the post", + decision.Subject.URI, communityDID) + return nil + } + + indexedCommunity := stored.communityDID switch { case !indexed: if err := c.convergeOnAcceptedSubject(ctx, communityDID, decision, subjectCollection); err != nil { @@ -1033,7 +1183,7 @@ func (c *PostEventConsumer) convergeOnAcceptedSubject( // event time to stamp: an empty rev bypasses the gate (which is correct — a // later real event for this record still wins on its own rev) and the // watermark falls back to wall clock. - if _, err := c.insertAuthorPost(ctx, authorPostInsert{ + applied, err := c.insertAuthorPost(ctx, authorPostInsert{ uri: decision.Subject.URI, authorDID: authorDID, record: record, @@ -1043,19 +1193,38 @@ func (c *PostEventConsumer) convergeOnAcceptedSubject( }, facets: facetsJSON, embed: embedJSON, labels: labelsJSON, bridgedUpvotes: up, bridgedDownvotes: down, bridgedAsOf: asOf, - }); err != nil { + }) + if err != nil { return err } - // The pending admission comes with it. ApplyAcceptance would create a row - // on its own, but one with no evaluated content: recording what was indexed - // is what lets the next author edit be recognised as an edit. - if _, err := c.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ - CommunityDID: communityDID, - PostURI: decision.Subject.URI, - EvaluatedCID: fetched.CID, - }); err != nil { - return fmt.Errorf("recording the pending admission for fetched post %s: %w", decision.Subject.URI, err) + // THE FETCH IS A CATCH-UP, NOT AN AUTHORITY, and applied=false is how it + // learns it lost the race. The acceptance and the post live in different + // repos, so Jetstream parallelises them and the post's own event can land + // while this fetch is in flight — which is not a rare interleaving but the + // normal one, since the fetch exists precisely because the event had not + // arrived. The insert then conflicts and writes nothing. + // + // UpsertPending is last-write-wins, so running it anyway would stamp the + // FETCHED CID over the newer one the real event just recorded. evaluated_cid + // is what the next decision judges, so the engine would evaluate content the + // author has already replaced, and an acceptance written from that verdict + // would pin a version the AppView is no longer serving. + // + // Skipping it costs nothing: ApplyAcceptance below classifies against + // whatever evaluated_cid the row actually holds, which is exactly the + // comparison that turns a stale pin into pending_reacceptance. + if applied { + // The pending admission comes with it. ApplyAcceptance would create a row + // on its own, but one with no evaluated content: recording what was indexed + // is what lets the next author edit be recognised as an edit. + if _, err := c.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: communityDID, + PostURI: decision.Subject.URI, + EvaluatedCID: fetched.CID, + }); err != nil { + return fmt.Errorf("recording the pending admission for fetched post %s: %w", decision.Subject.URI, err) + } } c.hydrateAuthorOpportunistically(ctx, authorDID) @@ -1111,48 +1280,121 @@ func (c *PostEventConsumer) applyRemoval( // (§5.3), is the one delete that arrives unpaired — and it is the one this // lookup resolves. func (c *PostEventConsumer) applyCommunityDecisionDelete(ctx context.Context, communityDID string, commit *CommitEvent) error { - if commit.Collection != posts.AcceptanceCollection { - log.Printf("INFO: %s deletion in %s carries no subject and is superseded by its paired write; skipping", - commit.Collection, communityDID) - return nil + postURI, found, err := c.subjectOfDeletedRecord(ctx, communityDID, commit) + if err != nil { + return err } - - var postURI string - err := c.db.QueryRowContext(ctx, - `SELECT post_uri FROM community_post_admissions - WHERE community_did = $1 AND acceptance_rkey = $2`, - communityDID, commit.RKey, - ).Scan(&postURI) - if errors.Is(err, sql.ErrNoRows) { - // No acceptance of that rkey stands here — the removal half of the same - // commit already cleared it, or this AppView never saw the acceptance. - // Either way there is nothing to withdraw. - log.Printf("INFO: acceptance deletion %s/%s matches no standing acceptance; nothing to withdraw", - communityDID, commit.RKey) + if !found { + // Nothing here matches that record key: the paired write of the same + // commit already superseded it, or this AppView never saw the record + // being deleted. Either way there is nothing to withdraw. + log.Printf("INFO: %s deletion %s/%s matches no standing record; nothing to withdraw", + commit.Collection, communityDID, commit.RKey) return nil } - if err != nil { - return fmt.Errorf("resolving the subject of acceptance deletion %s/%s: %w", communityDID, commit.RKey, err) - } - result, err := c.admissions.ApplyAcceptanceDelete(ctx, posts.CommunityDeleteCommand{ + cmd := posts.CommunityDeleteCommand{ CommunityDID: communityDID, PostURI: postURI, Watermark: posts.CommunityWatermark{Rev: commit.Rev}, - }) + } + + var result posts.AdmissionResult + if commit.Collection == posts.RemovalCollection { + result, err = c.admissions.ApplyRemovalDelete(ctx, cmd) + } else { + result, err = c.admissions.ApplyAcceptanceDelete(ctx, cmd) + } if err != nil { - return fmt.Errorf("withdrawing the acceptance of %s in %s: %w", postURI, communityDID, err) + return fmt.Errorf("withdrawing the %s of %s in %s: %w", commit.Collection, postURI, communityDID, err) } - logAdmissionOutcome(posts.AcceptanceCollection+"#delete", communityDID, postURI, result.Outcome) + logAdmissionOutcome(commit.Collection+"#delete", communityDID, postURI, result.Outcome) return nil } +// subjectOfDeletedRecord recovers which post a deleted acceptance or removal was +// about. +// +// A delete event carries NO record, and the rkey is a SHA-256 digest of the +// subject URI (§3.2), which is one-way — so the subject can only come from state +// the AppView already holds. The two collections need different lookups because +// only one of them has a column: +// +// - An ACCEPTANCE stores its rkey on the admission row, so the reverse lookup +// is an exact match. +// - A REMOVAL stores none, so its subject is found by recomputing the digest +// over the rows that could be its subject. The candidate set is bounded to +// this community's `removed` rows — moderation-sized, and served by the +// (community_did, status, created_at) index migration 034 already carries. +// +// WHY THE REMOVAL CASE CANNOT STAY A NO-OP, which is what it was. The paired +// commits do converge without it: a removal commit is {acceptance-delete, +// removal-put} and a restore is {removal-delete, acceptance-put}, and in both +// the put carries the subject in-record and outranks its delete under the §5.2 +// tuple. But a moderator can also simply WITHDRAW a removal — deleting the +// record and writing nothing — and that commit carries only this event. Ignored, +// the post stays `removed` forever while the community's own repo no longer says +// so: the signed record and the AppView disagree, and only the AppView is +// consulted when the post is served. +func (c *PostEventConsumer) subjectOfDeletedRecord( + ctx context.Context, communityDID string, commit *CommitEvent, +) (string, bool, error) { + if commit.Collection == posts.AcceptanceCollection { + var postURI string + err := c.db.QueryRowContext(ctx, + `SELECT post_uri FROM community_post_admissions + WHERE community_did = $1 AND acceptance_rkey = $2`, + communityDID, commit.RKey, + ).Scan(&postURI) + if errors.Is(err, sql.ErrNoRows) { + return "", false, nil + } + if err != nil { + return "", false, fmt.Errorf("resolving the subject of acceptance deletion %s/%s: %w", + communityDID, commit.RKey, err) + } + return postURI, true, nil + } + + rows, err := c.db.QueryContext(ctx, + `SELECT post_uri FROM community_post_admissions + WHERE community_did = $1 AND status = 'removed'`, communityDID) + if err != nil { + return "", false, fmt.Errorf("resolving the subject of removal deletion %s/%s: %w", + communityDID, commit.RKey, err) + } + defer func() { _ = rows.Close() }() + + for rows.Next() { + var postURI string + if err := rows.Scan(&postURI); err != nil { + return "", false, fmt.Errorf("scanning a removed subject of %s: %w", communityDID, err) + } + if posts.SubjectRkey(postURI) == commit.RKey { + return postURI, true, nil + } + } + if err := rows.Err(); err != nil { + return "", false, fmt.Errorf("reading the removed subjects of %s: %w", communityDID, err) + } + return "", false, nil +} + // logAdmissionOutcome records what a community event DID, including the skips. // // A skip is the ordering gate working — a multi-feed duplicate, a dead-letter -// redrive, an event superseded by its own commit's other half — so it is logged -// rather than returned as an error, which would bury healthy skips in the -// dead-letter queue among genuine failures. +// redrive, an event superseded by its own commit's other half, or this +// AppView's own write coming back to it — so it is logged rather than returned +// as an error, which would bury healthy skips in the dead-letter queue among +// genuine failures. +// +// THE LAST OF THOSE IS THE COMMON ONE ON A HOSTING INSTANCE and is easy to +// misread in the logs. When this AppView hosts the community, the engine writes +// the acceptance into the repo and stamps the row optimistically; the firehose +// then delivers that same commit back, and the watermark CAS answers +// skipped_stale because the row already carries that exact rev. Nothing is +// wrong — the write landed twice by design, once locally and once as its own +// echo — and the engine's own doc calls that echo a success. func logAdmissionOutcome(collection, communityDID, postURI string, outcome posts.AdmissionOutcome) { if outcome == posts.AdmissionApplied { log.Printf("✓ Applied %s for %s in %s", collection, postURI, communityDID) diff --git a/internal/atproto/jetstream/post_consumer.go b/internal/atproto/jetstream/post_consumer.go index 5ee6dbb..fc02bf8 100644 --- a/internal/atproto/jetstream/post_consumer.go +++ b/internal/atproto/jetstream/post_consumer.go @@ -30,12 +30,14 @@ type PostEventConsumer struct { // passes bridgeTrust. identityResolver identity.Resolver - // RED STUB fields (task 5, cycle 1) — the collaborators author-owned post - // ingestion needs. Declared here so the options in authorpost.go compile; - // nothing reads them yet. See docs/PRD_AUTHOR_OWNED_POSTS.md §5.3-§5.6. + // The collaborators author-owned post ingestion needs + // (docs/PRD_AUTHOR_OWNED_POSTS.md §5.3-§5.6). All four are read by the + // handlers in authorpost.go, and a nil one disables a capability rather + // than degrading it — see each field. // // admissions holds the per-(community, post) decision state. nil means the - // consumer is running in its pre-034 shape and records no admissions. + // consumer is running in its pre-034 shape and ignores all three + // author-owned collections rather than indexing them undecided. admissions posts.AdmissionRepository // deletedAccounts gates events from erased accounts. nil means no gate. deletedAccounts DeletedAccountLookup diff --git a/internal/atproto/oauth/transport.go b/internal/atproto/oauth/transport.go index 9e7c463..21ab957 100644 --- a/internal/atproto/oauth/transport.go +++ b/internal/atproto/oauth/transport.go @@ -1,6 +1,7 @@ package oauth import ( + "context" "fmt" "net" "net/http" @@ -62,14 +63,43 @@ func isPrivateIP(ip net.IP) bool { return false } +// vettedAddrsKeyType keys the addresses RoundTrip approved, so the dialler can +// read them off the request's own context. A private type, so nothing outside +// this file can plant a value under the same key. +type vettedAddrsKeyType struct{} + +var vettedAddrsKey vettedAddrsKeyType + +// RoundTrip vets the hostname's addresses and then makes the dial use THOSE +// ADDRESSES rather than the name. +// +// PASSING THE NAME ON WOULD BE A SECOND DECISION. The base transport resolves +// whatever host it is given, so a guard that approved answer A and then handed +// over the hostname lets the dialler act on answer B — and nothing binds the two +// together. DNS rebinding is the name for exploiting that: an attacker who +// controls the zone answers the first query publicly and the second with +// 169.254.169.254, and the approval describes a host the connection never went +// to. Every input that reaches this transport is chosen by a stranger (a DID +// document's PDS endpoint, an acceptance record's subject), so the attacker +// picks the moment to flip as well. +// +// The addresses ride on the request context and the dialler below consumes +// them, which also means the hostname is resolved exactly ONCE per request. +// That is the property, not an optimisation: any later answer is one the guard +// never saw. func (t *ssrfSafeTransport) RoundTrip(req *http.Request) (*http.Response, error) { host := req.URL.Hostname() - // Resolve hostname to IP + // A literal address is already the thing that will be dialled, so there is + // no second resolution to defend against — but it still has to pass the + // private check below. ips, err := t.resolveHost(host) if err != nil { return nil, fmt.Errorf("failed to resolve host: %w", err) } + if len(ips) == 0 { + return nil, fmt.Errorf("failed to resolve host: %s resolved to no addresses", host) + } // Check all resolved IPs if !t.allowPrivate { @@ -80,17 +110,53 @@ func (t *ssrfSafeTransport) RoundTrip(req *http.Request) (*http.Response, error) } } - return t.base.RoundTrip(req) + return t.base.RoundTrip(req.WithContext(context.WithValue(req.Context(), vettedAddrsKey, ips))) } // NewSSRFSafeHTTPClient creates an HTTP client with SSRF protections func NewSSRFSafeHTTPClient(allowPrivate bool) *http.Client { + dialer := &net.Dialer{ + Timeout: 10 * time.Second, + KeepAlive: 30 * time.Second, + } + transport := &ssrfSafeTransport{ base: &http.Transport{ - DialContext: (&net.Dialer{ - Timeout: 10 * time.Second, - KeepAlive: 30 * time.Second, - }).DialContext, + // The dial IGNORES the hostname in addr and connects to an address + // RoundTrip already approved, which is what closes the + // check-then-dial window. It takes only the port from addr, because + // the port is the one part of the destination the guard has no + // opinion about. + // + // FAIL CLOSED when there is nothing vetted: reaching here without a + // context value means this base transport was used directly rather + // than through RoundTrip, which is exactly the unguarded path the + // wrapper exists to prevent. + // + // TLS is unaffected. http.Transport derives the handshake's server + // name from the request URL rather than from the dial address, so + // certificate verification and SNI still name the host the caller + // asked for. + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + vetted, _ := ctx.Value(vettedAddrsKey).([]net.IP) + if len(vetted) == 0 { + return nil, fmt.Errorf("SSRF blocked: refusing to dial %s with no vetted address "+ + "(the SSRF-safe transport was bypassed)", addr) + } + _, port, err := net.SplitHostPort(addr) + if err != nil { + return nil, fmt.Errorf("SSRF blocked: cannot read a port from %q: %w", addr, err) + } + var lastErr error + for _, ip := range vetted { + conn, dialErr := dialer.DialContext(ctx, network, net.JoinHostPort(ip.String(), port)) + if dialErr == nil { + return conn, nil + } + lastErr = dialErr + } + return nil, lastErr + }, MaxIdleConns: 100, IdleConnTimeout: 90 * time.Second, TLSHandshakeTimeout: 10 * time.Second, diff --git a/internal/core/posts/decider.go b/internal/core/posts/decider.go index 432834c..cda3b16 100644 --- a/internal/core/posts/decider.go +++ b/internal/core/posts/decider.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "log" "os" "strings" "time" @@ -29,9 +28,11 @@ import ( // - THE ACTOR CLASS. A misclassification is a privilege decision. Trusted // aggregators skip visibility, ban and authorization entirely, so guessing // UPWARD on a failed lookup would hand the widest privileges in the system -// to whoever made the lookup fail. Every uncertain path therefore falls to -// the STRICTER class, matching CreatePost's existing behaviour (service.go -// step 3 treats a failed IsAggregator lookup as an ordinary user). +// to whoever made the lookup fail. But guessing DOWNWARD is not free here +// either, which is where this path parts company with CreatePost: a lookup +// that could not be made is reported as UNDECIDED rather than resolved to +// the stricter class, because this decision gets written into the admission +// row with redrivable = false. See classify for the full asymmetry. // // It reuses evaluateAdmissionPolicy rather than admitPost, and that is the split // task 3 built for: admitPost RESERVES a ledger slot, and the engine is not a @@ -187,7 +188,12 @@ func (d *AdmissionEngineDecider) DecideAdmission(ctx context.Context, communityD postURI, communityDID, ErrSubjectGone)) } - return evaluateAdmissionPolicy(ctx, admissionDeps{ + actor, err := d.classify(ctx, post.AuthorDID) + if err != nil { + return undecided(fmt.Errorf("deciding %s for %s: %w", postURI, communityDID, err)) + } + + decision, err := evaluateAdmissionPolicy(ctx, admissionDeps{ communities: d.deps.Communities, bans: d.deps.Policy.Bans, aggregators: d.deps.Authorizer, @@ -195,7 +201,7 @@ func (d *AdmissionEngineDecider) DecideAdmission(ctx context.Context, communityD limits: d.deps.Policy.Limits, now: d.deps.Policy.Now, }, AdmissionRequest{ - Actor: d.classify(ctx, post.AuthorDID), + Actor: actor, AuthorDID: post.AuthorDID, // The community DID, which resolves to itself. The engine's input is an // admission row, and its key is already the resolved DID — there is no @@ -209,42 +215,106 @@ func (d *AdmissionEngineDecider) DecideAdmission(ctx context.Context, communityD // redeciding. Fingerprint: "", }) + if err != nil || !decision.Admitted() { + return decision, err + } + + return d.applyQuota(ctx, communityDID, post.AuthorDID, actor, decision) } -// classify decides what class of actor the author is. +// applyQuota is the firehose path's §8 submission limit, and the last thing +// between an admitted decision and the engine writing an acceptance. +// +// IT COUNTS ADMISSION ROWS, NOT LEDGER ROWS, and that substitution is the whole +// reason it exists separately from admitPost's step 6. post_submissions is +// written by CreatePost, so a post that arrived over the firehose from an author +// on another server has no ledger row and never will — counting it would hold +// LOCAL users to the limit while exempting precisely the remote ones §8 is +// about. Anyone can write unlimited postv2 records naming any community, and the +// admission layer is what absorbs that. +// +// THE LIMIT IS THE SAME NUMBER the write path uses, taken from the same config, +// so an author is held to one quota rather than to two that drift. +// +// ACTOR CLASSES ARE TREATED EXACTLY AS admitPost TREATS THEM: only ActorUser is +// metered. A registered aggregator is already governed by its own hourly quota +// inside ValidateAggregatorPost, and applying this as well would silently halve +// an authorized aggregator's throughput; a trusted one has never had a +// submission limit, and inventing one here would stop the bridge dead at a +// number nobody chose. +func (d *AdmissionEngineDecider) applyQuota( + ctx context.Context, communityDID, authorDID string, actor ActorClass, decision AdmissionDecision, +) (AdmissionDecision, error) { + if actor != ActorUser || d.deps.Admissions == nil { + return decision, nil + } + limits := d.deps.Policy.Limits + if limits.MaxPerAuthorPerCommunity <= 0 || limits.Window <= 0 { + return decision, nil + } + + since := d.deps.Policy.Now().Add(-limits.Window) + count, err := d.deps.Admissions.CountRecentAdmissions(ctx, communityDID, authorDID, since) + if err != nil { + // UNDECIDED, never a refusal. The engine persists a refusal code and + // marks it non-redrivable, so a count that could not be taken must not + // become a permanent rate-limit verdict on somebody's post. + return undecided(fmt.Errorf("counting recent admissions for %s in %s: %w", authorDID, communityDID, err)) + } + + // EXCEEDS, not reaches. The subject being decided already has its own + // admission row — the consumer opens it before the engine ever runs — so it + // is inside this count, exactly as admitPost's reservation is inside its + // own. Comparing with >= would refuse the author's very first post. + if count > limits.MaxPerAuthorPerCommunity { + return AdmissionDecision{Code: DecisionRateLimitExceeded}, nil + } + return decision, nil +} + +// classify decides what class of actor the author is, or reports that it could +// not. +// +// A FAILED LOOKUP IS UNDECIDED HERE, AND A DOWNGRADE ON THE WRITE PATH. The two +// answers are deliberate opposites, and the difference is what each caller does +// with the result afterwards. CreatePost is talking to a live client: a +// downgrade to ActorUser applies the strict checks, costs an aggregator a few +// posts until the table recovers, and hands back something the caller can +// retry. Nothing is written down. +// +// The engine writes its verdict INTO the admission row, and a policy refusal is +// stamped redrivable = false — terminal, never revisited by the redrive pass. +// The same downgrade there does not cost a retry; it permanently marks a post +// refused for a reason that was never true, because a table was briefly +// unreachable, and nothing in the system would ever look at it again. // -// EVERY UNCERTAIN PATH FALLS TO ActorUser, the stricter class. A trusted -// aggregator skips visibility, ban and authorization entirely, so resolving a -// failed lookup UPWARD would hand the widest privileges in the system to -// whoever managed to make the lookup fail. Guessing downward costs an -// aggregator some refused posts until the lookup recovers — and CreatePost -// already made exactly this choice (service.go step 3), so the engine agreeing -// with it is also what keeps the write path and the ingestion path from -// disagreeing about who someone is. -func (d *AdmissionEngineDecider) classify(ctx context.Context, authorDID string) ActorClass { +// So the rule this encodes is: A DECISION THAT PERSISTS MAY ONLY BE MADE FROM +// AN ANSWER THAT WAS ACTUALLY OBTAINED. Guessing upward is never available +// either — a trusted aggregator skips visibility, ban and authorization +// entirely, so resolving uncertainty in that direction would hand the widest +// privileges in the system to whoever could make a lookup fail. +func (d *AdmissionEngineDecider) classify(ctx context.Context, authorDID string) (ActorClass, error) { // The trusted set is checked FIRST, which is both the cheaper path and the // only one that costs nothing: it is an in-memory set resolved at // construction, so a trusted actor never pays for a database lookup to // learn what the process already knew. if d.deps.TrustedAggregatorDIDs[authorDID] { - return ActorTrustedAggregator + return ActorTrustedAggregator, nil } // With no aggregator collaborators wired — a deployment with no aggregator - // support at all — nobody can be classified as one, which is the strict - // answer rather than a degraded one. + // support at all — nobody can be classified as one. That is a CONFIGURED + // fact rather than a failed lookup, so it answers rather than defers. if d.deps.Aggregators == nil || d.deps.Authorizer == nil { - return ActorUser + return ActorUser, nil } registered, err := d.deps.Aggregators.IsAggregator(ctx, authorDID) if err != nil { - log.Printf("[ADMISSION-DECIDER] Warning: classifying %s fell back to the user class, IsAggregator failed: %v", - authorDID, err) - return ActorUser + return "", fmt.Errorf("classifying %s: %w", authorDID, err) } if registered { - return ActorRegisteredAggregator + return ActorRegisteredAggregator, nil } - return ActorUser + return ActorUser, nil } diff --git a/internal/core/posts/engine.go b/internal/core/posts/engine.go index 2cc78da..834b49a 100644 --- a/internal/core/posts/engine.go +++ b/internal/core/posts/engine.go @@ -26,15 +26,21 @@ import ( // 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. +// queue driver, a firehose redelivery, and (once task 6 lands it) the +// synchronous fast path — from processing the same subject at the same moment, +// and no lock or per-subject claim is taken. Only the first two exist today; +// the fast path is named because the safety argument has to hold when it +// arrives, not because it is calling now. // 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. +// stamps first. Serializing the passes properly is QueueDriver's job (queue.go): +// one goroutine walks one ordered list, grouped by community, so no two subjects +// of a community are ever in flight together. The layered safety above still +// matters, because the driver is not the only caller — the synchronous fast +// path and the firehose consumer both reach the engine too. // EngineOutcome reports what one pass over one subject DID. // @@ -62,11 +68,14 @@ 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. + // NOT PRODUCED BY ANY PATH YET, and deliberately still deferred. + // 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 — needs an + // old-record snapshot the consumer does not keep, so wiring it is a + // recorded decision rather than an oversight (see loop_state.md). The + // outcome is declared so that path lands against a named contract instead + // of minting one. EngineRepinned EngineOutcome = "repinned" // EngineDeferred means the subject is still owed a decision and nothing diff --git a/internal/core/posts/queue.go b/internal/core/posts/queue.go index f8caca0..51160ce 100644 --- a/internal/core/posts/queue.go +++ b/internal/core/posts/queue.go @@ -2,6 +2,7 @@ package posts import ( "context" + "errors" "fmt" "log" "sync" @@ -12,9 +13,9 @@ import ( // and on what (docs/PRD_AUTHOR_OWNED_POSTS.md §5.6, §8). // // The engine settles one subject. Nothing until now decided which subjects, in -// what order, or how often — the fast path and the firehose consumer both push -// work at it, and neither can see a subject that was left pending because a -// credential expired or a lookup blipped. This is the pull side: a periodic pass +// what order, or how often — the firehose consumer pushes work at it today, and +// task 6's synchronous fast path will push more, and neither can see a subject +// that was left pending because a credential expired or a lookup blipped. This is the pull side: a periodic pass // over the undecided backlog that eventually reaches every stranded row. // // # IT IS A SINGLE GOROUTINE, AND THAT IS THE PER-COMMUNITY SERIALIZATION @@ -218,7 +219,7 @@ func (d *QueueDriver) RunPass(ctx context.Context) (PassReport, error) { startedAt := d.now() report := PassReport{StartedAt: startedAt} - subjects, err := d.subjects.ListPendingSubjects(ctx, d.batchSize) + subjects, err := d.subjects.ListPendingSubjects(ctx, d.fetchSize()) if err != nil { // The one failure a pass has nothing to do about. Every other outcome // below is per-subject and counted; this one means there is no work @@ -228,6 +229,12 @@ func (d *QueueDriver) RunPass(ctx context.Context) (PassReport, error) { report.Listed = len(subjects) for _, subject := range groupByCommunity(subjects) { + // THE BATCH IS FILLED WITH WORK, not with rows. Skipping a backed-off + // subject must not consume a slot, or a stuck prefix would spend the + // whole pass on subjects it never touched — see fetchSize. + if report.Processed >= d.batchSize { + break + } if d.heldBack(subject, startedAt) { continue } @@ -249,6 +256,17 @@ func (d *QueueDriver) RunPass(ctx context.Context) (PassReport, error) { // community's PDS returns errors, not deferrals, so exempting failures // would leave the loudest case as the one thing nothing paced. switch { + case errors.Is(err, ErrSubjectGone): + // NOT A FAILURE, and counting it as one would make an ordinary race + // look like an outage. The backlog query excludes tombstoned and + // unindexed posts, but a post can be deleted between the listing and + // the decision — so this is the queue meeting a subject that stopped + // existing while it worked, which is exactly what the exclusion is + // for and needs no operator's attention. It is counted as deferred + // and backed off like any other "nothing to do yet"; the next pass + // will not list it at all. + report.Deferred++ + d.deferSubject(subject, startedAt) case err != nil: report.Failed++ log.Printf("[ACCEPTANCE-QUEUE] Warning: %s in %s could not be settled: %v", @@ -263,10 +281,81 @@ func (d *QueueDriver) RunPass(ctx context.Context) (PassReport, error) { } } + d.pruneDeferrals(subjects) d.record(subjects, report, startedAt) return report, nil } +// queueOverFetchFactor bounds how far past a backed-off prefix one pass may +// reach: a pass never asks the query for more than batchSize × this. +// +// FOUR, and the number is a trade rather than a preference. The backlog is +// ordered oldest-first, so the subjects most likely to be stuck are exactly the +// ones at the front of it — a community whose credentials expired weeks ago sits +// there forever. Without over-fetching, a pass asks for LIMIT rows, gets LIMIT +// stuck ones, skips them all for backoff and does nothing; every pass, while a +// healthy post two rows behind is never decided. Over-fetching without a bound +// would instead let one pass drag the entire backlog into memory to find one +// live subject. Four buys three batches of headroom against a query whose cost +// grows with it, and a prefix deeper than that drains as backoffs expire. +const queueOverFetchFactor = 4 + +// fetchSize is how many rows to ask for so the pass can still fill its batch +// after skipping the subjects it already knows are held back. +// +// It asks for the batch plus exactly the number of deferrals currently held — +// the measured size of the prefix that may be skipped — rather than always +// over-fetching. A driver with nothing backed off has nothing to skip, so it +// asks for precisely what it will use. +func (d *QueueDriver) fetchSize() int { + held := d.deferredCount() + size := d.batchSize + held + if ceiling := d.batchSize * queueOverFetchFactor; size > ceiling { + size = ceiling + } + return size +} + +func (d *QueueDriver) deferredCount() int { + d.mu.Lock() + defer d.mu.Unlock() + return len(d.deferrals) +} + +// pruneDeferrals forgets the backoffs of subjects that are no longer listed. +// +// A deferral outlives its subject otherwise. The row gets settled by somebody +// else — the synchronous fast path, a firehose acceptance, a moderator's +// removal — and simply stops being listed, without ever telling the driver. The +// map is keyed by subject and swept by nothing, so on a busy instance it is an +// unbounded leak held for the life of the process. +// +// It is also wrong on RE-ENTRY, which is the part a leak metric would not show: +// a subject that leaves the backlog and comes back — an edit reopening an +// accepted post — would arrive carrying a stale backoff it did nothing to earn, +// and wait out a delay that was about a completely different decision. +// +// Pruning against the LISTING rather than against what the pass processed is +// deliberate: a subject skipped for backoff is still in the backlog, and +// forgetting it would defeat the backoff on the very next pass. +func (d *QueueDriver) pruneDeferrals(listed []PendingSubject) { + d.mu.Lock() + defer d.mu.Unlock() + + if len(d.deferrals) == 0 { + return + } + stillListed := make(map[subjectKey]bool, len(listed)) + for _, subject := range listed { + stillListed[keyOf(subject)] = true + } + for key := range d.deferrals { + if !stillListed[key] { + delete(d.deferrals, key) + } + } +} + // Snapshot returns the driver's health surface as of the last completed pass. func (d *QueueDriver) Snapshot() QueueSnapshot { d.mu.Lock() @@ -321,6 +410,7 @@ func (d *QueueDriver) record(subjects []PendingSubject, report PassReport, at ti PendingBacklog: report.Listed, LastPassDeferred: report.Deferred, LastPassFailed: report.Failed, + DeferredSubjects: len(d.deferrals), } // Taken as a MINIMUM rather than as subjects[0], even though the query // orders by age. The oldest entry's age is the queue's only early warning, diff --git a/internal/core/users/interfaces.go b/internal/core/users/interfaces.go index cf0f849..620ee01 100644 --- a/internal/core/users/interfaces.go +++ b/internal/core/users/interfaces.go @@ -13,6 +13,21 @@ type UpdateProfileInput struct { } // UserRepository defines the interface for user data persistence +// ErasureLookup reports whether a DID names an account this AppView was asked +// to erase (migration 036). +// +// It is a SEPARATE, OPTIONAL interface rather than a method on UserRepository, +// and detected with a type assertion at the one call site that needs it. Adding +// it to UserRepository would oblige every implementation to answer a question +// only the PostgreSQL one can — and a double that answered "not erased" by +// default would be a gate that fails open, which is the single outcome this +// marker exists to prevent. A repository that does not implement it disables +// the gate rather than weakening it, and nothing in production is such a +// repository. +type ErasureLookup interface { + IsAccountDeleted(ctx context.Context, did string) (bool, error) +} + type UserRepository interface { Create(ctx context.Context, user *User) (*User, error) GetByDID(ctx context.Context, did string) (*User, error) diff --git a/internal/core/users/service.go b/internal/core/users/service.go index de0b035..99e2b2f 100644 --- a/internal/core/users/service.go +++ b/internal/core/users/service.go @@ -455,6 +455,37 @@ func (s *userService) mintInviteCode(ctx context.Context) (string, error) { // best-effort — this heals users whose profile firehose event was never delivered // without ever blocking or failing the IndexUser call itself. func (s *userService) IndexUser(ctx context.Context, did, handle, pdsURL string) error { + // THE ERASURE GATE, and IndexUser is where it belongs because this is the + // FIREHOSE's door into the users table. A DID appearing in a profile or + // identity event means only that some repo somewhere emitted a record — a + // bridge, a replay, an overlapping feed — and any of those can arrive months + // after the account was erased. Letting it through would make the erasure + // undone by exactly the replays the marker exists to defend against, and + // undone silently: the users row reappears, repo.Create clears the marker on + // its way past, and the next replayed post indexes normally. + // + // The marker's only exit is a genuine re-registration, which reaches the + // repository's insert directly rather than through here. + // + // A LOOKUP FAILURE REFUSES. "I could not read the marker table" and "there + // is no marker" must never be the same answer, because the second one + // indexes. + if lookup, ok := s.userRepo.(ErasureLookup); ok { + erased, err := lookup.IsAccountDeleted(ctx, did) + if err != nil { + return fmt.Errorf("checking the erasure marker for %s before indexing: %w", did, err) + } + if erased { + // Nil, not an error. Every caller is a firehose consumer, and the + // connector dead-letters what a handler returns — so refusing with + // an error would turn each erased account into a permanent stream of + // redriving profile events. This is not a failure; it is an event + // with nothing to do. + log.Printf("INFO: not indexing %s from the firehose: the account was erased (migration 036 marker)", did) + return nil + } + } + // Try to create the user (idempotent - CreateUser returns existing user if DID exists) user, err := s.CreateUser(ctx, CreateUserRequest{ DID: did, diff --git a/internal/db/postgres/admission_queue_repo.go b/internal/db/postgres/admission_queue_repo.go index b232c9a..70a9f20 100644 --- a/internal/db/postgres/admission_queue_repo.go +++ b/internal/db/postgres/admission_queue_repo.go @@ -99,5 +99,28 @@ func (r *postgresAdmissionRepo) ListPendingSubjects(ctx context.Context, limit i func (r *postgresAdmissionRepo) CountRecentAdmissions( ctx context.Context, communityDID, authorDID string, since time.Time, ) (int, error) { - return 0, nil + // starts_with on the authority segment, the same shape userRepo.Delete uses + // to sweep an author's admissions. The trailing "/" matters: without it + // did:plc:abc would also match did:plc:abcdef, and one author would consume + // another's quota. + // + // accepted and pending together, and NOTHING else. §8 is explicit that a + // refusal consumes no quota — counting rejected or removed rows would let an + // author past their limit extend their own lockout every time they tried + // again. pending_reacceptance counts too: the post is admitted and visible + // history, merely awaiting a re-decision on an edit. + const query = ` + SELECT count(*) + FROM community_post_admissions + WHERE community_did = $1 + AND starts_with(post_uri, 'at://' || $2 || '/') + AND status IN ('accepted', 'pending', 'pending_reacceptance') + AND created_at >= $3 + ` + + var count int + if err := r.db.QueryRowContext(ctx, query, communityDID, authorDID, since).Scan(&count); err != nil { + return 0, fmt.Errorf("counting recent admissions for %s in %s: %w", authorDID, communityDID, err) + } + return count, nil } diff --git a/internal/db/postgres/deleted_account_repo.go b/internal/db/postgres/deleted_account_repo.go index 1f9c71c..4990aac 100644 --- a/internal/db/postgres/deleted_account_repo.go +++ b/internal/db/postgres/deleted_account_repo.go @@ -29,9 +29,24 @@ func NewDeletedAccountRepository(db *sql.DB) *DeletedAccountRepository { // is indistinguishable from a healthy answer — a database blip would silently // re-index the content a deletion erased, which is the exact outcome the marker // table exists to prevent. +func (r *postgresUserRepo) IsAccountDeleted(ctx context.Context, did string) (bool, error) { + return accountIsErased(ctx, r.db, did) +} + +// IsAccountDeleted implements the same lookup for the standalone repository. func (r *DeletedAccountRepository) IsAccountDeleted(ctx context.Context, did string) (bool, error) { + return accountIsErased(ctx, r.db, did) +} + +// accountIsErased is the single statement behind both lookups above. +// +// It is one function because the two callers are the two halves of the same +// guard — the ingestion consumer refusing an erased author's events, and the +// user service refusing to re-index them — and a second spelling would be a +// second chance for one of them to drift into failing open. +func accountIsErased(ctx context.Context, db *sql.DB, did string) (bool, error) { var deleted bool - if err := r.db.QueryRowContext(ctx, + if err := db.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM deleted_accounts WHERE did = $1)`, did, ).Scan(&deleted); err != nil { return false, fmt.Errorf("checking whether %s was erased: %w", did, err) -- 2.51.2