feat(client): Container Extensions operation during CatchingUp (#25757)

- "connectedToService" is emitted for CatchingUp (ahead of Connected)
- Signal only Audience will be sent to provide information sooner and
allow early activity.

Presence limits quorum use to ordering of fallback Join responses, which
are expected to happen later and often when quorum is more complete.

Author: jason-ha

packages/framework/presence/src/presenceDatastoreManager.ts

@@ -186,11 +186,43 @@ export class PresenceDatastoreManagerImpl implements PresenceDatastoreManager {
 		this.targetedSignalSupport = this.runtime.supportedFeatures.has("submit_signals_v2");
 	}
 
-	public joinSession(clientId: ClientConnectionId): void {
+	private getInteractiveMembersExcludingSelf(selfClientId: ClientConnectionId): {
+		all: Set<ClientConnectionId>;
+		writers: Set<ClientConnectionId>;
+	} {
+		const audience = this.runtime.getAudience();
+		const members = audience.getMembers();
+		const all = new Set<ClientConnectionId>();
+		const writers = new Set<ClientConnectionId>();
+		// Remove self (if present)
+		members.delete(selfClientId);
+		// Gather interactive client IDs
+		for (const [id, client] of members) {
+			if (client.details.capabilities.interactive) {
+				all.add(id);
+				if (client.mode === "write") {
+					writers.add(id);
+				}
+			}
+		}
+		return {
+			all,
+			writers,
+		};
+	}
+
+	public joinSession(selfClientId: ClientConnectionId): void {
+		const interactiveMembersExcludingSelf =
+			this.getInteractiveMembersExcludingSelf(selfClientId);
+
 		// Broadcast join message to all clients
-		const updateProviders = [...this.runtime.getQuorum().getMembers().keys()].filter(
-			(quorumClientId) => quorumClientId !== clientId,
-		);
+		// Select primary update providers
+		// Use write members if any, then fallback to read-only members.
+		const updateProviders = [
+			...(interactiveMembersExcludingSelf.writers.size > 0
+				? interactiveMembersExcludingSelf.writers
+				: interactiveMembersExcludingSelf.all),
+		];
 		// Limit to three providers to prevent flooding the network.
 		// If none respond, others present will (should) after a delay.
 		if (updateProviders.length > 3) {
@@ -330,8 +362,8 @@ export class PresenceDatastoreManagerImpl implements PresenceDatastoreManager {
 		assert(clientConnectionId !== undefined, 0xa59 /* Client connected without clientId */);
 		const currentClientToSessionValueState =
 			// When connected, `clientToSessionId` must always have current connection entry.
-			// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-			this.datastore["system:presence"].clientToSessionId[clientConnectionId]!;
+			this.datastore["system:presence"].clientToSessionId[clientConnectionId];
+		assert(currentClientToSessionValueState !== undefined, "Client connection update missing");
 
 		const newMessage = {
 			sendTimestamp: Date.now(),

packages/runtime/container-runtime-definitions/src/containerExtension.ts

@@ -282,7 +282,7 @@ export interface ExtensionHost<TRuntimeProperties extends ExtensionRuntimeProper
 	 * @remarks
 	 * Status changes are signaled through :
 	 * - {@link ExtensionHostEvents.disconnected}: Transitioning to Disconnected state
-	 * - {@link ExtensionHostEvents.joined}: Transition to Connected state (either for reading or writing)
+	 * - {@link ExtensionHostEvents.joined}: Transition to CatchingUp or Connected state (either for reading or writing)
 	 * - {@link ExtensionHostEvents.operabilityChanged}: When operability has changed (e.g., write to read)
 	 */
 	readonly getJoinedStatus: () => JoinedStatus;
@@ -313,6 +313,13 @@ export interface ExtensionHost<TRuntimeProperties extends ExtensionRuntimeProper
 	 */
 	getQuorum: () => IQuorumClients;
 
+	/**
+	 * The collection of all clients as enumerated by the service.
+	 *
+	 * @remarks This may include/exclude those found within the quorum.
+	 * It produces results faster than {@link ExtensionHost.getQuorum}, but
+	 * will be inaccurate if any signals are lost.
+	 */
 	getAudience: () => IAudience;
 }
 

packages/runtime/container-runtime/src/containerRuntime.ts

@@ -1568,6 +1568,7 @@ export class ContainerRuntime
 			deltaManager,
 			quorum,
 			audience,
+			signalAudience,
 			pendingLocalState,
 			supportedFeatures,
 			snapshotWithContents,
@@ -1722,7 +1723,8 @@ export class ContainerRuntime
 		// Later updates come through calls to setConnectionState/Status.
 		this.canSendOps = connected;
 		this.canSendSignals = this.getConnectionState
-			? this.getConnectionState() === ConnectionState.Connected
+			? this.getConnectionState() === ConnectionState.Connected ||
+				this.getConnectionState() === ConnectionState.CatchingUp
 			: undefined;
 
 		this.mc.logger.sendTelemetryEvent({
@@ -2038,6 +2040,8 @@ export class ContainerRuntime
 			});
 		}
 
+		this.signalAudience = signalAudience;
+
 		const closeSummarizerDelayOverride = this.mc.config.getNumber(
 			"Fluid.ContainerRuntime.Test.CloseSummarizerDelayOverrideMs",
 		);
@@ -2947,7 +2951,7 @@ export class ContainerRuntime
 	 * Emits service connection events based on connection state changes.
 	 *
 	 * @remarks
-	 * "connectedToService" is emitted when container connection state transitions to 'Connected' regardless of connection mode.
+	 * "connectedToService" is emitted when container connection state transitions to 'CatchingUp' or 'Connected' regardless of connection mode.
 	 * "disconnectedFromService" excludes false "disconnected" events that happen when readonly client transitions to 'Connected'.
 	 */
 	private emitServiceConnectionEvents(
@@ -2959,16 +2963,20 @@ export class ContainerRuntime
 			return;
 		}
 
-		const canSendSignals = this.getConnectionState() === ConnectionState.Connected;
+		const connectionState = this.getConnectionState();
+		const canSendSignals =
+			connectionState === ConnectionState.Connected ||
+			connectionState === ConnectionState.CatchingUp;
 		const canSendSignalsChanged = this.canSendSignals !== canSendSignals;
 		this.canSendSignals = canSendSignals;
 		if (canSendSignalsChanged) {
-			// If canSendSignals changed, we either transitioned from Connected to Disconnected or CatchingUp to Connected
+			// If canSendSignals changed, we either transitioned from CatchingUp or
+			// Connected to Disconnected or EstablishingConnection to CatchingUp.
 			if (canSendSignals) {
-				// Emit for CatchingUp to Connected transition
+				// Emit for EstablishingConnection to CatchingUp or Connected transition
 				this.emit("connectedToService", clientId, canSendOps);
 			} else {
-				// Emit for Connected to Disconnected transition
+				// Emit for CatchingUp or Connected to Disconnected transition
 				this.emit("disconnectedFromService");
 			}
 		} else if (canSendOpsChanged) {
@@ -3727,6 +3735,14 @@ export class ContainerRuntime
 		return this._audience;
 	}
 
+	/**
+	 * When defined, this {@link @fluidframework/container-definitions#IAudience}
+	 * maintains member list using signals only.
+	 * Thus "write" members may be known earlier than quorum and avoid noise from
+	 * un-summarized quorum history.
+	 */
+	private readonly signalAudience?: IAudience;
+
 	/**
 	 * Returns true of container is dirty, i.e. there are some pending local changes that
 	 * either were not sent out to delta stream or were not yet acknowledged.
@@ -5271,7 +5287,11 @@ export class ContainerRuntime
 		const getConnectionState = this.getConnectionState;
 		if (getConnectionState) {
 			const connectionState = getConnectionState();
-			if (connectionState === ConnectionState.Connected) {
+			if (
+				connectionState === ConnectionState.Connected ||
+				connectionState === ConnectionState.CatchingUp
+			) {
+				// Note: when CatchingUp, canSendOps will always be false.
 				return this.canSendOps ? "joinedForWriting" : "joinedForReading";
 			}
 		} else if (this.canSendOps) {
@@ -5297,9 +5317,10 @@ export class ContainerRuntime
 	): T {
 		let entry = this.extensions.get(id);
 		if (entry === undefined) {
+			const audience = this.signalAudience;
 			const runtime = {
 				getJoinedStatus: this.getJoinedStatus.bind(this),
-				getClientId: () => this.clientId,
+				getClientId: audience ? () => audience.getSelf()?.clientId : () => this.clientId,
 				events: this.lazyEventsForExtensions.value,
 				logger: this.baseLogger,
 				submitAddressedSignal: (
@@ -5309,7 +5330,7 @@ export class ContainerRuntime
 					this.submitExtensionSignal(id, addressChain, message);
 				},
 				getQuorum: this.getQuorum.bind(this),
-				getAudience: this.getAudience.bind(this),
+				getAudience: audience ? () => audience : this.getAudience.bind(this),
 				supportedFeatures: this.ILayerCompatDetails.supportedFeatures,
 			} satisfies ExtensionHost<TRuntimeProperties>;
 			entry = new factory(runtime, ...useContext);

packages/runtime/container-runtime/src/test/containerRuntime.extensions.spec.ts

@@ -269,8 +269,8 @@ describe("Runtime", () => {
 
 					assert.strictEqual(
 						extension.connectedToService,
-						false,
-						"Extension should be disconnected during CatchingUp state",
+						true,
+						"Extension should be connected during CatchingUp state",
 					);
 				});
 
@@ -379,22 +379,26 @@ describe("Runtime", () => {
 
 					// Transition to CatchingUp
 					updateConnectionState(runtime, context, ConnectionState.CatchingUp, "mockClientId");
+					assert.strictEqual(events.length, 1, "Should have received one joined event");
+					assert.deepStrictEqual(
+						events[0],
+						{ type: "joined", clientId: "mockClientId", canWrite: false },
+						"First event should be joined for reading",
+					);
 					assert.strictEqual(
 						extension.connectedToService,
-						false,
-						"Extension should be disconnected during CatchingUp",
+						true,
+						"Extension should be connected during CatchingUp",
 					);
-					assert.strictEqual(events.length, 0, "Should have no events during CatchingUp");
 
 					// Transition to Connected
 					updateConnectionState(runtime, context, ConnectionState.Connected, "mockClientId");
-					assert.strictEqual(events.length, 1, "Should have received one joined events");
+					assert.strictEqual(events.length, 2, "Should have received two events total");
 					assert.deepStrictEqual(
-						events[0],
-						{ type: "joined", clientId: "mockClientId", canWrite: true },
-						"First event should be joined for writing",
+						events[1],
+						{ type: "operabilityChanged", canWrite: true },
+						"Second event should be operabilityChanged to can write",
 					);
-
 					assert.strictEqual(
 						extension.connectedToService,
 						true,
@@ -403,11 +407,11 @@ describe("Runtime", () => {
 
 					// Transition back to Disconnected
 					updateConnectionState(runtime, context, ConnectionState.Disconnected);
-					assert.strictEqual(events.length, 2, "Should have received two events total");
+					assert.strictEqual(events.length, 3, "Should have received three events total");
 					assert.deepStrictEqual(
-						events[1],
+						events[2],
 						{ type: "disconnected" },
-						"Second event should be disconnected",
+						"Third event should be disconnected",
 					);
 					assert.strictEqual(
 						extension.connectedToService,
@@ -462,13 +466,18 @@ describe("Runtime", () => {
 					);
 					assert.strictEqual(
 						readOnlyExtension.connectedToService,
-						false,
-						"Extension should be disconnected during CatchingUp",
+						true,
+						"Extension should be connected during CatchingUp",
 					);
 					assert.strictEqual(
 						readOnlyEvents.length,
-						0,
-						"Should have no events during CatchingUp",
+						1,
+						"Should have received one event: joined for reading only",
+					);
+					assert.deepStrictEqual(
+						readOnlyEvents[0],
+						{ type: "joined", clientId: "mockClientId", canWrite: false },
+						"Event should be joined for reading",
 					);
 
 					// Transition to Connected
@@ -478,16 +487,7 @@ describe("Runtime", () => {
 						ConnectionState.Connected,
 						"mockClientId",
 					);
-					assert.strictEqual(
-						readOnlyEvents.length,
-						1,
-						"Should have received one event: joined for reading only",
-					);
-					assert.deepStrictEqual(
-						readOnlyEvents[0],
-						{ type: "joined", clientId: "mockClientId", canWrite: false },
-						"Event should be joined for reading",
-					);
+					assert.strictEqual(readOnlyEvents.length, 1, "Should have received one event total");
 					assert.strictEqual(
 						readOnlyExtension.connectedToService,
 						true,