Activities

Activity execution, scheduling, and retry behavior in Temporal Server

Activities

Activities are the mechanism in Temporal for executing non-deterministic operations or side effects like API calls, database operations, or file I/O. Unlike workflow code which must be deterministic, activity code can interact with the outside world.

Activity Execution Model

Activities are scheduled by workflow code and executed by workers. The History Service orchestrates the activity lifecycle through events and task queues.

Activity Scheduling Flow

Activity State Machine

The History Service tracks activity state in Mutable State:

1. Scheduled

   Activity task created but not yet picked up by a worker. The ActivityTaskScheduled event is written to history.

2. Started

   Worker has picked up the task and begun execution. The ActivityTaskStarted event records which worker and when.

3. Completed / Failed / Timed Out / Canceled

   Activity reaches terminal state. Result or failure is recorded in corresponding event.

// From service/history/workflow/activity.go
func GetActivityState(ai *persistencespb.ActivityInfo) enumspb.PendingActivityState {
    if ai.CancelRequested {
        return enumspb.PENDING_ACTIVITY_STATE_CANCEL_REQUESTED
    }
    if ai.StartedEventId != common.EmptyEventID {
        return enumspb.PENDING_ACTIVITY_STATE_STARTED
    }
    return enumspb.PENDING_ACTIVITY_STATE_SCHEDULED
}

Activity Info

The server maintains rich metadata for each activity in ActivityInfo:

ActivityInfo Structure

type ActivityInfo struct {
    ScheduledEventId int64      // Event ID when activity was scheduled
    StartedEventId   int64      // Event ID when activity started (or empty)
    ActivityId       string     // User-provided activity ID
    RequestId        string     // Unique request ID for deduplication
    
    ScheduledTime    *timestamp  // When activity was scheduled
    StartedTime      *timestamp  // When worker picked up task
    
    TaskQueue        string      // Which task queue to dispatch to
    ActivityType     string      // Activity function name
    
    // Retry state
    Attempt               int32
    RetryLastFailure      *Failure
    RetryLastWorkerIdentity string
    
    // Heartbeat tracking
    LastHeartbeatUpdateTime *timestamp
    LastHeartbeatDetails    *Payloads
    
    // Timeout configuration
    ScheduleToStartTimeout time.Duration
    ScheduleToCloseTimeout time.Duration
    StartToCloseTimeout    time.Duration
    HeartbeatTimeout       time.Duration
    
    CancelRequested bool
    Paused          bool
}

Activity Timeouts

Temporal enforces four types of activity timeouts to prevent stuck activities:

Schedule-To-Start

Maximum time activity can wait in the task queue before being picked up by a worker. Detects unavailable workers.

Start-To-Close

Maximum execution time from when worker starts until completion. Detects hung activity execution.

Schedule-To-Close

End-to-end timeout including queue time, execution, and all retries. Overall deadline for activity.

Heartbeat

Maximum time between heartbeats from long-running activities. Detects crashed workers.

Timeout Implementation

Timeouts are implemented as Timer Tasks in the History Service:

// From service/history/tasks/activity_task_timer.go
// Timer tasks are created when activity is scheduled/started:
// - ActivityRetryTimerTask: For retry backoff
// - ActivityTimeoutTask: For timeout enforcement

When a timer fires, the queue processor:

1. Loads the workflow’s mutable state
2. Checks if activity is still in expected state
3. Times out the activity or triggers retry
4. Schedules a new workflow task

Activity Retry

Activities support automatic retry with exponential backoff:

Retry Policy

{
  "initialInterval": "1s",
  "backoffCoefficient": 2.0,
  "maximumInterval": "100s",
  "maximumAttempts": 0,  // infinite
  "nonRetryableErrorTypes": ["InvalidArgument"]
}

Retry Logic

1. Activity fails

   Worker sends RespondActivityTaskFailed with failure information

2. History Service evaluates retry policy

   Checks attempt count, error type, and calculates next backoff interval

3. Schedule retry or fail workflow

   If retryable: Create ActivityRetryTimer task and keep activity in Scheduled state If not retryable: Append ActivityTaskFailed event and schedule workflow task

4. Timer fires

   Activity is re-dispatched to Matching Service for another attempt

// From service/history/workflow/activity.go
func UpdateActivityInfoForRetries(
    ai *persistencespb.ActivityInfo,
    version int64,
    attempt int32,
    failure *failurepb.Failure,
    nextScheduledTime *timestamppb.Timestamp,
    isActivityRetryStampIncrementEnabled bool,
) {
    ai.Attempt = attempt
    ai.ScheduledTime = nextScheduledTime
    ai.StartedEventId = common.EmptyEventID  // Reset started state
    ai.RetryLastFailure = failure
    ai.RetryLastWorkerIdentity = ai.StartedIdentity
    
    // Mark timers for recreation
    ai.TimerTaskStatus &^= TimerTaskStatusCreatedHeartbeat | 
                          TimerTaskStatusCreatedStartToClose
}

Activity retry attempts do NOT create new history events. Only the final success or non-retryable failure creates an event. This keeps history bounded for activities with many retries.

Activity Heartbeating

Long-running activities should heartbeat to prove they’re still alive:

Heartbeat Mechanism

1. Activity code calls heartbeat API with optional progress payload
2. Worker sends HeartbeatActivityTask RPC to History Service
3. History Service updates activity info with heartbeat time and details
4. Heartbeat timer is reset to detect future missed heartbeats

Heartbeat Benefits

Failure Detection

Quickly detect when worker or activity has crashed without waiting for full timeout

Progress Tracking

Application can query activity details to show progress to users

Resumption

Activity can resume from last heartbeat point if worker crashes

Observability

Heartbeat timestamps visible in workflow execution history

// Heartbeat details are stored in ActivityInfo
if ai.LastHeartbeatUpdateTime != nil && !ai.LastHeartbeatUpdateTime.AsTime().IsZero() {
    p.LastHeartbeatTime = ai.LastHeartbeatUpdateTime
    p.HeartbeatDetails = ai.LastHeartbeatDetails
}

Activity Cancellation

Workflows can request activity cancellation:

Cancellation Flow

Activity cancellation is a request, not a guarantee. The activity code must explicitly check for cancellation and handle it gracefully.

Local Activities

Local activities are a special optimization for very short activities:

• Execute in the same worker process as workflow
• Not recorded in history until completion
• Lower latency (no History Service roundtrip)
• Limited to short operations (seconds)
• No cross-worker routing

Use local activities for fast operations like cache lookups or simple calculations that don’t justify the overhead of full activity scheduling.

Activity Dispatch

Activities are dispatched through the Matching Service:

1. Transfer Task created in History Service
2. Queue Processor reads task and calls Matching Service
3. Matching Service adds to appropriate task queue partition
4. Worker polls and receives activity task
5. Task includes activity input, attempt number, heartbeat details

// From service/history/transfer_queue_active_task_executor.go
func processActivityTask(task *tasks.ActivityTask) error {
    // Load workflow mutable state
    // Get activity info
    // Call Matching Service to enqueue activity task
    return tm.matchingClient.AddActivityTask(ctx, &matchingservice.AddActivityTaskRequest{
        NamespaceId: task.NamespaceID,
        Execution:   task.WorkflowExecution,
        TaskQueue:   activityInfo.TaskQueue,
        // ...
    })
}

Related Concepts

• Workflows - Schedule and orchestrate activities
• Task Queues - Route activity tasks to workers
• Workers - Execute activity code