Fix assert 0xb7b (microsoft#25735)

There's a corner case where processing incoming ops (join/leave ops in
particular) can trigger a connection state transition which can trigger
a flush (but there are no changes to flush), and assert 0xb7b fires - it
asserts that we should never flush while processing incoming ops. This
protects against reentrancy, which is not an issue in this case.

The fix is to only assert if we have something to flush.

Author: markfields

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

@@ -365,6 +365,16 @@ export class Outbox {
 	 * @param resubmitInfo - Key information when flushing a resubmitted batch. Undefined means this is not resubmit.
 	 */
 	public flush(resubmitInfo?: BatchResubmitInfo): void {
+		// We have nothing to flush if all batchManagers are empty, and we we're not needing to resubmit an empty batch placeholder
+		if (
+			this.idAllocationBatch.empty &&
+			this.blobAttachBatch.empty &&
+			this.mainBatch.empty &&
+			resubmitInfo?.batchId === undefined
+		) {
+			return;
+		}
+
 		assert(
 			!this.isContextReentrant(),
 			0xb7b /* Flushing must not happen while incoming changes are being processed */,

packages/runtime/container-runtime/src/test/opLifecycle/outbox.spec.ts

@@ -17,6 +17,7 @@ import type {
 	ISequencedDocumentMessage,
 } from "@fluidframework/driver-definitions/internal";
 import { MockLogger } from "@fluidframework/telemetry-utils/internal";
+import { validateAssertionError } from "@fluidframework/test-runtime-utils/internal";
 
 import type { ICompressionRuntimeOptions } from "../../compressionDefinitions.js";
 import { CompressionAlgorithms } from "../../compressionDefinitions.js";
@@ -1092,6 +1093,57 @@ describe("Outbox", () => {
 			assert.strictEqual(state.opsResubmitted, opsResubmitted, "unexpected opsResubmitted");
 		}
 
+		it("should not assert when flushing while reentrant with empty batches", () => {
+			const outbox = getOutbox({
+				context: getMockContext(),
+				opGroupingConfig: {
+					groupedBatchingEnabled: true,
+				},
+			});
+
+			// Mark context as reentrant
+			state.isReentrant = true;
+
+			// Flush with no messages - should not throw
+			assert.doesNotThrow(() => {
+				outbox.flush();
+			}, "Should not assert when flushing empty batches while reentrant");
+
+			validateCounts(0, 0, 0);
+		});
+
+		it("should assert when flushing while reentrant with non-empty batches", () => {
+			const outbox = getOutbox({
+				context: getMockContext(),
+				opGroupingConfig: {
+					groupedBatchingEnabled: true,
+				},
+			});
+
+			const messages = [createMessage(ContainerMessageType.FluidDataStoreOp, "0")];
+
+			// Submit a message (not reentrant)
+			state.isReentrant = false;
+			outbox.submit(messages[0]);
+
+			// Now mark context as reentrant and try to flush - should throw
+			state.isReentrant = true;
+
+			assert.throws(
+				() => outbox.flush(),
+				(error: Error) => {
+					return validateAssertionError(
+						error,
+						"Flushing must not happen while incoming changes are being processed",
+					);
+				},
+				"Should assert when flushing non-empty batches while reentrant",
+			);
+
+			// Verify nothing was submitted
+			validateCounts(0, 0, 0);
+		});
+
 		it("batch has reentrant ops, but grouped batching is off", () => {
 			const outbox = getOutbox({
 				context: getMockContext(),