fix: dont check idempotency keys for dag steps

fix: tests + cleanup

fix: naming

fix: naming

Author: mrkaye97

pkg/repository/durable_events.go

@@ -683,29 +683,25 @@ 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    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 +818,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 +1113,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 +1272,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 +2096,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 +2137,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_dag_replay_test.go

@@ -25,7 +25,7 @@ func ingestDagStepTrigger(
 ) *IngestTriggerRunsEntry {
 	t.Helper()
 
-	result, err := ingestDagStepTriggerResult(ctx, repo, tenantID, task, invocationCount, actionId, childIndex, false)
+	result, err := ingestDagStepTriggerResult(ctx, repo, tenantID, task, invocationCount, actionId, childIndex, []byte(`{"x":1}`), false)
 	require.NoError(t, err)
 	require.NotNil(t, result.TriggerRunsResult)
 	require.Len(t, result.TriggerRunsResult.Entries, 1)
@@ -41,6 +41,7 @@ func ingestDagStepTriggerResult(
 	invocationCount int32,
 	actionId string,
 	childIndex int64,
+	input []byte,
 	isSkipped bool,
 ) (*IngestDurableTaskEventResult, error) {
 	parentExternalId := task.ExternalID
@@ -61,7 +62,7 @@ func ingestDagStepTriggerResult(
 				TriggerTaskData: &TriggerTaskData{
 					IsSkipped:            isSkipped,
 					WorkflowName:         "my-dag",
-					Data:                 []byte(`{"x":1}`),
+					Data:                 input,
 					TargetActionId:       &actionId,
 					ParentExternalId:     &parentExternalId,
 					ParentTaskId:         &parentTaskId,
@@ -84,7 +85,7 @@ func reinvokeDurableTask(t *testing.T, ctx context.Context, repos userEventScope
 	require.NoError(t, err)
 }
 
-func TestDagStepReplayAtAnotherStepsNodeIsNondeterministic(t *testing.T) {
+func TestDagStepReplayReusesTheEntryEvenWhenTheInputBytesDiffer(t *testing.T) {
 	pool, cleanup := setupPostgresWithMigration(t)
 	defer cleanup()
 
@@ -93,17 +94,20 @@ func TestDagStepReplayAtAnotherStepsNodeIsNondeterministic(t *testing.T) {
 	repos := newUserEventScopeTestRepositories(t, pool)
 	task := createUserEventScopeTestTask(t, ctx, repos, tenantID, 201)
 
-	dedupe := ingestDagStepTrigger(t, ctx, repos.durable, tenantID, task, 1, "my-dag:dedupe-action-items", 13)
-	require.False(t, dedupe.AlreadyExisted)
-	require.EqualValues(t, 1, dedupe.NodeId)
+	first := ingestDagStepTrigger(t, ctx, repos.durable, tenantID, task, 1, "my-dag:step-a", 1)
+	require.False(t, first.AlreadyExisted)
+	require.EqualValues(t, 1, first.NodeId)
 
 	reinvokeDurableTask(t, ctx, repos, tenantID, task)
 
-	_, err := ingestDagStepTriggerResult(ctx, repos.durable, tenantID, task, 2, "my-dag:bold-dates-and-quantities", 9, false)
+	result, err := ingestDagStepTriggerResult(ctx, repos.durable, tenantID, task, 2, "my-dag:step-a", 1, []byte(`{"x": 1}`), false)
+	require.NoError(t, err)
+	require.Len(t, result.TriggerRunsResult.Entries, 1)
 
-	var nonDeterminismErr *NonDeterminismError
-	require.ErrorAs(t, err, &nonDeterminismErr)
-	require.EqualValues(t, 1, nonDeterminismErr.NodeId)
+	replayed := result.TriggerRunsResult.Entries[0]
+	require.True(t, replayed.AlreadyExisted)
+	require.EqualValues(t, 1, replayed.NodeId)
+	require.Equal(t, first.WorkflowRunExternalId, replayed.WorkflowRunExternalId)
 }
 
 func TestDagStepEntriesOfSameWorkflowHaveDistinctIdempotencyKeys(t *testing.T) {
@@ -141,7 +145,7 @@ func TestDagStepCreatedSkippedTakesNextSatisfiedOrder(t *testing.T) {
 	require.False(t, running.IsSatisfied)
 	require.Nil(t, running.SatisfiedOrder)
 
-	result, err := ingestDagStepTriggerResult(ctx, repos.durable, tenantID, task, 1, "my-dag:step-b", 2, true)
+	result, err := ingestDagStepTriggerResult(ctx, repos.durable, tenantID, task, 1, "my-dag:step-b", 2, []byte(`{"x":1}`), true)
 	require.NoError(t, err)
 
 	skipped := result.TriggerRunsResult.Entries[0]

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 }