feat(client): internal Signal only based Audience

that will be sent to Container Extensions to provide information
sooner and allow early activity.

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

Author: jason-ha

packages/common/container-definitions/api-report/container-definitions.legacy.beta.api.md

@@ -157,6 +157,8 @@ export interface IContainerContext {
     // (undocumented)
     readonly quorum: IQuorumClients;
     readonly scope: FluidObject;
+    // @system
+    readonly signalAudience?: IAudience;
     readonly snapshotWithContents?: ISnapshot;
     // (undocumented)
     readonly storage: IContainerStorageService;

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

@@ -302,6 +302,16 @@ export interface IContainerContext {
 	readonly deltaManager: IDeltaManager<ISequencedDocumentMessage, IDocumentMessage>;
 	readonly quorum: IQuorumClients;
 	readonly audience: IAudience;
+	/**
+	 * Signal-based audience provides a view of the audience that only relies
+	 * on system signals which will be updated more quickly than
+	 * {@link IContainerContext#audience} that relies on ops for write clients.
+	 * Being signal-based the write members are inherently less reliable than
+	 * {@link IContainerContext#audience}.
+	 *
+	 * @system
+	 */
+	readonly signalAudience?: IAudience;
 	readonly loader: ILoader;
 	// The logger implementation, which would support tagged events, should be provided by the loader.
 	readonly taggedLogger: ITelemetryBaseLogger;

packages/framework/presence/src/presenceDatastoreManager.ts

@@ -254,24 +254,29 @@ export class PresenceDatastoreManagerImpl implements PresenceDatastoreManager {
 
 	private getAudienceInformation(selfClientId: ClientConnectionId): {
 		audience: IAudience;
-		otherMembers: Map<ClientConnectionId, IClient>;
-		othersWorthIgnoring: ClientConnectionId[];
+		interactiveMembersExcludingSelf: {
+			all: Set<ClientConnectionId>;
+			writers: Set<ClientConnectionId>;
+		};
 	} {
 		const audience = this.runtime.getAudience();
-		const otherMembers = audience.getMembers();
+		const members = audience.getMembers();
+		const all = new Set<ClientConnectionId>();
+		const writers = new Set<ClientConnectionId>();
 		// Remove self (if present)
-		otherMembers.delete(selfClientId);
-		// It is worth ignoring non-interactive clients
-		const othersWorthIgnoring: ClientConnectionId[] = [];
-		for (const [id, client] of otherMembers) {
-			if (!client.details.capabilities.interactive) {
-				othersWorthIgnoring.push(id);
+		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 {
 			audience,
-			otherMembers,
-			othersWorthIgnoring,
+			interactiveMembersExcludingSelf: { all, writers },
 		};
 	}
 
@@ -287,27 +292,22 @@ export class PresenceDatastoreManagerImpl implements PresenceDatastoreManager {
 		// Lack of others might mean that this client is very freshly joined
 		// and has not received any Join Ops or Signals (type="join") from
 		// the service yet.
-		const { audience, otherMembers, othersWorthIgnoring } =
+		const { audience, interactiveMembersExcludingSelf } =
 			this.getAudienceInformation(selfClientId);
 		// When no others (and not forced), delay join until there are.
-		if (otherMembers.size === othersWorthIgnoring.length && !forceEvenWhenAlone) {
+		if (interactiveMembersExcludingSelf.all.size === 0 && !forceEvenWhenAlone) {
 			// No one else known to be present.
 			this.deferJoinUntilAudienceMember(selfClientId, audience);
 			return;
 		}
 
 		// Broadcast join message to all clients
 		// Select primary update providers
-		const quorumMembers = this.runtime.getQuorum().getMembers();
-		// Remove self and non-interactive members from possibilities
-		othersWorthIgnoring.push(selfClientId);
-		for (const clientIdToDelete of othersWorthIgnoring) {
-			quorumMembers.delete(clientIdToDelete);
-			otherMembers.delete(clientIdToDelete);
-		}
-		// Use quorum members if available, then fallback to audience.
+		// Use write members if any, then fallback to read-only members.
 		const updateProviders = [
-			...(quorumMembers.size > 0 ? quorumMembers : otherMembers).keys(),
+			...(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.
@@ -904,8 +904,8 @@ export class PresenceDatastoreManagerImpl implements PresenceDatastoreManager {
 			// Check if requestor count meets or exceeds count of other audience
 			// members indicating that we effectively have a complete snapshot
 			// (once the current message being processed is processed).
-			const { otherMembers, othersWorthIgnoring } = this.getAudienceInformation(selfClientId);
-			if (this.broadcastRequests.size >= otherMembers.size - othersWorthIgnoring.length) {
+			const { interactiveMembersExcludingSelf } = this.getAudienceInformation(selfClientId);
+			if (this.broadcastRequests.size >= interactiveMembersExcludingSelf.all.size) {
 				// Note that no action is taken here specifically.
 				// We want action to be queued so that it takes place after
 				// current message is completely processed. All of the actions

packages/loader/container-loader/src/container.ts

@@ -145,6 +145,7 @@ import {
 	type ProtocolHandlerBuilder,
 	type ProtocolHandlerInternal,
 	protocolHandlerShouldProcessSignal,
+	wrapProtocolHandlerBuilder,
 } from "./protocol.js";
 import { initQuorumValuesFromCodeDetails } from "./quorum.js";
 import {
@@ -497,6 +498,7 @@ export class Container
 	private readonly subLogger: ITelemetryLoggerExt;
 	private readonly detachedBlobStorage: MemoryDetachedBlobStorage | undefined;
 	private readonly protocolHandlerBuilder: InternalProtocolHandlerBuilder;
+	private readonly signalAudience = new Audience();
 	private readonly client: IClient;
 
 	private readonly mc: MonitoringContext;
@@ -825,20 +827,22 @@ export class Container
 		// Tracking alternative ways to handle this in AB#4129.
 		this.options = { ...options };
 		this.scope = scope;
-		this.protocolHandlerBuilder =
+		this.protocolHandlerBuilder = wrapProtocolHandlerBuilder(
 			protocolHandlerBuilder ??
-			((
-				attributes: IDocumentAttributes,
-				quorumSnapshot: IQuorumSnapshot,
-				sendProposal: (key: string, value: unknown) => number,
-			): ProtocolHandlerInternal =>
-				new ProtocolHandler(
-					attributes,
-					quorumSnapshot,
-					sendProposal,
-					new Audience(),
-					(clientId: string) => this.clientsWhoShouldHaveLeft.has(clientId),
-				));
+				((
+					attributes: IDocumentAttributes,
+					quorumSnapshot: IQuorumSnapshot,
+					sendProposal: (key: string, value: unknown) => number,
+				): ProtocolHandlerInternal =>
+					new ProtocolHandler(
+						attributes,
+						quorumSnapshot,
+						sendProposal,
+						new Audience(),
+						(clientId: string) => this.clientsWhoShouldHaveLeft.has(clientId),
+					)),
+			this.signalAudience,
+		);
 
 		// Note that we capture the createProps here so we can replicate the creation call when we want to clone.
 		this.clone = async (
@@ -2222,6 +2226,13 @@ export class Container
 			const clientId = this.connectionStateHandler.clientId;
 			assert(clientId !== undefined, 0x96e /* there has to be clientId */);
 			this.protocolHandler.audience.setCurrentClientId(clientId);
+			this.signalAudience.setCurrentClientId(clientId);
+		} else if (this.connectionState === ConnectionState.CatchingUp) {
+			// Signal-based Audience does not wait for ops. So provide clientId
+			// as soon as possible.
+			const clientId = this.connectionStateHandler.pendingClientId;
+			assert(clientId !== undefined, "catching up without clientId");
+			this.signalAudience.setCurrentClientId(clientId);
 		}
 
 		// We communicate only transitions to Connected & Disconnected states, skipping all other states.

packages/loader/container-loader/src/protocol.ts

@@ -212,3 +212,52 @@ export function protocolHandlerShouldProcessSignal(
 	}
 	return false;
 }
+
+export function wrapProtocolHandlerBuilder(
+	builder: ProtocolHandlerBuilder,
+	signalAudience: IAudienceOwner,
+): InternalProtocolHandlerBuilder {
+	return (
+		attributes: IDocumentAttributes,
+		snapshot: IQuorumSnapshot,
+		sendProposal: (key: string, value: unknown) => number,
+	): ProtocolHandlerInternal => {
+		const baseHandler = builder(attributes, snapshot, sendProposal);
+		return {
+			quorum: baseHandler.quorum,
+			attributes: baseHandler.attributes,
+			setConnectionState: baseHandler.setConnectionState.bind(baseHandler),
+			snapshot: baseHandler.snapshot.bind(baseHandler),
+			close: baseHandler.close.bind(baseHandler),
+			processMessage: baseHandler.processMessage.bind(baseHandler),
+			getProtocolState: baseHandler.getProtocolState.bind(baseHandler),
+			audience: baseHandler.audience,
+			processSignal: (message: AudienceSignal) => {
+				const innerContent = message.content;
+				switch (innerContent.type) {
+					case SignalType.Clear: {
+						const members = signalAudience.getMembers();
+						for (const clientId of members.keys()) {
+							signalAudience.removeMember(clientId);
+						}
+						break;
+					}
+					case SignalType.ClientJoin: {
+						const newClient = innerContent.content;
+						signalAudience.addMember(newClient.clientId, newClient.client);
+						break;
+					}
+					case SignalType.ClientLeave: {
+						const leftClientId = innerContent.content;
+						signalAudience.removeMember(leftClientId);
+						break;
+					}
+					default: {
+						break;
+					}
+				}
+				baseHandler.processSignal(message);
+			},
+		};
+	};
+}

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

@@ -274,6 +274,13 @@ export interface ExtensionHost<TRuntimeProperties extends ExtensionRuntimeProper
 	 */
 	getQuorum: () => IQuorumClients;
 
+	/**
+	 * The collection of all clients as enumerated by the service.
+	 *
+	 * @remarks This may be a 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

@@ -1533,6 +1533,7 @@ export class ContainerRuntime
 			deltaManager,
 			quorum,
 			audience,
+			signalAudience,
 			pendingLocalState,
 			supportedFeatures,
 			snapshotWithContents,
@@ -2002,6 +2003,8 @@ export class ContainerRuntime
 			});
 		}
 
+		this.signalAudience = signalAudience;
+
 		const closeSummarizerDelayOverride = this.mc.config.getNumber(
 			"Fluid.ContainerRuntime.Test.CloseSummarizerDelayOverrideMs",
 		);
@@ -3638,6 +3641,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.
@@ -5158,9 +5169,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: (
@@ -5170,7 +5182,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);