Temporal for Durable Workflows: Replacing Fragile Cron Jobs and Distributed Sagas

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# The fastest way: Temporal CLI with embedded server
brew install temporal  # macOS
# or: curl -sSf https://temporal.download/cli.sh | sh

temporal server start-dev \
  --ui-port 8080 \
  --db-filename /tmp/temporal.db  # SQLite persistence for dev

# Open http://localhost:8080 to see the Temporal Web UI
# gRPC endpoint: localhost:7233

1
2
3
4
5
6

go get go.temporal.io/sdk@latest
go get go.temporal.io/sdk/client
go get go.temporal.io/sdk/worker
go get go.temporal.io/sdk/workflow
go get go.temporal.io/sdk/activity
go get go.temporal.io/sdk/temporal

1

pip install temporalio

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101

// activities/payment.go
package activities

import (
    "context"
    "fmt"
    "time"

    "go.temporal.io/sdk/activity"
    "go.temporal.io/sdk/temporal"
)

type PaymentActivities struct {
    paymentClient *PaymentGatewayClient
    emailClient   *EmailClient
    inventoryDB   *InventoryDB
}

type OrderInput struct {
    OrderID    string  `json:"order_id"`
    CustomerID string  `json:"customer_id"`
    Amount     float64 `json:"amount"`
    Items      []Item  `json:"items"`
}

type PaymentResult struct {
    TransactionID string    `json:"transaction_id"`
    ChargedAt     time.Time `json:"charged_at"`
}

// ValidateInventory checks that all items are in stock.
// If an item is temporarily out of stock, return a retryable error.
// If an item is permanently unavailable, return a non-retryable error.
func (a *PaymentActivities) ValidateInventory(ctx context.Context, input OrderInput) error {
    logger := activity.GetLogger(ctx)
    logger.Info("Validating inventory", "order_id", input.OrderID)

    for _, item := range input.Items {
        available, err := a.inventoryDB.CheckAvailability(ctx, item.SKU, item.Quantity)
        if err != nil {
            // Transient DB error — Temporal will retry
            return fmt.Errorf("checking inventory for %s: %w", item.SKU, err)
        }
        if !available {
            // Permanent business error — don't retry
            return temporal.NewNonRetryableApplicationError(
                fmt.Sprintf("item %s is out of stock", item.SKU),
                "OUT_OF_STOCK",
                nil,
            )
        }
    }
    return nil
}

func (a *PaymentActivities) ChargePayment(ctx context.Context, input OrderInput) (*PaymentResult, error) {
    logger := activity.GetLogger(ctx)
    logger.Info("Charging payment", "order_id", input.OrderID, "amount", input.Amount)

    // Idempotency key: use order ID so double-charging is impossible even if retried
    result, err := a.paymentClient.Charge(ctx, PaymentRequest{
        IdempotencyKey: "order-charge-" + input.OrderID,
        CustomerID:     input.CustomerID,
        Amount:         input.Amount,
        Currency:       "USD",
    })
    if err != nil {
        return nil, fmt.Errorf("payment charge failed: %w", err)
    }

    return &PaymentResult{
        TransactionID: result.TransactionID,
        ChargedAt:     time.Now(),
    }, nil
}

func (a *PaymentActivities) ReserveInventory(ctx context.Context, input OrderInput) error {
    for _, item := range input.Items {
        if err := a.inventoryDB.Reserve(ctx, item.SKU, item.Quantity, input.OrderID); err != nil {
            return fmt.Errorf("reserving %s: %w", item.SKU, err)
        }
    }
    return nil
}

func (a *PaymentActivities) SendConfirmationEmail(ctx context.Context, input OrderInput, txnID string) error {
    return a.emailClient.Send(ctx, EmailRequest{
        To:       input.CustomerID,
        Template: "order-confirmation",
        Data: map[string]any{
            "order_id":       input.OrderID,
            "transaction_id": txnID,
            "items":          input.Items,
            "amount":         input.Amount,
        },
    })
}

