fix: send agent processed-event ACKs directly to avoid replay loops#1030
fix: send agent processed-event ACKs directly to avoid replay loops#1030rayl15 wants to merge 1 commit into
Conversation
rayl15
requested review from
chetan-rns,
drewbailey,
jannfis,
jgwest,
jparsai and
mikeshng
as code owners
📝 WalkthroughWalkthroughThis PR changes how processed-event ACKs are delivered to the principal. Instead of enqueuing ACKs onto the per-resource send queue, Estimated code review effort: 3 (Moderate) | ~25 minutes ChangesACK Direct Send
Possibly related issues
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ 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. 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 `@internal/event/event_writer.go`:
- Around line 259-273: SendDirect currently shares sendMu with SendWaitingEvents
and can hold it across the blocking target.Send path, which can stall the
receiver under backpressure. Update EventWriter.SendDirect and the related send
path in sendOnStream/SendWaitingEvents so the mutex is only used to protect
shared state, not the actual gRPC send call; capture the target/stream state
first, release the lock, then perform the send outside the critical section.
🪄 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: c708277c-cc2f-42fc-8641-a5d5f4b1d6b5
📒 Files selected for processing (3)
agent/connection.gointernal/event/event_writer.gointernal/event/event_writer_test.go
| func (ew *EventWriter) SendDirect(ev *cloudevents.Event) error { | ||
| pev, err := format.ToProto(ev) | ||
| if err != nil { | ||
| return fmt.Errorf("could not wire event: %w", err) | ||
| } | ||
|
|
||
| ew.mu.RLock() | ||
| target := ew.target | ||
| ew.mu.RUnlock() | ||
|
|
||
| if target == nil { | ||
| return fmt.Errorf("no target stream available") | ||
| } | ||
|
|
||
| return ew.sendOnStream(target, &eventstreamapi.Event{Event: pev}) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🌐 Web query:
Does grpc-go ClientStream SendMsg block when the HTTP/2 flow control window is exhausted?
💡 Result:
Yes, in grpc-go, the ClientStream SendMsg method blocks when the HTTP/2 flow control window is exhausted [1][2][3]. According to the official grpc-go documentation and source code, SendMsg blocks until one of the following conditions is met [1][4]: - There is sufficient flow control to schedule the message with the transport [1][2]. - The stream is finished [1][2]. - The stream encounters an error (breaks) [1][2]. SendMsg does not wait for the server to acknowledge receipt of the message; it only waits until it can successfully hand the data off to the transport layer based on the available flow control window [1][2]. Because this blocking behavior is tied to the underlying HTTP/2 flow control mechanism, an RPC can remain blocked indefinitely if the receiver stops consuming data from the stream and no timeout is enforced by the application context [5].
Citations:
- 1: https://github.com/grpc/grpc-go/blob/master/stream.go
- 2: https://github.com/grpc/grpc-go/blob/v1.78.0/stream.go
- 3: https://stackoverflow.com/questions/57375360/how-does-grpc-client-streaming-flow-control-work-in-go
- 4: https://github.com/grpc/grpc-go/blob/v1.65.0/stream.go
- 5: Terminating RPCs waiting on flow control grpc/grpc-go#8447
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Files containing EventWriter/SendWaitingEvents/sendMu/SendDirect/receiver path:\n'
git ls-files | rg 'internal/event/event_writer\.go|agent/connection\.go|connection\.go|event_writer\.go'
printf '\nSearch references:\n'
rg -n 'sendMu|SendDirect|SendWaitingEvents|sendOnStream|target\.Send|Receiver|receiver' internal/event agent -SRepository: argoproj-labs/argocd-agent
Length of output: 3212
Avoid blocking the receiver under backpressure. SendDirect shares sendMu with SendWaitingEvents and holds it across target.Send; because gRPC sends can block when flow control is exhausted, a stuck send can stall inbound event processing on the receiver path.
🤖 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 `@internal/event/event_writer.go` around lines 259 - 273, SendDirect currently
shares sendMu with SendWaitingEvents and can hold it across the blocking
target.Send path, which can stall the receiver under backpressure. Update
EventWriter.SendDirect and the related send path in
sendOnStream/SendWaitingEvents so the mutex is only used to protect shared
state, not the actual gRPC send call; capture the target/stream state first,
release the lock, then perform the send outside the critical section.
There was a problem hiding this comment.
Thanks for flagging the trade-off. However, the suggested fix (releasing the lock before Send) can’t be applied as-is. The sendMu function is specifically designed to serialise target.Send because gRPC streams don’t support concurrent SendMsg. Sending outside the critical section would allow the SendWaitingEvents loop and SendDirect call Send on the same stream concurrently, effectively reintroducing the race condition the mutex prevents.
Regarding the underlying concern, yes SendDirect runs on the receiver goroutine. Therefore, under sustained outbound backpressure, the receiver can block on the ACK send. Two things limit this in practice:
- It’s not indefinite. The send is tied to the stream context, so a genuinely stuck peer triggers gRPC keepalive. This cancels the
streamCtx, returns an error toSend, and unblocks the receiver. - It only triggers if the principal keeps sending to the agent while it stops reading the agent’s stream. At that point, outbound progress is already stalled regardless.
This is the deliberately minimal approach discussed in the issue: “send ACKs directly instead of queuing them behind other updates”. Fully decoupling the receiver, for example by creating a dedicated highest-priority ACK lane drained by the single sender goroutine, is a clean follow-up. I’m happy to add it if preferred over the minimal fix here.
There was a problem hiding this comment.
@rayl15, that's a solid rebuttal — you're right that releasing sendMu around target.Send would reopen the exact concurrent-SendMsg race the mutex exists to prevent, so my suggested fix doesn't hold up here. The bounded-blocking argument (stream context/keepalive eventually unblocks a genuinely stuck peer, and the failure mode only manifests when the principal is simultaneously sending but not draining its own stream) is a reasonable trade-off for a minimal, targeted fix to the replay-loop issue.
I agree the fully decoupled approach (dedicated ACK lane drained by a single sender goroutine) is better suited as a follow-up rather than blocking this PR. Let me know if you'd like me to open a follow-up issue to track that enhancement — happy to draft one with the rationale from this thread if so.
There was a problem hiding this comment.
No need to open a ticket. The rationale is covered here and the QoS/priority-lane direction is already mentioned in #839. It’s better left to the maintainers to prioritise. I’m happy to take on the dedicated ACK-lane work as a follow-up if they decide it’s worthwhile.
Processed-event ACKs were enqueued on the normal send queue and routed through the per-resource EventWriter. Because an ACK carries the inbound event's resourceID, it shared a queue with the application's own status updates and could not be drained while a status update for that resource was still awaiting its own ACK. Under heavy status churn the ACK was delayed long enough for the principal to retry the original spec-update, producing replay loops and temporary principal-side lag. Send processed ACKs directly on the stream via a new EventWriter.SendDirect, bypassing the per-resource queue and coalescing. All stream writes are now serialized through a dedicated send mutex, since SendDirect and the SendWaitingEvents loop can otherwise call the gRPC stream's Send concurrently, which is not supported. Fixes argoproj-labs#839 Signed-off-by: Rahul Sharma <er.csrahul@gmail.com>
rayl15
force-pushed
the
fix/ack-direct-send-839
branch
from
c0a2d16 to
1b1befd
Compare
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #1030 +/- ##
==========================================
+ Coverage 48.06% 48.07% +0.01%
==========================================
Files 125 125
Lines 18789 18802 +13
==========================================
+ Hits 9030 9040 +10
- Misses 8943 8944 +1
- Partials 816 818 +2
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
ValidationUnit (deterministic): Local managed-mode e2e real principal + managed agent over the gRPC event stream (guestbook Application, driven with 60 rapid spec-updates plus workload restarts to force concurrent status churn on the same app):
Reproduce the per-resource counts from the agent trace log: # every inbound spec-update event_id should be unique (no re-delivery)
grep 'Event received:.*spec-update' agent.log | grep -oE 'event_id=[^ ]+' | sort | uniq -d # -> empty
# ACKs never enter the send queue:
grep 'Adding an event to the event writer' agent.log | grep -c 'event_type=io.argoproj.argocd-agent.event.processed' # -> 0Screenshot Note: localhost latencies are too low to reproduce the original retry-timing race, so the unit test is the authoritative proof of the mechanism; the e2e confirms the change behaves correctly end-to-end and that processed ACKs no longer traverse the per-resource queue. |
2 participants
Summary
Fixes #839. In managed mode the agent could repeatedly re-process the same inbound principal
spec-updatefor anApplicationbecause the processed-event ACK shared the normal per-resource send path with the app's own status updates.A processed ACK carries the inbound event's
resourceID, so it lands in the sameEventWriterqueue as that application's status updates.EventWriterkeeps one sent-but-unacked event per resource and drains nothing else for that resource until it is ACK'd, so the outbound ACK could sit behind a status update that was itself awaiting an ACK. Under heavy status churn this delayed the ACK long enough for the principal to retry the originalspec-update, producing replay loops and temporary principal-side lag.Change
EventWriter.SendDirect, which writes an event straight to the stream, bypassing the per-resource queue and coalescing logic. The agent now sends processed ACKs through it (agent/connection.go) instead of enqueuing them on the send queue.sendMu.SendDirectand theSendWaitingEventsloop can otherwise call the gRPC stream'sSendconcurrently, which a gRPC stream does not support.This matches the first fix option suggested by @jannfis on the issue ("sending ACKs directly without using event writer to queue those behind other updates").
Scope
The principal emits processed ACKs the same way (
principal/event.go). That path is entangled with the broader per-agent queue QoS/prioritization discussion raised on the issue, so it is intentionally left out of this focused fix.Testing
internal/event/event_writer_test.go:SendDirectreturns an error when no target stream is set.go test ./internal/event/... ./agent/... -racepasses.Summary by CodeRabbit