Flow control

Options that decide when and how many runs go. You set them on the function; the engine does the rest. The code is in examples/ts/flow-control.ts and examples/go/flowcontrol/main.go.

Problem	Option	What happens to extra work
An outside API takes only N calls at once	`concurrency`	waits
Start at most N runs a minute, evenly	`throttle`	waits
Start at most N runs an hour, drop the rest	`rateLimit`	dropped
Many events in a burst; only the last one matters	`debounce`	replaced by the newest
Handle events in groups (one insert for many)	`batchEvents`	grouped
Some runs should go first	`priority`	goes first
One run per order ID, whatever happens	`idempotency`	ignored
Only one run per user at a time	`singleton`	skipped, or the old run cancelled

Most options take a key: an expression over event, like "event.data.accountId". With a key, the limit counts each value on its own: "5 at once per account" instead of "5 at once in all".

Durations are text like "30s", "1m", "24h" in TypeScript, and a time.Duration in Go.

The engine checks the options when you sync. A bad value is refused with a message such as function f: throttle period must be between 1s and 168h.

Concurrency

Limits how many steps of the function run at the same time.

TypeScriptGo

const syncCrm = flow.createFunction(
  {
    id: "sync-crm",
    // At most 5 steps at once per account, and 50 at once in all.
    concurrency: [{ limit: 50 }, { limit: 5, key: "event.data.accountId" }],
  },
  { event: "crm/sync" },
  async ({ event, step }) => {
    await step.run("sync", () => syncAccount(event.data.accountId));
  },
);

{"crm/sync", flow.FunctionOpts{
	ID: "sync-crm",
	// At most 5 steps at once per account, and 50 at once in all.
	Concurrency: []flow.Concurrency{
		{Limit: 50},
		{Limit: 5, Key: flow.StrPtr("event.data.accountId")},
	},
}},

Steps, not runs. The limit counts calls to your app that are in progress. A run that is sleeping or waiting for an event holds no slot. So 1,000 runs can be in progress with a limit of 5; only 5 do work at any moment.
When the limit is full, new work waits its turn (see priority). Nothing is dropped.
concurrency: 5 (TypeScript) is short for { limit: 5 }. A limit of 0 means no limit. Up to 100,000.
Up to 2 limits per function. Two limits can not have the same scope and key.
If your app dies in the middle of a step, its slot is freed after the step timeout plus 2 minutes.

With scope: "account" and a key, every function in the workspace that uses the same key expression and gets the same value shares one limit. Good for "all our code may call this partner API 10 times at once". scope: "env" means the same as "account", and both need a key.

TypeScriptGo

// Two functions share one limit: 10 calls at once to the partner's API, in all.
const partnerLimit = { limit: 10, key: "'partner-api'", scope: "account" as const };

const pushOrders = flow.createFunction(
  { id: "push-orders", concurrency: partnerLimit },
  { event: "partner/orders-changed" },
  async ({ event, step }) => {
    await step.run("push", () => callPartner("orders", event.data));
  },
);

const pushStock = flow.createFunction(
  { id: "push-stock", concurrency: partnerLimit },
  { event: "partner/stock-changed" },
  async ({ event, step }) => {
    await step.run("push", () => callPartner("stock", event.data));
  },
);

// Two functions share one limit: 10 calls at once to the partner's API, in all.
{"partner/orders-changed", flow.FunctionOpts{
	ID:          "push-orders",
	Concurrency: []flow.Concurrency{{Limit: 10, Key: flow.StrPtr("'partner-api'"), Scope: "account"}},
}},
{"partner/stock-changed", flow.FunctionOpts{
	ID:          "push-stock",
	Concurrency: []flow.Concurrency{{Limit: 10, Key: flow.StrPtr("'partner-api'"), Scope: "account"}},
}},

The key 'partner-api' is a fixed text, so every run of both functions gets the same value.

Throttle

Starts at most limit runs per period, spread out evenly. Extra runs wait; none are lost.

TypeScriptGo

const resizeImage = flow.createFunction(
  {
    id: "resize-image",
    // Start at most 60 runs a minute per user (one a second), plus 10 at once
    // after a quiet time. Extra runs wait; none are lost.
    throttle: { limit: 60, period: "1m", burst: 10, key: "event.data.userId" },
  },
  { event: "image/uploaded" },
  async ({ event, step }) => {
    await step.run("resize", () => resize(event.data.imageId));
  },
);

{"image/uploaded", flow.FunctionOpts{
	ID: "resize-image",
	// Start at most 60 runs a minute per user (one a second), plus 10 at once
	// after a quiet time. Extra runs wait; none are lost.
	Throttle: &flow.Throttle{Limit: 60, Period: time.Minute, Burst: 10, Key: flow.StrPtr("event.data.userId")},
}},

