Inngest Durable Functions

Inngest Durable Functions

Master Inngest's durable execution model for building fault-tolerant, long-running workflows. This skill covers the complete lifecycle from triggers to error handling.

These skills are focused on TypeScript. For Python or Go, refer to the Inngest documentation for language-specific guidance. Core concepts apply across all languages.

Core Concepts You Need to Know

Durable Execution Model

• Each step should encapsulate side-effects and non-deterministic code
• Memoization prevents re-execution of completed steps
• State persistence survives infrastructure failures
• Automatic retries with configurable retry count

Step Execution Flow

// ❌ BAD: Non-deterministic logic outside steps
async ({ event, step }) => {
  const timestamp = Date.now(); // This runs multiple times!

  const result = await step.run("process-data", () => {
    return processData(event.data);
  });
};

// ✅ GOOD: All non-deterministic logic in steps
async ({ event, step }) => {
  const result = await step.run("process-with-timestamp", () => {
    const timestamp = Date.now(); // Only runs once
    return processData(event.data, timestamp);
  });
};

Function Limits

Every Inngest function has these hard limits:

• Maximum 1,000 steps per function run
• Maximum 4MB returned data for each step
• Maximum 32MB combined function run state including, event data, step output, and function output
• Each step = separate HTTP request (~50-100ms overhead)

If you're hitting these limits, break your function into smaller functions connected via step.invoke() or step.sendEvent().

When to Use Steps

Always wrap in `step.run()`:

• API calls and network requests
• Database reads and writes
• File I/O operations
• Any non-deterministic operation
• Anything you want retried independently on failure

Never wrap in `step.run()`:

• Pure calculations and data transformations
• Simple validation logic
• Deterministic operations with no side effects
• Logging (use outside steps)

Function Creation

Basic Function Structure

const processOrder = inngest.createFunction(
  {
    id: "process-order", // Unique, never change this
    triggers: [{ event: "order/created" }],
    retries: 4, // Default: 4 retries per step
    concurrency: 10 // Max concurrent executions
  },
  async ({ event, step }) => {
    // Your durable workflow
  }
);

Step IDs and Memoization

// Step IDs can be reused - Inngest handles counters automatically
const data = await step.run("fetch-data", () => fetchUserData());
const more = await step.run("fetch-data", () => fetchOrderData()); // Different execution

// Use descriptive IDs for clarity
await step.run("validate-payment", () => validatePayment(event.data.paymentId));
await step.run("charge-customer", () => chargeCustomer(event.data));
await step.run("send-confirmation", () => sendEmail(event.data.email));

Triggers and Events

Event Triggers

Triggers are defined in the triggers array in the first argument of createFunction:

// Single event trigger
inngest.createFunction(
  { id: "my-fn", triggers: [{ event: "user/signup" }] },
  async ({ event }) => { /* ... */ }
);

// Event with conditional filter
inngest.createFunction(
  { id: "my-fn", triggers: [{ event: "user/action", if: 'event.data.action == "purchase" && event.data.amount > 100' }] },
  async ({ event }) => { /* ... */ }
);

// Multiple triggers (up to 10)
inngest.createFunction(
  {
    id: "my-fn",
    triggers: [
      { event: "user/signup" },
      { event: "user/login", if: 'event.data.firstLogin == true' },
      { cron: "0 9 * * *" } // Daily at 9 AM
    ]
  },
  async ({ event }) => { /* ... */ }
);

Cron Triggers

// Basic cron
inngest.createFunction(
  { id: "my-fn", triggers: [{ cron: "0 */6 * * *" }] }, // Every 6 hours
  async ({ step }) => { /* ... */ }
);

// With timezone
inngest.createFunction(
  { id: "my-fn", triggers: [{ cron: "TZ=Europe/Paris 0 12 * * 5" }] }, // Fridays at noon Paris time
  async ({ step }) => { /* ... */ }
);

// Combine with events
inngest.createFunction(
  {
    id: "my-fn",
    triggers: [
      { event: "manual/report.requested" },
      { cron: "0 0 * * 0" } // Weekly on Sunday
    ]
  },
  async ({ event, step }) => { /* ... */ }
);

Function Invocation

// Invoke another function as a step
const result = await step.invoke("generate-report", {
  function: generateReportFunction,
  data: { userId: event.data.userId }
});

// Use returned data
await step.run("process-report", () => {
  return processReport(result);
});

Idempotency Strategies

Event-Level Idempotency (Producer Side)

// Prevent duplicate events with custom ID
await inngest.send({
  id: `checkout-completed-${cartId}`, // 24-hour deduplication
  name: "cart/checkout.completed",
  data: { cartId, email: "user@example.com" }
});

Function-Level Idempotency (Consumer Side)

const sendEmail = inngest.createFunction(
  {
    id: "send-checkout-email",
    triggers: [{ event: "cart/checkout.completed" }],
    // Only run once per cartId per 24 hours
    idempotency: "event.data.cartId"
  },
  async ({ event, step }) => {
    // This function won't run twice for same cartId
  }
);

// Complex idempotency keys
const processUserAction = inngest.createFunction(
  {
    id: "process-user-action",
    triggers: [{ event: "user/action.performed" }],
    // Unique per user + organization combination
    idempotency: 'event.data.userId + "-" + event.data.organizationId'
  },
  async ({ event, step }) => {
    /* ... */
  }
);

Cancellation Patterns

Event-Based Cancellation

In expressions, event = the original triggering event, async = the new event being matched. See Expression Syntax Reference for full details.

const processOrder = inngest.createFunction(
  {
    id: "process-order",
    triggers: [{ event: "order/created" }],
    cancelOn: [
      {
        event: "order/cancelled",
        if: "event.data.orderId == async.data.orderId"
      }
    ]
  },
  async ({ event, step }) => {
    await step.sleepUntil("wait-for-payment", event.data.paymentDue);
    // Will be cancelled if order/cancelled event received
    await step.run("charge-payment", () => processPayment(event.data));
  }
);

Timeout Cancellation

const processWithTimeout = inngest.createFunction(
  {
    id: "process-with-timeout",
    triggers: [{ event: "long/process.requested" }],
    timeouts: {
      start: "5m", // Cancel if not started within 5 minutes
      finish: "30m" // Cancel if not finished within 30 minutes
    }
  },
  async ({ event, step }) => {
    /* ... */
  }
);

Handling Cancellation Cleanup

// Listen for cancellation events
const cleanupCancelled = inngest.createFunction(
  { id: "cleanup-cancelled-process", triggers: [{ event: "inngest/function.cancelled" }] },
  async ({ event, step }) => {
    if (event.data.function_id === "process-order") {
      await step.run("cleanup-resources", () => {
        return cleanupOrderResources(event.data.run_id);
      });
    }
  }
);

Error Handling and Retries

Default Retry Behavior

• 5 total attempts (1 initial + 4 retries) per step
• Exponential backoff with jitter
• Independent retry counters per step

Custom Retry Configuration

const reliableFunction = inngest.createFunction(
  {
    id: "reliable-function",
    triggers: [{ event: "critical/task" }],
    retries: 10 // Up to 10 retries per step
  },
  async ({ event, step, attempt }) => {
    // `attempt` is the function-level attempt counter (0-indexed)
    // It tracks retries for the currently executing step, not the overall function
    if (attempt > 5) {
      // Different logic for later attempts of the current step
    }
  }
);

