import { ITelemetryLogger } from "@fluidframework/common-definitions" ; import { assert } from "@fluidframework/common-utils" ; import { DataCorruptionError } from "@fluidframework/container-utils" ; import { IFluidHandle } from "@fluidframework/core-interfaces" ; import { IChannel , IChannelAttributes , IFluidDataStoreRuntime , IChannelFactory , } from "@fluidframework/datastore-definitions" ; import { IDocumentStorageService } from "@fluidframework/driver-definitions" ; import { readAndParse } from "@fluidframework/driver-utils" ; import { ISequencedDocumentMessage , ISnapshotTree , } from "@fluidframework/protocol-definitions" ; import { CreateChildSummarizerNodeFn , IFluidDataStoreContext , IGarbageCollectionData , IGarbageCollectionDetailsBase , ISummarizeInternalResult , ISummarizeResult , ISummarizerNodeWithGC , } from "@fluidframework/runtime-definitions" ; import { ChildLogger , TelemetryDataTag , ThresholdCounter } from "@fluidframework/telemetry-utils" ; import { attributesBlobKey , createServiceEndpoints , IChannelContext , summarizeChannelAsync , } from "./channelContext" ; import { ChannelDeltaConnection } from "./channelDeltaConnection" ; import { ChannelStorageService } from "./channelStorageService" ; import { ISharedObjectRegistry } from "./dataStoreRuntime" ; export class RemoteChannelContext implements IChannelContext { private isLoaded = false ; private pending : ISequencedDocumentMessage [ ] | undefined = [ ] ; private channelP : Promise < IChannel > | undefined ; private channel : IChannel | undefined ; private readonly services : { readonly deltaConnection : ChannelDeltaConnection ; readonly objectStorage : ChannelStorageService ; } ; private readonly summarizerNode : ISummarizerNodeWithGC ; private readonly subLogger : ITelemetryLogger ; private readonly thresholdOpsCounter : ThresholdCounter ; private static readonly pendingOpsCountThreshold = 1000 ; constructor ( private readonly runtime , private readonly dataStoreContext , storageService , submitFn , dirtyFn , addedGCOutboundReferenceFn , private readonly id , baseSnapshot , private readonly registry , extraBlobs , createSummarizerNode , getBaseGCDetails , private readonly attachMessageType ? ) { this . services = createServiceEndpoints ( this . id , this . dataStoreContext . connected , submitFn , ( ) => dirtyFn ( this . id ) , addedGCOutboundReferenceFn , storageService , baseSnapshot , extraBlobs ) ; const thisSummarizeInternal = async ( fullTree , trackState ) => this . summarizeInternal ( fullTree , trackState ) ; this . summarizerNode = createSummarizerNode ( thisSummarizeInternal , async ( fullGC ? ) => this . getGCDataInternal ( fullGC ) , async ( ) => getBaseGCDetails ( ) ) ; this . subLogger = ChildLogger . create ( this . runtime . logger , "RemoteChannelContext" ) ; this . thresholdOpsCounter = new ThresholdCounter ( RemoteChannelContext . pendingOpsCountThreshold , this . subLogger ) ; } public getChannel ( ) { if ( this . channelP === undefined ) { this . channelP = this . loadChannel ( ) ; } return this . channelP ; } public setConnectionState ( connected , clientId ? ) { if ( ! this . isLoaded ) { return ; } this . services . deltaConnection . setConnectionState ( connected ) ; } public applyStashedOp ( message ) { assert ( this . isLoaded , 0x194 ) ; return this . services . deltaConnection . applyStashedOp ( message ) ; } public processOp ( message , local , localOpMetadata ) { this . summarizerNode . invalidate ( message . sequenceNumber ) ; if ( this . isLoaded ) { this . services . deltaConnection . process ( message , local , localOpMetadata ) ; } else { assert ( ! local , 0x195 ) ; assert ( this . pending !== undefined , 0x23e ) ; this . pending . push ( message ) ; this . thresholdOpsCounter . sendIfMultiple ( "StorePendingOps" , this . pending . length ) ; } } public reSubmit ( content , localOpMetadata ) { assert ( this . isLoaded , 0x196 ) ; this . services . deltaConnection . reSubmit ( content , localOpMetadata ) ; } param fullTree param trackState public async summarize ( fullTree = false , trackState = true ) { return this . summarizerNode . summarize ( fullTree , trackState ) ; } private async summarizeInternal ( fullTree , trackState ) { const channel = await this . getChannel ( ) ; const summarizeResult = await summarizeChannelAsync ( channel , fullTree , trackState ) ; return { ... summarizeResult , id : this . id } ; } private async loadChannel ( ) { assert ( ! this . isLoaded , 0x197 ) ; let attributes ; if ( await this . services . objectStorage . contains ( attributesBlobKey ) ) { attributes = await readAndParse < IChannelAttributes | undefined > ( this . services . objectStorage , attributesBlobKey ) ; } let factory ; if ( attributes === undefined ) { if ( this . attachMessageType === undefined ) { throw new DataCorruptionError ( "channelTypeNotAvailable" , { channelId : this . id , dataStoreId : { value : this . dataStoreContext . id , tag : TelemetryDataTag . PackageData } , dataStorePackagePath : this . dataStoreContext . packagePath . join ( "/" ) } ) ; } factory = this . registry . get ( this . attachMessageType ) ; if ( factory === undefined ) { throw new DataCorruptionError ( "channelFactoryNotRegisteredForAttachMessageType" , { channelId : this . id , dataStoreId : { value : this . dataStoreContext . id , tag : TelemetryDataTag . PackageData } , dataStorePackagePath : this . dataStoreContext . packagePath . join ( "/" ) , channelFactoryType : this . attachMessageType } ) ; } attributes = factory . attributes ; } else { factory = this . registry . get ( attributes . type ) ; if ( factory === undefined ) { throw new DataCorruptionError ( "channelFactoryNotRegisteredForGivenType" , { channelId : this . id , dataStoreId : { value : this . dataStoreContext . id , tag : TelemetryDataTag . PackageData } , dataStorePackagePath : this . dataStoreContext . packagePath . join ( "/" ) , channelFactoryType : attributes . type } ) ; } } if ( attributes . snapshotFormatVersion !== undefined && attributes . snapshotFormatVersion !== factory . attributes . snapshotFormatVersion ) { this . subLogger . sendTelemetryEvent ( { eventName : "ChannelAttributesVersionMismatch" , channelType : { value : attributes . type , tag : TelemetryDataTag . PackageData } , channelSnapshotVersion : { value : `${ attributes . snapshotFormatVersion }@${ attributes . packageVersion }` , tag : TelemetryDataTag . PackageData } , channelCodeVersion : { value : `${ factory . attributes . snapshotFormatVersion }@${ factory . attributes . packageVersion }` , tag : TelemetryDataTag . PackageData } } ) ; } const channel = await factory . load ( this . runtime , this . id , this . services , attributes ) ; assert ( this . pending !== undefined , 0x23f ) ; for ( const message of this . pending ) { this . services . deltaConnection . process ( message , false , undefined ) ; } this . thresholdOpsCounter . send ( "ProcessPendingOps" , this . pending . length ) ; this . channel = channel ; this . pending = undefined ; this . isLoaded = true ; this . services . deltaConnection . setConnectionState ( this . dataStoreContext . connected ) ; return this . channel ; } param fullGC public async getGCData ( fullGC = false ) { return this . summarizerNode . getGCData ( fullGC ) ; } param fullGC private async getGCDataInternal ( fullGC = false ) { const channel = await this . getChannel ( ) ; return channel . getGCData ( fullGC ) ; } public updateUsedRoutes ( usedRoutes , gcTimestamp ? ) { /**
         * Currently, DDSs are always considered referenced and are not garbage collected. Update the summarizer node's
         * used routes to contain a route to this channel context.
         * Once we have GC at DDS level, this will be updated to use the passed usedRoutes. See -
         * https://github.com/microsoft/FluidFramework/issues/4611
         */ this . summarizerNode . updateUsedRoutes ( [ "" ] ) ; } }