feat(search-sync-worker): read HR JetStream domain from HR_JETSTREAM_DOMAIN#466
Merged
Merged
Conversation
Read the HR JetStream domain (OrgSyncStream) from HR_JETSTREAM_DOMAIN so a worker at one site can consume the HR stream in a remote NATS domain, while local collections keep the shared otel-traced JetStream context. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Sgh3aGBn7fPpK3kF13xqqc
… from env Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Sgh3aGBn7fPpK3kF13xqqc
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Sgh3aGBn7fPpK3kF13xqqc
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Sgh3aGBn7fPpK3kF13xqqc
…DOMAIN Route the spotlight-org (OrgSyncStream) consumer to a domain-scoped JetStream context when HR_JETSTREAM_DOMAIN is set, so a worker at one site can consume the HR stream in a remote NATS domain. Empty domain keeps the shared otel-traced context, matching current behavior. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Sgh3aGBn7fPpK3kF13xqqc
The otelBatch adapter re-channels messages via a goroutine that blocks on an unbuffered send; document at the drain site that batch.Messages() must be consumed to completion so a future early-break can't leak it or stall shutdown. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Sgh3aGBn7fPpK3kF13xqqc
📝 WalkthroughWalkthroughThis PR adds design/plan documentation and search-sync-worker code enabling the HR (spotlight-org / OrgSyncStream) consumer to fetch from a remote NATS JetStream domain via a new HR_JETSTREAM_DOMAIN env var. It introduces a msgFetcher/msgBatch seam with raw and otel adapters, updates main.go wiring and runConsumer's signature, and adds tests. ChangesHR JetStream Domain Env Support
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant main as main.go startup
participant hrJS as Domain-scoped JetStream
participant js as Shared otel JetStream
participant fetcher as msgFetcher
participant runConsumer
main->>hrJS: NewWithDomain (if HR_JETSTREAM_DOMAIN set)
alt HR stream and hrJS present
main->>hrJS: CreateOrUpdateConsumer
main->>fetcher: wrap as rawConsumerAdapter
else other collections
main->>js: CreateOrUpdateConsumer
main->>fetcher: wrap as otelConsumerAdapter
end
main->>runConsumer: start(fetcher)
runConsumer->>fetcher: Fetch()
fetcher-->>runConsumer: msgBatch
runConsumer->>runConsumer: range batch.Messages() and add(msg)
Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@search-sync-worker/consumer_source.go`:
- Around line 34-40: Both Fetch methods in rawConsumerAdapter and
otelConsumerAdapter return a bare underlying error, which violates the repo’s
wrapping rule. Update each Fetch implementation to wrap the fetch failure with
contextual text using fmt.Errorf(...: %w, err) before returning, while keeping
the successful path unchanged. Use the Fetch methods on rawConsumerAdapter and
otelConsumerAdapter as the fix points so the error chain remains intact for
existing tests.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 3a14d88d-3101-4896-af54-8c7edf190c43
📒 Files selected for processing (7)
docs/superpowers/plans/2026-07-07-hr-jetstream-domain-env.mddocs/superpowers/specs/2026-07-07-hr-jetstream-domain-env-design.mdsearch-sync-worker/config_test.gosearch-sync-worker/consumer_source.gosearch-sync-worker/consumer_source_test.gosearch-sync-worker/deploy/docker-compose.ymlsearch-sync-worker/main.go
Comment on lines
+34
to
+40
| func (a rawConsumerAdapter) Fetch(n int, opts ...jetstream.FetchOpt) (msgBatch, error) { | ||
| b, err := a.c.Fetch(n, opts...) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return b, nil | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Bare err returns violate the repo's error-wrapping rule.
Both rawConsumerAdapter.Fetch and otelConsumerAdapter.Fetch return the underlying err unwrapped. The coding guidelines require wrapping errors with context via fmt.Errorf("...: %w", err) and never returning a bare err. Since %w preserves the chain, this won't break the require.ErrorIs assertions in the tests.
As per coding guidelines: "Wrap errors with context using fmt.Errorf("...: %w", err); never return a bare err".
🐛 Proposed fix
func (a rawConsumerAdapter) Fetch(n int, opts ...jetstream.FetchOpt) (msgBatch, error) {
b, err := a.c.Fetch(n, opts...)
if err != nil {
- return nil, err
+ return nil, fmt.Errorf("raw consumer fetch: %w", err)
}
return b, nil
} func (a otelConsumerAdapter) Fetch(n int, opts ...jetstream.FetchOpt) (msgBatch, error) {
b, err := a.c.Fetch(n, opts...)
if err != nil {
- return nil, err
+ return nil, fmt.Errorf("otel consumer fetch: %w", err)
}
return otelBatch{b}, nil
}Also applies to: 47-53
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@search-sync-worker/consumer_source.go` around lines 34 - 40, Both Fetch
methods in rawConsumerAdapter and otelConsumerAdapter return a bare underlying
error, which violates the repo’s wrapping rule. Update each Fetch implementation
to wrap the fetch failure with contextual text using fmt.Errorf(...: %w, err)
before returning, while keeping the successful path unchanged. Use the Fetch
methods on rawConsumerAdapter and otelConsumerAdapter as the fix points so the
error chain remains intact for existing tests.
Source: Coding guidelines
GITMateuszCharczuk
approved these changes
Jul 8, 2026
GITMateuszCharczuk
left a comment
GITMateuszCharczuk
left a comment
Collaborator
There was a problem hiding this comment.
LGTM. Thanks!
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What & why
search-sync-worker'sspotlight-orgcollection consumes HR org data fromOrgSyncStream(HR_CENTRAL_SITE_ID), which is owned byhr-syncerand can live in a different NATS JetStream domain (a remote cluster in the supercluster). A worker at site A must be able to create a durable consumer on, and fetch from, the HR stream in site B's domain.Accessing a stream in another JetStream domain requires a JetStream context whose API prefix targets that domain (
$JS.<domain>.API.*), fixed at construction viajetstream.NewWithDomain. Todaymain.gobuilds a singleoteljetstream.New(nc)context used for all collections — local (INBOX / MESSAGES_CANONICAL) and HR — and one context can only serve one domain.This PR lets the HR domain be supplied via
HR_JETSTREAM_DOMAIN, routing only the HR consumer to a domain-scoped context while local collections keep the shared context. Unset behaves exactly as today.Approach
consumer_source.go). A minimalmsgFetcher/msgBatchinterface that both consumer flavors satisfy, normalized to rawjetstream.Msg(whichhandler.Addalready consumes):rawConsumerAdapterwraps a rawjetstream.Consumer(its batch already yields rawjetstream.Msg— pass-through).otelConsumerAdapter/otelBatchwrapoteljetstream.Consumer, unwrapping eachoteljetstream.Msgto its embedded rawjetstream.Msg.runConsumerrefactor. Parameter widened fromoteljetstream.ConsumertomsgFetcher; loop unwrapadd(msg.Msg)→add(msg); call site wrapsotelConsumerAdapter{cons}. Behavior-preserving.main.go). NewHRJetStreamDomainconfig field (env:"HR_JETSTREAM_DOMAIN" envDefault:""). When set, buildhrJS := jetstream.NewWithDomain(nc.NatsConn(), domain)(fail-fast on error). The collection whose stream isOrgSyncStream(streamCfg.Name == hrName) is created againsthrJSviarawConsumerAdapter; every other collection uses the shared otel-tracedjs.Why the HR context is raw (not otel-wrapped)
oteljetstream(instrumentation-go v0.2.0) exposes onlyNew(conn)— no domain variant — so the domain-scoped context comes from the rawjetstreampackage. This loses nothing in use:runConsumeralready unwraps to rawjetstream.Msgandhandler.Addnever reads the per-message otel trace context, so the consume-side span for this one stream was already discarded.Behavior when unset
HR_JETSTREAM_DOMAIN=""→hrJS == nil→ HR collection takes the shared-jspath viaotelConsumerAdapter. The only delta vs. today is that local consumers now flow through a pure-unwrap adapter over the same messages — observably identical. HR stream bootstrap is untouched (already skipped; owned byhr-syncer/ops).Testing
HR_JETSTREAM_DOMAIN(default""and set value).make test SERVICE=search-sync-workergreen (race on);make build/make lintclean.TestSearchSyncSpotlightOrg_Integration(empty-domain path) stays green.Ops
search-sync-worker/deploy/docker-compose.ymlgets a commentedHR_JETSTREAM_DOMAINexample. Set it in deployments wherehr-syncerruns in a different domain; leave unset for single-cluster.No client-API doc impact (internal consumer wiring, not a
chat.user.handler).🤖 Generated with Claude Code
Generated by Claude Code
Summary by CodeRabbit
New Features
Bug Fixes
Tests