🐛 Fix PubSub race condition with sequential event processing. - #215
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
WalkthroughIntroduces a queued, single-worker message processing model for Pub/Sub that sequentially runs the receive handler via a new Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes 🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
📝 Coding Plan
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
🧹 Nitpick comments (2)
test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go (2)
119-146: Good concurrent publish test structure.The test correctly publishes two events concurrently via goroutines and waits for completion. This exercises the sequential channel fix path.
For stronger race condition validation, consider running this test with the
-raceflag in CI to detect data races at runtime. The current assertions verify functional correctness but won't catch race detector violations.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go` around lines 119 - 146, The concurrent-publish test labeled "THE RACE CONDITION TEST" (the ginkgo.By call with "sending create_request and resync_response concurrently") should be executed under the Go race detector to catch data races; update the CI/test invocation that runs this integration test to include the -race flag (for example run go test ./... -race for the integration suite) or add a dedicated CI job that runs this package with -race so the goroutine interactions around sourceCloudEventsClient.Publish and the errChan handling are validated by the race detector.
67-68: Consider polling for readiness instead of fixed delay.Fixed 1-second waits may be insufficient in slow CI environments or excessive in fast environments. Consider polling for agent readiness (e.g., checking a health endpoint or store state) to make the test more robust.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go` around lines 67 - 68, Replace the fixed one-second sleep ("<-time.After(time.Second)") used to wait for "agent ready" with a polling loop that repeatedly checks agent readiness until a timeout; implement a helper like isAgentReady() (or call the agent health endpoint / inspect store state) in a loop using time.Ticker and a total timeout (e.g., 10-30s) and fail the test if the timeout elapses—this makes the test in manifestworkclients_resync_pubsub_test.go deterministic across CI speeds.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@pkg/cloudevents/generic/options/v2/pubsub/transport.go`:
- Around line 156-167: The goroutine draining eventChan can outlive Receive
because it only stops on ctx.Done(); add a dedicated done signal to stop it when
Receive returns: create a done channel (e.g., done := make(chan struct{})) in
the Receive scope, add a case <-done to the goroutine's select alongside
ctx.Done(), and ensure Receive closes(done) before returning (including the
branch where errChan yields an error) so the goroutine exits immediately and
doesn't leak while still draining eventChan; reference symbols: Receive,
eventChan, errChan, ctx.Done(), fn, em.msg.Ack().
---
Nitpick comments:
In `@test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go`:
- Around line 119-146: The concurrent-publish test labeled "THE RACE CONDITION
TEST" (the ginkgo.By call with "sending create_request and resync_response
concurrently") should be executed under the Go race detector to catch data
races; update the CI/test invocation that runs this integration test to include
the -race flag (for example run go test ./... -race for the integration suite)
or add a dedicated CI job that runs this package with -race so the goroutine
interactions around sourceCloudEventsClient.Publish and the errChan handling are
validated by the race detector.
- Around line 67-68: Replace the fixed one-second sleep
("<-time.After(time.Second)") used to wait for "agent ready" with a polling loop
that repeatedly checks agent readiness until a timeout; implement a helper like
isAgentReady() (or call the agent health endpoint / inspect store state) in a
loop using time.Ticker and a total timeout (e.g., 10-30s) and fail the test if
the timeout elapses—this makes the test in
manifestworkclients_resync_pubsub_test.go deterministic across CI speeds.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 9ea7f589-3aa9-44d2-bfa6-0c64077e306f
📒 Files selected for processing (2)
pkg/cloudevents/generic/options/v2/pubsub/transport.gotest/integration/cloudevents/manifestworkclients_resync_pubsub_test.go
0df40d6 to
941251d
Compare
There was a problem hiding this comment.
🧹 Nitpick comments (2)
test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go (2)
67-68: Consider replacing arbitrary sleeps with readiness polling.The
<-time.After(time.Second)pattern introduces fixed delays that may be insufficient under load or wasteful when systems are ready sooner. For more reliable tests, consider polling for readiness:// Example: poll until agent is ready gomega.Eventually(func() bool { // check agent readiness condition return agentClientHolder != nil && agentClientHolder.IsReady() }, 5*time.Second, 100*time.Millisecond).Should(gomega.BeTrue())If no readiness API exists,
time.Sleep(time.Second)is at least clearer in intent than the channel receive pattern.Also applies to: 90-91
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go` around lines 67 - 68, Replace the arbitrary receive-from-time.After pattern used to wait for the agent (the "<-time.After(time.Second)" call) with a readiness poll using gomega.Eventually: call a small closure that checks the agent readiness (e.g., agentClientHolder != nil && agentClientHolder.IsReady()) and assert it becomes true within a timeout (and short poll interval); if no readiness API exists, replace the channel receive with an explicit time.Sleep(time.Second) to make intent clearer; apply the same change to the other similar sleep at the later wait in this test.
128-146: Consider adding a barrier for stronger concurrency guarantees.The current pattern spawns two goroutines, but there's no guarantee they execute simultaneously. One goroutine could complete before the other starts, reducing the effectiveness of the race condition test.
A barrier pattern ensures both goroutines start their publish calls at the same instant:
♻️ Suggested enhancement for true concurrent execution
errChan := make(chan error, 2) +startBarrier := make(chan struct{}) // Send create_request go func() { + <-startBarrier // wait for signal err := sourceCloudEventsClient.Publish(ctx, createRequest, work) errChan <- err }() // Send resync_response go func() { + <-startBarrier // wait for signal err := sourceCloudEventsClient.Publish(ctx, resyncResponse, work.DeepCopy()) errChan <- err }() +close(startBarrier) // release both goroutines simultaneously🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go` around lines 128 - 146, The two publish goroutines (calls to sourceCloudEventsClient.Publish with createRequest and resyncResponse on work/work.DeepCopy()) should be coordinated with a start barrier to ensure they actually begin Publish at the same time: introduce a start channel (e.g. start := make(chan struct{})) and a readiness sync (e.g. a small readyWG or readyChan) so each goroutine signals it is ready, then close(start) to release both goroutines to call Publish concurrently; keep the existing errChan to collect errors and the existing expectation checks. Ensure both goroutines wait on <-start immediately before calling sourceCloudEventsClient.Publish.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go`:
- Around line 67-68: Replace the arbitrary receive-from-time.After pattern used
to wait for the agent (the "<-time.After(time.Second)" call) with a readiness
poll using gomega.Eventually: call a small closure that checks the agent
readiness (e.g., agentClientHolder != nil && agentClientHolder.IsReady()) and
assert it becomes true within a timeout (and short poll interval); if no
readiness API exists, replace the channel receive with an explicit
time.Sleep(time.Second) to make intent clearer; apply the same change to the
other similar sleep at the later wait in this test.
- Around line 128-146: The two publish goroutines (calls to
sourceCloudEventsClient.Publish with createRequest and resyncResponse on
work/work.DeepCopy()) should be coordinated with a start barrier to ensure they
actually begin Publish at the same time: introduce a start channel (e.g. start
:= make(chan struct{})) and a readiness sync (e.g. a small readyWG or readyChan)
so each goroutine signals it is ready, then close(start) to release both
goroutines to call Publish concurrently; keep the existing errChan to collect
errors and the existing expectation checks. Ensure both goroutines wait on
<-start immediately before calling sourceCloudEventsClient.Publish.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 83d9535b-cbeb-4b97-9b9f-e932cbd2163b
📒 Files selected for processing (2)
pkg/cloudevents/generic/options/v2/pubsub/transport.gotest/integration/cloudevents/manifestworkclients_resync_pubsub_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
- pkg/cloudevents/generic/options/v2/pubsub/transport.go
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@pkg/cloudevents/generic/options/v2/pubsub/transport.go`:
- Around line 141-178: The current design enqueues messages from
receiveFromSubscriber into eventChan and ACKs them in the separate goroutine in
pubsubTransport.Receive, which violates the cloud pubsub Receive callback
contract and bypasses flow control; change receiveFromSubscriber and
pubsubTransport.Receive so ACK/NACK is called inside the Subscriber.Receive
callback (i.e., inside receiveFromSubscriber) instead of in the processing
goroutine, and remove the external Ack call on em.msg.Ack(); to preserve
sequential processing use a shared sync.Mutex (or similar) inside the callback
to serialize handler invocation (call fn(ctx, evt) while holding the mutex) so
you don't need eventChan buffering, and ensure receiveFromSubscriber checks
ctx.Done() and returns immediately on errors so other subscriber goroutines
can't block trying to enqueue — also ensure errChan sends/errors are handled
non-blocking so Receive can return cleanly when one subscriber fails.
In `@test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go`:
- Around line 128-140: The two Publish goroutines (calling
sourceCloudEventsClient.Publish with createRequest and resyncResponse
respectively) must be synchronized so they start at the same time; change the
test to have both goroutines wait on a shared start signal (e.g., a startCh or
sync.WaitGroup) before calling Publish and then release that signal to run them
concurrently, while still sending errors into errChan; ensure you reference the
existing errChan, createRequest, resyncResponse, work.DeepCopy(), and
sourceCloudEventsClient.Publish when implementing the start synchronization to
reproduce the cross-subscription race deterministically.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: fae4ae3e-ec5d-4dde-a8f6-fbcbe10fdae1
📒 Files selected for processing (2)
pkg/cloudevents/generic/options/v2/pubsub/transport.gotest/integration/cloudevents/manifestworkclients_resync_pubsub_test.go
Signed-off-by: Morven Cao <lcao@redhat.com>
|
/hold |
There was a problem hiding this comment.
🧹 Nitpick comments (2)
test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go (1)
67-68: Consider using readiness polling instead of fixed sleeps.The fixed 1-second waits (
<-time.After(time.Second)) could make tests flaky under load or unnecessarily slow in fast environments. If the agent and source client expose readiness signals, polling them would be more robust.Also applies to: 90-91
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go` around lines 67 - 68, The test uses fixed sleeps ("<-time.After(time.Second)") which can make it flaky; replace these sleeps in manifestworkclients_resync_pubsub_test.go (the occurrences of <-time.After(time.Second) around the agent and source client startup) with a readiness polling loop that checks the agent/source client readiness (e.g., call the agent readiness method or probe the source client's health/connection) using a bounded retry loop or wait.PollImmediate with a context timeout; poll until the ready condition returns true or the timeout elapses and fail the test on timeout so tests are robust and faster in fast environments.pkg/cloudevents/generic/options/v2/pubsub/transport.go (1)
176-184: Document the throughput trade-off of holding the mutex during handler execution.Holding the mutex while calling
fn(ctx, evt)means all concurrent message processing (from both subscribers and allNumGoroutinesper subscriber) is serialized. Per theReceiveHandlerFncontract (context snippet 1), blocking operations in the handler will now block all message reception globally, not just the current subscription.This is intentional for correctness, but consider adding a brief comment noting the throughput trade-off, especially since
NumGoroutines(context snippet 3) can spawn multiple goroutines that will all contend for this lock.📝 Suggested documentation
// Lock to ensure sequential processing across both subscribers. // This prevents race conditions when concurrent events for the same // resource arrive on different subscriptions. + // Note: This serializes ALL message processing, reducing throughput. + // If high throughput is needed, consider per-resource locking instead. mu.Lock() defer mu.Unlock() fn(ctx, evt)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/cloudevents/generic/options/v2/pubsub/transport.go` around lines 176 - 184, Add a short inline comment before mu.Lock documenting that holding mu across the call to fn(ctx, evt) serializes processing across both subscribers and all goroutines created by NumGoroutines (per the ReceiveHandlerFn contract), reducing throughput when handlers perform blocking work; mention that this is intentional for correctness (to avoid races) and note that a future optimization could instead lock only the critical section if handler-level parallelism is required.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@pkg/cloudevents/generic/options/v2/pubsub/transport.go`:
- Around line 176-184: Add a short inline comment before mu.Lock documenting
that holding mu across the call to fn(ctx, evt) serializes processing across
both subscribers and all goroutines created by NumGoroutines (per the
ReceiveHandlerFn contract), reducing throughput when handlers perform blocking
work; mention that this is intentional for correctness (to avoid races) and note
that a future optimization could instead lock only the critical section if
handler-level parallelism is required.
In `@test/integration/cloudevents/manifestworkclients_resync_pubsub_test.go`:
- Around line 67-68: The test uses fixed sleeps ("<-time.After(time.Second)")
which can make it flaky; replace these sleeps in
manifestworkclients_resync_pubsub_test.go (the occurrences of
<-time.After(time.Second) around the agent and source client startup) with a
readiness polling loop that checks the agent/source client readiness (e.g., call
the agent readiness method or probe the source client's health/connection) using
a bounded retry loop or wait.PollImmediate with a context timeout; poll until
the ready condition returns true or the timeout elapses and fail the test on
timeout so tests are robust and faster in fast environments.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 9f3088d8-404a-4b84-86f5-a3bdd199d3cd
📒 Files selected for processing (2)
pkg/cloudevents/generic/options/v2/pubsub/transport.gotest/integration/cloudevents/manifestworkclients_resync_pubsub_test.go
| // Lock to ensure sequential processing across both subscribers. | ||
| // This prevents race conditions when concurrent events for the same | ||
| // resource arrive on different subscriptions. | ||
| mu.Lock() |
There was a problem hiding this comment.
I am not sure having lock is good here. Can we consider put this into a threadsafe queue, and start another gorouting to read from queue and call handler?
There was a problem hiding this comment.
code updated to use a separate goroutine to process messages/events using a queue, but ACK message is still called in Receive function to respect Pub/Sub flow control by only Ack'ing after the handler completes.
Signed-off-by: Morven Cao <lcao@redhat.com>
| <-work.done | ||
|
|
||
| // now that processing is complete, Ack the message. | ||
| msg.Ack() |
There was a problem hiding this comment.
do we need to block here. What if we directly put msg into the queue, decode msg in processMessages and ACK after handler?
There was a problem hiding this comment.
we can't ACK message in processMessages, because it's in another go routine; Pub/Sub requires Ack() or Nack() to be called within the Receive handler, not from another goroutine, see:
Do not call Ack or Nack from a different goroutine. The Receive method handles concurrency for you, and all acknowledgment should happen synchronously in the callback.
— Pub/Sub Go Client: Subscription.Receive
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: morvencao, qiujian16 The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
|
/unhold |
e885cce
into
open-cluster-management-io:main
2 participants
Summary
Fix PubSub race condition by funneling events from both subscriptions through a sequential channel to prevent concurrent processing.
Related issue(s)
Fixes #
Summary by CodeRabbit
Bug Fixes
Tests