Non-Retriable Errors

Prevent retries for code that won't succeed upon retry.

import { NonRetriableError } from "inngest";

const processUser = inngest.createFunction(
  { id: "process-user", triggers: [{ event: "user/process.requested" }] },
  async ({ event, step }) => {
    const user = await step.run("fetch-user", async () => {
      const user = await db.users.findOne(event.data.userId);

      if (!user) {
        // Don't retry - user doesn't exist
        throw new NonRetriableError("User not found, stopping execution");
      }

      return user;
    });

    // Continue processing...
  }
);

Custom Retry Timing

import { RetryAfterError } from "inngest";

const respectRateLimit = inngest.createFunction(
  { id: "api-call", triggers: [{ event: "api/call.requested" }] },
  async ({ event, step }) => {
    await step.run("call-api", async () => {
      const response = await externalAPI.call(event.data);

      if (response.status === 429) {
        // Retry after specific time from API
        const retryAfter = response.headers["retry-after"];
        throw new RetryAfterError("Rate limited", `${retryAfter}s`);
      }

      return response.data;
    });
  }
);

Logging Best Practices

Proper Logging Setup

import winston from "winston";

// Configure logger
const logger = winston.createLogger({
  level: "info",
  format: winston.format.json(),
  transports: [new winston.transports.Console()]
});

const inngest = new Inngest({
  id: "my-app",
  logger // Pass logger to client
});

// Or use the built-in ConsoleLogger for simple log level control
import { ConsoleLogger, Inngest } from "inngest";

const inngest = new Inngest({
  id: "my-app",
  logger: new ConsoleLogger({ level: "debug" }) // "debug" | "info" | "warn" | "error"
});

⚠️ v4 Breaking Change: The logLevel option has been removed. Use the logger option with ConsoleLogger or a custom logger instead.

Function Logging Patterns

const processData = inngest.createFunction(
  { id: "process-data", triggers: [{ event: "data/process.requested" }] },
  async ({ event, step, logger }) => {
    // ✅ GOOD: Log inside steps to avoid duplicates
    const result = await step.run("fetch-data", async () => {
      logger.info("Fetching data for user", { userId: event.data.userId });
      return await fetchUserData(event.data.userId);
    });

    // ❌ AVOID: Logging outside steps can duplicate
    // logger.info("Processing complete"); // This could run multiple times!

    await step.run("log-completion", async () => {
      logger.info("Processing complete", { resultCount: result.length });
    });
  }
);

Performance Optimization

Checkpointing

Checkpointing is enabled by default in v4. It allows functions to persist state periodically during execution, reducing latency between steps.

// Checkpointing is enabled by default in v4
// Configure maxRuntime for serverless platforms (set to 60-80% of platform timeout)
const realTimeFunction = inngest.createFunction(
  {
    id: "real-time-function",
    triggers: [{ event: "realtime/process" }],
    checkpointing: {
      maxRuntime: "50s", // For serverless with 60s timeout
    }
  },
  async ({ event, step }) => {
    // Steps execute immediately with periodic checkpointing
    const result1 = await step.run("step-1", () => process1(event.data));
    const result2 = await step.run("step-2", () => process2(result1));
    return { result2 };
  }
);

// Disable checkpointing if needed
const legacyFunction = inngest.createFunction(
  {
    id: "legacy-function",
    triggers: [{ event: "legacy/process" }],
    checkpointing: false
  },
  async ({ event, step }) => { /* ... */ }
);

Advanced Patterns

Conditional Step Execution

const conditionalProcess = inngest.createFunction(
  { id: "conditional-process", triggers: [{ event: "process/conditional" }] },
  async ({ event, step }) => {
    const userData = await step.run("fetch-user", () => {
      return getUserData(event.data.userId);
    });

    // Conditional step execution
    if (userData.isPremium) {
      await step.run("premium-processing", () => {
        return processPremiumFeatures(userData);
      });
    }

    // Always runs
    await step.run("standard-processing", () => {
      return processStandardFeatures(userData);
    });
  }
);

Error Recovery Patterns

const robustProcess = inngest.createFunction(
  { id: "robust-process", triggers: [{ event: "process/robust" }] },
  async ({ event, step }) => {
    let primaryResult;

    try {
      primaryResult = await step.run("primary-service", () => {
        return callPrimaryService(event.data);
      });
    } catch (error) {
      // Fallback to secondary service
      primaryResult = await step.run("fallback-service", () => {
        return callSecondaryService(event.data);
      });
    }

    return { result: primaryResult };
  }
);

Common Mistakes to Avoid

1. ❌ Non-deterministic code outside steps 2. ❌ Database calls outside steps 3. ❌ Logging outside steps (causes duplicates) 4. ❌ Changing step IDs after deployment 5. ❌ Not handling NonRetriableError cases 6. ❌ Ignoring idempotency for critical functions

Next Steps

• See inngest-steps for detailed step method reference
• See references/step-execution.md for detailed step patterns
• See references/error-handling.md for comprehensive error strategies
• See references/observability.md for monitoring and tracing setup
• See references/checkpointing.md for performance optimization details

---

_This skill covers Inngest's durable function patterns. For event sending and webhook handling, see the inngest-events skill._

Checkpointing for Performance Optimization

Guide to using Inngest's checkpointing feature for dramatically lower latency in real-time workflows.

What is Checkpointing?

Checkpointing executes steps immediately on your server rather than waiting for orchestration from Inngest. Steps run eagerly with periodic state checkpoints sent to Inngest for safety.

Performance Comparison

• Without checkpointing: ~50-100ms per step (HTTP round-trip to Inngest)
• With checkpointing: Near-zero latency between steps, periodic checkpoints

// Traditional execution: Each step = separate HTTP request
const traditional = inngest.createFunction(
  { id: "traditional-execution", triggers: [{ event: "process/traditional" }] },
  async ({ event, step }) => {
    // Step 1: HTTP request to Inngest → response → continue
    const data = await step.run("fetch-data", () => fetchData());

    // Step 2: HTTP request to Inngest → response → continue
    const processed = await step.run("process", () => process(data));

    // Step 3: HTTP request to Inngest → response → complete
    return await step.run("save", () => save(processed));
  }
);

// Checkpointed execution: Steps run immediately, periodic checkpoints
const checkpointed = inngest.createFunction(
  {
    id: "checkpointed-execution",
    triggers: [{ event: "process/checkpointed" }]
  },
  async ({ event, step }) => {
    // All steps run immediately, checkpoints sent periodically
    const data = await step.run("fetch-data", () => fetchData());
    const processed = await step.run("process", () => process(data));
    return await step.run("save", () => save(processed));
  }
);

Basic Checkpointing Setup

Checkpointing is enabled by default in v4. All functions automatically use checkpointing for optimal performance. You can configure maxRuntime for serverless environments or disable it per-function if needed.

TypeScript Configuration

import { Inngest } from "inngest";

// Checkpointing is enabled by default - no configuration needed
export const inngest = new Inngest({
  id: "my-app"
});

