From c7f5d4573fc76efad1dbab9bb1f41cb90d8c23f9 Mon Sep 17 00:00:00 2001 From: Crispin-2713 Date: Mon, 19 Oct 2026 03:58:47 -0400 Subject: [PATCH] fix(coalescing): an unread render stays when a later render or notice arrives, and a retraction or plain occurrence takes the whole run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A render describes what changed since the server's previous render (RFC-006 §5.2), so a later render can't stand in for an earlier one the model hasn't read (§3.3's last paragraph, vector 49). The coalescer kept one occupant per subject and treated a render as it treats a plain occurrence: - a render settling while an earlier render was still unread took the subject's slot, so a retraction withdrew it alone and left the earlier one queued (#548); - a notice arriving after a render had landed unread displaced it, as §5.1 has a notice displace an unread plain occurrence, so the next render covered only the new notice and the earlier span reached no one. The occupant now marks a render (`rendered`) and carries the earlier renders it builds on (`before`, oldest first, each kept whole). A render settling while such a run is there joins it. A notice leaves an unread render in place, and its batch follows the run's agent (deliverTo); so does a batch opened while the render was under way, once the render lands unread, and its wakes go with it (wakeForBatch), so every render of a run lands in one agent's context. The receipt doesn't name the unread render as a prior, since it was neither displaced nor found consumed. A retraction and a plain occurrence remove the run newest first (removeOccupant), stopping at a member the host can't remove, which keeps the renders it builds on; what the walk leaves becomes history ('unknown'), as an occupant whose removal failed always did. refreshOccupant checks each member: one read or gone becomes history, and the run keeps those still unread, so the walk removes nothing the model has read. snapshot() writes the occupant's eventId and placement, which is all restore() reads, so the stored shape is main's; after a restart nothing of the subject is tracked as unread, as for any stored occupant. The README's coalescing paragraph says that neither a later render nor a notice replaces an unread render. Tests: mcpl-coalescer-render-run, over a stub host whose deliveries wait unread in one agent's deferred queue, covers #548's sequence, a batch opened during a render, a notice between two renders, a plain occurrence over a run, removals that fail at the newest and at an earlier render, a render read before a retraction, a three-render run, a gone render in the middle of one, a lone render that stays a render and stays removable, and the snapshot. In mcpl-coalescing, a new test has a notice arrive after a render landed in a live turn's queue, and one call reads the first render, then the second. Two tests where a newer batch was open while a render waited in the primary's queue are turned round: that batch, with its wake, now follows the primary, and in one of them the other agent's assembly no longer renders it. The changelog fragment says what agents see. Co-Authored-By: Claude Opus 5.5 --- README.md | 5 +- changelog.d/548-render-runs.fixed.md | 9 + src/mcpl/push-coalescer.ts | 126 +++++++++--- test/mcpl-coalescer-render-run.test.ts | 260 +++++++++++++++++++++++++ test/mcpl-coalescing.test.ts | 51 ++++- 5 files changed, 414 insertions(+), 37 deletions(-) create mode 100644 changelog.d/548-render-runs.fixed.md create mode 100644 test/mcpl-coalescer-render-run.test.ts diff --git a/README.md b/README.md index b0ec17d8b..cf1d43096 100644 --- a/README.md +++ b/README.md @@ -399,7 +399,10 @@ some version was read (or history is unknown). `initial: true` on a create lets the host know the subject's history is complete, so a create → edit → delete that nobody read leaves no trace. `deferred: true` (push only) holds a batch of notices outside context and asks the server for the content with `push/render` -when a turn is assembled. +when a turn is assembled. Each render describes what changed since the one +before it, so neither a later render nor a notice replaces an earlier render +the model hasn't read: the newer render lands after it, and a retraction or a +later plain occurrence withdraws the unread renders, newest first. Plain content stays in context and is delivered on arrival through the ordinary channel/push path; the host only remembers where it landed. "Unread" means: diff --git a/changelog.d/548-render-runs.fixed.md b/changelog.d/548-render-runs.fixed.md new file mode 100644 index 000000000..694bec5c9 --- /dev/null +++ b/changelog.d/548-render-runs.fixed.md @@ -0,0 +1,9 @@ +- A deferred push's render that's still unread now stays where it was + placed when a later render or notice of the same subject arrives, as + RFC-006 §5.2 and §3.3 have it: each render describes what changed since + the one before, so the later render lands after it. A retraction removes + the subject's unread renders, newest first, and a plain occurrence + displaces them. A later render used to take the subject's place, so a + retraction withdrew only that one and left the earlier render queued + (#548), and a notice arriving after a render had landed unread removed + it, so the model never read what it described. diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index 0080f48a2..fae7d094b 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -146,9 +146,31 @@ export interface AssemblyResult { materialized: Set; } +/** What holds a subject's single slot (§3.2). A render describes what changed + * since the server's previous render (§5.2), so the renders before it that + * are still unread stay where they were placed, and the newest carries them + * as `before` (§3.3, vector 49). A plain occurrence displaces the run and a + * retraction removes it (§6), newest first (removeOccupant). */ +interface Occupant { + eventId: string; + placement: CoalescingPlacement; + /** A materialized batch (its result or its fallback), which later renders build on. */ + rendered?: true; + /** The run's earlier renders, oldest first, each placed ahead of this one. */ + before?: Array>; +} + +/** The occupant holding a run's renders, given oldest first: the newest, + * carrying the rest as `before`. */ +function runOf(members: Array>): Occupant | undefined { + if (members.length === 0) return undefined; + const newest = members[members.length - 1]; + return members.length > 1 ? { ...newest, before: members.slice(0, -1) } : newest; +} + interface SubjectState { history: 'none' | 'some' | 'unknown'; - occupant?: { eventId: string; placement: CoalescingPlacement }; + occupant?: Occupant; batch?: DeferredBatch; rendering?: Rendering; consumedEventId?: string; @@ -480,7 +502,10 @@ export class PushCoalescer { const born = this.subjects.has(subject) ? undefined : (occurrence.initial && !occurrence.retract ? 'none' as const : 'unknown' as const); const state = this.subjectFor(subject, occurrence); const occupantUnread = this.refreshOccupant(state); - const priorEventId = state.occupant?.eventId ?? state.batch?.latest.eventId ?? state.rendering?.batch.latest.eventId ?? state.consumedEventId; + // A notice leaves an unread render in place (acceptDeferred): neither + // displaced nor found consumed, it isn't the prior (§3.3). + const occupantPrior = occurrence.deferred && state.occupant?.rendered ? undefined : state.occupant?.eventId; + const priorEventId = occupantPrior ?? state.batch?.latest.eventId ?? state.rendering?.batch.latest.eventId ?? state.consumedEventId; this.host.audit({ kind: 'received', subject, eventId: occurrence.eventId, retract: occurrence.retract, deferred: occurrence.deferred }); let outcome: CoalesceOutcome; @@ -534,14 +559,48 @@ export class PushCoalescer { return state; } - /** Re-check the occupant against the host: consumed occupants become history. */ + /** + * Re-check the occupant against the host, member by member: a member a + * request has read, or that is gone (isUnread is false either way), becomes + * history, and the run keeps the members still unread. Returns whether any + * are. + */ private refreshOccupant(state: SubjectState): boolean { - if (!state.occupant) return false; - if (this.host.isUnread(state.occupant.placement)) return true; + const occupant = state.occupant; + if (!occupant) return false; + const { before = [], ...newest } = occupant; + const members = [...before, newest]; + const unread = members.filter((member) => this.host.isUnread(member.placement)); + if (unread.length === members.length) return true; state.history = 'some'; - state.consumedEventId = state.occupant.eventId; - state.occupant = undefined; - return false; + state.consumedEventId = members.filter((member) => !unread.includes(member)).at(-1)!.eventId; + state.occupant = runOf(unread); + return unread.length > 0; + } + + /** + * Remove the occupant and the run it builds on, newest first, auditing each + * as `kind`. Every member is unread: refreshOccupant has just run, under the + * lock, and the host's removal doesn't itself refuse a read message. A + * member the host can't remove may be read, so the walk stops there, leaving + * it and the renders it builds on, and the history becomes `unknown`. + * Returns false then. It doesn't touch `state.occupant`, which still names + * the members it removed: each caller clears it, so what the walk left is + * history from then on, as an occupant whose removal failed always was. + */ + private removeOccupant(subject: string, state: SubjectState, kind: 'removed' | 'displaced'): boolean { + const occupant = state.occupant!; + for (const member of [occupant, ...[...(occupant.before ?? [])].reverse()]) { + this.host.audit({ kind, subject, eventId: member.eventId }); + if (!this.host.remove(member.placement)) { + // Still in context and possibly about to be read. + this.host.audit({ kind: 'remove-failed', subject, eventId: member.eventId }); + state.history = 'unknown'; + state.consumedEventId = member.eventId; + return false; + } + } + return true; } private scheduleRecovery(subject: string, state: SubjectState): void { @@ -592,14 +651,8 @@ export class PushCoalescer { } private async retract(subject: string, state: SubjectState, occurrence: CoalescedOccurrence, occupantUnread: boolean): Promise { - if (occupantUnread && state.occupant) { - this.host.audit({ kind: 'removed', subject, eventId: state.occupant.eventId }); - if (!this.host.remove(state.occupant.placement)) { - // Still in context and possibly about to be read: say so, append the notice. - this.host.audit({ kind: 'remove-failed', subject, eventId: state.occupant.eventId }); - state.history = 'unknown'; - } - } + // Whatever the walk leaves may be read, and the history it sets says so (§6). + if (occupantUnread && state.occupant) this.removeOccupant(subject, state, 'removed'); state.occupant = undefined; this.dropBatches(subject, state); this.host.cancelWake(subject); @@ -623,8 +676,8 @@ export class PushCoalescer { const hadBatch = this.dropBatches(subject, state); if (occupantUnread && state.occupant) { const prior = state.occupant; - this.host.audit({ kind: 'displaced', subject, eventId: prior.eventId }); - if (this.host.remove(prior.placement)) { + // A plain occurrence is a complete snapshot (rule 6): it displaces the run. + if (this.removeOccupant(subject, state, 'displaced')) { this.host.cancelWake(subject); // The replacement follows the prior delivery's audience (§3.2). const placement = await this.host.deliver({ ...occurrence, deliverTo: prior.placement.agent }); @@ -632,10 +685,7 @@ export class PushCoalescer { if (placement) state.audienceAgent = placement.agent; return 'replaced'; } - // Could not remove: the prior may still be read. Append, as for consumed. - this.host.audit({ kind: 'remove-failed', subject, eventId: prior.eventId }); - state.history = 'unknown'; - state.consumedEventId = prior.eventId; + // The walk stopped: what it left may be read. Append, as for consumed. state.occupant = undefined; } if (hadBatch) this.host.cancelWake(subject); @@ -648,7 +698,12 @@ export class PushCoalescer { private async acceptDeferred(subject: string, state: SubjectState, occurrence: CoalescedOccurrence, occupantUnread: boolean): Promise { let outcome: CoalesceOutcome = 'first'; - if (occupantUnread && state.occupant) { + if (occupantUnread && state.occupant?.rendered) { + // An unread render is a delta the next render builds on (§5.2), not a + // plain occurrence: it stays, and the batch follows its agent (§3.2), so + // the next render reaches the context that holds it (§3.3, vector 49). + occurrence = { ...occurrence, deliverTo: state.occupant.placement.agent }; + } else if (occupantUnread && state.occupant) { // §5.1: a notice displaces an unread plain occurrence and opens a batch // — in the audience that held the plain occurrence (§3.2). this.host.audit({ kind: 'displaced', subject, eventId: state.occupant.eventId }); @@ -855,6 +910,11 @@ export class PushCoalescer { const empty = isVisiblyEmptyContent(content); this.host.audit({ kind: 'rendered', subject, eventId: occurrence.eventId, source, empty }); if (empty) return; // §5.2: nothing happened (blank text included) + // A render describes what changed since the server's previous render + // (§5.2), so the renders before it stay where they are, and this one + // joins their run after them (§3.3, vector 49). Any a request has + // read by then, refreshOccupant drops at its next look. + const run = state.occupant?.rendered ? state.occupant : undefined; const placement = await this.host.deliver({ ...occurrence, timestamp, content }, content, assemblingFor); if (placement?.deferredId) { // Landed in another agent's deferred queue (its turn is alive): still @@ -864,12 +924,24 @@ export class PushCoalescer { // queue's agent's wake for this batch; give it an ordinary wake, as // any delivery during its turn has, beside any wakes a newer batch // has, so that its next turn reads the content. - state.occupant = { eventId: occurrence.eventId, placement }; + // The run's renders keep their own records, whatever else an + // occupant carries. + let before: Occupant['before']; + if (run) { const { before: earlier, ...member } = run; before = [...(earlier ?? []), member]; } + state.occupant = { eventId: occurrence.eventId, placement, rendered: true, ...(before ? { before } : {}) }; + // A batch opened while this render was under way (§5.4 rule 1) + // follows the run's agent from here, wakes included, as a later + // notice's does (acceptDeferred), so the run stays in one context. + if (state.batch && !state.batch.latest.deliverTo) { + state.batch.latest = { ...state.batch.latest, deliverTo: placement.agent }; + await this.host.wakeForBatch(state.batch.latest); + } state.audienceAgent = placement.agent; await this.host.wakeToRead(occurrence, placement.agent); return; } - // The materialized occurrence is consumed by the request being assembled. + // The materialized occurrence is consumed by the request being + // assembled, and becomes history with the run placed ahead of it. rendering.materialized = true; state.history = 'some'; state.consumedEventId = occurrence.eventId; @@ -971,7 +1043,9 @@ export class PushCoalescer { subjects: [...this.subjects].map(([subject, s]) => ({ subject, history: s.history, consumedEventId: s.consumedEventId, identity: s.identity, ...(s.audienceAgent ? { audienceAgent: s.audienceAgent } : {}), - ...(s.occupant ? { occupant: s.occupant } : {}), + // restore() reads only these. After a restart nothing of the subject + // is tracked as unread, a run included: its history counts as read. + ...(s.occupant ? { occupant: { eventId: s.occupant.eventId, placement: s.occupant.placement } } : {}), ...(s.batch ? { batch: { latest: s.batch.latest as CoalescedOccurrence, notices: s.batch.notices, dropped: s.batch.dropped, ...(s.batch.noRender ? { rendering: true } : {}) } } : s.rendering ? { batch: { latest: s.rendering.batch.latest as CoalescedOccurrence, notices: s.rendering.batch.notices, dropped: s.rendering.batch.dropped, rendering: true } } : {}), diff --git a/test/mcpl-coalescer-render-run.test.ts b/test/mcpl-coalescer-render-run.test.ts new file mode 100644 index 000000000..5e4838bae --- /dev/null +++ b/test/mcpl-coalescer-render-run.test.ts @@ -0,0 +1,260 @@ +// A render describes what changed since the server's previous render (RFC-006 +// §5.2), so a later render can't replace an earlier one still unread: the +// earlier one stays where it was placed, the later one lands after it, and a +// retraction or a plain occurrence takes the whole run, newest first (§3.3, +// vector 49; §6). #548 found the second render taking the subject's one slot +// over the first; a notice between two renders dropped the first as well. +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { PushCoalescer, type CoalescerHost, type CoalescedOccurrence, type CoalescingSnapshot } from '../src/mcpl/push-coalescer.js'; +import { TS } from './helpers/coalescing-fixture.js'; + +interface Options { + /** Placements the host fails to remove (CoalescerHost.remove returns false). */ + unremovable?: Set; +} + +/** A host whose every delivery waits, unread, in the deferred queue of + * 'other', an agent whose turn is alive, until it's removed, read, or gone, + * whichever agent assembled it. CoalescerHost leaves placement to the host, + * so this is one it allows; the framework stores the assembling agent's own + * render directly, which mcpl-coalescing's tests run. Its removal, like the + * framework's, doesn't refuse a message a request has read. Each render says + * which notices it covers, as a notice-only server's does. */ +function makeHost(options: Options = {}) { + const queue: Array<{ id: string; text: string }> = []; + const read = new Set(); + const removed: string[] = []; + const audits: string[] = []; + const wakes: string[] = []; + const deliveries: Array<{ eventId: string; deliverTo?: string }> = []; + const held = new Map void>(); + const hold = { n1: false }; + let renderStarted: (eventId: string) => void = () => {}; + let saved: () => CoalescingSnapshot = () => { throw new Error('nothing saved'); }; + const host: CoalescerHost = { + isUnread: (placement) => queue.some((m) => m.id === placement.deferredId) && !read.has(placement.deferredId ?? ''), + remove: (placement) => { + const id = placement.deferredId ?? ''; + if (options.unremovable?.has(id)) return false; + const at = queue.findIndex((m) => m.id === id); + if (at < 0) return false; + queue.splice(at, 1); + removed.push(id); + return true; + }, + deliver: async (occurrence, materialized) => { + const id = `deferred-${deliveries.length + 1}`; + deliveries.push({ eventId: occurrence.eventId, deliverTo: occurrence.deliverTo }); + queue.push({ id, text: (materialized ?? occurrence.content).map((block) => (block.type === 'text' ? block.text : '')).join('') }); + return { agent: occurrence.deliverTo ?? 'other', deferredId: id }; + }, + wakeForBatch: async (occurrence) => { wakes.push(`${occurrence.eventId}>${occurrence.deliverTo ?? 'routing'}`); }, + wakeToRead: async () => {}, cancelWake: () => {}, authorized: () => true, + audience: async (occurrence) => (occurrence.deliverTo ? [occurrence.deliverTo] : ['agent', 'other']), + render: async (occurrence, params) => { + renderStarted(occurrence.eventId); + await new Promise((resolve) => { if (occurrence.eventId === 'n1' && hold.n1) held.set('n1', resolve); else resolve(); }); + return { content: [{ type: 'text', text: `changes in ${params.notices.map((notice) => notice.eventId).join(', ')}` }] }; + }, + audit: (record) => { + if (['removed', 'displaced', 'remove-failed'].includes(String(record.kind))) audits.push(`${String(record.kind)}:${String(record.eventId)}`); + }, + save: (snapshot) => { saved = snapshot; }, + }; + return { + host, queue, read, removed, audits, wakes, deliveries, hold, + snapshot: () => saved(), + texts: () => queue.map((m) => m.text), + started: () => new Promise((resolve) => { renderStarted = resolve; }), + release: (eventId: string) => held.get(eventId)?.(), + /** A message gone from the queue without the coalescer, as a module's removal takes one. */ + gone: (id: string) => { queue.splice(queue.findIndex((m) => m.id === id), 1); }, + }; +} + +const occurrence = (eventId: string, flags: Partial> = {}): CoalescedOccurrence => ({ + serverId: 'editor', binding: 'binding', scope: { kind: 'featureSet', id: 'doc' }, key: 'document', eventId, timestamp: TS, + retract: false, deferred: false, initial: false, content: [{ type: 'text', text: eventId }], event: null, ...flags, +}); +const notice = (eventId: string, initial = false) => occurrence(eventId, { deferred: true, initial }); +const retraction = (text?: string) => occurrence('r1', { retract: true, content: text ? [{ type: 'text', text }] : [] }); + +/** #548's sequence: n1's batch renders for 'agent'; n2 arrives during that + * render and opens a batch of its own (§5.4 rule 1); n1's render lands + * unread in 'other''s queue, and n2's batch then follows 'other'; 'other''s + * next assembly renders it. */ +async function renderAfterRender(h: ReturnType, coalescer: PushCoalescer): Promise { + h.hold.n1 = true; + await coalescer.accept(notice('n1', true)); + const started = h.started(); + const first = coalescer.assemble('agent'); + await started; + await coalescer.accept(notice('n2')); + h.release('n1'); + await first; + await coalescer.assemble('other'); +} + +/** Three renders, each settling unread in 'other''s queue after a notice. */ +async function threeRenders(coalescer: PushCoalescer): Promise { + await coalescer.accept(notice('n1', true)); + await coalescer.assemble('agent'); + await coalescer.accept(notice('n2')); + await coalescer.assemble('other'); + await coalescer.accept(notice('n3')); + await coalescer.assemble('other'); +} + +test("a render settling while an earlier render is unread lands after it, and a retraction removes both, newest first (#548)", async () => { + const h = makeHost(); + const coalescer = new PushCoalescer(h.host); + await renderAfterRender(h, coalescer); + assert.deepEqual(h.texts(), ['changes in n1', 'changes in n2'], "n1's render stays, and n2's lands after it"); + const result = await coalescer.accept(retraction()); + assert.deepEqual(h.removed, ['deferred-2', 'deferred-1'], 'the retraction removes the run, newest first'); + assert.deepEqual(h.audits, ['removed:n2', 'removed:n1']); + assert.deepEqual(h.texts(), []); + assert.equal(result.coalesce?.outcome, 'retracted', 'nothing of the subject was read'); +}); + +test("a batch opened while a render was under way follows the run's agent once that render lands unread", async () => { + const h = makeHost(); + const coalescer = new PushCoalescer(h.host); + h.hold.n1 = true; + await coalescer.accept(notice('n1', true)); + const started = h.started(); + const first = coalescer.assemble('agent'); + await started; + await coalescer.accept(notice('n2')); + h.release('n1'); + await first; + assert.equal(h.wakes.at(-1), 'n2>other', "n2's batch's wakes follow it to 'other'"); + await coalescer.assemble('agent'); + assert.deepEqual(h.texts(), ['changes in n1'], "'agent' isn't in n2's batch's audience any more, so its assembly renders nothing"); + await coalescer.assemble('other'); + assert.deepEqual(h.texts(), ['changes in n1', 'changes in n2'], "'other''s assembly renders it, after the render it builds on"); + assert.equal(h.deliveries.at(-1)?.deliverTo, 'other'); +}); + +test("a notice after a render still unread leaves it in place, and its batch renders in that render's agent, after it", async () => { + const h = makeHost(); + const coalescer = new PushCoalescer(h.host); + await coalescer.accept(notice('n1', true)); + await coalescer.assemble('agent'); + assert.deepEqual(h.texts(), ['changes in n1']); + const result = await coalescer.accept(notice('n2')); + assert.equal(result.coalesce?.outcome, 'first', 'the notice opens a batch, displacing nothing'); + assert.equal(result.coalesce?.priorEventId, undefined, "n1's render was neither displaced nor found consumed, and nothing else of the subject was"); + assert.deepEqual(h.texts(), ['changes in n1'], "n1's render stays unread"); + assert.equal(h.wakes.at(-1), 'n2>other', 'the batch follows the agent whose context holds the run'); + await coalescer.assemble('other'); + assert.deepEqual(h.texts(), ['changes in n1', 'changes in n2'], "n2's render lands after the render it builds on"); + const withdrawn = await coalescer.accept(retraction()); + assert.deepEqual(h.removed, ['deferred-2', 'deferred-1']); + assert.equal(withdrawn.coalesce?.outcome, 'retracted'); +}); + +test('a notice over an unread render names, as its prior, an earlier render a request has read', async () => { + const h = makeHost(); + const coalescer = new PushCoalescer(h.host); + await renderAfterRender(h, coalescer); + h.read.add('deferred-1'); + const result = await coalescer.accept(notice('n3')); + assert.equal(result.coalesce?.priorEventId, 'n1', "n1's render was found consumed; n2's, still unread, isn't the prior"); + assert.deepEqual(h.removed, [], "n2's render stays"); +}); + +test("a plain occurrence displaces the whole run, newest first, and takes its place in the run's agent", async () => { + const h = makeHost(); + const coalescer = new PushCoalescer(h.host); + await renderAfterRender(h, coalescer); + const result = await coalescer.accept(occurrence('p1')); + assert.equal(result.coalesce?.outcome, 'replaced'); + assert.equal(result.coalesce?.priorEventId, 'n2'); + assert.deepEqual(h.removed, ['deferred-2', 'deferred-1'], 'a complete snapshot supersedes every render'); + assert.deepEqual(h.audits, ['displaced:n2', 'displaced:n1']); + assert.deepEqual(h.texts(), ['p1']); + assert.equal(h.deliveries.at(-1)?.deliverTo, 'other', 'it goes where the run was (§3.2)'); +}); + +test("a plain occurrence over a run whose newest render can't be removed is appended, as after a consumed version", async () => { + const h = makeHost({ unremovable: new Set(['deferred-2']) }); + const coalescer = new PushCoalescer(h.host); + await renderAfterRender(h, coalescer); + const result = await coalescer.accept(occurrence('p1')); + assert.equal(result.coalesce?.outcome, 'appended'); + assert.deepEqual(h.audits, ['displaced:n2', 'remove-failed:n2']); + assert.deepEqual(h.removed, [], "n1's render stays as the baseline of n2's, which may be read"); + assert.deepEqual(h.texts(), ['changes in n1', 'changes in n2', 'p1']); + assert.equal(h.deliveries.at(-1)?.deliverTo, undefined, 'appended as a fresh occurrence'); +}); + +test("a retraction stops at a render it can't remove, so that render keeps the renders it builds on, and the notice follows", async () => { + const h = makeHost({ unremovable: new Set(['deferred-2']) }); + const coalescer = new PushCoalescer(h.host); + await renderAfterRender(h, coalescer); + const result = await coalescer.accept(retraction('n1 and n2 withdrawn')); + assert.deepEqual(h.removed, [], "n1's render stays as n2's baseline"); + assert.deepEqual(h.texts(), ['changes in n1', 'changes in n2', 'n1 and n2 withdrawn']); + assert.equal(result.coalesce?.outcome, 'noted'); +}); + +test("a retraction that can't remove an earlier render of the run appends its notice", async () => { + const h = makeHost({ unremovable: new Set(['deferred-1']) }); + const coalescer = new PushCoalescer(h.host); + await renderAfterRender(h, coalescer); + const result = await coalescer.accept(retraction('withdrawn')); + assert.deepEqual(h.removed, ['deferred-2']); + assert.deepEqual(h.audits, ['removed:n2', 'removed:n1', 'remove-failed:n1']); + assert.deepEqual(h.texts(), ['changes in n1', 'withdrawn']); + assert.equal(result.coalesce?.outcome, 'noted', "n1's render may be read, so the notice follows it"); +}); + +test('an earlier render of the run that a request read is history: a retraction removes only what is unread, and appends its notice', async () => { + const h = makeHost(); + const coalescer = new PushCoalescer(h.host); + await renderAfterRender(h, coalescer); + h.read.add('deferred-1'); + const result = await coalescer.accept(retraction('withdrawn')); + assert.deepEqual(h.removed, ['deferred-2'], 'the render the model read stays'); + assert.deepEqual(h.texts(), ['changes in n1', 'withdrawn']); + assert.equal(result.coalesce?.outcome, 'noted'); +}); + +test('a run of three renders stays in placement order, and a retraction removes it newest first', async () => { + const h = makeHost(); + const coalescer = new PushCoalescer(h.host); + await threeRenders(coalescer); + assert.deepEqual(h.texts(), ['changes in n1', 'changes in n2', 'changes in n3']); + const stored = h.snapshot().subjects.find((s) => s.occupant)?.occupant; + assert.deepEqual(stored, { eventId: 'n3', placement: { agent: 'other', deferredId: 'deferred-3' } }, 'a snapshot keeps the occupant as restore() reads it, without its run'); + const result = await coalescer.accept(retraction()); + assert.deepEqual(h.removed, ['deferred-3', 'deferred-2', 'deferred-1']); + assert.equal(result.coalesce?.outcome, 'retracted'); +}); + +test("a render of the run that's gone without the coalescer leaves the rest of the run tracked", async () => { + const h = makeHost(); + const coalescer = new PushCoalescer(h.host); + await threeRenders(coalescer); + h.gone('deferred-2'); + const result = await coalescer.accept(retraction('withdrawn')); + assert.deepEqual(h.removed, ['deferred-3', 'deferred-1'], 'the renders around the gone one are still removed'); + assert.deepEqual(h.texts(), ['withdrawn']); + assert.equal(result.coalesce?.outcome, 'noted', 'a gone render may have been read'); +}); + +test("a render left unread after a later one was read stays a render: a notice leaves it in place", async () => { + const h = makeHost(); + const coalescer = new PushCoalescer(h.host); + await renderAfterRender(h, coalescer); + h.read.add('deferred-2'); + const result = await coalescer.accept(notice('n3')); + assert.equal(result.coalesce?.outcome, 'first'); + assert.deepEqual(h.removed, [], "n1's render, still unread, isn't displaced as if it were a plain occurrence"); + assert.deepEqual(h.texts(), ['changes in n1', 'changes in n2']); + const withdrawn = await coalescer.accept(retraction('withdrawn')); + assert.deepEqual(h.removed, ['deferred-1'], "n1's render is still tracked, so the retraction removes it"); + assert.equal(withdrawn.coalesce?.outcome, 'noted', "n2's render was read"); +}); diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index 6a5c77561..2805142c2 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -1232,7 +1232,7 @@ test("a render placed in a live turn's deferred queue isn't read by the resident assert(f.lastRequest().includes('document_diff'), "the primary's turn read the render once its queue was flushed"); }); -test("a render placed in a live turn's deferred queue leaves the wakes of a newer batch of its subject as they are", async (t) => { +test("a render placed in a live turn's deferred queue takes a newer batch of its subject, and its wakes, to its agent", async (t) => { const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); f.alwaysRespond(); let release!: (value: unknown) => void; @@ -1248,7 +1248,8 @@ test("a render placed in a live turn's deferred queue leaves the wakes of a newe const internals = f.framework as unknown as TurnInternals; // As above, the primary's turn is alive and 'other''s assembly renders the // batch, here held. A notice during the render opens a newer batch of the - // subject, which wakes its audience. + // subject, which wakes its audience. Once the render lands in the + // primary's queue, the newer batch follows the primary, wakes included. internals.activeTurnTokens.set('agent', -1); try { const pass = internals.processInferenceRequests(); @@ -1258,8 +1259,8 @@ test("a render placed in a live turn's deferred queue leaves the wakes of a newe assert.deepEqual(woken, ['agent', 'other']); release({ content: [{ type: 'text', text: 'first_render' }] }); await settles(pass, 'the scheduler pass'); - assert.deepEqual(internals.pendingRequests.map((r) => r.agentName).sort(), [...woken, 'agent'].sort(), - "the newer batch's wakes are as they were, beside the primary's wake for the render"); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent', 'agent'], + "the newer batch's wake is the primary's now, beside the primary's wake for the render"); } finally { endStandInTurn(internals); } @@ -1301,7 +1302,33 @@ test("a render placed in a live turn's deferred queue wakes its agent though a n assert(f.lastRequest().includes('first_render'), "the primary's turn read the render"); }); -test("a render placed in a live turn's deferred queue keeps its agent's wake when another agent's assembly freezes a newer batch of its subject that gives nothing", async (t) => { +test("a notice after a render landed unread in a live turn's deferred queue leaves it there, and the turn reads it before the newer render (#548)", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + let renders = 0; + f.renderer(() => Promise.resolve({ content: [{ type: 'text', text: ++renders > 1 ? 'second_render' : 'first_render' }] })); + await f.send('push/event', f.params('1', 'first_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // The primary's turn is alive, so 'other''s assembly places the render in + // the primary's deferred queue, unread. A notice then arrives. The render + // is a delta the next one builds on (RFC-006 §5.2), so it stays, and the + // notice's batch follows the primary. + internals.activeTurnTokens.set('agent', -1); + try { + await settles(internals.processInferenceRequests(), 'the scheduler pass'); + assert.equal(renders, 1, "'other''s assembly rendered the first batch"); + await f.send('push/event', f.params('2', 'second_fallback', { deferred: true })); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent', 'agent'], "the render's wake and the newer batch's, both the primary's"); + } finally { + endStandInTurn(internals); + } + await settles(f.framework.runUntilIdle(), 'the drain'); + assert.equal(f.membrane.calls.length, 1); + const request = f.lastRequest(); + assert(request.includes('first_render') && request.indexOf('first_render') < request.indexOf('second_render'), 'one call read the first render, then the second'); +}); + +test("a render placed in a live turn's deferred queue takes a newer batch of its subject to its agent, so another agent's assembly leaves that batch and the render's wake alone", async (t) => { const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); f.alwaysRespond(); const withdrawn = withdrawals(f); @@ -1318,9 +1345,11 @@ test("a render placed in a live turn's deferred queue keeps its agent's wake whe const internals = f.framework as unknown as TurnInternals; // As above: the primary's turn is alive, 'other''s assembly renders the // batch into the primary's queue, and a notice during the render opens a - // newer batch, which wakes both. In the next pass the primary is still - // busy, and 'other''s assembly freezes the newer batch, whose render is - // empty: the freeze withdraws the subject's wakes, and nothing is placed. + // newer batch, which wakes both. Once the render lands, the newer batch + // follows the primary, whose queue holds the render it builds on (RFC-006 + // §5.2), wakes included. In the next pass the primary is still busy, and + // 'other' has no wake left for the batch: the batch waits, with its wake, + // for the primary. internals.activeTurnTokens.set('agent', -1); try { const pass = internals.processInferenceRequests(); @@ -1329,12 +1358,14 @@ test("a render placed in a live turn's deferred queue keeps its agent's wake whe release({ content: [{ type: 'text', text: 'first_render' }] }); await settles(pass, 'the first pass'); await settles(internals.processInferenceRequests(), 'the second pass'); - assert.deepEqual(withdrawn, ['other', 'other']); - assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent'], "the primary's wake for the render outlived the newer batch's freeze"); + assert.deepEqual(withdrawn, ['other'], "only 'other''s turn that rendered the first batch, which gave it nothing"); + assert.equal(renders, 1, "'other''s assembly didn't render the newer batch"); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent', 'agent'], "the primary's wakes for the render and for the newer batch both wait for its turn"); } finally { endStandInTurn(internals); } await settles(f.framework.runUntilIdle(), "the primary's turn"); + assert.equal(renders, 2, "the primary's assembly rendered the newer batch, which gave nothing"); assert.equal(f.membrane.calls.length, 1); assert(f.lastRequest().includes('first_render'), "the primary's turn read the render"); });