Implementation for resolving data corruption issue for distributed data ordering service

packages/loader/container-definitions/src/loader.ts

@@ -261,6 +261,11 @@ export type ILoaderOptions = {
      * Set min op frequency with which noops would be sent in case of active connection which is not sending any op.
      */
     noopCountFrequency?: number;
+
+    /**
+     * Max time(in ms) container will wait for a leave message of a disconnected client.
+    */
+    maxClientLeaveWaitTime?: number,
 };
 
 /**

packages/loader/container-loader/src/connectionStateHandler.ts

@@ -3,12 +3,12 @@
  * Licensed under the MIT License.
  */
 
-import { EventEmitter } from "events";
 import { IEvent, ITelemetryLogger } from "@fluidframework/common-definitions";
 import { IConnectionDetails } from "@fluidframework/container-definitions";
-import { ProtocolOpHandler, Quorum } from "@fluidframework/protocol-base";
-import { ConnectionMode, ISequencedClient } from "@fluidframework/protocol-definitions";
-import { EventEmitterWithErrorHandling } from "@fluidframework/telemetry-utils";
+import { ProtocolOpHandler } from "@fluidframework/protocol-base";
+import { ConnectionMode, IClient, ISequencedClient } from "@fluidframework/protocol-definitions";
+import { EventEmitterWithErrorHandling, PerformanceEvent } from "@fluidframework/telemetry-utils";
+import { assert, Timer } from "@fluidframework/common-utils";
 import { connectEventName, ConnectionState } from "./container";
 
 export interface IConnectionStateHandler {
@@ -17,6 +17,9 @@ export interface IConnectionStateHandler {
         (value: ConnectionState, oldState: ConnectionState, reason?: string | undefined) => void,
     propagateConnectionState: () => void,
     isContainerLoaded: () => boolean,
+    client: () => IClient,
+    shouldClientJoinWrite: () => boolean,
+    maxClientLeaveWaitTime: number | undefined,
 }
 
 export interface ILocalSequencedClient extends ISequencedClient {
@@ -37,6 +40,12 @@ export class ConnectionStateHandler extends EventEmitterWithErrorHandling<IConne
     private _connectionState = ConnectionState.Disconnected;
     private _pendingClientId: string | undefined;
     private _clientId: string | undefined;
+    private prevClientLeftTimer: Timer;
+    // This is client id of client for which we have received the addMember event but we are waiting on some previous
+    // client to leave before moving to Connected state.
+    private waitingClientId: string | undefined;
+    private waitEvent: PerformanceEvent | undefined;
+    private _clientSentOps: boolean = false;
 
     public get connectionState(): ConnectionState {
         return this._connectionState;
@@ -59,21 +68,62 @@ export class ConnectionStateHandler extends EventEmitterWithErrorHandling<IConne
         private readonly logger: ITelemetryLogger,
     ) {
         super();
+        this.prevClientLeftTimer = new Timer(
+            // Default is 90 sec for which we are going to wait for its own "leave" message.
+            this.handler.maxClientLeaveWaitTime ?? 90000,
+            () => {
+                this.clientLeaveWaitEnded(false);
+            },
+        );
     }
 
-    public receivedAddMemberEvent(clientId: string, quorum: Quorum) {
+    // This is true when this client submitted any ops.
+    public clientSentOps() {
+        this._clientSentOps = true;
+    }
+
+    public receivedAddMemberEvent(clientId: string) {
         // This is the only one that requires the pending client ID
         if (clientId === this.pendingClientId) {
+            // Wait for previous client to leave the quorum before firing "connected" event.
+            if (this.prevClientLeftTimer.hasTimer) {
+                this.waitEvent = PerformanceEvent.start(this.logger, {
+                    eventName: "WaitBeforeClientLeave",
+                    waitOnClientId: this._clientId,
+                    hadOutstandingOps: this.handler.shouldClientJoinWrite(),
+                });
+                this.waitingClientId = clientId;
+            } else {
+                this.setConnectionState(ConnectionState.Connected);
+            }
+        }
+    }
+
+    private clientLeaveWaitEnded(leaveReceived: boolean) {
+        // Move to connected state only if there was a client waiting for client leave. It may happen that
+        // during wait, the waiting client again got Disconnected/Connecting and then we don't want to
+        // move to connected state here for a non waiting client.
+        if (this.waitingClientId !== undefined) {
+            assert(this.waitingClientId === this.pendingClientId, "Pending Client Id should match the waiting client");
+            assert(this.waitEvent !== undefined, "Wait event should be set");
+            this.waitEvent.end({ leaveReceived });
             this.setConnectionState(ConnectionState.Connected);
         }
     }
 
+    public receivedRemoveMemberEvent(clientId: string) {
+        // If the client which has left was us, then finish the timer.
+        if (this.clientId === clientId) {
+            this.prevClientLeftTimer?.clear();
+            this.clientLeaveWaitEnded(true);
+        }
+    }
+
     public receivedDisconnectEvent(reason: string) {
         this.setConnectionState(ConnectionState.Disconnected, reason);
     }
 
     public receivedConnectEvent(
-        emitter: EventEmitter,
         connectionMode: ConnectionMode,
         details: IConnectionDetails,
         opsBehind?: number,
@@ -118,8 +168,10 @@ export class ConnectionStateHandler extends EventEmitterWithErrorHandling<IConne
 
         const oldState = this._connectionState;
         this._connectionState = value;
-
+        // Set it to undefined as now there is no client waiting to get connected.
+        this.waitingClientId = undefined;
         if (value === ConnectionState.Connected) {
+            assert(oldState === ConnectionState.Connecting, "Should only transition from Connecting state");
             // Mark our old client should have left in the quorum if it's still there
             if (this._clientId !== undefined) {
                 const client: ILocalSequencedClient | undefined =
@@ -129,9 +181,27 @@ export class ConnectionStateHandler extends EventEmitterWithErrorHandling<IConne
                 }
             }
             this._clientId = this.pendingClientId;
+            // Set _clientSentOps to false as this is a fresh connection.
+            this._clientSentOps = false;
         } else if (value === ConnectionState.Disconnected) {
             // Important as we process our own joinSession message through delta request
             this._pendingClientId = undefined;
+            // Only wait for "leave" message if we have some outstanding ops and the client was write client as
+            // server would not accept ops from read client. Also check if the timer is not already running as we
+            // could receive "Disconnected" event multiple times without getting connected and in that case we
+            // don't want to reset the timer as we still want to wait on original client which started this timer.
+            // We also check the dirty state of this connection as we only want to wait for the client leave of the
+            // client which created the ops. This helps with situation where a client disconnects immediately after
+            // getting connected without sending any ops(from previous client). In this case, we would join as write
+            // because there would be a diff between client seq number and clientSeqNumberObserved but then we don't
+            // want to wait for newly disconnected client to leave as it has not sent any ops yet.
+            if (this.handler.shouldClientJoinWrite()
+                && this.handler.client().mode === "write"
+                && this.prevClientLeftTimer.hasTimer === false
+                && this._clientSentOps
+            ) {
+                this.prevClientLeftTimer.restart();
+            }
         }
 
         if (this.handler.isContainerLoaded()) {

packages/loader/container-loader/src/container.ts

@@ -552,6 +552,7 @@ export class Container extends EventEmitterWithErrorHandling<IContainerEvents> i
     public get options(): ILoaderOptions { return this.loader.services.options; }
     private get scope() { return this.loader.services.scope;}
     private get codeLoader() { return this.loader.services.codeLoader;}
+
     constructor(
         private readonly loader: Loader,
         config: IContainerConfig,
@@ -599,6 +600,9 @@ export class Container extends EventEmitterWithErrorHandling<IContainerEvents> i
                     this.logConnectionStateChangeTelemetry(value, oldState, reason),
                 propagateConnectionState: () => this.propagateConnectionState(),
                 isContainerLoaded: () => this.loaded,
+                client: () => this.client,
+                shouldClientJoinWrite: () => this._deltaManager.shouldJoinWrite(),
+                maxClientLeaveWaitTime: this.loader.services.options.maxClientLeaveWaitTime,
             },
             this.logger,
         );
@@ -1396,10 +1400,14 @@ export class Container extends EventEmitterWithErrorHandling<IContainerEvents> i
 
         // Track membership changes and update connection state accordingly
         protocol.quorum.on("addMember", (clientId, details) => {
-            this.connectionStateHandler.receivedAddMemberEvent(clientId, this.protocolHandler.quorum);
+            this.connectionStateHandler.receivedAddMemberEvent(clientId);
+        });
+
+        protocol.quorum.on("removeMember", (clientId) => {
+            this.connectionStateHandler.receivedRemoveMemberEvent(clientId);
         });
 
-        protocol.quorum.on("addProposal",(proposal: IPendingProposal) => {
+        protocol.quorum.on("addProposal", (proposal: IPendingProposal) => {
             if (proposal.key === "code" || proposal.key === "code2") {
                 this.emit("codeDetailsProposed", proposal.value, proposal);
             }
@@ -1515,7 +1523,6 @@ export class Container extends EventEmitterWithErrorHandling<IContainerEvents> i
 
         deltaManager.on(connectEventName, (details: IConnectionDetails, opsBehind?: number) => {
             this.connectionStateHandler.receivedConnectEvent(
-                this,
                 this._deltaManager.connectionMode,
                 details,
                 opsBehind,
@@ -1529,6 +1536,10 @@ export class Container extends EventEmitterWithErrorHandling<IContainerEvents> i
             }
         });
 
+        deltaManager.once("submitOp", (message: IDocumentMessage) => {
+            this.connectionStateHandler.clientSentOps();
+        });
+
         deltaManager.on("disconnect", (reason: string) => {
             this.manualReconnectInProgress = false;
             this.connectionStateHandler.receivedDisconnectEvent(reason);

packages/loader/container-loader/src/deltaManager.ts

@@ -349,6 +349,10 @@ export class DeltaManager
         return this._reconnectMode;
     }
 
+    public shouldJoinWrite(): boolean {
+        return this.clientSequenceNumber !== this.clientSequenceNumberObserved;
+    }
+
     public async connectToStorage(): Promise<IDocumentStorageService> {
         if (this.storageService !== undefined) {
             return this.storageService;
@@ -580,7 +584,7 @@ export class DeltaManager
         // firing of "connected" event from Container and switch of current clientId (as tracked
         // by all DDSes). This will make it impossible to figure out if ops actually made it through,
         // so DDSes will immediately resubmit all pending ops, and some of them will be duplicates, corrupting document
-        if (this.clientSequenceNumberObserved !== this.clientSequenceNumber) {
+        if (this.shouldJoinWrite()) {
             requestedMode = "write";
         }