Deploy v0.10.1 to Staging

.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:

CHANGELOG.md

@@ -1,5 +1,28 @@
 <!-- hypercerts: project-owned release history; review this file during every Indigo upstream sync. -->
 
+## 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

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.

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
 

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",

cmd/rainbow/Dockerfile

@@ -1,4 +1,4 @@
-ARG VERSION=0.10.0
+ARG VERSION=0.10.1
 ARG COMMIT=unknown
 ARG DATE=unknown
 

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
 

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"

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

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

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"),

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,11 +250,26 @@ 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
 }
 
 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
+	}
 }

jetstream/internal/ingest/live/consumer.go

@@ -74,6 +74,11 @@ type Consumer struct {
 	// processBatch in isolation.
 	client atomic.Pointer[streaming.Client]
 
+	// 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
+
 	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,8 @@ 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
 			// will either reconnect or yield the next batch.

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