Skip to content

[CRE-6175] routed trigger event #23608

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
agparadiso merged 3 commits into from
Sep 2, 2026
Merged

[CRE-6175] routed trigger event #23608

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2d8e9e9
feat: use deadline stamped at dispatch instead of calculating the age…
agparadiso Sep 1, 2026
70d64c6
test: build test that documents/proves the behavioural change in whic…
agparadiso Sep 1, 2026
5f856ef
chore: improve doc on deadline
agparadiso Sep 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions core/services/workflows/v2/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ type EngineLimiters struct {
TriggerSubscriptionTime limits.TimeLimiter
TriggerRegistrationsTime limits.TimeLimiter
TriggerSubscription limits.BoundLimiter[int]
TriggerEventQueue limits.QueueLimiter[enqueuedTriggerEvent]
TriggerEventQueue limits.QueueLimiter[RoutedTriggerEvent]
TriggerEventQueueTime limits.TimeLimiter
ExecutionConcurrency limits.ResourcePoolLimiter[int]

Expand Down Expand Up @@ -154,7 +154,7 @@ func (l *EngineLimiters) init(lf limits.Factory, cfgFn func(*cresettings.Workflo
if err != nil {
return
}
l.TriggerEventQueue, err = limits.MakeQueueLimiter[enqueuedTriggerEvent](lf, cfg.TriggerEventQueueLimit)
l.TriggerEventQueue, err = limits.MakeQueueLimiter[RoutedTriggerEvent](lf, cfg.TriggerEventQueueLimit)
if err != nil {
return
}
Expand Down
70 changes: 31 additions & 39 deletions core/services/workflows/v2/engine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ type Engine struct {
// used to separate registration and unregistration phases
triggersRegMu sync.Mutex

allTriggerEventsQueueCh limits.QueueLimiter[enqueuedTriggerEvent]
allTriggerEventsQueueCh limits.QueueLimiter[RoutedTriggerEvent]
executionsSemaphore limits.ResourcePoolLimiter[int]
capCallsSemaphore limits.ResourcePoolLimiter[int]

Expand Down Expand Up @@ -125,13 +125,6 @@ type triggerCapability struct {
method string
}

type enqueuedTriggerEvent struct {
triggerCapID string
triggerIndex int
timestamp time.Time
event capabilities.TriggerResponse
}

func TriggerRegistrationID(workflowID string, triggerIndex int) string {
return fmt.Sprintf("trigger_reg_%s_%d", workflowID, triggerIndex)
}
Expand Down Expand Up @@ -396,12 +389,22 @@ func (e *Engine) Put(ctx context.Context, event RoutedTriggerEvent) error { // t
e.cfg.Hooks.OnTriggerEventDropped(triggerID, eventID, "draining")
return ErrEngineDraining
}
if err := e.allTriggerEventsQueueCh.Put(ctx, enqueuedTriggerEvent{
triggerCapID: triggerID,
triggerIndex: idx,
timestamp: e.cfg.Clock.Now(),
event: event.Event,
}); err != nil {

// Stamp the deadline once at dispatch: observedAt + queue timeout.
// If ObservedAt is not set, use the current time.
if event.ObservedAt.IsZero() {

@cedric-cordenier cedric-cordenier Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@agparadiso What is the correct behaviour here? I would expect that we control both sides of the interface here, inside two components of the core node -- i.e. we should either always be setting it, or never set it here.

If it's missing when it shouldn't be, let's treat it as a validation error.

@agparadiso agparadiso Sep 2, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The end correct behavior is that the dispatcher will be the only one setting this and it will always set it. I agree this should return a validation error. This Put method is going to be removed in a couple of tasks, but I'll address it on the current pr (the one building the dispatcher) as this had auto-merge enabled. 🙇🏼

event.ObservedAt = e.cfg.Clock.Now()
}
queueTimeout, err := e.cfg.LocalLimiters.TriggerEventQueueTime.Limit(ctx)

@cedric-cordenier cedric-cordenier Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: consider calling this ...QueueTimeout to be consistent with the variable name

if err != nil {
e.logger().Errorw("Failed to get trigger event queue time limit", "err", err)
tm := e.metrics.With(platform.KeyTriggerID, triggerID)
tm.IncrementTriggerEventDroppedTotal(ctx, monitoring.TriggerDropReasonQueueAgeLimitReadFailed)
return ErrEnqueueFailed
}
event.Deadline = event.ObservedAt.Add(queueTimeout)

if err := e.allTriggerEventsQueueCh.Put(ctx, event); err != nil {
tm := e.metrics.With(platform.KeyTriggerID, triggerID)
tm.IncrementTriggerEventEnqueueDroppedCounter(ctx)
retErr := ErrEnqueueFailed
Expand Down Expand Up @@ -808,30 +811,27 @@ func (e *Engine) handleAllTriggerEvents(ctx context.Context) {
if err != nil {
return
}
eventID := queueHead.event.Event.ID
triggerMetricLabels := e.metrics.With(platform.KeyTriggerID, queueHead.triggerCapID)
eventID := queueHead.Event.Event.ID
triggerMetricLabels := e.metrics.With(platform.KeyTriggerID, queueHead.TriggerCapID)
if e.Draining() {
triggerMetricLabels.IncrementTriggerEventDequeueDroppedCounter(ctx)
triggerMetricLabels.IncrementTriggerEventDroppedTotal(ctx, monitoring.TriggerDropReasonDequeueDraining)
e.cfg.Hooks.OnTriggerEventDropped(queueHead.triggerCapID, eventID, "draining")
e.logger().Infow("Engine is draining, stopping trigger handling loop", "eventID", eventID, "triggerID", queueHead.triggerCapID)
e.cfg.Hooks.OnTriggerEventDropped(queueHead.TriggerCapID, eventID, "draining")
e.logger().Infow("Engine is draining, stopping trigger handling loop", "eventID", eventID, "triggerID", queueHead.TriggerCapID)
return
}
eventAge := e.cfg.Clock.Now().Sub(queueHead.timestamp)

now := e.cfg.Clock.Now()
eventAge := now.Sub(queueHead.ObservedAt)
e.logger().Debugw("Popped a trigger event from the queue", "eventID", eventID, "eventAgeMs", eventAge.Milliseconds())
triggerMetricLabels.RecordTriggerEventQueueWaitSeconds(ctx, eventAge.Seconds())
triggerEventMaxAge, err := e.cfg.LocalLimiters.TriggerEventQueueTime.Limit(ctx)
if err != nil {
e.logger().Errorw("Failed to get trigger event queue time limit", "err", err)
triggerMetricLabels.IncrementTriggerEventDroppedTotal(ctx, monitoring.TriggerDropReasonQueueAgeLimitReadFailed)
continue
}
if eventAge > triggerEventMaxAge {
e.logger().Warnw("Trigger event is too old, skipping execution", "triggerID", queueHead.triggerCapID, "eventID", eventID, "eventAgeMs", eventAge.Milliseconds())
if now.After(queueHead.Deadline) {
e.logger().Warnw("Trigger event is too old, skipping execution", "triggerID", queueHead.TriggerCapID, "eventID", eventID, "eventAgeMs", eventAge.Milliseconds())
triggerMetricLabels.IncrementTriggerEventExpiredCounter(ctx)
triggerMetricLabels.IncrementTriggerEventDroppedTotal(ctx, monitoring.TriggerDropReasonExpired)
continue
}

semWaitStart := e.cfg.Clock.Now()
free, err := e.executionsSemaphore.Wait(ctx, 1) // block if too many concurrent workflow executions
triggerMetricLabels.RecordExecutionSemaphoreWaitSeconds(ctx, e.cfg.Clock.Now().Sub(semWaitStart).Seconds())
Expand All @@ -843,25 +843,17 @@ func (e *Engine) handleAllTriggerEvents(ctx context.Context) {

e.srvcEng.GoCtx(context.WithoutCancel(ctx), func(ctx context.Context) {
defer free()
routed := RoutedTriggerEvent{
WorkflowID: e.cfg.WorkflowID,
TriggerCapID: queueHead.triggerCapID,
TriggerIndex: queueHead.triggerIndex,
ObservedAt: queueHead.timestamp,
// Deadline: TODO: will be addressed on CRE-6175
SequenceNumber: 0, // reserved, always 0 in M1
Event: queueHead.event,
}

// Legacy path: startExecution handles all errors internally (metrics, ACK, hooks).
// This logs eventID context at the call site; the future dispatcher admitter
// (CRE-6176) will use this error for admission decisions.
if err := e.ExecuteTrigger(ctx, routed); err != nil {
if err := e.ExecuteTrigger(ctx, queueHead); err != nil {
// Dedup and shard-denial are expected outcomes (the event is handled,
// just not executed here), so they log at info rather than error level.
if errors.Is(err, ErrDuplicateExecution) || errors.Is(err, ErrShardDeniedNotOwner) {
e.logger().Infow("Skipping trigger event execution", "triggerID", routed.TriggerCapID, "eventID", routed.Event.Event.ID, "err", err)
e.logger().Infow("Skipping trigger event execution", "triggerID", queueHead.TriggerCapID, "eventID", queueHead.Event.Event.ID, "err", err)
} else {
e.logger().Errorw("Failed to execute trigger event", "triggerID", routed.TriggerCapID, "eventID", routed.Event.Event.ID, "err", err)
e.logger().Errorw("Failed to execute trigger event", "triggerID", queueHead.TriggerCapID, "eventID", queueHead.Event.Event.ID, "err", err)
}
}
})
Expand Down
Loading
Loading