// Functions automatically use checkpointing
const realTimeFunction = inngest.createFunction(
  {
    id: "real-time-function",
    triggers: [{ event: "realtime/process" }]
  },
  async ({ event, step }) => {
    // Steps execute immediately with periodic checkpointing
    const result1 = await step.run("immediate-step-1", () =>
      process1(event.data)
    );
    const result2 = await step.run("immediate-step-2", () => process2(result1));
    const result3 = await step.run("immediate-step-3", () => process3(result2));

    return { result: result3 };
  }
);

// Configure maxRuntime for serverless environments
const serverlessFunction = inngest.createFunction(
  {
    id: "serverless-function",
    triggers: [{ event: "realtime/serverless" }],
    checkpointing: {
      maxRuntime: "4m45s" // Leave buffer before platform timeout
    }
  },
  async ({ event, step }) => {
    // Steps execute immediately with checkpointing
    const result = await step.run("process", () => process(event.data));
    return { result };
  }
);

// Disable checkpointing for a specific function if needed
const noCheckpointFunction = inngest.createFunction(
  {
    id: "no-checkpoint-function",
    triggers: [{ event: "legacy/process" }],
    checkpointing: false
  },
  async ({ event, step }) => {
    // Uses traditional step-by-step orchestration
    const result = await step.run("process", () => process(event.data));
    return { result };
  }
);

Go Configuration

import (
  "github.com/inngest/inngestgo"
  "github.com/inngest/inngestgo/pkg/checkpoint"
)

_, err := inngestgo.CreateFunction(
  client,
  inngestgo.FunctionOpts{
    ID:         "checkpointed-function",
    Name:       "Checkpointed Function",
    Checkpoint: checkpoint.ConfigSafe, // Enable checkpointing
  },
  inngestgo.EventTrigger("process/checkpointed", nil),
  func(ctx context.Context, input inngestgo.Input[ProcessEvent]) (any, error) {
    // Function implementation
    return processWithCheckpoints(input.Event.Data)
  },
)

Advanced Configuration

Detailed Checkpointing Options

const advancedCheckpointing = inngest.createFunction(
  {
    id: "advanced-checkpointing",
    triggers: [{ event: "process/advanced" }],
    checkpointing: {
      // Maximum time to execute continuously before returning response
      maxRuntime: "300s", // Default: unlimited (0)

      // Number of steps to buffer before checkpointing
      bufferedSteps: 3, // Default: 1 (no buffering)

      // Maximum time to wait before checkpointing buffered steps
      maxInterval: "10s" // Default: immediate
    }
  },
  async ({ event, step }) => {
    // With bufferedSteps: 3, first 3 steps execute without checkpointing
    const step1 = await step.run("step-1", () => process1(event.data));
    const step2 = await step.run("step-2", () => process2(step1));
    const step3 = await step.run("step-3", () => process3(step2));
    // Checkpoint sent after step 3

    const step4 = await step.run("step-4", () => process4(step3));
    const step5 = await step.run("step-5", () => process5(step4));
    const step6 = await step.run("step-6", () => process6(step5));
    // Checkpoint sent after step 6

    return { result: step6 };
  }
);

Platform-Specific Runtime Limits

// Vercel Functions (5-minute timeout)
const vercelFunction = inngest.createFunction(
  {
    id: "vercel-optimized",
    triggers: [{ event: "process/vercel" }],
    checkpointing: {
      maxRuntime: "4m45s" // Leave 15s buffer for cleanup
    }
  },
  async ({ event, step }) => {
    /* ... */
  }
);

// AWS Lambda (15-minute timeout)
const lambdaFunction = inngest.createFunction(
  {
    id: "lambda-optimized",
    triggers: [{ event: "process/lambda" }],
    checkpointing: {
      maxRuntime: "14m30s" // Leave 30s buffer
    }
  },
  async ({ event, step }) => {
    /* ... */
  }
);

// Long-running server (unlimited)
const serverFunction = inngest.createFunction(
  {
    id: "server-optimized",
    triggers: [{ event: "process/server" }],
    checkpointing: {
      maxRuntime: "0", // Unlimited
      bufferedSteps: 5, // More aggressive buffering
      maxInterval: "30s"
    }
  },
  async ({ event, step }) => {
    /* ... */
  }
);

Checkpointing Patterns for Different Use Cases

Real-Time AI/ML Workflows

const aiWorkflow = inngest.createFunction(
  {
    id: "ai-workflow",
    triggers: [{ event: "ai/process.requested" }],
    checkpointing: {
      maxRuntime: "10m",
      bufferedSteps: 2,
      maxInterval: "5s"
    }
  },
  async ({ event, step }) => {
    // Immediate execution for real-time feel
    const preprocessed = await step.run("preprocess-data", () => {
      return preprocessInputData(event.data.input);
    });

    const modelResult = await step.run("run-ml-model", () => {
      return mlModel.predict(preprocessed);
    });

    const postprocessed = await step.run("postprocess-result", () => {
      return postprocessResult(modelResult, event.data.options);
    });

    const saved = await step.run("save-result", () => {
      return saveToDatabase(postprocessed, event.data.userId);
    });

    // Real-time response to user
    await step.run("send-realtime-response", () => {
      return websocketService.send(event.data.userId, {
        result: postprocessed,
        resultId: saved.id
      });
    });

    return { resultId: saved.id };
  }
);

High-Throughput Data Processing

const dataProcessingPipeline = inngest.createFunction(
  {
    id: "data-processing-pipeline",
    triggers: [{ event: "data/batch.received" }],
    checkpointing: {
      maxRuntime: "15m",
      bufferedSteps: 10, // High buffering for throughput
      maxInterval: "60s" // Less frequent checkpoints
    }
  },
  async ({ event, step }) => {
    const batchId = event.data.batchId;

    // Fast processing of many items
    const items = await step.run("fetch-batch-items", () => {
      return fetchBatchItems(batchId);
    });

    // Process items in parallel chunks
    const processedChunks = [];
    const chunkSize = 100;

    for (let i = 0; i < items.length; i += chunkSize) {
      const chunk = items.slice(i, i + chunkSize);

      const processedChunk = await step.run(`process-chunk-${i}`, () => {
        return Promise.all(chunk.map((item) => processItem(item)));
      });

      processedChunks.push(processedChunk);
    }

    // Flatten and validate results
    const allResults = await step.run("aggregate-results", () => {
      const flattened = processedChunks.flat();
      return validateAndAggregateResults(flattened);
    });

    // Bulk save to database
    await step.run("bulk-save-results", () => {
      return database.bulkInsert("processed_items", allResults);
    });

    return {
      batchId,
      itemsProcessed: allResults.length,
      processingTime: Date.now() - event.ts
    };
  }
);

Interactive User Workflows

