refactor(sync): enable concurrent sync start/stop [WPB-15262]

[vitorhugods]

WPB-15262 [Android] Sync engine refactoring

---

PR Submission Checklist for internal contributors

• The PR Title

  • conforms to the style of semantic commits messages¹ supported in Wire's Github Workflow²
  • contains a reference JIRA issue number like SQPIT-764
  • answers the question: If merged, this PR will: ... ³

• The PR Description

  • is free of optional paragraphs and you have filled the relevant parts to the best of your ability

---

What's new in this PR?

Issues

It is a pain to start/stop sync from multiple places using ConnectionPolicy.
For instance, what if you want to start sync, perform an operation, and then stop?
You'd need to set the policy, wait, and then reset it to whatever value it was. But what about race conditions? What if another piece of code is changing that too?

PAIN

Solutions

Rename SyncManager

SyncManager is now SyncStateObserver. It had the name of "Manager" but it didn't manage anything. It still has the same function: observe state and passively wait for specific states.

Move Sync start/stop logic to SyncExecutor

Instead of having only autonomous SlowSyncManager and IncrementalSyncManager that would start/stop on their own and follow the policy: Now SyncExecutor will allow multiple requests to be made at the same time. If at least one request is alive, it will start Sync.

Multiple entities can easily call syncExecutor.request {}.

Some examples:

// Start/stop sync for a synchronous operation
syncExecutor.request {
    waitUntilLiveOrFailure().onSuccess {
        sendMessage()
    }
}


// Start / Stop sync assynchronously
class LifecycleObserver {

    val job: Job? = null

    fun onBackground()
        job = launch {
            syncExecutor.request {
                awaitCancellation()
            }
        }
    }

    fun onBackground() {
        job?.cancel()
    }
}

// Start sync and never stop (useful for test-service and similar)
@OptIn(DelicateKaliumApi::class)
syncExecutor.request {
    keepSyncAlwaysOn()
}

Although not visible in this PR, the code in wire-android regarding sync policy will be reduced and simplified A LOT.

Note

Maybe SyncStateObserver and SyncExecutor could be the same class and be named SyncManager. HOWEVER, I've identified a few cyclic dependencies during this process. Some classes that are used during event processing have dependencies on SyncManager. It is possible to go around it, but in the end I opted to have these two entities: one responsible for passively watching Sync. Another responsible for actively requesting it.

Testing

Test Coverage

• I have added automated test to this contribution

PR Post Merge Checklist for internal contributors

• If any soft of configuration variable was introduced by this PR, it has been added to the relevant documents and the CI jobs have been updated.

---

References

1. https://sparkbox.com/foundry/semantic_commit_messages
2. https://github.com/wireapp/.github#usage
3. E.g. feat(conversation-list): Sort conversations by most emojis in the title #SQPIT-764.

[github-actions]

Test Results

3 524 tests  +4   3 416 ✅ +4   6m 11s ⏱️ -18s
  603 suites ±0     108 💤 ±0 
  603 files   ±0       0 ❌ ±0 

Results for commit 5502dae. ± Comparison against base commit 7c462ea.

This pull request removes 40 and adds 44 tests. Note that renamed tests count towards both.

com.wire.kalium.logic.data.sync.IncrementalSyncRepositoryTest ‑ givenASlowPolicyCollector_whenPolicyIsUpdatedManyTimes_thenUpdateEmissionShouldNotBeBlockedByOverflownBuffer[jvm]
com.wire.kalium.logic.data.sync.IncrementalSyncRepositoryTest ‑ givenConnectionPolicyIsUpdatedWithRepeatedValue_whenCollectingPolicy_thenShouldNotCollectRepeatedValues[jvm]
com.wire.kalium.logic.data.sync.IncrementalSyncRepositoryTest ‑ givenConnectionPolicyUpdatedMultipleTimes_whenCollectingConnectionPolicy_thenAllUpdatesShouldBeCollected[jvm]
com.wire.kalium.logic.data.sync.IncrementalSyncRepositoryTest ‑ givenNoConnectionPolicyUpdate_whenCollectingConnectionPolicy_thenShouldEmitKeepAliveByDefault[jvm]
com.wire.kalium.logic.sync.ObserveSyncStateUseCaseTest ‑ givenIncrementalSyncStateEmitsFailedState_whenRunningUseCase_thenEmitFailedState[jvm]
com.wire.kalium.logic.sync.ObserveSyncStateUseCaseTest ‑ givenIncrementalSyncStateEmitsFetchingPendingEventsState_whenRunningUseCase_thenEmitGatheringPendingEventsState[jvm]
com.wire.kalium.logic.sync.ObserveSyncStateUseCaseTest ‑ givenIncrementalSyncStateEmitsLiveState_whenRunningUseCase_thenEmitLiveState[jvm]
com.wire.kalium.logic.sync.ObserveSyncStateUseCaseTest ‑ givenIncrementalSyncStateEmitsPendingState_whenRunningUseCase_thenEmitGatheringPendingEventsState[jvm]
com.wire.kalium.logic.sync.ObserveSyncStateUseCaseTest ‑ givenSlowSyncStatusEmitsFailedState_whenRunningUseCase_thenEmitFailedState[jvm]
com.wire.kalium.logic.sync.ObserveSyncStateUseCaseTest ‑ givenSlowSyncStatusEmitsOngoingState_whenRunningUseCase_thenEmitSlowSyncState[jvm]
…

