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: