agent: better handling errors during connection handshake retries (#1578)

* agent: handle invitation connection handshake errors * fix/test retries for connecting via address
2026-08-06 23:40:18 +00:00 · 2025-06-25 19:06:00 +01:00
parent 976bd3a389
commit b4bcfd325b
8 changed files with 293 additions and 75 deletions
@@ -787,7 +787,7 @@ ackMessageAsync' c corrId connId msgId rcptInfo_ = do
  case cType of
    SCDuplex -> enqueueAck
    SCRcv -> enqueueAck
-    SCSnd -> throwE $ CONN SIMPLEX
+    SCSnd -> throwE $ CONN SIMPLEX "ackMessageAsync"
    SCContact -> throwE $ CMD PROHIBITED "ackMessageAsync: SCContact"
    SCNew -> throwE $ CMD PROHIBITED "ackMessageAsync: SCNew"
  where
@@ -851,7 +851,7 @@ setConnShortLink' c connId cMode userData clientData =
    pure sl
  where
    prepareContactLinkData :: RcvQueue -> AM (RcvQueue, SMP.LinkId, ConnShortLink 'CMContact, QueueLinkData)
-    prepareContactLinkData rq@RcvQueue {server, sndId, e2ePrivKey, shortLink} = do
+    prepareContactLinkData rq@RcvQueue {shortLink} = do
      g <- asks random
      AgentConfig {smpClientVRange = vr, smpAgentVRange} <- asks config
      let cslContact = CSLContact SLSServer CCTContact (qServer rq)
@@ -863,7 +863,7 @@ setConnShortLink' c connId cMode userData clientData =
          pure (rq, linkId, cslContact shortLinkKey, (linkEncFixedData, d))
        Nothing -> do
          sigKeys@(_, privSigKey) <- atomically $ C.generateKeyPair @'C.Ed25519 g
-          let qUri = SMPQueueUri vr $ SMPQueueAddress server sndId (C.publicKey e2ePrivKey) (Just QMContact)
+          let qUri = SMPQueueUri vr $ (rcvSMPQueueAddress rq) {queueMode = Just QMContact}
              connReq = CRContactUri $ ConnReqUriData SSSimplex smpAgentVRange [qUri] clientData
              (linkKey, linkData) = SL.encodeSignLinkData sigKeys smpAgentVRange connReq userData
              (linkId, k) = SL.contactShortLinkKdf linkKey
@@ -1129,23 +1129,46 @@ joinConnSrv c userId connId enableNtfs inv@CRInvitationUri {} cInfo pqSup subMod
  withInvLock c (strEncode inv) "joinConnSrv" $ do
    SomeConn cType conn <- withStore c (`getConn` connId)
    case conn of
-      NewConnection _ -> doJoin Nothing
-      SndConnection _ sq -> doJoin $ Just sq
-      DuplexConnection _ (RcvQueue {status = New} :| _) (sq@SndQueue {status = New} :| _) -> doJoin $ Just sq
+      NewConnection _ -> doJoin Nothing Nothing
+      SndConnection _ sq -> doJoin Nothing (Just sq)
+      DuplexConnection _ (rq@RcvQueue {status = New} :| _) (sq@SndQueue {status = sqStatus} :| _)
+        | sqStatus == New || sqStatus == Secured -> doJoin (Just rq) (Just sq)
      _ -> throwE $ CMD PROHIBITED $ "joinConnSrv: bad connection " <> show cType
  where
-    doJoin :: Maybe SndQueue -> AM (SndQueueSecured, Maybe ClientServiceId)
-    doJoin sq_ = do
+    doJoin :: Maybe RcvQueue -> Maybe SndQueue -> AM (SndQueueSecured, Maybe ClientServiceId)
+    doJoin rq_ sq_ = do
      (cData, sq, e2eSndParams, lnkId_) <- startJoinInvitation c userId connId sq_ enableNtfs inv pqSup
-      secureConfirmQueue c cData sq srv cInfo (Just e2eSndParams) subMode
+      secureConfirmQueue c cData rq_ sq srv cInfo (Just e2eSndParams) subMode
        >>= (mapM_ (delInvSL c connId srv) lnkId_ $>)
 joinConnSrv c userId connId enableNtfs cReqUri@CRContactUri {} cInfo pqSup subMode srv =
  lift (compatibleContactUri cReqUri) >>= \case
-    Just (qInfo, vrsn@(Compatible v)) -> do
-      let pqInitKeys = CR.joinContactInitialKeys (v >= pqdrSMPAgentVersion) pqSup
-      (CCLink cReq _, service) <- newRcvConnSrv c userId connId enableNtfs SCMInvitation Nothing Nothing pqInitKeys subMode srv
-      void $ sendInvitation c userId connId qInfo vrsn cReq cInfo
-      pure (False, service)
+    Just (qInfo, vrsn@(Compatible v)) ->
+      withInvLock c (strEncode cReqUri) "joinConnSrv" $ do
+        SomeConn cType conn <- withStore c (`getConn` connId)
+        let pqInitKeys = CR.joinContactInitialKeys (v >= pqdrSMPAgentVersion) pqSup
+        (CCLink cReq _, service) <- case conn of
+          NewConnection _ -> newRcvConnSrv c userId connId enableNtfs SCMInvitation Nothing Nothing pqInitKeys subMode srv
+          RcvConnection _ rq -> mkJoinInvitation rq pqInitKeys
+          _ -> throwE $ CMD PROHIBITED $ "joinConnSrv: bad connection " <> show cType
+        void $ sendInvitation c userId connId qInfo vrsn cReq cInfo
+        pure (False, service)
+      where
+        mkJoinInvitation rq@RcvQueue {clientService} pqInitKeys = do
+          g <- asks random
+          AgentConfig {smpClientVRange = vr, smpAgentVRange, e2eEncryptVRange = e2eVR} <- asks config
+          let qUri = SMPQueueUri vr $ (rcvSMPQueueAddress rq) {queueMode = Just QMMessaging}
+              crData = ConnReqUriData SSSimplex smpAgentVRange [qUri] Nothing
+          e2eRcvParams <- withStore' c $ \db ->
+            getRatchetX3dhKeys db connId >>= \case
+              Right keys -> pure $ CR.mkRcvE2ERatchetParams (maxVersion e2eVR) keys
+              Left e -> do
+                nonBlockingWriteTBQueue (subQ c) ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "no rcv ratchet " <> show e))
+                let pqEnc = CR.initialPQEncryption False pqInitKeys
+                (pk1, pk2, pKem, e2eRcvParams) <- liftIO $ CR.generateRcvE2EParams g (maxVersion e2eVR) pqEnc
+                createRatchetX3dhKeys db connId pk1 pk2 pKem
+                pure e2eRcvParams
+          let cReq = CRInvitationUri crData $ toVersionRangeT e2eRcvParams e2eVR
+          pure (CCLink cReq Nothing, dbServiceId <$> clientService)
    Nothing -> throwE $ AGENT A_VERSION

 delInvSL :: AgentClient -> ConnId -> SMPServerWithAuth -> SMP.LinkId -> AM ()
