From cd62d5b534fd234fe94d42c964848170cd59d7e0 Mon Sep 17 00:00:00 2001 From: mrkaye97 Date: Thu, 2 Jul 2026 12:40:20 -0400 Subject: [PATCH] fix: populate external ids inside of the existing trigger tx --- internal/services/dispatcher/server_v1.go | 4 --- pkg/repository/durable_events.go | 4 +++ pkg/repository/ids.go | 32 +++++++++++++---------- 3 files changed, 22 insertions(+), 18 deletions(-) diff --git a/internal/services/dispatcher/server_v1.go b/internal/services/dispatcher/server_v1.go index 3ad5034e19..a0575190a7 100644 --- a/internal/services/dispatcher/server_v1.go +++ b/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, diff --git a/pkg/repository/durable_events.go b/pkg/repository/durable_events.go index 5647823abe..3a50a41dde 100644 --- a/pkg/repository/durable_events.go +++ b/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) diff --git a/pkg/repository/ids.go b/pkg/repository/ids.go index 454eb64272..c4dddd3103 100644 --- a/pkg/repository/ids.go +++ b/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 }