import { setTimeout } from "timers/promises" ; import { ReadWriteMutex } from "@miniflare/durable-objects" ; import { triggerPromise } from "@miniflare/shared-test" ; import test from "ava" ; test ( "ReadWriteMutex: runs read closures concurrently" , async ( t ) => { const mutex = new ReadWriteMutex ( ) ; let remaining = 3 ; const [ trigger , promise ] = triggerPromise < void > ( ) ; const closure = async ( ) => { await setTimeout ( ) ; if ( -- remaining === 0 ) trigger ( ) ; await promise ; } ; await Promise . all ( [ mutex . runWithRead ( closure ) , mutex . runWithRead ( closure ) , mutex . runWithRead ( closure ) , ] ) ; t . pass ( ) ; } ) ; test ( "ReadWriteMutex: runs write closures exclusively" , async ( t ) => { const mutex = new ReadWriteMutex ( ) ; const events = [ ] ; await Promise . all ( [ mutex . runWithWrite ( async ( ) => { events . push ( 1 ) ; await setTimeout ( ) ; events . push ( 2 ) ; } ) , mutex . runWithWrite ( async ( ) => { events . push ( 3 ) ; } ) , ] ) ; if ( events [ 0 ] === 1 ) t . deepEqual ( events , [ 1 , 2 , 3 ] ) ; else t . deepEqual ( events , [ 3 , 1 , 2 ] ) ; } ) ; test ( "ReadWriteMutex: read and write locks can be acquired synchronously" , ( t ) => { const mutex = new ReadWriteMutex ( ) ; let acquired = false ; mutex . runWithRead ( ( ) => ( acquired = true ) ) ; t . true ( acquired ) ; acquired = false ; mutex . runWithWrite ( ( ) => ( acquired = true ) ) ; t . true ( acquired ) ; } ) ; test ( "ReadWriteMutex: writer waits for readers to finish, but future readers wait for writer" , async ( t ) => { const mutex = new ReadWriteMutex ( ) ; let writerReleased = false ; const [ read1Trigger , read1Promise ] = triggerPromise < void > ( ) ; const [ read2Trigger , read2Promise ] = triggerPromise < void > ( ) ; void mutex . runWithRead ( ( ) => read1Promise ) ; void mutex . runWithRead ( ( ) => read2Promise ) ; void mutex . runWithWrite ( ( ) => ( writerReleased = true ) ) ; await setTimeout ( ) ; t . false ( writerReleased ) ; let remaining = 2 ; const [ trigger , promise ] = triggerPromise < void > ( ) ; const closure = async ( ) => { if ( -- remaining === 0 ) trigger ( ) ; await promise ; } ; const readPromises = Promise . all ( [ mutex . runWithRead ( closure ) , mutex . runWithRead ( closure ) , ] ) ; read1Trigger ( ) ; await setTimeout ( ) ; t . false ( writerReleased ) ; read2Trigger ( ) ; await setTimeout ( ) ; t . true ( writerReleased ) ; await readPromises ; let readerAcquired = false ; void mutex . runWithRead ( ( ) => ( readerAcquired = true ) ) ; t . true ( readerAcquired ) ; } ) ; test ( "ReadWriteMutex: writers prioritised over readers when releasing" , async ( t ) => { const mutex = new ReadWriteMutex ( ) ; let events = [ ] ; let [ trigger , promise ] = triggerPromise < void > ( ) ; mutex . runWithWrite ( ( ) => promise ) ; let promise2 = mutex . runWithRead ( ( ) => events . push ( 2 ) ) ; let promise1 = mutex . runWithWrite ( ( ) => events . push ( 1 ) ) ; trigger ( ) ; await Promise . all ( [ promise2 , promise1 ] ) ; t . deepEqual ( events , [ 1 , 2 ] ) ; events = [ ] ; [ trigger , promise ] = triggerPromise < void > ( ) ; mutex . runWithRead ( ( ) => promise ) ; promise1 = mutex . runWithWrite ( ( ) => events . push ( 1 ) ) ; promise2 = mutex . runWithRead ( ( ) => events . push ( 2 ) ) ; trigger ( ) ; await setTimeout ( ) ; await Promise . all ( [ promise1 , promise2 ] ) ; t . deepEqual ( events , [ 1 , 2 ] ) ; } ) ;