@@ -1157,14 +1180,18 @@ joinConnSrvAsync :: AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequest
 joinConnSrvAsync c userId connId enableNtfs inv@CRInvitationUri {} cInfo pqSupport subMode srv = do
  SomeConn cType conn <- withStore c (`getConn` connId)
  case conn of
-    NewConnection _ -> doJoin Nothing
-    SndConnection _ sq -> doJoin $ Just sq
+    NewConnection _ -> doJoin Nothing Nothing
+    SndConnection _ sq -> doJoin Nothing (Just sq)
+    -- this branch should never be reached with async flow because once receive queue is created,
+    -- there are not more failure points (sending confirmation is asynchronous)
+    DuplexConnection _ (rq@RcvQueue {status = New} :| _) (sq@SndQueue {status = sqStatus} :| _)
+      | sqStatus == New || sqStatus == Secured -> doJoin (Just rq) (Just sq)
    _ -> throwE $ CMD PROHIBITED $ "joinConnSrvAsync: bad connection " <> show cType
  where
-    doJoin :: Maybe SndQueue -> AM (SndQueueSecured, Maybe ClientServiceId)
-    doJoin sq_ = do
+    doJoin :: Maybe RcvQueue -> Maybe SndQueue -> AM (SndQueueSecured, Maybe ClientServiceId)
+    doJoin rq_ sq_ = do
      (cData, sq, e2eSndParams, lnkId_) <- startJoinInvitation c userId connId sq_ enableNtfs inv pqSupport
