Skip to content

Graceful Cancel, Budget-Bound Shard Connections, Measured Pool Sizing

Choose a tag to compare

View all tags
@bw19 bw19 released this 28 Sep 02:21
· 271 commits to main since this release
v0.11.0
a98c3a8

v0.11.0 gives a flow a way to stop gracefully. The immediate, unconditional stop is renamed Terminate,
and Cancel now delivers the cancellation to every step in progress as an error on its node's onError
transition, so a workflow can compensate, clean up, or carry on. The flow ends cancelled, failed, or
completed, depending on what its handlers do. Beneath that, a shard's connection budget no longer grows with the
fleet. A few replicas per shard dispatch it and split its budget, and every other replica holds a
two-connection reader pool there, so a host scaled wide for its own traffic can no longer over-connect the
database. Pool sizes come from measured sweeps on six database sizes instead of a model, sleeping steps get an
index of their own so they stop slowing every scheduling pass, and graph validation rejects field names that a
when expression could never reference.

Highlights

  • Graceful Cancel. Every step in progress receives the cancellation on its onError transition, where
    workflow.IsCancelled(err) reports true. The flow resolves cancelled when every unrecovered loss was a
    cancellation. (Breaking — see below.)
  • Terminate and the terminated status. The old immediate stop keeps its behavior under a new name and a
    terminal status of its own. (Breaking — see below.)
  • Cancellation crosses subgraphs. A cancelled child reaches its caller as an error workflow.IsCancelled
    recognizes, and a real child error now keeps its status code and properties across the boundary.
  • Each shard's connections are bounded by its budget, not by the fleet. Dispatcher replicas split the budget;
    every other replica holds two connections. A new dwarf_shard_dispatching gauge shows which is which.
  • Measured pool sizing. The connections-per-vCPU table is measured on 1–64 vCPU databases across 0.25–2 ms of
    round-trip time and interpolated between tiers. Short-RTT pools are markedly smaller than before.
  • Sleeping steps stop costing dispatch. A step waiting on flow.Sleep, a retry backoff or a long poll no
    longer adds to every scheduling pass.
  • Stricter graph validation. forEach, as and SetReducer field names must be identifiers.
    (Breaking — see below.)

Graceful Cancel

Engine.Cancel(ctx, rootFlowKey, reason) asks a flow to stop and lets the flow decide how. Every step in
progress when it is called is covered, and each receives the cancellation as an error on its node's onError
transition:

  • A step already running finishes first. Its changes are kept, but its next transition is replaced by the
    onError route.
  • A step that has not started is not run. It goes straight to its onError route.
  • An error the task returns itself wins over the cancellation and takes the ordinary error path.
  • A step with no onError transition fails its branch. The flow ends cancelled when every unrecovered
    loss in it was a cancellation, and failed when a real error is mixed in. The outcome is decided per loss,
    never from whether Cancel was called, so a flow whose handlers all recover ends completed.
  • Subgraphs. Cancellation reaches the whole tree under the root key. A subgraph caller runs once more when its
    child returns, so its code sees the child's result (a cancelled child's error is one workflow.IsCancelled
    recognizes) before its own cancellation is delivered.
  • Interrupts. An interrupted step is cancelled when it is next resumed, and its resume data is not acted on.
  • Once per flow. A second Cancel does nothing, so a call whose response was lost is safe to retry, even
    after a handler caught the first cancellation and the flow carried on. To stop a flow that recovered,
    Terminate it.

Cancel returns once the request is recorded, not when the flow stops; use Await for that. A Fork of a
cancelled flow is not under cancellation.

g.AddTransition("Charge", "Ship")
g.AddTransitionOnError("Charge", "Refund") // runs on a cancellation, with workflow.IsCancelled(onErr) true

err := engine.Cancel(ctx, flowKey, "customer withdrew the order")

Terminate(ctx, flowKey, reason) remains the unconditional alternative: in-flight work is abandoned, no handler
runs, and the flow reads terminated as soon as the call returns. FlowOutcome and FlowSummary report the two
reasons separately, as TerminateReason and CancelReason.

Budget-Bound Shard Connections

In v0.10.x every replica took at least two connections on every shard, so past a certain fleet size the shard's
connection budget stopped dividing and every added replica pushed the database further past it. Over-connection
is the direction in which a database collapses rather than merely slowing down, and a host scaled wide for its own
traffic could reach it without doing much workflow work at all.

