From eeafbde9270ee08d646d67111eea3765fd9c321b Mon Sep 17 00:00:00 2001 From: Maxence Maireaux Date: Thu, 11 Jun 2026 10:10:41 +0200 Subject: [PATCH] fix(triggers): make event processing idempotent under redelivery and retries The event bus is at-least-once. Previously only SAVED_PAYMENT/SAVED_ACCOUNT got a deterministic, dedup-ing Temporal workflow id; every other event type ran with a server-generated id, so a redelivery (or a partial-failure NACK after some triggers had already started) re-executed triggers and replayed side-effecting stages such as money movements. - listener: derive a deterministic workflow id for ALL events (taskIDPrefix-triggerID-) with REJECT_DUPLICATE, so a redelivery is rejected as a duplicate instead of starting a second run. (H1) - occurrence id: ExecuteTrigger built the occurrence with uuid.NewString() in workflow code, yielding a different id on every Temporal replay. Use the (deterministic) workflow execution id instead. (M4) - insert activities: InsertNewInstance, InsertNewStage and InsertTriggerOccurrence now use ON CONFLICT DO NOTHING. With deterministic primary keys, a retry after a lost ack (row committed, result lost) would otherwise fail forever on the duplicate key and wedge the workflow. (M2) Adds a redelivery regression test for a non-payment event. --- internal/triggers/activities.go | 5 ++++ internal/triggers/listener.go | 24 ++++++++++++----- internal/triggers/listener_test.go | 38 +++++++++++++++++++++++++++ internal/triggers/trigger.go | 8 ++++-- internal/triggers/workflow_trigger.go | 6 ++++- internal/workflow/activities.go | 9 +++++++ 6 files changed, 80 insertions(+), 10 deletions(-) diff --git a/internal/triggers/activities.go b/internal/triggers/activities.go index d4eedd1..8b17a8c 100644 --- a/internal/triggers/activities.go +++ b/internal/triggers/activities.go @@ -81,8 +81,13 @@ func (a Activities) EvalTriggerVariables(ctx context.Context, trigger Trigger, r } func (a Activities) InsertTriggerOccurrence(ctx context.Context, occurrence Occurrence) error { + // Idempotent: a Temporal retry after a lost ack (row committed but the + // activity result never reached the server) must not fail on the duplicate + // primary key. The occurrence id is deterministic (see + // NewTriggerOccurrence), so DO NOTHING is safe. _, err := a.db.NewInsert(). Model(pointer.For(occurrence)). + On("CONFLICT DO NOTHING"). Exec(ctx) return err } diff --git a/internal/triggers/listener.go b/internal/triggers/listener.go index de0cc17..c9c4760 100644 --- a/internal/triggers/listener.go +++ b/internal/triggers/listener.go @@ -148,14 +148,24 @@ func handleMessage( searchAttributes[workflow.SearchAttributeTriggerID] = trigger.ID } - options := client.StartWorkflowOptions{ - TaskQueue: taskQueue, - SearchAttributes: searchAttributes, - } + // Derive a deterministic workflow ID so an at-least-once redelivery of + // the same event does not start a second trigger execution (which would + // replay side-effecting stages such as money movements). For + // SAVED_PAYMENT/SAVED_ACCOUNT we key on the object id (dedup across + // distinct deliveries of the same object); for every other event type + // we fall back to the message UUID, which is preserved across + // redeliveries. + dedupKey := msg.UUID if objectID != nil { - options.ID = taskIDPrefix + "-" + trigger.ID + "-" + *objectID - options.WorkflowIDReusePolicy = enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE - options.WorkflowExecutionErrorWhenAlreadyStarted = true + dedupKey = *objectID + } + + options := client.StartWorkflowOptions{ + TaskQueue: taskQueue, + SearchAttributes: searchAttributes, + ID: taskIDPrefix + "-" + trigger.ID + "-" + dedupKey, + WorkflowIDReusePolicy: enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE, + WorkflowExecutionErrorWhenAlreadyStarted: true, } _, execErr := temporalClient.ExecuteWorkflow(ctx, options, ExecuteTrigger, ProcessEventRequest{ diff --git a/internal/triggers/listener_test.go b/internal/triggers/listener_test.go index fb25dfd..c89d6ef 100644 --- a/internal/triggers/listener_test.go +++ b/internal/triggers/listener_test.go @@ -379,4 +379,42 @@ func TestHandleMessage(t *testing.T) { require.NoError(t, err) require.Equal(t, 1, count) }) + + t.Run("redelivery of a non-payment event is skipped", func(t *testing.T) { + t.Parallel() + + db := setupTestDB(t) + taskQueue := setupWorker(t, db) + + w := insertNoOpWorkflow(t, db) + insertTrigger(t, db, w.ID, "NEW_TRANSACTION", nil, nil) + + event := makeMessage("NEW_TRANSACTION", "v1", map[string]any{}) + evaluator := NewDefaultExpressionEvaluator() + + // Simulate an at-least-once redelivery: the broker re-delivers the same + // message, preserving its UUID. + msg1 := publish.NewMessage(logging.TestingContext(), *event) + msg2 := publish.NewMessage(logging.TestingContext(), *event) + msg2.UUID = msg1.UUID + + require.NoError(t, handleMessage(devServer.Client(), db, evaluator, "test", "test", taskQueue, false, msg1)) + require.Eventually(t, func() bool { + count, err := db.NewSelect(). + Model((*Occurrence)(nil)). + Count(logging.TestingContext()) + return err == nil && count == 1 + }, 10*time.Second, 200*time.Millisecond) + + require.NoError(t, handleMessage(devServer.Client(), db, evaluator, "test", "test", taskQueue, false, msg2)) + + // The deterministic workflow id (keyed on the message UUID) must reject + // the duplicate, so no second trigger execution / occurrence is created. + time.Sleep(2 * time.Second) + count, err := db.NewSelect(). + Model((*Occurrence)(nil)). + Count(logging.TestingContext()) + require.NoError(t, err) + require.Equal(t, 1, count) + }) } diff --git a/internal/triggers/trigger.go b/internal/triggers/trigger.go index ad2860c..ce12f27 100644 --- a/internal/triggers/trigger.go +++ b/internal/triggers/trigger.go @@ -112,9 +112,13 @@ type Occurrence struct { Error *string `json:"error,omitempty" bun:"error"` } -func NewTriggerOccurrence(triggerID string, event publish.EventMessage, at time.Time) Occurrence { +// NewTriggerOccurrence builds an occurrence with a caller-provided id. The id +// must be deterministic when called from workflow code (e.g. the workflow +// execution id) so that it stays stable across Temporal replays — generating a +// random uuid here would yield a different occurrence id on every replay. +func NewTriggerOccurrence(id, triggerID string, event publish.EventMessage, at time.Time) Occurrence { return Occurrence{ - ID: uuid.NewString(), + ID: id, TriggerID: triggerID, Date: at, Event: event, diff --git a/internal/triggers/workflow_trigger.go b/internal/triggers/workflow_trigger.go index 16bbf2a..44ebb28 100644 --- a/internal/triggers/workflow_trigger.go +++ b/internal/triggers/workflow_trigger.go @@ -68,7 +68,11 @@ func (w triggerWorkflow) RunTrigger(ctx temporalworkflow.Context, req ProcessEve func (w triggerWorkflow) ExecuteTrigger(ctx temporalworkflow.Context, req ProcessEventRequest, trigger Trigger) error { vars := make(map[string]string) - occurrence := NewTriggerOccurrence(trigger.ID, req.Event, temporalworkflow.Now(ctx)) + // Use the (deterministic, replay-stable) workflow execution id as the + // occurrence id rather than a random uuid generated in workflow code. + occurrence := NewTriggerOccurrence( + temporalworkflow.GetInfo(ctx).WorkflowExecution.ID, + trigger.ID, req.Event, temporalworkflow.Now(ctx)) err := temporalworkflow.ExecuteActivity( temporalworkflow.WithActivityOptions(ctx, temporalworkflow.ActivityOptions{ StartToCloseTimeout: 10 * time.Second, diff --git a/internal/workflow/activities.go b/internal/workflow/activities.go index b06d487..79e4d65 100644 --- a/internal/workflow/activities.go +++ b/internal/workflow/activities.go @@ -70,9 +70,14 @@ func (a Activities) SendWorkflowStageTerminationEvent(ctx context.Context, insta func (a Activities) InsertNewInstance(ctx context.Context, workflowID string) (*Instance, error) { instance := NewInstance(activity.GetInfo(ctx).WorkflowExecution.ID, workflowID) + // Idempotent: the primary key is the (deterministic) workflow execution id, + // so a Temporal retry after a lost ack must not fail on the duplicate key. + // The returned instance is rebuilt deterministically, so it is correct even + // when the row already exists. if _, err := a.db. NewInsert(). Model(&instance). + On("CONFLICT DO NOTHING"). Exec(ctx); err != nil { return nil, err } @@ -90,8 +95,12 @@ func (a Activities) UpdateInstance(ctx context.Context, instance *Instance) erro func (a Activities) InsertNewStage(ctx context.Context, instance Instance, ind int) (*Stage, error) { stage := NewStage(instance.ID, activity.GetInfo(ctx).WorkflowExecution.RunID, ind) + // Idempotent: the primary key is deterministic (instance id + run id + + // index), so a Temporal retry after a lost ack must not fail on the + // duplicate key. if _, err := a.db.NewInsert(). Model(&stage). + On("CONFLICT DO NOTHING"). Exec(ctx); err != nil { return nil, err }