fabric/sweeper

Automatic recovery for a leased store: a sweeper that scans for runs whose lease expired (their node died), for idle graph waits that are due, and for approval requests whose deadline passed (agent.with_approval_expiry, graph.with_approval_expiry), and recovers each through its registered root. Recovery rejects an expired agent request, so the model sees it, and fails a graph run whose request expired (graph.ExpiredApproval).

Register every root agent and root graph whose runs the sweeper may meet, then put supervised under the application’s supervisor in place of store.supervised: it starts the store’s subtree and the sweeper in the order recovery needs, so the sweeper always stops before the store’s runners drain.

let assert Ok(runs) =
  store.leased(name, node: "node-a", lease: duration.seconds(30), backend:)
let assert Ok(subtree) =
  sweeper.supervised(
    runs,
    [sweeper.agent(desk, context: fn(_run) { Context(user: "sweeper") })],
    every: duration.seconds(5),
  )
static_supervisor.new(static_supervisor.OneForOne)
|> static_supervisor.add(subtree)
|> static_supervisor.start

Types

Root

</>

A registered root and its deployed recovery code. Build one with agent or graph.

pub type Root =
  @internal Root

SweeperError

</>

Why supervised refused its configuration.

pub type SweeperError {
  EveryNotPositive(duration.Duration)
  EveryTooLarge(
    value: duration.Duration,
    limit: duration.Duration,
  )
  DuplicateRoot(run.DefinitionId)
  StoreNotLeased
  StoreNotRunning
}

Constructors

```
EveryNotPositive(duration.Duration)
```
Shorter than 1 ms.
```
EveryTooLarge(value: duration.Duration, limit: duration.Duration)
```
Longer than the longest timer the runtime can set (limit, 2^32 - 1 ms).
```
DuplicateRoot(run.DefinitionId)
```
Two roots share this agent or graph definition.
```
StoreNotLeased
```
The sweeper recovers through leases: only a store.leased store has them.
```
StoreNotRunning
```
start found no running process of the store (store.start it first).

Values

agent

</>

pub fn agent(
  agent: agent.Agent(context, answer),
  context context: fn(run.RunId) -> context,
) -> Root

The root agent agent, whose runs the sweeper recovers with the context context builds for each run (in 5 seconds at most).

describe_error

</>

pub fn describe_error(error: SweeperError) -> String

graph

</>

pub fn graph(
  identity: run.DefinitionId,
  build build: fn(store.Store) -> graph.Runtime(
    context,
    state,
    answer,
  ),
) -> Root

The root graph identity. build rebuilds its complete runtime, and its children’s, against the store it is given, in 5 seconds at most; the sweeper also discovers the graph’s idle waits that are due, including those whose local wakeup was lost.

start

</>

pub fn start(
  store: store.Store,
  roots: List(Root),
  every every: duration.Duration,
) -> Result(process.Pid, List(SweeperError))

Starts the sweeper of a store already started with store.start, for scripts and tests, linked to the caller; it stops when the store or the caller stops, also when the caller exits normally. Returns the sweeper’s process. Under a supervisor use supervised, which also orders the sweeper after the store.

supervised

</>

pub fn supervised(
  store: store.Store,
  roots: List(Root),
  every every: duration.Duration,
) -> Result(
  supervision.ChildSpecification(Nil),
  List(SweeperError),
)

The leased store’s subtree (see store.supervised) followed by its sweeper, for the application’s supervisor: use it instead of store.supervised, not beside it. The sweeper restarts with the store and stops before the store’s runners drain.

The sweeper scans at boot, then waits every after each bounded batch of at most 50 expired leases and 50 changed idle dependencies; no scans overlap. Each candidate is recovered through its registered root, in 30 seconds at most; a crashed running effect becomes uncertain according to its recovery contract. A failure or an unregistered root leaves its claim to expire and does not prevent the next root’s recovery. telemetry.sweep reports each scan; synchronous sweep handlers have 1 second before their emitter is stopped. Held work waits for its lease to expire (fabric.recover can take an earlier local lease at once). A signal wait without a deadline needs explicit delivery.

Every problem is reported at once.