Stop addressing subscription events to a request id

A subscribe arrives as a request, so the source address the subscription
controller handles it with carries the request id used to route the
response back to the caller. That address was stored verbatim as the
destination for the subscription, so every event broadcast to it, and the
replay event sent when the subscription is created, went out addressed to
a request that had long since been answered.

A request id identifies one request in flight and has no meaning on an
event. The receiving link keys response routing off it, so events carrying
one are put through paths meant for response traffic.

Strip the request id when taking the subscription's destination from the
request that created it. Add Address.withoutRequestId for that, as the
same address is used in three places here.

The subscription map is keyed on addressIndex, which already excludes the
request id, so subscriptions were never duplicated by this and only the
address events are sent to was wrong.

Fixes #847

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/lib/src/data/messages_core.ts

@@ -114,6 +114,24 @@ export class Address {
 		return `${this.type}:${this.id}` as const;
 	}
 
+	/**
+	 * Returns this address with any request id removed
+	 *
+	 * A request id identifies one request in flight and is set on the
+	 * source address of a request so that its response can be matched back
+	 * up with it.  It is meaningless on anything else, so an address taken
+	 * from a request has to be stripped of it before it is used as the
+	 * destination of an event.
+	 *
+	 * @returns this address without a request id.
+	 */
+	withoutRequestId() {
+		if (this.requestId === undefined) {
+			return this;
+		}
+		return new Address(this.type, this.id);
+	}
+
 	/**
 	 * Returns true if this address targets the given destination
 	 * @param dst - Destination to check

packages/lib/src/subscriptions.ts

@@ -345,6 +345,9 @@ export class SubscriptionController {
 		}
 		const addressIndex = src.addressIndex();
 		const subscriber = eventData.subscriptions.get(addressIndex);
+		// src is the source of a request and carries its request id, which
+		// has no meaning on the events sent back to the subscriber.
+		const subscriberDst = src.withoutRequestId();
 		switch (request.action) {
 			case "unsubscribe":
 				if (!subscriber) {
@@ -358,7 +361,9 @@ export class SubscriptionController {
 
 			case "subscribe":
 				if (!subscriber) {
-					eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters });
+					eventData.subscriptions.set(
+						addressIndex, { link: link, dst: subscriberDst, filters: request.filters }
+					);
 				} else {
 					subscriber.filters.union(request.filters);
 				}
@@ -371,7 +376,9 @@ export class SubscriptionController {
 					}
 					eventData.subscriptions.delete(addressIndex);
 				} else if (!subscriber) {
-					eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters });
+					eventData.subscriptions.set(
+						addressIndex, { link: link, dst: subscriberDst, filters: request.filters }
+					);
 				} else {
 					subscriber.filters = request.filters;
 				}
@@ -384,7 +391,7 @@ export class SubscriptionController {
 		if (eventData.subscriptionUpdate) {
 			const eventReplay = await eventData.subscriptionUpdate(request, src, dst);
 			if (eventReplay) {
-				link.sendTo(src, eventReplay);
+				link.sendTo(subscriberDst, eventReplay);
 				return true;
 			}
 		}

test/lib/subscriptions.test.js

@@ -520,6 +520,72 @@ describe("lib/subscriptions", function() {
 			});
 		});
 
+		describe("subscriber addressing", function() {
+			// A subscribe arrives as a request, so the source address it is
+			// handled with carries the request id used to route the response
+			// back to the caller. That id identifies one request in flight and
+			// means nothing on an event, see #847. These use a controller of
+			// their own because a subscription is only ever addressed when it
+			// is first created, and the shared one is already subscribed.
+			const connectorData = connectorSetupDate[0];
+
+			function requestSource(requestId) {
+				return new lib.Address(connectorData.dst.type, connectorData.dst.id, requestId);
+			}
+
+			function lastSentMessage() {
+				return getLink(connectorData.id).connector.sentMessages.at(-1);
+			}
+
+			it("should strip the request id from broadcast events", async function() {
+				const ownSubscriptions = new lib.SubscriptionController(mockController);
+				ownSubscriptions.handle(RegisteredEvent);
+				await ownSubscriptions.handleRequest(
+					getLink(connectorData.id),
+					new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"),
+					requestSource(213), connectorData.src
+				);
+
+				ownSubscriptions.broadcast(new RegisteredEvent());
+				await onceConnectorSend(connectorData.id);
+
+				assert.equal(lastSentMessage().name, RegisteredEvent.name);
+				assert.equal(lastSentMessage().dst.requestId, undefined, "event addressed to a request id");
+				assert.deepEqual(lastSentMessage().dst.toJSON(), connectorData.dst.toJSON());
+			});
+
+			it("should strip the request id from the replayed event", async function() {
+				const ownSubscriptions = new lib.SubscriptionController(mockController);
+				ownSubscriptions.handle(RegisteredEvent, async () => new RegisteredEvent());
+				await ownSubscriptions.handleRequest(
+					getLink(connectorData.id),
+					new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"),
+					requestSource(417), connectorData.src
+				);
+
+				assert.equal(lastSentMessage().name, RegisteredEvent.name);
+				assert.equal(lastSentMessage().dst.requestId, undefined, "replay addressed to a request id");
+				assert.deepEqual(lastSentMessage().dst.toJSON(), connectorData.dst.toJSON());
+			});
+
+			it("should strip the request id when replacing filters creates the subscription", async function() {
+				const ownSubscriptions = new lib.SubscriptionController(mockController);
+				ownSubscriptions.handle(RegisteredEvent);
+				await ownSubscriptions.handleRequest(
+					getLink(connectorData.id),
+					new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "foo"),
+					requestSource(918), connectorData.src
+				);
+
+				ownSubscriptions.broadcast(new RegisteredEvent(), "foo");
+				await onceConnectorSend(connectorData.id);
+
+				assert.equal(lastSentMessage().name, RegisteredEvent.name);
+				assert.equal(lastSentMessage().dst.requestId, undefined, "event addressed to a request id");
+				assert.deepEqual(lastSentMessage().dst.toJSON(), connectorData.dst.toJSON());
+			});
+		});
+
 		describe(".unsubscribeLink() / .unsubscribeAddress()", function() {
 			beforeEach(function() {
 				subscriptions.handle(RegisteredEvent);