const interactiveWorkflow = inngest.createFunction(
  {
    id: "interactive-workflow",
    triggers: [{ event: "user/workflow.started" }],
    checkpointing: {
      maxRuntime: "5m",
      bufferedSteps: 1, // Immediate checkpoints for user feedback
      maxInterval: "2s"
    }
  },
  async ({ event, step }) => {
    const userId = event.data.userId;

    // Step 1: Immediate user feedback
    await step.run("send-progress-update", () => {
      return notificationService.send(userId, {
        message: "Processing started...",
        progress: 10
      });
    });

    // Step 2: Quick validation
    const validationResult = await step.run("validate-request", () => {
      const result = validateUserRequest(event.data);

      notificationService.send(userId, {
        message: result.valid ? "Request validated" : "Validation failed",
        progress: 25
      });

      return result;
    });

    if (!validationResult.valid) {
      return { error: "Validation failed", details: validationResult.errors };
    }

    // Step 3: Main processing with progress updates
    const processedData = await step.run("main-processing", () => {
      notificationService.send(userId, {
        message: "Processing your request...",
        progress: 50
      });

      const result = performMainProcessing(event.data);

      notificationService.send(userId, {
        message: "Processing complete",
        progress: 90
      });

      return result;
    });

    // Step 4: Final completion
    await step.run("complete-workflow", () => {
      notificationService.send(userId, {
        message: "Workflow completed successfully!",
        progress: 100,
        result: processedData
      });

      return logWorkflowCompletion(userId, processedData);
    });

    return { success: true, result: processedData };
  }
);

Checkpointing Best Practices

When to Use Checkpointing

// ✅ IDEAL for checkpointing: Real-time workflows
const idealForCheckpointing = inngest.createFunction(
  {
    id: "ideal-checkpointing",
    triggers: [{ event: "realtime/user.request" }]
  },
  async ({ event, step }) => {
    // Fast operations that benefit from immediate execution
    const validated = await step.run("validate", () => validate(event.data));
    const processed = await step.run("process", () => process(validated));
    const response = await step.run("respond", () => sendResponse(processed));

    return response;
  }
);

// ❌ NOT ideal for checkpointing: Very long-running steps
const notIdealForCheckpointing = inngest.createFunction(
  {
    id: "not-ideal-checkpointing",
    triggers: [{ event: "batch/long.process" }],
    checkpointing: false // Disable for very long-running steps
  },
  async ({ event, step }) => {
    // Very long-running operations
    const largeDataset = await step.run("fetch-large-dataset", () => {
      return fetchMillionsOfRecords(); // Takes 10+ minutes
    });

    const processed = await step.run("heavy-processing", () => {
      return processLargeDataset(largeDataset); // Takes 30+ minutes
    });

    return processed;
  }
);

Configuration Guidelines

// High-frequency, low-latency functions
const highFrequencyConfig = {
  checkpointing: {
    maxRuntime: "1m",
    bufferedSteps: 1, // Immediate checkpoints
    maxInterval: "1s"
  }
};

// Medium complexity workflows
const mediumComplexityConfig = {
  checkpointing: {
    maxRuntime: "5m",
    bufferedSteps: 3, // Balance performance and safety
    maxInterval: "10s"
  }
};

// High-throughput batch processing
const highThroughputConfig = {
  checkpointing: {
    maxRuntime: "10m",
    bufferedSteps: 10, // Maximize performance
    maxInterval: "30s"
  }
};

Current Limitations

Known Limitations

• Parallel step execution: Switches to standard orchestration when function branches into parallel steps
• No checkpointing resume: After parallel execution, checkpointing doesn't resume
• Middleware compatibility: Ensure SDK version >=3.51.0 for proper middleware transforms

Feature Support Matrix

Checkpointing dramatically improves function latency for real-time workflows while maintaining all the durability and reliability benefits of Inngest's execution model.

Error Handling and Retries

Comprehensive guide to handling errors, configuring retries, and building resilient Inngest functions.

Understanding Inngest Error Types

Errors vs Failures

• Error: Causes a step to retry (transient issues)
• Failed Step: Step that exhausted all retry attempts
• Failed Function: Function marked as "Failed" when unhandled step failure occurs

const errorHandlingExample = inngest.createFunction(
  { id: "error-handling-demo", retries: 3, triggers: [{ event: "demo/error-handling" }] },
  async ({ event, step }) => {
    try {
      // This step can error and retry up to 3 times
      const result = await step.run("might-fail", async () => {
        const data = await unreliableAPI.call();
        if (!data) throw new Error("API returned empty data");
        return data;
      });

      // If step fails after 3 retries, catch the failure here
    } catch (error) {
      // Handle failed step - this runs after all retries exhausted
      await step.run("handle-failure", () => {
        return logFailureAndNotify(error.message);
      });
    }
  }
);

Retry Configuration

Function-Level Retry Settings

const customRetries = inngest.createFunction(
  {
    id: "custom-retries",
    retries: 10, // Each step gets up to 10 retries (11 total attempts)
    triggers: [{ event: "critical/task" }]
  },
  async ({ event, step, attempt }) => {
    // attempt is 0-indexed: 0, 1, 2, ..., 10

    const result = await step.run("critical-operation", async () => {
      if (attempt < 5) {
        // Use different strategy for early attempts
        return await primaryService.process(event.data);
      } else {
        // Switch to backup service for later attempts
        return await backupService.process(event.data);
      }
    });
  }
);

Per-Step Independent Retries

const independentRetries = inngest.createFunction(
  { id: "independent-retries", retries: 4, triggers: [{ event: "multi/step.process" }] },
  async ({ event, step }) => {
    // Step 1: Can retry up to 4 times independently
    const userData = await step.run("fetch-user", async () => {
      return await userService.getUser(event.data.userId);
    });

    // Step 2: Also gets its own 4 retries, regardless of Step 1's attempts
    const processedData = await step.run("process-data", async () => {
      return await dataProcessor.process(userData);
    });

    // Step 3: Independent retry counter as well
    await step.run("save-results", async () => {
      return await database.save(processedData);
    });

    // If any step fails all retries, it becomes a "failed step"
    // Other steps are unaffected
  }
);

Non-Retriable Errors

When to Use NonRetriableError

import { NonRetriableError } from "inngest";

const smartErrorHandling = inngest.createFunction(
  { id: "smart-error-handling", triggers: [{ event: "process/user" }] },
  async ({ event, step }) => {
    const user = await step.run("validate-and-fetch-user", async () => {
      // Check if user exists
      const user = await database.users.findById(event.data.userId);

      if (!user) {
        // Don't retry - user doesn't exist
        throw new NonRetriableError("User not found");
      }

      if (user.status === "deleted") {
        // Don't retry - user is deleted
        throw new NonRetriableError("User account deleted");
      }

      if (!user.hasPermission("process")) {
        // Don't retry - insufficient permissions
        throw new NonRetriableError("User lacks required permissions");
      }

      return user;
    });

    // This step only runs if user validation passed
    await step.run("process-user-data", async () => {
      return await processUserData(user);
    });
  }
);

Common NonRetriableError Scenarios

