import { ISummarizerNode , ISummarizerNodeConfig , ISummarizeInternalResult , ISummarizeResult , ISummaryTreeWithStats , CreateChildSummarizerNodeParam , CreateSummarizerNodeSource , } from "@fluidframework/runtime-definitions" ; import { ISequencedDocumentMessage , SummaryType , ISnapshotTree , SummaryObject , } from "@fluidframework/protocol-definitions" ; import { ITelemetryLogger } from "@fluidframework/common-definitions" ; import { assert , unreachableCase } from "@fluidframework/common-utils" ; import { mergeStats , convertToSummaryTree , calculateStats } from "../summaryUtils" ; import { ReadAndParseBlob } from "../utils" ; import { decodeSummary , encodeSummary , EncodeSummaryParam , EscapedPath , ICreateChildDetails , IInitialSummary , ISummarizerNodeRootContract , parseSummaryForSubtrees , parseSummaryTreeForSubtrees , RefreshSummaryResult , SummaryNode , } from "./summarizerNodeUtils" ; export interface IRootSummarizerNode extends ISummarizerNode , ISummarizerNodeRootContract { } /**
 * Encapsulates the summarizing work and state of an individual tree node in the
 * summary tree. It tracks changes and allows for optimizations when unchanged, or
 * can allow for fallback summaries to be generated when an error is encountered.
 * Usage is for the root node to call startSummary first to begin tracking a WIP
 * (work in progress) summary. Then all nodes will call summarize to summaries their
 * individual parts. Once completed and uploaded to storage, the root node will call
 * completeSummary or clearSummary to clear the WIP summary tracking state if something
 * went wrong. The SummarizerNodes will track all pending summaries that have been
 * recorded by the completeSummary call. When one of them is acked, the root node should
 * call refreshLatestSummary to inform the tree of SummarizerNodes of the new baseline
 * latest successful summary.
 */ export class SummarizerNode implements IRootSummarizerNode { /**
     * The reference sequence number of the most recent acked summary.
     * Returns 0 if there is not yet an acked summary.
     */ public get referenceSequenceNumber ( ) { return this . _latestSummary ?. referenceSequenceNumber ?? 0 ; } protected readonly children = new Map < string , SummarizerNode > ( ) ; protected readonly pendingSummaries = new Map < string , SummaryNode > ( ) ; private readonly outstandingOps : ISequencedDocumentMessage [ ] = [ ] ; private wipReferenceSequenceNumber : number | undefined ; private wipLocalPaths : { localPath : EscapedPath ; additionalPath ? : EscapedPath ; } | undefined ; private wipSkipRecursion = false ; public startSummary ( referenceSequenceNumber , summaryLogger ) { assert ( this . wipSummaryLogger === undefined , 0x19f ) ; assert ( this . wipReferenceSequenceNumber === undefined , 0x1a0 ) ; this . wipSummaryLogger = summaryLogger ; for ( const child of this . children . values ( ) ) { child . startSummary ( referenceSequenceNumber , this . wipSummaryLogger ) ; } this . wipReferenceSequenceNumber = referenceSequenceNumber ; } public async summarize ( fullTree ) { assert ( this . isTrackingInProgress ( ) , 0x1a1 ) ; assert ( this . wipSummaryLogger !== undefined , 0x1a2 ) ; if ( this . canReuseHandle && ! fullTree && ! this . hasChanged ( ) ) { const latestSummary = this . _latestSummary ; if ( latestSummary !== undefined ) { this . wipLocalPaths = { localPath : latestSummary . localPath , additionalPath : latestSummary . additionalPath } ; this . wipSkipRecursion = true ; const stats = mergeStats ( ) ; stats . handleNodeCount ++ ; return { summary : { type : SummaryType . Handle , handle : latestSummary . fullPath . path , handleType : SummaryType . Tree } , stats } ; } } try { const result = await this . summarizeInternalFn ( fullTree ) ; this . wipLocalPaths = { localPath : EscapedPath . create ( result . id ) } ; if ( result . pathPartsForChildren !== undefined ) { this . wipLocalPaths . additionalPath = EscapedPath . createAndConcat ( result . pathPartsForChildren ) ; } return { summary : result . summary , stats : result . stats } ; } catch ( error ) { if ( this . throwOnError || this . trackingSequenceNumber < this . _changeSequenceNumber ) { throw error ; } const latestSummary = this . _latestSummary ; const initialSummary = this . initialSummary ; let encodeParam ; let localPath ; if ( latestSummary !== undefined ) { encodeParam = { fromSummary : true , summaryNode : latestSummary } ; localPath = latestSummary . localPath ; } else if ( initialSummary ?. summary !== undefined ) { encodeParam = { fromSummary : false , initialSummary : initialSummary . summary } ; localPath = EscapedPath . create ( initialSummary . id ) ; } else { throw error ; } this . wipSummaryLogger . sendErrorEvent ( { eventName : "SummarizingWithBasePlusOps" } , error ) ; const summary = encodeSummary ( encodeParam , this . outstandingOps ) ; this . wipLocalPaths = { localPath , additionalPath : summary . additionalPath } ; this . wipSkipRecursion = true ; return { summary : summary . summary , stats : summary . stats } ; } } /**
     * Complete the WIP summary for the given proposalHandle
     */ public completeSummary ( proposalHandle ) { this . completeSummaryCore ( proposalHandle , undefined , false ) ; } /**
     * Recursive implementation for completeSummary, with additional internal-only parameters
     */ protected completeSummaryCore ( proposalHandle , parentPath , parentSkipRecursion ) { assert ( this . wipSummaryLogger !== undefined , 0x1a3 ) ; assert ( this . wipReferenceSequenceNumber !== undefined , 0x1a4 ) ; let localPathsToUse = this . wipLocalPaths ; if ( parentSkipRecursion ) { const latestSummary = this . _latestSummary ; if ( latestSummary !== undefined ) { localPathsToUse = { localPath : latestSummary . localPath , additionalPath : latestSummary . additionalPath } ; } else { this . clearSummary ( ) ; return ; } } assert ( ! ! localPathsToUse , 0x1a5 ) ; const summary = new SummaryNode ( { ... localPathsToUse , referenceSequenceNumber : this . wipReferenceSequenceNumber , basePath : parentPath } ) ; const fullPathForChildren = summary . fullPathForChildren ; for ( const child of this . children . values ( ) ) { child . completeSummaryCore ( proposalHandle , fullPathForChildren , this . wipSkipRecursion || parentSkipRecursion ) ; } this . pendingSummaries . set ( proposalHandle , summary ) ; this . clearSummary ( ) ; } public clearSummary ( ) { this . wipReferenceSequenceNumber = undefined ; this . wipLocalPaths = undefined ; this . wipSkipRecursion = false ; this . wipSummaryLogger = undefined ; for ( const child of this . children . values ( ) ) { child . clearSummary ( ) ; } } returns public async refreshLatestSummary ( proposalHandle , summaryRefSeq , getSnapshot , readAndParseBlob , correlatedSummaryLogger ) { if ( proposalHandle !== undefined ) { const maybeSummaryNode = this . pendingSummaries . get ( proposalHandle ) ; if ( maybeSummaryNode !== undefined ) { this . refreshLatestSummaryFromPending ( proposalHandle , maybeSummaryNode . referenceSequenceNumber ) ; return { latestSummaryUpdated : true , wasSummaryTracked : true } ; } } if ( this . referenceSequenceNumber >= summaryRefSeq ) { return { latestSummaryUpdated : false } ; } const snapshotTree = await getSnapshot ( ) ; await this . refreshLatestSummaryFromSnapshot ( summaryRefSeq , snapshotTree , undefined , EscapedPath . create ( "" ) , correlatedSummaryLogger , readAndParseBlob ) ; return { latestSummaryUpdated : true , wasSummaryTracked : false , snapshot : snapshotTree } ; } protected refreshLatestSummaryFromPending ( proposalHandle , referenceSequenceNumber ) { const summaryNode = this . pendingSummaries . get ( proposalHandle ) ; if ( summaryNode === undefined ) { assert ( this . _latestSummary === undefined , 0x1a6 ) ; return ; } else { assert ( referenceSequenceNumber === summaryNode . referenceSequenceNumber , 0x1a7 ) ; this . pendingSummaries . delete ( proposalHandle ) ; } this . refreshLatestSummaryCore ( referenceSequenceNumber ) ; this . _latestSummary = summaryNode ; for ( const child of this . children . values ( ) ) { child . refreshLatestSummaryFromPending ( proposalHandle , referenceSequenceNumber ) ; } } protected async refreshLatestSummaryFromSnapshot ( referenceSequenceNumber , snapshotTree , basePath , localPath , correlatedSummaryLogger , readAndParseBlob ) { if ( this . referenceSequenceNumber >= referenceSequenceNumber ) { return ; } this . refreshLatestSummaryCore ( referenceSequenceNumber ) ; const { baseSummary , pathParts } = decodeSummary ( snapshotTree , correlatedSummaryLogger ) ; this . _latestSummary = new SummaryNode ( { referenceSequenceNumber , basePath , localPath } ) ; const { childrenTree , childrenPathPart } = parseSummaryForSubtrees ( baseSummary ) ; if ( childrenPathPart !== undefined ) { pathParts . push ( childrenPathPart ) ; } if ( pathParts . length > 0 ) { this . _latestSummary . additionalPath = EscapedPath . createAndConcat ( pathParts ) ; } const pathForChildren = this . _latestSummary . fullPathForChildren ; await Promise . all ( Array . from ( this . children ) . filter ( ( [ id ] ) => { return childrenTree . trees [ id ] !== undefined ; } ) . map ( async ( [ id , child ] ) => { return child . refreshLatestSummaryFromSnapshot ( referenceSequenceNumber , childrenTree . trees [ id ] , pathForChildren , EscapedPath . create ( id ) , correlatedSummaryLogger , readAndParseBlob ) ; } ) ) ; } private refreshLatestSummaryCore ( referenceSequenceNumber ) { for ( const [ key , value ] of this . pendingSummaries ) { if ( value . referenceSequenceNumber < referenceSequenceNumber ) { this . pendingSummaries . delete ( key ) ; } } while ( this . outstandingOps . length > 0 && this . outstandingOps [ 0 ] . sequenceNumber <= referenceSequenceNumber ) { this . outstandingOps . shift ( ) ; } } public loadBaseSummaryWithoutDifferential ( snapshot ) { const { childrenPathPart } = parseSummaryForSubtrees ( snapshot ) ; if ( childrenPathPart !== undefined && this . _latestSummary !== undefined ) { this . _latestSummary . additionalPath = EscapedPath . create ( childrenPathPart ) ; } } public async loadBaseSummary ( snapshot , readAndParseBlob ) { const decodedSummary = decodeSummary ( snapshot , this . defaultLogger ) ; const outstandingOps = await decodedSummary . getOutstandingOps ( readAndParseBlob ) ; const { childrenPathPart } = parseSummaryForSubtrees ( decodedSummary . baseSummary ) ; if ( childrenPathPart !== undefined ) { decodedSummary . pathParts . push ( childrenPathPart ) ; } if ( decodedSummary . pathParts . length > 0 && this . _latestSummary !== undefined ) { this . _latestSummary . additionalPath = EscapedPath . createAndConcat ( decodedSummary . pathParts ) ; } if ( outstandingOps . length > 0 ) { const newOpsLatestSeq = outstandingOps [ outstandingOps . length - 1 ] . sequenceNumber ; assert ( newOpsLatestSeq <= this . trackingSequenceNumber , 0x1a9 ) ; } return { baseSummary : decodedSummary . baseSummary , outstandingOps } ; } public recordChange ( op ) { const lastOp = this . outstandingOps [ this . outstandingOps . length - 1 ] ; if ( lastOp !== undefined ) { assert ( lastOp . sequenceNumber < op . sequenceNumber , 0x1aa ) ; } this . invalidate ( op . sequenceNumber ) ; this . trackingSequenceNumber = op . sequenceNumber ; this . outstandingOps . push ( op ) ; } public invalidate ( sequenceNumber ) { if ( sequenceNumber > this . _changeSequenceNumber ) { this . _changeSequenceNumber = sequenceNumber ; } } /**
     * True if a change has been recorded with sequence number exceeding
     * the latest successfully acked summary reference sequence number.
     * False implies that the previous summary can be reused.
     */ protected hasChanged ( ) { return this . _changeSequenceNumber > this . referenceSequenceNumber ; } public get latestSummary ( ) : Readonly < SummaryNode > | undefined { return this . _latestSummary ; } private readonly canReuseHandle : boolean ; private readonly throwOnError : boolean ; /**
     * Sequence number of latest tracked op. This updates during recordChange,
     * but not for invalidate since we don't have the op. If this drifts from
     * changeSequenceNumber and we try to create a differential summary we assert.
     */ private trackingSequenceNumber : number ; /**
     * Do not call constructor directly.
     * Use createRootSummarizerNode to create root node, or createChild to create child nodes.
     */ public constructor ( protected readonly defaultLogger , private readonly summarizeInternalFn , config , private _changeSequenceNumber , /** Undefined means created without summary */ private _latestSummary ? , private readonly initialSummary ? , protected wipSummaryLogger ? ) { this . canReuseHandle = config . canReuseHandle ?? true ; this . throwOnError = true ; this . trackingSequenceNumber = this . _changeSequenceNumber ; } public createChild ( /** Summarize function */ summarizeInternalFn , /** Initial id or path part of this node */ id , /**
     * Information needed to create the node.
     * If it is from a base summary, it will assert that a summary has been seen.
     * Attach information if it is created from an attach op.
     */ createParam , config = { } ) { assert ( ! this . children . has ( id ) , 0x1ab ) ; const createDetails = this . getCreateDetailsForChild ( id , createParam ) ; const child = new SummarizerNode ( this . defaultLogger , summarizeInternalFn , config , createDetails . changeSequenceNumber , createDetails . latestSummary , createDetails . initialSummary , this . wipSummaryLogger ) ; this . maybeUpdateChildState ( child ) ; this . children . set ( id , child ) ; return child ; } public getChild ( id ) { return this . children . get ( id ) ; } param id param createParam returns protected getCreateDetailsForChild ( id , createParam ) { let initialSummary ; let latestSummary ; let changeSequenceNumber ; const parentLatestSummary = this . _latestSummary ; switch ( createParam . type ) { case CreateSummarizerNodeSource . FromAttach : { if ( parentLatestSummary !== undefined && createParam . sequenceNumber <= parentLatestSummary . referenceSequenceNumber ) { latestSummary = parentLatestSummary . createForChild ( id ) ; } else { const summary = convertToSummaryTree ( createParam . snapshot ) as ISummaryTreeWithStats ; initialSummary = { sequenceNumber : createParam . sequenceNumber , id , summary } ; } changeSequenceNumber = createParam . sequenceNumber ; break ; } case CreateSummarizerNodeSource . FromSummary : { if ( this . initialSummary === undefined ) { assert ( ! ! parentLatestSummary , 0x1ac ) ; } } case CreateSummarizerNodeSource . Local : { const parentInitialSummary = this . initialSummary ; if ( parentInitialSummary !== undefined ) { let childSummary ; if ( parentInitialSummary . summary !== undefined ) { const { childrenTree } = parseSummaryTreeForSubtrees ( parentInitialSummary . summary . summary ) ; assert ( childrenTree . type === SummaryType . Tree , 0x1d6 ) ; childSummary = childrenTree . tree [ id ] ; } if ( createParam . type === CreateSummarizerNodeSource . FromSummary ) { assert ( ! ! childSummary , 0x1ad ) ; } let childSummaryWithStats ; if ( childSummary !== undefined ) { assert ( childSummary . type === SummaryType . Tree , 0x1ae ) ; childSummaryWithStats = { summary : childSummary , stats : calculateStats ( childSummary ) } ; } initialSummary = { sequenceNumber : parentInitialSummary . sequenceNumber , id , summary : childSummaryWithStats } ; } latestSummary = parentLatestSummary ?. createForChild ( id ) ; changeSequenceNumber = parentLatestSummary ?. referenceSequenceNumber ?? - 1 ; break ; } default : { const type = ( createParam as unknown as CreateChildSummarizerNodeParam ) . type ; unreachableCase ( createParam , `Unexpected CreateSummarizerNodeSource: ${ type }` ) ; } } return { initialSummary , latestSummary , changeSequenceNumber } ; } param child protected maybeUpdateChildState ( child ) { if ( this . isTrackingInProgress ( ) ) { child . wipReferenceSequenceNumber = this . wipReferenceSequenceNumber ; } } /**
     * Tells whether summary tracking is in progress. True if "startSummary" API is called before summarize.
     */ protected isTrackingInProgress ( ) { return this . wipReferenceSequenceNumber !== undefined ; } } param logger param summarizeInternalFn param changeSequenceNumber param referenceSequenceNumber param config export const createRootSummarizerNode = ( logger , summarizeInternalFn , changeSequenceNumber , referenceSequenceNumber , config = { } ) : IRootSummarizerNode => new SummarizerNode ( logger , summarizeInternalFn , config , changeSequenceNumber , referenceSequenceNumber === undefined ? undefined : SummaryNode . createForRoot ( referenceSequenceNumber ) ) ;