Conversation
skipi
marked this pull request as ready for review
September 2, 2026 22:04
skipi
requested review from
DamjanBecirovic,
adbatista,
dexyk and
loadez
as code owners
September 2, 2026 22:04
skipi
force-pushed
the
mk/velocity/emitter-persistent-publisher
branch
from
September 3, 2026 04:59
1aca3a2 to
96effba
Compare
skipi
force-pushed
the
mk/velocity/emitter-persistent-publisher
branch
from
September 3, 2026 07:32
96effba to
6c46ff2
Compare
skipi
added a commit
that referenced
this pull request
Sep 3, 2026
Replace the hand-rolled persistent publisher with go-tackle's own Publisher, addressing the review on #1213. - Bump go-tackle to v0.0.0-20231226193542-c913a4af4f94 (matches self_hosted_hub). Its Publisher dials with a bounded 5s timeout (net.DialTimeout + handshake deadline) and opens a fresh channel per publish over one shared connection, so we drop the hand-rolled ensurePublisher/resetPublisher and the amqp091 default 30s+30s dial. - Connect once at the start of each tick and Close() at the end, instead of holding a connection idle for ~24h between daily ticks. One connection per run: no Cloud NAT idle-reap, no orphan-on-rollout, and the unpinned heartbeat becomes irrelevant. The per-publish channel close-ok plus the end-of-tick connection close restore a flush barrier, so a mid-flight connection death surfaces as an error rather than a silent publish_message{success}. - Publish via PublishWithContext with a bounded 15s ctx instead of Publish()'s context.Background(), which would otherwise retry every 1s forever against a down broker. Keep a mutex, but scoped to only the PublishWithContext call. go-tackle's shared Publisher has a data race on its reconnect path: reconnect() reassigns p.connectOnce = sync.Once{} under reconnectionLock (publisher.go:232) while getConnection() reads that same Once via .Do() (publisher.go:179) without the lock. The emitter is the first caller to fan multiple goroutines at one shared Publisher, so it is the first to hit it; -race flags it deterministically once the broker drops mid-tick. The mutex serialises only the fast channel-open/publish/channel-close — the dial happens once per tick in openPublisher(), outside the lock, and describeProject()'s gRPC calls stay parallel across the worker pool. A go-tackle fix for the connectOnce race can let us drop the mutex later. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
skipi
added a commit
that referenced
this pull request
Sep 3, 2026
Applies the agreed points from a 4-reviewer pass on #1213: - Skip the broker-backed tests when RABBITMQ_URL is unset instead of hard-failing, so `go test ./...` on a host without the compose stack skips rather than reporting spurious failures. - Add TestConcurrentReconnectAfterLiveDropIsRaceFree: establish a live connection, drop it, then fan 50 concurrent publishes that must all reconnect. Verified load-bearing — with the publish mutex removed it trips -race in go-tackle's reconnectAndPublish (the connectOnce reset outside reconnectionLock); with the mutex it is clean and re-dials exactly once. The prior suite could pass with the mutex removed, so it did not guard the headline fix; this test does. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
loadez
approved these changes
Sep 9, 2026
tackle.PublishMessage dials, handshakes and closes a connection per message. The emitter publishes every pending metric on each run with 20 workers, so a run with thousands of metrics opened thousands of short-lived connections; on a node behind a shared NAT the ports linger in TIME_WAIT and run out, the next dials time out, and the emitter keeps retrying into the same wall. Keep one publisher (connection + channel + exchange declared once) for the emitter's lifetime, guarded by a mutex since the workers publish concurrently. On a publish error reconnect once and retry; if that fails, drop the publisher so the next call reconnects. Co-Authored-By: Claude Code <noreply@anthropic.com>
The rationale lives in the previous commit message and the PR body. Co-Authored-By: Claude Code <noreply@anthropic.com>
Replace the hand-rolled persistent publisher with go-tackle's own Publisher, addressing the review on #1213. - Bump go-tackle to v0.0.0-20231226193542-c913a4af4f94 (matches self_hosted_hub). Its Publisher dials with a bounded 5s timeout (net.DialTimeout + handshake deadline) and opens a fresh channel per publish over one shared connection, so we drop the hand-rolled ensurePublisher/resetPublisher and the amqp091 default 30s+30s dial. - Connect once at the start of each tick and Close() at the end, instead of holding a connection idle for ~24h between daily ticks. One connection per run: no Cloud NAT idle-reap, no orphan-on-rollout, and the unpinned heartbeat becomes irrelevant. The per-publish channel close-ok plus the end-of-tick connection close restore a flush barrier, so a mid-flight connection death surfaces as an error rather than a silent publish_message{success}. - Publish via PublishWithContext with a bounded 15s ctx instead of Publish()'s context.Background(), which would otherwise retry every 1s forever against a down broker. Keep a mutex, but scoped to only the PublishWithContext call. go-tackle's shared Publisher has a data race on its reconnect path: reconnect() reassigns p.connectOnce = sync.Once{} under reconnectionLock (publisher.go:232) while getConnection() reads that same Once via .Do() (publisher.go:179) without the lock. The emitter is the first caller to fan multiple goroutines at one shared Publisher, so it is the first to hit it; -race flags it deterministically once the broker drops mid-tick. The mutex serialises only the fast channel-open/publish/channel-close — the dial happens once per tick in openPublisher(), outside the lock, and describeProject()'s gRPC calls stay parallel across the worker pool. A go-tackle fix for the connectOnce race can let us drop the mutex later. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Add pkg/emitter/pending_metrics_test.go, driving the publisher through tackle's PublisherOptions.ConnectFunc hook against the compose RabbitMQ: - TestPublisherDialsOncePerTick: 200 concurrent publishes over one connection dial exactly once. - TestPublisherReconnectsAfterConnectionDrop: a dropped connection is re-dialled and the retry succeeds. - TestPublisherClosesConnectionAtEndOfTick: closePublisher() actually closes the underlying connection. - TestOpenPublisherFailsFastWhenBrokerIsUnreachable: a blackholed broker aborts within the bounded connection timeout, not amqp091's 30s default. - TestPublishStaysBoundedWhenBrokerDiesMidTick: 20 concurrent publishes against a broker that dies mid-tick return within a bounded wall-clock instead of N x timeout. The suite passes under CI's make test (-p 1) and, with the scoped publish mutex, is clean under -race (run natively; the amd64 image's ThreadSanitizer aborts under emulation). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Applies the agreed points from a 4-reviewer pass on #1213: - Skip the broker-backed tests when RABBITMQ_URL is unset instead of hard-failing, so `go test ./...` on a host without the compose stack skips rather than reporting spurious failures. - Add TestConcurrentReconnectAfterLiveDropIsRaceFree: establish a live connection, drop it, then fan 50 concurrent publishes that must all reconnect. Verified load-bearing — with the publish mutex removed it trips -race in go-tackle's reconnectAndPublish (the connectOnce reset outside reconnectionLock); with the mutex it is clean and re-dials exactly once. The prior suite could pass with the mutex removed, so it did not guard the headline fix; this test does. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
skipi
force-pushed
the
mk/velocity/emitter-persistent-publisher
branch
from
September 15, 2026 08:41
3bad9dd to
7f3abc0
Compare
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
2 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.
tackle.PublishMessageopens and closes a new AMQP connection per message. The emitter publishes every pending metric on each cron tick with 20 workers, so one tick can open thousands of short-lived connections — enough to exhaust the egress NAT's per-node ports, after which publishes time out and the node's other outbound traffic is throttled.Hold one publisher (connection, channel, exchange declared once) for the emitter's lifetime, serialized with a mutex; on a publish error reconnect once and retry.
Tested:
go build,go vet, andpkg/e2e(drives the emitter against the compose RabbitMQ) pass.