import { ITelemetryLogger } from "@fluidframework/common-definitions" ; import { IFluidHandle , IFluidHandleContext , IRequest , IResponse , } from "@fluidframework/core-interfaces" ; import { IAudience , IDeltaManager , ContainerWarning , BindState , AttachState , ILoaderOptions , } from "@fluidframework/container-definitions" ; import { CreateProcessingError } from "@fluidframework/container-utils" ; import { assert , Deferred , LazyPromise , TypedEventEmitter , unreachableCase , } from "@fluidframework/common-utils" ; import { ChildLogger , raiseConnectedEvent , } from "@fluidframework/telemetry-utils" ; import { buildSnapshotTree } from "@fluidframework/driver-utils" ; import { IClientDetails , IDocumentMessage , ISequencedDocumentMessage , SummaryType , ISummaryBlob , ISummaryTree , IQuorumClients , } from "@fluidframework/protocol-definitions" ; import { CreateSummarizerNodeSource , IAttachMessage , IEnvelope , IFluidDataStoreContext , IFluidDataStoreChannel , IGarbageCollectionData , IGarbageCollectionDetailsBase , IInboundSignalMessage , ISummaryTreeWithStats , } from "@fluidframework/runtime-definitions" ; import { convertSnapshotTreeToSummaryTree , convertSummaryTreeToITree , generateHandleContextPath , RequestParser , SummaryTreeBuilder , create404Response , createResponseError , exceptionToResponse , } from "@fluidframework/runtime-utils" ; import { IChannel , IFluidDataStoreRuntime , IFluidDataStoreRuntimeEvents , IChannelFactory , } from "@fluidframework/datastore-definitions" ; import { GCDataBuilder , unpackChildNodesGCDetails , unpackChildNodesUsedRoutes , } from "@fluidframework/garbage-collector" ; import { v4 as uuid } from "uuid" ; import { IChannelContext , summarizeChannel } from "./channelContext" ; import { LocalChannelContext , LocalChannelContextBase , RehydratedLocalChannelContext } from "./localChannelContext" ; import { RemoteChannelContext } from "./remoteChannelContext" ; export enum DataStoreMessageType { Attach = "attach" , ChannelOp = "op" } export interface ISharedObjectRegistry { get ( name ) : IChannelFactory | undefined ; } /**
 * Base data store class
 */ export class FluidDataStoreRuntime extends TypedEventEmitter < IFluidDataStoreRuntimeEvents > implements IFluidDataStoreChannel , IFluidDataStoreRuntime , IFluidHandleContext { param context param sharedObjectRegistry param existing public static load ( context , sharedObjectRegistry , existing ) { return new FluidDataStoreRuntime ( context , sharedObjectRegistry , existing ) ; } public get IFluidRouter ( ) { return this ; } public get connected ( ) : boolean { return this . dataStoreContext . connected ; } public get clientId ( ) : string | undefined { return this . dataStoreContext . clientId ; } public get clientDetails ( ) : IClientDetails { return this . dataStoreContext . clientDetails ; } public get isAttached ( ) : boolean { return this . attachState !== AttachState . Detached ; } public get attachState ( ) : AttachState { return this . _attachState ; } public get absolutePath ( ) : string { return generateHandleContextPath ( this . id , this . routeContext ) ; } public get routeContext ( ) : IFluidHandleContext { return this . dataStoreContext . IFluidHandleContext ; } public get IFluidHandleContext ( ) { return this ; } public get rootRoutingContext ( ) { return this ; } public get channelsRoutingContext ( ) { return this ; } public get objectsRoutingContext ( ) { return this ; } private _disposed = false ; public get disposed ( ) { return this . _disposed ; } private readonly contexts = new Map < string , IChannelContext > ( ) ; private readonly contextsDeferred = new Map < string , Deferred < IChannelContext > > ( ) ; private readonly pendingAttach = new Map < string , IAttachMessage > ( ) ; private bindState : BindState ; private graphAttachState : AttachState = AttachState . Detached ; private readonly deferredAttached = new Deferred < void > ( ) ; private readonly localChannelContextQueue = new Map < string , LocalChannelContextBase > ( ) ; private readonly notBoundedChannelContextSet = new Set < string > ( ) ; private boundhandles : Set < IFluidHandle > | undefined ; private _attachState : AttachState ; public readonly id : string ; public readonly options : ILoaderOptions ; public readonly deltaManager : IDeltaManager < ISequencedDocumentMessage , IDocumentMessage > ; private readonly quorum : IQuorumClients ; private readonly audience : IAudience ; public readonly logger : ITelemetryLogger ; private readonly channelsBaseGCDetails : LazyPromise < Map < string , IGarbageCollectionDetailsBase > > ; public constructor ( private readonly dataStoreContext , private readonly sharedObjectRegistry , existing ) { super ( ) ; this . logger = ChildLogger . create ( dataStoreContext . logger , "FluidDataStoreRuntime" , { all : { dataStoreId : uuid ( ) } } ) ; this . id = dataStoreContext . id ; this . options = dataStoreContext . options ; this . deltaManager = dataStoreContext . deltaManager ; this . quorum = dataStoreContext . getQuorum ( ) ; this . audience = dataStoreContext . getAudience ( ) ; const tree = dataStoreContext . baseSnapshot ; this . channelsBaseGCDetails = new LazyPromise ( async ( ) => { const baseGCDetails = await ( this . dataStoreContext . getBaseGCDetails ?. ( ) ?? this . dataStoreContext . getInitialGCSummaryDetails ( ) ) ; return unpackChildNodesGCDetails ( baseGCDetails ) ; } ) ; if ( tree ?. trees !== undefined ) { Object . keys ( tree . trees ) . forEach ( ( path ) => { if ( path === "_search" ) { return ; } let channelContext ; if ( dataStoreContext . isLocalDataStore ) { channelContext = new RehydratedLocalChannelContext ( path , this . sharedObjectRegistry , this , this . dataStoreContext , this . dataStoreContext . storage , ( content , localOpMetadata ) => this . submitChannelOp ( path , content , localOpMetadata ) , ( address ) => this . setChannelDirty ( address ) , ( srcHandle , outboundHandle ) => this . addedGCOutboundReference ( srcHandle , outboundHandle ) , tree . trees [ path ] ) ; if ( dataStoreContext . attachState !== AttachState . Detached ) { ( channelContext as LocalChannelContextBase ) . markAttached ( ) ; } else { this . localChannelContextQueue . set ( path , channelContext as LocalChannelContextBase ) ; } } else { channelContext = new RemoteChannelContext ( this , dataStoreContext , dataStoreContext . storage , ( content , localOpMetadata ) => this . submitChannelOp ( path , content , localOpMetadata ) , ( address ) => this . setChannelDirty ( address ) , ( srcHandle , outboundHandle ) => this . addedGCOutboundReference ( srcHandle , outboundHandle ) , path , tree . trees [ path ] , this . sharedObjectRegistry , undefined , this . dataStoreContext . getCreateChildSummarizerNodeFn ( path , { type : CreateSummarizerNodeSource . FromSummary } ) , async ( ) => this . getChannelBaseGCDetails ( path ) ) ; } const deferred = new Deferred < IChannelContext > ( ) ; deferred . resolve ( channelContext ) ; this . contexts . set ( path , channelContext ) ; this . contextsDeferred . set ( path , deferred ) ; } ) ; } this . attachListener ( ) ; this . bindState = existing ? BindState . Bound : BindState . NotBound ; this . _attachState = dataStoreContext . attachState ; if ( existing ) { this . deferredAttached . resolve ( ) ; } } public dispose ( ) { if ( this . _disposed ) { return ; } this . _disposed = true ; this . emit ( "dispose" ) ; this . removeAllListeners ( ) ; } public async resolveHandle ( request ) { return this . request ( request ) ; } public async request ( request ) { try { const parser = RequestParser . create ( request ) ; const id = parser . pathParts [ 0 ] ; if ( id === "_channels" || id === "_custom" ) { return this . request ( parser . createSubRequest ( 1 ) ) ; } if ( this . contextsDeferred . has ( id ) && parser . isLeaf ( 1 ) ) { try { const value = await this . contextsDeferred . get ( id ) ! . promise ; const channel = await value . getChannel ( ) ; return { mimeType : "fluid/object" , status : 200 , value : channel } ; } catch ( error ) { this . logger . sendErrorEvent ( { eventName : "GetChannelFailedInRequest" } , error ) ; return createResponseError ( 500 , `Failed to get Channel: ${ error }` , request ) ; } } return create404Response ( request ) ; } catch ( error ) { return exceptionToResponse ( error ) ; } } public async getChannel ( id ) { this . verifyNotClosed ( ) ; if ( ! this . contextsDeferred . has ( id ) ) { this . contextsDeferred . set ( id , new Deferred < IChannelContext > ( ) ) ; } const context = await this . contextsDeferred . get ( id ) ! . promise ; const channel = await context . getChannel ( ) ; return channel ; } public createChannel ( id = uuid ( ) , type ) { this . verifyNotClosed ( ) ; assert ( ! this . contexts . has ( id ) , 0x179 ) ; this . notBoundedChannelContextSet . add ( id ) ; const context = new LocalChannelContext ( id , this . sharedObjectRegistry , type , this , this . dataStoreContext , this . dataStoreContext . storage , ( content , localOpMetadata ) => this . submitChannelOp ( id , content , localOpMetadata ) , ( address ) => this . setChannelDirty ( address ) , ( srcHandle , outboundHandle ) => this . addedGCOutboundReference ( srcHandle , outboundHandle ) ) ; this . contexts . set ( id , context ) ; if ( this . contextsDeferred . has ( id ) ) { this . contextsDeferred . get ( id ) ! . resolve ( context ) ; } else { const deferred = new Deferred < IChannelContext > ( ) ; deferred . resolve ( context ) ; this . contextsDeferred . set ( id , deferred ) ; } assert ( ! ! context . channel , 0x17a ) ; return context . channel ; } param channel public bindChannel ( channel ) { assert ( this . notBoundedChannelContextSet . has ( channel . id ) , 0x17b ) ; this . notBoundedChannelContextSet . delete ( channel . id ) ; if ( this . isAttached ) { this . attachChannel ( channel ) ; return ; } else { this . bind ( channel . handle ) ; if ( ! this . localChannelContextQueue . has ( channel . id ) ) { this . localChannelContextQueue . set ( channel . id , this . contexts . get ( channel . id ) as LocalChannelContextBase ) ; } } } public attachGraph ( ) { if ( this . graphAttachState !== AttachState . Detached ) { return ; } this . graphAttachState = AttachState . Attaching ; if ( this . boundhandles !== undefined ) { this . boundhandles . forEach ( ( handle ) => { handle . attachGraph ( ) ; } ) ; this . boundhandles = undefined ; } this . localChannelContextQueue . forEach ( ( channel ) => { channel . markAttached ( ) ; } ) ; this . localChannelContextQueue . clear ( ) ; this . bindToContext ( ) ; this . graphAttachState = AttachState . Attached ; } /**
     * Binds this runtime to the container
     * This includes the following:
     * 1. Sending an Attach op that includes all existing state
     * 2. Attaching the graph if the data store becomes attached.
     */ public bindToContext ( ) { if ( this . bindState !== BindState . NotBound ) { return ; } this . bindState = BindState . Binding ; this . dataStoreContext . bindToContext ( ) ; this . bindState = BindState . Bound ; } public bind ( handle ) { if ( this . isAttached || this . graphAttachState !== AttachState . Detached ) { handle . attachGraph ( ) ; return ; } if ( this . boundhandles === undefined ) { this . boundhandles = new Set < IFluidHandle > ( ) ; } this . boundhandles . add ( handle ) ; } public setConnectionState ( connected , clientId ? ) { this . verifyNotClosed ( ) ; for ( const [ , object ] of this . contexts ) { object . setConnectionState ( connected , clientId ) ; } raiseConnectedEvent ( this . logger , this , connected , clientId ) ; } public getQuorum ( ) { return this . quorum ; } public getAudience ( ) { return this . audience ; } public async uploadBlob ( blob ) { this . verifyNotClosed ( ) ; return this . dataStoreContext . uploadBlob ( blob ) ; } public process ( message , local , localOpMetadata ) { this . verifyNotClosed ( ) ; try { switch ( message . type ) { case DataStoreMessageType . Attach : { const attachMessage = message . contents as IAttachMessage ; const id = attachMessage . id ; if ( local ) { assert ( this . pendingAttach . has ( id ) , 0x17c ) ; this . pendingAttach . delete ( id ) ; } else { assert ( ! this . contexts . has ( id ) , 0x17d ) ; const flatBlobs = new Map < string , ArrayBufferLike > ( ) ; const snapshotTree = buildSnapshotTree ( attachMessage . snapshot . entries , flatBlobs ) ; const remoteChannelContext = new RemoteChannelContext ( this , this . dataStoreContext , this . dataStoreContext . storage , ( content , localContentMetadata ) => this . submitChannelOp ( id , content , localContentMetadata ) , ( address ) => this . setChannelDirty ( address ) , ( srcHandle , outboundHandle ) => this . addedGCOutboundReference ( srcHandle , outboundHandle ) , id , snapshotTree , this . sharedObjectRegistry , flatBlobs , this . dataStoreContext . getCreateChildSummarizerNodeFn ( id , { type : CreateSummarizerNodeSource . FromAttach , sequenceNumber : message . sequenceNumber , snapshot : attachMessage . snapshot } ) , async ( ) => this . getChannelBaseGCDetails ( id ) , attachMessage . type ) ; this . contexts . set ( id , remoteChannelContext ) ; if ( this . contextsDeferred . has ( id ) ) { this . contextsDeferred . get ( id ) ! . resolve ( remoteChannelContext ) ; } else { const deferred = new Deferred < IChannelContext > ( ) ; deferred . resolve ( remoteChannelContext ) ; this . contextsDeferred . set ( id , deferred ) ; } } break ; } case DataStoreMessageType . ChannelOp : this . processChannelOp ( message , local , localOpMetadata ) ; break ; default : } this . emit ( "op" , message ) ; } catch ( error ) { throw CreateProcessingError ( error , "fluidDataStoreRuntimeFailedToProcessMessage" , message ) ; } } public processSignal ( message , local ) { this . emit ( "signal" , message , local ) ; } private isChannelAttached ( id ) { return ( ! this . notBoundedChannelContextSet . has ( id ) && ! this . localChannelContextQueue . has ( id ) && ! this . pendingAttach . has ( id ) ) ; } /**
     * Returns the outbound routes of this channel. Currently, all contexts in this channel are considered
     * referenced and are hence outbound. This will change when we have root and non-root channel contexts.
     * The only root contexts will be considered as referenced.
     */ private getOutboundRoutes ( ) { const outboundRoutes = [ ] ; for ( const [ contextId ] of this . contexts ) { outboundRoutes . push ( `${ this . absolutePath }/${ contextId }` ) ; } return outboundRoutes ; } param builder private updateGCNodes ( builder ) { builder . addRouteToAllNodes ( this . absolutePath ) ; builder . addNode ( "/" , this . getOutboundRoutes ( ) ) ; } param fullGC public async getGCData ( fullGC = false ) { const builder = new GCDataBuilder ( ) ; await Promise . all ( Array . from ( this . contexts ) . filter ( ( [ contextId , _ ] ) => { return this . isChannelAttached ( contextId ) ; } ) . map ( async ( [ contextId , context ] ) => { const contextGCData = await context . getGCData ( fullGC ) ; builder . prefixAndAddNodes ( contextId , contextGCData . gcNodes ) ; } ) ) ; this . updateGCNodes ( builder ) ; return builder . getGCData ( ) ; } param usedRoutes param gcTimestamp public updateUsedRoutes ( usedRoutes , gcTimestamp ? ) { const usedContextRoutes = unpackChildNodesUsedRoutes ( usedRoutes ) ; for ( const [ id ] of usedContextRoutes ) { assert ( this . contexts . has ( id ) , 0x17e ) ; } for ( const [ contextId , context ] of this . contexts ) { context . updateUsedRoutes ( usedContextRoutes . get ( contextId ) ?? [ ] , gcTimestamp ) ; } } param srcHandle param outboundHandle private addedGCOutboundReference ( srcHandle , outboundHandle ) { this . dataStoreContext . addedGCOutboundReference ?. ( srcHandle , outboundHandle ) ; } param channelId returns private async getChannelBaseGCDetails ( channelId ) { let channelBaseGCDetails = ( await this . channelsBaseGCDetails ) . get ( channelId ) ; if ( channelBaseGCDetails === undefined ) { channelBaseGCDetails = { } ; } if ( channelBaseGCDetails . usedRoutes === undefined || channelBaseGCDetails . usedRoutes . length === 0 ) { channelBaseGCDetails . usedRoutes = [ "" ] ; } return channelBaseGCDetails ; } param fullTree param trackState public async summarize ( fullTree = false , trackState = true ) { const summaryBuilder = new SummaryTreeBuilder ( ) ; await Promise . all ( Array . from ( this . contexts ) . filter ( ( [ contextId , _ ] ) => { const isAttached = this . isChannelAttached ( contextId ) ; assert ( isAttached , 0x17f ) ; return isAttached ; } ) . map ( async ( [ contextId , context ] ) => { const contextSummary = await context . summarize ( true , trackState ) ; summaryBuilder . addWithStats ( contextId , contextSummary ) ; } ) ) ; return summaryBuilder . getSummaryTree ( ) ; } public getAttachSummary ( ) { this . attachGraph ( ) ; const summaryBuilder = new SummaryTreeBuilder ( ) ; for ( const [ contextId , context ] of this . contexts ) { if ( ! ( context instanceof LocalChannelContextBase ) ) { throw new Error ( "Should only be called with local channel handles" ) ; } if ( ! this . notBoundedChannelContextSet . has ( contextId ) ) { let summaryTree ; if ( context . isLoaded ) { const contextSummary = context . getAttachSummary ( ) ; assert ( contextSummary . summary . type === SummaryType . Tree , 0x180 ) ; summaryTree = { stats : contextSummary . stats , summary : contextSummary . summary } ; } else { assert ( ! ! this . dataStoreContext . baseSnapshot , 0x181 ) ; summaryTree = convertSnapshotTreeToSummaryTree ( this . dataStoreContext . baseSnapshot . trees [ contextId ] ) ; } summaryBuilder . addWithStats ( contextId , summaryTree ) ; } } return summaryBuilder . getSummaryTree ( ) ; } public submitMessage ( type , content , localOpMetadata ) { this . submit ( type , content , localOpMetadata ) ; } public submitSignal ( type , content ) { this . verifyNotClosed ( ) ; return this . dataStoreContext . submitSignal ( type , content ) ; } /**
     * Will return when the data store is attached.
     */ public async waitAttached ( ) { return this . deferredAttached . promise ; } public raiseContainerWarning ( warning ) { this . dataStoreContext . raiseContainerWarning ( warning ) ; } /**
     * Attach channel should only be called after the data store has been attached
     */ private attachChannel ( channel ) { this . verifyNotClosed ( ) ; if ( channel . handle . isAttached ) { return ; } channel . handle . attachGraph ( ) ; assert ( this . isAttached , 0x182 ) ; if ( this . bindState === BindState . Bound && this . graphAttachState === AttachState . Attached ) { const summarizeResult = summarizeChannel ( channel , true , false ) ; const snapshot = convertSummaryTreeToITree ( summarizeResult . summary ) ; const message = { id : channel . id , snapshot , type : channel . attributes . type } ; this . pendingAttach . set ( channel . id , message ) ; this . submit ( DataStoreMessageType . Attach , message ) ; } const context = this . contexts . get ( channel . id ) as LocalChannelContextBase ; context . markAttached ( ) ; } private submitChannelOp ( address , contents , localOpMetadata ) { const envelope = { address , contents } ; this . submit ( DataStoreMessageType . ChannelOp , envelope , localOpMetadata ) ; } private submit ( type , content , localOpMetadata = undefined ) { this . verifyNotClosed ( ) ; this . dataStoreContext . submitMessage ( type , content , localOpMetadata ) ; } param content param localOpMetadata public reSubmit ( type , content , localOpMetadata ) { this . verifyNotClosed ( ) ; switch ( type ) { case DataStoreMessageType . ChannelOp : { const envelope = content as IEnvelope ; const channelContext = this . contexts . get ( envelope . address ) ; assert ( ! ! channelContext , 0x183 ) ; channelContext . reSubmit ( envelope . contents , localOpMetadata ) ; break ; } case DataStoreMessageType . Attach : this . submit ( type , content , localOpMetadata ) ; break ; default : unreachableCase ( type ) ; } } public async applyStashedOp ( content ) { const envelope = content as IEnvelope ; const channelContext = this . contexts . get ( envelope . address ) ; assert ( ! ! channelContext , 0x184 ) ; await channelContext . getChannel ( ) ; return channelContext . applyStashedOp ( envelope . contents ) ; } private setChannelDirty ( address ) { this . verifyNotClosed ( ) ; this . dataStoreContext . setChannelDirty ( address ) ; } private processChannelOp ( message , local , localOpMetadata ) { this . verifyNotClosed ( ) ; const envelope = message . contents as IEnvelope ; const transformed = { ... message , contents : envelope . contents } ; const channelContext = this . contexts . get ( envelope . address ) ; assert ( ! ! channelContext , 0x185 ) ; channelContext . processOp ( transformed , local , localOpMetadata ) ; return channelContext ; } private attachListener ( ) { this . setMaxListeners ( Number . MAX_SAFE_INTEGER ) ; this . dataStoreContext . once ( "attaching" , ( ) => { assert ( this . bindState !== BindState . NotBound , 0x186 ) ; this . _attachState = AttachState . Attaching ; this . deferredAttached . resolve ( ) ; this . emit ( "attaching" ) ; } ) ; this . dataStoreContext . once ( "attached" , ( ) => { assert ( this . bindState === BindState . Bound , 0x187 ) ; this . _attachState = AttachState . Attached ; this . emit ( "attached" ) ; } ) ; } private verifyNotClosed ( ) { if ( this . _disposed ) { throw new Error ( "Runtime is closed" ) ; } } } param Base param requestHandler export const mixinRequestHandler = ( requestHandler , Base = FluidDataStoreRuntime ) => class RuntimeWithRequestHandler extends Base { public async request ( request ) { const response = await super . request ( request ) ; if ( response . status === 404 ) { return requestHandler ( request , this ) ; } return response ; } } as typeof FluidDataStoreRuntime ; param handler param Base export const mixinSummaryHandler = ( handler , Base = FluidDataStoreRuntime ) => class RuntimeWithSummarizerHandler extends Base { private addBlob ( summary , path , content ) { const firstName = path . shift ( ) ; if ( firstName === undefined ) { throw new Error ( "Path can't be empty" ) ; } let blob = { type : SummaryType . Blob , content } ; summary . stats . blobNodeCount ++ ; summary . stats . totalBlobSize += content . length ; for ( const name of path . reverse ( ) ) { blob = { type : SummaryType . Tree , tree : { [ name ] : blob } } ; summary . stats . treeNodeCount ++ ; } summary . summary . tree [ firstName ] = blob ; } async summarize ( ... args ) { const summary = await super . summarize ( ... args ) ; const content = await handler ( this ) ; if ( content !== undefined ) { this . addBlob ( summary , content . path , content . content ) ; } return summary ; } } as typeof FluidDataStoreRuntime ;