func (a *PaymentActivities) RefundPayment(ctx context.Context, txnID string, amount float64) error {
    return a.paymentClient.Refund(ctx, txnID, amount)
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107

// workflows/order.go
package workflows

import (
    "time"

    "go.temporal.io/sdk/temporal"
    "go.temporal.io/sdk/workflow"
    "myapp/activities"
)

// OrderWorkflow orchestrates the complete order processing pipeline.
// This function MUST be deterministic — no direct I/O, no time.Now(), no random.
func OrderWorkflow(ctx workflow.Context, input activities.OrderInput) error {
    logger := workflow.GetLogger(ctx)
    logger.Info("Starting order workflow", "order_id", input.OrderID)

    // Activity options: how Temporal should call and retry each activity
    activityOpts := workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,  // Max time for one attempt
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    time.Second,
            BackoffCoefficient: 2.0,
            MaximumInterval:    30 * time.Second,
            MaximumAttempts:    5,
            // Don't retry business logic errors
            NonRetryableErrorTypes: []string{"OUT_OF_STOCK", "INVALID_CARD"},
        },
    }
    ctx = workflow.WithActivityOptions(ctx, activityOpts)

    var acts *activities.PaymentActivities // nil — Temporal resolves this at runtime

    // Step 1: Validate inventory
    if err := workflow.ExecuteActivity(ctx, acts.ValidateInventory, input).Get(ctx, nil); err != nil {
        return fmt.Errorf("inventory validation failed: %w", err)
    }

    // Step 2: Charge payment
    var payment activities.PaymentResult
    if err := workflow.ExecuteActivity(ctx, acts.ChargePayment, input).Get(ctx, &payment); err != nil {
        return fmt.Errorf("payment failed: %w", err)
    }

    // Step 3: Reserve inventory (compensate if this fails after payment)
    if err := workflow.ExecuteActivity(ctx, acts.ReserveInventory, input).Get(ctx, nil); err != nil {
        // Compensation: refund the payment before returning the error
        logger.Error("Inventory reservation failed, refunding", "error", err)

        refundCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
            StartToCloseTimeout: 60 * time.Second,
            RetryPolicy: &temporal.RetryPolicy{
                MaximumAttempts: 10, // Try harder to refund
            },
        })
        _ = workflow.ExecuteActivity(refundCtx, acts.RefundPayment,
            payment.TransactionID, input.Amount).Get(ctx, nil)

        return fmt.Errorf("inventory reservation failed: %w", err)
    }

    // Step 4: Wait up to 5 minutes for fulfillment system to acknowledge
    // (demonstrates timer + signal pattern)
    fulfillmentCtx, cancel := workflow.WithCancel(ctx)
    defer cancel()

    fulfillmentSignal := workflow.GetSignalChannel(ctx, "fulfillment-ack")

    selector := workflow.NewSelector(ctx)

    var fulfilled bool
    selector.AddReceive(fulfillmentSignal, func(c workflow.ReceiveChannel, more bool) {
        c.Receive(ctx, nil)
        fulfilled = true
    })

    selector.AddFuture(workflow.NewTimer(fulfillmentCtx, 5*time.Minute), func(f workflow.Future) {
        fulfilled = false
    })

    selector.Select(ctx) // Block until signal OR timer fires

    // Step 5: Send confirmation email regardless of fulfillment ack
    emailOpts := workflow.ActivityOptions{
        StartToCloseTimeout: 10 * time.Second,
        RetryPolicy: &temporal.RetryPolicy{
            MaximumAttempts: 20, // Really want this email to go out
            MaximumInterval: 5 * time.Minute,
        },
    }
    emailCtx := workflow.WithActivityOptions(ctx, emailOpts)

    if err := workflow.ExecuteActivity(emailCtx, acts.SendConfirmationEmail,
        input, payment.TransactionID).Get(ctx, nil); err != nil {
        // Log but don't fail the order — email failure is non-critical
        logger.Error("Failed to send confirmation email", "error", err)
    }

    if !fulfilled {
        logger.Warn("Fulfillment acknowledgment not received within 5 minutes",
            "order_id", input.OrderID)
    }

    logger.Info("Order workflow completed", "order_id", input.OrderID,
        "transaction_id", payment.TransactionID)
    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

// worker/main.go
package main

