fix: dont check idempotency keys for dag steps

Author: mrkaye97

pkg/repository/durable_events.go

@@ -683,29 +683,28 @@ func nonDeterminismDetail(opts IngestDurableTaskEventOpts, expectedKind sqlcv1.V
 }
 
 type GetOrCreateLogEntryOpt struct {
-	Kind                 sqlcv1.V1DurableEventLogKind
-	IdempotencyKey       []byte
-	LegacyIdempotencyKey []byte
-	InputPayload         []byte
-	ResultPayload        []byte
-	NodeId               int64
-	BranchId             int64
-	IsSatisfied          bool
-	UserMessage          *string
-	WaitData             string // JSON-encoded WaitData, empty string means no wait data
-	ChildTaskExternalId  uuid.UUID
-	ShouldSkip           bool
+	Kind           sqlcv1.V1DurableEventLogKind
+	IdempotencyKey []byte
+	InputPayload   []byte
+	ResultPayload  []byte
+	NodeId         int64
+	BranchId       int64
+	IsSatisfied    bool
+	UserMessage    *string
+	WaitData       string // JSON-encoded WaitData, empty string means no wait data
+	// IsDagStepTrigger skips the idempotency check against an existing entry: the engine orchestrates the DAG
+	// itself from a pinned workflow version, so its replays are deterministic by construction, whereas the key
+	// hashes the run's input bytes, which change with the engine version that serialised them.
+	IsDagStepTrigger    bool
+	ChildTaskExternalId uuid.UUID
+	ShouldSkip          bool
 	// IsTriggered creates the entry with triggered_at already set: the child's task rows exist
 	// (an in-place replay), so there is no run left for the trigger path to spawn.
 	IsTriggered bool
 }
 
