From 8144a90d009fe289c41ff1a76bf6b97cdb833116 Mon Sep 17 00:00:00 2001 From: Bretton <36870434+BrettM86@users.noreply.github.com> Date: Fri, 10 Jul 2026 15:06:16 -0700 Subject: [PATCH] =?UTF-8?q?Task=2009:=20BigSky=20relay=20in=20the=20e2e=20?= =?UTF-8?q?pipeline=20=E2=80=94=20Lemmy=20=E2=86=92=20tidepool=20=E2=86=92?= =?UTF-8?q?=20relay=20=E2=86=92=20Jetstream?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every e2e scenario now transits a real relay (the strictest consumer that exists: PLC-backed DID resolution, per-commit signature verification, getRepo crawl), and the bridge announces itself through the production RequestCrawlAll path — closing the "requestCrawl never exercised against a real relay" follow-up. Also plants the v1.1 loop (tasks 09–12 specs, PLAN/LOOP_STATE tables) and the votes-as-records design-revisit note. Infra: - compose: pinned bigsky + relay-postgres (bgs + carstore DBs) + one-shot relay-bootstrap (fresh bigsky refuses ALL non-admin requestCrawl: new-PDS-per-day limit defaults to 0, checked before trusted domains); Jetstream re-pointed at the relay; direct bridge tap behind profile `direct`. Every env var verified against the image — Coves' stanza sets three that don't exist (--crawl-insecure-ws is arg-only). - HANDLE_RESOLVER_HOSTS=tidepool makes bridged handles verify via the bridge's /.well-known/atproto-did (no DNS wildcard needed in compose). - ALLOW_DEV_REQUEST_CRAWL (dev-only, refused in production) lets the harness drive the real announce path; dev sending is dial-time restricted to loopback/private via ap.NewPrivateOnlyHTTPClient (inverse SSRF guard) so a dev misconfig can never announce to a public relay. - RequestCrawlAll: per-relay goroutines with bounded retry (24×5s, 10s per-attempt cap) — bigsky's requestCrawl handler calls back into the announcer's describeServer, racing the bridge's own listener at startup. Terminal-vs-retryable is decided by pre-flight checks only; ALL wire errors including 4xx retry deliberately (bigsky answers the callback race with HTTP 400 — comment + regression test pin this). Tests: - tests/e2e/relay_test.go: bridge-originated crawl asserted in the relay's PDS registry (registered + actively subscribed); repos listed and served (getLatestCommit rev >= observed event's rev). - Order-tolerant listener (pending buffer): bigsky's parallel indexer preserves only per-repo order, so cross-repo assertions relaxed to presence+linkage — while vetEvent gains a per-DID rev-monotonicity check so the per-repo ordering that DOES survive the relay is now asserted suite-wide. drain() returns+clears the pending buffer (no vacuous-pass trap for task 10's negative assertions). - New unit pins: private-only dial guard (public IP refused, loopback ok), 400-is-retried, per-attempt timeout bounds a hanging relay, validation test rewritten to actually detect retrying (was vacuous). Discovered and documented (FOLLOWUPS "Relay pipeline"): cross-repo event ordering does not survive a relay — the Coves AppView's author-before-post validation needs a park-and-retry path; tombstone active:false is unobservable through bigsky until the #account frame (task 11); indigo's slurper drops a PDS after the 16th consecutive dial failure. Review: 7/7 (4 Claude specialty + codex gpt-5.6-sol, gemini-3.1-pro, glm-5.2). Codex unique catch: the dev override's missing public-address guard. Full clean make e2e: 10/10 scenarios, 96.7s. Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 7 +- FOLLOWUPS.md | 106 ++++++++++++- LOOP_STATE.md | 61 +++++++- Makefile | 2 +- PLAN.md | 9 ++ README.md | 112 +++++++++----- cmd/tidepool/main.go | 16 +- docker-compose.e2e.yml | 206 +++++++++++++++++++++++-- e2e/relay/init-carstore.sql | 7 + internal/ap/egress.go | 48 +++++- internal/ap/egress_test.go | 58 +++++++ internal/config/config.go | 25 ++- internal/config/config_test.go | 30 ++++ internal/sync/crawl.go | 124 +++++++++++++-- internal/sync/sync_test.go | 160 +++++++++++++++++++ tasks/09-e2e-relay.md | 72 +++++++++ tasks/10-e2e-scenarios.md | 55 +++++++ tasks/11-hardening.md | 63 ++++++++ tasks/12-perf-scale.md | 48 ++++++ tests/e2e/bridge_test.go | 108 ++++++++----- tests/e2e/helpers.go | 271 ++++++++++++++++++++++++++++++--- tests/e2e/relay_test.go | 134 ++++++++++++++++ 22 files changed, 1575 insertions(+), 147 deletions(-) create mode 100644 e2e/relay/init-carstore.sql create mode 100644 tasks/09-e2e-relay.md create mode 100644 tasks/10-e2e-scenarios.md create mode 100644 tasks/11-hardening.md create mode 100644 tasks/12-perf-scale.md create mode 100644 tests/e2e/relay_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d497928..838cf2e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,9 +7,10 @@ # plc.directory). # # e2e: on demand — the full docker-compose stack (Lemmy built from source in -# debug mode, did:plc directory, Jetstream). Heavy (the Lemmy debug build -# is a full Rust compile), so it runs only via workflow_dispatch or the -# `run-e2e` PR label. +# debug mode, did:plc directory, BigSky relay, Jetstream — every scenario +# transits Lemmy → bridge → relay → Jetstream). Heavy (the Lemmy debug +# build is a full Rust compile), so it runs only via workflow_dispatch or +# the `run-e2e` PR label. name: ci diff --git a/FOLLOWUPS.md b/FOLLOWUPS.md index 703a10b..5e6a611 100644 --- a/FOLLOWUPS.md +++ b/FOLLOWUPS.md @@ -2,8 +2,85 @@ Everything known-deferred at the end of the v1 build loop (tasks 01–08), collected from `LOOP_STATE.md`'s cross-task notes plus discoveries made -while building the e2e harness. Organized by area; items marked **(e2e)** -were discovered or confirmed by the task-08 harness. +while building the e2e harness, updated as v1.1 tasks land. Organized by +area; items marked **(e2e)** were discovered or confirmed by the task-08 +harness, **(relay)** by task 09's relay pipeline. + +## Relay pipeline (task 09 discoveries) + +- **Cross-repo event ordering does NOT survive a relay (relay).** bigsky + indexes its inbound firehose with a parallel scheduler keyed by repo DID + (indigo `events/schedulers/parallel`, 100 workers): per-repo order is + preserved, cross-repo order is not. The bridge's profile-before-post + emission discipline (locked decision 3's "author indexed before post") + therefore only holds on the bridge's OWN firehose — through a relay, a + post in the community repo routinely overtakes its author's first-ever + `actor.profile` (the profile costs the relay a PLC resolution + handle + verification; the community DID is already cached). **Carry to Coves:** + the AppView consumes through relay infrastructure, so its "author must be + indexed before the post" validation needs a retry/park-and-reprocess + path, or Coves consumes the bridge firehose directly. The e2e suite's + listener buffers out-of-order events (`tests/e2e/helpers.go` `pending`) + and asserts presence + linkage instead of cross-repo arrival order. +- **Tombstoned (`active:false`) repo status is not observable through + bigsky.** It filters tombstoned/taken-down repos out of `listRepos`, + serves no `getRepoStatus`, and learns account state upstream only from + `#account` frames — which the bridge does not emit yet (task 11). When + the `#account` frame lands, add a relay-side e2e assertion that a + consent-revoked repo disappears from the relay's `listRepos`. +- **A fresh bigsky refuses all non-admin `requestCrawl`**: the new-PDS + per-day limit defaults to 0 and is checked BEFORE the trusted-domain + list. Any deployment announcing to a self-hosted relay needs the + `setPerDayLimit` admin bootstrap (compose does it in `relay-bootstrap`). +- **Verify env vars against `--help`, not other people's compose files.** + Coves' relay stanza sets `BGS_CRAWL_INSECURE_WS`, `BGS_PORT`, and + `LOG_LEVEL` — none of which exist in the pinned image + (`--crawl-insecure-ws` has no env binding and must be a command arg; the + log knob is `BSKYLOG_LOG_LEVEL`). Also local-only traps: bigsky defaults + to the public `plc.directory` (`ATP_PLC_HOST`) AND the public `1.1.1.1` + DNS resolver (`RESOLVE_ADDRESS`). +- The pinned bigsky image has **no arm64 manifest** — Apple Silicon runs it + under emulation (`platform: linux/amd64`), fine at e2e volume; a future + image bump could revisit. +- indigo's slurper reconnect backoff is effectively sub-second for the + first 10 attempts (`sleepForBackoff` multiplies by 2 **nanoseconds**, so + only the additive +0–1s jitter — rand milliseconds — matters), then 30s; + on the 16th consecutive dial failure (backoff > 15) it marks the PDS + `registered=false` and STOPS retrying — a + bridge outage longer than ~3 minutes needs a fresh `requestCrawl` (ours + re-announces on every startup, which covers the bridge's own restarts but + not a long bridge outage while the relay stays up). + +## Design revisits (decide before/with the write side) + +- **Votes-as-records (2026-07-10 discussion).** The aggregate side channel + is a v1 decision worth re-opening when write-back is designed, not a law + of nature. Facts to carry into that decision: + - A faithful **going-forward** per-voter record stream is buildable + today: every live Like/Dislike arrives with the voter's AP identity + (that's how `vote_events` dedupes). Flips/undos are already solved + state-tracking. + - **History is counts-only forever**: Lemmy's per-voter list endpoints + (`listPostLikes`/`listCommentLikes`) are origin-instance admin/mod + APIs — the bridge is a federated peer, not an admin. Any records + design still needs an aggregate baseline for pre-subscribe history. + - Naïve design (vote record in the voter's repo) mints a permanent + public did:plc per drive-by voter — a global-registry externality, + not a compute cost. **Leading alternative:** votes in the + *community's* repo with a `voter` field + deterministic rkey + (voter+subject hash), mirroring how posts already live in the + community repo with an `author` field. Zero new DIDs, votes on the + firehose. Requires a new lexicon + Coves AppView consumer — decide + WITH Coves. + - Write amplification (one commit + firehose event per vote) is only + viable after the perf items land: per-DID MST cache, block GC, + batched pruning (tasks 11–12). Hardening first is a prerequisite, + not a competing priority. + - Write-back symmetry favors records: Coves users' votes on bridged + posts are already native `social.coves.feed.vote` records; outbound + translation is records→Like. Symmetric records would let frontends + drop the `getVoteAggregates` XRPC for everything except historical + baselines. ## Federation & interop @@ -45,8 +122,13 @@ were discovered or confirmed by the task-08 harness. callers should eventually REUSE an orphaned minted DID via a PLC updateHandle op instead of re-minting (task 03 note). - Moderation federation, DMs. -- Relay `requestCrawl` is wired but has never been exercised against a real - relay (dev logs instead of sending). +- ~~Relay `requestCrawl` is wired but has never been exercised against a + real relay (dev logs instead of sending).~~ **Closed by task 09**: the + e2e stack runs a real BigSky; the bridge announces itself through the + production `RequestCrawlAll` path (now with a bounded retry — the relay + calls back into `describeServer`, racing process start) under the + dev-only `ALLOW_DEV_REQUEST_CRAWL` override, and the suite asserts the + bridge is registered + actively subscribed in the relay's PDS registry. - `ENVIRONMENT=production` has never been end-to-end tested (the harness runs development mode for migrations-on-start, strict validation, http scheme, private fetch). @@ -69,6 +151,10 @@ were discovered or confirmed by the task-08 harness. - `SigningKeys` could become a `SignCommit` capability (keeps key plaintext inside identity; enables KMS later) — revisit before the interface calcifies. +- `getRepo`'s optional `since` parameter (diff export) is not implemented — + a `since` request gets the full CAR, which the spec permits (extra blocks + are legal); consumers needing incremental sync use subscribeRepos + (`internal/sync/server.go`). ## Ingestion (task 06 notes) @@ -88,6 +174,12 @@ were discovered or confirmed by the task-08 harness. - Mint-gate ("retry via queue backoff") is verified at unit level only — the harness never drives minting into the rate limiter (a low `MINT_RATE_PER_MINUTE` stack variant would need its own compose profile). +- The `activityID` rand-failure path is guarded but unit-untestable + (Go 1.24+ makes a `crypto/rand` failure a fatal crash, not a returnable + error) — permanent test gap unless the reader is injected. +- `ingest.NewNoopVotes` (`internal/ingest/votes.go`) is dead code: task 07 + wired the real `votes.NewAggregator` in main.go and nothing (prod or + test) references the noop anymore — delete candidate. ## Votes (task 07 notes) @@ -148,8 +240,10 @@ were discovered or confirmed by the task-08 harness. (`PLC_COMMIT` in `e2e/plc/Dockerfile`); bump it deliberately via `git ls-remote` when upstream fixes/features are needed. - Jetstream **exits** when its upstream drops; `restart: unless-stopped` - papers over it. If Jetstream grows reconnect logic upstream, drop the - policy. + papers over it. Since task 09 its upstream is the relay (which stays up + across the bridge-restart scenario — the relay's own slurper reconnects), + so the policy only matters if the relay itself dies. If Jetstream grows + reconnect logic upstream, drop the policy. - Unexpected-collection enforcement runs on **every** commit event any listener consumes (await and drain both fail fast on a collection outside the four emitted ones), and the vote scenario watches the firehose diff --git a/LOOP_STATE.md b/LOOP_STATE.md index 786c44b..933a711 100644 --- a/LOOP_STATE.md +++ b/LOOP_STATE.md @@ -16,7 +16,25 @@ update this file → schedule next. Stop the loop when every task is `done`. | 8 | 08-e2e-harness (infra: Dockerfile, compose, Lemmy federation, Makefile, CI, lexicon-sync) | done | (see git log) | 7/7 reviewers (4 Claude + codex/gemini/glm, first full external panel since 03); fixes: PRODUCTION https→http redirect-downgrade guard (codex unique catch), webfinger fallback narrowed to transport failures + both-legs errors + 4 tests, minter PDS-endpoint scheme threading, PLC image commit-pin, --wait-timeout + CI logs if:always() + Makefile up-failure cleanup/teardown-status, loopback-only host binds, check-lexicons fail-open holes, sync-lexicons bridge-nesting guard, 2 false compose-header claims rewritten (invented env var, wrong --wait semantics) | | 9 | 08-e2e-harness (tests: tests/e2e helpers + 8 scenarios, FOLLOWUPS.md, README) | done | (see git log) | 8/8 reviewers (5 Claude + codex/gemini/glm; gemini zero-issue "excellent", codex sharpest); fixes: drain() dead-listener vacuous-pass (5/8 flagged), centralized vetEvent (unknown-collection Fatalf + lexicon-validate every consumed create/update, suite-wide locked-decision-7 enforcement), scenario-7 backfill-completion poll + gap-post cursor-resume proof + op-agnostic dup keys, readLoop goroutine join, subscribe fail-fast on explicit reject, embed.external e2e coverage, seeder e2e assertion. NEW ASSERTION CAUGHT REAL BUG: Lemmy vote-clear federates Undo with RECONSTRUCTED inner vote (fresh id, type Like even for live dislike; flips are bare opposite votes, no Undo) → id-targeted RetractVote no-oped every production vote-clear; fixed with known-id replay probe + live-vote fallback + 3 unit tests | -ALL TASKS DONE — loop complete. `make e2e` green (8/8 scenarios, ~100s), full unit suite green. +v1 loop (tasks 01–08) COMPLETE — `make e2e` green (8/8 scenarios, ~100s), full unit suite green. + +## v1.1 loop (tasks 09–12) — added 2026-07-10 + +Goal: work the FOLLOWUPS.md backlog and put a real relay in the e2e +pipeline. Locked requirement: every state has a full e2e pipeline of +Lemmy → PDS record → firehose ingestion (where applicable) — task 09 +re-points Jetstream through the relay so every scenario transits it. +Same protocol: implement (Fable agent) → /second-opinion → fix → verify → +commit → update this file → next task. Votes-as-records was deliberately +NOT scheduled — see FOLLOWUPS.md "Design revisits" (decide with the +write-back design; tasks 11–12 are its prerequisites). + +| # | Task | Status | Commit | Notes | +|---|------|--------|--------|-------| +| 10 | 09-e2e-relay | done | (see git log) | 7/7 reviewers (4 Claude emulated + codex/gemini/glm); fixes: dev requestCrawl PUBLIC-relay dial guard (codex unique catch — NewPrivateOnlyHTTPClient, inverse SSRF guard), terminal-error classification made pre-flight-only (whole-chain IsValidation was abandoning a relay on attempt 1 for transient DNS), 10s per-attempt timeout (budget arithmetic was 14min worst-case, not 2min), vacuous validation-no-retry test rewritten + 400-is-retried pin, vetEvent per-DID rev-monotonicity (restores per-repo ordering assertion suite-wide), drain() returns+clears pending (closes task-10 vacuous-pass trap), relay poll robustness + pagination cap, doc corrections (RESOLVE_ADDRESS overstatement, spec BGS_CRAWL_INSECURE_WS annotation, FOLLOWUPS 16th-failure off-by-one). KEPT DELIBERATE over 3 reviewers' objection: all wire errors incl. 4xx retried — bigsky answers the describeServer callback race with HTTP 400 (comment + test pin it). Final clean make e2e: 10/10, 96.7s | +| 11 | 10-e2e-scenarios | pending | | image/consent/Delete(Actor)/unsubscribe/community-update/vote-hammer/suite-end sweep | +| 12 | 11-hardening | pending | | inbox+sync rate limits, #account frame, delete-before-create, pruners, housekeeping | +| 13 | 12-perf-scale | pending | | MST cache, getRepo streaming/reachable-set, blocks GC, ClaimNext scan | Statuses: pending → in-progress → review → done (or blocked: ). @@ -384,6 +402,47 @@ and deferred TODOs here) never affects run outcome). SEED_COUNTS_FROM_API (default on; strict bool parse rejects typos). Known accepted drift: a baseline voter who flips sends Undo for a like we never saw (no-op) — stale until re-seed. +### From task 09 (relay pipeline — tasks 10/11 MUST know) +- The e2e pipeline is now Lemmy → tidepool → BigSky relay → Jetstream: + EVERY scenario's events transit relay validation (DID resolution against + the local PLC, per-commit sig verification). jetstream-direct (compose + profile `direct`, host :6038) is a debug tap only. Relay on host :2480 + (admin key e2e-relay-admin-key); helpers: relayPDSList/relayListRepos/ + relayGetLatestCommit in tests/e2e/helpers.go. +- CRITICAL for scenario design: bigsky's parallel indexer (100 workers + keyed by repo DID) preserves per-repo order but NOT cross-repo order — + author profiles and community posts routinely swap. The e2e listener now + BUFFERS consumed-but-unmatched commit events and rescans them on later + awaits; await predicates MUST be PURE (side-effecting accounting belongs + in drain loops — scenario 8 was rewritten that way). Cross-repo ordering + assertions were removed from scenarios 2/3/6; presence + linkage is what + survives a relay. Carry to Coves AppView: it cannot rely on + profile-before-post through relay infra (FOLLOWUPS "Relay pipeline"). +- New config: ALLOW_DEV_REQUEST_CRAWL (dev-only, refused in production) + makes dev actually send requestCrawl; RequestCrawlAll now retries per + relay (5s × 24, vars compressible in tests) because bigsky's requestCrawl + handler calls BACK into the announcing host's describeServer before + subscribing — the first attempt races the bridge's own listener (observed + live: attempt 1 fails, attempt 2 lands). +- bigsky env facts (verified against image --help + source; Coves' compose + stanza is wrong): ATP_PLC_HOST (plc), RELAY_ADMIN_KEY, DATA_DIR, + RESOLVE_ADDRESS (defaults to PUBLIC 1.1.1.1 — set 127.0.0.11), + HANDLE_RESOLVER_HOSTS=tidepool (trial-host resolver GETs + /.well-known/atproto-did with the handle as Host header — handles VERIFY, + 0 failures in a full run), BSKYLOG_LOG_LEVEL; --crawl-insecure-ws has NO + env binding (command arg). Fresh relay refuses ALL non-admin requestCrawl + (new-PDS-per-day limit 0, checked before trusted domains) → one-shot + relay-bootstrap service raises it; tidepool depends_on its completion. + Image has no arm64 manifest (platform: linux/amd64, emulated on Apple + Silicon). Relay gets its own postgres with TWO dbs (bgs + carstore). +- Tombstone visibility through the relay is a DEAD END until task 11: + bigsky has no getRepoStatus, filters tombstoned repos from listRepos, and + only learns account state from #account frames the bridge doesn't emit + yet. When task 11 adds the #account frame, add the relay-side assertion + (repo disappears from relay listRepos after consent revocation). +- Timing budgets: eventTimeout 90s→120s (relay hop + first-sight DID work + + amd64 emulation), burstTimeout 3m (scenario 8 is drain-based now), + crawlTimeout 3m (covers the announce retry window). Suite runs ~114s. - DEFERRED (task 08+): vote_events grows unbounded (no pruning of superseded/undone rows — mirror FIREHOSE_RETENTION treatment alongside ap_tombstones); actor-delete/consent revocation does NOT scrub that diff --git a/Makefile b/Makefile index a87ee68..5b0e535 100644 --- a/Makefile +++ b/Makefile @@ -117,7 +117,7 @@ e2e: ## Full e2e run: build + start the stack, run the suite, tear down (-v) e2e-up: ## Start the e2e stack and leave it running (for iterating on tests) @$(E2E_COMPOSE) up -d --build --wait --wait-timeout 600 - @echo "$(GREEN)✓ e2e stack up: tidepool 127.0.0.1:8092, lemmy 127.0.0.1:8541, jetstream 127.0.0.1:6028$(RESET)" + @echo "$(GREEN)✓ e2e stack up: tidepool 127.0.0.1:8092, lemmy 127.0.0.1:8541, relay 127.0.0.1:2480, jetstream 127.0.0.1:6028$(RESET)" e2e-test: ## Run the e2e suite against an already-running stack (make e2e-up) @go test -tags e2e -count=1 -v -timeout 20m ./tests/e2e/... diff --git a/PLAN.md b/PLAN.md index 1499b39..c9fa508 100644 --- a/PLAN.md +++ b/PLAN.md @@ -89,6 +89,15 @@ sanctioned side-channel XRPC. Nothing ever strongRefs a vote. | 7 | tasks/07-vote-aggregates.md | Vote ingestion, aggregate store, side-channel XRPC | 800 | 6 | | 8 | tasks/08-e2e-harness.md | docker-compose with real Lemmy + PLC + Jetstream, E2E tests, lexicon conformance | 1000 | 4,6 | +### v1.1 sections (added 2026-07-10 — FOLLOWUPS backlog + relay pipeline) + +| # | Task file | What | ~LOC | Depends on | +|---|-----------|------|------|-----------| +| 9 | tasks/09-e2e-relay.md | BigSky relay in e2e; full pipeline Lemmy → bridge → relay → Jetstream; real requestCrawl | 600 | 8 | +| 10 | tasks/10-e2e-scenarios.md | Scenario completion: image, consent, Delete(Actor), unsubscribe, community update, vote hammer, suite-end sweep | 800 | 9 | +| 11 | tasks/11-hardening.md | Pre-internet-facing hardening: rate limits, #account frame, ordering gaps, pruners, housekeeping | 1000 | 10 | +| 12 | tasks/12-perf-scale.md | MST cache, getRepo streaming/reachable-set, blocks GC, ClaimNext — prerequisite for any votes-as-records revisit | 800 | 11 | + ## Loop protocol Each iteration (one section per iteration): diff --git a/README.md b/README.md index 00550e9..c792ee8 100644 --- a/README.md +++ b/README.md @@ -36,30 +36,67 @@ your directory is on a different port. The end-to-end harness runs the whole read path against **real infrastructure** — a real Lemmy federating with the bridge, a real did:plc -directory backing DID minting, and a real Jetstream decoding the firehose — -in one compose network: +directory backing DID minting, a real atproto relay (indigo **BigSky**) +crawling the bridge, and a real Jetstream decoding the **relay's** firehose +— in one compose network: ``` docker-compose.e2e.yml - ┌─────────────────────────────────────────────────────────────────────┐ - │ http://lemmy (debug build) http://tidepool (this repo) │ - │ ┌───────────────────────┐ Follow ◀─ ┌───────────────────────────┐ │ - │ │ lemmy + lemmy-postgres│ Accept ─▶ │ tidepool + its postgres │ │ - │ │ + pictrs │ Announce ─▶│ inbox → queue → material │ │ - │ └───────────────────────┘ (plain │ izer → virtual PDS │ │ - │ HTTP) └─────┬──────────────┬──────┘ │ - │ mint DIDs │ CBOR frames│ │ - │ ┌─────────────▼──┐ ┌────────▼──────┐ │ - │ │ plc (did:plc │ │ jetstream │ │ - │ │ + plc-postgres)│ │ (JSON events) │ │ - │ └────────────────┘ └────────┬──────┘ │ - └───────────────────────────────────────────────────────────┼─────────┘ - host (127.0.0.1 only): tidepool :8092, lemmy :8541, jetstream :6028 ◀─┘ + ┌───────────────────────────────────────────────────────────────────────┐ + │ http://lemmy (debug build) http://tidepool (this repo) │ + │ ┌───────────────────────┐ Follow ◀─ ┌───────────────────────────┐ │ + │ │ lemmy + lemmy-postgres│ Accept ─▶ │ tidepool + its postgres │ │ + │ │ + pictrs │ Announce ─▶│ inbox → queue → material │ │ + │ └───────────────────────┘ (plain │ izer → virtual PDS │ │ + │ HTTP) └─────┬──────────────┬──────┘ │ + │ mint DIDs │ CBOR frames│ ▲ │ + │ ┌─────────────▼──┐ │ │request │ + │ │ plc (did:plc │◀────┐ │ │Crawl │ + │ │ + plc-postgres)│ │ ┌──▼─┴───────┐ │ + │ └────────────────┘ DID │ │ relay │ │ + │ resolution └──│ (BigSky) │ │ + │ │ + postgres │ │ + │ └──────┬─────┘ │ + │ CBOR frames│ │ + │ ┌───────▼─────┐ │ + │ │ jetstream │ │ + │ │ (JSON) │ │ + │ └───────┬─────┘ │ + └───────────────────────────────────────────────────────────────┼───────┘ + host (127.0.0.1 only): tidepool :8092, lemmy :8541, │ + relay :2480, jetstream :6028 ◀──────┘ tests/e2e (go test -tags e2e) ``` -The host ports bind **loopback-only** (`127.0.0.1:8092/8541/6028`): the -stack carries an admin token and runs with `ALLOW_PRIVATE_FETCH=1`, so it +Every event the suite consumes has therefore survived the strictest consumer +that exists: the relay resolves each repo's DID against the local PLC, +verifies every commit signature, and crawls repos over the bridge's +`getRepo`. The bridge announces itself to the relay on startup via the +**real** `RequestCrawlAll` code path (`RELAY_HOSTS=http://relay:2470` plus +the dev-only `ALLOW_DEV_REQUEST_CRAWL=1` — production always sends and +refuses the flag), retrying internally because the relay validates the +hostname by calling back into the bridge's `describeServer`. A fresh +relay's new-PDS-per-day limit is 0 (refuses all non-admin `requestCrawl`), +so a one-shot `relay-bootstrap` service raises the limit over the admin API +before the bridge starts — the announcement itself is still +bridge-originated, and the suite asserts it landed (`relay_test.go`). +Bridged handles verify for real too: `HANDLE_RESOLVER_HOSTS=tidepool` +points bigsky's trial-host resolver at the bridge's Host-header-keyed +`/.well-known/atproto-did` (compose-DNS-invisible names like +`alice.lemmy.tidepool` would otherwise fail handle verification, which +bigsky treats as non-fatal). For debugging, a direct bridge→Jetstream tap +exists behind the `direct` compose profile (`jetstream-direct`, host port +6038) — it is not the tested path. + +One ordering consequence worth knowing: bigsky indexes its inbound firehose +with a parallel scheduler keyed by repo DID, so **per-repo event order +survives the relay but cross-repo order does not** — an author's +`actor.profile` (author repo) and their post (community repo) may swap on +the relay's output, and any AppView consuming through relay infrastructure +must tolerate that (see FOLLOWUPS.md). + +The host ports bind **loopback-only** (`127.0.0.1:8092/8541/2480/6028`): +the stack carries admin tokens and runs with `ALLOW_PRIVATE_FETCH=1`, so it must not be reachable from the local network. ```sh @@ -81,23 +118,29 @@ network, with `BRIDGE_SCHEME=http` (dev-only) making the bridge emit plain-HTTP AP ids to match. See the header comments in `docker-compose.e2e.yml` and `e2e/lemmy/Dockerfile` for the full story. -The suite (`tests/e2e/bridge_test.go`, build tag `e2e`) covers: subscribe → -`community.profile` on the firehose; a link post → `actor.profile` then -`community.post` in order, with the shared url crossing the wire as -`embed.external`; comment threads in the authors' repos with resolving -strongRefs; edits/deletes as update/delete ops; post **and** comment votes -reaching `getVoteAggregates` — upvote, flip, retract — while **never** -appearing as records; pre-subscribe backfill with each author's profile -emitted before their posts and Lemmy's pre-existing vote counts seeded into -the aggregates; a mid-test container restart proving deterministic-rkey +The suite (`tests/e2e/bridge_test.go` + `relay_test.go`, build tag `e2e`) +covers: subscribe → `community.profile` on the firehose; a link post → +`actor.profile` and `community.post` (presence + author linkage; arrival +order across the two repos is relay-dependent, see above), with the shared +url crossing the wire as `embed.external`; comment threads in the authors' +repos with resolving strongRefs; edits/deletes as update/delete ops; post +**and** comment votes reaching `getVoteAggregates` — upvote, flip, retract — +while **never** appearing as records; pre-subscribe backfill with every +post's author profile arriving and Lemmy's pre-existing vote counts seeded +into the aggregates; a mid-test container restart proving deterministic-rkey replay idempotency (the forced backfill redo is confirmed complete via the admin API's `last_backfill_at` before asserting it emitted nothing) plus -exactly-once delivery of a post created in the recovery window and Jetstream -cursor resume; and a concurrent-ingestion burst accounted per -(did, collection/rkey). Every create/update the tests consume from Jetstream -is validated against the vendored Coves lexicons on the consumer side of the -wire, and any collection outside the four the bridge emits fails the suite -immediately (votes must never become records). Two scripts keep the vendored +exactly-once delivery of a post created in the recovery window — now proven +**through the relay**, whose slurper reconnects to the bounced bridge with +its stored cursor; a concurrent-ingestion burst accounted per +(did, collection/rkey); and relay-state assertions — the bridge registered +and actively subscribed in the relay's PDS registry (the bridge-originated +`requestCrawl`, asserted rather than eyeballed), and bridged repos listed +and served by the relay's own `listRepos`/`getLatestCommit`. Every +create/update the tests consume from Jetstream has passed the relay's +signature verification AND is validated against the vendored Coves lexicons +on the consumer side of the wire, and any collection outside the four the +bridge emits fails the suite immediately (votes must never become records). Two scripts keep the vendored lexicons honest: `scripts/sync-lexicons.sh` copies them from a Coves checkout; `scripts/check-lexicons.sh` verifies the committed manifest (the layer that runs in CI) and additionally byte-compares against @@ -124,7 +167,8 @@ production**: | `USER_AGENT` | derived | outbound HTTP user agent | | `ALLOW_PRIVATE_FETCH` | off | dev-only: disables the SSRF egress guard (AP fetches **and** PLC directory requests) so localhost targets work | | `FIREHOSE_RETENTION` | `72h` | how long `firehose_events` rows are kept for `subscribeRepos` cursor replay (Go duration; a background pruner trims older events hourly) | -| `RELAY_HOSTS` | *(optional)* | comma-separated relays to send `com.atproto.sync.requestCrawl` to on startup; in development the request is logged, never sent | +| `RELAY_HOSTS` | *(optional)* | comma-separated relays to send `com.atproto.sync.requestCrawl` to on startup (each retried on a bounded budget — the relay calls back into `describeServer` before subscribing, which can race process start); in development the request is logged, never sent, unless `ALLOW_DEV_REQUEST_CRAWL` opts in | +| `ALLOW_DEV_REQUEST_CRAWL` | off | dev-only: actually SEND `requestCrawl` to `RELAY_HOSTS` in development (exists for the e2e stack's local BigSky); refused in production, where sending is already the behavior | | `ADMIN_TOKEN` | `dev-admin-token` | bearer token protecting the `/admin` API | | `BACKFILL_MAX_POSTS` | `100` | posts materialized per community backfill run | | `MINT_RATE_PER_MINUTE` / `MINT_BURST` | `60` / `120` | rate gate on inbound DID minting (PLC registrations are forever; unseen authors in delivered content trigger mints) | diff --git a/cmd/tidepool/main.go b/cmd/tidepool/main.go index ae1f793..d15696b 100644 --- a/cmd/tidepool/main.go +++ b/cmd/tidepool/main.go @@ -126,13 +126,23 @@ func run(logger *slog.Logger) error { // Ask configured relays to crawl us. Development hosts are not publicly // reachable, so dev only logs what it would have sent (never touches a - // live relay from a laptop). + // live relay from a laptop) — unless ALLOW_DEV_REQUEST_CRAWL opts in, + // which exists for the e2e harness's LOCAL BigSky relay and is refused + // by config in production. Even then dev must not be able to poke public + // infrastructure, so the dev-override path uses the private-only client: + // it refuses any non-loopback/private/link-local destination at dial + // time. Production keeps the standard SSRF-guarded client, where public + // relays are the point. if len(cfg.RelayHosts) > 0 { - if cfg.IsDevelopment() { - logger.Info("development environment: skipping requestCrawl", + if cfg.IsDevelopment() && !cfg.AllowDevRequestCrawl { + logger.Info("development environment: skipping requestCrawl (set ALLOW_DEV_REQUEST_CRAWL=1 to send to a local relay)", "relays", cfg.RelayHosts, "hostname", cfg.BridgeHostname) } else { crawlClient := ap.NewGuardedHTTPClient(cfg.AllowPrivateAddresses, 30*time.Second) + if cfg.IsDevelopment() { + // Reached only under ALLOW_DEV_REQUEST_CRAWL: local relays only. + crawlClient = ap.NewPrivateOnlyHTTPClient(30 * time.Second) + } go tidepoolsync.RequestCrawlAll(ctx, crawlClient, cfg.RelayHosts, cfg.BridgeHostname, logger) } } diff --git a/docker-compose.e2e.yml b/docker-compose.e2e.yml index 7ff60c1..9763a32 100644 --- a/docker-compose.e2e.yml +++ b/docker-compose.e2e.yml @@ -1,7 +1,54 @@ # Tidepool end-to-end stack: a real Lemmy federating with the bridge, a real -# did:plc directory backing DID minting, and a real Jetstream decoding the -# bridge's subscribeRepos firehose. `make e2e` builds it, waits for health, -# runs tests/e2e (tagged `e2e`) from the host, and tears it down. +# did:plc directory backing DID minting, a real atproto relay (indigo +# BigSky) crawling the bridge like a hostile consumer — DID resolution +# against the local PLC, per-commit signature verification, getRepo crawls — +# and a real Jetstream decoding the RELAY's firehose. `make e2e` builds it, +# waits for health, runs tests/e2e (tagged `e2e`) from the host, and tears +# it down. The tested pipeline is: +# +# Lemmy ──AP──▶ tidepool ──subscribeRepos──▶ relay (BigSky) ──▶ jetstream +# +# so every scenario's records transit relay validation for free. A direct +# bridge→Jetstream tap exists behind the `direct` compose profile for +# debugging only (jetstream-direct below) — it is NOT the tested path. +# +# ── BigSky facts this file depends on (verified against the pinned image's +# --help output and the indigo source, bgs/ + cmd/bigsky/) ───────────── +# * PLC host is the `--plc-host` flag / ATP_PLC_HOST env (defaults to the +# PUBLIC https://plc.directory — it MUST be pointed at the compose `plc` +# service or the relay fails closed resolving our did:plc identities). +# * `--crawl-insecure-ws` (ws:// to the PDS, and http:// for requestCrawl +# hostname validation) has NO env-var binding — it must be a command-line +# arg. Coves' BGS_CRAWL_INSECURE_WS env var does not exist in the image +# and is silently ignored; same for BGS_PORT and LOG_LEVEL (the real log +# knob is BSKYLOG_LOG_LEVEL). +# * A fresh relay's slurp config has new_pds_per_day_limit = 0, and that +# limiter is checked BEFORE the trusted-domain list — a brand-new relay +# refuses every non-admin requestCrawl until an admin raises the limit. +# The one-shot `relay-bootstrap` service does exactly that (and nothing +# else): the crawl itself is still bridge-originated (RequestCrawlAll), +# which is the thing under test. +# * requestCrawl validates the hostname by calling back into the PDS's +# /xrpc/com.atproto.server.describeServer before subscribing — hence the +# bridge's retrying announcement (internal/sync/crawl.go) and tidepool's +# depends_on relay-bootstrap. +# * Handle verification: HANDLE_RESOLVER_HOSTS switches bigsky to its +# TestHandleResolver, which GETs http:///.well-known/atproto-did +# with the HANDLE in the Host header — exactly the resolution surface the +# bridge already serves. Bridged handles (alice.lemmy.tidepool) do not +# exist in compose DNS, so without this the prod resolver would fail — +# which is NON-fatal in bigsky (bgs.createExternalUser logs and keeps the +# repo with an invalid handle), but with it handles actually verify. +# * RESOLVE_ADDRESS defaults to the PUBLIC 1.1.1.1:53 resolver; pointed at +# Docker's embedded DNS (127.0.0.11) as defense-in-depth on a dead code +# path — NOT what makes the stack local-only. Docker's embedded DNS +# forwards unknown names to its upstream resolver, so this alone would +# not be hermetic; the invariant actually holds because +# HANDLE_RESOLVER_HOSTS fully REPLACES the DNS-TXT handle resolver with +# the TestHandleResolver (above) and ATP_PLC_HOST keeps did:plc +# resolution on the compose plc, leaving no code path that queries DNS. +# * The image has no arm64 manifest (`platform: linux/amd64`; Apple Silicon +# runs it under emulation — fine at e2e volume). # # ── How Lemmy↔Tidepool federation works over plain HTTP ────────────────── # Lemmy only accepts http:// federation URLs in a DEBUG build (a compile-time @@ -27,11 +74,12 @@ # trust store is not reliably extensible at runtime, and the proxy adds # nothing for a local harness. # -# Host ports (loopback-only — the stack carries an admin token and +# Host ports (loopback-only — the stack carries admin tokens and # ALLOW_PRIVATE_FETCH=1, so it must not be reachable from the local -# network; offset from the dev stack so both can run): +# network; offset from the dev stack AND the Coves dev relay so all can run): # 127.0.0.1:8092 Tidepool HTTP (admin API, XRPC, healthz) # 127.0.0.1:8541 Lemmy HTTP API (8541 nods to lemmy_alpha in upstream's federation compose) +# 127.0.0.1:2480 Relay (BigSky) XRPC + admin API # 127.0.0.1:6028 Jetstream WebSocket (/subscribe) # # LOCAL-ONLY: nothing here talks to plc.directory, public relays, or public @@ -73,6 +121,13 @@ services: PLC_DIRECTORY_URL: http://plc:3000 ADMIN_TOKEN: e2e-admin-token SEED_COUNTS_FROM_API: "1" + # Announce ourselves to the local relay on startup via the REAL + # RequestCrawlAll path. ALLOW_DEV_REQUEST_CRAWL exists precisely for + # this stack (dev is otherwise log-only, and production refuses the + # flag); the announcement retries internally because the relay + # validates the hostname by calling back into our describeServer. + RELAY_HOSTS: http://relay:2470 + ALLOW_DEV_REQUEST_CRAWL: "1" LOG_LEVEL: debug ports: - "127.0.0.1:${TIDEPOOL_E2E_PORT:-8092}:80" @@ -82,6 +137,11 @@ services: condition: service_healthy plc: condition: service_healthy + # The relay must be up AND willing to accept new PDS subscriptions + # (see relay-bootstrap) before the bridge's startup requestCrawl fires, + # or the announcement burns its retry budget against a 0-limit relay. + relay-bootstrap: + condition: service_completed_successfully healthcheck: # /healthz pings the database, so healthy == migrated and serving. test: ["CMD", "wget", "--spider", "-q", "http://localhost/healthz"] @@ -125,29 +185,117 @@ services: retries: 30 start_period: 30s - # ── Jetstream (consumes the bridge's firehose) ────────────────────────── + # ── BigSky relay (indigo) ─────────────────────────────────────────────── + relay-postgres: + image: postgres:16 + environment: + POSTGRES_DB: bgs + POSTGRES_USER: relay + POSTGRES_PASSWORD: relay + volumes: + # First-boot init: creates the second database (`carstore`) — bigsky + # keeps BGS state and carstore shard metadata in separate databases. + - ./e2e/relay/init-carstore.sql:/docker-entrypoint-initdb.d/init-carstore.sql:ro + networks: [tidepool-e2e] + healthcheck: + test: ["CMD-SHELL", "pg_isready -U relay -d bgs"] + interval: 2s + timeout: 5s + retries: 30 + + relay: + # Pinned bigsky build (same pin Coves uses). All flag/env names below + # are verified against this image's --help output — see the header + # comment for the ones that look plausible but do NOT exist. + image: ghcr.io/bluesky-social/indigo:bigsky-0a2d4173e6e89e49b448f6bb0a6e1ab58d12b385 + # No arm64 manifest; Apple Silicon hosts run it under emulation. + platform: linux/amd64 + # --crawl-insecure-ws has no env binding (verified): ws:// to the bridge + # and http:// hostname validation. The image entrypoint is `dumb-init --` + # so the command must restate the binary. + command: ["/bigsky", "--crawl-insecure-ws"] + environment: + DATABASE_URL: postgresql://relay:relay@relay-postgres:5432/bgs?sslmode=disable + CARSTORE_DATABASE_URL: postgresql://relay:relay@relay-postgres:5432/carstore?sslmode=disable + DATA_DIR: /data + # LOCAL-ONLY, twice over: DID resolution against the compose PLC (the + # default is the public plc.directory), and DNS against Docker's + # embedded resolver (the default is the public 1.1.1.1). + ATP_PLC_HOST: http://plc:3000 + RESOLVE_ADDRESS: 127.0.0.11:53 + # Bridged handles live under the bridge hostname and resolve via the + # bridge's own /.well-known/atproto-did (Host-header keyed) — bigsky's + # trial-host resolver speaks exactly that. Without this, handle + # verification fails non-fatally (repo kept, handle marked invalid). + HANDLE_RESOLVER_HOSTS: tidepool + RELAY_ADMIN_KEY: e2e-relay-admin-key + # Default is 100 repos per PDS — a long-lived stack under repeated + # suite runs mints past that. + RELAY_DEFAULT_REPO_LIMIT: "10000" + # Disk persister gives real event playback (with TTL) for downstream + # cursor replay instead of the DB persister's unbounded table. + RELAY_PERSISTER_DIR: /data/persister + BSKYLOG_LOG_LEVEL: info + ports: + - "127.0.0.1:${RELAY_E2E_PORT:-2480}:2470" + networks: [tidepool-e2e] + depends_on: + relay-postgres: + condition: service_healthy + plc: + condition: service_healthy + healthcheck: + test: ["CMD", "wget", "--spider", "-q", "http://localhost:2470/xrpc/_health"] + interval: 2s + timeout: 5s + retries: 30 + start_period: 30s + + # One-shot bootstrap: raise the new-PDS-per-day limit from its fresh-DB + # default of 0 (which refuses ALL non-admin requestCrawl — checked before + # the trusted-domain list, so there is no config-only bypass). This is the + # sanctioned admin-API bootstrap; the crawl announcement itself still + # comes from the bridge (the code path under test). Reuses the bigsky + # image (has busybox wget) to avoid pulling another one. + relay-bootstrap: + image: ghcr.io/bluesky-social/indigo:bigsky-0a2d4173e6e89e49b448f6bb0a6e1ab58d12b385 + platform: linux/amd64 + entrypoint: ["/bin/sh", "-ec"] + command: + - >- + wget -q -O- --post-data='' + --header='Authorization: Bearer e2e-relay-admin-key' + 'http://relay:2470/admin/subs/setPerDayLimit?limit=1000' + restart: "no" + networks: [tidepool-e2e] + depends_on: + relay: + condition: service_healthy + + # ── Jetstream (consumes the RELAY's firehose — the tested pipeline) ───── jetstream: image: ghcr.io/bluesky-social/jetstream:sha-306e463693365e21a5ffd3ec051a5a7920000214 - # Jetstream EXITS when its upstream websocket dies ("shutting down on - # events kill"), which is exactly what the restart scenario provokes by - # bouncing tidepool. Docker revives it; it reconnects with the cursor - # persisted in /data (container-local, survives restarts) and resumes - # replay from the bridge without gaps or duplicates. + # Jetstream EXITS when its upstream websocket dies. Its upstream is now + # the relay, which stays up across the restart scenario (the RELAY + # reconnects to the bounced bridge with its stored per-PDS cursor), so + # the policy only matters if the relay itself dies. Docker revives it; + # it reconnects with the cursor persisted in /data (container-local, + # survives restarts) and resumes replay without gaps or duplicates. restart: unless-stopped environment: - JETSTREAM_WS_URL: ws://tidepool:80/xrpc/com.atproto.sync.subscribeRepos + JETSTREAM_WS_URL: ws://relay:2470/xrpc/com.atproto.sync.subscribeRepos JETSTREAM_LISTEN_ADDR: ":6018" JETSTREAM_METRICS_LISTEN_ADDR: ":6019" JETSTREAM_DATA_DIR: /data JETSTREAM_EVENT_TTL: 24h - # The bridge is quiet between test steps; don't self-restart on idle. + # The stack is quiet between test steps; don't self-restart on idle. JETSTREAM_LIVENESS_TTL: 24h LOG_LEVEL: debug ports: - "127.0.0.1:${JETSTREAM_E2E_PORT:-6028}:6018" networks: [tidepool-e2e] depends_on: - tidepool: + relay: condition: service_healthy # No wget/curl in the image, but it is debian-based with bash — a # /dev/tcp probe of the subscribe port is a real readiness gate. Without @@ -160,6 +308,36 @@ services: retries: 30 start_period: 10s + # Debug-only tap: Jetstream wired straight to the bridge, bypassing the + # relay. NOT the tested path — bring it up alongside the stack with + # docker compose -f docker-compose.e2e.yml --profile direct up -d jetstream-direct + # and point a listener at ws://127.0.0.1:6038 to compare the bridge's raw + # emissions against what survives relay validation. + jetstream-direct: + image: ghcr.io/bluesky-social/jetstream:sha-306e463693365e21a5ffd3ec051a5a7920000214 + profiles: [direct] + restart: unless-stopped + environment: + JETSTREAM_WS_URL: ws://tidepool:80/xrpc/com.atproto.sync.subscribeRepos + JETSTREAM_LISTEN_ADDR: ":6018" + JETSTREAM_METRICS_LISTEN_ADDR: ":6019" + JETSTREAM_DATA_DIR: /data + JETSTREAM_EVENT_TTL: 24h + JETSTREAM_LIVENESS_TTL: 24h + LOG_LEVEL: debug + ports: + - "127.0.0.1:${JETSTREAM_DIRECT_E2E_PORT:-6038}:6018" + networks: [tidepool-e2e] + depends_on: + tidepool: + condition: service_healthy + healthcheck: + test: ["CMD", "bash", "-c", "exec 3<>/dev/tcp/localhost/6018"] + interval: 2s + timeout: 5s + retries: 30 + start_period: 10s + # ── Lemmy ─────────────────────────────────────────────────────────────── lemmy-postgres: image: postgres:16 diff --git a/e2e/relay/init-carstore.sql b/e2e/relay/init-carstore.sql new file mode 100644 index 0000000..3d7c5f2 --- /dev/null +++ b/e2e/relay/init-carstore.sql @@ -0,0 +1,7 @@ +-- The BigSky relay wants TWO databases: its main BGS state (users, PDS +-- registry, slurp config) and the carstore's shard metadata. One postgres +-- container serves both; this init script (run once by the postgres image's +-- entrypoint on first boot) creates the second database next to the +-- POSTGRES_DB-created `bgs`. The CAR shard FILES live on the relay +-- container's own disk (DATA_DIR) — only the metadata is relational. +CREATE DATABASE carstore OWNER relay; diff --git a/internal/ap/egress.go b/internal/ap/egress.go index 92988e8..3749ee0 100644 --- a/internal/ap/egress.go +++ b/internal/ap/egress.go @@ -28,6 +28,29 @@ func NewGuardedHTTPClient(allowPrivate bool, timeout time.Duration) *http.Client } } +// NewPrivateOnlyHTTPClient returns an *http.Client whose transport enforces +// the INVERSE of the SSRF guard at dial time: every destination must be +// loopback, RFC1918/ULA private, or link-local — anything public is refused. +// It backs dev-only paths that talk to local containers (currently the +// ALLOW_DEV_REQUEST_CRAWL startup crawl against the e2e stack's local relay), +// enforcing in code the project invariant that development and tests NEVER +// contact public infrastructure. Production paths must keep the standard +// guarded client (NewGuardedHTTPClient), where public destinations are the +// point. +func NewPrivateOnlyHTTPClient(timeout time.Duration) *http.Client { + if timeout == 0 { + timeout = DefaultRequestTimeout + } + guard := &egressGuard{ + privateOnly: true, + lookupIPAddr: net.DefaultResolver.LookupIPAddr, + } + return &http.Client{ + Timeout: timeout, + Transport: guardedTransport(nil, guard), + } +} + // egressGuard blocks outbound requests to addresses an SSRF attacker would // pivot through: loopback, RFC1918 private, link-local, unique-local, // multicast, unspecified, and the cloud-metadata endpoint. It rejects @@ -42,6 +65,10 @@ func NewGuardedHTTPClient(allowPrivate bool, timeout time.Duration) *http.Client // the httptest-based tests that hit 127.0.0.1 turn it off. type egressGuard struct { allowPrivate bool + // privateOnly inverts the guard: ONLY loopback/private/link-local + // destinations are dialable, everything public is refused. Dev-only + // (NewPrivateOnlyHTTPClient); it takes precedence over allowPrivate. + privateOnly bool // lookupIPAddr resolves a hostname to candidate IPs. It is a field (not a // direct net.Resolver call) so tests can inject a DNS-rebinding scenario — // a hostname that resolves to a blocked address — without a real DNS server. @@ -76,8 +103,16 @@ func (g *egressGuard) checkURL(u *url.URL) error { } // checkIP rejects addresses in the blocked ranges unless the guard is -// disabled for dev/test. +// disabled for dev/test. In privateOnly mode the check is inverted: only +// local (loopback/private/link-local) addresses pass. func (g *egressGuard) checkIP(ip net.IP) error { + if g.privateOnly { + if !isLocalIP(ip) { + return errors.NewValidationError("url", + fmt.Sprintf("address %s is public; this dev-only client permits local egress only", ip)) + } + return nil + } if g.allowPrivate { return nil } @@ -93,6 +128,17 @@ func (g *egressGuard) checkIP(ip net.IP) error { // explicit and covers the mapped forms. var metadataV4 = net.IPv4(169, 254, 169, 254) +// isLocalIP reports whether ip is a local destination the private-only +// dev client may dial: loopback, RFC1918 / unique-local private, or +// link-local unicast. Everything else — including unspecified and +// multicast — is refused. +func isLocalIP(ip net.IP) bool { + if ip == nil { + return false + } + return ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() +} + func isBlockedIP(ip net.IP) bool { if ip == nil { return true diff --git a/internal/ap/egress_test.go b/internal/ap/egress_test.go index 49a3144..54e2790 100644 --- a/internal/ap/egress_test.go +++ b/internal/ap/egress_test.go @@ -7,6 +7,7 @@ import ( "net/http/httptest" "net/url" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -159,5 +160,62 @@ func TestDialGuard_IPLiteralAddr(t *testing.T) { require.Error(t, err, "a blocked IP literal must be refused at dial time") } +// TestPrivateOnlyHTTPClient_RefusesPublicDestinations: the dev-only client +// (ALLOW_DEV_REQUEST_CRAWL's crawl path) is the INVERSE of the SSRF guard — +// it must refuse anything public at dial time, so a misconfigured +// RELAY_HOSTS in dev can never contact live infrastructure. 203.0.113.10 is +// TEST-NET-3: the refusal happens before any dial, so nothing is contacted. +func TestPrivateOnlyHTTPClient_RefusesPublicDestinations(t *testing.T) { + client := NewPrivateOnlyHTTPClient(2 * time.Second) + resp, err := client.Get("http://203.0.113.10/") + if resp != nil { + _ = resp.Body.Close() + } + require.Error(t, err, "a public literal-IP destination must be refused at dial time") + assert.True(t, errors.IsValidation(err), "refusal must be the guard's validation error, got %v", err) + assert.Contains(t, err.Error(), "public") +} + +// TestPrivateOnlyHTTPClient_AllowsLoopback: the same client must still reach +// local servers — that is its whole purpose (local e2e relays). +func TestPrivateOnlyHTTPClient_AllowsLoopback(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + client := NewPrivateOnlyHTTPClient(2 * time.Second) + resp, err := client.Get(server.URL) + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + assert.Equal(t, http.StatusNoContent, resp.StatusCode) +} + +// TestPrivateOnlyDialGuard_ResolvedPublicIP: a hostname that resolves to a +// public address is refused by the private-only guard at dial time, same +// resolved-IP mechanics as the SSRF guard but inverted. +func TestPrivateOnlyDialGuard_ResolvedPublicIP(t *testing.T) { + g := &egressGuard{privateOnly: true} + g.lookupIPAddr = func(_ context.Context, _ string) ([]net.IPAddr, error) { + return []net.IPAddr{{IP: net.IPv4(93, 184, 216, 34)}}, nil + } + dialed := false + base := func(_ context.Context, _, _ string) (net.Conn, error) { + dialed = true + return stubConn{}, nil + } + _, err := g.dialContext(base)(context.Background(), "tcp", "relay.example:443") + require.Error(t, err, "a hostname resolving to a public address must be refused") + assert.False(t, dialed, "the private-only guard must not dial a public resolved address") + + // The same guard dials a loopback resolution. + g.lookupIPAddr = func(_ context.Context, _ string) ([]net.IPAddr, error) { + return []net.IPAddr{{IP: net.IPv4(127, 0, 0, 1)}}, nil + } + _, err = g.dialContext(base)(context.Background(), "tcp", "relay.local:443") + require.NoError(t, err) + assert.True(t, dialed) +} + // stubConn is a no-op net.Conn so the dial guard tests never touch the network. type stubConn struct{ net.Conn } diff --git a/internal/config/config.go b/internal/config/config.go index 1f84e18..5058732 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -75,7 +75,9 @@ type Config struct { // RelayHosts lists relays to send com.atproto.sync.requestCrawl to on // startup (RELAY_HOSTS, comma-separated, optional). In development the // request is logged instead of sent — dev hosts are not publicly - // reachable and must never poke real relays. + // reachable and must never poke real relays — unless + // ALLOW_DEV_REQUEST_CRAWL opts in, which sends but only to local + // (loopback/private/link-local) relay addresses. RelayHosts []string // AllowPrivateAddresses disables the SSRF egress guard, letting the AP // client fetch loopback/private/link-local/metadata addresses. It defaults @@ -83,6 +85,19 @@ type Config struct { // tests that hit httptest servers on 127.0.0.1. Set ALLOW_PRIVATE_FETCH=1 // to enable. AllowPrivateAddresses bool + // AllowDevRequestCrawl makes ENVIRONMENT=development actually SEND + // com.atproto.sync.requestCrawl to RelayHosts on startup instead of only + // logging the would-be request. It exists for harnesses that run a REAL + // local relay (the e2e stack's BigSky) — dev hosts are otherwise not + // publicly reachable and must never poke live relays. The local-only + // invariant is enforced at dial time: when this flag is active the + // crawl client (ap.NewPrivateOnlyHTTPClient) refuses any destination + // that is not loopback/private/link-local, so a public relay in + // RELAY_HOSTS cannot be contacted from dev. Refused in production, + // where sending is already the behavior and the flag could only + // mislead (the ALLOW_PRIVATE_FETCH pattern). Set + // ALLOW_DEV_REQUEST_CRAWL=1 to enable. + AllowDevRequestCrawl bool // AdminToken is the bearer token protecting the /admin API (community // subscribe/unsubscribe/backfill). ADMIN_TOKEN; required in production, // dev default is a fixed, publicly known value. @@ -237,6 +252,14 @@ func Load(logger *slog.Logger) (*Config, error) { return nil, fmt.Errorf("config: ALLOW_PRIVATE_FETCH must not be set in production") } + // Dev-only requestCrawl override, same posture as ALLOW_PRIVATE_FETCH: + // meaningful only where dev would otherwise log instead of send, refused + // where it could mask a config mistake. + cfg.AllowDevRequestCrawl = boolVar("ALLOW_DEV_REQUEST_CRAWL") + if cfg.AllowDevRequestCrawl && !isDevelopment { + return nil, fmt.Errorf("config: ALLOW_DEV_REQUEST_CRAWL must not be set in production (production always sends requestCrawl)") + } + // Admin API auth: like the KEK, the dev default is fixed and public — // required in production. cfg.AdminToken, err = stringVar(logger, isDevelopment, "ADMIN_TOKEN", "dev-admin-token") diff --git a/internal/config/config_test.go b/internal/config/config_test.go index e40a6d9..dd2af7d 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -20,6 +20,7 @@ func clearConfigEnv(t *testing.T) { "PLC_DIRECTORY_URL", "BRIDGE_SERVICE_DID", "USER_AGENT", "BRIDGE_KEK", "ADMIN_TOKEN", "BACKFILL_MAX_POSTS", "MINT_RATE_PER_MINUTE", "MINT_BURST", "INGEST_WORKERS", "BRIDGE_SCHEME", + "ALLOW_PRIVATE_FETCH", "ALLOW_DEV_REQUEST_CRAWL", "RELAY_HOSTS", } { t.Setenv(name, "") } @@ -156,6 +157,35 @@ func TestLoad_BridgeScheme(t *testing.T) { require.Error(t, err, "unknown schemes are rejected") } +func TestLoad_AllowDevRequestCrawl(t *testing.T) { + clearConfigEnv(t) + + cfg, err := Load(discardLogger()) + require.NoError(t, err) + assert.False(t, cfg.AllowDevRequestCrawl, "off by default: dev logs requestCrawl instead of sending") + + t.Setenv("ALLOW_DEV_REQUEST_CRAWL", "1") + cfg, err = Load(discardLogger()) + require.NoError(t, err) + assert.True(t, cfg.AllowDevRequestCrawl, "the e2e harness turns real dev requestCrawl on") +} + +func TestLoad_AllowDevRequestCrawlRefusedInProduction(t *testing.T) { + clearConfigEnv(t) + t.Setenv("ENVIRONMENT", EnvironmentProduction) + t.Setenv("DATABASE_URL", "postgres://prod/db") + t.Setenv("LISTEN_ADDR", ":8080") + t.Setenv("BRIDGE_HOSTNAME", "tidepool.example") + t.Setenv("PLC_DIRECTORY_URL", "https://plc.directory") + t.Setenv("BRIDGE_KEK", "sfDrM4bIeCJp01ZBTArLPJXNQlD7pcYFsod2An6UAF0=") + t.Setenv("ADMIN_TOKEN", "prod-admin-token") + t.Setenv("ALLOW_DEV_REQUEST_CRAWL", "1") + + _, err := Load(discardLogger()) + require.Error(t, err, "production always sends; the dev override set there is a config mistake") + assert.Contains(t, err.Error(), "ALLOW_DEV_REQUEST_CRAWL") +} + func TestLoad_BridgeSchemeHTTPRefusedInProduction(t *testing.T) { clearConfigEnv(t) t.Setenv("ENVIRONMENT", EnvironmentProduction) diff --git a/internal/sync/crawl.go b/internal/sync/crawl.go index 8bd5654..b4850d2 100644 --- a/internal/sync/crawl.go +++ b/internal/sync/crawl.go @@ -9,12 +9,36 @@ import ( "log/slog" "net/http" "strings" + gosync "sync" + "time" comatproto "github.com/bluesky-social/indigo/api/atproto" "tidepool/internal/errors" ) +// Startup crawl announcements retry on a bounded budget: requestCrawl is +// sent while the process is still booting, and the receiving relay VALIDATES +// the hostname by calling straight back into this bridge's +// /xrpc/com.atproto.server.describeServer before subscribing (verified in +// indigo bgs/handlers.go) — so the very first attempt can race our own HTTP +// listener coming up, and the relay itself may still be starting. Transient +// startup failures are normal; a relay that stays unreachable for the whole +// budget is logged and dropped (the next process restart re-announces). +// Vars, not consts, so tests can compress the schedule. +// +// Budget arithmetic: each attempt is capped at crawlAttemptTimeout, so the +// worst case against a relay that hangs every attempt is +// crawlMaxAttempts × (crawlAttemptTimeout + crawlRetryInterval) +// = 24 × (10s + 5s) = 6 minutes. The common failure mode (connection +// refused while the relay boots) fails near-instantly, giving +// ≈ 24 × 5s = 2 minutes. +var ( + crawlRetryInterval = 5 * time.Second + crawlAttemptTimeout = 10 * time.Second // per-attempt cap so a hanging relay cannot blow the budget + crawlMaxAttempts = 24 +) + // RequestCrawl asks one relay to start crawling this host, via // com.atproto.sync.requestCrawl. relay may be a bare hostname (https is // assumed) or a full http(s) URL; hostname is this bridge's public hostname. @@ -22,16 +46,10 @@ import ( // egress transport (ap.NewGuardedHTTPClient) and tests can hit httptest // servers. func RequestCrawl(ctx context.Context, client *http.Client, relay, hostname string) error { - if client == nil { - return errors.NewValidationError("client", "must not be nil") - } - if hostname == "" { - return errors.NewValidationError("hostname", "must not be empty") + if err := validateRequestCrawlInput(client, relay, hostname); err != nil { + return err } base := strings.TrimSuffix(strings.TrimSpace(relay), "/") - if base == "" { - return errors.NewValidationError("relay", "must not be empty") - } if !strings.HasPrefix(base, "http://") && !strings.HasPrefix(base, "https://") { base = "https://" + base } @@ -47,6 +65,9 @@ func RequestCrawl(ctx context.Context, client *http.Client, relay, hostname stri } req.Header.Set("Content-Type", "application/json") + // Everything past this point touches the wire; failures here may be + // transient and are retried by requestCrawlWithRetry. + resp, err := client.Do(req) if err != nil { return fmt.Errorf("sync: requestCrawl to %s: %w", base, err) @@ -59,21 +80,92 @@ func RequestCrawl(ctx context.Context, client *http.Client, relay, hostname stri return nil } +// validateRequestCrawlInput runs the pre-flight caller-bug checks that +// RequestCrawl performs before touching the wire: nil client, empty +// hostname, empty relay. requestCrawlWithRetry runs these ONCE ahead of its +// retry loop and treats only these as terminal — a caller bug cannot be +// fixed by retrying, but any error from an actual HTTP attempt might be +// transient and stays retryable. +func validateRequestCrawlInput(client *http.Client, relay, hostname string) error { + if client == nil { + return errors.NewValidationError("client", "must not be nil") + } + if hostname == "" { + return errors.NewValidationError("hostname", "must not be empty") + } + if strings.TrimSuffix(strings.TrimSpace(relay), "/") == "" { + return errors.NewValidationError("relay", "must not be empty") + } + return nil +} + // RequestCrawlAll announces this host to every configured relay, logging // per-relay outcomes instead of failing fast — relays are independent and a -// dead one must not block the others. Callers gate on environment: in -// development this function should not be invoked at all (main logs the -// would-be request instead), because dev hosts are not publicly reachable -// and must never poke real relays. +// dead one must not block the others (each gets its own goroutine and retry +// budget; the call returns when every relay has succeeded or exhausted its +// budget). Callers gate on environment: in development this function is +// only invoked under ALLOW_DEV_REQUEST_CRAWL (the e2e stack's local relay); +// otherwise dev logs the would-be request, because dev hosts are not +// publicly reachable and must never poke real relays. func RequestCrawlAll(ctx context.Context, client *http.Client, relays []string, hostname string, logger *slog.Logger) { if logger == nil { logger = slog.Default() } + var wg gosync.WaitGroup for _, relay := range relays { - if err := RequestCrawl(ctx, client, relay, hostname); err != nil { - logger.Error("requestCrawl failed", "relay", relay, "error", err) - continue + wg.Add(1) + go func(relay string) { + defer wg.Done() + requestCrawlWithRetry(ctx, client, relay, hostname, logger) + }(relay) + } + wg.Wait() +} + +// requestCrawlWithRetry drives one relay's announcement to a terminal +// outcome: success, a pre-flight validation error (caller bug — retrying +// cannot fix it), context cancellation, or budget exhaustion. +func requestCrawlWithRetry(ctx context.Context, client *http.Client, relay, hostname string, logger *slog.Logger) { + // Pre-flight caller-bug checks run once, before the loop, and are the + // ONLY terminal errors. Everything that comes back from an actual HTTP + // attempt is retried within the budget — deliberately including + // transport-layer ValidationErrors from the egress guard (a transient + // "host did not resolve to any address" must not abandon the relay) and + // ALL HTTP statuses including 4xx: bigsky answers 400 while the + // describeServer callback race is unresolved (it probes our listener + // before subscribing), and that 400 clears once our listener is up. + // Do not "fix" this by short-circuiting on client errors. + if err := validateRequestCrawlInput(client, relay, hostname); err != nil { + logger.Error("requestCrawl rejected (not retrying)", "relay", relay, "error", err) + return + } + for attempt := 1; ; attempt++ { + // Cap each attempt so a hanging relay cannot stretch the budget past + // the arithmetic documented on crawlMaxAttempts (the injected + // client's own timeout is typically longer, e.g. 30s in main). + attemptCtx, cancel := context.WithTimeout(ctx, crawlAttemptTimeout) + err := RequestCrawl(attemptCtx, client, relay, hostname) + cancel() + if err == nil { + logger.Info("requested crawl", "relay", relay, "hostname", hostname, "attempt", attempt) + return + } + // A shutdown can surface as the in-flight attempt's error; report it + // as shutdown, not as a relay failure or budget exhaustion. + if ctx.Err() != nil { + logger.Warn("requestCrawl abandoned", "relay", relay, "cause", ctx.Err()) + return + } + if attempt >= crawlMaxAttempts { + logger.Error("requestCrawl failed, budget exhausted", "relay", relay, "attempts", attempt, "error", err) + return + } + logger.Warn("requestCrawl failed, will retry", "relay", relay, "attempt", attempt, "error", err) + select { + case <-ctx.Done(): + logger.Warn("requestCrawl abandoned", "relay", relay, "cause", ctx.Err()) + return + case <-time.After(crawlRetryInterval): } - logger.Info("requested crawl", "relay", relay, "hostname", hostname) } } diff --git a/internal/sync/sync_test.go b/internal/sync/sync_test.go index 25fc6dc..ee8a4c9 100644 --- a/internal/sync/sync_test.go +++ b/internal/sync/sync_test.go @@ -6,6 +6,7 @@ import ( "encoding/json" "fmt" "io" + "log/slog" "net" "net/http" "net/http/httptest" @@ -843,3 +844,162 @@ func TestRequestCrawl(t *testing.T) { assert.Error(t, RequestCrawl(ctx, client, relay.URL, ""), "empty hostname must be rejected") assert.Error(t, RequestCrawl(ctx, nil, relay.URL, testHostname), "nil client must be rejected") } + +// testLogger returns a discard logger: the crawl tests assert outcomes via +// call counts, and retry chatter should never pollute test output. +func testLogger(t *testing.T) *slog.Logger { + t.Helper() + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// compressCrawlRetries shrinks the startup-crawl retry schedule for tests. +func compressCrawlRetries(t *testing.T, interval time.Duration, attempts int) { + t.Helper() + prevInterval, prevAttempts := crawlRetryInterval, crawlMaxAttempts + crawlRetryInterval, crawlMaxAttempts = interval, attempts + t.Cleanup(func() { crawlRetryInterval, crawlMaxAttempts = prevInterval, prevAttempts }) +} + +func TestRequestCrawlAll_RetriesTransientFailures(t *testing.T) { + compressCrawlRetries(t, 10*time.Millisecond, 10) + + // The relay fails twice before accepting — the startup race in miniature + // (relay booting / bridge listener not yet up when the relay probes back). + var calls atomic.Int64 + relay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if calls.Add(1) <= 2 { + http.Error(w, "still booting", http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + })) + defer relay.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + RequestCrawlAll(ctx, &http.Client{Timeout: time.Second}, []string{relay.URL}, testHostname, testLogger(t)) + assert.EqualValues(t, 3, calls.Load(), "two failures then the success — no extra attempts after that") +} + +func TestRequestCrawlAll_BudgetExhaustionAndIndependence(t *testing.T) { + compressCrawlRetries(t, time.Millisecond, 3) + + var deadCalls atomic.Int64 + dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + deadCalls.Add(1) + http.Error(w, "nope", http.StatusInternalServerError) + })) + defer dead.Close() + var liveCalls atomic.Int64 + live := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + liveCalls.Add(1) + w.WriteHeader(http.StatusOK) + })) + defer live.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + RequestCrawlAll(ctx, &http.Client{Timeout: time.Second}, []string{dead.URL, live.URL}, testHostname, testLogger(t)) + assert.EqualValues(t, 3, deadCalls.Load(), "the dead relay gets exactly its budget") + assert.EqualValues(t, 1, liveCalls.Load(), "a dead relay must not block or repeat the healthy one") +} + +func TestRequestCrawlAll_ContextCancelStopsRetrying(t *testing.T) { + compressCrawlRetries(t, time.Hour, 100) // only cancellation can end the wait + + relay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "nope", http.StatusInternalServerError) + })) + defer relay.Close() + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + RequestCrawlAll(ctx, &http.Client{Timeout: time.Second}, []string{relay.URL}, testHostname, testLogger(t)) + }() + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("RequestCrawlAll kept waiting after context cancellation") + } +} + +func TestRequestCrawlAll_ValidationErrorNotRetried(t *testing.T) { + // The hour-long interval is the tripwire: if the pre-flight caller-bug + // error were classified as retryable, the goroutine would sleep an hour + // before attempt 2 and the deadline below would fire. + compressCrawlRetries(t, time.Hour, 5) + + var calls atomic.Int64 + relay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + w.WriteHeader(http.StatusOK) + })) + defer relay.Close() + + done := make(chan struct{}) + go func() { + defer close(done) + // Empty hostname is a caller bug (validation error): terminal, no retry. + RequestCrawlAll(context.Background(), &http.Client{Timeout: time.Second}, []string{relay.URL}, "", testLogger(t)) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("a pre-flight validation error was retried instead of being terminal") + } + assert.EqualValues(t, 0, calls.Load(), "validation failures must not reach the wire, let alone retry") +} + +// TestRequestCrawlAll_BadRequestIsRetried pins DELIBERATE behavior: HTTP 4xx +// from a relay is retried like any other wire-level failure. Bigsky answers +// 400 while the describeServer callback race is unresolved (it probes our +// listener before subscribing), so treating 400 as terminal would defeat the +// exact race this retry loop exists for. +func TestRequestCrawlAll_BadRequestIsRetried(t *testing.T) { + compressCrawlRetries(t, time.Millisecond, 10) + + var calls atomic.Int64 + relay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if calls.Add(1) <= 2 { + http.Error(w, "host failed to verify", http.StatusBadRequest) + return + } + w.WriteHeader(http.StatusOK) + })) + defer relay.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + RequestCrawlAll(ctx, &http.Client{Timeout: time.Second}, []string{relay.URL}, testHostname, testLogger(t)) + assert.EqualValues(t, 3, calls.Load(), "two 400s then the success — 4xx must be retried, and success must stop the loop") +} + +// TestRequestCrawlAll_AttemptTimeoutBoundsHangingRelay pins the per-attempt +// cap (crawlAttemptTimeout): a relay that hangs cannot stretch each attempt +// to the HTTP client's full timeout, keeping the documented budget honest. +func TestRequestCrawlAll_AttemptTimeoutBoundsHangingRelay(t *testing.T) { + compressCrawlRetries(t, time.Millisecond, 2) + prevTimeout := crawlAttemptTimeout + crawlAttemptTimeout = 50 * time.Millisecond + t.Cleanup(func() { crawlAttemptTimeout = prevTimeout }) + + release := make(chan struct{}) + var calls atomic.Int64 + relay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + <-release // hang until the test ends + })) + defer relay.Close() + defer close(release) // LIFO: unblock hung handlers before relay.Close waits on them + + start := time.Now() + // The client's own timeout (10s) is far above the attempt cap: only the + // per-attempt context can keep this fast. + RequestCrawlAll(context.Background(), &http.Client{Timeout: 10 * time.Second}, []string{relay.URL}, testHostname, testLogger(t)) + elapsed := time.Since(start) + assert.EqualValues(t, 2, calls.Load(), "a hanging relay still gets its full attempt budget") + assert.Less(t, elapsed, 5*time.Second, "attempts must be bounded by crawlAttemptTimeout, not the client timeout") +} diff --git a/tasks/09-e2e-relay.md b/tasks/09-e2e-relay.md new file mode 100644 index 0000000..dfb778c --- /dev/null +++ b/tasks/09-e2e-relay.md @@ -0,0 +1,72 @@ +# Task 09 — E2E relay: full pipeline Lemmy → Tidepool → relay → Jetstream (~600 LOC) + +## Goal +Put a real atproto relay (indigo BigSky) between the bridge and Jetstream +in the e2e stack, so every scenario's records transit the strictest +consumer that exists: DID resolution against the local PLC, per-commit +signature verification, and initial `getRepo` crawls. Closes the FOLLOWUPS +item "relay `requestCrawl` has never been exercised against a real relay" +and turns the sync surface's correctness claims (CAR exports, proofs, +cursor semantics) into things a hostile consumer verifies on every run. + +## Locked decisions +- **One pipeline, through the relay.** Jetstream consumes from the relay, + not the bridge, so every existing and future scenario exercises relay + validation for free (locked requirement: every state has a full + Lemmy → PDS record → firehose pipeline where applicable). Keep the + direct bridge→Jetstream wiring available behind a compose profile for + debugging, not as the tested path. +- **The bridge's own `RequestCrawlAll` must send the real request.** The + e2e stack runs `ENVIRONMENT=development`, where `RELAY_HOSTS` is + log-only. Add a narrowly-scoped dev override (mirror the + `ALLOW_PRIVATE_FETCH` pattern: dev-only, refused/ignored in production + where sending is already the behavior) so the harness drives the real + code path. The relay's admin API (`POST /admin/pds/requestCrawl`, + Coves' pattern) is an acceptable suite *fallback/bootstrap*, but the + DoD is bridge-originated crawl. +- Local-only invariant holds: loopback-only host binds, nothing ever + contacts a public relay or plc.directory. + +## Deliverables +- `docker-compose.e2e.yml`: relay service — the pinned bigsky image Coves + uses (`ghcr.io/bluesky-social/indigo:bigsky-0a2d4173e6e89e49b448f6bb0a6e1ab58d12b385`, + bump deliberately if needed), its own postgres (or a second database in + an existing one — decide and comment), `BGS_CRAWL_INSECURE_WS=true` + (ws:// upstream) *[verified outcome: this env var does not exist in the + pinned image — `--crawl-insecure-ws` is arg-only and is passed as a + command argument instead; see FOLLOWUPS.md]*, admin key, healthcheck, + loopback-only host port. + **Identity resolution must point at the compose `plc` service** — find + bigsky's actual PLC-host flag/env in the indigo source (verify, don't + guess) or the relay will try the public directory and fail closed. +- Jetstream re-pointed at the relay's firehose; existing scenarios pass + unchanged through the longer pipeline (expect timing budgets to need + loosening — crawl + validation adds latency). +- Bridge sends `requestCrawl` to the relay on startup via the real + `RequestCrawlAll` path (`RELAY_HOSTS=` + the new dev override). +- `tests/e2e`: relay-specific assertions — the bridge host appears in the + relay's crawled-PDS state; repos are listed/crawled; the restart/replay + scenario still proves cursor resume *through the relay*; tombstoned + (`active:false`) repo status observed through the relay where its API + exposes it. +- **Known rock, investigate and document:** bridged handles + (`alice.lemmy.tidepool`) do not resolve in compose DNS — determine + whether bigsky's handle verification failure is non-fatal (likely: + marks handle invalid, keeps repo) and document the posture; add + network aliases only if actually required. +- README stack diagram + runbook update; FOLLOWUPS updates for anything + discovered and deferred. + +## Definition of done +- `make e2e` green from a clean checkout with all scenarios transiting + Lemmy → tidepool → relay → Jetstream. +- Bridge-originated `requestCrawl` observed in relay state/logs (asserted, + not eyeballed). +- Unit suite still green; nothing contacts public infrastructure. + +## References +- `~/Code/coves/docker-compose.dev.yml` relay stanza (lines ~248–296): + image pin, `BGS_CRAWL_INSECURE_WS`, admin requestCrawl curl. +- indigo source (`cmd/bigsky`) for flags/env: PLC host, admin API routes. +- `internal/sync/crawl.go` `RequestCrawlAll`; `cmd/tidepool/main.go` + dev-mode skip. diff --git a/tasks/10-e2e-scenarios.md b/tasks/10-e2e-scenarios.md new file mode 100644 index 0000000..02729b4 --- /dev/null +++ b/tasks/10-e2e-scenarios.md @@ -0,0 +1,55 @@ +# Task 10 — E2E scenario completion: every state, full pipeline (~800 LOC) + +## Goal +Close the FOLLOWUPS "scenario ideas not yet covered" list so every +user-visible state transition the bridge implements is proven end-to-end +(Lemmy → tidepool → relay → Jetstream, task 09's pipeline), with lexicon +validation on the wire. After this task, "the e2e suite passes" means +every materialization arm and lifecycle flow has crossed real +infrastructure. + +## Deliverables (each a scenario in `tests/e2e`, using existing helpers) +1. **Image post**: upload via pictrs → post with image → blob fetched and + stored → `embed.images` (+ nsfw label shape if expressible via Lemmy) + crosses the wire and lexicon-validates. Closes the materializer test + gap ("embed.images never appears on the wire"). +2. **Consent — `#nobridge`**: Lemmy user with `#nobridge` in bio posts → + nothing materializes (and no DID mint); remove the marker → bridging + resumes on profile refresh. Then the reverse: bridged actor adds the + marker → scrub delete-commits observed on the firehose, repo stays + active (reversible posture). +3. **`Delete(Actor)`**: Lemmy account deletion → all their records + scrubbed (delete ops on firehose), repo terminally tombstoned — + `getRepoStatus`/`listRepos` report `active:false`, handle stops + resolving, content endpoints refuse. Assert through the relay where + its API exposes repo state. +4. **Unsubscribe**: `DELETE /admin/communities` → `Undo{Follow}` → new + Lemmy posts in that community produce NO bridge output (bounded + negative assertion), while a still-subscribed community keeps flowing + (positive control in the same window). +5. **Community profile update**: rename/description change in Lemmy → + `community.profile` update event with rkey `self`. +6. **Vote concurrency hammer**: many voters, one post, delivered + concurrently (parallel inbox deliveries / multiple Lemmy users voting + in a burst) → final `getVoteAggregates` exactly correct. First real + exercise of the aggregate-row locking claim beyond unit level. +7. **Suite-end sweep**: after all scenarios, replay the entire firehose + from cursor 0 with an unfiltered listener and assert no event ever + carried a collection outside the four emitted ones and every + create/update lexicon-validates — closes the "events emitted while no + unfiltered listener was subscribed" gap. +- Stretch (skip if it needs its own compose profile): low + `MINT_RATE_PER_MINUTE` variant driving the mint gate end-to-end. + +## Definition of done +- `make e2e` green, all scenarios through the full relay pipeline. +- Every new record shape that crosses the wire is lexicon-validated by + the consuming listener (vetEvent path), not by unit fixtures. +- FOLLOWUPS updated: covered items removed, new discoveries added. + +## References +- `tests/e2e/bridge_test.go`, `helpers.go` (vetEvent, drain, listener + conventions — extend, don't fork). +- FOLLOWUPS.md "E2E harness itself" + materializer/vote test-gap items. +- Lemmy API: pictrs upload, account deletion, community edit endpoints + (0.19.x, `/api/v3`). diff --git a/tasks/11-hardening.md b/tasks/11-hardening.md new file mode 100644 index 0000000..f587b0e --- /dev/null +++ b/tasks/11-hardening.md @@ -0,0 +1,63 @@ +# Task 11 — Hardening: the pre-internet-facing FOLLOWUPS (~1000 LOC) + +## Goal +Work the correctness/security backlog in FOLLOWUPS.md so the bridge can +face the open internet: admission control on every public surface, the +missing firehose account signal, ordering/tombstone gaps, unbounded table +growth, and the small housekeeping items. Verified under the task 09/10 +harness — the strictest pipeline we have. + +## Deliverables (FOLLOWUPS is the checklist; this is the triage) +Security / admission control: +- **Per-signer AND per-IP rate limit on `/inbox`** (top item: + queue-flood DoS via many self-signed identities). Token-bucket, + fail-closed cap, mirroring the votes XRPC limiter's discipline. +- **Connection cap + per-IP rate limit on the public sync surface** + (subscribeRepos, getRepo and friends). +- Seeded-count upper sanity cap (a hostile origin API can't inject + absurd baselines). + +Protocol correctness: +- **`#account{active:false}` firehose frame** on `Delete(Actor)` / + consent revocation, so subscribers purge instead of relying on scrub + delete-commits. Assert it in an e2e scenario (task 10's Delete(Actor) + scenario gains the frame check). +- **Delete-before-Create**: reconcile the README claim ("remembered via + `ap_tombstones`") against the FOLLOWUPS gap ("a Delete arriving before + its object was ever materialized leaves nothing to tombstone") — + verify which is true, close the gap, kill the false doc either way. +- **Automatic Follow re-send** when a subscription stays `pending` past + a threshold (Lemmy first-contact Accept race — currently only the test + harness retries; production operators shouldn't have to). +- Actor-delete / consent revocation scrubs that actor's `vote_events` + rows (consistency with scrub posture elsewhere). + +Unbounded growth / housekeeping: +- Pruners for `ap_tombstones` and superseded/undone `vote_events` rows + (mirror `FIREHOSE_RETENTION` treatment); batch the `PruneEvents` + DELETE while there. +- `MAX_BLOB_BYTES` above 5 MiB: wire the AP client's response cap to the + config value instead of silently clamping. +- Transient media-fetch failure on profile refresh carries forward + existing blobs instead of dropping them; `DeleteActor` scrubs blobs + stored under community DIDs. +- `commitRecord` PutRecord→PutMapping in one tx. +- Delete dead `ingest.NewNoopVotes`; rename + `service_keys.private_key_pem` (it holds sealed ciphertext for the + plc-rotation row) via migration. +- Strict-validation failure metric (production logs-and-writes today — + make it observable; strict-first rollout stays deferred). + +## Definition of done +- Full unit suite + `make e2e` green. +- Every FOLLOWUPS item this task closes is deleted from FOLLOWUPS.md; + anything triaged out is annotated with why. +- New public-surface limits have tests proving both enforcement and + non-interference with legitimate load (the e2e suite itself is the + canary — it must pass under the new limits). + +## References +- FOLLOWUPS.md (Ingestion, Sync surface, Votes, Materializer, + Storage/housekeeping sections). +- `internal/votes/xrpc.go` limiter (pattern to reuse), LOOP_STATE task + 06/07 notes (queue fencing, outcome contract — do not violate). diff --git a/tasks/12-perf-scale.md b/tasks/12-perf-scale.md new file mode 100644 index 0000000..0b43612 --- /dev/null +++ b/tasks/12-perf-scale.md @@ -0,0 +1,48 @@ +# Task 12 — Perf & scale: big-community readiness (~800 LOC) + +## Goal +Remove the known scaling cliffs before any big community hits them — and +as the explicit prerequisite for ever revisiting votes-as-records +(FOLLOWUPS "Design revisits": write amplification is only viable after +this task). Every change here is behavior-preserving; the e2e suite and +golden-value tests (deterministic TIDs, at-uris) must not move. + +## Deliverables +- **Per-DID MST tree cache**: `PutRecord` is currently O(repo size) with + one SELECT per node, full-tree, per commit. Cache the tree per DID with + invalidation tied to the commit path (single-writer discipline + the + global advisory lock make this tractable). Benchmark before/after with + a realistic big-repo fixture (thousands of records) and record numbers + in the commit message. +- **`getRepo` memory + reachable-set**: stream the CAR instead of + buffering it whole; export the reachable set from the current commit + rather than every historical block. Coordinate with GetRecord's + read-consistency dependence on append-only `blocks` (LOOP_STATE task + 03/04 notes) — reachable-set-only export must not break proof reads. +- **`blocks` GC**: prune blocks unreachable from any live commit, as an + explicit background sweep with a retention guard. This is the + load-bearing append-only table — design the invariant first (what do + GetRecord/getRepo/replay need?), write it down in the code, then + implement to it. If a safe GC needs the sync `since`/diff-export work, + say so and defer that half explicitly. +- **`ClaimNext` O(N) scan** when one community's queue backs up behind a + failing event: bound the scan (indexed skip / per-key cursor) without + breaking the per-community serial-ordering guarantee or the fencing + contract. +- Cheap wins while in the area: `getRepo` `since` param (diff export) if + the reachable-set work makes it nearly free — otherwise leave the + documented full-CAR fallback. + +## Definition of done +- Full unit suite + `make e2e` green; golden TID/at-uri tests untouched. +- Before/after benchmarks for PutRecord (big repo) and getRepo (memory) + recorded in the commit message. +- FOLLOWUPS updated (items closed/annotated); the votes-as-records + design-revisit note updated to reflect which prerequisites now hold. + +## References +- FOLLOWUPS.md Sync surface + Storage sections; LOOP_STATE task 03/04 + notes (commit serialization, advisory locks, why blocks is + append-only). +- `internal/repo/` (MST load path, ExportCAR), `internal/store` + (ClaimNext). diff --git a/tests/e2e/bridge_test.go b/tests/e2e/bridge_test.go index 61455cd..6be8d56 100644 --- a/tests/e2e/bridge_test.go +++ b/tests/e2e/bridge_test.go @@ -47,11 +47,23 @@ func TestSubscribe_CommunityProfileOnFirehose(t *testing.T) { } } -// Scenario 2: a Lemmy user shares a link → actor.profile first, then -// community.post in the COMMUNITY's repo with author = the user's DID -// (PLAN.md locked decision 3), in that order — and the shared url crosses -// the wire as an embed.external whose uri survives byte-identical (the -// classic CBOR→JSON breakage point). +// Scenario 2: a Lemmy user shares a link → actor.profile AND community.post +// in the COMMUNITY's repo with author = the user's DID (PLAN.md locked +// decision 3) — and the shared url crosses the wire as an embed.external +// whose uri survives byte-identical (the classic CBOR→JSON breakage point). +// +// Ordering caveat (task 09 discovery): the bridge emits the author's +// profile strictly before the post on its OWN firehose, but the two records +// live in different repos (author vs community) and bigsky indexes its +// inbound stream with a parallel scheduler keyed by repo DID (indigo +// events/schedulers/parallel, 100 workers) — per-repo order survives the +// relay, CROSS-repo order does not. In practice the post usually overtakes +// the profile: the community DID is already known to the relay while the +// author's first-ever event costs a PLC resolution + handle verification. +// So this scenario asserts presence + the author linkage, not arrival +// order — and the Coves AppView, which consumes through relay +// infrastructure too, cannot rely on profile-before-post either +// (FOLLOWUPS.md "Relay pipeline"). func TestPost_ActorProfileThenPost(t *testing.T) { h := newHarness(t) community, sub := setupSubscribedCommunity(t, h, "post") @@ -70,24 +82,12 @@ func TestPost_ActorProfileThenPost(t *testing.T) { post := user.createLinkPost(t, community.ID, title, "first bridged post", linkURL) t.Logf("created lemmy post %d (%s)", post.ID, post.APID) - // Both events funnel through one listener; the FIRST match for this - // scenario must be the author's profile — the AppView rejects posts - // whose author isn't indexed yet, so emission order is load-bearing. - first := l.await("actor.profile or community.post for "+username, func(e *jsEvent) bool { - switch e.Commit.Collection { - case colActorProfile: - name, _ := fieldOf(e.Commit.Record, "displayName") - return name == username - case colPost: - got, _ := fieldOf(e.Commit.Record, "title") - return e.Did == sub.DID && got == title - } - return false + // Await both events in either order (await buffers non-matches, so a + // post that overtook the profile through the relay is not lost). + profileEv := l.await("actor.profile for "+username, func(e *jsEvent) bool { + name, _ := fieldOf(e.Commit.Record, "displayName") + return e.Commit.Collection == colActorProfile && name == username }) - if first.Commit.Collection != colActorProfile { - t.Fatalf("first event was %s — actor.profile must be emitted before the post", first) - } - profileEv := first if profileEv.Commit.RKey != rkeySelf { t.Errorf("actor.profile rkey = %q, want %q", profileEv.Commit.RKey, rkeySelf) } @@ -127,8 +127,9 @@ func TestComments_StrongRefsResolve(t *testing.T) { title := "Comment thread " + h.suffix post := user.createPost(t, community.ID, title, "root post") - // The author's profile precedes their first content (scenario 2 asserts - // the ordering; here it pins the author's DID). + // The author's profile pins the author's DID. Arrival order vs the post + // is relay-dependent (see scenario 2) — await's buffering makes these + // sequential awaits order-tolerant. profileEv := l.await("author actor.profile", func(e *jsEvent) bool { name, _ := fieldOf(e.Commit.Record, "displayName") return e.Commit.Collection == colActorProfile && name == username @@ -355,19 +356,21 @@ func TestBackfill_PreexistingPosts(t *testing.T) { }) // All three historical posts must materialize (accept triggers the - // outbox backfill). Order among the posts is newest-first outbox order — - // only presence is asserted — but each post's AUTHOR must have hit the - // firehose as an actor.profile before the post itself (the Coves AppView - // rejects posts by unindexed authors, so backfill emission order is as - // load-bearing as the live path's). + // outbox backfill), and every post's AUTHOR must appear on the firehose + // as an actor.profile. The bridge emits each profile strictly before + // its posts, but profile and post live in different repos and the relay + // preserves only per-repo order (scenario 2's caveat) — so authorship + // is asserted as presence (each author's profile arrives), not arrival + // order. profileDIDs := map[string]bool{} + postAuthors := map[string]string{} // author DID → one of their post titles votedURI := "" remaining := len(titles) for remaining > 0 { ev := l.await(fmt.Sprintf("backfilled post or author profile (%d posts to go)", remaining), func(e *jsEvent) bool { switch e.Commit.Collection { case colActorProfile: - return true // consume every profile to track author-first ordering + return true // consume every profile to track the author set case colPost: title, _ := fieldOf(e.Commit.Record, "title") return e.Did == sub.DID && e.Commit.Operation == opCreate && titles[title] @@ -379,15 +382,23 @@ func TestBackfill_PreexistingPosts(t *testing.T) { continue } title := recordField(t, ev.Commit.Record, "title") - if got := recordField(t, ev.Commit.Record, "author"); !profileDIDs[got] { - t.Errorf("backfilled post %q emitted before its author's actor.profile (author %s)", title, got) - } + postAuthors[recordField(t, ev.Commit.Record, "author")] = title if title == votedTitle { votedURI = ev.atURI() } delete(titles, title) remaining-- } + // Any author whose profile hasn't been consumed yet must still be in + // flight (or already buffered by await): wait for each explicitly. + for author, title := range postAuthors { + if profileDIDs[author] { + continue + } + l.await(fmt.Sprintf("actor.profile for the author of %q (%s)", title, author), func(e *jsEvent) bool { + return e.Commit.Collection == colActorProfile && e.Did == author + }) + } // The seeded baseline shows through the side channel: 2 up (author // auto-like + the pre-subscribe vote), 0 down. @@ -561,7 +572,12 @@ func TestBurst_ConcurrentIngestionExactlyOnce(t *testing.T) { // Account for EVERY commit on the two communities' repos by // (did, collection/rkey) — an update sneaking in after a create is a - // duplicate commit on that record, not a separate event. + // duplicate commit on that record, not a separate event. Collection is + // drain-based, not await-based: drain sees every live event exactly + // once, so the accounting cannot double-count (await predicates must + // stay pure now that non-matches are buffered and rescanned), and it is + // inherently order-agnostic — which the relay's cross-repo reordering + // demands anyway. seen := map[string]int{} keyTitle := map[string]string{} count := func(e *jsEvent) { @@ -580,15 +596,23 @@ func TestBurst_ConcurrentIngestionExactlyOnce(t *testing.T) { // Every post arrives… matched := map[string]bool{} + deadline := time.Now().Add(burstTimeout) for len(matched) < len(titles) { - ev := l.await(fmt.Sprintf("burst post (%d to go)", len(titles)-len(matched)), func(e *jsEvent) bool { - count(e) - title, _ := fieldOf(e.Commit.Record, "title") - return e.Commit.Operation == opCreate && - (e.Did == subA.DID || e.Did == subB.DID) && - titles[title] && !matched[title] - }) - matched[recordField(t, ev.Commit.Record, "title")] = true + if time.Now().After(deadline) { + t.Fatalf("only %d/%d burst posts arrived within %s", len(matched), len(titles), burstTimeout) + } + for _, ev := range l.drain(time.Second) { + count(ev) + if ev.Kind != kindCommit || ev.Commit == nil || ev.Commit.Operation != opCreate { + continue + } + if ev.Did != subA.DID && ev.Did != subB.DID { + continue + } + if title, ok := fieldOf(ev.Commit.Record, "title"); ok && titles[title] { + matched[title] = true + } + } } // …and exactly once: nothing trailing, no second commit on any rkey. for _, ev := range l.drain(4 * time.Second) { diff --git a/tests/e2e/helpers.go b/tests/e2e/helpers.go index 5530013..cb507b8 100644 --- a/tests/e2e/helpers.go +++ b/tests/e2e/helpers.go @@ -2,8 +2,10 @@ // Package e2e drives the docker-compose.e2e.yml stack end to end: a real // Lemmy (debug build, plain-HTTP federation) federating with Tidepool, a -// real did:plc directory backing DID minting, and a real Jetstream decoding -// the bridge's subscribeRepos firehose. +// real did:plc directory backing DID minting, a real BigSky relay crawling +// the bridge (DID resolution against the local PLC, per-commit signature +// verification), and a real Jetstream decoding the RELAY's firehose — every +// event the suite consumes has therefore survived relay validation. // // Coves-style: E2E tests test REAL infrastructure, not mocks. Run them with // `make e2e` (compose up --build → wait for health → go test -tags e2e → @@ -50,20 +52,41 @@ func envOr(name, fallback string) string { func tidepoolURL() string { return envOr("TIDEPOOL_E2E_URL", "http://localhost:8092") } func lemmyURL() string { return envOr("LEMMY_E2E_URL", "http://localhost:8541") } +func relayURL() string { return envOr("RELAY_E2E_URL", "http://localhost:2480") } func jetstreamURL() string { return envOr("JETSTREAM_E2E_URL", "ws://localhost:6028") } -func adminToken() string { return envOr("TIDEPOOL_E2E_ADMIN_TOKEN", "e2e-admin-token") } +func adminToken() string { return envOr("TIDEPOOL_E2E_ADMIN_TOKEN", "e2e-admin-token") } +func relayAdminKey() string { return envOr("RELAY_E2E_ADMIN_KEY", "e2e-relay-admin-key") } +func bridgeHostname() string { return envOr("TIDEPOOL_E2E_HOSTNAME", "tidepool") } // stackTimeout bounds the initial wait-for-healthy loop. Container startup // (Lemmy migrations, PLC boot) can be slow on a cold machine; `make e2e` // already waited for compose health, so this is usually instant. const stackTimeout = 5 * time.Minute -// eventTimeout bounds one wait for a federation → firehose → Jetstream -// round trip. LEMMY_TEST_FAST_FEDERATION makes deliveries near-instant, but -// leave generous slack for slow CI. -const eventTimeout = 90 * time.Second +// eventTimeout bounds one wait for a federation → bridge firehose → relay → +// Jetstream round trip. LEMMY_TEST_FAST_FEDERATION makes Lemmy deliveries +// near-instant, but the relay hop adds real work per event — first sight of +// a DID costs the relay a PLC resolution plus handle verification, and on +// Apple Silicon the (amd64-only) relay image runs emulated — so this budget +// is deliberately looser than the pre-relay 90s. +const eventTimeout = 120 * time.Second + +// burstTimeout bounds scenario 8's whole 12-post burst arriving (a single +// shared budget: the collection loop is drain-based rather than one +// eventTimeout per await). +const burstTimeout = 3 * time.Minute + +// crawlTimeout bounds waiting for the bridge's startup requestCrawl to land +// in the relay's PDS registry. The announcement retries inside the bridge +// (internal/sync/crawl.go: 24 attempts × 5s interval, 10s per-attempt cap) +// because the relay validates the hostname by calling back into the bridge's +// describeServer — the budget must cover the realistic retry window (attempts +// fail near-instantly in this stack: connection-refused or bigsky's fast 400) +// plus the relay's first subscribe. A stack where every attempt hangs to the +// 10s cap is broken, and failing at this deadline is then the right signal. +const crawlTimeout = 3 * time.Minute // ── Firehose vocabulary ──────────────────────────────────────────────────── @@ -88,6 +111,12 @@ const ( rkeySelf = "self" ) +// pendingSoftCap is the listener's early-warning threshold for the pending +// buffer (await's consumed-but-unmatched events): crossing it t.Logf's once +// as a leak signal, without failing — a busy shared stack can legitimately +// buffer plenty of foreign traffic. +const pendingSoftCap = 512 + // expectedCollections is the complete set of record collections that may // legally appear on the firehose. Anything else — vote records above all // (PLAN.md locked decision 7: votes NEVER become records) — is a bug, and @@ -101,7 +130,7 @@ var expectedCollections = map[string]bool{ // ── Harness ──────────────────────────────────────────────────────────────── -// harness bundles the three service endpoints plus a logged-in Lemmy admin. +// harness bundles the service endpoints plus a logged-in Lemmy admin. type harness struct { http *http.Client admin *lemmyClient // Lemmy admin (setup credentials from lemmy.hjson) @@ -177,6 +206,9 @@ func waitForStack() error { {"lemmy /api/v3/site", func() error { return probeHTTP(client, lemmyURL()+"/api/v3/site") }}, + {"relay /xrpc/_health", func() error { + return probeHTTP(client, relayURL()+"/xrpc/_health") + }}, {"jetstream /subscribe", func() error { u := jetstreamURL() + "/subscribe?cursor=" + fmt.Sprint(time.Now().UnixMicro()) conn, _, err := websocket.DefaultDialer.Dial(u, nil) @@ -623,6 +655,128 @@ func (h *harness) getVoteAggregates(t *testing.T, uris ...string) map[string]vot return byURI } +// ── Relay (BigSky) clients ───────────────────────────────────────────────── + +// relayPDS is the slice of bigsky's /admin/pds/list response the suite +// asserts on (an enriched gorm models.PDS; Go field names, no json tags). +type relayPDS struct { + Host string `json:"Host"` + Registered bool `json:"Registered"` + HasActiveConnection bool `json:"HasActiveConnection"` + RepoCount int64 `json:"RepoCount"` +} + +// relayPDSList reads the relay's crawled-PDS registry via its admin API. +// Error-returning (not Fatalf) because its only callers are poll loops: one +// transient admin-API hiccup must not kill a minutes-long wait — the loop +// logs the error and Fatalf's at its own deadline. +func (h *harness) relayPDSList() ([]relayPDS, error) { + req, err := http.NewRequest(http.MethodGet, relayURL()+"/admin/pds/list", nil) + if err != nil { + return nil, fmt.Errorf("relay pds/list: build request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+relayAdminKey()) + resp, err := h.http.Do(req) + if err != nil { + return nil, fmt.Errorf("relay pds/list: %w", err) + } + defer func() { _ = resp.Body.Close() }() + raw, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("relay pds/list: read body: %w", err) + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("relay pds/list: status %d: %s", resp.StatusCode, truncate(raw, 300)) + } + var out []relayPDS + if err := json.Unmarshal(raw, &out); err != nil { + return nil, fmt.Errorf("relay pds/list: decode: %w (%s)", err, truncate(raw, 300)) + } + return out, nil +} + +// relayListReposPageCap bounds relayListRepos pagination: the e2e relay +// holds a few dozen repos (≪ one 500-repo page), so hitting the cap — or a +// cursor that stops advancing — means bigsky's paging is broken, and the +// walk must error out rather than spin until the 20m test timeout. +const relayListReposPageCap = 100 + +// relayListRepos walks the relay's public com.atproto.sync.listRepos and +// returns did → head. Tombstoned/taken-down repos are filtered out by +// bigsky itself, so absence after a takedown is the observable signal. +// Error-returning for the same poll-loop reason as relayPDSList. +func (h *harness) relayListRepos() (map[string]string, error) { + repos := map[string]string{} + cursor := "" + for page := 0; ; page++ { + if page >= relayListReposPageCap { + return nil, fmt.Errorf("relay listRepos: still paginating after %d pages (cursor %q) — runaway paging", relayListReposPageCap, cursor) + } + u := relayURL() + "/xrpc/com.atproto.sync.listRepos?limit=500" + if cursor != "" { + u += "&cursor=" + url.QueryEscape(cursor) + } + resp, err := h.http.Get(u) + if err != nil { + return nil, fmt.Errorf("relay listRepos: %w", err) + } + raw, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if err != nil { + return nil, fmt.Errorf("relay listRepos: read body: %w", err) + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("relay listRepos: status %d: %s", resp.StatusCode, truncate(raw, 300)) + } + var out struct { + Cursor string `json:"cursor"` + Repos []struct { + Did string `json:"did"` + Head string `json:"head"` + } `json:"repos"` + } + if err := json.Unmarshal(raw, &out); err != nil { + return nil, fmt.Errorf("relay listRepos: decode: %w", err) + } + for _, r := range out.Repos { + repos[r.Did] = r.Head + } + if out.Cursor == "" { + return repos, nil + } + if out.Cursor == cursor { + return nil, fmt.Errorf("relay listRepos: cursor %q did not advance between pages", cursor) + } + cursor = out.Cursor + } +} + +// relayGetLatestCommit reads the relay's view of a repo head. Returns an +// error (rather than failing) so poll loops can wait out the relay's +// asynchronous indexing of a commit it just received. +func (h *harness) relayGetLatestCommit(did string) (cid, rev string, err error) { + resp, err := h.http.Get(relayURL() + "/xrpc/com.atproto.sync.getLatestCommit?did=" + url.QueryEscape(did)) + if err != nil { + return "", "", err + } + defer func() { _ = resp.Body.Close() }() + raw, err := io.ReadAll(resp.Body) + if err != nil { + return "", "", err + } + if resp.StatusCode != http.StatusOK { + return "", "", fmt.Errorf("relay getLatestCommit(%s): status %d: %s", did, resp.StatusCode, truncate(raw, 200)) + } + var out struct { + CID string `json:"cid"` + Rev string `json:"rev"` + } + if err := json.Unmarshal(raw, &out); err != nil { + return "", "", fmt.Errorf("relay getLatestCommit(%s): decode: %w", did, err) + } + return out.CID, out.Rev, nil +} + // ── Jetstream WebSocket listener ─────────────────────────────────────────── // jsEvent is Jetstream's JSON event shape (kind "commit" only — the bridge @@ -664,6 +818,30 @@ type jsListener struct { done chan struct{} // closed by readLoop's defer: goroutine exited once sync.Once + // pending holds events consumed by an await that did not match its + // predicate, for rescanning by LATER awaits. Load-bearing since the + // relay entered the pipeline: bigsky indexes its inbound firehose with a + // parallel scheduler keyed by repo DID (indigo events/schedulers/ + // parallel, 100 workers), so per-repo event order is preserved but + // CROSS-repo order is not — an author's actor.profile (author repo) and + // their post (community repo) may legally swap on the relay's output. + // Sequential awaits would otherwise silently discard the reordered + // event and burn a full eventTimeout. Only touched from the test + // goroutine (await/drain), never from readLoop. + pending []*jsEvent + + // lastRev tracks the newest commit rev THIS listener has consumed per + // repo DID (revs are TIDs: strictly increasing per repo). The pending + // buffer made the suite order-tolerant, but PLAN.md locked decision 3's + // per-repo discipline — and the FOLLOWUPS "Relay pipeline" carve-out — + // lean on bigsky preserving PER-repo order even while shuffling repos + // against each other; vetEvent asserts that property on every consumed + // event so a relay/bridge ordering regression cannot hide behind the + // listener's own tolerance. Per-listener (like pending): overlapping + // listeners each see a per-repo-ordered stream of their own. Only + // touched from the test goroutine. + lastRev map[string]string + mu sync.Mutex readErr error // readLoop's terminal error, nil on deliberate close } @@ -727,11 +905,12 @@ func (h *harness) newListener(t *testing.T, cursorMicros int64, collections ...s time.Sleep(2 * time.Second) } l := &jsListener{ - t: t, - conn: conn, - events: make(chan *jsEvent, 1024), - closed: make(chan struct{}), - done: make(chan struct{}), + t: t, + conn: conn, + events: make(chan *jsEvent, 1024), + closed: make(chan struct{}), + done: make(chan struct{}), + lastRev: map[string]string{}, } go l.readLoop() t.Cleanup(l.close) @@ -797,7 +976,10 @@ func (l *jsListener) close() { // an immediate failure, no matter which scenario's await/drain window it // lands in; // - every create/update record must validate against the vendored Coves -// lexicons (deletes carry no record). +// lexicons (deletes carry no record); +// - commit revs must be strictly increasing per repo DID (see lastRev): +// the relay guarantees per-repo order even though cross-repo order is +// lost, and this is the one place every consumed event passes through. func (l *jsListener) vetEvent(ev *jsEvent) { l.t.Helper() if ev.Kind != kindCommit || ev.Commit == nil { @@ -806,17 +988,39 @@ func (l *jsListener) vetEvent(ev *jsEvent) { if !expectedCollections[ev.Commit.Collection] { l.t.Fatalf("unexpected collection on firehose: %s — only community/actor profiles, posts, and comments may ever appear (votes never become records)", ev) } + if prev, ok := l.lastRev[ev.Did]; ok && ev.Commit.Rev <= prev { + l.t.Fatalf("per-repo rev order violated on firehose: %s has rev %q after rev %q — bigsky preserves per-repo commit order, so this is a relay/bridge ordering bug", + ev, ev.Commit.Rev, prev) + } + l.lastRev[ev.Did] = ev.Commit.Rev if op := ev.Commit.Operation; op == opCreate || op == opUpdate { validateLexicon(l.t, ev.Commit) } } -// await returns the first commit event matching pred, failing the test -// after eventTimeout. Non-matching events are vetted (vetEvent), logged, -// and kept out of the way (each scenario matches on its own -// community/author to stay independent of concurrent traffic). +// await returns the first commit event matching pred — scanning events an +// earlier await consumed-but-buffered FIRST (see pending: the relay does +// not preserve cross-repo ordering), then the live stream — failing the +// test after eventTimeout. Non-matching events are vetted (vetEvent), +// logged, and buffered for later awaits (each scenario matches on its own +// community/author to stay independent of concurrent traffic). A matched +// buffered event is removed, so repeat awaits with the same predicate +// consume distinct events. +// +// Because buffered events are re-tested by later awaits' predicates, +// predicates must be PURE — accounting sweeps belong in drain loops (which +// return every consumed event exactly once), not in predicate side effects. func (l *jsListener) await(desc string, pred func(*jsEvent) bool) *jsEvent { l.t.Helper() + for i, ev := range l.pending { + if pred(ev) { + copy(l.pending[i:], l.pending[i+1:]) + l.pending[len(l.pending)-1] = nil // let the shifted-out *jsEvent GC + l.pending = l.pending[:len(l.pending)-1] + l.t.Logf("await %s: matched buffered %s", desc, ev) + return ev + } + } timer := time.NewTimer(eventTimeout) defer timer.Stop() for { @@ -827,11 +1031,19 @@ func (l *jsListener) await(desc string, pred func(*jsEvent) bool) *jsEvent { return nil } l.vetEvent(ev) - if ev.Kind == kindCommit && ev.Commit != nil && pred(ev) { + if ev.Kind != kindCommit || ev.Commit == nil { + l.t.Logf("await %s: skipping %s", desc, ev) + continue + } + if pred(ev) { l.t.Logf("await %s: matched %s", desc, ev) return ev } - l.t.Logf("await %s: skipping %s", desc, ev) + l.pending = append(l.pending, ev) + if len(l.pending) == pendingSoftCap { + l.t.Logf("await %s: pending buffer reached %d unmatched events — a scenario may be leaking unmatched traffic (soft warning, not fatal)", desc, pendingSoftCap) + } + l.t.Logf("await %s: buffering %s", desc, ev) case <-timer.C: l.t.Fatalf("await %s: no matching jetstream event within %s", desc, eventTimeout) return nil @@ -839,15 +1051,24 @@ func (l *jsListener) await(desc string, pred func(*jsEvent) bool) *jsEvent { } } -// drain collects everything that arrives within d (for negative -// assertions: "nothing else showed up"). A dead reader would make every -// negative assertion pass vacuously, so an unexpectedly closed channel is -// fatal — silence must mean "connected and nothing arrived". +// drain returns everything the listener has consumed-but-not-matched so +// far (the pending buffer, cleared here so repeated drains cannot return +// an event twice) plus everything that arrives LIVE within d — for +// negative assertions ("nothing else showed up") and order-agnostic +// accounting sweeps. Including pending is load-bearing, not a convenience: +// a drain running AFTER an await on the same listener would otherwise +// silently miss events that await consumed and buffered, turning negative +// assertions vacuous. It cannot double-count either — pending events were +// vetted at consumption but never matched/counted by any await. A dead +// reader would make every negative assertion pass vacuously, so an +// unexpectedly closed channel is fatal — silence must mean "connected and +// nothing arrived". func (l *jsListener) drain(d time.Duration) []*jsEvent { l.t.Helper() timer := time.NewTimer(d) defer timer.Stop() - var out []*jsEvent + out := l.pending + l.pending = nil for { select { case ev, ok := <-l.events: diff --git a/tests/e2e/relay_test.go b/tests/e2e/relay_test.go new file mode 100644 index 0000000..2115030 --- /dev/null +++ b/tests/e2e/relay_test.go @@ -0,0 +1,134 @@ +//go:build e2e + +package e2e + +// Task 09: relay-specific assertions. Every OTHER scenario already +// exercises the relay implicitly — Jetstream consumes the relay's firehose, +// so no event reaches any listener without surviving bigsky's DID +// resolution (against the local PLC), per-commit signature verification, +// and indexing. The tests here assert the relay-side STATE that implicit +// transit doesn't: the bridge present in the crawled-PDS registry (the +// bridge-originated requestCrawl, task 09's definition of done) and repos +// listed/served by the relay's own sync surface. +// +// Posture notes (verified against the pinned image's source, indigo bgs/): +// - Handle verification: the compose stack points bigsky's trial-host +// resolver at the bridge (HANDLE_RESOLVER_HOSTS=tidepool), so bridged +// handles — DNS-invisible names like alice.lemmy.tidepool — resolve via +// the bridge's Host-header-keyed /.well-known/atproto-did and actually +// verify. Had that failed, bigsky's createExternalUser treats handle +// failure as NON-fatal (repo kept, handle marked invalid), so this is +// belt on top of a safe default. +// - Tombstones: bigsky FILTERS tombstoned/taken-down repos out of +// listRepos and serves no getRepoStatus, and the bridge does not emit +// #account frames yet (deferred to task 11) — so consent-revoked +// (active:false) repo status is NOT yet observable through the relay. +// Tracked in FOLLOWUPS.md. + +import ( + "testing" + "time" +) + +// TestRelay_BridgeCrawledBySelfAnnouncement proves the bridge-originated +// requestCrawl: the bridge's startup RequestCrawlAll (RELAY_HOSTS + +// ALLOW_DEV_REQUEST_CRAWL, the REAL production code path) is the only +// requestCrawl sender in the stack — the compose bootstrap merely raises +// the relay's new-PDS-per-day limit from its fresh-database 0 — so the +// bridge hostname appearing in the relay's PDS registry as registered, +// with a live subscribeRepos connection, is the announcement observed +// end-to-end rather than eyeballed in logs. +func TestRelay_BridgeCrawledBySelfAnnouncement(t *testing.T) { + h := newHarness(t) + + deadline := time.Now().Add(crawlTimeout) + var last []relayPDS + var lastErr error + for { + // Transient admin-API errors must not kill the poll — log and keep + // waiting; only the deadline is fatal (with the last error shown). + pdsList, err := h.relayPDSList() + if err != nil { + lastErr = err + t.Logf("relay pds/list poll: %v (retrying)", err) + } else { + last = pdsList + for _, pds := range last { + if pds.Host != bridgeHostname() { + continue + } + if !pds.Registered { + break // present but unregistered: keep polling + } + if !pds.HasActiveConnection { + break // crawl accepted, subscription still dialing + } + t.Logf("relay crawled the bridge: host=%s registered=%v active=%v repos=%d", + pds.Host, pds.Registered, pds.HasActiveConnection, pds.RepoCount) + return + } + } + if time.Now().After(deadline) { + t.Fatalf("bridge host %q never became a registered+connected PDS on the relay within %s (registry: %+v, last poll error: %v)", + bridgeHostname(), crawlTimeout, last, lastErr) + } + time.Sleep(2 * time.Second) + } +} + +// TestRelay_RepoCrawledAndServed drives one community through the full +// pipeline and then asserts the RELAY's own view of the resulting repo: +// it appears in the relay's listRepos with a head, and the relay serves +// getLatestCommit for it — i.e. the relay didn't just forward frames, it +// built (and can serve) validated repo state from the bridge's commits. +func TestRelay_RepoCrawledAndServed(t *testing.T) { + h := newHarness(t) + + cursor := cursorNow() + l := h.newListener(t, cursor, colCommunityProfile) + + _, sub := setupSubscribedCommunity(t, h, "rly") + + // The event arriving at Jetstream proves the relay accepted and + // re-emitted the commit (Jetstream's upstream IS the relay). + ev := l.await("community.profile through the relay", func(e *jsEvent) bool { + return e.Did == sub.DID && e.Commit.Collection == colCommunityProfile && + e.Commit.Operation == opCreate + }) + + // The relay indexes asynchronously; poll its sync surface for the repo. + // Transient listRepos errors are logged and waited out — only the + // deadline is fatal. + deadline := time.Now().Add(eventTimeout) + var lastErr error + for { + repos, err := h.relayListRepos() + if err != nil { + lastErr = err + t.Logf("relay listRepos poll: %v (retrying)", err) + } else if head, ok := repos[sub.DID]; ok { + if head == "" { + t.Fatalf("relay listRepos shows %s with an empty head", sub.DID) + } + break + } + if time.Now().After(deadline) { + t.Fatalf("repo %s never appeared in the relay's listRepos within %s (last poll error: %v)", sub.DID, eventTimeout, lastErr) + } + time.Sleep(time.Second) + } + + // getLatestCommit must serve the repo, and — since the profile commit + // has demonstrably been processed (it reached Jetstream) — at a rev no + // older than that event's. + cid, rev, err := h.relayGetLatestCommit(sub.DID) + if err != nil { + t.Fatalf("relay getLatestCommit(%s): %v", sub.DID, err) + } + if cid == "" || rev == "" { + t.Fatalf("relay getLatestCommit(%s) returned empty cid/rev (%q/%q)", sub.DID, cid, rev) + } + if rev < ev.Commit.Rev { + t.Errorf("relay head rev %q is OLDER than the profile event's rev %q — relay state lagging its own emissions", rev, ev.Commit.Rev) + } +} -- 2.51.2