Error Handling

Patterns for handling errors in workflows.

Task-Level Errors

Returning TaskFailure

class MyTaskExecutor extends TaskExecutor {
  @override
  String get schemaType => 'task.myTask';

  @override
  String get name => 'My Task';

  @override
  Future<TaskResult> execute(ExecutionContext context) async {
    // Validation errors - don't retry
    if (context.get<dynamic>('required') == null) {
      return TaskFailure(
        errorType: ErrorType.validation,
        message: 'Required field is missing',
        isRetryable: false,
      );
    }

    try {
      final result = await externalService.call();
      return TaskSuccess(output: result);
    } on TimeoutException {
      // Transient error - retry
      return TaskFailure(
        errorType: ErrorType.timeout,
        message: 'Service timeout',
        isRetryable: true,
      );
    } on ActivityException {
      // Activity error - don't retry
      return TaskFailure(
        errorType: ErrorType.activity,
        message: 'Activity execution failed',
        isRetryable: false,
      );
    } catch (e) {
      // Unknown error
      return TaskFailure(
        errorType: ErrorType.internal,
        message: 'Unexpected error: $e',
        isRetryable: false,
      );
    }
  }
}

Error Types

Type	Description	Retryable
validation	Invalid input	No
timeout	Operation timeout	Yes
activity	Activity execution failed	No
condition	Gateway condition evaluation failed	No
internal	Internal engine error	No
cancelled	Operation was cancelled	No

Workflow-Level Error Handling

Error Routing with Gateway

builder
    .task('riskyOperation', execute: (ctx) async {
      try {
        final result = await doWork();
        return TaskSuccess(output: {'success': true, 'result': result});
      } catch (e) {
        return TaskSuccess(output: {
          'success': false,
          'error': {'message': e.toString(), 'retryable': e is TimeoutException},
        });
      }
    })

    .oneOf('checkResult', [
      Branch.whenFn((o) => o['success'] == true, then: 'continueNormal'),
      Branch.whenFn((o) => o['error']?['retryable'] == true, then: 'retryOperation'),
      Branch.otherwise(then: 'handleError'),
    ])

    .task('retryOperation', execute: (ctx) async {
      // Increment retry count - use getAny for accumulated output
      final retries = (ctx.getAny<int>('retryCount') ?? 0) + 1;
      return {'retryCount': retries, 'shouldRetry': retries < 3};
    })

    .oneOf('checkRetry', [
      Branch.whenFn((o) => o['shouldRetry'] == true, then: 'riskyOperation'),
      Branch.otherwise(then: 'handleError'),
    ])

    .task('handleError', execute: (ctx) async {
      await notifyService.alertError(ctx.input);  // Previous node output
      return {'errorHandled': true};
    })

    .end('success')
    .end('failure');

Compensation Pattern

Undo completed work if later step fails:

builder
    // Step 1: Reserve inventory
    .task('reserveInventory', execute: (ctx) async {
      final reservation = await inventoryService.reserve(ctx.getRequired<List>('items'));
      return {'reservationId': reservation.id};
    })

    // Step 2: Charge payment
    .task('chargePayment', execute: (ctx) async {
      try {
        final charge = await paymentService.charge(
          ctx.getAny<String>('paymentMethod')!,  // From earlier node via accumulated
          ctx.getAny<num>('amount')!,
        );
        return {'success': true, 'chargeId': charge.id};
      } catch (e) {
        return {'success': false, 'error': e.toString()};
      }
    })

    // Check payment result
    .oneOf('checkPayment', [
      Branch.whenFn((o) => o['success'] == true, then: 'confirmOrder'),
      Branch.otherwise(then: 'compensateInventory'),
    ])

    // Compensation: Release inventory
    .task('compensateInventory', execute: (ctx) async {
      await inventoryService.release(ctx.getAny<String>('reservationId')!);
      return {'compensated': true};
    })

    .end('completed')
    .end('paymentFailed');

Saga Pattern

Long-running transactions with compensation:

// Each step has a compensating action
final steps = [
  {'action': 'createOrder', 'compensate': 'cancelOrder'},
  {'action': 'reserveInventory', 'compensate': 'releaseInventory'},
  {'action': 'chargePayment', 'compensate': 'refundPayment'},
  {'action': 'shipOrder', 'compensate': 'cancelShipment'},
];

builder
    .task('executeStep', execute: (ctx) async {
      final stepIndex = ctx.getAny<int>('currentStep') ?? 0;
      final step = steps[stepIndex];

      try {
        await executeAction(step['action'], ctx.input);
        return {
          'currentStep': stepIndex + 1,
          'completedSteps': [...(ctx.getAny<List>('completedSteps') ?? []), step],
          'success': true,
        };
      } catch (e) {
        return {
          'success': false,
          'failedAt': stepIndex,
          'error': e.toString(),
        };
      }
    })

    .oneOf('checkStep', [
      Branch.whenFn((o) => o['success'] == false, then: 'startCompensation'),
      Branch.whenFn((o) => o['currentStep'] >= steps.length, then: 'complete'),
      Branch.otherwise(then: 'executeStep'),
    ])

    .task('compensate', execute: (ctx) async {
      final completed = ctx.getAny<List>('completedSteps') ?? [];
      // Compensate in reverse order
      for (final step in completed.reversed) {
        await executeAction(step['compensate'], ctx.input);
      }
      return {'compensated': true};
    });

Timeout Handling

Task Timeout

class TimeoutAwareTaskExecutor extends TaskExecutor {
  final Duration timeout;

  TimeoutAwareTaskExecutor({this.timeout = const Duration(minutes: 5)});

  @override
  String get schemaType => 'task.timeoutAware';

  @override
  String get name => 'Timeout Aware Task';

  @override
  Future<TaskResult> execute(ExecutionContext context) async {
    try {
      final result = await doWork().timeout(timeout);
      return TaskSuccess(output: result);
    } on TimeoutException {
      return TaskFailure(
        errorType: ErrorType.timeout,
        message: 'Task exceeded ${timeout.inMinutes} minute timeout',
        isRetryable: true,
      );
    }
  }
}

Workflow Timeout

// External timer service sends signal after deadline
builder
    .signalWait('awaitCompletion', signal: 'work_done', storeAs: 'result')

    // Timer job runs periodically, checks for expired workflows
    // and sends 'workflow_timeout' signal

    .oneOf('checkResult', [
      Branch.whenFn((o) => o['result']?['completed'] == true, then: 'success'),
      Branch.whenFn((o) => o['result']?['timeout'] == true, then: 'handleTimeout'),
    ])

    .task('handleTimeout', execute: (ctx) async {
      await notifyService.notifyTimeout(ctx.input);  // Previous node output
      return {'timedOut': true};
    });

Best Practices

1. Categorize errors - Different handling for different types
2. Use retry sparingly - Only for transient errors
3. Limit retries - Prevent infinite loops
4. Implement compensation - Undo partial work
5. Log errors - For debugging and monitoring
6. Notify on failures - Don't let errors go unnoticed

Next Steps

• Idempotency - Crash recovery
• Best Practices - Design guidance
• Storage Adapters - Persistence