const commonNonRetriableErrors = inngest.createFunction(
  { id: "non-retriable-examples", triggers: [{ event: "example/errors" }] },
  async ({ event, step }) => {
    // Authentication/Authorization errors
    await step.run("check-permissions", async () => {
      const user = await getUser(event.data.userId);
      if (!user.isActive) {
        throw new NonRetriableError("Account deactivated");
      }
    });

    // Validation errors
    await step.run("validate-data", async () => {
      if (!event.data.email || !isValidEmail(event.data.email)) {
        throw new NonRetriableError("Invalid email format");
      }
    });

    // Business logic violations
    await step.run("check-business-rules", async () => {
      const account = await getAccount(event.data.accountId);
      if (account.trialExpired && !account.isPaid) {
        throw new NonRetriableError("Trial expired and no payment method");
      }
    });

    // Resource not found (after initial creation)
    await step.run("process-resource", async () => {
      const resource = await getResource(event.data.resourceId);
      if (!resource) {
        throw new NonRetriableError("Resource was deleted during processing");
      }
      return processResource(resource);
    });
  }
);

Custom Retry Timing

RetryAfterError for Rate Limits

import { RetryAfterError } from "inngest";

const respectRateLimit = inngest.createFunction(
  { id: "rate-limited-api", triggers: [{ event: "api/call.requested" }] },
  async ({ event, step }) => {
    const result = await step.run("call-external-api", async () => {
      try {
        const response = await externalAPI.makeRequest(event.data.payload);
        return response.data;
      } catch (error) {
        if (error.response?.status === 429) {
          // Respect the API's rate limit
          const retryAfter = error.response.headers["retry-after"];
          const retryAfterMs = parseInt(retryAfter) * 1000;

          throw new RetryAfterError(
            "API rate limit exceeded",
            new Date(Date.now() + retryAfterMs)
          );
        }

        if (error.response?.status === 503) {
          // Service unavailable - retry after 30 seconds
          throw new RetryAfterError("Service temporarily unavailable", "30s");
        }

        throw error; // Regular retry for other errors
      }
    });
  }
);

Dynamic Retry Strategies

const dynamicRetryStrategy = inngest.createFunction(
  { id: "dynamic-retry", triggers: [{ event: "process/adaptive" }] },
  async ({ event, step, attempt }) => {
    const result = await step.run("adaptive-processing", async () => {
      try {
        // Try different strategies based on attempt number
        if (attempt < 2) {
          // Fast path for first few attempts
          return await fastService.process(event.data);
        } else if (attempt < 4) {
          // Reliable but slower service
          return await reliableService.process(event.data);
        } else {
          // Last resort - manual processing tracking
          return await db.manualProcessing.insert(event.data);
        }
      } catch (error) {
        // Dynamic retry timing based on error type
        if (error.code === "RATE_LIMITED") {
          const delay = Math.min(Math.pow(2, attempt) * 1000, 60000); // Max 1 minute
          throw new RetryAfterError("Rate limited", `${delay}ms`);
        }

        if (error.code === "SERVER_OVERLOADED") {
          const delay = 5000 + attempt * 2000; // Increasing delay
          throw new RetryAfterError("Server overloaded", `${delay}ms`);
        }

        throw error; // Use default retry timing
      }
    });
  }
);

Error Recovery Patterns

Fallback Services Pattern

const fallbackPattern = inngest.createFunction(
  { id: "fallback-services", triggers: [{ event: "process/with-fallback" }] },
  async ({ event, step }) => {
    let result;
    let service = "primary";

    try {
      // Try primary service
      result = await step.run("try-primary-service", async () => {
        return await primaryService.process(event.data);
      });
    } catch (primaryError) {
      // Fallback to secondary service
      service = "secondary";
      result = await step.run("try-secondary-service", async () => {
        return await secondaryService.process(event.data);
      });
    }

    // Log which service was used
    await step.run("log-service-usage", async () => {
      return await analytics.track("service-usage", {
        eventId: event.id,
        serviceUsed: service,
        success: true
      });
    });

    return { result, serviceUsed: service };
  }
);

Failure Handlers

Function-Level Failure Handling

const processWithFailureHandler = inngest.createFunction(
  {
    id: "process-with-failure-handler",
    retries: 3,
    triggers: [{ event: "risky/process" }],
    onFailure: async ({ event, error }) => {
      // This runs when function fails after all retries
      // Access run_id from the failure event data
      const runId = event.data.run_id;

      console.error("Function failed:", {
        eventName: event.name,
        runId,
        error: error.message,
        eventData: event.data
      });

      // Send alert
      await notificationService.sendAlert({
        type: "function-failure",
        functionId: "process-with-failure-handler",
        runId,
        error: error.message,
        eventData: event.data
      });
    }
  },
  async ({ event, step }) => {
    // This might fail after retries
    const result = await step.run("risky-operation", async () => {
      return await riskyExternalService.process(event.data);
    });

    return result;
  }
);

Error Monitoring and Alerting

Structured Error Logging

const structuredErrorLogging = inngest.createFunction(
  { id: "structured-error-logging", triggers: [{ event: "process/with-logging" }] },
  async ({ event, step, logger }) => {
    const baseContext = {
      eventName: event.name,
      eventId: event.id,
      userId: event.data.userId
    };

    try {
      const result = await step.run("api-call-with-logging", async () => {
        try {
          logger.info("Starting API call", {
            ...baseContext,
            step: "api-call-with-logging",
            endpoint: event.data.endpoint
          });

          const response = await externalAPI.call(event.data);

          logger.info("API call succeeded", {
            ...baseContext,
            step: "api-call-with-logging",
            responseStatus: response.status,
            responseSize: JSON.stringify(response.data).length
          });

          return response.data;
        } catch (error) {
          const errorContext = {
            ...baseContext,
            step: "api-call-with-logging",
            error: {
              message: error.message,
              code: error.code,
              status: error.response?.status,
              headers: error.response?.headers
            },
            retryAttempt: step.attempt || 0
          };

          logger.error("API call failed", errorContext);

          // Add to error tracking service
          await errorTracker.captureException(error, errorContext);

          throw error;
        }
      });
    } catch (stepFailure) {
      // Final error handling after all retries
      logger.error("Step failed after all retries", {
        ...baseContext,
        finalError: stepFailure.message,
        totalAttempts: (stepFailure.attempt || 0) + 1
      });

      throw stepFailure;
    }
  }
);

Best Practices Summary

Error Handling Checklist

• ✅ Use NonRetriableError for permanent failures (auth, validation, not found)
• ✅ Configure appropriate retry counts based on failure impact
• ✅ Implement fallback strategies for critical operations
• ✅ Add structured logging with sufficient context
• ✅ Use failure handlers for alerting and cleanup
• ✅ Monitor retry rates to identify systemic issues
• ✅ Respect external service rate limits with RetryAfterError

Common Anti-Patterns

• ❌ Retrying permanent errors (NonRetriableError exists for a reason)
• ❌ Not logging error context (makes debugging impossible)
• ❌ Ignoring failure handlers (missed opportunity for cleanup)
• ❌ No fallback strategies (single points of failure)
• ❌ Not monitoring error rates (missing early warning signs)
• ❌ Overly aggressive retries (can overwhelm downstream services)

Error Handling Strategy Framework

1. Identify error types: Permanent vs transient 2. Configure retries: Based on error impact and recovery time 3. Implement fallbacks: For critical business operations 4. Add monitoring: Structured logging and alerting 5. Test failure scenarios: Ensure error paths work as expected 6. Document error behaviors: For team knowledge sharing

Observability and Extended Traces

Comprehensive guide to monitoring, tracing, and observing Inngest functions with OpenTelemetry integration.

