import { v4 as uuid } from "uuid" ; import { assert , Deferred } from "@fluidframework/common-utils" ; import { ITelemetryLogger } from "@fluidframework/common-definitions" ; import { ThrottlingError , RateLimiter , NonRetryableError } from "@fluidframework/driver-utils" ; import { IConnected } from "@fluidframework/protocol-definitions" ; import { snapshotKey , ICacheEntry , IEntry , IFileEntry , IPersistedCache , IOdspError , } from "@fluidframework/odsp-driver-definitions" ; import { DriverErrorType } from "@fluidframework/driver-definitions" ; import { PerformanceEvent , isFluidError , normalizeError } from "@fluidframework/telemetry-utils" ; import { fetchAndParseAsJSONHelper , fetchArray , IOdspResponse } from "./odspUtils" ; import { IOdspCache , INonPersistentCache , IPersistedFileCache , } from "./odspCache" ; import { IVersionedValueWithEpoch , persistedCacheValueVersion } from "./contracts" ; export type FetchType = "blob" | "createBlob" | "createFile" | "joinSession" | "ops" | "test" | "snapshotTree" | "treesLatest" | "uploadSummary" | "push" | "versions" ; export type FetchTypeInternal = FetchType | "cache" ; export const Odsp409Error = "Odsp409Error" ; export const defaultCacheExpiryTimeoutMs = 2 * 24 * 60 * 60 * 1000 ; /**
 * This class is a wrapper around fetch calls. It adds epoch to the request made so that the
 * server can match it with its epoch value in order to match the version.
 * It also validates the epoch value received in response of fetch calls. If the epoch does not match,
 * then it also clears all the cached entries for the given container.
 */ export class EpochTracker implements IPersistedFileCache { private _fluidEpoch : string | undefined ; public readonly rateLimiter : RateLimiter ; private readonly driverId = uuid ( ) ; private networkCallNumber = 1 ; constructor ( protected readonly cache , protected readonly fileEntry , protected readonly logger ) { this . rateLimiter = new RateLimiter ( 24 ) ; } public setEpoch ( epoch , fromCache , fetchType ) { assert ( this . _fluidEpoch === undefined , 0x1db ) ; this . _fluidEpoch = epoch ; this . logger . sendTelemetryEvent ( { eventName : "EpochLearnedFirstTime" , epoch , fetchType , fromCache } ) ; } public async get ( entry ) { try { const value = await this . cache . get ( this . fileEntryFromEntry ( entry ) ) ; if ( value === undefined || value . version !== persistedCacheValueVersion ) { return undefined ; } assert ( value . fluidEpoch !== undefined , 0x1dc ) ; if ( this . _fluidEpoch === undefined ) { this . setEpoch ( value . fluidEpoch , true , "cache" ) ; } else if ( this . _fluidEpoch !== value . fluidEpoch ) { return undefined ; } if ( entry . type === snapshotKey ) { const cacheTime = value . value ?. cacheEntryTime ; const currentTime = Date . now ( ) ; if ( cacheTime === undefined || currentTime - cacheTime >= defaultCacheExpiryTimeoutMs ) { this . logger . sendTelemetryEvent ( { eventName : "odspVersionsCacheExpired" , duration : currentTime - cacheTime , maxCacheAgeMs : defaultCacheExpiryTimeoutMs } ) ; await this . removeEntries ( ) ; return undefined ; } } return value . value ; } catch ( error ) { this . logger . sendErrorEvent ( { eventName : "cacheFetchError" , type : entry . type } , error ) ; return undefined ; } } public async put ( entry , value ) { assert ( this . _fluidEpoch !== undefined , 0x1dd ) ; if ( entry . type === snapshotKey ) { value . cacheEntryTime = value . cacheEntryTime ?? Date . now ( ) ; } const data = { value , version : persistedCacheValueVersion , fluidEpoch : this . _fluidEpoch } ; return this . cache . put ( this . fileEntryFromEntry ( entry ) , data ) . catch ( ( error ) => { this . logger . sendErrorEvent ( { eventName : "cachePutError" , type : entry . type } , error ) ; throw error ; } ) ; } public async removeEntries ( ) { try { return await this . cache . removeEntries ( this . fileEntry ) ; } catch ( error ) { this . logger . sendErrorEvent ( { eventName : "removeCacheEntries" } , error ) ; } } public get fluidEpoch ( ) { return this . _fluidEpoch ; } public async validateEpochFromPush ( details ) { const epoch = details . epoch ; assert ( epoch !== undefined , 0x09d ) ; try { this . validateEpochFromResponse ( epoch , "push" ) ; } catch ( error ) { await this . checkForEpochError ( error , epoch , "push" ) ; throw error ; } } param url param fetchOptions param fetchType param addInBody public async fetchAndParseAsJSON < T > ( url , fetchOptions , fetchType , addInBody = false ) { const clientCorrelationId = this . formatClientCorrelationId ( ) ; this . addEpochInRequest ( fetchOptions , addInBody , clientCorrelationId ) ; let epochFromResponse ; return this . rateLimiter . schedule ( async ( ) => fetchAndParseAsJSONHelper < T > ( url , fetchOptions ) ) . then ( ( response ) => { epochFromResponse = response . headers . get ( "x-fluid-epoch" ) ; this . validateEpochFromResponse ( epochFromResponse , fetchType ) ; response . commonSpoHeaders = { ... response . commonSpoHeaders , "X-RequestStats" : clientCorrelationId } ; return response ; } ) . catch ( async ( error ) => { if ( epochFromResponse === undefined ) { epochFromResponse = ( error as IOdspError ) . serverEpoch ; } await this . checkForEpochError ( error , epochFromResponse , fetchType ) ; throw error ; } ) . catch ( ( error ) => { const fluidError = normalizeError ( error , { props : { XRequestStatsHeader : clientCorrelationId } } ) ; throw fluidError ; } ) ; } param url param fetchOptions param fetchType param addInBody public async fetchArray ( url , fetchOptions , fetchType , addInBody = false ) { const clientCorrelationId = this . formatClientCorrelationId ( ) ; this . addEpochInRequest ( fetchOptions , addInBody , clientCorrelationId ) ; let epochFromResponse ; return this . rateLimiter . schedule ( async ( ) => fetchArray ( url , fetchOptions ) ) . then ( ( response ) => { epochFromResponse = response . headers . get ( "x-fluid-epoch" ) ; this . validateEpochFromResponse ( epochFromResponse , fetchType ) ; response . commonSpoHeaders = { ... response . commonSpoHeaders , "X-RequestStats" : clientCorrelationId } ; return response ; } ) . catch ( async ( error ) => { if ( epochFromResponse === undefined ) { epochFromResponse = ( error as IOdspError ) . serverEpoch ; } await this . checkForEpochError ( error , epochFromResponse , fetchType ) ; throw error ; } ) . catch ( ( error ) => { const fluidError = normalizeError ( error , { props : { XRequestStatsHeader : clientCorrelationId } } ) ; throw fluidError ; } ) ; } private addEpochInRequest ( fetchOptions , addInBody , clientCorrelationId ) { if ( addInBody ) { const headers = { } ; headers [ "X-RequestStats" ] = clientCorrelationId ; if ( this . fluidEpoch !== undefined ) { headers [ "x-fluid-epoch" ] = this . fluidEpoch ; } this . addParamInBody ( fetchOptions , headers ) ; } else { const addHeader = ( key , val ) => { fetchOptions . headers = { ... fetchOptions . headers } ; assert ( fetchOptions . headers !== undefined , 0x282 ) ; fetchOptions . headers [ key ] = val ; } ; addHeader ( "X-RequestStats" , clientCorrelationId ) ; if ( this . fluidEpoch !== undefined ) { addHeader ( "x-fluid-epoch" , this . fluidEpoch ) ; } } } private addParamInBody ( fetchOptions , headers ) { const body = fetchOptions . body ; assert ( typeof body === "string" , 0x21d ) ; const splitBody = body . split ( "\r\n" ) ; const firstLine = splitBody . shift ( ) ; assert ( firstLine !== undefined && firstLine . startsWith ( "--" ) , 0x21e ) ; const formParams = [ firstLine ] ; Object . entries ( headers ) . forEach ( ( [ key , value ] ) => { formParams . push ( `${ key }: ${ value }` ) ; } ) ; splitBody . forEach ( ( value ) => { formParams . push ( value ) ; } ) ; fetchOptions . body = formParams . join ( "\r\n" ) ; } private formatClientCorrelationId ( ) { return `driverId=${ this . driverId }, RequestNumber=${ this . networkCallNumber ++ }` ; } protected validateEpochFromResponse ( epochFromResponse , fetchType , fromCache = false ) { const error = this . checkForEpochErrorCore ( epochFromResponse ) ; if ( error !== undefined ) { throw error ; } if ( epochFromResponse !== undefined ) { if ( this . _fluidEpoch === undefined ) { this . setEpoch ( epochFromResponse , fromCache , fetchType ) ; } } } private async checkForEpochError ( error , epochFromResponse , fetchType , fromCache = false ) { if ( isFluidError ( error ) && error . errorType === DriverErrorType . fileOverwrittenInStorage ) { const epochError = this . checkForEpochErrorCore ( epochFromResponse ) ; if ( epochError !== undefined ) { assert ( isFluidError ( epochError ) , 0x21f ) ; epochError . addTelemetryProperties ( { fromCache , clientEpoch : this . fluidEpoch , fetchType } ) ; this . logger . sendErrorEvent ( { eventName : "fileOverwrittenInStorage" } , epochError ) ; await this . removeEntries ( ) ; throw epochError ; } throw new ThrottlingError ( "coherency409" , error . message , 1 , { [ Odsp409Error ] : true } ) ; } } private checkForEpochErrorCore ( epochFromResponse ) { if ( this . fluidEpoch && epochFromResponse && ( this . fluidEpoch !== epochFromResponse ) ) { return new NonRetryableError ( "epochMismatch" , "Epoch mismatch" , DriverErrorType . fileOverwrittenInStorage ) ; } } private fileEntryFromEntry ( entry ) { return { ... entry , file : this . fileEntry } ; } } export class EpochTrackerWithRedemption extends EpochTracker { private readonly treesLatestDeferral = new Deferred < void > ( ) ; protected validateEpochFromResponse ( epochFromResponse , fetchType , fromCache = false ) { super . validateEpochFromResponse ( epochFromResponse , fetchType , fromCache ) ; this . treesLatestDeferral . resolve ( ) ; } public async get ( entry ) { let result = super . get ( entry ) ; if ( entry . type === snapshotKey ) { result = result . then ( ( value ) => { if ( value !== undefined ) { this . treesLatestDeferral . resolve ( ) ; } return value ; } ) . catch ( ( error ) => { this . treesLatestDeferral . reject ( error ) ; throw error ; } ) ; } return result ; } public async fetchAndParseAsJSON < T > ( url , fetchOptions , fetchType , addInBody = false ) { const completed = this . treesLatestDeferral . isCompleted ; try { return await super . fetchAndParseAsJSON < T > ( url , fetchOptions , fetchType , addInBody ) ; } catch ( error ) { if ( fetchType === "treesLatest" ) { this . treesLatestDeferral . reject ( error ) ; } if ( fetchType !== "joinSession" || error . statusCode < 401 || error . statusCode > 404 || completed ) { throw error ; } } await PerformanceEvent . timedExecAsync ( this . logger , { eventName : "JoinSessionSyncWait" } , async ( event ) => { const timeoutRes = 51 ; let timer ; const timeoutP = new Promise < number > ( ( resolve ) => { timer = setTimeout ( ( ) => { resolve ( timeoutRes ) ; } , 15000 ) ; } ) ; const res = await Promise . race ( [ timeoutP , this . treesLatestDeferral . promise . finally ( ( ) => clearTimeout ( timer ) ) ] ) ; if ( res === timeoutRes ) { event . cancel ( ) ; } } , { start : true , end : true , cancel : "generic" } ) ; return super . fetchAndParseAsJSON < T > ( url , fetchOptions , fetchType , addInBody ) ; } } export interface ICacheAndTracker { cache : IOdspCache ; epochTracker : EpochTracker ; } export function createOdspCacheAndTracker ( persistedCacheArg , nonpersistentCache , fileEntry , logger ) { const epochTracker = new EpochTrackerWithRedemption ( persistedCacheArg , fileEntry , logger ) ; return { cache : { ... nonpersistentCache , persistedCache : epochTracker } , epochTracker } ; }