fix: tests + cleanup

Author: mrkaye97

pkg/repository/durable_events.go

@@ -683,19 +683,16 @@ func nonDeterminismDetail(opts IngestDurableTaskEventOpts, expectedKind sqlcv1.V
 }
 
 type GetOrCreateLogEntryOpt struct {
-	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
+	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
@@ -704,7 +701,7 @@ type GetOrCreateLogEntryOpt struct {
 }
 
 func (o GetOrCreateLogEntryOpt) matchesExistingEntry(existing *sqlcv1.BulkGetDurableEventLogEntriesRow) bool {
-	return o.IsDagStepTrigger || bytes.Equal(o.IdempotencyKey, existing.IdempotencyKey)
+	return o.IsDAGStepTrigger || bytes.Equal(o.IdempotencyKey, existing.IdempotencyKey)
 }
 
 type GetOrCreateLogEntryOpts struct {

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]