Extended Traces Setup

Basic Extended Traces Configuration

// IMPORTANT: Import and run extendedTracesMiddleware() FIRST
import { extendedTracesMiddleware } from "inngest/experimental";
const extendedTraces = extendedTracesMiddleware();

// Then import everything else
import { Inngest } from "inngest";

const inngest = new Inngest({
  id: "my-app",
  middleware: [extendedTraces]
});

Advanced Extended Traces Configuration

import { extendedTracesMiddleware } from "inngest/experimental";
import { PrismaInstrumentation } from "@prisma/instrumentation";

const extendedTraces = extendedTracesMiddleware({
  // Provider behavior options
  behaviour: "auto", // "auto" | "extendProvider" | "createProvider" | "off"

  // Custom instrumentations
  instrumentations: [
    new PrismaInstrumentation()
    // Add other custom instrumentations
  ]
});

export const inngest = new Inngest({
  id: "my-app",
  middleware: [extendedTraces]
});

Integration with Existing Providers (Sentry Example)

import * as Sentry from "@sentry/node";
import { extendedTracesMiddleware } from "inngest/experimental";

// Initialize Sentry first
Sentry.init({
  dsn: process.env.SENTRY_DSN,
  tracesSampleRate: 1.0
});

// Extended traces will extend Sentry's provider
const extendedTraces = extendedTracesMiddleware({
  behaviour: "auto" // Will extend Sentry's existing provider
});

export const inngest = new Inngest({
  id: "my-app",
  middleware: [extendedTraces]
});

Manual Provider Integration

import { Inngest } from "inngest";
import {
  extendedTracesMiddleware,
  InngestSpanProcessor
} from "inngest/experimental";
import { BasicTracerProvider } from "@opentelemetry/sdk-trace-base";
import { NodeSDK } from "@opentelemetry/auto-instrumentations-node";

// Create client with disabled auto-instrumentation
export const inngest = new Inngest({
  id: "my-app",
  middleware: [
    extendedTracesMiddleware({
      behaviour: "off" // Don't auto-instrument
    })
  ]
});

// Manually create and configure provider
const provider = new BasicTracerProvider({
  spanProcessors: [
    new InngestSpanProcessor(inngest) // Add Inngest span processor
  ]
});

// Register the provider
provider.register();

// Initialize Node SDK with custom provider
const sdk = new NodeSDK({
  traceExporter: yourTraceExporter
});

sdk.start();

Automatic Instrumentation Coverage

Extended traces automatically instruments these libraries when creating a new provider:

Network and HTTP

• http and https (Node.js built-in)
• undici (Node.js global fetch API)
• @grpc/grpc-js

Databases

• mongodb
• mongoose
• pg (PostgreSQL)
• mysql and mysql2
• redis and ioredis
• cassandra-driver
• knex

Web Frameworks

• express
• koa
• @hapi/hapi
• restify
• connect
• @nestjs/core

Message Queues

• amqplib
• kafkajs

Cloud Services

• @aws-sdk/client-* (AWS SDK v3)

Logging

• winston
• pino
• bunyan

Other

• dns and net (Node.js built-in)
• fs (Node.js built-in)
• dataloader
• generic-pool
• memcached
• socket.io

Logging Best Practices

Tip: Use a logger that supports a child logger for automatic function metadata insertion. Supported libraries include Winston, Pino, Bunyan, and Roarr.

Logger Configuration with Extended Traces

import winston from "winston";
import { extendedTracesMiddleware } from "inngest/experimental";

// Configure Winston with JSON format for structured logging
const logger = winston.createLogger({
  level: "info",
  exitOnError: false,
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.errors({ stack: true }),
    winston.format.json()
  ),
  transports: [
    new winston.transports.Console(),
    new winston.transports.File({ filename: "app.log" })
  ]
});

// Enable extended traces
const extendedTraces = extendedTracesMiddleware();

export const inngest = new Inngest({
  id: "my-app",
  logger, // Pass logger to client
  middleware: [extendedTraces]
});

Function Logging Patterns

const observableFunction = inngest.createFunction(
  { id: "observable-function", triggers: [{ event: "process/observable" }] },
  async ({ event, step, logger, runId }) => {
    // Logger automatically includes function metadata when using .child()
    logger.info("Function started", {
      eventData: event.data,
      runId
    });

    const userData = await step.run("fetch-user-data", async () => {
      logger.info("Fetching user data", {
        userId: event.data.userId
      });

      const data = await userService.getUser(event.data.userId);

      logger.info("User data fetched", {
        userId: data.id,
        userType: data.type,
        dataSize: JSON.stringify(data).length
      });

      return data;
    });

    const result = await step.run("process-data", async () => {
      // Using Date.now() within a step is OK!
      const startTime = Date.now();

      try {
        const processed = await dataProcessor.process(userData);

        logger.info("Data processing completed", {
          processingTime: Date.now() - startTime,
          resultSize: processed.length
        });

        return processed;
      } catch (error) {
        logger.error("Data processing failed", {
          processingTime: Date.now() - startTime,
          error: error.message,
          userId: userData.id
        });
        throw error;
      }
    });

    logger.info("Function completed successfully", {
      totalExecutionTime: Date.now() - event.ts,
      stepsExecuted: 2,
      resultCount: result.length
    });

    return result;
  }
);

Structured Logging with Correlation IDs

const correlatedLogging = inngest.createFunction(
  { id: "correlated-logging", triggers: [{ event: "process/correlated" }] },
  async ({ event, step, logger, runId }) => {
    // Create correlation context
    const correlationId = event.data.correlationId || runId;
    const baseContext = {
      correlationId,
      userId: event.data.userId,
      requestId: event.data.requestId,
      runId
    };

    logger.info("Starting correlated process", baseContext);

    const step1Result = await step.run("external-api-call", async () => {
      logger.info("Calling external API", {
        ...baseContext,
        step: "external-api-call",
        endpoint: "/api/user-data"
      });

      try {
        const response = await externalAPI.getUserData(event.data.userId, {
          headers: { "X-Correlation-ID": correlationId }
        });

        logger.info("External API call successful", {
          ...baseContext,
          step: "external-api-call",
          responseStatus: response.status,
          responseTime: response.responseTime
        });

        return response.data;
      } catch (error) {
        logger.error("External API call failed", {
          ...baseContext,
          step: "external-api-call",
          error: error.message,
          statusCode: error.status
        });
        throw error;
      }
    });

    await step.run("database-operation", async () => {
      logger.info("Performing database operation", {
        ...baseContext,
        step: "database-operation",
        operation: "upsert"
      });

      const result = await database.users.upsert({
        id: event.data.userId,
        data: step1Result,
        correlationId // Include in database record
      });

      logger.info("Database operation completed", {
        ...baseContext,
        step: "database-operation",
        recordId: result.id,
        operation: "upsert"
      });

      return result;
    });
  }
);

Performance Monitoring

Custom Metrics and Traces

import { trace, context, SpanStatusCode } from "@opentelemetry/api";

