Fan-out over a list

MapAsync runs a step for each item of a list and returns the results, as an array like Task.WhenAll, in the order of the items. It is Task.WhenAll over a list, with an id for every step, a limit on how many steps run at the same time, and a choice of what a failure does.

var invoices = await context.Step.RunAsync("load-invoices",
    ct => billing.GetOpenInvoicesAsync(customerId, ct));

string[] paymentIds = await context.Step.MapAsync("charge", invoices,
    (invoice, ct) => billing.ChargeAsync(invoice.Id, ct),
    new FanOutOptions { MaxParallelism = 5 });

Each item is a step with the id charge[0], charge[1] and so on, retried on its own like any other step. The methods are extensions of IStepTools, so they work on context.Step and on any fake of it. ForEachAsync is the same for a handler without a result:

await context.Step.ForEachAsync("notify", customerIds, (customerId, ct) => billing.NotifyAsync(customerId, ct));

You have	Use
A list of up to a few hundred items, and want the results	MapAsync
The same, and want no results	ForEachAsync
The same, and want to know which items failed instead of failing the run	MapOutcomesAsync
More items than one run can hold	MapInChunksAsync or MapOutcomesInChunksAsync, which run the list in child runs
Work per item that has steps of its own	InvokeAsync per item, or an event per item

The handler of an item is the handler of a step, so it cannot call steps. Parallel.ForEachAsync is not a replacement: it would run the handlers in the call itself and repeat them on every replay.

The list has to be the same on every call

Inngest runs a function by calling it again for every step, and the code before the fan-out runs each time. The steps of a fan-out are matched to the results Inngest recorded by their position, so the list must come out the same on every call: take it from the event, or from the result of a step, as load-invoices does above. A list read outside a step, from a database or from the clock, may differ between calls, and then charge[3] gets the result that was recorded for another item.

MapAsync takes an IReadOnlyList<T> on purpose. A lazy IEnumerable<T> would run its query again on every call.

Give each fan-out in a function an id of its own. With the same id, the second one gets numbered ids like any repeated step id, charge[0]:1, and which of two fan-outs that run in parallel is the second can depend on timing.

How many run at once

Without MaxParallelism every item starts at once. With it, the first items fill the window, and each time Inngest calls the function again with a result, the next item takes the place of the one that finished:

MaxParallelism = 2, five items	Items started
First call	0 and 1
After both have finished	2 and 3
After both have finished	4

Inngest calls the function again once all the steps started together have finished, as the parallel steps guide describes, so by default the window moves on in groups: a slow item holds back the next group. ParallelMode = ParallelMode.Race has Inngest call the function after each item, and the window moves on as soon as one item has finished, at the cost of more calls:

var paymentIds = await context.Step.MapAsync("charge", invoices,
    (invoice, ct) => billing.ChargeAsync(invoice.Id, ct),
    new FanOutOptions { MaxParallelism = 5, ParallelMode = ParallelMode.Race });

MaxParallelism bounds the items of this run. The function’s own concurrency limit bounds the steps of all its runs together, and the plan of your Inngest account bounds the steps that run at once overall.

What a list costs

Every call replays the function: the results of all the steps that finished are in the request, and the function reads them one by one before it reaches new work. A list of n items with MaxParallelism = w takes these calls:

Mode	Calls
Default	For each window of two or more items, one call that finds the items and one call per item. A window of one item runs in the call that finds it, when Inngest allows that. One more call returns the result.
Race	One call that finds the first items, one per item, and one per finished item that calls the function again: 2n + 1. A step of a race never runs in the call that finds it.

Seven items with MaxParallelism = 3 are windows of 3, 3 and 1, so (1 + 3) + (1 + 3) + 1 + 1 = 10 calls, which is what the dev server makes. The replay work and the size of each request grow with the number of results so far, so keep the result of an item small, an id rather than a document, and run long lists in chunks.

When an item fails

An item whose step fails for good, after its retries, throws StepFailedException where it is awaited. MapAsync and ForEachAsync stop at the first one:

• no item starts after the failure was seen;
• the call throws that item’s StepFailedException, whose StepId is charge[3], without waiting for the items in flight. The SDK cannot recall a step it has planned. By default Inngest calls the function again only once the whole window has finished, so the failure surfaces then; in race mode it surfaces as soon as that item has failed;
• when several items failed, the exception is the one Inngest recorded first, the same on every replay.

An item that started before the failure was seen is not recalled either. If the function catches the exception and goes on, the SDK still reports that item to Inngest, as it does for any step the function started and did not wait for.

