Support rebasing ops for testing. Part 1, the mock.

api-report/test-runtime-utils.api.md

@@ -115,6 +115,8 @@ export class MockContainerRuntime {
     // (undocumented)
     process(message: ISequencedDocumentMessage): void;
     // (undocumented)
+    protected get referenceSequenceNumber(): number;
+    // (undocumented)
     submit(messageContent: any, localOpMetadata: unknown): number;
 }
 
@@ -142,6 +144,20 @@ export class MockContainerRuntimeFactory {
     sequenceNumber: number;
 }
 
+// @public
+export class MockContainerRuntimeFactoryForRebasing extends MockContainerRuntimeFactory {
+    // (undocumented)
+    createContainerRuntime(dataStoreRuntime: MockFluidDataStoreRuntime, overrides?: {
+        minimumSequenceNumber?: number;
+    }): MockContainerRuntimeForRebasing;
+    // (undocumented)
+    processAllMessages(): void;
+    // (undocumented)
+    processOneMessage(): void;
+    // (undocumented)
+    processSomeMessages(count: number): void;
+}
+
 // @public
 export class MockContainerRuntimeFactoryForReconnection extends MockContainerRuntimeFactory {
     // (undocumented)
@@ -152,6 +168,19 @@ export class MockContainerRuntimeFactoryForReconnection extends MockContainerRun
     }): MockContainerRuntimeForReconnection;
 }
 
