fix(engine): populate external ids for durable task children inside of the existing trigger tx

internal/services/dispatcher/server_v1.go

@@ -669,10 +669,6 @@ func (d *DispatcherServiceImpl) handleTriggerRuns(
 		})
 	}
 
-	if populateErr := d.repo.Triggers().PopulateExternalIdsForWorkflow(ctx, invocation.tenantId, triggerOpts); populateErr != nil {
-		return status.Errorf(codes.Internal, "failed to populate external ids for workflow: %v", populateErr)
-	}
-
 	ingestionResult, err := d.repo.DurableEvents().IngestDurableTaskEvent(ctx, v1.IngestDurableTaskEventOpts{
 		BaseIngestEventOpts: &v1.BaseIngestEventOpts{
 			TenantId:        invocation.tenantId,

pkg/repository/durable_events.go

@@ -1064,6 +1064,10 @@ func (r *durableEventsRepository) IngestDurableTaskEvent(ctx context.Context, op
 
 	switch opts.Kind {
 	case sqlcv1.V1DurableEventLogKindRUN:
+		if populateErr := r.populateExternalIdsForWorkflow(ctx, tx, tenantId, opts.TriggerRuns.TriggerOpts); populateErr != nil {
+			return nil, fmt.Errorf("failed to populate external ids for workflow: %w", populateErr)
+		}
+
 		innerOpts := make([]GetOrCreateLogEntryOpt, len(opts.TriggerRuns.TriggerOpts))
 
 		nonSkipOffset := int64(0)

pkg/repository/ids.go

@@ -85,6 +85,22 @@ func (c *ChildWorkflowSignalCreatedData) Bytes() []byte {
 // GenerateExternalIdsForWorkflow generates external ids and additional looks up child workflows and whether they
 // already exist.
 func (s *sharedRepository) PopulateExternalIdsForWorkflow(ctx context.Context, tenantId uuid.UUID, opts []*WorkflowNameTriggerOpts) error {
+	tx, commit, rollback, err := sqlchelpers.PrepareTx(ctx, s.pool, s.l)
+
+	if err != nil {
+		return err
+	}
+
+	defer rollback()
+
+	if err := s.populateExternalIdsForWorkflow(ctx, tx, tenantId, opts); err != nil {
+		return err
+	}
+
+	return commit(ctx)
+}
+
+func (s *sharedRepository) populateExternalIdsForWorkflow(ctx context.Context, tx sqlcv1.DBTX, tenantId uuid.UUID, opts []*WorkflowNameTriggerOpts) error {
 	// get child workflow data first
 	optsWithParents := make([]*WorkflowNameTriggerOpts, 0, len(opts))
 
@@ -99,7 +115,7 @@ func (s *sharedRepository) PopulateExternalIdsForWorkflow(ctx context.Context, t
 	}
 
 	if len(optsWithParents) > 0 {
-		err := s.generateExternalIdsForChildWorkflows(ctx, tenantId, optsWithParents)
+		err := s.generateExternalIdsForChildWorkflows(ctx, tx, tenantId, optsWithParents)
 
 		if err != nil {
 			return err
@@ -109,15 +125,7 @@ func (s *sharedRepository) PopulateExternalIdsForWorkflow(ctx context.Context, t
 	return nil
 }
 
-func (s *sharedRepository) generateExternalIdsForChildWorkflows(ctx context.Context, tenantId uuid.UUID, opts []*WorkflowNameTriggerOpts) error {
-	tx, commit, rollback, err := sqlchelpers.PrepareTx(ctx, s.pool, s.l)
-
-	if err != nil {
-		return err
-	}
-
-	defer rollback()
-
+func (s *sharedRepository) generateExternalIdsForChildWorkflows(ctx context.Context, tx sqlcv1.DBTX, tenantId uuid.UUID, opts []*WorkflowNameTriggerOpts) error {
 	externalIds := make([]uuid.UUID, 0, len(opts))
 	spawnKeyToOpt := make(map[string]*WorkflowNameTriggerOpts)
 
@@ -262,9 +270,5 @@ func (s *sharedRepository) generateExternalIdsForChildWorkflows(ctx context.Cont
 		return err
 	}
 
-	if err := commit(ctx); err != nil {
-		return err
-	}
-
 	return nil
 }