-      secureConfirmQueueAsync c cData sq srv cInfo (Just e2eSndParams) subMode
+      secureConfirmQueueAsync c cData rq_ sq srv cInfo (Just e2eSndParams) subMode
        >>= (mapM_ (delInvSL c connId srv) lnkId_ $>)
 joinConnSrvAsync _c _userId _connId _enableNtfs (CRContactUri _) _cInfo _subMode _pqSupport _srv = do
  throwE $ CMD PROHIBITED "joinConnSrvAsync"
@@ -1254,7 +1281,7 @@ subscribeConnections' c connIds = do
    sndSubResult :: SndQueue -> Either AgentErrorType (Maybe ClientServiceId)
    sndSubResult SndQueue {status} = case status of
      Confirmed -> Right Nothing
-      Active -> Left $ CONN SIMPLEX
+      Active -> Left $ CONN SIMPLEX "subscribeConnections"
      _ -> Left $ INTERNAL "unexpected queue status"
    connResults :: [(RcvQueue, Either AgentErrorType (Maybe SMP.ServiceId))] -> Map ConnId (Either AgentErrorType (Maybe SMP.ServiceId))
    connResults = M.map snd . foldl' addResult M.empty
@@ -1327,7 +1354,7 @@ getConnectionMessages' c = mapM $ tryAgentError' . getConnectionMessage
        DuplexConnection _ (rq :| _) _ -> pure rq
        RcvConnection _ rq -> pure rq
        ContactConnection _ rq -> pure rq
-        SndConnection _ _ -> throwE $ CONN SIMPLEX
+        SndConnection _ _ -> throwE $ CONN SIMPLEX "getConnectionMessage"
        NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionMessage: NewConnection"
      msg_ <- getQueueMessage c rq `catchAgentError` \e -> atomically (releaseGetLock c rq) >> throwError e
      when (isNothing msg_) $ do
@@ -1412,7 +1439,7 @@ sendMessagesB_ c reqs connIds = withConnLocks c connIds "sendMessages" $ do
    prepareConn s (Right ((_, pqEnc, msgFlags, msgOrRef), SomeConn _ conn)) = case conn of
      DuplexConnection cData _ sqs -> prepareMsg cData sqs
      SndConnection cData sq -> prepareMsg cData [sq]
-      _ -> (s, Left $ CONN SIMPLEX)
+      _ -> (s, Left $ CONN SIMPLEX "sendMessagesB_")
      where
        prepareMsg :: ConnData -> NonEmpty SndQueue -> (Set ConnId, Either AgentErrorType (ConnData, NonEmpty SndQueue, Maybe PQEncryption, MsgFlags, ValueOrRef AMessage))
        prepareMsg cData@ConnData {connId, pqSupport} sqs