const customTracing = inngest.createFunction(
  { id: "custom-tracing", triggers: [{ event: "process/traced" }] },
  async ({ event, step }) => {
    const tracer = trace.getTracer("my-app");

    const result = await step.run("traced-operation", async () => {
      // Create custom span
      return tracer.startActiveSpan(
        "business-logic-operation",
        async (span) => {
          try {
            // Add custom attributes
            span.setAttributes({
              "user.id": event.data.userId,
              "operation.type": "data-processing",
              "input.size": JSON.stringify(event.data).length
            });

            // Simulate some work with nested spans
            const processedData = await tracer.startActiveSpan(
              "data-transformation",
              async (childSpan) => {
                childSpan.setAttributes({
                  "transformation.type": "normalize"
                });

                const result = await transformData(event.data);

                childSpan.setAttributes({
                  "transformation.output_records": result.length
                });

                childSpan.setStatus({ code: SpanStatusCode.OK });
                childSpan.end();

                return result;
              }
            );

            // Another nested operation
            const savedData = await tracer.startActiveSpan(
              "data-persistence",
              async (childSpan) => {
                childSpan.setAttributes({
                  "db.operation": "bulk_insert",
                  "db.table": "processed_data"
                });

                const saveResult = await database.bulkInsert(processedData);

                childSpan.setAttributes({
                  "db.records_inserted": saveResult.insertedCount
                });

                childSpan.setStatus({ code: SpanStatusCode.OK });
                childSpan.end();

                return saveResult;
              }
            );

            // Add result attributes to main span
            span.setAttributes({
              "result.records_processed": processedData.length,
              "result.records_saved": savedData.insertedCount
            });

            span.setStatus({ code: SpanStatusCode.OK });

            return savedData;
          } catch (error) {
            span.recordException(error);
            span.setStatus({
              code: SpanStatusCode.ERROR,
              message: error.message
            });
            throw error;
          } finally {
            span.end();
          }
        }
      );
    });

    return result;
  }
);

External Service Integration

Datadog Integration

import winston from "winston";

const datadogLogger = winston.createLogger({
  level: "info",
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.json()
  ),
  transports: [
    new winston.transports.Console(),
    new winston.transports.Http({
      host: "http-intake.logs.datadoghq.com",
      path: `/api/v2/logs?dd-api-key=${process.env.DD_API_KEY}&ddsource=inngest&service=my-app&ddtags=env:${process.env.NODE_ENV}`,
      ssl: true
    })
  ]
});

export const inngest = new Inngest({
  id: "my-app",
  logger: datadogLogger,
  middleware: [extendedTracesMiddleware()]
});

Observability Best Practices

Key Metrics to Track

• Function execution time: Total duration from trigger to completion
• Step execution time: Individual step performance
• Retry rates: Which steps/functions fail most often
• Queue depth: How many functions are waiting to execute
• Error rates: Function and step failure percentages
• Throughput: Functions processed per minute/hour

Health Check Functions

const healthCheck = inngest.createFunction(
  { id: "health-check", triggers: [{ cron: "*/5 * * * *" }] }, // Every 5 minutes
  async ({ step, logger }) => {
    const healthStatus = {
      services: {}
    };

    // Check database connectivity
    healthStatus.services.database = await step.run(
      "check-database",
      async () => {
        try {
          const result = await database.query("SELECT 1");
          return { status: "healthy" };
        } catch (error) {
          return { status: "unhealthy", error: error.message };
        }
      }
    );

    // Check external API connectivity
    healthStatus.services.externalAPI = await step.run(
      "check-external-api",
      async () => {
        try {
          const startTime = Date.now();
          const response = await externalAPI.healthCheck();
          return {
            status: "healthy",
            apiStatus: response.status
          };
        } catch (error) {
          return { status: "unhealthy", error: error.message };
        }
      }
    );

    // Send health status to monitoring
    await step.run("report-health-status", async () => {
      const overallHealth = Object.values(healthStatus.services).every(
        (service) => service.status === "healthy"
      )
        ? "healthy"
        : "degraded";

      await monitoringService.reportHealth({
        ...healthStatus,
        overallStatus: overallHealth
      });

      if (overallHealth === "degraded") {
        await alertingService.send({
          level: "warning",
          message: "System health check detected issues",
          healthStatus
        });
      }
    });

    return healthStatus;
  }
);

Debugging and Troubleshooting

Debug Logging Configuration

const debugFunction = inngest.createFunction(
  { id: "debug-function", triggers: [{ event: "debug/test" }] },
  async ({ event, step, logger }) => {
    // Enable debug logging conditionally
    const isDebugMode =
      event.data.debug || process.env.NODE_ENV === "development";

    if (isDebugMode) {
      logger.debug("Debug mode enabled", {
        eventData: event.data,
        environment: process.env.NODE_ENV
      });
    }

    const result = await step.run("debug-operation", async () => {
      if (isDebugMode) {
        logger.debug("Starting debug operation", {
          input: event.data,
          timestamp: Date.now()
        });
      }

      try {
        const result = await someOperation(event.data);

        if (isDebugMode) {
          logger.debug("Operation completed", {
            result: result,
            executionTime: Date.now() - startTime
          });
        }

        return result;
      } catch (error) {
        logger.error("Operation failed", {
          error: error.message,
          stack: isDebugMode ? error.stack : undefined,
          input: event.data
        });
        throw error;
      }
    });

    return result;
  }
);

This comprehensive observability setup ensures you have full visibility into your Inngest functions' performance, errors, and behavior across all environments.

Step Execution and Memoization

Deep dive into how Inngest executes steps, handles memoization, and manages state persistence.

How Step Execution Works

Execution Flow

1. Initial execution: Function called with event payload 2. Step discovery: First step encountered, code executes 3. State persistence: Result sent to Inngest, stored in state 4. Function interruption: Execution stops after first step 5. Subsequent execution: Function re-called with event + previous state 6. Memoization: Previous step result injected, execution continues

Each Step = HTTP Request

Tip: See references/checkpointing.md to handle multiple steps on a single HTTP request, optimizing performance for low latency.

const importContacts = inngest.createFunction(
  { id: "import-contacts", triggers: [{ event: "contacts/csv.uploaded" }] },
  async ({ event, step }) => {
    // HTTP Request #1 - Executes and returns
    const rows = await step.run("parse-csv", async () => {
      return await parseCsv(event.data.fileURI);
    });

    // HTTP Request #2 - Gets rows from state, executes this step
    const normalizedRows = await step.run("normalize-csv", async () => {
      return normalizeRows(rows, getColumnMapping());
    });

    // HTTP Request #3 - Gets previous results, executes final step
    const results = await step.run("import-contacts", async () => {
      return await importContacts(normalizedRows);
    });

    return { results };
  }
);

Step ID Management

Step ID Hashing

• Inngest hashes step IDs as state identifiers
• Index position also included in result
• Same ID can be reused - Inngest handles counters automatically

async ({ event, step }) => {
  // These are DIFFERENT executions even with same ID
  const userData = await step.run("fetch-data", () =>
    fetchUser(event.data.userId)
  );
  const orderData = await step.run("fetch-data", () =>
    fetchOrders(event.data.userId)
  );

  // Inngest internally tracks: fetch-data[0] and fetch-data[1]
};

Best Practices for Step IDs

