Add stagingModeAutoFlushThreshold for staging mode batch control

.claude/skills/trigger-pipelines-for-copilot-pr/SKILL.md

@@ -26,7 +26,8 @@ Second comment:
 Posting those comments will trigger all our pipelines, which is necessary for PRs that are created by Copilot.
 
 To post the comments first check if the GitHub CLI is available,
-and if so use `gh pr comment <PULL_REQUEST_NUMBER> --repo microsoft/FluidFramework --body "<COMMENT_TEXT>"`.
+and if so use `MSYS_NO_PATHCONV=1 gh pr comment <PULL_REQUEST_NUMBER> --repo microsoft/FluidFramework --body "<COMMENT_TEXT>"`.
+Note: `MSYS_NO_PATHCONV=1` is required on Windows (Git Bash) to prevent `/azp` from being expanded to `C:/Program Files/Git/azp`.
 If `gh` is not available but `$GITHUB_TOKEN` is, you can try the GitHub REST API directly, e.g.:
 
 ```

packages/runtime/container-runtime/api-report/container-runtime.legacy.beta.api.md

@@ -46,6 +46,7 @@ export interface ContainerRuntimeOptions {
     readonly gcOptions: IGCRuntimeOptions;
     readonly loadSequenceNumberVerification: "close" | "log" | "bypass";
     readonly maxBatchSizeInBytes: number;
+    readonly stagingModeAutoFlushThreshold: number;
     // (undocumented)
     readonly summaryOptions: ISummaryRuntimeOptions;
 }

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

@@ -43,6 +43,7 @@ export type RuntimeOptionsAffectingDocSchema = Omit<
 	| "maxBatchSizeInBytes"
 	| "loadSequenceNumberVerification"
 	| "summaryOptions"
+	| "stagingModeAutoFlushThreshold"
 	| "disableSchemaUpgrade"
 >;
 

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

@@ -242,6 +242,7 @@ import {
 	DuplicateBatchDetector,
 	ensureContentsDeserialized,
 	type IBatchCheckpoint,
+	largeBatchThreshold,
 	OpCompressor,
 	OpDecompressor,
 	OpGroupingManager,
@@ -257,6 +258,7 @@ import {
 	type IPendingLocalState,
 	PendingStateManager,
 	type PendingBatchResubmitMetadata,
+	type IPendingMessage,
 } from "./pendingStateManager.js";
 import { BatchRunCounter, RunCounter } from "./runCounter.js";
 import {
@@ -475,6 +477,18 @@ export interface ContainerRuntimeOptions {
 	 */
 	readonly createBlobPayloadPending: true | undefined;
 
+	/**
+	 * Controls automatic batch flushing during staging mode.
+	 * Normal turn-based/async flush scheduling is suppressed while in staging mode
+	 * until the accumulated batch reaches this many ops, at which point the batch
+	 * is flushed. Incoming ops always break the current batch regardless of this setting.
+	 *
+	 * Set to Infinity to only break batches on system events (incoming ops).
+	 *
+	 * @defaultValue `largeBatchThreshold` (currently 1000)
+	 */
+	readonly stagingModeAutoFlushThreshold: number;
+
 	/**
 	 * When this property is set to true, the runtime will never send DocumentSchemaChange ops
 	 * and will throw an error if any incoming DocumentSchemaChange ops are received.
@@ -613,6 +627,16 @@ const defaultMaxBatchSizeInBytes = 700 * 1024;
 
 const defaultChunkSizeInBytes = 204800;
 
+/**
+ * Default maximum ops per staging-mode batch before automatic flush scheduling resumes.
+ *
+ * Chosen based on production telemetry: copy-paste operations routinely produce batches
+ * of 1000+ ops (435K instances over 30 days), and receivers on modern Fluid versions
+ * handle them without issues. Uses {@link largeBatchThreshold} to stay aligned with
+ * the existing "large batch" telemetry threshold ({@link OpGroupingManager}).
+ */
+const defaultStagingModeAutoFlushThreshold = largeBatchThreshold;
+
 /**
  * The default time to wait for pending ops to be processed during summarization
  */
@@ -967,6 +991,7 @@ export class ContainerRuntime
 			loadSequenceNumberVerification: "close",
 			maxBatchSizeInBytes: defaultMaxBatchSizeInBytes,
 			chunkSizeInBytes: defaultChunkSizeInBytes,
+			stagingModeAutoFlushThreshold: defaultStagingModeAutoFlushThreshold,
 			disableSchemaUpgrade: false,
 		};
 
@@ -993,6 +1018,7 @@ export class ContainerRuntime
 				? disabledCompressionConfig
 				: defaultConfigs.compressionOptions,
 			createBlobPayloadPending = defaultConfigs.createBlobPayloadPending,
+			stagingModeAutoFlushThreshold = defaultConfigs.stagingModeAutoFlushThreshold,
 			disableSchemaUpgrade = defaultConfigs.disableSchemaUpgrade,
 		}: IContainerRuntimeOptionsInternal = runtimeOptions;
 
@@ -1223,6 +1249,7 @@ export class ContainerRuntime
 			enableGroupedBatching,
 			explicitSchemaControl,
 			createBlobPayloadPending,
+			stagingModeAutoFlushThreshold,
 			disableSchemaUpgrade,
 		};
 
@@ -1402,6 +1429,7 @@ export class ContainerRuntime
 
 	private readonly batchRunner = new BatchRunCounter();
 	private readonly _flushMode: FlushMode;
+	private readonly stagingModeAutoFlushThreshold: number;
 	/**
 	 * BatchId tracking is needed whenever there's a possibility of a "forked Container",
 	 * where the same local state is pending in two different running Containers, each of
@@ -1834,6 +1862,10 @@ export class ContainerRuntime
 			this.closeFn(error);
 			throw error;
 		}
+		this.stagingModeAutoFlushThreshold =
+			this.mc.config.getNumber("Fluid.ContainerRuntime.StagingModeAutoFlushThreshold") ??
+			runtimeOptions.stagingModeAutoFlushThreshold ??
+			defaultStagingModeAutoFlushThreshold;
 		this.batchIdTrackingEnabled =
 			this.mc.config.getBoolean("Fluid.Container.enableOfflineFull") ??
 			this.mc.config.getBoolean("Fluid.ContainerRuntime.enableBatchIdTracking") ??
@@ -3613,20 +3645,39 @@ export class ContainerRuntime
 		// since we mark whole batches as "staged" or not to indicate whether to submit them.
 		this.flush();
 
-		const exitStagingMode = (discardOrCommit: () => void): void => {
+		const exitStagingMode = (
+			discardOrCommit: () => IPendingMessage["batchInfo"][],
+			exitMethod: "commit" | "discard",
+		): void => {
 			try {
-				// Final flush of any last staged changes
-				// NOTE: We can't use this.flush() here, because orderSequentially uses StagingMode and in the rollback case we'll hit assert 0x24c
-				this.outbox.flush();
-
-				this.stageControls = undefined;
-
-				// During Staging Mode, we avoid submitting any ID Allocation ops (apart from resubmitting pre-staging ops).
-				// Now that we've exited, we need to submit an ID Allocation op for any IDs that were generated while in Staging Mode.
-				this.submitIdAllocationOpIfNeeded({ staged: false });
-				discardOrCommit();
-
-				this.channelCollection.notifyStagingMode(false);
+				PerformanceEvent.timedExec(
+					this.mc.logger,
+					{
+						eventName: `ExitStagingMode_${exitMethod}`,
+					},
+					(event) => {
+						// Final flush of any last staged changes
+						// NOTE: We can't use this.flush() here, because orderSequentially uses StagingMode and in the rollback case we'll hit assert 0x24c
+						this.outbox.flush();
+
+						this.stageControls = undefined;
+
+						// During Staging Mode, we avoid submitting any ID Allocation ops (apart from resubmitting pre-staging ops).
+						// Now that we've exited, we need to submit an ID Allocation op for any IDs that were generated while in Staging Mode.
+						this.submitIdAllocationOpIfNeeded({ staged: false });
+						const batchInfos = discardOrCommit();
+						event.reportProgress({
+							details: {
+								autoFlushThreshold: this.stagingModeAutoFlushThreshold,
+								batches: batchInfos.length,
+								batchesAtOrOverThreshold: batchInfos.filter(
+									(b) => b.length >= this.stagingModeAutoFlushThreshold,
+								).length,
+							},
+						});
+						this.channelCollection.notifyStagingMode(false);
+					},
+				);
 			} catch (error) {
 				const normalizedError = normalizeError(error);
 				this.closeFn(normalizedError);
@@ -3638,21 +3689,24 @@ export class ContainerRuntime
 			discardChanges: () =>
 				exitStagingMode(() => {
 					// Pop all staged batches from the PSM and roll them back in LIFO order
-					this.pendingStateManager.popStagedBatches(({ runtimeOp, localOpMetadata }) => {
-						this.rollbackStagedChange(runtimeOp, localOpMetadata);
-					});
+					const batchInfos = this.pendingStateManager.popStagedBatches(
+						({ runtimeOp, localOpMetadata }) => {
+							this.rollbackStagedChange(runtimeOp, localOpMetadata);
+						},
+					);
 					this.updateDocumentDirtyState();
-				}),
+					return batchInfos;
+				}, "discard"),
 			commitChanges: (options) => {
 				const { squash } = { ...defaultStagingCommitOptions, ...options };
 				exitStagingMode(() => {
 					// Replay all staged batches in typical FIFO order.
 					// We'll be out of staging mode so they'll be sent to the service finally.
-					this.pendingStateManager.replayPendingStates({
+					return this.pendingStateManager.replayPendingStates({
 						committingStagedBatches: true,
 						squash,
 					});
-				});
+				}, "commit");
 			},
 		};
 
@@ -4783,6 +4837,20 @@ export class ContainerRuntime
 	}
 
 	private scheduleFlush(): void {
+		// During staging mode, suppress automatic flush scheduling until the main batch
+		// reaches or exceeds the threshold.
+		// Incoming ops still break the batch via direct this.flush() calls elsewhere
+		// (deltaManager "op" handler, process(), connection changes, getPendingLocalState,
+		// exitStagingMode). Those all bypass scheduleFlush(), so they're unaffected by this check.
+		// Additionally, outbox.maybeFlushPartialBatch() (called on every submit) detects
+		// sequence number changes and throws if unexpected changes are detected.
+		if (
+			this.inStagingMode &&
+			this.outbox.mainBatchMessageCount < this.stagingModeAutoFlushThreshold
+		) {
+			return;
+		}
+
 		if (this.flushScheduled) {
 			return;
 		}

packages/runtime/container-runtime/src/opLifecycle/index.ts

@@ -45,6 +45,7 @@ export {
 } from "./remoteMessageProcessor.js";
 export {
 	type EmptyGroupedBatch,
+	largeBatchThreshold,
 	OpGroupingManager,
 	type OpGroupingManagerConfig,
 	isGroupedBatch,

packages/runtime/container-runtime/src/opLifecycle/opGroupingManager.ts

@@ -17,6 +17,13 @@ import type {
 	OutboundSingletonBatch,
 } from "./definitions.js";
 
+/**
+ * The number of ops in a batch above which the batch is considered "large"
+ * for telemetry purposes. Used by both {@link OpGroupingManager} (GroupLargeBatch event)
+ * and as the default staging-mode auto-flush threshold.
+ */
+export const largeBatchThreshold = 1000;
+
 /**
  * Grouping makes assumptions about the shape of message contents. This interface codifies those assumptions, but does not validate them.
  */
@@ -123,7 +130,10 @@ export class OpGroupingManager {
 			return batch as OutboundSingletonBatch;
 		}
 
-		if (batch.messages.length >= 1000) {
+		// Use > (not >=) so that batches flushed exactly at the staging-mode
+		// auto-flush threshold (which defaults to largeBatchThreshold) don't
+		// trigger this event. Only genuinely oversized batches are logged.
+		if (batch.messages.length > largeBatchThreshold) {
 			this.logger.sendTelemetryEvent({
 				eventName: "GroupLargeBatch",
 				length: batch.messages.length,

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

@@ -431,7 +431,7 @@ export class PendingStateManager implements IDisposable {
 			clientId !== undefined,
 			0xa33 /* clientId (from stateHandler) could only be undefined if we've never connected, but we have a CSN so we know that's not the case */,
 		);
-
+		const batchInfo = { clientId, batchStartCsn, length: batch.length, ignoreBatchId, staged };
 		for (const message of batch) {
 			const {
 				runtimeOp,
@@ -446,8 +446,7 @@ export class PendingStateManager implements IDisposable {
 				runtimeOp,
 				localOpMetadata,
 				opMetadata,
-				// Note: We only will read this off the first message, but put it on all for simplicity
-				batchInfo: { clientId, batchStartCsn, length: batch.length, ignoreBatchId, staged },
+				batchInfo,
 			};
 			this.pendingMessages.push(pendingMessage);
 		}
@@ -751,8 +750,12 @@ export class PendingStateManager implements IDisposable {
 	 * Called when the Container's connection state changes. If the Container gets connected, it replays all the pending
 	 * states in its queue. This includes triggering resubmission of unacked ops.
 	 * ! Note: successfully resubmitting an op that has been successfully sequenced is not possible due to checks in the ConnectionStateHandler (Loader layer)
+	 *
+	 * @returns The unique batch infos for all batches that were replayed.
 	 */
-	public replayPendingStates(options?: ReplayPendingStateOptions): void {
+	public replayPendingStates(
+		options?: ReplayPendingStateOptions,
+	): IPendingMessage["batchInfo"][] {
 		const { committingStagedBatches, squash } = {
 			...defaultReplayPendingStatesOptions,
 			...options,
@@ -779,6 +782,7 @@ export class PendingStateManager implements IDisposable {
 
 		const initialPendingMessagesCount = this.pendingMessages.length;
 		let remainingPendingMessagesCount = this.pendingMessages.length;
+		const replayedBatchSet = new Set<IPendingMessage["batchInfo"]>();
 
 		let seenStagedBatch = false;
 
@@ -816,6 +820,7 @@ export class PendingStateManager implements IDisposable {
 			if (asEmptyBatchLocalOpMetadata(pendingMessage.localOpMetadata)?.emptyBatch === true) {
 				// Resubmit no messages, with the batchId. Will result in another empty batch marker.
 				this.stateHandler.reSubmitBatch([], { batchId, staged, squash });
+				replayedBatchSet.add(pendingMessage.batchInfo);
 				continue;
 			}
 
@@ -842,6 +847,7 @@ export class PendingStateManager implements IDisposable {
 					],
 					{ batchId, staged, squash },
 				);
+				replayedBatchSet.add(pendingMessage.batchInfo);
 				continue;
 			}
 			// else: batchMetadataFlag === true  (It's a typical multi-message batch)
@@ -881,6 +887,7 @@ export class PendingStateManager implements IDisposable {
 			}
 
 			this.stateHandler.reSubmitBatch(batch, { batchId, staged, squash });
+			replayedBatchSet.add(pendingMessage.batchInfo);
 		}
 
 		if (!committingStagedBatches) {
@@ -898,6 +905,8 @@ export class PendingStateManager implements IDisposable {
 				clientId: this.stateHandler.clientId(),
 			});
 		}
+
+		return [...replayedBatchSet];
 	}
 
 	/**
@@ -908,11 +917,13 @@ export class PendingStateManager implements IDisposable {
 			// callback will only be given staged messages with a valid runtime op (i.e. not empty batch and not an initial message with only serialized content)
 			stagedMessage: IPendingMessage & { runtimeOp: LocalContainerRuntimeMessage },
 		) => void,
-	): void {
+	): IPendingMessage["batchInfo"][] {
+		const batchSet = new Set<IPendingMessage["batchInfo"]>();
 		while (!this.pendingMessages.isEmpty()) {
 			const stagedMessage = this.pendingMessages.peekBack();
 			if (stagedMessage?.batchInfo.staged === true) {
 				this.pendingMessages.pop();
+				batchSet.add(stagedMessage.batchInfo);
 
 				if (hasTypicalRuntimeOp(stagedMessage)) {
 					callback(stagedMessage);
@@ -925,6 +936,7 @@ export class PendingStateManager implements IDisposable {
 			this.pendingMessages.toArray().every((m) => m.batchInfo.staged !== true),
 			0xb89 /* Shouldn't be any more staged messages */,
 		);
+		return [...batchSet];
 	}
 }