@@ -1881,7 +1908,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq@SndQueue {userI
    notify cmd = atomically $ writeTBQueue subQ ("", connId, AEvt (sAEntity @e) cmd)
    notifyDel :: AEntityI e => InternalId -> AEvent e -> AM ()
    notifyDel msgId cmd = notify cmd >> delMsg msgId
-    connError msgId = notifyDel msgId . ERR . CONN
+    connError msgId = notifyDel msgId . ERR . (`CONN` "")
    qError msgId = notifyDel msgId . ERR . AGENT . A_QUEUE
    internalErr msgId = notifyDel msgId . ERR . INTERNAL

@@ -1899,7 +1926,7 @@ ackMessage' c connId msgId rcptInfo_ = withConnLock c connId "ackMessage" $ do
  case conn of
    DuplexConnection {} -> ack >> sendRcpt conn >> del
    RcvConnection {} -> ack >> del
-    SndConnection {} -> throwE $ CONN SIMPLEX
+    SndConnection {} -> throwE $ CONN SIMPLEX "ackMessage"
    ContactConnection {} -> throwE $ CMD PROHIBITED "ackMessage: ContactConnection"
    NewConnection _ -> throwE $ CMD PROHIBITED "ackMessage: NewConnection"
  where
@@ -1933,8 +1960,8 @@ getConnectionQueueInfo' c connId = do
    DuplexConnection _ (rq :| _) _ -> getQueueInfo c rq
    RcvConnection _ rq -> getQueueInfo c rq
    ContactConnection _ rq -> getQueueInfo c rq
-    SndConnection {} -> throwE $ CONN SIMPLEX
-    NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionQueueInfo': NewConnection"
+    SndConnection {} -> throwE $ CONN SIMPLEX "getConnectionQueueInfo"
+    NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionQueueInfo: NewConnection"

 switchConnection' :: AgentClient -> ConnId -> AM ConnectionStats
 switchConnection' c connId =
@@ -2035,7 +2062,7 @@ suspendConnection' c connId = withConnLock c connId "suspendConnection" $ do
    DuplexConnection _ rqs _ -> mapM_ (suspendQueue c) rqs
    RcvConnection _ rq -> suspendQueue c rq
    ContactConnection _ rq -> suspendQueue c rq
-    SndConnection _ _ -> throwE $ CONN SIMPLEX
+    SndConnection _ _ -> throwE $ CONN SIMPLEX "suspendConnection"
    NewConnection _ -> throwE $ CMD PROHIBITED "suspendConnection"

 -- | Delete SMP agent connection (DEL command) in Reader monad
@@ -2354,7 +2381,7 @@ toggleConnectionNtfs' c connId enable = do
    DuplexConnection cData _ _ -> toggle cData
    RcvConnection cData _ -> toggle cData
    ContactConnection cData _ -> toggle cData
-    _ -> throwE $ CONN SIMPLEX
+    _ -> throwE $ CONN SIMPLEX "toggleConnectionNtfs"
  where
    toggle :: ConnData -> AM ()
    toggle cData
@@ -2865,8 +2892,11 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
                      rc = CR.initRcvRatchet rcVs rcDHRs rcParams pqSupport'
                  g <- asks random
                  (agentMsgBody_, rc', skipped) <- liftError cryptoError $ CR.rcDecrypt g rc M.empty encConnInfo
-                  case (agentMsgBody_, skipped) of
-                    (Right agentMsgBody, CR.SMDNoChange) ->
+                  case skipped of
+                    CR.SMDNoChange -> pure ()
+                    _ -> logWarn "conf: skipped confirmations"
+                  case agentMsgBody_ of
+                    Right agentMsgBody ->
                      parseMessage agentMsgBody >>= \case
                        AgentConnInfoReply smpQueues connInfo -> do
                          processConf connInfo SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues = L.toList smpQueues, smpClientVersion = phVer}
@@ -2893,7 +2923,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
                            createConfirmation db g newConfirmation
                          let srvs = map qServer $ smpReplyQueues senderConf
                          notify $ CONF confId pqSupport' srvs connInfo
-                    _ -> prohibited "conf: decrypt error or skipped"
+                    _ -> prohibited "conf: decrypt error"
                -- party accepting connection
                (DuplexConnection _ (rq'@RcvQueue {smpClientVersion = v'} :| _) _, Nothing) -> do
                  g <- asks random
