Coordinate Replicas

A component runs as several replicas, and a few requirements concern all of them at once:

• “The nightly settlement must never run twice at the same time”: only one replica may do it.
• “Every replica is woken to drain the mail queue”: the work has to be split between them.
• “A customer gets ten appraisals a minute”: the limit counts calls on every replica.

context.atom answers these three. Every component has it; nothing is declared in the manifest. When they are needed and when they are not is in the model article Shared Decisions.

§Set it up

context.atom needs Redis, named in the Context. toa create wrote the address of the one npm run dock starts; give production its own:

# context.toa.yaml
atomicity:
  redis: redis://localhost
  redis@production: redis://redis.example.com

Pulses and the recovery of events that failed to publish use it too, so an application with those needs the annotation even if no operation touches context.atom.

§Only one replica at a time

lock runs a function while no other replica of the component holds the same key. Here it is a new operation of store.orders:

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

export async function effect (input: null, context: Context) {
  return await context.atom.lock('settlement', async () => {
    const batch = await collect(context)

    await submit(batch, context)

    return { submitted: batch.length }
  })
}

• The first argument is a key, or an array of keys to hold together.
• The call waits for as long as it takes to get the lock. There is no timeout; other replicas asking for the same key queue behind the holder. The wait counts toward the operation’s duration.
• The lock is held while the function runs and released when it returns or throws. What the function returns is what lock returns.
• A replica that dies while holding a lock loses it: the lock expires.

§When the lock is lost

A lock can be lost while the function runs, when Redis becomes unreachable for instance. The function is told through an AbortSignal passed as its argument. Check it before a step that must not overlap:

await context.atom.lock('settlement', async (signal) => {
  const batch = await collect(context)

  if (signal.aborted) throw signal.error

  await submit(batch, context)
})

§What not to lock

Do not lock around changes to an entity: concurrent changes and repeated calls are already handled there. Use a lock for work that reaches outside the application and must not overlap with itself.

§Split work between replicas

slots(total) divides total slots among the replicas that are running and answers which of them this replica owns. Each slot belongs to one replica at a time:

context.atom.slots(128) // [0, 2, 4, …] with two replicas, or null

notifications has a pulse that drains an outgoing queue of an external mail system, divided into 16 partitions. Every replica gets the pulse; each takes its share:

// components/notifications/operations/drain.ts
import type { Context } from '../types/index.d.ts'

export async function effect (input: null, context: Context) {
  const slots = context.atom.slots(16)

  if (slots === null) return

  for (const partition of slots)
    await drain(partition, context)
}

slots returns null while the replica owns nothing: at startup, during a rollout, or while Redis is unreachable. Do nothing in that case. Do not treat null as “all of them”.

A replica that has just started owns nothing for about two intervals. Slots are redistributed whenever a replica starts or stops, so ask every time; do not remember the answer.

With one replica, it owns every slot: slots(4) is [0, 1, 2, 3].

§Be told when the share changes

Code that asks rarely, once an hour say, may ask exactly during a rollout and lose the whole hour. onassigned calls a listener with the assignment instead, at once and again each time it changes:

const off = context.atom.onassigned((assignment) => {
  if (assignment === null) return pause()

  const { i, n } = assignment // this replica is number i of n

  resume(i, n)
})

• The listener is called once as it is added, with the assignment as it stands, and then only on a change: one arrived, was lost, or the number of replicas changed.
• assignment is { i, n }, or null while the replica owns nothing. The slots owned out of total are i, i + n, i + 2n, … below total.
• The returned function removes the listener.
• A listener must not throw. One that does leaves this replica owning nothing from then on, and slots answers null.

§Limit calls across replicas

meter(keys, deltas) counts usage per key over all replicas. Usage is kept as a debt in milliseconds that drains by itself, one millisecond per millisecond. The call adds each delta to the debt under the matching key and returns what each debt has reached:

const [debt] = await context.atom.meter(['customer:42'], [1000])

To allow a customer ten requests a minute to a paid external service, charge each request a tenth of a minute and refuse when more than a minute is owed:

# components/store.catalog/manifest.toa.yaml
name: catalog
namespace: store

operations:
  appraise:
    input:
      type: object
      properties:
        customer: { type: string }
        product: { type: string }
      required: [customer, product]
    errors: [TOO_MANY_REQUESTS]

// components/store.catalog/operations/appraise.ts
import type { AppraiseInput, Context } from '../types/index.d.ts'

const COST = 6_000 // one request is worth 6 seconds: ten per minute
const WINDOW = 60_000

export async function effect (input: AppraiseInput, context: Context) {
  const [debt] = await context.atom.meter([`appraise:${input.customer}`], [COST])

  if (debt > WINDOW) return new Error('TOO_MANY_REQUESTS')

  return await appraise(input.product, context)
}

appraise stands for the request to the paid service.

A refused request has still added its cost, so a caller that keeps asking stays refused. To read a debt without adding to it, pass a delta of 0.

Several keys can be charged in one call, a per-customer and a global one for instance:

const [mine, everyone] = await context.atom.meter(
  [`appraise:${input.customer}`, 'appraise'],
  [COST, COST]
)

§Try it

$ npm run restart

$ npx toa call store.catalog.appraise "{ input: { customer: 'c1', product: 'a1' } }"

The eleventh call within a minute replies with the error TOO_MANY_REQUESTS. A single local replica owns every slot and never waits for a lock; start the same components a second time, in another terminal, to see the slots divide between the two:

$ npx toa compose ./components/*

§Keys belong to the component

A key used by store.orders is not the same key in billing.wallets, and applications sharing one Redis do not meet.

§The annotation

# context.toa.yaml
atomicity:
  redis@local: redis://localhost:6379
  redis@production:
    - redis://redis-0.example.com
    - redis://redis-1.example.com
    - redis://redis-2.example.com
  interval: 5000

Several addresses are independent servers, not the nodes of one cluster. Give three or more where a lock must survive the loss of a Redis server: with a single address, a Redis restart or failover can leave two holders of one lock. slots and meter use the first address only. An even number is refused when the process starts:

Atomicity takes an odd number of addresses, 2 given.

The address carries no credentials.

§When Redis is missing or unreachable

An unreachable Redis does not prevent a component from starting. Work that depends on slots pauses and resumes by itself.

In rare cases a piece of work coordinated this way is still done twice. Write it as you write an operation that can run more than once.

§What you see in metrics

With metrics on:

group is the component. A rising lock.wait means replicas are queueing for the same key.