🐛 respawn cloudevents receiver after spontaneous Receive error - #222
🐛 respawn cloudevents receiver after spontaneous Receive error#222hbhushan3 wants to merge 1 commit into
Conversation
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: hbhushan3 The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
WalkthroughThe receiver goroutine is enhanced to implement recovery logic. When Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~22 minutes Suggested labels
Suggested reviewers
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 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. Review rate limit: 0/1 reviews remaining, refill in 60 minutes.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 the current code and only fix it if needed.
Inline comments:
In `@pkg/cloudevents/generic/clients/baseclient.go`:
- Around line 321-346: In the recovery branch where the receiver unexpectedly
exits (logging "cloudevents receiver exited unexpectedly, triggering
resubscribe"), call c.transport.Close(receiverCtx) to close the previous
transport before sending stopReceiverSignal and re-triggering subscribe; this
ensures OpenInbound's goroutines and connection are cleaned up (referencing
c.transport.Close, OpenInbound, stopReceiverSignal, receiverChan, subscribeChan
and receiverCtx) — place the Close call immediately prior to writing to
c.receiverChan and c.subscribeChan and handle any returned error with a log.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 16312172-f863-4eb0-80ed-f74fe0f63cc0
📒 Files selected for processing (2)
pkg/cloudevents/generic/clients/baseclient.gopkg/cloudevents/generic/clients/baseclient_test.go
| // Receive returned spontaneously (e.g. transient gRPC Canceled / | ||
| // Unavailable on the inbound subscribe stream) while the parent | ||
| // context is still alive and the transport-level ErrorChan never | ||
| // fired. Without this branch, startReceiving stays true, no future | ||
| // startReceiverSignal can respawn the receiver, and bundle status | ||
| // updates are silently dropped until the pod restarts. | ||
| // | ||
| // Recover by sending stopReceiverSignal (clears startReceiving and | ||
| // cancels the now-dead receiver context) and re-triggering the | ||
| // subscribe goroutine, which will retry Subscribe with backoff and | ||
| // emit a fresh startReceiverSignal on success. | ||
| logger.V(2).Info("cloudevents receiver exited unexpectedly, triggering resubscribe") | ||
| select { | ||
| case c.receiverChan <- stopReceiverSignal: | ||
| // Signal sent successfully | ||
| default: | ||
| // Receiver channel is unavailable, that's okay - don't block | ||
| logger.V(2).Info("stopReceiverSignal not sent, receiver channel unavailable") | ||
| } | ||
| select { | ||
| case c.subscribeChan <- struct{}{}: | ||
| // Signal sent successfully | ||
| default: | ||
| // Subscribe channel is unavailable, that's okay - don't block | ||
| logger.V(2).Info("subscribe signal not sent, subscribe channel is unavailable") | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== CloudEventTransport interface =="
rg -nC3 'type CloudEventTransport interface|Connect\(ctx context.Context\) error|Subscribe\(ctx context.Context\) error|Receive\(ctx context.Context, .*ReceiveHandlerFn\) error|Close\(ctx context.Context\) error|ErrorChan\(\) <-chan error' pkg/cloudevents
echo
echo "== Transport implementations: subscribe/receive/close/error handling =="
rg -nC4 'func .*Subscribe\(ctx context.Context\) error|func .*Receive\(ctx context.Context, .*ReceiveHandlerFn\) error|func .*Close\(ctx context.Context\) error|reconnectErrorChan|OpenInbound|heartbeat|ErrorChan\(' pkg/cloudevents
echo
echo "== Existing cleanup path around receive-side failures =="
rg -nC5 'reconnectErrorChan|Close\(\)|OpenInbound|context canceled|Unavailable|Canceled' pkg/cloudeventsRepository: open-cluster-management-io/sdk-go
Length of output: 50390
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== gRPC protocol OpenInbound and receiver implementation ==="
rg -nA30 'func \(p \*Protocol\) OpenInbound' pkg/cloudevents/generic/options/grpc/protocol/protocol.go
echo
echo "=== startEventsReceiver and heartbeat watcher in protocol ==="
rg -nA20 'func.*startEventsReceiver|startHeartbeatWatcher|cancelReceiverAndWatcher' pkg/cloudevents/generic/options/grpc/protocol/protocol.go
echo
echo "=== closeChan usage and Close cleanup ==="
rg -nB5 -A10 'closeChan' pkg/cloudevents/generic/options/grpc/protocol/protocol.goRepository: open-cluster-management-io/sdk-go
Length of output: 3815
Call transport.Close(ctx) in the recovery path before re-triggering subscribe.
When a transport's Receive() returns spontaneously (e.g., transient gRPC Unavailable while parent context is alive), the recovery code sends stopReceiverSignal and re-tickles subscribe. However, it does not call transport.Close(), leaving the previous OpenInbound() goroutine stuck.
In the gRPC protocol implementation, OpenInbound() blocks at a select waiting for either subCtx.Done() or closeChan (line 145). If Close() is not called, closeChan is never signaled, so OpenInbound() never reaches the cancel() call (line 149) that terminates the receiver and heartbeat watcher goroutines. This results in:
- The old subscription context remaining active indefinitely
- The healthChecker goroutine continuing to run on the stale context
- The old gRPC connection never being closed (line 152)
Add c.transport.Close(receiverCtx) before sending the resubscribe signal to ensure proper cleanup of the previous subscription's internal goroutines and connection.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pkg/cloudevents/generic/clients/baseclient.go` around lines 321 - 346, In the
recovery branch where the receiver unexpectedly exits (logging "cloudevents
receiver exited unexpectedly, triggering resubscribe"), call
c.transport.Close(receiverCtx) to close the previous transport before sending
stopReceiverSignal and re-triggering subscribe; this ensures OpenInbound's
goroutines and connection are cleaned up (referencing c.transport.Close,
OpenInbound, stopReceiverSignal, receiverChan, subscribeChan and receiverCtx) —
place the Close call immediately prior to writing to c.receiverChan and
c.subscribeChan and handle any returned error with a log.
2 participants
What
Recover the
baseClientreceiver lifecycle whentransport.Receivereturns an error spontaneously while both the parent context and the receiver context are still alive.When this happens, the receiver goroutine launched inside
(*baseClient).subscribe(pkg/cloudevents/generic/clients/baseclient.go) now:stopReceiverSignalonreceiverChan— clears thestartReceivingflag and cancels the dead receiver context.subscribeChan— the subscribe goroutine retriestransport.Subscribe()(withDelayFnbackoff) and, on success, emits a freshstartReceiverSignalthat respawns the receiver.Both sends are non-blocking (
select/default) and the target channels are size-1 buffered, so the recovery path is idempotent: if the transport'sErrorChanalso fires for the same failure, the duplicate signals are coalesced and no extra reconnect is performed.A new white-box test,
TestReceiverRecoveryAfterSpontaneousReceiveError, drives a fakeCloudEventTransportwhose firstReceivecall returnsrpc error: code = Canceled desc = context canceledand whoseErrorChanis never written. It asserts thatReceiveis invoked at least twice within 10s — proving the recovery path runs end-to-end.Why
Production symptom
In ARO HCP,
aro-hcp-backendpods (which embed this SDK as the maestro consumer) silently stopped processing maestro bundle status updates. A single log line was emitted at the moment of failure and then the pod went permanently silent on the receive path until it was manually restarted hours later:The line originates here: https://github.com/open-cluster-management-io/sdk-go/blob/v1.2.0/pkg/cloudevents/generic/clients/baseclient.go#L308
After this point, no further
Receiveinvocation was ever made. Both maestro consumer controllers (cluster-scoped and nodepool-scoped) sharing the brokenbaseClientstopped making progress; cluster/nodepool ARM resources stayed in transient states until the pod was rolled.Internal tracking: AROSLSRE-755
Root cause
Inside
(*baseClient).subscribe, thecase startReceiverSignal:branch spawns:If
Receivereturns on its own, this goroutine exits and nothing restarts it. The state machine relies on eitherstopReceiverSignal(sent by the connection monitor when it sees a transport error) orstartReceiverSignal(sent by the subscribe goroutine after a successfulSubscribe) to resetstartReceivingand respawn — but neither will fire, because:startReceivingstaystrue, so subsequentstartReceiverSignals are ignored (baseclient.go:284-289).subscribeChan.transport.ErrorChan(). For the gRPC transport, errors from a spontaneously-failing inbound subscribe stream are reported viaselect { case errorChan <- err: default: }over an unbufferederrorChan(pkg/cloudevents/generic/options/grpc/protocol/protocol.go:178-183,pkg/cloudevents/generic/options/grpc/agentoptions.go:29). The single reader is only available during the steady-state select; if it is mid-Connect/Close/DelayFn, the error is silently dropped — which is exactly what the production traces show.This is a different gap than #183 (which also touches reconnect, but addresses the connection-monitor reader path; it does not respawn a receiver goroutine that exited on its own).
Reproduction (deterministic)
The new regression test drives the exact production failure mode — a transport whose
Receivereturns a transient error whileErrorChanstays silent — and asserts that the client recovers by invokingReceiveagain. Without this PR the test fails after its 10s deadline; with this PR it passes in tens of milliseconds.Before this PR (run against
main@ v1.3.0, with only the new test cherry-picked):receive=1 subscribe=1is the smoking gun: after the spontaneous error the receiver goroutine exited,startReceivingstayedtrue, no newReceivewas ever started — exactly the behavior observed in production.After this PR:
Full package:
ok ... 14.037s.go vet ./...andgo build ./...are clean.Notes for reviewers
ctx.Err() == nil && receiverCtx.Err() == nil, so genuine shutdown / explicitstopReceiverSignalpaths are unaffected.select/defaultagainst size-1 buffered channels (matching the existing house-style atbaseclient.go:117-141,:218-224,:228-234), so the new path cannot deadlock and cannot double-trigger ifErrorChanalso fires for the same failure.Summary by CodeRabbit
Release Notes
Bug Fixes
Tests