-func (o GetOrCreateLogEntryOpt) idempotencyKeyMatches(existingKey []byte) bool {
-	if bytes.Equal(o.IdempotencyKey, existingKey) {
-		return true
-	}
-
-	return len(o.LegacyIdempotencyKey) > 0 && bytes.Equal(o.LegacyIdempotencyKey, existingKey)
+func (o GetOrCreateLogEntryOpt) matchesExistingEntry(existing *sqlcv1.BulkGetDurableEventLogEntriesRow) bool {
+	return o.IsDagStepTrigger || bytes.Equal(o.IdempotencyKey, existing.IdempotencyKey)
 }
 
 type GetOrCreateLogEntryOpts struct {
@@ -822,26 +821,14 @@ func satisfiedOrderPtr(v pgtype.Int8) *int64 {
 }
 
 func (r *durableEventsRepository) createIdempotencyKey(kind sqlcv1.V1DurableEventLogKind, triggerOpts *WorkflowNameTriggerOpts, waitForConditions []CreateExternalSignalConditionOpt) ([]byte, error) {
-	return hashIdempotencyKey(kind, triggerOpts, waitForConditions, true)
-}
-
-func (r *durableEventsRepository) createLegacyDagStepIdempotencyKey(triggerOpts *WorkflowNameTriggerOpts) ([]byte, error) {
-	if !triggerOpts.IsDagStepTrigger {
-		return nil, nil
-	}
-
-	return hashIdempotencyKey(sqlcv1.V1DurableEventLogKindRUN, triggerOpts, nil, false)
-}
-
-func hashIdempotencyKey(kind sqlcv1.V1DurableEventLogKind, triggerOpts *WorkflowNameTriggerOpts, waitForConditions []CreateExternalSignalConditionOpt, includeDagStepIdentity bool) ([]byte, error) {
 	// note: can't use additional metadata here because it's not stable, since we store trace information in it w/ the otel instrumentors
 	dataToHash := []byte(kind)
 
 	if triggerOpts != nil {
 		dataToHash = append(dataToHash, triggerOpts.Data...)
 		dataToHash = append(dataToHash, []byte(triggerOpts.WorkflowName)...)
 
-		if includeDagStepIdentity && triggerOpts.IsDagStepTrigger {
+		if triggerOpts.IsDagStepTrigger {
 			if triggerOpts.TargetActionId != nil {
 				dataToHash = append(dataToHash, []byte(*triggerOpts.TargetActionId)...)
 			}
@@ -1129,7 +1116,7 @@ func (r *durableEventsRepository) getOrCreateEventLogEntriesForTasks(
 				state.newEntryByKey[key] = o
 				continue
 			}
-			if !o.idempotencyKeyMatches(e.IdempotencyKey) {
+			if !o.matchesExistingEntry(e) {
 				nonDeterminismErr = &NonDeterminismError{
 					BranchId:                o.BranchId,
 					NodeId:                  o.NodeId,
@@ -1288,7 +1275,7 @@ func (r *durableEventsRepository) getOrCreateEventLogEntriesForTasks(
 				return nil, nil, nil, fmt.Errorf("expected to find log entry for skipped child task external id %s", o.ChildTaskExternalId)
 			}
 
-			if len(o.IdempotencyKey) > 0 && !o.idempotencyKeyMatches(e.IdempotencyKey) {
+			if len(o.IdempotencyKey) > 0 && !o.matchesExistingEntry(e) {
 				return nil, nil, nil, &NonDeterminismError{
 					BranchId:                e.BranchID,
 					NodeId:                  e.NodeID,
@@ -2112,31 +2099,25 @@ func (r *durableEventsRepository) planDurableEventLogAppend(
 				// only index-based dedupe is validated against the existing entry's
 				// idempotency key: an explicit child_key intentionally reuses the
 				// cached child even when the inputs differ
-				var idempotencyKey, legacyIdempotencyKey []byte
+				var idempotencyKey []byte
 				if triggerOpts.ChildKey == nil {
 					key, keyErr := r.createIdempotencyKey(sqlcv1.V1DurableEventLogKindRUN, triggerOpts, nil)
 					if keyErr != nil {
 						return nil, fmt.Errorf("failed to create idempotency key: %w", keyErr), nil
 					}
 					idempotencyKey = key
-
-					legacyKey, legacyKeyErr := r.createLegacyDagStepIdempotencyKey(triggerOpts)
-					if legacyKeyErr != nil {
-						return nil, fmt.Errorf("failed to create legacy idempotency key: %w", legacyKeyErr), nil
-					}
-					legacyIdempotencyKey = legacyKey
 				}
 
 				if _, exists := childExternalIdToTriggerOpts[triggerOpts.ExternalId]; !exists {
 					childExternalIdToTriggerOpts[triggerOpts.ExternalId] = triggerOpts
 				}
 
 				innerOpts[i] = GetOrCreateLogEntryOpt{
-					Kind:                 sqlcv1.V1DurableEventLogKindRUN,
-					ChildTaskExternalId:  triggerOpts.ExternalId,
-					IdempotencyKey:       idempotencyKey,
-					LegacyIdempotencyKey: legacyIdempotencyKey,
-					ShouldSkip:           true,
+					Kind:                sqlcv1.V1DurableEventLogKindRUN,
+					ChildTaskExternalId: triggerOpts.ExternalId,
+					IdempotencyKey:      idempotencyKey,
+					IsDagStepTrigger:    triggerOpts.IsDagStepTrigger,
+					ShouldSkip:          true,
 				}
 				continue
 			}
@@ -2159,29 +2140,24 @@ func (r *durableEventsRepository) planDurableEventLogAppend(
 				return nil, fmt.Errorf("failed to create idempotency key: %w", keyErr), nil
 			}
 
-			legacyIdempotencyKey, legacyKeyErr := r.createLegacyDagStepIdempotencyKey(triggerOpts)
-			if legacyKeyErr != nil {
-				return nil, fmt.Errorf("failed to create legacy idempotency key: %w", legacyKeyErr), nil
-			}
-
 			// A child that is being cancelled or skipped never runs, so no completion event
 			// will arrive for it (trigger.go creates it directly in a terminal state,
 			// bypassing the match pipeline): its entry is terminal at creation regardless of
 			// whether this is a first run or a replay.
 			isSatisfied := triggerOpts.IsCancelled || triggerOpts.IsSkipped
 
 			innerOpts[i] = GetOrCreateLogEntryOpt{
-				Kind:                 sqlcv1.V1DurableEventLogKindRUN,
-				NodeId:               nodeId,
-				BranchId:             branchId,
-				ChildTaskExternalId:  triggerOpts.ExternalId,
-				IdempotencyKey:       idempotencyKey,
-				LegacyIdempotencyKey: legacyIdempotencyKey,
-				InputPayload:         inputPayload,
-				WaitData:             marshalWaitData(waitDataFromTriggerOpt(triggerOpts)),
-				UserMessage:          triggerOpts.UserMessage,
-				IsSatisfied:          isSatisfied,
-				IsTriggered:          childrenToReplay[triggerOpts.ExternalId],
+				Kind:                sqlcv1.V1DurableEventLogKindRUN,
+				NodeId:              nodeId,
+				BranchId:            branchId,
+				ChildTaskExternalId: triggerOpts.ExternalId,
+				IdempotencyKey:      idempotencyKey,
+				InputPayload:        inputPayload,
+				WaitData:            marshalWaitData(waitDataFromTriggerOpt(triggerOpts)),
+				UserMessage:         triggerOpts.UserMessage,
+				IsDagStepTrigger:    triggerOpts.IsDagStepTrigger,
+				IsSatisfied:         isSatisfied,
+				IsTriggered:         childrenToReplay[triggerOpts.ExternalId],
 			}
 		}
 

pkg/repository/durable_events_test.go

@@ -307,28 +307,17 @@ func TestCreateIdempotencyKey_DagStepsOfSameWorkflowDiffer(t *testing.T) {
 	assert.NotEqual(t, stepA, stepB)
 }
 
-func TestIdempotencyKeyMatches_AcceptsLegacyDagStepKey(t *testing.T) {
-	r := &durableEventsRepository{}
-	triggerOpts := dagStepTriggerOpts("my-dag:step-a", 9)
-
-	current, err := r.createIdempotencyKey(sqlcv1.V1DurableEventLogKindRUN, triggerOpts, nil)
-	assert.NoError(t, err)
-
-	legacy, err := r.createLegacyDagStepIdempotencyKey(triggerOpts)
-	assert.NoError(t, err)
-
-	preUpgrade := keyFromKind(t, sqlcv1.V1DurableEventLogKindRUN, &WorkflowNameTriggerOpts{
-		TriggerTaskData: &TriggerTaskData{WorkflowName: "my-dag", Data: []byte(`{"x":1}`)},
-	}, nil)
+func TestMatchesExistingEntry_DagStepIsNeverCheckedForNonDeterminism(t *testing.T) {
+	opt := GetOrCreateLogEntryOpt{IdempotencyKey: []byte("key-hashed-from-compact-input"), IsDagStepTrigger: true}
 
-	assert.Equal(t, preUpgrade, string(legacy))
-	assert.NotEqual(t, string(current), string(legacy))
+	assert.True(t, opt.matchesExistingEntry(&sqlcv1.BulkGetDurableEventLogEntriesRow{IdempotencyKey: []byte("key-hashed-from-spaced-input")}))
+}
 
-	opt := GetOrCreateLogEntryOpt{IdempotencyKey: current, LegacyIdempotencyKey: legacy}
+func TestMatchesExistingEntry_ChildRunMatchesByKey(t *testing.T) {
+	opt := GetOrCreateLogEntryOpt{IdempotencyKey: []byte("key")}
 
-	assert.True(t, opt.idempotencyKeyMatches(current))
-	assert.True(t, opt.idempotencyKeyMatches(legacy))
-	assert.False(t, opt.idempotencyKeyMatches([]byte(keyFromKind(t, sqlcv1.V1DurableEventLogKindRUN, dagStepTriggerOpts("my-dag:step-b", 13), nil))))
+	assert.True(t, opt.matchesExistingEntry(&sqlcv1.BulkGetDurableEventLogEntriesRow{IdempotencyKey: []byte("key")}))
+	assert.False(t, opt.matchesExistingEntry(&sqlcv1.BulkGetDurableEventLogEntriesRow{IdempotencyKey: []byte("other key")}))
 }
 
 func int32Ptr(i int32) *int32 { return &i }