import { strict as assert } from "assert" ; import { ContainerRuntimeFactoryWithDefaultDataStore } from "@fluidframework/aqueduct" ; import { IContainer , IHostLoader } from "@fluidframework/container-definitions" ; import { ConnectionState , Container , Loader } from "@fluidframework/container-loader" ; import { ContainerMessageType , isRuntimeMessage , unpackRuntimeMessage , } from "@fluidframework/container-runtime" ; import { IContainerRuntime } from "@fluidframework/container-runtime-definitions" ; import { IFluidCodeDetails , IFluidHandle , IFluidLoadable , IRequest } from "@fluidframework/core-interfaces" ; import { LocalDocumentServiceFactory , LocalResolver } from "@fluidframework/local-driver" ; import { SharedMap , SharedDirectory } from "@fluidframework/map" ; import { ISequencedDocumentMessage } from "@fluidframework/protocol-definitions" ; import { IEnvelope , FlushMode , IContainerRuntimeBase } from "@fluidframework/runtime-definitions" ; import { requestFluidObject , createDataStoreFactory } from "@fluidframework/runtime-utils" ; import { ILocalDeltaConnectionServer , LocalDeltaConnectionServer } from "@fluidframework/server-local-server" ; import { SharedString } from "@fluidframework/sequence" ; import { createAndAttachContainer , ITestFluidObject , LoaderContainerTracker , LocalCodeLoader , TestFluidObjectFactory , } from "@fluidframework/test-utils" ; describe ( "Ops on Reconnect" , ( ) => { const documentId = "opsOnReconnectTest" ; const documentLoadUrl = `fluid-test://localhost/${ documentId }` ; const map1Id = "map1Key" ; const map2Id = "map2Key" ; const directoryId = "directoryKey" ; const stringId = "sharedStringKey" ; const codeDetails = { package : "opsOnReconnectTestPackage" , config : { } } ; let urlResolver ; let deltaConnectionServer ; let documentServiceFactory ; let loaderContainerTracker ; let container1 ; let container1Object1 ; let container1Object1Map1 ; let container1Object1Map2 ; let container1Object1Directory ; let container1Object1String ; let receivedValues = [ ] ; /**
     * Waits for the "connected" event from the given container.
     */ async function waitForContainerReconnection ( container ) { await new Promise < void > ( ( resolve ) => container . once ( "connected" , ( ) => resolve ( ) ) ) ; } async function createLoader ( ) { const factory = new TestFluidObjectFactory ( [ [ map1Id , SharedMap . getFactory ( ) ] , [ map2Id , SharedMap . getFactory ( ) ] , [ directoryId , SharedDirectory . getFactory ( ) ] , [ stringId , SharedString . getFactory ( ) ] , ] ) ; const defaultFactory = createDataStoreFactory ( "default" , factory ) ; const dataObject2Factory = createDataStoreFactory ( "dataObject2" , factory ) ; const innerRequestHandler = async ( request , runtime ) => runtime . IFluidHandleContext . resolveHandle ( request ) ; const runtimeFactory = new ContainerRuntimeFactoryWithDefaultDataStore ( defaultFactory , [ [ defaultFactory . type , Promise . resolve ( defaultFactory ) ] , [ dataObject2Factory . type , Promise . resolve ( dataObject2Factory ) ] , ] , undefined , [ innerRequestHandler ] ) ; const codeLoader = new LocalCodeLoader ( [ [ codeDetails , runtimeFactory ] ] ) ; const loader = new Loader ( { urlResolver , documentServiceFactory , codeLoader } ) ; loaderContainerTracker . add ( loader ) ; return loader ; } async function createContainer ( ) { const loader = await createLoader ( ) ; return createAndAttachContainer ( codeDetails , loader , urlResolver . createCreateNewRequest ( documentId ) ) ; } async function setupSecondContainersDataObject ( ) { const loader = await createLoader ( ) ; const container2 = await loader . resolve ( { url : documentLoadUrl } ) ; await waitForContainerReconnection ( container2 ) ; container2 . on ( "op" , ( containerMessage ) => { if ( ! isRuntimeMessage ( containerMessage ) ) { return ; } const message = unpackRuntimeMessage ( containerMessage ) ; if ( message . type === ContainerMessageType . FluidDataStoreOp ) { const envelope = message . contents as IEnvelope ; const address = envelope . contents . content . address ; const content = envelope . contents . content . contents ; const batch = message . metadata ?. batch ; let value1 ; let value2 ; if ( address === stringId ) { value1 = content . pos1 ; value2 = content . seg ; } else { value1 = content . key ; value2 = content . value . value ; } receivedValues . push ( [ value1 , value2 , batch ] ) ; } } ) ; const container2Object1 = await requestFluidObject < ITestFluidObject & IFluidLoadable > ( container2 , "default" ) ; return container2Object1 ; } const flush = ( containerObject ) => ( containerObject . context . containerRuntime as IContainerRuntime ) . flush ( ) ; beforeEach ( async ( ) => { urlResolver = new LocalResolver ( ) ; deltaConnectionServer = LocalDeltaConnectionServer . create ( ) ; documentServiceFactory = new LocalDocumentServiceFactory ( deltaConnectionServer ) ; loaderContainerTracker = new LoaderContainerTracker ( ) ; container1 = await createContainer ( ) as Container ; container1Object1 = await requestFluidObject < ITestFluidObject & IFluidLoadable > ( container1 , "default" ) ; container1Object1 . context . containerRuntime . setFlushMode ( FlushMode . Immediate ) ; container1Object1Map1 = await container1Object1 . getSharedObject < SharedMap > ( map1Id ) ; container1Object1Map2 = await container1Object1 . getSharedObject < SharedMap > ( map2Id ) ; container1Object1Directory = await container1Object1 . getSharedObject < SharedDirectory > ( directoryId ) ; container1Object1String = await container1Object1 . getSharedObject < SharedString > ( stringId ) ; await loaderContainerTracker . ensureSynchronized ( ) ; } ) ; afterEach ( ( ) => { loaderContainerTracker . reset ( ) ; } ) ; describe ( "Ops on Container reconnect" , ( ) => { it ( "can resend ops on reconnection that were sent in disconnected state" , async ( ) => { await setupSecondContainersDataObject ( ) ; container1Object1 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; assert ( container1 . clientId ) ; documentServiceFactory . disconnectClient ( container1 . clientId , "Disconnected for testing" ) ; assert . equal ( container1 . connectionState , ConnectionState . Disconnected ) ; container1Object1Map1 . set ( "key1" , "value1" ) ; container1Object1Map1 . set ( "key2" , "value2" ) ; container1Object1Map2 . set ( "key3" , "value3" ) ; container1Object1Map2 . set ( "key4" , "value4" ) ; container1Object1String . insertText ( 0 , "value5" ) ; flush ( container1Object1 ) ; await waitForContainerReconnection ( container1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const expectedValues = [ [ "key1" , "value1" , true ] , [ "key2" , "value2" , undefined ] , [ "key3" , "value3" , undefined ] , [ "key4" , "value4" , undefined ] , [ 0 , "value5" , false ] , ] ; assert . deepStrictEqual ( receivedValues , expectedValues , "Did not receive the ops that were sent in disconnected state" ) ; } ) ; it ( "can resend ops on reconnection that were sent in Nack'd state" , async ( ) => { await setupSecondContainersDataObject ( ) ; container1Object1 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; assert ( container1 . clientId ) ; documentServiceFactory . nackClient ( container1 . clientId ) ; assert . equal ( container1 . connectionState , ConnectionState . Disconnected ) ; container1Object1Map1 . set ( "key1" , "value1" ) ; container1Object1Map1 . set ( "key2" , "value2" ) ; container1Object1Directory . set ( "key3" , "value3" ) ; container1Object1Directory . set ( "key4" , "value4" ) ; flush ( container1Object1 ) ; await waitForContainerReconnection ( container1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const expectedValues = [ [ "key1" , "value1" , true ] , [ "key2" , "value2" , undefined ] , [ "key3" , "value3" , undefined ] , [ "key4" , "value4" , false ] , ] ; assert . deepStrictEqual ( receivedValues , expectedValues , "Did not receive the ops that were sent in Nack'd state" ) ; } ) ; } ) ; describe ( "Ordering of ops that are sent in disconnected state" , ( ) => { it ( "can resend ops in a dataObject in right order on reconnect" , async ( ) => { await setupSecondContainersDataObject ( ) ; container1Object1 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; assert ( container1 . clientId ) ; documentServiceFactory . disconnectClient ( container1 . clientId , "Disconnected for testing" ) ; assert . equal ( container1 . connectionState , ConnectionState . Disconnected ) ; container1Object1Map1 . set ( "key1" , "value1" ) ; container1Object1Map2 . set ( "key2" , "value2" ) ; container1Object1Directory . set ( "key3" , "value3" ) ; container1Object1Map1 . set ( "key4" , "value4" ) ; container1Object1Map2 . set ( "key5" , "value5" ) ; container1Object1Directory . set ( "key6" , "value6" ) ; flush ( container1Object1 ) ; await waitForContainerReconnection ( container1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const expectedValues = [ [ "key1" , "value1" , true ] , [ "key2" , "value2" , undefined ] , [ "key3" , "value3" , undefined ] , [ "key4" , "value4" , undefined ] , [ "key5" , "value5" , undefined ] , [ "key6" , "value6" , false ] , ] ; assert . deepStrictEqual ( receivedValues , expectedValues , "Did not receive the ops that were sent in disconnected state" ) ; } ) ; it ( "can resend ops in multiple dataObjects in right order on reconnect" , async ( ) => { container1Object1 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; const container1Object2 = await requestFluidObject < ITestFluidObject & IFluidLoadable > ( await container1Object1 . context . containerRuntime . createDataStore ( "dataObject2" ) , "/" ) ; const container1Object2Map1 = await container1Object2 . getSharedObject < SharedMap > ( map1Id ) ; const container1Object2Map2 = await container1Object2 . getSharedObject < SharedMap > ( map2Id ) ; container1Object1Map1 . set ( "dataStore2Key" , container1Object2 . handle ) ; flush ( container1Object1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const container2Object1 = await setupSecondContainersDataObject ( ) ; const container2Object1Map1 = await container2Object1 . getSharedObject < SharedMap > ( map1Id ) ; assert ( container2Object1Map1 ) ; const container2Object2Handle = container2Object1Map1 . get < IFluidHandle < ITestFluidObject & IFluidLoadable > > ( "dataStore2Key" ) ; assert ( container2Object2Handle ) ; const container2Object2 = await container2Object2Handle . get ( ) ; assert . ok ( container2Object2 , "Could not get dataStore2 in the second container" ) ; assert ( container1 . clientId ) ; documentServiceFactory . disconnectClient ( container1 . clientId , "Disconnected for testing" ) ; assert . equal ( container1 . connectionState , ConnectionState . Disconnected ) ; container1Object1Map1 . set ( "key1" , "value1" ) ; container1Object2Map1 . set ( "key2" , "value2" ) ; container1Object1Map2 . set ( "key3" , "value3" ) ; container1Object2Map2 . set ( "key4" , "value4" ) ; container1Object1Map1 . set ( "key5" , "value5" ) ; container1Object2Map1 . set ( "key6" , "value6" ) ; container1Object1Map2 . set ( "key7" , "value7" ) ; container1Object2Map2 . set ( "key8" , "value8" ) ; flush ( container1Object1 ) ; await waitForContainerReconnection ( container1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const expectedValues = [ [ "key1" , "value1" , true ] , [ "key2" , "value2" , undefined ] , [ "key3" , "value3" , undefined ] , [ "key4" , "value4" , undefined ] , [ "key5" , "value5" , undefined ] , [ "key6" , "value6" , undefined ] , [ "key7" , "value7" , undefined ] , [ "key8" , "value8" , false ] , ] ; assert . deepStrictEqual ( receivedValues , expectedValues , "Did not receive the ops that were sent in disconnected state" ) ; } ) ; } ) ; describe ( "Ordering of ops when disconnecting after ops are sent" , ( ) => { it ( "can resend ops in a dataObject in right order on connect" , async ( ) => { await setupSecondContainersDataObject ( ) ; container1Object1Map1 . set ( "key1" , "value1" ) ; container1Object1Map2 . set ( "key2" , "value2" ) ; container1Object1Directory . set ( "key3" , "value3" ) ; container1Object1Map1 . set ( "key4" , "value4" ) ; container1Object1Map2 . set ( "key5" , "value5" ) ; container1Object1Directory . set ( "key6" , "value6" ) ; assert ( container1 . clientId ) ; documentServiceFactory . disconnectClient ( container1 . clientId , "Disconnected for testing" ) ; assert . equal ( container1 . connectionState , ConnectionState . Disconnected ) ; await waitForContainerReconnection ( container1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const expectedValues = [ [ "key1" , "value1" , undefined ] , [ "key2" , "value2" , undefined ] , [ "key3" , "value3" , undefined ] , [ "key4" , "value4" , undefined ] , [ "key5" , "value5" , undefined ] , [ "key6" , "value6" , undefined ] , ] ; assert . deepStrictEqual ( receivedValues , expectedValues , "Did not receive the ops that were sent in disconnected state" ) ; } ) ; it ( "can resend ops in multiple dataObjects in right order on connect" , async ( ) => { container1Object1 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; const container1Object2 = await requestFluidObject < ITestFluidObject & IFluidLoadable > ( await container1Object1 . context . containerRuntime . createDataStore ( "dataObject2" ) , "/" ) ; const container1Object2Map1 = await container1Object2 . getSharedObject < SharedMap > ( map1Id ) ; const container1Object2Map2 = await container1Object2 . getSharedObject < SharedMap > ( map2Id ) ; container1Object1Map1 . set ( "dataStore2Key" , container1Object2 . handle ) ; flush ( container1Object1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const container2Object1 = await setupSecondContainersDataObject ( ) ; const container2Object1Map1 = await container2Object1 . getSharedObject < SharedMap > ( map1Id ) ; const container2Object2Handle = container2Object1Map1 . get < IFluidHandle < ITestFluidObject & IFluidLoadable > > ( "dataStore2Key" ) ; assert ( container2Object2Handle ) ; const container2Object2 = await container2Object2Handle . get ( ) ; assert . ok ( container2Object2 , "Could not get dataStore2 in the second container" ) ; container1Object1Map1 . set ( "key1" , "value1" ) ; container1Object2Map1 . set ( "key2" , "value2" ) ; container1Object1Map2 . set ( "key3" , "value3" ) ; container1Object2Map2 . set ( "key4" , "value4" ) ; container1Object1Map1 . set ( "key5" , "value5" ) ; container1Object2Map1 . set ( "key6" , "value6" ) ; container1Object1Map2 . set ( "key7" , "value7" ) ; container1Object2Map2 . set ( "key8" , "value8" ) ; flush ( container1Object1 ) ; assert ( container1 . clientId ) ; documentServiceFactory . disconnectClient ( container1 . clientId , "Disconnected for testing" ) ; assert . equal ( container1 . connectionState , ConnectionState . Disconnected ) ; await waitForContainerReconnection ( container1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const expectedValues = [ [ "key1" , "value1" , true ] , [ "key2" , "value2" , undefined ] , [ "key3" , "value3" , undefined ] , [ "key4" , "value4" , undefined ] , [ "key5" , "value5" , undefined ] , [ "key6" , "value6" , undefined ] , [ "key7" , "value7" , undefined ] , [ "key8" , "value8" , false ] , ] ; assert . deepStrictEqual ( receivedValues , expectedValues , "Did not receive the ops that were sent in disconnected state" ) ; } ) ; } ) ; describe ( "Op batching on Container reconnect" , ( ) => { it ( "can resend batch ops in a dataObject in right order on connect" , async ( ) => { await setupSecondContainersDataObject ( ) ; container1Object1 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; assert ( container1 . clientId ) ; documentServiceFactory . disconnectClient ( container1 . clientId , "Disconnected for testing" ) ; assert . equal ( container1 . connectionState , ConnectionState . Disconnected ) ; container1Object1 . context . containerRuntime . orderSequentially ( ( ) => { container1Object1Map1 . set ( "key1" , "value1" ) ; container1Object1Map2 . set ( "key2" , "value2" ) ; container1Object1Directory . set ( "key3" , "value3" ) ; } ) ; flush ( container1Object1 ) ; container1Object1 . context . containerRuntime . orderSequentially ( ( ) => { container1Object1Map1 . set ( "key4" , "value4" ) ; container1Object1Map2 . set ( "key5" , "value5" ) ; container1Object1Directory . set ( "key6" , "value6" ) ; } ) ; ( container1Object1 . context . containerRuntime as IContainerRuntime ) . flush ( ) ; await waitForContainerReconnection ( container1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const expectedValues = [ [ "key1" , "value1" , true ] , [ "key2" , "value2" , undefined ] , [ "key3" , "value3" , false ] , [ "key4" , "value4" , true ] , [ "key5" , "value5" , undefined ] , [ "key6" , "value6" , false ] , ] ; assert . deepStrictEqual ( receivedValues , expectedValues , "Did not receive the ops that were sent in disconnected state" ) ; } ) ; it ( "can resend consecutive manually flushed batches in right order on connect" , async ( ) => { await setupSecondContainersDataObject ( ) ; container1Object1 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; assert ( container1 . clientId ) ; documentServiceFactory . disconnectClient ( container1 . clientId , "Disconnected for testing" ) ; assert . equal ( container1 . connectionState , ConnectionState . Disconnected ) ; container1Object1Map1 . set ( "key1" , "value1" ) ; container1Object1Map2 . set ( "key2" , "value2" ) ; container1Object1Directory . set ( "key3" , "value3" ) ; flush ( container1Object1 ) ; container1Object1Map1 . set ( "key4" , "value4" ) ; container1Object1Map2 . set ( "key5" , "value5" ) ; container1Object1Directory . set ( "key6" , "value6" ) ; flush ( container1Object1 ) ; await waitForContainerReconnection ( container1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const expectedValues = [ [ "key1" , "value1" , true ] , [ "key2" , "value2" , undefined ] , [ "key3" , "value3" , false ] , [ "key4" , "value4" , true ] , [ "key5" , "value5" , undefined ] , [ "key6" , "value6" , false ] , ] ; assert . deepStrictEqual ( receivedValues , expectedValues , "Did not receive the ops that were sent in disconnected state" ) ; } ) ; it ( "can resend manually flushed batch in right order on connect" , async ( ) => { await setupSecondContainersDataObject ( ) ; container1Object1 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; assert ( container1 . clientId ) ; documentServiceFactory . disconnectClient ( container1 . clientId , "Disconnected for testing" ) ; assert . equal ( container1 . connectionState , ConnectionState . Disconnected ) ; container1Object1Map1 . set ( "key1" , "value1" ) ; container1Object1Map2 . set ( "key2" , "value2" ) ; container1Object1Directory . set ( "key3" , "value3" ) ; flush ( container1Object1 ) ; await waitForContainerReconnection ( container1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const expectedValues = [ [ "key1" , "value1" , true ] , [ "key2" , "value2" , undefined ] , [ "key3" , "value3" , false ] , ] ; assert . deepStrictEqual ( receivedValues , expectedValues , "Did not receive the ops that were sent in disconnected state" ) ; } ) ; it ( "can resend batch ops after reconnect if disconnect happened during the batch" , async ( ) => { await setupSecondContainersDataObject ( ) ; container1Object1 . context . containerRuntime . setFlushMode ( FlushMode . TurnBased ) ; container1Object1Map1 . set ( "key1" , "value1" ) ; container1Object1Map2 . set ( "key2" , "value2" ) ; container1Object1Directory . set ( "key3" , "value3" ) ; assert ( container1 . clientId ) ; documentServiceFactory . disconnectClient ( container1 . clientId , "Disconnected for testing" ) ; assert . strictEqual ( container1 . deltaManager . outbound . length , 0 ) ; assert . deepStrictEqual ( receivedValues , [ ] , "Values have been sent unexpectedly" ) ; await waitForContainerReconnection ( container1 ) ; await loaderContainerTracker . ensureSynchronized ( ) ; const expectedValues = [ [ "key1" , "value1" , true ] , [ "key2" , "value2" , undefined ] , [ "key3" , "value3" , false ] , ] ; assert . deepStrictEqual ( receivedValues , expectedValues , "Did not receive the ops that were re-sent" ) ; } ) ; } ) ; afterEach ( async ( ) => { await deltaConnectionServer . webSocketServer . close ( ) ; receivedValues = [ ] ; } ) ; } ) ;