fix(propeller): record events Admin already has in the dedup filter

flytepropeller/events/admin_eventsink.go

@@ -66,37 +66,42 @@ func (s *adminEventSink) Sink(ctx context.Context, message proto.Message) error
 
 	// Validate submission with rate limiter and send admin event
 	if s.rateLimiter.Allow() {
+		var sendErr error
 		switch eventMessage := message.(type) {
 		case *event.WorkflowExecutionEvent:
 			request := &admin.WorkflowExecutionEventRequest{
 				Event: eventMessage,
 			}
 
-			_, err := s.adminClient.CreateWorkflowEvent(ctx, request)
-			if err != nil {
-				return errors.WrapError(err)
-			}
+			_, sendErr = s.adminClient.CreateWorkflowEvent(ctx, request)
 		case *event.NodeExecutionEvent:
 			request := &admin.NodeExecutionEventRequest{
 				Event: eventMessage,
 			}
 
-			_, err := s.adminClient.CreateNodeEvent(ctx, request)
-			if err != nil {
-				return errors.WrapError(err)
-			}
+			_, sendErr = s.adminClient.CreateNodeEvent(ctx, request)
 		case *event.TaskExecutionEvent:
 			request := &admin.TaskExecutionEventRequest{
 				Event: eventMessage,
 			}
 
-			_, err := s.adminClient.CreateTaskEvent(ctx, request)
-			if err != nil {
-				return errors.WrapError(err)
-			}
+			_, sendErr = s.adminClient.CreateTaskEvent(ctx, request)
 		default:
 			return fmt.Errorf("unknown event type [%s]", eventMessage.String())
 		}
+
+		if sendErr != nil {
+			wrapped := errors.WrapError(sendErr)
+			if errors.IsAlreadyExists(wrapped) {
+				// Admin already holds this event, so re-sending it can never succeed. Record the id
+				// so later re-emits are answered by the filter.Contains check above instead of
+				// spending rate limiter capacity that other executions need. The check returns this
+				// same error code, so callers see no change.
+				s.filter.Add(ctx, id)
+			}
+
+			return wrapped
+		}
 	} else {
 		return &errors.EventError{Code: errors.ResourceExhausted,
 			Cause: fmt.Errorf("Admin EventSink throttling admin traffic"), Message: "Resource Exhausted"}

flytepropeller/events/eventsink_drain_repro_test.go

@@ -0,0 +1,87 @@
+package events
+
+import (
+	"context"
+	"testing"
+
+	"github.com/golang/protobuf/ptypes"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/flyteorg/flyte/flyteidl/clients/go/admin/mocks"
+	"github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
+	"github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/core"
+	"github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/event"
+	"github.com/flyteorg/flyte/flytestdlib/fastcheck"
+	"github.com/flyteorg/flyte/flytestdlib/promutils"
+)
+
+func taskEventFor(name string) *event.TaskExecutionEvent {
+	return &event.TaskExecutionEvent{
+		Phase:        core.TaskExecution_SUCCEEDED,
+		OccurredAt:   ptypes.TimestampNow(),
+		TaskId:       &core.Identifier{ResourceType: core.ResourceType_TASK, Name: "task-id"},
+		RetryAttempt: 1,
+		ParentNodeExecutionId: &core.NodeExecutionIdentifier{
+			NodeId: "node-id",
+			ExecutionId: &core.WorkflowExecutionIdentifier{
+				Project: "p",
+				Domain:  "d",
+				Name:    name,
+			},
+		},
+		Logs: []*core.TaskLog{{Uri: "logs.txt"}},
+	}
+}
+
+// A workflow that keeps re-emitting an event Admin has already durably recorded must not be able
+// to drain the process-wide event-sink rate limiter and starve other executions.
+//
+// The sink is built exactly as flytepropeller builds it in NewAdminEventSink: a real
+// OppoBloomFilter and a real token-bucket limiter. Only the Admin client is a stub, and it
+// answers exactly as a real Admin does for an event it already holds: codes.AlreadyExists.
+func TestAdminEventSink_AlreadyExistsDrainsRateLimiter(t *testing.T) {
+	ctx := context.Background()
+	scope := promutils.NewTestScope()
+
+	filter, err := fastcheck.NewOppoBloomFilter(50000, scope.NewSubScope("filter"))
+	assert.NoError(t, err)
+
+	// Capacity 10: a real deployment's shared budget, scaled down so the drain is observable.
+	adminClient := &mocks.AdminServiceClient{}
+	sink, err := NewAdminEventSink(ctx, adminClient, &Config{Rate: 1, Capacity: 10}, filter)
+	assert.NoError(t, err)
+
+	stuck := taskEventFor("stuck-execution")
+	healthy := taskEventFor("healthy-execution")
+
+	adminCalls := 0
+	adminClient.On("CreateTaskEvent", ctx,
+		mock.MatchedBy(func(req *admin.TaskExecutionEventRequest) bool {
+			return req.GetEvent().GetParentNodeExecutionId().GetExecutionId().GetName() == "stuck-execution"
+		}),
+	).Run(func(mock.Arguments) { adminCalls++ }).
+		Return(nil, status.Error(codes.AlreadyExists, "Grpc AlreadyExists error"))
+
+	adminClient.On("CreateTaskEvent", ctx,
+		mock.MatchedBy(func(req *admin.TaskExecutionEventRequest) bool {
+			return req.GetEvent().GetParentNodeExecutionId().GetExecutionId().GetName() == "healthy-execution"
+		}),
+	).Return(&admin.TaskExecutionEventResponse{}, nil)
+
+	// Admin already holds this event, so re-emitting it can never succeed. Every attempt should
+	// be answered from the local dedup filter, not by spending a token on a doomed round trip.
+	for i := 0; i < 10; i++ {
+		_ = sink.Sink(ctx, stuck)
+	}
+	t.Logf("re-emits of an already-recorded event that reached Admin: %d/10", adminCalls)
+	assert.Equal(t, 1, adminCalls,
+		"an already-recorded event should reach Admin once; afterwards the dedup filter should answer")
+
+	// The unrelated execution must still be able to record its event.
+	healthyErr := sink.Sink(ctx, healthy)
+	assert.NoError(t, healthyErr,
+		"a healthy execution was starved of rate-limiter capacity by an already-recorded event")
+}