Each shard now has a small number of dispatcher replicas that split its connection budget and run its steps.
Every other replica holds a reader pool of two connections there, enough to create, await and read flows, and
runs none of that shard's steps. The shard's total stays within its budget at any fleet size.

  • Selection is automatic. Each shard's dispatchers are chosen by a rank every replica computes the same way
    from the shared peer registry, and it re-balances as replicas join and leave. A replica that shuts down gives up
    its slot at the start of its drain, not the end, so a small shard is never left waiting on the slowest
    in-flight task.
  • A crashed dispatcher holds its slot for about 40 s, until its registry row ages out. The shard runs on its
    other dispatchers meanwhile; it stops only if all of them crash at once.
  • SetMaxOpenConns opts out. With an explicit pool size, every replica dispatches every shard at that size.
  • Observability. dwarf_shard_dispatching{shard} is 1 where this replica dispatches and 0 where it only
    reads. Sum it across the fleet: a shard summing to zero runs nothing. The engine also reads the database's
    connection limit at startup and logs a warning when the fleet approaches it.

Measured Pool Sizing

The connections-per-vCPU table that sizes each dispatcher's pool now comes from pool sweeps on Cloud SQL at 1, 2,
4, 8, 16, 32 and 64 vCPU across round-trip times of 0.25–2 ms. It replaces a model that over-provisioned at short
round-trip times on every tier measured, and worst on the smallest: a 1-vCPU shard derived 15 connections where 3
sustain its load. Each cell is the smallest pool that sustained that distance's achievable throughput, with a
safety margin applied on top. A shard between two measured sizes interpolates between them, and one declared
larger than 64 vCPU uses the 64-vCPU row.

Sleeping steps get an index of their own. Measured at 400k pending steps in one priority band, 396k of them
asleep: a scheduling pass goes from 47 to 5 ms on PostgreSQL, 38 to 5 ms on SQL Server, 295 to 12 ms on SQLite,
and 192 to 93 ms on MySQL. MariaDB's optimizer does not use it and is unchanged.

The scheduler's scan for work and its supply of work to the workers now run as two independently paced loops,
so a slow scan over a deep backlog no longer throttles how fast workers receive steps.

Other Changes

  • Many fairness keys work on SQL Server. More than about 2,000 active fairness keys exceeded SQL Server's
    per-statement parameter limit, and that shard quietly stopped supplying work. On PostgreSQL and SQL Server the
    fetch also stops early per key instead of ranking every due row.
  • when expressions compare integers exactly rather than through a float64, and their literals are exactly
    true, false and finite numbers.
  • A shard on a slow database kept its readers from being promoted, so it ran nothing after its dispatchers
    left.
  • A rolling restart could briefly exceed a shard's connection budget.
  • A cancel reason containing % was garbled.
  • Dependencies. boolexp v1.3.0.

Breaking Changes

  • Cancel is renamed Terminate, and Cancel is now the graceful operation. Code that called Cancel for
    an immediate stop still compiles, but it now gets graceful cancellation: steps are not abandoned, and onError
    handlers run. Change those call sites to Terminate.
  • New terminal status terminated. workflow.StatusTerminated joins completed, failed and cancelled,
    and cancelled now means a graceful cancellation. Code that switches on a status string must handle the new
    value; workflow.IsTerminalStatus and workflow.TerminalStatuses include it.
  • FlowOutcome and FlowSummary gain TerminateReason, and CancelReason now carries only a Cancel
    reason.
  • Validate rejects a forEach, as or SetReducer field name that is not an ASCII identifier (a letter or
    underscore, then letters, digits or underscores). A dotted name such as order.items was a nested path in
    when but a flat key in forEach, so a fan-out over it found no field and ran zero branches without an error.
    State itself may still carry fields under any name.

Operational Prerequisites

  • Stop the whole fleet to upgrade, as for every dwarf release: no replica of v0.10.x may run against a
    v0.11.0 database.
  • Apply the schema changes by hand, on every shard, before starting v0.11.0. The engine does not migrate a
    database created by v0.10.x forward: without the statements below, the new columns and index are never added,
    and the first query that uses them fails. Flows in progress carry over. The two UPDATE statements relabel
    flows and steps that v0.10.x stopped with its Cancel, which v0.11.0 calls terminated; without them those
    rows read as graceful cancellations.

MySQL and MariaDB:

ALTER TABLE dwarf_flows ADD COLUMN cancel_watermark BIGINT NOT NULL DEFAULT 0;
ALTER TABLE dwarf_steps
    ADD COLUMN cohort_cancellations INT NOT NULL DEFAULT 0,
    ADD COLUMN cancelling TINYINT NOT NULL DEFAULT 0;
ALTER TABLE dwarf_peers
    ADD COLUMN working SMALLINT NOT NULL DEFAULT 1,
    ADD COLUMN rtt_us BIGINT NOT NULL DEFAULT 0;
CREATE INDEX idx_dwarf_steps_due ON dwarf_steps
    (status, parked, priority, not_before, fairness_key, created_at, step_id, lease_expires, fairness_weight);
