Scoped steps and Entity Framework Core

A function’s services come from a dependency injection scope that lives for one call from Inngest, and scoped services such as an Entity Framework DbContext work as they do in a controller. A call is not a step, though. Inngest calls the function again for every step, the function’s own code runs again each time, and with checkpointing one call runs several steps. A DbContext injected into the function is a single instance for all of that. Entity Framework describes a DbContext as a unit of work, and a step is the natural boundary for one.

context.Step.RunScopedAsync runs a step in a scope of its own instead. The handler resolves what it needs from that scope, and the scope is disposed when the handler ends, so nothing a step tracks or leaves open reaches the next step, the next attempt or the next replay. It is part of the Inngest package and knows nothing about Entity Framework; this page uses DbContext because it is the service that needs it most.

The examples use this model:

using Microsoft.EntityFrameworkCore;

public sealed class ShopDbContext(DbContextOptions<ShopDbContext> options) : DbContext(options)
{
    public DbSet<Order> Orders => Set<Order>();

    public DbSet<Payment> Payments => Set<Payment>();

    protected override void OnModelCreating(ModelBuilder modelBuilder) =>
        modelBuilder.Entity<Payment>().HasIndex(payment => new { payment.RunId, payment.StepId }).IsUnique();
}

public sealed class Order
{
    public required string Id { get; init; }

    public decimal Total { get; init; }

    public string Status { get; set; } = "placed";

    public string? Note { get; set; }

    public List<OrderLine> Lines { get; } = [];
}

public sealed class OrderLine
{
    public int Id { get; set; }

    public required string Sku { get; init; }

    public int Quantity { get; init; }

    public Order? Order { get; set; }
}

public sealed class Payment
{
    public int Id { get; set; }

    public required string RunId { get; init; }

    public required string StepId { get; init; }

    public required string OrderId { get; init; }

    public decimal Amount { get; init; }
}

public sealed record PaymentRecorded(int PaymentId);

public sealed record LineSummary(string Sku, int Quantity);

public sealed record OrderSummary(string Id, string Status, List<LineSummary> Lines);

What the call’s scope shares

A DbContext tracks every entity it loads or changes, and SaveChangesAsync saves all of them. The context injected into the function belongs to the call, so a step that saves also saves whatever the rest of the call left in it:

using Microsoft.EntityFrameworkCore;

// A read and an unsaved change made outside a step, so they happen again on every call.
var order = await db.Orders.SingleAsync(o => o.Id == context.Event.Data.OrderId, cancellationToken);
order.Note = $"Seen at {DateTimeOffset.UtcNow:O}";

await context.Step.RunAsync("record-payment", async ct =>
{
    db.Payments.Add(new Payment { RunId = context.RunId, StepId = "record-payment", OrderId = order.Id, Amount = order.Total });

    // Saves the payment, and the note too, which nothing asked this step to save.
    await db.SaveChangesAsync(ct);
});

It works the other way round too: when earlier code has already loaded a row, a tracking query returns that instance with the values it holds, not the values it just read, and FindAsync does not query at all. A scoped step starts from an empty context every time:

var order = context.Event.Data;

var payment = await context.Step.RunScopedAsync("record-payment", async (ShopDbContext db, CancellationToken ct) =>
{
    var row = new Payment { RunId = context.RunId, StepId = "record-payment", OrderId = order.OrderId, Amount = order.Total };
    db.Payments.Add(row);
    await db.SaveChangesAsync(ct);
    return new PaymentRecorded(row.Id);
});

RunScopedAsync

The handler can ask for the scope’s IServiceProvider, or for one service, which the SDK resolves with GetRequiredService. A handler without a result returns Task, and every overload also takes a StepOptions in place of the id, like the other step tools. Prefer the form that names the service, with the type of the first parameter written so that the compiler can infer it:

using Microsoft.EntityFrameworkCore;

// One service: the type of the first parameter is written, and the SDK resolves it from the scope.
var open = await context.Step.RunScopedAsync("count-open", (ShopDbContext db, CancellationToken ct) =>
    db.Orders.CountAsync(o => o.Status == "placed", ct));

// Several services: ask for the scope's provider and resolve them from it.
var total = await context.Step.RunScopedAsync("sum-orders", async (services, ct) =>
{
    var db = services.GetRequiredService<ShopDbContext>();
    return await db.Orders.SumAsync(o => o.Total, ct);
});

