[BUG] use deep copy of bit arrays when getting array node state

flytepropeller/events/admin_eventsink.go

@@ -57,7 +57,11 @@ func (s *adminEventSink) Sink(ctx context.Context, message proto.Message) error
 
 	if s.filter.Contains(ctx, id) {
 		logger.Debugf(ctx, "event '%s' has already been sent", string(id))
-		return nil
+		return &errors.EventError{
+			Code:    errors.AlreadyExists,
+			Cause:   fmt.Errorf("event has already been sent"),
+			Message: "Event Already Exists",
+		}
 	}
 
 	// Validate submission with rate limiter and send admin event

flytepropeller/events/admin_eventsink_test.go

@@ -184,13 +184,16 @@ func TestAdminFilterContains(t *testing.T) {
 	filter.OnContainsMatch(mock.Anything, mock.Anything).Return(true)
 
 	wfErr := adminEventSink.Sink(ctx, wfEvent)
-	assert.NoError(t, wfErr)
+	assert.Error(t, wfErr)
+	assert.True(t, errors.IsAlreadyExists(wfErr))
 
 	nodeErr := adminEventSink.Sink(ctx, nodeEvent)
-	assert.NoError(t, nodeErr)
+	assert.Error(t, nodeErr)
+	assert.True(t, errors.IsAlreadyExists(nodeErr))
 
 	taskErr := adminEventSink.Sink(ctx, taskEvent)
-	assert.NoError(t, taskErr)
+	assert.Error(t, taskErr)
+	assert.True(t, errors.IsAlreadyExists(taskErr))
 }
 
 func TestIDFromMessage(t *testing.T) {

flytepropeller/events/errors/errors.go

@@ -33,7 +33,11 @@
 }
 
 func (r EventError) Error() string {
-	return fmt.Sprintf("%s: %s, caused by [%s]", r.Code, r.Message, r.Cause.Error())
+	var cause string
+	if r.Cause != nil {
+		cause = r.Cause.Error()
+	}
+	return fmt.Sprintf("%s: %s, caused by [%s]", r.Code, r.Message, cause)
 }
 
 func (r *EventError) Is(target error) bool {

flytepropeller/pkg/apis/flyteworkflow/v1alpha1/node_status.go

@@ -316,6 +316,27 @@
 	}
 }
 