import (
    "log"

    "go.temporal.io/sdk/client"
    "go.temporal.io/sdk/worker"
    "myapp/activities"
    "myapp/workflows"
)

func main() {
    c, err := client.Dial(client.Options{
        HostPort: "localhost:7233",
    })
    if err != nil {
        log.Fatalf("Unable to create Temporal client: %v", err)
    }
    defer c.Close()

    // Create a worker that listens on the "orders" task queue
    w := worker.New(c, "orders", worker.Options{
        MaxConcurrentActivityExecutionSize: 100,
        MaxConcurrentWorkflowTaskExecutionSize: 50,
    })

    // Register workflow and activity implementations
    w.RegisterWorkflow(workflows.OrderWorkflow)

    acts := &activities.PaymentActivities{
        paymentClient: newPaymentClient(),
        emailClient:   newEmailClient(),
        inventoryDB:   newInventoryDB(),
    }
    w.RegisterActivity(acts)

    // Start the worker (blocks until interrupted)
    if err := w.Run(worker.InterruptCh()); err != nil {
        log.Fatalf("Worker failed: %v", err)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

// Starting a workflow from your HTTP handler or CLI
func StartOrder(c client.Client, order OrderInput) (string, error) {
    options := client.StartWorkflowOptions{
        // Deterministic ID: re-starting the same order is idempotent
        ID:        "order-" + order.OrderID,
        TaskQueue: "orders",

        // Workflow-level timeout (distinct from activity timeouts)
        WorkflowExecutionTimeout: 24 * time.Hour,

        // Prevent concurrent runs for same order
        WorkflowIDReusePolicy: enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
    }

    we, err := c.ExecuteWorkflow(context.Background(), options,
        workflows.OrderWorkflow, order)
    if err != nil {
        return "", fmt.Errorf("starting workflow: %w", err)
    }

    return we.GetID(), nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

// workflows/approval.go — a workflow that waits for human approval
const (
    SignalApprove = "approve"
    SignalReject  = "reject"
)

type ApprovalInput struct {
    RequestID   string  `json:"request_id"`
    Requester   string  `json:"requester"`
    Amount      float64 `json:"amount"`
    Description string  `json:"description"`
}

func ApprovalWorkflow(ctx workflow.Context, input ApprovalInput) (bool, error) {
    logger := workflow.GetLogger(ctx)

    // Set up signal channels
    approveSignal := workflow.GetSignalChannel(ctx, SignalApprove)
    rejectSignal := workflow.GetSignalChannel(ctx, SignalReject)

    // Notify approvers (via activity)
    actCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,
    })
    var notifyActs *activities.NotificationActivities
    if err := workflow.ExecuteActivity(actCtx, notifyActs.NotifyApprovers, input).Get(ctx, nil); err != nil {
        return false, err
    }

    // Wait for approval, rejection, or timeout (48 hours)
    selector := workflow.NewSelector(ctx)

    var approved bool
    var decided bool

    selector.AddReceive(approveSignal, func(c workflow.ReceiveChannel, more bool) {
        var approverID string
        c.Receive(ctx, &approverID)
        approved = true
        decided = true
        logger.Info("Request approved", "request_id", input.RequestID, "approver", approverID)
    })

    selector.AddReceive(rejectSignal, func(c workflow.ReceiveChannel, more bool) {
        var reason string
        c.Receive(ctx, &reason)
        approved = false
        decided = true
        logger.Info("Request rejected", "request_id", input.RequestID, "reason", reason)
    })

    selector.AddFuture(workflow.NewTimer(ctx, 48*time.Hour), func(f workflow.Future) {
        approved = false
        decided = true
        logger.Warn("Approval timed out", "request_id", input.RequestID)
    })

    // Wait until a decision is made
    for !decided {
        selector.Select(ctx)
    }

    // Send outcome notification via activity
    _ = workflow.ExecuteActivity(actCtx, notifyActs.NotifyOutcome, input, approved).Get(ctx, nil)

    return approved, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// Sending a signal from another service
func ApproveRequest(c client.Client, workflowID string, approverID string) error {
    return c.SignalWorkflow(
        context.Background(),
        workflowID,
        "",  // empty runID = signal the latest run
        workflows.SignalApprove,
        approverID,
    )
}

func RejectRequest(c client.Client, workflowID string, reason string) error {
    return c.SignalWorkflow(
        context.Background(),
        workflowID,
        "",
        workflows.SignalReject,
        reason,
    )
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

// workflows/order_with_state.go
const QueryCurrentState = "current-state"

type OrderState struct {
    Status        string    `json:"status"`
    CurrentStep   string    `json:"current_step"`
    PaymentTxnID  string    `json:"payment_txn_id,omitempty"`
    StartedAt     time.Time `json:"started_at"`
    LastUpdatedAt time.Time `json:"last_updated_at"`
}

func OrderWorkflowWithState(ctx workflow.Context, input activities.OrderInput) error {
    state := OrderState{
        Status:    "started",
        StartedAt: workflow.Now(ctx),  // workflow.Now(), NOT time.Now()
    }

    // Register a query handler — can be called any time the workflow is running
    if err := workflow.SetQueryHandler(ctx, QueryCurrentState, func() (OrderState, error) {
        return state, nil
    }); err != nil {
        return err
    }

    // Update state as the workflow progresses
    state.CurrentStep = "validating_inventory"
    state.LastUpdatedAt = workflow.Now(ctx)

    // ... run activities, updating state along the way ...

    state.CurrentStep = "charging_payment"
    // ... etc

    state.Status = "completed"
    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// Querying a running workflow from an HTTP handler
func GetOrderStatus(c client.Client, workflowID string) (*OrderState, error) {
    resp, err := c.QueryWorkflow(
        context.Background(),
        workflowID,
        "",
        workflows.QueryCurrentState,
    )
    if err != nil {
        return nil, fmt.Errorf("querying workflow: %w", err)
    }

    var state workflows.OrderState
    if err := resp.Get(&state); err != nil {
        return nil, fmt.Errorf("decoding state: %w", err)
    }
    return &state, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

// schedules/create.go
package schedules

import (
    "context"
    "log"

    "go.temporal.io/sdk/client"
    "go.temporal.io/sdk/temporal"
)

func CreateDailyReportSchedule(c client.Client) error {
    scheduleClient := c.ScheduleClient()

    handle, err := scheduleClient.Create(context.Background(),
        client.ScheduleOptions{
            ID: "daily-revenue-report",
            Spec: client.ScheduleSpec{
                // CronExpressions uses standard cron syntax
                CronExpressions: []string{"0 8 * * *"},  // 8am UTC daily

                // Or use structured intervals:
                // Intervals: []client.ScheduleIntervalSpec{{
                //     Every: 24 * time.Hour,
                //     Offset: 8 * time.Hour,  // Start at 8am
                // }},

                // Jitter: randomize start time by up to 5 minutes to avoid thundering herds
                Jitter: 5 * time.Minute,
            },
            Action: &client.ScheduleWorkflowAction{
                Workflow:  "GenerateDailyReport",
                TaskQueue: "reports",
                Args:      []interface{}{ReportInput{Type: "revenue", Region: "all"}},
            },
            Policy: client.SchedulePolicies{
                // If previous run is still going when next is scheduled, skip it
                Overlap: temporal.ScheduleOverlapPolicySkip,

                // If the server was down and we missed runs, catch up on at most 1
                CatchupWindow: 1 * time.Hour,
            },
            // Start paused (useful for initially reviewing a schedule)
            State: client.ScheduleState{
                Paused: false,
                Note:   "Daily revenue report — see runbook/reports.md",
            },
        },
    )
    if err != nil {
        return err
    }

    log.Printf("Created schedule: %s", handle.GetID())
    return nil
}

// Trigger a schedule immediately (useful for backfilling or testing)
func TriggerSchedule(c client.Client, scheduleID string) error {
    handle := c.ScheduleClient().GetHandle(context.Background(), scheduleID)
    return handle.Trigger(context.Background(), client.ScheduleTriggerOptions{
        Overlap: temporal.ScheduleOverlapPolicyAllowAll,
    })
}

// Pause a schedule (e.g., during a maintenance window)
func PauseSchedule(c client.Client, scheduleID, reason string) error {
    handle := c.ScheduleClient().GetHandle(context.Background(), scheduleID)
    return handle.Pause(context.Background(), client.SchedulePauseOptions{
        Note: reason,
    })
}

1
2

# crontab — fragile, no observability, dies silently
0 8 * * * /usr/local/bin/python /app/generate_report.py --type revenue 2>&1 | logger -t report

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

# Python SDK example for the report workflow
import asyncio
from datetime import timedelta
from temporalio import workflow, activity
from temporalio.client import Client
from temporalio.worker import Worker
from temporalio.common import RetryPolicy

@activity.defn
async def fetch_sales_data(start_date: str, end_date: str) -> dict:
    """Fetch sales data from the data warehouse."""
    # All the actual work happens here — retryable, observable
    async with aiohttp.ClientSession() as session:
        resp = await session.get(
            "http://data-warehouse/api/sales",
            params={"start": start_date, "end": end_date},
        )
        resp.raise_for_status()
        return await resp.json()

@activity.defn
async def generate_pdf_report(data: dict, report_type: str) -> str:
    """Generate PDF and upload to S3. Returns the S3 key."""
    pdf_bytes = render_report_template(data, report_type)
    key = f"reports/{report_type}/{datetime.utcnow():%Y/%m/%d}/report.pdf"
    await upload_to_s3(pdf_bytes, key)
    return key

@activity.defn
async def send_report_email(s3_key: str, recipients: list[str]) -> None:
    """Email the report link to recipients."""
    presigned_url = generate_presigned_url(s3_key, expiry_seconds=3600 * 24)
    await send_email(
        to=recipients,
        subject=f"Daily Revenue Report — {datetime.utcnow():%Y-%m-%d}",
        body=f"Your report is ready: {presigned_url}",
    )

@workflow.defn
class DailyReportWorkflow:
    @workflow.run
    async def run(self, report_type: str) -> str:
        # Use workflow.now() for deterministic timestamps
        today = workflow.now()
        start_date = (today - timedelta(days=1)).strftime("%Y-%m-%d")
        end_date = today.strftime("%Y-%m-%d")

        activity_opts = dict(
            start_to_close_timeout=timedelta(minutes=10),
            retry_policy=RetryPolicy(
                initial_interval=timedelta(seconds=5),
                backoff_coefficient=2.0,
                maximum_interval=timedelta(minutes=5),
                maximum_attempts=5,
            ),
        )

        # Each step is automatically retried if it fails
        data = await workflow.execute_activity(
            fetch_sales_data,
            args=[start_date, end_date],
            **activity_opts,
        )

        s3_key = await workflow.execute_activity(
            generate_pdf_report,
            args=[data, report_type],
            **activity_opts,
        )

        recipients = ["finance@company.com", "ceo@company.com"]
        await workflow.execute_activity(
            send_report_email,
            args=[s3_key, recipients],
            **activity_opts,
        )

        return s3_key


async def main():
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="reports",
        workflows=[DailyReportWorkflow],
        activities=[fetch_sales_data, generate_pdf_report, send_report_email],
    )
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

// workflows/batch_processor.go
// Process a large batch by fanning out to child workflows

type BatchInput struct {
    BatchID string   `json:"batch_id"`
    ItemIDs []string `json:"item_ids"`
}

func BatchProcessorWorkflow(ctx workflow.Context, input BatchInput) error {
    logger := workflow.GetLogger(ctx)
    logger.Info("Starting batch", "batch_id", input.BatchID, "count", len(input.ItemIDs))

    // Split into chunks of 100
    chunkSize := 100
    var futures []workflow.ChildWorkflowFuture

    for i := 0; i < len(input.ItemIDs); i += chunkSize {
        end := i + chunkSize
        if end > len(input.ItemIDs) {
            end = len(input.ItemIDs)
        }
        chunk := input.ItemIDs[i:end]

        childCtx := workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{
            WorkflowID:               fmt.Sprintf("%s-chunk-%d", input.BatchID, i/chunkSize),
            WorkflowExecutionTimeout: 30 * time.Minute,
            RetryPolicy: &temporal.RetryPolicy{
                MaximumAttempts: 3,
            },
        })

        // Start all child workflows in parallel (don't .Get() yet)
        future := workflow.ExecuteChildWorkflow(childCtx, ChunkProcessorWorkflow, ChunkInput{
            BatchID: input.BatchID,
            Items:   chunk,
        })
        futures = append(futures, future)
    }

    // Now wait for all children to complete, collecting errors
    var errs []error
    for _, future := range futures {
        if err := future.Get(ctx, nil); err != nil {
            errs = append(errs, err)
            logger.Error("Chunk failed", "error", err)
        }
    }

    if len(errs) > 0 {
        return fmt.Errorf("%d chunks failed out of %d", len(errs), len(futures))
    }

    logger.Info("Batch complete", "batch_id", input.BatchID)
    return nil
}

func ChunkProcessorWorkflow(ctx workflow.Context, input ChunkInput) error {
    actCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
        RetryPolicy:         &temporal.RetryPolicy{MaximumAttempts: 5},
    })

    var acts *activities.ProcessingActivities
    for _, itemID := range input.Items {
        if err := workflow.ExecuteActivity(actCtx, acts.ProcessItem, itemID).Get(ctx, nil); err != nil {
            return fmt.Errorf("processing item %s: %w", itemID, err)
        }
    }
    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

// workflows/saga.go — distributed transaction with compensation
func TravelBookingWorkflow(ctx workflow.Context, booking TravelBooking) error {
    logger := workflow.GetLogger(ctx)

    // Track compensations in reverse order
    var compensations []func(workflow.Context)

    actCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,
        RetryPolicy:         &temporal.RetryPolicy{MaximumAttempts: 3},
    })

    compensate := func() {
        // Run compensations in reverse order (LIFO)
        compensationCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
            StartToCloseTimeout: 60 * time.Second,
            RetryPolicy: &temporal.RetryPolicy{
                MaximumAttempts: 10, // Compensations must succeed
            },
        })
        for i := len(compensations) - 1; i >= 0; i-- {
            compensations[i](compensationCtx)
        }
    }

    var acts *activities.TravelActivities

    // Step 1: Book flight
    var flightBooking FlightBooking
    if err := workflow.ExecuteActivity(actCtx, acts.BookFlight, booking.Flight).
        Get(ctx, &flightBooking); err != nil {
        return fmt.Errorf("flight booking failed: %w", err)
    }
    compensations = append(compensations, func(cCtx workflow.Context) {
        if err := workflow.ExecuteActivity(cCtx, acts.CancelFlight,
            flightBooking.BookingID).Get(cCtx, nil); err != nil {
            logger.Error("Failed to cancel flight", "booking_id", flightBooking.BookingID, "error", err)
        }
    })

    // Step 2: Book hotel
    var hotelBooking HotelBooking
    if err := workflow.ExecuteActivity(actCtx, acts.BookHotel, booking.Hotel).
        Get(ctx, &hotelBooking); err != nil {
        logger.Error("Hotel booking failed, compensating", "error", err)
        compensate()
        return fmt.Errorf("hotel booking failed: %w", err)
    }
    compensations = append(compensations, func(cCtx workflow.Context) {
        if err := workflow.ExecuteActivity(cCtx, acts.CancelHotel,
            hotelBooking.BookingID).Get(cCtx, nil); err != nil {
            logger.Error("Failed to cancel hotel", "booking_id", hotelBooking.BookingID, "error", err)
        }
    })

    // Step 3: Book car rental
    if err := workflow.ExecuteActivity(actCtx, acts.BookCar, booking.Car).
        Get(ctx, nil); err != nil {
        logger.Error("Car rental failed, compensating", "error", err)
        compensate()
        return fmt.Errorf("car rental failed: %w", err)
    }

    logger.Info("All bookings confirmed",
        "flight", flightBooking.BookingID,
        "hotel", hotelBooking.BookingID)
    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

