From 8c7b72ec052e7e4e945e63a634ca0a389ebe94e1 Mon Sep 17 00:00:00 2001 From: Evelyn Osman Date: Thu, 12 Nov 2026 12:00:18 +0100 Subject: [PATCH 1/5] Push the validated release tag from the Release workflow The Changesets v2 migration set push-git-tags to false, and scripts/release.mjs only tags the runner's checkout, so the v0.10.0 release run created and validated its tag without publishing it. Push the tag after validation, refusing a tag that does not resolve to the checked-out release commit. A plain push never moves an existing tag. This work was assisted by a machine --- .github/workflows/release.yml | 12 ++++++++++++ RELEASING.md | 2 +- 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 60f47daa..e2cf659b 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -83,6 +83,18 @@ jobs: (github.event_name == 'workflow_dispatch' && inputs.tag_current_version) || steps.changesets.outputs.published == 'true' run: npm run check-component-versions -- --require-tag + # The Changesets step and the current-version path tag only the runner's + # checkout. Publish the validated tag, refusing one that does not resolve to + # the checked-out release commit; a plain push never moves an existing tag. + - name: Push validated release tag + if: >- + (github.event_name == 'workflow_dispatch' && inputs.tag_current_version) || + steps.changesets.outputs.published == 'true' + run: | + set -euo pipefail + tag="v$(node -p "require('./package.json').version")" + test "$(git rev-list -n 1 "refs/tags/$tag")" = "$(git rev-parse HEAD)" + git push origin "refs/tags/$tag" - name: Rename release pull request with version if: steps.changesets.outputs.pr-number != '' env: diff --git a/RELEASING.md b/RELEASING.md index c4f5844d..79b3a656 100644 --- a/RELEASING.md +++ b/RELEASING.md @@ -17,7 +17,7 @@ The `Release` workflow uses the standard Hypercerts release bot credentials, `RE 1. Open **Actions → Release → Run workflow** on `main`. 2. The workflow verifies Relay and Rainbow, then opens or updates a `prepare for release vX.Y.Z` pull request. That pull request updates the root package, Administration package, Relay/Rainbow and Jetstream build versions, Docker build defaults, and `CHANGELOG.md` together. 3. Review and merge the version pull request normally. -4. Its merge triggers the workflow again. Before Changesets creates `vX.Y.Z`, the workflow checks that the source tag name, root package, Administration package, Relay/Rainbow runtime metadata, Jetstream runtime metadata, and all four Docker build defaults use the same version. It then verifies that the tag resolves to the merged release commit. +4. Its merge triggers the workflow again. Before Changesets creates `vX.Y.Z`, the workflow checks that the source tag name, root package, Administration package, Relay/Rainbow runtime metadata, Jetstream runtime metadata, and all four Docker build defaults use the same version. It then verifies that the tag resolves to the merged release commit and pushes that tag to GitHub. 5. Use the `hypercerts-github-release` skill to create the GitHub Release from that existing validated tag with GitHub CLI generated notes. Review the generated notes for accurate operator impact before creating it. For the one-time initial tag of an already-versioned tree, a maintainer can dispatch the workflow from `main` with **Create the missing tag for the checked-out package version** enabled. It validates and creates only the current `vX.Y.Z` tag; it does not bypass review for later Changesets releases or create a GitHub Release. From ff3e3a9394dcaf825981a617ac33f90708019dbc Mon Sep 17 00:00:00 2001 From: Evelyn Osman Date: Thu, 12 Nov 2026 12:35:51 +0100 Subject: [PATCH 2/5] fix(jetstream): keep live ingest running across relay restarts A single HTTP 502/503 from the relay during reconnect made atmos end the upstream iterator, and the live consumer returned nil, leaving Jetstream running without ingestion for hours. On restart the stale cursor replayed a large backlog that overflowed the per-DID verify queue (TECH-883). - Retry transient dial rejections (408, 425, 429, 5xx) with backoff and log each successful upstream connection. - Return ErrUpstreamStreamEnded when the iterator ends while the consumer is still running, so the process exits instead of stalling silently. - Raise verify parallelism to 128 (per-DID queue 256) and pace upstream reads at 500 frames/s by default; both are configurable. - Summarise repeated verification failure log lines per DID and class. This work was assisted by a machine --- .../jetstream-live-ingest-resilience.md | 22 ++ jetstream/README.md | 34 ++ jetstream/cmd/jetstream/main.go | 24 ++ jetstream/internal/ingest/live/config.go | 30 ++ jetstream/internal/ingest/live/consumer.go | 35 +- jetstream/internal/ingest/live/errors.go | 6 + jetstream/internal/ingest/live/metrics.go | 29 +- .../internal/ingest/live/upstream_dial.go | 144 +++++++++ .../ingest/live/upstream_dial_test.go | 302 ++++++++++++++++++ .../internal/ingest/orchestrator/bootstrap.go | 3 + .../internal/ingest/orchestrator/config.go | 7 + .../internal/ingest/orchestrator/steady.go | 3 + jetstream/internal/jetstreamd/options.go | 11 + jetstream/internal/jetstreamd/runtime.go | 23 +- .../internal/jetstreamd/verify_failure_log.go | 146 +++++++++ .../jetstreamd/verify_failure_log_test.go | 123 +++++++ 16 files changed, 931 insertions(+), 11 deletions(-) create mode 100644 .changeset/jetstream-live-ingest-resilience.md create mode 100644 jetstream/internal/ingest/live/upstream_dial.go create mode 100644 jetstream/internal/ingest/live/upstream_dial_test.go create mode 100644 jetstream/internal/jetstreamd/verify_failure_log.go create mode 100644 jetstream/internal/jetstreamd/verify_failure_log_test.go diff --git a/.changeset/jetstream-live-ingest-resilience.md b/.changeset/jetstream-live-ingest-resilience.md new file mode 100644 index 00000000..908f7140 --- /dev/null +++ b/.changeset/jetstream-live-ingest-resilience.md @@ -0,0 +1,22 @@ +--- +'hypercerts-relay': patch +--- + +Jetstream live ingestion no longer stops silently when the Relay restarts. +Previously, a single HTTP 502 or 503 from the Relay or its proxy during a +reconnect ended the upstream stream. Jetstream then kept running without +ingesting until its next restart. Transient rejections (408, 425, 429 and 5xx) +are now retried with backoff from the same cursor. Any other end of the stream +exits the process with an `upstream stream ended unexpectedly` error so it is +restarted. Successful connections are logged as `upstream connected`. + +Catching up after a restart no longer overflows the per-account verify queue as +easily. Verification now uses 128 workers by default, up from 32, so each +account's queue holds 256 events instead of 64. Upstream reads are paced at 500 +frames per second by default. Tune these with `JETSTREAM_LIVE_VERIFY_PARALLELISM`, +`JETSTREAM_LIVE_UPSTREAM_READ_RATE` and `JETSTREAM_LIVE_UPSTREAM_READ_BURST`. +Pacing limits how fast Jetstream replays a large backlog. + +Repeated `verification failure` warnings for the same account are now logged +once per minute, followed by a `verification failures repeated` summary with +the count and the last error. The verifier failure metric is unchanged. diff --git a/jetstream/README.md b/jetstream/README.md index 3da7023e..92ea4fec 100644 --- a/jetstream/README.md +++ b/jetstream/README.md @@ -73,6 +73,40 @@ safe to combine it with Relay's database or event-store backup. `Dockerfile` builds this module with `docker build -f jetstream/Dockerfile jetstream`. It does not publish a service or make a deployment decision. +### Live ingestion reconnects and pacing + +The live consumer retries an upstream dial that the Relay or its proxy rejects +with HTTP 408, 425, 429 or any 5xx. It backs off and resumes from the same +cursor. These rejections are expected while the Relay restarts. Each retry logs +`upstream dial rejected; retrying`, and each successful connection logs +`upstream connected` with the resume cursor. Other HTTP rejections, such as a +wrong URL, end the stream. The process then exits with an `upstream stream ended +unexpectedly` error so its supervisor restarts it, instead of running without +ingestion. Watch `jetstream_livestream_upstream_connects_total` and +`jetstream_livestream_upstream_transient_dial_rejections_total` alongside the +reconnect counter. + +Upstream reads are paced, and verification is sized, so catching up after a +restart does not overflow the per-account verify queue: + +| Flag | Environment | Default | Meaning | +| --- | --- | --- | --- | +| `--live-verify-parallelism` | `JETSTREAM_LIVE_VERIFY_PARALLELISM` | 128 | Verification workers. Each account's verify queue holds twice this many events. `1` gives strict ordering with an unbounded queue and full backpressure. | +| `--live-upstream-read-rate` | `JETSTREAM_LIVE_UPSTREAM_READ_RATE` | 500 | Maximum upstream frames read per second. `0` disables pacing. | +| `--live-upstream-read-burst` | `JETSTREAM_LIVE_UPSTREAM_READ_BURST` | 0 | Frames read without waiting. `0` uses the verify parallelism. | + +Pacing bounds catch-up speed. At 500 frames per second, a backlog of one million +Relay events takes about 35 minutes to replay. Raise the rate if the Relay's +sustained event rate approaches it; `jetstream_livestream_verify_queue_dropped_events_total` +shows whether the queue still overflows. A dropped event still books repository +recovery, as described below. + +Verification failures are logged once per account and failure class each minute. +Later failures in that minute are counted and reported in one `verification +failures repeated` line with the count and the last error. A resyncing account +reports a failure for every commit it buffers. The +`jetstream_verifier_failures_total` metric still counts every failure. + ## Hypercerts policy work `serve` opens a durable global exact-NSID policy before ingestion. On a new data diff --git a/jetstream/cmd/jetstream/main.go b/jetstream/cmd/jetstream/main.go index 39ee979f..0bac5283 100644 --- a/jetstream/cmd/jetstream/main.go +++ b/jetstream/cmd/jetstream/main.go @@ -390,6 +390,27 @@ func serveCommand() *cli.Command { Sources: cli.EnvVars("JETSTREAM_SUBSCRIBE_SLOW_MIN_RATE"), Value: 5, }, + // hypercerts: Size the upstream verify queue and pace catch-up reads + // so a restart backlog cannot overflow it (TECH-883). + &cli.IntFlag{ + Name: "live-verify-parallelism", + Usage: "Upstream verification workers. Each DID's verify queue holds twice this many events; " + + "events beyond it are dropped and repaired by repository recovery. 1 gives strict ordering with full backpressure.", + Sources: cli.EnvVars("JETSTREAM_LIVE_VERIFY_PARALLELISM"), + Value: jetstreamd.DefaultLiveVerifyParallelism, + }, + &cli.FloatFlag{ + Name: "live-upstream-read-rate", + Usage: "Maximum upstream subscribeRepos frames read per second, so catch-up cannot outrun verification. 0 disables pacing.", + Sources: cli.EnvVars("JETSTREAM_LIVE_UPSTREAM_READ_RATE"), + Value: jetstreamd.DefaultLiveUpstreamReadRate, + }, + &cli.IntFlag{ + Name: "live-upstream-read-burst", + Usage: "Upstream frames that may be read without waiting for pacing. 0 uses --live-verify-parallelism.", + Sources: cli.EnvVars("JETSTREAM_LIVE_UPSTREAM_READ_BURST"), + Value: 0, + }, &cli.IntFlag{ Name: "cursor-block-index-cache-size", Usage: "Deprecated compatibility no-op: sealed segment metadata is always resident in the manifest.", @@ -531,6 +552,9 @@ func serveOptionsFromCommand(cmd *cli.Command) (jetstreamd.Options, error) { SubscribeReadBatch: cmd.Int("subscribe-read-batch"), SubscribeSlowWindow: cmd.Duration("subscribe-slow-window"), SubscribeSlowMinRate: cmd.Float("subscribe-slow-min-rate"), + LiveVerifyParallelism: cmd.Int("live-verify-parallelism"), + LiveUpstreamReadRate: cmd.Float("live-upstream-read-rate"), + LiveUpstreamReadBurst: cmd.Int("live-upstream-read-burst"), CursorBlockIndexCacheSize: cmd.Int("cursor-block-index-cache-size"), CompactionInterval: cmd.Duration("compaction-interval"), CompactionTombstoneCap: cmd.Int("compaction-tombstone-cap"), diff --git a/jetstream/internal/ingest/live/config.go b/jetstream/internal/ingest/live/config.go index 5be0eac3..e6a65a3f 100644 --- a/jetstream/internal/ingest/live/config.go +++ b/jetstream/internal/ingest/live/config.go @@ -7,6 +7,7 @@ import ( "github.com/bluesky-social/jetstream/internal/hypercerts/policygate" "github.com/bluesky-social/jetstream/internal/hypercerts/selection" "log/slog" + "math" "time" "github.com/bluesky-social/jetstream/internal/ingest" @@ -208,6 +209,20 @@ type Config struct { // Intended for deterministic harnesses that feed the firehose over an // in-memory connection. Dial streaming.DialFunc + + // hypercerts: VerifyParallelism is the number of atmos verification + // workers. atmos caps each DID's verify queue at twice this value and + // drops events beyond it. Zero selects DefaultVerifyParallelism; 1 gives + // strict ordering with an unbounded queue and full backpressure. + VerifyParallelism int + + // hypercerts: UpstreamReadRate paces upstream frame reads, in frames per + // second, so catch-up cannot outrun verification. Zero disables pacing. + UpstreamReadRate float64 + + // hypercerts: UpstreamReadBurst is the number of frames that may be read + // without waiting. Zero selects VerifyParallelism (half the per-DID queue). + UpstreamReadBurst int } func (c *Config) validate() error { @@ -235,6 +250,15 @@ func (c *Config) validate() error { if c.Verifier == nil { return fmt.Errorf("%w: Verifier is required", ErrInvalidConfig) } + if c.VerifyParallelism < 0 { + return fmt.Errorf("%w: VerifyParallelism must not be negative", ErrInvalidConfig) + } + if c.UpstreamReadRate < 0 || math.IsNaN(c.UpstreamReadRate) || math.IsInf(c.UpstreamReadRate, 0) { + return fmt.Errorf("%w: UpstreamReadRate must be a finite non-negative number", ErrInvalidConfig) + } + if c.UpstreamReadBurst < 0 { + return fmt.Errorf("%w: UpstreamReadBurst must not be negative", ErrInvalidConfig) + } return nil } @@ -242,4 +266,10 @@ func (c *Config) applyDefaults() { if c.now == nil { c.now = time.Now } + if c.VerifyParallelism == 0 { + c.VerifyParallelism = DefaultVerifyParallelism + } + if c.UpstreamReadBurst == 0 { + c.UpstreamReadBurst = c.VerifyParallelism + } } diff --git a/jetstream/internal/ingest/live/consumer.go b/jetstream/internal/ingest/live/consumer.go index fc4c36f1..e3253390 100644 --- a/jetstream/internal/ingest/live/consumer.go +++ b/jetstream/internal/ingest/live/consumer.go @@ -74,6 +74,11 @@ type Consumer struct { // processBatch in isolation. client atomic.Pointer[streaming.Client] + // lastStreamErr is the most recent stream-level error yielded by the + // upstream iterator. Owned by the Run goroutine; reported when the + // iterator ends while the consumer is still meant to run. + lastStreamErr error + closeMu sync.Mutex closed bool } @@ -440,7 +445,9 @@ func (c *Consumer) saveCursorAndSyncState(cur int64) error { // // Reconnects with exponential backoff are handled internally by // atmos's streaming.Client; Run does not see transient network -// errors as terminal. +// errors or transient HTTP rejections as terminal. If the upstream +// iterator ends while ctx is still live, Run returns an error wrapping +// ErrUpstreamStreamEnded rather than stopping silently. func (c *Consumer) Run(ctx context.Context) error { c.closeMu.Lock() closed := c.closed @@ -500,9 +507,15 @@ func (c *Consumer) Run(ctx context.Context) error { if c.cfg.ReconnectBackoff != nil { opts.Backoff = gt.Some(*c.cfg.ReconnectBackoff) } - if c.cfg.Dial != nil { - opts.Dial = gt.Some(c.cfg.Dial) - } + // hypercerts: Retry transient relay rejections, log connections, pace + // reads and size the per-DID verify queue (TECH-883). + opts.Parallelism = gt.Some(c.cfg.VerifyParallelism) + opts.Dial = gt.Some(upstreamDial( + c.cfg.Dial, + c.logger, + c.cfg.Metrics, + newUpstreamReadLimiter(c.cfg.UpstreamReadRate, c.cfg.UpstreamReadBurst), + )) client, err := streaming.NewClient(opts) if err != nil { @@ -550,7 +563,18 @@ func (c *Consumer) Run(ctx context.Context) error { } }() - return c.consumeEvents(ctx, client.Events(ctx)) + err = c.consumeEvents(ctx, client.Events(ctx)) + if err == nil && ctx.Err() == nil { + // hypercerts: atmos ends the iterator only on a non-retryable dial + // error. Returning nil here would leave the process up with no live + // ingestion; fail so the orchestrator tears the process down and the + // supervisor restarts it from the persisted cursor (TECH-883). + if c.lastStreamErr == nil { + return ErrUpstreamStreamEnded + } + return fmt.Errorf("%w: last stream error: %w", ErrUpstreamStreamEnded, c.lastStreamErr) + } + return err } // consumeEvents drains one upstream event sequence. It is separate from Run so @@ -564,6 +588,7 @@ func (c *Consumer) consumeEvents(ctx context.Context, events iter.Seq2[[]streami // continuity loss rather than halting, so operators recover // without a crash loop. The loss stays explicitly logged and // recorded; it is never silently absorbed. + c.lastStreamErr = err c.recordUpstreamLoss(ctx, err) // Classify, log, and continue — the next iteration // will either reconnect or yield the next batch. diff --git a/jetstream/internal/ingest/live/errors.go b/jetstream/internal/ingest/live/errors.go index 9f428189..30bdf9d6 100644 --- a/jetstream/internal/ingest/live/errors.go +++ b/jetstream/internal/ingest/live/errors.go @@ -24,6 +24,12 @@ var ( // cursor for such events, so a later jetstream build that knows // the new kind can be replayed from the gap. ErrUnknownEventKind = errors.New("livestream: unknown event kind") + + // hypercerts: ErrUpstreamStreamEnded is returned by Run when the upstream + // event iterator ends while the consumer's context is still live, so a + // non-retryable dial failure stops the process instead of leaving it + // running without ingestion. + ErrUpstreamStreamEnded = errors.New("livestream: upstream stream ended unexpectedly") ) // InvalidEventError is returned by ConvertEvent when a whole upstream diff --git a/jetstream/internal/ingest/live/metrics.go b/jetstream/internal/ingest/live/metrics.go index e240864d..d4f94363 100644 --- a/jetstream/internal/ingest/live/metrics.go +++ b/jetstream/internal/ingest/live/metrics.go @@ -18,6 +18,8 @@ const ( type Metrics struct { EventsReceived prometheus.Counter Reconnects prometheus.Counter + UpstreamConnects prometheus.Counter + TransientDialRejections prometheus.Counter DecodeErrors prometheus.Counter SequenceGaps prometheus.Counter SequenceGapMissedSeqs prometheus.Counter @@ -48,6 +50,19 @@ func NewMetrics(reg prometheus.Registerer) *Metrics { Name: "reconnects_total", Help: "Number of websocket reconnect attempts the atmos client has made.", }), + // hypercerts: Successful dials and retried HTTP rejections make a + // stalled live consumer visible (TECH-883). + UpstreamConnects: prometheus.NewCounter(prometheus.CounterOpts{ + Namespace: metricsNamespace, Subsystem: metricsSubsystem, + Name: "upstream_connects_total", + Help: "Number of successful upstream subscribeRepos websocket connections.", + }), + TransientDialRejections: prometheus.NewCounter(prometheus.CounterOpts{ + Namespace: metricsNamespace, Subsystem: metricsSubsystem, + Name: "upstream_transient_dial_rejections_total", + Help: "Number of upstream dials rejected with a transient HTTP status " + + "(408, 425, 429 or 5xx) and retried with backoff.", + }), DecodeErrors: prometheus.NewCounter(prometheus.CounterOpts{ Namespace: metricsNamespace, Subsystem: metricsSubsystem, Name: "decode_errors_total", @@ -140,7 +155,7 @@ func NewMetrics(reg prometheus.Registerer) *Metrics { }), } reg.MustRegister( - m.EventsReceived, m.Reconnects, + m.EventsReceived, m.Reconnects, m.UpstreamConnects, m.TransientDialRejections, m.DecodeErrors, m.SequenceGaps, m.SequenceGapMissedSeqs, m.UnknownEvents, m.VerifyQueueDrops, m.StreamErrorFrames, m.StaleResyncsDropped, m.ReplayedAccountsDrop, m.ReplayedIdentityDrop, m.UpstreamCursor, m.LastSeenUpstreamEventTimestamp, @@ -160,6 +175,18 @@ func (m *Metrics) incReconnects() { } } +func (m *Metrics) incUpstreamConnects() { + if m != nil { + m.UpstreamConnects.Inc() + } +} + +func (m *Metrics) incTransientDialRejections() { + if m != nil { + m.TransientDialRejections.Inc() + } +} + func (m *Metrics) incDecodeErrors() { if m != nil { m.DecodeErrors.Inc() diff --git a/jetstream/internal/ingest/live/upstream_dial.go b/jetstream/internal/ingest/live/upstream_dial.go new file mode 100644 index 00000000..835fbad0 --- /dev/null +++ b/jetstream/internal/ingest/live/upstream_dial.go @@ -0,0 +1,144 @@ +package live + +// hypercerts: This file adapts the upstream subscribeRepos dial for two +// operational faults observed on Railway staging (TECH-883): +// +// - atmos v0.3.6 classifies every failed dial that carried an HTTP response +// as a permanent *streaming.DialError and ends the event iterator. A relay +// restart behind a proxy answers 502/503 for a few seconds, so one restart +// silently stopped live ingestion for hours. Transient statuses are +// reported to atmos without the response so its own backoff retries them. +// - atmos caps each DID's verify queue at Parallelism*2 and sheds events +// instead of blocking. Catch-up after a restart read the backlog faster +// than it could be verified. Reads are paced by a token bucket so the +// dispatch goroutine cannot outrun verification by an unbounded margin. + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net/http" + "net/url" + + "github.com/bluesky-social/jetstream/internal/version" + "github.com/coder/websocket" + "github.com/jcalabro/atmos/streaming" + "golang.org/x/time/rate" +) + +const ( + // DefaultVerifyParallelism is the number of atmos verification workers + // used when Config.VerifyParallelism is zero. atmos caps each DID's verify + // queue at twice this value; atmos's own default is 32 (queue 64), which + // overflowed while replaying a few hours of backlog after a restart. + DefaultVerifyParallelism = 128 + + // DefaultUpstreamReadRate is the recommended upstream frame read rate, in + // frames per second, for production. It is applied by the command-line + // default; Config.UpstreamReadRate zero disables pacing. + DefaultUpstreamReadRate = 500 +) + +// errTransientDial marks an upstream HTTP rejection that should be retried. +var errTransientDial = errors.New("transient upstream dial rejection") + +// transientDialStatus reports whether an HTTP status returned instead of a +// WebSocket upgrade is expected to clear without operator action: request +// timeouts, rate limits and every server-side 5xx (including proxy 502/503/504 +// responses while the relay restarts). +func transientDialStatus(code int) bool { + switch code { + case http.StatusRequestTimeout, http.StatusTooEarly, http.StatusTooManyRequests: + return true + } + return code >= 500 && code <= 599 +} + +// defaultUpstreamDial mirrors atmos's built-in dial; atmos does not export it. +func defaultUpstreamDial(ctx context.Context, u string, cfg streaming.DialConfig) (streaming.Conn, *http.Response, error) { + var subprotocols []string + for _, s := range cfg.Subprotocols { + subprotocols = append(subprotocols, string(s)) + } + conn, resp, err := websocket.Dial(ctx, u, &websocket.DialOptions{ + HTTPHeader: http.Header{"User-Agent": []string{"hypercerts-jetstream/" + version.Version}}, + Subprotocols: subprotocols, + CompressionMode: cfg.Compression, + }) + if err != nil { + return nil, resp, err + } + return conn, resp, nil +} + +// upstreamDial wraps base with transient-rejection retry, connection logging +// and optional read pacing. limiter may be nil to disable pacing. +func upstreamDial(base streaming.DialFunc, logger *slog.Logger, metrics *Metrics, limiter *rate.Limiter) streaming.DialFunc { + if base == nil { + base = defaultUpstreamDial + } + return func(ctx context.Context, u string, cfg streaming.DialConfig) (streaming.Conn, *http.Response, error) { + cursor := dialCursor(u) + conn, resp, err := base(ctx, u, cfg) + if err != nil { + if resp != nil && transientDialStatus(resp.StatusCode) { + status := resp.StatusCode + if resp.Body != nil { + _ = resp.Body.Close() + } + metrics.incTransientDialRejections() + logger.WarnContext(ctx, "upstream dial rejected; retrying", + "status", status, + "cursor", cursor, + "err", err, + ) + // Returning no response makes atmos treat the failure as + // retryable and apply its reconnect backoff. + return nil, nil, fmt.Errorf("%w: HTTP %d: %w", errTransientDial, status, err) + } + return nil, resp, err + } + metrics.incUpstreamConnects() + logger.InfoContext(ctx, "upstream connected", "cursor", cursor) + if limiter != nil { + conn = &pacedConn{Conn: conn, limiter: limiter} + } + return conn, resp, nil + } +} + +// dialCursor extracts the resume cursor atmos placed in the dial URL, for logs. +func dialCursor(u string) string { + parsed, err := url.Parse(u) + if err != nil { + return "" + } + return parsed.Query().Get("cursor") +} + +// pacedConn waits for a token before each upstream read. Blocking the read +// leaves frames in the socket, so TCP flow control pushes back on the relay +// instead of atmos shedding events from a full per-DID verify queue. +type pacedConn struct { + streaming.Conn + limiter *rate.Limiter +} + +func (p *pacedConn) Read(ctx context.Context) (websocket.MessageType, []byte, error) { + if err := p.limiter.Wait(ctx); err != nil { + return 0, nil, err + } + return p.Conn.Read(ctx) +} + +// newUpstreamReadLimiter returns nil when pacing is disabled. +func newUpstreamReadLimiter(perSecond float64, burst int) *rate.Limiter { + if perSecond <= 0 { + return nil + } + if burst < 1 { + burst = 1 + } + return rate.NewLimiter(rate.Limit(perSecond), burst) +} diff --git a/jetstream/internal/ingest/live/upstream_dial_test.go b/jetstream/internal/ingest/live/upstream_dial_test.go new file mode 100644 index 00000000..162848f9 --- /dev/null +++ b/jetstream/internal/ingest/live/upstream_dial_test.go @@ -0,0 +1,302 @@ +package live + +import ( + "context" + "errors" + "io" + "log/slog" + "math" + "net/http" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/bluesky-social/jetstream/segment" + "github.com/jcalabro/atmos/streaming" + "github.com/jcalabro/gt" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" +) + +// trackedBody records whether the dial wrapper closed a rejected response. +type trackedBody struct { + io.Reader + closed atomic.Bool +} + +func (b *trackedBody) Close() error { b.closed.Store(true); return nil } + +func httpRejection(code int) (*http.Response, *trackedBody) { + body := &trackedBody{Reader: strings.NewReader("")} + return &http.Response{StatusCode: code, Body: body}, body +} + +func fastBackoff() *streaming.BackoffPolicy { + return &streaming.BackoffPolicy{ + InitialDelay: gt.Some(time.Millisecond), + MaxDelay: gt.Some(5 * time.Millisecond), + Multiplier: gt.Some(1.0), + Jitter: gt.Some(false), + } +} + +func TestTransientDialStatus(t *testing.T) { + t.Parallel() + for code, want := range map[int]bool{ + http.StatusRequestTimeout: true, + http.StatusTooEarly: true, + http.StatusTooManyRequests: true, + http.StatusInternalServerError: true, + http.StatusBadGateway: true, + http.StatusServiceUnavailable: true, + http.StatusGatewayTimeout: true, + http.StatusOK: false, + http.StatusBadRequest: false, + http.StatusUnauthorized: false, + http.StatusNotFound: false, + } { + require.Equal(t, want, transientDialStatus(code), "status %d", code) + } +} + +func TestUpstreamDial_TransientRejectionIsRetryable(t *testing.T) { + t.Parallel() + resp, body := httpRejection(http.StatusBadGateway) + metrics := NewMetrics(prometheus.NewRegistry()) + dial := upstreamDial(func(context.Context, string, streaming.DialConfig) (streaming.Conn, *http.Response, error) { + return nil, resp, errors.New("expected handshake response status code 101 but got 502") + }, slog.New(slog.DiscardHandler), metrics, nil) + + conn, gotResp, err := dial(t.Context(), "wss://relay.invalid/xrpc/com.atproto.sync.subscribeRepos?cursor=42", streaming.DialConfig{}) + require.Nil(t, conn) + require.Nil(t, gotResp, "a transient rejection must not carry its response, or atmos treats it as permanent") + require.ErrorIs(t, err, errTransientDial) + require.True(t, body.closed.Load(), "the rejected response body must be closed") + require.Equal(t, 1.0, testutil.ToFloat64(metrics.TransientDialRejections)) + require.Equal(t, 0.0, testutil.ToFloat64(metrics.UpstreamConnects)) +} + +func TestUpstreamDial_PermanentRejectionKeepsResponse(t *testing.T) { + t.Parallel() + resp, _ := httpRejection(http.StatusNotFound) + dial := upstreamDial(func(context.Context, string, streaming.DialConfig) (streaming.Conn, *http.Response, error) { + return nil, resp, errors.New("got 404") + }, slog.New(slog.DiscardHandler), nil, nil) + + _, gotResp, err := dial(t.Context(), "wss://relay.invalid/x", streaming.DialConfig{}) + require.Error(t, err) + require.NotErrorIs(t, err, errTransientDial) + require.Same(t, resp, gotResp) +} + +func TestUpstreamDial_SuccessWrapsConnOnlyWhenPaced(t *testing.T) { + t.Parallel() + metrics := NewMetrics(prometheus.NewRegistry()) + base := func(context.Context, string, streaming.DialConfig) (streaming.Conn, *http.Response, error) { + return newMemConn(), nil, nil + } + + conn, _, err := upstreamDial(base, slog.New(slog.DiscardHandler), metrics, nil)(t.Context(), "wss://relay.invalid/x", streaming.DialConfig{}) + require.NoError(t, err) + require.IsType(t, &memConn{}, conn) + + conn, _, err = upstreamDial(base, slog.New(slog.DiscardHandler), metrics, newUpstreamReadLimiter(10, 1))(t.Context(), "wss://relay.invalid/x", streaming.DialConfig{}) + require.NoError(t, err) + require.IsType(t, &pacedConn{}, conn) + require.Equal(t, 2.0, testutil.ToFloat64(metrics.UpstreamConnects)) +} + +func TestDialCursor(t *testing.T) { + t.Parallel() + require.Equal(t, "189045", dialCursor("wss://relay.invalid/xrpc/com.atproto.sync.subscribeRepos?cursor=189045")) + require.Empty(t, dialCursor("wss://relay.invalid/xrpc/com.atproto.sync.subscribeRepos")) + require.Empty(t, dialCursor("://bad")) +} + +func TestNewUpstreamReadLimiter(t *testing.T) { + t.Parallel() + require.Nil(t, newUpstreamReadLimiter(0, 10)) + require.Nil(t, newUpstreamReadLimiter(-1, 10)) + l := newUpstreamReadLimiter(100, 0) + require.NotNil(t, l) + require.Equal(t, 1, l.Burst(), "a zero burst would make every Wait fail") +} + +func TestPacedConn_LimitsReadRate(t *testing.T) { + t.Parallel() + frames := [][]byte{{1}, {2}, {3}, {4}, {5}} + conn := &pacedConn{Conn: newMemConn(frames...), limiter: newUpstreamReadLimiter(50, 1)} + + start := time.Now() + for range frames { + _, _, err := conn.Read(t.Context()) + require.NoError(t, err) + } + // One token is available immediately; the remaining four arrive at 50/s. + require.GreaterOrEqual(t, time.Since(start), 70*time.Millisecond) +} + +func TestPacedConn_ReadHonoursContext(t *testing.T) { + t.Parallel() + conn := &pacedConn{Conn: newMemConn([]byte{1}, []byte{2}), limiter: newUpstreamReadLimiter(0.001, 1)} + _, _, err := conn.Read(t.Context()) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, _, err = conn.Read(ctx) + require.Error(t, err) +} + +func TestConfig_LiveIngestTuning(t *testing.T) { + t.Parallel() + base := func() Config { + return Config{ + SegmentsDir: "x", Store: newTestStore(t), SeqKey: "s", CursorKey: "c", + RelayURL: "https://relay.invalid", Logger: slog.New(slog.DiscardHandler), + Verifier: newTestVerifier(t), + } + } + + cfg := base() + require.NoError(t, cfg.validate()) + cfg.applyDefaults() + require.Equal(t, DefaultVerifyParallelism, cfg.VerifyParallelism) + require.Equal(t, DefaultVerifyParallelism, cfg.UpstreamReadBurst) + require.Zero(t, cfg.UpstreamReadRate, "pacing stays opt-in below the command line") + + cfg = base() + cfg.VerifyParallelism = 8 + cfg.applyDefaults() + require.Equal(t, 8, cfg.UpstreamReadBurst) + + for name, mutate := range map[string]func(*Config){ + "negative parallelism": func(c *Config) { c.VerifyParallelism = -1 }, + "negative rate": func(c *Config) { c.UpstreamReadRate = -1 }, + "infinite rate": func(c *Config) { c.UpstreamReadRate = math.Inf(1) }, + "negative burst": func(c *Config) { c.UpstreamReadBurst = -1 }, + } { + cfg := base() + mutate(&cfg) + require.ErrorIs(t, cfg.validate(), ErrInvalidConfig, name) + } +} + +// TestConsumer_Run_RetriesTransientRelayRejection reproduces the staging +// stall (TECH-883): the relay answers 503 while it restarts. Ingestion must +// back off, reconnect from the same cursor and deliver the events. +func TestConsumer_Run_RetriesTransientRelayRejection(t *testing.T) { + t.Parallel() + + frames := [][]byte{ + encodeIdentityFrame(t, "did:plc:aaa", 1), + encodeIdentityFrame(t, "did:plc:bbb", 2), + } + conn := newMemConn(frames...) + metrics := NewMetrics(prometheus.NewRegistry()) + + var dials atomic.Int64 + var delivered atomic.Int64 + c, err := Open(Config{ + SegmentsDir: filepath.Join(t.TempDir(), "live_segments"), + Store: newTestStore(t), + SeqKey: "live_segments/seq/next", + CursorKey: "relay/cursor", + RelayURL: "https://relay.invalid", + Logger: slog.New(slog.DiscardHandler), + Metrics: metrics, + Verifier: newTestVerifier(t), + MaxEventsPerBlock: 2, + ReconnectBackoff: fastBackoff(), + UpstreamReadRate: 1000, + OnEvent: func(*segment.Event) { delivered.Add(1) }, + Dial: func(context.Context, string, streaming.DialConfig) (streaming.Conn, *http.Response, error) { + if dials.Add(1) <= 2 { + resp, _ := httpRejection(http.StatusServiceUnavailable) + return nil, resp, errors.New("expected handshake response status code 101 but got 503") + } + return conn, nil, nil + }, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = c.Close() }) + + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + t.Cleanup(cancel) + runErr := make(chan error, 1) + go func() { runErr <- c.Run(ctx) }() + + require.Eventually(t, func() bool { return delivered.Load() >= int64(len(frames)) }, + 3*time.Second, 10*time.Millisecond, "ingestion did not resume after transient relay rejections") + require.Equal(t, 2.0, testutil.ToFloat64(metrics.TransientDialRejections)) + require.Equal(t, 1.0, testutil.ToFloat64(metrics.UpstreamConnects)) + require.GreaterOrEqual(t, testutil.ToFloat64(metrics.Reconnects), 2.0) + + select { + case err := <-runErr: + t.Fatalf("Run returned while the stream was healthy: %v", err) + default: + } + cancel() + select { + case <-runErr: + case <-time.After(3 * time.Second): + t.Fatal("Run did not return after cancel") + } +} + +// TestConsumer_Run_FailsWhenUpstreamStreamEnds pins the no-silent-stall +// contract: when atmos ends the iterator on a permanent dial rejection while +// the consumer is still meant to run, Run returns an error. +func TestConsumer_Run_FailsWhenUpstreamStreamEnds(t *testing.T) { + t.Parallel() + + c, err := Open(Config{ + SegmentsDir: filepath.Join(t.TempDir(), "live_segments"), + Store: newTestStore(t), + SeqKey: "live_segments/seq/next", + CursorKey: "relay/cursor", + RelayURL: "https://relay.invalid", + Logger: slog.New(slog.DiscardHandler), + Verifier: newTestVerifier(t), + ReconnectBackoff: fastBackoff(), + Dial: func(context.Context, string, streaming.DialConfig) (streaming.Conn, *http.Response, error) { + resp, _ := httpRejection(http.StatusNotFound) + return nil, resp, errors.New("expected handshake response status code 101 but got 404") + }, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = c.Close() }) + + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + t.Cleanup(cancel) + err = c.Run(ctx) + require.ErrorIs(t, err, ErrUpstreamStreamEnded) + var dialErr *streaming.DialError + require.ErrorAs(t, err, &dialErr, "the terminal dial error must be reported") + require.Equal(t, http.StatusNotFound, dialErr.StatusCode) + require.NoError(t, ctx.Err(), "Run must fail promptly rather than wait for cancellation") +} + +// TestConsumer_Run_DefaultsVerifyParallelism checks that Open applies the +// larger verify worker default rather than atmos's default of 32. +func TestConsumer_Run_DefaultsVerifyParallelism(t *testing.T) { + t.Parallel() + c, err := Open(Config{ + SegmentsDir: filepath.Join(t.TempDir(), "live_segments"), + Store: newTestStore(t), + SeqKey: "live_segments/seq/next", + CursorKey: "relay/cursor", + RelayURL: "https://relay.invalid", + Logger: slog.New(slog.DiscardHandler), + Verifier: newTestVerifier(t), + }) + require.NoError(t, err) + t.Cleanup(func() { _ = c.Close() }) + require.Equal(t, DefaultVerifyParallelism, c.cfg.VerifyParallelism) +} + +var _ streaming.Conn = (*pacedConn)(nil) diff --git a/jetstream/internal/ingest/orchestrator/bootstrap.go b/jetstream/internal/ingest/orchestrator/bootstrap.go index 27292d78..609be298 100644 --- a/jetstream/internal/ingest/orchestrator/bootstrap.go +++ b/jetstream/internal/ingest/orchestrator/bootstrap.go @@ -155,6 +155,9 @@ func (o *Orchestrator) openBootstrapLive(liveSegmentsDir string) (*live.Consumer OnUpstreamLoss: o.cfg.OnUpstreamLoss, ReconnectBackoff: o.cfg.LiveReconnectBackoff, Dial: o.cfg.LiveDial, + VerifyParallelism: o.cfg.LiveVerifyParallelism, + UpstreamReadRate: o.cfg.LiveUpstreamReadRate, + UpstreamReadBurst: o.cfg.LiveUpstreamReadBurst, SegmentIOFaultInjector: o.cfg.SegmentIOFaultInjector, }) diff --git a/jetstream/internal/ingest/orchestrator/config.go b/jetstream/internal/ingest/orchestrator/config.go index 71e3afa8..706bb078 100644 --- a/jetstream/internal/ingest/orchestrator/config.go +++ b/jetstream/internal/ingest/orchestrator/config.go @@ -228,6 +228,13 @@ type Config struct { // nil; deterministic harnesses feed the firehose in-memory. LiveDial streaming.DialFunc + // hypercerts: LiveVerifyParallelism, LiveUpstreamReadRate and + // LiveUpstreamReadBurst size atmos's verify workers and pace upstream + // reads for both live consumers. See live.Config for zero-value meaning. + LiveVerifyParallelism int + LiveUpstreamReadRate float64 + LiveUpstreamReadBurst int + // IngestOnAfterSeal is forwarded to every writer that appends to // /segments. Used by cmd/jetstream to wire the manifest's // OnSegmentSealed callback. Optional. diff --git a/jetstream/internal/ingest/orchestrator/steady.go b/jetstream/internal/ingest/orchestrator/steady.go index 68120fce..323303a5 100644 --- a/jetstream/internal/ingest/orchestrator/steady.go +++ b/jetstream/internal/ingest/orchestrator/steady.go @@ -91,6 +91,9 @@ func (o *Orchestrator) runSteadyState(ctx context.Context) error { TimestampStamper: o.cfg.TimestampStamper, ReconnectBackoff: o.cfg.LiveReconnectBackoff, Dial: o.cfg.LiveDial, + VerifyParallelism: o.cfg.LiveVerifyParallelism, + UpstreamReadRate: o.cfg.LiveUpstreamReadRate, + UpstreamReadBurst: o.cfg.LiveUpstreamReadBurst, SegmentIOFaultInjector: o.cfg.SegmentIOFaultInjector, }) diff --git a/jetstream/internal/jetstreamd/options.go b/jetstream/internal/jetstreamd/options.go index 694e3b01..56dbdc1e 100644 --- a/jetstream/internal/jetstreamd/options.go +++ b/jetstream/internal/jetstreamd/options.go @@ -12,6 +12,7 @@ import ( "github.com/bluesky-social/jetstream/internal/crashpoint" "github.com/bluesky-social/jetstream/internal/ingest" "github.com/bluesky-social/jetstream/internal/ingest/backfill" + "github.com/bluesky-social/jetstream/internal/ingest/live" "github.com/bluesky-social/jetstream/internal/store" "github.com/bluesky-social/jetstream/segment" "github.com/cockroachdb/pebble/vfs" @@ -30,6 +31,9 @@ const ( DefaultFailedRepoRetryWorkers = backfill.DefaultFailedRepoRetryWorkers DefaultFailedRepoRetryHostWorkers = backfill.DefaultFailedRepoRetryHostWorkers DefaultFailedRepoRetryMaxDelay = backfill.DefaultFailedRepoRetryMaxDelay + // hypercerts: Live ingestion verify sizing and read pacing (TECH-883). + DefaultLiveVerifyParallelism = live.DefaultVerifyParallelism + DefaultLiveUpstreamReadRate = live.DefaultUpstreamReadRate ) // PhaseBarrier is a test hook that can pause execution after a major @@ -117,6 +121,13 @@ type Options struct { // firehose over an in-memory connection. LiveDial streaming.DialFunc + // hypercerts: LiveVerifyParallelism, LiveUpstreamReadRate and + // LiveUpstreamReadBurst size atmos's verify workers and pace upstream + // reads for both live consumers. See live.Config for zero-value meaning. + LiveVerifyParallelism int + LiveUpstreamReadRate float64 + LiveUpstreamReadBurst int + // HTTPTransport, when non-nil, is the RoundTripper for every outbound // HTTP client (backfill getRepo/listRepos, identity/PLC resolution). // Production leaves it nil (real sockets); deterministic harnesses serve diff --git a/jetstream/internal/jetstreamd/runtime.go b/jetstream/internal/jetstreamd/runtime.go index 796d6a13..318ee936 100644 --- a/jetstream/internal/jetstreamd/runtime.go +++ b/jetstream/internal/jetstreamd/runtime.go @@ -74,6 +74,8 @@ type Runtime struct { manifest *manifest.Manifest tail *subscribe.Tail verifier *atmossync.Verifier + // verifyFailures summarises repeated verification failure log lines. + verifyFailures *verifyFailureLog // recovery books durable jobs for events shed by the verify queue. recovery *recoveryCoalescer orchestrator *orchestrator.Orchestrator @@ -402,16 +404,18 @@ func Build(ctx context.Context, opts Options) (*Runtime, error) { rt.tail = tail verifierLogger := processLogger.With(slog.String("component", "verifier")) + // hypercerts: Repeated failures for one DID are summarised rather than + // logged per commit; the failure metric still counts every one (TECH-883). + verifyFailures := newVerifyFailureLog(verifierLogger) + rt.verifyFailures = verifyFailures verifier, err := atmossync.NewVerifier(atmossync.VerifierOptions{ Directory: directory, StateStore: stateStore, SyncClient: gt.Some(syncClient), OnVerificationFailure: gt.Some(func(did atmos.DID, vErr error) { - verifierMetrics.IncFailure(obs.Classify(vErr)) - verifierLogger.Warn("verification failure", - "did", did, - "err", vErr, - ) + class := classifyVerifyFailure(vErr) + verifierMetrics.IncFailure(class) + verifyFailures.observe(string(did), class, vErr) }), }) if err != nil { @@ -567,6 +571,9 @@ func Build(ctx context.Context, opts Options) (*Runtime, error) { FailedRepoRetryMaxDelay: opts.FailedRepoRetryMaxDelay, LiveReconnectBackoff: opts.LiveReconnectBackoff, LiveDial: opts.LiveDial, + LiveVerifyParallelism: opts.LiveVerifyParallelism, + LiveUpstreamReadRate: opts.LiveUpstreamReadRate, + LiveUpstreamReadBurst: opts.LiveUpstreamReadBurst, IngestOnAfterSeal: mft.OnSegmentSealed, OnSegmentCompacted: onSegmentCompacted, SegmentManifestChecksums: mft.SegmentChecksums, @@ -946,6 +953,12 @@ func (r *Runtime) Run(ctx context.Context) (runErr error) { })) } + if r.verifyFailures != nil { + g.Go(r.goroutineRoot("verify_failure_log", func() error { + return r.verifyFailures.run(gctx) + })) + } + // hypercerts: Start the recovery coalescer before the orchestrator opens the // live consumer, so the first bootstrap-phase drops reach a running sink. if r.recovery != nil { diff --git a/jetstream/internal/jetstreamd/verify_failure_log.go b/jetstream/internal/jetstreamd/verify_failure_log.go new file mode 100644 index 00000000..a1a4861e --- /dev/null +++ b/jetstream/internal/jetstreamd/verify_failure_log.go @@ -0,0 +1,146 @@ +package jetstreamd + +// hypercerts: The atmos verifier reports a verification failure for every +// commit it buffers while a DID resyncs, each against the same stale chain +// state. One gap after a catch-up overflow produced hundreds of identical +// warnings per second and Railway discarded thousands of log lines +// (TECH-883). This summariser logs the first failure per DID and class in full, +// counts repeats, and reports them once per window. Metrics stay exact; only +// log volume is reduced. + +import ( + "context" + "log/slog" + "sync" + "time" + + "github.com/bluesky-social/jetstream/internal/obs" +) + +const ( + // verifyFailureLogWindow is how long repeats of one DID and failure + // class are counted before they are summarised. + verifyFailureLogWindow = time.Minute + // verifyFailureLogMaxKeys bounds the per-DID state. Failures for further + // DIDs in a window are counted in aggregate only. + verifyFailureLogMaxKeys = 10_000 +) + +type verifyFailureKey struct { + did string + class string +} + +type verifyFailureEntry struct { + first time.Time + suppressed uint64 + lastErr error +} + +type verifyFailureLog struct { + logger *slog.Logger + window time.Duration + maxKeys int + now func() time.Time + + mu sync.Mutex + entries map[verifyFailureKey]*verifyFailureEntry + overflow uint64 +} + +func newVerifyFailureLog(logger *slog.Logger) *verifyFailureLog { + return &verifyFailureLog{ + logger: logger, + window: verifyFailureLogWindow, + maxKeys: verifyFailureLogMaxKeys, + now: time.Now, + entries: map[verifyFailureKey]*verifyFailureEntry{}, + } +} + +// observe records one failure. It never blocks on I/O while holding the lock. +func (l *verifyFailureLog) observe(did string, class string, err error) { + key := verifyFailureKey{did: did, class: class} + l.mu.Lock() + if e, ok := l.entries[key]; ok { + e.suppressed++ + e.lastErr = err + l.mu.Unlock() + return + } + if len(l.entries) >= l.maxKeys { + l.overflow++ + l.mu.Unlock() + return + } + l.entries[key] = &verifyFailureEntry{first: l.now()} + l.mu.Unlock() + l.logger.Warn("verification failure", "did", did, "class", class, "err", err) +} + +type verifyFailureSummary struct { + key verifyFailureKey + count uint64 + lastErr error +} + +// flush summarises and clears entries whose window has elapsed, or all +// entries when force is set. Logging happens after the lock is released. +func (l *verifyFailureLog) flush(force bool) { + now := l.now() + var summaries []verifyFailureSummary + l.mu.Lock() + for key, e := range l.entries { + if !force && now.Sub(e.first) < l.window { + continue + } + if e.suppressed > 0 { + summaries = append(summaries, verifyFailureSummary{key: key, count: e.suppressed, lastErr: e.lastErr}) + } + delete(l.entries, key) + } + overflow := l.overflow + l.overflow = 0 + l.mu.Unlock() + + for _, s := range summaries { + l.logger.Warn("verification failures repeated", + "did", s.key.did, + "class", s.key.class, + "suppressed", s.count, + "window", l.window, + "last_err", s.lastErr, + ) + } + if overflow > 0 { + l.logger.Warn("verification failures not logged per DID", + "count", overflow, + "tracked_dids_limit", l.maxKeys, + ) + } +} + +// run flushes elapsed windows until ctx ends, then flushes everything. +func (l *verifyFailureLog) run(ctx context.Context) error { + interval := l.window / 4 + if interval <= 0 { + interval = time.Second + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + l.flush(true) + return nil + case <-ticker.C: + l.flush(false) + } + } +} + +// classifyVerifyFailure is the metric class used for the log key, so log +// summaries line up with jetstream_verifier_failures_total labels. +func classifyVerifyFailure(err error) string { + return obs.Classify(err) +} diff --git a/jetstream/internal/jetstreamd/verify_failure_log_test.go b/jetstream/internal/jetstreamd/verify_failure_log_test.go new file mode 100644 index 00000000..873878f8 --- /dev/null +++ b/jetstream/internal/jetstreamd/verify_failure_log_test.go @@ -0,0 +1,123 @@ +package jetstreamd + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "log/slog" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncBuffer) records(t *testing.T) []map[string]any { + t.Helper() + b.mu.Lock() + defer b.mu.Unlock() + var out []map[string]any + for _, line := range bytes.Split(bytes.TrimSpace(b.buf.Bytes()), []byte("\n")) { + if len(line) == 0 { + continue + } + var rec map[string]any + require.NoError(t, json.Unmarshal(line, &rec)) + out = append(out, rec) + } + return out +} + +func newTestVerifyFailureLog() (*verifyFailureLog, *syncBuffer, *time.Time) { + buf := &syncBuffer{} + l := newVerifyFailureLog(slog.New(slog.NewJSONHandler(buf, nil))) + now := time.Date(2026, 11, 12, 11, 17, 24, 0, time.UTC) + l.now = func() time.Time { return now } + return l, buf, &now +} + +func TestVerifyFailureLog_SummarisesRepeatsPerDID(t *testing.T) { + t.Parallel() + l, buf, now := newTestVerifyFailureLog() + + for i := range 180 { + l.observe("did:plc:a", "chain", errors.New("chain break "+string(rune('a'+i%26)))) + } + l.observe("did:plc:b", "chain", errors.New("chain break b")) + l.observe("did:plc:a", "signature", errors.New("bad signature")) + + recs := buf.records(t) + require.Len(t, recs, 3, "only the first failure per DID and class is logged immediately") + for _, r := range recs { + require.Equal(t, "verification failure", r["msg"]) + } + + l.flush(false) + require.Len(t, buf.records(t), 3, "nothing is summarised before the window elapses") + + *now = now.Add(verifyFailureLogWindow) + l.flush(false) + recs = buf.records(t) + require.Len(t, recs, 4, "one summary for the DID with repeats; none for single failures") + summary := recs[3] + require.Equal(t, "verification failures repeated", summary["msg"]) + require.Equal(t, "did:plc:a", summary["did"]) + require.Equal(t, "chain", summary["class"]) + require.EqualValues(t, 179, summary["suppressed"]) + require.Contains(t, summary["last_err"], "chain break") + + // The window restarts: the next failure is logged in full again. + l.observe("did:plc:a", "chain", errors.New("chain break again")) + recs = buf.records(t) + require.Len(t, recs, 5) + require.Equal(t, "verification failure", recs[4]["msg"]) +} + +func TestVerifyFailureLog_BoundsTrackedDIDs(t *testing.T) { + t.Parallel() + l, buf, _ := newTestVerifyFailureLog() + l.maxKeys = 2 + + l.observe("did:plc:a", "chain", errors.New("x")) + l.observe("did:plc:b", "chain", errors.New("x")) + l.observe("did:plc:c", "chain", errors.New("x")) + l.observe("did:plc:d", "chain", errors.New("x")) + require.Len(t, buf.records(t), 2) + + l.flush(true) + recs := buf.records(t) + require.Len(t, recs, 3) + require.Equal(t, "verification failures not logged per DID", recs[2]["msg"]) + require.EqualValues(t, 2, recs[2]["count"]) + require.Empty(t, l.entries, "a forced flush clears all state") +} + +func TestVerifyFailureLog_RunFlushesOnShutdown(t *testing.T) { + t.Parallel() + l, buf, _ := newTestVerifyFailureLog() + l.observe("did:plc:a", "chain", errors.New("x")) + l.observe("did:plc:a", "chain", errors.New("y")) + + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { done <- l.run(ctx) }() + cancel() + require.NoError(t, <-done) + + recs := buf.records(t) + require.Len(t, recs, 2) + require.Equal(t, "verification failures repeated", recs[1]["msg"]) + require.EqualValues(t, 1, recs[1]["suppressed"]) +} From 619d5716e8012ad72a5967e061b08cea6b2cf565 Mon Sep 17 00:00:00 2001 From: Evelyn Osman Date: Thu, 12 Nov 2026 14:42:58 +0100 Subject: [PATCH 3/5] fix(jetstream): address review of live ingest resilience - Mark the changeset minor: it adds three configuration options. - Assert that redials after transient rejections resume from the persisted cursor. - Add hypercerts markers to the remaining upstream-file hunks and drop a pass-through classifier wrapper. This work was assisted by a machine --- .../jetstream-live-ingest-resilience.md | 2 +- jetstream/internal/ingest/live/consumer.go | 3 ++- jetstream/internal/ingest/live/metrics.go | 6 ++++-- .../ingest/live/upstream_dial_test.go | 21 ++++++++++++++----- jetstream/internal/jetstreamd/runtime.go | 5 +++-- .../internal/jetstreamd/verify_failure_log.go | 8 ------- 6 files changed, 26 insertions(+), 19 deletions(-) diff --git a/.changeset/jetstream-live-ingest-resilience.md b/.changeset/jetstream-live-ingest-resilience.md index 908f7140..c0420b1e 100644 --- a/.changeset/jetstream-live-ingest-resilience.md +++ b/.changeset/jetstream-live-ingest-resilience.md @@ -1,5 +1,5 @@ --- -'hypercerts-relay': patch +'hypercerts-relay': minor --- Jetstream live ingestion no longer stops silently when the Relay restarts. diff --git a/jetstream/internal/ingest/live/consumer.go b/jetstream/internal/ingest/live/consumer.go index e3253390..ad5ad557 100644 --- a/jetstream/internal/ingest/live/consumer.go +++ b/jetstream/internal/ingest/live/consumer.go @@ -74,7 +74,7 @@ type Consumer struct { // processBatch in isolation. client atomic.Pointer[streaming.Client] - // lastStreamErr is the most recent stream-level error yielded by the + // hypercerts: lastStreamErr is the most recent stream-level error yielded by the // upstream iterator. Owned by the Run goroutine; reported when the // iterator ends while the consumer is still meant to run. lastStreamErr error @@ -588,6 +588,7 @@ func (c *Consumer) consumeEvents(ctx context.Context, events iter.Seq2[[]streami // continuity loss rather than halting, so operators recover // without a crash loop. The loss stays explicitly logged and // recorded; it is never silently absorbed. + // hypercerts: Kept to report why the iterator ended (TECH-883). c.lastStreamErr = err c.recordUpstreamLoss(ctx, err) // Classify, log, and continue — the next iteration diff --git a/jetstream/internal/ingest/live/metrics.go b/jetstream/internal/ingest/live/metrics.go index d4f94363..03bdb534 100644 --- a/jetstream/internal/ingest/live/metrics.go +++ b/jetstream/internal/ingest/live/metrics.go @@ -16,8 +16,9 @@ const ( // consumer. A nil *Metrics is a valid zero-value: every method is a // no-op, so tests can skip metric registration entirely. type Metrics struct { - EventsReceived prometheus.Counter - Reconnects prometheus.Counter + EventsReceived prometheus.Counter + Reconnects prometheus.Counter + // hypercerts: Upstream connection visibility (TECH-883). UpstreamConnects prometheus.Counter TransientDialRejections prometheus.Counter DecodeErrors prometheus.Counter @@ -175,6 +176,7 @@ func (m *Metrics) incReconnects() { } } +// hypercerts: Helpers for the TECH-883 upstream connection counters. func (m *Metrics) incUpstreamConnects() { if m != nil { m.UpstreamConnects.Inc() diff --git a/jetstream/internal/ingest/live/upstream_dial_test.go b/jetstream/internal/ingest/live/upstream_dial_test.go index 162848f9..11f2c557 100644 --- a/jetstream/internal/ingest/live/upstream_dial_test.go +++ b/jetstream/internal/ingest/live/upstream_dial_test.go @@ -9,6 +9,7 @@ import ( "net/http" "path/filepath" "strings" + "sync" "sync/atomic" "testing" "time" @@ -192,17 +193,19 @@ func TestConsumer_Run_RetriesTransientRelayRejection(t *testing.T) { t.Parallel() frames := [][]byte{ - encodeIdentityFrame(t, "did:plc:aaa", 1), - encodeIdentityFrame(t, "did:plc:bbb", 2), + encodeIdentityFrame(t, "did:plc:aaa", 42), + encodeIdentityFrame(t, "did:plc:bbb", 43), } conn := newMemConn(frames...) metrics := NewMetrics(prometheus.NewRegistry()) + st := newTestStore(t) var dials atomic.Int64 + var dialedCursors sync.Map var delivered atomic.Int64 c, err := Open(Config{ SegmentsDir: filepath.Join(t.TempDir(), "live_segments"), - Store: newTestStore(t), + Store: st, SeqKey: "live_segments/seq/next", CursorKey: "relay/cursor", RelayURL: "https://relay.invalid", @@ -213,8 +216,10 @@ func TestConsumer_Run_RetriesTransientRelayRejection(t *testing.T) { ReconnectBackoff: fastBackoff(), UpstreamReadRate: 1000, OnEvent: func(*segment.Event) { delivered.Add(1) }, - Dial: func(context.Context, string, streaming.DialConfig) (streaming.Conn, *http.Response, error) { - if dials.Add(1) <= 2 { + Dial: func(_ context.Context, u string, _ streaming.DialConfig) (streaming.Conn, *http.Response, error) { + n := dials.Add(1) + dialedCursors.Store(n, dialCursor(u)) + if n <= 2 { resp, _ := httpRejection(http.StatusServiceUnavailable) return nil, resp, errors.New("expected handshake response status code 101 but got 503") } @@ -223,6 +228,7 @@ func TestConsumer_Run_RetriesTransientRelayRejection(t *testing.T) { }) require.NoError(t, err) t.Cleanup(func() { _ = c.Close() }) + require.NoError(t, c.saveCursorAndSyncState(41)) ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) t.Cleanup(cancel) @@ -234,6 +240,11 @@ func TestConsumer_Run_RetriesTransientRelayRejection(t *testing.T) { require.Equal(t, 2.0, testutil.ToFloat64(metrics.TransientDialRejections)) require.Equal(t, 1.0, testutil.ToFloat64(metrics.UpstreamConnects)) require.GreaterOrEqual(t, testutil.ToFloat64(metrics.Reconnects), 2.0) + for i := int64(1); i <= 3; i++ { + cur, ok := dialedCursors.Load(i) + require.True(t, ok, "dial %d not observed", i) + require.Equal(t, "41", cur, "dial %d must resume from the persisted cursor", i) + } select { case err := <-runErr: diff --git a/jetstream/internal/jetstreamd/runtime.go b/jetstream/internal/jetstreamd/runtime.go index 318ee936..fed065fc 100644 --- a/jetstream/internal/jetstreamd/runtime.go +++ b/jetstream/internal/jetstreamd/runtime.go @@ -74,7 +74,7 @@ type Runtime struct { manifest *manifest.Manifest tail *subscribe.Tail verifier *atmossync.Verifier - // verifyFailures summarises repeated verification failure log lines. + // hypercerts: verifyFailures summarises repeated verification failure log lines. verifyFailures *verifyFailureLog // recovery books durable jobs for events shed by the verify queue. recovery *recoveryCoalescer @@ -413,7 +413,7 @@ func Build(ctx context.Context, opts Options) (*Runtime, error) { StateStore: stateStore, SyncClient: gt.Some(syncClient), OnVerificationFailure: gt.Some(func(did atmos.DID, vErr error) { - class := classifyVerifyFailure(vErr) + class := obs.Classify(vErr) verifierMetrics.IncFailure(class) verifyFailures.observe(string(did), class, vErr) }), @@ -953,6 +953,7 @@ func (r *Runtime) Run(ctx context.Context) (runErr error) { })) } + // hypercerts: Flush verification failure log summaries (TECH-883). if r.verifyFailures != nil { g.Go(r.goroutineRoot("verify_failure_log", func() error { return r.verifyFailures.run(gctx) diff --git a/jetstream/internal/jetstreamd/verify_failure_log.go b/jetstream/internal/jetstreamd/verify_failure_log.go index a1a4861e..e4fc4caf 100644 --- a/jetstream/internal/jetstreamd/verify_failure_log.go +++ b/jetstream/internal/jetstreamd/verify_failure_log.go @@ -13,8 +13,6 @@ import ( "log/slog" "sync" "time" - - "github.com/bluesky-social/jetstream/internal/obs" ) const ( @@ -138,9 +136,3 @@ func (l *verifyFailureLog) run(ctx context.Context) error { } } } - -// classifyVerifyFailure is the metric class used for the log key, so log -// summaries line up with jetstream_verifier_failures_total labels. -func classifyVerifyFailure(err error) string { - return obs.Classify(err) -} From 2edd3d2408ea61f6f332406615655118ad3f7a79 Mon Sep 17 00:00:00 2001 From: Evelyn Osman Date: Thu, 12 Nov 2026 14:47:25 +0100 Subject: [PATCH 4/5] Switch changeset to patch --- .changeset/jetstream-live-ingest-resilience.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/jetstream-live-ingest-resilience.md b/.changeset/jetstream-live-ingest-resilience.md index c0420b1e..908f7140 100644 --- a/.changeset/jetstream-live-ingest-resilience.md +++ b/.changeset/jetstream-live-ingest-resilience.md @@ -1,5 +1,5 @@ --- -'hypercerts-relay': minor +'hypercerts-relay': patch --- Jetstream live ingestion no longer stops silently when the Relay restarts. From 04deceb464f8dc9e52e49937ad149cb37650f2ad Mon Sep 17 00:00:00 2001 From: "hypercerts-release-bot[bot]" <253337036+hypercerts-release-bot[bot]@users.noreply.github.com> Date: Thu, 12 Nov 2026 14:00:18 +0000 Subject: [PATCH 5/5] chore: release --- .../jetstream-live-ingest-resilience.md | 22 ------------------ CHANGELOG.md | 23 +++++++++++++++++++ administration/Dockerfile | 2 +- administration/package-lock.json | 4 ++-- administration/package.json | 2 +- cmd/rainbow/Dockerfile | 2 +- cmd/relay/Dockerfile | 2 +- hypercerts/version/version.go | 2 +- jetstream/Dockerfile | 4 ++-- jetstream/internal/version/version.go | 2 +- package-lock.json | 4 ++-- package.json | 2 +- 12 files changed, 36 insertions(+), 35 deletions(-) delete mode 100644 .changeset/jetstream-live-ingest-resilience.md diff --git a/.changeset/jetstream-live-ingest-resilience.md b/.changeset/jetstream-live-ingest-resilience.md deleted file mode 100644 index 908f7140..00000000 --- a/.changeset/jetstream-live-ingest-resilience.md +++ /dev/null @@ -1,22 +0,0 @@ ---- -'hypercerts-relay': patch ---- - -Jetstream live ingestion no longer stops silently when the Relay restarts. -Previously, a single HTTP 502 or 503 from the Relay or its proxy during a -reconnect ended the upstream stream. Jetstream then kept running without -ingesting until its next restart. Transient rejections (408, 425, 429 and 5xx) -are now retried with backoff from the same cursor. Any other end of the stream -exits the process with an `upstream stream ended unexpectedly` error so it is -restarted. Successful connections are logged as `upstream connected`. - -Catching up after a restart no longer overflows the per-account verify queue as -easily. Verification now uses 128 workers by default, up from 32, so each -account's queue holds 256 events instead of 64. Upstream reads are paced at 500 -frames per second by default. Tune these with `JETSTREAM_LIVE_VERIFY_PARALLELISM`, -`JETSTREAM_LIVE_UPSTREAM_READ_RATE` and `JETSTREAM_LIVE_UPSTREAM_READ_BURST`. -Pacing limits how fast Jetstream replays a large backlog. - -Repeated `verification failure` warnings for the same account are now logged -once per minute, followed by a `verification failures repeated` summary with -the count and the last error. The verifier failure metric is unchanged. diff --git a/CHANGELOG.md b/CHANGELOG.md index 21b5d2f8..17cce65e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,28 @@ +## 0.10.1 + +### Patch Changes + +- [#111](https://github.com/hypercerts-org/hypercerts-relay/pull/111) [`ff3e3a9`](https://github.com/hypercerts-org/hypercerts-relay/commit/ff3e3a9394dcaf825981a617ac33f90708019dbc) Thanks [@Ashex](https://github.com/Ashex)! - Jetstream live ingestion no longer stops silently when the Relay restarts. + Previously, a single HTTP 502 or 503 from the Relay or its proxy during a + reconnect ended the upstream stream. Jetstream then kept running without + ingesting until its next restart. Transient rejections (408, 425, 429 and 5xx) + are now retried with backoff from the same cursor. Any other end of the stream + exits the process with an `upstream stream ended unexpectedly` error so it is + restarted. Successful connections are logged as `upstream connected`. + + Catching up after a restart no longer overflows the per-account verify queue as + easily. Verification now uses 128 workers by default, up from 32, so each + account's queue holds 256 events instead of 64. Upstream reads are paced at 500 + frames per second by default. Tune these with `JETSTREAM_LIVE_VERIFY_PARALLELISM`, + `JETSTREAM_LIVE_UPSTREAM_READ_RATE` and `JETSTREAM_LIVE_UPSTREAM_READ_BURST`. + Pacing limits how fast Jetstream replays a large backlog. + + Repeated `verification failure` warnings for the same account are now logged + once per minute, followed by a `verification failures repeated` summary with + the count and the last error. The verifier failure metric is unchanged. + ## 0.10.0 ### Minor Changes diff --git a/administration/Dockerfile b/administration/Dockerfile index 9b4ead47..1ac1fbcf 100644 --- a/administration/Dockerfile +++ b/administration/Dockerfile @@ -1,5 +1,5 @@ # Shared bootstrap can also be built from Jetstream's isolated context. -ARG VERSION=0.10.0 +ARG VERSION=0.10.1 ARG COMMIT=unknown ARG DATE=unknown diff --git a/administration/package-lock.json b/administration/package-lock.json index 7a88e5be..906f1586 100644 --- a/administration/package-lock.json +++ b/administration/package-lock.json @@ -1,12 +1,12 @@ { "name": "@hypercerts/relay-administration", - "version": "0.10.0", + "version": "0.10.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@hypercerts/relay-administration", - "version": "0.10.0", + "version": "0.10.1", "license": "MIT OR Apache-2.0", "dependencies": { "@atproto/oauth-client-node": "0.5.4", diff --git a/administration/package.json b/administration/package.json index 4406efa5..92b1e33e 100644 --- a/administration/package.json +++ b/administration/package.json @@ -1,6 +1,6 @@ { "name": "@hypercerts/relay-administration", - "version": "0.10.0", + "version": "0.10.1", "license": "MIT OR Apache-2.0", "private": true, "type": "module", diff --git a/cmd/rainbow/Dockerfile b/cmd/rainbow/Dockerfile index 64dbadec..724c142a 100644 --- a/cmd/rainbow/Dockerfile +++ b/cmd/rainbow/Dockerfile @@ -1,4 +1,4 @@ -ARG VERSION=0.10.0 +ARG VERSION=0.10.1 ARG COMMIT=unknown ARG DATE=unknown diff --git a/cmd/relay/Dockerfile b/cmd/relay/Dockerfile index d1fdf716..53320b33 100644 --- a/cmd/relay/Dockerfile +++ b/cmd/relay/Dockerfile @@ -1,6 +1,6 @@ # Build from the repository root: # docker build -f cmd/relay/Dockerfile -t hypercerts-relay . -ARG VERSION=0.10.0 +ARG VERSION=0.10.1 ARG COMMIT=unknown ARG DATE=unknown diff --git a/hypercerts/version/version.go b/hypercerts/version/version.go index a1af553c..cb7a0a4f 100644 --- a/hypercerts/version/version.go +++ b/hypercerts/version/version.go @@ -3,4 +3,4 @@ package version // Version is the release version reported by Relay and Rainbow binaries. -var Version = "0.10.0" +var Version = "0.10.1" diff --git a/jetstream/Dockerfile b/jetstream/Dockerfile index f49a0551..4ffc7552 100644 --- a/jetstream/Dockerfile +++ b/jetstream/Dockerfile @@ -7,7 +7,7 @@ FROM --platform=$BUILDPLATFORM golang:1.27.3-bookworm AS build # Build-args set by `docker buildx build` (or stamped manually from CI). # VERSION/COMMIT/DATE are linked into internal/version via -ldflags. -ARG VERSION=0.10.0 +ARG VERSION=0.10.1 ARG COMMIT=unknown ARG DATE=unknown # Empty for canonical images. The disposable acceptance Compose build sets @@ -64,7 +64,7 @@ RUN GOOS=${TARGETOS} GOARCH=${TARGETARCH} go build -trimpath -buildvcs=false \ # ────────────────────────────────────────────────────────────────────────────── FROM gcr.io/distroless/static-debian12:nonroot AS runtime -ARG VERSION=0.10.0 +ARG VERSION=0.10.1 ARG COMMIT=unknown ARG DATE=unknown ARG SOURCE=https://github.com/bluesky-social/jetstream diff --git a/jetstream/internal/version/version.go b/jetstream/internal/version/version.go index 2dcd946d..96453095 100644 --- a/jetstream/internal/version/version.go +++ b/jetstream/internal/version/version.go @@ -9,7 +9,7 @@ package version // go build -ldflags "-X github.com/bluesky-social/jetstream/internal/version.Version=v1.2.3 ..." var ( // hypercerts: Keep the local default aligned with the repository release version. - Version = "0.10.0" + Version = "0.10.1" Commit = "unknown" Date = "unknown" ) diff --git a/package-lock.json b/package-lock.json index 5c8216fd..592b593b 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "hypercerts-relay", - "version": "0.10.0", + "version": "0.10.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "hypercerts-relay", - "version": "0.10.0", + "version": "0.10.1", "license": "MIT OR Apache-2.0", "devDependencies": { "@changesets/changelog-github": "1.0.1", diff --git a/package.json b/package.json index c1042272..2dbf6026 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "hypercerts-relay", - "version": "0.10.0", + "version": "0.10.1", "description": "Hypercerts-maintained Indigo Relay and Rainbow fork", "license": "MIT OR Apache-2.0", "private": true,