@@ -3180,18 +3210,18 @@ connectReplyQueues c cData@ConnData {userId, connId} ownConnInfo sq_ (qInfo :| _
          (sq, _) <- lift $ newSndQueue userId connId qInfo' Nothing
          withStore c $ \db -> upgradeRcvConnToDuplex db connId sq

-secureConfirmQueueAsync :: AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> SubscriptionMode -> AM (SndQueueSecured, Maybe ClientServiceId)
-secureConfirmQueueAsync c cData sq srv connInfo e2eEncryption_ subMode = do
+secureConfirmQueueAsync :: AgentClient -> ConnData -> Maybe RcvQueue -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> SubscriptionMode -> AM (SndQueueSecured, Maybe ClientServiceId)
+secureConfirmQueueAsync c cData rq_ sq srv connInfo e2eEncryption_ subMode = do
  sqSecured <- agentSecureSndQueue c cData sq
-  (qInfo, service) <- mkAgentConfirmation c cData sq srv connInfo subMode
+  (qInfo, service) <- mkAgentConfirmation c cData rq_ sq srv connInfo subMode
  storeConfirmation c cData sq e2eEncryption_ qInfo
  lift $ submitPendingMsg c cData sq
  pure (sqSecured, service)

-secureConfirmQueue :: AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> SubscriptionMode -> AM (SndQueueSecured, Maybe ClientServiceId)
-secureConfirmQueue c cData@ConnData {connId, connAgentVersion, pqSupport} sq srv connInfo e2eEncryption_ subMode = do
+secureConfirmQueue :: AgentClient -> ConnData -> Maybe RcvQueue -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> SubscriptionMode -> AM (SndQueueSecured, Maybe ClientServiceId)
+secureConfirmQueue c cData@ConnData {connId, connAgentVersion, pqSupport} rq_ sq srv connInfo e2eEncryption_ subMode = do
  sqSecured <- agentSecureSndQueue c cData sq
-  (qInfo, service) <- mkAgentConfirmation c cData sq srv connInfo subMode
+  (qInfo, service) <- mkAgentConfirmation c cData rq_ sq srv connInfo subMode
  msg <- mkConfirmation qInfo
  void $ sendConfirmation c sq msg
  withStore' c $ \db -> setSndQueueStatus db sq Confirmed
@@ -3221,9 +3251,11 @@ agentSecureSndQueue c ConnData {connAgentVersion} sq@SndQueue {queueMode, status
    sndSecure = senderCanSecure queueMode
    initiatorRatchetOnConf = connAgentVersion >= ratchetOnConfSMPAgentVersion

-mkAgentConfirmation :: AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> SubscriptionMode -> AM (AgentMessage, Maybe ClientServiceId)
-mkAgentConfirmation c cData sq srv connInfo subMode = do
-  (qInfo, service) <- createReplyQueue c cData sq subMode srv
+mkAgentConfirmation :: AgentClient -> ConnData -> Maybe RcvQueue -> SndQueue -> SMPServerWithAuth -> ConnInfo -> SubscriptionMode -> AM (AgentMessage, Maybe ClientServiceId)
+mkAgentConfirmation c cData rq_ sq srv connInfo subMode = do
+  (qInfo, service) <- case rq_ of
+    Nothing -> createReplyQueue c cData sq subMode srv
+    Just rq@RcvQueue {smpClientVersion = v, clientService} -> pure (SMPQueueInfo v $ rcvSMPQueueAddress rq, dbServiceId <$> clientService)
  pure (AgentConnInfoReply (qInfo :| []) connInfo, service)

 enqueueConfirmation :: AgentClient -> ConnData -> SndQueue -> ConnInfo -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> AM ()
@@ -867,7 +867,7 @@ newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v =
    Right client -> do
      logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
      atomically $ putTMVar (sessionVar v) (Right client)
-      atomically $ writeTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent CONNECT client)
+      liftIO $ nonBlockingWriteTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent CONNECT client)
      pure client
    Left e -> do
      ei <- asks $ persistErrorInterval . config
@@ -2126,12 +2126,12 @@ withStoreBatch' c actions = withStoreBatch c (fmap (fmap Right) . actions)

 storeError :: StoreError -> AgentErrorType
 storeError = \case