try
{
    await context.Step.MapAsync("charge", invoices, (invoice, ct) => billing.ChargeAsync(invoice.Id, ct));
}
catch (StepFailedException exception)
{
    context.Logger.LogWarning("{StepId} failed: {Error}", exception.StepId, exception.Error.Message);
    throw;
}

Left uncaught, the exception fails the run without retries, like any uncaught StepFailedException.

To keep going and learn which items failed, use MapOutcomesAsync. It runs every item and returns one FanOutOutcome per item, in the order of the items, instead of throwing:

var outcomes = await context.Step.MapOutcomesAsync("charge", invoices,
    (invoice, ct) => billing.ChargeAsync(invoice.Id, ct));

foreach (var outcome in outcomes.Where(outcome => !outcome.Succeeded))
{
    context.Logger.LogWarning("{Invoice} was not charged: {Error}", invoices[outcome.Index].Id, outcome.Error!.Message);
}

var paymentIds = outcomes.Where(outcome => outcome.Succeeded).Select(outcome => outcome.Value!).ToList();

Index is the position in the list, Value is the result, and Error is the FunctionError the step failed with. Only failed steps become outcomes. Anything else the fan-out meets, such as a result that cannot be read as the item’s type, is thrown.

Cancellation

Every method takes a CancellationToken. Once it is canceled, the next item does not start, and the call throws OperationCanceledException. Items already planned are not recalled. The token has to be canceled the same way on every replay, or two calls start different items. Cancel it from what the function decides from the results of steps, and not from a timer or another thread. The token of the call, which Inngest cancels when it abandons the call, is safe to pass and changes nothing, since the call ends then anyway.

Lists too long for one run

Inngest fails a run once 1,000 of its steps have completed, so a run holds at most 999, and every step of the run counts: the items of a fan-out, the other steps of the function and, in the parent of chunks, the invoke steps. On the dev server a fan-out of 999 items completes, and one of 1,000 fails with “exceeded the step limit of 1000 steps” after all its steps ran. MapAsync therefore refuses 1,000 items or more with a NonRetriableException, since a retry would meet the same list, and fewer fit when the function has other steps. A longer list goes to child runs in chunks: the parent invokes one function per chunk, and each of those runs has a step for every item of its chunk.

The limits that decide the sizes are below. This SDK checks only the number of steps; the rest is for choosing sizes. The last column says what was tried against the dev server (inngest/inngest:v1.49.0) while this page was written. Nothing was tried on Inngest Cloud.

Limit	Value	What it bounds	Tried on the dev server
Steps in a run	1,000, documented	The items of a fan-out, the size of a chunk, the number of chunks	999 steps complete, 1,000 fail
Result of a step	4 MiB, documented	The result of one item	Not tried
Response of the app	8 MiB (output_too_large), not documented	The items of all the chunks planned in one call, and the outcomes a chunk’s run ends with	Items of 3.5 MB accepted, of 10 MB refused; outcomes of 7.8 MB accepted, of 11.7 MB refused after the items ran
Size of an event	256 KiB on Free, 3 MiB on Pro and Business, documented	A bound for the items of one chunk, which are the data of the event that starts its run	Chunks of 300 KiB, 1 MiB and 3.5 MB accepted
State of a run	32 MiB, documented	The outcomes of all the chunks, which land in the parent’s state	Not tried

Inngest documents no separate limit for the payload of an invoke. The dev server did not apply the limit of events to it, and what Inngest Cloud does is not verified here, so use the event size of your plan as a conservative bound.

The function of a chunk

The app provides the child by deriving from FanOutFunction<TItem, TResult> and implementing ProcessAsync for one item. The base class runs it for each item of its chunk, as a step of its own with the id item[index], and returns the outcomes. Register the class like any function, with an id and no trigger, because only the fan-out starts it:

ChargeInvoices.cs

using Inngest;

[InngestFunction("charge-invoices", Retries = 3)]
[Concurrency(10)]
public sealed class ChargeInvoices(IBilling billing) : FanOutFunction<Invoice, string>
{
    // How many items of a chunk run at the same time. The default is all of them.
    protected override int? MaxParallelism => 5;

    protected override Task<string> ProcessAsync(Invoice invoice, CancellationToken cancellationToken) =>
        billing.ChargeAsync(invoice.Id, cancellationToken);
}

Attributes

inngest.AddFunction<ChargeInvoices>();

Code first

inngest.AddFunction<ChargeInvoices>(new FunctionOptions
{
    Id = "charge-invoices",
    Triggers = [],
    Retries = 3,
    Concurrency = [new ConcurrencyLimit(10)],
});

The options of the function are those of the chunk’s run. Its Retries are the retries of each item’s step, and a concurrency limit bounds the steps of all the chunks at once.

Running the chunks

The parent names the class and the types, and says how many items go to a chunk:

