Synchronize sync workers

engine/src/main/java/com/google/android/fhir/sync/FhirSynchronizer.kt

@@ -25,6 +25,8 @@ import java.time.OffsetDateTime
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import org.hl7.fhir.r4.model.ResourceType
 
 enum class SyncOperation {
@@ -79,18 +81,20 @@ internal class FhirSynchronizer(
   }
 
   suspend fun synchronize(): SyncJobStatus {
-    setSyncState(SyncJobStatus.Started())
-
-    return listOf(download(), upload())
-      .filterIsInstance<SyncResult.Error>()
-      .flatMap { it.exceptions }
-      .let {
-        if (it.isEmpty()) {
-          setSyncState(SyncResult.Success())
-        } else {
-          setSyncState(SyncResult.Error(it))
+    mutex.withLock(mutexOwner) {
+      setSyncState(SyncJobStatus.Started())
+
+      return listOf(download(), upload())
+        .filterIsInstance<SyncResult.Error>()
+        .flatMap { it.exceptions }
+        .let {
+          if (it.isEmpty()) {
+            setSyncState(SyncResult.Success())
+          } else {
+            setSyncState(SyncResult.Error(it))
+          }
         }
-      }
+    }
   }
 
   private suspend fun download(): SyncResult {
@@ -141,4 +145,19 @@ internal class FhirSynchronizer(
       SyncResult.Error(exceptions)
     }
   }
+
+  companion object {
+    private val mutex = Mutex()
+    private var mutexOwner: Any? = null
+
+    /**
+     * This is used in testing. In testing the [mutexOwner] should be non-null to catch an
+     * [IllegalStateException] thrown when trying to re-synchronize. In production the [mutexOwner]
+     * should be null so that successive synchronize requests are suspended and resumed when lock is
+     * released.
+     */
+    fun setMutexOwner(owner: Any) {
+      mutexOwner = owner
+    }
+  }
 }

engine/src/test/java/com/google/android/fhir/sync/FhirSynchronizerTest.kt

@@ -23,6 +23,7 @@ import com.google.android.fhir.sync.upload.UploadSyncResult
 import com.google.android.fhir.sync.upload.Uploader
 import com.google.android.fhir.testing.TestFhirEngineImpl
 import com.google.common.truth.Truth.assertThat
+import java.lang.IllegalStateException
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.launch
@@ -53,8 +54,11 @@ class FhirSynchronizerTest {
 
   private lateinit var fhirSynchronizer: FhirSynchronizer
 
+  private val mutexOwner = Object()
+
   @Before
   fun setUp() {
+    FhirSynchronizer.setMutexOwner(mutexOwner)
     MockitoAnnotations.openMocks(this)
     fhirSynchronizer =
       FhirSynchronizer(
@@ -142,4 +146,36 @@ class FhirSynchronizerTest {
       assertThat(result).isInstanceOf(SyncJobStatus.Failed::class.java)
       assertThat(listOf(error)).isEqualTo((result as SyncJobStatus.Failed).exceptions)
     }
+
+  @Test
+  fun `re-synchronize should emit IllegalStateException on trying to acquire mutex lock`() =
+    runTest(UnconfinedTestDispatcher()) {
+      `when`(downloader.download())
+        .thenReturn(
+          flowOf(DownloadState.Success(listOf(), 0, 0)),
+        )
+      `when`(uploader.upload(any()))
+        .thenReturn(
+          UploadSyncResult.Success(
+            listOf(),
+          ),
+        )
+
+      backgroundScope.launch {
+        fhirSynchronizer.syncState.collect {
+          if (it is SyncJobStatus.Started) {
+            try {
+              fhirSynchronizer.synchronize()
+            } catch (e: Exception) {
+              assertThat(e).isInstanceOf(IllegalStateException::class.java)
+              assertThat(e.localizedMessage).isNotNull()
+              assertThat(e.localizedMessage)
+                .isEqualTo("This mutex is already locked by the specified owner: $mutexOwner")
+            }
+          }
+        }
+      }
+
+      val result = fhirSynchronizer.synchronize()
+    }
 }