+// @public
+export class MockContainerRuntimeForRebasing extends MockContainerRuntime {
+    constructor(dataStoreRuntime: MockFluidDataStoreRuntime, factory: MockContainerRuntimeFactoryForRebasing, overrides?: {
+        minimumSequenceNumber?: number;
+    });
+    // (undocumented)
+    process(message: ISequencedDocumentMessage): void;
+    // (undocumented)
+    rebase(): void;
+    // (undocumented)
+    submit(messageContent: any, localOpMetadata: unknown): number;
+}
+
 // @public
 export class MockContainerRuntimeForReconnection extends MockContainerRuntime {
     constructor(dataStoreRuntime: MockFluidDataStoreRuntime, factory: MockContainerRuntimeFactoryForReconnection, overrides?: {

experimental/dds/attributable-map/src/test/mocha/rebasing.spec.ts

@@ -0,0 +1,76 @@
+/*!
+ * Copyright (c) Microsoft Corporation and contributors. All rights reserved.
+ * Licensed under the MIT License.
+ */
+
+import { strict as assert } from "assert";
+import {
+	MockFluidDataStoreRuntime,
+	MockContainerRuntimeFactoryForRebasing,
+	MockContainerRuntimeForRebasing,
+	MockStorage,
+} from "@fluidframework/test-runtime-utils";
+import { MapFactory, AttributableMap } from "../../map";
+
+describe("Rebasing", () => {
+	let containerRuntimeFactory: MockContainerRuntimeFactoryForRebasing;
+	let containerRuntime1: MockContainerRuntimeForRebasing;
+	let containerRuntime2: MockContainerRuntimeForRebasing;
+	let map1: AttributableMap;
+	let map2: AttributableMap;
+
+	beforeEach(async () => {
+		containerRuntimeFactory = new MockContainerRuntimeFactoryForRebasing();
+
+		// Create the first SharedMap.
+		const dataStoreRuntime1 = new MockFluidDataStoreRuntime();
+		containerRuntime1 = containerRuntimeFactory.createContainerRuntime(dataStoreRuntime1);
+		const services1 = {
+			deltaConnection: containerRuntime1.createDeltaConnection(),
+			objectStorage: new MockStorage(),
+		};
+		map1 = new AttributableMap("shared-map-1", dataStoreRuntime1, MapFactory.Attributes);
+		map1.connect(services1);
+
+		// Create the second SharedMap.
+		const dataStoreRuntime2 = new MockFluidDataStoreRuntime();
+		containerRuntime2 = containerRuntimeFactory.createContainerRuntime(dataStoreRuntime2);
+		const services2 = {
+			deltaConnection: containerRuntime2.createDeltaConnection(),
+			objectStorage: new MockStorage(),
+		};
+		map2 = new AttributableMap("shared-map-2", dataStoreRuntime2, MapFactory.Attributes);
+		map2.connect(services2);
+	});
+
+	it("Rebasing ops maintains eventual consistency", async () => {
+		const keyCount = 10;
+		for (let i = 0; i < keyCount; i++) {
+			map1.set(`${i}`, map1.size);
+		}
+
+		containerRuntime1.rebase();
+		containerRuntimeFactory.processAllMessages();
+
+		for (let i = 0; i < keyCount; i++) {
+			assert.strictEqual(map1.get(`${i}`), i);
+			assert.strictEqual(map2.get(`${i}`), i);
+		}
+
+		const deleteThreshold = 5;
+		for (let i = 0; i < deleteThreshold - 1; i++) {
+			map2.delete(`${i}`);
+		}
+
+		map1.delete(`${deleteThreshold - 1}`);
+
+		containerRuntime2.rebase();
+		containerRuntimeFactory.processAllMessages();
+
+		for (let i = 0; i < 10; i++) {
+			const expected = i < deleteThreshold ? undefined : i;
+			assert.strictEqual(map1.get(`${i}`), expected);
+			assert.strictEqual(map2.get(`${i}`), expected);
+		}
+	});
+});

packages/runtime/test-runtime-utils/src/index.ts

@@ -23,5 +23,9 @@ export {
 	MockContainerRuntimeFactoryForReconnection,
 	MockContainerRuntimeForReconnection,
 } from "./mocksForReconnection";
+export {
+	MockContainerRuntimeFactoryForRebasing,
+	MockContainerRuntimeForRebasing,
+} from "./mocksForRebasing";
 export { MockStorage } from "./mockStorage";
 export { validateAssertionError } from "./validateAssertionError";

packages/runtime/test-runtime-utils/src/mocks.ts

@@ -145,7 +145,7 @@ export class MockContainerRuntime {
 			clientId: this.clientId,
 			clientSequenceNumber,
 			contents: messageContent,
-			referenceSequenceNumber: this.deltaManager.lastSequenceNumber,
+			referenceSequenceNumber: this.referenceSequenceNumber,
 			type: MessageType.Operation,
 		};
 		this.factory.pushMessage(msg);
@@ -193,6 +193,10 @@ export class MockContainerRuntime {
 		}
 		return [local, localOpMetadata];
 	}
+
+	protected get referenceSequenceNumber() {
+		return this.deltaManager.lastSequenceNumber;
+	}
 }
 
 /**

packages/runtime/test-runtime-utils/src/mocksForRebasing.ts

@@ -0,0 +1,155 @@
+/*!
+ * Copyright (c) Microsoft Corporation and contributors. All rights reserved.
+ * Licensed under the MIT License.
+ */
+
+import { v4 as uuid } from "uuid";
+import { ISequencedDocumentMessage, MessageType } from "@fluidframework/protocol-definitions";
+import {
+	MockContainerRuntime,
+	MockContainerRuntimeFactory,
+	MockFluidDataStoreRuntime,
+} from "./mocks";
+
+/**
+ * Specialized implementation of MockContainerRuntime for testing op rebasing, when the runtime will resend
+ * ops to the datastores and all ops within the same batch will have the same sequence number.
+ */
+export class MockContainerRuntimeForRebasing extends MockContainerRuntime {
+	private readonly currentBatch: ITrackableMessage[] = [];
+
+	constructor(
+		dataStoreRuntime: MockFluidDataStoreRuntime,
+		factory: MockContainerRuntimeFactoryForRebasing,
+		overrides?: { minimumSequenceNumber?: number },
+	) {
+		super(dataStoreRuntime, factory, overrides);
+	}
+
+	public process(message: ISequencedDocumentMessage) {
+		// Processing ops will happen in a separate JS turn, so by then, we'd increase
+		// the sequence number and flush the current batch.
+		this.clientSequenceNumber++;
+		this.currentBatch.splice(0);
+
+		super.process(message);
+	}
+
+	public submit(messageContent: any, localOpMetadata: unknown) {
+		const message = {
+			content: messageContent,
+			localOpMetadata,
+			opId: uuid(),
+			timesSubmitted: 0,
+		};
+		this.submitInternal(message);
+		this.currentBatch.push(message);
+
+		// Messages in the same batch will have the same clientSequenceNumber
+		return this.clientSequenceNumber;
+	}
+
+	private submitInternal(message: ITrackableMessage) {
+		message.timesSubmitted++;
+
+		const metadata = { opId: message.opId, timesSubmitted: message.timesSubmitted };
+		this.factory.pushMessage({
+			clientId: this.clientId,
+			clientSequenceNumber: this.clientSequenceNumber,
+			contents: message.content,
+			referenceSequenceNumber: this.referenceSequenceNumber,
+			type: MessageType.Operation,
+			metadata,
+		});
+		this.addPendingMessage(
+			message.content,
+			{ ...(message.localOpMetadata as object), ...metadata },
+			this.clientSequenceNumber,
+		);
+	}
+
+	public rebase() {
+		this.currentBatch.forEach((message) => this.submitInternal(message));
+	}
+}
+
+/**
+ * To help debugging eventual consistency tests, all ops produced by this mock
+ * can be tracked using an unique id. The tracking information will also be included
+ * in the message metadata and local op metadata.
+ */
+interface ITrackableMessage {
+	/**
+	 * Message content
+	 */
+	content: any;
+	/**
+	 * local op metadata
+	 */
+	localOpMetadata: unknown;
+	/**
+	 * Unique identifier
+	 */
+	opId: string;
+	/**
+	 * How many times has this op been resubmitted
+	 */
+	timesSubmitted: number;
+}
+
+/**
+ * Specialized implementation of MockContainerRuntimeFactory for testing op rebasing.
+ */
+export class MockContainerRuntimeFactoryForRebasing extends MockContainerRuntimeFactory {
+	public createContainerRuntime(
+		dataStoreRuntime: MockFluidDataStoreRuntime,
+		overrides?: { minimumSequenceNumber?: number },
+	): MockContainerRuntimeForRebasing {
+		const containerRuntime = new MockContainerRuntimeForRebasing(
+			dataStoreRuntime,
+			this,
+			overrides,
+		);
+		this.runtimes.push(containerRuntime);
+		return containerRuntime;
+	}
+
+	private processMessage() {
+		if (this.messages.length === 0) {
+			throw new Error("Tried to process a message that did not exist");
+		}
+
+		// Explicitly JSON clone the value to match the behavior of going thru the wire.
+		const message = JSON.parse(
+			JSON.stringify(this.messages.shift()),
+		) as ISequencedDocumentMessage;
+
+		this.minSeq.set(message.clientId, message.referenceSequenceNumber);
+		// Messages from the same batch have the same sequence number
+		message.sequenceNumber = this.sequenceNumber;
+		message.minimumSequenceNumber = this.getMinSeq();
+		for (const runtime of this.runtimes) {
+			runtime.process(message);
+		}
+	}
+
+	public processOneMessage() {
+		// Increase the sequence number between batches
+		this.sequenceNumber++;
+		this.processMessage();
+	}
+
+	public processSomeMessages(count: number) {
+		// Increase the sequence number between batches
+		this.sequenceNumber++;
+		for (let i = 0; i < count; i++) {
+			this.processMessage();
+		}
+	}
+
+	public processAllMessages() {
+		// Increase the sequence number between batches
+		this.sequenceNumber++;
+		this.processSomeMessages(this.messages.length);
+	}
+}

packages/runtime/test-runtime-utils/src/mocksForReconnection.ts

@@ -14,7 +14,7 @@ import {
 } from "./mocks";
 
 /**
- * Specalized implementation of MockContainerRuntime for testing ops during reconnection.
+ * Specialized implementation of MockContainerRuntime for testing ops during reconnection.
  */
 export class MockContainerRuntimeForReconnection extends MockContainerRuntime {
 	/**
@@ -105,7 +105,7 @@ export class MockContainerRuntimeForReconnection extends MockContainerRuntime {
 }
 
 /**
- * Specalized implementation of MockContainerRuntimeFactory for testing ops during reconnection.
+ * Specialized implementation of MockContainerRuntimeFactory for testing ops during reconnection.
  */
 export class MockContainerRuntimeFactoryForReconnection extends MockContainerRuntimeFactory {
 	public createContainerRuntime(