From ff4af821bb60b143d042802d6f4452f0f3fb4109 Mon Sep 17 00:00:00 2001 From: Rami Date: Fri, 9 Oct 2026 09:56:58 +0100 Subject: [PATCH 1/5] test(workflow-host): the start rate test of several triggers makes two event starts, not 61 The test published 61 events one after another into the host's SQLite file while the host's follower started a run for each on a second connection to the same file, so its length grew with the start rate and with the machine's load: 94 ms alone, 507 to 643 ms beside 64 busy threads, and 5,049 ms against the 5,000 ms limit in the full gate. The test now records that the workflow was started 59 times in the current minute, as the backlog test of start-rates.test.ts records the starts already waiting, and then publishes two events: the every schedule still starts its run, the first event makes the 60th start and the second waits for a later minute. A schedule start counted in the rate leaves both events waiting, and an event start left out of it starts both, and each fails the test. That 60 starts are admitted one by one stays proven by start-rates.test.ts. Co-Authored-By: Claude Opus 5.5 --- .../src/triggers/several-triggers.test.ts | 52 +++++++++++++------ 1 file changed, 35 insertions(+), 17 deletions(-) diff --git a/packages/workflow-host/src/triggers/several-triggers.test.ts b/packages/workflow-host/src/triggers/several-triggers.test.ts index 61886ae33..340339f47 100644 --- a/packages/workflow-host/src/triggers/several-triggers.test.ts +++ b/packages/workflow-host/src/triggers/several-triggers.test.ts @@ -3,19 +3,21 @@ import { describe, expect, it } from 'vitest'; import { statement } from '../database/statement.ts'; import { + alpha, at, cronTrigger, eventRecordOf, eventTrigger, everyTrigger, published, + publishedInTurn, runRecorded, specRecordAt, specRecorded, } from '../reaction-testing/brain-writes.ts'; import { startsReaching, untilScheduleRuns } from '../reaction-testing/kept-triggers.ts'; import { movedClock } from '../reaction-testing/moved-clock.ts'; -import { reactingHost } from '../reaction-testing/reacting-host.ts'; +import { reactingHost, type ReactingHost } from '../reaction-testing/reacting-host.ts'; import { until } from '../reaction-testing/until.ts'; import { reactionExecutionIdOf } from '../reactions/reaction-ids.ts'; import { mostStartsAMinute } from '../reactions/start-rates.ts'; @@ -30,15 +32,33 @@ function isoAt(minutes: number): string { return new Date(activatedAt + minutes * aMinute).toISOString(); } -async function closingAt(start: number, onEvents = closed) { +async function openedAt(start: number) { const clock = movedClock(start); const reacting = await reactingHost({ clock }); - await specRecorded(reacting.database.store, { + return { reacting, clock, starts: reacting.reactions.starts }; +} + +function closingSaved({ database }: ReactingHost, onEvents = closed) { + return specRecorded(database.store, { name: 'close', version: 1, triggers: [onEvents, cronTrigger('30 9 * * *'), everyTrigger(15 * aMinute)], }); - return { reacting, clock, starts: reacting.reactions.starts }; +} + +async function closingAt(start: number, onEvents = closed) { + const opened = await openedAt(start); + await closingSaved(opened.reacting, onEvents); + return opened; +} + +function closingStartedIn({ database }: ReactingHost, minute: number, starts: number) { + return Effect.runPromise( + database.write( + statement`INSERT INTO workflow_reaction_rates (brain_key, workflow, minute, starts) + VALUES (${alpha}, 'close', ${minute}, ${starts})`, + ), + ); } function executionIdOf(start: { readonly executionId: string } | undefined): string { @@ -112,26 +132,24 @@ describe('a due time of a schedule asked for again', () => { }); describe('the start rate of a workflow with several triggers', () => { - it(`counts the ${mostStartsAMinute} starts of its event trigger a minute, and never a start of its schedule`, async () => { - const { reacting } = await closingAt(activatedAt + 15 * aMinute + 1000); + it(`counts the starts of its event trigger to ${mostStartsAMinute} a minute, and never a start of its schedule`, async () => { + const minute = activatedAt + 15 * aMinute; + const { reacting } = await openedAt(minute + 1000); + await closingStartedIn(reacting, minute, mostStartsAMinute - 1); + await closingSaved(reacting); await startsReaching(reacting.reactions.starts, 1); - await Array.from({ length: mostStartsAMinute + 1 }, (_, index) => index).reduce>( - (before, index) => - before.then(() => published(reacting.database.store, { id: `e${index}`, type: 'com.acme.closed' })), - Promise.resolve(), - ); - const starts = await startsReaching(reacting.reactions.starts, mostStartsAMinute + 1); + await publishedInTurn(reacting.database.store, [ + { id: 'e1', type: 'com.acme.closed' }, + { id: 'e2', type: 'com.acme.closed' }, + ]); + const starts = await startsReaching(reacting.reactions.starts, 2); const waiting = await until( () => Effect.runPromise(reacting.database.read(statement`SELECT execution_id FROM workflow_reaction_backlog`)), (rows) => rows.length > 0, ); - expect([starts.filter(({ trigger }) => trigger.kind === 'event').length, waiting.length]).toEqual([ - mostStartsAMinute, - 1, - ]); - expect(starts.filter(({ trigger }) => trigger.kind === 'every')).toHaveLength(1); + expect([starts.map(({ trigger }) => trigger.kind), waiting.length]).toEqual([['every', 'event'], 1]); }); }); From 73e4ec04e17dbdbefb71347d517a9aceab190a8a Mon Sep 17 00:00:00 2001 From: Rami Date: Fri, 9 Oct 2026 09:57:05 +0100 Subject: [PATCH 2/5] test(server): the rebuilding recall test asks once the rebuild holds its first fold The test waited for the recall function's standing to read rebuilding, which the standing also reads before the projector has added the view's row (recall-standing.ts answers rebuilding when no view is kept). The projector adds the row as waiting and moves it to rebuilding in a second write (view-reconciling.ts), so the recall the test made next answered whichever of three details held at that moment: not begun, waits behind other recall functions, or is building. Alone, one run in eight saw the standing before the row existed, and beside 64 busy threads one run in twelve failed on the waiting detail. The test now waits until the gate holds the rebuild's first fold, which the projector asks for only after it has set the row to rebuilding and which holds the view at 0 folded events, so the building detail is the one the situation gives. The test of a function that waits behind another, which shows the waiting detail, waits the same way, with the bound the standing waits use rather than vi.waitFor's default second. Co-Authored-By: Claude Opus 5.5 --- .../server/src/recall/recall-versions.test.ts | 9 ++++---- .../src/testing/servers/recall-server.ts | 23 +++++++++++-------- 2 files changed, 18 insertions(+), 14 deletions(-) diff --git a/packages/server/src/recall/recall-versions.test.ts b/packages/server/src/recall/recall-versions.test.ts index 6571108ba..8596dfb6a 100644 --- a/packages/server/src/recall/recall-versions.test.ts +++ b/packages/server/src/recall/recall-versions.test.ts @@ -1,11 +1,12 @@ import { internalTermsIn } from '@beonauto/api/testing'; import { recallDocument } from '@beonauto/recollection/testing'; import { Schema } from 'effect'; -import { afterEach, describe, expect, it, vi } from 'vitest'; +import { afterEach, describe, expect, it } from 'vitest'; import { alpha, type ReasoningServer } from '../testing/servers/reasoning-server.ts'; import { brainWithReviews, + foldsHeld, foldsHeldUntilOpened, inState, liveAt, @@ -65,7 +66,7 @@ describe('a recall function whose view is being built', { timeout: recallTestTim const gate = foldsHeldUntilOpened(); const server = await servingWith({}, gate); await brainWithReviews(server, 1); - await standingUntil(server, 'reviews', inState('rebuilding')); + await foldsHeld(gate, 1); const askedAt = Date.now(); const meanwhile = await recalled(server, 'reviews', { campaign: 'spring' }); @@ -114,9 +115,7 @@ describe('a recall function waiting for its view to be built', { timeout: recall const gate = foldsHeldUntilOpened(); const server = await servingWith({ RECOLLECTION_MAX_REBUILDS: '1', RECOLLECTION_BRAINS_AT_ONCE: '1' }, gate); await brainWithReviews(server, 1, 'beta'); - await vi.waitFor(() => { - expect(gate.held()).toBe(1); - }); + await foldsHeld(gate, 1); await brainWithReviews(server, 1); await saved(server, 'count', '. + 1'); gate.letThrough(1); diff --git a/packages/server/src/testing/servers/recall-server.ts b/packages/server/src/testing/servers/recall-server.ts index 3c1d2aa52..1f0db589a 100644 --- a/packages/server/src/testing/servers/recall-server.ts +++ b/packages/server/src/testing/servers/recall-server.ts @@ -9,6 +9,8 @@ import { alpha, servingReasoning, type ReasoningServer } from './reasoning-serve export const recallTestTimeoutMs = 60_000; +const untilNearTheTestTimeout = { timeout: recallTestTimeoutMs - 10_000, interval: 50 }; + export const anyReview = [ '---', 'description: Reviews a campaign brief, answering whatever the model writes', @@ -73,6 +75,12 @@ export function foldsHeldUntilOpened(): Gate { }; } +export function foldsHeld(gate: Gate, folds: number): Promise { + return vi.waitFor(() => { + expect(gate.held()).toBe(folds); + }, untilNearTheTestTimeout); +} + export async function brainWithReviews(server: ReasoningServer, briefsFirst = 0, brain = 'alpha'): Promise { const path = `/v1/orgs/acme/brains/${brain}`; await server.call('POST', '/v1/orgs/acme/brains', { body: { brain, name: brain } }); @@ -102,15 +110,12 @@ export function standingUntil( name: string, holds: (standing: Standing) => boolean, ): Promise { - return vi.waitFor( - async () => { - const { body } = await server.call('GET', `${alpha}/specs/recollection/${name}`); - const standing = Option.getOrUndefined(decodeStanding(body))?.standing; - expect(standing !== undefined && holds(standing)).toBe(true); - return body; - }, - { timeout: recallTestTimeoutMs - 10_000, interval: 50 }, - ); + return vi.waitFor(async () => { + const { body } = await server.call('GET', `${alpha}/specs/recollection/${name}`); + const standing = Option.getOrUndefined(decodeStanding(body))?.standing; + expect(standing !== undefined && holds(standing)).toBe(true); + return body; + }, untilNearTheTestTimeout); } export function liveWith(folded: number): (standing: Standing) => boolean { From 1a4d8fb8b35232a630bcffb9631d0b3662581530 Mon Sep 17 00:00:00 2001 From: Rami Date: Fri, 9 Oct 2026 10:04:52 +0100 Subject: [PATCH 3/5] fix(workflow-host): a lone recall function is never shown waiting behind others The projector wrote a new or renewed view as waiting and then, in a second write, moved the views that took a slot to rebuilding. Between the two writes a brain's only recall function read as waiting, and a run of it answered that it waits behind other recall functions of the brain being built, when there were none. A new or renewed view is now written in the phase its place gives it: the slots are counted over the views still building and the fresh ones, in the order saved, before anything is written, and a dropped view is removed first, so a fresh view never takes a slot a dropped one still holds. The rows suite watches the row after every write a reconcile makes and sees a view built alone as rebuilding from its first write, new and renewed; on the earlier code it sees waiting after each. Co-Authored-By: Claude Opus 5.5 --- packages/workflow-host/README.md | 2 +- .../src/projector/view-reconciling.ts | 92 +++++++++++++------ .../src/views-testing/rows-suite.ts | 39 +++++++- 3 files changed, 100 insertions(+), 33 deletions(-) diff --git a/packages/workflow-host/README.md b/packages/workflow-host/README.md index 449f2f305..699d476c9 100644 --- a/packages/workflow-host/README.md +++ b/packages/workflow-host/README.md @@ -184,7 +184,7 @@ The projector reads the rows of a brain without their views, and reads a view on The projector (`src/projector`) runs while the host holds the database's workflows, woken by its sweep and by the ledger's `AppendSignal`, raised in the process on every append. Each sweep it asks the ledger for the definition streams of the recall type, `EventStore.definitionStreams`, through the ledger's partial index on their names, and takes every brain whose stream moved or that has rows; an append to a brain it does not know asks for them at once, so a brain's first recall function, and a brain made after the start, are found without a restart. A brain is passed in a fiber of its own, at most `brainsAtOnce` at once. A pass of a brain: -1. reads the brain's definition stream from where it last read it and brings the rows in line: a function without a row gets one, a newer version resets its row, a retired function's row is dropped, and the first `rebuildsAtOnce` rows that wait or rebuild, in the order saved, rebuild while the others wait; a stalled row holds no slot; +1. reads the brain's definition stream from where it last read it and brings the rows in line: a function without a row gets one, a newer version resets its row, a retired function's row is dropped, and the first `rebuildsAtOnce` rows that wait or rebuild, in the order saved, rebuild while the others wait; a stalled row holds no slot. A new or reset row is written in the phase its place gives it, so a view that takes a slot, as a brain's only recall function does, is never seen waiting; 2. reads the brain from the smallest checkpoint of its live views, a page of up to 1,000 records, the most one read of the store examines, of the types the views can match, the fact types their filters name, `event_published` when a filter names any other type, and the three definition types, oldest first behind PostgreSQL's horizon, so an event committed late is folded late, never passed over; 3. folds the page in a worker of the pool, warm when one of its module is idle (`src/pages`), the projector holding at most half the pool's workers and at least one: each view folds the events after its own checkpoint that its filters match, never the runs of its own function, each fold under its own work, deadline and size and the view's schema, and each filter's `data` test under the same limits, with work of its own, sharing the fold's deadline; the worker checks the page's 2 seconds of folding, counted from its first fold, before each fold, and ends the page early, before the next fold, once they are spent, so at most one fold runs past them; 4. writes each view that moved once, its checkpoint alone when it only read past events, with one statement conditional on the version and the checkpoint it read, so a host that lost the claim, or one that read a row a newer version has since reset, writes nothing, and a dropped row is never made again; diff --git a/packages/workflow-host/src/projector/view-reconciling.ts b/packages/workflow-host/src/projector/view-reconciling.ts index 67aaf09c0..a095f66a7 100644 --- a/packages/workflow-host/src/projector/view-reconciling.ts +++ b/packages/workflow-host/src/projector/view-reconciling.ts @@ -2,8 +2,8 @@ import { Effect } from 'effect'; import { rowsOf, type DatabaseFailed, type HostDatabase } from '../database/host-database.ts'; import type { Statement } from '../database/statement.ts'; -import { viewDetailsOf } from '../views/view-details.ts'; -import type { ViewRow } from '../views/view-rows.ts'; +import { viewDetailsOf, type ViewDetails } from '../views/view-details.ts'; +import type { ViewPhase, ViewRow } from '../views/view-rows.ts'; import { viewRowOf, ViewRowSchema } from '../views/view-rows.ts'; import { phaseSet, viewAdded, viewDropped, viewRenewed, viewsOfBrain, type NewView } from '../views/view-statements.ts'; import { @@ -38,41 +38,75 @@ function definitionsOf({ database, definitionType, definitions }: Reconciling, b ); } -function newViewOf(brain: string, name: string, kept: KeptFunction): NewView | undefined { +interface Saved { + readonly name: string; + readonly saved: number; +} + +interface FreshFunction extends Saved { + readonly version: number; + readonly details: ViewDetails; +} + +function isBuilding({ phase }: ViewRow): boolean { + return phase === 'waiting' || phase === 'rebuilding'; +} + +function slotsOf(building: readonly Saved[], rebuildsAtOnce: number): ReadonlySet { + const inOrderSaved = building.toSorted((one, other) => one.saved - other.saved); + return new Set(inOrderSaved.slice(0, rebuildsAtOnce).map(({ name }) => name)); +} + +function phaseIn(slots: ReadonlySet, name: string): ViewPhase { + return slots.has(name) ? 'rebuilding' : 'waiting'; +} + +function freshOf(name: string, kept: KeptFunction, row: ViewRow | undefined): readonly FreshFunction[] { const details = viewDetailsOf(kept.details); - return details === undefined - ? undefined - : { - brain, - name, - version: kept.version, - saved: kept.saved, - details: JSON.stringify(details), - phase: 'waiting', - view: JSON.stringify(details.initial), - }; + return details === undefined || (row !== undefined && row.version >= kept.version) + ? [] + : [{ name, saved: kept.saved, version: kept.version, details }]; +} + +function newViewOf(brain: string, { name, saved, version, details }: FreshFunction, phase: ViewPhase): NewView { + return { + brain, + name, + version, + saved, + details: JSON.stringify(details), + phase, + view: JSON.stringify(details.initial), + }; } -function changesOf(brain: string, rows: readonly ViewRow[], { functions }: BrainDefinitions): readonly Statement[] { +function changesOf( + brain: string, + rows: readonly ViewRow[], + { functions }: BrainDefinitions, + rebuildsAtOnce: number, +): readonly Statement[] { const byName = new Map(rows.map((row) => [row.name, row])); - const kept = [...functions].flatMap(([name, function_]: readonly [string, KeptFunction]): readonly Statement[] => { - const added = newViewOf(brain, name, function_); - const row = byName.get(name); - if (added === undefined || (row !== undefined && row.version >= added.version)) { - return []; - } - return [row === undefined ? viewAdded(added) : viewRenewed(added)]; - }); + const fresh = [...functions].flatMap(([name, kept]: readonly [string, KeptFunction]) => + freshOf(name, kept, byName.get(name)), + ); + const freshNames = new Set(fresh.map(({ name }) => name)); + const stillBuilding = rows.filter((row) => isBuilding(row) && functions.has(row.name) && !freshNames.has(row.name)); + const slots = slotsOf([...stillBuilding, ...fresh], rebuildsAtOnce); const dropped = rows.filter(({ name }) => !functions.has(name)).map(({ name }) => viewDropped(brain, name)); - return [...kept, ...dropped]; + const written = fresh.map((function_) => { + const view = newViewOf(brain, function_, phaseIn(slots, function_.name)); + return byName.has(view.name) ? viewRenewed(view) : viewAdded(view); + }); + return [...dropped, ...written]; } function phasesOf(rows: readonly ViewRow[], rebuildsAtOnce: number): readonly ViewRow[] { - const building = rows.filter(({ phase }) => phase === 'waiting' || phase === 'rebuilding'); - const slots = new Set(building.slice(0, rebuildsAtOnce).map(({ name }) => name)); - return rows.map((row) => - building.includes(row) ? { ...row, phase: slots.has(row.name) ? 'rebuilding' : 'waiting' } : row, + const slots = slotsOf( + rows.filter((row) => isBuilding(row)), + rebuildsAtOnce, ); + return rows.map((row) => (isBuilding(row) ? { ...row, phase: phaseIn(slots, row.name) } : row)); } export function reconciled(parts: Reconciling, brain: string): Effect.Effect { @@ -80,7 +114,7 @@ export function reconciled(parts: Reconciling, brain: string): Effect.Effect> { return Object.fromEntries(rows.map(({ name, phase }) => [name, phase])); } -function reconciling(views: ViewHarness, rebuildsAtOnce: number): () => Promise { +function reconciling( + views: ViewHarness, + rebuildsAtOnce: number, + database = views.store.database, +): () => Promise { const parts: Reconciling = { - database: views.store.database, + database, definitionType: 'recollection', rebuildsAtOnce, definitions: new Map(), @@ -46,6 +50,19 @@ function rowNamed(views: ViewHarness, name: string): Promise { return Effect.runPromise(Effect.map(read, (row) => viewRowOf(row))); } +function phasesAfterEachWrite(views: ViewHarness, name: string) { + const { database } = views.store; + const seen: string[] = []; + const phaseRead = Effect.map(rowsOf(ViewRowSchema, database.read(viewNamed(alphaKey, name))), (rows) => { + seen.push(...rows.map(({ phase }) => phase)); + }); + const watched: HostDatabase = { + ...database, + write: (statement) => Effect.tap(database.write(statement), () => phaseRead), + }; + return { database: watched, seen: () => seen }; +} + function rowsOfAlpha(views: ViewHarness): Promise { return Effect.runPromise(rowsOfBrain(views.store.database, alphaKey)); } @@ -96,6 +113,21 @@ function slotTests(settingsOf: SettingsOf): void { }); } +function loneViewTests(settingsOf: SettingsOf): void { + it('write a view built alone as rebuilding from its first write, new or renewed, and never as waiting', async () => { + const views = await viewHarness(await settingsOf()); + const phases = phasesAfterEachWrite(views, 'runs'); + const reconcile = reconciling(views, 1, phases.database); + + await views.saved('runs', counting); + await reconcile(); + await views.saved('runs', counting); + const [renewed] = await reconcile(); + + expect([phases.seen(), renewed?.version]).toEqual([['rebuilding', 'rebuilding'], 2]); + }); +} + function conditionalWriteTests(settingsOf: SettingsOf): void { it('take no write from a host that read a row before another wrote it or a newer version reset it', async () => { const views = await viewHarness(await settingsOf()); @@ -148,6 +180,7 @@ function racingTests(settingsOf: SettingsOf): void { export function rowsSuite(settingsOf: SettingsOf): void { describe('the rows of the views', { timeout: viewTestTimeoutMs }, () => { slotTests(settingsOf); + loneViewTests(settingsOf); conditionalWriteTests(settingsOf); racingTests(settingsOf); }); From bcc9e257694e71c6a68f6b56f8eba62e8533a74a Mon Sep 17 00:00:00 2001 From: Rami Date: Fri, 9 Oct 2026 10:04:59 +0100 Subject: [PATCH 4/5] test(server): the rebuilding recall test orders its answer by the held fold, not a clock The test measured that the recall answered within 2,000 ms. The gate already holds the rebuild's first fold until after the answer, so the rebuild cannot finish before it: a recall that waited for its view would never answer, and the answer's detail says the view has folded 0 events. The wall-clock bound added only a dependence on the machine's load, and is removed. Co-Authored-By: Claude Opus 5.5 --- packages/server/src/recall/recall-versions.test.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/packages/server/src/recall/recall-versions.test.ts b/packages/server/src/recall/recall-versions.test.ts index 8596dfb6a..a1a2809d5 100644 --- a/packages/server/src/recall/recall-versions.test.ts +++ b/packages/server/src/recall/recall-versions.test.ts @@ -62,15 +62,13 @@ function updated(server: ReasoningServer, name: string, fold: string) { } describe('a recall function whose view is being built', { timeout: recallTestTimeoutMs }, () => { - it('answers at once, unavailable, rebuilding, with Retry-After, until its view has caught up', async () => { + it('answers before its view is built, unavailable, rebuilding, with Retry-After, until its view has caught up', async () => { const gate = foldsHeldUntilOpened(); const server = await servingWith({}, gate); await brainWithReviews(server, 1); await foldsHeld(gate, 1); - const askedAt = Date.now(); const meanwhile = await recalled(server, 'reviews', { campaign: 'spring' }); - const answeredInMs = Date.now() - askedAt; gate.open(); await standingUntil(server, 'reviews', liveWith(1)); const caughtUp = await recalled(server, 'reviews', { campaign: 'spring' }); @@ -87,7 +85,6 @@ describe('a recall function whose view is being built', { timeout: recallTestTim }); expect(meanwhile.headers.get('retry-after')).toBe('5'); expect(internalTermsIn(decodeDetail(meanwhile.body).detail)).toEqual([]); - expect(answeredInMs).toBeLessThan(2000); expect(caughtUp).toMatchObject({ status: 200, body: { output: [{ verdict: 'approve' }] } }); }); From 6be47d067c0f84ec3dbcade915f643f24469eb29 Mon Sep 17 00:00:00 2001 From: Rami Date: Fri, 9 Oct 2026 10:24:31 +0100 Subject: [PATCH 5/5] fix(workflow-host): a waiting view takes a freed slot before the holder's row changes When the view holding a slot was retired or renewed in the reconcile that freed it, the view waiting behind it kept reading waiting beside the free slot until the phase writes after the next read of the rows: a retired holder's row was deleted first, and a renewed holder left both rows waiting for a moment. A still-building view that takes a slot is now moved to rebuilding first, before the drops and before the new and renewed rows. With the drops first, the rows suite sees the waiting view alone and waiting between the delete and its move; moving it first leaves the retired holder and it both rebuilding until the delete, which no standing shows, since a retired function has none. The suite records the rows after every write of a reconcile, with the holder retired while another view is saved and renewed while another waits, and no row reads waiting while fewer rows rebuild than there are slots. Since a view that takes a slot now moves with the other changes, the writes after the rows are read again only move back to waiting the rows past the slots, as when the brain may build fewer views at once than it did; the suite now covers that case too. Co-Authored-By: Claude Opus 5.5 --- .../src/projector/view-reconciling.ts | 5 +- .../src/views-testing/rows-suite.ts | 86 +++++++++++++++++-- 2 files changed, 81 insertions(+), 10 deletions(-) diff --git a/packages/workflow-host/src/projector/view-reconciling.ts b/packages/workflow-host/src/projector/view-reconciling.ts index a095f66a7..6c07280a3 100644 --- a/packages/workflow-host/src/projector/view-reconciling.ts +++ b/packages/workflow-host/src/projector/view-reconciling.ts @@ -93,12 +93,15 @@ function changesOf( const freshNames = new Set(fresh.map(({ name }) => name)); const stillBuilding = rows.filter((row) => isBuilding(row) && functions.has(row.name) && !freshNames.has(row.name)); const slots = slotsOf([...stillBuilding, ...fresh], rebuildsAtOnce); + const promoted = stillBuilding + .filter((row) => row.phase === 'waiting' && slots.has(row.name)) + .map((row) => phaseSet(row, 'rebuilding')); const dropped = rows.filter(({ name }) => !functions.has(name)).map(({ name }) => viewDropped(brain, name)); const written = fresh.map((function_) => { const view = newViewOf(brain, function_, phaseIn(slots, function_.name)); return byName.has(view.name) ? viewRenewed(view) : viewAdded(view); }); - return [...dropped, ...written]; + return [...promoted, ...dropped, ...written]; } function phasesOf(rows: readonly ViewRow[], rebuildsAtOnce: number): readonly ViewRow[] { diff --git a/packages/workflow-host/src/views-testing/rows-suite.ts b/packages/workflow-host/src/views-testing/rows-suite.ts index b6aaa9ad5..f1d3a579c 100644 --- a/packages/workflow-host/src/views-testing/rows-suite.ts +++ b/packages/workflow-host/src/views-testing/rows-suite.ts @@ -12,7 +12,9 @@ import { viewHarness, type ViewHarness } from './view-harness.ts'; const NameRow = Schema.Struct({ name: Schema.String }); -function phasesOf(rows: readonly ViewRow[]): Readonly> { +type PhasedRows = readonly Pick[]; + +function phasesOf(rows: PhasedRows): Readonly> { return Object.fromEntries(rows.map(({ name, phase }) => [name, phase])); } @@ -50,15 +52,15 @@ function rowNamed(views: ViewHarness, name: string): Promise { return Effect.runPromise(Effect.map(read, (row) => viewRowOf(row))); } -function phasesAfterEachWrite(views: ViewHarness, name: string) { +function rowsAfterEachWrite(views: ViewHarness) { const { database } = views.store; - const seen: string[] = []; - const phaseRead = Effect.map(rowsOf(ViewRowSchema, database.read(viewNamed(alphaKey, name))), (rows) => { - seen.push(...rows.map(({ phase }) => phase)); + const seen: PhasedRows[] = []; + const rowsRead = Effect.map(rowsOfBrain(database, alphaKey), (rows) => { + seen.push(rows.map(({ name, phase }) => ({ name, phase }))); }); const watched: HostDatabase = { ...database, - write: (statement) => Effect.tap(database.write(statement), () => phaseRead), + write: (statement) => Effect.tap(database.write(statement), () => rowsRead), }; return { database: watched, seen: () => seen }; } @@ -116,15 +118,79 @@ function slotTests(settingsOf: SettingsOf): void { function loneViewTests(settingsOf: SettingsOf): void { it('write a view built alone as rebuilding from its first write, new or renewed, and never as waiting', async () => { const views = await viewHarness(await settingsOf()); - const phases = phasesAfterEachWrite(views, 'runs'); - const reconcile = reconciling(views, 1, phases.database); + const watched = rowsAfterEachWrite(views); + const reconcile = reconciling(views, 1, watched.database); await views.saved('runs', counting); await reconcile(); await views.saved('runs', counting); const [renewed] = await reconcile(); - expect([phases.seen(), renewed?.version]).toEqual([['rebuilding', 'rebuilding'], 2]); + expect([watched.seen().map((rows) => phasesOf(rows)), renewed?.version]).toEqual([ + [{ runs: 'rebuilding' }, { runs: 'rebuilding' }], + 2, + ]); + }); +} + +function handOverTests(settingsOf: SettingsOf): void { + it('hand the slot of a retired view to the view waiting, as another is saved, never leaving one waiting beside a free slot', async () => { + const views = await viewHarness(await settingsOf()); + const watched = rowsAfterEachWrite(views); + const reconcile = reconciling(views, 1, watched.database); + await views.saved('first', counting); + await views.saved('second', counting); + await reconcile(); + + await views.retired('first'); + await views.saved('third', counting); + const rows = await reconcile(); + + expect(watched.seen().map((seen) => phasesOf(seen))).toEqual([ + { first: 'rebuilding' }, + { first: 'rebuilding', second: 'waiting' }, + { first: 'rebuilding', second: 'rebuilding' }, + { second: 'rebuilding' }, + { second: 'rebuilding', third: 'waiting' }, + ]); + expect(phasesOf(rows)).toEqual({ second: 'rebuilding', third: 'waiting' }); + }); + + it('hand the slot of a renewed view to the view waiting, never leaving one waiting beside a free slot', async () => { + const views = await viewHarness(await settingsOf()); + const watched = rowsAfterEachWrite(views); + const reconcile = reconciling(views, 1, watched.database); + await views.saved('first', counting); + await views.saved('second', counting); + await reconcile(); + + await views.saved('first', counting); + const rows = await reconcile(); + + expect(watched.seen().map((seen) => phasesOf(seen))).toEqual([ + { first: 'rebuilding' }, + { first: 'rebuilding', second: 'waiting' }, + { first: 'rebuilding', second: 'rebuilding' }, + { second: 'rebuilding', first: 'waiting' }, + ]); + expect(phasesOf(rows)).toEqual({ second: 'rebuilding', first: 'waiting' }); + }); +} + +function fewerSlotsTests(settingsOf: SettingsOf): void { + it('move back to waiting the views past the slots when the brain may build fewer at once', async () => { + const views = await viewHarness(await settingsOf()); + await views.saved('first', counting); + await views.saved('second', counting); + await reconciling(views, 2)(); + + const rows = await reconciling(views, 1)(); + const kept = await rowsOfAlpha(views); + + expect([phasesOf(rows), phasesOf(kept)]).toEqual([ + { first: 'rebuilding', second: 'waiting' }, + { first: 'rebuilding', second: 'waiting' }, + ]); }); } @@ -181,6 +247,8 @@ export function rowsSuite(settingsOf: SettingsOf): void { describe('the rows of the views', { timeout: viewTestTimeoutMs }, () => { slotTests(settingsOf); loneViewTests(settingsOf); + handOverTests(settingsOf); + fewerSlotsTests(settingsOf); conditionalWriteTests(settingsOf); racingTests(settingsOf); });