Add server timestamps to version mark resolution APIs (microsoft#28037)

Author: markfields

.changeset/lazy-teams-tan.md

@@ -0,0 +1,11 @@
+---
+"@fluidframework/container-runtime": minor
+"__section": feature
+---
+
+Add server timestamps to beta version mark resolution APIs
+
+The `@beta` `IVersionMarkResolver` APIs now expose the server timestamp of a resolved mark while remaining compatible with callers using the previous result and listener shapes:
+
+- The resolved variants returned by `sealAndCaptureVersionMark()` and `resolve()` populate an optional `timestamp` property.
+- The `onBatchSequenced()` listener can receive `timestamp` as its optional third argument when an incoming batch resolves a pending mark.

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

@@ -340,7 +340,7 @@ export interface IUploadSummaryResult extends Omit<IGenerateSummaryTreeResult, "
 
 // @beta @legacy
 export interface IVersionMarkResolver {
-    onBatchSequenced(listener: (batchId: string, sequenceNumber: number) => void): () => void;
+    onBatchSequenced(listener: (batchId: string, sequenceNumber: number, timestamp?: number) => void): () => void;
     resolve(batchId: string, sequenceNumberLowerBound: number): Promise<ResolveResult>;
     sealAndCaptureVersionMark(): VersionMarkCapture;
 }
@@ -384,6 +384,7 @@ export type ReadFluidDataStoreAttributes = IFluidDataStoreAttributes0 | IFluidDa
 export type ResolveResult = {
     readonly kind: "resolved";
     readonly sequenceNumber: number;
+    readonly timestamp?: number;
 } | {
     readonly kind: "pending";
 } | {
@@ -448,6 +449,7 @@ export type VersionMarkCapture = {
 } | {
     readonly kind: "resolved";
     readonly sequenceNumber: number;
+    readonly timestamp?: number;
 };
 
 // (No @packageDocumentation comment for this package)

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

@@ -340,7 +340,7 @@ export interface IUploadSummaryResult extends Omit<IGenerateSummaryTreeResult, "
 
 // @beta @legacy
 export interface IVersionMarkResolver {
-    onBatchSequenced(listener: (batchId: string, sequenceNumber: number) => void): () => void;
+    onBatchSequenced(listener: (batchId: string, sequenceNumber: number, timestamp?: number) => void): () => void;
     resolve(batchId: string, sequenceNumberLowerBound: number): Promise<ResolveResult>;
     sealAndCaptureVersionMark(): VersionMarkCapture;
 }
@@ -379,6 +379,7 @@ export type ReadFluidDataStoreAttributes = IFluidDataStoreAttributes0 | IFluidDa
 export type ResolveResult = {
     readonly kind: "resolved";
     readonly sequenceNumber: number;
+    readonly timestamp?: number;
 } | {
     readonly kind: "pending";
 } | {
@@ -443,6 +444,7 @@ export type VersionMarkCapture = {
 } | {
     readonly kind: "resolved";
     readonly sequenceNumber: number;
+    readonly timestamp?: number;
 };
 
 // (No @packageDocumentation comment for this package)

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

@@ -1943,6 +1943,7 @@ export class ContainerRuntime
 		const fetchOps = (context as IContainerContextInternal).fetchOps;
 		this.versionMarkResolverInternal = new VersionMarkResolver({
 			getCurrentSequenceNumber: () => this.deltaManager.lastSequenceNumber,
+			getCurrentTimestamp: () => this.getCurrentReferenceTimestampMs(),
 			getCurrentMinimumSequenceNumber: () => this.deltaManager.minimumSequenceNumber,
 			getCurrentPendingBatchId: () => this.pendingStateManager.getMostRecentPendingBatchId(),
 			logger: createChildLogger({
@@ -3422,6 +3423,7 @@ export class ContainerRuntime
 					this.versionMarkResolverInternal.processInboundBatch(
 						versionMarkUpdate.sequenced.batchId,
 						versionMarkUpdate.sequenced.sequenceNumber,
+						versionMarkUpdate.sequenced.timestamp,
 					);
 				}
 				this.versionMarkInboundBatchId = versionMarkUpdate.carriedBatchId;

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

@@ -104,7 +104,9 @@ import { pkgVersion } from "../packageVersion.js";
 import type { IPendingMessage, PendingStateManager } from "../pendingStateManager.js";
 import {
 	type ISummaryCancellationToken,
+	type IContainerRuntimeMetadata,
 	neverCancelledSummaryToken,
+	metadataBlobName,
 	recentBatchInfoBlobName,
 	type IRefreshSummaryAckOptions,
 } from "../summary/index.js";
@@ -2886,16 +2888,117 @@ describe("Runtime", () => {
 					"Expected DuplicateBatch telemetry to still be logged",
 				);
 			});
+
+			it("Version mark tracking keeps the earlier resolution for a tolerated duplicate batch with no explicit batchId", async () => {
+				const logger = new MockLogger();
+				const { runtime: containerRuntime } = await ContainerRuntime.loadRuntime2({
+					context: getMockContext({ logger }) as IContainerContext,
+					registry: new FluidDataStoreRegistry([]),
+					existing: false,
+					runtimeOptions: {
+						enableRuntimeIdCompressor: "on",
+					},
+					provideEntryPoint: mockProvideEntryPoint,
+				});
+
+				// Subscribing activates version-mark tracking on the inbound batch path.
+				const promotions: { sequenceNumber: number; timestamp: number | undefined }[] = [];
+				containerRuntime.versionMarkResolver.onBatchSequenced(
+					(_batchId, sequenceNumber, timestamp) => {
+						promotions.push({ sequenceNumber, timestamp });
+					},
+				);
+
+				const makeMessage = (
+					sequenceNumber: number,
+					timestamp: number,
+				): ISequencedDocumentMessage =>
+					({
+						clientId: "clientId",
+						clientSequenceNumber: 1,
+						sequenceNumber,
+						timestamp,
+						type: MessageType.Operation,
+						contents: { type: ContainerMessageType.Rejoin, contents: undefined },
+					}) satisfies Partial<ISequencedDocumentMessage> as ISequencedDocumentMessage;
+
+				containerRuntime.process(makeMessage(123, 1000), false);
+
+				// The service re-broadcasts the same batch under a different sequence number/timestamp.
+				// DuplicateBatchDetector tolerates this (no explicit batchId), so it must not throw, and
+				// VersionMarkResolver must keep the earlier (first-landed) resolution rather than remapping
+				// to the duplicate's sequence number.
+				assert.doesNotThrow(
+					() => containerRuntime.process(makeMessage(234, 2000), false),
+					"Should not throw when the duplicate batch has no explicit batchId",
+				);
+				logger.assertMatchAny(
+					[{ eventName: "ContainerRuntime:DuplicateBatch" }],
+					"Expected DuplicateBatch telemetry to be logged",
+				);
+				assert.deepStrictEqual(
+					promotions,
+					[{ sequenceNumber: 123, timestamp: 1000 }],
+					"Only the first (original) resolution should have been promoted to listeners",
+				);
+			});
 		});
 
 		describe("Version mark inbound update", () => {
+			it("captures the summary timestamp after load before any new ops arrive", async () => {
+				const sequenceNumber = 42;
+				const timestamp = 123456;
+				const metadata: IContainerRuntimeMetadata = {
+					summaryFormatVersion: 1,
+					message: {
+						clientId: mockClientId,
+						clientSequenceNumber: 1,
+						minimumSequenceNumber: 0,
+						referenceSequenceNumber: 41,
+						sequenceNumber,
+						timestamp,
+						type: MessageType.Operation,
+					},
+				};
+				const context = getMockContext({
+					baseSnapshot: {
+						trees: { ".channels": { trees: {}, blobs: {} } },
+						blobs: { [metadataBlobName]: "metadata-id" },
+					},
+					mockStorage: {
+						...defaultMockStorage,
+						readBlob: async (id) => {
+							assert.equal(id, "metadata-id");
+							return stringToBuffer(JSON.stringify(metadata), "utf8");
+						},
+					},
+				});
+				const deltaManager = context.deltaManager as MockDeltaManager;
+				deltaManager.initialSequenceNumber = sequenceNumber;
+				deltaManager.lastSequenceNumber = sequenceNumber;
+				deltaManager.lastMessage = undefined;
+
+				const { runtime: containerRuntime } = await ContainerRuntime.loadRuntime2({
+					context: context as IContainerContext,
+					registry: new FluidDataStoreRegistry([]),
+					existing: true,
+					provideEntryPoint: mockProvideEntryPoint,
+				});
+
+				assert.deepEqual(containerRuntime.versionMarkResolver.sealAndCaptureVersionMark(), {
+					kind: "resolved",
+					sequenceNumber,
+					timestamp,
+				});
+			});
+
 			it("resolves through context fetchOps and the historical unpack pipeline", async () => {
 				let capturedSignal: AbortSignal | undefined;
 				let readCount = 0;
 				const context = {
 					...getMockContext(),
 					fetchOps: async (from, to, abortSignal) => {
-						assert.equal(from, 11, "the history read starts at the inclusive lower bound");
+						assert.equal(from, 10, "the history read starts at the inclusive lower bound");
 						assert.equal(to, undefined, "the history read has no fixed upper bound");
 						capturedSignal = abortSignal;
 						return {
@@ -2919,6 +3022,7 @@ describe("Runtime", () => {
 											{
 												type: MessageType.Operation,
 												sequenceNumber: 13,
+												timestamp: 13000,
 												clientId: "targetClient",
 												clientSequenceNumber: 7,
 												contents: {
@@ -2943,8 +3047,8 @@ describe("Runtime", () => {
 				});
 
 				assert.deepEqual(
-					await containerRuntime.versionMarkResolver.resolve("targetBatch", 11),
-					{ kind: "resolved", sequenceNumber: 13 },
+					await containerRuntime.versionMarkResolver.resolve("targetBatch", 10),
+					{ kind: "resolved", sequenceNumber: 13, timestamp: 13000 },
 				);
 				assert.equal(readCount, 1, "the stream stops reading after the target batch");
 				assert.equal(capturedSignal?.aborted, true, "the fetch is aborted after the match");

packages/runtime/container-runtime/src/test/versionMarks/inboundBatch.spec.ts

@@ -18,17 +18,23 @@ import {
 	// eslint-disable-next-line import-x/no-internal-modules -- test targets the inbound-batch helper directly
 } from "../../versionMarks/inboundBatch.js";
 
-/** The helper only reads `sequenceNumber` off messages. */
+/** The helper reads the sequence number and server timestamp off messages. */
 function msg(sequenceNumber: number): InboundSequencedContainerRuntimeMessage {
-	return { sequenceNumber } as unknown as InboundSequencedContainerRuntimeMessage;
+	return {
+		sequenceNumber,
+		timestamp: sequenceNumber * 1000,
+	} as unknown as InboundSequencedContainerRuntimeMessage;
 }
 
 function batchStart(batchId: string | undefined, keyMessageSeq: number): BatchStartInfo {
 	return {
 		batchId,
 		clientId: "client",
 		batchStartCsn: 3,
-		keyMessage: { sequenceNumber: keyMessageSeq } as unknown as ISequencedDocumentMessage,
+		keyMessage: {
+			sequenceNumber: keyMessageSeq,
+			timestamp: keyMessageSeq * 1000,
+		} as unknown as ISequencedDocumentMessage,
 	};
 }
 
@@ -58,7 +64,7 @@ describe("inboundVersionMarkUpdate", () => {
 				"stale_[9]",
 			);
 			assert.deepEqual(result, {
-				sequenced: { batchId: "b_[3]", sequenceNumber: 12 },
+				sequenced: { batchId: "b_[3]", sequenceNumber: 12, timestamp: 12000 },
 				carriedBatchId: undefined,
 			});
 		});
@@ -69,7 +75,7 @@ describe("inboundVersionMarkUpdate", () => {
 				undefined,
 			);
 			assert.deepEqual(result, {
-				sequenced: { batchId: "b_[3]", sequenceNumber: 42 },
+				sequenced: { batchId: "b_[3]", sequenceNumber: 42, timestamp: 42000 },
 				carriedBatchId: undefined,
 			});
 		});
@@ -80,7 +86,11 @@ describe("inboundVersionMarkUpdate", () => {
 				undefined,
 			);
 			assert.deepEqual(result, {
-				sequenced: { batchId: generateBatchId("client", 3), sequenceNumber: 7 },
+				sequenced: {
+					batchId: generateBatchId("client", 3),
+					sequenceNumber: 7,
+					timestamp: 7000,
+				},
 				carriedBatchId: undefined,
 			});
 		});
@@ -103,7 +113,7 @@ describe("inboundVersionMarkUpdate", () => {
 		it("records the carried id at the batch's last op and clears the carry", () => {
 			const result = inboundVersionMarkUpdate(nextBatchMessage(22, true), "b_[3]");
 			assert.deepEqual(result, {
-				sequenced: { batchId: "b_[3]", sequenceNumber: 22 },
+				sequenced: { batchId: "b_[3]", sequenceNumber: 22, timestamp: 22000 },
 				carriedBatchId: undefined,
 			});
 		});
@@ -123,7 +133,7 @@ describe("inboundVersionMarkUpdate", () => {
 
 			const end = inboundVersionMarkUpdate(nextBatchMessage(32, true), mid.carriedBatchId);
 			assert.deepEqual(end, {
-				sequenced: { batchId: "b_[3]", sequenceNumber: 32 },
+				sequenced: { batchId: "b_[3]", sequenceNumber: 32, timestamp: 32000 },
 				carriedBatchId: undefined,
 			});
 		});