-  SEConnNotFound -> CONN NOT_FOUND
+  SEConnNotFound -> CONN NOT_FOUND ""
  SEUserNotFound -> NO_USER
-  SERatchetNotFound -> CONN NOT_FOUND
-  SEConnDuplicate -> CONN DUPLICATE
-  SEBadConnType CRcv -> CONN SIMPLEX
-  SEBadConnType CSnd -> CONN SIMPLEX
+  SERatchetNotFound -> CONN NOT_FOUND ""
+  SEConnDuplicate -> CONN DUPLICATE ""
+  SEBadConnType cxt CRcv -> CONN SIMPLEX cxt
+  SEBadConnType cxt CSnd -> CONN SIMPLEX cxt
  SEInvitationNotFound cxt invId -> CMD PROHIBITED $ "SEInvitationNotFound " <> cxt <> ", invitationId = " <> show invId
  -- this error is never reported as store error,
  -- it is used to wrap agent operations when "transaction-like" store access is needed
@@ -1837,7 +1837,7 @@ data AgentErrorType
  = -- | command or response error
    CMD {cmdErr :: CommandErrorType, errContext :: String}
  | -- | connection errors
-    CONN {connErr :: ConnectionErrorType}
+    CONN {connErr :: ConnectionErrorType, errContext :: String}
  | -- | user not found in database
    NO_USER
  | -- | SMP protocol errors forwarded to agent clients
@@ -119,6 +119,10 @@ rcvQueueInfo :: RcvQueue -> RcvQueueInfo
 rcvQueueInfo rq@RcvQueue {server, rcvSwchStatus} =
  RcvQueueInfo {rcvServer = server, rcvSwitchStatus = rcvSwchStatus, canAbortSwitch = canAbortRcvSwitch rq}

+rcvSMPQueueAddress :: RcvQueue -> SMPQueueAddress
+rcvSMPQueueAddress RcvQueue {server, sndId, e2ePrivKey, queueMode} =
+  SMPQueueAddress server sndId (C.publicKey e2ePrivKey) queueMode
+
 canAbortRcvSwitch :: RcvQueue -> Bool
 canAbortRcvSwitch = maybe False canAbort . rcvSwchStatus
  where
@@ -662,7 +666,7 @@ data StoreError
    SESndQueueExists
  | -- | Wrong connection type, e.g. "send" connection when "receive" or "duplex" is expected, or vice versa.
    -- 'upgradeRcvConnToDuplex' and 'upgradeSndConnToDuplex' do not allow duplex connections - they would also return this error.
-    SEBadConnType ConnType
+    SEBadConnType String ConnType
  | -- | Confirmation not found.
    SEConfirmationNotFound
  | -- | Invitation not found
@@ -387,7 +387,7 @@ updateNewConnRcv db connId rq =
  getConn db connId $>>= \case
    (SomeConn _ NewConnection {}) -> updateConn
    (SomeConn _ RcvConnection {}) -> updateConn -- to allow retries
