From 414d0d8e2a0b314bb5d81d75e97c1476fe2024ce Mon Sep 17 00:00:00 2001 From: Bretton Date: Thu, 15 Oct 2026 14:55:14 -0700 Subject: [PATCH] feat(identity): refresh bridged handles and serve fediverse origins Tidepool is renaming every bridged actor to ..ap.tdpl.io and announcing it with #identity events. Coves trusted the event's handle for users, ignored identity events for communities, and had no way to show or look up a bridged account's fediverse address. This stores only freshly verified handles, serializes refreshes per DID, adds a sweep for rows that missed their event, and adds origin fields to profiles and views. Changes: - User #identity events ignore the event's handle: the consumer purges the DID, resolves it through the new uncached resolver (identity.NewUncachedResolver), verifies the handle and writes it. Unknown DIDs cost one indexed read. A per-DID timeout debounce (4096 entries) keeps a hanging handle domain from stalling the user lane. - The community consumer handles #identity events for indexed communities through the same RefreshIdentityHandle path, checks the new handle with communities.ValidateHandle and the stored hostedBy domain, and writes it with a new Repository.UpdateHandle. Failures leave the row unchanged. - Refreshes of one DID are serialized by a session-scoped Postgres advisory lock on a dedicated pool (8 connections, separate from the application pool), held from the DID purge until the handle write. The stored handle is re-read inside the lock. A busy lock (25s lock_timeout) returns ErrIdentityRefreshBusy as a deferred redrive; a missing lock fails closed. - New cmd/reconcile-handles: selects user and community rows on trusted bridge hosts, optionally clears identity_cache, refreshes each DID under the shared lock, and reports still-differing rows with reason codes. -dry-run is read-only. Exit status is non-zero while any row differs. Runbook in docs/PRODUCTION_DEPLOY.md. - actor.profile and profileViewDetailed gain origin, originUsername, originUrl and originSoftware; community.profile, communityView and communityViewDetailed gain originUrl and originSoftware. They are stored only after a fresh, document-first verification on a trusted bridge PDS with the handle under the bridge's registrable domain, and are sanitized (UTF-8 byte limits, origin pair set together, https originUrl on the same host, originSoftware lowercased). Migrations users_profile_origin and communities_origin_display add the columns. - Post and comment authorView gains origin, originUsername and originSoftware through postViewSelectColumns/scanPostView and comment author hydration. actor.defs#profileView gains all four fields, used for notification actors and recentUpvoters. Native users and deleted-comment placeholders carry none of the keys. - The actor parameter of actor.getProfile, getPosts and getComments accepts name@host and @name@host. GetDIDByOriginUsername matches exactly, then by a unique case-insensitive match; more than one match is 409 AmbiguousActor. The scoped form only reads the database. Migration users_origin_username_index adds the lookup indexes. - docs/BRIDGED_IDENTITY.md describes the design, rollout and known open items; SECURITY_MODEL.md, TEST_ARCHITECTURE.md and LEXICON_PUBLISHING.md are updated. Deploy and lexicon notes: Coves must deploy before Tidepool renames anything and before Tidepool writes origin fields. Tidepool must re-sync its vendored lexicons from Coves main before its branch merges. None of the lexicon changes are published; run goat lex publish --update per docs/LEXICON_PUBLISHING.md. The actor parameter change (format at-identifier dropped for a plain string, maxLength 2048) is a recorded input-loosening exception. After Tidepool's rename job, run reconcile-handles with -clear-identity-cache and re-run until it exits 0. The trusted-bridge origin admission path is covered by integration tests but has not been exercised end to end against Tidepool. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../helpers_integration_test.go | 336 +++++++ .../identity_lock_race_integration_test.go | 333 +++++++ cmd/reconcile-handles/main.go | 216 +++++ cmd/reconcile-handles/open.go | 95 ++ .../reconcile_integration_test.go | 85 ++ .../reconcile_outcomes_integration_test.go | 285 ++++++ .../run_stderr_integration_test.go | 64 ++ cmd/reconcile-handles/run_test.go | 74 ++ cmd/reconcile-handles/sweep.go | 441 +++++++++ cmd/reconcile-handles/sweep_test.go | 95 ++ cmd/server/consumers.go | 10 + cmd/server/database.go | 43 + cmd/server/main.go | 12 +- .../uncached_identity_resolver_wiring_test.go | 105 ++ cmd/server/wiring.go | 23 +- docs/BRIDGED_IDENTITY.md | 87 ++ docs/LEXICON_PUBLISHING.md | 27 +- docs/NOTIFICATIONS.md | 2 +- docs/PRODUCTION_DEPLOY.md | 38 + docs/README.md | 1 + docs/SECURITY_MODEL.md | 2 +- docs/TEST_ARCHITECTURE.md | 2 +- internal/api/handlers/actor/errors.go | 44 + internal/api/handlers/actor/get_comments.go | 43 +- .../api/handlers/actor/get_comments_test.go | 15 +- internal/api/handlers/actor/get_posts.go | 55 +- internal/api/handlers/actor/get_posts_test.go | 15 +- .../api/handlers/actor/resolve_actor_test.go | 177 ++++ .../community/community_origin_view_test.go | 157 +++ internal/api/handlers/community/list_test.go | 4 + .../handlers/community/viewer_state_test.go | 7 +- .../notification/list_notifications_test.go | 46 + .../list_notifications_upvote_test.go | 35 + internal/api/handlers/user/delete_test.go | 5 + .../routes/author_origin_integration_test.go | 157 +++ ...deration_content_label_integration_test.go | 11 +- internal/api/routes/user.go | 29 +- internal/api/routes/user_get_profile_test.go | 118 ++- ...user_scoped_identifier_integration_test.go | 82 ++ .../viewer_moderation_integration_test.go | 80 +- internal/api/xrpc/mapper.go | 7 + internal/atproto/identity/base_resolver.go | 2 +- .../atproto/identity/uncached_resolver.go | 71 ++ .../uncached_resolver_integration_test.go | 116 +++ .../identity/uncached_resolver_test.go | 95 ++ .../atproto/identity/well_known_hosts_test.go | 5 +- .../atproto/jetstream/community_consumer.go | 428 +++++++- .../jetstream/community_consumer_test.go | 77 +- .../community_did_verification_test.go | 6 + .../community_handle_binding_test.go | 19 +- .../jetstream/community_origin_test.go | 15 +- .../community_profile_origin_debounce_test.go | 275 ++++++ .../atproto/jetstream/error_taxonomy_test.go | 19 +- internal/atproto/jetstream/errors.go | 11 +- .../atproto/jetstream/identity_refresh.go | 279 ++++++ .../identity_refresh_debounce_test.go | 534 ++++++++++ .../jetstream/identity_refresh_lock.go | 186 ++++ ...ity_refresh_lock_statement_timeout_test.go | 101 ++ .../jetstream/identity_refresh_lock_test.go | 187 ++++ .../identity_refresh_lock_wait_test.go | 14 + .../jetstream/resolve_deadline_test.go | 26 + .../uncached_identity_resolver_wiring.go | 17 + internal/atproto/jetstream/user_consumer.go | 415 +++++--- .../atproto/jetstream/user_consumer_test.go | 44 +- .../atproto/jetstream/user_profile_origin.go | 61 ++ .../user_profile_origin_debounce_test.go | 293 ++++++ .../jetstream/user_profile_origin_test.go | 206 ++++ .../lexicon/social/coves/actor/defs.json | 40 + .../social/coves/actor/getComments.json | 4 +- .../lexicon/social/coves/actor/getPosts.json | 4 +- .../social/coves/actor/getProfile.json | 4 +- .../lexicon/social/coves/actor/profile.json | 20 + .../lexicon/social/coves/community/defs.json | 20 + .../social/coves/community/post/defs.json | 15 + .../social/coves/community/profile.json | 10 + .../comments/comment_placeholder_view_test.go | 80 ++ internal/core/comments/comment_service.go | 13 +- .../core/comments/comment_service_test.go | 8 + internal/core/communities/community.go | 10 + .../community_identity_consumer_test.go | 399 ++++++++ .../community_identity_refresh_race_test.go | 315 ++++++ .../community_profile_origin_consumer_test.go | 916 ++++++++++++++++++ .../core/communities/community_view_test.go | 2 +- .../consumer_profile_indexing_test.go | 62 +- .../core/communities/fake_repository_test.go | 4 + internal/core/communities/interfaces.go | 1 + internal/core/communities/service.go | 9 +- .../communities/service_validation_test.go | 18 + internal/core/notifications/list_service.go | 4 + internal/core/notifications/view.go | 12 +- internal/core/posts/post.go | 4 + internal/core/users/errors.go | 4 + internal/core/users/interfaces.go | 7 + .../users/resolve_actor_identifier_test.go | 248 +++++ internal/core/users/service.go | 73 +- internal/core/users/service_test.go | 5 + internal/core/users/user.go | 10 + .../core/users/user_identity_consumer_test.go | 574 +++++++---- .../users/user_identity_refresh_lock_test.go | 803 +++++++++++++++ internal/core/users/user_integration_test.go | 70 ++ .../user_profile_origin_consumer_test.go | 389 ++++++++ .../migrations/057_users_profile_origin.sql | 13 + .../058_communities_origin_display.sql | 9 + .../059_users_origin_username_index.sql | 14 + .../db/postgres/admission_repo_schema_test.go | 10 +- .../bridged_votes_peak_migration_test.go | 6 + internal/db/postgres/community_repo.go | 56 +- ...unity_subscriber_recount_migration_test.go | 28 +- .../credential_reencrypt_migration_test.go | 12 + .../postgres/discover_hot_migration_test.go | 6 + ...uture_comment_created_at_migration_test.go | 6 + .../postgres/notification_migration_test.go | 12 + internal/db/postgres/post_repo.go | 5 + .../read_path_indexes_migration_test.go | 6 + .../rematerialize_ledger_schema_test.go | 6 + .../user_origin_username_repo_test.go | 131 +++ internal/db/postgres/user_repo.go | 140 ++- internal/db/postgres/user_repo_test.go | 34 + .../vote_drift_recount_migration_test.go | 6 + ...tes_upvote_history_index_migration_test.go | 6 + tests/e2e/community_identity_rename_test.go | 57 ++ tests/e2e/contracts_test.go | 3 + tests/e2e/reconcile_handles_test.go | 58 ++ tests/e2e/user_identity_rename_test.go | 50 + ...nvalid-origin-software-too-many-bytes.json | 4 + ...profile-invalid-origin-too-many-bytes.json | 4 + .../profile-invalid-origin-url-not-uri.json | 4 + ...nvalid-origin-username-too-many-bytes.json | 4 + .../actor/profile-valid-origin.json | 7 + ...nvalid-origin-software-too-many-bytes.json | 5 + ...etailed-invalid-origin-too-many-bytes.json | 5 + ...w-detailed-invalid-origin-url-not-uri.json | 5 + ...nvalid-origin-username-too-many-bytes.json | 5 + .../profile-view-detailed-valid-origin.json | 8 + ...nvalid-origin-software-too-many-bytes.json | 12 + ...ty-profile-invalid-origin-url-not-uri.json | 12 + ...ommunity-profile-valid-origin-display.json | 12 + ...nvalid-origin-software-too-many-bytes.json | 10 + ...w-detailed-invalid-origin-url-not-uri.json | 10 + ...ty-view-detailed-valid-origin-display.json | 10 + ...nvalid-origin-software-too-many-bytes.json | 9 + ...unity-view-invalid-origin-url-not-uri.json | 9 + .../community-view-valid-origin-display.json | 9 + ...author-origin-software-too-many-bytes.json | 24 + ...-invalid-author-origin-too-many-bytes.json | 24 + ...author-origin-username-too-many-bytes.json | 24 + .../post/post-view-valid-author-origin.json | 24 + tests/lexicon_actor_identifier_test.go | 59 ++ tests/lexicon_fixtures_test.go | 69 +- ...exicon_notification_bridged_upvote_test.go | 57 ++ 150 files changed, 11620 insertions(+), 601 deletions(-) create mode 100644 cmd/reconcile-handles/helpers_integration_test.go create mode 100644 cmd/reconcile-handles/identity_lock_race_integration_test.go create mode 100644 cmd/reconcile-handles/main.go create mode 100644 cmd/reconcile-handles/open.go create mode 100644 cmd/reconcile-handles/reconcile_integration_test.go create mode 100644 cmd/reconcile-handles/reconcile_outcomes_integration_test.go create mode 100644 cmd/reconcile-handles/run_stderr_integration_test.go create mode 100644 cmd/reconcile-handles/run_test.go create mode 100644 cmd/reconcile-handles/sweep.go create mode 100644 cmd/reconcile-handles/sweep_test.go create mode 100644 cmd/server/uncached_identity_resolver_wiring_test.go create mode 100644 docs/BRIDGED_IDENTITY.md create mode 100644 internal/api/handlers/actor/resolve_actor_test.go create mode 100644 internal/api/handlers/community/community_origin_view_test.go create mode 100644 internal/api/routes/author_origin_integration_test.go create mode 100644 internal/api/routes/user_scoped_identifier_integration_test.go create mode 100644 internal/atproto/identity/uncached_resolver.go create mode 100644 internal/atproto/identity/uncached_resolver_integration_test.go create mode 100644 internal/atproto/identity/uncached_resolver_test.go create mode 100644 internal/atproto/jetstream/community_profile_origin_debounce_test.go create mode 100644 internal/atproto/jetstream/identity_refresh.go create mode 100644 internal/atproto/jetstream/identity_refresh_debounce_test.go create mode 100644 internal/atproto/jetstream/identity_refresh_lock.go create mode 100644 internal/atproto/jetstream/identity_refresh_lock_statement_timeout_test.go create mode 100644 internal/atproto/jetstream/identity_refresh_lock_test.go create mode 100644 internal/atproto/jetstream/identity_refresh_lock_wait_test.go create mode 100644 internal/atproto/jetstream/uncached_identity_resolver_wiring.go create mode 100644 internal/atproto/jetstream/user_profile_origin.go create mode 100644 internal/atproto/jetstream/user_profile_origin_debounce_test.go create mode 100644 internal/atproto/jetstream/user_profile_origin_test.go create mode 100644 internal/core/communities/community_identity_consumer_test.go create mode 100644 internal/core/communities/community_identity_refresh_race_test.go create mode 100644 internal/core/communities/community_profile_origin_consumer_test.go create mode 100644 internal/core/users/resolve_actor_identifier_test.go create mode 100644 internal/core/users/user_identity_refresh_lock_test.go create mode 100644 internal/core/users/user_profile_origin_consumer_test.go create mode 100644 internal/db/migrations/057_users_profile_origin.sql create mode 100644 internal/db/migrations/058_communities_origin_display.sql create mode 100644 internal/db/migrations/059_users_origin_username_index.sql create mode 100644 internal/db/postgres/user_origin_username_repo_test.go create mode 100644 tests/e2e/community_identity_rename_test.go create mode 100644 tests/e2e/reconcile_handles_test.go create mode 100644 tests/e2e/user_identity_rename_test.go create mode 100644 tests/lexicon-test-data/actor/profile-invalid-origin-software-too-many-bytes.json create mode 100644 tests/lexicon-test-data/actor/profile-invalid-origin-too-many-bytes.json create mode 100644 tests/lexicon-test-data/actor/profile-invalid-origin-url-not-uri.json create mode 100644 tests/lexicon-test-data/actor/profile-invalid-origin-username-too-many-bytes.json create mode 100644 tests/lexicon-test-data/actor/profile-valid-origin.json create mode 100644 tests/lexicon-test-data/actor/profile-view-detailed-invalid-origin-software-too-many-bytes.json create mode 100644 tests/lexicon-test-data/actor/profile-view-detailed-invalid-origin-too-many-bytes.json create mode 100644 tests/lexicon-test-data/actor/profile-view-detailed-invalid-origin-url-not-uri.json create mode 100644 tests/lexicon-test-data/actor/profile-view-detailed-invalid-origin-username-too-many-bytes.json create mode 100644 tests/lexicon-test-data/actor/profile-view-detailed-valid-origin.json create mode 100644 tests/lexicon-test-data/community/community-profile-invalid-origin-software-too-many-bytes.json create mode 100644 tests/lexicon-test-data/community/community-profile-invalid-origin-url-not-uri.json create mode 100644 tests/lexicon-test-data/community/community-profile-valid-origin-display.json create mode 100644 tests/lexicon-test-data/community/community-view-detailed-invalid-origin-software-too-many-bytes.json create mode 100644 tests/lexicon-test-data/community/community-view-detailed-invalid-origin-url-not-uri.json create mode 100644 tests/lexicon-test-data/community/community-view-detailed-valid-origin-display.json create mode 100644 tests/lexicon-test-data/community/community-view-invalid-origin-software-too-many-bytes.json create mode 100644 tests/lexicon-test-data/community/community-view-invalid-origin-url-not-uri.json create mode 100644 tests/lexicon-test-data/community/community-view-valid-origin-display.json create mode 100644 tests/lexicon-test-data/post/post-view-invalid-author-origin-software-too-many-bytes.json create mode 100644 tests/lexicon-test-data/post/post-view-invalid-author-origin-too-many-bytes.json create mode 100644 tests/lexicon-test-data/post/post-view-invalid-author-origin-username-too-many-bytes.json create mode 100644 tests/lexicon-test-data/post/post-view-valid-author-origin.json create mode 100644 tests/lexicon_actor_identifier_test.go diff --git a/cmd/reconcile-handles/helpers_integration_test.go b/cmd/reconcile-handles/helpers_integration_test.go new file mode 100644 index 0000000..4eebb91 --- /dev/null +++ b/cmd/reconcile-handles/helpers_integration_test.go @@ -0,0 +1,336 @@ +//go:build integration + +package main + +import ( + "context" + "database/sql" + "errors" + "net/http" + "net/http/httptest" + "net/url" + "regexp" + "strings" + "sync" + "testing" + "time" + + "Coves/internal/atproto/identity" + "Coves/internal/atproto/jetstream" + "Coves/internal/core/communities" + "Coves/internal/core/users" + "Coves/internal/crypto/credentialcipher/credentialciphertest" + "Coves/internal/db/postgres" + "Coves/tests/testkit" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// bridgeHost is the trusted bridge PDS. Rows whose pds_url is this host are +// bridged; the trust list is given in the same full-URL form. +const bridgeHost = "https://bridge.test" + +// scriptedAnswer is the uncached resolver's answer for one DID. subject is the +// DID the answer claims to be about; empty means the asked DID. +type scriptedAnswer struct { + handle string + subject string + err error +} + +// scriptedIdentityResolver is an identity.UncachedResolver answering each DID +// from a fixed script. It records how often each identifier was resolved, and +// runs onFirstResolve (when set before the sweep starts) once, on the first +// Resolve call, before answering it. +type scriptedIdentityResolver struct { + onFirstResolve func(ctx context.Context, did string) + + mu sync.Mutex + answers map[string]scriptedAnswer + calls map[string]int + firstResolve sync.Once +} + +func newScriptedIdentityResolver(answers map[string]scriptedAnswer) *scriptedIdentityResolver { + return &scriptedIdentityResolver{answers: answers, calls: map[string]int{}} +} + +func (r *scriptedIdentityResolver) Resolve(ctx context.Context, did string) (*identity.Identity, error) { + r.firstResolve.Do(func() { + if r.onFirstResolve != nil { + r.onFirstResolve(ctx, did) + } + }) + r.mu.Lock() + r.calls[did]++ + answer, ok := r.answers[did] + r.mu.Unlock() + if !ok { + return nil, &identity.ErrNotFound{Identifier: did, Reason: "no scripted answer"} + } + if answer.err != nil { + return nil, answer.err + } + subject := answer.subject + if subject == "" { + subject = did + } + return &identity.Identity{ + DID: subject, Handle: answer.handle, PDSURL: bridgeHost, + Method: identity.MethodDNS, ResolvedAt: time.Now().UTC(), + }, nil +} + +func (r *scriptedIdentityResolver) ResolveHandle(context.Context, string) (string, string, error) { + return "", "", errors.New("scripted identity resolver: ResolveHandle is not scripted") +} + +func (r *scriptedIdentityResolver) ResolveDID(context.Context, string) (*identity.DIDDocument, error) { + return nil, errors.New("scripted identity resolver: ResolveDID is not scripted") +} + +func (r *scriptedIdentityResolver) Purge(context.Context, string) error { return nil } + +func (r *scriptedIdentityResolver) ResolveDeclaredIdentity(_ context.Context, did string) (*identity.DeclaredIdentity, error) { + r.mu.Lock() + answer, ok := r.answers[did] + r.mu.Unlock() + if !ok { + return nil, &identity.ErrNotFound{Identifier: did, Reason: "no scripted document"} + } + subject := answer.subject + if subject == "" { + subject = did + } + return &identity.DeclaredIdentity{DID: subject, DeclaredHandle: answer.handle, PDSURL: bridgeHost}, nil +} + +// callCount is how many times did was resolved. +func (r *scriptedIdentityResolver) callCount(did string) int { + r.mu.Lock() + defer r.mu.Unlock() + return r.calls[did] +} + +// newFailingPLCCachingResolver is a real Postgres-backed caching resolver whose +// PLC directory is a loopback server that fails the test on any request. The +// sweep uses the caching resolver only to purge, so nothing should reach it. +func newFailingPLCCachingResolver(t *testing.T, db *sql.DB) identity.Resolver { + t.Helper() + plc := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Failf(t, "PLC directory contacted", "the caching resolver requested %s %s", r.Method, r.URL.Path) + http.Error(w, "test PLC directory refuses every request", http.StatusTeapot) + })) + t.Cleanup(plc.Close) + // The hatch lets a stray request reach the failing server instead of being + // refused by the SSRF guard, so it fails the test rather than an error path. + config := identity.DefaultConfig(identity.WithPrivateHostsAllowed()) + config.PLCURL = plc.URL + return identity.NewResolver(db, config) +} + +// openSecondPool opens another pool to the same clone as db, tagged with +// applicationName. It is closed before the clone is dropped. +func openSecondPool(t *testing.T, db *sql.DB, applicationName string) *sql.DB { + t.Helper() + ctx := context.Background() + var database string + require.NoError(t, db.QueryRowContext(ctx, `SELECT current_database()`).Scan(&database), "read clone name") + pg := testkit.Endpoints().Postgres + pool, err := sql.Open("postgres", pg.URL(database)+"&application_name="+url.QueryEscape(applicationName)) + require.NoError(t, err, "open pool %s on %s", applicationName, pg.Redacted(database)) + t.Cleanup(func() { _ = pool.Close() }) + require.NoError(t, pool.PingContext(ctx), "ping pool %s on %s", applicationName, pg.Redacted(database)) + return pool +} + +// sweepLockApplicationName tags the sweep's lock pool, so pg_locks +// observations can tell its backends apart. +const sweepLockApplicationName = "reconcile-handles-lock" + +// newSweepDependencies wires a sweep against db: a dedicated lock pool on the +// same clone, the failing-PLC caching resolver, the given uncached resolver, +// trust in bridgeHost only, and hostedBy verification skipped. +func newSweepDependencies(t *testing.T, db *sql.DB, uncached identity.UncachedResolver) sweepDependencies { + t.Helper() + return sweepDependencies{ + Database: db, + LockDatabase: openSecondPool(t, db, sweepLockApplicationName), + CachingResolver: newFailingPLCCachingResolver(t, db), + UncachedResolver: uncached, + Trust: jetstream.NewBridgeTrust([]string{bridgeHost}), + SkipHostedByVerification: true, + } +} + +// uniqueDID is a did:plc identifier unique to this test process. +func uniqueDID(t *testing.T, prefix string) string { + t.Helper() + return "did:plc:" + testkit.UniqueIDWithPrefix(t, prefix) +} + +func seedUser(t *testing.T, db *sql.DB, did, handle, pdsURL string) { + t.Helper() + _, err := postgres.NewUserRepository(db).Create(context.Background(), &users.User{DID: did, Handle: handle, PDSURL: pdsURL}) + require.NoError(t, err, "seed user %s", did) +} + +func seedCommunity(t *testing.T, db *sql.DB, did, handle, pdsURL string) { + t.Helper() + name, _, _ := strings.Cut(handle, ".") + now := time.Now().UTC() + _, err := postgres.NewCommunityRepository(db, credentialciphertest.Fixed()).Create(context.Background(), &communities.Community{ + DID: did, Handle: handle, Name: name, DisplayName: name, PDSURL: pdsURL, + OwnerDID: did, CreatedByDID: did, HostedByDID: "did:web:bridge.test", + Visibility: "public", CreatedAt: now, UpdatedAt: now, + }) + require.NoError(t, err, "seed community %s", did) +} + +// seedIdentityCacheRow stores an unexpired identity_cache row keyed by identifier. +func seedIdentityCacheRow(t *testing.T, db *sql.DB, identifier, did, handle string) { + t.Helper() + _, err := db.ExecContext(context.Background(), ` + INSERT INTO identity_cache (identifier, did, handle, pds_url, resolved_at, resolution_method, expires_at) + VALUES ($1, $2, $3, $4, NOW(), 'dns', NOW() + INTERVAL '1 day')`, + identifier, did, handle, bridgeHost) + require.NoError(t, err, "seed identity_cache row %s", identifier) +} + +// seedOAuthSession stores an unexpired OAuth session for did carrying handle. +func seedOAuthSession(t *testing.T, db *sql.DB, did, handle, pdsURL string) { + t.Helper() + _, err := db.ExecContext(context.Background(), ` + INSERT INTO oauth_sessions (did, handle, pds_url, access_token, refresh_token, dpop_private_jwk, auth_server_iss, expires_at, session_id) + VALUES ($1, $2, $3, 'test-access', 'test-refresh', '{}', 'https://auth.bridge.test', NOW() + INTERVAL '1 day', 'reconcile-session')`, + did, handle, pdsURL) + require.NoError(t, err, "seed OAuth session for %s", did) +} + +func storedUserHandle(t *testing.T, db *sql.DB, did string) string { + t.Helper() + var handle string + require.NoError(t, db.QueryRowContext(context.Background(), `SELECT handle FROM users WHERE did = $1`, did).Scan(&handle), + "read user %s", did) + return handle +} + +func storedCommunityHandle(t *testing.T, db *sql.DB, did string) string { + t.Helper() + var handle string + require.NoError(t, db.QueryRowContext(context.Background(), `SELECT handle FROM communities WHERE did = $1`, did).Scan(&handle), + "read community %s", did) + return handle +} + +func oauthSessionHandle(t *testing.T, db *sql.DB, did string) string { + t.Helper() + var handle string + require.NoError(t, db.QueryRowContext(context.Background(), `SELECT handle FROM oauth_sessions WHERE did = $1`, did).Scan(&handle), + "read OAuth session for %s", did) + return handle +} + +func identityCacheHasRow(t *testing.T, db *sql.DB, identifier string) bool { + t.Helper() + var exists bool + require.NoError(t, db.QueryRowContext(context.Background(), + `SELECT EXISTS (SELECT 1 FROM identity_cache WHERE identifier = $1)`, identifier).Scan(&exists), + "read identity_cache row %s", identifier) + return exists +} + +// lastLine is the final non-empty line of a printed report. +func lastLine(output string) string { + lines := strings.Split(strings.TrimRight(output, "\n"), "\n") + return lines[len(lines)-1] +} + +// recordingResolver is an identity.Resolver that records every identifier it +// is asked about and forwards to inner. With a nil inner it fails the test on +// any call. +type recordingResolver struct { + t *testing.T + inner identity.Resolver + + mu sync.Mutex + identifiers []string +} + +func (r *recordingResolver) record(method, identifier string) error { + r.mu.Lock() + r.identifiers = append(r.identifiers, identifier) + r.mu.Unlock() + if r.inner == nil { + assert.Failf(r.t, "caching resolver called", "%s(%s)", method, identifier) + return errors.New("recording resolver: no inner resolver") + } + return nil +} + +func (r *recordingResolver) Resolve(ctx context.Context, identifier string) (*identity.Identity, error) { + if err := r.record("Resolve", identifier); err != nil { + return nil, err + } + return r.inner.Resolve(ctx, identifier) +} + +func (r *recordingResolver) ResolveHandle(ctx context.Context, handle string) (string, string, error) { + if err := r.record("ResolveHandle", handle); err != nil { + return "", "", err + } + return r.inner.ResolveHandle(ctx, handle) +} + +func (r *recordingResolver) ResolveDID(ctx context.Context, did string) (*identity.DIDDocument, error) { + if err := r.record("ResolveDID", did); err != nil { + return nil, err + } + return r.inner.ResolveDID(ctx, did) +} + +func (r *recordingResolver) Purge(ctx context.Context, identifier string) error { + if err := r.record("Purge", identifier); err != nil { + return err + } + return r.inner.Purge(ctx, identifier) +} + +// seen is every identifier passed to any method, in call order. +func (r *recordingResolver) seen() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.identifiers...) +} + +func identityCacheRowCount(t *testing.T, db *sql.DB) int { + t.Helper() + var count int + require.NoError(t, db.QueryRowContext(context.Background(), `SELECT COUNT(*) FROM identity_cache`).Scan(&count), + "count identity_cache rows") + return count +} + +// identityCacheRowContents is the did and handle stored under identifier. +func identityCacheRowContents(t *testing.T, db *sql.DB, identifier string) (did, handle string) { + t.Helper() + require.NoError(t, db.QueryRowContext(context.Background(), + `SELECT did, handle FROM identity_cache WHERE identifier = $1`, identifier).Scan(&did, &handle), + "read identity_cache row %s", identifier) + return did, handle +} + +// differsLineVerified finds the report line for one still-differing row and +// returns its verified field. found is false when no line matches. +func differsLineVerified(output, kind, did, stored, reason string) (verified string, found bool) { + pattern := regexp.MustCompile(`(?m)^differs kind=` + regexp.QuoteMeta(kind) + + ` did=` + regexp.QuoteMeta(did) + + ` stored=` + regexp.QuoteMeta(stored) + + ` verified=(\S+) reason=` + regexp.QuoteMeta(reason) + `$`) + match := pattern.FindStringSubmatch(output) + if match == nil { + return "", false + } + return match[1], true +} diff --git a/cmd/reconcile-handles/identity_lock_race_integration_test.go b/cmd/reconcile-handles/identity_lock_race_integration_test.go new file mode 100644 index 0000000..9e390c1 --- /dev/null +++ b/cmd/reconcile-handles/identity_lock_race_integration_test.go @@ -0,0 +1,333 @@ +//go:build integration + +// The sweep and the identity-event consumers refresh handles under one per-DID +// lock. In these races the sweep and a consumer each run on their own pools to +// the same clone, with their own scripted uncached resolver. One side holds the +// lock while blocked at its resolution; the other is released only after it +// has returned or is seen waiting on the advisory lock in pg_locks. +package main + +import ( + "bytes" + "context" + "database/sql" + "sync" + "testing" + "time" + + "Coves/internal/atproto/identity" + "Coves/internal/atproto/jetstream" + "Coves/internal/core/users" + "Coves/internal/crypto/credentialcipher/credentialciphertest" + "Coves/internal/db/postgres" + "Coves/tests/testkit" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// lockRacePhaseLimit bounds each phase in which one side is held at its +// resolution barrier. It stays under the refresh's 5s resolve timeout, so the +// test releases the holder before that timeout can decide the outcome. +const lockRacePhaseLimit = 3 * time.Second + +// lockRaceFinishLimit bounds the wait for both sides after the release. +const lockRaceFinishLimit = 10 * time.Second + +const lockRacePollInterval = 10 * time.Millisecond + +// consumerApplicationName tags the consumer's pool in pg_locks observations. +const consumerApplicationName = "reconcile-handles-consumer" + +// resolveBarrier holds a resolution until opened or until the resolution's +// context ends. Its hold method fits scriptedIdentityResolver.onFirstResolve. +type resolveBarrier struct { + entered chan struct{} + enterOnce sync.Once + release chan struct{} + releaseOnce sync.Once +} + +func newResolveBarrier() *resolveBarrier { + return &resolveBarrier{entered: make(chan struct{}), release: make(chan struct{})} +} + +func (b *resolveBarrier) hold(ctx context.Context, _ string) { + b.enterOnce.Do(func() { close(b.entered) }) + select { + case <-b.release: + case <-ctx.Done(): + } +} + +func (b *resolveBarrier) open() { + b.releaseOnce.Do(func() { close(b.release) }) +} + +// lockRaceRun is one side of the race running in its own goroutine. +type lockRaceRun struct { + name string + applicationName string + done chan struct{} + err error +} + +func (run *lockRaceRun) returned() bool { + select { + case <-run.done: + return true + default: + return false + } +} + +func (run *lockRaceRun) result(t *testing.T, ctx context.Context) error { + t.Helper() + select { + case <-run.done: + return run.err + case <-ctx.Done(): + require.FailNowf(t, "race side did not return", "%s: %v", run.name, ctx.Err()) + return nil + } +} + +// identityEventHandler is the public event entry point both consumers share. +type identityEventHandler interface { + HandleEvent(ctx context.Context, event *jetstream.JetstreamEvent) error +} + +// newIdentityEventConsumer builds the consumer for kind on its own pool to the +// clone behind db: its own repository, failing-PLC caching resolver and +// per-DID refresh lock, with uncached as its uncached resolver. +func newIdentityEventConsumer(t *testing.T, db *sql.DB, kind string, uncached identity.UncachedResolver) identityEventHandler { + t.Helper() + pool := openSecondPool(t, db, consumerApplicationName) + caching := newFailingPLCCachingResolver(t, pool) + lock := jetstream.NewPostgresIdentityRefreshLock(pool, jetstream.IdentityRefreshLockWaitTimeout) + switch kind { + case kindUser: + service := users.NewUserService(postgres.NewUserRepository(pool), caching, "", nil, "") + return jetstream.NewUserEventConsumer(service, caching, + jetstream.WithUserUncachedIdentityResolver(uncached), + jetstream.WithUserIdentityRefreshLock(lock)) + case kindCommunity: + return jetstream.NewCommunityEventConsumer(postgres.NewCommunityRepository(pool, credentialciphertest.Fixed()), "", + true, caching, + jetstream.WithCommunityUncachedIdentityResolver(uncached), + jetstream.WithCommunityIdentityRefreshLock(lock)) + } + require.FailNowf(t, "unknown row kind", "%q", kind) + return nil +} + +// advisoryLockWaiting reports whether a backend of applicationName on this +// clone waits for an advisory lock whose tag is granted to another backend. +func advisoryLockWaiting(ctx context.Context, observer *sql.DB, applicationName string) (bool, error) { + var waiting bool + err := observer.QueryRowContext(ctx, ` + SELECT EXISTS ( + SELECT 1 + FROM pg_locks waiter + JOIN pg_stat_activity activity ON activity.pid = waiter.pid + JOIN pg_locks holder + ON holder.locktype = 'advisory' AND holder.granted + AND holder.pid <> waiter.pid + AND holder.database = waiter.database + AND holder.classid = waiter.classid + AND holder.objid = waiter.objid + AND holder.objsubid = waiter.objsubid + WHERE waiter.locktype = 'advisory' AND NOT waiter.granted + AND waiter.database = (SELECT oid FROM pg_database WHERE datname = current_database()) + AND activity.application_name = $1)`, applicationName).Scan(&waiting) + return waiting, err +} + +// awaitContenderSettled polls until contender has returned or one of its +// backends is seen waiting on the per-DID advisory lock. It reports whether +// the contender is waiting, and fails the test when neither happens within +// the phase. +func awaitContenderSettled(t *testing.T, phase context.Context, observer *sql.DB, contender *lockRaceRun) bool { + t.Helper() + ticker := time.NewTicker(lockRacePollInterval) + defer ticker.Stop() + for { + if contender.returned() { + return false + } + waiting, err := advisoryLockWaiting(phase, observer, contender.applicationName) + if err != nil && phase.Err() == nil { + require.NoError(t, err, "poll pg_locks for %s", contender.applicationName) + } + if waiting { + return true + } + select { + case <-ticker.C: + case <-phase.Done(): + require.FailNowf(t, "contender not settled", + "%s neither returned nor waited on the per-DID advisory lock within %s", contender.name, lockRacePhaseLimit) + } + } +} + +// lockRaceCase is one race between the sweep and an identity-event consumer +// over a single bridged row. +type lockRaceCase struct { + name string + kind string + // sweepHoldsLock: the sweep blocks at its resolution while holding the + // lock and the consumer contends; otherwise the consumer holds. + sweepHoldsLock bool + // contenderMustWait fails the race unless the contender is seen waiting on + // the advisory lock before the holder is released. + contenderMustWait bool + seedHandle string + sweepAnswer string + consumerAnswer string + wantHandle string +} + +// runLockRace seeds the row, runs the race and returns the stored handle once +// both sides returned, together with the sweep's report. Neither side may fail. +func runLockRace(t *testing.T, tc lockRaceCase) (storedHandle string, report sweepReport) { + t.Helper() + db := testkit.DB(t) + did := uniqueDID(t, tc.kind) + switch tc.kind { + case kindUser: + seedUser(t, db, did, tc.seedHandle, bridgeHost) + case kindCommunity: + seedCommunity(t, db, did, tc.seedHandle, bridgeHost) + } + + barrier := newResolveBarrier() + sweepResolver := newScriptedIdentityResolver(map[string]scriptedAnswer{did: {handle: tc.sweepAnswer}}) + consumerResolver := newScriptedIdentityResolver(map[string]scriptedAnswer{did: {handle: tc.consumerAnswer}}) + if tc.sweepHoldsLock { + sweepResolver.onFirstResolve = barrier.hold + } else { + consumerResolver.onFirstResolve = barrier.hold + } + dependencies := newSweepDependencies(t, db, sweepResolver) + consumer := newIdentityEventConsumer(t, db, tc.kind, consumerResolver) + + ctx, cancel := context.WithCancel(context.Background()) + var sides sync.WaitGroup + // Registered after every pool, so it runs before they close. + t.Cleanup(func() { + barrier.open() + cancel() + sides.Wait() + }) + start := func(name, applicationName string, side func(context.Context) error) *lockRaceRun { + run := &lockRaceRun{name: name, applicationName: applicationName, done: make(chan struct{})} + sides.Add(1) + go func() { + defer sides.Done() + defer close(run.done) + run.err = side(ctx) + }() + return run + } + var out bytes.Buffer + startSweep := func() *lockRaceRun { + return start("sweep", sweepLockApplicationName, func(ctx context.Context) error { + var err error + report, err = reconcile(ctx, sweepOptions{Concurrency: 1}, dependencies, &out) + return err + }) + } + startConsumer := func() *lockRaceRun { + return start(tc.kind+" consumer", consumerApplicationName, func(ctx context.Context) error { + return consumer.HandleEvent(ctx, &jetstream.JetstreamEvent{ + Did: did, Kind: "identity", + Identity: &jetstream.IdentityEvent{Did: did, Handle: "event.bridge.test", Seq: 1, Time: time.Now().UTC().Format(time.RFC3339)}, + }) + }) + } + + var holder, contender *lockRaceRun + if tc.sweepHoldsLock { + holder = startSweep() + } else { + holder = startConsumer() + } + phase, cancelPhase := context.WithTimeout(ctx, lockRacePhaseLimit) + defer cancelPhase() + select { + case <-barrier.entered: + case <-holder.done: + require.FailNowf(t, "holder returned early", "%s returned before reaching its resolution barrier: %v", holder.name, holder.err) + case <-phase.Done(): + require.FailNowf(t, "holder not at barrier", "%s did not reach its resolution barrier within %s", holder.name, lockRacePhaseLimit) + } + + if tc.sweepHoldsLock { + contender = startConsumer() + } else { + contender = startSweep() + } + waiting := awaitContenderSettled(t, phase, db, contender) + if tc.contenderMustWait { + require.Truef(t, waiting, "%s returned (err=%v) instead of waiting on the per-DID advisory lock", contender.name, contender.err) + } + barrier.open() + + finish, cancelFinish := context.WithTimeout(ctx, lockRaceFinishLimit) + defer cancelFinish() + require.NoError(t, holder.result(t, finish), holder.name) + require.NoError(t, contender.result(t, finish), contender.name) + + if tc.kind == kindCommunity { + return storedCommunityHandle(t, db, did), report + } + return storedUserHandle(t, db, did), report +} + +// TestReconcile_SweepHoldingLockKeepsConsumerHandle (DoD 9): the sweep resolves +// an old handle and stalls inside the per-DID lock while a user consumer on +// another pool handles an identity event for the same DID. The consumer waits +// for the lock, so its newer handle lands after the sweep's write and stays. +func TestReconcile_SweepHoldingLockKeepsConsumerHandle(t *testing.T) { + t.Parallel() + stored, _ := runLockRace(t, lockRaceCase{ + kind: kindUser, + sweepHoldsLock: true, + seedHandle: "seedalice.bridge.test", + sweepAnswer: "oldalice.bridge.test", + consumerAnswer: "newalice.bridge.test", + }) + assert.Equal(t, "newalice.bridge.test", stored, "users row handle") +} + +// TestReconcile_SweepRereadsHandleInsideLock (DoD 11): a consumer stalls inside +// the per-DID lock after resolving B; the sweep selects the row while it still +// stores A and waits for the lock. The consumer then stores B, and the sweep, +// whose resolution answers A, compares against the handle read inside the lock +// (B), so it writes A back and reports the row updated. +func TestReconcile_SweepRereadsHandleInsideLock(t *testing.T) { + t.Parallel() + for _, tc := range []lockRaceCase{ + { + name: "user", kind: kindUser, contenderMustWait: true, + seedHandle: "alicea.bridge.test", sweepAnswer: "alicea.bridge.test", consumerAnswer: "aliceb.bridge.test", + wantHandle: "alicea.bridge.test", + }, + { + name: "community", kind: kindCommunity, contenderMustWait: true, + seedHandle: "gardena.bridge.test", sweepAnswer: "gardena.bridge.test", consumerAnswer: "gardenb.bridge.test", + wantHandle: "gardena.bridge.test", + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + stored, report := runLockRace(t, tc) + assert.Equal(t, tc.wantHandle, stored, "stored handle") + assert.Equal(t, 1, report.Checked, "Checked") + assert.Equal(t, 1, report.Updated, "Updated") + assert.Equal(t, 0, report.Unchanged, "Unchanged") + assert.Equal(t, 0, report.StillDiffering, "StillDiffering") + }) + } +} diff --git a/cmd/reconcile-handles/main.go b/cmd/reconcile-handles/main.go new file mode 100644 index 0000000..3f08566 --- /dev/null +++ b/cmd/reconcile-handles/main.go @@ -0,0 +1,216 @@ +// Command reconcile-handles re-resolves the handles of bridged users and +// communities (rows whose stored pds_url is on a trusted bridge host), updates +// stored handles through the per-DID locked identity refresh, and reports the +// rows whose stored handle still differs from a verified one. +// +// It is the Coves half of the bridge handle-rename rollout: step 6 runs it +// after the rename job finishes, and step 6a reads its report as the gate (exit +// status 0 only when every bridged row matches its verified handle). +// +// A normal run takes, per DID, the same locked refresh the identity-event +// consumers run: purge the cached DID answer, resolve the DID fresh through the +// uncached resolver, verify the handle, re-read the stored handle inside the +// lock, purge the old handle and write the new one. Users also get their OAuth +// session handles updated; communities pass ValidateHandle and the hostedBy +// verification first. A dry run only resolves and verifies: it takes no lock, +// purges nothing and writes nothing. +// +// It loads the AppView's own configuration (internal/config), so run it with +// the AppView's environment: it resolves through the same PLC directory, +// well-known overrides and private-host rules as cmd/server. +// +// Usage: +// +// DATABASE_URL=postgres://... reconcile-handles [-dry-run] [-clear-identity-cache] +// [-concurrency N] [-trusted-hosts https://bridge.example,...] +// +// Flags: +// +// -dry-run report what would change; write nothing +// -clear-identity-cache delete every identity_cache row before the sweep +// (refused together with -dry-run) +// -concurrency N DIDs refreshed at once (default 4) +// -trusted-hosts LIST comma-separated bridge PDS URLs; replaces +// TRUSTED_BRIDGE_PDS_HOSTS +// +// Output, one item per line: identity_cache_deleted= when the cache was +// cleared; "differs kind= did= stored= +// verified= reason=" per row that still differs; and a +// totals line, "checked= updated= unchanged= still_differing=" +// (a dry run prints would_update instead of updated). Reason codes: +// resolve-error, unverified, subject-mismatch, handle-conflict, +// invalid-handle, hostedby-mismatch, hostedby-pending (the hostedBy DID +// document could not be fetched in this run, and a transient failure stays +// cached for the run; re-run), lock-busy (another refresh held the DID's lock; +// re-run), and error (any other failure). +// +// On stderr it prints "reconcile-handles: plc_url=" once, naming the PLC +// directory the resolvers use, and one "reconcile-handles: cause +// kind= did= reason=: " line per failed row, +// in the order of the differs lines. +package main + +import ( + "context" + "database/sql" + "flag" + "fmt" + "io" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "Coves/internal/atproto/identity" + "Coves/internal/atproto/jetstream" + "Coves/internal/core/bridgedvotes" +) + +// trustedHostsVariable is the environment variable -trusted-hosts replaces. +const trustedHostsVariable = "TRUSTED_BRIDGE_PDS_HOSTS" + +type sweepOptions struct { + DryRun bool + ClearIdentityCache bool + Concurrency int +} + +type sweepDependencies struct { + Database *sql.DB // selection, writes, identity_cache clear, OAuth session updates + LockDatabase *sql.DB // dedicated pool for the per-DID advisory lock only + CachingResolver identity.Resolver + UncachedResolver identity.UncachedResolver + Trust *jetstream.BridgeTrust + SkipHostedByVerification bool // true skips the whole hostedBy check, domain match included + AllowPrivateHosts bool // IS_DEV_ENV: opens the hostedBy fetch's private-address hatch, as cmd/server does + LockWaitTimeout time.Duration // 0 = jetstream.IdentityRefreshLockWaitTimeout + InstanceDomain string // users.WithInstanceDomain, as cmd/server sets it; "" skips that check + PLCURL string // the PLC directory the resolvers use; printed once on stderr +} + +type sweepReport struct { + ClearedIdentityCacheRows int64 // -1 when the cache was not cleared + Checked int + Updated int // normal run only + WouldUpdate int // dry run only + Unchanged int + StillDiffering int + Differing []differingRow +} + +type differingRow struct { + Kind string // "user" | "community" + DID string + StoredHandle string + VerifiedHandle string // empty when no handle verified + Reason string // "" on a dry-run would-update row; otherwise a reason code + Cause error // the failure behind Reason; nil on a dry-run would-update row +} + +// exitStatus is 0 only when err is nil and no row still differs. +func exitStatus(report sweepReport, err error) int { + if err != nil || report.StillDiffering > 0 { + return 1 + } + return 0 +} + +// openFunc opens the database pools and resolvers. run calls it only after +// every flag/env validation passed. run itself sets dependencies.Trust from the +// effective host list (-trusted-hosts replaces TRUSTED_BRIDGE_PDS_HOSTS; both are +// comma-separated; empty entries are dropped), overwriting whatever open returned. +type openFunc func(ctx context.Context, getenv func(string) string, concurrency int) (sweepDependencies, func(), error) + +// run parses flags (-dry-run, -clear-identity-cache, -concurrency (default 4), +// -trusted-hosts) and env, validates, calls open, then reconcile, then prints +// errors to stderr. Returns the process exit status (open error → non-zero). +func run(ctx context.Context, args []string, getenv func(string) string, stdout, stderr io.Writer, open openFunc) int { + flags := flag.NewFlagSet("reconcile-handles", flag.ContinueOnError) + flags.SetOutput(stderr) + dryRun := flags.Bool("dry-run", false, + "resolve and verify every bridged DID and report what would change; take no lock, purge nothing, write nothing") + clearIdentityCache := flags.Bool("clear-identity-cache", false, + "delete every identity_cache row before the sweep (not allowed with -dry-run)") + concurrency := flags.Int("concurrency", 4, "number of DIDs refreshed at once") + trustedHosts := flags.String("trusted-hosts", "", + "comma-separated trusted bridge PDS URLs (scheme and host); replaces "+trustedHostsVariable) + complain := func(format string, args ...any) { + _, _ = fmt.Fprintf(stderr, "reconcile-handles: "+format+"\n", args...) + } + if err := flags.Parse(args); err != nil { + return 2 + } + if flags.NArg() > 0 { + complain("unexpected arguments: %s", strings.Join(flags.Args(), " ")) + return 2 + } + if *dryRun && *clearIdentityCache { + complain("-dry-run and -clear-identity-cache cannot be combined: a dry run must not change identity_cache") + return 2 + } + if *concurrency < 1 { + complain("-concurrency must be at least 1 (got %d)", *concurrency) + return 2 + } + + rawHosts, hostsSource := getenv(trustedHostsVariable), trustedHostsVariable + flags.Visit(func(f *flag.Flag) { + if f.Name == "trusted-hosts" { + rawHosts, hostsSource = *trustedHosts, "-trusted-hosts" + } + }) + hosts := splitHostList(rawHosts) + if len(hosts) == 0 { + complain("no trusted bridge hosts: set %s or pass -trusted-hosts", trustedHostsVariable) + return 2 + } + for _, host := range hosts { + if _, err := bridgedvotes.ParseTrustedHost(host); err != nil { + complain("%s: %v (scheme and host only, e.g. https://tdpl.io)", hostsSource, err) + return 2 + } + } + + dependencies, closeDependencies, err := open(ctx, getenv, *concurrency) + if err != nil { + complain("%v", err) + return 1 + } + if closeDependencies != nil { + defer closeDependencies() + } + dependencies.Trust = jetstream.NewBridgeTrust(hosts) + complain("plc_url=%s", dashIfEmpty(dependencies.PLCURL)) + + options := sweepOptions{DryRun: *dryRun, ClearIdentityCache: *clearIdentityCache, Concurrency: *concurrency} + report, err := reconcile(ctx, options, dependencies, stdout) + for _, row := range report.Differing { + if row.Cause != nil { + complain("cause kind=%s did=%s reason=%s: %v", row.Kind, row.DID, row.Reason, row.Cause) + } + } + if err != nil { + complain("%v", err) + } + return exitStatus(report, err) +} + +// splitHostList splits a comma-separated host list, trimming each entry and +// dropping empty ones. +func splitHostList(raw string) []string { + var hosts []string + for _, part := range strings.Split(raw, ",") { + if part = strings.TrimSpace(part); part != "" { + hosts = append(hosts, part) + } + } + return hosts +} + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + status := run(ctx, os.Args[1:], os.Getenv, os.Stdout, os.Stderr, openProduction) + stop() + os.Exit(status) +} diff --git a/cmd/reconcile-handles/open.go b/cmd/reconcile-handles/open.go new file mode 100644 index 0000000..0c1ad6d --- /dev/null +++ b/cmd/reconcile-handles/open.go @@ -0,0 +1,95 @@ +package main + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" + + "Coves/internal/atproto/identity" + "Coves/internal/config" + + _ "github.com/lib/pq" +) + +// openProduction loads the AppView's configuration and opens what the sweep +// runs against: the application pool, a separate lock pool, and the caching +// and uncached identity resolvers built the way cmd/server's buildIdentity +// builds them. +// +// config.Load reads the process environment, which is what getenv reads in +// production. DATABASE_URL is required here even though config.Load would +// fall back to the local dev database: an operator tool must not pick a +// database nobody named. +func openProduction(ctx context.Context, getenv func(string) string, concurrency int) (sweepDependencies, func(), error) { + if strings.TrimSpace(getenv("DATABASE_URL")) == "" { + return sweepDependencies{}, nil, errors.New("DATABASE_URL is required") + } + cfg, err := config.Load() + if err != nil { + return sweepDependencies{}, nil, fmt.Errorf("load configuration: %w", err) + } + dsn, err := cfg.Database.AppDSN() + if err != nil { + return sweepDependencies{}, nil, fmt.Errorf("build database DSN: %w", err) + } + + db, err := openPool(ctx, dsn, cfg.Database.MaxOpenConns, cfg.Database.MaxIdleConns, cfg.Database) + if err != nil { + return sweepDependencies{}, nil, fmt.Errorf("open database: %w", err) + } + // Lock waiters must not share a capped pool with the writes that release + // them: every refresh holds one lock connection while it waits or holds, + // so this pool has one connection per concurrent refresh. + lockDB, err := openPool(ctx, dsn, concurrency, concurrency, cfg.Database) + if err != nil { + _ = db.Close() + return sweepDependencies{}, nil, fmt.Errorf("open identity refresh lock database: %w", err) + } + + allowPrivateHosts := cfg.IsDevEnv + identityOptions := identity.PrivateHostOptions(allowPrivateHosts) + if len(cfg.Identity.WellKnownHosts) > 0 { + identityOptions = append(identityOptions, identity.WithWellKnownHosts(cfg.Identity.WellKnownHosts)) + } + identityConfig := identity.DefaultConfig(identityOptions...) + identityConfig.PLCURL = cfg.Identity.ResolverPLCURL + if cfg.Identity.CacheTTL > 0 { + identityConfig.CacheTTL = cfg.Identity.CacheTTL + } + identityConfig.NegativeCacheTTL = cfg.Identity.NegativeCacheTTL + + dependencies := sweepDependencies{ + Database: db, + LockDatabase: lockDB, + CachingResolver: identity.NewResolver(db, identityConfig), + UncachedResolver: identity.NewUncachedResolver(identityConfig), + SkipHostedByVerification: cfg.Instance.SkipDIDWebVerification, + InstanceDomain: cfg.Instance.Domain, + PLCURL: identityConfig.PLCURL, + AllowPrivateHosts: allowPrivateHosts, + } + closeAll := func() { + _ = lockDB.Close() + _ = db.Close() + } + return dependencies, closeAll, nil +} + +// openPool opens and pings a bounded pool on dsn. +func openPool(ctx context.Context, dsn string, maxOpen, maxIdle int, cfg config.DatabaseConfig) (*sql.DB, error) { + db, err := sql.Open("postgres", dsn) + if err != nil { + return nil, err + } + db.SetMaxOpenConns(maxOpen) + db.SetMaxIdleConns(maxIdle) + db.SetConnMaxLifetime(cfg.ConnMaxLifetime) + db.SetConnMaxIdleTime(cfg.ConnMaxIdleTime) + if err := db.PingContext(ctx); err != nil { + _ = db.Close() + return nil, err + } + return db, nil +} diff --git a/cmd/reconcile-handles/reconcile_integration_test.go b/cmd/reconcile-handles/reconcile_integration_test.go new file mode 100644 index 0000000..85fa744 --- /dev/null +++ b/cmd/reconcile-handles/reconcile_integration_test.go @@ -0,0 +1,85 @@ +//go:build integration + +package main + +import ( + "bytes" + "context" + "os" + "testing" + + "Coves/tests/testkit" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestMain(m *testing.M) { + os.Exit(testkit.Main(m, testkit.RequirePostgres)) +} + +// TestReconcile_UpdatesBridgedHandles is the acceptance case: a normal run +// moves a bridged user and a bridged community to their verified handles, +// carries the user's new handle into its OAuth session, purges the cache rows +// for the user's DID and old handle, and prints the totals line. +func TestReconcile_UpdatesBridgedHandles(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + + userDID := uniqueDID(t, "user") + communityDID := uniqueDID(t, "community") + unrelatedDID := uniqueDID(t, "unrelated") + + seedUser(t, db, userDID, "oldalice.bridge.test", bridgeHost) + seedOAuthSession(t, db, userDID, "oldalice.bridge.test", bridgeHost) + seedCommunity(t, db, communityDID, "oldgarden.bridge.test", bridgeHost) + seedIdentityCacheRow(t, db, userDID, userDID, "cachedalice.bridge.test") + seedIdentityCacheRow(t, db, "oldalice.bridge.test", userDID, "oldalice.bridge.test") + seedIdentityCacheRow(t, db, "unrelated.bridge.test", unrelatedDID, "unrelated.bridge.test") + + uncached := newScriptedIdentityResolver(map[string]scriptedAnswer{ + userDID: {handle: "newalice.bridge.test"}, + communityDID: {handle: "newgarden.bridge.test"}, + }) + var out bytes.Buffer + report, err := reconcile(context.Background(), sweepOptions{Concurrency: 4}, newSweepDependencies(t, db, uncached), &out) + + assert.Equal(t, "newalice.bridge.test", storedUserHandle(t, db, userDID), "users row handle") + assert.Equal(t, "newgarden.bridge.test", storedCommunityHandle(t, db, communityDID), "communities row handle") + assert.Equal(t, "newalice.bridge.test", oauthSessionHandle(t, db, userDID), "OAuth session handle") + assert.False(t, identityCacheHasRow(t, db, userDID), "identity_cache row for the user's DID") + assert.False(t, identityCacheHasRow(t, db, "oldalice.bridge.test"), "identity_cache row for the user's old handle") + assert.True(t, identityCacheHasRow(t, db, "unrelated.bridge.test"), "unrelated identity_cache row") + + require.NoError(t, err) + assert.Equal(t, 2, report.Checked, "Checked") + assert.Equal(t, 2, report.Updated, "Updated") + assert.Equal(t, 0, report.Unchanged, "Unchanged") + assert.Equal(t, 0, report.StillDiffering, "StillDiffering") + assert.Equal(t, "checked=2 updated=2 unchanged=0 still_differing=0", lastLine(out.String()), "totals line") + assert.Equal(t, 0, exitStatus(report, err), "exit status") +} + +// TestReconcile_LeavesMatchingHandleUnchanged: a bridged user whose stored +// handle already equals its verified handle is counted unchanged. +func TestReconcile_LeavesMatchingHandleUnchanged(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + + userDID := uniqueDID(t, "user") + seedUser(t, db, userDID, "alice.bridge.test", bridgeHost) + + uncached := newScriptedIdentityResolver(map[string]scriptedAnswer{ + userDID: {handle: "alice.bridge.test"}, + }) + var out bytes.Buffer + report, err := reconcile(context.Background(), sweepOptions{Concurrency: 4}, newSweepDependencies(t, db, uncached), &out) + + require.NoError(t, err) + assert.Equal(t, "alice.bridge.test", storedUserHandle(t, db, userDID), "users row handle") + assert.Equal(t, 1, report.Checked, "Checked") + assert.Equal(t, 1, report.Unchanged, "Unchanged") + assert.Equal(t, 0, report.Updated, "Updated") + assert.Equal(t, 0, report.StillDiffering, "StillDiffering") + assert.Equal(t, 0, exitStatus(report, err), "exit status") +} diff --git a/cmd/reconcile-handles/reconcile_outcomes_integration_test.go b/cmd/reconcile-handles/reconcile_outcomes_integration_test.go new file mode 100644 index 0000000..937be23 --- /dev/null +++ b/cmd/reconcile-handles/reconcile_outcomes_integration_test.go @@ -0,0 +1,285 @@ +//go:build integration + +package main + +import ( + "bytes" + "context" + "database/sql" + "errors" + "strings" + "sync" + "testing" + "time" + + "Coves/internal/atproto/jetstream" + "Coves/tests/testkit" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// nativeHost is a PDS that is not a trusted bridge, so rows on it are native. +const nativeHost = "https://native.test" + +// TestRun_ClearsIdentityCacheBeforeResolving: with -clear-identity-cache, +// every identity_cache row is gone before the first fresh resolution, and the +// number deleted is reported. +func TestRun_ClearsIdentityCacheBeforeResolving(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + + userDID := uniqueDID(t, "user") + seedUser(t, db, userDID, "alice.bridge.test", bridgeHost) + seedIdentityCacheRow(t, db, "first.unrelated.test", uniqueDID(t, "first"), "first.unrelated.test") + seedIdentityCacheRow(t, db, "second.unrelated.test", uniqueDID(t, "second"), "second.unrelated.test") + seedIdentityCacheRow(t, db, "third.unrelated.test", uniqueDID(t, "third"), "third.unrelated.test") + + uncached := newScriptedIdentityResolver(map[string]scriptedAnswer{ + userDID: {handle: "alice.bridge.test"}, + }) + cacheRowsAtFirstResolve := -1 + uncached.onFirstResolve = func(ctx context.Context, _ string) { + assert.NoError(t, db.QueryRowContext(ctx, `SELECT COUNT(*) FROM identity_cache`).Scan(&cacheRowsAtFirstResolve), + "count identity_cache rows at first resolve") + } + dependencies := newSweepDependencies(t, db, uncached) + open := func(context.Context, func(string) string, int) (sweepDependencies, func(), error) { + return dependencies, func() {}, nil + } + getenv := func(string) string { return "" } + var stdout, stderr bytes.Buffer + + status := run(context.Background(), + []string{"-clear-identity-cache", "-concurrency", "4", "-trusted-hosts", bridgeHost}, + getenv, &stdout, &stderr, open) + + assert.Equal(t, 0, cacheRowsAtFirstResolve, "identity_cache rows when the first resolution started") + assert.Contains(t, strings.Split(stdout.String(), "\n"), "identity_cache_deleted=3", "report lines") + assert.Equal(t, "checked=1 updated=0 unchanged=1 still_differing=0", lastLine(stdout.String()), "totals line") + assert.Equal(t, 0, status, "exit status; stderr: %s", stderr.String()) +} + +// TestRun_TrustedHostsFlagReplacesEnvironment: -trusted-hosts replaces +// TRUSTED_BRIDGE_PDS_HOSTS, so rows on the environment's host are native and +// never resolved, while rows on the flag's host are reconciled. +func TestRun_TrustedHostsFlagReplacesEnvironment(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + + nativeUserDID := uniqueDID(t, "native") + nativeCommunityDID := uniqueDID(t, "nativecomm") + bridgedDID := uniqueDID(t, "bridged") + seedUser(t, db, nativeUserDID, "nora.native.test", nativeHost) + seedCommunity(t, db, nativeCommunityDID, "pond.native.test", nativeHost) + seedUser(t, db, bridgedDID, "oldalice.bridge.test", bridgeHost) + + uncached := newScriptedIdentityResolver(map[string]scriptedAnswer{ + nativeUserDID: {handle: "newnora.native.test"}, + nativeCommunityDID: {handle: "newpond.native.test"}, + bridgedDID: {handle: "newalice.bridge.test"}, + }) + dependencies := newSweepDependencies(t, db, uncached) + caching := &recordingResolver{t: t, inner: dependencies.CachingResolver} + dependencies.CachingResolver = caching + dependencies.Trust = nil + open := func(context.Context, func(string) string, int) (sweepDependencies, func(), error) { + return dependencies, func() {}, nil + } + environment := map[string]string{"TRUSTED_BRIDGE_PDS_HOSTS": nativeHost} + getenv := func(key string) string { return environment[key] } + var stdout, stderr bytes.Buffer + + status := run(context.Background(), []string{"-trusted-hosts", bridgeHost}, getenv, &stdout, &stderr, open) + + assert.Equal(t, 0, uncached.callCount(nativeUserDID), "native user resolutions") + assert.Equal(t, 0, uncached.callCount(nativeCommunityDID), "native community resolutions") + assert.NotContains(t, caching.seen(), nativeUserDID, "caching resolver identifiers") + assert.NotContains(t, caching.seen(), nativeCommunityDID, "caching resolver identifiers") + assert.Equal(t, "nora.native.test", storedUserHandle(t, db, nativeUserDID), "native users row handle") + assert.Equal(t, "pond.native.test", storedCommunityHandle(t, db, nativeCommunityDID), "native communities row handle") + assert.Equal(t, "newalice.bridge.test", storedUserHandle(t, db, bridgedDID), "bridged users row handle") + assert.Equal(t, "checked=1 updated=1 unchanged=0 still_differing=0", lastLine(stdout.String()), "totals line") + assert.Equal(t, 0, status, "exit status; stderr: %s", stderr.String()) +} + +// holdIdentityRefreshLock holds did's per-DID refresh lock from a separate +// pool until the returned release runs (also at cleanup). It returns once the +// lock is held. +func holdIdentityRefreshLock(t *testing.T, db *sql.DB, did string) (release func()) { + t.Helper() + lock := jetstream.NewPostgresIdentityRefreshLock(openSecondPool(t, db, "reconcile-handles-holder"), 5*time.Second) + acquired := make(chan struct{}) + released := make(chan struct{}) + holderDone := make(chan struct{}) + var holderErr error + go func() { + defer close(holderDone) + holderErr = lock.WithDIDLock(context.Background(), did, func(context.Context) error { + close(acquired) + <-released + return nil + }) + }() + releaseOnce := sync.OnceFunc(func() { + close(released) + <-holderDone + assert.NoError(t, holderErr, "holder of %s's refresh lock", did) + }) + // Registered after the holder pool, so it runs before that pool closes. + t.Cleanup(releaseOnce) + select { + case <-acquired: + case <-holderDone: + require.Failf(t, "hold refresh lock", "the holder returned before acquiring %s's lock: %v", did, holderErr) + } + return releaseOnce +} + +// TestReconcile_ReportsEachReasonARowStillDiffers: one sweep over rows that +// each fail for a different reason, plus one that updates. Every failed row +// keeps its stored handle and is reported with its reason code. +func TestReconcile_ReportsEachReasonARowStillDiffers(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + + type differing struct { + kind, did, stored, reason string + } + resolveErrorDID := uniqueDID(t, "resolveerr") + unverifiedDID := uniqueDID(t, "unverified") + mismatchDID := uniqueDID(t, "mismatch") + conflictDID := uniqueDID(t, "conflict") + invalidDID := uniqueDID(t, "invalid") + busyDID := uniqueDID(t, "busy") + hostedByDID := uniqueDID(t, "hostedby") + updatableDID := uniqueDID(t, "updatable") + nativeDID := uniqueDID(t, "native") + + rows := []differing{ + {kind: "user", did: resolveErrorDID, stored: "anna.bridge.test", reason: "resolve-error"}, + {kind: "user", did: unverifiedDID, stored: "ben.bridge.test", reason: "unverified"}, + {kind: "user", did: mismatchDID, stored: "cleo.bridge.test", reason: "subject-mismatch"}, + {kind: "user", did: conflictDID, stored: "dora.bridge.test", reason: "handle-conflict"}, + {kind: "user", did: invalidDID, stored: "erin.bridge.test", reason: "invalid-handle"}, + {kind: "user", did: busyDID, stored: "finn.bridge.test", reason: "lock-busy"}, + {kind: "community", did: hostedByDID, stored: "oldgarden.bridge.test", reason: "hostedby-mismatch"}, + } + for _, row := range rows { + if row.kind == "community" { + seedCommunity(t, db, row.did, row.stored, bridgeHost) + } else { + seedUser(t, db, row.did, row.stored, bridgeHost) + } + } + seedUser(t, db, updatableDID, "gina.bridge.test", bridgeHost) + seedUser(t, db, nativeDID, "newdora.bridge.test", nativeHost) + + uncached := newScriptedIdentityResolver(map[string]scriptedAnswer{ + resolveErrorDID: {err: errors.New("scripted resolution failure")}, + unverifiedDID: {handle: "handle.invalid"}, + mismatchDID: {handle: "newcleo.bridge.test", subject: uniqueDID(t, "other")}, + conflictDID: {handle: "newdora.bridge.test"}, + invalidDID: {handle: "new_erin.bridge.test"}, + busyDID: {handle: "newfinn.bridge.test"}, + hostedByDID: {handle: "newgarden.elsewhere.test"}, + updatableDID: {handle: "newgina.bridge.test"}, + }) + dependencies := newSweepDependencies(t, db, uncached) + dependencies.SkipHostedByVerification = false + dependencies.LockWaitTimeout = 300 * time.Millisecond + releaseBusyLock := holdIdentityRefreshLock(t, db, busyDID) + + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + var out bytes.Buffer + report, err := reconcile(ctx, sweepOptions{Concurrency: 4}, dependencies, &out) + releaseBusyLock() + + verifiedByDID := map[string]string{} + for _, row := range rows { + if row.kind == "community" { + assert.Equal(t, row.stored, storedCommunityHandle(t, db, row.did), "%s communities row handle", row.reason) + } else { + assert.Equal(t, row.stored, storedUserHandle(t, db, row.did), "%s users row handle", row.reason) + } + verified, found := differsLineVerified(out.String(), row.kind, row.did, row.stored, row.reason) + assert.True(t, found, "differs line for the %s row in report:\n%s", row.reason, out.String()) + verifiedByDID[row.did] = verified + } + assert.Equal(t, "-", verifiedByDID[resolveErrorDID], "verified field of the resolve-error line") + assert.Equal(t, "-", verifiedByDID[unverifiedDID], "verified field of the unverified line") + assert.Equal(t, "newdora.bridge.test", verifiedByDID[conflictDID], "verified field of the handle-conflict line") + assert.Equal(t, "newgina.bridge.test", storedUserHandle(t, db, updatableDID), "updatable users row handle") + assert.Equal(t, "newdora.bridge.test", storedUserHandle(t, db, nativeDID), "native users row handle") + + assert.Equal(t, 8, report.Checked, "Checked") + assert.Equal(t, 1, report.Updated, "Updated") + assert.Equal(t, 7, report.StillDiffering, "StillDiffering") + wantDiffering := map[string]differing{} + for _, row := range rows { + wantDiffering[row.did] = row + } + gotDiffering := map[string]differing{} + for _, row := range report.Differing { + gotDiffering[row.DID] = differing{kind: row.Kind, did: row.DID, stored: row.StoredHandle, reason: row.Reason} + } + assert.Equal(t, wantDiffering, gotDiffering, "Differing rows") + assert.NotZero(t, exitStatus(report, err), "exit status") +} + +// TestRun_DryRunWritesNothing: -dry-run reports what it would change, lists +// every row that differs or fails, writes nothing (no row, cache entry or +// session changes, and the caching resolver is never touched) and exits +// non-zero because rows still differ. +func TestRun_DryRunWritesNothing(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + + staleUserDID := uniqueDID(t, "staleuser") + staleCommunityDID := uniqueDID(t, "stalecomm") + matchingDID := uniqueDID(t, "matching") + failingDID := uniqueDID(t, "failing") + seedUser(t, db, staleUserDID, "oldalice.bridge.test", bridgeHost) + seedIdentityCacheRow(t, db, staleUserDID, staleUserDID, "oldalice.bridge.test") + seedOAuthSession(t, db, staleUserDID, "oldalice.bridge.test", bridgeHost) + seedCommunity(t, db, staleCommunityDID, "oldgarden.bridge.test", bridgeHost) + seedUser(t, db, matchingDID, "bob.bridge.test", bridgeHost) + seedUser(t, db, failingDID, "carl.bridge.test", bridgeHost) + + uncached := newScriptedIdentityResolver(map[string]scriptedAnswer{ + staleUserDID: {handle: "newalice.bridge.test"}, + staleCommunityDID: {handle: "newgarden.bridge.test"}, + matchingDID: {handle: "bob.bridge.test"}, + failingDID: {err: errors.New("scripted resolution failure")}, + }) + dependencies := newSweepDependencies(t, db, uncached) + caching := &recordingResolver{t: t} + dependencies.CachingResolver = caching + open := func(context.Context, func(string) string, int) (sweepDependencies, func(), error) { + return dependencies, func() {}, nil + } + getenv := func(string) string { return "" } + var stdout, stderr bytes.Buffer + + status := run(context.Background(), []string{"-dry-run", "-concurrency", "4", "-trusted-hosts", bridgeHost}, + getenv, &stdout, &stderr, open) + + assert.Equal(t, "oldalice.bridge.test", storedUserHandle(t, db, staleUserDID), "stale users row handle") + assert.Equal(t, "oldgarden.bridge.test", storedCommunityHandle(t, db, staleCommunityDID), "stale communities row handle") + assert.Equal(t, "bob.bridge.test", storedUserHandle(t, db, matchingDID), "matching users row handle") + assert.Equal(t, "carl.bridge.test", storedUserHandle(t, db, failingDID), "failing users row handle") + assert.Equal(t, 1, identityCacheRowCount(t, db), "identity_cache rows") + cachedDID, cachedHandle := identityCacheRowContents(t, db, staleUserDID) + assert.Equal(t, staleUserDID, cachedDID, "identity_cache row did") + assert.Equal(t, "oldalice.bridge.test", cachedHandle, "identity_cache row handle") + assert.Equal(t, "oldalice.bridge.test", oauthSessionHandle(t, db, staleUserDID), "OAuth session handle") + assert.Empty(t, caching.seen(), "caching resolver identifiers") + + assert.NotContains(t, stdout.String(), " reason=error\n", "report lines (a reason=error row means the sweep returned an error)") + assert.Contains(t, strings.Split(stdout.String(), "\n"), + "differs kind=user did="+staleUserDID+" stored=oldalice.bridge.test verified=newalice.bridge.test reason=-", + "report lines") + assert.Equal(t, "checked=4 would_update=2 unchanged=1 still_differing=3", lastLine(stdout.String()), "totals line") + assert.NotEqual(t, 0, status, "exit status; stderr: %s", stderr.String()) +} diff --git a/cmd/reconcile-handles/run_stderr_integration_test.go b/cmd/reconcile-handles/run_stderr_integration_test.go new file mode 100644 index 0000000..63c1cfa --- /dev/null +++ b/cmd/reconcile-handles/run_stderr_integration_test.go @@ -0,0 +1,64 @@ +//go:build integration + +package main + +import ( + "bytes" + "context" + "errors" + "strings" + "testing" + + "Coves/tests/testkit" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestRun_PrintsPLCURLAndEachFailedRowsCause: in a normal run and in a dry +// run, run prints the PLC directory the resolvers use once on stderr, and +// exactly one stderr cause line per failed row, carrying the row's error. +func TestRun_PrintsPLCURLAndEachFailedRowsCause(t *testing.T) { + t.Parallel() + const plcURL = "http://plc.reconcile.test" + cases := []struct { + name string + args []string + }{ + {name: "normal run", args: []string{"-trusted-hosts", bridgeHost}}, + {name: "dry run", args: []string{"-dry-run", "-trusted-hosts", bridgeHost}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + failingDID := uniqueDID(t, "failing") + seedUser(t, db, failingDID, "carl.bridge.test", bridgeHost) + uncached := newScriptedIdentityResolver(map[string]scriptedAnswer{ + failingDID: {err: errors.New("scripted resolution failure: directory unreachable")}, + }) + dependencies := newSweepDependencies(t, db, uncached) + dependencies.PLCURL = plcURL + open := func(context.Context, func(string) string, int) (sweepDependencies, func(), error) { + return dependencies, func() {}, nil + } + getenv := func(string) string { return "" } + var stdout, stderr bytes.Buffer + + status := run(context.Background(), tc.args, getenv, &stdout, &stderr, open) + + assert.Equal(t, 1, strings.Count(stderr.String(), "plc_url="+plcURL+"\n"), "plc_url lines; stderr:\n%s", stderr.String()) + var causeLines []string + for _, line := range strings.Split(stderr.String(), "\n") { + if strings.Contains(line, failingDID) { + causeLines = append(causeLines, line) + } + } + require.Len(t, causeLines, 1, "stderr lines naming the failed row; stderr:\n%s", stderr.String()) + assert.True(t, strings.HasPrefix(causeLines[0], + "reconcile-handles: cause kind=user did="+failingDID+" reason=resolve-error: "), "cause line: %s", causeLines[0]) + assert.Contains(t, causeLines[0], "scripted resolution failure: directory unreachable", "cause line") + assert.NotZero(t, status, "exit status") + }) + } +} diff --git a/cmd/reconcile-handles/run_test.go b/cmd/reconcile-handles/run_test.go new file mode 100644 index 0000000..68642c9 --- /dev/null +++ b/cmd/reconcile-handles/run_test.go @@ -0,0 +1,74 @@ +package main + +import ( + "bytes" + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" +) + +// TestRun_RefusesBeforeOpeningOrFailsOnOpenError: invalid flag combinations +// and a missing trusted-host list are refused with a non-zero exit before +// anything is opened; an open failure after valid settings exits non-zero. +func TestRun_RefusesBeforeOpeningOrFailsOnOpenError(t *testing.T) { + t.Parallel() + const hostsVariable = "TRUSTED_BRIDGE_PDS_HOSTS" + hostsSet := map[string]string{hostsVariable: "https://bridge.test"} + + cases := []struct { + name string + args []string + env map[string]string + openErr error + wantOpenCalled bool + wantStderr []string + }{ + { + name: "dry run with cache clear", + args: []string{"-dry-run", "-clear-identity-cache"}, + env: hostsSet, + wantStderr: []string{"-dry-run", "-clear-identity-cache"}, + }, + { + name: "no trusted hosts configured", + args: nil, + env: map[string]string{}, + wantStderr: []string{hostsVariable}, + }, + { + name: "trusted-hosts flag with only empty entries", + args: []string{"-trusted-hosts", " , "}, + env: map[string]string{}, + wantStderr: []string{hostsVariable}, + }, + { + name: "open fails", + args: nil, + env: hostsSet, + openErr: errors.New("open database: connection refused"), + wantOpenCalled: true, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + openCalled := false + open := func(context.Context, func(string) string, int) (sweepDependencies, func(), error) { + openCalled = true + return sweepDependencies{}, func() {}, tc.openErr + } + getenv := func(key string) string { return tc.env[key] } + var stdout, stderr bytes.Buffer + + status := run(context.Background(), tc.args, getenv, &stdout, &stderr, open) + + assert.NotZero(t, status, "exit status") + assert.Equal(t, tc.wantOpenCalled, openCalled, "open called") + for _, want := range tc.wantStderr { + assert.Contains(t, stderr.String(), want, "stderr") + } + }) + } +} diff --git a/cmd/reconcile-handles/sweep.go b/cmd/reconcile-handles/sweep.go new file mode 100644 index 0000000..2bbadc7 --- /dev/null +++ b/cmd/reconcile-handles/sweep.go @@ -0,0 +1,441 @@ +package main + +import ( + "context" + "crypto/rand" + "database/sql" + "errors" + "fmt" + "io" + "sort" + "sync" + + "Coves/internal/atproto/identity" + "Coves/internal/atproto/jetstream" + covesoauth "Coves/internal/atproto/oauth" + "Coves/internal/core/communities" + "Coves/internal/core/users" + "Coves/internal/crypto/credentialcipher" + "Coves/internal/db/postgres" +) + +const ( + kindUser = "user" + kindCommunity = "community" +) + +// Reason codes printed for a row that still differs. +const ( + reasonResolveError = "resolve-error" + reasonUnverified = "unverified" + reasonSubjectMismatch = "subject-mismatch" + reasonHandleConflict = "handle-conflict" + reasonInvalidHandle = "invalid-handle" + reasonHostedByMismatch = "hostedby-mismatch" + reasonLockBusy = "lock-busy" + // reasonHostedByPending: the community's hostedBy DID document could not + // be fetched in this run, even when retried alone after the parallel pass. + reasonHostedByPending = "hostedby-pending" + // reasonError is any failure outside the codes above (a database error, a + // purge failure, a hostedBy document that could not be fetched). reconcile + // returns an error counting these rows. + reasonError = "error" +) + +// bridgedRow is one selected users or communities row. +type bridgedRow struct { + kind string + did string + handle string +} + +type rowStatus int + +const ( + rowUnchanged rowStatus = iota + rowUpdated + rowWouldUpdate + rowFailed +) + +// rowOutcome is what the sweep established for one row. +type rowOutcome struct { + differing differingRow // set for rowWouldUpdate and rowFailed + status rowStatus +} + +// handleRefresher runs one DID's locked handle refresh. +type handleRefresher func(ctx context.Context, did string) (jetstream.IdentityHandleRefresh, error) + +// reconcile runs one sweep and writes the human report to out. +func reconcile(ctx context.Context, options sweepOptions, dependencies sweepDependencies, out io.Writer) (sweepReport, error) { + report := sweepReport{ClearedIdentityCacheRows: -1} + if err := checkSweep(options, dependencies); err != nil { + return report, err + } + + var inspect func(context.Context, bridgedRow) rowOutcome + if options.DryRun { + inspect = func(ctx context.Context, row bridgedRow) rowOutcome { + return inspectRow(ctx, dependencies.UncachedResolver, row) + } + } else { + refreshers, err := newHandleRefreshers(dependencies) + if err != nil { + return report, err + } + inspect = func(ctx context.Context, row bridgedRow) rowOutcome { + return refreshRow(ctx, refreshers[row.kind], row) + } + } + + printer := &reportPrinter{out: out} + if options.ClearIdentityCache { + deleted, err := clearIdentityCache(ctx, dependencies.Database) + if err != nil { + return report, err + } + report.ClearedIdentityCacheRows = deleted + printer.printf("identity_cache_deleted=%d\n", deleted) + } + + rows, err := selectBridgedRows(ctx, dependencies.Database, dependencies.Trust) + if err != nil { + return report, err + } + report.Checked = len(rows) + + outcomes := sweepRows(ctx, rows, options.Concurrency, inspect) + if err := ctx.Err(); err != nil { + return report, fmt.Errorf("sweep interrupted before every row was checked: %w", err) + } + + failedWithError := 0 + for _, outcome := range outcomes { + switch outcome.status { + case rowUnchanged: + report.Unchanged++ + case rowUpdated: + report.Updated++ + case rowWouldUpdate: + report.WouldUpdate++ + report.StillDiffering++ + report.Differing = append(report.Differing, outcome.differing) + case rowFailed: + report.StillDiffering++ + report.Differing = append(report.Differing, outcome.differing) + if outcome.differing.Reason == reasonError { + failedWithError++ + } + } + } + sort.Slice(report.Differing, func(i, j int) bool { + if report.Differing[i].Kind != report.Differing[j].Kind { + return report.Differing[i].Kind < report.Differing[j].Kind + } + return report.Differing[i].DID < report.Differing[j].DID + }) + + for _, row := range report.Differing { + printer.printf("differs kind=%s did=%s stored=%s verified=%s reason=%s\n", + row.Kind, row.DID, dashIfEmpty(row.StoredHandle), dashIfEmpty(row.VerifiedHandle), dashIfEmpty(row.Reason)) + } + if options.DryRun { + printer.printf("checked=%d would_update=%d unchanged=%d still_differing=%d\n", + report.Checked, report.WouldUpdate, report.Unchanged, report.StillDiffering) + } else { + printer.printf("checked=%d updated=%d unchanged=%d still_differing=%d\n", + report.Checked, report.Updated, report.Unchanged, report.StillDiffering) + } + + var failures []error + if failedWithError > 0 { + failures = append(failures, fmt.Errorf("%d row(s) failed with reason %s", failedWithError, reasonError)) + } + if printer.err != nil { + failures = append(failures, fmt.Errorf("write report: %w", printer.err)) + } + return report, errors.Join(failures...) +} + +// checkSweep refuses a sweep its options or dependencies cannot run safely. +func checkSweep(options sweepOptions, dependencies sweepDependencies) error { + if options.DryRun && options.ClearIdentityCache { + return errors.New("-dry-run and -clear-identity-cache cannot be combined") + } + if options.Concurrency < 1 { + return fmt.Errorf("concurrency must be at least 1 (got %d)", options.Concurrency) + } + if dependencies.Database == nil { + return errors.New("no database") + } + if dependencies.UncachedResolver == nil { + return errors.New("no uncached identity resolver") + } + if dependencies.Trust == nil { + return errors.New("no trusted bridge hosts") + } + if options.DryRun { + return nil + } + if dependencies.LockDatabase == nil { + return errors.New("no identity refresh lock database") + } + if dependencies.CachingResolver == nil { + return errors.New("no caching identity resolver") + } + if dependencies.LockWaitTimeout < 0 { + return fmt.Errorf("lock wait timeout must not be negative (got %s)", dependencies.LockWaitTimeout) + } + return nil +} + +// newHandleRefreshers builds the identity-event consumers' own refresh entry +// points, one per row kind, sharing one per-DID lock on the lock pool. +func newHandleRefreshers(dependencies sweepDependencies) (map[string]handleRefresher, error) { + wait := dependencies.LockWaitTimeout + if wait == 0 { + wait = jetstream.IdentityRefreshLockWaitTimeout + } + lock := jetstream.NewPostgresIdentityRefreshLock(dependencies.LockDatabase, wait) + + sessionStore := covesoauth.NewMobileAwareStoreWrapper(covesoauth.NewPostgresOAuthStore(dependencies.Database, 0)). + UnwrapPostgresStore() + if sessionStore == nil { + return nil, errors.New("OAuth session store is not PostgreSQL-backed, so session handles cannot be updated") + } + userService := users.NewUserService(postgres.NewUserRepository(dependencies.Database), dependencies.CachingResolver, + "", nil, "", users.WithInstanceDomain(dependencies.InstanceDomain)) + userConsumer := jetstream.NewUserEventConsumer(userService, dependencies.CachingResolver, + jetstream.WithUserUncachedIdentityResolver(dependencies.UncachedResolver), + jetstream.WithUserIdentityRefreshLock(lock), + jetstream.WithSessionHandleUpdater(sessionStore), + ) + + // The community refresh reads the row through the repository, which + // requires a credential cipher. The sweep never reads community + // credentials, and bridged communities hold none, so it gets a cipher with + // a fresh random key: a row that does carry credentials fails to decrypt + // and is reported, rather than being read with the AppView's key. + cipherKey := make([]byte, 32) + if _, err := rand.Read(cipherKey); err != nil { + return nil, fmt.Errorf("generate community repository cipher key: %w", err) + } + cipher, err := credentialcipher.New(cipherKey) + if err != nil { + return nil, fmt.Errorf("build community repository cipher: %w", err) + } + communityOptions := append([]jetstream.CommunityConsumerOption{ + jetstream.WithCommunityUncachedIdentityResolver(dependencies.UncachedResolver), + jetstream.WithCommunityIdentityRefreshLock(lock), + }, jetstream.PrivateHostOptions(dependencies.AllowPrivateHosts)...) + communityConsumer := jetstream.NewCommunityEventConsumer( + postgres.NewCommunityRepository(dependencies.Database, cipher), "", + dependencies.SkipHostedByVerification, dependencies.CachingResolver, communityOptions...) + + return map[string]handleRefresher{ + kindUser: userConsumer.RefreshUserHandle, + kindCommunity: communityConsumer.RefreshCommunityHandle, + }, nil +} + +// clearIdentityCache deletes every identity_cache row and returns how many. +func clearIdentityCache(ctx context.Context, db *sql.DB) (int64, error) { + result, err := db.ExecContext(ctx, `DELETE FROM identity_cache`) + if err != nil { + return 0, fmt.Errorf("clear identity_cache: %w", err) + } + deleted, err := result.RowsAffected() + if err != nil { + return 0, fmt.Errorf("count cleared identity_cache rows: %w", err) + } + return deleted, nil +} + +// selectBridgedRows returns every users and communities row whose stored +// pds_url is a trusted bridge host. +func selectBridgedRows(ctx context.Context, db *sql.DB, trust *jetstream.BridgeTrust) ([]bridgedRow, error) { + var selected []bridgedRow + for _, table := range []struct { + kind string + query string + }{ + {kindUser, `SELECT did, handle, COALESCE(pds_url, '') FROM users ORDER BY did`}, + {kindCommunity, `SELECT did, handle, COALESCE(pds_url, '') FROM communities ORDER BY did`}, + } { + rows, err := selectTrustedRows(ctx, db, trust, table.kind, table.query) + if err != nil { + return nil, err + } + selected = append(selected, rows...) + } + return selected, nil +} + +func selectTrustedRows(ctx context.Context, db *sql.DB, trust *jetstream.BridgeTrust, kind, query string) ([]bridgedRow, error) { + rows, err := db.QueryContext(ctx, query) + if err != nil { + return nil, fmt.Errorf("select %s rows: %w", kind, err) + } + defer func() { _ = rows.Close() }() + + var selected []bridgedRow + for rows.Next() { + var did, handle, pdsURL string + if err := rows.Scan(&did, &handle, &pdsURL); err != nil { + return nil, fmt.Errorf("read %s row: %w", kind, err) + } + if trust.TrustsPDS(pdsURL) { + selected = append(selected, bridgedRow{kind: kind, did: did, handle: handle}) + } + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("select %s rows: %w", kind, err) + } + return selected, nil +} + +// sweepRows inspects every row, then re-inspects each hostedby-pending row +// alone, in row order. Communities share one hostedBy verification per +// (hostedBy DID, handle domain), so in the parallel pass every refresh but the +// one fetching the DID document is answered "pending"; a row still pending +// after its retry keeps that reason. +func sweepRows(ctx context.Context, rows []bridgedRow, concurrency int, + inspect func(context.Context, bridgedRow) rowOutcome, +) []rowOutcome { + outcomes := inspectRows(ctx, rows, concurrency, inspect) + for index, outcome := range outcomes { + if ctx.Err() != nil { + break + } + if outcome.differing.Reason == reasonHostedByPending { + outcomes[index] = inspect(ctx, rows[index]) + } + } + return outcomes +} + +// inspectRows runs inspect over rows with at most concurrency at once. Rows +// not started before ctx ends keep a zero outcome; the caller checks ctx. +func inspectRows(ctx context.Context, rows []bridgedRow, concurrency int, + inspect func(context.Context, bridgedRow) rowOutcome, +) []rowOutcome { + outcomes := make([]rowOutcome, len(rows)) + indexes := make(chan int) + var workers sync.WaitGroup + for range min(concurrency, len(rows)) { + workers.Add(1) + go func() { + defer workers.Done() + for index := range indexes { + outcomes[index] = inspect(ctx, rows[index]) + } + }() + } +feed: + for index := range rows { + select { + case indexes <- index: + case <-ctx.Done(): + break feed + } + } + close(indexes) + workers.Wait() + return outcomes +} + +// refreshRow runs the locked refresh for one row and classifies the result. +func refreshRow(ctx context.Context, refresh handleRefresher, row bridgedRow) rowOutcome { + result, err := refresh(ctx, row.did) + if err == nil { + if result.Changed { + return rowOutcome{status: rowUpdated} + } + return rowOutcome{status: rowUnchanged} + } + + stored := result.StoredHandle + if stored == "" { + stored = row.handle + } + return rowOutcome{ + status: rowFailed, + differing: differingRow{ + Kind: row.kind, DID: row.did, StoredHandle: stored, + VerifiedHandle: result.VerifiedHandle, Reason: refreshFailureReason(err), Cause: err, + }, + } +} + +// refreshFailureReason maps a refresh error to its reason code. +func refreshFailureReason(err error) string { + switch { + case errors.Is(err, jetstream.ErrIdentityRefreshBusy): + return reasonLockBusy + case errors.Is(err, jetstream.ErrIdentityResolveFailed): + return reasonResolveError + case errors.Is(err, identity.ErrIdentitySubjectMismatch): + return reasonSubjectMismatch + case errors.Is(err, identity.ErrHandleUnverified): + return reasonUnverified + case errors.Is(err, jetstream.ErrVerifiedHandleInvalid): + return reasonInvalidHandle + case errors.Is(err, jetstream.ErrHostedByMismatch): + return reasonHostedByMismatch + case errors.Is(err, jetstream.ErrHostedByVerificationPending): + return reasonHostedByPending + case errors.Is(err, users.ErrHandleAlreadyTaken), errors.Is(err, communities.ErrHandleTaken): + return reasonHandleConflict + default: + return reasonError + } +} + +// inspectRow is the dry run's read-only check: a fresh uncached resolution and +// handle verification compared with the stored handle. No lock, no purge, no +// write. +func inspectRow(ctx context.Context, resolver identity.UncachedResolver, row bridgedRow) rowOutcome { + failed := func(reason string, cause error) rowOutcome { + return rowOutcome{status: rowFailed, differing: differingRow{ + Kind: row.kind, DID: row.did, StoredHandle: row.handle, Reason: reason, Cause: cause, + }} + } + resolved, err := resolver.Resolve(ctx, row.did) + if err != nil { + return failed(reasonResolveError, fmt.Errorf("resolve identity: %w", err)) + } + verified, err := identity.VerifiedHandle(resolved, row.did) + if errors.Is(err, identity.ErrIdentitySubjectMismatch) { + return failed(reasonSubjectMismatch, fmt.Errorf("verify handle: %w", err)) + } + if err != nil { + return failed(reasonUnverified, fmt.Errorf("verify handle: %w", err)) + } + if verified == row.handle { + return rowOutcome{status: rowUnchanged} + } + return rowOutcome{status: rowWouldUpdate, differing: differingRow{ + Kind: row.kind, DID: row.did, StoredHandle: row.handle, VerifiedHandle: verified, + }} +} + +// reportPrinter writes report lines and keeps the first write error. +type reportPrinter struct { + out io.Writer + err error +} + +func (p *reportPrinter) printf(format string, args ...any) { + if p.err != nil { + return + } + _, p.err = fmt.Fprintf(p.out, format, args...) +} + +func dashIfEmpty(value string) string { + if value == "" { + return "-" + } + return value +} diff --git a/cmd/reconcile-handles/sweep_test.go b/cmd/reconcile-handles/sweep_test.go new file mode 100644 index 0000000..cfad94d --- /dev/null +++ b/cmd/reconcile-handles/sweep_test.go @@ -0,0 +1,95 @@ +package main + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "Coves/internal/atproto/jetstream" + + "github.com/stretchr/testify/assert" +) + +// TestSweepRows_RetriesHostedByPendingRowsAlone: community refreshes answered +// "hostedBy verification pending" in the parallel pass (another refresh holds +// the shared verification in flight) are retried after it, one row at a time. +// A retry that succeeds counts as updated; a row still pending after its retry +// is reported as hostedby-pending, not error. +func TestSweepRows_RetriesHostedByPendingRowsAlone(t *testing.T) { + t.Parallel() + pending := fmt.Errorf("unresolved reference: bidirectional DID verification required: %w", + jetstream.ErrHostedByVerificationPending) + rows := []bridgedRow{ + {kind: kindCommunity, did: "did:plc:firstpond", handle: "firstpond.bridge.test"}, + {kind: kindCommunity, did: "did:plc:secondpond", handle: "secondpond.bridge.test"}, + {kind: kindCommunity, did: "did:plc:thirdpond", handle: "thirdpond.bridge.test"}, + } + cases := []struct { + name string + retrySucceeds bool + wantStatus rowStatus + wantReason string + }{ + {name: "retry alone succeeds", retrySucceeds: true, wantStatus: rowUpdated, wantReason: ""}, + {name: "still pending after retry", retrySucceeds: false, wantStatus: rowFailed, wantReason: "hostedby-pending"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + var mu sync.Mutex + calls := map[string]int{} + active, mostActiveDuringRetry := 0, 0 + var firstCallsStarted sync.WaitGroup + firstCallsStarted.Add(len(rows)) + everyFirstCallStarted := make(chan struct{}) + go func() { + firstCallsStarted.Wait() + close(everyFirstCallStarted) + }() + refresh := func(ctx context.Context, did string) (jetstream.IdentityHandleRefresh, error) { + mu.Lock() + calls[did]++ + call := calls[did] + active++ + if call > 1 { + mostActiveDuringRetry = max(mostActiveDuringRetry, active) + } + mu.Unlock() + defer func() { + mu.Lock() + active-- + mu.Unlock() + }() + if call == 1 { + // Every row's first refresh overlaps the others, as in the parallel pass. + firstCallsStarted.Done() + select { + case <-everyFirstCallStarted: + case <-ctx.Done(): + } + return jetstream.IdentityHandleRefresh{}, pending + } + if tc.retrySucceeds { + return jetstream.IdentityHandleRefresh{VerifiedHandle: "new" + did, Changed: true}, nil + } + return jetstream.IdentityHandleRefresh{}, pending + } + + outcomes := sweepRows(ctx, rows, len(rows), func(ctx context.Context, row bridgedRow) rowOutcome { + return refreshRow(ctx, refresh, row) + }) + + for index, row := range rows { + assert.Equal(t, tc.wantStatus, outcomes[index].status, "%s status", row.did) + assert.Equal(t, tc.wantReason, outcomes[index].differing.Reason, "%s reason", row.did) + assert.Equal(t, 2, calls[row.did], "%s refreshes", row.did) + } + assert.Equal(t, 1, mostActiveDuringRetry, "refreshes running at once during the retries") + }) + } +} diff --git a/cmd/server/consumers.go b/cmd/server/consumers.go index bbc9f1e..a8e17c5 100644 --- a/cmd/server/consumers.go +++ b/cmd/server/consumers.go @@ -134,6 +134,10 @@ func (s *consumerSet) start(ctx context.Context, wg *sync.WaitGroup, app *applic func (a *application) registerFeedConsumers() ([]feedConsumer, error) { var consumers []feedConsumer + // One lock instance serializes identity refreshes of a DID for the user + // and community consumers alike. + identityRefreshLock := jetstream.NewPostgresIdentityRefreshLock(a.identityRefreshLockDB, jetstream.IdentityRefreshLockWaitTimeout) + // Users: actor profiles and actor blocks. // // A trusted bridge hosts many virtual repos, and its profile records may @@ -144,6 +148,9 @@ func (a *application) registerFeedConsumers() ([]feedConsumer, error) { jetstream.WithUserBridgeTrust(a.bridgeTrust), jetstream.WithUserRevGate(a.revGate), jetstream.WithUserBlockRepo(a.userBlockRepo), + jetstream.WithUserUncachedIdentityResolver(a.uncachedIdentityResolver), + jetstream.WithUserIdentityRefreshFailureTTL(a.cfg.Identity.NegativeCacheTTL), + jetstream.WithUserIdentityRefreshLock(identityRefreshLock), } // Statically typed rather than asserted at runtime: a type assertion here // would silently disable handle sync if the store's method set ever @@ -185,6 +192,9 @@ func (a *application) registerFeedConsumers() ([]feedConsumer, error) { []jetstream.CommunityConsumerOption{ jetstream.WithCommunityRevGate(a.revGate), jetstream.WithCommunityBridgeTrust(a.bridgeTrust), + jetstream.WithCommunityUncachedIdentityResolver(a.uncachedIdentityResolver), + jetstream.WithCommunityIdentityRefreshFailureTTL(a.cfg.Identity.NegativeCacheTTL), + jetstream.WithCommunityIdentityRefreshLock(identityRefreshLock), }, jetstream.PrivateHostOptions(a.allowPrivateHosts())...) consumers = append(consumers, feedConsumer{ diff --git a/cmd/server/database.go b/cmd/server/database.go index c260e74..332d5cc 100644 --- a/cmd/server/database.go +++ b/cmd/server/database.go @@ -110,6 +110,49 @@ func openSessionCoordinationDatabase(ctx context.Context, cfg config.DatabaseCon return db, nil } +// identityRefreshLockMaxOpenConns caps the identity refresh lock pool. Each +// in-flight identity refresh holds one connection from it while it waits for +// or holds its DID's lock, and a process runs at most a few refreshes at once +// (one per feed's consumer, plus the dead letter redriver). When it is full, +// further refreshes queue for a connection under their own context. +const identityRefreshLockMaxOpenConns = 8 + +// openIdentityRefreshLockDatabase returns the small dedicated pool that +// identity refreshes hold their per-DID advisory locks on. Keeping the waiting +// and holding connections off the application pool means refreshes queued on +// one DID cannot starve the handle write that would release them. Each lock +// wait is bounded by the lock_timeout the lock sets on its connection +// (jetstream.IdentityRefreshLockWaitTimeout). +// +// The caller owns the returned pool and must Close it. +func openIdentityRefreshLockDatabase(ctx context.Context, cfg config.DatabaseConfig) (*sql.DB, error) { + dsn, err := cfg.AppDSN() + if err != nil { + return nil, fmt.Errorf("building identity refresh lock DSN: %w", err) + } + + db, err := sql.Open("postgres", dsn) + if err != nil { + return nil, fmt.Errorf("opening identity refresh lock database: %w", err) + } + db.SetMaxOpenConns(identityRefreshLockMaxOpenConns) + db.SetMaxIdleConns(identityRefreshLockMaxOpenConns) + db.SetConnMaxLifetime(cfg.ConnMaxLifetime) + db.SetConnMaxIdleTime(cfg.ConnMaxIdleTime) + + if err := db.PingContext(ctx); err != nil { + if closeErr := db.Close(); closeErr != nil { + slog.Error("failed to close identity refresh lock pool after failed ping", "error", closeErr) + } + return nil, fmt.Errorf("pinging identity refresh lock database: %w", err) + } + + slog.Info("connected identity refresh lock pool", + "max_open_conns", identityRefreshLockMaxOpenConns, + ) + return db, nil +} + // runMigrations applies pending migrations on a short-lived connection that // carries no statement timeout. Before goose runs, legacy credentials are // resealed with the application cipher because migration 046 refuses to drop diff --git a/cmd/server/main.go b/cmd/server/main.go index 265e71d..5bfd98d 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -79,6 +79,16 @@ func run() error { } }() + identityRefreshLockDB, err := openIdentityRefreshLockDatabase(ctx, cfg.Database) + if err != nil { + return err + } + defer func() { + if closeErr := identityRefreshLockDB.Close(); closeErr != nil { + slog.Error("failed to close identity refresh lock pool", "error", closeErr) + } + }() + otelProvider, err := startObservability(ctx) if err != nil { return err @@ -92,7 +102,7 @@ func run() error { } }() - app, err := buildApplication(ctx, cfg, db, sessionCoordinationDB, credentialCipher) + app, err := buildApplication(ctx, cfg, db, sessionCoordinationDB, identityRefreshLockDB, credentialCipher) if err != nil { return err } diff --git a/cmd/server/uncached_identity_resolver_wiring_test.go b/cmd/server/uncached_identity_resolver_wiring_test.go new file mode 100644 index 0000000..62f944c --- /dev/null +++ b/cmd/server/uncached_identity_resolver_wiring_test.go @@ -0,0 +1,105 @@ +package main + +import ( + "database/sql" + "testing" + + "Coves/internal/atproto/identity" + "Coves/internal/atproto/jetstream" + "Coves/internal/atproto/oauth" + "Coves/internal/config" + + "github.com/stretchr/testify/require" +) + +// unusedUncachedIdentityResolver satisfies identity.UncachedResolver without +// any implementation: building the consumers must not resolve anything, and a +// call would panic on the nil embedded interface rather than reach a network. +type unusedUncachedIdentityResolver struct{ identity.UncachedResolver } + +// unusedIdentityResolver is the cached-resolver counterpart, present only +// because the community consumer refuses to be built without one. +type unusedIdentityResolver struct{ identity.Resolver } + +// Without this test, deleting either WithUserUncachedIdentityResolver or +// WithCommunityUncachedIdentityResolver from registerFeedConsumers leaves every +// test green while production answers identity events and origin display +// fields from a cached identity. +func TestRegisterFeedConsumers_UserAndCommunityConsumersGetTheUncachedIdentityResolver(t *testing.T) { + a := &application{ + cfg: &config.Config{}, + identityResolver: unusedIdentityResolver{}, + uncachedIdentityResolver: unusedUncachedIdentityResolver{}, + oauthStore: oauth.NewMobileAwareStoreWrapper(nil), + } + + consumers, err := a.registerFeedConsumers() + require.NoError(t, err) + + var userConsumer *jetstream.UserEventConsumer + var communityConsumer *jetstream.CommunityEventConsumer + for _, consumer := range consumers { + switch consumer.name { + case jetstream.ConsumerUsers: + userConsumer = consumer.handler.(*jetstream.UserEventConsumer) + case jetstream.ConsumerCommunities: + communityConsumer = consumer.handler.(*jetstream.CommunityEventConsumer) + } + } + require.NotNil(t, userConsumer, "the user consumer must be registered") + require.NotNil(t, communityConsumer, "the community consumer must be registered") + + require.True(t, userConsumer.HasUncachedIdentityResolver(), + "the user consumer must refresh identity events through the uncached resolver") + require.True(t, communityConsumer.HasUncachedIdentityResolver(), + "the community consumer must verify origin display fields through the uncached resolver") +} + +// Without this test, dropping WithUserIdentityRefreshLock or +// WithCommunityIdentityRefreshLock from registerFeedConsumers leaves every +// unit test green while production refuses +// every identity refresh (the consumer fails closed without a lock), or, were +// the fail-closed check ever removed, lets two processes interleave refreshes of +// one DID and store the older handle. +// +// The pools never dial: sql.Open only validates its arguments, and building the +// consumers must not touch the database. The DSN points at a closed loopback +// port so a stray dial fails locally instead of reaching any host. +func TestRegisterFeedConsumers_UserConsumerGetsTheIdentityRefreshLock(t *testing.T) { + openUndialedPool := func() *sql.DB { + pool, err := sql.Open("postgres", "host=127.0.0.1 port=1 dbname=wiring_test sslmode=disable connect_timeout=1") + require.NoError(t, err) + t.Cleanup(func() { _ = pool.Close() }) + return pool + } + a := &application{ + cfg: &config.Config{}, + db: openUndialedPool(), + sessionCoordinationDB: openUndialedPool(), + identityRefreshLockDB: openUndialedPool(), + identityResolver: unusedIdentityResolver{}, + uncachedIdentityResolver: unusedUncachedIdentityResolver{}, + oauthStore: oauth.NewMobileAwareStoreWrapper(nil), + } + + consumers, err := a.registerFeedConsumers() + require.NoError(t, err) + + var userConsumer *jetstream.UserEventConsumer + var communityConsumer *jetstream.CommunityEventConsumer + for _, consumer := range consumers { + switch consumer.name { + case jetstream.ConsumerUsers: + userConsumer = consumer.handler.(*jetstream.UserEventConsumer) + case jetstream.ConsumerCommunities: + communityConsumer = consumer.handler.(*jetstream.CommunityEventConsumer) + } + } + require.NotNil(t, userConsumer, "the user consumer must be registered") + require.NotNil(t, communityConsumer, "the community consumer must be registered") + + require.True(t, userConsumer.HasIdentityRefreshLock(), + "the user consumer must serialize identity refreshes of one DID through the Postgres refresh lock") + require.True(t, communityConsumer.HasIdentityRefreshLock(), + "the community consumer must serialize identity refreshes of one DID through the Postgres refresh lock") +} diff --git a/cmd/server/wiring.go b/cmd/server/wiring.go index f49a61a..c2da3da 100644 --- a/cmd/server/wiring.go +++ b/cmd/server/wiring.go @@ -89,17 +89,21 @@ type application struct { // sessionCoordinationDB is the dedicated pool OAuth session operations hold // connections from; see openSessionCoordinationDatabase. sessionCoordinationDB *sql.DB + // identityRefreshLockDB is the dedicated pool identity refreshes hold + // their per-DID locks on; see openIdentityRefreshLockDatabase. + identityRefreshLockDB *sql.DB credentialCipher *credentialcipher.Cipher // Identity and authentication - identityResolver identity.Resolver - oauthClient *oauth.OAuthClient - oauthStore *oauth.MobileAwareStoreWrapper - oauthHandler *oauth.OAuthHandler - authMiddleware *middleware.OAuthAuthMiddleware - dualAuth *middleware.DualAuthMiddleware - serviceAuthValidator middleware.ServiceAuthValidator - instanceAdminAuth *middleware.InstanceAdminMiddleware + identityResolver identity.Resolver + uncachedIdentityResolver identity.UncachedResolver + oauthClient *oauth.OAuthClient + oauthStore *oauth.MobileAwareStoreWrapper + oauthHandler *oauth.OAuthHandler + authMiddleware *middleware.OAuthAuthMiddleware + dualAuth *middleware.DualAuthMiddleware + serviceAuthValidator middleware.ServiceAuthValidator + instanceAdminAuth *middleware.InstanceAdminMiddleware // moderationAuthority answers who is an instance admin. It is the // allowlist instanceAdminAuth checks, shared with the read paths that // hydrate viewer.moderation and with getViewerAuthority. @@ -195,12 +199,14 @@ func buildApplication( cfg *config.Config, db *sql.DB, sessionCoordinationDB *sql.DB, + identityRefreshLockDB *sql.DB, credentialCipher *credentialcipher.Cipher, ) (app *application, err error) { app = &application{ cfg: cfg, db: db, sessionCoordinationDB: sessionCoordinationDB, + identityRefreshLockDB: identityRefreshLockDB, credentialCipher: credentialCipher, stopImageProxyCleanup: func() {}, } @@ -304,6 +310,7 @@ func (a *application) buildIdentity() { } identityConfig.NegativeCacheTTL = a.cfg.Identity.NegativeCacheTTL a.identityResolver = identity.NewResolver(a.db, identityConfig) + a.uncachedIdentityResolver = identity.NewUncachedResolver(identityConfig) // The same expression the hatch above was derived from, not a second read of // the same config field: a log line describing the SSRF gate has to be unable diff --git a/docs/BRIDGED_IDENTITY.md b/docs/BRIDGED_IDENTITY.md new file mode 100644 index 0000000..c38b565 --- /dev/null +++ b/docs/BRIDGED_IDENTITY.md @@ -0,0 +1,87 @@ +# Bridged identity: handle refresh and origin fields + +Tidepool, the ActivityPub bridge, renamed every bridged actor from `..tdpl.io` to `..ap.tdpl.io` (for example `alice.lemmy.world.ap.tdpl.io`). The DID stays the same; the handle changes in PLC and Tidepool announces it with a firehose `#identity` event. The old scheme needed one DNS delegation per instance, could not be turned back into the fediverse address (`my-lemmy.org` and `my.lemmy.org` collided), and lost case and underscores. The design document lives in the Tidepool repository and is not public. + +Coves needed five changes for this, and this doc covers them: + +1. Store a fresh, verified handle on every identity event, for users and communities. +2. Serialize handle refreshes per DID, so that two feeds or the redriver cannot revert a handle. +3. A one-off sweep, `cmd/reconcile-handles`, to fix rows that missed their identity event. +4. Origin fields (`origin`, `originUsername`, `originUrl`, `originSoftware`) on bridged profiles, admitted only from a trusted bridge, and served on profile views, author views and notifications. +5. Resolving a bridged user by `name@host`. + +Bridged-account detection is unchanged: it uses `TRUSTED_BRIDGE_PDS_HOSTS` against the PDS, never the handle. The `hostedBy` registrable domain is still `tdpl.io`, so community `hostedBy` checks pass after the rename. + +## Identity events + +`#identity` events are no longer trusted as delivered. The user consumer (`internal/atproto/jetstream/user_consumer.go`) and the community consumer (`community_consumer.go`) both: + +- Look up the DID in `users` or `communities` first. An unknown DID costs one indexed read and no resolver call. Jetstream delivers identity events for the whole network, so this ordering matters. A known DID is refreshed on every identity event, including events with no handle. +- Ignore the event's handle and call `RefreshIdentityHandle` (`identity_refresh.go`). In order, it purges the DID from the caching resolver, checks the timeout debounce, resolves the DID through the uncached resolver (`identity.NewUncachedResolver`) under the 5-second `identityResolveTimeout`, runs `identity.VerifiedHandle`, re-reads the stored handle inside the per-DID lock, purges the old handle's cache row, and writes the new handle. The old handle is purged before the write, so a failed purge is retried on redrive instead of being lost. +- Return `ErrUnresolvedReference` for resolution or purge failures, so the event is redriven. + +The community path also validates the new handle (`communities.ValidateHandle`, permanent on failure) and re-runs `verifyHostedByClaim` against the stored `hosted_by_did`. A mismatch is `ErrPermanentEvent` and keeps the old handle. A fetch failure is `ErrUnresolvedReference`. A unique-handle conflict (`ErrHandleTaken`) is `ErrUnresolvedReference`. Users also get their OAuth session handles updated. + +Timeout debounce: a refresh whose resolve hit the deadline records the DID for `IDENTITY_NEGATIVE_CACHE_TTL` (4096 entries at most). Within that window the next identity event still purges the cached DID but skips resolution and defers without spending a redrive attempt. Fast failures are not recorded. This is the one accepted exception to "every identity event triggers a fresh resolution"; `docs/SECURITY_MODEL.md` has the reasoning. + +## Per-DID refresh lock + +`NewPostgresIdentityRefreshLock` (`identity_refresh_lock.go`) takes a session-scoped `pg_advisory_lock` on a dedicated connection. The key is the first 8 bytes of `sha256("coves/identity-refresh/" + did)`, so it works across every feed's consumer, the redriver and the separate sweep process. + +- Pool: `openIdentityRefreshLockDatabase` in `cmd/server/database.go`, capped at 8 connections and separate from the application pool. Lock waiters must never share a capped pool with the writes that release them, or the pool deadlocks. The sweep opens its own lock pool sized to `-concurrency`. +- Timeouts: each acquire sets `lock_timeout` to `IdentityRefreshLockWaitTimeout` (25 seconds: the resolve timeout, plus the 15-second well-known verification timeout, plus 5 seconds) and `statement_timeout` to 5 seconds more, so contention always surfaces as SQLSTATE 55P03. A same-DID contender can stall its serial lane for up to 25 seconds. +- A busy lock returns `jetstream.ErrIdentityRefreshBusy`, wrapped in `ErrUnresolvedReference` as a deferred redrive (no in-line retry, no redrive attempt spent). The sweep reports it as `lock-busy`. +- A nil lock (nil database or a wait of zero or less) makes `RefreshIdentityHandle` fail closed before any purge. The callback runs synchronously and is not re-entrant. + +## Reconciliation sweep + +`cmd/reconcile-handles` selects every user and community row whose `pds_url` is a trusted bridge host (`-trusted-hosts`, replacing `TRUSTED_BRIDGE_PDS_HOSTS`), resolves each DID fresh, and writes changes through the same locked refresh as the consumers, using the exported entry points `(*UserEventConsumer).RefreshUserHandle` and `(*CommunityEventConsumer).RefreshCommunityHandle`. `-dry-run` resolves and verifies only. `-clear-identity-cache` empties `identity_cache` first and cannot be combined with `-dry-run`. Exit status 0 means every bridged row matches its verified handle. The runbook, with the production invocation, reason codes and re-run rules, is "Bridged handle reconciliation" in `docs/PRODUCTION_DEPLOY.md`. + +## Origin fields on profiles + +Columns: `users.origin`, `origin_username`, `origin_url`, `origin_software` (migration 057) and `communities.origin_url`, `origin_software` (migration 058; community `origin` already existed). A cleared column holds `''` and an untouched one is `NULL`, so filter with `COALESCE(column, '') <> ''`, never `IS NOT NULL`. + +Admission (`resolveFreshUserProfileIdentity`, `resolveFreshCommunityIdentity`) requires all of the following for the event being processed. A stored `pds_url` or an `identity_cache` row never admits the fields, because a cache row lives 24 hours and can still name the bridge after a repo moves. + +1. The DID document, read uncached, names a PDS in `TRUSTED_BRIDGE_PDS_HOSTS`. +2. The declared handle's registrable domain (eTLD+1) equals the trusted PDS host's registrable domain. For Tidepool that means handles under `tdpl.io`. This rule was decided on 2026-10-12. Without it, an account could point its PLC PDS at the bridge host during consumer lag and be admitted. +3. A fresh uncached resolution verifies the handle both ways, names the same DID, and still names a trusted PDS whose registrable domain matches the verified handle's. + +If admission fails, the fields are not admitted: a known user or community has its origin fields cleared and the rest of the edit applied, and an unknown user's profile event is dropped without creating a row. This is never redriven. A transient resolution failure or an unverified handle on a record that asserts the fields returns `ErrUnresolvedReference` without writing, so Tidepool's one-time profile rewrite survives by redrive. Community admission logs one line per event when an asserted field is not stored or a stored one is cleared (DID and reason only). + +Sanitizing (`user_profile_origin.go`): `origin` is normalized as a domain and kept only together with a non-empty `originUsername` of at most 256 bytes. `originUrl` must be `https`, have a host equal to `origin`, pass atproto `uri` syntax and carry no userinfo. `originSoftware` is trimmed and lowercased, then must be 1 to 64 bytes. All lexicon `maxLength` limits count UTF-8 bytes, because indigo compares `len(value)`. + +## Origin fields on views + +- `actor.defs#profileViewDetailed` (`getProfile`) and `actor.defs#profileView` (notification reply authors and `recentUpvoters`) carry all four fields. +- `community.defs#communityView` and `#communityViewDetailed` (`community.get`, `list`, `search`) carry `originUrl` and `originSoftware` next to the existing `origin`. +- `community.post.defs#authorView` carries `origin`, `originUsername` and `originSoftware` on posts and comments. Posts have no software field of their own; the author's profile supplies it. Every post-view query selects these three columns through `postViewSelectColumns`; a new post-view query must do the same or the scan misaligns. Deleted-comment placeholders carry only the DID and `handle.invalid`. + +No field marks an account as bridged; a client can only tell from the presence of `origin`. This was left out on purpose. + +## Resolving `name@host` + +`users.Service.ResolveActorIdentifierToDID` accepts a DID, a handle, `name@host` or `@name@host`, splitting at the last `@` and keeping the name's case. A name over 256 bytes, with invalid UTF-8 or with control characters is a 400 before any query. The repository lookup `GetDIDByOriginUsername` tries an exact match on (`origin_username`, `origin`) first, then a unique case-insensitive match. Two case-insensitive matches return `users.ErrAmbiguousUser`, which `xrpc/mapper.go` maps to 409 `AmbiguousActor`, mirroring `AmbiguousCommunity`. This way, shared links (exact addresses) always resolve, typed or lowercased addresses resolve in the common case, and no account becomes unreachable. Migration 059 adds non-unique exact and `lower()` indexes; the pair is copied from the bridge and is not unique. `actor.getProfile`, `getPosts` and `getComments` share `resolveActorDID`; an outage is a logged 500, never a 404. + +There is no fallback that builds `