diff --git a/changelog.d/publish-waits-for-new-registration.fixed.md b/changelog.d/publish-waits-for-new-registration.fixed.md new file mode 100644 index 000000000..619dfafb9 --- /dev/null +++ b/changelog.d/publish-waits-for-new-registration.fixed.md @@ -0,0 +1,23 @@ +- Words for a channel that its server registered again or retargeted after + the publish's last wait, as it can while the channel's opening is + announced, were published before anything had confirmed the new + registration or target open. They now wait for that queued lifecycle + work, as they wait for any channel's, and go out only into the channel it + leaves open. If it leaves none open, nothing is published, and the failure + reads as at the first wait: "lifecycle work prevented delivery: …". A + close decided in that window, where the host directs the server's + lifecycle, now stands, as one decided while the words wait already does. +- A `channels.publish` grant withdrawn while the words wait now holds for + them too: the check just before the publish reads the grant as it is + then, and refuses as the check before the waits does. +- On a server whose channel lifecycle the host doesn't direct (no + `channels.lifecycle` grant), a channel decided closed no longer refuses + words that waited for queued lifecycle work: the host can't close it, so + they go into its open transport, as they already did when nothing was + queued. +- Words for a channel whose registration is no longer marked open, with no + lifecycle work queued for it, are now refused where they used to be + published. That happens, for example, when a server answers an open of + the channel for another one. On a server without `channels.lifecycle`, + words for a retargeted channel are refused the same way, unless the + server's traffic has confirmed the new target open first. diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index 4e558591f..a15ceebfb 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -855,6 +855,15 @@ export class ChannelRegistry { return CapabilityGrant.of(this.serverRegistry.getServer(serverId)).has('channels.lifecycle'); } + /** Whether a decided close holds words back from a channel: only where + * the host directs the server's lifecycle. Without that grant the host + * can't close the channel, and words go into a transport that is open, + * as the open step's engagesDecidedClosed reads it too. */ + private closeStands(entry: ChannelEntry): boolean { + return this.getDesiredState(entry.serverId, entry.descriptor.id) === 'closed' && + this.directsLifecycle(entry.serverId); + } + /** A registration's entry: nothing about its transport is known until a * receipt for its target confirms it open or closed. */ private registerEntry(key: string, serverId: string, descriptor: ChannelDescriptor): void { @@ -2663,8 +2672,9 @@ export class ChannelRegistry { /** * Join outstanding lifecycle work without recording new intent. Retargeting * may have invalidated open confirmation while the decision remains pending. - * A close that settles during the wait remains authoritative; waiting for - * it is not an instruction to open the channel again. + * A close that settles during the wait remains authoritative where the host + * directs the server's lifecycle (closeStands); waiting for it is not an + * instruction to open the channel again. */ private async waitForOpenChannel(entry: ChannelEntry): Promise { const lifecycleKey = this.lifecycleKey(entry.serverId, entry.descriptor.id); @@ -2686,7 +2696,7 @@ export class ChannelRegistry { } const current = this.channels.get(`${entry.serverId}:${entry.descriptor.id}`); if (!current) throw new Error('Channel registration was removed while delivery waited'); - if (!current.open || this.getDesiredState(current.serverId, current.descriptor.id) === 'closed') { + if (!current.open || this.closeStands(current)) { throw new Error('Channel is not confirmed open for delivery, or was selected for closure while delivery waited'); } return current; @@ -3733,16 +3743,21 @@ export class ChannelRegistry { * may declare the channel inbound-only meanwhile, or a removal take it * away. So the check just before the publish, after every await, reads * the channel's registration as it is then, and refuses one that is gone - * or inbound-only. A removal during a wait already fails it there - * (waitForOpenChannel); the check also catches one made after the last - * wait, as the opening's notice can. Words refused after an open for a - * declaration leave the channel open, and the agent gets the failure for - * the words. An open by delivery (`openSource` 'opened-by-delivery', the - * default) that turned a closed channel open also gives the notice that - * it's open, which says how to close it. One that only - * reconfirmed the transport of a channel already decided open or tuned out - * changes no decision, and gives none. When the registration is gone, the - * failure says the channel was removed. + * or inbound-only, or whose server's grant no longer has channels.publish. + * A removal during a wait already fails it there (waitForOpenChannel); the + * check also catches one made after the last wait, as the opening's notice + * can. A registration nothing has confirmed open, such as one that + * replaced the one waited for or a retarget of it, or one whose close now + * stands (where the host directs the server's lifecycle), is treated as at + * the first wait: the words wait for its queued lifecycle work, and the + * check runs again on the registration that work leaves. Words refused + * after an open for a declaration leave the channel open, and the agent + * gets the failure for the words. An open by delivery (`openSource` + * 'opened-by-delivery', the default) that turned a closed channel open + * also gives the notice that it's open, which says how to close it. + * One that only reconfirmed the transport of a channel already decided + * open or tuned out changes no decision, and gives none. When the + * registration is gone, the failure says the channel was removed. * * Only the connector's `delivered: true` confirms a post. `failed` means * nothing was posted: the attempt stopped before the request was written, @@ -3787,15 +3802,21 @@ export class ChannelRegistry { return { status: 'failed', destination, reason: `server "${entry.serverId}" not found`, at: at() }; } // §14.1: channels/publish requires channels.publish in the grant. The - // host not sending is the enforcement. - if (!CapabilityGrant.of(server).has('channels.publish')) { - return { - status: 'failed', - destination, - reason: `channels.publish not in "${entry.serverId}"'s effective grant (§14.1)`, - at: at(), - }; - } + // host not sending is the enforcement, here and at the last check. + const ungranted = (): PublishOutcome => ({ + status: 'failed', + destination, + reason: `channels.publish not in "${entry.serverId}"'s effective grant (§14.1)`, + at: at(), + }); + if (!CapabilityGrant.of(server).has('channels.publish')) return ungranted(); + + const prevented = (err: unknown): PublishOutcome => ({ + status: 'failed', + destination, + reason: `lifecycle work prevented delivery: ${connectorText(err instanceof Error ? err.message : String(err))}`, + at: at(), + }); // Lifecycle work already queued for the channel settles first: words go // out only into the channel it leaves, and fail visibly if it leaves @@ -3804,12 +3825,7 @@ export class ChannelRegistry { try { entry = await this.waitForOpenChannel(entry); } catch (err) { - return { - status: 'failed', - destination, - reason: `lifecycle work prevented delivery: ${connectorText(err instanceof Error ? err.message : String(err))}`, - at: at(), - }; + return prevented(err); } } @@ -3867,14 +3883,35 @@ export class ChannelRegistry { // open's or any other: a registration that is gone, or inbound-only, at // this moment takes nothing. The entry is read from the map rather than // taken from the last wait: the opening's notice runs in between, and - // publish() resumes a microtask after the wait. - const current = this.channels.get(`${entry.serverId}:${entry.descriptor.id}`); + // publish() resumes a microtask after the wait. A registration nothing + // has confirmed open, such as one that replaced the one waited for + // (registered with open: false) or a retarget of it (marked unconfirmed), + // or one whose close now stands (closeStands), is treated as at the + // first wait: the words wait for its queued lifecycle work, and + // waitForOpenChannel refuses what that work leaves unconfirmed or with + // its close standing. The loop opens nothing itself. The check then runs + // again on the registration that stands. waitForOpenChannel returns + // only a registration it found open and whose close doesn't stand, or + // throws, so the loop ends unless the channel keeps changing. + const key = `${entry.serverId}:${entry.descriptor.id}`; + let current = this.channels.get(key); + while (current && !isInboundOnly(current) && (!current.open || this.closeStands(current))) { + try { + entry = await this.waitForOpenChannel(current); + } catch (err) { + return prevented(err); + } + current = this.channels.get(key); + } if (!current) { return { status: 'failed', destination, reason: 'the channel was removed before the words were published', at: at() }; } if (isInboundOnly(current)) { return { status: 'failed', destination, reason: inboundOnlyReason(current), at: at() }; } + // A grant reduced during any of the waits holds at once (§6.7): the + // words already on their way take it too. + if (!CapabilityGrant.of(server).has('channels.publish')) return ungranted(); let result: unknown; try { diff --git a/test/channel-lifecycle-order.test.ts b/test/channel-lifecycle-order.test.ts index 3cd3b782a..bab5fbe42 100644 --- a/test/channel-lifecycle-order.test.ts +++ b/test/channel-lifecycle-order.test.ts @@ -317,6 +317,7 @@ test('a draft published into a channel with a close pending waits for it and pos const outcome = await draft; assert.equal(outcome.status, 'failed'); assert.match(outcome.reason ?? '', /^lifecycle work prevented delivery: /); + assert.deepEqual(outcome.destination, { serverId: 'test', channelId: 'x', label: 'x' }); assert.equal(f.registry.getDesiredState('test', 'x'), 'closed'); assert.equal(f.isOpen(), false); assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published'); @@ -369,6 +370,234 @@ for (const change of ['announced again as inbound-only', 'removed'] as const) { }); } +for (const answer of ['opens', 'fails'] as const) { + test(`a channel registered again while its opening is announced takes the words only once its new registration's open ${answer === 'opens' ? 'succeeds' : 'fails, and then not at all'}`, async () => { + // A channels/changed during the opening's notice can replace the + // channel's entry with a fresh registration, which nothing has confirmed + // open: its reconcile queues the open. The words wait for that work and + // go out only into the channel it leaves open. + let announced: Promise | undefined; + let f!: ReturnType; + f = fixture(() => { + f.held.add('x'); + announced = f.registry.handleChanged('test', { added: [descriptor('x', false, 'replacement')] }); + }); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + const speech = f.registry.routeSpeech('resident', 'hello', 'x'); + for (let i = 0; i < 5; i++) await tick(); + assert.equal(f.notices.length, 1, 'the opening was announced'); + assert.equal(f.pending.length, 1, "the new registration's open is in flight"); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing is published before that open is answered'); + if (answer === 'opens') f.pending.pop()!.finish(); + else f.pending.pop()!.fail(); + const outcome = await speech; + assert.deepEqual(f.calls.filter((c) => c.kind === 'open').map((c) => c.address), [{ value: 'original' }, { value: 'replacement' }]); + if (answer === 'opens') { + assert.equal(outcome?.delivered, true); + assert.deepEqual(f.publishedWhileOpen, [true]); + assert.deepEqual(f.routeFailures, []); + } else { + assert.equal(outcome, null); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published'); + assert.match(f.routeFailures[0]?.reason ?? '', /^lifecycle work prevented delivery: /); + } + await announced; + }); +} + +test('a channel retargeted while its opening is announced takes the words only once its new target is open', async () => { + // A retarget keeps the channel's registration but marks it unconfirmed, + // with an open of the new target queued, so the words wait for that open + // as they do for a new registration's. + let announced: Promise | undefined; + let f!: ReturnType; + f = fixture(() => { + f.held.add('x'); + announced = f.registry.handleChanged('test', { updated: [descriptor('x', false, 'retargeted')] }); + }); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + const speech = f.registry.routeSpeech('resident', 'hello', 'x'); + for (let i = 0; i < 5; i++) await tick(); + assert.equal(f.pending.length, 1, "the new target's open is in flight"); + assert.equal(f.isOpen('x'), false); + // The fixture's server keeps the channel open by id through a retarget, + // so this counts the publishes rather than reading publishedWhileOpen's values. + assert.equal(f.publishedWhileOpen.length, 0, "nothing is published before the new target's open is answered"); + f.pending.pop()!.finish(); + assert.equal((await speech)?.delivered, true); + assert.deepEqual(f.calls.filter((c) => c.kind === 'open').map((c) => c.address), [{ value: 'original' }, { value: 'retargeted' }]); + assert.equal(f.publishedWhileOpen.length, 1); + await announced; +}); + +test('a channel registered again while the words wait for a new registration takes them only once the newest is open', async () => { + // The server registers x again while the words wait for the first new + // registration's open. The words wait through the newer one's open too, + // and the check after the wait reads the registration that stands. + let announced: Promise | undefined; + let f!: ReturnType; + f = fixture(() => { + f.held.add('x'); + announced = f.registry.handleChanged('test', { added: [descriptor('x', false, 'second')] }); + }); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + const speech = f.registry.routeSpeech('resident', 'hello', 'x'); + for (let i = 0; i < 5; i++) await tick(); + assert.equal(f.pending.length, 1, "the second registration's open is in flight"); + const again = f.registry.handleChanged('test', { added: [descriptor('x', false, 'third')] }); + f.pending.pop()!.finish(); + for (let i = 0; i < 5; i++) await tick(); + assert.ok(f.pending.length >= 1, "the third registration's lifecycle work is in flight"); + assert.equal(f.publishedWhileOpen.length, 0, "nothing is published while it's in flight"); + await f.drain(speech, announced!, again); + assert.equal((await speech)?.delivered, true); + const opens = f.calls.filter((c) => c.kind === 'open').map((c) => (c.address as { value: string }).value); + assert.equal(opens.at(-1), 'third', 'the last open is the newest registration\'s'); + assert.equal(f.publishedWhileOpen.length, 1); +}); + +test('a grant without channels.publish by the time the words would be published refuses them', async () => { + // A grant reduced during the opening's notice holds at once (§6.7): the + // last check reads the grant as it is then. + let f!: ReturnType; + f = fixture(() => { f.setGrant(ALL_CAPABILITY_PATHS.filter((path) => path !== 'channels.publish')); }); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + const outcome = await f.registry.publish('resident', 'words', { serverId: 'test', channelId: 'x' }); + assert.equal(f.notices.length, 1, 'the opening was announced'); + assert.equal(outcome.status, 'failed'); + assert.equal(outcome.reason, 'channels.publish not in "test"\'s effective grant (§14.1)'); + assert.deepEqual(outcome.destination, { serverId: 'test', channelId: 'x', label: 'x' }); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published'); +}); + +test('a grant without channels.publish after the words wait for a new registration refuses them', async () => { + // The grant is read after the wait the last check runs, not only before it. + let announced: Promise | undefined; + let f!: ReturnType; + f = fixture(() => { + f.held.add('x'); + announced = f.registry.handleChanged('test', { added: [descriptor('x', false, 'replacement')] }); + }); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + const draft = f.registry.publish('resident', 'words', { serverId: 'test', channelId: 'x' }); + for (let i = 0; i < 5; i++) await tick(); + assert.equal(f.pending.length, 1, "the new registration's open is in flight"); + f.setGrant(ALL_CAPABILITY_PATHS.filter((path) => path !== 'channels.publish')); + f.pending.pop()!.finish(); + const outcome = await draft; + assert.equal(outcome.status, 'failed'); + assert.equal(outcome.reason, 'channels.publish not in "test"\'s effective grant (§14.1)'); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published'); + await announced; +}); + +test('a close decided while its opening is announced refuses the words, as a close decided while they wait does', async () => { + // The registration is the one the waits returned and still open, but a + // close is now decided and queued: the words wait for it and are refused. + let closing: Promise | undefined; + let f!: ReturnType; + f = fixture(() => { + f.held.add('x'); + closing = f.tool('close'); + }); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + const speech = f.registry.routeSpeech('resident', 'hello', 'x'); + for (let i = 0; i < 5; i++) await tick(); + assert.equal(f.registry.getDesiredState('test', 'x'), 'closed'); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing is published while the close is queued'); + await f.drain(closing!, speech); + assert.equal(await speech, null); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published'); + assert.match(f.routeFailures[0]?.reason ?? '', /^lifecycle work prevented delivery: /); + assert.equal(f.isOpen(), false, 'the close the agent decided stands'); +}); + +test('a retarget with nothing queued for it refuses the words, rather than publishing into it unconfirmed', async () => { + // Without channels.lifecycle the host queues no open for a retarget, so + // nothing will confirm the new target: the words are refused, not sent. + let announced: Promise | undefined; + let f!: ReturnType; + f = fixture(() => { + f.setGrant(ALL_CAPABILITY_PATHS.filter((path) => path !== 'channels.lifecycle')); + announced = f.registry.handleChanged('test', { updated: [descriptor('x', false, 'retargeted')] }); + }); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + const outcome = await f.registry.publish('resident', 'words', { serverId: 'test', channelId: 'x' }); + assert.equal(outcome.status, 'failed'); + assert.match(outcome.reason ?? '', /^lifecycle work prevented delivery: /); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published'); + assert.deepEqual(f.calls.filter((c) => c.kind === 'open').map((c) => c.address), [{ value: 'original' }], 'no open was queued for the new target'); + await announced; +}); + +test('a channel registered again and confirmed open by its traffic, once the host no longer directs its lifecycle, takes the words though decided closed', async () => { + // Without channels.lifecycle a decided close doesn't hold words back from + // a transport that is open, whichever registration carries it: the new + // registration's traffic confirms it open, and the words go out. + let closing: Promise<{ success: boolean }> | undefined; + let announced: Promise | undefined; + let traffic: Promise | undefined; + let f!: ReturnType; + f = fixture(() => { + f.setGrant(ALL_CAPABILITY_PATHS.filter((path) => path !== 'channels.lifecycle')); + closing = f.tool('close'); + announced = f.registry.handleChanged('test', { added: [descriptor('x', false, 'replacement')] }); + traffic = f.registry.handleIncoming('test', { messages: [message()] }); + }); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + const outcome = await f.registry.publish('resident', 'words', { serverId: 'test', channelId: 'x' }); + assert.equal((await closing!).success, false, 'the close is refused for the grant'); + assert.equal(f.registry.getDesiredState('test', 'x'), 'closed', 'the refused close is still recorded as a decision'); + assert.equal(outcome.status, 'delivered'); + assert.deepEqual(f.publishedWhileOpen, [true]); + await announced; + await traffic; +}); + +test("a close decided while the words wait, once the host no longer directs the channel's lifecycle, leaves them going into the open transport", async () => { + // waitForOpenChannel takes a decided close as standing only where the host + // directs the server's lifecycle, as the last check does. + let announced: Promise | undefined; + let f!: ReturnType; + f = fixture(() => { + f.held.add('x'); + announced = f.registry.handleChanged('test', { added: [descriptor('x', false, 'replacement')] }); + }); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + const draft = f.registry.publish('resident', 'words', { serverId: 'test', channelId: 'x' }); + for (let i = 0; i < 5; i++) await tick(); + assert.equal(f.pending.length, 1, "the new registration's open is in flight"); + f.setGrant(ALL_CAPABILITY_PATHS.filter((path) => path !== 'channels.lifecycle')); + const closing = await f.tool('close'); + assert.equal(closing.success, false, 'the close is refused for the grant'); + f.pending.pop()!.finish(); + const outcome = await draft; + assert.equal(f.registry.getDesiredState('test', 'x'), 'closed', 'the refused close is still recorded as a decision'); + assert.equal(outcome.status, 'delivered'); + assert.deepEqual(f.publishedWhileOpen, [true]); + await announced; +}); + +test('a channel registered again as inbound-only while its opening is announced is refused without waiting for its new open', async () => { + // Nothing would be published into it whatever its open does, so the words + // don't wait for that open: they're refused for the declaration at once. + let announced: Promise | undefined; + let f!: ReturnType; + f = fixture(() => { + f.held.add('x'); + announced = f.registry.handleChanged('test', { added: [{ ...descriptor('x', false, 'replacement'), direction: 'inbound' }] }); + }); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + const speech = f.registry.routeSpeech('resident', 'hello', 'x'); + for (let i = 0; i < 5; i++) await tick(); + assert.equal(f.pending.length, 1, "the new registration's open is in flight"); + assert.match(f.routeFailures[0]?.reason ?? '', /^inbound-only: server "test" declared this channel's direction "inbound"/); + assert.equal(await speech, null); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published'); + f.pending.pop()!.finish(); + await announced; +}); + for (const opener of ['tool', 'reconcile'] as const) { test(`an open answered after the agent closed the channel during it leaves it closed, and its inbound-only declaration applies (${opener})`, async () => { // The close is the newer decision, so the channel ends closed. The answer