(op bunching) Added logic to process messages in a bunch in SharedObject (#23836)

This PR adds support for DDSes to process a 'bunch' of messages. Op
bunching was added via the following PRs:
#22839,
#22840 and
#22841.

Added a `processMessagesCore` function to `SharedObjectCore` that DDSes
can override to process a message bunch. This is marked optional today
because it's a breaking change. It is similar to the `processCore`
function and will replace it eventually. For now, `processCore` has been
marked as deprecated.

This change updates the relative ordering of the `pre-op` and `op`
events emitted by the shared object. Previously, the `pre-op` event
would be emitted before a message was processed and `op` event was
emitted immediately afterwards. With this change, the ordering of these
events w.r.t. to the message being processed is still the same. However,
there may be other ops processed and other event fired in between.
Note that the only guarantee these events provide is that the `pre-op`
event is emitted before a message is procesed and the `op` event is
processed after. So, this change doesn't break that guarantee.

Author: agarwal-navin

.changeset/itchy-pants-brake.md

@@ -0,0 +1,14 @@
+---
+"@fluidframework/shared-object-base": minor
+---
+---
+"section": other
+---
+
+Change when the `pre-op` and `op` events on `ISharedObjectEvents` are emitted
+
+Previous behavior - `pre-op` was emitted immediately before an op was processed. Then the op was processed and `op` was emitted immediately after that.
+
+New behavior - `pre-op` will still be emitted before an op is processed and `op` will still be emitted after an op is processed. However, these won't be immediate and other ops in a batch for the shared object may be processed in between.
+
+Note that these events are for internal use only as mentioned in the @remarks section of their definition.

.changeset/witty-falcons-wonder.md

@@ -0,0 +1,12 @@
+---
+"@fluidframework/shared-object-base": minor
+---
+---
+"section": deprecation
+---
+
+Deprecate `processCore` on `SharedObject` and `SharedObjectCore` in favor of `processMessagesCore`
+
+A new function `processMessagesCore` has been added in place of `processCore`, which will be called to process multiple messages instead of a single one on the channel. This is part of a feature called "Op bunching" where contiguous ops in a grouped batch are bunched and processed together by the shared object.
+
+Implementations of `SharedObject` and `SharedObjectCore` must now also implement `processMessagesCore`. A basic implementation could be to iterate over the messages' content and process them one by one as it happens now. Note that some DDS may be able to optimize processing by processing the messages together.

packages/dds/shared-object-base/api-report/shared-object-base.legacy.alpha.api.md

@@ -78,7 +78,9 @@ export abstract class SharedObjectCore<TEvent extends ISharedObjectEvents = ISha
     protected newAckBasedPromise<T>(executor: (resolve: (value: T | PromiseLike<T>) => void, reject: (reason?: unknown) => void) => void): Promise<T>;
     protected onConnect(): void;
     protected abstract onDisconnect(): void;
+    // @deprecated
     protected abstract processCore(message: ISequencedDocumentMessage, local: boolean, localOpMetadata: unknown): void;
+    protected processMessagesCore?(messagesCollection: IRuntimeMessageCollection): void;
     protected reSubmitCore(content: unknown, localOpMetadata: unknown): void;
     protected rollback(content: unknown, localOpMetadata: unknown): void;
     // (undocumented)

packages/dds/shared-object-base/src/sharedObject.ts

@@ -33,6 +33,7 @@ import {
 	blobCountPropertyName,
 	totalBlobSizePropertyName,
 	type IRuntimeMessageCollection,
+	type IRuntimeMessagesContent,
 } from "@fluidframework/runtime-definitions/internal";
 import {
 	toDeltaManagerInternal,
@@ -161,6 +162,20 @@ export abstract class SharedObjectCore<
 		const { opProcessingHelper, callbacksHelper } = this.setUpSampledTelemetryHelpers();
 		this.opProcessingHelper = opProcessingHelper;
 		this.callbacksHelper = callbacksHelper;
+
+		const processMessagesCore = this.processMessagesCore?.bind(this);
+		this.processMessagesHelper =
+			processMessagesCore === undefined
+				? (messagesCollection: IRuntimeMessageCollection) =>
+						processHelper(messagesCollection, this.process.bind(this))
+				: (messagesCollection: IRuntimeMessageCollection) => {
+						processMessagesCoreHelper(
+							messagesCollection,
+							this.opProcessingHelper,
+							this.emitInternal.bind(this),
+							processMessagesCore,
+						);
+					};
 	}
 
 	/**
@@ -384,13 +399,43 @@ export abstract class SharedObjectCore<
 	 * @param local - True if the shared object is local
 	 * @param localOpMetadata - For local client messages, this is the metadata that was submitted with the message.
 	 * For messages from a remote client, this will be undefined.
+	 *
+	 * @deprecated Replaced by {@link SharedObjectCore.processMessagesCore}.
 	 */
 	protected abstract processCore(
 		message: ISequencedDocumentMessage,
 		local: boolean,
 		localOpMetadata: unknown,
 	): void;
 
+	/* eslint-disable jsdoc/check-indentation */
+	/**
+	 * Process a 'bunch' of messages for this shared object.
+	 *
+	 * @remarks
+	 * A 'bunch' is a group of messages that have the following properties:
+	 * - They are all part of the same grouped batch, which entails:
+	 *   - They are contiguous in sequencing order.
+	 *   - They are all from the same client.
+	 *   - They are all based on the same reference sequence number.
+	 *   - They are not interleaved with messages from other clients.
+	 * - They are not interleaved with messages from other DDS in the container.
+	 * Derived classes should override this if they need to do custom processing on a 'bunch' of remote messages.
+	 * @param messageCollection - The collection of messages to process.
+	 *
+	 */
+	/* eslint-enable jsdoc/check-indentation */
+	protected processMessagesCore?(messagesCollection: IRuntimeMessageCollection): void;
+
+	/**
+	 * Calls {@link SharedObjectCore.processCore} or {@link SharedObjectCore.processMessagesCore} depending on whether
+	 * processMessagesCore is defined. This helper is used to keep the code cleaner while we have to support both these
+	 * function.
+	 */
+	private readonly processMessagesHelper: (
+		messagesCollection: IRuntimeMessageCollection,
+	) => void;
+
 	/**
 	 * Called when the object has disconnected from the delta stream.
 	 */
@@ -557,6 +602,8 @@ export abstract class SharedObjectCore<
 	 * @param local - Whether the message originated from the local client
 	 * @param localOpMetadata - For local client messages, this is the metadata that was submitted with the message.
 	 * For messages from a remote client, this will be undefined.
+	 *
+	 * @deprecated Replaced by {@link SharedObjectCore.processMessages}.
 	 */
 	private process(
 		message: ISequencedDocumentMessage,
@@ -582,23 +629,42 @@ export abstract class SharedObjectCore<
 		this.emitInternal("op", message, local, this);
 	}
 
+	/* eslint-disable jsdoc/check-indentation */
 	/**
-	 * Process messages for this shared object. The messages here are contiguous messages for this object in a batch.
+	 * Process a bunch of messages for this shared object. A bunch is group of messages that have the following properties:
+	 * - They are all part of the same grouped batch, which entails:
+	 *   - They are contiguous in sequencing order.
+	 *   - They are all from the same client.
+	 *   - They are all based on the same reference sequence number.
+	 *   - They are not interleaved with messages from other clients.
+	 * - They are not interleaved with messages from other DDS in the container.
 	 * @param messageCollection - The collection of messages to process.
+	 *
 	 */
+	/* eslint-enable jsdoc/check-indentation */
 	private processMessages(messagesCollection: IRuntimeMessageCollection): void {
-		const { envelope, messagesContent, local } = messagesCollection;
-		for (const { contents, localOpMetadata, clientSequenceNumber } of messagesContent) {
-			this.process(
-				{
-					...envelope,
-					clientSequenceNumber,
-					contents: parseHandles(contents, this.serializer),
-				},
-				local,
+		this.verifyNotClosed(); // This will result in container closure.
+
+		// Decode any handles in the contents before processing the messages.
+		const decodedMessagesContent: IRuntimeMessagesContent[] = [];
+		for (const {
+			contents,
+			localOpMetadata,
+			clientSequenceNumber,
+		} of messagesCollection.messagesContent) {
+			const decodedMessageContent: IRuntimeMessagesContent = {
+				contents: parseHandles(contents, this.serializer),
 				localOpMetadata,
-			);
+				clientSequenceNumber,
+			};
+			decodedMessagesContent.push(decodedMessageContent);
 		}
+
+		const decodedMessagesCollection: IRuntimeMessageCollection = {
+			...messagesCollection,
+			messagesContent: decodedMessagesContent,
+		};
+		this.processMessagesHelper(decodedMessagesCollection);
 	}
 
 	/**
@@ -943,3 +1009,75 @@ function isChannel(loadable: IFluidLoadable): loadable is IChannel {
 	// This assumes no other IFluidLoadable has an `attributes` field, and thus may not be fully robust.
 	return (loadable as IChannel).attributes !== undefined;
 }
+
+/**
+ * Utility that processes the given messages in the message collection together by calling `processMessagesCore`.
+ * This will be called when {@link SharedObjectCore.processMessagesCore} is defined.
+ */
+function processMessagesCoreHelper(
+	messagesCollection: IRuntimeMessageCollection,
+	opProcessingHelper: SampledTelemetryHelper<void, ProcessTelemetryProperties>,
+	emitInternal: (
+		event: "pre-op" | "op",
+		op: ISequencedDocumentMessage,
+		local: boolean,
+	) => void,
+	processMessagesCore: (messagesCollection: IRuntimeMessageCollection) => void,
+): void {
+	const { envelope, local, messagesContent } = messagesCollection;
+
+	const emitEvents = (
+		event: "pre-op" | "op",
+		messagesContentForEvent: readonly IRuntimeMessagesContent[],
+	): void => {
+		for (const { contents, clientSequenceNumber } of messagesContentForEvent) {
+			const message: ISequencedDocumentMessage = {
+				...envelope,
+				contents,
+				clientSequenceNumber,
+			};
+			emitInternal(event, message, local);
+		}
+	};
+
+	emitEvents("pre-op", messagesContent);
+	opProcessingHelper.measure(
+		(): ICustomData<ProcessTelemetryProperties> => {
+			processMessagesCore(messagesCollection);
+			const telemetryProperties: ProcessTelemetryProperties = {
+				sequenceDifference: envelope.sequenceNumber - envelope.referenceSequenceNumber,
+			};
+			return {
+				customData: telemetryProperties,
+			};
+		},
+		local ? "local" : "remote",
+	);
+	emitEvents("op", messagesContent);
+}
+
+/**
+ * Utility that processes the given messages in the message collection one by one by calling `process`. This will
+ * be called when {@link SharedObjectCore.processMessagesCore} is not defined.
+ */
+function processHelper(
+	messagesCollection: IRuntimeMessageCollection,
+	process: (
+		message: ISequencedDocumentMessage,
+		local: boolean,
+		localOpMetadata: unknown,
+	) => void,
+): void {
+	const { envelope, local, messagesContent } = messagesCollection;
+	for (const { contents, localOpMetadata, clientSequenceNumber } of messagesContent) {
+		process(
+			{
+				...envelope,
+				contents,
+				clientSequenceNumber,
+			},
+			local,
+			localOpMetadata,
+		);
+	}
+}