Skip to main content

Delayed / Scheduled Actions

Overview​

Many workflows need to "wait" before doing the next thing — send a follow-up message three days after a purchase, or watch whether the customer clicks a link within 24 hours and only resend if they do not.

Rather than relying on the broker's delayed messages, the system records the work into the scheduled_action table and has a cron sweep for due rows every minute. This approach is inspectable, editable, and survives worker restarts.

Business Flow​

Scheduling​

  • When TriggerService.executeAction finds actionConfig.delay set, it calls ScheduleAction instead of dispatching the action immediately.
  • execute_at is computed per mode:
    • in the normal mode, it is the current time plus delay.seconds
    • in waitForTracking mode, execute_at is set to the expiry time (from expiresSeconds or delaySeconds) and expires_at is recorded alongside it, expressing "wait until the deadline unless someone clicks first"
  • If delay.reEvaluateConditions = true, the sourceConfig is stored as well so the conditions can be re-evaluated when the action actually comes due — the customer's situation may have changed by then.
  • Before writing the row, Redis is checked for a tracking click that already arrived, guarding against a race condition.

Processing due actions​

This runs every minute on two profiles: trigger-worker via ScheduledActionService.ProcessDueActions, and cron-scheduler via ScheduledActionScannerService.Run, which publishes onto the queue instead.

  1. recoverStaleProcessing recovers rows stuck in processing for more than five minutes, which happens when a worker dies mid-flight.
  2. claimBatch atomically claims 500 rows at a time using FOR UPDATE SKIP LOCKED, so multiple replicas can work concurrently without colliding.
  3. Each action is handled by processAction:
    • load the current line_user
    • if a sourceConfig was stored, call EvaluateConditions again; if it no longer passes, cancel the action
    • if it passes, build a synthetic TriggerRule and call the trigger engine's ExecuteAction
    • retry up to three times (maxRetries) before marking the action failed
  4. handleExpired deals with actions past their expires_at that were never unlocked, following whatever the configuration specifies — typically taking the "did not click" branch.
  5. Loop back to claimBatch until no work remains.

Early unlock​

If the user clicks the link while the system is still waiting, the tracking consumer calls checkAndCompleteDelayWait, which makes that workflow's action run immediately.

Key Files & Functions​

  • internal/scheduledaction/scheduledaction.go
    • ScheduledActionService.ScheduleAction(ctx, params) — writes into scheduled_action
    • ProcessDueActions(ctx) — the per-minute cron on the trigger-worker profile
    • claimBatch() (using FOR UPDATE SKIP LOCKED), processAction(), recoverStaleProcessing(), handleExpired(), updateStatus(), getLineUser()
    • the constants batchSize = 500, maxRetries = 3, staleProcessingMinute = 5
    • the TriggerExecutor interface (EvaluateConditions, ExecuteAction), which breaks the circular dependency
  • internal/actionexecute/scheduler.go — ActionSchedulerService.ScheduleAction(), a lightweight variant used on the main profile that only inserts and runs no cron
  • internal/cronscheduler/scheduled_action_scanner.go — ScheduledActionScannerService.Run() on the cron-scheduler profile, which claims rows and publishes them to scheduled_action_fire
  • internal/trigger/service_events.go — ProcessScheduledActionFire() and handleScheduledActionExpired()
  • cmd/worker/main.go — runTriggerWorker(), which cross-wires the two services
  • Queue: scheduled_action_fire, consumed on trigger-worker and published from cron-scheduler

Connections to Other Services​

  • The scheduled_action table — key columns are execute_at, expires_at, status (pending, processing, done, failed, cancelled), retry_count, action_type, action_config, source_config, workflow_id, and workflow_node_path
  • Other tables: trigger_rule (the originating rule) and line_user (for re-evaluating conditions)
  • Redis — checks for tracking clicks that arrive ahead of the deadline
  • RabbitMQ — the scheduled_action_fire queue, with action_execute as the final destination
  • Directly connected to the Trigger / Workflow Automation Engine and Workflow Action Execution