string[] paymentIds = await context.Step.MapInChunksAsync<ChargeInvoices, Invoice, string>(
    "charge", invoices, chunkSize: 100, new FanOutOptions { MaxParallelism = 4 });

The parent has an invoke step for every chunk, with the id charge[0-99], charge[100-199] and so on, which holds the position of both ends so that a chunk size changed between two calls cannot give a chunk the result of another. MaxParallelism is the number of chunks in flight, and ChunkTimeout is how long a chunk’s run may take; without it Inngest waits as long as its default for an invoke, a year. A chunk that times out fails its step and its run goes on.

Failures work as in MapAsync, with the difference that the failing step ran in another run:

• a chunk stops starting items once one has failed for good, and the parent starts no more chunks. The call throws a StepFailedException whose StepId is charge[137], the position of the item, and whose Error is the item’s error;
• a chunk whose run fails or times out throws a StepFailedException with the id of the chunk’s step;
• MapOutcomesInChunksAsync runs every item and returns an outcome for each. When a chunk’s run fails or times out, each of its items gets the error of the chunk’s step, although that run may have processed some or all of them. The parent cannot tell, so retrying the items it reports as failed can process an item twice; make ProcessAsync safe to repeat.

Two more things to know. A parent that fails does not cancel the chunk runs it has started; they run to the end and their results are dropped. And a chunkSize changed between two calls of a run never mixes results, because the ids carry both ends, but no chunk then has the id it had, so every chunk is invoked again and its items run again. Keep the chunk size the same for runs that are in flight.

Choosing the sizes

What decides the sizes in practice is what has to fit:

• The items of the chunks planned together fit in one response. The call that plans chunks answers with all of them, so chunkSize times the size of an item as JSON, times the chunks in flight (MaxParallelism, or all of them), should stay well under 8 MiB. One chunk should also stay under the event size of your plan.
• The outcomes of a chunk fit in the response that ends its run. An outcome is the result as JSON in {"index":…,"value":…}, which adds about 25 bytes to it. Each result is a step result, at most 4 MiB.
• All the outcomes fit in the parent’s state. They add up over the chunks, with the event and the other steps, to 32 MiB at most. Keep the result of an item small, a few numbers or an id. For work with no result, return a bool.
• The items themselves are small. A list that is too big for a response is a list of ids or ranges, such as page numbers, and ProcessAsync loads what it needs.

The call refuses a chunkSize of 1,000 or more, and a list that makes 1,000 chunks or more, with a NonRetriableException. A list beyond that is split between several runs, for example by an event per part.

Under Native AOT or trimming, add the chunk and its outcomes to your JsonSerializerContext, for each function of a chunk. The engine writes the output of a function by its runtime type, which is why the outcomes are an array and the context lists the array. See Native AOT:

using System.Text.Json.Serialization;
using Inngest;

[JsonSerializable(typeof(FanOutChunk<Invoice>))]
[JsonSerializable(typeof(FanOutOutcome<string>[]))]
internal sealed partial class FanOutJson : JsonSerializerContext;

Checkpointing

With checkpointing on (CheckpointingOptions, see the API overview), steps that follow one another run in one call. A fan-out with MaxParallelism = 1 is such a sequence, and runs in a single call up to the checkpointing MaxRuntime. A window of several items is a parallel group, which goes back to one call per step.

Testing

Mock the steps of the items by their ids, charge[0], charge[1], with the test host:

using Inngest.Testing;

var mocks = new StepMocks()
    .Run("charge[0]", "pay_1")
    .Run("charge[1]", "pay_2");

var run = await host.RunAsync("settle-invoices", new InngestEvent("billing/settle", new { customerId = "acme" }), mocks);

A parent that runs chunks mocks the invoke steps by their ids and class. The host runs one function at a time, so run the function of the chunk with its own event, a FanOutChunk<TItem> named inngest/function.invoked, and hand its output to the parent’s mock:

using Inngest.Testing;

var chunk = await host.RunAsync<ChargeInvoices>(
    new InngestEvent("inngest/function.invoked", new FanOutChunk<Invoice>(0, firstChunk, FailFast: true)));

var mocks = new StepMocks().Invoke<ChargeInvoices, FanOutOutcome<string>[]>(
    "charge[0-99]", chunk.GetOutput<FanOutOutcome<string>[]>()!);

A fake of IStepTools that runs handlers on the spot, as in the testing guide, runs MapAsync too. If the test is about failures, the fake has to throw StepFailedException for a handler that failed, as Inngest reports it.

Related

Parallel steps explains how Inngest runs steps started together, Invoking functions the child runs, and Fan-out with invoke one function per item with its own flow control. The Inngest docs describe parallel steps and fan-out with events.