With limit: 60, period: "1m", one run can start each second. burst lets that many more start at once after a quiet time (default 0).
Throttle acts when a run starts. After that, its steps are not slowed. To limit work in progress, use concurrency.
limit 1 to 100,000; period 1 second to 7 days.
Applies to runs from events, cron and flow.start. Not to step.invoke.

Rate limit

Starts at most limit runs per period. Extra events are dropped: no run, and no error to the sender.

TypeScriptGo

const lowStockAlert = flow.createFunction(
  {
    id: "low-stock-alert",
    // At most one run per product every 4 hours. Extra events start no run.
    rateLimit: { limit: 1, period: "4h", key: "event.data.productId" },
  },
  { event: "stock/low" },
  async ({ event, step }) => {
    await step.run("alert", () => alert(`low stock: ${event.data.productId}`));
  },
);

{"stock/low", flow.FunctionOpts{
	ID: "low-stock-alert",
	// At most one run per product every 4 hours. Extra events start no run.
	RateLimit: &flow.RateLimit{Limit: 1, Period: 4 * time.Hour, Key: flow.StrPtr("event.data.productId")},
}},

On a local engine, three stock/low events (two for p1, one for p2) started two runs: one for each product.

Use it when extra work is truly not wanted (alerts). When the work must happen, use throttle.
limit 1 to 100,000; period 1 second to 7 days.
Applies only to runs started by events. Not to cron or step.invoke.

Debounce

Waits until events stop coming for period, then starts one run with the last event.

TypeScriptGo

const indexDocument = flow.createFunction(
  {
    id: "index-document",
    // Wait until a document has had no edits for 5 minutes, then run once, with
    // the last edit. Never wait more than 30 minutes in all.
    debounce: { period: "5m", key: "event.data.documentId", timeout: "30m" },
  },
  { event: "document/edited" },
  async ({ event, step }) => {
    await step.run("index", () => indexDoc(event.data.documentId));
  },
);

{"document/edited", flow.FunctionOpts{
	ID: "index-document",
	// Wait until a document has had no edits for 5 minutes, then run once, with
	// the last edit. Never wait more than 30 minutes in all.
	Debounce: &flow.Debounce{Period: 5 * time.Minute, Key: flow.StrPtr("event.data.documentId"), Timeout: 30 * time.Minute},
}},

Each new event pushes the start back by period.
timeout caps the whole wait: the run starts at the latest timeout after the first event, even if events keep coming.
period and timeout 1 second to 7 days.
If the function is paused when the run would start, the event is dropped.
Debounce ignores idempotency and a future ts. It can not be used with batchEvents.

Batching

Collects events and starts one run with many of them.

TypeScriptGo

const saveClicks = flow.createFunction(
  {
    id: "save-clicks",
    // Up to 100 events per run, or what came in 10 seconds, per site.
    batchEvents: { maxSize: 100, timeout: "10s", key: "event.data.siteId" },
  },
  { event: "page/clicked" },
  async ({ events, step }) => {
    await step.run("insert", () => insertClicks(events.map((e) => e.data)));
    return { saved: events.length };
  },
);

{"page/clicked", flow.FunctionOpts{
	ID: "save-clicks",
	// Up to 100 events per run, or what came in 10 seconds, per site.
	BatchEvents: &flow.BatchEvents{MaxSize: 100, Timeout: 10 * time.Second, Key: flow.StrPtr("event.data.siteId")},
}},

In Go the handler reads the batch from in.Events:

// With batchEvents, in.Events holds every event of the batch.
count := len(in.Events)

On a local engine, three page/clicked events for one site started one run, 10 seconds after the first, with all three events.

The run starts when the batch has maxSize events, or timeout after its oldest event, or when its events add up to 4 MB, whichever comes first.
The handler gets all of them in events (Go: in.Events); event is the first.
if (optional) picks which events go into batches. An event that does not match starts a run of its own at once.
maxSize 1 to 100; timeout 1 second to 1 hour.
Batching ignores idempotency. It can not be used with debounce.

Priority

Lets some runs go first when they wait for a concurrency slot.

TypeScriptGo

const generateReport = flow.createFunction(
  {
    id: "generate-report",
    concurrency: { limit: 10 },
    // Runs waiting for a slot: enterprise ones go first, as if they had waited
    // 2 minutes longer.
    priority: { run: "event.data.plan == 'enterprise' ? 120 : 0" },
  },
  { event: "report/requested" },
  async ({ event, step }) => {
    await step.run("build", () => buildReport(event.data.accountId));
  },
);

{"report/requested", flow.FunctionOpts{
	ID:          "generate-report",
	Concurrency: []flow.Concurrency{{Limit: 10}},
	// Runs waiting for a slot: enterprise ones go first, as if they had waited
	// 2 minutes longer.
	Priority: &flow.Priority{Run: flow.StrPtr("event.data.plan == 'enterprise' ? 120 : 0")},
}},