// workflows/order_versioned.go

const (
    // Bump this each time you make an incompatible change
    addFraudCheckVersion = 1
    parallelEmailVersion = 2
)

func OrderWorkflow(ctx workflow.Context, input OrderInput) error {
    // Original code (version 0): no fraud check
    // New code (version 1): add fraud check before payment
    v := workflow.GetVersion(ctx, "add-fraud-check", workflow.DefaultVersion, addFraudCheckVersion)
    if v >= addFraudCheckVersion {
        // Only new workflows run this — old replaying workflows skip it
        if err := workflow.ExecuteActivity(ctx, acts.FraudCheck, input).Get(ctx, nil); err != nil {
            return err
        }
    }

    // ... rest of workflow ...

    // Version 2: send email and SMS in parallel instead of sequentially
    v2 := workflow.GetVersion(ctx, "parallel-notifications", workflow.DefaultVersion, parallelEmailVersion)
    if v2 >= parallelEmailVersion {
        // New parallel implementation
        emailFuture := workflow.ExecuteActivity(ctx, acts.SendEmail, input)
        smsFuture := workflow.ExecuteActivity(ctx, acts.SendSMS, input)
        emailFuture.Get(ctx, nil)
        smsFuture.Get(ctx, nil)
    } else {
        // Old sequential implementation (for in-flight workflows)
        workflow.ExecuteActivity(ctx, acts.SendEmail, input).Get(ctx, nil)
        workflow.ExecuteActivity(ctx, acts.SendSMS, input).Get(ctx, nil)
    }

    return nil
}

