import merge from "lodash/merge" ; import { v4 as uuid } from "uuid" ; import { IDisposable , } from "@fluidframework/common-definitions" ; import { assert , performance , unreachableCase } from "@fluidframework/common-utils" ; import { IRequest , IResponse , IFluidRouter , IFluidCodeDetails , isFluidCodeDetails , } from "@fluidframework/core-interfaces" ; import { IAudience , IConnectionDetails , IContainer , IContainerEvents , IDeltaManager , ICriticalContainerError , ContainerWarning , AttachState , IThrottlingWarning , IPendingLocalState , ReadOnlyInfo , IContainerLoadMode , } from "@fluidframework/container-definitions" ; import { DataCorruptionError , extractSafePropertiesFromMessage , GenericError , UsageError , } from "@fluidframework/container-utils" ; import { IDocumentService , IDocumentStorageService , IFluidResolvedUrl , IResolvedUrl , } from "@fluidframework/driver-definitions" ; import { readAndParse , OnlineStatus , isOnline , ensureFluidResolvedUrl , combineAppAndProtocolSummary , runWithRetry , isFluidResolvedUrl , } from "@fluidframework/driver-utils" ; import { isSystemMessage , ProtocolOpHandler , } from "@fluidframework/protocol-base" ; import { FileMode , IClient , IClientConfiguration , IClientDetails , ICommittedProposal , IDocumentAttributes , IDocumentMessage , IProcessMessageResult , IQuorum , ISequencedClient , ISequencedDocumentMessage , ISequencedProposal , ISignalClient , ISignalMessage , ISnapshotTree , ITree , ITreeEntry , IVersion , MessageType , TreeEntry , ISummaryTree , IPendingProposal , SummaryType , ISummaryContent , IQuorumProposals , } from "@fluidframework/protocol-definitions" ; import { ChildLogger , EventEmitterWithErrorHandling , PerformanceEvent , raiseConnectedEvent , TelemetryLogger , connectedEventName , disconnectedEventName , normalizeError , MonitoringContext , loggerToMonitoringContext , } from "@fluidframework/telemetry-utils" ; import { Audience } from "./audience" ; import { ContainerContext } from "./containerContext" ; import { ReconnectMode , IConnectionManagerFactoryArgs } from "./contracts" ; import { DeltaManager , IConnectionArgs } from "./deltaManager" ; import { DeltaManagerProxy } from "./deltaManagerProxy" ; import { ILoaderOptions , Loader , RelativeLoader } from "./loader" ; import { pkgVersion } from "./packageVersion" ; import { ConnectionStateHandler , ILocalSequencedClient } from "./connectionStateHandler" ; import { RetriableDocumentStorageService } from "./retriableDocumentStorageService" ; import { ProtocolTreeStorageService } from "./protocolTreeDocumentStorageService" ; import { BlobOnlyStorage , ContainerStorageAdapter } from "./containerStorageAdapter" ; import { getSnapshotTreeFromSerializedContainer } from "./utils" ; import { QuorumProxy } from "./quorum" ; import { CollabWindowTracker } from "./collabWindowTracker" ; import { ConnectionManager } from "./connectionManager" ; const detachedContainerRefSeqNumber = 0 ; const dirtyContainerEvent = "dirty" ; const savedContainerEvent = "saved" ; export interface IContainerLoadOptions { /**
     * Disables the Container from reconnecting if false, allows reconnect otherwise.
     */ canReconnect ? : boolean ; /**
     * Client details provided in the override will be merged over the default client.
     */ clientDetailsOverride ? : IClientDetails ; resolvedUrl : IFluidResolvedUrl ; /**
     * Control which snapshot version to load from.  See IParsedUrl for detailed information.
     */ version : string | undefined ; /**
     * Loads the Container in paused state if true, unpaused otherwise.
     */ loadMode ? : IContainerLoadMode ; } export interface IContainerConfig { resolvedUrl ? : IFluidResolvedUrl ; canReconnect ? : boolean ; /**
     * Client details provided in the override will be merged over the default client.
     */ clientDetailsOverride ? : IClientDetails ; } export enum ConnectionState { /**
     * The document is no longer connected to the delta server
     */ Disconnected , /**
     * The document has an inbound connection but is still pending for outbound deltas
     */ Connecting , /**
     * The document is fully connected
     */ Connected } returns export async function waitContainerToCatchUp ( container ) { if ( container . closed ) { throw new Error ( "Container is closed" ) ; } return new Promise < boolean > ( ( resolve , reject ) => { const deltaManager = container . deltaManager ; container . on ( "closed" , reject ) ; const waitForOps = ( ) => { assert ( container . connectionState !== ConnectionState . Disconnected , 0x0cd ) ; const hasCheckpointSequenceNumber = deltaManager . hasCheckpointSequenceNumber ; const connectionOpSeqNumber = deltaManager . lastKnownSeqNumber ; assert ( deltaManager . lastSequenceNumber <= connectionOpSeqNumber , 0x266 ) ; if ( deltaManager . lastSequenceNumber === connectionOpSeqNumber ) { resolve ( hasCheckpointSequenceNumber ) ; return ; } const callbackOps = ( message ) => { if ( connectionOpSeqNumber <= message . sequenceNumber ) { resolve ( hasCheckpointSequenceNumber ) ; deltaManager . off ( "op" , callbackOps ) ; } } ; deltaManager . on ( "op" , callbackOps ) ; } ; if ( container . connectionState === ConnectionState . Connected ) { waitForOps ( ) ; return ; } const callback = ( ) => { container . off ( connectedEventName , callback ) ; waitForOps ( ) ; } ; container . on ( connectedEventName , callback ) ; container . resume ! ( ) ; } ) ; } const getCodeProposal = ( quorum ) => quorum . get ( "code" ) ?? quorum . get ( "code2" ) ; export class Container extends EventEmitterWithErrorHandling < IContainerEvents > implements IContainer { public static version = "^0.1.0" ; /**
     * Load an existing container.
     */ public static async load ( loader , loadOptions , pendingLocalState ? ) { const container = new Container ( loader , { clientDetailsOverride : loadOptions . clientDetailsOverride , resolvedUrl : loadOptions . resolvedUrl , canReconnect : loadOptions . canReconnect } ) ; return PerformanceEvent . timedExecAsync ( container . mc . logger , { eventName : "Load" } , async ( event ) => new Promise < Container > ( ( resolve , reject ) => { container . _lifecycleState = "loading" ; const version = loadOptions . version ; const defaultMode = { opsBeforeReturn : "cached" } ; assert ( pendingLocalState === undefined || loadOptions . loadMode === undefined , 0x1e1 ) ; const mode = loadOptions . loadMode ?? defaultMode ; const onClosed = ( err ? ) => { reject ( err ?? new GenericError ( "containerClosedWithoutErrorDuringLoad" ) ) ; } ; container . on ( "closed" , onClosed ) ; container . load ( version , mode , pendingLocalState ) . finally ( ( ) => { container . removeListener ( "closed" , onClosed ) ; } ) . then ( ( props ) => { event . end ( { ... props , ... loadOptions . loadMode } ) ; resolve ( container ) ; } , ( error ) => { const err = normalizeError ( error ) ; container . close ( err ) ; onClosed ( err ) ; } ) ; } ) , { start : true , end : true , cancel : "generic" } ) ; } /**
     * Create a new container in a detached state.
     */ public static async createDetached ( loader , codeDetails ) { const container = new Container ( loader , { } ) ; return PerformanceEvent . timedExecAsync ( container . mc . logger , { eventName : "CreateDetached" } , async ( _event ) => { container . _lifecycleState = "loading" ; await container . createDetached ( codeDetails ) ; return container ; } , { start : true , end : true , cancel : "generic" } ) ; } /**
     * Create a new container in a detached state that is initialized with a
     * snapshot from a previous detached container.
     */ public static async rehydrateDetachedFromSnapshot ( loader , snapshot ) { const container = new Container ( loader , { } ) ; return PerformanceEvent . timedExecAsync ( container . mc . logger , { eventName : "RehydrateDetachedFromSnapshot" } , async ( _event ) => { const deserializedSummary = JSON . parse ( snapshot ) as ISummaryTree ; container . _lifecycleState = "loading" ; await container . rehydrateDetachedFromSnapshot ( deserializedSummary ) ; return container ; } , { start : true , end : true , cancel : "generic" } ) ; } public subLogger : TelemetryLogger ; private readonly _canReconnect : boolean = true ; private readonly mc : MonitoringContext ; private _lifecycleState : "created" | "loading" | "loaded" | "closing" | "closed" = "created" ; private get loaded ( ) : boolean { return ( this . _lifecycleState !== "created" && this . _lifecycleState !== "loading" ) ; } private set loaded ( t ) { assert ( t , 0x27d ) ; assert ( this . _lifecycleState !== "created" , 0x27e ) ; if ( this . _lifecycleState === "loading" ) { this . _lifecycleState = "loaded" ; } } public get closed ( ) : boolean { return ( this . _lifecycleState === "closing" || this . _lifecycleState === "closed" ) ; } private _attachState = AttachState . Detached ; private readonly _storage : ContainerStorageAdapter ; public get storage ( ) : IDocumentStorageService { return this . _storage ; } private _storageService : IDocumentStorageService & IDisposable | undefined ; private get storageService ( ) : IDocumentStorageService { if ( this . _storageService === undefined ) { throw new Error ( "Attempted to access storageService before it was defined" ) ; } return this . _storageService ; } private readonly clientDetailsOverride : IClientDetails | undefined ; private readonly _deltaManager : DeltaManager < ConnectionManager > ; private service : IDocumentService | undefined ; private readonly _audience : Audience ; private _context : ContainerContext | undefined ; private get context ( ) { if ( this . _context === undefined ) { throw new Error ( "Attempted to access context before it was defined" ) ; } return this . _context ; } private _protocolHandler : ProtocolOpHandler | undefined ; private get protocolHandler ( ) { if ( this . _protocolHandler === undefined ) { throw new Error ( "Attempted to access protocolHandler before it was defined" ) ; } return this . _protocolHandler ; } private resumedOpProcessingAfterLoad = false ; private firstConnection = true ; private manualReconnectInProgress = false ; private readonly connectionTransitionTimes : number [ ] = [ ] ; private messageCountAfterDisconnection : number = 0 ; private _loadedFromVersion : IVersion | undefined ; private _resolvedUrl : IFluidResolvedUrl | undefined ; private attachStarted = false ; private _dirtyContainer = false ; private lastVisible : number | undefined ; private readonly visibilityEventHandler : ( ( ) => void ) | undefined ; private readonly connectionStateHandler : ConnectionStateHandler ; private setAutoReconnectTime = performance . now ( ) ; private readonly collabWindowTracker = new CollabWindowTracker ( ( type , contents ) => this . submitMessage ( type , contents ) , ( ) => this . activeConnection ( ) , this . loader . services . options ?. noopTimeFrequency , this . loader . services . options ?. noopCountFrequency ) ; private get connectionMode ( ) { return this . _deltaManager . connectionManager . connectionMode ; } public get IFluidRouter ( ) : IFluidRouter { return this ; } public get resolvedUrl ( ) : IResolvedUrl | undefined { return this . _resolvedUrl ; } public get loadedFromVersion ( ) : IVersion | undefined { return this . _loadedFromVersion ; } public get readOnlyInfo ( ) : ReadOnlyInfo { return this . _deltaManager . readOnlyInfo ; } /**
     * Tracks host requiring read-only mode.
     */ public forceReadonly ( readonly ) { this . _deltaManager . connectionManager . forceReadonly ( readonly ) ; } public get id ( ) : string { return this . _resolvedUrl ?. id ?? "" ; } public get deltaManager ( ) : IDeltaManager < ISequencedDocumentMessage , IDocumentMessage > { return this . _deltaManager ; } public get connectionState ( ) : ConnectionState { return this . connectionStateHandler . connectionState ; } public get connected ( ) : boolean { return this . connectionStateHandler . connected ; } /**
     * Service configuration details. If running in offline mode will be undefined otherwise will contain service
     * configuration details returned as part of the initial connection.
     */ public get serviceConfiguration ( ) : IClientConfiguration | undefined { return this . _deltaManager . serviceConfiguration ; } /**
     * The server provided id of the client.
     * Set once this.connected is true, otherwise undefined
     */ public get clientId ( ) : string | undefined { return this . connectionStateHandler . clientId ; } /**
     * The server provided claims of the client.
     * Set once this.connected is true, otherwise undefined
     */ public get scopes ( ) : string [ ] | undefined { return this . _deltaManager . connectionManager . scopes ; } public get clientDetails ( ) : IClientDetails { return this . _deltaManager . clientDetails ; } deprecated public get codeDetails ( ) : IFluidCodeDetails | undefined { return this . _context ?. codeDetails ?? this . getCodeDetailsFromQuorum ( ) ; } returns public getSpecifiedCodeDetails ( ) { return this . getCodeDetailsFromQuorum ( ) ; } returns public getLoadedCodeDetails ( ) { return this . _context ?. codeDetails ; } /**
     * Retrieves the audience associated with the document
     */ public get audience ( ) : IAudience { return this . _audience ; } /**
     * Returns true if container is dirty.
     * Which means data loss if container is closed at that same moment
     * Most likely that happens when there is no network connection to ordering service
     */ public get isDirty ( ) { return this . _dirtyContainer ; } private get serviceFactory ( ) { return this . loader . services . documentServiceFactory ; } private get urlResolver ( ) { return this . loader . services . urlResolver ; } public readonly options : ILoaderOptions ; private get scope ( ) { return this . loader . services . scope ; } private get codeLoader ( ) { return this . loader . services . codeLoader ; } constructor ( private readonly loader , config ) { super ( ( name , error ) => { this . mc . logger . sendErrorEvent ( { eventName : "ContainerEventHandlerException" , name : typeof name === "string" ? name : undefined } , error ) ; } ) ; this . _audience = new Audience ( ) ; this . clientDetailsOverride = config . clientDetailsOverride ; this . _resolvedUrl = config . resolvedUrl ; if ( config . canReconnect !== undefined ) { this . _canReconnect = config . canReconnect ; } const type = this . client . details . type ; const interactive = this . client . details . capabilities . interactive ; const clientType = `${ interactive ? "interactive" : "noninteractive" }${ type !== undefined && type !== "" ? `/${ type }` : "" }` ; this . subLogger = ChildLogger . create ( loader . services . subLogger , undefined , { all : { clientType , containerId : uuid ( ) , docId : ( ) => this . id , containerAttachState : ( ) => this . _attachState , containerLifecycleState : ( ) => this . _lifecycleState , containerConnectionState : ( ) => ConnectionState [ this . connectionState ] } , error : { dmInitialSeqNumber : ( ) => this . _deltaManager ?. initialSequenceNumber , dmLastKnownSeqNumber : ( ) => this . _deltaManager ?. lastKnownSeqNumber , containerLoadedFromVersionId : ( ) => this . loadedFromVersion ?. id , containerLoadedFromVersionDate : ( ) => this . loadedFromVersion ?. date , dmLastMsqSeqNumber : ( ) => this . deltaManager ?. lastMessage ?. sequenceNumber , dmLastMsqSeqTimestamp : ( ) => this . deltaManager ?. lastMessage ?. timestamp , dmLastMsqSeqClientId : ( ) => this . deltaManager ?. lastMessage ?. clientId , connectionStateDuration : ( ) => performance . now ( ) - this . connectionTransitionTimes [ this . connectionState ] } } ) ; this . mc = loggerToMonitoringContext ( ChildLogger . create ( this . subLogger , "Container" ) ) ; const summarizeProtocolTree = this . mc . config . getBoolean ( "Fluid.Container.summarizeProtocolTree" ) ?? this . loader . services . options . summarizeProtocolTree ; this . options = { ... this . loader . services . options , summarizeProtocolTree } ; this . connectionStateHandler = new ConnectionStateHandler ( { protocolHandler : ( ) => this . _protocolHandler , logConnectionStateChangeTelemetry : ( value , oldState , reason ) => this . logConnectionStateChangeTelemetry ( value , oldState , reason ) , shouldClientJoinWrite : ( ) => this . _deltaManager . connectionManager . shouldJoinWrite ( ) , maxClientLeaveWaitTime : this . loader . services . options . maxClientLeaveWaitTime , logConnectionIssue : ( eventName ) => { this . _deltaManager . logConnectionIssue ( { eventName , duration : performance . now ( ) - this . connectionTransitionTimes [ ConnectionState . Connecting ] } ) ; } , connectionStateChanged : ( ) => { if ( this . loaded ) { this . propagateConnectionState ( ) ; } } } , this . mc . logger ) ; this . _deltaManager = this . createDeltaManager ( ) ; this . _storage = new ContainerStorageAdapter ( ( ) => { if ( this . attachState !== AttachState . Attached ) { if ( this . loader . services . detachedBlobStorage !== undefined ) { return new BlobOnlyStorage ( this . loader . services . detachedBlobStorage , this . mc . logger ) ; } this . mc . logger . sendErrorEvent ( { eventName : "NoRealStorageInDetachedContainer" } ) ; throw new Error ( "Real storage calls not allowed in Unattached container" ) ; } return this . storageService ; } ) ; const isDomAvailable = typeof document === "object" && document !== null && typeof document . addEventListener === "function" && document . addEventListener !== null ; if ( isDomAvailable ) { this . lastVisible = document . hidden ? performance . now ( ) : undefined ; this . visibilityEventHandler = ( ) => { if ( document . hidden ) { this . lastVisible = performance . now ( ) ; } else { setTimeout ( ( ) => this . lastVisible = undefined , 0 ) ; } } ; document . addEventListener ( "visibilitychange" , this . visibilityEventHandler ) ; } this . on ( "newListener" , ( event , listener ) => { Promise . resolve ( ) . then ( ( ) => { switch ( event ) { case dirtyContainerEvent : if ( this . _dirtyContainer ) { listener ( ) ; } break ; case savedContainerEvent : if ( ! this . _dirtyContainer ) { listener ( ) ; } break ; case connectedEventName : if ( this . connected ) { listener ( this . clientId ) ; } break ; case disconnectedEventName : if ( ! this . connected ) { listener ( ) ; } break ; default : } } ) . catch ( ( error ) => { this . mc . logger . sendErrorEvent ( { eventName : "RaiseConnectedEventError" } , error ) ; } ) ; } ) ; } /**
     * Retrieves the quorum associated with the document
     */ public getQuorum ( ) { return this . protocolHandler . quorum ; } public close ( error ? ) { if ( this . closed ) { return ; } try { this . _lifecycleState = "closing" ; try { this . _deltaManager . close ( error ) ; this . _protocolHandler ?. close ( ) ; this . connectionStateHandler . dispose ( ) ; this . _context ?. dispose ( error !== undefined ? new Error ( error . message ) : undefined ) ; assert ( this . connectionState === ConnectionState . Disconnected , 0x0cf ) ; this . _storageService ?. dispose ( ) ; this . service ?. dispose ( error ) ; } catch ( exception ) { this . mc . logger . sendErrorEvent ( { eventName : "ContainerCloseException" } , exception ) ; } this . mc . logger . sendTelemetryEvent ( { eventName : "ContainerClose" , category : error === undefined ? "generic" : "error" } , error ) ; this . emit ( "closed" , error ) ; this . removeAllListeners ( ) ; if ( this . visibilityEventHandler !== undefined ) { document . removeEventListener ( "visibilitychange" , this . visibilityEventHandler ) ; } } finally { this . _lifecycleState = "closed" ; } } public closeAndGetPendingLocalState ( ) { assert ( this . attachState === AttachState . Attached , 0x0d1 ) ; assert ( this . resolvedUrl !== undefined && this . resolvedUrl . type === "fluid" , 0x0d2 ) ; const pendingState = { pendingRuntimeState : this . context . getPendingLocalState ( ) , url : this . resolvedUrl . url } ; this . close ( ) ; return JSON . stringify ( pendingState ) ; } public get attachState ( ) : AttachState { return this . _attachState ; } public serialize ( ) { assert ( this . attachState === AttachState . Detached , 0x0d3 ) ; const appSummary = this . context . createSummary ( ) ; const protocolSummary = this . captureProtocolSummary ( ) ; const combinedSummary = combineAppAndProtocolSummary ( appSummary , protocolSummary ) ; if ( this . loader . services . detachedBlobStorage && this . loader . services . detachedBlobStorage . size > 0 ) { combinedSummary . tree [ ".hasAttachmentBlobs" ] = { type : SummaryType . Blob , content : "true" } ; } return JSON . stringify ( combinedSummary ) ; } public async attach ( request ) { await PerformanceEvent . timedExecAsync ( this . mc . logger , { eventName : "Attach" } , async ( ) => { if ( this . _lifecycleState !== "loaded" ) { throw new UsageError ( `containerNotValidForAttach [${ this . _lifecycleState }]` ) ; } assert ( this . _attachState === AttachState . Detached && ! this . attachStarted , 0x205 ) ; this . attachStarted = true ; const hasAttachmentBlobs = this . loader . services . detachedBlobStorage !== undefined && this . loader . services . detachedBlobStorage . size > 0 ; try { assert ( this . deltaManager . inbound . length === 0 , 0x0d6 ) ; let summary ; if ( ! hasAttachmentBlobs ) { const appSummary = this . context . createSummary ( ) ; const protocolSummary = this . captureProtocolSummary ( ) ; summary = combineAppAndProtocolSummary ( appSummary , protocolSummary ) ; this . _attachState = AttachState . Attaching ; this . context . notifyAttaching ( ) ; } const createNewResolvedUrl = await this . urlResolver . resolve ( request ) ; ensureFluidResolvedUrl ( createNewResolvedUrl ) ; if ( this . service === undefined ) { this . service = await runWithRetry ( async ( ) => this . serviceFactory . createContainer ( summary , createNewResolvedUrl , this . subLogger ) , "containerAttach" , this . mc . logger , { } ) ; } const resolvedUrl = this . service . resolvedUrl ; ensureFluidResolvedUrl ( resolvedUrl ) ; this . _resolvedUrl = resolvedUrl ; await this . connectStorageService ( ) ; if ( hasAttachmentBlobs ) { assert ( ! ! this . loader . services . detachedBlobStorage , 0x24e ) ; const redirectTable = new Map < string , string > ( ) ; while ( redirectTable . size < this . loader . services . detachedBlobStorage . size ) { const newIds = this . loader . services . detachedBlobStorage . getBlobIds ( ) . filter ( ( id ) => ! redirectTable . has ( id ) ) ; for ( const id of newIds ) { const blob = await this . loader . services . detachedBlobStorage . readBlob ( id ) ; const response = await this . storageService . createBlob ( blob ) ; redirectTable . set ( id , response . id ) ; } } const appSummary = this . context . createSummary ( redirectTable ) ; const protocolSummary = this . captureProtocolSummary ( ) ; summary = combineAppAndProtocolSummary ( appSummary , protocolSummary ) ; this . _attachState = AttachState . Attaching ; this . context . notifyAttaching ( ) ; await this . storageService . uploadSummaryWithContext ( summary , { referenceSequenceNumber : 0 , ackHandle : undefined , proposalHandle : undefined } ) ; } this . _attachState = AttachState . Attached ; this . emit ( "attached" ) ; this . propagateConnectionState ( ) ; if ( ! this . closed ) { this . resumeInternal ( { fetchOpsFromStorage : false , reason : "createDetached" } ) ; } } catch ( error ) { const newError = normalizeError ( error ) ; const resolvedUrl = this . resolvedUrl ; if ( isFluidResolvedUrl ( resolvedUrl ) ) { newError . addTelemetryProperties ( { resolvedUrl : resolvedUrl . url } ) ; } this . close ( newError ) ; throw newError ; } } , { start : true , end : true , cancel : "generic" } ) ; } public async request ( path ) { return PerformanceEvent . timedExecAsync ( this . mc . logger , { eventName : "Request" } , async ( ) => this . context . request ( path ) , { end : true , cancel : "error" } ) ; } public async snapshot ( tagMessage , fullTree = false ) { if ( ! this . protocolHandler . quorum . has ( "code" ) && ! this . protocolHandler . quorum . has ( "code2" ) ) { this . mc . logger . sendTelemetryEvent ( { eventName : "SkipSnapshot" } ) ; return ; } try { await this . deltaManager . inbound . pause ( ) ; await this . snapshotCore ( tagMessage , fullTree ) ; } catch ( ex ) { this . mc . logger . sendErrorEvent ( { eventName : "SnapshotExceptionError" } , ex ) ; throw ex ; } finally { this . deltaManager . inbound . resume ( ) ; } } public setAutoReconnect ( reconnect ) { if ( this . closed ) { throw new Error ( "Attempting to setAutoReconnect() a closed Container" ) ; } const mode = reconnect ? ReconnectMode . Enabled : ReconnectMode . Disabled ; const currentMode = this . _deltaManager . connectionManager . reconnectMode ; if ( currentMode === mode ) { return ; } const now = performance . now ( ) ; const duration = now - this . setAutoReconnectTime ; this . setAutoReconnectTime = now ; this . mc . logger . sendTelemetryEvent ( { eventName : reconnect ? "AutoReconnectEnabled" : "AutoReconnectDisabled" , connectionMode : this . connectionMode , connectionState : ConnectionState [ this . connectionState ] , duration } ) ; this . _deltaManager . connectionManager . setAutoReconnect ( mode ) ; if ( reconnect && this . _attachState === AttachState . Attached && this . resumedOpProcessingAfterLoad ) { if ( this . connectionState === ConnectionState . Disconnected ) { this . manualReconnectInProgress = true ; } this . connectToDeltaStream ( { reason : "autoReconnect" } ) ; } } public resume ( ) { if ( ! this . closed ) { this . resumeInternal ( { reason : "DocumentOpenResume" , fetchOpsFromStorage : false } ) ; } } private resumeInternal ( args ) { assert ( ! this . closed , 0x0d9 ) ; if ( ! this . resumedOpProcessingAfterLoad ) { this . resumedOpProcessingAfterLoad = true ; this . _deltaManager . inbound . resume ( ) ; this . _deltaManager . inboundSignal . resume ( ) ; } this . connectToDeltaStream ( args ) ; } param error public raiseContainerWarning ( warning ) { if ( warning . logged !== true ) { this . logContainerError ( warning ) ; } this . emit ( "warning" , warning ) ; } public async getAbsoluteUrl ( relativeUrl ) { if ( this . resolvedUrl === undefined ) { return undefined ; } return this . urlResolver . getAbsoluteUrl ( this . resolvedUrl , relativeUrl , this . _context ?. codeDetails ) ; } public async proposeCodeDetails ( codeDetails ) { if ( ! isFluidCodeDetails ( codeDetails ) ) { throw new Error ( "Provided codeDetails are not IFluidCodeDetails" ) ; } if ( this . codeLoader . IFluidCodeDetailsComparer ) { const comparison = await this . codeLoader . IFluidCodeDetailsComparer . compare ( codeDetails , this . getCodeDetailsFromQuorum ( ) ) ; if ( comparison !== undefined && comparison <= 0 ) { throw new Error ( "Proposed code details should be greater than the current" ) ; } } return this . protocolHandler . quorum . propose ( "code" , codeDetails ) . then ( ( ) => true ) . catch ( ( ) => false ) ; } private async processCodeProposal ( ) { const codeDetails = this . getCodeDetailsFromQuorum ( ) ; await Promise . all ( [ this . deltaManager . inbound . pause ( ) , this . deltaManager . inboundSignal . pause ( ) ] ) ; if ( ( await this . context . satisfies ( codeDetails ) === true ) ) { this . deltaManager . inbound . resume ( ) ; this . deltaManager . inboundSignal . resume ( ) ; return ; } this . close ( new GenericError ( "existingContextDoesNotSatisfyIncomingProposal" ) ) ; } private async snapshotCore ( tagMessage , fullTree = false ) { const root = this . snapshotBase ( ) ; const dataStoreEntries = await this . context . snapshot ( tagMessage , fullTree ) ; if ( dataStoreEntries !== null ) { root . entries . push ( ... dataStoreEntries . entries ) ; } const deltaDetails = `${ this . _deltaManager . lastSequenceNumber }:${ this . _deltaManager . minimumSequenceNumber }` ; const message = `Commit @${ deltaDetails } ${ tagMessage }` ; const lastVersion = await this . getVersion ( this . id ) ; const parents = lastVersion !== undefined ? [ lastVersion . id ] : [ ] ; return this . storageService . write ( root , parents , message , "" ) ; } private snapshotBase ( ) { const entries = [ ] ; const quorumSnapshot = this . protocolHandler . quorum . snapshot ( ) ; entries . push ( { mode : FileMode . File , path : "quorumMembers" , type : TreeEntry . Blob , value : { contents : JSON . stringify ( quorumSnapshot . members ) , encoding : "utf-8" } } ) ; entries . push ( { mode : FileMode . File , path : "quorumProposals" , type : TreeEntry . Blob , value : { contents : JSON . stringify ( quorumSnapshot . proposals ) , encoding : "utf-8" } } ) ; entries . push ( { mode : FileMode . File , path : "quorumValues" , type : TreeEntry . Blob , value : { contents : JSON . stringify ( quorumSnapshot . values ) , encoding : "utf-8" } } ) ; const documentAttributes = { branch : this . id , minimumSequenceNumber : this . _deltaManager . minimumSequenceNumber , sequenceNumber : this . _deltaManager . lastSequenceNumber , term : this . _deltaManager . referenceTerm } ; entries . push ( { mode : FileMode . File , path : ".attributes" , type : TreeEntry . Blob , value : { contents : JSON . stringify ( documentAttributes ) , encoding : "utf-8" } } ) ; const root = { entries } ; return root ; } private async getVersion ( version ) { const versions = await this . storageService . getVersions ( version , 1 ) ; return versions [ 0 ] ; } private recordConnectStartTime ( ) { if ( this . connectionTransitionTimes [ ConnectionState . Disconnected ] === undefined ) { this . connectionTransitionTimes [ ConnectionState . Disconnected ] = performance . now ( ) ; } } private connectToDeltaStream ( args ) { this . recordConnectStartTime ( ) ; if ( ! this . _canReconnect || ! this . client . details . capabilities . interactive ) { args . mode = "write" ; } this . _deltaManager . connect ( args ) ; } param specifiedVersion private async load ( specifiedVersion , loadMode , pendingLocalState ? ) { if ( this . _resolvedUrl === undefined ) { throw new Error ( "Attempting to load without a resolved url" ) ; } this . service = await this . serviceFactory . createDocumentService ( this . _resolvedUrl , this . subLogger ) ; const connectionArgs = { reason : "DocumentOpen" , mode : "write" , fetchOpsFromStorage : false } ; if ( loadMode . deltaConnection === undefined ) { this . connectToDeltaStream ( connectionArgs ) ; } await this . connectStorageService ( ) ; this . _attachState = AttachState . Attached ; const { snapshot , versionId } = await this . fetchSnapshotTree ( specifiedVersion ) ; assert ( snapshot !== undefined , 0x237 ) ; const attributes = await this . getDocumentAttributes ( this . storageService , snapshot ) ; let opsBeforeReturnP ; switch ( loadMode . opsBeforeReturn ) { case undefined : this . attachDeltaManagerOpHandler ( attributes , loadMode . deltaConnection !== "none" ? "all" : "none" ) ; break ; case "cached" : opsBeforeReturnP = this . attachDeltaManagerOpHandler ( attributes , "cached" ) ; break ; case "all" : opsBeforeReturnP = this . attachDeltaManagerOpHandler ( attributes , "all" ) ; break ; default : unreachableCase ( loadMode . opsBeforeReturn ) ; } this . _protocolHandler = await this . loadAndInitializeProtocolState ( attributes , this . storageService , snapshot ) ; const codeDetails = this . getCodeDetailsFromQuorum ( ) ; await this . instantiateContext ( true , codeDetails , snapshot , pendingLocalState ) ; this . propagateConnectionState ( ) ; this . loaded = true ; if ( ! this . closed ) { if ( opsBeforeReturnP !== undefined ) { this . _deltaManager . inbound . resume ( ) ; await opsBeforeReturnP ; await this . _deltaManager . inbound . waitTillProcessingDone ( ) ; this . _deltaManager . inbound . pause ( ) ; } switch ( loadMode . deltaConnection ) { case undefined : this . resume ( ) ; break ; case "delayed" : this . resumedOpProcessingAfterLoad = true ; this . _deltaManager . inbound . resume ( ) ; this . _deltaManager . inboundSignal . resume ( ) ; break ; case "none" : break ; default : unreachableCase ( loadMode . deltaConnection ) ; } } if ( this . closed ) { throw new Error ( "Container was closed while load()" ) ; } return { sequenceNumber : attributes . sequenceNumber , version : versionId } ; } private async createDetached ( source ) { const attributes = { sequenceNumber : detachedContainerRefSeqNumber , term : 1 , minimumSequenceNumber : 0 } ; const committedCodeProposal = { key : "code" , value : source , approvalSequenceNumber : 0 , commitSequenceNumber : 0 , sequenceNumber : 0 } ; const members = [ ] ; const proposals = [ ] ; const values = [ [ "code" , committedCodeProposal ] ] ; await this . attachDeltaManagerOpHandler ( attributes ) ; this . _protocolHandler = await this . initializeProtocolState ( attributes , members , proposals , values ) ; await this . instantiateContextDetached ( false ) ; this . propagateConnectionState ( ) ; this . loaded = true ; } private async rehydrateDetachedFromSnapshot ( detachedContainerSnapshot ) { if ( detachedContainerSnapshot . tree [ ".hasAttachmentBlobs" ] !== undefined ) { assert ( ! ! this . loader . services . detachedBlobStorage && this . loader . services . detachedBlobStorage . size > 0 , 0x250 ) ; delete detachedContainerSnapshot . tree [ ".hasAttachmentBlobs" ] ; } const snapshotTree = getSnapshotTreeFromSerializedContainer ( detachedContainerSnapshot ) ; this . _storage . loadSnapshotForRehydratingContainer ( snapshotTree ) ; const attributes = await this . getDocumentAttributes ( this . _storage , snapshotTree ) ; assert ( attributes . sequenceNumber === 0 , 0x0db ) ; await this . attachDeltaManagerOpHandler ( attributes ) ; this . _protocolHandler = await this . loadAndInitializeProtocolState ( attributes , this . _storage , snapshotTree ) ; await this . instantiateContextDetached ( true , snapshotTree ) ; this . loaded = true ; this . propagateConnectionState ( ) ; } private async connectStorageService ( ) { if ( this . _storageService !== undefined ) { return ; } assert ( this . service !== undefined , 0x1ef ) ; const storageService = await this . service . connectToStorage ( ) ; this . _storageService = new RetriableDocumentStorageService ( storageService , this . mc . logger ) ; if ( this . options . summarizeProtocolTree === true ) { this . mc . logger . sendTelemetryEvent ( { eventName : "summarizeProtocolTreeEnabled" } ) ; this . _storageService = new ProtocolTreeStorageService ( this . _storageService , ( ) => this . captureProtocolSummary ( ) ) ; } assert ( storageService . policies ?. minBlobSize === this . storageService . policies ?. minBlobSize , 0x0e0 ) ; } private async getDocumentAttributes ( storage , tree ) { if ( tree === undefined ) { return { minimumSequenceNumber : 0 , sequenceNumber : 0 , term : 1 } ; } const attributesHash = ".protocol" in tree . trees ? tree . trees [ ".protocol" ] . blobs . attributes : tree . blobs [ ".attributes" ] ; const attributes = await readAndParse < IDocumentAttributes > ( storage , attributesHash ) ; if ( attributes . term === undefined ) { attributes . term = 1 ; } return attributes ; } private async loadAndInitializeProtocolState ( attributes , storage , snapshot ) { let members = [ ] ; let proposals = [ ] ; let values = [ ] ; if ( snapshot !== undefined ) { const baseTree = ".protocol" in snapshot . trees ? snapshot . trees [ ".protocol" ] : snapshot ; [ members , proposals , values ] = await Promise . all ( [ readAndParse < [ string , ISequencedClient ] [ ] > ( storage , baseTree . blobs . quorumMembers ) , readAndParse < [ number , ISequencedProposal , string [ ] ] [ ] > ( storage , baseTree . blobs . quorumProposals ) , readAndParse < [ string , ICommittedProposal ] [ ] > ( storage , baseTree . blobs . quorumValues ) , ] ) ; } const protocolHandler = await this . initializeProtocolState ( attributes , members , proposals , values ) ; return protocolHandler ; } private async initializeProtocolState ( attributes , members , proposals , values ) { const protocol = new ProtocolOpHandler ( attributes . minimumSequenceNumber , attributes . sequenceNumber , attributes . term , members , proposals , values , ( key , value ) => this . submitMessage ( MessageType . Propose , { key , value } ) , ( sequenceNumber ) => this . submitMessage ( MessageType . Reject , sequenceNumber ) ) ; const protocolLogger = ChildLogger . create ( this . subLogger , "ProtocolHandler" ) ; protocol . quorum . on ( "error" , ( error ) => { protocolLogger . sendErrorEvent ( error ) ; } ) ; protocol . quorum . on ( "addMember" , ( clientId , details ) => { this . connectionStateHandler . receivedAddMemberEvent ( clientId ) ; } ) ; protocol . quorum . on ( "removeMember" , ( clientId ) => { this . connectionStateHandler . receivedRemoveMemberEvent ( clientId ) ; } ) ; protocol . quorum . on ( "addProposal" , ( proposal ) => { if ( proposal . key === "code" || proposal . key === "code2" ) { this . emit ( "codeDetailsProposed" , proposal . value , proposal ) ; } } ) ; protocol . quorum . on ( "approveProposal" , ( sequenceNumber , key , value ) => { if ( key === "code" || key === "code2" ) { if ( ! isFluidCodeDetails ( value ) ) { this . mc . logger . sendErrorEvent ( { eventName : "CodeProposalNotIFluidCodeDetails" } ) ; } this . processCodeProposal ( ) . catch ( ( error ) => { this . close ( normalizeError ( error ) ) ; throw error ; } ) ; } } ) ; return protocol ; } private captureProtocolSummary ( ) { const quorumSnapshot = this . protocolHandler . quorum . snapshot ( ) ; const documentAttributes = { minimumSequenceNumber : this . protocolHandler . minimumSequenceNumber , sequenceNumber : this . protocolHandler . sequenceNumber , term : this . protocolHandler . term } ; const summary = { tree : { attributes : { content : JSON . stringify ( documentAttributes ) , type : SummaryType . Blob } , quorumMembers : { content : JSON . stringify ( quorumSnapshot . members ) , type : SummaryType . Blob } , quorumProposals : { content : JSON . stringify ( quorumSnapshot . proposals ) , type : SummaryType . Blob } , quorumValues : { content : JSON . stringify ( quorumSnapshot . values ) , type : SummaryType . Blob } } , type : SummaryType . Tree } ; return summary ; } private getCodeDetailsFromQuorum ( ) { const quorum = this . protocolHandler . quorum ; const pkg = getCodeProposal ( quorum ) ; return pkg as IFluidCodeDetails ; } private get client ( ) : IClient { const client = this . options ?. client !== undefined ? ( this . options . client as IClient ) : { details : { capabilities : { interactive : true } } , mode : "read" , permission : [ ] , scopes : [ ] , user : { id : "" } } ; if ( this . clientDetailsOverride !== undefined ) { merge ( client . details , this . clientDetailsOverride ) ; } client . details . environment = [ client . details . environment , ` loaderVersion:${ pkgVersion }` ] . join ( ";" ) ; return client ; } /**
     * Returns true if connection is active, i.e. it's "write" connection and
     * container runtime was notified about this connection (i.e. we are up-to-date and could send ops).
     * This happens after client received its own joinOp and thus is in the quorum.
     * If it's not true, runtime is not in position to send ops.
     */ private activeConnection ( ) { return this . connectionState === ConnectionState . Connected && this . connectionMode === "write" ; } private createDeltaManager ( ) { const serviceProvider = ( ) => this . service ; const deltaManager = new DeltaManager < ConnectionManager > ( serviceProvider , ChildLogger . create ( this . subLogger , "DeltaManager" ) , ( ) => this . activeConnection ( ) , ( props ) => new ConnectionManager ( serviceProvider , this . client , this . _canReconnect , ChildLogger . create ( this . subLogger , "ConnectionManager" ) , props ) ) ; deltaManager . inbound . pause ( ) ; deltaManager . inboundSignal . pause ( ) ; deltaManager . on ( "connect" , ( details , opsBehind ? ) => { this . _audience . clear ( ) ; for ( const priorClient of details . initialClients ?? [ ] ) { this . _audience . addMember ( priorClient . clientId , priorClient . client ) ; } this . connectionStateHandler . receivedConnectEvent ( this . connectionMode , details ) ; } ) ; deltaManager . on ( "disconnect" , ( reason ) => { this . manualReconnectInProgress = false ; this . collabWindowTracker . stopSequenceNumberUpdate ( ) ; this . connectionStateHandler . receivedDisconnectEvent ( reason ) ; } ) ; deltaManager . on ( "throttled" , ( warning ) => { this . raiseContainerWarning ( warning ) ; } ) ; deltaManager . on ( "readonly" , ( readonly ) => { this . emit ( "readonly" , readonly ) ; } ) ; deltaManager . on ( "closed" , ( error ? ) => { this . close ( error ) ; } ) ; return deltaManager ; } private async attachDeltaManagerOpHandler ( attributes , prefetchType ? ) { return this . _deltaManager . attachOpHandler ( attributes . minimumSequenceNumber , attributes . sequenceNumber , attributes . term ?? 1 , { process : ( message ) => this . processRemoteMessage ( message ) , processSignal : ( message ) => { this . processSignal ( message ) ; } } , prefetchType ) ; } private logConnectionStateChangeTelemetry ( value , oldState , reason ? ) { const time = performance . now ( ) ; this . connectionTransitionTimes [ value ] = time ; const duration = time - this . connectionTransitionTimes [ oldState ] ; let durationFromDisconnected ; let connectionInitiationReason ; let autoReconnect ; let checkpointSequenceNumber ; let opsBehind ; if ( value === ConnectionState . Disconnected ) { autoReconnect = this . _deltaManager . connectionManager . reconnectMode ; } else { if ( value === ConnectionState . Connected ) { durationFromDisconnected = time - this . connectionTransitionTimes [ ConnectionState . Disconnected ] ; durationFromDisconnected = TelemetryLogger . formatTick ( durationFromDisconnected ) ; } else { checkpointSequenceNumber = this . deltaManager . lastKnownSeqNumber ; if ( this . deltaManager . hasCheckpointSequenceNumber ) { opsBehind = checkpointSequenceNumber - this . deltaManager . lastSequenceNumber ; } } if ( this . firstConnection ) { connectionInitiationReason = "InitialConnect" ; } else if ( this . manualReconnectInProgress ) { connectionInitiationReason = "ManualReconnect" ; } else { connectionInitiationReason = "AutoReconnect" ; } } this . mc . logger . sendPerformanceEvent ( { eventName : `ConnectionStateChange_${ ConnectionState [ value ] }` , from : ConnectionState [ oldState ] , duration , durationFromDisconnected , reason , connectionInitiationReason , pendingClientId : this . connectionStateHandler . pendingClientId , clientId : this . clientId , autoReconnect , opsBehind , online : OnlineStatus [ isOnline ( ) ] , lastVisible : this . lastVisible !== undefined ? performance . now ( ) - this . lastVisible : undefined , checkpointSequenceNumber , ... this . _deltaManager . connectionProps } ) ; if ( value === ConnectionState . Connected ) { this . firstConnection = false ; this . manualReconnectInProgress = false ; } } private propagateConnectionState ( ) { const logOpsOnReconnect = this . connectionState === ConnectionState . Connected && ! this . firstConnection && this . connectionMode === "write" ; if ( logOpsOnReconnect ) { this . messageCountAfterDisconnection = 0 ; } const state = this . connectionState === ConnectionState . Connected ; if ( ! this . context . disposed ) { this . context . setConnectionState ( state , this . clientId ) ; } assert ( this . protocolHandler !== undefined , 0x0dc ) ; this . protocolHandler . quorum . setConnectionState ( state , this . clientId ) ; raiseConnectedEvent ( this . mc . logger , this , state , this . clientId ) ; if ( logOpsOnReconnect ) { this . mc . logger . sendTelemetryEvent ( { eventName : "OpsSentOnReconnect" , count : this . messageCountAfterDisconnection } ) ; } } private submitContainerMessage ( type , contents , batch ? , metadata ? ) { const outboundMessageType = type ; switch ( outboundMessageType ) { case MessageType . Operation : case MessageType . RemoteHelp : break ; case MessageType . Summarize : { const summary = contents as ISummaryContent ; if ( summary . details === undefined ) { summary . details = { } ; } summary . details . includesProtocolTree = this . options . summarizeProtocolTree === true ; break ; } default : this . close ( new GenericError ( "invalidContainerSubmitOpType" , undefined , { messageType : type } ) ) ; return - 1 ; } return this . submitMessage ( type , contents , batch , metadata ) ; } private submitMessage ( type , contents , batch ? , metadata ? ) { if ( this . connectionState !== ConnectionState . Connected ) { this . mc . logger . sendErrorEvent ( { eventName : "SubmitMessageWithNoConnection" , type } ) ; return - 1 ; } this . messageCountAfterDisconnection += 1 ; this . collabWindowTracker . stopSequenceNumberUpdate ( ) ; return this . _deltaManager . submit ( type , contents , batch , metadata ) ; } private processRemoteMessage ( message ) { if ( message . clientId != null ) { let errorCode ; const client = this . getQuorum ( ) . getMember ( message . clientId ) ; if ( client === undefined && message . type !== MessageType . ClientJoin ) { errorCode = "messageClientIdMissingFromQuorum" ; } else if ( client ?. shouldHaveLeft === true && message . type !== MessageType . NoOp ) { errorCode = "messageClientIdShouldHaveLeft" ; } if ( errorCode !== undefined ) { const error = new DataCorruptionError ( errorCode , extractSafePropertiesFromMessage ( message ) ) ; this . close ( normalizeError ( error ) ) ; } } const local = this . clientId === message . clientId ; if ( ! isSystemMessage ( message ) ) { this . context . process ( message , local , undefined ) ; } const result = this . protocolHandler . processMessage ( message , local ) ; this . collabWindowTracker . scheduleSequenceNumberUpdate ( message , result . immediateNoOp === true ) ; this . emit ( "op" , message ) ; return result ; } private submitSignal ( message ) { this . _deltaManager . submitSignal ( JSON . stringify ( message ) ) ; } private processSignal ( message ) { if ( message . clientId === null ) { const innerContent = message . content as { content : any ; type : string ; } ; if ( innerContent . type === MessageType . ClientJoin ) { const newClient = innerContent . content as ISignalClient ; this . _audience . addMember ( newClient . clientId , newClient . client ) ; } else if ( innerContent . type === MessageType . ClientLeave ) { const leftClientId = innerContent . content as string ; this . _audience . removeMember ( leftClientId ) ; } } else { const local = this . clientId === message . clientId ; this . context . processSignal ( message , local ) ; } } param specifiedVersion returns private async fetchSnapshotTree ( specifiedVersion ) { const version = await this . getVersion ( specifiedVersion ?? this . id ) ; if ( version === undefined && specifiedVersion !== undefined ) { this . mc . logger . sendErrorEvent ( { eventName : "NoVersionFoundWhenSpecified" , id : specifiedVersion } ) ; } this . _loadedFromVersion = version ; const snapshot = await this . storageService . getSnapshotTree ( version ) ?? undefined ; if ( snapshot === undefined && version !== undefined ) { this . mc . logger . sendErrorEvent ( { eventName : "getSnapshotTreeFailed" , id : version . id } ) ; } return { snapshot , versionId : version ?. id } ; } private async instantiateContextDetached ( existing , snapshot ? , pendingLocalState ? ) { const codeDetails = this . getCodeDetailsFromQuorum ( ) ; if ( codeDetails === undefined ) { throw new Error ( "pkg should be provided in create flow!!" ) ; } await this . instantiateContext ( existing , codeDetails , snapshot , pendingLocalState ) ; } private async instantiateContext ( existing , codeDetails , snapshot ? , pendingLocalState ? ) { assert ( this . _context ?. disposed !== false , 0x0dd ) ; if ( this . _dirtyContainer ) { this . mc . logger . sendErrorEvent ( { eventName : "DirtyContainerReloadContainer" } ) ; } const loader = new RelativeLoader ( this , this . loader ) ; this . _context = await ContainerContext . createOrLoad ( this , this . scope , this . codeLoader , codeDetails , snapshot , new DeltaManagerProxy ( this . _deltaManager ) , new QuorumProxy ( this . protocolHandler . quorum ) , loader , ( warning ) => this . raiseContainerWarning ( warning ) , ( type , contents , batch , metadata ) => this . submitContainerMessage ( type , contents , batch , metadata ) , ( message ) => this . submitSignal ( message ) , ( error ? ) => this . close ( error ) , Container . version , ( dirty ) => { this . _dirtyContainer = dirty ; this . emit ( dirty ? dirtyContainerEvent : savedContainerEvent ) ; } , existing , pendingLocalState ) ; this . emit ( "contextChanged" , codeDetails ) ; } private logContainerError ( warning ) { this . mc . logger . sendErrorEvent ( { eventName : "ContainerWarning" } , warning ) ; } }