import { inspect } from "util" ; import { IContextErrorData , IPartitionConfig , IPartitionLambda , IPartitionLambdaConfig , IPartitionLambdaFactory , IQueuedMessage , LambdaCloseType , } from "@fluidframework/server-services-core" ; import { getLumberBaseProperties , Lumberjack } from "@fluidframework/server-services-telemetry" ; import { QueueObject , queue } from "async" ; import * as _ from "lodash" ; import { DocumentContext } from "./documentContext" ; export class DocumentPartition { private readonly q : QueueObject < IQueuedMessage > ; private readonly lambdaP : Promise < IPartitionLambda > ; private lambda : IPartitionLambda | undefined ; private corrupt = false ; private closed = false ; private activityTimeoutTime : number | undefined ; constructor ( factory , config , private readonly tenantId , private readonly documentId , public readonly context , private readonly activityTimeout ) { this . updateActivityTime ( ) ; const documentConfig = { leaderEpoch : config . leaderEpoch , tenantId , documentId } ; this . q = queue ( ( message , callback ) => { try { if ( ! this . corrupt ) { const optionalPromise = this . lambda ! . handler ( message ) ; if ( optionalPromise ) { optionalPromise . then ( callback as any ) . catch ( ( error ) => { this . markAsCorrupt ( error , message ) ; callback ( ) ; } ) ; return ; } } else { this . context . checkpoint ( message ) ; } } catch ( error ) { this . markAsCorrupt ( error , message ) ; } callback ( ) ; } , 1 ) ; this . q . pause ( ) ; this . context . on ( "error" , ( error , errorData ) => { if ( errorData . restart ) { this . close ( LambdaCloseType . Error ) ; } } ) ; this . lambdaP = factory . create ( documentConfig , context , this . updateActivityTime . bind ( this ) ) ; this . lambdaP . then ( ( lambda ) => { this . lambda = lambda ; this . q . resume ( ) ; } , ( error ) => { this . markAsCorrupt ( error ) ; this . q . resume ( ) ; } ) ; } public process ( message ) { if ( this . closed ) { return ; } void this . q . push ( message ) ; this . updateActivityTime ( ) ; } public close ( closeType ) { if ( this . closed ) { return ; } this . closed = true ; this . q . kill ( ) ; if ( this . lambda ) { this . lambda . close ( closeType ) ; } else { this . lambdaP . then ( ( lambda ) => { lambda . close ( closeType ) ; } , ( error ) => { } ) ; } } public isInactive ( now = Date . now ( ) ) { return ! this . context . hasPendingWork ( ) && this . activityTimeoutTime && now > this . activityTimeoutTime ; } /**
     * Marks this document partition as corrupt
     * Future messages will be checkpointed but no real processing will happen
     */ private markAsCorrupt ( error , message ? ) { this . corrupt = true ; this . context . log ?. error ( `Marking document as corrupted due to error: ${ inspect ( error ) }` , { messageMetaData : { documentId : this . documentId , tenantId : this . tenantId } } ) ; Lumberjack . error ( `Marking document as corrupted due to error` , getLumberBaseProperties ( this . documentId , this . tenantId ) , error ) ; this . context . error ( error , { restart : false , tenantId : this . tenantId , documentId : this . documentId } ) ; if ( message ) { this . context . checkpoint ( message ) ; } } private updateActivityTime ( ) { this . activityTimeoutTime = Date . now ( ) + this . activityTimeout ; } }