import { strict as assert } from "assert" ; import { ContainerMessageType , isRuntimeMessage } from "@fluidframework/container-runtime" ; import { IContainerRuntime } from "@fluidframework/container-runtime-definitions" ; import { SharedMap } from "@fluidframework/map" ; import { ISequencedDocumentMessage } from "@fluidframework/protocol-definitions" ; import { FlushMode } from "@fluidframework/runtime-definitions" ; import { requestFluidObject } from "@fluidframework/runtime-utils" ; import { ITestFluidObject , ChannelFactoryRegistry , timeoutPromise , ITestObjectProvider , ITestContainerConfig , DataObjectFactoryType , } from "@fluidframework/test-utils" ; import { describeFullCompat , } from "@fluidframework/test-version-utils" ; const map1Id = "map1Key" ; const map2Id = "map2Key" ; const registry = [ [ map1Id , SharedMap . getFactory ( ) ] , [ map2Id , SharedMap . getFactory ( ) ] , ] ; const testContainerConfig = { fluidDataObjectType : DataObjectFactoryType . Test , registry } ; describeFullCompat ( "Batching" , ( getTestObjectProvider ) => { let provider ; beforeEach ( ( ) => { provider = getTestObjectProvider ( ) ; } ) ; let dataObject1 ; let dataObject2 ; let dataObject1map1 ; let dataObject1map2 ; let dataObject2map1 ; let dataObject2map2 ; function setupBatchMessageListener ( dataStore , receivedMessages ) { dataStore . context . containerRuntime . on ( "op" , ( message ) => { if ( isRuntimeMessage ( message ) ) { receivedMessages . push ( message ) ; } } ) ; } function verifyBatchMetadata ( batchMessages ) { const batchCount = batchMessages . length ; assert ( batchCount !== 0 , "No messages in the batch" ) ; const batchBeginMetadata = batchMessages [ 0 ] . metadata ?. batch ; const batchEndMetadata = batchMessages [ batchCount - 1 ] . metadata ?. batch ; if ( batchCount === 1 ) { assert . equal ( batchBeginMetadata , undefined , "Batch with one message should not have batch metadata" ) ; return ; } assert . equal ( batchBeginMetadata , true , "Batch begin metadata not found" ) ; assert . equal ( batchEndMetadata , false , "Batch end metadata not found" ) ; } const filterDatastoreOps = ( messages ) => { return messages . filter ( ( m ) => m . type === ContainerMessageType . FluidDataStoreOp ) ; } ; async function waitForCleanContainers ( ... dataStores ) { return Promise . all ( dataStores . map ( async ( dataStore ) => { const runtime = dataStore . context . containerRuntime as IContainerRuntime ; while ( runtime . isDirty ) { await timeoutPromise ( ( resolve ) => runtime . once ( "batchEnd" , resolve ) ) ; } } ) ) ; } beforeEach ( async ( ) => { const container1 = await provider . makeTestContainer ( testContainerConfig ) ; dataObject1 = await requestFluidObject < ITestFluidObject > ( container1 , "default" ) ; dataObject1 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; dataObject1map1 = await dataObject1 . getSharedObject < SharedMap > ( map1Id ) ; dataObject1map2 = await dataObject1 . getSharedObject < SharedMap > ( map2Id ) ; const container2 = await provider . loadTestContainer ( testContainerConfig ) ; dataObject2 = await requestFluidObject < ITestFluidObject > ( container2 , "default" ) ; dataObject2 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; dataObject2map1 = await dataObject2 . getSharedObject < SharedMap > ( map1Id ) ; dataObject2map2 = await dataObject2 . getSharedObject < SharedMap > ( map2Id ) ; await waitForCleanContainers ( dataObject1 , dataObject2 ) ; await provider . ensureSynchronized ( ) ; } ) ; describe ( "Local ops batch metadata verification" , ( ) => { let dataObject1BatchMessages = [ ] ; let dataObject2BatchMessages = [ ] ; beforeEach ( ( ) => { setupBatchMessageListener ( dataObject1 , dataObject1BatchMessages ) ; setupBatchMessageListener ( dataObject2 , dataObject2BatchMessages ) ; } ) ; describe ( "Automatic batches via orderSequentially" , ( ) => { it ( "can send and receive multiple batch ops correctly" , async ( ) => { dataObject1 . context . containerRuntime . orderSequentially ( ( ) => { dataObject1map1 . set ( "key1" , "value1" ) ; dataObject1map2 . set ( "key2" , "value2" ) ; dataObject1map1 . set ( "key3" , "value3" ) ; dataObject1map2 . set ( "key4" , "value4" ) ; } ) ; await provider . ensureSynchronized ( ) ; assert . equal ( filterDatastoreOps ( dataObject1BatchMessages ) . length , 4 , "Incorrect number of messages received on local client" ) ; assert . equal ( filterDatastoreOps ( dataObject2BatchMessages ) . length , 4 , "Incorrect number of messages received on remote client" ) ; verifyBatchMetadata ( dataObject1BatchMessages ) ; verifyBatchMetadata ( dataObject2BatchMessages ) ; } ) ; it ( "can send and receive single batch op correctly" , async ( ) => { dataObject2 . context . containerRuntime . orderSequentially ( ( ) => { dataObject2map1 . set ( "key1" , "value1" ) ; } ) ; await provider . ensureSynchronized ( ) ; assert . equal ( dataObject1BatchMessages . length , 1 , "Incorrect number of messages received on local client" ) ; assert . equal ( dataObject2BatchMessages . length , 1 , "Incorrect number of messages received on remote client" ) ; verifyBatchMetadata ( dataObject1BatchMessages ) ; verifyBatchMetadata ( dataObject2BatchMessages ) ; } ) ; it ( "can send and receive consecutive batches correctly" , async ( ) => { /**
                 * This test verifies that among other things, the PendingStateManager's algorithm of handling
                 * consecutive batches is correct.
                 */ dataObject2 . context . containerRuntime . orderSequentially ( ( ) => { dataObject2map1 . set ( "key1" , "value1" ) ; dataObject2map2 . set ( "key2" , "value2" ) ; } ) ; ( dataObject2 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; dataObject2 . context . containerRuntime . orderSequentially ( ( ) => { dataObject2map1 . set ( "key3" , "value3" ) ; dataObject2map2 . set ( "key4" , "value4" ) ; } ) ; ( dataObject2 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; await provider . ensureSynchronized ( ) ; assert . equal ( dataObject1BatchMessages . length , 4 , "Incorrect number of messages received on local client" ) ; assert . equal ( dataObject2BatchMessages . length , 4 , "Incorrect number of messages received on remote client" ) ; verifyBatchMetadata ( dataObject1BatchMessages . slice ( 0 , 2 ) ) ; verifyBatchMetadata ( dataObject1BatchMessages . slice ( 2 , 4 ) ) ; verifyBatchMetadata ( dataObject2BatchMessages . slice ( 0 , 2 ) ) ; verifyBatchMetadata ( dataObject2BatchMessages . slice ( 2 , 4 ) ) ; } ) ; it ( "can handle calls to orderSequentially with no batch messages" , async ( ) => { /**
                 * This test verifies that among other things, the PendingStateManager's algorithm of handling batches
                 * with no messages is correct.
                 */ dataObject1 . context . containerRuntime . orderSequentially ( ( ) => { } ) ; await provider . ensureSynchronized ( ) ; assert . equal ( dataObject1BatchMessages . length , 0 , "Incorrect number of messages received on local client" ) ; assert . equal ( dataObject2BatchMessages . length , 0 , "Incorrect number of messages received on remote client" ) ; } ) ; it ( "can handle nested orderSequentially by ignoring inner calls to it" , async ( ) => { dataObject1 . context . containerRuntime . orderSequentially ( ( ) => { dataObject1map1 . set ( "key1" , "value1" ) ; dataObject1 . context . containerRuntime . orderSequentially ( ( ) => { dataObject1map2 . set ( "key2" , "value2" ) ; dataObject1 . context . containerRuntime . orderSequentially ( ( ) => { dataObject1map1 . set ( "key3" , "value3" ) ; } ) ; } ) ; dataObject1map2 . set ( "key4" , "value4" ) ; } ) ; await provider . ensureSynchronized ( ) ; assert . equal ( filterDatastoreOps ( dataObject1BatchMessages ) . length , 4 , "Incorrect number of messages received on local client" ) ; assert . equal ( filterDatastoreOps ( dataObject2BatchMessages ) . length , 4 , "Incorrect number of messages received on remote client" ) ; verifyBatchMetadata ( dataObject1BatchMessages ) ; verifyBatchMetadata ( dataObject2BatchMessages ) ; } ) ; } ) ; describe ( "Manually flushed batches" , ( ) => { it ( "can send and receive multiple batch ops that are manually flushed" , async ( ) => { dataObject1 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; dataObject1map1 . set ( "key1" , "value1" ) ; dataObject1map2 . set ( "key2" , "value2" ) ; dataObject1map1 . set ( "key3" , "value3" ) ; dataObject1map2 . set ( "key4" , "value4" ) ; ( dataObject1 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; await provider . ensureSynchronized ( ) ; assert . equal ( filterDatastoreOps ( dataObject1BatchMessages ) . length , 4 , "Incorrect number of messages received on local client" ) ; assert . equal ( filterDatastoreOps ( dataObject2BatchMessages ) . length , 4 , "Incorrect number of messages received on remote client" ) ; verifyBatchMetadata ( dataObject1BatchMessages ) ; verifyBatchMetadata ( dataObject2BatchMessages ) ; } ) ; it ( "can send and receive single batch op that is manually flushed" , async ( ) => { dataObject2 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; dataObject2map1 . set ( "key1" , "value1" ) ; ( dataObject2 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; await provider . ensureSynchronized ( ) ; assert . equal ( dataObject1BatchMessages . length , 1 , "Incorrect number of messages received on local client" ) ; assert . equal ( dataObject2BatchMessages . length , 1 , "Incorrect number of messages received on remote client" ) ; verifyBatchMetadata ( dataObject1BatchMessages ) ; verifyBatchMetadata ( dataObject2BatchMessages ) ; } ) ; it ( "can send and receive consecutive batches that are manually flushed" , async ( ) => { /**
                 * This test verifies that among other things, the PendingStateManager's algorithm of handling
                 * consecutive batches is correct.
                 */ dataObject2 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; dataObject2map1 . set ( "key1" , "value1" ) ; dataObject2map2 . set ( "key2" , "value2" ) ; ( dataObject2 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; dataObject2map1 . set ( "key3" , "value3" ) ; dataObject2map2 . set ( "key4" , "value4" ) ; ( dataObject2 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; dataObject2map1 . set ( "key5" , "value5" ) ; dataObject2map2 . set ( "key6" , "value6" ) ; ( dataObject2 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; await provider . ensureSynchronized ( ) ; assert . equal ( dataObject1BatchMessages . length , 6 , "Incorrect number of messages received on local client" ) ; assert . equal ( dataObject2BatchMessages . length , 6 , "Incorrect number of messages received on remote client" ) ; verifyBatchMetadata ( dataObject1BatchMessages . slice ( 0 , 2 ) ) ; verifyBatchMetadata ( dataObject1BatchMessages . slice ( 2 , 4 ) ) ; verifyBatchMetadata ( dataObject1BatchMessages . slice ( 4 , 6 ) ) ; verifyBatchMetadata ( dataObject2BatchMessages . slice ( 0 , 2 ) ) ; verifyBatchMetadata ( dataObject2BatchMessages . slice ( 2 , 4 ) ) ; verifyBatchMetadata ( dataObject2BatchMessages . slice ( 4 , 6 ) ) ; } ) ; } ) ; afterEach ( async ( ) => { dataObject1BatchMessages = [ ] ; dataObject2BatchMessages = [ ] ; } ) ; } ) ; describe ( "Document Dirty State" , ( ) => { function verifyDocumentDirtyState ( dataStore , expectedState ) { const dirty = ( dataStore . context . containerRuntime as IContainerRuntime ) . isDirty ; assert . equal ( dirty , expectedState , "The document dirty state is not as expected" ) ; } describe ( "Automatic batches via orderSequentially" , ( ) => { it ( "should clean document dirty state after a batch with single message is sent" , async ( ) => { dataObject1 . context . containerRuntime . orderSequentially ( ( ) => { dataObject1map1 . set ( "key1" , "value1" ) ; } ) ; verifyDocumentDirtyState ( dataObject1 , true ) ; await provider . ensureSynchronized ( ) ; verifyDocumentDirtyState ( dataObject1 , false ) ; } ) ; it ( "should clean document dirty state after a batch with multiple messages is sent" , async ( ) => { dataObject1 . context . containerRuntime . orderSequentially ( ( ) => { dataObject1map1 . set ( "key1" , "value1" ) ; dataObject1map2 . set ( "key2" , "value2" ) ; dataObject1map1 . set ( "key3" , "value3" ) ; } ) ; verifyDocumentDirtyState ( dataObject1 , true ) ; await provider . ensureSynchronized ( ) ; verifyDocumentDirtyState ( dataObject1 , false ) ; } ) ; it ( "should clean document dirty state after consecutive batches are sent" , async ( ) => { dataObject1 . context . containerRuntime . orderSequentially ( ( ) => { dataObject1map1 . set ( "key1" , "value1" ) ; } ) ; dataObject1 . context . containerRuntime . orderSequentially ( ( ) => { dataObject1map2 . set ( "key2" , "value2" ) ; dataObject1map1 . set ( "key3" , "value3" ) ; dataObject1map2 . set ( "key4" , "value4" ) ; } ) ; verifyDocumentDirtyState ( dataObject1 , true ) ; await provider . ensureSynchronized ( ) ; verifyDocumentDirtyState ( dataObject1 , false ) ; } ) ; it ( "should clean document dirty state after batch and non-batch messages are sent" , async ( ) => { dataObject1map1 . set ( "key1" , "value1" ) ; dataObject1 . context . containerRuntime . orderSequentially ( ( ) => { dataObject1map2 . set ( "key2" , "value2" ) ; dataObject1map1 . set ( "key3" , "value3" ) ; dataObject1map2 . set ( "key4" , "value4" ) ; } ) ; dataObject1 . context . containerRuntime . orderSequentially ( ( ) => { dataObject1map1 . set ( "key5" , "value5" ) ; } ) ; dataObject1map1 . set ( "key5" , "value5" ) ; verifyDocumentDirtyState ( dataObject1 , true ) ; await provider . ensureSynchronized ( ) ; verifyDocumentDirtyState ( dataObject1 , false ) ; } ) ; } ) ; describe ( "Manually flushed batches" , ( ) => { it ( "should clean document dirty state after a batch with single message is flushed" , async ( ) => { dataObject1map1 . set ( "key1" , "value1" ) ; ( dataObject1 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; verifyDocumentDirtyState ( dataObject1 , true ) ; await provider . ensureSynchronized ( ) ; verifyDocumentDirtyState ( dataObject1 , false ) ; } ) ; it ( "should clean document dirty state after a batch with multiple messages is flushed" , async ( ) => { dataObject1map1 . set ( "key1" , "value1" ) ; dataObject1map2 . set ( "key2" , "value2" ) ; dataObject1map1 . set ( "key3" , "value3" ) ; ( dataObject1 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; verifyDocumentDirtyState ( dataObject1 , true ) ; await provider . ensureSynchronized ( ) ; verifyDocumentDirtyState ( dataObject1 , false ) ; } ) ; it ( "should clean document dirty state after consecutive batches are flushed" , async ( ) => { dataObject1map1 . set ( "key1" , "value1" ) ; ( dataObject1 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; dataObject1map2 . set ( "key2" , "value2" ) ; dataObject1map1 . set ( "key3" , "value3" ) ; dataObject1map2 . set ( "key4" , "value4" ) ; ( dataObject1 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; verifyDocumentDirtyState ( dataObject1 , true ) ; await provider . ensureSynchronized ( ) ; verifyDocumentDirtyState ( dataObject1 , false ) ; } ) ; it ( "should clean document dirty state after batch and non-batch messages are flushed" , async ( ) => { dataObject1map1 . set ( "key1" , "value1" ) ; dataObject1map2 . set ( "key2" , "value2" ) ; dataObject1map1 . set ( "key3" , "value3" ) ; dataObject1map2 . set ( "key4" , "value4" ) ; ( dataObject1 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; dataObject1map1 . set ( "key5" , "value5" ) ; ( dataObject1 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; dataObject1map1 . set ( "key5" , "value5" ) ; verifyDocumentDirtyState ( dataObject1 , true ) ; await provider . ensureSynchronized ( ) ; verifyDocumentDirtyState ( dataObject1 , false ) ; } ) ; } ) ; } ) ; } ) ;