agent: resolve potential race conditions in double ratchet (#1923)

* agent: resolve potential race conditions in double ratchet * update * improve * diff * diff * diff * simpler * more races --------- Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com>
2026-10-23 00:02:44 +00:00 · 2026-10-04 22:28:50 +01:00
1 parent fcc5670ae2
commit 65893a359a
@@ -208,7 +208,7 @@ This syntax of decrypted SMP client message body is defined by `decryptedAgentMe
 Decrypted SMP message client body can be one of 4 types:
 - `agentConnInfo` - used by the initiating party when confirming reply queue - sent in `agentConfirmation` envelope.
 - `agentConnInfoReply` - used by accepting party, includes reply queue(s) in the initial confirmation - sent in `agentConfirmation` envelope.
- `agentRatchetInfo` - used to pass additional information when renegotiating double ratchet encryption - sent in `agentRatchetKey` envelope.
+- `agentRatchetInfo` - used to pass additional information when renegotiating double ratchet encryption - sent in `agentRatchetKey` envelope. A key sent in reply to another key includes the hash of that key; agents do not reply to such keys.
 - `agentMessage` - all other agent messages.

 `agentMessage` contains these parts:
@@ -233,7 +233,8 @@ connInfo = *OCTET
 agentConnInfoReply = %s"D" smpQueues connInfo
 smpQueues = length 1*newQueueInfo ; NonEmpty list of reply queues
 agentRatchetInfo = %s"R" ratchetInfo
-ratchetInfo = *OCTET
+ratchetInfo = [answeredKeyHash *OCTET] ; bytes after answeredKeyHash are ignored
+answeredKeyHash = %s"0" / (%s"1" shortString) ; "0" in a key that starts renegotiation, otherwise SHA-256 of the two raw public keys of the answered key

 agentMessage = %s"M" agentMsgHeader aMessage
 agentMsgHeader = agentMsgId prevMsgHash
@@ -2435,62 +2435,66 @@ enqueueMessageB c reqs = do
  cfg <- asks config
  (_, reqMids) <- unsafeWithStore c $ \db -> do
    mapAccumLM (\ids r -> storeSentMsg db cfg ids r `E.catchAny` \e -> (ids,) <$> handleInternal e) IM.empty reqs
-  forME reqMids $ \((csqs_, _, _, _), InternalId msgId, pqSecr) -> forM csqs_ $ \(cData, sq :| sqs) -> do
-    submitPendingMsg c sq
-    let sqs' = filter (isActiveSndQ cData) sqs
-    pure ((msgId, pqSecr), if null sqs' then Nothing else Just (sqs', msgId))
+  forME reqMids $ submitSentMsg c
  where
-    storeSentMsg ::
-      DB.Connection ->
-      AgentConfig ->
-      IntMap (Maybe Int64, AMessage) ->
-      Either AgentErrorType (Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage) ->
-      IO (IntMap (Maybe Int64, AMessage), Either AgentErrorType ((Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage), InternalId, PQEncryption))
-    storeSentMsg db cfg aMessageIds = \case
-      Left e -> pure (aMessageIds, Left e)
-      Right req@(csqs_, pqEnc_, msgFlags, mbr) -> case mbr of
-        VRValue i_ aMessage -> case i_ >>= (`IM.lookup` aMessageIds) of
-          Just _ -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg duplicate saved message body")
-          Nothing -> do
-            (mbId_, r) <- case csqs_ of
-              Left e -> pure (Nothing, Left e)
-              Right (cData, sq :| _) -> do
-                mbId <- createSndMsgBody db aMessage
-                (Just mbId,) <$> storeSentMsg_ cData sq mbId aMessage
-            let aMessageIds' = maybe id (`IM.insert` (mbId_, aMessage)) i_ aMessageIds
-            pure (aMessageIds', r)
-        VRRef i -> case csqs_ of
-          Left e -> pure $ (aMessageIds, Left e)
-          Right (cData, sq :| _) -> case IM.lookup i aMessageIds of
-            Just (Just mbId, aMessage) -> (aMessageIds,) <$> storeSentMsg_ cData sq mbId aMessage
-            Just (Nothing, aMessage) -> do
-              mbId <- createSndMsgBody db aMessage
-              let aMessageIds' = IM.insert i (Just mbId, aMessage) aMessageIds
-              (aMessageIds',) <$> storeSentMsg_ cData sq mbId aMessage
-            Nothing -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg missing saved message body id")
-        where
-          storeSentMsg_ cData@ConnData {connId} sq sndMsgBodyId aMessage = fmap (first storeError) $ runExceptT $ do
-            let AgentConfig {e2eEncryptVRange} = cfg
-            internalTs <- liftIO getCurrentTime
-            (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId
-            -- We need to do pre-flight encoding that is not stored in database
-            -- to calculate its hash and remember it on connection (createSndMsg -> updateSndMsgHash)
-            -- to enable next enqueue.
-            -- (As encoding is different per connection, we can't store shared body, so it's repeated on delivery)
-            let agentMsgStr = encodeAgentMsgStr aMessage internalSndId prevMsgHash
-                internalHash = C.sha256Hash agentMsgStr
-                currentE2EVersion = maxVersion e2eEncryptVRange
-            (mek, paddedLen, pqEnc) <- agentRatchetEncryptHeader db cData e2eEncAgentMsgLength pqEnc_ currentE2EVersion
-            withExceptT (SEAgentError . cryptoError) $ CR.rcCheckCanPad paddedLen agentMsgStr
-            let msgType = aMessageType aMessage
-                -- msgBody is empty, because snd_messages record is linked to snd_message_bodies
-                msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody = "", pqEncryption = pqEnc, internalHash, prevMsgHash, sndMsgPrepData_ = Just SndMsgPrepData {encryptKey = mek, paddedLen, sndMsgBodyId}}
-            liftIO $ createSndMsg db connId msgData
-            liftIO $ createSndMsgDelivery db sq internalId
-            pure (req, internalId, pqEnc)
    handleInternal :: E.SomeException -> IO (Either AgentErrorType b)
    handleInternal = pure . Left . INTERNAL . show

+submitSentMsg :: AgentClient -> ((Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage), InternalId, PQEncryption) -> AM' (Either AgentErrorType ((AgentMsgId, PQEncryption), Maybe ([SndQueue], AgentMsgId)))
+submitSentMsg c ((csqs_, _, _, _), InternalId msgId, pqSecr) = forM csqs_ $ \(cData, sq :| sqs) -> do
+  submitPendingMsg c sq
+  let sqs' = filter (isActiveSndQ cData) sqs
+  pure ((msgId, pqSecr), if null sqs' then Nothing else Just (sqs', msgId))
+
+storeSentMsg ::
+  DB.Connection ->
+  AgentConfig ->
+  IntMap (Maybe Int64, AMessage) ->
+  Either AgentErrorType (Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage) ->
+  IO (IntMap (Maybe Int64, AMessage), Either AgentErrorType ((Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage), InternalId, PQEncryption))
+storeSentMsg db cfg aMessageIds = \case
+  Left e -> pure (aMessageIds, Left e)
+  Right req@(csqs_, pqEnc_, msgFlags, mbr) -> case mbr of
+    VRValue i_ aMessage -> case i_ >>= (`IM.lookup` aMessageIds) of
+      Just _ -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg duplicate saved message body")
+      Nothing -> do
+        (mbId_, r) <- case csqs_ of
+          Left e -> pure (Nothing, Left e)
+          Right (cData, sq :| _) -> do
+            mbId <- createSndMsgBody db aMessage
+            (Just mbId,) <$> storeSentMsg_ cData sq mbId aMessage
+        let aMessageIds' = maybe id (`IM.insert` (mbId_, aMessage)) i_ aMessageIds
+        pure (aMessageIds', r)
+    VRRef i -> case csqs_ of
+      Left e -> pure $ (aMessageIds, Left e)
+      Right (cData, sq :| _) -> case IM.lookup i aMessageIds of
+        Just (Just mbId, aMessage) -> (aMessageIds,) <$> storeSentMsg_ cData sq mbId aMessage
+        Just (Nothing, aMessage) -> do
+          mbId <- createSndMsgBody db aMessage
+          let aMessageIds' = IM.insert i (Just mbId, aMessage) aMessageIds
+          (aMessageIds',) <$> storeSentMsg_ cData sq mbId aMessage
+        Nothing -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg missing saved message body id")
+    where
+      storeSentMsg_ cData@ConnData {connId} sq sndMsgBodyId aMessage = fmap (first storeError) $ runExceptT $ do
+        let AgentConfig {e2eEncryptVRange} = cfg
+        internalTs <- liftIO getCurrentTime
+        (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId
+        -- We need to do pre-flight encoding that is not stored in database
+        -- to calculate its hash and remember it on connection (createSndMsg -> updateSndMsgHash)
+        -- to enable next enqueue.
+        -- (As encoding is different per connection, we can't store shared body, so it's repeated on delivery)
+        let agentMsgStr = encodeAgentMsgStr aMessage internalSndId prevMsgHash
+            internalHash = C.sha256Hash agentMsgStr
+            currentE2EVersion = maxVersion e2eEncryptVRange
+        (mek, paddedLen, pqEnc) <- agentRatchetEncryptHeader db cData e2eEncAgentMsgLength pqEnc_ currentE2EVersion
+        withExceptT (SEAgentError . cryptoError) $ CR.rcCheckCanPad paddedLen agentMsgStr
+        let msgType = aMessageType aMessage
+            -- msgBody is empty, because snd_messages record is linked to snd_message_bodies
+            msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody = "", pqEncryption = pqEnc, internalHash, prevMsgHash, sndMsgPrepData_ = Just SndMsgPrepData {encryptKey = mek, paddedLen, sndMsgBodyId}}
+        liftIO $ createSndMsg db connId msgData
+        liftIO $ createSndMsgDelivery db sq internalId
+        pure (req, internalId, pqEnc)
+
 encodeAgentMsgStr :: AMessage -> InternalSndId -> PrevSndMsgHash -> ByteString
 encodeAgentMsgStr aMessage internalSndId prevMsgHash = do
  let privHeader = APrivHeader (unSndId internalSndId) prevMsgHash
@@ -2865,15 +2869,18 @@ synchronizeRatchet' c connId pqSupport' force = withConnLock c connId "synchroni
    SomeConn _ (DuplexConnection cData@ConnData {pqSupport} rqs sqs)
      | ratchetSyncAllowed cData || force -> do
          -- check queues are not switching?
-          when (pqSupport' /= pqSupport) $ withStore' c $ \db -> setConnPQSupport db connId pqSupport'
          let cData' = cData {pqSupport = pqSupport'} :: ConnData
-          AgentConfig {e2eEncryptVRange} <- asks config
+          AgentConfig {e2eEncryptVRange, smpAgentVRange} <- asks config
          g <- asks random
          (pks, e2eParams) <- liftIO $ CR.generateRcvE2EParams g (maxVersion e2eEncryptVRange) pqSupport'
-          enqueueRatchetKeyMsgs c cData' sqs e2eParams
-          withStore' c $ \db -> do
-            setConnRatchetSync db connId RSStarted
-            setRatchetX3dhKeys db connId pks
+          msgId <- withStore c $ \db -> runExceptT $ do
+            msgId <- storeRatchetKey db (L.head sqs) (maxVersion smpAgentVRange) e2eParams Nothing
+            liftIO $ do
+              when (pqSupport' /= pqSupport) $ setConnPQSupport db connId pqSupport'
+              setConnRatchetSync db connId RSStarted
+              setRatchetX3dhKeys db connId pks
+            pure msgId
+          lift $ enqueueRatchetKeyMsgs c cData' sqs msgId
          let cData'' = cData' {ratchetSyncState = RSStarted} :: ConnData
              conn' = DuplexConnection cData'' rqs sqs
          connectionStats c conn'
@@ -3613,9 +3620,9 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                      _ -> prohibited "handshake: incorrect state" >> ack
                  (Just e2eDh, Nothing) -> do
                    decryptClientMessage e2eDh clientMsg >>= \case
-                      (SMP.PHEmpty, AgentRatchetKey {agentVersion, e2eEncryption}) -> do
+                      (SMP.PHEmpty, AgentRatchetKey {agentVersion, e2eEncryption, info}) -> do
                        conn' <- updateConnVersion conn cData agentVersion
-                        qDuplex conn' "AgentRatchetKey" $ \a -> newRatchetKey e2eEncryption a >> ack
+                        qDuplex conn' "AgentRatchetKey" $ \a -> newRatchetKey e2eEncryption info a >> ack
                      (SMP.PHEmpty, AgentMsgEnvelope {agentVersion, encAgentMessage}) -> do
                        conn' <- updateConnVersion conn cData agentVersion
                        -- primary queue is set as Active in helloMsg, below is to set additional queues Active
@@ -3658,7 +3665,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                              qDuplexAckDel conn'' name a = qDuplex conn'' name a >> ackDel msgId
                              resetRatchetSync :: AM (Connection c)
                              resetRatchetSync
-                                | rss `notElem` ([RSOk, RSStarted] :: [RatchetSyncState]) = do
+                                | rss `notElem` ([RSOk, RSStarted] :: [RatchetSyncState]) = ifM ((RSStarted ==) <$> getRatchetSyncState) (pure conn') $ do
                                    let cData'' = (toConnData conn') {ratchetSyncState = RSOk} :: ConnData
                                        conn'' = updateConnection cData'' conn'
                                    cStats <- connectionStats c conn''
@@ -3699,7 +3706,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                              notifySync :: AM ()
                              notifySync = qDuplex conn' "AGENT A_CRYPTO error" $ \connDuplex -> do
                                let rss' = cryptoErrToSyncState e
-                                when (rss `elem` ([RSOk, RSAllowed, RSRequired] :: [RatchetSyncState])) $ do
+                                whenM ((`elem` ([RSOk, RSAllowed, RSRequired] :: [RatchetSyncState])) <$> getRatchetSyncState) $ do
                                  let cData'' = (toConnData conn') {ratchetSyncState = rss'} :: ConnData
                                      conn'' = updateConnection cData'' connDuplex
                                  cStats <- connectionStats c conn''
@@ -3795,6 +3802,9 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
          enqueueCmd :: InternalCommand -> AM ()
          enqueueCmd = enqueueCommand c "" connId (Just srv) . AInternalCommand

+          getRatchetSyncState :: AM RatchetSyncState
+          getRatchetSyncState = withStore c (`getConnRatchetSync` connId)
+
          unexpected :: BrokerMsg -> AM ()
          unexpected r = do
            logServer "<--" c srv rId $ "unexpected: " <> bshow r
@@ -4163,42 +4173,47 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
            DuplexConnection {} -> action conn'
            _ -> qError $ name <> ": message must be sent to duplex connection"

-          newRatchetKey :: CR.RcvE2ERatchetParams 'C.X448 -> Connection 'CDuplex -> AM ()
-          newRatchetKey e2eOtherPartyParams@(CR.E2ERatchetParams e2eVersion k1Rcv k2Rcv _) conn'@(DuplexConnection cData'@ConnData {lastExternalSndId, pqSupport} _ sqs) =
+          newRatchetKey :: CR.RcvE2ERatchetParams 'C.X448 -> ByteString -> Connection 'CDuplex -> AM ()
+          newRatchetKey e2eOtherPartyParams@(CR.E2ERatchetParams e2eVersion k1Rcv k2Rcv _) info conn'@(DuplexConnection cData'@ConnData {lastExternalSndId, pqSupport} _ sqs) =
            unlessM ratchetExists $ do
              AgentConfig {e2eEncryptVRange} <- asks config
              unless (e2eVersion `isCompatible` e2eEncryptVRange) (throwE $ AGENT A_VERSION)
-              keys <- getSendRatchetKeys
+              keys_ <- getSendRatchetKeys
              let rcVs = CR.RatchetVersions {current = e2eVersion, maxSupported = maxVersion e2eEncryptVRange}
-              initRatchet rcVs keys
-              notifyAgreed
+              case keys_ of
+                Just (keys, replyKey_) -> initRatchet rcVs keys replyKey_ >> notifyAgreed
+                Nothing -> withStore' c $ \db -> addProcessedRatchetKeyHash db connId rkHashRcv
            where
              rkHashRcv = rkHash k1Rcv k2Rcv
              rkHash k1 k2 = C.sha256Hash $ C.pubKeyBytes k1 <> C.pubKeyBytes k2
+              answeredKeyHash_ = case smpDecode info of
+                Right (AgentRatchetInfo RatchetInfo {answeredKeyHash}) -> answeredKeyHash
+                _ -> Nothing
              ratchetExists :: AM Bool
              ratchetExists = withStore' c $ \db -> do
                exists <- checkRatchetKeyHashExists db connId rkHashRcv
-                unless exists $ addProcessedRatchetKeyHash db connId rkHashRcv
                pure exists
-              getSendRatchetKeys :: AM (CR.RcvE2EPrivRatchetParams 'C.X448)
-              getSendRatchetKeys = case rss of
-                RSOk -> sendReplyKey -- receiving client
-                RSAllowed -> sendReplyKey
-                RSRequired -> sendReplyKey
-                RSStarted -> withStore c (`getRatchetX3dhKeys` connId) -- initiating client
-                RSAgreed -> do
-                  withStore' c $ \db -> setConnRatchetSync db connId RSRequired
+              getSendRatchetKeys :: AM (Maybe (CR.RcvE2EPrivRatchetParams 'C.X448, Maybe (CR.RcvE2ERatchetParams 'C.X448)))
+              getSendRatchetKeys = getRatchetSyncState >>= \rss' -> case (rss', answeredKeyHash_) of
+                (RSStarted, Nothing) -> Just . (,Nothing) <$> withStore c (`getRatchetX3dhKeys` connId) -- initiating client
+                (RSStarted, Just h) -> fmap (,Nothing) . answeredKeys h <$> withStore c (`getRatchetX3dhKeys` connId)
+                (_, Nothing) -> Just <$> sendReplyKey -- receiving client
+                (RSAgreed, Just _) -> do
+                  withStore' c $ \db -> addProcessedRatchetKeyHash db connId rkHashRcv >> setConnRatchetSync db connId RSRequired
                  notifyRatchetSyncError
                  -- can communicate for other client to reset to RSRequired
                  -- - need to add new AgentMsgEnvelope, AgentMessage, AgentMessageType
                  -- - need to deduplicate on receiving side
                  throwE $ AGENT (A_CRYPTO RATCHET_SYNC)
+                (_, Just _) -> pure Nothing
                where
+                  answeredKeys h keys@(pk1, pk2, _)
+                    | rkHash (C.publicKey pk1) (C.publicKey pk2) == h = Just keys
+                    | otherwise = Nothing
                  sendReplyKey = do
                    g <- asks random
                    (pks, e2eParams) <- liftIO $ CR.generateRcvE2EParams g e2eVersion pqSupport
-                    enqueueRatchetKeyMsgs c cData' sqs e2eParams
-                    pure pks
+                    pure (pks, Just e2eParams)
                  notifyRatchetSyncError = do
                    let cData'' = cData' {ratchetSyncState = RSRequired} :: ConnData
                        conn'' = updateConnection cData'' conn'
@@ -4210,23 +4225,33 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                    conn'' = updateConnection cData'' conn'
                cStats <- connectionStats c conn''
                notify $ RSYNC RSAgreed Nothing cStats
-              recreateRatchet :: CR.Ratchet 'C.X448 -> AM ()
-              recreateRatchet rc = withStore' c $ \db -> do
-                setConnRatchetSync db connId RSAgreed
-                deleteRatchet db connId
-                createRatchet db connId rc
+              recreateRatchet :: Maybe (CR.RcvE2ERatchetParams 'C.X448) -> Maybe AMessage -> CR.Ratchet 'C.X448 -> AM ()
+              recreateRatchet replyKey_ eready_ rc = do
+                cfg@AgentConfig {smpAgentVRange} <- asks config
+                (msgId_, r_) <- withStore c $ \db -> runExceptT $ do
+                  msgId_ <- forM replyKey_ $ \e2eParams -> storeRatchetKey db (L.head sqs) (maxVersion smpAgentVRange) e2eParams (Just rkHashRcv)
+                  liftIO $ do
+                    addProcessedRatchetKeyHash db connId rkHashRcv
+                    setConnRatchetSync db connId RSAgreed
+                    deleteRatchet db connId
+                    createRatchet db connId rc
+                  r_ <- forM eready_ $ \msg -> liftIO $ storeSentMsg db cfg IM.empty (Right (Right (cData', sqs), Nothing, SMP.MsgFlags {notification = True}, vrValue msg)) >>= either E.throwIO pure . snd
+                  pure (msgId_, r_)
+                forM_ msgId_ $ lift . enqueueRatchetKeyMsgs c cData' sqs
+                forM_ r_ $ \r -> lift $ do
+                  r' <- submitSentMsg c r
+                  enqueueSavedMessageB c $ mapMaybe snd $ rights [r']
              -- compare public keys `k1` in AgentRatchetKey messages sent by self and other party
              -- to determine ratchet initilization ordering
-              initRatchet :: CR.RatchetVersions -> CR.RcvE2EPrivRatchetParams 'C.X448 -> AM ()
-              initRatchet rcVs (pk1, pk2, pKem)
+              initRatchet :: CR.RatchetVersions -> CR.RcvE2EPrivRatchetParams 'C.X448 -> Maybe (CR.RcvE2ERatchetParams 'C.X448) -> AM ()
+              initRatchet rcVs (pk1, pk2, pKem) replyKey_
                | rkHash (C.publicKey pk1) (C.publicKey pk2) <= rkHashRcv = do
                    rcParams <- liftError cryptoError $ CR.pqX3dhRcv (pk1, pk2, pKem) e2eOtherPartyParams
-                    recreateRatchet $ CR.initRcvRatchet rcVs pk2 rcParams pqSupport
+                    recreateRatchet replyKey_ Nothing $ CR.initRcvRatchet rcVs pk2 rcParams pqSupport
                | otherwise = do
                    (_, rcDHRs) <- atomically . C.generateKeyPair =<< asks random
                    rcParams <- liftEitherWith cryptoError $ CR.pqX3dhSnd (pk1, pk2, CR.APRKP CR.SRKSProposed <$> pKem) e2eOtherPartyParams
-                    recreateRatchet $ CR.initSndRatchet rcVs k2Rcv rcDHRs rcParams
-                    void . enqueueMessages' c cData' sqs SMP.MsgFlags {notification = True} $ EREADY lastExternalSndId
+                    recreateRatchet replyKey_ (Just $ EREADY lastExternalSndId) $ CR.initSndRatchet rcVs k2Rcv rcDHRs rcParams

          checkMsgIntegrity :: PrevExternalSndId -> ExternalSndId -> PrevRcvMsgHash -> ByteString -> MsgIntegrity
          checkMsgIntegrity prevExtSndId extSndId internalPrevMsgHash receivedPrevMsgHash
@@ -4339,32 +4364,25 @@ storeConfirmation c cData@ConnData {connId, pqSupport, connAgentVersion = v} sq
    liftIO $ createSndMsg db connId msgData
    liftIO $ createSndMsgDelivery db sq internalId

-enqueueRatchetKeyMsgs :: AgentClient -> ConnData -> NonEmpty SndQueue -> CR.RcvE2ERatchetParams 'C.X448 -> AM ()
-enqueueRatchetKeyMsgs c cData (sq :| sqs) e2eEncryption = do
-  msgId <- enqueueRatchetKey c sq e2eEncryption
-  mapM_ (lift . enqueueSavedMessage c msgId) $ filter (isActiveSndQ cData) sqs
+enqueueRatchetKeyMsgs :: AgentClient -> ConnData -> NonEmpty SndQueue -> InternalId -> AM' ()
+enqueueRatchetKeyMsgs c cData (sq :| sqs) (InternalId msgId) = do
+  submitPendingMsg c sq
+  mapM_ (enqueueSavedMessage c msgId) $ filter (isActiveSndQ cData) sqs

-enqueueRatchetKey :: AgentClient -> SndQueue -> CR.RcvE2ERatchetParams 'C.X448 -> AM AgentMsgId
-enqueueRatchetKey c sq@SndQueue {connId} e2eEncryption = do
-  aVRange <- asks $ smpAgentVRange . config
-  msgId <- storeRatchetKey $ maxVersion aVRange
-  lift $ submitPendingMsg c sq
-  pure $ unId msgId
-  where
-    storeRatchetKey :: VersionSMPA -> AM InternalId
-    storeRatchetKey agentVersion = withStore c $ \db -> runExceptT $ do
-      internalTs <- liftIO getCurrentTime
-      (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId
-      let agentMsg = AgentRatchetInfo ""
-          agentMsgStr = smpEncode agentMsg
-          internalHash = C.sha256Hash agentMsgStr
-      let msgBody = smpEncode $ AgentRatchetKey {agentVersion, e2eEncryption, info = agentMsgStr}
-          msgType = agentMessageType agentMsg
-          -- this message is e2e encrypted with queue key, not with double ratchet
-          msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, pqEncryption = PQEncOff, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash, sndMsgPrepData_ = Nothing}
-      liftIO $ createSndMsg db connId msgData
-      liftIO $ createSndMsgDelivery db sq internalId
-      pure internalId
+storeRatchetKey :: DB.Connection -> SndQueue -> VersionSMPA -> CR.RcvE2ERatchetParams 'C.X448 -> Maybe ByteString -> ExceptT StoreError IO InternalId
+storeRatchetKey db sq@SndQueue {connId} agentVersion e2eEncryption answeredKeyHash_ = do
+  internalTs <- liftIO getCurrentTime
+  (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId
+  let agentMsg = AgentRatchetInfo RatchetInfo {answeredKeyHash = answeredKeyHash_}
+      agentMsgStr = smpEncode agentMsg
+      internalHash = C.sha256Hash agentMsgStr
+  let msgBody = smpEncode $ AgentRatchetKey {agentVersion, e2eEncryption, info = agentMsgStr}
+      msgType = agentMessageType agentMsg
+      -- this message is e2e encrypted with queue key, not with double ratchet
+      msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, pqEncryption = PQEncOff, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash, sndMsgPrepData_ = Nothing}
+  liftIO $ createSndMsg db connId msgData
+  liftIO $ createSndMsgDelivery db sq internalId
+  pure internalId

 -- encoded AgentMessage -> encoded EncAgentMessage
 agentRatchetEncrypt :: DB.Connection -> ConnData -> ByteString -> (PQSupport -> Int) -> Maybe PQEncryption -> CR.VersionE2E -> ExceptT StoreError IO (ByteString, PQEncryption)
@@ -79,6 +79,7 @@ module Simplex.Messaging.Agent.Protocol
    SMPConfirmation (..),
    AgentMsgEnvelope (..),
    AgentMessage (..),
+    RatchetInfo (..),
    RequestSignature (..),
    AgentMessageType (..),
    APrivHeader (..),
@@ -928,7 +929,7 @@ data AgentMessage
  | -- AgentConnInfoReply is used by accepting party in duplexHandshake mode (v2), allowing to include reply queue(s) in the initial confirmation.
    -- It made removed REPLY message unnecessary.
    AgentConnInfoReply (NonEmpty SMPQueueInfo) ConnInfo
-  | AgentRatchetInfo ByteString
+  | AgentRatchetInfo RatchetInfo
  | AgentMessage APrivHeader AMessage
  | AgentServiceRequest (NonEmpty SMPQueueInfo) (Maybe RequestSignature) MsgBody
  | AgentServiceResponse MsgBody
@@ -939,7 +940,7 @@ instance Encoding AgentMessage where
  smpEncode = \case
    AgentConnInfo cInfo -> smpEncode ('I', Tail cInfo)
    AgentConnInfoReply smpQueues cInfo -> smpEncode ('D', smpQueues, Tail cInfo) -- 'D' stands for "duplex"
-    AgentRatchetInfo info -> smpEncode ('R', Tail info)
+    AgentRatchetInfo info -> smpEncode ('R', info)
    AgentMessage hdr aMsg -> smpEncode ('M', hdr, aMsg)
    AgentServiceRequest qs sig_ body -> smpEncode ('A', qs, sig_, Tail body)
    AgentServiceResponse body -> smpEncode ('P', Tail body)
@@ -948,13 +949,23 @@ instance Encoding AgentMessage where
    smpP >>= \case
      'I' -> AgentConnInfo . unTail <$> smpP
      'D' -> AgentConnInfoReply <$> smpP <*> (unTail <$> smpP)
-      'R' -> AgentRatchetInfo . unTail <$> smpP
+      'R' -> AgentRatchetInfo <$> smpP
      'M' -> AgentMessage <$> smpP <*> smpP
      'A' -> AgentServiceRequest <$> smpP <*> smpP <*> (unTail <$> smpP)
      'P' -> AgentServiceResponse . unTail <$> smpP
      'J' -> AgentRejection . unTail <$> smpP
      _ -> fail "bad AgentMessage"

+data RatchetInfo = RatchetInfo {answeredKeyHash :: Maybe ByteString}
+  deriving (Eq, Show)
+
+instance Encoding RatchetInfo where
+  smpEncode RatchetInfo {answeredKeyHash} = smpEncode answeredKeyHash
+  smpP = do
+    answeredKeyHash <- smpP <|> pure Nothing
+    _ <- A.takeByteString
+    pure RatchetInfo {answeredKeyHash}
+
 -- internal type for storing message type in the database
 data AgentMessageType
  = AM_CONN_INFO
@@ -76,6 +76,7 @@ module Simplex.Messaging.Agent.Store.AgentStore
    getExpiredServiceConns,
    deleteExpiredServiceRequests,
    getDeletedWaitingDeliveryConnIds,
+    getConnRatchetSync,
    setConnRatchetSync,
    addProcessedRatchetKeyHash,
    checkRatchetKeyHashExists,
@@ -2797,6 +2798,11 @@ getDeletedWaitingDeliveryConnIds :: DB.Connection -> IO [ConnId]
 getDeletedWaitingDeliveryConnIds db =
  map fromOnly <$> DB.query_ db "SELECT conn_id FROM connections WHERE deleted_at_wait_delivery IS NOT NULL"

+getConnRatchetSync :: DB.Connection -> ConnId -> IO (Either StoreError RatchetSyncState)
+getConnRatchetSync db connId =
+  firstRow fromOnly SEConnNotFound $
+    DB.query db "SELECT ratchet_sync_state FROM connections WHERE conn_id = ?" (Only connId)
+
 setConnRatchetSync :: DB.Connection -> ConnId -> RatchetSyncState -> IO ()
 setConnRatchetSync db connId ratchetSyncState =
  DB.execute db "UPDATE connections SET ratchet_sync_state = ? WHERE conn_id = ?" (ratchetSyncState, connId)
@@ -11,6 +11,15 @@ m20260929_ratchet_indexes =
  [r|
 DROP INDEX idx_skipped_messages_conn_id;
 CREATE INDEX idx_skipped_messages_conn_id ON skipped_messages(conn_id, skipped_message_id);
+
+DELETE FROM processed_ratchet_key_hashes AS h
+WHERE EXISTS (
+  SELECT 1 FROM processed_ratchet_key_hashes d
+  WHERE d.conn_id = h.conn_id AND d.hash = h.hash AND d.processed_ratchet_key_hash_id < h.processed_ratchet_key_hash_id
+);
+
+DROP INDEX idx_processed_ratchet_key_hashes_hash;
+CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash);
 CREATE INDEX idx_processed_ratchet_key_hashes_conn_id ON processed_ratchet_key_hashes(conn_id, processed_ratchet_key_hash_id);
 |]

@@ -18,6 +27,8 @@ down_m20260929_ratchet_indexes :: Text
 down_m20260929_ratchet_indexes =
  [r|
 DROP INDEX idx_processed_ratchet_key_hashes_conn_id;
+DROP INDEX idx_processed_ratchet_key_hashes_hash;
+CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash);
 DROP INDEX idx_skipped_messages_conn_id;
 CREATE INDEX idx_skipped_messages_conn_id ON skipped_messages(conn_id);
 |]
@@ -1177,7 +1177,7 @@ CREATE INDEX idx_processed_ratchet_key_hashes_created_at ON smp_agent_test_proto



-CREATE INDEX idx_processed_ratchet_key_hashes_hash ON smp_agent_test_protocol_schema.processed_ratchet_key_hashes USING btree (conn_id, hash);
+CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON smp_agent_test_protocol_schema.processed_ratchet_key_hashes USING btree (conn_id, hash);



@@ -9,6 +9,14 @@ import Database.SQLite.Simple.QQ (sql)
 m20260929_ratchet_indexes :: Query
 m20260929_ratchet_indexes =
  [sql|
+DELETE FROM processed_ratchet_key_hashes AS h
+WHERE EXISTS (
+  SELECT 1 FROM processed_ratchet_key_hashes d
+  WHERE d.conn_id = h.conn_id AND d.hash = h.hash AND d.processed_ratchet_key_hash_id < h.processed_ratchet_key_hash_id
+);
+
+DROP INDEX idx_processed_ratchet_key_hashes_hash;
+CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash);
 CREATE INDEX idx_processed_ratchet_key_hashes_conn_id ON processed_ratchet_key_hashes(conn_id, processed_ratchet_key_hash_id);
  |]

@@ -16,4 +24,6 @@ down_m20260929_ratchet_indexes :: Query
 down_m20260929_ratchet_indexes =
  [sql|
 DROP INDEX idx_processed_ratchet_key_hashes_conn_id;
+DROP INDEX idx_processed_ratchet_key_hashes_hash;
+CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash);
  |]
@@ -573,10 +573,6 @@ CREATE INDEX idx_encrypted_rcv_message_hashes_hash ON encrypted_rcv_message_hash
  conn_id,
  hash
 );
-CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(
-  conn_id,
-  hash
-);
 CREATE INDEX idx_snd_messages_rcpt_internal_id ON snd_messages(
  conn_id,
  rcpt_internal_id
@@ -641,6 +637,10 @@ CREATE INDEX idx_connections_deleted ON connections(deleted);
 CREATE INDEX idx_connections_service_request_expires_at ON connections(
  service_request_expires_at
 );
+CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(
+  conn_id,
+  hash
+);
 CREATE INDEX idx_processed_ratchet_key_hashes_conn_id ON processed_ratchet_key_hashes(
  conn_id,
  processed_ratchet_key_hash_id
@@ -362,6 +362,10 @@ connectionRequestTests =
      smpEncodingTest $ AgentServiceRequest [qInfo] Nothing "service request payload"
      smpEncodingTest $ AgentServiceResponse "service response payload"
      smpEncodingTest $ AgentRejection "rejected: not allowed"
+    it "should serialize and parse ratchet key info" $ do
+      smpDecode "R" `shouldBe` Right (AgentRatchetInfo RatchetInfo {answeredKeyHash = Nothing})
+      smpDecode "R1\3abcdef" `shouldBe` Right (AgentRatchetInfo RatchetInfo {answeredKeyHash = Just "abc"})
+      smpEncodingTest $ AgentRatchetInfo RatchetInfo {answeredKeyHash = Just "0123456789abcdef0123456789abcdef"}
  where
    smpEncodingTest :: (Encoding a, Eq a, Show a, HasCallStack) => a -> Expectation
    smpEncodingTest a = smpDecode (smpEncode a) `shouldBe` Right a
@@ -70,7 +70,7 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (isRight)
 import Data.Int (Int64)
-import Data.List (find, isPrefixOf, isSuffixOf)
+import Data.List (find, isInfixOf, isPrefixOf, isSuffixOf)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.Map as M
 import Data.Maybe (isJust, isNothing)
@@ -465,6 +465,14 @@ functionalAPITests ps = do
        testRatchetSyncSimultaneous ps
      it "should ignore replayed ratchet key after expired hashes are deleted" $
        testRatchetSyncReplayedKey ps
+      it "should synchronize ratchets when synchronization is forced again" $
+        testRatchetSyncRepeated ps
+      it "should not mark ratchet key as processed when ratchet recreation fails" $
+        testRatchetSyncFailedKeyNotProcessed ps
+      it "should not store reply ratchet key when ratchet recreation fails" $
+        testRatchetSyncFailedRecreationNoReply ps
+      it "should not store ratchet key when starting synchronization fails" $
+        testRatchetSyncStartFailedNoKey ps
 #endif
    describe "Subscription mode OnlyCreate" $ do
      it "messages delivered only when polled" $
@@ -2771,6 +2779,96 @@ testRatchetSyncReplayedKey ps = withAgentClients2 $ \alice bob -> do
    runRight_ $ exchangeGreetingsMsgIds alice bobId 10 bob2 aliceId 7
  disposeAgentClient bob2

+testRatchetSyncRepeated :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testRatchetSyncRepeated ps = withAgentClients2 $ \alice bob -> do
+  (aliceId, bobId, bob2) <- startRatchetSyncOffline ps alice bob
+  ConnectionStats {ratchetSyncState = rss2} <- runRight $ synchronizeRatchet bob2 aliceId PQSupportOn True
+  rss2 `shouldBe` RSStarted
+
+  withSmpServerStoreMsgLogOn ps testPort $ \_ -> do
+    concurrently_
+      (getInAnyOrder alice [ratchetSyncP' bobId RSAgreed, serverUpP])
+      (getInAnyOrder bob2 [ratchetSyncP' aliceId RSAgreed, serverUpP])
+    runRight_ $ do
+      get alice =##> ratchetSyncP bobId RSAgreed
+      get alice =##> ratchetSyncP bobId RSOk
+      get bob2 =##> ratchetSyncP aliceId RSOk
+      msgId <- sendMessage alice bobId SMP.noMsgFlags "hello"
+      get alice ##> ("", bobId, SENT msgId)
+      get bob2 =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
+      ackMessage bob2 aliceId 8 Nothing
+  map fst <$> processedRatchetKeyHashes bob2 `shouldReturn` [aliceId, aliceId]
+  disposeAgentClient bob2
+
+testRatchetSyncFailedKeyNotProcessed :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testRatchetSyncFailedKeyNotProcessed ps = withAgentClients2 $ \alice bob -> do
+  (aliceId, bobId, bob2) <- startRatchetSyncOffline ps alice bob
+  withTransaction (store $ agentEnv bob2) $ \db ->
+    DB.execute_ db "UPDATE ratchets SET x3dh_priv_key_1 = NULL"
+
+  withSmpServerStoreMsgLogOn ps testPort $ \_ ->
+    concurrently_
+      (getInAnyOrder alice [ratchetSyncP' bobId RSAgreed, serverUpP])
+      (getInAnyOrder bob2 [x3dhKeysNotFoundP aliceId, serverUpP])
+  map fst <$> processedRatchetKeyHashes alice `shouldReturn` [bobId]
+  processedRatchetKeyHashes bob2 `shouldReturn` []
+  disposeAgentClient bob2
+  where
+    x3dhKeysNotFoundP :: ConnId -> ATransmission -> Bool
+    x3dhKeysNotFoundP cId = \case
+      (_, cId', AEvt SAEConn (ERR (A.INTERNAL e))) -> cId' == cId && "SEX3dhKeysNotFound" `isPrefixOf` e
+      _ -> False
+
+testRatchetSyncFailedRecreationNoReply :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testRatchetSyncFailedRecreationNoReply ps = withAgentClients2 $ \alice bob -> do
+  (_, bobId, bob2) <- startRatchetSyncOffline ps alice bob
+  aliceSndMsgs <- sndMessages alice
+  withTransaction (store $ agentEnv alice) $ \db ->
+    DB.execute_ db "CREATE TRIGGER fail_ratchet_insert BEFORE INSERT ON ratchets BEGIN SELECT RAISE(ABORT, 'ratchet insert failed'); END"
+  withSmpServerStoreMsgLogOn ps testPort $ \_ ->
+    concurrently_
+      (getInAnyOrder alice [ratchetInsertFailedP bobId, serverUpP])
+      (getInAnyOrder bob2 [serverUpP])
+  processedRatchetKeyHashes alice `shouldReturn` []
+  sndMessages alice `shouldReturn` aliceSndMsgs
+  disposeAgentClient bob2
+  where
+    ratchetInsertFailedP :: ConnId -> ATransmission -> Bool
+    ratchetInsertFailedP cId = \case
+      (_, cId', AEvt SAEConn (ERR (A.INTERNAL e))) -> cId' == cId && "ratchet insert failed" `isInfixOf` e
+      _ -> False
+
+testRatchetSyncStartFailedNoKey :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testRatchetSyncStartFailedNoKey ps = withAgentClients2 $ \alice bob -> do
+  (aliceId, _, bob2) <- withSmpServerStoreMsgLogOn ps testPort $ \_ ->
+    setupDesynchronizedRatchet alice bob
+  bobSndMsgs <- sndMessages bob2
+  withTransaction (store $ agentEnv bob2) $ \db ->
+    DB.execute_ db "CREATE TRIGGER fail_ratchet_update BEFORE UPDATE ON ratchets BEGIN SELECT RAISE(ABORT, 'ratchet update failed'); END"
+  Left (A.INTERNAL e) <- runExceptT $ synchronizeRatchet bob2 aliceId PQSupportOff False
+  e `shouldContain` "ratchet update failed"
+  ConnectionStats {ratchetSyncState} <- runRight $ getConnectionServers bob2 aliceId
+  ratchetSyncState `shouldBe` RSRequired
+  withTransaction (store $ agentEnv bob2) (`DB.query_` "SELECT conn_id, pq_support FROM connections") `shouldReturn` [(aliceId, PQSupportOn)]
+  sndMessages bob2 `shouldReturn` bobSndMsgs
+  disposeAgentClient bob2
+
+startRatchetSyncOffline :: HasCallStack => (ASrvTransport, AStoreType) -> AgentClient -> AgentClient -> IO (ConnId, ConnId, AgentClient)
+startRatchetSyncOffline ps alice bob = do
+  (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn ps testPort $ \_ ->
+    setupDesynchronizedRatchet alice bob
+  ("", "", DOWN _ _) <- nGet alice
+  ("", "", DOWN _ _) <- nGet bob2
+  ConnectionStats {ratchetSyncState} <- runRight $ synchronizeRatchet bob2 aliceId PQSupportOn False
+  ratchetSyncState `shouldBe` RSStarted
+  pure (aliceId, bobId, bob2)
+
+processedRatchetKeyHashes :: AgentClient -> IO [(ConnId, ByteString)]
+processedRatchetKeyHashes c = withTransaction (store $ agentEnv c) (`DB.query_` "SELECT conn_id, hash FROM processed_ratchet_key_hashes")
+
+sndMessages :: AgentClient -> IO [(ConnId, Int64)]
+sndMessages c = withTransaction (store $ agentEnv c) (`DB.query_` "SELECT conn_id, internal_id FROM snd_messages ORDER BY conn_id, internal_id")
+
 getMsg :: AgentClient -> ConnId -> ExceptT AgentErrorType IO a -> ExceptT AgentErrorType IO a
 getMsg c cId action = do
  liftIO $ noMessages c "nothing should be delivered before GET"
@@ -48,6 +48,7 @@ schemaDumpTest = do
  it "verify strict tables" testVerifyStrict
  it "should NOT create user record for new database" testUsersMigrationNew
  it "should create user record for old database" testUsersMigrationOld
+  it "should remove duplicate ratchet key hashes before adding unique index" testRatchetKeyHashesUniqueMigration

 testVerifySchemaDump :: IO ()
 testVerifySchemaDump = do
@@ -114,12 +115,28 @@ testUsersMigrationOld = do
    `shouldReturn` ([Only (1 :: Int)])
  closeDBStore st'

+testRatchetKeyHashesUniqueMigration :: IO ()
+testRatchetKeyHashesUniqueMigration = do
+  let beforeUnique = takeWhile (("m20260929_ratchet_indexes" /=) . name) appMigrations
+  Right st <- createDBStore (DBOpts testDB [] "" False True TQOff) beforeUnique (MigrationConfig MCError Nothing)
+  withTransaction' st $ \db -> do
+    SQL.execute_ db "INSERT INTO users (user_id) VALUES (1)"
+    SQL.execute_ db "INSERT INTO connections (conn_id, conn_mode, user_id) VALUES (x'01', 'INV', 1), (x'02', 'INV', 1)"
+    SQL.execute_ db "INSERT INTO processed_ratchet_key_hashes (conn_id, hash) VALUES (x'01', x'aa'), (x'01', x'aa'), (x'01', x'bb'), (x'02', x'aa'), (x'01', x'aa')"
+  closeDBStore st
+  Right st' <- createDBStore (DBOpts testDB [] "" False True TQOff) appMigrations (MigrationConfig MCYesUp Nothing)
+  withTransaction' st' (`SQL.query_` "SELECT processed_ratchet_key_hash_id FROM processed_ratchet_key_hashes ORDER BY processed_ratchet_key_hash_id")
+    `shouldReturn` [Only (1 :: Int), Only 3, Only 4]
+  closeDBStore st'
+
 skipComparisonForDownMigrations :: [String]
 skipComparisonForDownMigrations =
  [ -- on down migration idx_messages_internal_snd_id_ts index moves down to the end of the file
    "m20230814_indexes",
    -- snd_secure and last_broker_ts columns swap order on down migration
-    "m20250322_short_links"
+    "m20250322_short_links",
+    -- on down migration idx_processed_ratchet_key_hashes_hash index moves down to the end of the file
+    "m20260929_ratchet_indexes"
  ]

 getSchema :: FilePath -> FilePath -> IO String