import { strict as assert } from "assert" ; import { EventEmitter } from "events" ; import { DebugLogger , MockLogger } from "@fluidframework/telemetry-utils" ; import { ISequencedDocumentMessage , MessageType , } from "@fluidframework/protocol-definitions" ; import { IContainerContext , ICriticalContainerError } from "@fluidframework/container-definitions" ; import { MockDeltaManager , MockQuorum } from "@fluidframework/test-runtime-utils" ; import { ContainerRuntime , ScheduleManager } from "../containerRuntime" ; describe ( "Runtime" , ( ) => { describe ( "Container Runtime" , ( ) => { describe ( "ContainerRuntime" , ( ) => { describe ( "orderSequentially" , ( ) => { let containerRuntime ; const containerErrors = [ ] ; const getMockContext = ( ( ) : Partial < IContainerContext > => { return { deltaManager : new MockDeltaManager ( ) , quorum : new MockQuorum ( ) , logger : new MockLogger ( ) , clientDetails : { capabilities : { interactive : true } } , closeFn : ( error ? ) : void => { if ( error !== undefined ) { containerErrors . push ( error ) ; } } } ; } ) ; const getSingleContainerError = ( ) : ICriticalContainerError => { assert . strictEqual ( containerErrors . length , 1 ) ; return containerErrors [ 0 ] ; } ; beforeEach ( async ( ) => { containerRuntime = await ContainerRuntime . load ( getMockContext ( ) as IContainerContext , [ ] , undefined , { summaryOptions : { disableSummaries : true } } ) ; containerErrors . length = 0 ; } ) ; it ( "Can't call flush() inside orderSequentially's callback" , ( ) => { assert . throws ( ( ) => containerRuntime . orderSequentially ( ( ) => containerRuntime . flush ( ) ) ) ; assert . strictEqual ( getSingleContainerError ( ) . errorType , "dataProcessingError" ) ; } ) ; it ( "Can't call flush() inside orderSequentially's callback when nested" , ( ) => { assert . throws ( ( ) => containerRuntime . orderSequentially ( ( ) => containerRuntime . orderSequentially ( ( ) => containerRuntime . flush ( ) ) ) ) ; assert . strictEqual ( getSingleContainerError ( ) . errorType , "dataProcessingError" ) ; } ) ; it ( "Errors propagate to the container" , ( ) => { assert . throws ( ( ) => containerRuntime . orderSequentially ( ( ) => { throw new Error ( "Any" ) ; } ) ) ; const error = getSingleContainerError ( ) ; assert . strictEqual ( error . errorType , "dataProcessingError" ) ; assert . ok ( error . message . includes ( "Any" ) ) ; } ) ; } ) ; } ) ; describe ( "ScheduleManager" , ( ) => { describe ( "Batch processing events" , ( ) => { let batchBegin = 0 ; let batchEnd = 0 ; let sequenceNumber = 0 ; let emitter ; let deltaManager ; let scheduleManager ; beforeEach ( ( ) => { emitter = new EventEmitter ( ) ; deltaManager = new MockDeltaManager ( ) ; deltaManager . inbound . processCallback = ( message ) => { scheduleManager . beforeOpProcessing ( message ) ; scheduleManager . afterOpProcessing ( undefined , message ) ; } ; scheduleManager = new ScheduleManager ( deltaManager , emitter , DebugLogger . create ( "fluid:testScheduleManager" ) ) ; emitter . on ( "batchBegin" , ( ) => { assert . strictEqual ( batchBegin , batchEnd , "Received batchBegin before previous batchEnd" ) ; batchBegin ++ ; } ) ; emitter . on ( "batchEnd" , ( ) => { batchEnd ++ ; assert . strictEqual ( batchBegin , batchEnd , "Received batchEnd without corresponding batchBegin" ) ; } ) ; } ) ; afterEach ( ( ) => { batchBegin = 0 ; batchEnd = 0 ; sequenceNumber = 0 ; } ) ; /**
                 * Pushes single op to the inbound queue. Adds proper sequence numbers to them
                 */ function pushOp ( partialMessage ) { sequenceNumber ++ ; const message = { ... partialMessage , sequenceNumber } ; deltaManager . inbound . push ( message as ISequencedDocumentMessage ) ; } /**
                 * awaits until all ops that could be processed are processed.
                 */ async function processOps ( ) { const inbound = deltaManager . inbound ; while ( ! inbound . paused && inbound . length > 0 ) { await Promise . resolve ( ) ; } } it ( "Single non-batch message" , async ( ) => { const clientId = "test-client" ; const message = { clientId , type : MessageType . Operation } ; pushOp ( message ) ; await processOps ( ) ; assert . strictEqual ( deltaManager . inbound . length , 0 , "Did not process all ops" ) ; assert . strictEqual ( 1 , batchBegin , "Did not receive correct batchBegin events" ) ; assert . strictEqual ( 1 , batchEnd , "Did not receive correct batchEnd events" ) ; } ) ; it ( "Multiple non-batch messages" , async ( ) => { const clientId = "test-client" ; const message = { clientId , type : MessageType . Operation } ; pushOp ( message ) ; pushOp ( message ) ; pushOp ( message ) ; pushOp ( message ) ; pushOp ( message ) ; await processOps ( ) ; assert . strictEqual ( deltaManager . inbound . length , 0 , "Did not process all ops" ) ; assert . strictEqual ( 5 , batchBegin , "Did not receive correct batchBegin events" ) ; assert . strictEqual ( 5 , batchEnd , "Did not receive correct batchEnd events" ) ; } ) ; it ( "Message with non batch-related metadata" , async ( ) => { const clientId = "test-client" ; const message = { clientId , type : MessageType . Operation , metadata : { foo : 1 } } ; pushOp ( message ) ; await processOps ( ) ; assert . strictEqual ( deltaManager . inbound . length , 0 , "Did not process all ops" ) ; assert . strictEqual ( 1 , batchBegin , "Did not receive correct batchBegin event for the batch" ) ; assert . strictEqual ( 1 , batchEnd , "Did not receive correct batchEnd event for the batch" ) ; } ) ; it ( "Messages in a single batch" , async ( ) => { const clientId = "test-client" ; const batchBeginMessage = { clientId , type : MessageType . Operation , metadata : { batch : true } } ; const batchMessage = { clientId , type : MessageType . Operation } ; const batchEndMessage = { clientId , type : MessageType . Operation , metadata : { batch : false } } ; pushOp ( batchBeginMessage ) ; pushOp ( batchMessage ) ; pushOp ( batchMessage ) ; await processOps ( ) ; assert . strictEqual ( deltaManager . inbound . length , 3 , "Some of partial batch ops were processed" ) ; pushOp ( batchEndMessage ) ; await processOps ( ) ; assert . strictEqual ( deltaManager . inbound . length , 0 , "Did not process all ops" ) ; assert . strictEqual ( 1 , batchBegin , "Did not receive correct batchBegin event for the batch" ) ; assert . strictEqual ( 1 , batchEnd , "Did not receive correct batchEnd event for the batch" ) ; } ) ; it ( "two batches" , async ( ) => { const clientId = "test-client" ; const batchBeginMessage = { clientId , type : MessageType . Operation , metadata : { batch : true } } ; const batchMessage = { clientId , type : MessageType . Operation } ; const batchEndMessage = { clientId , type : MessageType . Operation , metadata : { batch : false } } ; await deltaManager . inbound . pause ( ) ; pushOp ( batchBeginMessage ) ; pushOp ( batchMessage ) ; pushOp ( batchMessage ) ; pushOp ( batchEndMessage ) ; pushOp ( batchBeginMessage ) ; pushOp ( batchMessage ) ; pushOp ( batchMessage ) ; assert . strictEqual ( deltaManager . inbound . length , 7 , "none of the batched ops are processed yet" ) ; void deltaManager . inbound . resume ( ) ; await processOps ( ) ; assert . strictEqual ( deltaManager . inbound . length , 3 , "none of the second batch ops are processed yet" ) ; assert . strictEqual ( 1 , batchBegin , "Did not receive correct batchBegin event for the batch" ) ; assert . strictEqual ( 1 , batchEnd , "Did not receive correct batchEnd event for the batch" ) ; pushOp ( batchEndMessage ) ; await processOps ( ) ; assert . strictEqual ( deltaManager . inbound . length , 0 , "processed all ops" ) ; assert . strictEqual ( 2 , batchBegin , "Did not receive correct batchBegin event for the batch" ) ; assert . strictEqual ( 2 , batchEnd , "Did not receive correct batchEnd event for the batch" ) ; } ) ; it ( "non-batched ops followed by batch" , async ( ) => { const clientId = "test-client" ; const batchBeginMessage = { clientId , type : MessageType . Operation , metadata : { batch : true } } ; const batchMessage = { clientId , type : MessageType . Operation } ; const batchEndMessage = { clientId , type : MessageType . Operation , metadata : { batch : false } } ; await deltaManager . inbound . pause ( ) ; pushOp ( batchMessage ) ; pushOp ( batchMessage ) ; pushOp ( batchBeginMessage ) ; pushOp ( batchMessage ) ; pushOp ( batchMessage ) ; await processOps ( ) ; assert . strictEqual ( deltaManager . inbound . length , 5 , "none of the batched ops are processed yet" ) ; void deltaManager . inbound . resume ( ) ; await processOps ( ) ; assert . strictEqual ( deltaManager . inbound . length , 3 , "none of the second batch ops are processed yet" ) ; pushOp ( batchEndMessage ) ; await processOps ( ) ; assert . strictEqual ( deltaManager . inbound . length , 0 , "processed all ops" ) ; assert . strictEqual ( 3 , batchBegin , "Did not receive correct batchBegin event for the batch" ) ; assert . strictEqual ( 3 , batchEnd , "Did not receive correct batchEnd event for the batch" ) ; } ) ; function testWrongBatches ( ) { const clientId1 = "test-client-1" ; const clientId2 = "test-client-2" ; const batchBeginMessage = { clientId : clientId1 , type : MessageType . Operation , metadata : { batch : true } } ; const batchMessage = { clientId : clientId1 , type : MessageType . Operation } ; const messagesToFail = [ { clientId : clientId1 , type : MessageType . NoOp } , batchBeginMessage , { clientId : clientId2 , type : MessageType . Operation } , { clientId : clientId2 , type : MessageType . Operation , metadata : { foo : 1 } } , { clientId : clientId2 , type : MessageType . Operation , metadata : { batch : true } } , ] ; let counter = 0 ; for ( const messageToFail of messagesToFail ) { counter ++ ; it ( `Partial batch messages, case ${ counter }` , async ( ) => { pushOp ( batchBeginMessage ) ; pushOp ( batchMessage ) ; pushOp ( batchMessage ) ; await processOps ( ) ; assert . strictEqual ( deltaManager . inbound . length , 3 , "Some of partial batch ops were processed" ) ; assert . throws ( ( ) => pushOp ( messageToFail ) ) ; assert . strictEqual ( deltaManager . inbound . length , 4 , "Some of batch ops were processed" ) ; assert . strictEqual ( 0 , batchBegin , "Did not receive correct batchBegin event for the batch" ) ; assert . strictEqual ( 0 , batchEnd , "Did not receive correct batchBegin event for the batch" ) ; } ) ; } } testWrongBatches ( ) ; } ) ; } ) ; } ) ; } ) ;