import { strict as assert } from "assert" ; import sinon from "sinon" ; import { Deferred } from "@fluidframework/common-utils" ; import { ISequencedDocumentMessage , ISummaryAck , ISummaryConfiguration , ISummaryNack , ISummaryProposal , MessageType , SummaryType , } from "@fluidframework/protocol-definitions" ; import { MockLogger } from "@fluidframework/telemetry-utils" ; import { MockDeltaManager } from "@fluidframework/test-runtime-utils" ; import { neverCancelledSummaryToken } from "../runWhileConnectedCoordinator" ; import { RunningSummarizer } from "../runningSummarizer" ; import { ISummarizerOptions } from "../summarizerTypes" ; import { SummaryCollection } from "../summaryCollection" ; import { SummarizeHeuristicData } from "../summarizerHeuristics" ; describe ( "Runtime" , ( ) => { describe ( "Summarization" , ( ) => { describe ( "RunningSummarizer" , ( ) => { let stopCall ; let runCount ; let fullTreeRunCount ; let refreshLatestAckRunCount ; let clock ; let mockLogger ; let mockDeltaManager ; let summaryCollection ; let summarizer ; const summarizerClientId = "test" ; let lastRefSeq = 0 ; let lastClientSeq ; let lastSummarySeq ; const summaryConfig = { idleTime : 5000 , maxTime : 5000 * 12 , maxOps : 1000 , maxAckWaitTime : 120000 } ; let shouldDeferGenerateSummary = false ; let deferGenerateSummary ; const flushPromises = async ( ) => new Promise ( ( resolve ) => process . nextTick ( resolve ) ) ; async function emitNextOp ( increment = 1 , timestamp = Date . now ( ) ) { lastRefSeq += increment ; const op = { sequenceNumber : lastRefSeq , timestamp } ; summarizer . handleOp ( undefined , op as ISequencedDocumentMessage ) ; mockDeltaManager . emit ( "op" , op ) ; await flushPromises ( ) ; } function emitBroadcast ( timestamp = Date . now ( ) ) { mockDeltaManager . emit ( "op" , { type : MessageType . Summarize , clientId : summarizerClientId , referenceSequenceNumber : lastRefSeq , clientSequenceNumber : -- lastClientSeq , sequenceNumber : -- lastSummarySeq , contents : { handle : "test-broadcast-handle" } , timestamp } ) ; } async function emitAck ( ) { const summaryProposal = { summarySequenceNumber : lastSummarySeq } ; const contents = { handle : "test-ack-handle" , summaryProposal } ; mockDeltaManager . emit ( "op" , { contents , type : MessageType . SummaryAck } ) ; await flushPromises ( ) ; } async function emitNack ( retryAfterSeconds ? ) { const summaryProposal = { summarySequenceNumber : lastSummarySeq } ; const contents = { summaryProposal , retryAfter : retryAfterSeconds , errorMessage : "test-nack" } ; mockDeltaManager . emit ( "op" , { contents , type : MessageType . SummaryNack } ) ; await flushPromises ( ) ; } async function tickAndFlushPromises ( ms ) { clock . tick ( ms ) ; await flushPromises ( ) ; } function assertRunCounts ( expectedTotalRunCount , expectedFullTreeRunCount , expectedRefreshLatestAckRunCount , errorMessage ? ) { const errorPrefix = errorMessage ? `${ errorMessage }: ` : "" ; assert . strictEqual ( runCount , expectedTotalRunCount , `${ errorPrefix }unexpected total run count` ) ; assert . strictEqual ( fullTreeRunCount , expectedFullTreeRunCount , `${ errorPrefix }unexpected fullTree count` ) ; assert . strictEqual ( refreshLatestAckRunCount , expectedRefreshLatestAckRunCount , `${ errorPrefix }unexpected refreshLatestAck count` ) ; } const startRunningSummarizer = async ( summarizerOptions ? ) : Promise < void > => { summarizer = await RunningSummarizer . start ( mockLogger , summaryCollection . createWatcher ( summarizerClientId ) , summaryConfig , async ( options ) => { runCount ++ ; const { fullTree = false , refreshLatestAck = false } = options ; if ( fullTree ) { fullTreeRunCount ++ ; } if ( refreshLatestAck ) { refreshLatestAckRunCount ++ ; } emitBroadcast ( ) ; if ( shouldDeferGenerateSummary ) { deferGenerateSummary = new Deferred < void > ( ) ; await deferGenerateSummary . promise ; deferGenerateSummary = undefined ; } return { stage : "submit" , referenceSequenceNumber : lastRefSeq , generateDuration : 0 , uploadDuration : 0 , submitOpDuration : 0 , summaryTree : { type : SummaryType . Tree , tree : { } } , summaryStats : { treeNodeCount : 0 , blobNodeCount : 0 , handleNodeCount : 0 , totalBlobSize : 0 , dataStoreCount : 0 , summarizedDataStoreCount : 0 , unreferencedBlobSize : 0 } , handle : "test-handle" , clientSequenceNumber : lastClientSeq } as const ; } , new SummarizeHeuristicData ( 0 , { refSequenceNumber : 0 , summaryTime : Date . now ( ) } ) , ( ) => { } , summaryCollection , neverCancelledSummaryToken , ( reason ) => { stopCall ++ ; } , summarizerOptions ) ; } ; before ( ( ) => { clock = sinon . useFakeTimers ( ) ; } ) ; after ( ( ) => { clock . restore ( ) ; } ) ; beforeEach ( async ( ) => { shouldDeferGenerateSummary = false ; deferGenerateSummary = undefined ; clock . reset ( ) ; runCount = 0 ; stopCall = 0 ; fullTreeRunCount = 0 ; refreshLatestAckRunCount = 0 ; lastRefSeq = 0 ; lastClientSeq = - 1000 ; lastSummarySeq = 0 ; mockLogger = new MockLogger ( ) ; mockDeltaManager = new MockDeltaManager ( ) ; summaryCollection = new SummaryCollection ( mockDeltaManager , mockLogger ) ; } ) ; describe ( "Summary Schedule" , ( ) => { beforeEach ( async ( ) => { await startRunningSummarizer ( ) ; } ) ; it ( "Should summarize after configured number of ops when not pending" , async ( ) => { await emitNextOp ( ) ; await emitNextOp ( summaryConfig . maxOps - 1 ) ; assertRunCounts ( 0 , 0 , 0 ) ; await emitNextOp ( 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:GenerateSummary" , summaryGenTag : runCount } , { eventName : "Running:SummaryOp" , summaryGenTag : runCount } , ] ) , "unexpected log sequence" ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; await emitAck ( ) ; assertRunCounts ( 2 , 0 , 0 ) ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:Summarize_end" , summaryGenTag : ( runCount - 1 ) } , { eventName : "Running:GenerateSummary" , summaryGenTag : runCount } , { eventName : "Running:SummaryOp" , summaryGenTag : runCount } , ] ) , "unexpected log sequence" ) ; await emitNextOp ( ) ; assertRunCounts ( 2 , 0 , 0 ) ; assert ( ! mockLogger . matchEvents ( [ { eventName : "Running:Summarize_end" } , ] ) , "No ack expected yet" ) ; } ) ; it ( "Should summarize after configured idle time when not pending" , async ( ) => { await emitNextOp ( ) ; await tickAndFlushPromises ( summaryConfig . idleTime - 1 ) ; assertRunCounts ( 0 , 0 , 0 ) ; await tickAndFlushPromises ( 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; await emitNextOp ( ) ; await tickAndFlushPromises ( summaryConfig . idleTime ) ; assertRunCounts ( 1 , 0 , 0 ) ; await emitAck ( ) ; await emitNextOp ( ) ; await tickAndFlushPromises ( summaryConfig . idleTime ) ; assertRunCounts ( 2 , 0 , 0 ) ; } ) ; it ( "Should summarize after configured active time when not pending" , async ( ) => { const idlesPerActive = Math . floor ( ( summaryConfig . maxTime + 1 ) / ( summaryConfig . idleTime - 1 ) ) ; const remainingTime = ( summaryConfig . maxTime + 1 ) % ( summaryConfig . idleTime - 1 ) ; await emitNextOp ( ) ; for ( let i = 0 ; i < idlesPerActive ; i ++ ) { await tickAndFlushPromises ( summaryConfig . idleTime - 1 ) ; await emitNextOp ( ) ; } await tickAndFlushPromises ( remainingTime - 1 ) ; await emitNextOp ( ) ; assertRunCounts ( 0 , 0 , 0 ) ; await tickAndFlushPromises ( 1 ) ; await emitNextOp ( ) ; assertRunCounts ( 1 , 0 , 0 ) ; for ( let i = 0 ; i < idlesPerActive ; i ++ ) { await tickAndFlushPromises ( summaryConfig . idleTime - 1 ) ; await emitNextOp ( ) ; } await tickAndFlushPromises ( remainingTime ) ; await emitNextOp ( ) ; assertRunCounts ( 1 , 0 , 0 ) ; await emitAck ( ) ; await emitNextOp ( ) ; assertRunCounts ( 2 , 0 , 0 ) ; } ) ; it ( "Should summarize after pending timeout" , async ( ) => { await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; await tickAndFlushPromises ( summaryConfig . maxAckWaitTime - 1 ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; await tickAndFlushPromises ( 1 ) ; await emitNextOp ( ) ; assertRunCounts ( 2 , 0 , 1 ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 2 , 0 , 1 ) ; await emitAck ( ) ; await emitNextOp ( ) ; assertRunCounts ( 3 , 0 , 1 ) ; } ) ; it ( "Should not cause pending ack timeouts using older summary time" , async ( ) => { shouldDeferGenerateSummary = true ; await emitNextOp ( ) ; await emitNextOp ( summaryConfig . maxOps ) ; assertRunCounts ( 1 , 0 , 0 ) ; assert ( deferGenerateSummary !== undefined , "submitSummary was not called" ) ; deferGenerateSummary . resolve ( ) ; await emitAck ( ) ; await tickAndFlushPromises ( summaryConfig . maxAckWaitTime ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 2 , 0 , 0 ) ; await emitNextOp ( ) ; await tickAndFlushPromises ( 1 ) ; await emitNextOp ( ) ; assert ( deferGenerateSummary !== undefined , "submitSummary was not called" ) ; deferGenerateSummary . resolve ( ) ; await emitNextOp ( ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 2 , 0 , 0 ) ; } ) ; it ( "Should summarize one last time before closing >50 ops" , async ( ) => { await emitNextOp ( 51 ) ; const stopP = summarizer . waitStop ( true ) ; await flushPromises ( ) ; await emitAck ( ) ; await stopP ; assertRunCounts ( 1 , 0 , 0 , "should perform lastSummary" ) ; } ) ; it ( "Should not summarize one last time before closing <=50 ops" , async ( ) => { await emitNextOp ( 50 ) ; const stopP = summarizer . waitStop ( true ) ; await flushPromises ( ) ; await emitAck ( ) ; await stopP ; assertRunCounts ( 0 , 0 , 0 , "should not perform lastSummary" ) ; } ) ; } ) ; describe ( "Safe Retries" , ( ) => { beforeEach ( async ( ) => { shouldDeferGenerateSummary = false ; deferGenerateSummary = undefined ; await startRunningSummarizer ( ) ; } ) ; it ( "Should retry on failures" , async ( ) => { await emitNextOp ( ) ; await emitNextOp ( summaryConfig . maxOps - 1 ) ; assertRunCounts ( 0 , 0 , 0 ) ; await emitNextOp ( 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; const retryProps1 = { summaryGenTag : 1 , summarizeAttemptsPerPhase : 1 , summarizeTotalAttempts : 1 , summarizeAttemptPhase : 1 } ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:GenerateSummary" , ... retryProps1 } , { eventName : "Running:SummaryOp" , ... retryProps1 } , ] ) , "unexpected log sequence" ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; await emitNack ( ) ; assertRunCounts ( 2 , 0 , 1 , "retry1 should be refreshLatestAck" ) ; const retryProps2 = { summaryGenTag : 1 , summarizeAttemptsPerPhase : 1 , summarizeTotalAttempts : 2 , summarizeAttemptPhase : 2 } ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:Summarize_cancel" , ... retryProps1 , reason : "summaryNack" } , { eventName : "Running:GenerateSummary" , ... retryProps2 } , { eventName : "Running:SummaryOp" , ... retryProps2 } , ] ) , "unexpected log sequence" ) ; await emitNack ( ) ; await tickAndFlushPromises ( 2 * 60 * 1000 - 1 ) ; assertRunCounts ( 2 , 0 , 1 , "retry2 should not start until after delay" ) ; await tickAndFlushPromises ( 1 ) ; assertRunCounts ( 3 , 0 , 2 , "retry2 should be refreshLatestAck" ) ; const retryProps3 = { summaryGenTag : 1 , summarizeAttemptsPerPhase : 1 , summarizeTotalAttempts : 3 , summarizeAttemptPhase : 3 } ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:Summarize_cancel" , ... retryProps2 , reason : "summaryNack" } , { eventName : "Running:GenerateSummary" , ... retryProps3 } , { eventName : "Running:SummaryOp" , ... retryProps3 } , ] ) , "unexpected log sequence" ) ; await emitNack ( ) ; await tickAndFlushPromises ( 10 * 60 * 1000 - 1 ) ; assertRunCounts ( 3 , 0 , 2 , "retry3 should not start until after delay" ) ; await tickAndFlushPromises ( 1 ) ; assertRunCounts ( 4 , 1 , 3 , "retry3 should be fullTree and refreshLatestAck" ) ; const retryProps4 = { summaryGenTag : 1 , summarizeAttemptsPerPhase : 1 , summarizeTotalAttempts : 4 , summarizeAttemptPhase : 4 } ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:Summarize_cancel" , ... retryProps3 , reason : "summaryNack" } , { eventName : "Running:GenerateSummary" , ... retryProps4 } , { eventName : "Running:SummaryOp" , ... retryProps4 } , ] ) , "unexpected log sequence" ) ; assert . strictEqual ( stopCall , 0 ) ; await emitNack ( ) ; assert . strictEqual ( stopCall , 1 ) ; } ) ; it ( "Should retry after delay on failures with retryAfter" , async ( ) => { await emitNextOp ( ) ; await emitNextOp ( summaryConfig . maxOps - 1 ) ; assertRunCounts ( 0 , 0 , 0 , "too early" ) ; await emitNextOp ( 1 ) ; assertRunCounts ( 1 , 0 , 0 , "normal run" ) ; const retryProps1 = { summaryGenTag : 1 , summarizeAttemptsPerPhase : 1 , summarizeTotalAttempts : 1 , summarizeAttemptPhase : 1 } ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:GenerateSummary" , ... retryProps1 } , { eventName : "Running:SummaryOp" , ... retryProps1 } , ] ) , "unexpected log sequence" ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 1 , 0 , 0 , "waiting for ack/nack" ) ; await emitNack ( 30 ) ; await tickAndFlushPromises ( 30 * 1000 - 1 ) ; assertRunCounts ( 1 , 0 , 0 , "waiting for retryAfter delay" ) ; await tickAndFlushPromises ( 1 ) ; assertRunCounts ( 2 , 0 , 0 , "rerun after retryAfter delay" ) ; const retryProps2 = { summaryGenTag : 1 , summarizeAttemptsPerPhase : 2 , summarizeTotalAttempts : 2 , summarizeAttemptPhase : 1 } ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:Summarize_cancel" , summaryGenTag : 1 , reason : "summaryNack" } , { eventName : "Running:SummarizeAttemptDelay" , ... retryProps2 } , { eventName : "Running:GenerateSummary" , ... retryProps2 } , { eventName : "Running:SummaryOp" , ... retryProps2 } , ] ) , "unexpected log sequence" ) ; await emitNack ( 30 ) ; await tickAndFlushPromises ( 30 * 1000 - 1 ) ; assertRunCounts ( 2 , 0 , 0 , "wait for another retryAfter delay" ) ; await tickAndFlushPromises ( 1 ) ; assertRunCounts ( 3 , 0 , 1 , "retry again with refreshLatestAck" ) ; const retryProps3 = { summaryGenTag : 1 , summarizeAttemptsPerPhase : 1 , summarizeTotalAttempts : 3 , summarizeAttemptPhase : 2 } ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:Summarize_cancel" , ... retryProps2 , reason : "summaryNack" } , { eventName : "Running:SummarizeAttemptDelay" , ... retryProps3 } , { eventName : "Running:GenerateSummary" , ... retryProps3 } , { eventName : "Running:SummaryOp" , ... retryProps3 } , ] ) , "unexpected log sequence" ) ; } ) ; it ( "Should wait on 429 from uploadSummaryWithContext" , async ( ) => { shouldDeferGenerateSummary = true ; await emitNextOp ( ) ; await emitNextOp ( summaryConfig . maxOps ) ; assert ( deferGenerateSummary !== undefined , "submitSummary was not called" ) ; deferGenerateSummary . reject ( { message : "error" , retryAfterSeconds : 30 } ) ; await flushPromises ( ) ; await tickAndFlushPromises ( 30 * 1000 - 1 ) ; assertRunCounts ( 1 , 0 , 0 , "failed upload" ) ; const retryProps1 = { summaryGenTag : 1 , summarizeAttemptsPerPhase : 1 , summarizeTotalAttempts : 1 , summarizeAttemptPhase : 1 } ; const retryProps2 = { summaryGenTag : 1 , summarizeAttemptsPerPhase : 2 , summarizeTotalAttempts : 2 , summarizeAttemptPhase : 1 } ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:Summarize_cancel" , ... retryProps1 } , { eventName : "Running:SummarizeAttemptDelay" , ... retryProps2 } , ] ) , "unexpected log sequence" ) ; shouldDeferGenerateSummary = false ; await tickAndFlushPromises ( 1 ) ; assertRunCounts ( 2 , 0 , 0 , "normal run" ) ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:GenerateSummary" , ... retryProps2 } , { eventName : "Running:SummaryOp" , ... retryProps2 } , ] ) , "unexpected log sequence" ) ; } ) ; } ) ; describe ( "On-demand Summaries" , ( ) => { beforeEach ( async ( ) => { await startRunningSummarizer ( ) ; } ) ; it ( "Should create an on-demand summary" , async ( ) => { await emitNextOp ( 2 ) ; const result = summarizer . summarizeOnDemand ( undefined , { reason : "test" } ) ; const submitResult = await result . summarySubmitted ; assertRunCounts ( 1 , 0 , 0 , "on-demand should run" ) ; assert ( submitResult . success , "on-demand summary should submit" ) ; assert ( submitResult . data . stage === "submit" , "on-demand summary submitted data stage should be submit" ) ; assert . strictEqual ( submitResult . data . referenceSequenceNumber , 2 , "ref seq num" ) ; assert ( submitResult . data . summaryTree !== undefined , "summary tree should exist" ) ; const broadcastResult = await result . summaryOpBroadcasted ; assert ( broadcastResult . success , "summary op should be broadcast" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . referenceSequenceNumber , 2 , "summarize op ref seq num should be same as summary seq" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . sequenceNumber , - 1 , "summarize op seq number should match test negative counter" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . contents . handle , "test-broadcast-handle" , "summarize op handle should be test-broadcast-handle" ) ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:GenerateSummary" , summaryGenTag : runCount } , { eventName : "Running:SummaryOp" , summaryGenTag : runCount } , ] ) , "unexpected log sequence" ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; await emitAck ( ) ; const ackNackResult = await result . receivedSummaryAckOrNack ; assert ( ackNackResult . success , "on-demand summary should succeed" ) ; assert ( ackNackResult . data . summaryAckOp . type === MessageType . SummaryAck , "should be ack" ) ; assert ( ackNackResult . data . summaryAckOp . contents . handle === "test-ack-handle" , "summary ack handle should be test-ack-handle" ) ; } ) ; it ( "Should return already running for on-demand summary" , async ( ) => { await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; let resolved = false ; try { summarizer . summarizeOnDemand ( undefined , { reason : "test" } ) ; resolved = true ; } catch { } await flushPromises ( ) ; assert ( resolved === false , "already running promise should not resolve yet" ) ; } ) ; it ( "On-demand summary should fail on nack" , async ( ) => { await emitNextOp ( 2 ) ; const result = summarizer . summarizeOnDemand ( undefined , { reason : "test" } ) ; const submitResult = await result . summarySubmitted ; assertRunCounts ( 1 , 0 , 0 , "on-demand should run" ) ; assert ( submitResult . success , "on-demand summary should submit" ) ; assert ( submitResult . data . stage === "submit" , "on-demand summary submitted data stage should be submit" ) ; assert . strictEqual ( submitResult . data . referenceSequenceNumber , 2 , "ref seq num" ) ; assert ( submitResult . data . summaryTree !== undefined , "summary tree should exist" ) ; const broadcastResult = await result . summaryOpBroadcasted ; assert ( broadcastResult . success , "summary op should be broadcast" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . referenceSequenceNumber , 2 , "summarize op ref seq num should be same as summary seq" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . sequenceNumber , - 1 , "summarize op seq number should match test negative counter" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . contents . handle , "test-broadcast-handle" , "summarize op handle should be test-broadcast-handle" ) ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:GenerateSummary" , summaryGenTag : runCount } , { eventName : "Running:SummaryOp" , summaryGenTag : runCount } , ] ) , "unexpected log sequence" ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; await emitNack ( ) ; const ackNackResult = await result . receivedSummaryAckOrNack ; assert ( ! ackNackResult . success , "on-demand summary should fail" ) ; assert ( ackNackResult . data ?. summaryNackOp . type === MessageType . SummaryNack , "should be nack" ) ; assert ( ackNackResult . data . summaryNackOp . contents . errorMessage === "test-nack" , "summary nack error should be test-nack" ) ; } ) ; it ( "Should fail an on-demand summary if stopping" , async ( ) => { summarizer . waitStop ( true ) . catch ( ( ) => { } ) ; const [ refreshLatestAck , fullTree ] = [ true , true ] ; const result1 = summarizer . summarizeOnDemand ( undefined , { reason : "test1" } ) ; const result2 = summarizer . summarizeOnDemand ( undefined , { reason : "test2" , refreshLatestAck } ) ; const result3 = summarizer . summarizeOnDemand ( undefined , { reason : "test3" , fullTree } ) ; const result4 = summarizer . summarizeOnDemand ( undefined , { reason : "test4" , refreshLatestAck , fullTree } ) ; const allResults = ( await Promise . all ( [ result1 . summarySubmitted , result1 . summaryOpBroadcasted , result1 . receivedSummaryAckOrNack , result2 . summarySubmitted , result2 . summaryOpBroadcasted , result2 . receivedSummaryAckOrNack , ] ) ) . concat ( await Promise . all ( [ result3 . summarySubmitted , result3 . summaryOpBroadcasted , result3 . receivedSummaryAckOrNack , result4 . summarySubmitted , result4 . summaryOpBroadcasted , result4 . receivedSummaryAckOrNack , ] ) ) ; for ( const result of allResults ) { assert ( ! result . success , "all results should fail" ) ; } } ) ; it ( "Should fail an on-demand summary if disposed" , async ( ) => { summarizer . dispose ( ) ; const [ refreshLatestAck , fullTree ] = [ true , true ] ; const result1 = summarizer . summarizeOnDemand ( undefined , { reason : "test1" } ) ; const result2 = summarizer . summarizeOnDemand ( undefined , { reason : "test2" , refreshLatestAck } ) ; const result3 = summarizer . summarizeOnDemand ( undefined , { reason : "test3" , fullTree } ) ; const result4 = summarizer . summarizeOnDemand ( undefined , { reason : "test4" , refreshLatestAck , fullTree } ) ; const allResults = ( await Promise . all ( [ result1 . summarySubmitted , result1 . summaryOpBroadcasted , result1 . receivedSummaryAckOrNack , result2 . summarySubmitted , result2 . summaryOpBroadcasted , result2 . receivedSummaryAckOrNack , ] ) ) . concat ( await Promise . all ( [ result3 . summarySubmitted , result3 . summaryOpBroadcasted , result3 . receivedSummaryAckOrNack , result4 . summarySubmitted , result4 . summaryOpBroadcasted , result4 . receivedSummaryAckOrNack , ] ) ) ; for ( const result of allResults ) { assert ( ! result . success , "all results should fail" ) ; } } ) ; } ) ; describe ( "Enqueue Summaries" , ( ) => { beforeEach ( async ( ) => { await startRunningSummarizer ( ) ; } ) ; it ( "Should summarize after specified sequence number" , async ( ) => { await emitNextOp ( 2 ) ; const afterSequenceNumber = 9 ; const result = summarizer . enqueueSummarize ( { reason : "test" , afterSequenceNumber } ) ; assert ( result . alreadyEnqueued === undefined , "should not be already enqueued" ) ; await emitNextOp ( 6 ) ; assertRunCounts ( 0 , 0 , 0 , "enqueued should not run yet, still 1 op short" ) ; await emitNextOp ( 1 ) ; assertRunCounts ( 1 , 0 , 0 , "enqueued should run" ) ; const submitResult = await result . summarySubmitted ; assert ( submitResult . success , "enqueued summary should submit" ) ; assert ( submitResult . data . stage === "submit" , "enqueued summary submitted data stage should be submit" ) ; assert . strictEqual ( submitResult . data . referenceSequenceNumber , 9 , "ref seq num" ) ; assert ( submitResult . data . summaryTree !== undefined , "summary tree should exist" ) ; const broadcastResult = await result . summaryOpBroadcasted ; assert ( broadcastResult . success , "summary op should be broadcast" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . referenceSequenceNumber , 9 , "summarize op ref seq num should be same as summary seq" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . sequenceNumber , - 1 , "summarize op seq number should match test negative counter" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . contents . handle , "test-broadcast-handle" , "summarize op handle should be test-broadcast-handle" ) ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:GenerateSummary" , summaryGenTag : runCount } , { eventName : "Running:SummaryOp" , summaryGenTag : runCount } , ] ) , "unexpected log sequence" ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; await emitAck ( ) ; const ackNackResult = await result . receivedSummaryAckOrNack ; assert ( ackNackResult . success , "enqueued summary should succeed" ) ; assert ( ackNackResult . data . summaryAckOp . type === MessageType . SummaryAck , "should be ack" ) ; assert ( ackNackResult . data . summaryAckOp . contents . handle === "test-ack-handle" , "summary ack handle should be test-ack-handle" ) ; } ) ; it ( "Should summarize after specified sequence number after heuristics attempt finishes" , async ( ) => { const afterSequenceNumber = summaryConfig . maxOps * 2 + 10 ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 1 , 0 , 0 ) ; const result = summarizer . enqueueSummarize ( { reason : "test" , afterSequenceNumber } ) ; assert ( result . alreadyEnqueued === undefined , "should not be already enqueued" ) ; let submitRan = false ; result . summarySubmitted . then ( ( ) => { submitRan = true ; } , ( ) => { } ) ; await emitAck ( ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 2 , 0 , 0 ) ; await emitNextOp ( 20 ) ; assert ( submitRan === false , "enqueued summary should not run until 2nd heuristic ack" ) ; await emitAck ( ) ; assert ( ( submitRan as boolean ) === true , "enqueued summary should run" ) ; assertRunCounts ( 3 , 0 , 0 ) ; const submitResult = await result . summarySubmitted ; assert ( submitResult . success , "enqueued summary should submit" ) ; assert ( submitResult . data . stage === "submit" , "enqueued summary submitted data stage should be submit" ) ; const expectedRefSeqNum = summaryConfig . maxOps * 2 + 22 ; assert . strictEqual ( submitResult . data . referenceSequenceNumber , expectedRefSeqNum , "ref seq num" ) ; assert ( submitResult . data . summaryTree !== undefined , "summary tree should exist" ) ; const broadcastResult = await result . summaryOpBroadcasted ; assert ( broadcastResult . success , "summary op should be broadcast" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . referenceSequenceNumber , expectedRefSeqNum , "summarize op ref seq num should be same as summary seq" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . sequenceNumber , - 3 , "summarize op seq number should match test negative counter" ) ; assert . strictEqual ( broadcastResult . data . summarizeOp . contents . handle , "test-broadcast-handle" , "summarize op handle should be test-broadcast-handle" ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 3 , 0 , 0 ) ; await emitAck ( ) ; const ackNackResult = await result . receivedSummaryAckOrNack ; assert ( ackNackResult . success , "enqueued summary should succeed" ) ; assert ( ackNackResult . data . summaryAckOp . type === MessageType . SummaryAck , "should be ack" ) ; assert ( ackNackResult . data . summaryAckOp . contents . handle === "test-ack-handle" , "summary ack handle should be test-ack-handle" ) ; } ) ; it ( "Should reject subsequent enqueued summarize attempt unless overridden" , async ( ) => { await emitNextOp ( 2 ) ; const afterSequenceNumber = 9 ; const result = summarizer . enqueueSummarize ( { reason : "test" , afterSequenceNumber } ) ; assert ( result . alreadyEnqueued === undefined , "should not be already enqueued" ) ; const result2 = summarizer . enqueueSummarize ( { reason : "test-fail" } ) ; assert ( result2 . alreadyEnqueued === true , "should be already enqueued" ) ; assert ( result2 . overridden === undefined , "should not be overridden" ) ; const result3 = summarizer . enqueueSummarize ( { reason : "test-override" , override : true } ) ; assert ( result3 . alreadyEnqueued === true , "should be already enqueued" ) ; assert ( result3 . overridden === true , "should be overridden" ) ; const firstResults = await Promise . all ( [ result . summarySubmitted , result . summaryOpBroadcasted , result . receivedSummaryAckOrNack , ] ) ; for ( const firstResult of firstResults ) { assert ( firstResult . success === false , "should fail because of override" ) ; } await emitAck ( ) ; const newResults = await Promise . all ( [ result3 . summarySubmitted , result3 . summaryOpBroadcasted , result3 . receivedSummaryAckOrNack , ] ) ; for ( const newResult of newResults ) { assert ( newResult . success === true , "should succeed" ) ; } } ) ; it ( "Should fail an enqueue summarize attempt if stopping" , async ( ) => { summarizer . waitStop ( true ) . catch ( ( ) => { } ) ; const result1 = summarizer . enqueueSummarize ( { reason : "test1" } ) ; assert ( result1 . alreadyEnqueued === undefined , "should not be already enqueued" ) ; const result2 = summarizer . enqueueSummarize ( { reason : "test2" , afterSequenceNumber : 123 } ) ; assert ( result2 . alreadyEnqueued === undefined , "should not be already enqueued" ) ; const allResults = await Promise . all ( [ result1 . summarySubmitted , result1 . summaryOpBroadcasted , result1 . receivedSummaryAckOrNack , result2 . summarySubmitted , result2 . summaryOpBroadcasted , result2 . receivedSummaryAckOrNack , ] ) ; for ( const result of allResults ) { assert ( ! result . success , "all results should fail" ) ; } } ) ; it ( "Should fail an enqueue summarize attempt if disposed" , async ( ) => { summarizer . dispose ( ) ; const result1 = summarizer . enqueueSummarize ( { reason : "test1" } ) ; assert ( result1 . alreadyEnqueued === undefined , "should not be already enqueued" ) ; const result2 = summarizer . enqueueSummarize ( { reason : "test2" , afterSequenceNumber : 123 } ) ; assert ( result2 . alreadyEnqueued === undefined , "should not be already enqueued" ) ; const allResults = await Promise . all ( [ result1 . summarySubmitted , result1 . summaryOpBroadcasted , result1 . receivedSummaryAckOrNack , result2 . summarySubmitted , result2 . summaryOpBroadcasted , result2 . receivedSummaryAckOrNack , ] ) ; for ( const result of allResults ) { assert ( ! result . success , "all results should fail" ) ; } } ) ; } ) ; describe ( "Summary Start" , ( ) => { it ( "Should summarize immediately if summary ack is missing at startup" , async ( ) => { assertRunCounts ( 0 , 0 , 0 ) ; const summaryTimestamp = Date . now ( ) ; emitBroadcast ( summaryTimestamp ) ; let startStatus = "starting" ; startRunningSummarizer ( ) . then ( ( ) => startStatus = "started" , ( ) => startStatus = "failed" ) ; await flushPromises ( ) ; assert . strictEqual ( startStatus , "starting" , "RunningSummarizer should still be starting since outstanding summary op" ) ; await emitNextOp ( 1 , summaryTimestamp + summaryConfig . maxAckWaitTime - 1 ) ; assert . strictEqual ( startStatus , "starting" , "RunningSummarizer should still be starting since timestamp is within maxAckWaitTime" ) ; await emitNextOp ( 1 , summaryTimestamp + summaryConfig . maxAckWaitTime ) ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:MissingSummaryAckFoundByOps" } , ] ) , "unexpected log sequence 1" ) ; assert . strictEqual ( startStatus , "started" , "RunningSummarizer should be started from the above op" ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 1 , 0 , 0 , "Should run summarizer once" ) ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:GenerateSummary" , summaryGenTag : runCount } , { eventName : "Running:SummaryOp" , summaryGenTag : runCount } , ] ) , "unexpected log sequence 2" ) ; assert ( ! mockLogger . matchEvents ( [ { eventName : "Running:Summarize_end" } , ] ) , "No ack expected yet" ) ; await emitAck ( ) ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:Summarize_end" , summaryGenTag : runCount , summarizeReason : "maxOps" } , ] ) , "unexpected log sequence 3" ) ; } ) ; } ) ; describe ( "Disabled Heuristics" , ( ) => { it ( "Should not summarize after time or ops" , async ( ) => { await startRunningSummarizer ( { disableHeuristics : true } ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 0 , 0 , 0 , "should not summarize after maxOps" ) ; await tickAndFlushPromises ( summaryConfig . idleTime + 1 ) ; assertRunCounts ( 0 , 0 , 0 , "should not summarize after idleTime" ) ; await tickAndFlushPromises ( summaryConfig . maxTime + 1 ) ; assertRunCounts ( 0 , 0 , 0 , "should not summarize after maxTime" ) ; await emitNextOp ( summaryConfig . maxOps * 3 + 10000 ) ; await tickAndFlushPromises ( summaryConfig . maxTime * 3 + summaryConfig . idleTime * 3 + 10000 ) ; assertRunCounts ( 0 , 0 , 0 , "make extra sure" ) ; } ) ; it ( "Should not summarize before closing" , async ( ) => { await startRunningSummarizer ( { disableHeuristics : true } ) ; await emitNextOp ( 51 ) ; const stopP = summarizer . waitStop ( true ) ; await flushPromises ( ) ; await emitAck ( ) ; await stopP ; assertRunCounts ( 0 , 0 , 0 , "should not perform lastSummary" ) ; } ) ; it ( "Should not summarize immediately if summary ack is missing at startup when disabled" , async ( ) => { assertRunCounts ( 0 , 0 , 0 ) ; const summaryTimestamp = Date . now ( ) ; emitBroadcast ( summaryTimestamp ) ; let startStatus = "starting" ; startRunningSummarizer ( { disableHeuristics : true } ) . then ( ( ) => startStatus = "started" , ( ) => startStatus = "failed" ) ; await flushPromises ( ) ; assert . strictEqual ( startStatus , "starting" , "RunningSummarizer should still be starting since outstanding summary op" ) ; await emitNextOp ( 1 , summaryTimestamp + summaryConfig . maxAckWaitTime - 1 ) ; assert . strictEqual ( startStatus , "starting" , "RunningSummarizer should still be starting since timestamp is within maxAckWaitTime" ) ; await emitNextOp ( 1 , summaryTimestamp + summaryConfig . maxAckWaitTime ) ; assert ( mockLogger . matchEvents ( [ { eventName : "Running:MissingSummaryAckFoundByOps" } , ] ) , "unexpected log sequence 1" ) ; assert . strictEqual ( startStatus , "started" , "RunningSummarizer should be started from the above op" ) ; await emitNextOp ( summaryConfig . maxOps + 1 ) ; assertRunCounts ( 0 , 0 , 0 , "Should not run summarizer" ) ; } ) ; } ) ; } ) ; } ) ; } ) ;