Refactor concurrent uploading of local changes

RELEASE-NOTES.txt

@@ -24,6 +24,7 @@
 * [*] Block editor: [List block] Fix an issue when merging a list item into a Paragraph would remove its nested list items [https://github.com/wordpress-mobile/gutenberg-mobile/pull/5785]
 * [*] Block editor: [Gallery block] Fixes a compatibility issue with Gallery block [https://github.com/wordpress-mobile/WordPress-Android/pull/18519]
 * [**] Block editor: Tapping any type of nested block moves focus to the nested block directly, rather than requiring multiple taps to navigate down each nesting levels. [https://github.com/wordpress-mobile/gutenberg-mobile/pull/5781]
+* [*] Fixed an error when syncing offline data to the cloud. [https://github.com/wordpress-mobile/WordPress-Android/pull/18449]
 
 22.4
 -----

WordPress/src/main/java/org/wordpress/android/modules/ThreadModule.kt

@@ -7,8 +7,10 @@ import dagger.hilt.components.SingletonComponent
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.sync.Mutex
 import org.wordpress.android.util.helpers.Debouncer
 import javax.inject.Named
+import javax.inject.Singleton
 
 const val APPLICATION_SCOPE = "APPLICATION_SCOPE"
 
@@ -53,4 +55,10 @@ class ThreadModule {
     fun provideDebouncer(): Debouncer {
         return Debouncer()
     }
+
+    @Provides
+    @Singleton
+    fun provideMutex(): Mutex {
+        return Mutex()
+    }
 }

WordPress/src/main/java/org/wordpress/android/ui/uploads/UploadStarter.kt

@@ -4,9 +4,7 @@ import android.content.Context
 import androidx.lifecycle.DefaultLifecycleObserver
 import androidx.lifecycle.LifecycleOwner
 import androidx.lifecycle.LiveData
-import androidx.lifecycle.Observer
 import androidx.lifecycle.ProcessLifecycleOwner
-import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
@@ -18,6 +16,7 @@ import kotlinx.coroutines.sync.withLock
 import org.wordpress.android.analytics.AnalyticsTracker.Stat
 import org.wordpress.android.fluxc.Dispatcher
 import org.wordpress.android.fluxc.generated.UploadActionBuilder
+import org.wordpress.android.fluxc.model.PostModel
 import org.wordpress.android.fluxc.model.SiteModel
 import org.wordpress.android.fluxc.store.PageStore
 import org.wordpress.android.fluxc.store.PostStore
@@ -45,14 +44,15 @@ import kotlin.coroutines.CoroutineContext
  * addition to this, call sites can also request an immediate execution by calling [checkConnectionAndUpload].
  *
  * The method [activateAutoUploading] must be called once, preferably during app creation, for the auto-uploads to work.
+ *
+ * @param mutex When the app comes to foreground both [queueUploadFromAllSites] and [queueUploadFromSite] are invoked.
+ * The problem is that they can run in parallel and `uploadServiceFacade.isPostUploadingOrQueued(it)` might return
+ * out-of-date result and a same post is added twice.
  */
 @Singleton
 @OpenForTesting
 class UploadStarter @Inject constructor(
-    /**
-     * The Application context
-     */
-    private val context: Context,
+    private val appContext: Context,
     private val dispatcher: Dispatcher,
     private val postStore: PostStore,
     private val pageStore: PageStore,
@@ -63,17 +63,11 @@ class UploadStarter @Inject constructor(
     @Named(IO_THREAD) private val ioDispatcher: CoroutineDispatcher,
     private val uploadServiceFacade: UploadServiceFacade,
     private val networkUtilsWrapper: NetworkUtilsWrapper,
-    private val connectionStatus: LiveData<ConnectionStatus>
+    private val connectionStatus: LiveData<ConnectionStatus>,
+    private val mutex: Mutex,
 ) : CoroutineScope {
     private val job = Job()
 
-    /**
-     * When the app comes to foreground both `queueUploadFromAllSites` and `queueUploadFromSite` are invoked.
-     * The problem is that they can run in parallel and `uploadServiceFacade.isPostUploadingOrQueued(it)` might return
-     * out-of-date result and a same post is added twice.
-     */
-    private val mutex = Mutex()
-
     override val coroutineContext: CoroutineContext get() = job + bgDispatcher
 
     /**
@@ -97,48 +91,34 @@ class UploadStarter @Inject constructor(
     fun activateAutoUploading(processLifecycleOwner: ProcessLifecycleOwner) {
         // We're skipping the first emitted value because the processLifecycleObserver below will also trigger an
         // immediate upload.
-        connectionStatus.skip(1).observe(processLifecycleOwner, Observer {
-            queueUploadFromAllSites()
-        })
+        connectionStatus.skip(1).observe(processLifecycleOwner) { queueUploadFromAllSites() }
 
         processLifecycleOwner.lifecycle.addObserver(processLifecycleObserver)
     }
 
-    fun queueUploadFromAllSites() = launch {
-        val sites = siteStore.sites
-        try {
-            checkConnectionAndUpload(sites = sites)
-        } catch (e: Exception) {
-            AppLog.e(T.MEDIA, e)
-        }
-    }
+    fun queueUploadFromAllSites() = launch { checkConnectionAndUpload(sites = siteStore.sites) }
 
     /**
      * Upload all local drafts from the given [site].
      */
-    fun queueUploadFromSite(site: SiteModel) = launch {
-        try {
-            checkConnectionAndUpload(sites = listOf(site))
-        } catch (e: Exception) {
-            AppLog.e(T.MEDIA, e)
-        }
-    }
+    fun queueUploadFromSite(site: SiteModel) = launch { checkConnectionAndUpload(sites = listOf(site)) }
 
     /**
      * If there is an internet connection, uploads all posts with local changes belonging to [sites].
      *
      * This coroutine will suspend until all the [upload] operations have completed. If one of them fails, all query
-     * and queuing attempts ([upload]) will be canceled. The exception will be thrown by this method.
+     * and queuing attempts ([upload]) will continue. The last exception will be thrown by this method.
      */
     private suspend fun checkConnectionAndUpload(sites: List<SiteModel>) = coroutineScope {
-        if (!networkUtilsWrapper.isNetworkAvailable()) {
-            return@coroutineScope
-        }
-
-        sites.forEach {
-            launch(ioDispatcher) {
-                upload(site = it)
+        if (!networkUtilsWrapper.isNetworkAvailable()) return@coroutineScope
+        try {
+            sites.forEach {
+                launch(ioDispatcher) {
+                    upload(site = it)
+                }
             }
+        } catch (e: Exception) {
+            AppLog.e(T.MEDIA, e)
         }
     }
 
@@ -148,51 +128,31 @@ class UploadStarter @Inject constructor(
     private suspend fun upload(site: SiteModel) = coroutineScope {
         try {
             mutex.withLock {
-                val posts = async { postStore.getPostsWithLocalChanges(site) }
-                val pages = async { pageStore.getPagesWithLocalChanges(site) }
+                val posts = async { postStore.getPostsWithLocalChanges(site).orEmpty() }
+                val pages = async { (pageStore.getPagesWithLocalChanges(site) as? List<PostModel>).orEmpty() }
                 val list = posts.await() + pages.await()
-
                 list.asSequence()
                     .map { post ->
                         val action = uploadActionUseCase.getAutoUploadAction(post, site)
                         Pair(post, action)
                     }
-                    .filter { (_, action) ->
-                        action != DO_NOTHING
-                    }
+                    .filter { (_, action) -> action != DO_NOTHING }
                     .toList()
                     .forEach { (post, action) ->
                         trackAutoUploadAction(action, post.status, post.isPage)
-                        AppLog.d(
-                            AppLog.T.POSTS,
-                            "UploadStarter for post (isPage: ${post.isPage}) title: ${post.title}, action: $action"
-                        )
-                        dispatcher.dispatch(
-                            UploadActionBuilder.newIncrementNumberOfAutoUploadAttemptsAction(
-                                post
-                            )
-                        )
-                        uploadServiceFacade.uploadPost(
-                            context = context,
-                            post = post,
-                            trackAnalytics = false
-                        )
+                        AppLog.d(T.POSTS, "UploadStarter for ${post.toStringLog()}; action: $action")
+                        dispatcher.dispatch(UploadActionBuilder.newIncrementNumberOfAutoUploadAttemptsAction(post))
+                        uploadServiceFacade.uploadPost(appContext, post, trackAnalytics = false)
                     }
             }
-        } catch (e: CancellationException) {
-            AppLog.e(T.MEDIA, e)
-            // Do any needed actions while we are still holding the mutex lock, then release it and rethrow the
-            // exception so it can be handled upstream
-            mutex.unlock()
-            throw e
+        } catch (e: Exception) {
+            AppLog.e(T.POSTS, e)
         }
     }
 
-    private fun trackAutoUploadAction(
-        action: UploadAction,
-        status: String,
-        isPage: Boolean
-    ) {
+    private fun PostModel.toStringLog() = "${if (isPage) "page" else "post"} with title: $title"
+
+    private fun trackAutoUploadAction(action: UploadAction, status: String, isPage: Boolean) {
         tracker.track(
             if (isPage) Stat.AUTO_UPLOAD_PAGE_INVOKED else Stat.AUTO_UPLOAD_POST_INVOKED,
             mapOf(

WordPress/src/test/java/org/wordpress/android/ui/uploads/UploadFixtures.kt

@@ -0,0 +1,60 @@
+package org.wordpress.android.ui.uploads
+
+import androidx.lifecycle.Lifecycle
+import androidx.lifecycle.MutableLiveData
+import androidx.lifecycle.ProcessLifecycleOwner
+import org.mockito.kotlin.any
+import org.mockito.kotlin.doReturn
+import org.mockito.kotlin.mock
+import org.wordpress.android.fluxc.model.PostModel
+import org.wordpress.android.fluxc.model.SiteModel
+import org.wordpress.android.fluxc.model.post.PostStatus
+import org.wordpress.android.fluxc.store.UploadStore
+import org.wordpress.android.ui.posts.PostUtilsWrapper
+import org.wordpress.android.util.DateTimeUtils
+import org.wordpress.android.util.NetworkUtilsWrapper
+import org.wordpress.android.viewmodel.helpers.ConnectionStatus
+import java.util.Date
+
+internal object UploadFixtures {
+    private var postIdIndex = 0
+
+    private fun makePostTitleFromId() = postIdIndex.toString().padStart(2, '0')
+
+    fun resetTestPostIdIndex() { postIdIndex = 0 }
+
+    fun createMockedNetworkUtilsWrapper() = mock<NetworkUtilsWrapper> { on { isNetworkAvailable() } doReturn true }
+
+    fun createConnectionStatusLiveData(initialValue: ConnectionStatus?): MutableLiveData<ConnectionStatus> {
+        return MutableLiveData<ConnectionStatus>().apply { value = initialValue }
+    }
+
+    fun createMockedPostUtilsWrapper() = mock<PostUtilsWrapper> {
+        on { isPublishable(any()) } doReturn true
+        on { isPostInConflictWithRemote(any()) } doReturn false
+    }
+
+    fun createMockedUploadStore(numberOfAutoUploadAttempts: Int) = mock<UploadStore> {
+        on { getNumberOfPostAutoUploadAttempts(any()) } doReturn numberOfAutoUploadAttempts
+    }
+
+    fun createMockedUploadServiceFacade() = mock<UploadServiceFacade> {
+        on { isPostUploadingOrQueued(any()) } doReturn false
+    }
+
+    fun createMockedProcessLifecycleOwner(lifecycle: Lifecycle = mock()) = mock<ProcessLifecycleOwner> {
+        on { this.lifecycle } doReturn lifecycle
+    }
+
+    fun createLocallyChangedPostModel(postStatus: PostStatus = PostStatus.DRAFT, page: Boolean = false) =
+        PostModel().apply {
+            setId(++postIdIndex)
+            setTitle(makePostTitleFromId())
+            setStatus(postStatus.toString())
+            setIsLocallyChanged(true)
+            setDateLocallyChanged(DateTimeUtils.iso8601FromTimestamp(Date().time / 1000))
+            setIsPage(page)
+        }
+
+    fun createSiteModel(isWpCom: Boolean = true) = SiteModel().apply { setIsWPCom(isWpCom) }
+}