From be94dd532676a79cfa7723931a3f39d92722f0c8 Mon Sep 17 00:00:00 2001
From: santosh-pingle <86107848+santosh-pingle@users.noreply.github.com>
Date: Wed, 29 Jan 2025 14:27:14 +0530
Subject: [PATCH] api support for sync job cancellation.  (#2717)

* cancel api to cancel the sync job.

* remove  commented code.

* datastore to store uuid.

* code cleanup.

---------

Co-authored-by: Santosh Pingle <spingle@google.com>
---
 .../android/fhir/demo/PeriodicSyncFragment.kt |  28 ++-
 .../fhir/demo/PeriodicSyncViewModel.kt        |  45 ++--
 .../google/android/fhir/demo/SyncFragment.kt  |  29 ++-
 .../fhir/demo/SyncFragmentViewModel.kt        |  12 +-
 demo/src/main/res/layout/periodic_sync.xml    |  38 +++
 demo/src/main/res/layout/sync.xml             |  18 ++
 .../android/fhir/sync/SyncInstrumentedTest.kt | 231 +++++++++++-------
 .../google/android/fhir/sync/FhirDataStore.kt |  39 ++-
 .../java/com/google/android/fhir/sync/Sync.kt | 187 ++++++++------
 9 files changed, 441 insertions(+), 186 deletions(-)

diff --git a/demo/src/main/java/com/google/android/fhir/demo/PeriodicSyncFragment.kt b/demo/src/main/java/com/google/android/fhir/demo/PeriodicSyncFragment.kt
index 4fc6d3c830..912ebbf517 100644
--- a/demo/src/main/java/com/google/android/fhir/demo/PeriodicSyncFragment.kt
+++ b/demo/src/main/java/com/google/android/fhir/demo/PeriodicSyncFragment.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 Google LLC
+ * Copyright 2024-2025 Google LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@ import android.view.LayoutInflater
 import android.view.MenuItem
 import android.view.View
 import android.view.ViewGroup
+import android.widget.Button
 import android.widget.ProgressBar
 import android.widget.TextView
 import androidx.appcompat.app.AppCompatActivity
@@ -48,6 +49,7 @@ class PeriodicSyncFragment : Fragment() {
     setUpActionBar()
     setHasOptionsMenu(true)
     refreshPeriodicSynUi()
+    setUpSyncButtons(view)
   }
 
   override fun onOptionsItemSelected(item: MenuItem): Boolean {
@@ -67,6 +69,30 @@ class PeriodicSyncFragment : Fragment() {
     }
   }
 
+  private fun setUpSyncButtons(view: View) {
+    val syncNowButton = view.findViewById<Button>(R.id.sync_now_button)
+    val cancelSyncButton = view.findViewById<Button>(R.id.cancel_sync_button)
+    syncNowButton.apply {
+      setOnClickListener {
+        periodicSyncViewModel.collectPeriodicSyncJobStatus()
+        toggleButtonVisibility(hiddenButton = syncNowButton, visibleButton = cancelSyncButton)
+        visibility = View.GONE
+      }
+    }
+    cancelSyncButton.apply {
+      setOnClickListener {
+        periodicSyncViewModel.cancelPeriodicSyncJob()
+        toggleButtonVisibility(hiddenButton = cancelSyncButton, visibleButton = syncNowButton)
+        visibility = View.GONE
+      }
+    }
+  }
+
+  private fun toggleButtonVisibility(hiddenButton: View, visibleButton: View) {
+    hiddenButton.visibility = View.GONE
+    visibleButton.visibility = View.VISIBLE
+  }
+
   private fun refreshPeriodicSynUi() {
     viewLifecycleOwner.lifecycleScope.launch {
       viewLifecycleOwner.lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
diff --git a/demo/src/main/java/com/google/android/fhir/demo/PeriodicSyncViewModel.kt b/demo/src/main/java/com/google/android/fhir/demo/PeriodicSyncViewModel.kt
index f5a230bd10..f68f85b011 100644
--- a/demo/src/main/java/com/google/android/fhir/demo/PeriodicSyncViewModel.kt
+++ b/demo/src/main/java/com/google/android/fhir/demo/PeriodicSyncViewModel.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 Google LLC
+ * Copyright 2024-2025 Google LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -31,36 +31,43 @@ import com.google.android.fhir.sync.RepeatInterval
 import com.google.android.fhir.sync.Sync
 import com.google.android.fhir.sync.SyncJobStatus
 import java.util.concurrent.TimeUnit
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.SharedFlow
-import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.shareIn
 import kotlinx.coroutines.launch
+import timber.log.Timber
 
 class PeriodicSyncViewModel(application: Application) : AndroidViewModel(application) {
 
-  val pollPeriodicSyncJobStatus: SharedFlow<PeriodicSyncJobStatus> =
-    Sync.periodicSync<DemoFhirSyncWorker>(
-        application.applicationContext,
+  private val _uiStateFlow = MutableStateFlow(PeriodicSyncUiState())
+  val uiStateFlow: StateFlow<PeriodicSyncUiState> = _uiStateFlow
+
+  private val _pollPeriodicSyncJobStatus = MutableSharedFlow<PeriodicSyncJobStatus>(replay = 10)
+
+  init {
+    viewModelScope.launch { initializePeriodicSync() }
+  }
+
+  private suspend fun initializePeriodicSync() {
+    val periodicSyncJobStatusFlow =
+      Sync.periodicSync<DemoFhirSyncWorker>(
+        context = getApplication<Application>().applicationContext,
         periodicSyncConfiguration =
           PeriodicSyncConfiguration(
             syncConstraints = Constraints.Builder().build(),
             repeat = RepeatInterval(interval = 15, timeUnit = TimeUnit.MINUTES),
           ),
       )
-      .shareIn(viewModelScope, SharingStarted.Eagerly, 10)
-
-  private val _uiStateFlow = MutableStateFlow(PeriodicSyncUiState())
-  val uiStateFlow: StateFlow<PeriodicSyncUiState> = _uiStateFlow
 
-  init {
-    collectPeriodicSyncJobStatus()
+    periodicSyncJobStatusFlow.collect { status -> _pollPeriodicSyncJobStatus.emit(status) }
   }
 
-  private fun collectPeriodicSyncJobStatus() {
+  fun collectPeriodicSyncJobStatus() {
     viewModelScope.launch {
-      pollPeriodicSyncJobStatus.collect { periodicSyncJobStatus ->
+      _pollPeriodicSyncJobStatus.collect { periodicSyncJobStatus ->
+        Timber.d(
+          "currentSyncJobStatus: ${periodicSyncJobStatus.currentSyncJobStatus}  lastSyncJobStatus ${periodicSyncJobStatus.lastSyncJobStatus}",
+        )
         val lastSyncStatus = getLastSyncStatus(periodicSyncJobStatus.lastSyncJobStatus)
         val lastSyncTime = getLastSyncTime(periodicSyncJobStatus.lastSyncJobStatus)
         val currentSyncStatus =
@@ -83,6 +90,14 @@ class PeriodicSyncViewModel(application: Application) : AndroidViewModel(applica
     }
   }
 
+  fun cancelPeriodicSyncJob() {
+    viewModelScope.launch {
+      Sync.cancelPeriodicSync<DemoFhirSyncWorker>(
+        getApplication<FhirApplication>().applicationContext,
+      )
+    }
+  }
+
   private fun getLastSyncStatus(lastSyncJobStatus: LastSyncJobStatus?): String? {
     return when (lastSyncJobStatus) {
       is LastSyncJobStatus.Succeeded ->
diff --git a/demo/src/main/java/com/google/android/fhir/demo/SyncFragment.kt b/demo/src/main/java/com/google/android/fhir/demo/SyncFragment.kt
index 4aaafe8950..f0fa743699 100644
--- a/demo/src/main/java/com/google/android/fhir/demo/SyncFragment.kt
+++ b/demo/src/main/java/com/google/android/fhir/demo/SyncFragment.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 Google LLC
+ * Copyright 2024-2025 Google LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -30,6 +30,7 @@ import androidx.fragment.app.viewModels
 import androidx.navigation.fragment.NavHostFragment
 import com.google.android.fhir.demo.extensions.launchAndRepeatStarted
 import com.google.android.fhir.sync.CurrentSyncJobStatus
+import timber.log.Timber
 
 class SyncFragment : Fragment() {
   private val syncFragmentViewModel: SyncFragmentViewModel by viewModels()
@@ -49,6 +50,9 @@ class SyncFragment : Fragment() {
     view.findViewById<Button>(R.id.sync_now_button).setOnClickListener {
       syncFragmentViewModel.triggerOneTimeSync()
     }
+    view.findViewById<Button>(R.id.cancel_sync_button).setOnClickListener {
+      syncFragmentViewModel.cancelOneTimeSyncWork()
+    }
     observeLastSyncTime()
     launchAndRepeatStarted(
       { syncFragmentViewModel.pollState.collect(::currentSyncJobStatus) },
@@ -73,24 +77,41 @@ class SyncFragment : Fragment() {
   }
 
   private fun currentSyncJobStatus(currentSyncJobStatus: CurrentSyncJobStatus) {
-    requireView().findViewById<TextView>(R.id.current_status).text =
+    Timber.d("currentSyncJobStatus: $currentSyncJobStatus")
+    // Update status text
+    val statusTextView = requireView().findViewById<TextView>(R.id.current_status)
+    statusTextView.text =
       getString(R.string.current_status, currentSyncJobStatus::class.java.simpleName)
 
-    // Update progress indicator visibility and handle status-specific actions
+    // Get views once to avoid repeated lookups
     val syncIndicator = requireView().findViewById<ProgressBar>(R.id.sync_indicator)
+    val syncNowButton = requireView().findViewById<Button>(R.id.sync_now_button)
+    val cancelSyncButton = requireView().findViewById<Button>(R.id.cancel_sync_button)
+
+    // Update view states based on sync status
     when (currentSyncJobStatus) {
       is CurrentSyncJobStatus.Running -> {
         syncIndicator.visibility = View.VISIBLE
+        syncNowButton.visibility = View.GONE
+        cancelSyncButton.visibility = View.VISIBLE
       }
       is CurrentSyncJobStatus.Succeeded -> {
         syncIndicator.visibility = View.GONE
         syncFragmentViewModel.updateLastSyncTimestamp(currentSyncJobStatus.timestamp)
+        syncNowButton.visibility = View.VISIBLE
+        cancelSyncButton.visibility = View.GONE
       }
       is CurrentSyncJobStatus.Failed,
+      is CurrentSyncJobStatus.Cancelled, -> {
+        syncIndicator.visibility = View.GONE
+        syncNowButton.visibility = View.VISIBLE
+        cancelSyncButton.visibility = View.GONE
+      }
       is CurrentSyncJobStatus.Enqueued,
-      is CurrentSyncJobStatus.Cancelled,
       is CurrentSyncJobStatus.Blocked, -> {
         syncIndicator.visibility = View.GONE
+        syncNowButton.visibility = View.GONE
+        cancelSyncButton.visibility = View.VISIBLE
       }
     }
   }
diff --git a/demo/src/main/java/com/google/android/fhir/demo/SyncFragmentViewModel.kt b/demo/src/main/java/com/google/android/fhir/demo/SyncFragmentViewModel.kt
index b1f4063b97..cdd16a0e61 100644
--- a/demo/src/main/java/com/google/android/fhir/demo/SyncFragmentViewModel.kt
+++ b/demo/src/main/java/com/google/android/fhir/demo/SyncFragmentViewModel.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2024 Google LLC
+ * Copyright 2023-2025 Google LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -53,15 +53,21 @@ class SyncFragmentViewModel(application: Application) : AndroidViewModel(applica
   val pollState: SharedFlow<CurrentSyncJobStatus> =
     _oneTimeSyncTrigger
       .flatMapLatest {
-        Sync.oneTimeSync<DemoFhirSyncWorker>(context = application.applicationContext)
+        Sync.oneTimeSync<DemoFhirSyncWorker>(
+          context = application.applicationContext,
+        )
       }
       .map { it }
-      .shareIn(viewModelScope, SharingStarted.Eagerly, 0)
+      .shareIn(viewModelScope, SharingStarted.Eagerly, replay = 0)
 
   fun triggerOneTimeSync() {
     viewModelScope.launch { _oneTimeSyncTrigger.emit(true) }
   }
 
+  fun cancelOneTimeSyncWork() {
+    viewModelScope.launch { Sync.cancelOneTimeSync<DemoFhirSyncWorker>(getApplication()) }
+  }
+
   /** Emits last sync time. */
   fun updateLastSyncTimestamp(lastSync: OffsetDateTime? = null) {
     val formatter =
diff --git a/demo/src/main/res/layout/periodic_sync.xml b/demo/src/main/res/layout/periodic_sync.xml
index 3a3b835342..911f40cfa9 100644
--- a/demo/src/main/res/layout/periodic_sync.xml
+++ b/demo/src/main/res/layout/periodic_sync.xml
@@ -87,4 +87,42 @@
         android:layout_marginTop="8dp"
     />
 
+  <!-- Sync Now Button -->
+  <Button
+        android:id="@+id/sync_now_button"
+        android:layout_width="wrap_content"
+        android:layout_height="wrap_content"
+        app:layout_constraintStart_toStartOf="parent"
+        app:layout_constraintEnd_toEndOf="parent"
+        app:layout_constraintTop_toBottomOf="@id/progress_percentage_label"
+        app:layout_constraintBottom_toBottomOf="parent"
+        android:layout_marginTop="16dp"
+        android:layout_marginBottom="64dp"
+        android:paddingLeft="24dp"
+        android:paddingRight="24dp"
+        android:text="Sync Now"
+        android:backgroundTint="?attr/colorPrimary"
+        android:textColor="?attr/colorOnPrimary"
+        app:layout_goneMarginTop="16dp"
+    />
+
+  <Button
+        android:id="@+id/cancel_sync_button"
+        android:layout_width="wrap_content"
+        android:layout_height="wrap_content"
+        app:layout_constraintStart_toStartOf="parent"
+        app:layout_constraintEnd_toEndOf="parent"
+        app:layout_constraintTop_toBottomOf="@id/progress_percentage_label"
+        app:layout_constraintBottom_toBottomOf="parent"
+        android:layout_marginTop="16dp"
+        android:layout_marginBottom="64dp"
+        android:paddingLeft="24dp"
+        android:paddingRight="24dp"
+        android:text="Cancel Sync"
+        android:backgroundTint="?attr/colorPrimary"
+        android:textColor="?attr/colorOnPrimary"
+        android:visibility="gone"
+        app:layout_goneMarginTop="16dp"
+    />
+
 </androidx.constraintlayout.widget.ConstraintLayout>
diff --git a/demo/src/main/res/layout/sync.xml b/demo/src/main/res/layout/sync.xml
index c7687e4f84..0bed92c94d 100644
--- a/demo/src/main/res/layout/sync.xml
+++ b/demo/src/main/res/layout/sync.xml
@@ -79,4 +79,22 @@
         android:textColor="?attr/colorOnPrimary"
     />
 
+  <Button
+        android:id="@+id/cancel_sync_button"
+        android:layout_width="wrap_content"
+        android:layout_height="wrap_content"
+        app:layout_constraintStart_toStartOf="parent"
+        app:layout_constraintEnd_toEndOf="parent"
+        app:layout_constraintTop_toBottomOf="@id/sync_now_button"
+        app:layout_constraintBottom_toBottomOf="parent"
+        android:layout_marginTop="16dp"
+        android:layout_marginBottom="64dp"
+        android:paddingLeft="24dp"
+        android:paddingRight="24dp"
+        android:text="Cancel Sync"
+        android:backgroundTint="?attr/colorPrimary"
+        android:textColor="?attr/colorOnPrimary"
+        android:visibility="gone"
+    />
+
 </androidx.constraintlayout.widget.ConstraintLayout>
diff --git a/engine/src/androidTest/java/com/google/android/fhir/sync/SyncInstrumentedTest.kt b/engine/src/androidTest/java/com/google/android/fhir/sync/SyncInstrumentedTest.kt
index 5bd8117f1f..491e7f3169 100644
--- a/engine/src/androidTest/java/com/google/android/fhir/sync/SyncInstrumentedTest.kt
+++ b/engine/src/androidTest/java/com/google/android/fhir/sync/SyncInstrumentedTest.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2024 Google LLC
+ * Copyright 2023-2025 Google LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -86,12 +86,13 @@ class SyncInstrumentedTest {
     WorkManagerTestInitHelper.initializeTestWorkManager(context)
     val workManager = WorkManager.getInstance(context)
     runBlocking {
-      Sync.oneTimeSync<TestSyncWorker>(context = context)
-        .transformWhile {
-          emit(it is CurrentSyncJobStatus.Succeeded)
-          it !is CurrentSyncJobStatus.Succeeded
+      val statusFlow = Sync.oneTimeSync<TestSyncWorker>(context = context)
+      statusFlow
+        .transformWhile { status ->
+          emit(status is CurrentSyncJobStatus.Succeeded)
+          status !is CurrentSyncJobStatus.Succeeded
         }
-        .shareIn(this, SharingStarted.Eagerly, 5)
+        .shareIn(this, SharingStarted.Eagerly, replay = 5)
     }
 
     assertThat(workManager.getWorkInfosByTag(TestSyncWorker::class.java.name).get().first().state)
@@ -103,13 +104,20 @@ class SyncInstrumentedTest {
     WorkManagerTestInitHelper.initializeTestWorkManager(context)
     val states = mutableListOf<CurrentSyncJobStatus>()
     runBlocking {
-      Sync.oneTimeSync<TestSyncWorker>(context = context)
-        .transformWhile {
-          states.add(it)
-          emit(it is CurrentSyncJobStatus.Succeeded)
-          it !is CurrentSyncJobStatus.Succeeded
-        }
-        .shareIn(this, SharingStarted.Eagerly, 5)
+      val statusFlow = Sync.oneTimeSync<TestSyncWorker>(context = context)
+      launch {
+        statusFlow
+          .transformWhile { status ->
+            states.add(status)
+            emit(status is CurrentSyncJobStatus.Succeeded)
+            status !is CurrentSyncJobStatus.Succeeded
+          }
+          .collect { isSuccess ->
+            if (isSuccess) {
+              // Flow has completed successfully
+            }
+          }
+      }
     }
     assertThat(states.first()).isInstanceOf(CurrentSyncJobStatus.Running::class.java)
     assertThat(states.last()).isInstanceOf(CurrentSyncJobStatus.Succeeded::class.java)
@@ -122,20 +130,24 @@ class SyncInstrumentedTest {
       val states = mutableListOf<CurrentSyncJobStatus>()
       val nextExecutionStates = mutableListOf<CurrentSyncJobStatus>()
 
+      // First one-time sync
       launch {
-          Sync.oneTimeSync<TestSyncWorker>(context = context).collect {
-            states.add(it)
-            if (it is CurrentSyncJobStatus.Succeeded) {
+          val statusFlow = Sync.oneTimeSync<TestSyncWorker>(context = context)
+          statusFlow.collect { status ->
+            states.add(status)
+            if (status is CurrentSyncJobStatus.Succeeded) {
               cancel()
             }
           }
         }
         .join()
 
+      // Second one-time sync
       launch {
-          Sync.oneTimeSync<TestSyncWorker>(context = context).collect {
-            nextExecutionStates.add(it)
-            if (it is CurrentSyncJobStatus.Succeeded) {
+          val statusFlow = Sync.oneTimeSync<TestSyncWorker>(context = context)
+          statusFlow.collect { status ->
+            nextExecutionStates.add(status)
+            if (status is CurrentSyncJobStatus.Succeeded) {
               cancel()
             }
           }
@@ -151,25 +163,38 @@ class SyncInstrumentedTest {
   fun oneTime_worker_failedSyncState() {
     WorkManagerTestInitHelper.initializeTestWorkManager(context)
     val states = mutableListOf<CurrentSyncJobStatus>()
+
     runBlocking {
-      Sync.oneTimeSync<TestSyncWorkerForDownloadFailing>(
+      val statusFlow =
+        Sync.oneTimeSync<TestSyncWorkerForDownloadFailing>(
           context = context,
-          RetryConfiguration(
-            BackoffCriteria(
-              BackoffPolicy.LINEAR,
-              30,
-              TimeUnit.SECONDS,
+          retryConfiguration =
+            RetryConfiguration(
+              backoffCriteria =
+                BackoffCriteria(
+                  BackoffPolicy.LINEAR,
+                  30,
+                  TimeUnit.SECONDS,
+                ),
+              maxRetries = 0,
             ),
-            0,
-          ),
         )
-        .transformWhile {
-          states.add(it)
-          emit(it is CurrentSyncJobStatus.Failed)
-          it !is CurrentSyncJobStatus.Failed
-        }
-        .shareIn(this, SharingStarted.Eagerly, 5)
+
+      launch {
+        statusFlow
+          .transformWhile { status ->
+            states.add(status)
+            emit(status is CurrentSyncJobStatus.Failed)
+            status !is CurrentSyncJobStatus.Failed
+          }
+          .collect { isFailed ->
+            if (isFailed) {
+              // Flow has completed with a failure
+            }
+          }
+      }
     }
+
     assertThat(states.first()).isInstanceOf(CurrentSyncJobStatus.Running::class.java)
     assertThat(states.last()).isInstanceOf(CurrentSyncJobStatus.Failed::class.java)
   }
@@ -182,44 +207,52 @@ class SyncInstrumentedTest {
       val nextExecutionStates = mutableListOf<CurrentSyncJobStatus>()
 
       launch {
-          Sync.oneTimeSync<TestSyncWorkerForDownloadFailing>(
+          val statusFlow =
+            Sync.oneTimeSync<TestSyncWorkerForDownloadFailing>(
               context = context,
-              RetryConfiguration(
-                BackoffCriteria(
-                  BackoffPolicy.LINEAR,
-                  30,
-                  TimeUnit.SECONDS,
+              retryConfiguration =
+                RetryConfiguration(
+                  backoffCriteria =
+                    BackoffCriteria(
+                      BackoffPolicy.LINEAR,
+                      30,
+                      TimeUnit.SECONDS,
+                    ),
+                  maxRetries = 0,
                 ),
-                0,
-              ),
             )
-            .collect {
-              states.add(it)
-              if (it is CurrentSyncJobStatus.Failed) {
-                cancel()
-              }
+
+          statusFlow.collect { status ->
+            states.add(status)
+            if (status is CurrentSyncJobStatus.Failed) {
+              cancel()
             }
+          }
         }
         .join()
 
       launch {
-          Sync.oneTimeSync<TestSyncWorkerForDownloadFailing>(
+          val nextStatusFlow =
+            Sync.oneTimeSync<TestSyncWorkerForDownloadFailing>(
               context = context,
-              RetryConfiguration(
-                BackoffCriteria(
-                  BackoffPolicy.LINEAR,
-                  30,
-                  TimeUnit.SECONDS,
+              retryConfiguration =
+                RetryConfiguration(
+                  backoffCriteria =
+                    BackoffCriteria(
+                      BackoffPolicy.LINEAR,
+                      30,
+                      TimeUnit.SECONDS,
+                    ),
+                  maxRetries = 0,
                 ),
-                0,
-              ),
             )
-            .collect {
-              nextExecutionStates.add(it)
-              if (it is CurrentSyncJobStatus.Failed) {
-                cancel()
-              }
+
+          nextStatusFlow.collect { status ->
+            nextExecutionStates.add(status)
+            if (status is CurrentSyncJobStatus.Failed) {
+              cancel()
             }
+          }
         }
         .join()
 
@@ -234,7 +267,8 @@ class SyncInstrumentedTest {
     val states = mutableListOf<PeriodicSyncJobStatus>()
     // run and wait for periodic worker to finish
     runBlocking {
-      Sync.periodicSync<TestSyncWorker>(
+      val flow =
+        Sync.periodicSync<TestSyncWorker>(
           context = context,
           periodicSyncConfiguration =
             PeriodicSyncConfiguration(
@@ -242,6 +276,7 @@ class SyncInstrumentedTest {
               repeat = RepeatInterval(interval = 15, timeUnit = TimeUnit.MINUTES),
             ),
         )
+      flow
         .transformWhile {
           states.add(it)
           emit(it)
@@ -263,9 +298,10 @@ class SyncInstrumentedTest {
   fun periodic_worker_failedPeriodicSyncState() {
     WorkManagerTestInitHelper.initializeTestWorkManager(context)
     val states = mutableListOf<PeriodicSyncJobStatus>()
-    // run and wait for periodic worker to finish
+
     runBlocking {
-      Sync.periodicSync<TestSyncWorkerForDownloadFailing>(
+      val statusFlow =
+        Sync.periodicSync<TestSyncWorkerForDownloadFailing>(
           context = context,
           periodicSyncConfiguration =
             PeriodicSyncConfiguration(
@@ -273,12 +309,18 @@ class SyncInstrumentedTest {
               repeat = RepeatInterval(interval = 15, timeUnit = TimeUnit.MINUTES),
             ),
         )
-        .transformWhile {
-          states.add(it)
-          emit(it)
-          it.currentSyncJobStatus !is CurrentSyncJobStatus.Enqueued
-        }
-        .shareIn(this, SharingStarted.Eagerly, 5)
+
+      launch {
+        statusFlow
+          .transformWhile { status ->
+            states.add(status)
+            emit(status)
+            status.currentSyncJobStatus !is CurrentSyncJobStatus.Enqueued
+          }
+          .collect { status ->
+            // Flow has emitted a value
+          }
+      }
     }
 
     assertThat(states.first().currentSyncJobStatus)
@@ -293,9 +335,11 @@ class SyncInstrumentedTest {
   fun periodic_worker_still_queued_to_run_after_oneTime_worker_started() {
     WorkManagerTestInitHelper.initializeTestWorkManager(context)
     val workManager = WorkManager.getInstance(context)
-    // run and wait for periodic worker to finish
+
+    // Run and wait for periodic worker to finish
     runBlocking {
-      Sync.periodicSync<TestSyncWorker>(
+      val statusFlow =
+        Sync.periodicSync<TestSyncWorker>(
           context = context,
           periodicSyncConfiguration =
             PeriodicSyncConfiguration(
@@ -303,36 +347,47 @@ class SyncInstrumentedTest {
               repeat = RepeatInterval(interval = 15, timeUnit = TimeUnit.MINUTES),
             ),
         )
-        .transformWhile {
-          emit(it)
-          it.currentSyncJobStatus !is CurrentSyncJobStatus.Enqueued
-        }
-        .shareIn(this, SharingStarted.Eagerly, 5)
+
+      launch {
+        statusFlow
+          .transformWhile { status ->
+            emit(status)
+            status.currentSyncJobStatus !is CurrentSyncJobStatus.Enqueued
+          }
+          .collect { status -> }
+      }
     }
 
-    // Verify the periodic worker completed the run, and is queued to run again
-    val periodicWorkerId =
-      workManager.getWorkInfosByTag(TestSyncWorker::class.java.name).get().first().id
-    assertThat(workManager.getWorkInfoById(periodicWorkerId).get().state)
-      .isEqualTo(WorkInfo.State.ENQUEUED)
+    // Verify the periodic worker completed the run and is queued to run again
+    val periodicWorkerInfo =
+      workManager
+        .getWorkInfoById(
+          workManager.getWorkInfosByTag(TestSyncWorker::class.java.name).get().first().id,
+        )
+        .get()
+    assertThat(periodicWorkerInfo.state).isEqualTo(WorkInfo.State.ENQUEUED)
 
-    // Start and complete a oneTime job, and verify it does not remove the periodic worker
+    // Start and complete a one-time job, and verify it does not remove the periodic worker
     runBlocking {
-      Sync.oneTimeSync<TestSyncWorker>(context = context)
-        .transformWhile {
-          emit(it)
-          it !is CurrentSyncJobStatus.Succeeded
+      val oneTimeStatusFlow = Sync.oneTimeSync<TestSyncWorker>(context = context)
+
+      oneTimeStatusFlow
+        .transformWhile { status ->
+          emit(status)
+          status !is CurrentSyncJobStatus.Succeeded
         }
-        .shareIn(this, SharingStarted.Eagerly, 5)
+        .collect { status -> }
     }
+
     assertThat(workManager.getWorkInfosByTag(TestSyncWorker::class.java.name).get().size)
       .isEqualTo(2)
 
     // Move forward to the next epoch to trigger the periodic sync
     val testDriver = WorkManagerTestInitHelper.getTestDriver(context)
-    testDriver?.setPeriodDelayMet(periodicWorkerId)
+    testDriver?.setPeriodDelayMet(periodicWorkerInfo.id)
 
-    assertThat(workManager.getWorkInfoById(periodicWorkerId).get().state)
+    // Confirm the periodic worker is running after advancing the period
+    assertThat(workManager.getWorkInfoById(periodicWorkerInfo.id).get().state)
       .isEqualTo(WorkInfo.State.RUNNING)
   }
 }
diff --git a/engine/src/main/java/com/google/android/fhir/sync/FhirDataStore.kt b/engine/src/main/java/com/google/android/fhir/sync/FhirDataStore.kt
index 5cb1245796..3e55d38c4b 100644
--- a/engine/src/main/java/com/google/android/fhir/sync/FhirDataStore.kt
+++ b/engine/src/main/java/com/google/android/fhir/sync/FhirDataStore.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2024 Google LLC
+ * Copyright 2023-2025 Google LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,11 +23,14 @@ import androidx.datastore.preferences.core.stringPreferencesKey
 import androidx.datastore.preferences.preferencesDataStore
 import java.io.IOException
 import java.time.OffsetDateTime
+import java.util.concurrent.ConcurrentHashMap
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import timber.log.Timber
 
 @PublishedApi
@@ -100,6 +103,40 @@ internal class FhirDataStore(context: Context) {
     runBlocking { dataStore.edit { pref -> pref[lastSyncTimestampKey] = datetime.toString() } }
   }
 
+  private val mutexMap = ConcurrentHashMap<String, Mutex>()
+
+  private fun getOrCreateMutex(key: String): Mutex {
+    // computeIfAbsent is thread-safe and only creates a Mutex if one doesn't exist.
+    return mutexMap.computeIfAbsent(key) { Mutex() }
+  }
+
+  /** Stores the given unique-work-name in DataStore. */
+  @PublishedApi
+  internal suspend fun storeUniqueWorkName(key: String, value: String) {
+    getOrCreateMutex(key).withLock {
+      dataStore.edit { preferences -> preferences[stringPreferencesKey("$key-key")] = value }
+    }
+  }
+
+  @PublishedApi
+  internal suspend fun removeUniqueWorkName(key: String) {
+    getOrCreateMutex(key).withLock {
+      dataStore.edit { preferences ->
+        val value = preferences.remove(stringPreferencesKey("$key-key"))
+        Timber.d("Removed value: $value")
+      }
+    }
+  }
+
+  /** Fetches the stored unique-work-name from DataStore. */
+  @PublishedApi
+  internal suspend fun fetchUniqueWorkName(key: String): String? {
+    return getOrCreateMutex(key).withLock {
+      val preferences = dataStore.data.first()
+      preferences[stringPreferencesKey("$key-key")]
+    }
+  }
+
   companion object {
     private const val FHIR_PREFERENCES_NAME = "FHIR_ENGINE_PREF_DATASTORE"
     private const val LAST_SYNC_TIMESTAMP = "LAST_SYNC_TIMESTAMP"
diff --git a/engine/src/main/java/com/google/android/fhir/sync/Sync.kt b/engine/src/main/java/com/google/android/fhir/sync/Sync.kt
index c54a39685a..3048f5f328 100644
--- a/engine/src/main/java/com/google/android/fhir/sync/Sync.kt
+++ b/engine/src/main/java/com/google/android/fhir/sync/Sync.kt
@@ -24,15 +24,9 @@ import androidx.work.ExistingWorkPolicy
 import androidx.work.OneTimeWorkRequest
 import androidx.work.PeriodicWorkRequest
 import androidx.work.WorkInfo
-import androidx.work.WorkInfo.State.BLOCKED
-import androidx.work.WorkInfo.State.CANCELLED
-import androidx.work.WorkInfo.State.ENQUEUED
-import androidx.work.WorkInfo.State.FAILED
-import androidx.work.WorkInfo.State.RUNNING
-import androidx.work.WorkInfo.State.SUCCEEDED
 import androidx.work.WorkManager
 import androidx.work.hasKeyWithValueOfType
-import com.google.android.fhir.FhirEngineProvider
+import com.google.android.fhir.FhirEngineProvider.getFhirDataStore
 import com.google.android.fhir.OffsetDateTimeTypeAdapter
 import com.google.android.fhir.sync.CurrentSyncJobStatus.Cancelled
 import com.google.android.fhir.sync.CurrentSyncJobStatus.Enqueued
@@ -48,6 +42,7 @@ import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.flatMapConcat
 import kotlinx.coroutines.flow.mapNotNull
+import timber.log.Timber
 
 object Sync {
   val gson: Gson =
@@ -64,12 +59,12 @@ object Sync {
    * @param retryConfiguration configuration to guide the retry mechanism, or `null` to stop retry.
    * @return a [Flow] of [CurrentSyncJobStatus]
    */
-  inline fun <reified W : FhirSyncWorker> oneTimeSync(
+  suspend inline fun <reified W : FhirSyncWorker> oneTimeSync(
     context: Context,
     retryConfiguration: RetryConfiguration? = defaultRetryConfiguration,
     existingWorkPolicy: ExistingWorkPolicy = ExistingWorkPolicy.KEEP,
   ): Flow<CurrentSyncJobStatus> {
-    val uniqueWorkName = "${W::class.java.name}-oneTimeSync"
+    val uniqueWorkName = createSyncUniqueName<W>("oneTimeSync")
     val flow = getWorkerInfo(context, uniqueWorkName)
     val oneTimeWorkRequest =
       createOneTimeWorkRequest(retryConfiguration, W::class.java, uniqueWorkName)
@@ -79,6 +74,7 @@ object Sync {
         existingWorkPolicy,
         oneTimeWorkRequest,
       )
+    storeUniqueWorkNameInDataStore(context, uniqueWorkName)
     return combineSyncStateForOneTimeSync(context, uniqueWorkName, flow)
   }
 
@@ -93,12 +89,12 @@ object Sync {
    * @return a [Flow] of [PeriodicSyncJobStatus]
    */
   @ExperimentalCoroutinesApi
-  inline fun <reified W : FhirSyncWorker> periodicSync(
+  suspend inline fun <reified W : FhirSyncWorker> periodicSync(
     context: Context,
     periodicSyncConfiguration: PeriodicSyncConfiguration,
     existingPeriodicWorkPolicy: ExistingPeriodicWorkPolicy = ExistingPeriodicWorkPolicy.KEEP,
   ): Flow<PeriodicSyncJobStatus> {
-    val uniqueWorkName = "${W::class.java.name}-periodicSync"
+    val uniqueWorkName = createSyncUniqueName<W>("periodicSync")
     val flow = getWorkerInfo(context, uniqueWorkName)
     val periodicWorkRequest =
       createPeriodicWorkRequest(periodicSyncConfiguration, W::class.java, uniqueWorkName)
@@ -108,9 +104,33 @@ object Sync {
         existingPeriodicWorkPolicy,
         periodicWorkRequest,
       )
+    storeUniqueWorkNameInDataStore(context, uniqueWorkName)
     return combineSyncStateForPeriodicSync(context, uniqueWorkName, flow)
   }
 
+  suspend inline fun <reified W : FhirSyncWorker> cancelOneTimeSync(context: Context) {
+    cancelSync<W>(context, "oneTimeSync")
+  }
+
+  suspend inline fun <reified W : FhirSyncWorker> cancelPeriodicSync(context: Context) {
+    cancelSync<W>(context, "periodicSync")
+  }
+
+  @PublishedApi
+  internal suspend inline fun <reified W : FhirSyncWorker> cancelSync(
+    context: Context,
+    syncType: String,
+  ) {
+    val uniqueWorkNameAsKey = createSyncUniqueName<W>(syncType)
+    val uniqueWorkNameValueFromDataStore =
+      getFhirDataStore(context).fetchUniqueWorkName(uniqueWorkNameAsKey)
+    if (uniqueWorkNameValueFromDataStore != null) {
+      WorkManager.getInstance(context).cancelUniqueWork(uniqueWorkNameValueFromDataStore)
+    } else {
+      Timber.w("No value found for uniqueWorkName: $uniqueWorkNameValueFromDataStore")
+    }
+  }
+
   /**
    * Retrieves the work information for a specific unique work name as a flow of pairs containing
    * the work state and the corresponding progress data if available.
@@ -154,7 +174,7 @@ object Sync {
     workerInfoSyncJobStatusPairFromWorkManagerFlow: Flow<Pair<WorkInfo.State, SyncJobStatus?>>,
   ): Flow<PeriodicSyncJobStatus> {
     val syncJobStatusInDataStoreFlow: Flow<SyncJobStatus?> =
-      FhirEngineProvider.getFhirDataStore(context).observeTerminalSyncJobStatus(workName)
+      getFhirDataStore(context).observeTerminalSyncJobStatus(workName)
     return combine(workerInfoSyncJobStatusPairFromWorkManagerFlow, syncJobStatusInDataStoreFlow) {
       workerInfoSyncJobStatusPairFromWorkManager,
       syncJobStatusFromDataStore,
@@ -162,11 +182,11 @@ object Sync {
       PeriodicSyncJobStatus(
         lastSyncJobStatus = mapSyncJobStatusToResult(syncJobStatusFromDataStore),
         currentSyncJobStatus =
-          createSyncState(
-            WorkRequest.PERIODIC,
+          createSyncStateForPeriodicSync(
+            context,
+            workName,
             workerInfoSyncJobStatusPairFromWorkManager.first,
             workerInfoSyncJobStatusPairFromWorkManager.second,
-            syncJobStatusFromDataStore,
           ),
       )
     }
@@ -188,14 +208,15 @@ object Sync {
     workerInfoSyncJobStatusPairFromWorkManagerFlow: Flow<Pair<WorkInfo.State, SyncJobStatus?>>,
   ): Flow<CurrentSyncJobStatus> {
     val syncJobStatusInDataStoreFlow: Flow<SyncJobStatus?> =
-      FhirEngineProvider.getFhirDataStore(context).observeTerminalSyncJobStatus(workName)
+      getFhirDataStore(context).observeTerminalSyncJobStatus(workName)
 
     return combine(workerInfoSyncJobStatusPairFromWorkManagerFlow, syncJobStatusInDataStoreFlow) {
       workerInfoSyncJobStatusPairFromWorkManager,
       syncJobStatusFromDataStore,
       ->
-      createSyncState(
-        WorkRequest.ONE_TIME,
+      createSyncStateForOneTimeSync(
+        context,
+        workName,
         workerInfoSyncJobStatusPairFromWorkManager.first,
         workerInfoSyncJobStatusPairFromWorkManager.second,
         syncJobStatusFromDataStore,
@@ -258,43 +279,31 @@ object Sync {
 
   /** Gets the timestamp of the last sync job. */
   fun getLastSyncTimestamp(context: Context): OffsetDateTime? {
-    return FhirEngineProvider.getFhirDataStore(context).readLastSyncTimestamp()
+    return getFhirDataStore(context).readLastSyncTimestamp()
   }
 
-  private fun createSyncState(
-    workRequest: WorkRequest,
+  private suspend fun createSyncStateForOneTimeSync(
+    context: Context,
+    uniqueWorkName: String,
     workInfoState: WorkInfo.State,
     syncJobStatusFromWorkManager: SyncJobStatus?,
     syncJobStatusFromDataStore: SyncJobStatus?,
   ): CurrentSyncJobStatus {
-    return when (syncJobStatusFromWorkManager) {
-      is SyncJobStatus.Started,
-      is SyncJobStatus.InProgress, -> Running(syncJobStatusFromWorkManager)
-      null -> {
-        when (workRequest) {
-          WorkRequest.ONE_TIME ->
-            handleNullWorkManagerStatusForOneTimeSync(workInfoState, syncJobStatusFromDataStore)
-          WorkRequest.PERIODIC -> handleNullWorkManagerStatusForPeriodicSync(workInfoState)
+    return when (workInfoState) {
+      WorkInfo.State.ENQUEUED -> {
+        Enqueued
+      }
+      WorkInfo.State.RUNNING -> {
+        return when (syncJobStatusFromWorkManager) {
+          is SyncJobStatus.Started,
+          is SyncJobStatus.InProgress, -> Running(syncJobStatusFromWorkManager)
+          is SyncJobStatus.Succeeded -> Succeeded(syncJobStatusFromWorkManager.timestamp)
+          is SyncJobStatus.Failed -> Failed(syncJobStatusFromWorkManager.timestamp)
+          null -> Running(SyncJobStatus.Started())
         }
       }
-      else -> error("Inconsistent syncJobStatus: $syncJobStatusFromWorkManager.")
-    }
-  }
-
-  /**
-   * Creates terminal states of [CurrentSyncJobStatus] from [syncJobStatusFromDataStore]; and
-   * intermediate states of [CurrentSyncJobStatus] from [WorkInfo.State].
-   *
-   * Note : Only call this API when `syncJobStatusFromWorkManager` is null.
-   */
-  private fun handleNullWorkManagerStatusForOneTimeSync(
-    workInfoState: WorkInfo.State,
-    syncJobStatusFromDataStore: SyncJobStatus?,
-  ): CurrentSyncJobStatus =
-    when (workInfoState) {
-      ENQUEUED -> Enqueued
-      RUNNING -> Running(SyncJobStatus.Started())
-      SUCCEEDED ->
+      WorkInfo.State.SUCCEEDED -> {
+        removeUniqueWorkNameInDataStore(context, uniqueWorkName)
         syncJobStatusFromDataStore?.let {
           when (it) {
             is SyncJobStatus.Succeeded -> Succeeded(it.timestamp)
@@ -302,7 +311,9 @@ object Sync {
           }
         }
           ?: error("Inconsistent terminal syncJobStatus.")
-      FAILED ->
+      }
+      WorkInfo.State.FAILED -> {
+        removeUniqueWorkNameInDataStore(context, uniqueWorkName)
         syncJobStatusFromDataStore?.let {
           when (it) {
             is SyncJobStatus.Failed -> Failed(it.timestamp)
@@ -310,34 +321,40 @@ object Sync {
           }
         }
           ?: error("Inconsistent terminal syncJobStatus.")
-      CANCELLED -> Cancelled
-      BLOCKED -> CurrentSyncJobStatus.Blocked
+      }
+      WorkInfo.State.CANCELLED -> {
+        removeUniqueWorkNameInDataStore(context, uniqueWorkName)
+        Cancelled
+      }
+      WorkInfo.State.BLOCKED -> CurrentSyncJobStatus.Blocked
     }
+  }
 
-  /**
-   * Only call this API when syncJobStatus From WorkManager is null. Create a [CurrentSyncJobStatus]
-   * from [WorkInfo.State]. (Note: syncJobStatusFromDataStore is updated as lastSynJobStatus, which
-   * is the terminalSyncJobStatus.)
-   */
-  private fun handleNullWorkManagerStatusForPeriodicSync(
+  private suspend fun createSyncStateForPeriodicSync(
+    context: Context,
+    uniqueWorkName: String,
     workInfoState: WorkInfo.State,
-  ): CurrentSyncJobStatus =
-    when (workInfoState) {
-      RUNNING -> Running(SyncJobStatus.Started())
-      ENQUEUED -> Enqueued
-      CANCELLED -> Cancelled
-      BLOCKED -> CurrentSyncJobStatus.Blocked
+    syncJobStatusFromWorkManager: SyncJobStatus?,
+  ): CurrentSyncJobStatus {
+    return when (workInfoState) {
+      WorkInfo.State.ENQUEUED -> Enqueued
+      WorkInfo.State.RUNNING -> {
+        return when (syncJobStatusFromWorkManager) {
+          is SyncJobStatus.Started,
+          is SyncJobStatus.InProgress, -> Running(syncJobStatusFromWorkManager)
+          is SyncJobStatus.Succeeded -> Succeeded(syncJobStatusFromWorkManager.timestamp)
+          is SyncJobStatus.Failed -> Failed(syncJobStatusFromWorkManager.timestamp)
+          null -> Running(SyncJobStatus.Started())
+        }
+      }
+      WorkInfo.State.CANCELLED -> {
+        removeUniqueWorkNameInDataStore(context, uniqueWorkName)
+        Cancelled
+      }
+      WorkInfo.State.BLOCKED -> CurrentSyncJobStatus.Blocked
       else -> error("Inconsistent WorkInfo.State in periodic sync : $workInfoState.")
     }
-
-  private fun mapDataStoreSyncJobStatusToCurrentSyncJobStatus(
-    syncJobStatusFromDataStore: SyncJobStatus,
-  ) =
-    when (syncJobStatusFromDataStore) {
-      is SyncJobStatus.Succeeded -> Succeeded(syncJobStatusFromDataStore.timestamp)
-      is SyncJobStatus.Failed -> Failed(syncJobStatusFromDataStore.timestamp)
-      else -> error("Inconsistent syncJobStatus in the dataStore : $syncJobStatusFromDataStore.")
-    }
+  }
 
   /**
    * Maps the [lastSyncJobStatus] to a specific [LastSyncJobStatus] based on the provided status.
@@ -361,8 +378,30 @@ object Sync {
       }
     }
 
-  private enum class WorkRequest {
-    ONE_TIME,
-    PERIODIC,
+  @PublishedApi
+  internal inline fun <reified W : FhirSyncWorker> createSyncUniqueName(syncType: String): String {
+    return "${W::class.java.name}-$syncType"
+  }
+
+  @PublishedApi
+  internal suspend fun storeUniqueWorkNameInDataStore(
+    context: Context,
+    uniqueWorkName: String,
+  ) {
+    val dataStore = getFhirDataStore(context)
+    if (dataStore.fetchUniqueWorkName(uniqueWorkName) == null) {
+      dataStore.storeUniqueWorkName(key = uniqueWorkName, value = uniqueWorkName)
+    }
+  }
+
+  @PublishedApi
+  internal suspend fun removeUniqueWorkNameInDataStore(
+    context: Context,
+    uniqueWorkName: String,
+  ) {
+    val dataStore = getFhirDataStore(context)
+    if (dataStore.fetchUniqueWorkName(uniqueWorkName) != null) {
+      dataStore.removeUniqueWorkName(key = uniqueWorkName)
+    }
   }
 }