[1fanwang]

Why are the changes needed?

adminEventSink.Sink keeps a dedup filter so an event that has already been recorded doesn't go back to Admin. The id is only added to that filter after a clean send:

	s.filter.Add(ctx, id)
	return nil

When Admin answers AlreadyExists, the sink returns the wrapped error and the id is never recorded. Every later re-emit of that event passes filter.Contains, takes a token from the sink's rate limiter, and makes a round trip that cannot succeed.

The limiter is shared by every execution the propeller is running. One execution re-emitting a recorded event can drain it, and the executions that needed those tokens get ResourceExhausted:

re-emits of an already-recorded event that reached Admin: 10/10
Error:      Received unexpected error:
            ResourceExhausted: Resource Exhausted, caused by [Admin EventSink throttling admin traffic]
Messages:   a healthy execution was starved of rate-limiter capacity by an already-recorded event

What changes were proposed in this pull request?

Add the id to the dedup filter when Admin reports it already has the event, so later re-emits are answered by the filter.Contains check instead of spending limiter capacity.

The error is still returned, and filter.Contains returns that same code for a locally known duplicate, so callers see no change either way.

Only AlreadyExists is filtered. EventAlreadyInTerminalState is deliberately left out: callers act on it — nodeerrors.IllegalStateError in node_exec_context.go, a forced transition to WorkflowPhaseFailed in workflow/executor.go — and folding it into the filter would replay it as AlreadyExists, which those same callers map to nil.

How was this patch tested?

TestAdminEventSink_AlreadyExistsDrainsRateLimiter builds the sink the way NewAdminEventSink does in production — a real OppoBloomFilter and a real token-bucket limiter, with only the Admin client stubbed, answering codes.AlreadyExists as a real Admin does for an event it already holds. It re-emits that event ten times against a limiter with ten tokens, then checks an unrelated execution can still record its own event.

cd flytepropeller
git checkout upstream/master -- events/admin_eventsink.go   # drop the fix
go test ./events/ -run TestAdminEventSink_AlreadyExistsDrainsRateLimiter -count=1
git checkout HEAD -- events/admin_eventsink.go              # restore it
go test ./events/ -run TestAdminEventSink_AlreadyExistsDrainsRateLimiter -count=1 -v
go test ./events/... -count=1

Raw output

Without the fix:

--- FAIL: TestAdminEventSink_AlreadyExistsDrainsRateLimiter (0.00s)
    eventsink_drain_repro_test.go:79: re-emits of an already-recorded event that reached Admin: 10/10
    eventsink_drain_repro_test.go:80:
        	Error:      	Not equal:
        	            	expected: 1
        	            	actual  : 10
        	Messages:   	an already-recorded event should reach Admin once; afterwards the dedup filter should answer
    eventsink_drain_repro_test.go:85:
        	Error:      	Received unexpected error:
        	            	ResourceExhausted: Resource Exhausted, caused by [Admin EventSink throttling admin traffic]
        	Messages:   	a healthy execution was starved of rate-limiter capacity by an already-recorded event
FAIL
FAIL	github.com/flyteorg/flyte/flytepropeller/events	0.850s

With the fix:

=== RUN   TestAdminEventSink_AlreadyExistsDrainsRateLimiter
    eventsink_drain_repro_test.go:79: re-emits of an already-recorded event that reached Admin: 1/10
--- PASS: TestAdminEventSink_AlreadyExistsDrainsRateLimiter (0.00s)
PASS
ok  	github.com/flyteorg/flyte/flytepropeller/events	0.739s

The rest of the package, including TestAdminAlreadyExistsError which pins the returned error:

ok  	github.com/flyteorg/flyte/flytepropeller/events	0.983s
ok  	github.com/flyteorg/flyte/flytepropeller/events/errors	0.305s

Labels

fixed

Check all the applicable boxes

• I updated the documentation accordingly.
• All new and existing tests passed.
• All commits are signed-off.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 57.24%. Comparing base (2c7f086) to head (a56821c).

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #7852      +/-   ##
==========================================
+ Coverage   57.22%   57.24%   +0.01%     
==========================================
  Files         931      931              
  Lines       58269    58269              
==========================================
+ Hits        33347    33354       +7     
+ Misses      21859    21854       -5     
+ Partials     3063     3061       -2     

Flag	Coverage Δ
unittests-datacatalog	53.51% <ø> (ø)
unittests-flyteadmin	53.23% <ø> (ø)
unittests-flytecopilot	48.05% <ø> (ø)
unittests-flytectl	64.14% <ø> (+0.04%)	⬆️
unittests-flyteidl	75.71% <ø> (ø)
unittests-flyteplugins	60.39% <ø> (ø)
unittests-flytepropeller	53.81% <100.00%> (+0.02%)	⬆️
unittests-flytestdlib	64.41% <ø> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Sovietaced]

I've been meaning to look into this issue after seeing HUGE amounts of AlreadyExists errors when rebooting propeller and wiping this in-memory send cache. Great find!