feat(dds): squash-on-resubmit for non-tree DDSes

.changeset/squash-non-tree-dds.md

@@ -0,0 +1,38 @@
+---
+"@fluidframework/cell": minor
+"@fluidframework/counter": minor
+"@fluid-experimental/ink": minor
+"@fluidframework/legacy-dds": minor
+"@fluidframework/map": minor
+"@fluidframework/matrix": minor
+"@fluidframework/merge-tree": minor
+"@fluidframework/ordered-collection": minor
+"@fluid-experimental/pact-map": minor
+"@fluidframework/register-collection": minor
+"@fluidframework/sequence": minor
+"@fluidframework/task-manager": minor
+---
+
+Implement squash-on-resubmit for non-tree DDSes
+
+All non-tree DDSes now have explicit `reSubmitSquashed` overrides so staging-mode commits (`commitChanges({squash: true})`) can drop intermediate values before they reach the wire. Values written and removed within a single staging session (for example, a sensitive string set and then deleted before commit) are no longer transmitted as part of the squashed batch.
+
+The model is uniform across DDSes: the runtime walks staged pending changes oldest-to-newest and asks the DDS, for each change, whether a later staged change subsumes it. Subsumed changes are dropped (with the same kind of pending-state cleanup that rollback performs); non-subsumed changes are resubmitted unchanged. Pre-staging ops still in flight are never touched.
+
+Per-DDS treatment:
+
+- `SharedCell`, `SharedMap`, `SharedDirectory`, `SharedMatrix`: subsumption-aware squash drops superseded ops (per-cell / per-key LWW; for `clear` and `delete`, a later clear or a later op on the same key subsumes). For `SharedDirectory` subdirectory lifecycle ops, a staged `createSubDirectory(name) + deleteSubDirectory(name)` pair is also dropped so user-supplied subdirectory names don't leak when the pair nets to no-op.
+- `SharedCounter`, `SharedTaskManager`: identity squash—increments and volunteer/abandon ops carry intent that is not subsumable by a later staged op of the same shape.
+- `SharedSequence` text and endpoints: unchanged—squash was already wired end-to-end via merge-tree's `regeneratePendingOp(squash)`.
+- `SharedSequence` / `SharedString` segment properties: merge-tree's `resetPendingDeltaToOps` now filters `annotate` `props` and `insert` `seg.properties` against keys touched by later staged annotates on the same segment. If every key is overridden the op is dropped entirely.
+- `SharedSequence` interval properties: `IntervalCollection.resubmitMessage` now filters `add` / `change` `value.properties` against keys touched by later staged `add` / `change` ops on the same interval, and drops an `add` / `change` entirely when a later `delete` on the same interval subsumes it.
+- legacy `SharedArray`: subsumption-aware squash drops a staged `insertEntry` (with its user-supplied `value`) when a later staged `deleteEntry`, deleting `toggle`, or move chain ending in either subsumes it; intervening `toggleMove` ops disable the optimization and the chain is resubmitted unchanged.
+- `Ink`, `ConsensusRegisterCollection`, `ConsensusOrderedCollection`, `PactMap`, legacy `SharedSignal`: identity squash with documented rationale. These DDSes have append-only, order-preserving, or consensus-bound semantics where collapsing pending ops would change observable behavior.
+
+Together this removes the dependency on the `Fluid.SharedObject.AllowStagingModeWithoutSquashing` configuration flag fallback for the listed DDSes.
+
+Known limitations (documented in code; not addressed by these changes):
+
+- `ConsensusOrderedCollection.add` carries a serialized user value; an `add(secret) → acquire → complete` chain inside a staging session still transmits the `add` op on commit.
+- `ConsensusRegisterCollection` writes participate in `readVersions()` history; collapsing pending writes would alter observable semantics, so intermediate writes during staging remain visible.
+- `Ink` and legacy `SharedSignal` ops carry user-supplied pen / metadata; staging-mode notifications are intentionally transmitted on commit.

packages/dds/cell/package.json

