Write a Multi-Step Workflow

Approving an order means more than a debit: the stock is reserved, a courier is booked through the courier’s API, and the customer is told. If the last step fails, the work is tried again, and it must not reserve the stock or book a courier a second time.

Write the steps as one effect and declare it under continuity. When it is tried again, the steps that already answered are not made again: the effect is given their answers and goes on from the step that failed.

§Declare the effect

fulfilment.dispatch ships an approved order:

# components/fulfilment.dispatch/manifest.toa.yaml
namespace: fulfilment
name: dispatch

configuration:
  schema:
    properties:
      courier: { type: string, format: uri }
    required: [courier]

operations:
  ship:
    type: effect
    input:
      type: object
      properties:
        order: { type: string }
      required: [order]

continuity:
  ship: 604800 # seconds an unfinished run can still be picked up

// components/fulfilment.dispatch/operations/ship.ts
import type { Context, ShipInput } from '../types/index.d.ts'

export async function effect (input: ShipInput, context: Context) {
  const order = await context.remote.store.orders.observe({ query: { id: input.order } })

  for (const item of order.items)
    await context.remote.fulfilment.stock.reserve({
      query: { id: item.product },
      input: { quantity: item.quantity }
    })

  const response = await context.fetch(`${context.configuration.courier}/bookings`, {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify({ order: order.id })
  })

  if (!response.ok) throw new Error(`Courier answered ${response.status}`)

  const booking = await response.json()

  await context.remote.notifications.send({
    input: { to: order.customer, text: `Your order is on its way: ${booking.tracking}` }
  })
}

Its steps are calls to what the rest of the shop has:

• observe of store.orders, an operation every component with an entity inherits; Create a Component writes the component.
• reserve of fulfilment.stock, the transition Two Writers at Once writes: it takes a quantity of a product out of stock. Each item of the order is a step of its own.
• The courier’s API, through context.fetch.
• send of notifications, the effect Call an External Service writes.

The operation is written like any other; nothing in it is about resuming. If send throws, the next attempt gets the order observe answered, the replies of reserve and the response of the courier without asking again, and makes only send.

continuity is a section of the manifest, beside operations. Each key names an effect of the component. The value is how long, in seconds, a run that has not finished can still be picked up. There is no default, and the minimum is 600. A manifest that states less, names an operation the component does not have, or names one that is not an effect is refused: 'transit' is not an effect.

§Start it

As a task, from approve, the transition Create a Component writes:

// components/store.orders/operations/approve.ts
import type { Context, State } from '../types/index.d.ts'

export async function transition (input: null, entry: State, context: Context) {
  if (entry.status !== 'pending')
    return new Error('NOT_PENDING')

  entry.status = 'approved'

  await context.remote.fulfilment.dispatch.ship({ input: { order: entry.id }, task: true })

  return entry
}

or from a receiver of the event store.orders.approved, which carries the order:

# components/fulfilment.dispatch/manifest.toa.yaml
receivers:
  store.orders.approved: ship

// components/fulfilment.dispatch/receivers/store.orders.approved.ts
import type { Entity as Order } from '@components/store.orders'

export function request (order: Order) {
  return { input: { order: order.id } }
}

Each task, and each event, is one run. Attempts of one run share the answers; two runs share nothing.

A call that waits for the answer throws, and a pulse may not name the operation. A debit of the customer’s wallet stays in approve for that reason: the customer waits to learn whether the money was there, and the debit is keyed by the order, so making it again takes nothing.

§Start the service that keeps the answers

Restart, approve a pending order, and nothing is shipped:

$ npm run restart
$ npx toa call store.orders.approve "{ query: { id: '7a1c5e0c2b8f4d0e9a3b6c1d2e4f5a6b' } }"

The approval answers, the task is accepted, and ship never runs. Nothing is logged. A continued effect is run by a service of Toa, continuity, which keeps the answers of its steps, and the application toa create wrote does not run that service. An extension that is declared while its service is not running fails silently: no error at the start and none afterwards.

sys is the process that runs what Toa serves beside the application. Add the service to its line in ecosystem.config.js, and restart:

// ecosystem.config.js
{ ...common, name: 'sys', args: 'serve configuration exposition introspection continuity' },

$ npm run restart

The opposite mistake is reported. A service named on that line while no component declares continuity stops sys when it starts:

Error: TOA_MONGODB_CONTINUITY_JOURNAL is not set.

npm run env writes that variable only once a component declares the extension. Declare first, then name the service.

§Try it

Make the last step fail: add throw new Error('down') to notifications.send for a moment, restart, and approve a pending order by its id.

$ npm run restart
$ npx toa call store.orders.approve "{ query: { id: '7a1c5e0c2b8f4d0e9a3b6c1d2e4f5a6b' } }"

The approval answers at once. In the logs (pm2 logs) ship is attempted again and again; every reserve and the courier are each called once. Remove the throw, run npm run restart, and the run finishes with the next attempt, still without a second reservation or booking.

§What counts as a step

A step is something the operation asks of its context:

A step is recognized by what it asks, not by when. Two attempts that make the same request are making the same step, so concurrent branches are fine:

await Promise.all([
  reserve(context, 'warehouse-a').then((r) => label(context, r)),
  reserve(context, 'warehouse-b').then((r) => label(context, r))
])

§Rules for the code

§Take time, randomness and ids from the context

An argument that differs between attempts makes a new step, and the step is made again:

// a new timestamp every attempt: the call is made again every attempt
await context.remote.billing.invoices.issue({ input: { at: Date.now() } })

// the same timestamp on every attempt: the call is made once
await context.remote.billing.invoices.issue({ input: { at: context.now() } })

A run sees one moment, one random value and one new id however many attempts it takes. They are on every operation context.

The same goes for anything else an argument is built from. An effect that reads its own entity reads it again on every attempt, and a step whose arguments come from it is made again if the entity changed.

§Keep nothing in the process

The next attempt may run in another process, after a restart or a deployment:

// counts in whichever process the attempt runs in, not across attempts
context.state.sent = (context.state.sent ?? 0) + 1

// kept with the run, and given back wherever the next attempt runs
const sent = await context.remote.notifications.send({ input })

A module variable, a cache in memory, a file on the local disk and context.state do not travel with the run. Anything a later attempt depends on comes from a step, or from state the operation writes and reads back.

§Take ids before branching

Two concurrent steps with the same arguments are the same request, and which branch gets which answer is not fixed. context.id() called in two concurrent branches is the common case: take the ids first, then branch.

§What can still repeat

A step that took effect just before the process stopped is made again, because its answer was not kept yet.

• A call to another component arrives as the same call. Declare the callee once, and the repeat changes nothing.
• A fetch to a provider is sent again. Pass a key the provider recognizes repeats by; context.id() gives one that is the same on every attempt:

const key = context.id()

const response = await context.fetch(`${context.configuration.courier}/bookings`, {
  method: 'POST',
  headers: { 'content-type': 'application/json', 'idempotency-key': key },
  body: JSON.stringify({ order: order.id })
})

If one run is delivered twice at the same moment, both copies go on with the answer the first of them kept. context.id(), context.now() and context.random() are the exception: each copy has its own value until one is kept, so a step built from it may be made by both.

§Failures you will see

• A returned error ends the run, as for any task. It is not tried again, and nobody reads it.
• A step that answers with a stream throws Unrecordable, and the run is set aside at once. A continued effect asks for values, and answers values.
• An answer larger than 16 MB cannot be kept, and the step throws.
• A run that keeps throwing is set aside for an operator after a few attempts; see Operating. Put back within the time stated under continuity, it picks up where it stopped. Put back later, it starts from the beginning.
• A run delivered again after it finished makes nothing.

§What it needs to run

The answers are kept by a service the runtime brings along, in the application’s MongoDB. It is deployed with the application as a service of its own when a component declares continuity.

A Context whose mongodb names one address for everything has nothing to add. One that names addresses per namespace names this one too:

# context.toa.yaml
mongodb:
  store: mongodb://store.mongo.example.com
  continuity: mongodb://mongo.example.com

While that database is away, runs wait and pick up when it is back.

Locally it is added to sys: above. Where the processes are started by hand, toa mono runs the service with the components; with toa compose, name it:

$ npx toa compose ./components/* --service continuity

§Checklist

• The operation is an effect, named under continuity with at least 600 seconds.
• It is started with task: true or from a receiver, never waited for.
• Every value that would differ between attempts comes from context.
• Nothing a later attempt needs lives in the process.
• Callees that change state declare once; requests to providers carry a key.
• No step answers with a stream.

What continuity is and where it applies: Continuity.