run is an expression that gives a number of seconds, from -600 to 600. 120 means "treat this run as if it had waited 2 minutes longer". A negative number puts a run back.
It only changes the order of work waiting for a concurrency slot. Without a concurrency limit (yours, or the plan's limit for the workspace, see below) nothing waits, and priority does nothing.
If the expression fails or does not give a number, the priority is 0.

Idempotency

At most one run per value of an expression.

TypeScriptGo

const sendReceipt = flow.createFunction(
  {
    id: "send-receipt",
    // One run per order, however many order/paid events come for it.
    idempotency: "event.data.orderId",
  },
  { event: "order/paid" },
  async ({ event, step }) => {
    await step.run("email", () => emailReceipt(event.data.orderId));
  },
);

{"order/paid", flow.FunctionOpts{
	ID: "send-receipt",
	// One run per order, however many order/paid events come for it.
	Idempotency: flow.StrPtr("event.data.orderId"),
}},

On a local engine, two order/paid events for o1 started one run.

An event with a value already seen gets no new run, while the engine keeps the first run: while it runs, and 3 days after it ends (4 days after it failed).
The value is tied to the function's full ID. Deleting the app and adding it again with the same ID does not forget the values.
Build the value from several fields with an expression: "event.data.userId + '-' + event.data.day".
If the expression fails (a missing field), the value is "null", and all those events share one run. Make sure the field is always there.
Applies only to runs started by events, and not with debounce or batchEvents.
To drop a repeated event (the same thing sent twice), give the event an id.

Singleton

Only one active run at a time, in all or per key.

TypeScriptGo

const rebuildSearch = flow.createFunction(
  {
    id: "rebuild-search",
    // One run at a time per site. While one runs, a new event starts no run.
    singleton: { key: "event.data.siteId", mode: "skip" },
  },
  { event: "search/rebuild" },
  async ({ event, step }) => {
    await step.run("rebuild", () => rebuild(event.data.siteId));
  },
);

{"search/rebuild", flow.FunctionOpts{
	ID: "rebuild-search",
	// One run at a time per site. While one runs, a new event starts no run.
	Singleton: &flow.Singleton{Key: flow.StrPtr("event.data.siteId"), Mode: "skip"},
}},

mode: "skip": while a run is active, new events start no run.
mode: "cancel": the active run is cancelled, and the new one starts. Good for "only the newest request matters".
A run is active until it ends (completed, failed, cancelled), also while it sleeps or waits.
Applies to runs from events, cron and flow.start. Not to step.invoke.
To learn which run holds a key, use flow.runs.singleton(fn, key). To cancel it, flow.runs.cancel({ function, singletonKey }). Both read the singleton itself, with no lag.

Start a run directly

flow.send answers as soon as the engine has the event; the singleton check comes later. flow.start starts a run of one function at once, and tells you if it did not start, for example because a singleton run holds the key:

// Starts the run now and says if it did not. `id` makes a retry of this call safe.
const result = await flow.start(runJob, { data: { jobId, key }, id: `job-${jobId}` });
if (!result.started && result.reason === "singleton") {
  console.log(`job ${key} already runs as ${result.activeRunId}`);
}

Singleton, throttle, concurrency and priority apply, as for an event.
Rate limit, debounce and batching do not: they are about streams of events.
idempotency does not apply either. Pass id instead: a retry with the same id lands on the same run, and answers existed: true.
The run exists when start answers, so a cancel sent right after it cannot miss.

How they work together

For each event and each matching function, the engine goes in this order:

The trigger's if. No match: no run.
idempotency gives the run its ID. A repeat ends here.
rateLimit may drop the event.
debounce or batchEvents hold the event; their run goes on at step 5 later.
singleton may skip the run (or cancel the old one).
throttle may move the start later.
While it runs, concurrency and priority decide which steps go next.

You can combine any of them, except debounce with batchEvents.

	Event	Cron	`flow.start`	`step.invoke` child
`concurrency`, `priority`	yes	yes	yes	yes
`throttle`, `singleton`	yes	yes	yes	no
`rateLimit`, `debounce`, `batchEvents`, `idempotency`	yes	no	no	no

Runs keep the flow control they started with. If you change these options and sync, only new runs use the new values.

The workspace's plan can also cap how many calls run at once over the whole workspace ("runs at once, in each workspace" on the plan). That limit works like concurrency: a sleeping or waiting run holds no slot.

Flow control ​

Concurrency ​

Share a limit between functions ​

Throttle ​

Rate limit ​

Debounce ​

Batching ​

Priority ​

Idempotency ​

Singleton ​

Start a run directly ​

How they work together ​

Flow control

Concurrency

Share a limit between functions

Throttle

Rate limit

Debounce

Batching

Priority

Idempotency

Singleton

Start a run directly

How they work together