feat(client-presence): Create Acknowledgment message interface and ackRequested flow

packages/framework/presence/src/internalTypes.ts

@@ -34,7 +34,9 @@ export type IEphemeralRuntime = Pick<
 	(IContainerRuntime & IRuntimeInternal) | IFluidDataStoreRuntime,
 	"clientId" | "connected" | "getAudience" | "getQuorum" | "off" | "on" | "submitSignal"
 > &
-	Partial<Pick<IFluidDataStoreRuntime, "logger">>;
+	Partial<Pick<IFluidDataStoreRuntime, "logger">> & {
+		ILayerCompatDetails?: unknown;
+	};
 
 /**
  * @internal

packages/framework/presence/src/presenceDatastoreManager.ts

@@ -3,6 +3,7 @@
  * Licensed under the MIT License.
  */
 
+import type { ILayerCompatDetails } from "@fluid-internal/client-utils";
 import type { IEmitter } from "@fluidframework/core-interfaces/internal";
 import { assert } from "@fluidframework/core-utils/internal";
 import type { IInboundSignalMessage } from "@fluidframework/runtime-definitions/internal";
@@ -75,6 +76,7 @@ interface DatastoreUpdateMessage extends IInboundSignalMessage {
 		sendTimestamp: number;
 		avgLatency: number;
 		isComplete?: true;
+		acknowledgementId?: number;
 		data: DatastoreMessageContent;
 	};
 }
@@ -90,11 +92,34 @@ interface ClientJoinMessage extends IInboundSignalMessage {
 	};
 }
 
+// Message type identifier for presence acknowledgment messages.
+const acknowledgementMessageType = "Pres:Ack";
+
+/**
+ * Message type used to acknowledge receipt of remote update messages.
+ */
+interface AcknowledgementMessage extends IInboundSignalMessage {
+	type: typeof acknowledgementMessageType;
+	content: {
+		id: number;
+	};
+}
+
 function isPresenceMessage(
 	message: IInboundSignalMessage,
 ): message is DatastoreUpdateMessage | ClientJoinMessage {
 	return message.type.startsWith("Pres:");
 }
+
+function isLayerCompatDetails(value: unknown): value is ILayerCompatDetails {
+	return (
+		typeof value === "object" &&
+		value !== null &&
+		"supportedFeatures" in value &&
+		value.supportedFeatures instanceof Set
+	);
+}
+
 /**
  * @internal
  */
@@ -157,6 +182,9 @@ export class PresenceDatastoreManagerImpl implements PresenceDatastoreManager {
 	private refreshBroadcastRequested = false;
 	private readonly timer = new TimerManager();
 	private readonly workspaces = new Map<string, AnyWorkspaceEntry<StatesWorkspaceSchema>>();
+	private readonly supportsTargetedSignals =
+		isLayerCompatDetails(this.runtime.ILayerCompatDetails) &&
+		this.runtime.ILayerCompatDetails.supportedFeatures.has("submit_signals_v2");
 
 	public constructor(
 		private readonly attendeeId: AttendeeId,
@@ -364,6 +392,7 @@ export class PresenceDatastoreManagerImpl implements PresenceDatastoreManager {
 		if (!isPresenceMessage(message)) {
 			return;
 		}
+
 		if (local) {
 			const deliveryDelta = received - message.content.sendTimestamp;
 			// Limit returnedMessages count to 256 such that newest message
@@ -395,6 +424,24 @@ export class PresenceDatastoreManagerImpl implements PresenceDatastoreManager {
 			if (message.content.isComplete) {
 				this.refreshBroadcastRequested = false;
 			}
+			// If the message requests an acknowledgement, we will send a targeted acknowledgement message back to just the requestor.
+			if (message.content.acknowledgementId !== undefined) {
+				assert(this.supportsTargetedSignals, "Targeted signals not supported");
+				this.runtime.submitSignal(
+					acknowledgementMessageType,
+					{
+						id: message.content.acknowledgementId,
+					} satisfies AcknowledgementMessage["content"],
+					message.clientId,
+				);
+				this.logger?.sendTelemetryEvent({
+					eventName: "AckSent",
+					details: {
+						requestor: message.clientId,
+						responder: this.runtime.clientId,
+					},
+				});
+			}
 		}
 
 		// Handle activation of unregistered workspaces before processing updates.

packages/framework/presence/src/test/mockEphemeralRuntime.ts

@@ -5,6 +5,7 @@
 
 import { strict as assert } from "node:assert";
 
+import type { ILayerCompatDetails } from "@fluid-internal/client-utils";
 import type { ITelemetryBaseLogger } from "@fluidframework/core-interfaces";
 import type { IClient, ISequencedClient } from "@fluidframework/driver-definitions";
 import { MockAudience, MockQuorumClients } from "@fluidframework/test-runtime-utils/internal";
@@ -78,6 +79,7 @@ export class MockEphemeralRuntime implements IEphemeralRuntime {
 		connected: [],
 		disconnected: [],
 	};
+	private readonly supportedFeatures = new Set(["submit_signals_v2"]);
 	private isSupportedEvent(event: string): event is keyof typeof this.listeners {
 		return event in this.listeners;
 	}
@@ -136,6 +138,14 @@ export class MockEphemeralRuntime implements IEphemeralRuntime {
 		);
 	}
 
+	public setTargetSignalSupport(supported: boolean): void {
+		if (supported) {
+			this.supportedFeatures.add("submit_signals_v2");
+		} else {
+			this.supportedFeatures.delete("submit_signals_v2");
+		}
+	}
+
 	public removeMember(clientId: ClientConnectionId): void {
 		const client = this.audience.getMember(clientId);
 		assert(client !== undefined, `Attempting to remove unknown connection: ${clientId}`);
@@ -187,5 +197,13 @@ export class MockEphemeralRuntime implements IEphemeralRuntime {
 		assert.deepStrictEqual(args, expected, "Unexpected signal");
 	};
 
+	public get ILayerCompatDetails(): ILayerCompatDetails {
+		return {
+			supportedFeatures: this.supportedFeatures,
+			generation: 1,
+			pkgVersion: "2.33.0",
+		};
+	}
+
 	// #endregion
 }

packages/framework/presence/src/test/presenceDatastoreManager.spec.ts

@@ -437,6 +437,47 @@ describe("Presence", () => {
 				// Verify
 				assert.strictEqual(listener.callCount, 1);
 			});
+
+			it("with acknowledgementId sends targeted acknowledgment messsage back to requestor", () => {
+				// Setup
+				logger.registerExpectedEvent({
+					eventName: "Presence:AckSent",
+					details: JSON.stringify({
+						requestor: "client4",
+						responder: "client2",
+					}),
+				});
+				// We expect to send a targeted acknowledgment back to the requestor
+				runtime.signalsExpected.push([
+					"Pres:Ack",
+					{
+						"id": 1,
+					},
+					"client4" /* targetClientId */,
+				]);
+
+				// Act - send generic datastore update with acknowledgement id specified
+				presence.processSignal(
+					"",
+					{
+						type: "Pres:DatastoreUpdate",
+						content: {
+							sendTimestamp: clock.now - 10,
+							avgLatency: 20,
+							data: {
+								"system:presence": systemWorkspaceUpdate,
+								"s:name:testStateWorkspace": statesWorkspaceUpdate,
+							},
+							acknowledgementId: 1,
+						},
+						clientId: "client4",
+					},
+					false,
+				);
+
+				// Verify
+				assertFinalExpectations(runtime, logger);
+			});
 		});
 	});
 });