From 94fcb91ca0768dd4b89b0798dde2faa56165f1f9 Mon Sep 17 00:00:00 2001 From: Rami Date: Fri, 23 Oct 2026 13:42:15 +0100 Subject: [PATCH 1/2] fix(workflow-host): the follower notes a cancel request it cannot read, as it notes an ending A run_cancel_requested whose context names no run was passed over with no note, where an unreadable ending was noted. Each consumer of waiting calls now reads its record before it makes a delivery of it, and one step passes over every record the follower cannot read and notes it once: an event, a fact, a definition, an ending or a cancel request. Co-Authored-By: Claude Fable 5.1 --- packages/workflow-host/README.md | 6 +- .../workflow-host/src/follower/consumers.ts | 33 ++++++--- .../src/follower/follower-records.test.ts | 38 +++++++++++ .../src/follower/record-steps.test.ts | 7 +- .../src/follower/record-steps.ts | 68 +++++++++++-------- .../src/waiting/cancel-requests.test.ts | 41 ++++++----- .../src/waiting/cancel-requests.ts | 35 +++++++--- .../src/waiting/child-endings.test.ts | 27 +++++--- .../src/waiting/child-endings.ts | 21 ++++-- 9 files changed, 192 insertions(+), 84 deletions(-) diff --git a/packages/workflow-host/README.md b/packages/workflow-host/README.md index 298de404a..743202002 100644 --- a/packages/workflow-host/README.md +++ b/packages/workflow-host/README.md @@ -121,7 +121,7 @@ The follower keeps a place in every playbook: at its start it reads every regist Every `sweepEveryMs` the follower sweeps, so that a playbook written by another process, which raises no signal here, is caught up too. A sweep reads which streams were appended to since the last, through the ledger's `readAppended`, from the point where the last read stopped, at most 10,000 records at a time; it reads the registries among them, naming a registry again at every sweep until a read of it succeeded, while a read that fails is said and the sweep's upkeep and passes go on, and passes up to 128 playbooks: up to 64 of those whose deliveries or run records wait, those a sweep took longest ago first, and in the places left the playbooks appended to by anything but a run's log, so that playbooks that wait never take every place. Playbooks it has no place for, and records past the 10,000, are taken by later sweeps: the next comes at once while the read stopped at its 10,000 records or playbooks appended to are still left over, and otherwise after a whole `sweepEveryMs`, so playbooks that wait past the 64, the rest of the round after a start, and failed playbooks handed back that do not fit, wait for it. A pass that fails is said, the playbooks after it in the sweep are passed all the same, and its playbook is handed back to the sweeps, which queue it in order with the playbooks appended to without hurrying the next sweep: at least 64 places of every sweep go to that queue, so a playbook with n playbooks queued before it is passed within ⌈(n + 1) / 64⌉ of the sweeps that reach their passes, however many are appended to after it. A sweep that fails before its passes hands back every playbook it took, and takes nothing out of what is queued until it has read the round. At its start the follower takes its place in the ledger before it reads the registries, and its sweeps go once over every playbook it follows, in the places left, since nothing read the ledger while the host was down. A sweep that finds nothing new reads the ledger once and the playbooks that wait once, and passes no playbook. -A pass reads the playbook's records oldest first through the ledger's recorded read, 100 records a page and at most 10 pages, and keeps its place in `workflow_followed_playbooks` after each record it made a delivery of and once more when the pass ends, so a pass that fails makes again only the deliveries of the record it was at, and reads again the records it passed over since its last delivery. It first glances at the records without their data, the definitions' records apart, and reads them with their data only for the event types that the playbook's triggers and listeners and the consumers its caller gave name (`workflow_subscription_types` and `workflow_listener_types`): a fact is read with its data when its type is named, and every published event when any type that is not a fact's is named, since an event's type lies in its data. A record whose data the pass did not read is passed over, so a playbook where nothing names a type costs one read of identifiers a wake; a record that names more types ends the pass: a run's record that armed a listener is passed first, and a definition that activates a trigger is read again by the next pass, which reads it and the records after it with the data of those types too. A record of a run's log is passed only once the dispatch watermark of the run covers its version, so the listeners a record armed are kept by the time the follower reaches it, and passing that record marks them as passed: a listener takes only the events recorded after the record that armed it. A record the watermark has not covered by the twentieth sweep that met it is passed all the same, as a delivery is skipped, so a run whose dispatch is stuck does not hold back its playbook: the host notes `run_record_passed`, keeps in `workflow_passed_runs` how far it passed the run's log, and then marks as passed every listener of the run armed by that record or before; a listener's insert, for its part, marks the listener passed afterwards when that row is there, so each side writes before it reads, and a listener the dispatch keeps while the record is passed or later is marked, and takes the events recorded from then on. The row is written only while the run is going and its dispatch is behind the record, taken back at once if the dispatch reached it meanwhile, and deleted once the run's dispatch reaches that far or the run ends. A record the follower cannot read is noted (`record_unreadable`) and passed over. +A pass reads the playbook's records oldest first through the ledger's recorded read, 100 records a page and at most 10 pages, and keeps its place in `workflow_followed_playbooks` after each record it made a delivery of and once more when the pass ends, so a pass that fails makes again only the deliveries of the record it was at, and reads again the records it passed over since its last delivery. It first glances at the records without their data, the definitions' records apart, and reads them with their data only for the event types that the playbook's triggers and listeners and the consumers its caller gave name (`workflow_subscription_types` and `workflow_listener_types`): a fact is read with its data when its type is named, and every published event when any type that is not a fact's is named, since an event's type lies in its data. A record whose data the pass did not read is passed over, so a playbook where nothing names a type costs one read of identifiers a wake; a record that names more types ends the pass: a run's record that armed a listener is passed first, and a definition that activates a trigger is read again by the next pass, which reads it and the records after it with the data of those types too. A record of a run's log is passed only once the dispatch watermark of the run covers its version, so the listeners a record armed are kept by the time the follower reaches it, and passing that record marks them as passed: a listener takes only the events recorded after the record that armed it. A record the watermark has not covered by the twentieth sweep that met it is passed all the same, as a delivery is skipped, so a run whose dispatch is stuck does not hold back its playbook: the host notes `run_record_passed`, keeps in `workflow_passed_runs` how far it passed the run's log, and then marks as passed every listener of the run armed by that record or before; a listener's insert, for its part, marks the listener passed afterwards when that row is there, so each side writes before it reads, and a listener the dispatch keeps while the record is passed or later is marked, and takes the events recorded from then on. The row is written only while the run is going and its dispatch is behind the record, taken back at once if the dispatch reached it meanwhile, and deleted once the run's dispatch reaches that far or the run ends. A record the follower cannot read is noted once (`record_unreadable`) and passed over with nothing delivered of it, by the one step that passes over every such record (`src/follower/record-steps.ts`): an event or a fact the consumers would be handed that does not decode, a definition record of the workflows that cannot be decoded, and a record one of the consumers of [waiting calls](#waiting-calls) cannot read as what it takes, an ending of a run or a cancel request. Each event, and each fact the playbook records about a run or a definition (`playbookFactOf` of `@on.auto/definitions`), whose data the pass read is handed to the consumers in turn. A consumer is the shape the loop knows: @@ -141,7 +141,7 @@ interface Delivery { } ``` -`batchOf` answers up to `most` deliveries after the key of the last one made, each with a key, the workflow it is for and the effect that makes it, the key of the last candidate it looked at and whether more are left; at most 100 deliveries of a record are made in a pass, counted across all the consumers, and the next pass goes on from the last key, at the consumer it stopped at. A delivery that fails holds the record and is tried again at every sweep, not at a signal; after `skippedAfterSweeps` failed sweeps the delivery's `skipped` says so and the follower goes on; the delivery says it, since it knows the workflow and the version it was for. The consumers here are the offers to listeners and the starts of triggered workflows, which take their types from the playbook's listeners and triggers rather than from `types` and skip after 20 sweeps, and the two of [waiting calls](#waiting-calls), the endings of the runs calls wait for and the cancel requests, which name their types and are never skipped, since a skip would lose an answer or a cancel. A caller gives the host more through the option `consumers`, with `Consumer`, `Delivery`, `DeliveryFailed`, `FollowedRecord` and `FollowedEvent` exported from the package. Such a consumer names the event types it wants in `types`, the `type` of the event or of the fact, and is handed, after the two, only the events and facts of those types, in every playbook; a consumer whose `types` is empty wants nothing and is handed nothing, and there is no type that stands for every type. A consumer that must never skip a delivery sets `skippedAfterSweeps` to `Infinity`. +`batchOf` answers up to `most` deliveries after the key of the last one made, each with a key, the workflow it is for and the effect that makes it, the key of the last candidate it looked at and whether more are left; at most 100 deliveries of a record are made in a pass, counted across all the consumers, and the next pass goes on from the last key, at the consumer it stopped at. A delivery that fails holds the record and is tried again at every sweep, not at a signal; after `skippedAfterSweeps` failed sweeps the delivery's `skipped` says so and the follower goes on; the delivery says it, since it knows the workflow and the version it was for. The consumers here are the offers to listeners and the starts of triggered workflows, which take their types from the playbook's listeners and triggers rather than from `types` and skip after 20 sweeps, and the two of [waiting calls](#waiting-calls), the endings of the runs calls wait for and the cancel requests, which name their types, read each record as an ending or a request before they make a delivery of it, and are never skipped, since a skip would lose an answer or a cancel. A caller gives the host more through the option `consumers`, with `Consumer`, `Delivery`, `DeliveryFailed`, `FollowedRecord` and `FollowedEvent` exported from the package. Such a consumer names the event types it wants in `types`, the `type` of the event or of the fact, and is handed, after the two, only the events and facts of those types, in every playbook; a consumer whose `types` is empty wants nothing and is handed nothing, and there is no type that stands for every type. A consumer that must never skip a delivery sets `skippedAfterSweeps` to `Infinity`. ### Listeners and offers @@ -178,7 +178,7 @@ A call whose function answers that its run finishes later, `{ status: 'waiting', The run a call waits for records the call it answers on its start, `called_by`, and every finish of that run copies it, so the ending is a fact on the run's stream that names the call. The follower's consumer of child endings (`src/waiting/child-endings.ts`), which names the three types of a finish, gives a call that is `waiting` the ending of its run as `call_answered`, mapped by `waiting.resultOf`, then marks the row `answered` and the answer given; the engine takes a key once, so a delivery made again after a crash is stale and appends nothing. An ending that reaches a call that is closed, or one that is still `running`, whose `perform` answers it, is a receipt and no failure. When the run ended before the row was marked `waiting`, the executor finds its ending as it marks the row and answers the call at once; and since a host can die between marking the row and reading the ending, the first resume after a host opens or takes the claim on the workflows reads every `waiting` row and answers each whose run has ended. The run's waiting rows are also read before any timer of the run fires, and the endings of their runs given first, so a run that ended while the server was down answers before the deadline that came due meanwhile. -The consumer of cancel requests (`src/waiting/cancel-requests.ts`) takes `run_cancel_requested`, which `cancel_run` of `@on.auto/definitions` records on the run's stream on any server. For a workflow it gives the run the input `cancel_requested`, with who asked, the kind and the reason, caused by the request; a run whose start has not reached the host yet takes nothing, and the consumer records the request it passed over in `workflow_pending_cancels`, one row a run, with the request's message id as `cause`, `cancelled_by`, `kind` and `reason` (`src/waiting/pending-cancel-rows.ts`), before it moves on. The host's start of that run reads the run's stream for a cancel asked since its start and gives it at once (`src/waiting/pending-cancels.ts`), so the run starts and ends cancelled. Since a host can die between applying the start and that read, the first resume after a host opens or takes the claim on the workflows reads that table alone, a hundred rows at a time, and gives each row's cancel to its run, four at a time: a row whose run took it, or had ended, is deleted, one whose run has not started stays for the start to meet, and a run that cannot take its cancel is reported without holding the others, and given it again at the next sweep. A row whose run answers that it has not started is deleted once the newest record of the run, read newest first without its data, is a finish, so a run that ended without ever reaching the host leaves no row after the next first resume; a run never handed to the host and never ended keeps its row, one a stranded run. The consumer takes no finish, so the finish of a run nothing waits on writes nothing. The first resume reads none of the runs still going, so it costs the same however many there are. A cancel recorded before the run's start, which only its caller records, is no delivery: the start it precedes is refused. For a run of another capability it hands the request to `waiting.cancelDeferred`, the capability's pure decision over the run's record and the settlement. A conflict because the run has already ended is a success for both consumers, which never skip a delivery: one that fails holds the record until it succeeds. +The consumer of cancel requests (`src/waiting/cancel-requests.ts`) takes `run_cancel_requested`, which `cancel_run` of `@on.auto/definitions` records on the run's stream on any server. For a workflow it gives the run the input `cancel_requested`, with who asked, the kind and the reason, caused by the request; a run whose start has not reached the host yet takes nothing, and the consumer records the request it passed over in `workflow_pending_cancels`, one row a run, with the request's message id as `cause`, `cancelled_by`, `kind` and `reason` (`src/waiting/pending-cancel-rows.ts`), before it moves on. The host's start of that run reads the run's stream for a cancel asked since its start and gives it at once (`src/waiting/pending-cancels.ts`), so the run starts and ends cancelled. Since a host can die between applying the start and that read, the first resume after a host opens or takes the claim on the workflows reads that table alone, a hundred rows at a time, and gives each row's cancel to its run, four at a time: a row whose run took it, or had ended, is deleted, one whose run has not started stays for the start to meet, and a run that cannot take its cancel is reported without holding the others, and given it again at the next sweep. A row whose run answers that it has not started is deleted once the newest record of the run, read newest first without its data, is a finish, so a run that ended without ever reaching the host leaves no row after the next first resume; a run never handed to the host and never ended keeps its row, one a stranded run. The consumer takes no finish, so the finish of a run nothing waits on writes nothing. The first resume reads none of the runs still going, so it costs the same however many there are. A cancel recorded before the run's start, which only its caller records, names the run and no definition, and is no delivery: the start it precedes is refused. A `run_cancel_requested` that cannot be read as a request, as one whose context names no run, is noted as `record_unreadable` and passed over, as an ending the consumer of child endings cannot read is. For a run of another capability it hands the request to `waiting.cancelDeferred`, the capability's pure decision over the run's record and the settlement. A conflict because the run has already ended is a success for both consumers, which never skip a delivery: one that fails holds the record until it succeeds. A cancel of a call (`src/calls/call-cancels.ts`) handles each state of its row by name, and cancels the run the call waits for, by the id in `child`, before it marks the row: a `waiting` call is then marked `cancelled`; a `running` call is marked and its fiber interrupted, the run its arguments name being the one the function's `childOf` derived when it started; a `cancelled` call cancels its run again, which is harmless; an `answered` call is `already_answered`; and a call never seen leaves a tombstone. A cancel that fails, or whose run has an address its playbook cannot hold, fails its dispatch, which is made again with the row unchanged. The cancel the host records is a fact on the child's stream whatever state the child is in: on a stream with no run yet it is the first record, and the start that lands after it is refused as `cancelled`, so the ledger orders the cancel and the start and no child outlives the call that wanted it; on a run within its call it is recorded too, and the run its interrupt stops ends `rejected` as `cancelled` with the kind of the cancel rather than `failed`. A run cancelled by a call ends with the kind of the call's cancel, `deadline` or `parent_ended`, caused by the entry of the run's log that cancelled the call. diff --git a/packages/workflow-host/src/follower/consumers.ts b/packages/workflow-host/src/follower/consumers.ts index 59914033d..37ad6f058 100644 --- a/packages/workflow-host/src/follower/consumers.ts +++ b/packages/workflow-host/src/follower/consumers.ts @@ -28,7 +28,7 @@ interface Batch { readonly more: boolean; } -export interface RecordConsumer { +export interface RecordConsumer { readonly name: string; readonly skippedAfterSweeps: number; readonly batchOf: (followed: Followed, after: string | undefined, most: number) => Effect.Effect; @@ -38,20 +38,13 @@ export interface Consumer extends RecordConsumer { readonly types: readonly string[]; } -export interface CallConsumer extends RecordConsumer { - readonly types: readonly string[]; -} - export interface BoundConsumer { readonly name: string; readonly skippedAfterSweeps: number; readonly batchOf: (after: string | undefined, most: number) => Effect.Effect; } -export function boundTo( - consumer: RecordConsumer, - followed: Followed, -): BoundConsumer { +export function boundTo(consumer: RecordConsumer, followed: Followed): BoundConsumer { return { name: consumer.name, skippedAfterSweeps: consumer.skippedAfterSweeps, @@ -59,6 +52,28 @@ export function boundTo( }; } +export type Reading = BoundConsumer | 'unreadable'; + +export interface CallConsumer { + readonly types: readonly string[]; + readonly read: (call: CallRecord) => Reading; +} + +interface CallReader extends RecordConsumer { + readonly types: readonly string[]; + readonly readOf: (call: CallRecord) => Read | undefined; +} + +export function callConsumerOf(reader: CallReader): CallConsumer { + return { + types: reader.types, + read: (call) => { + const read = reader.readOf(call); + return read === undefined ? 'unreadable' : boundTo(reader, read); + }, + }; +} + export const deliveriesOfARecordInAPass = 100; export const deliverySweeps = 20; diff --git a/packages/workflow-host/src/follower/follower-records.test.ts b/packages/workflow-host/src/follower/follower-records.test.ts index 706f18f82..13dfda283 100644 --- a/packages/workflow-host/src/follower/follower-records.test.ts +++ b/packages/workflow-host/src/follower/follower-records.test.ts @@ -61,6 +61,44 @@ describe('a record of a playbook the follower cannot read', () => { }); }); +describe('a record the consumers of waiting calls cannot read', () => { + it('is said once and passed over, a cancel request whose context names no run as an ending whose context names none', async () => { + const reacting = await reactingHost(); + const { store } = reacting.database; + await definitionRecorded(store, { + name: 'watch', + version: 1, + triggers: [eventTrigger({ type: 'com.acme.sentinel' })], + }); + const stream = `${alpha}runs/r-bad`; + await recorded(store, stream, { type: 'run_cancel_requested', data: { kind: 'requested', reason: 'Gone' } }); + await recorded(store, stream, { type: 'run_succeeded', data: { output: null, record: {} } }); + + await published(store, { id: 's1', type: 'com.acme.sentinel' }); + await until( + () => Promise.resolve(reacting.reactions.starts()), + (starts) => starts.some(({ cause }) => cause === messageIdOf(`${alpha}events/s1`, 1)), + ); + + expect(reacting.notes()).toEqual([ + { + kind: 'record_unreadable', + org: 'acme', + playbook: 'alpha', + recordId: messageIdOf(stream, 1), + type: 'run_cancel_requested', + }, + { + kind: 'record_unreadable', + org: 'acme', + playbook: 'alpha', + recordId: messageIdOf(stream, 2), + type: 'run_succeeded', + }, + ]); + }); +}); + describe('a consumer the host is given', () => { it('is offered the events of the types it names, though nothing in the playbook reacts, and one naming none is offered nothing', async () => { const received: string[] = []; diff --git a/packages/workflow-host/src/follower/record-steps.test.ts b/packages/workflow-host/src/follower/record-steps.test.ts index 91cdcf2f5..a942d407a 100644 --- a/packages/workflow-host/src/follower/record-steps.test.ts +++ b/packages/workflow-host/src/follower/record-steps.test.ts @@ -3,7 +3,7 @@ import { recordingTracer } from '@on.auto/operations/testing'; import { Effect } from 'effect'; import { describe, expect, it } from 'vitest'; -import { deliverySweeps, type CallConsumer } from './consumers.ts'; +import { callConsumerOf, deliverySweeps } from './consumers.ts'; import { playbookOfKey, stepOf, type StepParts } from './record-steps.ts'; import type { GateVerdict } from './run-gate.ts'; @@ -58,17 +58,18 @@ describe('the step over the record of a run that passed listeners', () => { }); }); -const answering: CallConsumer = { +const answering = callConsumerOf({ name: 'call_answers', types: ['call_answered'], skippedAfterSweeps: deliverySweeps, + readOf: (call) => call, batchOf: () => Effect.succeed({ deliveries: [{ key: 'notify', workflow: 'close', deliver: Effect.void, skipped: () => Effect.void }], through: undefined, more: false, }), -}; +}); const answered: RecordedEvent = { ...record, diff --git a/packages/workflow-host/src/follower/record-steps.ts b/packages/workflow-host/src/follower/record-steps.ts index 96dd224df..2896bf636 100644 --- a/packages/workflow-host/src/follower/record-steps.ts +++ b/packages/workflow-host/src/follower/record-steps.ts @@ -8,7 +8,7 @@ import { deliverySweeps, type CallConsumer, type Consumer, - type FollowedRecord, + type Reading, type RecordConsumer, } from './consumers.ts'; import { deliveredAll, type Mode } from './delivery-loop.ts'; @@ -54,18 +54,8 @@ function passedOver(record: RecordedEvent): Progress { return { cursor: record.cursor, delivered: null, attempts: 0, waiting: false }; } -function followedOf( - parts: StepParts, - playbookKey: string, - relative: RecordedEvent, -): Effect.Effect { - const event = followedEventOf(relative, parts.definitionType); - if (event === 'unreadable') { - return Effect.as(parts.unreadable(playbookKey, relative), null); - } - return Effect.succeed( - event === 'none' ? null : { playbook: playbookOfKey(playbookKey), playbookKey, record: relative, event }, - ); +function unreadableStep(parts: StepParts, playbookKey: string, record: RecordedEvent): Effect.Effect { + return Effect.as(parts.unreadable(playbookKey, record), { progress: passedOver(record) }); } function runRecordStep(parts: StepParts, stepping: Stepping, progress: Progress, record: RecordedEvent) { @@ -85,31 +75,49 @@ function runRecordStep(parts: StepParts, stepping: Stepping, progress: Progress, }); } -function boundOf(parts: StepParts, followed: FollowedRecord | null) { - if (followed === null) { +function followedReadingsOf(parts: StepParts, playbookKey: string, relative: RecordedEvent): readonly Reading[] { + const event = followedEventOf(relative, parts.definitionType); + if (event === 'none') { return []; } - const wanting = parts.registered.filter(({ types }) => types.includes(followed.event.event.type)); + if (event === 'unreadable') { + return ['unreadable']; + } + const followed = { playbook: playbookOfKey(playbookKey), playbookKey, record: relative, event }; + const wanting = parts.registered.filter(({ types }) => types.includes(event.event.type)); return [...parts.consumers, ...wanting].map((consumer) => boundTo(consumer, followed)); } -function deliveredStep(parts: StepParts, stepping: Stepping, progress: Progress, record: RecordedEvent) { - return Effect.gen(function* () { - const { playbookKey } = stepping; - const relative = relativeRecord(playbookKey, record); - const calls = parts.calls - .filter(({ types }) => types.includes(record.type)) - .map((consumer) => boundTo(consumer, { playbook: playbookOfKey(playbookKey), playbookKey, record: relative })); - const consumers = [...calls, ...boundOf(parts, yield* followedOf(parts, playbookKey, relative))]; +function readingsOf(parts: StepParts, playbookKey: string, record: RecordedEvent): readonly Reading[] { + const relative = relativeRecord(playbookKey, record); + const call = { playbook: playbookOfKey(playbookKey), playbookKey, record: relative }; + const calls = parts.calls.filter(({ types }) => types.includes(record.type)).map((consumer) => consumer.read(call)); + return [...calls, ...followedReadingsOf(parts, playbookKey, relative)]; +} + +function deliveredStep( + parts: StepParts, + stepping: Stepping, + progress: Progress, + record: RecordedEvent, +): Effect.Effect { + return Effect.suspend(() => { + const readings = readingsOf(parts, stepping.playbookKey, record); + const consumers = readings.filter((reading) => reading !== 'unreadable'); + if (consumers.length < readings.length) { + return unreadableStep(parts, stepping.playbookKey, record); + } if (consumers.length === 0) { - return { progress: passedOver(record) }; + return Effect.succeed({ progress: passedOver(record) }); } - const { made, ...delivered } = yield* deliveredAll(consumers, progress, stepping.mode).pipe( + return deliveredAll(consumers, progress, stepping.mode).pipe( Effect.withSpan(`follow ${record.type}`, followSpanOf(record)), + Effect.map(({ made, ...delivered }) => + delivered.end === undefined + ? { progress: passedOver(record), delivered: made } + : { ...delivered, delivered: made }, + ), ); - return delivered.end === undefined - ? { progress: passedOver(record), delivered: made } - : { ...delivered, delivered: made }; }); } @@ -130,7 +138,7 @@ export function stepOf( const readAgain: Effect.Effect = Effect.succeed({ progress, end: 'more' }); return Effect.flatMap(parts.applyDefinitionRecord(playbookKey, record), (applied) => applied === 'unreadable' - ? Effect.as(parts.unreadable(playbookKey, record), { progress: passedOver(record) }) + ? unreadableStep(parts, playbookKey, record) : Effect.flatMap(stepping.wantsMore(), (more) => (more ? readAgain : step)), ); } diff --git a/packages/workflow-host/src/waiting/cancel-requests.test.ts b/packages/workflow-host/src/waiting/cancel-requests.test.ts index c00681403..cbd0a5997 100644 --- a/packages/workflow-host/src/waiting/cancel-requests.test.ts +++ b/packages/workflow-host/src/waiting/cancel-requests.test.ts @@ -137,10 +137,18 @@ async function failingConsumer() { const endingOfAnotherCapability = { type: 'run_succeeded', data: { output: null, record: {} } }; +function readBy(consumer: ReturnType, fact: Fact, context?: Context) { + const reading = consumer.read(asked(fact, context)); + if (reading === 'unreadable') { + throw new Error(`The consumer of cancel requests could not read ${fact.type}`); + } + return reading; +} + describe('the consumer of cancel requests', () => { it('never skips a request, and fails a delivery the run cannot take now, to be made again', async () => { - const consumer = await failingConsumer(); - const { deliveries } = await Effect.runPromise(consumer.batchOf(asked(askedToCancel), undefined, 100)); + const consumer = readBy(await failingConsumer(), askedToCancel); + const { deliveries } = await Effect.runPromise(consumer.batchOf(undefined, 100)); const failure = await Effect.runPromise(Effect.flip(Effect.forEach(deliveries, ({ deliver }) => deliver))); const skipped = Effect.forEach(deliveries, (delivery) => delivery.skipped('')); @@ -148,23 +156,24 @@ describe('the consumer of cancel requests', () => { expect(await Effect.runPromise(Effect.as(skipped, 'nothing skipped'))).toBe('nothing skipped'); }); - it('makes no delivery of a cancel before a start, nor of a record that is no request or ends another capability’s run', async () => { + it('makes no delivery of a cancel its caller recorded before the run’s start, nor again of one it made', async () => { const consumer = await failingConsumer(); - const beforeTheStart = asked( - { type: 'run_cancel_requested', data: { kind: 'requested', reason: 'Gone' } }, - { by: 'playbook:alpha', at }, - ); + const beforeTheStart = readBy(consumer, askedToCancel, { by: 'playbook:alpha', at, runId }); + const made = readBy(consumer, askedToCancel); const batches = await Effect.runPromise( - Effect.all([ - consumer.batchOf(beforeTheStart, undefined, 100), - consumer.batchOf(asked({ type: 'event_published', data: {} }), undefined, 100), - ]), - ); - const ending = await Effect.runPromise( - consumer.batchOf(asked(endingOfAnotherCapability, runOf('interaction', 'playbook:alpha')), undefined, 100), + Effect.all([beforeTheStart.batchOf(undefined, 100), made.batchOf('cancel', 100)]), ); - const resumed = await Effect.runPromise(consumer.batchOf(asked(askedToCancel), 'cancel', 100)); - expect([...batches, ending, resumed].map(({ deliveries }) => deliveries)).toEqual([[], [], [], []]); + expect(batches.map(({ deliveries }) => deliveries)).toEqual([[], []]); + }); + + it('cannot read a request whose context names no run, nor a record that is no request', async () => { + const consumer = await failingConsumer(); + + expect([ + consumer.read(asked(askedToCancel, { by: 'playbook:alpha', at })), + consumer.read(asked({ type: 'event_published', data: {} })), + consumer.read(asked(endingOfAnotherCapability, runOf('interaction', 'playbook:alpha'))), + ]).toEqual(['unreadable', 'unreadable', 'unreadable']); }); }); diff --git a/packages/workflow-host/src/waiting/cancel-requests.ts b/packages/workflow-host/src/waiting/cancel-requests.ts index 932fec77d..20afba9b7 100644 --- a/packages/workflow-host/src/waiting/cancel-requests.ts +++ b/packages/workflow-host/src/waiting/cancel-requests.ts @@ -4,7 +4,13 @@ import type { RunInput, Submission } from '@on.auto/workflow-engine'; import { Effect } from 'effect'; import type { HostDatabase } from '../database/host-database.ts'; -import { DeliveryFailed, type CallConsumer, type CallRecord, type Delivery } from '../follower/consumers.ts'; +import { + callConsumerOf, + DeliveryFailed, + type CallConsumer, + type CallRecord, + type Delivery, +} from '../follower/consumers.ts'; import { runKeyOf } from '../runs/run-address.ts'; import { passedOverRow } from './pending-cancel-rows.ts'; import type { WaitingOptions } from './waiting-options.ts'; @@ -21,6 +27,9 @@ interface Asked { readonly playbook: PlaybookAddress; readonly record: RecordedEvent; readonly request: CancelRequested; +} + +interface AskedOfARun extends Asked { readonly type: string; readonly runId: string; } @@ -33,7 +42,7 @@ function lineageOf({ id, correlationId }: RecordedEvent): Lineage { return { causationId: id, correlationId }; } -function cancelledWorkflow(parts: CancelParts, { playbook, record, request, runId }: Asked) { +function cancelledWorkflow(parts: CancelParts, { playbook, record, request, runId }: AskedOfARun) { const { kind, reason } = request.data; const { by } = request.context; const runKey = runKeyOf({ ...playbook, runId }); @@ -44,7 +53,7 @@ function cancelledWorkflow(parts: CancelParts, { playbook, record, request, runI ); } -function cancelled(parts: CancelParts, asked: Asked): Effect.Effect { +function cancelled(parts: CancelParts, asked: AskedOfARun): Effect.Effect { const { playbook, record, request, type, runId } = asked; if (type === parts.workflows) { return cancelledWorkflow(parts, asked); @@ -56,32 +65,38 @@ function cancelled(parts: CancelParts, asked: Asked): Effect.Effect Effect.void, }; return [delivery]; } export function cancelRequests(parts: CancelParts): CallConsumer { - return { + return callConsumerOf({ name: 'cancel_requests', types: ['run_cancel_requested'], skippedAfterSweeps: Number.POSITIVE_INFINITY, - batchOf: (followed, after) => + readOf: askedOf, + batchOf: (asked, after) => Effect.sync(() => ({ - deliveries: after === undefined ? deliveriesOf(parts, followed) : [], + deliveries: after === undefined ? deliveriesOf(parts, asked) : [], through: undefined, more: false, })), - }; + }); } diff --git a/packages/workflow-host/src/waiting/child-endings.test.ts b/packages/workflow-host/src/waiting/child-endings.test.ts index e72fb6999..f24a9f419 100644 --- a/packages/workflow-host/src/waiting/child-endings.test.ts +++ b/packages/workflow-host/src/waiting/child-endings.test.ts @@ -2,6 +2,7 @@ import { Conflict } from '@on.auto/operations'; import { Effect } from 'effect'; import { describe, expect, it } from 'vitest'; +import type { BoundConsumer, CallConsumer } from '../follower/consumers.ts'; import { alpha, at, recorded } from '../reaction-testing/playbook-writes.ts'; import { faultyDatabase } from '../testing/faulty-database.ts'; import { until } from '../testing/until.ts'; @@ -78,20 +79,30 @@ describe('the ending of a run that answers a call', () => { }); }); +function endingReadBy(consumer: CallConsumer): BoundConsumer { + const reading = consumer.read(endingRecord); + if (reading === 'unreadable') { + throw new Error('The consumer of child endings could not read the ending'); + } + return reading; +} + describe('a delivery of an ending that the run cannot take now', () => { it('fails, to be made again, by a consumer that never skips a delivery', async () => { const { database } = await waitingParent(child); - const consumer = childEndings({ - database, - submitted: () => Effect.fail(new Conflict({ detail: 'The log of the run kept changing' })), - resultOf: resultOfEnding, - now: () => 1, - }); + const consumer = endingReadBy( + childEndings({ + database, + submitted: () => Effect.fail(new Conflict({ detail: 'The log of the run kept changing' })), + resultOf: resultOfEnding, + now: () => 1, + }), + ); - const { deliveries } = await Effect.runPromise(consumer.batchOf(endingRecord, undefined, 100)); + const { deliveries } = await Effect.runPromise(consumer.batchOf(undefined, 100)); const failure = await Effect.runPromise(Effect.flip(Effect.forEach(deliveries, ({ deliver }) => deliver))); const skipped = Effect.forEach(deliveries, (delivery) => delivery.skipped('')); - const resumed = await Effect.runPromise(consumer.batchOf(endingRecord, 'call', 100)); + const resumed = await Effect.runPromise(consumer.batchOf('call', 100)); expect([failure.detail, consumer.skippedAfterSweeps]).toEqual(['The log of the run kept changing', Infinity]); expect(resumed.deliveries).toEqual([]); diff --git a/packages/workflow-host/src/waiting/child-endings.ts b/packages/workflow-host/src/waiting/child-endings.ts index fab96f003..e77a22922 100644 --- a/packages/workflow-host/src/waiting/child-endings.ts +++ b/packages/workflow-host/src/waiting/child-endings.ts @@ -11,7 +11,7 @@ import { Effect } from 'effect'; import { answeredRow, callRowOf, deliveredRow, waitingCallsOf } from '../calls/call-rows.ts'; import type { HostDatabase } from '../database/host-database.ts'; -import { DeliveryFailed, type CallConsumer } from '../follower/consumers.ts'; +import { callConsumerOf, DeliveryFailed, type CallConsumer, type CallRecord } from '../follower/consumers.ts'; import { addressOfRun, runKeyOf } from '../runs/run-address.ts'; import type { WaitingOptions } from './waiting-options.ts'; @@ -27,6 +27,11 @@ interface CalledEnding { readonly calledBy: CalledBy; } +interface Ended { + readonly playbook: PlaybookAddress; + readonly ending: RunEnding; +} + function calledEndingOf(ending: RunEnding | undefined): CalledEnding | undefined { const calledBy = ending?.context.calledBy; return ending === undefined || calledBy === undefined ? undefined : { ending, calledBy }; @@ -72,14 +77,20 @@ export function childAnswersOn( }; } +function endedOf({ playbook, record }: CallRecord): Ended | undefined { + const ending = runEndingOf(record); + return ending === undefined ? undefined : { playbook, ending }; +} + export function childEndings(parts: EndingParts): CallConsumer { - return { + return callConsumerOf({ name: 'child_endings', types: ['run_succeeded', 'run_rejected', 'run_failed'], skippedAfterSweeps: Number.POSITIVE_INFINITY, - batchOf: ({ playbook, record }, after) => + readOf: endedOf, + batchOf: ({ playbook, ending }, after) => Effect.sync(() => { - const called = after === undefined ? calledEndingOf(runEndingOf(record)) : undefined; + const called = after === undefined ? calledEndingOf(ending) : undefined; return { deliveries: called === undefined @@ -96,7 +107,7 @@ export function childEndings(parts: EndingParts): CallConsumer { more: false, }; }), - }; + }); } export function endedChildrenOn(parts: EndingParts): (runKey: string) => Effect.Effect { From 868015b7fffa01441b68d91a8ea6b0c8a67cd61e Mon Sep 17 00:00:00 2001 From: Rami Date: Fri, 23 Oct 2026 13:53:12 +0100 Subject: [PATCH 2/2] fix(workflow-host): a playbook first followed starts where the host opened, so it misses nothing At a host's first start the follower followed a playbook it had never followed from its latest record, read after the host opened, so an ending or a cancel appended in between was passed over and its call was answered only at its deadline. The host now takes its place in the ledger as it opens, before it claims the workflows, and the follower follows such a playbook from that place. Co-Authored-By: Claude Fable 5.1 --- packages/workflow-host/README.md | 8 +- .../workflow-host/measure/finish-catch-up.ts | 11 ++- .../follower/followed-from-the-open.test.ts | 82 +++++++++++++++++++ .../src/follower/follower-assembly.ts | 2 +- .../src/follower/follower-host.ts | 2 + .../src/follower/playbook-discovery.test.ts | 2 +- .../src/follower/playbook-discovery.ts | 19 +++-- .../src/follower/playbook-records.ts | 5 -- .../workflow-host/src/host/host-serving.ts | 8 +- .../workflow-host/src/host/workflow-host.ts | 5 +- .../src/reaction-testing/follower-over.ts | 4 +- .../src/reactions/reaction-consumers.test.ts | 1 - .../src/reactions/reaction-consumers.ts | 2 +- .../src/sweeps/registry-retries.test.ts | 2 +- .../src/waiting-testing/held-follower.ts | 46 +++++++++++ 15 files changed, 165 insertions(+), 34 deletions(-) create mode 100644 packages/workflow-host/src/follower/followed-from-the-open.test.ts create mode 100644 packages/workflow-host/src/waiting-testing/held-follower.ts diff --git a/packages/workflow-host/README.md b/packages/workflow-host/README.md index 743202002..afd1f9ac8 100644 --- a/packages/workflow-host/README.md +++ b/packages/workflow-host/README.md @@ -4,7 +4,7 @@ The workflow engine of [`@on.auto/workflow-engine`](../workflow-engine) on Node: ## Entry -`@on.auto/workflow-host` (`src/index.ts`) exports `openWorkflowStore(settings, lostConnection, folders)`, which opens the host's database and migrates its tables and the tables of the folders it is given, answering a `WorkflowStore` with the database, `viewsPortOn(database)`, the `ViewsPort` of [the views of recall functions](#the-views-of-recall-functions) on that database, and `openWorkflowHost(options)`, which opens the host's database, or takes the store it is given, migrates its tables, claims the database's workflows (see [One host for a database](#one-host-for-a-database)), starts its loop if it holds them, and answers with a `WorkflowHost`: +`@on.auto/workflow-host` (`src/index.ts`) exports `openWorkflowStore(settings, lostConnection, folders)`, which opens the host's database and migrates its tables and the tables of the folders it is given, answering a `WorkflowStore` with the database, `viewsPortOn(database)`, the `ViewsPort` of [the views of recall functions](#the-views-of-recall-functions) on that database, and `openWorkflowHost(options)`, which opens the host's database, or takes the store it is given, migrates its tables, takes its place in the ledger (see [The follower](#the-follower)), claims the database's workflows (see [One host for a database](#one-host-for-a-database)), starts its loop if it holds them, and answers with a `WorkflowHost`: - `start(run, start)`: starts the run's log with the `started` input, and answers `started`; `going` when the run started before and is not settled, so nothing starts again, after trying again at once the settlement of a run that ended; or `settled` when the run ended and was settled, since a run never starts twice in one log. - `deliver(run, event)`: gives the run an `event_received` input, and answers `delivered`, also for an event the run took before; `not_started` for a run whose `started` has not arrived; or `ended` for a run that ended. @@ -117,9 +117,9 @@ A workflow with an event trigger is started for every event of its playbook that ### The follower -The follower keeps a place in every playbook: at its start it reads every registry of playbooks of the orgs (`org//playbooks`) and, the first time it sees a registry, follows each playbook it names from its latest record and then makes the playbook's triggers again from its definitions; after the start, a sweep, like a signal, follows from its first record any playbook one of whose streams is appended to, before it passes the playbook, and a registry is read, by its name, only when an append to it is signalled or swept, which follows from its first record every playbook the registry names that is not followed yet, even one with records in the ledger already. A playbook is passed when an append to one of its streams other than a run's log raises the ledger's signal (`streamAppends` of `@on.auto/ledger`, raised by the event store in this process), since a run's records matter only once an event follows them, and a host whose runs append thousands of records a second would otherwise pass its playbooks as often. +The follower keeps a place in every playbook: at its start it reads every registry of playbooks of the orgs (`org//playbooks`) and, the first time it sees a registry, follows each playbook it names from the place the host took in the ledger as it opened, before it claimed the workflows or started anything, the point the ledger's `readAppended` answers without a point, and then makes the playbook's triggers again from its definitions as they stand when it reads them, which its passes then read again from that place on, in order; so a record appended after the host opened is read, an ending or a cancel among them, however soon after the open it came and however late the follower took its place, and none of the records from before it, which no follower passed, is delivered; after the start, a sweep, like a signal, follows from its first record any playbook one of whose streams is appended to, before it passes the playbook, and a registry is read, by its name, only when an append to it is signalled or swept, which follows from its first record every playbook the registry names that is not followed yet, even one with records in the ledger already. A playbook is passed when an append to one of its streams other than a run's log raises the ledger's signal (`streamAppends` of `@on.auto/ledger`, raised by the event store in this process), since a run's records matter only once an event follows them, and a host whose runs append thousands of records a second would otherwise pass its playbooks as often. -Every `sweepEveryMs` the follower sweeps, so that a playbook written by another process, which raises no signal here, is caught up too. A sweep reads which streams were appended to since the last, through the ledger's `readAppended`, from the point where the last read stopped, at most 10,000 records at a time; it reads the registries among them, naming a registry again at every sweep until a read of it succeeded, while a read that fails is said and the sweep's upkeep and passes go on, and passes up to 128 playbooks: up to 64 of those whose deliveries or run records wait, those a sweep took longest ago first, and in the places left the playbooks appended to by anything but a run's log, so that playbooks that wait never take every place. Playbooks it has no place for, and records past the 10,000, are taken by later sweeps: the next comes at once while the read stopped at its 10,000 records or playbooks appended to are still left over, and otherwise after a whole `sweepEveryMs`, so playbooks that wait past the 64, the rest of the round after a start, and failed playbooks handed back that do not fit, wait for it. A pass that fails is said, the playbooks after it in the sweep are passed all the same, and its playbook is handed back to the sweeps, which queue it in order with the playbooks appended to without hurrying the next sweep: at least 64 places of every sweep go to that queue, so a playbook with n playbooks queued before it is passed within ⌈(n + 1) / 64⌉ of the sweeps that reach their passes, however many are appended to after it. A sweep that fails before its passes hands back every playbook it took, and takes nothing out of what is queued until it has read the round. At its start the follower takes its place in the ledger before it reads the registries, and its sweeps go once over every playbook it follows, in the places left, since nothing read the ledger while the host was down. A sweep that finds nothing new reads the ledger once and the playbooks that wait once, and passes no playbook. +Every `sweepEveryMs` the follower sweeps, so that a playbook written by another process, which raises no signal here, is caught up too. A sweep reads which streams were appended to since the last, through the ledger's `readAppended`, from the point where the last read stopped, at most 10,000 records at a time; it reads the registries among them, naming a registry again at every sweep until a read of it succeeded, while a read that fails is said and the sweep's upkeep and passes go on, and passes up to 128 playbooks: up to 64 of those whose deliveries or run records wait, those a sweep took longest ago first, and in the places left the playbooks appended to by anything but a run's log, so that playbooks that wait never take every place. Playbooks it has no place for, and records past the 10,000, are taken by later sweeps: the next comes at once while the read stopped at its 10,000 records or playbooks appended to are still left over, and otherwise after a whole `sweepEveryMs`, so playbooks that wait past the 64, the rest of the round after a start, and failed playbooks handed back that do not fit, wait for it. A pass that fails is said, the playbooks after it in the sweep are passed all the same, and its playbook is handed back to the sweeps, which queue it in order with the playbooks appended to without hurrying the next sweep: at least 64 places of every sweep go to that queue, so a playbook with n playbooks queued before it is passed within ⌈(n + 1) / 64⌉ of the sweeps that reach their passes, however many are appended to after it. A sweep that fails before its passes hands back every playbook it took, and takes nothing out of what is queued until it has read the round. At its start the follower takes its own place in the ledger, from which its sweeps read the streams appended to, before it reads the registries, and its sweeps go once over every playbook it follows, in the places left, since nothing read the ledger while the host was down. A sweep that finds nothing new reads the ledger once and the playbooks that wait once, and passes no playbook. A pass reads the playbook's records oldest first through the ledger's recorded read, 100 records a page and at most 10 pages, and keeps its place in `workflow_followed_playbooks` after each record it made a delivery of and once more when the pass ends, so a pass that fails makes again only the deliveries of the record it was at, and reads again the records it passed over since its last delivery. It first glances at the records without their data, the definitions' records apart, and reads them with their data only for the event types that the playbook's triggers and listeners and the consumers its caller gave name (`workflow_subscription_types` and `workflow_listener_types`): a fact is read with its data when its type is named, and every published event when any type that is not a fact's is named, since an event's type lies in its data. A record whose data the pass did not read is passed over, so a playbook where nothing names a type costs one read of identifiers a wake; a record that names more types ends the pass: a run's record that armed a listener is passed first, and a definition that activates a trigger is read again by the next pass, which reads it and the records after it with the data of those types too. A record of a run's log is passed only once the dispatch watermark of the run covers its version, so the listeners a record armed are kept by the time the follower reaches it, and passing that record marks them as passed: a listener takes only the events recorded after the record that armed it. A record the watermark has not covered by the twentieth sweep that met it is passed all the same, as a delivery is skipped, so a run whose dispatch is stuck does not hold back its playbook: the host notes `run_record_passed`, keeps in `workflow_passed_runs` how far it passed the run's log, and then marks as passed every listener of the run armed by that record or before; a listener's insert, for its part, marks the listener passed afterwards when that row is there, so each side writes before it reads, and a listener the dispatch keeps while the record is passed or later is marked, and takes the events recorded from then on. The row is written only while the run is going and its dispatch is behind the record, taken back at once if the dispatch reached it meanwhile, and deleted once the run's dispatch reaches that far or the run ends. A record the follower cannot read is noted once (`record_unreadable`) and passed over with nothing delivered of it, by the one step that passes over every such record (`src/follower/record-steps.ts`): an event or a fact the consumers would be handed that does not decode, a definition record of the workflows that cannot be decoded, and a record one of the consumers of [waiting calls](#waiting-calls) cannot read as what it takes, an ending of a run or a cancel request. @@ -149,7 +149,7 @@ The engine's `arm_listener` output inserts the run's listener with its filters a ### Triggers -A record of a workflow's definition carries the version's `triggers`, which the workflow's adapter parsed once, at save, and `definitionChangeOf` of `@on.auto/definitions` reads from the record; the host reads no document. A trigger is identified by its kind, `event`, `cron` or `every`, and its reference in the document, `/schedule/on`, `/schedule/cron` or `/schedule/every`, so a run names its trigger exactly and two versions compare their triggers like with like. Each trigger has a row of `workflow_subscriptions`, keyed by the playbook, the workflow and the reference, with the version it starts, its kind, its rule, the id and time of the record that activated it as it is, and, for a schedule, its next due time and its running run. When the follower passes a definition record, in the ledger's order, or replays the definition records of a playbook it begins to follow from its latest record (`src/triggers/definition-records.ts`), it compares the version's triggers with the workflow's rows by reference and rule (`src/triggers/trigger-changes.ts`): +A record of a workflow's definition carries the version's `triggers`, which the workflow's adapter parsed once, at save, and `definitionChangeOf` of `@on.auto/definitions` reads from the record; the host reads no document. A trigger is identified by its kind, `event`, `cron` or `every`, and its reference in the document, `/schedule/on`, `/schedule/cron` or `/schedule/every`, so a run names its trigger exactly and two versions compare their triggers like with like. Each trigger has a row of `workflow_subscriptions`, keyed by the playbook, the workflow and the reference, with the version it starts, its kind, its rule, the id and time of the record that activated it as it is, and, for a schedule, its next due time and its running run. When the follower passes a definition record, in the ledger's order, or replays the definition records of a playbook it begins to follow from the place the host opened at (`src/triggers/definition-records.ts`), it compares the version's triggers with the workflow's rows by reference and rule (`src/triggers/trigger-changes.ts`): - an unchanged trigger keeps its row with the version moved on, so its activation, its anchor, its next due time and its running run survive a save of the body; - a changed schedule moves its activation, its anchor and its next due time to that record and keeps its running run, so one run at a time holds whichever version started that run; diff --git a/packages/workflow-host/measure/finish-catch-up.ts b/packages/workflow-host/measure/finish-catch-up.ts index 4bcdbc5fc..5cdaaf276 100644 --- a/packages/workflow-host/measure/finish-catch-up.ts +++ b/packages/workflow-host/measure/finish-catch-up.ts @@ -45,11 +45,10 @@ async function latestOf(opened: Opened): Promise { return records[0]?.cursor ?? null; } -async function untilAt(opened: Opened, cursor: string | null): Promise { - const [place] = await placesOf(opened); - if (place !== cursor) { +async function untilPlaced(opened: Opened, reached: (places: readonly (string | null)[]) => boolean): Promise { + if (!reached(await placesOf(opened))) { await setTimeout(2); - await untilAt(opened, cursor); + await untilPlaced(opened, reached); } } @@ -65,13 +64,13 @@ export async function finishCatchUpOn(database: DatabaseSettings, finishes: numb const opened = await openHostDatabase(database, Function.constVoid); await playbookCreated(opened.store, 'alpha'); const following = await measuredHost(database); - await untilAt(opened, await latestOf(opened)); + await untilPlaced(opened, (places) => places.length > 0); await following.host.stop(); await finishedInTurn(opened, finishes); const latest = await latestOf(opened); const started = performance.now(); const catchingUp = await measuredHost(database); - await untilAt(opened, latest); + await untilPlaced(opened, ([place]) => place === latest); const milliseconds = performance.now() - started; await catchingUp.host.stop(); await opened.close(); diff --git a/packages/workflow-host/src/follower/followed-from-the-open.test.ts b/packages/workflow-host/src/follower/followed-from-the-open.test.ts new file mode 100644 index 000000000..555bda6a3 --- /dev/null +++ b/packages/workflow-host/src/follower/followed-from-the-open.test.ts @@ -0,0 +1,82 @@ +import { Effect } from 'effect'; +import { describe, expect, it } from 'vitest'; + +import { waitingCallsOf } from '../calls/call-rows.ts'; +import { alpha, at, recorded } from '../reaction-testing/playbook-writes.ts'; +import { runAt, startOf, workflow } from '../testing/host-documents.ts'; +import { until } from '../testing/until.ts'; +import { heldFollower } from '../waiting-testing/held-follower.ts'; +import { parentId, parentRun } from '../waiting-testing/waiting-parent.ts'; + +const child = '0199a3c4-7d2e-7c1a-9b3f-0000000000c1'; + +const asking = workflow('do:\n - ask: { call: notify, with: { to: ada } }'); + +const pausing = workflow('do:\n - pause: { wait: PT1H }'); + +const ofTheChild = { + by: 'playbook:alpha', + at, + runId: child, + definitionType: 'workflow', + definitionName: 'check', + definitionVersion: 1, + calledBy: { runId: parentId, reference: '/do/0/ask', run: 1 }, +}; + +const ofTheRun = { + by: 'acme-admin', + at, + runId: parentId, + definitionType: 'workflow', + definitionName: 'pause', + definitionVersion: 1, +}; + +describe('the ending of a run a call waits for, appended after the host opened and before its follower first passed the playbook', () => { + it('answers the call as the follower first passes the playbook from where the host opened, not at its deadline', async () => { + const { database, hosted, settled, release } = await heldFollower({ + answer: () => Effect.succeed({ status: 'waiting', child }), + }); + hosted.know(parentId); + await Effect.runPromise(hosted.host.start(runAt(parentId), startOf(asking))); + await until( + () => Effect.runPromise(waitingCallsOf(database, parentRun)), + (waiting) => waiting.length === 1, + ); + await recorded( + database.store, + `${alpha}runs/${child}`, + { type: 'run_succeeded', data: { output: 'checked', record: {} } }, + ofTheChild, + ); + + release(); + + expect(await settled(parentId)).toEqual({ status: 'succeeded', output: 'checked' }); + }); +}); + +describe('a cancel request appended after the host opened and before its follower first passed the playbook', () => { + it('ends its run cancelled as the follower first passes the playbook from where the host opened', async () => { + const { database, hosted, settled, release } = await heldFollower(); + hosted.know(parentId); + await Effect.runPromise(hosted.host.start(runAt(parentId), startOf(pausing))); + await recorded( + database.store, + `${alpha}runs/${parentId}`, + { type: 'run_cancel_requested', data: { kind: 'requested', reason: 'Not needed any more' } }, + ofTheRun, + ); + + release(); + + expect(await settled(parentId)).toEqual({ + status: 'rejected', + reason: 'cancelled', + kind: 'requested', + detail: 'Not needed any more', + by: 'acme-admin', + }); + }); +}); diff --git a/packages/workflow-host/src/follower/follower-assembly.ts b/packages/workflow-host/src/follower/follower-assembly.ts index adcc3e99d..5c448718c 100644 --- a/packages/workflow-host/src/follower/follower-assembly.ts +++ b/packages/workflow-host/src/follower/follower-assembly.ts @@ -44,7 +44,7 @@ export function followerOn(host: FollowerHost, assembly: FollowerAssembly): Foll const discovery = playbookDiscoveryOn({ database, playbooks, - records: assembly.records, + placeAtOpen: host.placeAtOpen, applyDefinitionRecord: reacting.applyDefinitionRecord, unreadable, definitionType: assembly.options.definitionType, diff --git a/packages/workflow-host/src/follower/follower-host.ts b/packages/workflow-host/src/follower/follower-host.ts index a9dbf7ce7..74677ebe5 100644 --- a/packages/workflow-host/src/follower/follower-host.ts +++ b/packages/workflow-host/src/follower/follower-host.ts @@ -1,3 +1,4 @@ +import type { RecordedPoint } from '@on.auto/ledger'; import type { Conflict } from '@on.auto/operations'; import type { RunInput, Submission } from '@on.auto/workflow-engine'; import type { Effect } from 'effect'; @@ -14,4 +15,5 @@ export interface FollowerHost { readonly sweepEveryMs: number; readonly reports: HostReports; readonly context?: HostContext; + readonly placeAtOpen: RecordedPoint; } diff --git a/packages/workflow-host/src/follower/playbook-discovery.test.ts b/packages/workflow-host/src/follower/playbook-discovery.test.ts index f12a463a6..39f7e1e64 100644 --- a/packages/workflow-host/src/follower/playbook-discovery.test.ts +++ b/packages/workflow-host/src/follower/playbook-discovery.test.ts @@ -31,7 +31,7 @@ function startsReaching(reacting: ReactingHost, count: number) { } describe('a playbook its org created before the host started', () => { - it('is followed from its latest record, with the triggers its workflows already had', async () => { + it('is followed from where the ledger stood as the host opened, with the triggers its workflows already had', async () => { const settings = await onSQLite(); const { store } = await openedOn(settings); await playbookCreated(store, 'alpha'); diff --git a/packages/workflow-host/src/follower/playbook-discovery.ts b/packages/workflow-host/src/follower/playbook-discovery.ts index 59721d73a..5186efc11 100644 --- a/packages/workflow-host/src/follower/playbook-discovery.ts +++ b/packages/workflow-host/src/follower/playbook-discovery.ts @@ -1,3 +1,4 @@ +import { recordedCursorOf, type RecordedPoint } from '@on.auto/ledger'; import { streamPrefixOfPlaybook, type PlaybookAddress } from '@on.auto/operations'; import { playbookCreatedOf, playbooksStreamOfOrg } from '@on.auto/playbooks'; import { Effect, Schema } from 'effect'; @@ -11,7 +12,6 @@ import { } from '../triggers/definition-records.ts'; import type { FollowedPlaybooks } from './followed-playbooks.ts'; import { scannedListeners } from './listener-scan.ts'; -import type { PlaybookRecords } from './playbook-records.ts'; export interface Discovery { readonly atStart: () => Effect.Effect; @@ -22,7 +22,7 @@ export interface Discovery { export interface DiscoveryParts { readonly database: HostDatabase; readonly playbooks: FollowedPlaybooks; - readonly records: PlaybookRecords; + readonly placeAtOpen: RecordedPoint; readonly applyDefinitionRecord: ApplyDefinitionRecord; readonly unreadable: (playbookKey: string, record: DefinitionRecord) => Effect.Effect; readonly definitionType: string; @@ -78,10 +78,10 @@ function seenOf(database: HostDatabase, stream: string): Effect.Effect rows.at(0)?.position)); } -function followedAtTailOn({ +function followedFromTheOpenOn({ database, playbooks, - records, + placeAtOpen, applyDefinitionRecord, unreadable, definitionType, @@ -93,7 +93,7 @@ function followedAtTailOn({ return (playbook: PlaybookAddress) => Effect.gen(function* () { const playbookKey = streamPrefixOfPlaybook(playbook); - yield* playbooks.follow(playbookKey, yield* records.tail(playbook)); + yield* playbooks.follow(playbookKey, recordedCursorOf(playbookKey, placeAtOpen)); const recorded = yield* Effect.promise(() => database.store.read(`${playbookKey}definitions/${definitionType}`, 0), ); @@ -101,7 +101,10 @@ function followedAtTailOn({ }); } -function registryReadOn({ database, playbooks }: DiscoveryParts, followedAtTail: ReturnType) { +function registryReadOn( + { database, playbooks }: DiscoveryParts, + followedFromTheOpen: ReturnType, +) { return ({ stream, org }: Registry, after: number | undefined, atStart: boolean) => Effect.gen(function* () { const { messages, version } = yield* Effect.promise(() => database.store.read(stream, after ?? 0)); @@ -116,7 +119,7 @@ function registryReadOn({ database, playbooks }: DiscoveryParts, followedAtTail: created, (playbook) => atStart && after === undefined - ? followedAtTail(playbook) + ? followedFromTheOpen(playbook) : playbooks.follow(streamPrefixOfPlaybook(playbook), null), { discard: true }, ); @@ -131,7 +134,7 @@ function registryReadOn({ database, playbooks }: DiscoveryParts, followedAtTail: export function playbookDiscoveryOn(parts: DiscoveryParts): Discovery { const { database, playbooks } = parts; - const registryRead = registryReadOn(parts, followedAtTailOn(parts)); + const registryRead = registryReadOn(parts, followedFromTheOpenOn(parts)); const everyRegistry = Effect.gen(function* () { const [streams, seen] = yield* orgsOf(database); const seenAt = new Map(seen.map(({ stream_id: stream, position }) => [stream, position])); diff --git a/packages/workflow-host/src/follower/playbook-records.ts b/packages/workflow-host/src/follower/playbook-records.ts index 7baba9b5d..8eade2dab 100644 --- a/packages/workflow-host/src/follower/playbook-records.ts +++ b/packages/workflow-host/src/follower/playbook-records.ts @@ -27,7 +27,6 @@ export interface PlaybookRecords { cursor: string | null, delivers: ReadonlySet, ) => Effect.Effect; - readonly tail: (playbook: PlaybookAddress) => Effect.Effect; } export function relativeRecord(playbookKey: string, record: RecordedEvent): RecordedEvent { @@ -50,9 +49,5 @@ export function playbookRecordsOf(store: EventStore): PlaybookRecords { }, ), ), - tail: (playbook) => - Effect.orDie(read(playbook, { kind: 'everything' }, { order: 'desc', limit: 1, dataOf: [] })).pipe( - Effect.map(({ records }) => records.reduce((_, { cursor }) => cursor, null)), - ), }; } diff --git a/packages/workflow-host/src/host/host-serving.ts b/packages/workflow-host/src/host/host-serving.ts index 6339b1452..075f24cc3 100644 --- a/packages/workflow-host/src/host/host-serving.ts +++ b/packages/workflow-host/src/host/host-serving.ts @@ -1,18 +1,19 @@ +import type { RecordedPoint } from '@on.auto/ledger'; import { Function } from 'effect'; import type { HostDatabase } from '../database/host-database.ts'; -import type { Consumer } from '../follower/consumers.ts'; import { startLoop } from '../loop/host-loop.ts'; import { hostWakeUp } from '../loop/host-wake-up.ts'; import type { ProjectorSettings } from '../projector/projector-settings.ts'; import { startProjector } from '../projector/projector.ts'; -import { startReacting } from '../reactions/host-reactions.ts'; +import { startReacting, type FollowerConsumers } from '../reactions/host-reactions.ts'; import { closeHostParts, openHostParts, type HostParts, type PartsOptions } from './host-parts.ts'; import type { WorkflowStore } from './workflow-store.ts'; export interface ServingOptions extends PartsOptions { - readonly consumers?: readonly Consumer[]; + readonly consumers?: FollowerConsumers['consumers']; readonly projector?: ProjectorSettings; + readonly placeAtOpen: RecordedPoint; } export interface Serving { @@ -30,6 +31,7 @@ function followerOf(database: HostDatabase, options: ServingOptions, parts: Host sweepEveryMs: options.sweepEveryMs, reports: options.reports, context: options.context, + placeAtOpen: options.placeAtOpen, }, options.reactions, parts.reacting, diff --git a/packages/workflow-host/src/host/workflow-host.ts b/packages/workflow-host/src/host/workflow-host.ts index 70ae09dac..544b4cecd 100644 --- a/packages/workflow-host/src/host/workflow-host.ts +++ b/packages/workflow-host/src/host/workflow-host.ts @@ -16,7 +16,7 @@ import { } from './run-requests.ts'; import { storeOf, type StoreSettings } from './workflow-store.ts'; -export interface HostOptions extends Omit { +export interface HostOptions extends Omit { readonly database: StoreSettings; readonly clock?: HostClock; readonly holder?: string; @@ -34,7 +34,8 @@ export interface WorkflowHost { export async function openWorkflowHost(options: HostOptions): Promise { const clock = options.clock ?? systemClock; const store = await storeOf(options.database, options.reports.lostConnection, options.projector?.folders ?? []); - const standing = await standingOn(store, { ...options, clock }, options.holder); + const { through: placeAtOpen } = await store.database.store.readAppended(undefined, 0); + const standing = await standingOn(store, { ...options, clock, placeAtOpen }, options.holder); const { guarded, closed } = gate(); const runs = runRequests({ database: store.database, clock, serving: standing.serving }); const stopped = async (): Promise => { diff --git a/packages/workflow-host/src/reaction-testing/follower-over.ts b/packages/workflow-host/src/reaction-testing/follower-over.ts index dc2ea1874..08d4a72de 100644 --- a/packages/workflow-host/src/reaction-testing/follower-over.ts +++ b/packages/workflow-host/src/reaction-testing/follower-over.ts @@ -10,8 +10,9 @@ import { refusalsOn } from '../reactions/refusals.ts'; import { recordingReports } from '../testing/recording-reports.ts'; import { recordedReactions, type RecordedReactions } from './recorded-reactions.ts'; -export function followerOver(database: HostDatabase, sweepEveryMs: number): RecordedReactions { +export async function followerOver(database: HostDatabase, sweepEveryMs: number): Promise { const reactions = recordedReactions(); + const { through: placeAtOpen } = await database.store.readAppended(undefined, 0); const follower = startReacting( { database, @@ -19,6 +20,7 @@ export function followerOver(database: HostDatabase, sweepEveryMs: number): Reco clock: systemClock, sweepEveryMs, reports: recordingReports().reports, + placeAtOpen, }, { ...reactions.options, appended: streamSignalOf() }, { refusals: refusalsOn(database, Date.now), stops: filterStops() }, diff --git a/packages/workflow-host/src/reactions/reaction-consumers.test.ts b/packages/workflow-host/src/reactions/reaction-consumers.test.ts index 588dea2dd..839a0adfe 100644 --- a/packages/workflow-host/src/reactions/reaction-consumers.test.ts +++ b/packages/workflow-host/src/reactions/reaction-consumers.test.ts @@ -52,7 +52,6 @@ function decliningEveryOffer(database: HostDatabase): Declining { return { outcome: 'stale', version: 2, declined: 'The run took no event of this kind' } as const; }), clock: { now: () => 0, sleep: () => Effect.void }, - sweepEveryMs: 1000, reports: { unsettled: () => Effect.void, trouble: () => Effect.void, diff --git a/packages/workflow-host/src/reactions/reaction-consumers.ts b/packages/workflow-host/src/reactions/reaction-consumers.ts index 2637c830f..ae674aa27 100644 --- a/packages/workflow-host/src/reactions/reaction-consumers.ts +++ b/packages/workflow-host/src/reactions/reaction-consumers.ts @@ -16,7 +16,7 @@ export interface ReactionUse { } export function reactionConsumersOf( - host: FollowerHost, + host: Pick, { options, refusals, stops }: ReactionUse, starting: Starting, ): readonly RecordConsumer[] { diff --git a/packages/workflow-host/src/sweeps/registry-retries.test.ts b/packages/workflow-host/src/sweeps/registry-retries.test.ts index b9cfb7740..7f7946a9d 100644 --- a/packages/workflow-host/src/sweeps/registry-retries.test.ts +++ b/packages/workflow-host/src/sweeps/registry-retries.test.ts @@ -53,7 +53,7 @@ describe('a playbook another process creates, swept before its registry could be it('is passed from its first record, so its first event starts a workflow without another append', async () => { const database = await openedOn(await onSQLite()); const held = sweptAtOnceAndRegistryFailing(database, beta); - const reactions = followerOver(held.database, 20); + const reactions = await followerOver(held.database, 20); const trigger = eventTrigger({ type: 'com.acme.closed' }); await playbookCreated(database.store, 'beta'); diff --git a/packages/workflow-host/src/waiting-testing/held-follower.ts b/packages/workflow-host/src/waiting-testing/held-follower.ts new file mode 100644 index 000000000..234881b27 --- /dev/null +++ b/packages/workflow-host/src/waiting-testing/held-follower.ts @@ -0,0 +1,46 @@ +import { playbooksStreamOfOrg } from '@on.auto/playbooks'; +import { Function } from 'effect'; + +import type { HostDatabase } from '../database/host-database.ts'; +import { openHostDatabase } from '../database/host-databases.ts'; +import { playbookCreated } from '../reaction-testing/playbook-writes.ts'; +import { aSQLiteFile, openedOn } from '../testing/host-files.ts'; +import { hostedOn, type HostedOptions } from '../testing/host-runs.ts'; +import { settledIn, type FollowedHost } from './followed-host.ts'; + +export interface HeldFollower extends FollowedHost { + readonly release: () => void; +} + +function registryHeldUntil(released: Promise, database: HostDatabase): HostDatabase { + const registry = playbooksStreamOfOrg('acme'); + return { + ...database, + store: { + ...database.store, + read: async (stream, after) => { + if (stream === registry) { + await released; + } + return database.store.read(stream, after); + }, + }, + }; +} + +export async function heldFollower(options: HostedOptions = {}): Promise { + const settings = { store: 'sqlite', file: aSQLiteFile() } as const; + const database = await openedOn(settings); + await playbookCreated(database.store, 'alpha'); + const released = Promise.withResolvers(); + const held = registryHeldUntil(released.promise, await openHostDatabase(settings, Function.constVoid)); + const hosted = await hostedOn({ database: held }, options); + return { + database, + hosted, + settled: settledIn(hosted), + release: () => { + released.resolve(); + }, + }; +}