import { default as AbortController } from "abort-controller" ; import { ITelemetryLogger } from "@fluidframework/common-definitions" ; import { assert , delay , } from "@fluidframework/common-utils" ; import { PerformanceEvent , } from "@fluidframework/telemetry-utils" ; import * as api from "@fluidframework/protocol-definitions" ; import { ISummaryContext , IDocumentStorageService , LoaderCachingPolicy , DriverErrorType , } from "@fluidframework/driver-definitions" ; import { RateLimiter , NonRetryableError } from "@fluidframework/driver-utils" ; import { IOdspResolvedUrl , ISnapshotOptions , OdspErrorType , InstrumentedStorageTokenFetcher , } from "@fluidframework/odsp-driver-definitions" ; import { IDocumentStorageGetVersionsResponse , HostStoragePolicyInternal , IVersionedValueWithEpoch , ISnapshotCachedEntry , } from "./contracts" ; import { downloadSnapshot , fetchSnapshot , fetchSnapshotWithRedeem } from "./fetchSnapshot" ; import { getUrlAndHeadersWithAuth } from "./getUrlAndHeadersWithAuth" ; import { IOdspCache } from "./odspCache" ; import { createCacheSnapshotKey , getWithRetryForTokenRefresh , ISnapshotContents , } from "./odspUtils" ; import { EpochTracker } from "./epochTracker" ; import { OdspSummaryUploadManager } from "./odspSummaryUploadManager" ; import { FlushResult } from "./odspDocumentDeltaConnection" ; async function promiseRaceWithWinner < T > ( promises ) { return new Promise ( ( resolve , reject ) => { promises . forEach ( ( p , index ) => { p . then ( ( v ) => resolve ( { index , value : v } ) ) . catch ( reject ) ; } ) ; } ) ; } class BlobCache { private blobCacheTimeout : ReturnType < typeof setTimeout > | undefined ; private deferBlobCacheClear : boolean = false ; private readonly _blobCache : Map < string , ArrayBuffer > = new Map ( ) ; private readonly blobsEvicted : Set < string > = new Set ( ) ; private blobCacheTimeoutDuration = 2 * 60 * 1000 ; private readonly purgeEnabled = false ; public get value ( ) { return this . _blobCache ; } public addBlobs ( blobs ) { blobs . forEach ( ( value , blobId ) => { this . _blobCache . set ( blobId , value ) ; } ) ; this . scheduleClearBlobsCache ( ) ; } /**
     * Schedule a timer for clearing the blob cache or defer the current one.
     */ private scheduleClearBlobsCache ( ) { if ( this . blobCacheTimeout !== undefined ) { this . deferBlobCacheClear = true ; } else if ( this . purgeEnabled ) { const clearCacheOrDefer = ( ) => { this . blobCacheTimeout = undefined ; if ( this . deferBlobCacheClear ) { this . deferBlobCacheClear = false ; this . scheduleClearBlobsCache ( ) ; } else { this . _blobCache . forEach ( ( _ , blobId ) => this . blobsEvicted . add ( blobId ) ) ; this . _blobCache . clear ( ) ; } } ; this . blobCacheTimeout = setTimeout ( clearCacheOrDefer , this . blobCacheTimeoutDuration ) ; this . blobCacheTimeoutDuration = 10 * 1000 ; } } public getBlob ( blobId ) { this . scheduleClearBlobsCache ( ) ; const blobContent = this . _blobCache . get ( blobId ) ; const evicted = this . blobsEvicted . has ( blobId ) ; return { blobContent , evicted } ; } public setBlob ( blobId , blob ) { const size = blob . byteLength ; if ( size < 256 * 1024 ) { this . scheduleClearBlobsCache ( ) ; return this . _blobCache . set ( blobId , blob ) ; } else { this . blobsEvicted . add ( blobId ) ; } } } export class OdspDocumentStorageService implements IDocumentStorageService { readonly policies = { caching : LoaderCachingPolicy . NoCaching , minBlobSize : 2048 } ; private readonly commitCache : Map < string , api . ISnapshotTree > = new Map ( ) ; private readonly attributesBlobHandles : Set < string > = new Set ( ) ; private readonly odspSummaryUploadManager : OdspSummaryUploadManager ; private _ops : api . ISequencedDocumentMessage [ ] | undefined ; private firstVersionCall = true ; private _snapshotSequenceNumber : number | undefined ; private readonly documentId : string ; private readonly snapshotUrl : string | undefined ; private readonly attachmentPOSTUrl : string | undefined ; private readonly attachmentGETUrl : string | undefined ; /**
     * NOTE: While commit cfff6e3 added restrictions to prevent large payloads, snapshot failures will continue to
     * happen until blob request throttling is implemented. Until then, as a temporary fix we set arbitrarily large
     * snapshot size and timeout limits so that such failures are unlikely to occur.
     */ private readonly maxSnapshotSizeLimit = 500000000 ; private readonly maxSnapshotFetchTimeout = 120000 ; private readonly createBlobRateLimiter = new RateLimiter ( 1 ) ; private readonly blobCache = new BlobCache ( ) ; public set ops ( ops ) { assert ( this . _ops === undefined , 0x0a5 ) ; this . _ops = ops ; } public get ops ( ) : api . ISequencedDocumentMessage [ ] | undefined { return this . _ops ; } public get snapshotSequenceNumber ( ) { return this . _snapshotSequenceNumber ; } constructor ( private readonly odspResolvedUrl , private readonly getStorageToken , private readonly logger , private readonly fetchFullSnapshot , private readonly cache , private readonly hostPolicy , private readonly epochTracker , private readonly flushCallback ) { this . documentId = this . odspResolvedUrl . hashedDocumentId ; this . snapshotUrl = this . odspResolvedUrl . endpoints . snapshotStorageUrl ; this . attachmentPOSTUrl = this . odspResolvedUrl . endpoints . attachmentPOSTStorageUrl ; this . attachmentGETUrl = this . odspResolvedUrl . endpoints . attachmentGETStorageUrl ; this . odspSummaryUploadManager = new OdspSummaryUploadManager ( this . snapshotUrl , getStorageToken , logger , epochTracker ) ; } public get repositoryUrl ( ) : string { return "" ; } public async createBlob ( file ) { this . checkAttachmentPOSTUrl ( ) ; const response = await getWithRetryForTokenRefresh ( async ( options ) => { const storageToken = await this . getStorageToken ( options , "CreateBlob" ) ; const { url , headers } = getUrlAndHeadersWithAuth ( `${ this . attachmentPOSTUrl }/content` , storageToken ) ; headers [ "Content-Type" ] = "application/octet-stream" ; return PerformanceEvent . timedExecAsync ( this . logger , { eventName : "createBlob" , size : file . byteLength , waitQueueLength : this . createBlobRateLimiter . waitQueueLength } , async ( event ) => { const res = await this . createBlobRateLimiter . schedule ( async ( ) => this . epochTracker . fetchAndParseAsJSON < api . ICreateBlobResponse > ( url , { body : file , headers , method : "POST" } , "createBlob" ) ) ; event . end ( { blobId : res . content . id , ... res . commonSpoHeaders } ) ; return res ; } ) ; } ) ; return response . content ; } private async readBlobCore ( blobId ) { const { blobContent , evicted } = this . blobCache . getBlob ( blobId ) ; let blob = blobContent ; if ( blob === undefined ) { this . checkAttachmentGETUrl ( ) ; blob = await getWithRetryForTokenRefresh ( async ( options ) => { const storageToken = await this . getStorageToken ( options , "GetBlob" ) ; const unAuthedUrl = `${ this . attachmentGETUrl }/${ encodeURIComponent ( blobId ) }/content` ; const { url , headers } = getUrlAndHeadersWithAuth ( unAuthedUrl , storageToken ) ; return PerformanceEvent . timedExecAsync ( this . logger , { eventName : "readDataBlob" , blobId , evicted , headers : Object . keys ( headers ) . length !== 0 ? true : undefined , waitQueueLength : this . epochTracker . rateLimiter . waitQueueLength } , async ( event ) => { const res = await this . epochTracker . fetchArray ( url , { headers } , "blob" ) ; event . end ( { waitQueueLength : this . epochTracker . rateLimiter . waitQueueLength , ... res . commonSpoHeaders , attempts : options . refresh ? 2 : 1 } ) ; const cacheControl = res . headers . get ( "cache-control" ) ; if ( cacheControl === undefined || ! ( cacheControl . includes ( "private" ) || cacheControl . includes ( "public" ) ) ) { this . logger . sendErrorEvent ( { eventName : "NonCacheableBlob" , cacheControl , blobId , ... res . commonSpoHeaders } ) ; } return res . content ; } ) ; } ) ; this . blobCache . setBlob ( blobId , blob ) ; } return blob ; } public async readBlob ( blobId ) { return this . readBlobCore ( blobId ) ; } public async getSnapshotTree ( version ? ) { if ( ! this . snapshotUrl ) { return null ; } let id ; if ( ! version || ! version . id ) { const versions = await this . getVersions ( null , 1 ) ; if ( ! versions || versions . length === 0 ) { return null ; } id = versions [ 0 ] . id ; } else { id = version . id ; } const snapshotTree = await this . readTree ( id ) ; if ( ! snapshotTree ) { return null ; } const commits = { } ; for ( const key of Object . keys ( snapshotTree . commits ) ) { commits [ decodeURIComponent ( key ) ] = snapshotTree . commits [ key ] ; } let finalTree ; if ( commits && commits [ ".app" ] ) { finalTree = await this . readSummaryTree ( commits [ ".protocol" ] || snapshotTree . trees [ ".protocol" ] , commits [ ".app" ] as string ) ; } else { if ( snapshotTree . blobs ) { const attributesBlob = snapshotTree . blobs . attributes ; if ( attributesBlob ) { this . attributesBlobHandles . add ( attributesBlob ) ; } } snapshotTree . commits = commits ; const appTree = snapshotTree . trees [ ".app" ] ; const protocolTree = snapshotTree . trees [ ".protocol" ] ; finalTree = this . combineProtocolAndAppSnapshotTree ( appTree , protocolTree ) ; } return finalTree ; } public async getVersions ( blobid , count ) { if ( blobid !== this . documentId && blobid ) { return [ { id : blobid , treeId : undefined ! } , ] ; } if ( ! this . snapshotUrl ) { return [ ] ; } if ( this . firstVersionCall && count === 1 && ( blobid === null || blobid === this . documentId ) ) { const hostSnapshotOptions = this . hostPolicy . snapshotOptions ; const odspSnapshotCacheValue = await PerformanceEvent . timedExecAsync ( this . logger , { eventName : "ObtainSnapshot" } , async ( event ) => { const props = { } ; let retrievedSnapshot ; const cachedSnapshotP = this . epochTracker . get ( createCacheSnapshotKey ( this . odspResolvedUrl ) ) . then ( async ( snapshotCachedEntry ) => { if ( snapshotCachedEntry !== undefined ) { const age = Date . now ( ) - ( snapshotCachedEntry . cacheEntryTime ?? ( Date . now ( ) - 30 * 24 * 60 * 60 * 1000 ) ) ; props [ "cacheEntryAge" ] = age ; } return snapshotCachedEntry ; } ) ; let method ; if ( this . hostPolicy . concurrentSnapshotFetch && ! this . hostPolicy . summarizerClient ) { const networkSnapshotP = this . fetchSnapshot ( hostSnapshotOptions ) ; const promiseRaceWinner = await promiseRaceWithWinner ( [ cachedSnapshotP . catch ( ( ) => undefined ) , networkSnapshotP . catch ( ( ) => undefined ) , ] ) ; retrievedSnapshot = promiseRaceWinner . value ; method = promiseRaceWinner . index === 0 ? "cache" : "network" ; if ( retrievedSnapshot === undefined ) { if ( promiseRaceWinner . index === 1 ) { retrievedSnapshot = await cachedSnapshotP ; method = "cache" ; } if ( retrievedSnapshot === undefined ) { retrievedSnapshot = await networkSnapshotP ; method = "network" ; } } } else { retrievedSnapshot = await cachedSnapshotP ; method = retrievedSnapshot !== undefined ? "cache" : "network" ; if ( retrievedSnapshot === undefined ) { retrievedSnapshot = await this . fetchSnapshot ( hostSnapshotOptions ) ; } } if ( method === "network" ) { props [ "cacheEntryAge" ] = undefined ; } event . end ( { ... props , method } ) ; return retrievedSnapshot ; } ) ; this . firstVersionCall = false ; this . _snapshotSequenceNumber = odspSnapshotCacheValue . sequenceNumber ; const { snapshotTree , blobs , ops } = odspSnapshotCacheValue ; let id ; if ( snapshotTree ) { id = snapshotTree . id ; assert ( id !== undefined , 0x221 ) ; this . setRootTree ( id , snapshotTree ) ; } if ( blobs ) { this . initBlobsCache ( blobs ) ; } this . ops = ops ; return id ? [ { id , treeId : undefined ! } ] : [ ] ; } return getWithRetryForTokenRefresh ( async ( options ) => { const storageToken = await this . getStorageToken ( options , "GetVersions" ) ; const { url , headers } = getUrlAndHeadersWithAuth ( `${ this . snapshotUrl }/versions?count=${ count }` , storageToken ) ; const response = await PerformanceEvent . timedExecAsync ( this . logger , { eventName : "getVersions" , headers : Object . keys ( headers ) . length !== 0 ? true : undefined } , async ( ) => this . epochTracker . fetchAndParseAsJSON < IDocumentStorageGetVersionsResponse > ( url , { headers } , "versions" ) ) ; const versionsResponse = response . content ; if ( ! versionsResponse ) { throw new NonRetryableError ( "getVersionsReturnedNoResponse" , "No response from /versions endpoint" , DriverErrorType . genericNetworkError ) ; } if ( ! Array . isArray ( versionsResponse . value ) ) { throw new NonRetryableError ( "getVersionsReturnedNonArrayResponse" , "Incorrect response from /versions endpoint" , DriverErrorType . genericNetworkError ) ; } return versionsResponse . value . map ( ( version ) => { let date ; for ( const rec of version . message . split ( "\n" ) ) { const index = rec . indexOf ( ":" ) ; if ( index !== - 1 && rec . substr ( 0 , index ) === "Date" ) { date = rec . substr ( index + 1 ) . trim ( ) ; break ; } } return { date , id : version . id , treeId : undefined ! } ; } ) ; } ) ; } private async fetchSnapshot ( hostSnapshotOptions ) { return this . fetchSnapshotCore ( hostSnapshotOptions ) . catch ( ( error ) => { if ( typeof error === "object" && error !== null ) { error . canRetry = false ; } throw error ; } ) ; } private async fetchSnapshotCore ( hostSnapshotOptions ) { const snapshotOptions = { mds : this . maxSnapshotSizeLimit , ... hostSnapshotOptions , timeout : hostSnapshotOptions ?. timeout ? Math . min ( hostSnapshotOptions . timeout , this . maxSnapshotFetchTimeout ) : this . maxSnapshotFetchTimeout } ; if ( this . hostPolicy . summarizerClient ) { snapshotOptions . mds = undefined ; snapshotOptions . timeout = undefined ; } const snapshotDownloader = async ( finalOdspResolvedUrl , storageToken , options , controller ? ) => { return downloadSnapshot ( finalOdspResolvedUrl , storageToken , this . logger , options , this . hostPolicy . fetchBinarySnapshotFormat , controller , this . epochTracker ) ; } ; const putInCache = async ( valueWithEpoch ) => { return this . cache . persistedCache . put ( createCacheSnapshotKey ( this . odspResolvedUrl ) , valueWithEpoch . value ) ; } ; const removeEntries = async ( ) => this . cache . persistedCache . removeEntries ( ) ; try { const odspSnapshot = await fetchSnapshotWithRedeem ( this . odspResolvedUrl , this . getStorageToken , snapshotOptions , this . logger , snapshotDownloader , putInCache , removeEntries , this . hostPolicy . enableRedeemFallback ) ; return odspSnapshot ; } catch ( error ) { const errorType = error . errorType ; if ( ( errorType === OdspErrorType . snapshotTooBig && hostSnapshotOptions ?. mds !== undefined ) && ( this . hostPolicy . summarizerClient !== true ) ) { throw error ; } if ( ( errorType === OdspErrorType . snapshotTooBig || errorType === OdspErrorType . fetchTimeout ) && snapshotOptions . blobs ) { this . logger . sendErrorEvent ( { eventName : "TreeLatest_SecondCall" , errorType } ) ; const snapshotOptionsWithoutBlobs = { ... snapshotOptions , blobs : 0 , mds : undefined , timeout : undefined } ; const odspSnapshot = await fetchSnapshotWithRedeem ( this . odspResolvedUrl , this . getStorageToken , snapshotOptionsWithoutBlobs , this . logger , snapshotDownloader , putInCache , removeEntries , this . hostPolicy . enableRedeemFallback ) ; return odspSnapshot ; } throw error ; } } public async write ( tree , parents , message ) { this . checkSnapshotUrl ( ) ; throw new Error ( "Not supported" ) ; } public async uploadSummaryWithContext ( summary , context ) { this . checkSnapshotUrl ( ) ; if ( ".protocol" in summary . tree && context . ackHandle !== undefined ) { let retry = 0 ; for ( ; ; ) { const result = await this . flushCallback ( ) ; const seq = result . lastPersistedSequenceNumber ; if ( seq !== undefined && seq >= context . referenceSequenceNumber ) { break ; } retry ++ ; if ( retry > 3 ) { this . logger . sendErrorEvent ( { eventName : "FlushFailure" , ... result , retry , referenceSequenceNumber : context . referenceSequenceNumber } ) ; break ; } this . logger . sendPerformanceEvent ( { eventName : "FlushExtraCall" , ... result , retry , referenceSequenceNumber : context . referenceSequenceNumber } ) ; await delay ( 1000 * ( result . retryAfter ?? 1 ) ) ; } } const id = await PerformanceEvent . timedExecAsync ( this . logger , { eventName : "uploadSummaryWithContext" } , async ( ) => this . odspSummaryUploadManager . writeSummaryTree ( summary , context ) ) ; return id ; } public async downloadSummary ( commit ) { throw new Error ( "Not implemented yet" ) ; } private setRootTree ( id , tree ) { this . commitCache . set ( id , tree ) ; } private initBlobsCache ( blobs ) { this . blobCache . addBlobs ( blobs ) ; } private checkSnapshotUrl ( ) { if ( ! this . snapshotUrl ) { throw new NonRetryableError ( "noSnapshotUrlProvided" , "Method failed because no snapshot url was available" , DriverErrorType . genericError ) ; } } private checkAttachmentPOSTUrl ( ) { if ( ! this . attachmentPOSTUrl ) { throw new NonRetryableError ( "noAttachmentPOSTUrlProvided" , "Method failed because no attachment POST url was available" , DriverErrorType . genericError ) ; } } private checkAttachmentGETUrl ( ) { if ( ! this . attachmentGETUrl ) { throw new NonRetryableError ( "noAttachmentGETUrlWasProvided" , "Method failed because no attachment GET url was available" , DriverErrorType . genericError ) ; } } private async readTree ( id ) { if ( ! this . snapshotUrl ) { return null ; } let tree = this . commitCache . get ( id ) ; if ( ! tree ) { tree = await getWithRetryForTokenRefresh ( async ( options ) => { const storageToken = await this . getStorageToken ( options , "ReadCommit" ) ; const snapshotDownloader = async ( url , fetchOptions ) => { return this . epochTracker . fetchAndParseAsJSON ( url , fetchOptions , "snapshotTree" ) ; } ; const snapshot = await fetchSnapshot ( this . snapshotUrl ! , storageToken , id , this . fetchFullSnapshot , this . logger , snapshotDownloader ) ; let treeId = "" ; if ( snapshot . snapshotTree ) { assert ( snapshot . snapshotTree . id !== undefined , 0x222 ) ; treeId = snapshot . snapshotTree . id ; this . setRootTree ( treeId , snapshot . snapshotTree ) ; } if ( snapshot . blobs ) { this . initBlobsCache ( snapshot . blobs ) ; } return this . commitCache . get ( id ) ?? this . commitCache . get ( treeId ) ; } ) ; } if ( ! tree ) { return null ; } return tree ; } param protocolTreeOrId param appTreeId private async readSummaryTree ( protocolTreeOrId , appTreeId ) { let hierarchicalProtocolTree ; if ( typeof ( protocolTreeOrId ) === "string" ) { hierarchicalProtocolTree = await this . readTree ( protocolTreeOrId ) ; } else { hierarchicalProtocolTree = protocolTreeOrId ; } const hierarchicalAppTree = await this . readTree ( appTreeId ) ; if ( ! hierarchicalProtocolTree ) { throw new Error ( "Invalid protocol tree" ) ; } if ( ! hierarchicalAppTree ) { throw new Error ( "Invalid app tree" ) ; } if ( hierarchicalProtocolTree . blobs ) { const attributesBlob = hierarchicalProtocolTree . blobs . attributes ; if ( attributesBlob ) { this . attributesBlobHandles . add ( attributesBlob ) ; } } return this . combineProtocolAndAppSnapshotTree ( hierarchicalAppTree , hierarchicalProtocolTree ) ; } private combineProtocolAndAppSnapshotTree ( hierarchicalAppTree , hierarchicalProtocolTree ) { const summarySnapshotTree = { blobs : { ... hierarchicalAppTree . blobs } , commits : { ... hierarchicalAppTree . commits } , trees : { ... hierarchicalAppTree . trees , ".protocol" : hierarchicalProtocolTree } } ; return summarySnapshotTree ; } }