-    (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+    (SomeConn c _) -> pure . Left . SEBadConnType "updateNewConnRcv" $ connType c
  where
    updateConn :: IO (Either StoreError RcvQueue)
    updateConn = Right <$> addConnRcvQueue_ db connId rq
@@ -396,7 +396,7 @@ updateNewConnSnd :: DB.Connection -> ConnId -> NewSndQueue -> IO (Either StoreEr
 updateNewConnSnd db connId sq =
  getConn db connId $>>= \case
    (SomeConn _ NewConnection {}) -> updateConn
-    (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+    (SomeConn c _) -> pure . Left . SEBadConnType "updateNewConnSnd" $ connType c
  where
    updateConn :: IO (Either StoreError SndQueue)
    updateConn = Right <$> addConnSndQueue_ db connId sq
@@ -472,14 +472,14 @@ upgradeRcvConnToDuplex :: DB.Connection -> ConnId -> NewSndQueue -> IO (Either S
 upgradeRcvConnToDuplex db connId sq =
  getConn db connId $>>= \case
    (SomeConn _ RcvConnection {}) -> Right <$> addConnSndQueue_ db connId sq
-    (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+    (SomeConn c _) -> pure . Left . SEBadConnType "upgradeRcvConnToDuplex" $ connType c

 -- TODO [certs rcv] store clientServiceId from NewRcvQueue
 upgradeSndConnToDuplex :: DB.Connection -> ConnId -> NewRcvQueue -> IO (Either StoreError RcvQueue)
 upgradeSndConnToDuplex db connId rq =
  getConn db connId >>= \case
    Right (SomeConn _ SndConnection {}) -> Right <$> addConnRcvQueue_ db connId rq
-    Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+    Right (SomeConn c _) -> pure . Left . SEBadConnType "upgradeSndConnToDuplex" $ connType c
    _ -> pure $ Left SEConnNotFound

 -- TODO [certs rcv] store clientServiceId from NewRcvQueue
@@ -487,7 +487,7 @@ addConnRcvQueue :: DB.Connection -> ConnId -> NewRcvQueue -> IO (Either StoreErr
 addConnRcvQueue db connId rq =
  getConn db connId >>= \case
    Right (SomeConn _ DuplexConnection {}) -> Right <$> addConnRcvQueue_ db connId rq
-    Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+    Right (SomeConn c _) -> pure . Left . SEBadConnType "addConnRcvQueue" $ connType c
    _ -> pure $ Left SEConnNotFound

 addConnRcvQueue_ :: DB.Connection -> ConnId -> NewRcvQueue -> IO RcvQueue
@@ -499,7 +499,7 @@ addConnSndQueue :: DB.Connection -> ConnId -> NewSndQueue -> IO (Either StoreErr
 addConnSndQueue db connId sq =
  getConn db connId >>= \case
    Right (SomeConn _ DuplexConnection {}) -> Right <$> addConnSndQueue_ db connId sq
-    Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+    Right (SomeConn c _) -> pure . Left . SEBadConnType "addConnSndQueue" $ connType c
    _ -> pure $ Left SEConnNotFound

 addConnSndQueue_ :: DB.Connection -> ConnId -> NewSndQueue -> IO SndQueue
@@ -55,6 +55,7 @@ module Simplex.Messaging.Crypto.Ratchet
    supportedE2EEncryptVRange,
    generateRcvE2EParams,
    generateSndE2EParams,
+    mkRcvE2ERatchetParams,
    initialPQEncryption,
    connPQEncryption,
    joinContactInitialKeys,
@@ -196,6 +197,8 @@ data ARKEMParams = forall s. RatchetKEMStateI s => ARKP (SRatchetKEMState s) (RK

 deriving instance Show ARKEMParams

+type RcvRKEMParams = RKEMParams 'RKSProposed
+
 instance RatchetKEMStateI s => Encoding (RKEMParams s) where
  smpEncode = \case
    RKParamsProposed k -> smpEncode ('P', k)
@@ -406,6 +409,12 @@ data UseKEM (s :: RatchetKEMState) where

 data AUseKEM = forall s. RatchetKEMStateI s => AUseKEM (SRatchetKEMState s) (UseKEM s)

+mkRcvE2ERatchetParams :: VersionE2E -> (PrivateKey a, PrivateKey a, Maybe RcvPrivRKEMParams) -> RcvE2ERatchetParams a
+mkRcvE2ERatchetParams v (pk1, pk2, pKem) = E2ERatchetParams v (publicKey pk1) (publicKey pk2) (mkKem <$> pKem)
+  where
+    mkKem :: RcvPrivRKEMParams -> RcvRKEMParams
+    mkKem (PrivateRKParamsProposed (k, _)) = RKParamsProposed k
+
 generateE2EParams :: forall s a. (AlgorithmI a, DhAlgorithm a) => TVar ChaChaDRG -> VersionE2E -> Maybe (UseKEM s) -> IO (PrivateKey a, PrivateKey a, Maybe (PrivRKEMParams s), E2ERatchetParams s a)
 generateE2EParams g v useKEM_ = do
  (k1, pk1) <- atomically $ generateKeyPair g