Clock management and kicked

2025-08-11 14:37:45 +00:00 · 2024-10-03 12:15:51 +10:00
parent 1f5fde0d9a
commit a5c89d8d5a
17 changed files with 429 additions and 290 deletions
--- a/libsession/src/main/java/org/session/libsession/messaging/MessagingModuleConfiguration.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/MessagingModuleConfiguration.kt
@@ -7,6 +7,7 @@ import org.session.libsession.database.StorageProtocol
 import org.session.libsession.messaging.groups.GroupManagerV2
 import org.session.libsession.messaging.notifications.TokenFetcher
 import org.session.libsession.snode.OwnedSwarmAuth
+import org.session.libsession.snode.SnodeClock
 import org.session.libsession.utilities.ConfigFactoryProtocol
 import org.session.libsession.utilities.Device
 import org.session.libsession.utilities.Toaster
@@ -22,6 +23,7 @@ class MessagingModuleConfiguration(
    val toaster: Toaster,
    val tokenFetcher: TokenFetcher,
    val groupManagerV2: GroupManagerV2,
+    val clock: SnodeClock,
 ) {

    companion object {
--- a/libsession/src/main/java/org/session/libsession/messaging/groups/GroupManagerV2.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/groups/GroupManagerV2.kt
@@ -29,7 +29,21 @@ interface GroupManagerV2 {
        removeMessages: Boolean
    )

-    suspend fun handleMemberLeft(message: GroupUpdated, closedGroupId: AccountId)
+    /**
+     * Remove all messages from the group for the given members.
+     *
+     * This will delete all messages locally, and, if the user is an admin, remotely as well.
+     *
+     * Note: unlike [handleDeleteMemberContent], [requestMessageDeletion], this method
+     * does not try to validate the validity of the request, it also does not ask other members
+     * to delete the messages. It simply removes what it can.
+     */
+    suspend fun removeMemberMessages(
+        groupAccountId: AccountId,
+        members: List<AccountId>
+    )
+
+    suspend fun handleMemberLeft(message: GroupUpdated, group: AccountId)

    suspend fun leaveGroup(group: AccountId, deleteOnLeave: Boolean)

@@ -59,8 +73,22 @@ interface GroupManagerV2 {

    suspend fun setName(groupId: AccountId, newName: String)

+    /**
+     * Send a request to the group to delete the given messages.
+     *
+     * It can be called by a regular member who wishes to delete their own messages.
+     * It can also called by an admin, who can delete any messages from any member.
+     */
    suspend fun requestMessageDeletion(groupId: AccountId, messageHashes: List<String>)

+    /**
+     * Handle a request to delete a member's content from the group. This is called when we receive
+     * a message from the server that a member's content needs to be deleted. (usually sent by
+     * [requestMessageDeletion], for example)
+     *
+     * In contrast to [removeMemberMessages], where it will remove the messages blindly, this method
+     * will check if the right conditions are met before removing the messages.
+     */
    suspend fun handleDeleteMemberContent(
        groupId: AccountId,
        deleteMemberContent: GroupUpdateDeleteMemberContentMessage,
--- a/libsession/src/main/java/org/session/libsession/messaging/groups/RemoveGroupMemberHandler.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/groups/RemoveGroupMemberHandler.kt
@@ -2,10 +2,8 @@ package org.session.libsession.messaging.groups

 import android.os.SystemClock
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.Job
-import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.async
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.first
@@ -23,6 +21,7 @@ import org.session.libsession.snode.SnodeMessage
 import org.session.libsession.snode.utilities.await
 import org.session.libsession.utilities.ConfigFactoryProtocol
 import org.session.libsession.utilities.TextSecurePreferences
+import org.session.libsession.utilities.waitUntilGroupConfigsPushed
 import org.session.libsignal.protos.SignalServiceProtos
 import org.session.libsignal.protos.SignalServiceProtos.DataMessage.GroupUpdateMessage
 import org.session.libsignal.utilities.AccountId
@@ -35,17 +34,22 @@ private const val TAG = "RemoveGroupMemberHandler"

 private const val MIN_PROCESS_INTERVAL_MILLS = 1_000L

+/**
+ * This handler is responsible for processing pending group member removals.
+ *
+ * It automatically does so by listening to the config updates changes and checking for any pending removals.
+ */
 class RemoveGroupMemberHandler @Inject constructor(
    private val configFactory: ConfigFactoryProtocol,
    private val textSecurePreferences: TextSecurePreferences,
+    private val groupManager: GroupManagerV2,
 ) {
-    private val scope: CoroutineScope = GlobalScope
    private var job: Job? = null

    fun start() {
        require(job == null) { "Already started" }

-        job = scope.launch {
+        job = GlobalScope.launch {
            while (true) {
                // Make sure we have a local number before we start processing
                textSecurePreferences.watchLocalNumber().first { it != null }
@@ -74,74 +78,54 @@ class RemoveGroupMemberHandler @Inject constructor(
    }

    private suspend fun processPendingMemberRemoval() {
-        // Run the removal process for each group in parallel
-        val removalTasks = configFactory.withUserConfigs { it.userGroups.allClosedGroupInfo() }
+        configFactory.withUserConfigs { it.userGroups.allClosedGroupInfo() }
            .asSequence()
            .filter { it.hasAdminKey() }
-            .associate { group ->
-                group.name to scope.async {
-                    processPendingRemovalsForGroup(
-                        groupAccountId = group.groupAccountId,
-                        groupName = group.name,
-                        adminKey = group.adminKey!!
-                    )
-                }
+            .forEach { group ->
+                processPendingRemovalsForGroup(group.groupAccountId, group.adminKey!!)
            }
-
-        // Wait and collect the results of the removal tasks
-        for ((groupName, task) in removalTasks) {
-            try {
-                task.await()
-            } catch (e: Exception) {
-                Log.e(TAG, "Error processing pending removals for group $groupName", e)
-            }
-        }
    }

    private suspend fun processPendingRemovalsForGroup(
        groupAccountId: AccountId,
-        groupName: String,
        adminKey: ByteArray
    ) {
-        val swarmAuth = OwnedSwarmAuth(
-            accountId = groupAccountId,
-            ed25519PublicKeyHex = null,
-            ed25519PrivateKey = adminKey
-        )
+        val groupAuth = OwnedSwarmAuth.ofClosedGroup(groupAccountId, adminKey)

-        val batchCalls = configFactory.withGroupConfigs(groupAccountId) { configs ->
+        val (pendingRemovals, batchCalls) = configFactory.withGroupConfigs(groupAccountId) { configs ->
            val pendingRemovals = configs.groupMembers.all().filter { it.removed }
            if (pendingRemovals.isEmpty()) {
                // Skip if there are no pending removals
-                return@withGroupConfigs emptyList()
+                return@withGroupConfigs pendingRemovals to emptyList()
            }

-            Log.d(TAG, "Processing ${pendingRemovals.size} pending removals for group $groupName")
+            Log.d(TAG, "Processing ${pendingRemovals.size} pending removals for group")

            // Perform a sequential call to group snode to:
            // 1. Revoke the member's sub key (by adding the key to a "revoked list" under the hood)
-            // 2. Send a message to a special namespace to inform the removed members they have been removed
-            // 3. Conditionally, delete removed-members' messages from the group's message store, if that option is selected by the actioning admin
+            // 2. Send a message to a special namespace on the group to inform the removed members they have been removed
+            // 3. Conditionally, send a `GroupUpdateDeleteMemberContent` to the group so the message deletion
+            //    can be performed by everyone in the group.
            val calls = ArrayList<SnodeAPI.SnodeBatchRequestInfo>(3)

            // Call No 1. Revoke sub-key. This call is crucial and must not fail for the rest of the operation to be successful.
            calls += checkNotNull(
                SnodeAPI.buildAuthenticatedRevokeSubKeyBatchRequest(
-                    groupAdminAuth = swarmAuth,
+                    groupAdminAuth = groupAuth,
                    subAccountTokens = pendingRemovals.map {
                        configs.groupKeys.getSubAccountToken(AccountId(it.sessionId))
                    }
                )
            ) { "Fail to create a revoke request" }

-            // Call No 2. Send a message to the removed members
+            // Call No 2. Send a "kicked" message to the revoked namespace
            calls += SnodeAPI.buildAuthenticatedStoreBatchInfo(
                namespace = Namespace.REVOKED_GROUP_MESSAGES(),
                message = buildGroupKickMessage(groupAccountId.hexString, pendingRemovals, configs.groupKeys, adminKey),
-                auth = swarmAuth,
+                auth = groupAuth,
            )

-            // Call No 3. Conditionally remove the message from the group's message store
+            // Call No 3. Conditionally send the `GroupUpdateDeleteMemberContent`
            if (pendingRemovals.any { it.shouldRemoveMessages }) {
                calls += SnodeAPI.buildAuthenticatedStoreBatchInfo(
                    namespace = Namespace.CLOSED_GROUP_MESSAGES(),
@@ -152,11 +136,11 @@ class RemoveGroupMemberHandler @Inject constructor(
                            .filter { it.shouldRemoveMessages }
                            .map { it.sessionId }
                    ),
-                    auth = swarmAuth,
+                    auth = groupAuth,
                )
            }

-            calls
+            pendingRemovals to (calls as List<SnodeAPI.SnodeBatchRequestInfo>)
        }

        if (batchCalls.isEmpty()) {
@@ -164,9 +148,39 @@ class RemoveGroupMemberHandler @Inject constructor(
        }

        val node = SnodeAPI.getSingleTargetSnode(groupAccountId.hexString).await()
-        SnodeAPI.getBatchResponse(node, groupAccountId.hexString, batchCalls, true)
+        val response = SnodeAPI.getBatchResponse(node, groupAccountId.hexString, batchCalls, sequence = true)

-        //TODO: Handle message removal
+        val firstError = response.results.firstOrNull { !it.isSuccessful }
+        check(firstError == null) {
+            "Error processing pending removals for group: code = ${firstError?.code}, body = ${firstError?.body}"
+        }
+
+        Log.d(TAG, "Essential steps for group removal are done")
+
+        // The essential part of the operation has been successful once we get to this point,
+        // now we can go ahead and update the configs
+        configFactory.withMutableGroupConfigs(groupAccountId) { configs ->
+            pendingRemovals.forEach(configs.groupMembers::erase)
+            configs.rekey()
+        }
+
+        configFactory.waitUntilGroupConfigsPushed(groupAccountId)
+
+        Log.d(TAG, "Group configs updated")
+
+        // Try to delete members' message. It's ok to fail as they will be re-tried in different
+        // cases (a.k.a the GroupUpdateDeleteMemberContent message handling) and could be by different admins.
+        val deletingMessagesForMembers = pendingRemovals.filter { it.shouldRemoveMessages }
+        if (deletingMessagesForMembers.isNotEmpty()) {
+            try {
+                groupManager.removeMemberMessages(
+                    groupAccountId,
+                    deletingMessagesForMembers.map { AccountId(it.sessionId) }
+                )
+            } catch (e: Exception) {
+                Log.e(TAG, "Error deleting messages for removed members", e)
+            }
+        }
    }

    private fun buildDeleteGroupMemberContentMessage(
@@ -211,7 +225,7 @@ class RemoveGroupMemberHandler @Inject constructor(
                domain = Sodium.KICKED_DOMAIN
            )
        ),
-        ttl = SnodeMessage.CONFIG_TTL,
+        ttl = SnodeMessage.DEFAULT_TTL,
        timestamp = SnodeAPI.nowWithOffset
    )
 }
--- a/libsession/src/main/java/org/session/libsession/messaging/messages/Message.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/messages/Message.kt
@@ -5,6 +5,7 @@ import org.session.libsession.database.StorageProtocol
 import org.session.libsession.messaging.MessagingModuleConfiguration
 import org.session.libsession.messaging.messages.control.ExpirationTimerUpdate
 import org.session.libsession.messaging.messages.visible.VisibleMessage
+import org.session.libsession.snode.SnodeMessage
 import org.session.libsignal.protos.SignalServiceProtos
 import org.session.libsignal.protos.SignalServiceProtos.Content.ExpirationType

@@ -25,7 +26,7 @@ abstract class Message {

    open val coerceDisappearAfterSendToRead = false

-    open val defaultTtl: Long = 14 * 24 * 60 * 60 * 1000
+    open val defaultTtl: Long = SnodeMessage.DEFAULT_TTL
    open val ttl: Long get() = specifiedTtl ?: defaultTtl
    open val isSelfSendValid: Boolean = false

--- a/libsession/src/main/java/org/session/libsession/messaging/sending_receiving/ReceivedMessageHandler.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/sending_receiving/ReceivedMessageHandler.kt
@@ -55,6 +55,7 @@ import org.session.libsignal.crypto.ecc.DjbECPublicKey
 import org.session.libsignal.crypto.ecc.ECKeyPair
 import org.session.libsignal.messages.SignalServiceGroup
 import org.session.libsignal.protos.SignalServiceProtos
+import org.session.libsignal.protos.SignalServiceProtos.DataMessage.GroupUpdateMemberChangeMessage
 import org.session.libsignal.protos.SignalServiceProtos.SharedConfigMessage
 import org.session.libsignal.utilities.AccountId
 import org.session.libsignal.utilities.Base64
--- a/libsession/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/ClosedGroupPoller.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/ClosedGroupPoller.kt
@@ -5,10 +5,10 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.async
-import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.supervisorScope
 import network.loki.messenger.libsession_util.util.Sodium
 import org.session.libsession.database.StorageProtocol
 import org.session.libsession.messaging.groups.GroupManagerV2
@@ -22,7 +22,6 @@ import org.session.libsession.snode.model.RetrieveMessageResponse
 import org.session.libsession.snode.utilities.await
 import org.session.libsession.utilities.ConfigFactoryProtocol
 import org.session.libsession.utilities.ConfigMessage
-import org.session.libsession.utilities.getClosedGroup
 import org.session.libsignal.database.LokiAPIDatabaseProtocol
 import org.session.libsignal.utilities.AccountId
 import org.session.libsignal.utilities.Log
@@ -102,9 +101,9 @@ class ClosedGroupPoller(
        job = null
    }

-    private suspend fun poll(snode: Snode): Unit = coroutineScope {
+    private suspend fun poll(snode: Snode): Unit = supervisorScope {
        val groupAuth =
-            configFactoryProtocol.getGroupAuth(closedGroupSessionId) ?: return@coroutineScope
+            configFactoryProtocol.getGroupAuth(closedGroupSessionId) ?: return@supervisorScope
        val configHashesToExtends = configFactoryProtocol.withGroupConfigs(closedGroupSessionId) {
            buildSet {
                addAll(it.groupKeys.currentHashes())
@@ -121,23 +120,21 @@ class ClosedGroupPoller(

        val pollingTasks = mutableListOf<Pair<String, Deferred<*>>>()

-        pollingTasks += "retrieving revoked messages" to async {
-            handleRevoked(
-                SnodeAPI.sendBatchRequest(
-                    snode,
-                    closedGroupSessionId.hexString,
-                    SnodeAPI.buildAuthenticatedRetrieveBatchRequest(
-                        lastHash = lokiApiDatabase.getLastMessageHashValue(
-                            snode,
-                            closedGroupSessionId.hexString,
-                            Namespace.REVOKED_GROUP_MESSAGES()
-                        ).orEmpty(),
-                        auth = groupAuth,
-                        namespace = Namespace.REVOKED_GROUP_MESSAGES(),
-                        maxSize = null,
-                    ),
-                    RetrieveMessageResponse::class.java
-                )
+        val receiveRevokeMessage = async {
+            SnodeAPI.sendBatchRequest(
+                snode,
+                closedGroupSessionId.hexString,
+                SnodeAPI.buildAuthenticatedRetrieveBatchRequest(
+                    lastHash = lokiApiDatabase.getLastMessageHashValue(
+                        snode,
+                        closedGroupSessionId.hexString,
+                        Namespace.REVOKED_GROUP_MESSAGES()
+                    ).orEmpty(),
+                    auth = groupAuth,
+                    namespace = Namespace.REVOKED_GROUP_MESSAGES(),
+                    maxSize = null,
+                ),
+                RetrieveMessageResponse::class.java
            )
        }

@@ -198,18 +195,28 @@ class ClosedGroupPoller(
            }
        }

-        // The retrieval of the config and regular messages can be done concurrently,
+        // The retrieval of the all group messages can be done concurrently,
        // however, in order for the messages to be able to be decrypted, the config messages
        // must be processed first.
        pollingTasks += "polling and handling group config keys and messages" to async {
-            val (keysMessage, infoMessage, membersMessage) = groupConfigRetrieval.map { it.await() }
-            saveLastMessageHash(snode, keysMessage, Namespace.ENCRYPTION_KEYS())
-            saveLastMessageHash(snode, infoMessage, Namespace.CLOSED_GROUP_INFO())
-            saveLastMessageHash(snode, membersMessage, Namespace.CLOSED_GROUP_MEMBERS())
-            handleGroupConfigMessages(keysMessage, infoMessage, membersMessage)
+            val result = runCatching {
+                val (keysMessage, infoMessage, membersMessage) = groupConfigRetrieval.map { it.await() }
+                handleGroupConfigMessages(keysMessage, infoMessage, membersMessage)
+                saveLastMessageHash(snode, keysMessage, Namespace.ENCRYPTION_KEYS())
+                saveLastMessageHash(snode, infoMessage, Namespace.CLOSED_GROUP_INFO())
+                saveLastMessageHash(snode, membersMessage, Namespace.CLOSED_GROUP_MEMBERS())

-            val regularMessages = groupMessageRetrieval.await()
-            handleMessages(regularMessages, snode)
+                val regularMessages = groupMessageRetrieval.await()
+                handleMessages(regularMessages, snode)
+            }
+
+            // Revoke message must be handled regardless, and at the end
+            val revokedMessages = receiveRevokeMessage.await()
+            handleRevoked(revokedMessages)
+            saveLastMessageHash(snode, revokedMessages, Namespace.REVOKED_GROUP_MESSAGES())
+
+            // Propagate any prior exceptions
+            result.getOrThrow()
        }

        // Wait for all tasks to complete, gather any exceptions happened during polling
@@ -254,23 +261,23 @@ class ClosedGroupPoller(
            )

            if (decoded != null) {
-                Log.d(TAG, "decoded kick message was for us")
                val message = decoded.decodeToString()
-                if (Sodium.KICKED_REGEX.matches(message)) {
-                    val (sessionId, generation) = message.split("-")
-                    val currentKeysGeneration by lazy {
-                        configFactoryProtocol.withGroupConfigs(closedGroupSessionId) {
-                            it.groupKeys.currentGeneration()
-                        }
+                val matcher = Sodium.KICKED_REGEX.matcher(message)
+                if (matcher.matches()) {
+                    val sessionId = matcher.group(1)
+                    val messageGeneration = matcher.group(2)!!.toInt()
+                    val currentKeysGeneration = configFactoryProtocol.withGroupConfigs(closedGroupSessionId) {
+                        it.groupKeys.currentGeneration()
                    }

-                    if (sessionId == storage.getUserPublicKey() && generation.toInt() >= currentKeysGeneration) {
-                        try {
-                            groupManagerV2.handleKicked(closedGroupSessionId)
-                        } catch (e: Exception) {
-                            Log.e("GroupPoller", "Error handling kicked message: $e")
-                        }
+                    val isForMe = sessionId == storage.getUserPublicKey()
+                    Log.d(TAG, "Received kicked message, for us? ${sessionId == storage.getUserPublicKey()}, message key generation = $messageGeneration, our key generation = $currentKeysGeneration")
+
+                    if (isForMe && messageGeneration >= currentKeysGeneration) {
+                        groupManagerV2.handleKicked(closedGroupSessionId)
                    }
+                } else {
+                    Log.w(TAG, "Received an invalid kicked message")
                }
            }
        }
--- a/libsession/src/main/java/org/session/libsession/snode/OnionRequestAPI.kt
+++ b/libsession/src/main/java/org/session/libsession/snode/OnionRequestAPI.kt
@@ -1,6 +1,5 @@
 package org.session.libsession.snode

-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.launch
 import nl.komponents.kovenant.Deferred
@@ -26,12 +25,10 @@ import org.session.libsignal.utilities.HTTP
 import org.session.libsignal.utilities.JsonUtil
 import org.session.libsignal.utilities.Log
 import org.session.libsignal.utilities.Snode
-import org.session.libsignal.utilities.ThreadUtils
 import org.session.libsignal.utilities.recover
 import org.session.libsignal.utilities.toHexString
 import java.util.concurrent.atomic.AtomicReference
 import kotlin.collections.set
-import kotlin.coroutines.EmptyCoroutineContext

 private typealias Path = List<Snode>

@@ -603,11 +600,7 @@ object OnionRequestAPI {
                                val bodyAsString = json["body"] as String
                                JsonUtil.fromJson(bodyAsString, Map::class.java)
                            }
-                            if (body["t"] != null) {
-                                val timestamp = body["t"] as Long
-                                val offset = timestamp - System.currentTimeMillis()
-                                SnodeAPI.clockOffset = offset
-                            }
+
                            if (body.containsKey("hf")) {
                                @Suppress("UNCHECKED_CAST")
                                val currentHf = body["hf"] as List<Int>
--- a/libsession/src/main/java/org/session/libsession/snode/SnodeAPI.kt
+++ b/libsession/src/main/java/org/session/libsession/snode/SnodeAPI.kt
@@ -25,6 +25,7 @@ import nl.komponents.kovenant.functional.bind
 import nl.komponents.kovenant.functional.map
 import nl.komponents.kovenant.task
 import nl.komponents.kovenant.unwrap
+import org.session.libsession.messaging.MessagingModuleConfiguration
 import org.session.libsession.messaging.utilities.MessageWrapper
 import org.session.libsession.messaging.utilities.SodiumUtilities.sodium
 import org.session.libsession.snode.model.BatchResponse
@@ -47,6 +48,7 @@ import org.session.libsignal.utilities.Log
 import org.session.libsignal.utilities.Snode
 import org.session.libsignal.utilities.prettifiedDescription
 import org.session.libsignal.utilities.retryIfNeeded
+import java.util.Date
 import java.util.Locale
 import kotlin.collections.component1
 import kotlin.collections.component2
@@ -63,15 +65,11 @@ object SnodeAPI {
    internal var snodePool: Set<Snode>
        get() = database.getSnodePool()
        set(newValue) { database.setSnodePool(newValue) }
-    /**
-     * The offset between the user's clock and the Service Node's clock. Used in cases where the
-     * user's clock is incorrect.
-     */
-    internal var clockOffset = 0L

+    @Deprecated("Use a dependency injected SnodeClock.currentTimeMills() instead")
    @JvmStatic
    val nowWithOffset
-        get() = System.currentTimeMillis() + clockOffset
+        get() = MessagingModuleConfiguration.shared.clock.currentTimeMills()

    internal var forkInfo by observable(database.getForkInfo()) { _, oldValue, newValue ->
        if (newValue > oldValue) {
@@ -418,7 +416,6 @@ object SnodeAPI {
            namespace = namespace,
            auth = auth,
            verificationData = { ns, t -> "${Snode.Method.SendMessage.rawValue}$ns$t" },
-            timestamp = message.timestamp
        ) {
            putAll(message.toJSON())
        }
@@ -785,7 +782,7 @@ object SnodeAPI {
        parseRawMessagesResponse(resp, snode, auth.accountId.hexString)
    }

-    private fun getNetworkTime(snode: Snode): Promise<Pair<Snode, Long>, Exception> =
+    fun getNetworkTime(snode: Snode): Promise<Pair<Snode, Long>, Exception> =
        invoke(Snode.Method.Info, snode, emptyMap()).map { rawResponse ->
            val timestamp = rawResponse["timestamp"] as? Long ?: -1
            snode to timestamp
@@ -805,13 +802,15 @@ object SnodeAPI {
                "Message sent to ${message.recipient} but authenticated with ${auth.accountId.hexString}"
            }

+            val timestamp = nowWithOffset
+
            buildAuthenticatedParameters(
                auth = auth,
                namespace = namespace,
                verificationData = { ns, t -> "${Snode.Method.SendMessage.rawValue}$ns$t" },
-                timestamp = message.timestamp
+                timestamp = timestamp
            ) {
-                put("sig_timestamp", message.timestamp)
+                put("sig_timestamp", timestamp)
                putAll(message.toJSON())
            }
        } else {
@@ -921,7 +920,7 @@ object SnodeAPI {
    fun deleteAllMessages(auth: SwarmAuth): Promise<Map<String, Boolean>, Exception> =
        scope.retrySuspendAsPromise(maxRetryCount) {
            val snode = getSingleTargetSnode(auth.accountId.hexString).await()
-            val (_, timestamp) = getNetworkTime(snode).await()
+            val timestamp = MessagingModuleConfiguration.shared.clock.waitForNetworkAdjustedTime()

            val params = buildAuthenticatedParameters(
                auth = auth,
--- a/libsession/src/main/java/org/session/libsession/snode/SnodeClock.kt
+++ b/libsession/src/main/java/org/session/libsession/snode/SnodeClock.kt
@@ -0,0 +1,88 @@
+package org.session.libsession.snode
+
+import android.os.SystemClock
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.launch
+import org.session.libsession.snode.utilities.await
+import org.session.libsignal.utilities.Log
+import java.util.Date
+
+/**
+ * A class that manages the network time by querying the network time from a random snode. The
+ * primary goal of this class is to provide a time that is not tied to current system time and not
+ * prone to time changes locally.
+ *
+ * Before the first network query is successfully, calling [currentTimeMills] will return the current
+ * system time.
+ */
+class SnodeClock() {
+    private val instantState = MutableStateFlow<Instant?>(null)
+    private var job: Job? = null
+
+    fun start() {
+        require(job == null) { "Already started" }
+
+        job = GlobalScope.launch {
+            while (true) {
+                try {
+                    val node = SnodeAPI.getRandomSnode().await()
+                    val requestStarted = SystemClock.uptimeMillis()
+
+                    var networkTime = SnodeAPI.getNetworkTime(node).await().second
+                    val requestEnded = SystemClock.uptimeMillis()
+
+                    // Adjust the network time to account for the time it took to make the request
+                    // so that the network time equals to the time when the request was started
+                    networkTime -= (requestEnded - requestStarted) / 2
+
+                    val inst = Instant(requestStarted, networkTime)
+
+                    Log.d("SnodeClock", "Network time: ${Date(inst.now())}, system time: ${Date()}")
+
+                    instantState.value = inst
+                } catch (e: Exception) {
+                    Log.e("SnodeClock", "Failed to get network time. Retrying in a few seconds", e)
+                } finally {
+                    // Retry frequently if we haven't got any result before
+                    val delayMills = if (instantState.value == null) {
+                        3_000L
+                    } else {
+                        3600_000L
+                    }
+
+                    delay(delayMills)
+                }
+            }
+        }
+    }
+
+    /**
+     * Wait for the network adjusted time to come through.
+     */
+    suspend fun waitForNetworkAdjustedTime(): Long {
+        return instantState.filterNotNull().first().now()
+    }
+
+    /**
+     * Get the current time in milliseconds. If the network time is not available yet, this method
+     * will return the current system time.
+     */
+    fun currentTimeMills(): Long {
+        return instantState.value?.now() ?: System.currentTimeMillis()
+    }
+
+    private class Instant(
+        val systemUptime: Long,
+        val networkTime: Long,
+    ) {
+        fun now(): Long {
+            val elapsed = SystemClock.uptimeMillis() - systemUptime
+            return networkTime + elapsed
+        }
+    }
+}
--- a/libsession/src/main/java/org/session/libsession/snode/SnodeMessage.kt
+++ b/libsession/src/main/java/org/session/libsession/snode/SnodeMessage.kt
@@ -32,6 +32,7 @@ data class SnodeMessage(
    }

    companion object {
-        const val CONFIG_TTL: Long = 30 * 24 * 60 * 60 * 1000L
+        const val CONFIG_TTL: Long = 30 * 24 * 60 * 60 * 1000L // 30 days
+        const val DEFAULT_TTL: Long = 14 * 24 * 60 * 60 * 1000L // 14 days
    }
 }
--- a/libsession/src/main/java/org/session/libsession/utilities/ConfigFactoryProtocol.kt
+++ b/libsession/src/main/java/org/session/libsession/utilities/ConfigFactoryProtocol.kt
@@ -1,6 +1,11 @@
 package org.session.libsession.utilities

 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.onStart
+import kotlinx.coroutines.withTimeoutOrNull
 import network.loki.messenger.libsession_util.MutableConfig
 import network.loki.messenger.libsession_util.MutableContacts
 import network.loki.messenger.libsession_util.MutableConversationVolatileConfig
@@ -94,6 +99,52 @@ fun ConfigFactoryProtocol.getClosedGroup(groupId: AccountId): GroupInfo.ClosedGr
    return withUserConfigs { it.userGroups.getClosedGroup(groupId.hexString) }
 }

+/**
+ * Wait until all user configs are pushed to the server.
+ *
+ * This function is not essential to the pushing of the configs, the config push will schedule
+ * itself upon changes, so this function is purely observatory.
+ *
+ * This function will check the user configs immediately, if nothing needs to be pushed, it will return immediately.
+ *
+ * @return True if all user configs are pushed, false if the timeout is reached.
+ */
+suspend fun ConfigFactoryProtocol.waitUntilUserConfigsPushed(timeoutMills: Long = 10_000L): Boolean {
+    fun needsPush() = withUserConfigs { configs ->
+        UserConfigType.entries.any { configs.getConfig(it).needsPush() }
+    }
+
+    return withTimeoutOrNull(timeoutMills){
+        configUpdateNotifications
+            .onStart { emit(ConfigUpdateNotification.UserConfigs) } // Trigger the filtering immediately
+            .filter { it == ConfigUpdateNotification.UserConfigs && !needsPush() }
+            .first()
+    } != null
+}
+
+/**
+ * Wait until all configs of given group are pushed to the server.
+ *
+ * This function is not essential to the pushing of the configs, the config push will schedule
+ * itself upon changes, so this function is purely observatory.
+ *
+ * This function will check the group configs immediately, if nothing needs to be pushed, it will return immediately.
+ *
+ * @return True if all group configs are pushed, false if the timeout is reached.
+ */
+suspend fun ConfigFactoryProtocol.waitUntilGroupConfigsPushed(groupId: AccountId, timeoutMills: Long = 10_000L): Boolean {
+    fun needsPush() = withGroupConfigs(groupId) { configs ->
+        configs.groupInfo.needsPush() || configs.groupMembers.needsPush()
+    }
+
+    return withTimeoutOrNull(timeoutMills) {
+        configUpdateNotifications
+            .onStart { emit(ConfigUpdateNotification.GroupConfigsUpdated(groupId)) } // Trigger the filtering immediately
+            .filter { it == ConfigUpdateNotification.GroupConfigsUpdated(groupId) && !needsPush() }
+            .first()
+    } != null
+}
+
 interface UserConfigs {
    val contacts: ReadableContacts
    val userGroups: ReadableUserGroupsConfig