@@ -107,14 +107,17 @@
 	"devDependencies": {
 		"@arethetypeswrong/cli": "^0.18.2",
 		"@biomejs/biome": "~2.4.5",
+		"@fluid-internal/client-utils": "workspace:~",
 		"@fluid-internal/mocha-test-setup": "workspace:~",
+		"@fluid-private/stochastic-test-utils": "workspace:~",
 		"@fluid-private/test-dds-utils": "workspace:~",
 		"@fluid-tools/build-cli": "catalog:buildTools",
 		"@fluidframework/build-common": "^2.0.3",
 		"@fluidframework/build-tools": "catalog:buildTools",
 		"@fluidframework/cell-previous": "npm:@fluidframework/cell@2.92.0",
 		"@fluidframework/container-definitions": "workspace:~",
 		"@fluidframework/eslint-config-fluid": "catalog:eslint",
+		"@fluidframework/runtime-utils": "workspace:~",
 		"@fluidframework/test-runtime-utils": "workspace:~",
 		"@microsoft/api-extractor": "7.58.1",
 		"@types/mocha": "^10.0.10",

packages/dds/cell/src/cell.ts

@@ -334,6 +334,23 @@ export class SharedCell<T = any>
 		}
 	}
 
+	/**
+	 * Cell is last-write-wins: any pending op other than the latest is superseded.
+	 * We drop superseded ops entirely (no wire emission) and resubmit only the final pending op,
+	 * which by construction represents the cell's tip state.
+	 */
+	protected override reSubmitSquashed(content: unknown, localOpMetadata: unknown): void {
+		const cellOpMetadata = localOpMetadata as ICellLocalOpMetadata;
+		const lastPendingMessageId = this.pendingMessageIds[this.pendingMessageIds.length - 1];
+		if (cellOpMetadata.pendingMessageId === lastPendingMessageId) {
+			this.submitLocalMessage(content, localOpMetadata);
+		} else {
+			const index = this.pendingMessageIds.indexOf(cellOpMetadata.pendingMessageId);
+			assert(index !== -1, 0xd01 /* Pending message id missing from queue during squash */);
+			this.pendingMessageIds.splice(index, 1);
+		}
+	}
+
 	/**
 	 * Rollback a local op.
 	 *

packages/dds/cell/src/test/cell.spec.ts

@@ -5,7 +5,12 @@
 
 import { strict as assert } from "node:assert";
 
-import { type IGCTestProvider, runGCTests } from "@fluid-private/test-dds-utils";
+import {
+	enterStagingMode,
+	type IGCTestProvider,
+	reconnectAndSquash,
+	runGCTests,
+} from "@fluid-private/test-dds-utils";
 import { AttachState } from "@fluidframework/container-definitions";
 import {
 	MockContainerRuntimeFactory,
@@ -488,6 +493,142 @@ describe("Cell", () => {
 		});
 	});
 
+	describe("Squash on resubmit", () => {
+		let containerRuntimeFactory: MockContainerRuntimeFactoryForReconnection;
+		let dataStoreRuntime1: MockFluidDataStoreRuntime;
+		let containerRuntime1: MockContainerRuntimeForReconnection;
+		let cell1: ISharedCell;
+		let cell2: ISharedCell;
+		let peerObservations: { event: "valueChanged" | "delete"; value: unknown }[];
+
+		function createCellForSquash(id: string): {
+			cell: ISharedCell;
+			dataStoreRuntime: MockFluidDataStoreRuntime;
+			containerRuntime: MockContainerRuntimeForReconnection;
+		} {
+			const dataStoreRuntime = new MockFluidDataStoreRuntime();
+			const containerRuntime =
+				containerRuntimeFactory.createContainerRuntime(dataStoreRuntime);
+			const services = {
+				deltaConnection: dataStoreRuntime.createDeltaConnection(),
+				objectStorage: new MockStorage(),
+			};
+			const cell = new SharedCell(id, dataStoreRuntime, CellFactory.Attributes);
+			cell.connect(services);
+			return { cell, dataStoreRuntime, containerRuntime };
+		}
+
+		beforeEach("createCellsForSquash", () => {
+			containerRuntimeFactory = new MockContainerRuntimeFactoryForReconnection();
+			const response1 = createCellForSquash("cell1");
+			cell1 = response1.cell;
+			dataStoreRuntime1 = response1.dataStoreRuntime;
+			containerRuntime1 = response1.containerRuntime;
+			const response2 = createCellForSquash("cell2");
+			cell2 = response2.cell;
+			peerObservations = [];
+			cell2.on("valueChanged", (value) =>
+				peerObservations.push({ event: "valueChanged", value }),
+			);
+			cell2.on("delete", () => peerObservations.push({ event: "delete", value: undefined }));
+		});
+
+		it("drops intermediate set when a later delete supersedes it", () => {
+			const secret = "SSN: 123-45-6789";
+			containerRuntime1.connected = false;
+			cell1.set(secret);
+			cell1.delete();
+			reconnectAndSquash(containerRuntime1, dataStoreRuntime1);
+			containerRuntimeFactory.processAllMessages();
+
+			assert.equal(cell1.get(), undefined, "cell1 final state should be empty");
+			assert.equal(cell2.get(), undefined, "cell2 final state should be empty");
+			assert.deepEqual(
+				peerObservations,
+				[{ event: "delete", value: undefined }],
+				"peer must only see the final delete; intermediate set must not leak",
+			);
+			for (const observation of peerObservations) {
+				assert.notEqual(
+					observation.value,
+					secret,
+					"secret value must never appear on the wire",
+				);
+			}
+		});
+
+		it("drops intermediate sets when a later set supersedes them", () => {
+			const secret = "secret-intermediate";
+			const finalValue = "final-value";
+			containerRuntime1.connected = false;
+			cell1.set(secret);
+			cell1.set(finalValue);
+			reconnectAndSquash(containerRuntime1, dataStoreRuntime1);
+			containerRuntimeFactory.processAllMessages();
+
+			assert.equal(cell1.get(), finalValue);
+			assert.equal(cell2.get(), finalValue);
+			assert.deepEqual(
+				peerObservations,
+				[{ event: "valueChanged", value: finalValue }],
+				"peer should see exactly one set carrying the final value",
+			);
+		});
+
+		it("squashes a long pending chain to the final state only", () => {
+			containerRuntime1.connected = false;
+			cell1.set("a");
+			cell1.set("b");
+			cell1.delete();
+			cell1.set("c");
+			cell1.delete();
+			cell1.set("d");
+			reconnectAndSquash(containerRuntime1, dataStoreRuntime1);
+			containerRuntimeFactory.processAllMessages();
+
+			assert.equal(cell1.get(), "d");
+			assert.equal(cell2.get(), "d");
+			assert.equal(peerObservations.length, 1);
+			assert.deepEqual(peerObservations[0], { event: "valueChanged", value: "d" });
+		});
+
+		it("passes through a single pending op unchanged", () => {
+			containerRuntime1.connected = false;
+			cell1.set("only");
+			reconnectAndSquash(containerRuntime1, dataStoreRuntime1);
+			containerRuntimeFactory.processAllMessages();
+
+			assert.equal(cell2.get(), "only");
+			assert.deepEqual(peerObservations, [{ event: "valueChanged", value: "only" }]);
+		});
+
+		it("preserves a pre-staging set still in flight when a staging set is squashed", () => {
+			// Submit a pre-staging set while connected so it's in flight at the runtime layer
+			// but not yet ACKed when we disconnect. The staging-mode set + delete should leave
+			// the pre-staging value to land normally, and "secret" must never reach the peer.
+			cell1.set("pre");
+			enterStagingMode(containerRuntime1);
+			cell1.set("secret");
+			cell1.delete();
+			reconnectAndSquash(containerRuntime1, dataStoreRuntime1);
+			containerRuntimeFactory.processAllMessages();
+
+			assert.equal(cell1.get(), undefined);
+			assert.equal(cell2.get(), undefined);
+			const sawPre = peerObservations.some(
+				(observation) => observation.event === "valueChanged" && observation.value === "pre",
+			);
+			assert.equal(
+				sawPre,
+				true,
+				"pre-staging set must land on the peer when only the staged ops are squashed",
+			);
+			for (const observation of peerObservations) {
+				assert.notEqual(observation.value, "secret", "staged secret must not leak");
+			}
+		});
+	});
+
 	describe("Garbage Collection", () => {
 		class GCSharedCellProvider implements IGCTestProvider {
 			private subCellCount = 0;

packages/dds/cell/src/test/cell.squash.fuzz.spec.ts

@@ -0,0 +1,142 @@
+/*!
+ * Copyright (c) Microsoft Corporation and contributors. All rights reserved.
+ * Licensed under the MIT License.
+ */
+
+import { strict as assert } from "node:assert";
+import * as path from "node:path";
+
+import { TypedEventEmitter } from "@fluid-internal/client-utils";
+import {
+	type Generator,
+	createWeightedAsyncGenerator,
+	done,
+	takeAsync,
+} from "@fluid-private/stochastic-test-utils";
+import {
+	type DDSFuzzHarnessEvents,
+	type SquashFuzzModel,
+	type SquashFuzzTestState,
+	createSquashFuzzSuite,
+} from "@fluid-private/test-dds-utils";
+import { isFluidHandle } from "@fluidframework/runtime-utils/internal";
+
+import { CellFactory } from "../cellFactory.js";
+
+import { _dirname } from "./dirname.cjs";
+
+interface SetOp {
+	type: "set";
+	value: string | number;
+}
+interface SetPoisonedOp {
+	type: "setPoisoned";
+}
+interface DeleteOp {
+	type: "delete";
+}
+
+type SquashOp = SetOp | SetPoisonedOp | DeleteOp;
+
+type FuzzState = SquashFuzzTestState<CellFactory>;
+
+function isPoisonedHandle(value: unknown): boolean {
+	return (
+		isFluidHandle(value) && (value as unknown as { poisoned?: unknown }).poisoned === true
+	);
+}
+
+function makeGenerator(): (state: FuzzState) => Promise<SquashOp | typeof done> {
+	const isInStaging = (state: FuzzState): boolean =>
+		state.client.stagingModeStatus === "staging";
+
+	const setOp = async (state: FuzzState): Promise<SetOp> => ({
+		type: "set",
+		value: state.random.pick([
+			(): string => state.random.string(state.random.integer(1, 4)),
+			(): number => state.random.integer(0, 100),
+		])(),
+	});
+	const setPoisoned = async (): Promise<SetPoisonedOp> => ({ type: "setPoisoned" });
+	const deleteOp = async (): Promise<DeleteOp> => ({ type: "delete" });
+
+	return createWeightedAsyncGenerator<SquashOp, FuzzState>([
+		[setOp, 6],
+		[setPoisoned, 4, isInStaging],
+		[deleteOp, 3],
+	]);
+}
+
+function makeExitingGenerator(): Generator<SquashOp, FuzzState> {
+	return (state): SquashOp | typeof done => {
+		const value = state.client.channel.get();
+		if (isPoisonedHandle(value)) {
+			return { type: "delete" };
+		}
+		return done;
+	};
+}
+
+function reducer(state: FuzzState, op: SquashOp): void {
+	const { client } = state;
+	switch (op.type) {
+		case "set": {
+			client.channel.set(op.value);
+			break;
+		}
+		case "setPoisoned": {
+			client.channel.set(state.random.poisonedHandle());
+			break;
+		}
+		case "delete": {
+			client.channel.delete();
+			break;
+		}
+		default: {
+			break;
+		}
+	}
+}
+
+const squashModel: SquashFuzzModel<CellFactory, SquashOp> = {
+	workloadName: "cell squashing",
+	generatorFactory: () => takeAsync(60, makeGenerator()),
+	reducer,
+	validateConsistency: async (a, b) => {
+		const vA: unknown = a.channel.get();
+		const vB: unknown = b.channel.get();
+		if (isFluidHandle(vA)) {
+			assert(isFluidHandle(vB));
+		} else {
+			assert.equal(vA, vB);
+		}
+	},
+	factory: new CellFactory(),
+	exitingStagingModeGeneratorFactory: makeExitingGenerator,
+	validatePoisonedContentRemoved: (client) => {
+		const value: unknown = client.channel.get();
+		assert(
+			!isPoisonedHandle(value),
+			"Poisoned handle in cell not removed before exiting staging",
+		);
+	},
+};
+
+const emitter = new TypedEventEmitter<DDSFuzzHarnessEvents>();
+
+describe("SharedCell squash fuzz", () => {
+	createSquashFuzzSuite(squashModel, {
+		validationStrategy: { type: "fixedInterval", interval: 10 },
+		reconnectProbability: 0.1,
+		numberOfClients: 3,
+		clientJoinOptions: {
+			maxNumberOfClients: 4,
+			clientAddProbability: 0.05,
+		},
+		detachedStartOptions: { numOpsBeforeAttach: 0 },
+		defaultTestCount: 50,
+		saveFailures: { directory: path.join(_dirname, "../../src/test/results-squash-cell") },
+		emitter,
+		stagingMode: { changeStagingModeProbability: 0.15 },
+	});
+});

packages/dds/cell/src/test/dirname.cts

@@ -0,0 +1,15 @@
+/*!
+ * Copyright (c) Microsoft Corporation and contributors. All rights reserved.
+ * Licensed under the MIT License.
+ */
+
+/**
+ * Problem:
+ * - `__dirname` is not defined in ESM
+ * - `import.meta.url` is not defined in CJS
+ * Solution:
+ * - Export '__dirname' from a .cjs file in the same directory.
+ *
+ * Note that *.cjs files are always CommonJS, but can be imported from ESM.
+ */
+export const _dirname = __dirname;