From b8fa7be29b46ebf28f1c65db8fdbd87b36c9cb5d Mon Sep 17 00:00:00 2001 From: Noah Pederson Date: Fri, 9 Oct 2026 22:08:11 -0500 Subject: [PATCH] Fix replay redirects and update dependencies Follow archive redirects without forwarding credentials to another origin. The previous client refused provider redirects and stopped before the first batch. Exclude live probes from default tests with an integration build tag. Update Prometheus to its latest stable release and Indigo to its latest available commit. Refresh the Nix vendor hash. --- docs/jetstream-replay.md | 7 +- go.mod | 22 ++--- go.sum | 60 +++++++------- internal/appview/indexing_test.go | 53 ------------ internal/appview/replay_live_test.go | 64 +++++++++++++++ internal/microcosm/replay.go | 1 + internal/microcosm/replay_live_test.go | 62 ++++++++++++++ internal/microcosm/replay_status.go | 18 +++++ internal/microcosm/replay_test.go | 107 ++++++++++++++----------- nix/package.nix | 2 +- 10 files changed, 252 insertions(+), 144 deletions(-) create mode 100644 internal/appview/replay_live_test.go create mode 100644 internal/microcosm/replay_live_test.go diff --git a/docs/jetstream-replay.md b/docs/jetstream-replay.md index d4850ba..fdef5ba 100644 --- a/docs/jetstream-replay.md +++ b/docs/jetstream-replay.md @@ -51,6 +51,8 @@ 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. +Archive downloads follow redirects through the same queue. Redirects to another origin do not receive the archive API key. +Replay rejects encrypted-to-unencrypted redirects and embedded credentials. Loopback development origins can redirect to loopback HTTP endpoints. `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. @@ -127,7 +129,7 @@ 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 +PROVENANCE_REPLAY_SMOKE=true go test -tags=integration ./internal/microcosm -run '^TestReplayLivePlan$' -v ``` The probe requires `JETSTREAM_API_KEY` in the process environment and permits one sequence in its archive plan. @@ -135,10 +137,11 @@ The probe requires `JETSTREAM_API_KEY` in the process environment and permits on For a subset trial with a separate temporary database, run: ```sh -PROVENANCE_REPLAY_SUBSET_TEST=true go test ./internal/appview -run '^TestLiveReplaySubset$' -v +PROVENANCE_REPLAY_SUBSET_TEST=true go test -tags=integration ./internal/appview -run '^TestLiveReplaySubset$' -v ``` The trial requires the runtime key and imports sequences `(0,1000]` with one external worker. +Default tests exclude both live probes. Each probe requires the `integration` build tag and its explicit environment flag. It also verifies the durable checkpoint after a database restart. This small sequence range does not bound compressed segment download size. diff --git a/go.mod b/go.mod index 9185974..46dd3f5 100644 --- a/go.mod +++ b/go.mod @@ -3,11 +3,11 @@ module provenance go 1.26.6 require ( - github.com/bluesky-social/indigo v0.0.0-20261006083258-ae9297b7d0f4 + github.com/bluesky-social/indigo v0.0.0-20261008210153-5568a3799fb6 github.com/bluesky-social/jetstream v0.3.10 github.com/coder/websocket v1.8.15 github.com/jcalabro/atmos v0.7.3 - github.com/prometheus/client_golang v1.24.1 + github.com/prometheus/client_golang v1.25.0 modernc.org/sqlite v1.60.1 ) @@ -24,12 +24,12 @@ require ( 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/go-querystring v1.2.0 // indirect github.com/google/uuid v1.6.0 // indirect github.com/hashicorp/golang-lru/v2 v2.0.7 // 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/compress v1.20.1 // 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 @@ -38,20 +38,20 @@ require ( github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect github.com/pkg/errors v0.9.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/prometheus/client_model v0.6.3 // indirect + github.com/prometheus/common v0.72.0 // indirect + github.com/prometheus/procfs v0.22.0 // 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/crypto v0.57.0 // indirect + golang.org/x/net v0.59.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 + golang.org/x/text v0.42.0 // indirect + golang.org/x/time v0.16.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 diff --git a/go.sum b/go.sum index 3915c07..4cdd131 100644 --- a/go.sum +++ b/go.sum @@ -2,8 +2,8 @@ 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/indigo v0.0.0-20261008210153-5568a3799fb6 h1:YTwEAQzRH2XnkhtlphEDPyjvztFt1ta1IskQaW3SYMU= +github.com/bluesky-social/indigo v0.0.0-20261008210153-5568a3799fb6/go.mod h1:QKojlbP6uRiB4jDlHGnl1eoCLVMQw5luR8HXTut4xzE= 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= @@ -23,8 +23,6 @@ github.com/cockroachdb/redact v1.1.5/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZ 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= @@ -43,11 +41,11 @@ 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.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= -github.com/google/go-querystring v1.1.0 h1:AnCroh3fv4ZBgVIf1Iwtovgjaw/GiKJo8M8yD/fhyJ8= -github.com/google/go-querystring v1.1.0/go.mod h1:Kcdr2DB4koayq7X8pmAG4sNG59So17icRSOU623lUBU= +github.com/google/go-querystring v1.2.0 h1:yhqkPbu2/OH+V9BfpCVPZkNmUXhb2gBxJArfhIxNtP0= +github.com/google/go-querystring v1.2.0/go.mod h1:8IFJqpSRITyJ8QhQ13bmbeMBDfmeEJZD5A0egEOmkqU= github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3 h1:LMLX+LgTNWpfvCBdFebv6EsYotImrt/Ppc5cXIriCSo= 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= @@ -66,8 +64,8 @@ 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/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ= +github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= 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= @@ -103,14 +101,14 @@ 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.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/prometheus/client_golang v1.25.0 h1:cDtW1TcFsMi1RJc8khwCKB4VhzwV04aKqbYtNOXuiK0= +github.com/prometheus/client_golang v1.25.0/go.mod h1:j2X/4/kaQHLPnmyEU8zjmkndo1T317VpCIQEml8LTKk= +github.com/prometheus/client_model v0.6.3 h1:O0jaTVAYNxTHYInEPFJt5I3+sN8zqBtVMPTB1qyxiEo= +github.com/prometheus/client_model v0.6.3/go.mod h1:gpN5P9S7Rr6Yr92PiQ+Ixvhf6JZEkF1dnxsYL2aPBEM= +github.com/prometheus/common v0.72.0 h1:tAYsE+sPJxIncDAobm4H5aQjmox9ZxEIIqPbiffa8G4= +github.com/prometheus/common v0.72.0/go.mod h1:77NWqAQ2tXT7BIK40qjJdw5Acrsrg1TlHAnsQi3i6mk= +github.com/prometheus/procfs v0.22.0 h1:6q9+/JL9IKAPbCmBrv9n5O5Ty3NKnciV5X7YGw0oics= +github.com/prometheus/procfs v0.22.0/go.mod h1:CvmFr/GVhIjIvWJZW3tgkODBQMRIf0EyWMQLHCHab58= 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= @@ -120,8 +118,8 @@ github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0t 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/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= 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= @@ -156,11 +154,13 @@ 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= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= 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/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= 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= @@ -169,8 +169,8 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn 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/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues= +golang.org/x/net v0.59.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= 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= @@ -183,16 +183,16 @@ 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/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= +golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= +golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= 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/tools v0.51.0 h1:k4Xc/1Om9jwkBJBo4NVLMSARBoWtK10mx+W5BnXCeAI= +golang.org/x/tools v0.51.0/go.mod h1:9eEncMayCV6zRMGhR5eZEC2iBx98qWcF1HZ9Z7wJOoA= 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= @@ -207,8 +207,6 @@ 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= -gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= lukechampine.com/blake3 v1.4.1 h1:I3Smz7gso8w4/TunLKec6K2fn+kyKtDxr/xcQEN84Wg= lukechampine.com/blake3 v1.4.1/go.mod h1:QFosUxmjB8mnrWFSNwKmvxHpfY72bmD2tQ0kBMM3kwo= modernc.org/cc/v4 v4.29.7 h1:q+NXGJ0bK3b4TXFYQQVr9pYETGnmwFWkrUzJnMya/Tg= diff --git a/internal/appview/indexing_test.go b/internal/appview/indexing_test.go index ecab8d4..eacda17 100644 --- a/internal/appview/indexing_test.go +++ b/internal/appview/indexing_test.go @@ -5,19 +5,13 @@ import ( "errors" "net/http" "net/http/httptest" - "os" - "path/filepath" "strings" "sync" "testing" "time" "provenance/internal/domain" - "provenance/internal/microcosm" - "provenance/internal/storage" - "provenance/internal/trust" "provenance/internal/trust/algorithms" - "provenance/internal/upstream" ) type failedReplay struct { @@ -139,50 +133,3 @@ func TestIndexReadiness(t *testing.T) { t.Fatal("subset reported production readiness") } } - -// The optional live trial uses a separate database and never connects the live stream. -func TestLiveReplaySubset(t *testing.T) { - if os.Getenv("PROVENANCE_REPLAY_SUBSET_TEST") != "true" { - t.Skip("live subset trial 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(), 2*time.Minute) - defer cancel() - queue, err := upstream.New(ctx, 1) - if err != nil { - t.Fatal(err) - } - defer queue.Close() - bounds := domain.SequenceRange{Before: 1000} - replay, err := microcosm.NewReplayWithRange(microcosm.DefaultReplayURL, key, domain.SoftAllAuthors, queue, 1, bounds) - if err != nil { - t.Fatal(err) - } - path := filepath.Join(t.TempDir(), "subset.db") - store, err := storage.OpenWithSoft(path, domain.SoftAllAuthors) - if err != nil { - t.Fatal(err) - } - defer func() { store.Close() }() - limits := trust.Limits{Edges: trust.DefaultEdgeLimit, States: trust.DefaultStateLimit, Penalty: domain.DefaultPenalty} - s := NewWithOptions(store, nil, limits, Options{SoftMode: domain.SoftAllAuthors, Indexing: domain.IndexStarting}) - s.RunReplay(ctx, replay) - status := s.index.snapshot(s.now()) - if status.Phase != domain.IndexComplete || status.Ready || status.Checkpoint != bounds.Before { - t.Fatalf("subset did not complete: phase %s checkpoint %d failure %s", status.Phase, status.Checkpoint, status.Failure) - } - t.Logf("subset (%d,%d], checkpoint %d, archive tip %d, last witnessed event %s", bounds.After, bounds.Before, status.Checkpoint, status.ArchiveTip, status.LastEvent) - store.Close() - store, err = storage.OpenWithSoft(path, domain.SoftAllAuthors) - if err != nil { - t.Fatal(err) - } - s = NewWithOptions(store, nil, limits, Options{SoftMode: domain.SoftAllAuthors, Indexing: domain.IndexStarting}) - s.RunReplay(ctx, replay) - if status := s.index.snapshot(s.now()); status.Phase != domain.IndexComplete || status.Checkpoint != bounds.Before { - t.Fatal("subset restart lost its completion checkpoint") - } -} diff --git a/internal/appview/replay_live_test.go b/internal/appview/replay_live_test.go new file mode 100644 index 0000000..fe23e3b --- /dev/null +++ b/internal/appview/replay_live_test.go @@ -0,0 +1,64 @@ +//go:build integration + +package appview + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "provenance/internal/domain" + "provenance/internal/microcosm" + "provenance/internal/storage" + "provenance/internal/trust" + "provenance/internal/upstream" +) + +// The optional live trial uses a separate database and never connects the live stream. +func TestLiveReplaySubset(t *testing.T) { + if os.Getenv("PROVENANCE_REPLAY_SUBSET_TEST") != "true" { + t.Skip("live subset trial 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(), 2*time.Minute) + defer cancel() + queue, err := upstream.New(ctx, 1) + if err != nil { + t.Fatal(err) + } + defer queue.Close() + bounds := domain.SequenceRange{Before: 1000} + replay, err := microcosm.NewReplayWithRange(microcosm.DefaultReplayURL, key, domain.SoftAllAuthors, queue, 1, bounds) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "subset.db") + store, err := storage.OpenWithSoft(path, domain.SoftAllAuthors) + if err != nil { + t.Fatal(err) + } + defer func() { store.Close() }() + limits := trust.Limits{Edges: trust.DefaultEdgeLimit, States: trust.DefaultStateLimit, Penalty: domain.DefaultPenalty} + s := NewWithOptions(store, nil, limits, Options{SoftMode: domain.SoftAllAuthors, Indexing: domain.IndexStarting}) + s.RunReplay(ctx, replay) + status := s.index.snapshot(s.now()) + if status.Phase != domain.IndexComplete || status.Ready || status.Checkpoint != bounds.Before { + t.Fatalf("subset did not complete: phase %s checkpoint %d failure %s", status.Phase, status.Checkpoint, status.Failure) + } + t.Logf("subset (%d,%d], checkpoint %d, archive tip %d, last witnessed event %s", bounds.After, bounds.Before, status.Checkpoint, status.ArchiveTip, status.LastEvent) + store.Close() + store, err = storage.OpenWithSoft(path, domain.SoftAllAuthors) + if err != nil { + t.Fatal(err) + } + s = NewWithOptions(store, nil, limits, Options{SoftMode: domain.SoftAllAuthors, Indexing: domain.IndexStarting}) + s.RunReplay(ctx, replay) + if status := s.index.snapshot(s.now()); status.Phase != domain.IndexComplete || status.Checkpoint != bounds.Before { + t.Fatal("subset restart lost its completion checkpoint") + } +} diff --git a/internal/microcosm/replay.go b/internal/microcosm/replay.go index 25c0120..bb430c6 100644 --- a/internal/microcosm/replay.go +++ b/internal/microcosm/replay.go @@ -57,6 +57,7 @@ func NewReplayWithRange(host, key string, mode domain.SoftMode, queue *upstream. } // Bulk reads use context cancellation, not a short whole-response timeout. c.http.Timeout = 0 + c.http.CheckRedirect = replayRedirect transport := http.DefaultTransport.(*http.Transport).Clone() transport.ResponseHeaderTimeout = atproto.RequestTimeout c.http.Transport = queue.TransportWithPolicy(transport, upstream.CallerRetries) diff --git a/internal/microcosm/replay_live_test.go b/internal/microcosm/replay_live_test.go new file mode 100644 index 0000000..723d55f --- /dev/null +++ b/internal/microcosm/replay_live_test.go @@ -0,0 +1,62 @@ +//go:build integration + +package microcosm + +import ( + "context" + "encoding/json" + "io" + "net/http" + "os" + "strings" + "testing" + "time" + + "provenance/internal/domain" + "provenance/internal/upstream" +) + +// 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/microcosm/replay_status.go b/internal/microcosm/replay_status.go index 0ad9874..4572eb5 100644 --- a/internal/microcosm/replay_status.go +++ b/internal/microcosm/replay_status.go @@ -14,6 +14,24 @@ import ( "provenance/internal/domain" ) +const replayRedirectLimit = 10 + +// Follow archive download redirects without forwarding the key to another origin. +func replayRedirect(req *http.Request, via []*http.Request) error { + if len(via) == 0 || len(via) >= replayRedirectLimit { + return errors.New("archive redirect limit reached") + } + origin := via[0].URL + development := origin.Scheme == "http" && loopback(origin.Hostname()) + if req.URL.User != nil || req.URL.Scheme != "https" && !(development && req.URL.Scheme == "http" && loopback(req.URL.Hostname())) { + return errors.New("archive redirect requires encrypted transport") + } + if req.URL.Scheme != origin.Scheme || req.URL.Host != origin.Host { + req.Header.Del("Authorization") + } + return nil +} + // Observe protocol boundaries without replacing the SDK's planner or archive decoder. type replayTransport struct { base http.RoundTripper diff --git a/internal/microcosm/replay_test.go b/internal/microcosm/replay_test.go index 3d98de6..e365d33 100644 --- a/internal/microcosm/replay_test.go +++ b/internal/microcosm/replay_test.go @@ -9,7 +9,6 @@ import ( "net/http/httptest" "os" "path/filepath" - "strings" "sync/atomic" "testing" "time" @@ -127,6 +126,67 @@ func TestReplayAuthentication(t *testing.T) { } } +func TestReplayArchiveRedirect(t *testing.T) { + archive := archiveFixture(t) + 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() + var downloads atomic.Int64 + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + downloads.Add(1) + if r.Header.Get("Authorization") != "" { + t.Error("redirect leaked archive credentials") + } + if r.Header.Get("User-Agent") != upstream.UserAgent { + t.Error("redirect lacks project User-Agent") + } + w.Write(archive) + })) + defer backend.Close() + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer test-key" { + t.Error("origin lacks archive credentials") + } + if r.URL.Path == "/xrpc/network.bsky.jetstream.planSnapshot" { + w.Write([]byte(`{"sealedTipSeq":1,"plannedThroughSeq":1,"segments":[{"name":"seg_0000000000.jss","index":0,"minSeq":1,"maxSeq":1,"mode":"segment"}],"stats":{}}`)) + return + } + http.Redirect(w, r, backend.URL+"/archive", http.StatusTemporaryRedirect) + })) + defer origin.Close() + replay, err := NewReplayWithRange(origin.URL, "test-key", domain.SoftDisabled, queue, 1, domain.SequenceRange{Before: 1}) + if err != nil { + t.Fatal(err) + } + var events int + err = replay.ConsumeReplay(ctx, 0, nil, func(_ context.Context, batch []domain.Change, _ int64) error { events += len(batch); return nil }, nil) + if err != nil || events != 1 || downloads.Load() != 1 { + t.Fatalf("redirect replay: events %d downloads %d error %v", events, downloads.Load(), err) + } +} + +func TestReplayRedirectPolicy(t *testing.T) { + origin := httptest.NewRequest(http.MethodGet, "https://archive.example/download", nil) + for _, target := range []string{"http://archive.example/download", "http://127.0.0.1/download", "https://user:secret@archive.example/download"} { + req := httptest.NewRequest(http.MethodGet, target, nil) + if err := replayRedirect(req, []*http.Request{origin}); err == nil { + t.Errorf("unsafe redirect accepted: %s", target) + } + } + req := httptest.NewRequest(http.MethodGet, "https://archive.example/download", nil) + via := make([]*http.Request, replayRedirectLimit) + for index := range via { + via[index] = origin + } + if err := replayRedirect(req, via); err == nil { + t.Fatal("redirect limit ignored") + } +} + func TestReplayFailureKinds(t *testing.T) { for _, test := range []struct { err error @@ -217,48 +277,3 @@ func TestReplayArchiveCutover(t *testing.T) { 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/nix/package.nix b/nix/package.nix index 60107d0..78902f5 100644 --- a/nix/package.nix +++ b/nix/package.nix @@ -23,7 +23,7 @@ buildGoModule { ]; }; - vendorHash = "sha256-PmbA9/0HDOEqSJHNnTfs23BirGRGQ/BUfLz5YaDmi/0="; + vendorHash = "sha256-jESp9LV36vpZJ7EVjo2CYS3lcKWdhMOAiL8ubzG2SJg="; env.CGO_ENABLED = "0"; # Include library tests even when the package contains only one service binary. -- 2.51.2