fix(workflow-host): the follower notes each unreadable record, and a first follow misses nothing

packages/workflow-host/measure/finish-catch-up.ts

@@ -45,11 +45,10 @@ async function latestOf(opened: Opened): Promise<string | null> {
   return records[0]?.cursor ?? null;
 }
 
-async function untilAt(opened: Opened, cursor: string | null): Promise<void> {
-  const [place] = await placesOf(opened);
-  if (place !== cursor) {
+async function untilPlaced(opened: Opened, reached: (places: readonly (string | null)[]) => boolean): Promise<void> {
+  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();

packages/workflow-host/src/follower/consumers.ts

@@ -28,7 +28,7 @@ interface Batch {
   readonly more: boolean;
 }
 
-export interface RecordConsumer<Followed extends CallRecord = FollowedRecord> {
+export interface RecordConsumer<Followed = FollowedRecord> {
   readonly name: string;
   readonly skippedAfterSweeps: number;
   readonly batchOf: (followed: Followed, after: string | undefined, most: number) => Effect.Effect<Batch>;
@@ -38,27 +38,42 @@ export interface Consumer extends RecordConsumer {
   readonly types: readonly string[];
 }
 
-export interface CallConsumer extends RecordConsumer<CallRecord> {
-  readonly types: readonly string[];
-}
-
 export interface BoundConsumer {
   readonly name: string;
   readonly skippedAfterSweeps: number;
   readonly batchOf: (after: string | undefined, most: number) => Effect.Effect<Batch>;
 }
 
-export function boundTo<Followed extends CallRecord>(
-  consumer: RecordConsumer<Followed>,
-  followed: Followed,
-): BoundConsumer {
+export function boundTo<Followed>(consumer: RecordConsumer<Followed>, followed: Followed): BoundConsumer {
   return {
     name: consumer.name,
     skippedAfterSweeps: consumer.skippedAfterSweeps,
     batchOf: (after, most) => consumer.batchOf(followed, after, most),
   };
 }
 
+export type Reading = BoundConsumer | 'unreadable';
+
+export interface CallConsumer {
+  readonly types: readonly string[];
+  readonly read: (call: CallRecord) => Reading;
+}
+
+interface CallReader<Read> extends RecordConsumer<Read> {
+  readonly types: readonly string[];
+  readonly readOf: (call: CallRecord) => Read | undefined;
+}
+
+export function callConsumerOf<Read>(reader: CallReader<Read>): 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;

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',
+    });
+  });
+});

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,

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;
 }

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[] = [];

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');

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<void>;
@@ -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<void>;
   readonly definitionType: string;
@@ -78,10 +78,10 @@ function seenOf(database: HostDatabase, stream: string): Effect.Effect<number |
   ).pipe(Effect.map((rows) => rows.at(0)?.position));
 }
 
-function followedAtTailOn({
+function followedFromTheOpenOn({
   database,
   playbooks,
-  records,
+  placeAtOpen,
   applyDefinitionRecord,
   unreadable,
   definitionType,
@@ -93,15 +93,18 @@ 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),
       );
       yield* Effect.forEach(definitionRecordsIn(recorded), (record) => applied(playbookKey, record), { discard: true });
     });
 }
 
-function registryReadOn({ database, playbooks }: DiscoveryParts, followedAtTail: ReturnType<typeof followedAtTailOn>) {
+function registryReadOn(
+  { database, playbooks }: DiscoveryParts,
+  followedFromTheOpen: ReturnType<typeof followedFromTheOpenOn>,
+) {
   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]));

packages/workflow-host/src/follower/playbook-records.ts

@@ -27,7 +27,6 @@ export interface PlaybookRecords {
     cursor: string | null,
     delivers: ReadonlySet<string>,
   ) => Effect.Effect<RecordedPage>;
-  readonly tail: (playbook: PlaybookAddress) => Effect.Effect<string | null>;
 }
 
 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<string | null>((_, { cursor }) => cursor, null)),
-      ),
   };
 }

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,