// ✅ GOOD: Descriptive and unique
await step.run("validate-payment-method", () => validatePayment());
await step.run("charge-customer-card", () => chargeCard());
await step.run("send-confirmation-email", () => sendEmail());

// ❌ AVOID: Too generic
await step.run("step1", () => validatePayment());
await step.run("step2", () => chargeCard());
await step.run("step3", () => sendEmail());

// ❌ DANGEROUS: Changing IDs breaks memoization
// Before deploy:
await step.run("process-data", () => processUserData());
// After deploy:
await step.run("process-user-data", () => processUserData()); // Will re-execute!

Note: Changing IDs will force re-execution, which can be used to evolve the functionality of a function.

State Persistence and Recovery

State Structure

{
  "parse-csv": {
    "data": [...], // Step result
    "index": 0
  },
  "normalize-csv": {
    "data": {...},
    "index": 1
  }
}

Recovery from Failures

const robustProcess = inngest.createFunction(
  { id: "robust-process", triggers: [{ event: "process/data" }] },
  async ({ event, step }) => {
    // Step 1: Completes successfully
    const data = await step.run("fetch-external-data", async () => {
      return await externalAPI.getData(event.data.id);
    });

    // Step 2: Fails with network timeout
    const processed = await step.run("process-data", async () => {
      // This throws an error on first attempt
      return await heavyProcessing(data);
    });

    // On retry:
    // - Step 1 is skipped (memoized result used)
    // - Step 2 is re-executed with same input data
    // - If successful, continues to step 3

    await step.run("save-results", async () => {
      return await database.save(processed);
    });
  }
);

Advanced Step Patterns

Parallel Step Execution

const parallelProcess = inngest.createFunction(
  { id: "parallel-process", triggers: [{ event: "process/parallel" }] },
  async ({ event, step }) => {
    const userData = await step.run("fetch-user", () => {
      return getUserData(event.data.userId);
    });

    // These run in parallel (separate HTTP requests)
    const [profile, orders, preferences] = await Promise.all([
      step.run("fetch-profile", () => getUserProfile(userData.id)),
      step.run("fetch-orders", () => getUserOrders(userData.id)),
      step.run("fetch-preferences", () => getUserPreferences(userData.id))
    ]);

    return { profile, orders, preferences };
  }
);

Conditional Step Execution

const conditionalSteps = inngest.createFunction(
  { id: "conditional-steps", triggers: [{ event: "user/signup" }] },
  async ({ event, step }) => {
    const user = await step.run("create-user", () => {
      return createUser(event.data);
    });

    // Conditional steps - only executed when conditions are met
    if (user.accountType === "premium") {
      await step.run("setup-premium-features", () => {
        return setupPremiumFeatures(user.id);
      });
    }

    if (user.company) {
      await step.run("create-company-profile", () => {
        return createCompanyProfile(user.company);
      });
    }

    // Always executed
    await step.run("send-welcome-email", () => {
      return sendWelcomeEmail(user.email);
    });
  }
);

Dynamic Step Generation

const dynamicSteps = inngest.createFunction(
  { id: "dynamic-steps", triggers: [{ event: "batch/process" }] },
  async ({ event, step }) => {
    const items = await step.run("fetch-items", () => {
      return getItemsToProcess(event.data.batchId);
    });

    // Process each item as separate step for individual retry
    const results = [];
    for (let i = 0; i < items.length; i++) {
      const result = await step.run(`process-item-${i}`, () => {
        return processItem(items[i]);
      });
      results.push(result);
    }

    return { processedCount: results.length, results };
  }
);

Step Timing and Performance

Step Overhead

• Each step adds ~50-100ms overhead (See "checkpointing" reference)
• Consider step granularity vs retry isolation

// ❌ TOO GRANULAR: Many small steps
const tooGranular = inngest.createFunction(
  { id: "too-granular", triggers: [{ event: "process/data" }] },
  async ({ event, step }) => {
    const a = await step.run("step-1", () => simpleOperation1());
    const b = await step.run("step-2", () => simpleOperation2(a));
    const c = await step.run("step-3", () => simpleOperation3(b));
    // 3 atomic requests for simple operations
  }
);

// ✅ BETTER: Logical grouping
const betterGrouping = inngest.createFunction(
  { id: "better-grouping", triggers: [{ event: "process/data" }] },
  async ({ event, step }) => {
    const processedData = await step.run("process-data-batch", () => {
      const a = simpleOperation1();
      const b = simpleOperation2(a);
      return simpleOperation3(b);
    });

    // Separate step for different failure domain
    const result = await step.run("save-to-database", () => {
      return database.save(processedData);
    });
  }
);

When to Use Steps vs Regular Code

// Use steps for:
// - API calls (can fail due to network)
// - Database operations (can fail due to locks)
// - File I/O operations
// - Any non-deterministic operations
// - Operations you want to retry independently

// Use regular code for:
// - Pure functions/calculations
// - Data transformations
// - Validation logic
// - Simple conditionals

const goodPatterns = inngest.createFunction(
  { id: "good-patterns", triggers: [{ event: "process/user" }] },
  async ({ event, step }) => {
    // ✅ Regular code: deterministic validation
    if (!event.data.email || !event.data.userId) {
      throw new Error("Missing required fields");
    }

    // ✅ Step: External API call
    const userData = await step.run("fetch-user-data", () => {
      return userAPI.getUser(event.data.userId);
    });

    // ✅ Regular code: data transformation
    const processedData = {
      ...userData,
      email: event.data.email.toLowerCase(),
      fullName: `${userData.firstName} ${userData.lastName}`
    };

    // ✅ Step: Database operation
    const savedUser = await step.run("save-user", () => {
      return database.users.upsert(processedData);
    });

    return { userId: savedUser.id };
  }
);

Troubleshooting Step Issues

Common Step Problems

1. Step never completes

// ❌ PROBLEM: Infinite loop or hanging operation
await step.run("broken-step", async () => {
  while (true) {} // This will timeout the function
});

// ✅ SOLUTION: Add proper exit conditions
await step.run("fixed-step", async () => {
  let attempts = 0;
  while (attempts < 10) {
    const result = await tryOperation();
    if (result.success) return result;
    attempts++;
  }
  throw new Error("Max attempts reached");
});

2. Step re-executes unexpectedly

// ❌ PROBLEM: Changed step ID
// Before:
await step.run("process-data", () => processData());
// After deploy:
await step.run("process-user-data", () => processData()); // Re-executes!

// ✅ SOLUTION: Keep step IDs consistent
await step.run("process-data", () => processUserData()); // Same ID, updated logic

3. Step data inconsistency

// ❌ PROBLEM: Non-deterministic data in step
await step.run("create-record", () => {
  return database.create({
    id: Math.random(), // Different on retry!
    timestamp: Date.now(), // Different on retry!
    data: event.data
  });
});

// ✅ SOLUTION: Use deterministic or external IDs
await step.run("create-record", () => {
  return database.create({
    id: event.data.userId + "-" + event.data.timestamp,
    timestamp: event.data.timestamp,
    data: event.data
  });
});

Step Performance Monitoring

Key Metrics to Track

• Step execution time: Individual step performance
• Step retry rate: Which steps fail most often
• Function duration: Total execution time across all steps
• Memoization hit rate: How often steps are skipped