// After ALL workflows running the old version have completed,
// you can clean up the version check:
// v := workflow.GetVersion(ctx, "add-fraud-check", addFraudCheckVersion, addFraudCheckVersion)
// (This asserts the version is always >= 1, removing the branch)

1
2
3
4
5
6
7
8

helm repo add temporal https://charts.temporal.io
helm repo update

# Deploy Temporal with PostgreSQL backend
helm install temporal temporal/temporal \
  --namespace temporal \
  --create-namespace \
  --values temporal-values.yaml

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

# temporal-values.yaml
server:
  replicaCount: 3
  config:
    persistence:
      defaultStore: postgres-default
      visibilityStore: postgres-visibility
      datastores:
        postgres-default:
          sql:
            driver: postgres
            host: postgres.temporal.svc.cluster.local
            port: 5432
            database: temporal
            user: temporal
            existingSecret: temporal-db-secret
            secretKey: password
            maxConns: 20
            maxConnLifetime: "1h"
        postgres-visibility:
          sql:
            driver: postgres
            host: postgres.temporal.svc.cluster.local
            port: 5432
            database: temporal_visibility
            user: temporal
            existingSecret: temporal-db-secret
            secretKey: password

  frontend:
    replicaCount: 2
    resources:
      requests:
        cpu: 200m
        memory: 256Mi
      limits:
        cpu: 1000m
        memory: 512Mi

  history:
    replicaCount: 3
    resources:
      requests:
        cpu: 500m
        memory: 512Mi
      limits:
        cpu: 2000m
        memory: 1Gi

  matching:
    replicaCount: 2
    resources:
      requests:
        cpu: 200m
        memory: 256Mi

  worker:
    replicaCount: 1