// No result.
await context.Step.RunScopedAsync("archive", async (ShopDbContext db, CancellationToken ct) =>
{
    await db.Orders.Where(o => o.Status == "shipped").ExecuteDeleteAsync(ct);
});

Type arguments work too, with one catch. Two, as in RunScopedAsync<ShopDbContext, int>, keep the result. One, as in RunScopedAsync<ShopDbContext>, picks the form without a result, so a value the lambda returns is dropped:

using Microsoft.EntityFrameworkCore;

var all = await context.Step.RunScopedAsync<ShopDbContext, int>("count-all", (db, ct) => db.Orders.CountAsync(ct));

// The number of rows deleted is dropped.
await context.Step.RunScopedAsync<ShopDbContext>("purge", (db, ct) =>
    db.Orders.Where(o => o.Status == "cancelled").ExecuteDeleteAsync(ct));

Both forms resolve services with GetRequiredService<T>() and use no reflection of their own, so they work in trimmed and Native AOT apps as far as the services do. To Inngest a scoped step is a run step, reported exactly as RunAsync reports one, with the same id rules, retries and results. What differs is the scope:

• It lives as long as the handler. The SDK creates it when the handler starts and disposes it when the handler ends, whether the handler returned or threw, asynchronously for services that are IAsyncDisposable. That is before the result is saved. Pass the handler’s CancellationToken to every call it makes: a handler that ignores it keeps its scope open until it returns, which can be after the call has ended.
• Every execution gets a new one. Each attempt of a step, each step of a checkpointed call and each step that runs in parallel has a scope of its own. Steps in parallel already run in calls of their own, so what the scope keeps apart is a step from the function’s own code, which uses the call’s scope, and from the other steps that one call runs one after another. A step whose result Inngest already has, a step that is only planned and a step a test mocks create none.
• It is a scope of its own, not the call’s. With Microsoft’s container, IServiceScopeFactory makes a scope next to the call’s, not inside it; a container that nests scopes may differ. Either way, scoped services in it are new instances and singletons are shared, but nothing that code stored in the call’s scope is there, such as a tenant that middleware set on a scoped service. Set it from the step: services.GetRequiredService<ITenantAccessor>().Set(tenantId).
• Middleware keeps seeing the call’s scope. FunctionContext.Services is the same for every hook. WrapStepHandlerAsync wraps the scope: it is created inside next() and gone when next() returns, and the step’s StepInfo.Type is Run. See Middleware.
• Disposal is for releasing, not for work. An exception thrown while the scope is disposed is only logged, at the error level. The step’s outcome stands and nothing retries it, and a handler that threw keeps its own exception. So never commit, flush or save in the Dispose or DisposeAsync of a scoped service: a failure there is lost while the step is recorded as a success. Do that in the handler, with a call that can fail the step.
• Steps cannot be nested. Calling a step tool inside the handler throws, as it does in RunAsync.

Results are plain data

The result is serialized to JSON after the scope has been disposed, and every replay reads it back (Results are JSON); it also counts toward Inngest’s limits on the size of a step’s result. Return records, lists and ids, never tracked entities. An entity with navigation properties usually refers back to itself, which System.Text.Json refuses, and a query that has not run yet, such as an IQueryable or a lazy Select, would need the context that is gone. Project in the query and return what the function needs next:

using Microsoft.EntityFrameworkCore;

var summary = await context.Step.RunScopedAsync("load-order", (ShopDbContext db, CancellationToken ct) =>
    db.Orders
        .Where(o => o.Id == orderId)
        .Select(o => new OrderSummary(o.Id, o.Status, o.Lines.Select(line => new LineSummary(line.Sku, line.Quantity)).ToList()))
        .SingleAsync(ct));

A result that cannot be serialized fails the step after its handler has finished. The work is already committed by then, and the retry does it again, which is a reason to make steps safe to repeat, as described below.

Do not hand an entity from one step to another either. Re-attaching a copy that a step returned, with Update, writes every column it holds, including values that changed in the meantime. Pass the id and load the row again where it is needed.

Entity Framework Core

One SaveChangesAsync per step

A step is the unit that Inngest retries, and it records the step’s result only when the handler returns. Give each step one SaveChangesAsync: for a relational provider it runs in a transaction of its own, so the step writes everything or nothing, and a retry starts from a clean state. Work that is separate, such as taking the payment and booking the shipment, belongs in separate steps, so that a failure retries only what failed. Return what the next step needs, such as the generated key, after the save.

