import { default as AbortController } from "abort-controller" ; import { v4 as uuid } from "uuid" ; import { ITelemetryLogger , IEventProvider , ITelemetryProperties , ITelemetryErrorEvent , } from "@fluidframework/common-definitions" ; import { IDeltaHandlerStrategy , IDeltaManager , IDeltaManagerEvents , IDeltaQueue , ICriticalContainerError , IThrottlingWarning , IConnectionDetails , } from "@fluidframework/container-definitions" ; import { assert , TypedEventEmitter } from "@fluidframework/common-utils" ; import { normalizeError , logIfFalse , safeRaiseEvent , } from "@fluidframework/telemetry-utils" ; import { IDocumentDeltaStorageService , IDocumentService , DriverErrorType , } from "@fluidframework/driver-definitions" ; import { isSystemMessage } from "@fluidframework/protocol-base" ; import { IDocumentMessage , ISequencedDocumentMessage , ISignalMessage , MessageType , ITrace , ConnectionMode , } from "@fluidframework/protocol-definitions" ; import { NonRetryableError , } from "@fluidframework/driver-utils" ; import { ThrottlingWarning , CreateProcessingError , DataCorruptionError , } from "@fluidframework/container-utils" ; import { DeltaQueue } from "./deltaQueue" ; import { IConnectionManagerFactoryArgs , IConnectionManager , } from "./contracts" ; export interface IConnectionArgs { mode ? : ConnectionMode ; fetchOpsFromStorage ? : boolean ; reason : string ; } /**
 * Includes events emitted by the concrete implementation DeltaManager
 * but not exposed on the public interface IDeltaManager
 */ export interface IDeltaManagerInternalEvents extends IDeltaManagerEvents { ( event , listener ) ; ( event , listener ) ; } /**
 * Manages the flow of both inbound and outbound messages. This class ensures that shared objects receive delta
 * messages in order regardless of possible network conditions or timings causing out of order delivery.
 */ export class DeltaManager < TConnectionManager extends IConnectionManager > extends TypedEventEmitter < IDeltaManagerInternalEvents > implements IDeltaManager < ISequencedDocumentMessage , IDocumentMessage > , IEventProvider < IDeltaManagerInternalEvents > { public readonly connectionManager : TConnectionManager ; public get active ( ) : boolean { return this . _active ( ) ; } public get disposed ( ) { return this . closed ; } public get IDeltaSender ( ) { return this ; } private pending : ISequencedDocumentMessage [ ] = [ ] ; private fetchReason : string | undefined ; private minSequenceNumber : number = 0 ; private lastQueuedSequenceNumber : number = 0 ; private lastObservedSeqNumber : number = 0 ; private lastProcessedSequenceNumber : number = 0 ; private lastProcessedMessage : ISequencedDocumentMessage | undefined ; private baseTerm : number = 0 ; private prevEnqueueMessagesReason : string | undefined ; private previouslyProcessedMessage : ISequencedDocumentMessage | undefined ; private initSequenceNumber : number = 0 ; private readonly _inbound : DeltaQueue < ISequencedDocumentMessage > ; private readonly _inboundSignal : DeltaQueue < ISignalMessage > ; private closed = false ; private handler : IDeltaHandlerStrategy | undefined ; private deltaStorage : IDocumentDeltaStorageService | undefined ; private readonly throttlingIdSet = new Set < string > ( ) ; private timeTillThrottling : number = 0 ; private readonly closeAbortController = new AbortController ( ) ; private readonly deltaStorageDelayId = uuid ( ) ; private readonly deltaStreamDelayId = uuid ( ) ; private messageBuffer : IDocumentMessage [ ] = [ ] ; private _checkpointSequenceNumber : number | undefined ; public get inbound ( ) : IDeltaQueue < ISequencedDocumentMessage > { return this . _inbound ; } public get inboundSignal ( ) : IDeltaQueue < ISignalMessage > { return this . _inboundSignal ; } public get initialSequenceNumber ( ) : number { return this . initSequenceNumber ; } public get lastSequenceNumber ( ) : number { return this . lastProcessedSequenceNumber ; } public get lastMessage ( ) { return this . lastProcessedMessage ; } public get lastKnownSeqNumber ( ) { return this . lastObservedSeqNumber ; } public get referenceTerm ( ) : number { return this . baseTerm ; } public get minimumSequenceNumber ( ) : number { return this . minSequenceNumber ; } /**
     * Tells if  current connection has checkpoint information.
     * I.e. we know how far behind the client was at the time of establishing connection
     */ public get hasCheckpointSequenceNumber ( ) { assert ( this . connectionManager . connected , 0x0df ) ; return this . _checkpointSequenceNumber !== undefined ; } public get maxMessageSize ( ) : number { return this . connectionManager . maxMessageSize ; } public get version ( ) { return this . connectionManager . version ; } public get serviceConfiguration ( ) { return this . connectionManager . serviceConfiguration ; } public get outbound ( ) { return this . connectionManager . outbound ; } public get readOnlyInfo ( ) { return this . connectionManager . readOnlyInfo ; } public get clientDetails ( ) { return this . connectionManager . clientDetails ; } deprecated public get readonly ( ) { return this . readOnlyInfo . readonly ; } public submit ( type , contents , batch = false , metadata ? ) { const traces = [ { action : "start" , service : "client" , timestamp : Date . now ( ) } ] ; const messagePartial = { contents : JSON . stringify ( contents ) , metadata , referenceSequenceNumber : this . lastProcessedSequenceNumber , traces , type } ; if ( ! batch ) { this . flush ( ) ; } const message = this . connectionManager . prepareMessageToSend ( messagePartial ) ; if ( message === undefined ) { return - 1 ; } this . messageBuffer . push ( message ) ; if ( ! batch ) { this . flush ( ) ; } this . emit ( "submitOp" , message ) ; return message . clientSequenceNumber ; } public submitSignal ( content ) { return this . connectionManager . submitSignal ( content ) ; } public flush ( ) { if ( this . messageBuffer . length === 0 ) { return ; } this . emit ( "prepareSend" , this . messageBuffer ) ; this . connectionManager . sendMessages ( this . messageBuffer ) ; this . messageBuffer = [ ] ; } public get connectionProps ( ) : ITelemetryProperties { return { sequenceNumber : this . lastSequenceNumber , ... this . connectionManager . connectionProps } ; } param event public logConnectionIssue ( event ) { assert ( this . connectionManager . connected , 0x238 ) ; const pendingSorted = this . pending . sort ( ( a , b ) => a . sequenceNumber - b . sequenceNumber ) ; this . logger . sendErrorEvent ( { ... event , fetchReason : this . fetchReason , lastQueuedSequenceNumber : this . lastQueuedSequenceNumber , lastProcessedSequenceNumber : this . lastProcessedSequenceNumber , lastObserved : this . lastObservedSeqNumber , ... this . connectionManager . connectionVerboseProps , pendingOps : this . pending . length , pendingFirst : pendingSorted [ 0 ] ?. sequenceNumber , haveHandler : this . handler !== undefined , inboundLength : this . inbound . length , inboundPaused : this . inbound . paused } ) ; } constructor ( private readonly serviceProvider , private readonly logger , private readonly _active , createConnectionManager ) { super ( ) ; const props = { incomingOpHandler : ( messages , reason ) => this . enqueueMessages ( messages , reason ) , signalHandler : ( message ) => this . _inboundSignal . push ( message ) , reconnectionDelayHandler : ( delayMs , error ) => this . emitDelayInfo ( this . deltaStreamDelayId , delayMs , error ) , closeHandler : ( error ) => this . close ( error ) , disconnectHandler : ( reason ) => this . disconnectHandler ( reason ) , connectHandler : ( connection ) => this . connectHandler ( connection ) , pongHandler : ( latency ) => this . emit ( "pong" , latency ) , readonlyChangeHandler : ( readonly ? ) => safeRaiseEvent ( this , this . logger , "readonly" , readonly ) } ; this . connectionManager = createConnectionManager ( props ) ; this . _inbound = new DeltaQueue < ISequencedDocumentMessage > ( ( op ) => { this . processInboundMessage ( op ) ; } ) ; this . _inbound . on ( "error" , ( error ) => { this . close ( CreateProcessingError ( error , "deltaManagerInboundErrorHandler" , this . lastMessage ) ) ; } ) ; this . _inboundSignal = new DeltaQueue < ISignalMessage > ( ( message ) => { if ( this . handler === undefined ) { throw new Error ( "Attempted to process an inbound signal without a handler attached" ) ; } this . handler . processSignal ( { clientId : message . clientId , content : JSON . parse ( message . content as string ) } ) ; } ) ; this . _inboundSignal . on ( "error" , ( error ) => { this . close ( normalizeError ( error ) ) ; } ) ; } private connectHandler ( connection ) { this . refreshDelayInfo ( this . deltaStreamDelayId ) ; const props = this . connectionManager . connectionVerboseProps ; props . connectionLastQueuedSequenceNumber = this . lastQueuedSequenceNumber ; props . connectionLastObservedSeqNumber = this . lastObservedSeqNumber ; const checkpointSequenceNumber = connection . checkpointSequenceNumber ; this . _checkpointSequenceNumber = checkpointSequenceNumber ; if ( checkpointSequenceNumber !== undefined ) { this . updateLatestKnownOpSeqNumber ( checkpointSequenceNumber ) ; } assert ( this . messageBuffer . length === 0 , 0x0e9 ) ; this . emit ( "connect" , connection , checkpointSequenceNumber !== undefined ? this . lastObservedSeqNumber - this . lastSequenceNumber : undefined ) ; if ( checkpointSequenceNumber !== undefined ) { if ( checkpointSequenceNumber > this . lastQueuedSequenceNumber ) { this . fetchMissingDeltas ( "AfterConnection" ) ; } } else if ( connection . mode === "read" ) { this . fetchMissingDeltas ( "AfterReadConnection" ) ; } } public dispose ( ) { throw new Error ( "Not implemented." ) ; } /**
     * Sets the sequence number from which inbound messages should be returned
     */ public async attachOpHandler ( minSequenceNumber , sequenceNumber , term , handler , prefetchType = "none" ) { this . initSequenceNumber = sequenceNumber ; this . lastProcessedSequenceNumber = sequenceNumber ; this . baseTerm = term ; this . minSequenceNumber = minSequenceNumber ; this . lastQueuedSequenceNumber = sequenceNumber ; this . lastObservedSeqNumber = sequenceNumber ; assert ( this . handler === undefined , 0x0e2 ) ; this . handler = handler ; assert ( ! ! ( this . handler as any ) , 0x0e3 ) ; assert ( this . fetchReason === undefined , 0x268 ) ; if ( this . closed ) { return ; } this . _inbound . resume ( ) ; this . _inboundSignal . resume ( ) ; if ( prefetchType !== "none" ) { const cacheOnly = prefetchType === "cached" ; await this . fetchMissingDeltasCore ( "DocumentOpen" , cacheOnly , this . lastQueuedSequenceNumber ) ; if ( cacheOnly ) { this . fetchMissingDeltas ( "DocumentOpen" ) ; } } assert ( this . fetchReason !== undefined || this . pending . length === 0 , 0x269 ) ; } public connect ( args ) { const fetchOpsFromStorage = args . fetchOpsFromStorage ?? true ; logIfFalse ( this . handler !== undefined || ! fetchOpsFromStorage , this . logger , "CantFetchWithoutBaseline" ) ; if ( fetchOpsFromStorage ) { this . fetchMissingDeltas ( args . reason ) ; } this . connectionManager . connect ( args . mode ) ; } private async getDeltas ( from , to , callback , cacheOnly ) { const docService = this . serviceProvider ( ) ; if ( docService === undefined ) { throw new Error ( "Delta manager is not attached" ) ; } if ( this . deltaStorage === undefined ) { this . deltaStorage = await docService . connectToDeltaStorage ( ) ; } let cancelFetch ; if ( to !== undefined ) { const lastExpectedOp = to - 1 ; if ( this . lastQueuedSequenceNumber >= lastExpectedOp ) { this . logger . sendPerformanceEvent ( { reason : this . fetchReason , eventName : "ExtraStorageCall" , early : true , from , to , ... this . connectionManager . connectionVerboseProps } ) ; return ; } cancelFetch = ( op ) => op . sequenceNumber >= lastExpectedOp ; } else { cancelFetch = ( op ) => op . sequenceNumber >= this . lastObservedSeqNumber ; } const controller = new AbortController ( ) ; let opsFromFetch = false ; const opListener = ( op ) => { assert ( op . sequenceNumber === this . lastQueuedSequenceNumber , 0x23a ) ; if ( ! opsFromFetch && cancelFetch ( op ) ) { controller . abort ( ) ; this . _inbound . off ( "push" , opListener ) ; } } ; try { this . _inbound . on ( "push" , opListener ) ; assert ( this . closeAbortController . signal . onabort === null , 0x1e8 ) ; this . closeAbortController . signal . onabort = ( ) => controller . abort ( ) ; const stream = this . deltaStorage . fetchMessages ( from , to , controller . signal , cacheOnly , this . fetchReason ) ; while ( true ) { const result = await stream . read ( ) ; if ( result . done ) { break ; } try { opsFromFetch = true ; callback ( result . value ) ; } finally { opsFromFetch = false ; } } } finally { this . closeAbortController . signal . onabort = null ; this . _inbound . off ( "push" , opListener ) ; assert ( ! opsFromFetch , 0x289 ) ; } } /**
     * Closes the connection and clears inbound & outbound queues.
     */ public close ( error ? ) { if ( this . closed ) { return ; } this . closed = true ; this . connectionManager . dispose ( error ) ; this . closeAbortController . abort ( ) ; this . _inbound . clear ( ) ; this . _inboundSignal . clear ( ) ; this . _inbound . pause ( ) ; this . _inboundSignal . pause ( ) ; this . pending = [ ] ; this . emit ( "closed" , error ) ; this . removeAllListeners ( ) ; } public refreshDelayInfo ( id ) { this . throttlingIdSet . delete ( id ) ; if ( this . throttlingIdSet . size === 0 ) { this . timeTillThrottling = 0 ; } } private disconnectHandler ( reason ) { if ( this . messageBuffer . length > 0 ) { this . logger . sendErrorEvent ( { eventName : "OpenBatchOnDisconnect" , length : this . messageBuffer . length } ) ; this . messageBuffer . length = 0 ; } this . emit ( "disconnect" , reason ) ; } param id param delayMs param error public emitDelayInfo ( id , delayMs , error ) { const timeNow = Date . now ( ) ; this . throttlingIdSet . add ( id ) ; if ( delayMs > 0 && ( timeNow + delayMs > this . timeTillThrottling ) ) { this . timeTillThrottling = timeNow + delayMs ; const throttlingWarning = ThrottlingWarning . wrap ( error , "deltaManagerEmitDelayInfo" , delayMs / 1000 , this . logger ) ; this . emit ( "throttled" , throttlingWarning ) ; } } private comparableMessagePayload ( m ) { return `${ m . clientId }-${ m . type }-${ m . minimumSequenceNumber }-${ m . referenceSequenceNumber }-${ m . timestamp }` ; } private enqueueMessages ( messages , reason , allowGaps = false ) { if ( this . handler === undefined ) { this . pending = this . pending . concat ( messages ) ; return ; } assert ( this . pending . length === 0 || this . fetchReason !== undefined , 0x1e9 ) ; if ( messages . length === 0 ) { return ; } const from = messages [ 0 ] . sequenceNumber ; const last = messages [ messages . length - 1 ] . sequenceNumber ; if ( last > this . lastQueuedSequenceNumber ) { let prev = from - 1 ; const initialGap = prev - this . lastQueuedSequenceNumber ; let firstMissing ; let duplicate = 0 ; let gap = 0 ; for ( const message of messages ) { if ( message . sequenceNumber === prev ) { duplicate ++ ; } else if ( message . sequenceNumber !== prev + 1 ) { gap ++ ; if ( firstMissing === undefined ) { firstMissing = prev + 1 ; } } prev = message . sequenceNumber ; } let eventName ; if ( duplicate !== 0 || gap !== 0 && ! allowGaps || initialGap > 0 && this . fetchReason === undefined ) { eventName = "enqueueMessages" ; } else if ( this . fetchReason !== undefined && this . fetchReason !== reason && ( from <= this . lastQueuedSequenceNumber + 1 && last > this . lastQueuedSequenceNumber ) ) { eventName = "enqueueMessagesExtraFetch" ; } if ( eventName !== undefined ) { this . logger . sendPerformanceEvent ( { eventName , reason , previousReason : this . prevEnqueueMessagesReason , from , to : last + 1 , length : messages . length , fetchReason : this . fetchReason , duplicate : duplicate > 0 ? duplicate : undefined , initialGap : initialGap !== 0 ? initialGap : undefined , gap : gap > 0 ? gap : undefined , firstMissing , dmInitialSeqNumber : this . initialSequenceNumber , ... this . connectionManager . connectionVerboseProps } ) ; } } this . updateLatestKnownOpSeqNumber ( messages [ messages . length - 1 ] . sequenceNumber ) ; const n = this . previouslyProcessedMessage ?. sequenceNumber ; assert ( n === undefined || n === this . lastQueuedSequenceNumber , 0x0ec ) ; for ( const message of messages ) { if ( message . sequenceNumber <= this . lastQueuedSequenceNumber ) { if ( this . previouslyProcessedMessage ?. sequenceNumber === message . sequenceNumber ) { const message1 = this . comparableMessagePayload ( this . previouslyProcessedMessage ) ; const message2 = this . comparableMessagePayload ( message ) ; if ( message1 !== message2 ) { const error = new NonRetryableError ( "twoMessagesWithSameSeqNumAndDifferentPayload" , undefined , DriverErrorType . fileOverwrittenInStorage , { clientId : this . connectionManager . clientId , sequenceNumber : message . sequenceNumber , message1 , message2 } ) ; this . close ( error ) ; } } } else if ( message . sequenceNumber !== this . lastQueuedSequenceNumber + 1 ) { this . pending . push ( message ) ; this . fetchMissingDeltas ( reason , message . sequenceNumber ) ; } else { this . lastQueuedSequenceNumber = message . sequenceNumber ; this . previouslyProcessedMessage = message ; this . _inbound . push ( message ) ; } } this . prevEnqueueMessagesReason = this . pending . length > 0 ? "unknown" : reason ; } private processInboundMessage ( message ) { const startTime = Date . now ( ) ; this . lastProcessedMessage = message ; assert ( message . clientId !== undefined || isSystemMessage ( message ) , 0x0ed ) ; if ( typeof message . contents === "string" && message . contents !== "" && message . type !== MessageType . ClientLeave ) { message . contents = JSON . parse ( message . contents ) ; } this . connectionManager . beforeProcessingIncomingOp ( message ) ; if ( message . traces !== undefined && message . traces . length > 0 ) { message . traces . push ( { action : "end" , service : "client" , timestamp : Date . now ( ) } ) ; } if ( this . minSequenceNumber > message . minimumSequenceNumber ) { throw new DataCorruptionError ( "msnMovesBackwards" , { ... extractLogSafeMessageProperties ( message ) , clientId : this . connectionManager . clientId } ) ; } this . minSequenceNumber = message . minimumSequenceNumber ; if ( message . sequenceNumber !== this . lastProcessedSequenceNumber + 1 ) { throw new DataCorruptionError ( "nonSequentialSequenceNumber" , { ... extractLogSafeMessageProperties ( message ) , clientId : this . connectionManager . clientId } ) ; } this . lastProcessedSequenceNumber = message . sequenceNumber ; assert ( this . lastProcessedSequenceNumber <= this . lastObservedSeqNumber , 0x267 ) ; if ( message . term === undefined ) { message . term = 1 ; } this . baseTerm = message . term ; if ( this . handler === undefined ) { throw new Error ( "Attempted to process an inbound message without a handler attached" ) ; } this . handler . process ( message ) ; const endTime = Date . now ( ) ; this . emit ( "op" , message , endTime - startTime ) ; } /**
     * Retrieves the missing deltas between the given sequence numbers
     */ private fetchMissingDeltas ( reasonArg , to ? ) { this . fetchMissingDeltasCore ( reasonArg , false , to ) . catch ( ( error ) => { this . logger . sendErrorEvent ( { eventName : "fetchMissingDeltasException" } , error ) ; } ) ; } /**
    * Retrieves the missing deltas between the given sequence numbers
    */ private async fetchMissingDeltasCore ( reason , cacheOnly , to ? ) { if ( this . fetchReason !== undefined ) { return ; } if ( this . closed ) { this . logger . sendTelemetryEvent ( { eventName : "fetchMissingDeltasClosedConnection" , reason } ) ; return ; } if ( this . handler === undefined ) { assert ( this . lastQueuedSequenceNumber === 0 , 0x26b ) ; return ; } try { let from = this . lastQueuedSequenceNumber + 1 ; const n = this . previouslyProcessedMessage ?. sequenceNumber ; if ( n !== undefined ) { assert ( n === this . lastQueuedSequenceNumber , 0x0f2 ) ; assert ( from > 1 , 0x0f3 ) ; from -- ; } const fetchReason = `${ reason }_fetch` ; this . fetchReason = fetchReason ; await this . getDeltas ( from , to , ( messages ) => { this . refreshDelayInfo ( this . deltaStorageDelayId ) ; this . enqueueMessages ( messages , fetchReason ) ; } , cacheOnly ) ; } catch ( error ) { this . logger . sendErrorEvent ( { eventName : "GetDeltas_Exception" } , error ) ; this . close ( normalizeError ( error ) ) ; } finally { this . refreshDelayInfo ( this . deltaStorageDelayId ) ; this . fetchReason = undefined ; this . processPendingOps ( reason ) ; } } /**
     * Sorts pending ops and attempts to apply them
     */ private processPendingOps ( reason ? ) { if ( this . closed ) { return ; } assert ( this . handler !== undefined , 0x26c ) ; const pendingSorted = this . pending . sort ( ( a , b ) => a . sequenceNumber - b . sequenceNumber ) ; this . pending = [ ] ; this . enqueueMessages ( pendingSorted , `${ reason }_pending` , true ) ; if ( this . fetchReason === undefined ) { if ( this . lastQueuedSequenceNumber < this . lastObservedSeqNumber ) { this . fetchMissingDeltas ( "OpsBehind" ) ; } } } private updateLatestKnownOpSeqNumber ( seq ) { if ( this . lastObservedSeqNumber < seq ) { this . lastObservedSeqNumber = seq ; } } } function extractLogSafeMessageProperties ( message ) { const safeProps = { messageClientId : message . clientId , sequenceNumber : message . sequenceNumber , clientSequenceNumber : message . clientSequenceNumber , referenceSequenceNumber : message . referenceSequenceNumber , minimumSequenceNumber : message . minimumSequenceNumber , messageTimestamp : message . timestamp } ; return safeProps ; }