From fdefc8f19643a7703453fe3c4c7cd4669e19910b Mon Sep 17 00:00:00 2001 From: Noah Pederson Date: Fri, 9 Oct 2026 17:20:45 -0500 Subject: [PATCH] Add historical replay and query-local traversal Reconstruct the soft index from Jetstream v2 archives and resume from atomic sequence checkpoints. Keep legacy ingestion optional. Read reachable neighbors instead of the complete graph so that unrelated records do not exhaust query limits. Share bounded work across reputation checks and graph-conflict retries. tsk-30 tsk-31 --- cmd/appview/main.go | 18 +- docs/jetstream-replay.md | 86 ++++++++ go.mod | 37 +++- go.sum | 141 +++++++++++-- internal/appview/neighbors.go | 54 +++++ internal/appview/neighbors_test.go | 86 ++++++++ internal/appview/replay.go | 54 +++++ internal/appview/reputation.go | 8 + internal/appview/service.go | 13 +- internal/appview/trust_cache.go | 52 +++-- internal/config/config.go | 15 ++ internal/config/config_test.go | 24 +++ internal/domain/change.go | 2 + internal/microcosm/replay.go | 130 ++++++++++++ internal/microcosm/replay_test.go | 154 ++++++++++++++ internal/storage/db/models.go | 6 + internal/storage/db/querier.go | 8 + internal/storage/db/queries.sql.go | 217 ++++++++++++++++++++ internal/storage/events.go | 19 +- internal/storage/lists.go | 22 +- internal/storage/neighbors.go | 57 +++++ internal/storage/neighbors_test.go | 58 ++++++ internal/storage/queries.sql | 37 ++++ internal/storage/replay.go | 97 +++++++++ internal/storage/replay_schema.sql | 7 + internal/storage/replay_test.go | 122 +++++++++++ internal/storage/scale_test.go | 40 ++-- internal/storage/soft.go | 4 +- internal/storage/sqlite.go | 17 +- internal/storage/store.go | 3 + internal/trust/algorithm.go | 14 +- internal/trust/algorithms/neighbors_test.go | 87 ++++++++ internal/trust/algorithms/optimistic.go | 4 +- internal/trust/algorithms/shortest_path.go | 19 +- internal/trust/algorithms/traversal.go | 72 ++++++- internal/trust/budget.go | 32 +++ internal/upstream/queue.go | 28 ++- internal/upstream/queue_test.go | 29 +++ nix/modules/appview.nix | 19 +- nix/package.nix | 2 +- nix/services/provenance.nix | 9 + sqlc.yaml | 1 + 42 files changed, 1806 insertions(+), 98 deletions(-) create mode 100644 docs/jetstream-replay.md create mode 100644 internal/appview/neighbors.go create mode 100644 internal/appview/neighbors_test.go create mode 100644 internal/appview/replay.go create mode 100644 internal/microcosm/replay.go create mode 100644 internal/microcosm/replay_test.go create mode 100644 internal/storage/neighbors.go create mode 100644 internal/storage/neighbors_test.go create mode 100644 internal/storage/replay.go create mode 100644 internal/storage/replay_schema.sql create mode 100644 internal/storage/replay_test.go create mode 100644 internal/trust/algorithms/neighbors_test.go create mode 100644 internal/trust/budget.go diff --git a/cmd/appview/main.go b/cmd/appview/main.go index 9c6c471..62ec1fc 100644 --- a/cmd/appview/main.go +++ b/cmd/appview/main.go @@ -43,12 +43,20 @@ func run() error { // This selection controls calculation, accepted query parameters, and the API schema. algorithm := algorithms.Optimistic{} service := appview.NewWithOptions(store, atproto.NewRepositoryWithQueue(cfg.SoftMode, queue), cfg.Limits, appview.Options{Trust: algorithm, Reputation: reputationalg.DirectSum{}, CacheTTL: cfg.TrustCacheTTL, SoftMode: cfg.SoftMode}) - stream, err := microcosm.NewWithQueue(cfg.JetstreamURL, cfg.SlingshotURL, cfg.SoftMode, queue) - if err != nil { - return err - } var workers sync.WaitGroup - workers.Go(func() { service.Run(ctx, stream) }) + if cfg.ReplayURL != "" { + stream, err := microcosm.NewReplay(cfg.ReplayURL, cfg.ReplayKey, cfg.SoftMode, queue, cfg.ExternalWorkers) + if err != nil { + return err + } + workers.Go(func() { service.RunReplay(ctx, stream) }) + } else { + stream, err := microcosm.NewWithQueue(cfg.JetstreamURL, cfg.SlingshotURL, cfg.SoftMode, queue) + if err != nil { + return err + } + workers.Go(func() { service.Run(ctx, stream) }) + } err = httpx.Serve(ctx, cfg.Address, appview.Handler(service, cfg.ServiceKey)) cancel() workers.Wait() diff --git a/docs/jetstream-replay.md b/docs/jetstream-replay.md new file mode 100644 index 0000000..da810a1 --- /dev/null +++ b/docs/jetstream-replay.md @@ -0,0 +1,86 @@ +# Historical Jetstream replay + +Historical replay is optional. The default appview still uses the legacy Microcosm stream. +The v2 adapter reconstructs current records, not a complete mutation audit. +The provider can compact deleted records from its archive. + +## Configuration + +1. Supply `JETSTREAM_API_KEY` through the service environment or a runtime secret. +2. Set `PROVENANCE_JETSTREAM_REPLAY=true`. +3. For a network-wide soft index, set `ENABLE_SOFT_VOUCH=true`. +4. Set `ENABLE_SOFT_VOUCH_ALL_AUTHORS=true`. +5. For blocked moderation lists, set `ENABLE_SOFT_VOUCH_LISTS=true`. +6. Start the appview. + +The default archive origin is `https://jetstream.us-west.bsky.network`. +`PROVENANCE_JETSTREAM_REPLAY_URL` selects another origin. +The API key authenticates archive requests only. It does not reach the public WebSocket. +No API key belongs in Git, a Nix expression, or a command argument. +Keep secret files outside the repository. A `path:.` Nix flake can copy ignored files, including `.env`, into the Nix store. +Use Git-filtered flake checks instead of `path:.` when the repository contains private data. + +NixOS exposes `services.provenance.appview.jetstreamReplay`, `jetstreamReplayURL`, and `jetstreamAPIKeyFile`. +The module loads the key with systemd credentials. +Development services expose `jetstreamReplay` and `jetstreamReplayURL` and accept the key through the environment. + +CAUTION: Do not enable replay before you assess archive charges and available disk space. +The provider meters compressed download bytes. A full import can consume substantial time, disk space, and memory. + +## Recovery and cutover + +The official Go SDK pins each archive plan, decodes selected collections, and connects the live tail without a gap. +It fills an expired live cursor from the archive. +SQLite commits each batch and its sequence checkpoint in one transaction. +Per-account sequence guards reject duplicate events and preserve ordering across interleaved accounts. +Witnessed timestamps preserve the existing snapshot and legacy replay guards. +Sequence order distinguishes record mutations with the same timestamp. +Snapshot and deletion watermarks still reject records at or before their timestamp. + +Account deletion and sync divergence remove obsolete records. +Replacement records after a sync marker restore current state. +Network-wide list replay retains list definitions and members before a subscriber references them. +Participant-only mode retains referenced-list snapshots. +Explicit ingestion never enrolls nonparticipants. + +Checkpoints include the archive origin and soft mode. +Different origins do not share sequence cursors. +Changing the origin or soft mode starts a new archive import. +Do not replace a server with an unrelated sequence namespace at the same origin. + +Downloads and WebSocket handshakes share the external worker queue and the project User-Agent. +`PROVENANCE_EXTERNAL_WORKERS` bounds HTTP concurrency and the SDK decode pool. +The SDK uses one range stream per segment and at most three attempts per request. +The queue retains per-origin `Retry-After` cooldowns without additional retries. +The SDK retains partial range downloads in memory during retries, not across process restarts. +A process restart resumes from the last committed event batch and can download part of a segment again. +The SDK also prefetches compressed segments. The worker count is not a strict memory limit. + +## Query limits + +Built-in trust algorithms read outgoing records only for reachable accounts. +Unrelated records no longer consume the query edge limit. +The 10-hop cap and the state-based breadth-first score calculation remain unchanged. +One query shares its neighbor cache, edge budget, and state budget across reputation author checks. +Concurrent graph changes prevent stale cache publication and cause bounded retries. +Custom algorithms without the neighbor capability retain the complete edge-buffer contract. + +The 100,000-record storage fixture allocates about 118 MB for a complete graph read and selection. +An outgoing read and selection for one author with 100 edges allocates about 106 KB. +These figures exclude traversal state, SDK buffers, peak memory, WAL, and SQLite caches. + +## Validation + +Local tests cover binary archive decode, live cutover, collection filters, bearer-key scope, atomic rollback, restart, and equal-time events. +The adapter does not start an import unless replay is enabled. +No full-network import forms part of the test suite. + +To verify the key without a segment download, run the opt-in archive-plan probe: + +```sh +PROVENANCE_REPLAY_SMOKE=true go test ./internal/microcosm -run '^TestReplayLivePlan$' -v +``` + +The probe requires `JETSTREAM_API_KEY` in the process environment and permits one sequence in its archive plan. + +Provider documentation: . diff --git a/go.mod b/go.mod index 7b0a358..0e3d893 100644 --- a/go.mod +++ b/go.mod @@ -1,36 +1,59 @@ module provenance -go 1.26.0 +go 1.26.6 require ( github.com/bluesky-social/indigo v0.0.0-20261006083258-ae9297b7d0f4 + github.com/bluesky-social/jetstream v0.3.10 github.com/coder/websocket v1.8.15 modernc.org/sqlite v1.60.1 ) require ( github.com/beorn7/perks v1.0.1 // indirect + github.com/bluesky-social/gttp v0.1.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/cockroachdb/errors v1.14.0 // indirect + github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b // indirect + github.com/cockroachdb/pebble v1.1.5 // indirect + github.com/cockroachdb/redact v1.1.5 // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/earthboundkid/versioninfo/v2 v2.24.1 // indirect - github.com/golang-jwt/jwt/v5 v5.2.2 // indirect + github.com/getsentry/sentry-go v0.46.0 // indirect + github.com/gogo/protobuf v1.3.2 // indirect + github.com/golang-jwt/jwt/v5 v5.3.1 // indirect github.com/google/go-querystring v1.1.0 // indirect github.com/google/uuid v1.6.0 // indirect github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect + github.com/jcalabro/atmos v0.7.3 // indirect + github.com/jcalabro/gloom v0.1.0 // indirect + github.com/jcalabro/gt v0.0.14 // indirect + github.com/klauspost/compress v1.19.2 // indirect + github.com/klauspost/cpuid/v2 v2.4.0 // indirect + github.com/kr/pretty v0.3.1 // indirect + github.com/kr/text v0.2.0 // indirect github.com/mattn/go-isatty v0.0.24 // indirect - github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect github.com/mr-tron/base58 v1.3.0 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect - github.com/prometheus/client_golang v1.17.0 // indirect - github.com/prometheus/client_model v0.5.0 // indirect - github.com/prometheus/common v0.45.0 // indirect - github.com/prometheus/procfs v0.12.0 // indirect + github.com/pkg/errors v0.9.1 // indirect + github.com/prometheus/client_golang v1.24.1 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.70.1 // indirect + github.com/prometheus/procfs v0.21.1 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + github.com/rogpeppe/go-internal v1.14.1 // indirect + github.com/zeebo/xxh3 v1.1.0 // indirect gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b // indirect gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02 // indirect golang.org/x/crypto v0.55.0 // indirect + golang.org/x/net v0.58.0 // indirect + golang.org/x/sync v0.23.0 // indirect golang.org/x/sys v0.48.0 // indirect + golang.org/x/text v0.41.0 // indirect golang.org/x/time v0.15.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260928230214-8a89bd6388cc // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260928230214-8a89bd6388cc // indirect google.golang.org/protobuf v1.36.12 // indirect modernc.org/libc v1.77.1 // indirect modernc.org/mathutil v1.7.1 // indirect diff --git a/go.sum b/go.sum index a76887d..a08dfd2 100644 --- a/go.sum +++ b/go.sum @@ -1,19 +1,48 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/bluesky-social/gttp v0.1.0 h1:jsG6ssqLYboxxdx+eBScIvKfMfBB6qvN8E4SW+NeoaM= +github.com/bluesky-social/gttp v0.1.0/go.mod h1:TPtZ67MOhePifYaztEI42hUhluaE2GJ33+oFxPermuw= github.com/bluesky-social/indigo v0.0.0-20261006083258-ae9297b7d0f4 h1:7kb1maxAQ/gW9Jxyjy59qMJh9wuugQG55BQgb3Xx5XA= github.com/bluesky-social/indigo v0.0.0-20261006083258-ae9297b7d0f4/go.mod h1:JfOxJR/TXBUxnT9ywiBnXDt/Uby5bwcq9re38wg9IEw= +github.com/bluesky-social/jetstream v0.3.10 h1:yNzbLIcllA16q9wnJSjDYlGNM0rZrKAnnaoog0X+XSs= +github.com/bluesky-social/jetstream v0.3.10/go.mod h1:8JQSi30b5o7p5NVUEpqD9JV7PQumti/ger+1o76b8hI= +github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= +github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/cockroachdb/datadriven v1.0.3-0.20230413201302-be42291fc80f h1:otljaYPt5hWxV3MUfO5dFPFiOXg9CyG5/kCfayTqsJ4= +github.com/cockroachdb/datadriven v1.0.3-0.20230413201302-be42291fc80f/go.mod h1:a9RdTaap04u637JoCzcUoIcDmvwSUtcUFtT/C3kJlTU= +github.com/cockroachdb/errors v1.14.0 h1:EfdVEJpN3z8rPMo43Yit59LxoiIa470fSXpZXuEs+ZI= +github.com/cockroachdb/errors v1.14.0/go.mod h1:xRa70jZ9sNBQmISt5KmJmAD++E4dQHm89oCRiZGEdq0= +github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b h1:r6VH0faHjZeQy818SGhaone5OnYfxFR/+AzdY3sf5aE= +github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b/go.mod h1:Vz9DsVWQQhf3vs21MhPMZpMGSht7O/2vFW2xusFUVOs= +github.com/cockroachdb/pebble v1.1.5 h1:5AAWCBWbat0uE0blr8qzufZP5tBjkRyy/jWe1QWLnvw= +github.com/cockroachdb/pebble v1.1.5/go.mod h1:17wO9el1YEigxkP/YtV8NtCivQDgoCyBg5c4VR/eOWo= +github.com/cockroachdb/redact v1.1.5 h1:u1PMllDkdFfPWaNGMyLD1+so+aq3uUItthCFqzwPJ30= +github.com/cockroachdb/redact v1.1.5/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg= github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/earthboundkid/versioninfo/v2 v2.24.1 h1:SJTMHaoUx3GzjjnUO1QzP3ZXK6Ee/nbWyCm58eY3oUg= github.com/earthboundkid/versioninfo/v2 v2.24.1/go.mod h1:VcWEooDEuyUJnMfbdTh0uFN4cfEIg+kHMuWB2CDCLjw= -github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeDy8= -github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk= +github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= +github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= +github.com/getsentry/sentry-go v0.46.0 h1:mbdDaarbUdOt9X+dx6kDdntkShLEX3/+KyOsVDTPDj0= +github.com/getsentry/sentry-go v0.46.0/go.mod h1:evVbw2qotNUdYG8KxXbAdjOQWWvWIwKxpjdZZIvcIPw= +github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA= +github.com/go-errors/errors v1.4.2/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og= +github.com/go-logr/logr v1.4.4 h1:tG4xh9yMsRCAiodLVTxyrkzSZ9+o0L1Kg/+cPVcbP/8= +github.com/go-logr/logr v1.4.4/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= @@ -23,16 +52,30 @@ github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3 h1:LMLX+LgTNWpfvCBdFe github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3/go.mod h1:jl5iWTm0/hd5PjEYEOuwAJ57L/CibdZfrqZ5XA5GrCk= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs= github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/ipfs/go-cid v0.6.2 h1:VuGwJd+KJTaMJ4S4d5EEf9SXc17YUblS5axCbocn9YE= github.com/ipfs/go-cid v0.6.2/go.mod h1:Xhwg8NzHeK9xPCEZkCw4idzPiuNMpX3fARuI5Iwj1Lo= +github.com/jcalabro/atmos v0.7.3 h1:UNyO78urlVHwCZ/KZpmtpMoXe2Q3FMDfR6w6/NYnSNk= +github.com/jcalabro/atmos v0.7.3/go.mod h1:xPADdLqXVyJwfkajuZIdm3X7KoPmmkWPbieTBSKSU7w= +github.com/jcalabro/gloom v0.1.0 h1:Z8VTAMrdRNXr11tt9c5pIpDDJxAHg3QIBrHFt5x2I2M= +github.com/jcalabro/gloom v0.1.0/go.mod h1:6uAawINljQYLN3a2YsekJ8LDYT/SobmJyfeKrSjHVx0= +github.com/jcalabro/gt v0.0.14 h1:m5Go+0xX+Atb3PbS1rLuNrFWiO0w1bdZlAoxbBITzXc= +github.com/jcalabro/gt v0.0.14/go.mod h1:iBnF7jE8LG8mW0zjXVYTuXoT+L9UfwP7RNgMrmuuOTs= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8= +github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/cpuid/v2 v2.4.0 h1:S6Hrbc7+ywsr0r+RLapfGBHfyefhCTwEh3A0tV913Dw= github.com/klauspost/cpuid/v2 v2.4.0/go.mod h1:19jmZ9mjzoF//ddRSUsv0zfBTJWh3QJh9FNxZTMrGxU= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI= github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A= -github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 h1:jWpvCLoY8Z/e3VKvlsiIGKtc+UG6U5vzxaoagmhXfyg= -github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0/go.mod h1:QUyp042oQthUoa9bqDv0ER0wrtXnBruoNd7aNjkbP+k= github.com/minio/sha256-simd v1.0.1 h1:6kaan5IFmwTNynnKKpDHe6FWHohJOHhCPchzK49dzMM= github.com/minio/sha256-simd v1.0.1/go.mod h1:Pz6AKMiUdngCLpeTL/RJY1M9rUuPMYujV5xJjtbRSN8= github.com/mr-tron/base58 v1.3.0 h1:K6Y13R2h+dku0wOqKtecgRnBUBPrZzLZy5aIj8lCcJI= @@ -47,45 +90,119 @@ github.com/multiformats/go-multihash v0.2.3 h1:7Lyc8XfX/IY2jWb/gI7JP+o7JEq9hOa7B github.com/multiformats/go-multihash v0.2.3/go.mod h1:dXgKXCXjBzdscBLk9JkjINiEsCKRVch90MdaGiKsvSM= github.com/multiformats/go-varint v0.1.0 h1:i2wqFp4sdl3IcIxfAonHQV9qU5OsZ4Ts9IOoETFs5dI= github.com/multiformats/go-varint v0.1.0/go.mod h1:5KVAVXegtfmNQQm/lCY+ATvDzvJJhSkUlGQV9wgObdI= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4= +github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/prometheus/client_golang v1.17.0 h1:rl2sfwZMtSthVU752MqfjQozy7blglC+1SOtjMAMh+Q= -github.com/prometheus/client_golang v1.17.0/go.mod h1:VeL+gMmOAxkS2IqfCq0ZmHSL+LjWfWDUmp1mBz9JgUY= -github.com/prometheus/client_model v0.5.0 h1:VQw1hfvPvk3Uv6Qf29VrPF32JB6rtbgI6cYPYQjL0Qw= -github.com/prometheus/client_model v0.5.0/go.mod h1:dTiFglRmd66nLR9Pv9f0mZi7B7fk5Pm3gvsjB5tr+kI= -github.com/prometheus/common v0.45.0 h1:2BGz0eBc2hdMDLnO/8n0jeB3oPrt2D08CekT0lneoxM= -github.com/prometheus/common v0.45.0/go.mod h1:YJmSTw9BoKxJplESWWxlbyttQR4uaEcGyv9MZjVOJsY= -github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo= -github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo= +github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU= +github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY= +github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc= +github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI= +github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY= +github.com/puzpuzpuz/xsync/v4 v4.5.0 h1:vOSWu6b57/emh+L/Cw0BeQfvxa/cogFywXHeGUxQxAg= +github.com/puzpuzpuz/xsync/v4 v4.5.0/go.mod h1:VJDmTCJMBt8igNxnkQd86r+8KUeN1quSfNKu5bLYFQo= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI= github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e h1:28X54ciEwwUxyHn9yrZfl5ojgF4CBNLWX7LR0rvBkf4= github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e/go.mod h1:pM99HXyEbSQHcosHc0iW7YFmwnscr+t9Te4ibko05so= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ= +github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b h1:CzigHMRySiX3drau9C6Q5CAbNIApmLdat5jPMqChvDA= gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b/go.mod h1:/y/V339mxv2sZmYYR64O07VuCpdNZqCTwO8ZcouTMI8= gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02 h1:qwDnMxjkyLmAFgcfgTnfJrmYKWhHnci3GjDqcZp1M3Q= gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02/go.mod h1:JTnUj0mpYiAsuZLmKjTx/ex3AtMowcCgnE7YNyCEP0I= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 h1:8tvICD4vSTOOsNrsI4Ljf6C+6UKvpTEH5XY3JMoyPoo= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0/go.mod h1:z9+yiacE0IHRqM4qFfkbt/JYlmYXgss8GY/jXoNuPJI= +go.opentelemetry.io/otel v1.45.0 h1:pdrWmLHofpubmArBv1LgFSv1Z0Ie/ppdZzu+kUN5EeU= +go.opentelemetry.io/otel v1.45.0/go.mod h1:XZxIqPapzEYnhNSScF5DIqXhm/rYi0FzCe2XddAwZfQ= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0 h1:QRefszxJmfPdjXUUm3j6iDzY03mTPXMjqErFqQ67vUg= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0/go.mod h1:Tiz03lTBVBrm7eWZBOidzEaYaJa8tjwGUGv6d8mlTyk= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.45.0 h1:QBajQ2SrwQijzHyZbQlPsuIzpl/ll8DY6wPWsajeGcI= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.45.0/go.mod h1:08ZQLjrPLQ6R4kAXvuOvODEer5Yh4CoFvll5qB2BCI8= +go.opentelemetry.io/otel/metric v1.45.0 h1:7Eg1uH7CJ5cXv9is6tnBe1FI6rj1nwUdbFypRm3br/M= +go.opentelemetry.io/otel/metric v1.45.0/go.mod h1:HAPbm1nd3p1PmFH7v2dR+6BjXxw+Lq4a2+pndMAm08s= +go.opentelemetry.io/otel/sdk v1.45.0 h1:4VVSMgQ83dUgW2aoX5f6JgLvHwIvzcuLnF9lUdCSpCw= +go.opentelemetry.io/otel/sdk v1.45.0/go.mod h1:Sr40LgXV7DsKMMJMKOhUWOgMWTfAaqvm2kF0g7ilwuA= +go.opentelemetry.io/otel/trace v1.45.0 h1:l/mP6Uv7oNO7/TblbhpbgMidxhq1uO/rPsikOyVhxag= +go.opentelemetry.io/otel/trace v1.45.0/go.mod h1:qoJJA2xNMnxRrdISU/kLtfUH2wNeQbiv+jhs/CxI8bc= +go.opentelemetry.io/proto/otlp v1.11.0 h1:5rrYs0Ykyj50sdU/JU0x8etU+LubXWb+gED6TbEdMIk= +go.opentelemetry.io/proto/otlp v1.11.0/go.mod h1:SmVizdCOAm3XBtG1g1NnOdhW6jtddT72hLMhv8VwA8E= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c= golang.org/x/mod v0.41.0/go.mod h1:Ek9pY8RKWXwsWvd3rQiHYtMqkjSUV+s1Rj7j4H5Ur6o= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.50.0 h1:c2ifzfcuY7L90lZ2aKd8S4K2NpASF08SZx9ZuJkHmSU= golang.org/x/tools v0.50.0/go.mod h1:7ulVMw3831Mwi5EZD6RomGyffr4VFjuNYXf2BbCEAV0= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 h1:+cNy6SZtPcJQH3LJVLOSmiC7MMxXNOb3PU/VUEz+EhU= golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90= +google.golang.org/genproto/googleapis/api v0.0.0-20260928230214-8a89bd6388cc h1:rZHRz0ogq4Pid1IKtA/ivHi8XONVF5tqUQXA2TtgpkE= +google.golang.org/genproto/googleapis/api v0.0.0-20260928230214-8a89bd6388cc/go.mod h1:K/n0XLazCJHjD73YuMyNJsN38BPPy1kfFnKxUXaThFU= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260928230214-8a89bd6388cc h1:4bNTbnb44EqGVy9HaQxSY2jnafSUdgV3qHtedvNpDKg= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260928230214-8a89bd6388cc/go.mod h1:OaIUM3+LpYcK2GXM4FTmhWoIq371Owdr+Cc7/BsYHHc= +google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU= +google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/internal/appview/neighbors.go b/internal/appview/neighbors.go new file mode 100644 index 0000000..e179efd --- /dev/null +++ b/internal/appview/neighbors.go @@ -0,0 +1,54 @@ +package appview + +import ( + "context" + "errors" + "sync" + + "provenance/internal/domain" + "provenance/internal/storage" + "provenance/internal/trust" +) + +var errGraphChanged = errors.New("graph changed during neighbor reads") + +type queryGraph struct { + mu sync.Mutex + store storage.Store + tag string + mode domain.Mode + revision int64 + limit int + rawBudget *trust.Budget + neighbors map[string][]domain.Vouch + records map[[2]string]*domain.Vouch +} + +func (g *queryGraph) Outgoing(ctx context.Context, author string, limit int) ([]domain.Vouch, error) { + g.mu.Lock() + defer g.mu.Unlock() + if rows, ok := g.neighbors[author]; ok { + return rows, nil + } + if g.rawBudget == nil { + g.rawBudget = trust.NewBudget(trust.Limits{Edges: g.limit}) + } + remaining := g.rawBudget.RemainingEdges() + rows, revision, err := g.store.Outgoing(ctx, author, g.tag, g.mode, min(limit, remaining+1)) + if err != nil { + return nil, err + } + if err := g.rawBudget.Use(len(rows), 0); err != nil { + return nil, err + } + if revision != g.revision { + return nil, errGraphChanged + } + selected := trust.Select(rows, g.tag, g.mode) + g.neighbors[author] = selected + for index := range selected { + v := &selected[index] + g.records[[2]string{v.Author, v.Subject}] = v + } + return selected, nil +} diff --git a/internal/appview/neighbors_test.go b/internal/appview/neighbors_test.go new file mode 100644 index 0000000..97ae233 --- /dev/null +++ b/internal/appview/neighbors_test.go @@ -0,0 +1,86 @@ +package appview + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "provenance/internal/domain" + "provenance/internal/storage" + "provenance/internal/trust" + "provenance/internal/trust/algorithms" +) + +func TestUnrelatedGraphCapacity(t *testing.T) { + s, _ := algorithmService(t, algorithms.Optimistic{}) + s.limits.Edges = 10 + ctx := context.Background() + for index := 0; index < 20; index++ { + author := fmt.Sprintf("did:plc:unrelated%d", index) + if err := s.store.Join(ctx, author); err != nil { + t.Fatal(err) + } + v := domain.Vouch{Author: author, Subject: bob, Score: 2} + if err := s.store.Apply(ctx, "fixture", domain.Change{Kind: domain.RecordChange, DID: author, RecordKey: domain.RecordKey(bob, ""), TimeUS: time.Now().Add(time.Hour).UnixMicro(), Vouch: &v}); err != nil { + t.Fatal(err) + } + } + result, err := s.Reputation(ctx, alice, bob, "", domain.Default) + if err != nil || result.Score == nil { + t.Fatalf("unrelated records blocked query: %+v %v", result, err) + } +} + +func TestNeighborRevisionConflict(t *testing.T) { + s, _ := algorithmService(t, algorithms.Optimistic{}) + ctx := context.Background() + revision, err := s.store.GraphRevision(ctx) + if err != nil { + t.Fatal(err) + } + g := &queryGraph{store: s.store, revision: revision, limit: 100, neighbors: map[string][]domain.Vouch{}, records: map[[2]string]*domain.Vouch{}} + if _, err := g.Outgoing(ctx, alice, 100); err != nil { + t.Fatal(err) + } + v := domain.Vouch{Author: alice, Subject: bob, Score: 2} + if err := s.store.Apply(ctx, "fixture", domain.Change{Kind: domain.RecordChange, DID: alice, RecordKey: domain.RecordKey(bob, ""), TimeUS: time.Now().Add(time.Hour).UnixMicro(), Vouch: &v}); err != nil { + t.Fatal(err) + } + if _, err := g.Outgoing(ctx, bob, 100); !errors.Is(err, errGraphChanged) { + t.Fatalf("mixed graph revisions: %v", err) + } +} + +func TestNeighborBudget(t *testing.T) { + s, _ := algorithmService(t, algorithms.Optimistic{}) + reader := &trustReader{service: s, input: trust.Input{Source: alice, Limits: s.limits}, budget: trust.NewBudget(trust.Limits{Edges: 1, States: 1})} + if _, err := reader.Score(context.Background(), bob); !errors.Is(err, trust.ErrLimit) { + t.Fatalf("shared budget: %v", err) + } +} + +type conflictingNeighbors struct { + storage.Store + calls int +} + +func (s *conflictingNeighbors) Outgoing(ctx context.Context, author, tag string, mode domain.Mode, limit int) ([]domain.Vouch, int64, error) { + s.calls++ + revision, err := s.Store.GraphRevision(ctx) + return []domain.Vouch{{Author: author, Subject: bob, Score: 1}}, revision + 1, err +} + +func TestNeighborRetryBudget(t *testing.T) { + s, _ := algorithmService(t, algorithms.Optimistic{}) + s.limits.Edges = 1 + store := &conflictingNeighbors{Store: s.store} + s.store = store + if _, err := s.Reputation(context.Background(), alice, bob, "", domain.Default); !errors.Is(err, trust.ErrLimit) { + t.Fatalf("revision retry: %v", err) + } + if store.calls != 2 { + t.Fatalf("retries reset the raw edge budget: %d", store.calls) + } +} diff --git a/internal/appview/replay.go b/internal/appview/replay.go new file mode 100644 index 0000000..0d8300a --- /dev/null +++ b/internal/appview/replay.go @@ -0,0 +1,54 @@ +package appview + +import ( + "context" + "log/slog" + "time" + + "provenance/internal/domain" + "provenance/internal/upstream" +) + +type ReplayStream interface { + Source() string + ConsumeReplay(context.Context, int64, func(context.Context, string, int64) (bool, error), func(context.Context, []domain.Change, int64) error) error +} + +// Use sequence checkpoints only for v2. Legacy streams retain their timestamp cursors. +func (s *Service) RunReplay(ctx context.Context, stream ReplayStream) { + delay := InitialRetryDelay + if err := s.BootstrapPending(ctx); err != nil { + slog.Warn("pending bootstrap failed", "error", err) + } + for ctx.Err() == nil { + cursor, err := s.store.Cursor(ctx, stream.Source(), 0) + processed := 0 + started := time.Now() + if err == nil { + err = stream.ConsumeReplay(ctx, cursor, s.wanted, func(ctx context.Context, changes []domain.Change, cursor int64) error { + s.syncMu.Lock() + defer s.syncMu.Unlock() + for index := range changes { + if err := s.prepareChange(ctx, &changes[index]); err != nil { + return err + } + } + if err := s.store.ApplyReplay(ctx, stream.Source(), changes, cursor); err != nil { + return err + } + processed += len(changes) + return nil + }) + } + if ctx.Err() != nil { + return + } + delay = reconnectDelay(delay, processed, time.Since(started)) + pause := upstream.Jitter(delay) + slog.Warn("replay stream disconnected", "error", err, "retry", pause) + if err := upstream.Wait(ctx, pause); err != nil { + return + } + delay = min(delay*2, MaxRetryDelay) + } +} diff --git a/internal/appview/reputation.go b/internal/appview/reputation.go index c35943f..4adda50 100644 --- a/internal/appview/reputation.go +++ b/internal/appview/reputation.go @@ -29,6 +29,14 @@ func (s *Service) RelativeReputation(ctx context.Context, source, target, tag st return reputation.Result{}, trust.ErrLimit } reader := &trustReader{service: s, input: input, tag: tag, mode: mode} + reader.budget = trust.NewBudget(input.Limits) + reader.rawBudget = trust.NewBudget(input.Limits) + if err := reader.rawBudget.Use(len(vouches), 0); err != nil { + return reputation.Result{}, err + } + if err := reader.budget.Use(len(vouches), 0); err != nil { + return reputation.Result{}, err + } selected := trust.Select(vouches, tag, mode) // An excluded intermediary cannot act as an authority in the aggregate. selected = slices.DeleteFunc(selected, func(v domain.Vouch) bool { diff --git a/internal/appview/service.go b/internal/appview/service.go index 72b00af..ff71381 100644 --- a/internal/appview/service.go +++ b/internal/appview/service.go @@ -367,6 +367,13 @@ func (s *Service) BootstrapPending(ctx context.Context) error { func (s *Service) Apply(ctx context.Context, change domain.Change) error { s.syncMu.Lock() defer s.syncMu.Unlock() + if err := s.prepareChange(ctx, &change); err != nil { + return err + } + return s.store.Apply(ctx, s.streamSource(), change) +} + +func (s *Service) prepareChange(ctx context.Context, change *domain.Change) error { if change.RefreshIdentity { if repo, ok := s.repo.(identityInvalidator); ok { if err := repo.InvalidateIdentity(ctx, change.DID); err != nil { @@ -374,8 +381,8 @@ func (s *Service) Apply(ctx context.Context, change domain.Change) error { } } } - if domain.SoftLists(s.softMode) && change.Source == domain.ListBlockCollection && change.ListRecord != nil { - needed, err := s.store.NeedsList(ctx, change) + if domain.SoftLists(s.softMode) && change.Source == domain.ListBlockCollection && change.ListRecord != nil && !(change.Sequence > 0 && domain.SoftAll(s.softMode)) { + needed, err := s.store.NeedsList(ctx, *change) if err != nil { return err } @@ -391,7 +398,7 @@ func (s *Service) Apply(ctx context.Context, change domain.Change) error { change.ListData = &data } } - return s.store.Apply(ctx, s.streamSource(), change) + return nil } // Separate cursors keep disabled periods from advancing an enabled mode's replay position. diff --git a/internal/appview/trust_cache.go b/internal/appview/trust_cache.go index 905b0e4..9d0f8fc 100644 --- a/internal/appview/trust_cache.go +++ b/internal/appview/trust_cache.go @@ -16,17 +16,20 @@ import ( const cacheAttempts = 3 -// A reputation query shares one lazy graph snapshot across its cache misses. +// A reputation query shares bounded neighbor reads across its author-trust cache misses. type trustReader struct { - mu sync.Mutex - service *Service - input trust.Input - tag string - mode domain.Mode - edges []domain.Vouch - records map[[2]string]*domain.Vouch - revision int64 - loaded bool + mu sync.Mutex + service *Service + input trust.Input + tag string + mode domain.Mode + edges []domain.Vouch + records map[[2]string]*domain.Vouch + revision int64 + loaded bool + graph *queryGraph + budget *trust.Budget + rawBudget *trust.Budget } func (r *trustReader) Score(ctx context.Context, target string) (trust.Result, error) { @@ -53,6 +56,19 @@ func (r *trustReader) Score(ctx context.Context, target string) (trust.Result, e if !errors.Is(err, sql.ErrNoRows) { return trust.Result{}, err } + if !r.loaded { + if _, ok := r.service.algorithm.(trust.NeighborAlgorithm); ok { + revision, err := r.service.store.GraphRevision(ctx) + if err != nil { + return trust.Result{}, err + } + if r.rawBudget == nil { + r.rawBudget = trust.NewBudget(input.Limits) + } + r.graph = &queryGraph{store: r.service.store, tag: r.tag, mode: r.mode, revision: revision, rawBudget: r.rawBudget, neighbors: map[string][]domain.Vouch{}, records: map[[2]string]*domain.Vouch{}} + r.revision, r.records, r.loaded = revision, r.graph.records, true + } + } if !r.loaded { limit := r.service.limits.Edges + 1 if input.Source == target { @@ -69,11 +85,23 @@ func (r *trustReader) Score(ctx context.Context, target string) (trust.Result, e r.records = pathRecords(r.edges) } input.Edges = r.edges - if input.Source == target { + input.Graph = nil + if r.graph != nil { + input.Graph = r.graph + } + if r.budget == nil { + r.budget = trust.NewBudget(input.Limits) + } + input.Budget = r.budget + if input.Source == target && r.graph == nil { input.Edges = nil r.loaded = false } result, err := r.service.algorithm.Calculate(ctx, input) + if errors.Is(err, errGraphChanged) { + r.loaded, r.graph, r.edges = false, nil, nil + continue + } if err != nil { return trust.Result{}, err } @@ -89,7 +117,7 @@ func (r *trustReader) Score(ctx context.Context, target string) (trust.Result, e if saved { return result, nil } - r.loaded = false + r.loaded, r.graph, r.edges = false, nil, nil } return trust.Result{}, trust.ErrLimit } diff --git a/internal/config/config.go b/internal/config/config.go index da8043d..8012bf9 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -29,6 +29,8 @@ type Appview struct { TrustCacheTTL time.Duration SoftMode domain.SoftMode ExternalWorkers int + ReplayURL string + ReplayKey string } type UI struct { @@ -98,10 +100,23 @@ func LoadAppview() (Appview, error) { softMode = domain.SoftAllAuthorLists } } + replay, err := strconv.ParseBool(value("PROVENANCE_JETSTREAM_REPLAY", "false")) + if err != nil { + return Appview{}, errors.New("PROVENANCE_JETSTREAM_REPLAY must be a boolean") + } + replayURL, replayKey := "", "" + if replay { + replayURL = value("PROVENANCE_JETSTREAM_REPLAY_URL", microcosm.DefaultReplayURL) + replayKey = os.Getenv("JETSTREAM_API_KEY") + if replayKey == "" { + return Appview{}, errors.New("JETSTREAM_API_KEY is required for historical replay") + } + } return Appview{ Address: value("PROVENANCE_APPVIEW_ADDR", defaultAppviewAddress), Database: value("PROVENANCE_APPVIEW_DB", "data/appview.db"), ServiceKey: serviceKey, JetstreamURL: value("PROVENANCE_JETSTREAM_URL", microcosm.DefaultJetstreamURL), SlingshotURL: value("PROVENANCE_SLINGSHOT_URL", microcosm.DefaultSlingshotURL), Limits: trust.Limits{States: trust.DefaultStateLimit, Edges: trust.DefaultEdgeLimit, Penalty: penalty}, TrustCacheTTL: ttl, SoftMode: softMode, ExternalWorkers: workers, + ReplayURL: replayURL, ReplayKey: replayKey, }, nil } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index d059a71..68f484d 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -50,6 +50,30 @@ func TestServicePorts(t *testing.T) { } } +func TestReplayConfiguration(t *testing.T) { + t.Setenv("PROVENANCE_SERVICE_KEY", strings.Repeat("x", MinServiceKeyLength)) + t.Setenv("PROVENANCE_JETSTREAM_REPLAY", "") + t.Setenv("JETSTREAM_API_KEY", "") + cfg, err := LoadAppview() + if err != nil || cfg.ReplayURL != "" || cfg.ReplayKey != "" { + t.Fatal("replay is not disabled by default", err) + } + t.Setenv("PROVENANCE_JETSTREAM_REPLAY", "true") + if _, err := LoadAppview(); err == nil { + t.Fatal("replay accepted a missing key") + } + t.Setenv("JETSTREAM_API_KEY", "test-key") + t.Setenv("PROVENANCE_JETSTREAM_REPLAY_URL", "https://archive.example.com") + cfg, err = LoadAppview() + if err != nil || cfg.ReplayURL != "https://archive.example.com" || cfg.ReplayKey != "test-key" { + t.Fatal("replay configuration failed", err) + } + t.Setenv("PROVENANCE_JETSTREAM_REPLAY", "invalid") + if _, err := LoadAppview(); err == nil { + t.Fatal("replay accepted a non-boolean switch") + } +} + func TestExternalWorkers(t *testing.T) { t.Setenv("PROVENANCE_SERVICE_KEY", strings.Repeat("x", MinServiceKeyLength)) t.Setenv("PROVENANCE_ENCRYPTION_KEY", "test-key") diff --git a/internal/domain/change.go b/internal/domain/change.go index 3cdbd41..58aab36 100644 --- a/internal/domain/change.go +++ b/internal/domain/change.go @@ -6,12 +6,14 @@ const ( RecordChange ChangeKind = "commit" AccountChange ChangeKind = "account" IdentityChange ChangeKind = "identity" + ResetChange ChangeKind = "sync" ) type Change struct { Kind ChangeKind DID string TimeUS int64 + Sequence int64 RecordKey string Source string ListRecord *ListRecord diff --git a/internal/microcosm/replay.go b/internal/microcosm/replay.go new file mode 100644 index 0000000..2bb69bd --- /dev/null +++ b/internal/microcosm/replay.go @@ -0,0 +1,130 @@ +package microcosm + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "math" + "net/http" + "net/url" + "strings" + + jetstream "github.com/bluesky-social/jetstream" + + "provenance/internal/atproto" + "provenance/internal/domain" + "provenance/internal/upstream" +) + +const DefaultReplayURL = "https://jetstream.us-west.bsky.network" +const replayBatchSize = 64 + +// Replay keeps archive transport and format details outside the application service. +type Replay struct { + client *Client + host, key string + workers int +} + +func NewReplay(host, key string, mode domain.SoftMode, queue *upstream.Queue, workers int) (*Replay, error) { + if err := validateURL(host, "https", "http"); err != nil { + return nil, err + } + u, _ := url.Parse(host) + if u.Path != "" && u.Path != "/" { + return nil, errors.New("replay URL must contain only an origin") + } + if strings.TrimSpace(key) == "" { + return nil, errors.New("JETSTREAM_API_KEY is required for historical replay") + } + if workers < 1 || workers > upstream.MaxWorkers { + return nil, errors.New("invalid replay worker count") + } + c, err := NewWithQueue(DefaultJetstreamURL, DefaultSlingshotURL, mode, queue) + if err != nil { + return nil, err + } + // Bulk reads use context cancellation, not a short whole-response timeout. + c.http.Timeout = 0 + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.ResponseHeaderTimeout = atproto.RequestTimeout + c.http.Transport = queue.TransportWithPolicy(transport, upstream.CallerRetries) + return &Replay{client: c, host: strings.TrimRight(host, "/"), key: key, workers: workers}, nil +} + +func (r *Replay) Source() string { + return fmt.Sprintf("jetstream-v2:%s:soft-%d", r.host, r.client.softMode) +} + +func (r *Replay) ConsumeReplay(ctx context.Context, cursor int64, wanted func(context.Context, string, int64) (bool, error), apply func(context.Context, []domain.Change, int64) error) error { + if cursor < 0 { + return errors.New("negative replay cursor") + } + collections := []string{domain.Collection} + if r.client.softMode != domain.SoftDisabled { + collections = append(collections, domain.FollowCollection, domain.BlockCollection) + } + if domain.SoftLists(r.client.softMode) { + collections = append(collections, domain.ListCollection, domain.ListItemCollection, domain.ListBlockCollection) + } + client, err := jetstream.Subscribe(r.host, jetstream.WithAPIKey(r.key), jetstream.WithAfterSeq(uint64(cursor)), jetstream.WithCollections(collections), jetstream.WithHTTPClient(r.client.http), jetstream.WithBatchSize(replayBatchSize), jetstream.WithDownloadConcurrency(r.workers), jetstream.WithSegmentStripes(1), jetstream.WithMaxDownloadAttempts(upstream.MaxAttempts)) + if err != nil { + return err + } + defer client.Close() + for batch, err := range client.Events(ctx) { + if err != nil { + return err + } + if batch.LastCursor() > math.MaxInt64 { + return errors.New("replay cursor exceeds storage range") + } + changes := make([]domain.Change, 0, len(batch.Events())) + for _, event := range batch.Events() { + change, err := r.decodeReplay(ctx, event, wanted) + if err != nil { + return err + } + changes = append(changes, change) + } + if err := apply(ctx, changes, int64(batch.LastCursor())); err != nil { + return err + } + } + return ctx.Err() +} + +func (r *Replay) decodeReplay(ctx context.Context, e jetstream.Event, wanted func(context.Context, string, int64) (bool, error)) (domain.Change, error) { + if e.Seq == 0 || e.Seq > math.MaxInt64 || e.WitnessedAtUS <= 0 { + return domain.Change{}, errors.New("replay event lacks a sequence or witnessed timestamp") + } + if err := atproto.ValidDID(e.DID); err != nil { + return domain.Change{}, err + } + if e.Kind != jetstream.KindSync && e.Kind != jetstream.KindCommit && e.Kind != jetstream.KindIdentity && e.Kind != jetstream.KindAccount { + return domain.Change{}, errors.New("unsupported replay event kind") + } + if e.Kind == jetstream.KindSync { + if e.Sync == nil || e.Sync.DID != e.DID { + return domain.Change{}, errors.New("invalid replay sync marker") + } + return domain.Change{Kind: domain.ResetChange, DID: e.DID, TimeUS: e.WitnessedAtUS, Sequence: int64(e.Seq)}, nil + } + // Reuse the legacy record validation without a network hydration fallback. + if e.Kind == jetstream.KindCommit && (e.Commit == nil || e.Commit.Operation != jetstream.OpDelete && e.Commit.Record == nil) { + return domain.Change{}, errors.New("replay commit lacks its decoded record") + } + e.TimeUS = e.WitnessedAtUS + data, err := json.Marshal(e) + if err != nil { + return domain.Change{}, err + } + // Archive list definitions can precede their first subscriber. Legacy ingestion remains reference-only. + if e.Commit != nil && domain.SoftAll(r.client.softMode) && domain.SoftLists(r.client.softMode) && domain.ListSource(e.Commit.Collection) { + wanted = nil + } + change, err := r.client.decode(ctx, data, wanted) + change.Sequence = int64(e.Seq) + return change, err +} diff --git a/internal/microcosm/replay_test.go b/internal/microcosm/replay_test.go new file mode 100644 index 0000000..6801683 --- /dev/null +++ b/internal/microcosm/replay_test.go @@ -0,0 +1,154 @@ +package microcosm + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/bluesky-social/jetstream/segment" + "github.com/coder/websocket" + + "provenance/internal/domain" + "provenance/internal/upstream" +) + +func TestReplayArchiveCutover(t *testing.T) { + path := filepath.Join(t.TempDir(), "archive.jss") + writer, err := segment.New(segment.Config{Path: path}) + if err != nil { + t.Fatal(err) + } + defer writer.Close() + if _, err := writer.Append(segment.Event{Seq: 1, WitnessedAt: 100, Kind: segment.KindDelete, DID: author, Collection: domain.Collection, Rkey: domain.RecordKey(subject, "work")}); err != nil { + t.Fatal(err) + } + if _, err := writer.Seal(); err != nil { + t.Fatal(err) + } + archive, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + queue, err := upstream.New(ctx, 1) + if err != nil { + t.Fatal(err) + } + defer queue.Close() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("User-Agent") != upstream.UserAgent { + t.Error("missing project User-Agent") + } + switch r.URL.Path { + case "/xrpc/network.bsky.jetstream.planSnapshot": + if r.Header.Get("Authorization") != "Bearer test-key" { + t.Error("archive lacks bearer key") + } + var plan struct { + Collections []string `json:"collections"` + } + if err := json.NewDecoder(r.Body).Decode(&plan); err != nil { + t.Error(err) + } + if len(plan.Collections) != 1 || plan.Collections[0] != domain.Collection { + t.Error("incorrect archive filters") + } + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"sealedTipSeq":1,"plannedThroughSeq":1,"segments":[{"name":"seg_0000000000.jss","index":0,"minSeq":1,"maxSeq":1,"mode":"segment"}],"stats":{}}`)) + case "/xrpc/network.bsky.jetstream.getSegment": + if r.Header.Get("Authorization") != "Bearer test-key" { + t.Error("download lacks bearer key") + } + w.Write(archive) + case "/xrpc/network.bsky.jetstream.subscribeEvents": + if r.Header.Get("Authorization") != "" { + t.Error("secret reached public stream") + } + conn, err := websocket.Accept(w, r, nil) + if err != nil { + t.Error(err) + return + } + defer conn.CloseNow() + event := map[string]any{"$type": "message", "payload": map[string]any{"$type": "network.bsky.jetstream.subscribeEvents#commit", "did": author, "seq": 2, "time": "2026-10-07T00:00:00Z", "witnessedAt": "2026-10-07T00:00:00Z", "rev": "r", "operation": "create", "collection": domain.Collection, "rkey": domain.RecordKey(subject, "work"), "record": validRecord()}} + data, _ := json.Marshal(event) + if err := conn.Write(ctx, websocket.MessageText, data); err != nil { + return + } + conn.Read(ctx) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + replay, err := NewReplay(server.URL, "test-key", domain.SoftDisabled, queue, 1) + if err != nil { + t.Fatal(err) + } + stop := errors.New("complete") + var changes []domain.Change + err = replay.ConsumeReplay(ctx, 0, nil, func(_ context.Context, batch []domain.Change, cursor int64) error { + changes = append(changes, batch...) + if cursor == 2 { + return stop + } + return nil + }) + if !errors.Is(err, stop) || len(changes) != 2 || changes[0].Sequence != 1 || changes[0].Vouch != nil || changes[1].Vouch == nil { + t.Fatalf("archive cutover: %+v %v", changes, err) + } +} + +// This opt-in probe verifies archive access without a segment download. +func TestReplayLivePlan(t *testing.T) { + if os.Getenv("PROVENANCE_REPLAY_SMOKE") != "true" { + t.Skip("live archive probe is disabled") + } + key := os.Getenv("JETSTREAM_API_KEY") + if key == "" { + t.Fatal("JETSTREAM_API_KEY is absent from this process") + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + queue, err := upstream.New(ctx, 1) + if err != nil { + t.Fatal(err) + } + defer queue.Close() + replay, err := NewReplay(DefaultReplayURL, key, domain.SoftAllAuthors, queue, 1) + if err != nil { + t.Fatal(err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, DefaultReplayURL+"/xrpc/network.bsky.jetstream.planSnapshot", strings.NewReader(`{"collections":["app.bsky.graph.follow"],"afterSeq":0,"beforeSeq":1}`)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+key) + resp, err := replay.client.http.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("archive plan returned HTTP %d", resp.StatusCode) + } + var plan struct { + PlannedThrough uint64 `json:"plannedThroughSeq"` + } + if err := json.NewDecoder(io.LimitReader(resp.Body, MaxEventBytes)).Decode(&plan); err != nil { + t.Fatal(err) + } + if plan.PlannedThrough > 1 { + t.Fatal("archive ignored the bounded sequence window") + } +} diff --git a/internal/storage/db/models.go b/internal/storage/db/models.go index 0628c0b..3b525de 100644 --- a/internal/storage/db/models.go +++ b/internal/storage/db/models.go @@ -73,6 +73,12 @@ type RecordVersion struct { TimeUs int64 } +type ReplayAccount struct { + Source string + Author int64 + Sequence int64 +} + type Secret struct { Kind string ID string diff --git a/internal/storage/db/querier.go b/internal/storage/db/querier.go index 978cd29..ff041f4 100644 --- a/internal/storage/db/querier.go +++ b/internal/storage/db/querier.go @@ -11,9 +11,11 @@ import ( type Querier interface { AddSecret(ctx context.Context, arg AddSecretParams) error AdvanceCursor(ctx context.Context, arg AdvanceCursorParams) error + AdvanceReplayAccount(ctx context.Context, arg AdvanceReplayAccountParams) error AuthorRecords(ctx context.Context, author string) ([]Vouch, error) CachedTrust(ctx context.Context, arg CachedTrustParams) (string, error) ClearListAuthor(ctx context.Context, author string) error + ClearListVersions(ctx context.Context, author string) error ClearSoft(ctx context.Context, author string) error ClearSoftVersions(ctx context.Context, author string) error ClearVersions(ctx context.Context, author string) error @@ -37,10 +39,13 @@ type Querier interface { ListBootstrap(ctx context.Context, author string) (int64, error) ListGraph(ctx context.Context, arg ListGraphParams) ([]ListEdge, error) ListIncoming(ctx context.Context, arg ListIncomingParams) ([]ListEdge, error) + ListOutgoing(ctx context.Context, arg ListOutgoingParams) ([]ListEdge, error) ListRecord(ctx context.Context, arg ListRecordParams) (ListRecordRow, error) ListURIRecords(ctx context.Context, listUri string) ([]ListURIRecordsRow, error) ListVersion(ctx context.Context, arg ListVersionParams) (int64, error) ListWatermark(ctx context.Context, listUri string) (int64, error) + // Outgoing reads preserve per-pair selection and use author indexes rather than a global graph scan. + Outgoing(ctx context.Context, arg OutgoingParams) ([]Vouch, error) OwnVouches(ctx context.Context, arg OwnVouchesParams) ([]Vouch, error) Participant(ctx context.Context, did string) (Participant, error) Participants(ctx context.Context, limit int64) ([]Participant, error) @@ -55,6 +60,8 @@ type Querier interface { RemoveSecret(ctx context.Context, arg RemoveSecretParams) error ReplaceSoft(ctx context.Context, arg ReplaceSoftParams) error ReplaceVouch(ctx context.Context, arg ReplaceVouchParams) error + ReplayClearLists(ctx context.Context, arg ReplayClearListsParams) error + ReplaySequence(ctx context.Context, arg ReplaySequenceParams) (int64, error) SaveDependency(ctx context.Context, arg SaveDependencyParams) error SaveTrust(ctx context.Context, arg SaveTrustParams) error SetAccount(ctx context.Context, arg SetAccountParams) error @@ -72,6 +79,7 @@ type Querier interface { // The caller sorts complete results. Avoid a global DID sort before the edge limit. SoftGraph(ctx context.Context, arg SoftGraphParams) ([]SoftEdge, error) SoftIncoming(ctx context.Context, arg SoftIncomingParams) ([]SoftEdge, error) + SoftOutgoing(ctx context.Context, arg SoftOutgoingParams) ([]SoftEdge, error) SoftRecords(ctx context.Context, author string) ([]SoftVouch, error) SoftVersion(ctx context.Context, arg SoftVersionParams) (int64, error) Tags(ctx context.Context, arg TagsParams) ([]string, error) diff --git a/internal/storage/db/queries.sql.go b/internal/storage/db/queries.sql.go index d5c157c..beeee3a 100644 --- a/internal/storage/db/queries.sql.go +++ b/internal/storage/db/queries.sql.go @@ -45,6 +45,22 @@ func (q *Queries) AdvanceCursor(ctx context.Context, arg AdvanceCursorParams) er return err } +const advanceReplayAccount = `-- name: AdvanceReplayAccount :exec +INSERT INTO replay_accounts(source, author, sequence) VALUES (?1, (SELECT id FROM identities WHERE did = CAST(?2 AS TEXT)), ?3) +ON CONFLICT(source, author) DO UPDATE SET sequence = excluded.sequence +` + +type AdvanceReplayAccountParams struct { + Source string + Author string + Sequence int64 +} + +func (q *Queries) AdvanceReplayAccount(ctx context.Context, arg AdvanceReplayAccountParams) error { + _, err := q.db.ExecContext(ctx, advanceReplayAccount, arg.Source, arg.Author, arg.Sequence) + return err +} + const authorRecords = `-- name: AuthorRecords :many SELECT author, subject, tag, score, created_at, record_key, comment, source FROM vouches WHERE author = ? ` @@ -106,6 +122,15 @@ func (q *Queries) ClearListAuthor(ctx context.Context, author string) error { return err } +const clearListVersions = `-- name: ClearListVersions :exec +DELETE FROM list_versions WHERE author = (SELECT id FROM identities WHERE did = CAST(?1 AS TEXT)) +` + +func (q *Queries) ClearListVersions(ctx context.Context, author string) error { + _, err := q.db.ExecContext(ctx, clearListVersions, author) + return err +} + const clearSoft = `-- name: ClearSoft :exec DELETE FROM soft_vouches WHERE author = (SELECT id FROM identities WHERE did = CAST(?1 AS TEXT)) ` @@ -611,6 +636,61 @@ func (q *Queries) ListIncoming(ctx context.Context, arg ListIncomingParams) ([]L return items, nil } +const listOutgoing = `-- name: ListOutgoing :many +SELECT e.author, e.subject, e.source, e.record_key, e.created_at, e.list_uri, e.list_author, e.list_key, e.item_key FROM list_edges e +WHERE e.author = ?1 +AND (CAST(?2 AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = e.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = e.author AND v.subject = e.subject AND (v.tag = ?3 OR (CAST(?4 AS INTEGER) = 1 AND v.tag = ''))) +LIMIT ?5 +` + +type ListOutgoingParams struct { + Author string + AllAuthors int64 + Tag string + Fallback int64 + MaxRows int64 +} + +func (q *Queries) ListOutgoing(ctx context.Context, arg ListOutgoingParams) ([]ListEdge, error) { + rows, err := q.db.QueryContext(ctx, listOutgoing, + arg.Author, + arg.AllAuthors, + arg.Tag, + arg.Fallback, + arg.MaxRows, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListEdge{} + for rows.Next() { + var i ListEdge + if err := rows.Scan( + &i.Author, + &i.Subject, + &i.Source, + &i.RecordKey, + &i.CreatedAt, + &i.ListUri, + &i.ListAuthor, + &i.ListKey, + &i.ItemKey, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listRecord = `-- name: ListRecord :one SELECT a.did AS author, r.source, r.record_key, r.list_uri, COALESCE(s.did, '') AS subject, r.purpose, r.created_at FROM list_records r JOIN identities a ON a.id = r.author LEFT JOIN identities s ON s.id = r.subject @@ -723,6 +803,59 @@ func (q *Queries) ListWatermark(ctx context.Context, listUri string) (int64, err return bootstrap_us, err } +const outgoing = `-- name: Outgoing :many +SELECT v.author, v.subject, v.tag, v.score, v.created_at, v.record_key, v.comment, v.source FROM vouches v JOIN participants p ON p.did = v.author +WHERE v.author = ?1 AND p.active = 1 +AND NOT EXISTS (SELECT 1 FROM participants target WHERE target.did = v.subject AND target.active = 0) +AND (v.tag = ?2 OR (CAST(?3 AS INTEGER) = 1 AND v.tag = '')) +ORDER BY v.subject, v.tag LIMIT ?4 +` + +type OutgoingParams struct { + Author string + Tag string + Fallback int64 + MaxRows int64 +} + +// Outgoing reads preserve per-pair selection and use author indexes rather than a global graph scan. +func (q *Queries) Outgoing(ctx context.Context, arg OutgoingParams) ([]Vouch, error) { + rows, err := q.db.QueryContext(ctx, outgoing, + arg.Author, + arg.Tag, + arg.Fallback, + arg.MaxRows, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []Vouch{} + for rows.Next() { + var i Vouch + if err := rows.Scan( + &i.Author, + &i.Subject, + &i.Tag, + &i.Score, + &i.CreatedAt, + &i.RecordKey, + &i.Comment, + &i.Source, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const ownVouches = `-- name: OwnVouches :many SELECT vouches.author, vouches.subject, vouches.tag, vouches.score, vouches.created_at, vouches.record_key, vouches.comment, vouches.source FROM vouches JOIN participants ON participants.did = vouches.author WHERE participants.active = 1 AND author = ? ORDER BY subject, tag LIMIT ? OFFSET ? @@ -1028,6 +1161,38 @@ func (q *Queries) ReplaceVouch(ctx context.Context, arg ReplaceVouchParams) erro return err } +const replayClearLists = `-- name: ReplayClearLists :exec +DELETE FROM list_records WHERE author = (SELECT id FROM identities WHERE did = CAST(?1 AS TEXT)) +AND NOT EXISTS (SELECT 1 FROM list_snapshots s WHERE s.list_uri = list_records.list_uri AND s.bootstrap_us >= ?2) +AND (source != 'app.bsky.graph.listblock' OR NOT EXISTS (SELECT 1 FROM list_bootstraps b WHERE b.author = list_records.author AND b.bootstrap_us >= ?2)) +` + +type ReplayClearListsParams struct { + Author string + TimeUs int64 +} + +func (q *Queries) ReplayClearLists(ctx context.Context, arg ReplayClearListsParams) error { + _, err := q.db.ExecContext(ctx, replayClearLists, arg.Author, arg.TimeUs) + return err +} + +const replaySequence = `-- name: ReplaySequence :one +SELECT sequence FROM replay_accounts WHERE source = ?1 AND author = (SELECT id FROM identities WHERE did = CAST(?2 AS TEXT)) +` + +type ReplaySequenceParams struct { + Source string + Author string +} + +func (q *Queries) ReplaySequence(ctx context.Context, arg ReplaySequenceParams) (int64, error) { + row := q.db.QueryRowContext(ctx, replaySequence, arg.Source, arg.Author) + var sequence int64 + err := row.Scan(&sequence) + return sequence, err +} + const saveDependency = `-- name: SaveDependency :exec INSERT OR IGNORE INTO trust_dependencies(cache_key, author, source, record_key) VALUES (?1, (SELECT id FROM identities WHERE did = CAST(?2 AS TEXT)), ?3, ?4) @@ -1388,6 +1553,58 @@ func (q *Queries) SoftIncoming(ctx context.Context, arg SoftIncomingParams) ([]S return items, nil } +const softOutgoing = `-- name: SoftOutgoing :many +SELECT s.author, s.subject, s.source, s.record_key, s.score, s.created_at FROM soft_edges s +WHERE s.author = ?1 +AND (CAST(?2 AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = s.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = s.author AND v.subject = s.subject AND (v.tag = ?3 OR (CAST(?4 AS INTEGER) = 1 AND v.tag = ''))) +LIMIT ?5 +` + +type SoftOutgoingParams struct { + Author string + AllAuthors int64 + Tag string + Fallback int64 + MaxRows int64 +} + +func (q *Queries) SoftOutgoing(ctx context.Context, arg SoftOutgoingParams) ([]SoftEdge, error) { + rows, err := q.db.QueryContext(ctx, softOutgoing, + arg.Author, + arg.AllAuthors, + arg.Tag, + arg.Fallback, + arg.MaxRows, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []SoftEdge{} + for rows.Next() { + var i SoftEdge + if err := rows.Scan( + &i.Author, + &i.Subject, + &i.Source, + &i.RecordKey, + &i.Score, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const softRecords = `-- name: SoftRecords :many SELECT author, subject, source, record_key, score, created_at FROM soft_vouches WHERE author = (SELECT id FROM identities WHERE did = CAST(?1 AS TEXT)) ` diff --git a/internal/storage/events.go b/internal/storage/events.go index 746fb98..8316d96 100644 --- a/internal/storage/events.go +++ b/internal/storage/events.go @@ -28,6 +28,16 @@ func (s *SQLite) Apply(ctx context.Context, source string, change domain.Change) } defer tx.Rollback() q := s.queries.WithTx(tx) + if err := s.applyEvent(ctx, q, change); err != nil { + return err + } + if err := q.AdvanceCursor(ctx, db.AdvanceCursorParams{Source: source, TimeUs: change.TimeUS}); err != nil { + return err + } + return tx.Commit() +} + +func (s *SQLite) applyEvent(ctx context.Context, q *db.Queries, change domain.Change) error { if domain.SoftLists(s.softMode) && (domain.ListSource(change.Source) || change.Kind == domain.AccountChange) { if err := s.applyList(ctx, q, change); err != nil { return err @@ -43,10 +53,7 @@ func (s *SQLite) Apply(ctx context.Context, source string, change domain.Change) return err } } - if err := q.AdvanceCursor(ctx, db.AdvanceCursorParams{Source: source, TimeUs: change.TimeUS}); err != nil { - return err - } - return tx.Commit() + return nil } func applyChange(ctx context.Context, q *db.Queries, change domain.Change) error { @@ -65,7 +72,7 @@ func applyChange(ctx context.Context, q *db.Queries, change domain.Change) error } at := time.UnixMicro(change.TimeUS).UTC().Format(time.RFC3339Nano) if change.Kind == domain.AccountChange { - if change.TimeUS <= p.AccountUs { + if oldEvent(change, p.AccountUs) { return nil } if !change.Active && change.Status == domain.DeletedAccount { @@ -93,7 +100,7 @@ func applyChange(ctx context.Context, q *db.Queries, change domain.Change) error if err != nil && !errors.Is(err, sql.ErrNoRows) { return err } - if change.TimeUS <= version { + if oldEvent(change, version) { return nil } if err := q.DeleteRecord(ctx, db.DeleteRecordParams{Author: change.DID, RecordKey: change.RecordKey}); err != nil { diff --git a/internal/storage/lists.go b/internal/storage/lists.go index e0089a2..13aa095 100644 --- a/internal/storage/lists.go +++ b/internal/storage/lists.go @@ -58,6 +58,22 @@ func listFresh(ctx context.Context, q *db.Queries, change domain.Change) (bool, if account.Active == 0 { version = max(version, account.AccountUs) } + if change.Sequence > 0 && change.TimeUS == version { + // Snapshot watermarks still override archive records with the same timestamp. + if change.TimeUS <= account.BootstrapUs { + return false, nil + } + if change.Source == domain.ListBlockCollection { + bootstrap, err := q.ListBootstrap(ctx, change.DID) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return false, err + } + if change.TimeUS <= bootstrap { + return false, nil + } + } + return true, nil + } return change.TimeUS > version, nil } @@ -185,7 +201,7 @@ func (s *SQLite) applyList(ctx context.Context, q *db.Queries, change domain.Cha } } else { wanted, err := q.WantsListOwner(ctx, change.DID) - if err != nil || !wanted { + if err != nil || !wanted && !(change.Sequence > 0 && domain.SoftAll(s.softMode)) { return err } } @@ -229,7 +245,7 @@ func (s *SQLite) applyList(ctx context.Context, q *db.Queries, change domain.Cha return err } // Retain unknown deletions as tombstones, but do not store unrelated list contents. - applyNew = applyNew && wanted + applyNew = applyNew && (wanted || change.Sequence > 0 && domain.SoftAll(s.softMode)) } if change.RecordKey == "" { return errors.New("list event key is required") @@ -264,7 +280,7 @@ func (s *SQLite) listAccount(ctx context.Context, q *db.Queries, change domain.C if err != nil { return err } - if change.TimeUS <= max(a.AccountUs, a.BootstrapUs) { + if oldEvent(change, a.AccountUs) || change.TimeUS <= a.BootstrapUs { return nil } if !change.Active && change.Status == domain.DeletedAccount { diff --git a/internal/storage/neighbors.go b/internal/storage/neighbors.go new file mode 100644 index 0000000..7bb402e --- /dev/null +++ b/internal/storage/neighbors.go @@ -0,0 +1,57 @@ +package storage + +import ( + "context" + "errors" + + "provenance/internal/domain" + "provenance/internal/storage/db" +) + +func (s *SQLite) GraphRevision(ctx context.Context) (int64, error) { + return s.queries.GraphRevision(ctx) +} + +// Read one author's records and the revision in a short transaction. No global graph load occurs. +func (s *SQLite) Outgoing(ctx context.Context, author, tag string, mode domain.Mode, limit int) ([]domain.Vouch, int64, error) { + if limit < 0 { + return nil, 0, errors.New("negative neighbor limit") + } + tx, err := s.conn.BeginTx(ctx, nil) + if err != nil { + return nil, 0, err + } + defer tx.Rollback() + q := s.queries.WithTx(tx) + revision, err := q.GraphRevision(ctx) + if err != nil { + return nil, 0, err + } + fallback, all := int64(0), int64(0) + if mode == domain.Fallback { + fallback = 1 + } + if domain.SoftAll(s.softMode) { + all = 1 + } + rows, err := q.Outgoing(ctx, db.OutgoingParams{Author: author, Tag: tag, Fallback: fallback, MaxRows: int64(limit)}) + if err != nil { + return nil, 0, err + } + result := vouchRows(rows) + if s.softMode != domain.SoftDisabled && mode != domain.Tagged && len(result) < limit { + soft, err := q.SoftOutgoing(ctx, db.SoftOutgoingParams{Author: author, Tag: tag, Fallback: fallback, AllAuthors: all, MaxRows: int64(limit - len(result))}) + if err != nil { + return nil, 0, err + } + result = appendSoftRows(result, soft) + if domain.SoftLists(s.softMode) && len(result) < limit { + lists, err := q.ListOutgoing(ctx, db.ListOutgoingParams{Author: author, Tag: tag, Fallback: fallback, AllAuthors: all, MaxRows: int64(limit - len(result))}) + if err != nil { + return nil, 0, err + } + result = appendListRows(result, lists) + } + } + return result, revision, tx.Commit() +} diff --git a/internal/storage/neighbors_test.go b/internal/storage/neighbors_test.go new file mode 100644 index 0000000..b5bff51 --- /dev/null +++ b/internal/storage/neighbors_test.go @@ -0,0 +1,58 @@ +package storage + +import ( + "context" + "path/filepath" + "reflect" + "testing" + + "provenance/internal/domain" + "provenance/internal/trust" +) + +func TestOutgoingSelection(t *testing.T) { + ctx := context.Background() + s, err := OpenWithSoft(filepath.Join(t.TempDir(), "graph.db"), domain.SoftAllAuthors) + if err != nil { + t.Fatal(err) + } + defer s.Close() + const author = "did:plc:author" + if err := s.Join(ctx, author); err != nil { + t.Fatal(err) + } + for index, v := range []domain.Vouch{ + {Author: author, Subject: "did:plc:target", Score: 0}, + {Author: author, Subject: "did:plc:target", Tag: "topic", Score: 2}, + {Author: author, Subject: "did:plc:soft", Source: domain.FollowCollection, RecordKey: "follow", Score: domain.FollowScore}, + {Author: "did:plc:unrelated", Subject: "did:plc:soft", Source: domain.FollowCollection, RecordKey: "other", Score: domain.FollowScore}, + } { + key := v.RecordKey + if key == "" { + key = domain.RecordKey(v.Subject, v.Tag) + } + if err := s.Apply(ctx, "fixture", domain.Change{Kind: domain.RecordChange, DID: v.Author, Source: v.Source, RecordKey: key, Vouch: &v, TimeUS: int64(index + 1)}); err != nil { + t.Fatal(err) + } + } + for _, mode := range []domain.Mode{domain.Default, domain.Tagged, domain.Fallback} { + tag := "topic" + if mode == domain.Default { + tag = "" + } + all, revision, err := s.TrustGraph(ctx, tag, mode, 100) + if err != nil { + t.Fatal(err) + } + var expected []domain.Vouch + for _, v := range trust.Select(all, tag, mode) { + if v.Author == author { + expected = append(expected, v) + } + } + rows, gotRevision, err := s.Outgoing(ctx, author, tag, mode, 100) + if err != nil || revision != gotRevision || !reflect.DeepEqual(expected, trust.Select(rows, tag, mode)) { + t.Fatalf("mode %v: %v %v %v", mode, expected, rows, err) + } + } +} diff --git a/internal/storage/queries.sql b/internal/storage/queries.sql index ada3d25..1820960 100644 --- a/internal/storage/queries.sql +++ b/internal/storage/queries.sql @@ -1,6 +1,13 @@ -- name: Join :exec INSERT INTO participants(did) VALUES (?) ON CONFLICT(did) DO NOTHING; +-- name: ReplaySequence :one +SELECT sequence FROM replay_accounts WHERE source = @source AND author = (SELECT id FROM identities WHERE did = CAST(@author AS TEXT)); + +-- name: AdvanceReplayAccount :exec +INSERT INTO replay_accounts(source, author, sequence) VALUES (@source, (SELECT id FROM identities WHERE did = CAST(@author AS TEXT)), @sequence) +ON CONFLICT(source, author) DO UPDATE SET sequence = excluded.sequence; + -- name: Participant :one SELECT * FROM participants WHERE did = ?; @@ -156,6 +163,14 @@ DELETE FROM list_records WHERE author = (SELECT id FROM identities WHERE did = C -- name: ClearListAuthor :exec DELETE FROM list_records WHERE author = (SELECT id FROM identities WHERE did = CAST(@author AS TEXT)); +-- name: ReplayClearLists :exec +DELETE FROM list_records WHERE author = (SELECT id FROM identities WHERE did = CAST(@author AS TEXT)) +AND NOT EXISTS (SELECT 1 FROM list_snapshots s WHERE s.list_uri = list_records.list_uri AND s.bootstrap_us >= @time_us) +AND (source != 'app.bsky.graph.listblock' OR NOT EXISTS (SELECT 1 FROM list_bootstraps b WHERE b.author = list_records.author AND b.bootstrap_us >= @time_us)); + +-- name: ClearListVersions :exec +DELETE FROM list_versions WHERE author = (SELECT id FROM identities WHERE did = CAST(@author AS TEXT)); + -- name: ListVersion :one SELECT time_us FROM list_versions WHERE author = (SELECT id FROM identities WHERE did = CAST(@author AS TEXT)) AND source = @source AND record_key = @record_key; @@ -190,6 +205,28 @@ AND NOT EXISTS (SELECT 1 FROM participants target WHERE target.did = vouches.sub AND (tag = sqlc.arg(tag) OR (CAST(sqlc.arg(fallback) AS INTEGER) = 1 AND tag = '')) ORDER BY author, subject, tag LIMIT sqlc.arg(max_rows); +-- Outgoing reads preserve per-pair selection and use author indexes rather than a global graph scan. +-- name: Outgoing :many +SELECT v.* FROM vouches v JOIN participants p ON p.did = v.author +WHERE v.author = @author AND p.active = 1 +AND NOT EXISTS (SELECT 1 FROM participants target WHERE target.did = v.subject AND target.active = 0) +AND (v.tag = @tag OR (CAST(@fallback AS INTEGER) = 1 AND v.tag = '')) +ORDER BY v.subject, v.tag LIMIT @max_rows; + +-- name: SoftOutgoing :many +SELECT s.* FROM soft_edges s +WHERE s.author = @author +AND (CAST(@all_authors AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = s.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = s.author AND v.subject = s.subject AND (v.tag = @tag OR (CAST(@fallback AS INTEGER) = 1 AND v.tag = ''))) +LIMIT @max_rows; + +-- name: ListOutgoing :many +SELECT e.* FROM list_edges e +WHERE e.author = @author +AND (CAST(@all_authors AS INTEGER) = 1 OR EXISTS (SELECT 1 FROM participants p WHERE p.did = e.author AND p.active = 1)) +AND NOT EXISTS (SELECT 1 FROM vouches v WHERE v.author = e.author AND v.subject = e.subject AND (v.tag = @tag OR (CAST(@fallback AS INTEGER) = 1 AND v.tag = ''))) +LIMIT @max_rows; + -- name: OwnVouches :many SELECT vouches.* FROM vouches JOIN participants ON participants.did = vouches.author WHERE participants.active = 1 AND author = ? ORDER BY subject, tag LIMIT ? OFFSET ?; diff --git a/internal/storage/replay.go b/internal/storage/replay.go new file mode 100644 index 0000000..e6e118e --- /dev/null +++ b/internal/storage/replay.go @@ -0,0 +1,97 @@ +package storage + +import ( + "context" + "database/sql" + "errors" + + "github.com/bluesky-social/indigo/atproto/syntax" + + "provenance/internal/domain" + "provenance/internal/storage/db" +) + +// Sequence guards reject replay duplicates before timestamp checks. Equal-time mutations retain their sequence order. +func oldEvent(change domain.Change, version int64) bool { + return change.TimeUS < version || change.TimeUS == version && change.Sequence == 0 +} + +// Commit every SDK batch with its resume cursor. A failed batch leaves no partial mutations. +func (s *SQLite) ApplyReplay(ctx context.Context, source string, changes []domain.Change, cursor int64) error { + if source == "" || cursor < 0 { + return errors.New("invalid replay checkpoint") + } + tx, err := s.conn.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + q := s.queries.WithTx(tx) + for _, change := range changes { + if change.Sequence <= 0 || change.Sequence > cursor || change.TimeUS <= 0 { + return errors.New("invalid replay sequence or timestamp") + } + if _, err := syntax.ParseDID(change.DID); err != nil { + return err + } + if err := q.PutIdentity(ctx, change.DID); err != nil { + return err + } + previous, err := q.ReplaySequence(ctx, db.ReplaySequenceParams{Source: source, Author: change.DID}) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + if change.Sequence <= previous { + continue + } + if change.Kind == domain.ResetChange { + if err := s.resetReplay(ctx, q, change); err != nil { + return err + } + } else if err := s.applyEvent(ctx, q, change); err != nil { + return err + } + if err := q.AdvanceReplayAccount(ctx, db.AdvanceReplayAccountParams{Source: source, Author: change.DID, Sequence: change.Sequence}); err != nil { + return err + } + } + if err := q.AdvanceCursor(ctx, db.AdvanceCursorParams{Source: source, TimeUs: cursor}); err != nil { + return err + } + return tx.Commit() +} + +func (s *SQLite) resetReplay(ctx context.Context, q *db.Queries, change domain.Change) error { + p, err := q.Participant(ctx, change.DID) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + if err == nil && change.TimeUS > p.BootstrapUs { + if err := q.ClearVouches(ctx, change.DID); err != nil { + return err + } + if err := q.ClearVersions(ctx, change.DID); err != nil { + return err + } + } + a, err := q.SoftAccount(ctx, change.DID) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + if err != nil { + return err + } + if change.TimeUS <= a.BootstrapUs { + return nil + } + if err := q.ClearSoft(ctx, change.DID); err != nil { + return err + } + if err := q.ClearSoftVersions(ctx, change.DID); err != nil { + return err + } + if err := q.ReplayClearLists(ctx, db.ReplayClearListsParams{Author: change.DID, TimeUs: change.TimeUS}); err != nil { + return err + } + return q.ClearListVersions(ctx, change.DID) +} diff --git a/internal/storage/replay_schema.sql b/internal/storage/replay_schema.sql new file mode 100644 index 0000000..e085eac --- /dev/null +++ b/internal/storage/replay_schema.sql @@ -0,0 +1,7 @@ +-- Per-account sequence guards retain order when archive batches interleave accounts. +CREATE TABLE IF NOT EXISTS replay_accounts ( + source TEXT NOT NULL, + author INTEGER NOT NULL REFERENCES identities(id), + sequence INTEGER NOT NULL CHECK(sequence > 0), + PRIMARY KEY(source, author) +) WITHOUT ROWID; diff --git a/internal/storage/replay_test.go b/internal/storage/replay_test.go new file mode 100644 index 0000000..231e6c2 --- /dev/null +++ b/internal/storage/replay_test.go @@ -0,0 +1,122 @@ +package storage + +import ( + "context" + "path/filepath" + "testing" + + "provenance/internal/domain" +) + +func TestReplayAtomicRestart(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "replay.db") + s, err := OpenWithSoft(path, domain.SoftAllAuthors) + if err != nil { + t.Fatal(err) + } + defer func() { s.Close() }() + const author = "did:plc:author" + const source = "instance:one" + v := domain.Vouch{Author: author, Subject: "did:plc:target", Source: domain.FollowCollection, RecordKey: "follow", Score: domain.FollowScore} + change := domain.Change{Kind: domain.RecordChange, DID: author, Source: v.Source, RecordKey: v.RecordKey, Vouch: &v, TimeUS: 100, Sequence: 1} + if err := s.ApplyReplay(ctx, source, []domain.Change{change}, 1); err != nil { + t.Fatal(err) + } + bad := change + bad.Sequence = 3 + bad.Vouch = &domain.Vouch{Author: author, Subject: v.Subject, Source: v.Source, RecordKey: v.RecordKey, Score: 2} + deleted := change + deleted.Sequence, deleted.Vouch = 2, nil + if err := s.ApplyReplay(ctx, source, []domain.Change{deleted, bad}, 3); err == nil { + t.Fatal("invalid batch succeeded") + } + if cursor, err := s.Cursor(ctx, source, 0); err != nil || cursor != 1 { + t.Fatalf("failed batch advanced cursor: %d %v", cursor, err) + } + if err := s.ApplyReplay(ctx, source, []domain.Change{deleted}, 2); err != nil { + t.Fatal(err) + } + s.Close() + s, err = OpenWithSoft(path, domain.SoftAllAuthors) + if err != nil { + t.Fatal(err) + } + if err := s.ApplyReplay(ctx, source, []domain.Change{change}, 2); err != nil { + t.Fatal(err) + } + rows, err := s.Graph(ctx, "", domain.Default, 10) + if err != nil || len(rows) != 0 { + t.Fatalf("equal-time deletion or replay guard failed: %+v %v", rows, err) + } + change.Sequence = 3 + old := change + reset := domain.Change{Kind: domain.ResetChange, DID: author, TimeUS: 100, Sequence: 4} + change.Sequence = 5 + replacement := v + replacement.RecordKey = "replacement" + change.RecordKey, change.Vouch = replacement.RecordKey, &replacement + if err := s.ApplyReplay(ctx, source, []domain.Change{old, reset, change}, 5); err != nil { + t.Fatal(err) + } + rows, err = s.Graph(ctx, "", domain.Default, 10) + if err != nil || len(rows) != 1 { + t.Fatalf("sync replacement failed: %+v %v", rows, err) + } +} + +func TestReplayV8Migration(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "v8.db") + s, err := Open(path) + if err != nil { + t.Fatal(err) + } + if err := s.Join(ctx, "did:plc:author"); err != nil { + t.Fatal(err) + } + if _, err := s.conn.Exec("DROP TABLE replay_accounts"); err != nil { + t.Fatal(err) + } + if _, err := s.conn.Exec("PRAGMA user_version=8"); err != nil { + t.Fatal(err) + } + s.Close() + s, err = Open(path) + if err != nil { + t.Fatal(err) + } + defer s.Close() + if _, err := s.Participant(ctx, "did:plc:author"); err != nil { + t.Fatal("migration lost participant", err) + } + change := domain.Change{Kind: domain.IdentityChange, DID: "did:plc:author", TimeUS: 100, Sequence: 1} + if err := s.ApplyReplay(ctx, "migration", []domain.Change{change}, 1); err != nil { + t.Fatal(err) + } +} + +func TestReplayListLifecycle(t *testing.T) { + ctx := context.Background() + s, err := OpenWithSoft(filepath.Join(t.TempDir(), "lists.db"), domain.SoftAllAuthorLists) + if err != nil { + t.Fatal(err) + } + defer s.Close() + const author = "did:plc:owner" + r := domain.ListRecord{Author: author, Source: domain.ListCollection, RecordKey: "list", ListURI: "at://did:plc:owner/app.bsky.graph.list/list", Purpose: domain.ModerationList, CreatedAt: "2026-10-07T00:00:00Z"} + create := domain.Change{Kind: domain.RecordChange, DID: author, Source: r.Source, RecordKey: r.RecordKey, ListRecord: &r, TimeUS: 100, Sequence: 1} + active := domain.Change{Kind: domain.AccountChange, DID: author, Active: true, TimeUS: 100, Sequence: 2} + deleted := active + deleted.Active, deleted.Status, deleted.Sequence = false, domain.DeletedAccount, 3 + if err := s.ApplyReplay(ctx, "lists", []domain.Change{create, active, deleted}, 3); err != nil { + t.Fatal(err) + } + var count int + if err := s.conn.QueryRowContext(ctx, "SELECT count(*) FROM list_records").Scan(&count); err != nil { + t.Fatal(err) + } + if count != 0 { + t.Fatal("equal-time account deletion retained list records") + } +} diff --git a/internal/storage/scale_test.go b/internal/storage/scale_test.go index 3e2542b..2fae228 100644 --- a/internal/storage/scale_test.go +++ b/internal/storage/scale_test.go @@ -54,19 +54,35 @@ func BenchmarkSoftScale(b *testing.B) { b.Fatal(err) } bytesPerRecord := float64(pages*pageSize) / float64(count) - b.ReportAllocs() - b.ResetTimer() - for index := 0; index < b.N; index++ { - rows, _, err := s.TrustGraph(ctx, "", domain.Default, count+1) - if err != nil || len(rows) != count { - b.Fatalf("graph: %d %v", len(rows), err) - } - selected := trust.Select(rows, "", domain.Default) - if len(selected) != count { - b.Fatal("selection lost edges") - } + for name, read := range map[string]func() ([]domain.Vouch, error){ + "graph": func() ([]domain.Vouch, error) { + rows, _, err := s.TrustGraph(ctx, "", domain.Default, count+1) + return rows, err + }, + "outgoing": func() ([]domain.Vouch, error) { + rows, _, err := s.Outgoing(ctx, fmt.Sprintf("did:plc:%024d", 0), "", domain.Default, count/authorCount+1) + return rows, err + }, + } { + b.Run(name, func(b *testing.B) { + b.ReportAllocs() + for index := 0; index < b.N; index++ { + rows, err := read() + if err != nil { + b.Fatal(err) + } + selected := trust.Select(rows, "", domain.Default) + expected := count + if name == "outgoing" { + expected /= authorCount + } + if len(selected) != expected { + b.Fatal("selection lost edges") + } + } + b.ReportMetric(bytesPerRecord, "disk-bytes/record") + }) } - b.ReportMetric(bytesPerRecord, "disk-bytes/record") }) } } diff --git a/internal/storage/soft.go b/internal/storage/soft.go index 690a33e..fc1ca6b 100644 --- a/internal/storage/soft.go +++ b/internal/storage/soft.go @@ -86,7 +86,7 @@ func (s *SQLite) applySoft(ctx context.Context, q *db.Queries, change domain.Cha return nil } if change.Kind == domain.AccountChange { - if change.TimeUS <= a.AccountUs { + if oldEvent(change, a.AccountUs) { return nil } active := int64(0) @@ -114,7 +114,7 @@ func (s *SQLite) applySoft(ctx context.Context, q *db.Queries, change domain.Cha if err != nil && !errors.Is(err, sql.ErrNoRows) { return err } - if change.TimeUS <= version { + if oldEvent(change, version) { return nil } if v := change.Vouch; v != nil { diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index 7c9d959..bf5b860 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -49,7 +49,10 @@ var migration8 string //go:embed migration8_copy.sql var migration8Copy string -const SchemaVersion = 8 +//go:embed replay_schema.sql +var replaySchema string + +const SchemaVersion = 9 type SQLite struct { conn *sql.DB @@ -114,6 +117,15 @@ func migrate(conn *sql.DB) error { if version == SchemaVersion { return tx.Commit() } + if version == 8 { + if _, err := tx.Exec(replaySchema); err != nil { + return err + } + if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version=%d", SchemaVersion)); err != nil { + return err + } + return tx.Commit() + } if version == 1 { if _, err := tx.Exec(migration2); err != nil { return err @@ -166,6 +178,9 @@ func migrate(conn *sql.DB) error { if _, err := tx.Exec(listsSchema); err != nil { return err } + if _, err := tx.Exec(replaySchema); err != nil { + return err + } if version > 0 { if _, err := tx.Exec(migration8Copy); err != nil { return err diff --git a/internal/storage/store.go b/internal/storage/store.go index 9ff3c2d..1ae10a3 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -28,7 +28,10 @@ type Store interface { Tags(context.Context, string, int) ([]string, error) Cursor(context.Context, string, int64) (int64, error) Apply(context.Context, string, domain.Change) error + ApplyReplay(context.Context, string, []domain.Change, int64) error TrustGraph(context.Context, string, domain.Mode, int) ([]domain.Vouch, int64, error) + GraphRevision(context.Context) (int64, error) + Outgoing(context.Context, string, string, domain.Mode, int) ([]domain.Vouch, int64, error) Incoming(context.Context, string, string, domain.Mode, int) ([]domain.Vouch, error) CachedTrust(context.Context, string, time.Time) (trust.Result, error) SaveTrust(context.Context, string, trust.Result, []trust.Dependency, int64, time.Time) (bool, error) diff --git a/internal/trust/algorithm.go b/internal/trust/algorithm.go index eb26854..ca2d7f8 100644 --- a/internal/trust/algorithm.go +++ b/internal/trust/algorithm.go @@ -6,13 +6,23 @@ import ( "provenance/internal/domain" ) -// Algorithm calculates relative trust without storage or network access. +// Algorithm calculates relative trust through supplied edges or a query-local graph reader. // Implementations must support concurrent requests and respect context cancellation and limits. type Algorithm interface { Describe() Descriptor Calculate(context.Context, Input) (Result, error) } +// NeighborAlgorithm accepts Input.Graph instead of a complete edge buffer. +type NeighborAlgorithm interface { + Algorithm + UsesNeighbors() +} + +type Graph interface { + Outgoing(context.Context, string, int) ([]domain.Vouch, error) +} + // Account parameters declare identity lists for validation and API documentation. type AccountParameter struct { Name string @@ -34,4 +44,6 @@ type Input struct { Target string Limits Limits Accounts Accounts + Graph Graph + Budget *Budget } diff --git a/internal/trust/algorithms/neighbors_test.go b/internal/trust/algorithms/neighbors_test.go new file mode 100644 index 0000000..d80278c --- /dev/null +++ b/internal/trust/algorithms/neighbors_test.go @@ -0,0 +1,87 @@ +package algorithms_test + +import ( + "context" + "fmt" + "math/rand/v2" + "reflect" + "testing" + + "provenance/internal/domain" + "provenance/internal/trust" + "provenance/internal/trust/algorithms" +) + +type neighbors struct { + edges []domain.Vouch + calls map[string]int +} + +func TestNeighborDifferential(t *testing.T) { + random := rand.New(rand.NewPCG(1, 2)) + for trial := 0; trial < 100; trial++ { + var edges []domain.Vouch + for author := 0; author < 15; author++ { + for subject := 0; subject < 15; subject++ { + if author == subject || random.IntN(5) != 0 { + continue + } + edges = append(edges, domain.Vouch{Author: fmt.Sprint(author), Subject: fmt.Sprint(subject), Score: float64(random.IntN(6) - 3)}) + } + } + for _, algorithm := range []trust.Algorithm{algorithms.Optimistic{}, algorithms.ShortestPath{}} { + input := trust.Input{Source: "0", Target: "14", Edges: edges, Limits: trust.Limits{Edges: 10000, States: 1000, Penalty: .1}} + if algorithm.Describe().Name == "ShortestPath" { + input.Accounts = trust.Accounts{"ignored": {"7"}} + } + want, err := algorithm.Calculate(context.Background(), input) + if err != nil { + t.Fatal(err) + } + input.Edges, input.Graph = nil, &neighbors{edges: edges, calls: map[string]int{}} + got, err := algorithm.Calculate(context.Background(), input) + if err != nil || !reflect.DeepEqual(want, got) { + t.Fatalf("trial %d %s: want %+v got %+v: %v", trial, algorithm.Describe().Name, want, got, err) + } + } + } +} + +func (g *neighbors) Outgoing(ctx context.Context, author string, limit int) ([]domain.Vouch, error) { + g.calls[author]++ + var result []domain.Vouch + for _, edge := range g.edges { + if edge.Author == author { + result = append(result, edge) + } + } + if len(result) > limit { + return nil, trust.ErrLimit + } + return result, nil +} + +func TestNeighborTraversal(t *testing.T) { + edges := []domain.Vouch{{Author: "alice", Subject: "bob", Score: 1}, {Author: "alice", Subject: "carol", Score: 2}, {Author: "carol", Subject: "bob", Score: 2}, {Author: "unreachable", Subject: "other", Score: 2}} + for _, algorithm := range []trust.Algorithm{algorithms.Optimistic{}, algorithms.ShortestPath{}} { + input := trust.Input{Source: "alice", Target: "bob", Edges: edges, Limits: trust.Limits{Edges: 100, States: 100, Penalty: .1}} + want, err := algorithm.Calculate(context.Background(), input) + if err != nil { + t.Fatal(err) + } + graph := &neighbors{edges: edges, calls: map[string]int{}} + input.Edges, input.Graph = nil, graph + got, err := algorithm.Calculate(context.Background(), input) + if err != nil || got.Score == nil || *got.Score != *want.Score || len(got.Path) != len(want.Path) { + t.Fatalf("%s: %+v %+v %v", algorithm.Describe().Name, got, want, err) + } + if graph.calls["unreachable"] != 0 { + t.Fatal("traversal read an unreachable author") + } + for _, count := range graph.calls { + if count != 1 { + t.Fatal("traversal fetched an author twice") + } + } + } +} diff --git a/internal/trust/algorithms/optimistic.go b/internal/trust/algorithms/optimistic.go index 29fc3bc..b4be00a 100644 --- a/internal/trust/algorithms/optimistic.go +++ b/internal/trust/algorithms/optimistic.go @@ -11,6 +11,8 @@ type Optimistic struct{} func (Optimistic) Describe() trust.Descriptor { return trust.Descriptor{Name: "Optimistic"} } +func (Optimistic) UsesNeighbors() {} + func (Optimistic) Calculate(ctx context.Context, input trust.Input) (trust.Result, error) { - return calculate(ctx, input.Edges, input.Source, input.Target, input.Limits, highestScore) + return calculate(ctx, input, highestScore, nil) } diff --git a/internal/trust/algorithms/shortest_path.go b/internal/trust/algorithms/shortest_path.go index 14c2e55..8dd28e6 100644 --- a/internal/trust/algorithms/shortest_path.go +++ b/internal/trust/algorithms/shortest_path.go @@ -4,7 +4,6 @@ import ( "context" "errors" - "provenance/internal/domain" "provenance/internal/trust" ) @@ -24,6 +23,8 @@ func (ShortestPath) Describe() trust.Descriptor { }}} } +func (ShortestPath) UsesNeighbors() {} + func (ShortestPath) Calculate(ctx context.Context, input trust.Input) (trust.Result, error) { if err := ctx.Err(); err != nil { return trust.Result{}, err @@ -42,19 +43,5 @@ func (ShortestPath) Calculate(ctx context.Context, input trust.Input) (trust.Res } excluded[did] = struct{}{} } - // Copy the graph so that an experiment cannot modify the caller's edges. - edges := make([]domain.Vouch, 0, len(input.Edges)) - for _, edge := range input.Edges { - if err := ctx.Err(); err != nil { - return trust.Result{}, err - } - if _, ok := excluded[edge.Author]; ok { - continue - } - if _, ok := excluded[edge.Subject]; ok { - continue - } - edges = append(edges, edge) - } - return calculate(ctx, edges, input.Source, input.Target, input.Limits, fewestHops) + return calculate(ctx, input, fewestHops, excluded) } diff --git a/internal/trust/algorithms/traversal.go b/internal/trust/algorithms/traversal.go index fdc7439..a323144 100644 --- a/internal/trust/algorithms/traversal.go +++ b/internal/trust/algorithms/traversal.go @@ -29,7 +29,11 @@ const ( fewestHops ) -func calculate(ctx context.Context, edges []domain.Vouch, source, target string, limits trust.Limits, policy pathPolicy) (trust.Result, error) { +func calculate(ctx context.Context, input trust.Input, policy pathPolicy, excluded map[string]struct{}) (trust.Result, error) { + edges, source, target, limits := input.Edges, input.Source, input.Target, input.Limits + if input.Graph != nil && len(edges) != 0 { + return trust.Result{}, errors.New("use either a graph reader or an edge buffer") + } if source == "" || target == "" { return trust.Result{}, errors.New("source and target are required") } @@ -45,17 +49,56 @@ func calculate(ctx context.Context, edges []domain.Vouch, source, target string, adj := make(map[string][]domain.Vouch) for _, v := range edges { + if _, ok := excluded[v.Author]; ok { + continue + } + if _, ok := excluded[v.Subject]; ok { + continue + } if err := domain.ValidateVouch(v); err != nil { return trust.Result{}, err } - if v.Author == source && v.Subject == target && v.Score == domain.Blocked { - return result(trust.BlockedStatus, domain.Blocked, []trust.PathScore{ - {DID: source, Score: domain.SelfScore}, - {DID: target, Assertion: &v.Score, Score: domain.Blocked}, - }), nil - } adj[v.Author] = append(adj[v.Author], v) } + loaded := map[string]bool{} + read := 0 + outgoing := func(author string) ([]domain.Vouch, error) { + if input.Graph == nil || loaded[author] { + return adj[author], nil + } + rows, err := input.Graph.Outgoing(ctx, author, limits.Edges-read+1) + if err != nil { + return nil, err + } + if len(rows) > limits.Edges-read { + return nil, trust.ErrLimit + } + read += len(rows) + for _, v := range rows { + if v.Author != author { + return nil, errors.New("neighbor author mismatch") + } + if _, ok := excluded[v.Subject]; ok { + continue + } + if err := domain.ValidateVouch(v); err != nil { + return nil, err + } + adj[author] = append(adj[author], v) + } + sort.Slice(adj[author], func(i, j int) bool { return adj[author][i].Subject < adj[author][j].Subject }) + loaded[author] = true + return adj[author], nil + } + first, err := outgoing(source) + if err != nil { + return trust.Result{}, err + } + for _, v := range first { + if v.Subject == target && v.Score == domain.Blocked { + return result(trust.BlockedStatus, domain.Blocked, []trust.PathScore{{DID: source, Score: domain.SelfScore}, {DID: target, Assertion: &v.Score, Score: domain.Blocked}}), nil + } + } for author := range adj { sort.Slice(adj[author], func(i, j int) bool { return adj[author][i].Subject < adj[author][j].Subject }) } @@ -64,6 +107,9 @@ func calculate(ctx context.Context, edges []domain.Vouch, source, target string, // Longer paths with the same bottleneck cannot improve the result, including cycles. initial := state{source, domain.MaxScore} queue := []step{{state: initial, parent: -1}} + if err := input.Budget.Use(0, 1); err != nil { + return trust.Result{}, err + } seen := map[state]bool{initial: true} bestIndex := -1 bestRaw := math.Inf(-1) @@ -89,7 +135,14 @@ func calculate(ctx context.Context, edges []domain.Vouch, source, target string, if policy == fewestHops && bestIndex >= 0 && current.depth == queue[bestIndex].depth { continue } - for _, edge := range adj[current.state.node] { + neighbors, err := outgoing(current.state.node) + if err != nil { + return trust.Result{}, err + } + for _, edge := range neighbors { + if err := input.Budget.Use(1, 0); err != nil { + return trust.Result{}, err + } visits++ if visits > limits.Edges { return trust.Result{}, trust.ErrLimit @@ -102,6 +155,9 @@ func calculate(ctx context.Context, edges []domain.Vouch, source, target string, return trust.Result{}, trust.ErrLimit } seen[next] = true + if err := input.Budget.Use(0, 1); err != nil { + return trust.Result{}, err + } queue = append(queue, step{next, head, current.depth + 1, edge.Score}) } } diff --git a/internal/trust/budget.go b/internal/trust/budget.go new file mode 100644 index 0000000..e8d7043 --- /dev/null +++ b/internal/trust/budget.go @@ -0,0 +1,32 @@ +package trust + +import "sync" + +// Budget shares a query's examined-edge and state limits across reputation author checks and retries. +type Budget struct { + mu sync.Mutex + limits Limits + edges, states int +} + +func NewBudget(limits Limits) *Budget { return &Budget{limits: limits} } + +func (b *Budget) RemainingEdges() int { + b.mu.Lock() + defer b.mu.Unlock() + return b.limits.Edges - b.edges +} + +func (b *Budget) Use(edges, states int) error { + if b == nil { + return nil + } + b.mu.Lock() + defer b.mu.Unlock() + if edges < 0 || states < 0 || edges > b.limits.Edges-b.edges || states > b.limits.States-b.states { + return ErrLimit + } + b.edges += edges + b.states += states + return nil +} diff --git a/internal/upstream/queue.go b/internal/upstream/queue.go index 88bb999..bb2b9a6 100644 --- a/internal/upstream/queue.go +++ b/internal/upstream/queue.go @@ -30,6 +30,7 @@ type job struct { req *http.Request base http.RoundTripper result chan result + policy RetryPolicy } type result struct { @@ -68,15 +69,28 @@ var defaultQueue = sync.OnceValue(func() *Queue { q, _ := New(context.Background func Default() *Queue { return defaultQueue() } type queued struct { - queue *Queue - base http.RoundTripper + queue *Queue + base http.RoundTripper + policy RetryPolicy } +type RetryPolicy int + +const ( + QueueRetries RetryPolicy = iota + CallerRetries +) + func (q *Queue) Transport(base http.RoundTripper) http.RoundTripper { + return q.TransportWithPolicy(base, QueueRetries) +} + +// An archive decoder owns range retries. Do not multiply its attempt budget in the queue. +func (q *Queue) TransportWithPolicy(base http.RoundTripper, policy RetryPolicy) http.RoundTripper { if base == nil { base = http.DefaultTransport } - return queued{q, base} + return queued{q, base, policy} } func (t queued) RoundTrip(req *http.Request) (resp *http.Response, err error) { @@ -91,7 +105,7 @@ func (t queued) RoundTrip(req *http.Request) (resp *http.Response, err error) { if err := t.queue.ctx.Err(); err != nil { return nil, err } - work := job{req: req, base: t.base, result: make(chan result)} + work := job{req: req, base: t.base, result: make(chan result), policy: t.policy} select { case <-req.Context().Done(): return nil, req.Context().Err() @@ -132,7 +146,7 @@ func (q *Queue) serve(work job) { cleanup := func() { stop(); cancel() } req := work.req.Clone(ctx) req.Header.Set("User-Agent", UserAgent) - resp, err := q.fetch(req, work.base) + resp, err := q.fetch(req, work.base, work.policy) var body *heldBody if resp != nil { body = &heldBody{ReadCloser: resp.Body, done: make(chan struct{}), cleanup: cleanup} @@ -204,7 +218,7 @@ func origin(req *http.Request) string { return req.URL.Scheme + "://" + net.JoinHostPort(strings.ToLower(req.URL.Hostname()), port) } -func (q *Queue) fetch(req *http.Request, base http.RoundTripper) (*http.Response, error) { +func (q *Queue) fetch(req *http.Request, base http.RoundTripper, policy RetryPolicy) (*http.Response, error) { host := origin(req) safe := (req.Method == http.MethodGet || req.Method == http.MethodHead) && (req.Body == nil || req.Body == http.NoBody) delay := retryBase @@ -239,7 +253,7 @@ func (q *Queue) fetch(req *http.Request, base http.RoundTripper) (*http.Response } } } - if !retry || !safe || attempt >= MaxAttempts { + if !retry || !safe || attempt >= MaxAttempts || policy == CallerRetries { return resp, err } if resp != nil { diff --git a/internal/upstream/queue_test.go b/internal/upstream/queue_test.go index 97aa631..dbd19ab 100644 --- a/internal/upstream/queue_test.go +++ b/internal/upstream/queue_test.go @@ -62,6 +62,35 @@ func TestWorkerBound(t *testing.T) { } } +func TestCallerRetryOwnership(t *testing.T) { + q, err := New(context.Background(), 1) + if err != nil { + t.Fatal(err) + } + defer q.Close() + var calls atomic.Int64 + client := &http.Client{Transport: q.TransportWithPolicy(transport(func(*http.Request) (*http.Response, error) { + calls.Add(1) + resp := response(http.StatusTooManyRequests) + resp.Header.Set("Retry-After", "60") + return resp, nil + }), CallerRetries)} + resp, err := client.Get("https://archive.example.com") + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if calls.Load() != 1 { + t.Fatal("queue multiplied caller retries") + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, "https://archive.example.com", nil) + if _, err := client.Do(req); !errors.Is(err, context.DeadlineExceeded) || calls.Load() != 1 { + t.Fatal("caller retries bypassed origin cooldown") + } +} + func TestRetries(t *testing.T) { for _, method := range []string{http.MethodGet, http.MethodPost, http.MethodPut, http.MethodDelete} { t.Run(method, func(t *testing.T) { diff --git a/nix/modules/appview.nix b/nix/modules/appview.nix index ac8dc9a..92ea026 100644 --- a/nix/modules/appview.nix +++ b/nix/modules/appview.nix @@ -39,6 +39,14 @@ in description = "Jetstream endpoint. Public Microcosm instances and self-hosted instances are supported."; }; + jetstreamReplay = lib.mkEnableOption "historical Jetstream v2 replay with live cutover"; + jetstreamReplayURL = lib.mkOption { + type = lib.types.str; + default = "https://jetstream.us-west.bsky.network"; + description = "Jetstream v2 archive origin. Sequence checkpoints belong to this origin."; + }; + jetstreamAPIKeyFile = common.secretOption "Runtime path to the Jetstream archive API key. Keep the file outside the Nix store."; + slingshotURL = lib.mkOption { type = lib.types.str; default = "https://slingshot.microcosm.blue"; @@ -72,16 +80,25 @@ in assertion = common.validSecret cfg.serviceKeyFile; message = "services.provenance.appview.serviceKeyFile must name an absolute runtime file outside the Nix store."; } + { + assertion = !cfg.jetstreamReplay || common.validSecret cfg.jetstreamAPIKeyFile; + message = "Historical replay requires jetstreamAPIKeyFile outside the Nix store."; + } ]; systemd.services.provenance-appview = common.unit { inherit (cfg) package stateDirectory; binary = "appview"; - credentials.PROVENANCE_SERVICE_KEY = cfg.serviceKeyFile; + credentials = { + PROVENANCE_SERVICE_KEY = cfg.serviceKeyFile; + } + // lib.optionalAttrs cfg.jetstreamReplay { JETSTREAM_API_KEY = cfg.jetstreamAPIKeyFile; }; environment = { PROVENANCE_APPVIEW_ADDR = cfg.listenAddress; PROVENANCE_APPVIEW_DB = "/var/lib/${cfg.stateDirectory}/${cfg.databaseFile}"; PROVENANCE_JETSTREAM_URL = cfg.jetstreamURL; + PROVENANCE_JETSTREAM_REPLAY = lib.boolToString cfg.jetstreamReplay; + PROVENANCE_JETSTREAM_REPLAY_URL = cfg.jetstreamReplayURL; PROVENANCE_SLINGSHOT_URL = cfg.slingshotURL; PROVENANCE_HOP_PENALTY = toString cfg.hopPenalty; PROVENANCE_TRUST_CACHE_TTL = cfg.trustCacheTTL; diff --git a/nix/package.nix b/nix/package.nix index ba1c248..89b8c6d 100644 --- a/nix/package.nix +++ b/nix/package.nix @@ -23,7 +23,7 @@ buildGoModule { ]; }; - vendorHash = "sha256-AREnluE97NFAXDW5Vb8+rhqscWqxsNirtQCH8nbLQ9E="; + vendorHash = "sha256-jnJjurJXFibeKd5t0K2Gl5wTzxTBoTHtRhHi9wBaQJc="; env.CGO_ENABLED = "0"; # Include library tests even when the package contains only one service binary. diff --git a/nix/services/provenance.nix b/nix/services/provenance.nix index 755867d..677abc0 100644 --- a/nix/services/provenance.nix +++ b/nix/services/provenance.nix @@ -72,6 +72,9 @@ let trust_cache_ttl=${lib.escapeShellArg cfg.trustCacheTTL} export PROVENANCE_TRUST_CACHE_TTL="''${PROVENANCE_TRUST_CACHE_TTL:-$trust_cache_ttl}" export PROVENANCE_EXTERNAL_WORKERS="''${PROVENANCE_EXTERNAL_WORKERS:-${toString cfg.externalWorkers}}" + export PROVENANCE_JETSTREAM_REPLAY="''${PROVENANCE_JETSTREAM_REPLAY:-${lib.boolToString cfg.jetstreamReplay}}" + replay_url=${lib.escapeShellArg cfg.jetstreamReplayURL} + export PROVENANCE_JETSTREAM_REPLAY_URL="''${PROVENANCE_JETSTREAM_REPLAY_URL:-$replay_url}" export ENABLE_SOFT_VOUCH="''${ENABLE_SOFT_VOUCH:-${lib.boolToString cfg.enableSoftVouch}}" export ENABLE_SOFT_VOUCH_LISTS="''${ENABLE_SOFT_VOUCH_LISTS:-${lib.boolToString cfg.softVouchLists}}" export ENABLE_SOFT_VOUCH_ALL_AUTHORS="''${ENABLE_SOFT_VOUCH_ALL_AUTHORS:-${lib.boolToString cfg.softVouchAllAuthors}}" @@ -107,6 +110,12 @@ in description = "Trust cache lifetime as a positive Go duration."; }; externalWorkers = (import ../modules/common.nix { inherit lib; }).externalWorkersOption; + jetstreamReplay = lib.mkEnableOption "historical Jetstream v2 replay with live cutover"; + jetstreamReplayURL = lib.mkOption { + type = lib.types.str; + default = "https://jetstream.us-west.bsky.network"; + description = "Jetstream v2 archive origin. Supply JETSTREAM_API_KEY through the runtime environment."; + }; enableSoftVouch = lib.mkEnableOption "Bluesky follows and blocks as virtual assertions"; softVouchLists = lib.mkEnableOption "blocked-list soft assertions when enableSoftVouch is active"; softVouchAllAuthors = lib.mkEnableOption "soft assertions from all observed authors"; diff --git a/sqlc.yaml b/sqlc.yaml index d6e86b2..638d0a9 100644 --- a/sqlc.yaml +++ b/sqlc.yaml @@ -5,6 +5,7 @@ sql: - internal/storage/schema.sql - internal/storage/soft_schema.sql - internal/storage/lists_schema.sql + - internal/storage/replay_schema.sql queries: internal/storage/queries.sql gen: go: -- 2.54.0