Register the context as usual. With AddDbContextPool a step takes a context from the pool when it resolves one and the scope gives it back, reset, when the step ends:

using Microsoft.EntityFrameworkCore;

builder.Services.AddDbContextPool<ShopDbContext>(options =>
{
    // Your provider, for example options.UseNpgsql(connectionString).
});

If a step needs only a DbContext, IDbContextFactory<T> does the same without the SDK. AddDbContextPool does not register the factory: register it with AddDbContextFactory, or AddPooledDbContextFactory for a pool, inject it into the function and create the context inside a plain RunAsync. A scoped step is for steps that use several scoped services that must share one context, such as repositories or a unit of work class registered as scoped. Such a class commits when the handler tells it to, with a call that can fail the step, never when it is disposed.

using Microsoft.EntityFrameworkCore;

builder.Services.AddDbContextFactory<ShopDbContext>(options =>
{
    // Your provider, for example options.UseNpgsql(connectionString).
});

using Microsoft.EntityFrameworkCore;

var count = await context.Step.RunAsync("count-open", async ct =>
{
    await using var db = await dbFactory.CreateDbContextAsync(ct);
    return await db.Orders.CountAsync(o => o.Status == "placed", ct);
});

Transactions and execution strategies

A step that needs several saves in one transaction opens the transaction itself. If the provider is set up with a retrying execution strategy, such as EnableRetryOnFailure(), Entity Framework throws on the first command you run in a transaction you started yourself, unless the whole transaction runs inside strategy.ExecuteAsync. The strategy runs the delegate again after a transient failure, so the delegate must start from a clean state: clear the change tracker and build the entities inside it. Pass the step’s token to ExecuteAsync as well, so that the strategy stops retrying, and waiting between retries, when the call is canceled. The delegate also looks for the payment that an earlier execution committed before it does anything else, for the reason given under idempotency keys:

using Microsoft.EntityFrameworkCore;

var order = context.Event.Data;

var captured = await context.Step.RunScopedAsync("capture-payment", async (ShopDbContext db, CancellationToken ct) =>
{
    var strategy = db.Database.CreateExecutionStrategy();
    return await strategy.ExecuteAsync(async token =>
    {
        db.ChangeTracker.Clear();
        await using var transaction = await db.Database.BeginTransactionAsync(token);

        // An earlier execution of this step may have committed and then been lost.
        var existing = await db.Payments.SingleOrDefaultAsync(p => p.RunId == context.RunId && p.StepId == "capture-payment", token);
        if (existing is not null)
        {
            return new PaymentRecorded(existing.Id);
        }

        var row = await db.Orders.SingleAsync(o => o.Id == order.OrderId, token);
        row.Status = "paid";
        await db.SaveChangesAsync(token);

        var payment = new Payment { RunId = context.RunId, StepId = "capture-payment", OrderId = row.Id, Amount = row.Total };
        db.Payments.Add(payment);
        await db.SaveChangesAsync(token);

        await transaction.CommitAsync(token);
        return new PaymentRecorded(payment.Id);
    }, ct);
});

Without a retrying strategy the delegate runs once. If the connection drops while the transaction commits, Entity Framework cannot tell whether it committed and may run the delegate again; the lookup at the top makes that harmless. If two executions of the step overlap, the unique index fails the save of the second, and its retry finds the payment. See Microsoft’s pages on connection resiliency and the lifetime of a DbContext.

Idempotency keys

A step’s handler can run again after its work has been committed. Inngest records a step’s result when the call that ran the step answers with it, or when checkpointing saves it. If the call is lost before then, because the process stops, a proxy times out or the connection drops, Inngest retries the step, and the handler finds its rows already written. With checkpointing and a BufferedSteps above 1, every step that was not saved yet runs again. The same happens when the result cannot be serialized. Inngest’s own page on idempotency makes the point about external services. Deduplicating the event or the function, as flow control does, decides whether a run starts, not whether a step runs twice.

So key the write by what stays the same across executions of the step: the run id and the step id. A new Guid or context.Attempt would differ each time, which defeats the key. Put a unique index on the key, as the model above does for Payment, so the database refuses a second row even when two executions overlap, and make the handler return the row that is there:

using Microsoft.EntityFrameworkCore;