web:
  enabled: true
  replicaCount: 1
  ingress:
    enabled: true
    hosts:
      - host: temporal.internal.mycompany.com
        paths:
          - path: /
            pathType: Prefix

# Enable Prometheus metrics
prometheus:
  enabled: false  # Use your existing Prometheus stack instead

# Namespace auto-provisioning
admintools:
  enabled: true

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# Your application worker deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-worker
  namespace: myapp
spec:
  replicas: 5
  selector:
    matchLabels:
      app: order-worker
  template:
    metadata:
      labels:
        app: order-worker
    spec:
      containers:
      - name: worker
        image: myregistry/order-worker:latest
        env:
        - name: TEMPORAL_HOST
          value: "temporal-frontend.temporal.svc.cluster.local:7233"
        - name: TEMPORAL_NAMESPACE
          value: "production"
        - name: TEMPORAL_TASK_QUEUE
          value: "orders"
        resources:
          requests:
            cpu: "500m"
            memory: "512Mi"
          limits:
            cpu: "2"
            memory: "2Gi"
        # Graceful shutdown: let in-flight activities complete
        lifecycle:
          preStop:
            exec:
              command: ["/bin/sh", "-c", "sleep 30"]
        terminationGracePeriodSeconds: 60

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# Create a namespace for production (done once)
temporal operator namespace create \
  --namespace production \
  --retention 30d \  # How long to keep completed workflow histories
  --description "Production workflows"

