import { IDisposable , ITelemetryLogger , ITelemetryProperties , } from "@fluidframework/common-definitions" ; import { IDeltaQueue , ReadOnlyInfo , IConnectionDetails , } from "@fluidframework/container-definitions" ; import { assert , performance , TypedEventEmitter } from "@fluidframework/common-utils" ; import { TelemetryLogger , normalizeError , wrapError , } from "@fluidframework/telemetry-utils" ; import { IDocumentService , IDocumentDeltaConnection , IDocumentDeltaConnectionEvents , DriverError , } from "@fluidframework/driver-definitions" ; import { ConnectionMode , IClient , IClientConfiguration , IClientDetails , IDocumentMessage , INack , INackContent , ISequencedDocumentMessage , ISignalClient , ISignalMessage , ITokenClaims , MessageType , ScopeType , ISequencedDocumentSystemMessage , } from "@fluidframework/protocol-definitions" ; import { canRetryOnError , createWriteError , createGenericNetworkError , getRetryDelayFromError , logNetworkFailure , waitForConnectedState , DeltaStreamConnectionForbiddenError , GenericNetworkError , } from "@fluidframework/driver-utils" ; import { GenericError , } from "@fluidframework/container-utils" ; import { DeltaQueue } from "./deltaQueue" ; import { ReconnectMode , IConnectionManager , IConnectionManagerFactoryArgs , } from "./contracts" ; const MaxReconnectDelayInMs = 8000 ; const InitialReconnectDelayInMs = 1000 ; const DefaultChunkSize = 16 * 1024 ; const fatalConnectErrorProp = { fatalConnectError : true } ; function getNackReconnectInfo ( nackContent ) { const message = `Nack (${ nackContent . type }): ${ nackContent . message }` ; const canRetry = nackContent . code !== 403 ; const retryAfterMs = nackContent . retryAfter !== undefined ? nackContent . retryAfter * 1000 : undefined ; return createGenericNetworkError ( `nack [${ nackContent . code }]` , message , canRetry , retryAfterMs , { statusCode : nackContent . code } ) ; } const createReconnectError = ( fluidErrorCode , err ) => wrapError ( err , ( errorMessage ) => new GenericNetworkError ( fluidErrorCode , errorMessage , true ) ) ; /**
 * Implementation of IDocumentDeltaConnection that does not support submitting
 * or receiving ops. Used in storage-only mode.
 */ class NoDeltaStream extends TypedEventEmitter < IDocumentDeltaConnectionEvents > implements IDocumentDeltaConnection , IDisposable { clientId : string = "storage-only client" ; claims : ITokenClaims = { scopes : [ ScopeType . DocRead ] } as any ; mode : ConnectionMode = "read" ; existing : boolean = true ; maxMessageSize : number = 0 ; version : string = "" ; initialMessages : ISequencedDocumentMessage [ ] = [ ] ; initialSignals : ISignalMessage [ ] = [ ] ; initialClients : ISignalClient [ ] = [ ] ; serviceConfiguration : IClientConfiguration = { maxMessageSize : 0 , blockSize : 0 , summary : undefined as any } ; checkpointSequenceNumber ? : number | undefined = undefined ; submit ( messages ) { this . emit ( "nack" , this . clientId , messages . map ( ( operation ) => { return { operation , content : { message : "Cannot submit with storage-only connection" , code : 403 } } ; } ) ) ; } submitSignal ( message ) { this . emit ( "nack" , this . clientId , { operation : message , content : { message : "Cannot submit signal with storage-only connection" , code : 403 } } ) ; } private _disposed = false ; public get disposed ( ) { return this . _disposed ; } public dispose ( ) { this . _disposed = true ; } } /**
 * Implementation of IConnectionManager, used by Container class
 * Implements constant connectivity to relay service, by reconnecting in case of loast connection or error.
 * Exposes various controls to influecen this process, including manual reconnects, forced read-only mode, etc.
 */ export class ConnectionManager implements IConnectionManager { /** Connection mode used when reconnecting on error or disconnect. */ private readonly defaultReconnectionMode : ConnectionMode ; private pendingConnection = false ; private connection : IDocumentDeltaConnection | undefined ; /** file ACL - whether user has only read-only access to a file */ private _readonlyPermissions : boolean | undefined ; /** tracks host requiring read-only mode. */ private _forceReadonly = false ; /**
     * Controls whether the DeltaManager will automatically reconnect to the delta stream after receiving a disconnect.
     */ private _reconnectMode : ReconnectMode ; /** True if there is pending (async) reconnection from "read" to "write" */ private pendingReconnect = false ; /** downgrade "write" connection to "read" */ private downgradedConnection = false ; private clientSequenceNumber = 0 ; private clientSequenceNumberObserved = 0 ; /** Counts the number of noops sent by the client which may not be acked. */ private trailingNoopCount = 0 ; /** track clientId used last time when we sent any ops */ private lastSubmittedClientId : string | undefined ; private connectFirstConnection = true ; private _connectionVerboseProps : Record < string , string | number > = { } ; private _connectionProps : ITelemetryProperties = { } ; private closed = false ; private readonly _outbound : DeltaQueue < IDocumentMessage [ ] > ; public get connectionVerboseProps ( ) { return this . _connectionVerboseProps ; } public readonly clientDetails : IClientDetails ; /**
     * The current connection mode, initially read.
     */ public get connectionMode ( ) : ConnectionMode { assert ( ! this . downgradedConnection || this . connection ?. mode === "write" , 0x277 ) ; if ( this . connection === undefined ) { return "read" ; } return this . connection . mode ; } public get connected ( ) { return this . connection !== undefined ; } public get clientId ( ) { return this . connection ?. clientId ; } /**
     * Automatic reconnecting enabled or disabled.
     * If set to Never, then reconnecting will never be allowed.
     */ public get reconnectMode ( ) : ReconnectMode { return this . _reconnectMode ; } public get maxMessageSize ( ) : number { return this . connection ?. serviceConfiguration ?. maxMessageSize ?? DefaultChunkSize ; } public get version ( ) : string { if ( this . connection === undefined ) { throw new Error ( "Cannot check version without a connection" ) ; } return this . connection . version ; } public get serviceConfiguration ( ) : IClientConfiguration | undefined { return this . connection ?. serviceConfiguration ; } public get scopes ( ) : string [ ] | undefined { return this . connection ?. claims . scopes ; } public get outbound ( ) : IDeltaQueue < IDocumentMessage [ ] > { return this . _outbound ; } /**
     * Returns set of props that can be logged in telemetry that provide some insights / statistics
     * about current or last connection (if there is no connection at the moment)
    */ public get connectionProps ( ) : ITelemetryProperties { if ( this . connection !== undefined ) { return this . _connectionProps ; } else { return { ... this . _connectionProps , sentOps : this . clientSequenceNumber } ; } } public shouldJoinWrite ( ) { return this . clientSequenceNumberObserved < ( this . clientSequenceNumber - this . trailingNoopCount ) ; } /**
     * Tells if container is in read-only mode.
     * Data stores should listen for "readonly" notifications and disallow user
     * making changes to data stores.
     * Readonly state can be because of no storage write permission,
     * or due to host forcing readonly mode for container.
     * It is undefined if we have not yet established websocket connection
     * and do not know if user has write access to a file.
     */ private get readonly ( ) { if ( this . _forceReadonly ) { return true ; } return this . _readonlyPermissions ; } public get readOnlyInfo ( ) : ReadOnlyInfo { const storageOnly = this . connection !== undefined && this . connection instanceof NoDeltaStream ; if ( storageOnly || this . _forceReadonly || this . _readonlyPermissions === true ) { return { readonly : true , forced : this . _forceReadonly , permissions : this . _readonlyPermissions , storageOnly } ; } return { readonly : this . _readonlyPermissions } ; } private static detailsFromConnection ( connection ) { return { claims : connection . claims , clientId : connection . clientId , existing : connection . existing , checkpointSequenceNumber : connection . checkpointSequenceNumber , get initialClients ( ) { return connection . initialClients ; } , mode : connection . mode , serviceConfiguration : connection . serviceConfiguration , version : connection . version } ; } constructor ( private readonly serviceProvider , private client , reconnectAllowed , private readonly logger , private readonly props ) { this . clientDetails = this . client . details ; this . defaultReconnectionMode = this . client . mode ; this . _reconnectMode = reconnectAllowed ? ReconnectMode . Enabled : ReconnectMode . Never ; this . _outbound = new DeltaQueue < IDocumentMessage [ ] > ( ( messages ) => { if ( this . connection === undefined ) { throw new Error ( "Attempted to submit an outbound message without connection" ) ; } this . connection . submit ( messages ) ; } ) ; this . _outbound . on ( "error" , ( error ) => { this . props . closeHandler ( normalizeError ( error ) ) ; } ) ; } public dispose ( error ) { if ( this . closed ) { return ; } this . closed = true ; this . pendingConnection = false ; this . _reconnectMode = ReconnectMode . Never ; this . _outbound . clear ( ) ; const disconnectReason = error !== undefined ? `Closing DeltaManager (${ error . message })` : "Closing DeltaManager" ; this . disconnectFromDeltaStream ( disconnectReason ) ; this . set_readonlyPermissions ( true ) ; } /**
     * Enables or disables automatic reconnecting.
     * Will throw an error if reconnectMode set to Never.
    */ public setAutoReconnect ( mode ) { assert ( mode !== ReconnectMode . Never && this . _reconnectMode !== ReconnectMode . Never , 0x278 ) ; this . _reconnectMode = mode ; if ( mode !== ReconnectMode . Enabled ) { this . disconnectFromDeltaStream ( "setAutoReconnect" ) ; } } param readonly public forceReadonly ( readonly ) { if ( readonly !== this . _forceReadonly ) { this . logger . sendTelemetryEvent ( { eventName : "ForceReadOnly" , value : readonly } ) ; } const oldValue = this . readonly ; this . _forceReadonly = readonly ; if ( oldValue !== this . readonly ) { assert ( this . _reconnectMode !== ReconnectMode . Never , 0x279 ) ; let reconnect = false ; if ( this . readonly === true ) { if ( this . shouldJoinWrite ( ) ) { this . logger . sendErrorEvent ( { eventName : "ForceReadonlyPendingChanged" } ) ; } reconnect = this . disconnectFromDeltaStream ( "Force readonly" ) ; } this . props . readonlyChangeHandler ( this . readonly ) ; if ( reconnect ) { this . triggerConnect ( "read" ) ; } } } private set_readonlyPermissions ( readonly ) { const oldValue = this . readonly ; this . _readonlyPermissions = readonly ; if ( oldValue !== this . readonly ) { this . props . readonlyChangeHandler ( this . readonly ) ; } } public connect ( connectionMode ? ) { this . connectCore ( connectionMode ) . catch ( ( error ) => { const normalizedError = normalizeError ( error , { props : fatalConnectErrorProp } ) ; this . props . closeHandler ( normalizedError ) ; } ) ; } private async connectCore ( connectionMode ? ) { assert ( ! this . closed , 0x26a ) ; if ( this . connection !== undefined || this . pendingConnection ) { return ; } let requestedMode = connectionMode ?? this . defaultReconnectionMode ; if ( this . shouldJoinWrite ( ) ) { requestedMode = "write" ; } const docService = this . serviceProvider ( ) ; assert ( docService !== undefined , 0x2a7 ) ; let connection ; if ( docService . policies ?. storageOnly === true ) { connection = new NoDeltaStream ( ) ; this . pendingConnection = true ; this . setupNewSuccessfulConnection ( connection , "read" ) ; assert ( ! this . pendingConnection , 0x2b3 ) ; return ; } this . pendingConnection = true ; let delayMs = InitialReconnectDelayInMs ; let connectRepeatCount = 0 ; const connectStartTime = performance . now ( ) ; let lastError ; while ( connection === undefined ) { if ( this . closed ) { throw new Error ( "Attempting to connect a closed DeltaManager" ) ; } connectRepeatCount ++ ; try { this . client . mode = requestedMode ; connection = await docService . connectToDeltaStream ( this . client ) ; if ( connection . disposed ) { this . logger . sendTelemetryEvent ( { eventName : "ReceivedClosedConnection" } ) ; connection = undefined ; } } catch ( origError ) { if ( typeof origError === "object" && origError !== null && origError ?. errorType === DeltaStreamConnectionForbiddenError . errorType ) { connection = new NoDeltaStream ( ) ; requestedMode = "read" ; break ; } if ( ! canRetryOnError ( origError ) ) { const error = normalizeError ( origError , { props : fatalConnectErrorProp } ) ; this . props . closeHandler ( error ) ; throw error ; } if ( connectRepeatCount === 1 ) { logNetworkFailure ( this . logger , { delay : delayMs , eventName : "DeltaConnectionFailureToConnect" , duration : TelemetryLogger . formatTick ( performance . now ( ) - connectStartTime ) } , origError ) ; } lastError = origError ; const retryDelayFromError = getRetryDelayFromError ( origError ) ; delayMs = retryDelayFromError ?? Math . min ( delayMs * 2 , MaxReconnectDelayInMs ) ; if ( retryDelayFromError !== undefined ) { this . props . reconnectionDelayHandler ( retryDelayFromError , origError ) ; } await waitForConnectedState ( delayMs ) ; } } if ( connectRepeatCount > 1 ) { this . logger . sendTelemetryEvent ( { eventName : "MultipleDeltaConnectionFailures" , attempts : connectRepeatCount , duration : TelemetryLogger . formatTick ( performance . now ( ) - connectStartTime ) } , lastError ) ; } this . setupNewSuccessfulConnection ( connection , requestedMode ) ; } param args private triggerConnect ( connectionMode ) { assert ( this . connection === undefined , 0x239 ) ; if ( this . reconnectMode !== ReconnectMode . Enabled ) { return ; } this . connect ( connectionMode ) ; } param reason private disconnectFromDeltaStream ( reason ) { this . pendingReconnect = false ; this . downgradedConnection = false ; if ( this . connection === undefined ) { return false ; } assert ( ! this . pendingConnection , 0x27b ) ; const connection = this . connection ; this . connection = undefined ; connection . off ( "op" , this . opHandler ) ; connection . off ( "signal" , this . props . signalHandler ) ; connection . off ( "nack" , this . nackHandler ) ; connection . off ( "disconnect" , this . disconnectHandlerInternal ) ; connection . off ( "error" , this . errorHandler ) ; connection . off ( "pong" , this . props . pongHandler ) ; this . _outbound . pause ( ) ; this . _outbound . clear ( ) ; this . props . disconnectHandler ( reason ) ; connection . dispose ( ) ; this . _connectionVerboseProps = { } ; return true ; } param connection private setupNewSuccessfulConnection ( connection , requestedMode ) { assert ( this . connection === undefined , 0x0e6 ) ; assert ( ! connection . disposed , 0x28a ) ; if ( this . pendingConnection ) { this . pendingConnection = false ; } else { assert ( this . closed , 0x27f ) ; } this . connection = connection ; const readonly = ! connection . claims . scopes . includes ( ScopeType . DocWrite ) ; assert ( requestedMode === "read" || readonly === ( this . connectionMode === "read" ) , 0x0e7 ) ; assert ( ! readonly || this . connectionMode === "read" , 0x0e8 ) ; this . set_readonlyPermissions ( readonly ) ; if ( this . closed ) { this . disconnectFromDeltaStream ( "ConnectionManager already closed" ) ; return ; } this . _outbound . resume ( ) ; connection . on ( "op" , this . opHandler ) ; connection . on ( "signal" , this . props . signalHandler ) ; connection . on ( "nack" , this . nackHandler ) ; connection . on ( "disconnect" , this . disconnectHandlerInternal ) ; connection . on ( "error" , this . errorHandler ) ; connection . on ( "pong" , this . props . pongHandler ) ; const initialMessages = connection . initialMessages . sort ( ( a , b ) => a . sequenceNumber - b . sequenceNumber ) ; let checkpointSequenceNumber = connection . checkpointSequenceNumber ; this . _connectionVerboseProps = { clientId : connection . clientId , mode : connection . mode } ; this . _connectionProps = { } ; if ( connection . relayServiceAgent !== undefined ) { this . _connectionVerboseProps . relayServiceAgent = connection . relayServiceAgent ; this . _connectionProps . relayServiceAgent = connection . relayServiceAgent ; } this . _connectionProps . socketDocumentId = connection . claims . documentId ; this . _connectionProps . connectionMode = connection . mode ; let last = - 1 ; if ( initialMessages . length !== 0 ) { this . _connectionVerboseProps . connectionInitialOpsFrom = initialMessages [ 0 ] . sequenceNumber ; last = initialMessages [ initialMessages . length - 1 ] . sequenceNumber ; this . _connectionVerboseProps . connectionInitialOpsTo = last + 1 ; if ( checkpointSequenceNumber === undefined || checkpointSequenceNumber < last ) { checkpointSequenceNumber = last ; } } this . props . incomingOpHandler ( initialMessages , this . connectFirstConnection ? "InitialOps" : "ReconnectOps" ) ; if ( connection . initialSignals !== undefined ) { for ( const signal of connection . initialSignals ) { this . props . signalHandler ( signal ) ; } } const details = ConnectionManager . detailsFromConnection ( connection ) ; details . checkpointSequenceNumber = checkpointSequenceNumber ; this . props . connectHandler ( details ) ; this . connectFirstConnection = false ; } param connection param requestedMode param error returns private reconnectOnError ( requestedMode , error ) { this . reconnectOnErrorCore ( requestedMode , error . message , error ) ; } param connection param requestedMode param error returns private async reconnectOnErrorCore ( requestedMode , disconnectMessage , error ? ) { assert ( this . connection !== undefined , 0x0eb ) ; this . disconnectFromDeltaStream ( disconnectMessage ) ; const canRetry = error !== undefined ? canRetryOnError ( error ) : true ; if ( ! canRetry ) { this . props . closeHandler ( normalizeError ( error , { props : fatalConnectErrorProp } ) ) ; } else if ( this . reconnectMode === ReconnectMode . Never ) { this . props . closeHandler ( ) ; } if ( this . closed || this . reconnectMode !== ReconnectMode . Enabled ) { return ; } const delayMs = error !== undefined ? getRetryDelayFromError ( error ) : undefined ; if ( delayMs !== undefined ) { this . props . reconnectionDelayHandler ( delayMs , error ) ; await waitForConnectedState ( delayMs ) ; } this . triggerConnect ( requestedMode ) ; } public prepareMessageToSend ( message ) { if ( this . readonly === true ) { assert ( this . readOnlyInfo . readonly === true , 0x1f0 ) ; const error = new GenericError ( "deltaManagerReadonlySubmit" , undefined , { readonly : this . readOnlyInfo . readonly , forcedReadonly : this . readOnlyInfo . forced , readonlyPermissions : this . readOnlyInfo . permissions , storageOnly : this . readOnlyInfo . storageOnly } ) ; this . props . closeHandler ( error ) ; return undefined ; } assert ( ! ! this . connection , 0x0e4 ) ; if ( this . lastSubmittedClientId !== this . connection ?. clientId ) { this . lastSubmittedClientId = this . connection ?. clientId ; this . clientSequenceNumber = 0 ; this . clientSequenceNumberObserved = 0 ; } if ( message . type === MessageType . NoOp ) { this . trailingNoopCount ++ ; } else { this . trailingNoopCount = 0 ; } return { ... message , clientSequenceNumber : ++ this . clientSequenceNumber } ; } public submitSignal ( content ) { if ( this . connection !== undefined ) { this . connection . submitSignal ( content ) ; } else { this . logger . sendErrorEvent ( { eventName : "submitSignalDisconnected" } ) ; } } public sendMessages ( messages ) { assert ( this . connected , 0x2b4 ) ; if ( this . connectionMode === "read" || this . downgradedConnection ) { if ( ! this . pendingReconnect ) { this . pendingReconnect = true ; Promise . resolve ( ) . then ( async ( ) => { if ( this . pendingReconnect ) { await this . reconnectOnErrorCore ( "write" , "Switch to write" ) ; } } ) . catch ( ( ) => { } ) ; } return ; } assert ( ! this . pendingReconnect , 0x2b5 ) ; this . _outbound . push ( messages ) ; } public beforeProcessingIncomingOp ( message ) { assert ( this . clientId !== message . clientId || this . lastSubmittedClientId === message . clientId , 0x0ee ) ; if ( this . lastSubmittedClientId !== undefined && this . lastSubmittedClientId === message . clientId ) { const clientSequenceNumber = message . clientSequenceNumber ; assert ( this . clientSequenceNumberObserved < clientSequenceNumber , 0x0ef ) ; assert ( clientSequenceNumber <= this . clientSequenceNumber , 0x0f0 ) ; this . clientSequenceNumberObserved = clientSequenceNumber ; } if ( message . type === MessageType . ClientLeave ) { const systemLeaveMessage = message as ISequencedDocumentSystemMessage ; const clientId = JSON . parse ( systemLeaveMessage . data ) as string ; if ( clientId === this . clientId ) { this . logger . sendPerformanceEvent ( { eventName : "ReadConnectionTransition" } ) ; this . downgradedConnection = true ; } } } private readonly opHandler = ( documentId , messagesArg ) => { const messages = Array . isArray ( messagesArg ) ? messagesArg : [ messagesArg ] ; this . props . incomingOpHandler ( messages , "opHandler" ) ; } ; private readonly nackHandler = ( documentId , messages ) => { const message = messages [ 0 ] ; if ( this . _readonlyPermissions ) { this . props . closeHandler ( createWriteError ( "writeOnReadOnlyDocument" ) ) ; } const reconnectInfo = message . content !== undefined ? getNackReconnectInfo ( message . content ) : createGenericNetworkError ( "nackReasonUnknown" , undefined , true ) ; this . reconnectOnError ( "write" , reconnectInfo ) ; } ; private readonly disconnectHandlerInternal = ( disconnectReason ) => { this . reconnectOnError ( this . defaultReconnectionMode , createReconnectError ( "dmDocumentDeltaConnectionDisconnected" , disconnectReason ) ) ; } ; private readonly errorHandler = ( error ) => { this . reconnectOnError ( this . defaultReconnectionMode , createReconnectError ( "dmDocumentDeltaConnectionError" , error ) ) ; } ; }