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