# Create a namespace for staging
temporal operator namespace create \
  --namespace staging \
  --retention 7d

# List namespaces
temporal operator namespace list

# Describe a namespace
temporal operator namespace describe --namespace production

1
2
3
4
5
6
7
8

# Temporal exposes Prometheus metrics — add to your scrape config
scrape_configs:
  - job_name: temporal
    static_configs:
      - targets:
        - temporal-frontend.temporal.svc.cluster.local:9090
        - temporal-history.temporal.svc.cluster.local:9090
        - temporal-matching.temporal.svc.cluster.local:9090

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# Workflow failure rate (alert if >1%)
rate(temporal_workflow_failed_total[5m])
/
rate(temporal_workflow_completed_total[5m])
> 0.01

# Activity failure rate before final success
rate(temporal_activity_failed_total[5m]) > 10

# Workflow task schedule-to-start latency (workers falling behind)
histogram_quantile(0.99, rate(temporal_workflow_task_schedule_to_start_latency_bucket[5m]))
> 5  # Alert if p99 > 5 seconds

# Activity task queue backlog
temporal_activity_task_schedule_to_start_latency_sum > 60  # >60s backlog

# Persistence errors (database problems)
rate(temporal_persistence_error_with_type_total[5m]) > 0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

func (a *Activities) ChargeCard(ctx context.Context, orderID string, amount float64) error {
    // Use a stable idempotency key derived from workflow inputs
    // This key must be the same on every retry attempt
    idempotencyKey := fmt.Sprintf("charge-%s-v1", orderID)

    _, err := a.stripe.Charges.New(&stripe.ChargeParams{
        Amount:         stripe.Int64(int64(amount * 100)),
        Currency:       stripe.String("usd"),
        IdempotencyKey: &idempotencyKey,
    })
    return err
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

func (a *Activities) ProcessLargeFile(ctx context.Context, fileKey string) error {
    file := downloadFile(fileKey)
    total := len(file.Chunks)

    for i, chunk := range file.Chunks {
        // Check if Temporal has requested cancellation
        if err := ctx.Err(); err != nil {
            return temporal.NewCanceledError()
        }

        // Process the chunk
        if err := processChunk(ctx, chunk); err != nil {
            return err
        }

        // Heartbeat with progress — Temporal will reschedule if heartbeat stops
        activity.RecordHeartbeat(ctx, map[string]int{
            "processed": i + 1,
            "total":     total,
        })
    }
    return nil
}

1
2
3
4
5
6

// Activity options for long-running activities
actCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
    StartToCloseTimeout:    2 * time.Hour,    // Total time budget
    HeartbeatTimeout:       30 * time.Second, // Must heartbeat within 30s or Temporal retries
    ScheduleToCloseTimeout: 4 * time.Hour,
})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

# Start the dev server
temporal server start-dev

# List running workflows
temporal workflow list --namespace production

# Describe a specific workflow
temporal workflow describe --workflow-id order-abc123 --namespace production

# Show full event history
temporal workflow show --workflow-id order-abc123 --namespace production

# Send a signal
temporal workflow signal \
  --workflow-id approval-req-456 \
  --name approve \
  --input '"approver@company.com"' \
  --namespace production

# Query a workflow
temporal workflow query \
  --workflow-id order-abc123 \
  --type current-state \
  --namespace production

# Terminate a stuck workflow
temporal workflow terminate \
  --workflow-id order-abc123 \
  --reason "stuck in invalid state, will re-process" \
  --namespace production

# Cancel a workflow gracefully (triggers cancellation handling)
temporal workflow cancel --workflow-id order-abc123 --namespace production

# List schedules
temporal schedule list --namespace production

# Trigger a schedule immediately
temporal schedule trigger --schedule-id daily-report --namespace production

# Reset a workflow to a previous event (for recovering from bugs)
temporal workflow reset \
  --workflow-id order-abc123 \
  --event-id 15 \
  --reason "resetting after bug fix in step 3" \
  --namespace production