var order = context.Event.Data;

var recorded = await context.Step.RunScopedAsync("record-payment", async (ShopDbContext db, CancellationToken ct) =>
{
    var existing = await db.Payments.SingleOrDefaultAsync(p => p.RunId == context.RunId && p.StepId == "record-payment", ct);
    if (existing is not null)
    {
        return new PaymentRecorded(existing.Id);
    }

    var payment = new Payment { RunId = context.RunId, StepId = "record-payment", OrderId = order.OrderId, Amount = order.Total };
    db.Payments.Add(payment);
    try
    {
        await db.SaveChangesAsync(ct);
    }
    catch (DbUpdateException)
    {
        // An execution that overlapped this one committed first, and the unique index refused this row.
        db.ChangeTracker.Clear();
        var winner = await db.Payments.SingleOrDefaultAsync(p => p.RunId == context.RunId && p.StepId == "record-payment", ct);
        if (winner is null)
        {
            throw;
        }

        return new PaymentRecorded(winner.Id);
    }

    return new PaymentRecorded(payment.Id);
});

A step id used in a loop is numbered (record-payment:1), so build the id from the item, as in $"record-payment-{invoice.Id}", and key by that. Other steps need other shapes:

• A step that creates a row keys it by the run and the step, as above, or by a natural key such as the order id, behind a unique index.
• A step that sets a value to a fixed one is safe to repeat as it is. So is an upsert, but EF Core has no upsert API: it takes raw SQL, ExecuteSqlInterpolatedAsync with the provider’s ON CONFLICT or MERGE, or the look-up-then-insert shown above.
• A step that changes a value relative to its old one, such as a stock count, is not safe to repeat. Record the change in a row keyed by the run and the step, in the same save: look for the row first and skip the change when it is there, and let the unique index catch executions that overlap.

Keep the keys as long as the run can still retry the step. A function triggered by inngest/function.finished can delete them when the run ends.

Reads between steps

Code outside steps runs on every call, and what a read finds can change between two calls. A branch taken on such a read can differ from call to call, so the function reaches different steps than the run recorded. Put every read that decides what the function does into a step, so that its result is saved:

using Microsoft.EntityFrameworkCore;

var status = await context.Step.RunScopedAsync("load-status", (ShopDbContext db, CancellationToken ct) =>
    db.Orders.Where(o => o.Id == orderId).Select(o => o.Status).SingleAsync(ct));

if (status == "paid")
{
    await context.Step.RunAsync("ship", () => Task.CompletedTask);
}

Testing

The test host runs scoped steps like any other, with a new scope for each, and a mock replaces one where its handler would run, so new StepMocks().Run("record-payment", new PaymentRecorded(1)) answers the step above without touching a database. To test the database code, run it against a real database: SQLite in memory or, better, the kind you deploy to. The EF Core in-memory provider does not support transactions, and a test of an idempotency key needs a database that enforces the unique index.

A fake IStepTools has to implement RunScopedAsync itself: the default throws NotSupportedException, so that fakes written before it keep compiling. Creating a scope from a provider of the test and running the handler in it is all it takes:

using Inngest;
using Microsoft.Extensions.DependencyInjection;

public sealed class ScopingStepTools(IServiceProvider services) : IStepTools
{
    public async Task<T> RunScopedAsync<T>(string id, Func<IServiceProvider, CancellationToken, Task<T>> handler)
    {
        await using var scope = services.CreateAsyncScope();
        return await handler(scope.ServiceProvider, CancellationToken.None);
    }

    public async Task<T> RunAsync<T>(string id, Func<CancellationToken, Task<T>> handler) => await handler(CancellationToken.None);

    public Task SleepAsync(string id, TimeSpan duration) => Task.CompletedTask;

    public Task SleepUntilAsync(string id, DateTimeOffset until) => Task.CompletedTask;

    public Task<InngestEvent<TData>?> WaitForEventAsync<TData>(string id, WaitForEventOptions options) =>
        Task.FromResult<InngestEvent<TData>?>(null);

    public Task<TResult> InvokeAsync<TResult>(string id, InvokeOptions options) => throw new NotSupportedException();

    public Task<SendEventsResult> SendEventAsync(string id, IReadOnlyList<InngestEvent> events) => throw new NotSupportedException();
}

The overloads that take a StepOptions, or a service type, or no result, all end up in that one method.