improvement(client-container-loader): tighten Signal expectations

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

@@ -10,6 +10,8 @@ import type {
 	ReadOnlyInfo,
 } from "@fluidframework/container-definitions/internal";
 import { type ITelemetryBaseProperties, LogLevel } from "@fluidframework/core-interfaces";
+import type { JsonString } from "@fluidframework/core-interfaces/internal";
+import { JsonStringify } from "@fluidframework/core-interfaces/internal";
 import { assert } from "@fluidframework/core-utils/internal";
 import type {
 	ConnectionMode,
@@ -113,6 +115,19 @@ interface IPendingConnection {
 	connectionMode: ConnectionMode;
 }
 
+function assertExpectedSignals(
+	signals: ISignalMessage[],
+): asserts signals is ISignalMessage<{ type: never; content: JsonString<unknown> }>[] {
+	for (const signal of signals) {
+		if ("type" in signal) {
+			throw new Error("Unexpected type in ISignalMessage");
+		}
+		if (typeof signal.content !== "string") {
+			throw new TypeError("Non-string content in ISignalMessage");
+		}
+	}
+}
+
 /**
  * Implementation of IConnectionManager, used by Container class
  * Implements constant connectivity to relay service, by reconnecting in case of lost connection or error.
@@ -921,29 +936,29 @@ export class ConnectionManager implements IConnectionManager {
 		// Synthesize clear & join signals out of initialClients state.
 		// This allows us to have single way to process signals, and makes it simpler to initialize
 		// protocol in Container.
-		const clearSignal: ISignalMessage = {
+		const clearSignal = {
 			// API uses null
 			// eslint-disable-next-line unicorn/no-null
 			clientId: null, // system message
-			content: JSON.stringify({
+			content: JsonStringify({
 				type: SignalType.Clear,
 			}),
 		};
 
 		// list of signals to process due to this new connection
-		let signalsToProcess: ISignalMessage[] = [clearSignal];
-
-		const clientJoinSignals: ISignalMessage[] = (connection.initialClients ?? []).map(
-			(priorClient) => ({
-				// API uses null
-				// eslint-disable-next-line unicorn/no-null
-				clientId: null, // system signal
-				content: JSON.stringify({
-					type: SignalType.ClientJoin,
-					content: priorClient, // ISignalClient
-				}),
+		let signalsToProcess: ISignalMessage<{ type: never; content: JsonString<unknown> }>[] = [
+			clearSignal,
+		];
+
+		const clientJoinSignals = (connection.initialClients ?? []).map((priorClient) => ({
+			// API uses null
+			// eslint-disable-next-line unicorn/no-null
+			clientId: null, // system signal
+			content: JsonStringify({
+				type: SignalType.ClientJoin,
+				content: priorClient, // ISignalClient
 			}),
-		);
+		}));
 		if (clientJoinSignals.length > 0) {
 			signalsToProcess = [...signalsToProcess, ...clientJoinSignals];
 		}
@@ -953,6 +968,7 @@ export class ConnectionManager implements IConnectionManager {
 		// for "self" and connection.initialClients does not contain "self", so we have to process them after
 		// "clear" signal above.
 		if (connection.initialSignals !== undefined && connection.initialSignals.length > 0) {
+			assertExpectedSignals(connection.initialSignals);
 			signalsToProcess = [...signalsToProcess, ...connection.initialSignals];
 		}
 
@@ -1179,6 +1195,7 @@ export class ConnectionManager implements IConnectionManager {
 
 	private readonly signalHandler = (signalsArg: ISignalMessage | ISignalMessage[]): void => {
 		const signals = Array.isArray(signalsArg) ? signalsArg : [signalsArg];
+		assertExpectedSignals(signals);
 		this.props.signalHandler(signals);
 	};
 

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

@@ -140,9 +140,10 @@ import { NoopHeuristic } from "./noopHeuristic.js";
 import { pkgVersion } from "./packageVersion.js";
 import type { IQuorumSnapshot } from "./protocol/index.js";
 import {
-	type IProtocolHandler,
+	type InternalProtocolHandlerBuilder,
 	ProtocolHandler,
 	type ProtocolHandlerBuilder,
+	type ProtocolHandlerInternal,
 	protocolHandlerShouldProcessSignal,
 } from "./protocol.js";
 import { initQuorumValuesFromCodeDetails } from "./quorum.js";
@@ -495,7 +496,7 @@ export class Container
 	private readonly scope: FluidObject;
 	private readonly subLogger: ITelemetryLoggerExt;
 	private readonly detachedBlobStorage: MemoryDetachedBlobStorage | undefined;
-	private readonly protocolHandlerBuilder: ProtocolHandlerBuilder;
+	private readonly protocolHandlerBuilder: InternalProtocolHandlerBuilder;
 	private readonly client: IClient;
 
 	private readonly mc: MonitoringContext;
@@ -597,8 +598,8 @@ export class Container
 		}
 		return this._runtime;
 	}
-	private _protocolHandler: IProtocolHandler | undefined;
-	private get protocolHandler(): IProtocolHandler {
+	private _protocolHandler: ProtocolHandlerInternal | undefined;
+	private get protocolHandler(): ProtocolHandlerInternal {
 		if (this._protocolHandler === undefined) {
 			throw new Error("Attempted to access protocolHandler before it was defined");
 		}
@@ -830,7 +831,7 @@ export class Container
 				attributes: IDocumentAttributes,
 				quorumSnapshot: IQuorumSnapshot,
 				sendProposal: (key: string, value: unknown) => number,
-			): ProtocolHandler =>
+			): ProtocolHandlerInternal =>
 				new ProtocolHandler(
 					attributes,
 					quorumSnapshot,

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

@@ -12,6 +12,7 @@ import {
 	type IConnectionDetails,
 } from "@fluidframework/container-definitions/internal";
 import type { IErrorBase, ITelemetryBaseProperties } from "@fluidframework/core-interfaces";
+import type { JsonString } from "@fluidframework/core-interfaces/internal";
 import type { ConnectionMode, IClientDetails } from "@fluidframework/driver-definitions";
 import type {
 	IContainerPackageInfo,
@@ -145,7 +146,9 @@ export interface IConnectionManagerFactoryArgs {
 	 * Called by connection manager for each incoming signal.
 	 * May be called before connectHandler is called (due to initial signals on socket connection)
 	 */
-	readonly signalHandler: (signals: ISignalMessage[]) => void;
+	readonly signalHandler: (
+		signals: ISignalMessage<{ type: never; content: JsonString<unknown> }>[],
+	) => void;
 
 	/**
 	 * Called when connection manager experiences delay in connecting to relay service.

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

@@ -16,7 +16,8 @@ import type {
 	ITelemetryBaseEvent,
 	ITelemetryBaseProperties,
 } from "@fluidframework/core-interfaces";
-import type { IThrottlingWarning } from "@fluidframework/core-interfaces/internal";
+import { JsonParse } from "@fluidframework/core-interfaces/internal";
+import type { IThrottlingWarning, JsonString } from "@fluidframework/core-interfaces/internal";
 import { assert } from "@fluidframework/core-utils/internal";
 import type { ConnectionMode } from "@fluidframework/driver-definitions";
 import {
@@ -210,7 +211,9 @@ export class DeltaManager<TConnectionManager extends IConnectionManager>
 	private initSequenceNumber: number = 0;
 
 	private readonly _inbound: DeltaQueue<ISequencedDocumentMessage>;
-	private readonly _inboundSignal: DeltaQueue<ISignalMessage>;
+	private readonly _inboundSignal: DeltaQueue<
+		ISignalMessage<{ type: never; content: JsonString<unknown> }>
+	>;
 
 	private _closed = false;
 	private _disposed = false;
@@ -433,7 +436,9 @@ export class DeltaManager<TConnectionManager extends IConnectionManager>
 					this.close(normalizeError(error));
 				}
 			},
-			signalHandler: (signals: ISignalMessage[]) => {
+			signalHandler: (
+				signals: ISignalMessage<{ type: never; content: JsonString<unknown> }>[],
+			) => {
 				for (const signal of signals) {
 					this._inboundSignal.push(signal);
 				}
@@ -474,14 +479,16 @@ export class DeltaManager<TConnectionManager extends IConnectionManager>
 		});
 
 		// Inbound signal queue
-		this._inboundSignal = new DeltaQueue<ISignalMessage>((message) => {
+		this._inboundSignal = new DeltaQueue<
+			ISignalMessage<{ type: never; content: JsonString<unknown> }>
+		>((message) => {
 			if (this.handler === undefined) {
 				throw new Error("Attempted to process an inbound signal without a handler attached");
 			}
 
 			this.handler.processSignal({
 				...message,
-				content: JSON.parse(message.content as string),
+				content: JsonParse(message.content),
 			});
 		});
 

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

@@ -21,12 +21,39 @@ import {
 } from "./protocol/index.js";
 
 // ADO: #1986: Start using enum from protocol-base.
-export enum SignalType {
-	ClientJoin = "join", // same value as MessageType.ClientJoin,
-	ClientLeave = "leave", // same value as MessageType.ClientLeave,
-	Clear = "clear", // used only by client for synthetic signals
+export const SignalType = {
+	ClientJoin: "join", // same value as MessageType.ClientJoin,
+	ClientLeave: "leave", // same value as MessageType.ClientLeave,
+	Clear: "clear", // used only by client for synthetic signals
+} as const;
+
+interface SystemSignalContent {
+	type: (typeof SignalType)[keyof typeof SignalType];
+	content?: unknown;
+}
+
+interface InboundSystemSignal<TSignalContent extends SystemSignalContent>
+	extends ISignalMessage<{ type: never; content: TSignalContent }> {
+	// eslint-disable-next-line @rushstack/no-new-null -- `null` is used in JSON protocol to indicate system message
+	readonly clientId: null;
 }
 
+type ClientJoinSignal = InboundSystemSignal<{
+	type: typeof SignalType.ClientJoin;
+	content: ISignalClient;
+}>;
+
+type ClientLeaveSignal = InboundSystemSignal<{
+	type: typeof SignalType.ClientLeave;
+	content: string; // clientId of leaving client
+}>;
+
+type ClearClientsSignal = InboundSystemSignal<{
+	type: typeof SignalType.Clear;
+}>;
+
+type AudienceSignal = ClientJoinSignal | ClientLeaveSignal | ClearClientsSignal;
+
 /**
  * Function to be used for creating a protocol handler.
  * @legacy @beta
@@ -47,7 +74,34 @@ export interface IProtocolHandler extends IBaseProtocolHandler {
 	processSignal(message: ISignalMessage);
 }
 
-export class ProtocolHandler extends ProtocolOpHandler implements IProtocolHandler {
+/**
+ * More specific version of {@link IProtocolHandler} with narrower call
+ * constraints for processSignal.
+ */
+export interface ProtocolHandlerInternal extends IProtocolHandler {
+	/**
+	 * Process the audience related signal.
+	 * @privateRemarks Internally only AudienceSignal messages need handled
+	 */
+	processSignal(message: AudienceSignal): void;
+}
+
+/**
+ * Function to be used for creating a protocol handler.
+ *
+ * @remarks This is the same are {@link ProtocolHandlerBuilder} but
+ * returns the {@link ProtocolHandlerInternal} which has narrower
+ * expectations for `processSignal`.
+ */
+export type InternalProtocolHandlerBuilder = (
+	attributes: IDocumentAttributes,
+	snapshot: IQuorumSnapshot,
+	// TODO: use a real type (breaking change)
+	// eslint-disable-next-line @typescript-eslint/no-explicit-any
+	sendProposal: (key: string, value: any) => number,
+) => ProtocolHandlerInternal;
+
+export class ProtocolHandler extends ProtocolOpHandler implements ProtocolHandlerInternal {
 	constructor(
 		attributes: IDocumentAttributes,
 		quorumSnapshot: IQuorumSnapshot,
@@ -104,8 +158,8 @@ export class ProtocolHandler extends ProtocolOpHandler implements IProtocolHandl
 		return super.processMessage(message, local);
 	}
 
-	public processSignal(message: ISignalMessage): void {
-		const innerContent = message.content as { content: unknown; type: string };
+	public processSignal(message: AudienceSignal): void {
+		const innerContent = message.content;
 		switch (innerContent.type) {
 			case SignalType.Clear: {
 				const members = this.audience.getMembers();
@@ -117,15 +171,15 @@ export class ProtocolHandler extends ProtocolOpHandler implements IProtocolHandl
 				break;
 			}
 			case SignalType.ClientJoin: {
-				const newClient = innerContent.content as ISignalClient;
+				const newClient = innerContent.content;
 				// Ignore write clients - quorum will control such clients.
 				if (newClient.client.mode === "read") {
 					this.audience.addMember(newClient.clientId, newClient.client);
 				}
 				break;
 			}
 			case SignalType.ClientLeave: {
-				const leftClientId = innerContent.content as string;
+				const leftClientId = innerContent.content;
 				// Ignore write clients - quorum will control such clients.
 				if (this.audience.getMember(leftClientId)?.mode === "read") {
 					this.audience.removeMember(leftClientId);
@@ -144,7 +198,9 @@ export class ProtocolHandler extends ProtocolOpHandler implements IProtocolHandl
  * The protocol handler should strictly handle only ClientJoin, ClientLeave
  * and Clear signal types.
  */
-export function protocolHandlerShouldProcessSignal(message: ISignalMessage): boolean {
+export function protocolHandlerShouldProcessSignal(
+	message: ISignalMessage,
+): message is AudienceSignal {
 	// Signal originates from server
 	if (message.clientId === null) {
 		const innerContent = message.content as { content: unknown; type: string };