com.wire.kalium.logic.sync.ObserveSyncStateUseCaseTest ‑ givenObserverFlowEmitsValues_whenInvoking_thenShouldForwardTheSameValues[jvm]
com.wire.kalium.logic.sync.SyncExecutorTest ‑ givenKeepSyncAlwaysOn_whenRequestBlockEnds_thenShouldKeepSync[jvm]
com.wire.kalium.logic.sync.SyncExecutorTest ‑ givenMultipleRequests_whenProcessingSync_thenShouldHaveOnlyOneSubscriptionToEachSyncManager[jvm]
com.wire.kalium.logic.sync.SyncExecutorTest ‑ givenNoMoreRequests_whenStartingAsNeeded_thenShouldStopSync[jvm]
com.wire.kalium.logic.sync.SyncExecutorTest ‑ givenNoRequests_whenStartingAsNeeded_thenShouldNotStartAnySync[jvm]
com.wire.kalium.logic.sync.SyncExecutorTest ‑ givenOneRequest_whenStartingAsNeeded_thenShouldStartWithSlowSyncFirst[jvm]
com.wire.kalium.logic.sync.SyncExecutorTest ‑ givenOneRequest_whenSyncScopeIsCancelled_thenShouldStopSyncManagerSubscriptions[jvm]
com.wire.kalium.logic.sync.SyncExecutorTest ‑ givenSlowSyncCompletes_whenSlowSyncIsReset_thenShouldStopIncrementalSync[jvm]
com.wire.kalium.logic.sync.SyncExecutorTest ‑ givenSlowSyncCompletes_whenStartingAsNeeded_thenShouldStartIncrementalSync[jvm]
com.wire.kalium.logic.sync.SyncExecutorTest ‑ givenWaitingForState_whenFailureHappens_thenShouldContinue[jvm]
…

♻️ This comment has been updated with latest results.

[github-actions]

Bencher Report

Branch	refactor/sync/allow-concurrent-sync-start-stopping
Testbed	ubuntu-latest

⚠️ WARNING: No Threshold found!

Without a Threshold, no Alerts will ever be generated.

• Latency (nanoseconds (ns))

Click here to create a new Threshold
For more information, see the Threshold documentation.
To only post results if a Threshold exists, set the --ci-only-thresholds flag.

Click to view all benchmark results

Benchmark	Latency	microseconds (µs)
com.wire.kalium.benchmarks.logic.CoreLogicBenchmark.createObjectInFiles	📈 view plot ⚠️ NO THRESHOLD	762.20
com.wire.kalium.benchmarks.logic.CoreLogicBenchmark.createObjectInMemory	📈 view plot ⚠️ NO THRESHOLD	408,452.04
com.wire.kalium.benchmarks.persistence.MessagesNoPragmaTuneBenchmark.messageInsertionBenchmark	📈 view plot ⚠️ NO THRESHOLD	1,416,548.46
com.wire.kalium.benchmarks.persistence.MessagesNoPragmaTuneBenchmark.queryMessagesBenchmark	📈 view plot ⚠️ NO THRESHOLD	28,176.11

🐰 View full continuous benchmarking report in Bencher

[datadog-wireapp]

Datadog Report

Branch report: refactor/sync/allow-concurrent-sync-start-stopping
Commit report: ca82b7d
Test service: kalium-jvm

✅ 0 Failed, 3416 Passed, 108 Skipped, 1m 2.01s Total Time

[Garzas]

Looking good 🚀 just one comment

[Garzas]

I think those mocks will throw exception

[vitorhugods]

These functions aren't used anywhere. The only place using this Fake is in SyncExecutorTest. And SyncExecutor only needs the syncState.

So it's fine to leave them like this.

If someone wants to use this Fake elsewhere, feel free to fill the gaps 😄

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 50.70%. Comparing base (7c462ea) to head (5502dae).
Report is 1 commits behind head on develop.

Additional details and impacted files

@@             Coverage Diff             @@
##           develop    #3304      +/-   ##
===========================================
+ Coverage    50.68%   50.70%   +0.01%     
===========================================
  Files         1611     1610       -1     
  Lines        58145    58188      +43     
  Branches      5246     5249       +3     
===========================================
+ Hits         29473    29503      +30     
- Misses       26645    26657      +12     
- Partials      2027     2028       +1     

see 19 files with indirect coverage changes

---

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 7c462ea...5502dae. Read the comment docs.