UPDATE dwarf_flows SET status='terminated' WHERE status='cancelled';
UPDATE dwarf_steps SET status='terminated' WHERE status='cancelled';

PostgreSQL:

ALTER TABLE dwarf_flows ADD COLUMN cancel_watermark BIGINT NOT NULL DEFAULT 0;
ALTER TABLE dwarf_steps
    ADD COLUMN cohort_cancellations INT NOT NULL DEFAULT 0,
    ADD COLUMN cancelling SMALLINT NOT NULL DEFAULT 0;
ALTER TABLE dwarf_peers
    ADD COLUMN working SMALLINT NOT NULL DEFAULT 1,
    ADD COLUMN rtt_us BIGINT NOT NULL DEFAULT 0;
CREATE INDEX idx_dwarf_steps_due ON dwarf_steps (status, parked, priority, not_before)
    INCLUDE (fairness_key, created_at, step_id, lease_expires, fairness_weight) WHERE status = 'pending';
UPDATE dwarf_flows SET status='terminated' WHERE status='cancelled';
UPDATE dwarf_steps SET status='terminated' WHERE status='cancelled';

SQL Server:

ALTER TABLE dwarf_flows ADD cancel_watermark BIGINT NOT NULL DEFAULT 0;
ALTER TABLE dwarf_steps ADD
    cohort_cancellations INT NOT NULL DEFAULT 0,
    cancelling TINYINT NOT NULL DEFAULT 0;
ALTER TABLE dwarf_peers ADD
    working SMALLINT NOT NULL DEFAULT 1,
    rtt_us BIGINT NOT NULL DEFAULT 0;
CREATE INDEX idx_dwarf_steps_due ON dwarf_steps (status, parked, priority, not_before)
    INCLUDE (fairness_key, created_at, step_id, lease_expires, fairness_weight) WHERE status = 'pending';
UPDATE dwarf_flows SET status='terminated' WHERE status='cancelled';
UPDATE dwarf_steps SET status='terminated' WHERE status='cancelled';

SQLite:

ALTER TABLE dwarf_flows ADD COLUMN cancel_watermark INTEGER NOT NULL DEFAULT 0;
ALTER TABLE dwarf_steps ADD COLUMN cohort_cancellations INTEGER NOT NULL DEFAULT 0;
ALTER TABLE dwarf_steps ADD COLUMN cancelling INTEGER NOT NULL DEFAULT 0;
ALTER TABLE dwarf_peers ADD COLUMN working INTEGER NOT NULL DEFAULT 1;
ALTER TABLE dwarf_peers ADD COLUMN rtt_us INTEGER NOT NULL DEFAULT 0;
CREATE INDEX idx_dwarf_steps_due ON dwarf_steps
    (status, parked, priority, not_before, fairness_key, created_at, step_id, lease_expires, fairness_weight)
    WHERE status = 'pending';
UPDATE dwarf_flows SET status='terminated' WHERE status='cancelled';
UPDATE dwarf_steps SET status='terminated' WHERE status='cancelled';

The column additions are quick on every dialect. The index build scales with the size of dwarf_steps; on a large
PostgreSQL table, CREATE INDEX CONCURRENTLY avoids blocking writes while it runs.

Known Issues

  • A replica that dies between finishing a step and starting its successor can leave the flow running
    forever.     The engine detects it (dwarf_flows_orphaned) but does not repair it; recover the flow with Fork.
  • Crash recovery of a claimed step is bounded by a 5-minute sweep, so it takes up to TimeBudget + 30 s + 5
    minutes, and a short TimeBudget barely shortens it.
  • A replica that loses its dispatcher slot while holding a very large in-flight backlog on that shard may
    finish it slowly enough that some steps re-run elsewhere. Execution is at-least-once, so no state is corrupted.

Documentation

Flows. Flows documents Cancel and Terminate side by side, the cancelled and
terminated statuses, and how a cancellation meets onError handlers, subgraphs and interrupts.
Fan-Out and Subgraphs, Tasks and Graphs
follow, the last with the identifier rule for declared field names.

Operations. Deployment covers dispatchers and reader pools, states the crash-recovery
bounds as they are, adds a worked drain-window example (a 15-minute task cap needs a termination grace period of
about 16 minutes), and adds reference configurations with measured throughput and a sizing example.
Observability adds dwarf_shard_dispatching, and Upgrading states
that a flow's graph is fixed when the flow is created, while task implementations may change behind stable names.

Benchmarks. Cloud benchmarks add the 64-vCPU tier and sustained-versus-peak
capacity: sustained throughput runs at about 81–84% of what a short measurement window reports.

Assets 2
Loading