+func (in *ArrayNodeStatus) DeepCopyInto(out *ArrayNodeStatus) {
+	*out = *in
+	out.MutableStruct = in.MutableStruct
+
+	if in.ExecutionError != nil {
+		in, out := &in.ExecutionError, &out.ExecutionError
+		*out = new(core.ExecutionError)
+		*out = *in
+	}
+}
+
+func (in *ArrayNodeStatus) DeepCopy() *ArrayNodeStatus {
+	if in == nil {
+		return nil
+	}
+
+	out := &ArrayNodeStatus{}
+	in.DeepCopyInto(out)
+	return out
+}
+
 type NodeStatus struct {
 	MutableStruct
 	Phase                NodePhase     `json:"phase,omitempty"`

flytepropeller/pkg/controller/config/config.go

@@ -259,6 +259,7 @@ const (
 type EventConfig struct {
 	RawOutputPolicy           RawOutputPolicy `json:"raw-output-policy" pflag:",How output data should be passed along in execution events."`
 	FallbackToOutputReference bool            `json:"fallback-to-output-reference" pflag:",Whether output data should be sent by reference when it is too large to be sent inline in execution events."`
+	ErrorOnAlreadyExists      bool            `json:"error-on-already-exists" pflag:",Whether to return an error when an event already exists."`
 }
 
 // ParallelismBehavior defines how ArrayNode should handle subNode parallelism by default

flytepropeller/pkg/controller/nodes/array/handler.go

@@ -11,6 +11,7 @@
 	"github.com/flyteorg/flyte/flyteplugins/go/tasks/pluginmachinery/ioutils"
 	"github.com/flyteorg/flyte/flyteplugins/go/tasks/plugins/array/errorcollector"
 	"github.com/flyteorg/flyte/flytepropeller/events"
+	eventsErr "github.com/flyteorg/flyte/flytepropeller/events/errors"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/apis/flyteworkflow/v1alpha1"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/compiler/validators"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/config"
@@ -21,6 +22,7 @@
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/interfaces"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/task/k8s"
 	"github.com/flyteorg/flyte/flytestdlib/bitarray"
+	stdConfig "github.com/flyteorg/flyte/flytestdlib/config"
 	"github.com/flyteorg/flyte/flytestdlib/logger"
 	"github.com/flyteorg/flyte/flytestdlib/promutils"
 	"github.com/flyteorg/flyte/flytestdlib/storage"
@@ -112,6 +114,10 @@
 
 	// update state for subNodes
 	if err := eventRecorder.finalize(ctx, nCtx, idlcore.TaskExecution_ABORTED, 0, a.eventConfig); err != nil {
+		// a task event with abort phase is already emitted when handling ArrayNodePhaseFailing
+		if eventsErr.IsAlreadyExists(err) {
+			return nil
+		}
 		logger.Errorf(ctx, "ArrayNode event recording failed: [%s]", err.Error())
 		return err
 	}
@@ -579,12 +585,35 @@
 
 		// increment taskPhaseVersion if we detect any changes in subNode state.
 		if incrementTaskPhaseVersion {
-			arrayNodeState.TaskPhaseVersion = arrayNodeState.TaskPhaseVersion + 1
+			arrayNodeState.TaskPhaseVersion++
 		}
 
-		if err := eventRecorder.finalize(ctx, nCtx, taskPhase, arrayNodeState.TaskPhaseVersion, a.eventConfig); err != nil {
-			logger.Errorf(ctx, "ArrayNode event recording failed: [%s]", err.Error())
-			return handler.UnknownTransition, err
+		const maxRetries = 3
+		retries := 0
+		for retries <= maxRetries {
+			err := eventRecorder.finalize(ctx, nCtx, taskPhase, arrayNodeState.TaskPhaseVersion, a.eventConfig)
+
+			if err == nil {
+				break
+			}
+
+			// Handle potential race condition if FlyteWorkflow CRD fails to get synced
+			if eventsErr.IsAlreadyExists(err) {
+				if !incrementTaskPhaseVersion {
+					break
+				}
+				logger.Warnf(ctx, "Event version already exists, bumping version and retrying (%d/%d): [%s]", retries+1, maxRetries, err.Error())
+				arrayNodeState.TaskPhaseVersion++
+			} else {
+				logger.Errorf(ctx, "ArrayNode event recording failed: [%s]", err.Error())
+				return handler.UnknownTransition, err
+			}
+
+			retries++
+			if retries > maxRetries {
+				logger.Errorf(ctx, "ArrayNode event recording failed after %d retries: [%s]", maxRetries, err.Error())
+				return handler.UnknownTransition, err
+			}
 		}
 
 		// if the ArrayNode phase has changed we need to reset the taskPhaseVersion to 0
@@ -632,9 +661,21 @@
 		return nil, err
 	}
 
+	eventConfigCopy, err := stdConfig.DeepCopyConfig(eventConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	deepCopiedEventConfig, ok := eventConfigCopy.(*config.EventConfig)
+	if !ok {
+		return nil, fmt.Errorf("deep copy error: expected *config.EventConfig, but got %T", eventConfigCopy)
+	}
+
+	deepCopiedEventConfig.ErrorOnAlreadyExists = true
+
 	arrayScope := scope.NewSubScope("array")
 	return &arrayNodeHandler{
-		eventConfig:                 eventConfig,
+		eventConfig:                 deepCopiedEventConfig,
 		gatherOutputsRequestChannel: make(chan *gatherOutputsRequest),
 		metrics:                     newMetrics(arrayScope),
 		nodeExecutionRequestChannel: make(chan *nodeExecutionRequest),