mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-22 17:32:16 +00:00
agent: use DB to queue async commands and messages (#929)
* agent: use DB to queue async commands and messages * fix message delivery, async commands dont work * fix async commands * remove comment * rename * comment * balance agent operation * empty lines * balance another agent operation * diff * worker * qAddress * fix incorrect queue ID in v1 * types for queues * fix test * update index
This commit is contained in:
Evgeny Poberezkin authored and GitHub committed
1 parent
fa457d1c25
commit
2489333c87
13 files changed
+414
-357
No files matched your search
@@ -98,6 +98,7 @@ library
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230722_indexes
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230814_indexes
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230829_crypto_files
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231222_command_created_at
|
||||
Simplex.Messaging.Agent.TAsyncs
|
||||
Simplex.Messaging.Agent.TRcvQueues
|
||||
Simplex.Messaging.Client
|
||||
|
||||
@@ -155,8 +155,7 @@ addWorker c wsSel runWorker runWorkerNoSrv srv_ = do
|
||||
Nothing -> runWorkerNoSrv c doWork
|
||||
worker <- async $ runWorker' `agentFinally` atomically (TM.delete srv_ ws)
|
||||
atomically $ TM.insert srv_ (doWork, worker) ws
|
||||
Just (doWork, _) ->
|
||||
void . atomically $ tryPutTMVar doWork ()
|
||||
Just (doWork, _) -> atomically $ hasWorkToDo' doWork
|
||||
|
||||
runXFTPRcvWorker :: forall m. AgentMonad m => AgentClient -> XFTPServer -> TMVar () -> m ()
|
||||
runXFTPRcvWorker c srv doWork = do
|
||||
@@ -526,8 +525,7 @@ addXFTPDelWorker c srv = do
|
||||
doWork <- newTMVarIO ()
|
||||
worker <- async $ runXFTPDelWorker c srv doWork `agentFinally` atomically (TM.delete srv ws)
|
||||
atomically $ TM.insert srv (doWork, worker) ws
|
||||
Just (doWork, _) ->
|
||||
void . atomically $ tryPutTMVar doWork ()
|
||||
Just (doWork, _) -> atomically $ hasWorkToDo' doWork
|
||||
|
||||
runXFTPDelWorker :: forall m. AgentMonad m => AgentClient -> XFTPServer -> TMVar () -> m ()
|
||||
runXFTPDelWorker c srv doWork = do
|
||||
|
||||
+100
-139
@@ -167,7 +167,7 @@ import Simplex.Messaging.Version
|
||||
import Simplex.RemoteControl.Client
|
||||
import Simplex.RemoteControl.Invitation
|
||||
import Simplex.RemoteControl.Types
|
||||
import UnliftIO.Async (async, race_)
|
||||
import UnliftIO.Async (race_)
|
||||
import UnliftIO.Concurrent (forkFinally, forkIO, threadDelay)
|
||||
import UnliftIO.STM
|
||||
|
||||
@@ -431,9 +431,8 @@ rcDiscoverCtrl' pairings = do
|
||||
liftError RCP $ discoverRCCtrl subs pairings
|
||||
|
||||
-- | Activate operations
|
||||
-- stopPending = True is used in iOS NSE
|
||||
foregroundAgent :: MonadUnliftIO m => AgentClient -> Bool -> m ()
|
||||
foregroundAgent c stopPending = withAgentEnv c $ foregroundAgent' c stopPending
|
||||
foregroundAgent :: MonadUnliftIO m => AgentClient -> m ()
|
||||
foregroundAgent c = withAgentEnv c $ foregroundAgent' c
|
||||
|
||||
-- | Suspend operations with max delay to deliver pending messages
|
||||
suspendAgent :: MonadUnliftIO m => AgentClient -> Int -> m ()
|
||||
@@ -627,10 +626,10 @@ newRcvConnSrv :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> SCon
|
||||
newRcvConnSrv c userId connId enableNtfs cMode clientData subMode srv = do
|
||||
AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config
|
||||
(rq, qUri) <- newRcvQueue c userId connId srv smpClientVRange subMode `catchAgentError` \e -> liftIO (print e) >> throwError e
|
||||
void . withStore c $ \db -> updateNewConnRcv db connId rq
|
||||
rq' <- withStore c $ \db -> updateNewConnRcv db connId rq
|
||||
case subMode of
|
||||
SMOnlyCreate -> pure ()
|
||||
SMSubscribe -> addSubscription c rq
|
||||
SMSubscribe -> addSubscription c rq'
|
||||
when enableNtfs $ do
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ sendNtfSubCommand ns (connId, NSCCreate)
|
||||
@@ -651,7 +650,7 @@ joinConn c userId connId enableNtfs cReq cInfo subMode = do
|
||||
_ -> getSMPServer c userId
|
||||
joinConnSrv c userId connId enableNtfs cReq cInfo subMode srv
|
||||
|
||||
startJoinInvitation :: AgentMonad m => UserId -> ConnId -> Bool -> ConnectionRequestUri 'CMInvitation -> m (Compatible Version, ConnData, SndQueue, CR.Ratchet 'C.X448, CR.E2ERatchetParams 'C.X448)
|
||||
startJoinInvitation :: AgentMonad m => UserId -> ConnId -> Bool -> ConnectionRequestUri 'CMInvitation -> m (Compatible Version, ConnData, NewSndQueue, CR.Ratchet 'C.X448, CR.E2ERatchetParams 'C.X448)
|
||||
startJoinInvitation userId connId enableNtfs (CRInvitationUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)} e2eRcvParamsUri) = do
|
||||
AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config
|
||||
case ( qUri `compatibleVersion` smpClientVRange,
|
||||
@@ -674,12 +673,11 @@ joinConnSrv c userId connId enableNtfs inv@CRInvitationUri {} cInfo subMode srv
|
||||
withInvLock c (strEncode inv) "joinConnSrv" $ do
|
||||
(aVersion, cData@ConnData {connAgentVersion}, q, rc, e2eSndParams) <- startJoinInvitation userId connId enableNtfs inv
|
||||
g <- asks random
|
||||
connId' <- withStore c $ \db -> runExceptT $ do
|
||||
connId' <- ExceptT $ createSndConn db g cData q
|
||||
(connId', sq) <- withStore c $ \db -> runExceptT $ do
|
||||
r@(connId', _) <- ExceptT $ createSndConn db g cData q
|
||||
liftIO $ createRatchet db connId' rc
|
||||
pure connId'
|
||||
let sq = (q :: SndQueue) {connId = connId'}
|
||||
cData' = (cData :: ConnData) {connId = connId'}
|
||||
pure r
|
||||
let cData' = (cData :: ConnData) {connId = connId'}
|
||||
duplexHS = connAgentVersion /= 1
|
||||
tryError (confirmQueue aVersion c cData' sq srv cInfo (Just e2eSndParams) subMode) >>= \case
|
||||
Right _ -> do
|
||||
@@ -704,10 +702,9 @@ joinConnSrv c userId connId enableNtfs (CRContactUri ConnReqUriData {crAgentVRan
|
||||
joinConnSrvAsync :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> SubscriptionMode -> SMPServerWithAuth -> m ()
|
||||
joinConnSrvAsync c userId connId enableNtfs inv@CRInvitationUri {} cInfo subMode srv = do
|
||||
(aVersion, cData, q, rc, e2eSndParams) <- startJoinInvitation userId connId enableNtfs inv
|
||||
dbQueueId <- withStore c $ \db -> runExceptT $ do
|
||||
q' <- withStore c $ \db -> runExceptT $ do
|
||||
liftIO $ createRatchet db connId rc
|
||||
ExceptT $ updateNewConnSnd db connId q
|
||||
let q' = (q :: SndQueue) {dbQueueId}
|
||||
confirmQueueAsync aVersion c cData q' srv cInfo (Just e2eSndParams) subMode
|
||||
joinConnSrvAsync _c _userId _connId _enableNtfs (CRContactUri _) _cInfo _subMode _srv = do
|
||||
throwError $ CMD PROHIBITED
|
||||
@@ -716,10 +713,10 @@ createReplyQueue :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> Subsc
|
||||
createReplyQueue c ConnData {userId, connId, enableNtfs} SndQueue {smpClientVersion} subMode srv = do
|
||||
(rq, qUri) <- newRcvQueue c userId connId srv (versionToRange smpClientVersion) subMode
|
||||
let qInfo = toVersionT qUri smpClientVersion
|
||||
rq' <- withStore c $ \db -> upgradeSndConnToDuplex db connId rq
|
||||
case subMode of
|
||||
SMOnlyCreate -> pure ()
|
||||
SMSubscribe -> addSubscription c rq
|
||||
void . withStore c $ \db -> upgradeSndConnToDuplex db connId rq
|
||||
SMSubscribe -> addSubscription c rq'
|
||||
when enableNtfs $ do
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ sendNtfSubCommand ns (connId, NSCCreate)
|
||||
@@ -910,59 +907,48 @@ sendMessagesB c reqs = withConnLocks c connIds "sendMessages" $ do
|
||||
|
||||
enqueueCommand :: AgentMonad m => AgentClient -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> m ()
|
||||
enqueueCommand c corrId connId server aCommand = do
|
||||
withStore c $ \db -> createCommand db corrId connId server aCommand
|
||||
resumeSrvCmds c server
|
||||
commandId <- withStore c $ \db -> createCommand db corrId connId server aCommand
|
||||
queuePendingCommands c server [commandId]
|
||||
|
||||
resumeSrvCmds :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> m ()
|
||||
resumeSrvCmds c server =
|
||||
unlessM (cmdProcessExists c server) $
|
||||
async (runCommandProcessing c server)
|
||||
>>= \a -> atomically (TM.insert server a $ asyncCmdProcesses c)
|
||||
resumeSrvCmds :: forall m. AgentMonad' m => AgentClient -> Maybe SMPServer -> m ()
|
||||
resumeSrvCmds = void .: getAsyncCmdWorker
|
||||
|
||||
resumeConnCmds :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
|
||||
resumeConnCmds c connId =
|
||||
unlessM connQueued $
|
||||
withStore' c (`getPendingCommands` connId)
|
||||
>>= mapM_ (uncurry enqueueConnCmds)
|
||||
withStore' c (`getPendingCommandServers` connId)
|
||||
>>= mapM_ (resumeSrvCmds c)
|
||||
where
|
||||
enqueueConnCmds srv cmdIds = do
|
||||
resumeSrvCmds c srv
|
||||
queuePendingCommands c srv cmdIds
|
||||
connQueued = atomically $ isJust <$> TM.lookupInsert connId True (connCmdsQueued c)
|
||||
|
||||
cmdProcessExists :: AgentMonad' m => AgentClient -> Maybe SMPServer -> m Bool
|
||||
cmdProcessExists c srv = atomically $ TM.member srv (asyncCmdProcesses c)
|
||||
|
||||
queuePendingCommands :: AgentMonad' m => AgentClient -> Maybe SMPServer -> [AsyncCmdId] -> m ()
|
||||
queuePendingCommands c server cmdIds = atomically $ do
|
||||
q <- getPendingCommandQ c server
|
||||
mapM_ (writeTQueue q) cmdIds
|
||||
|
||||
getPendingCommandQ :: AgentClient -> Maybe SMPServer -> STM (TQueue AsyncCmdId)
|
||||
getPendingCommandQ c server = do
|
||||
maybe newMsgQueue pure =<< TM.lookup server (asyncCmdQueues c)
|
||||
getAsyncCmdWorker :: AgentMonad' m => AgentClient -> Maybe SMPServer -> m Worker
|
||||
getAsyncCmdWorker c server =
|
||||
atomically (getWorker >>= maybe createWorker resumeWorker) >>= \w -> runWorker w $> w
|
||||
where
|
||||
newMsgQueue = do
|
||||
cq <- newTQueue
|
||||
TM.insert server cq $ asyncCmdQueues c
|
||||
pure cq
|
||||
getWorker = TM.lookup server $ asyncCmdWorkers c
|
||||
resumeWorker w = hasWorkToDo w $> w
|
||||
deleteWorker wId = mapM_ $ \w -> when (wId == workerId w) $ TM.delete server $ asyncCmdWorkers c
|
||||
createWorker = do
|
||||
w <- newWorker c
|
||||
TM.insert server w $ asyncCmdWorkers c
|
||||
pure w
|
||||
runWorker w@Worker {workerId = wId, doWork} =
|
||||
runWorkerAsync w . void . runExceptT $
|
||||
runCommandProcessing c server doWork
|
||||
`agentFinally` atomically (getWorker >>= deleteWorker wId)
|
||||
|
||||
runCommandProcessing :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> m ()
|
||||
runCommandProcessing c@AgentClient {subQ} server_ = do
|
||||
cq <- atomically $ getPendingCommandQ c server_
|
||||
runCommandProcessing :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> TMVar () -> m ()
|
||||
runCommandProcessing c@AgentClient {subQ} server_ doWork = do
|
||||
ri <- asks $ messageRetryInterval . config -- different retry interval?
|
||||
forever $ do
|
||||
atomically $ endAgentOperation c AOSndNetwork
|
||||
waitForWork doWork
|
||||
atomically $ throwWhenInactive c
|
||||
cmdId <- atomically $ readTQueue cq
|
||||
atomically $ beginAgentOperation c AOSndNetwork
|
||||
tryAgentError (withStore c $ \db -> getPendingCommand db cmdId) >>= \case
|
||||
Left e -> atomically $ writeTBQueue subQ ("", "", APC SAEConn $ ERR e)
|
||||
Right cmd -> processCmd (riFast ri) cmdId cmd
|
||||
withWork c doWork (\db -> getPendingServerCommand db server_) $ processCmd (riFast ri)
|
||||
where
|
||||
processCmd :: RetryInterval -> AsyncCmdId -> PendingCommand -> m ()
|
||||
processCmd ri cmdId PendingCommand {corrId, userId, connId, command} = case command of
|
||||
processCmd :: RetryInterval -> PendingCommand -> m ()
|
||||
processCmd ri PendingCommand {cmdId, corrId, userId, connId, command} = case command of
|
||||
AClientCommand (APC _ cmd) -> case cmd of
|
||||
NEW enableNtfs (ACM cMode) subMode -> noServer $ do
|
||||
usedSrvs <- newTVarIO ([] :: [SMPServer])
|
||||
@@ -1102,13 +1088,10 @@ enqueueMessage c cData sq msgFlags aMessage =
|
||||
-- this function is used only for sending messages in batch, it returns the list of successes to enqueue additional deliveries
|
||||
enqueueMessageB :: forall m t. (AgentMonad' m, Traversable t) => AgentClient -> t (Either AgentErrorType (ConnData, NonEmpty SndQueue, MsgFlags, AMessage)) -> m (t (Either AgentErrorType (AgentMsgId, Maybe (ConnData, [SndQueue], AgentMsgId))))
|
||||
enqueueMessageB c reqs = do
|
||||
void . forME reqs $ \(cData, sq :| _, _, _) ->
|
||||
runExceptT $ resumeMsgDelivery c cData sq
|
||||
aVRange <- asks $ maxVersion . smpAgentVRange . config
|
||||
reqMids <- withStoreBatch c $ \db -> fmap (bindRight $ storeSentMsg db aVRange) reqs
|
||||
forME reqMids $ \((cData, sq :| sqs, _, _), mId) -> do
|
||||
let InternalId msgId = mId
|
||||
queuePendingMsgs c sq [mId]
|
||||
forME reqMids $ \((cData, sq :| sqs, _, _), InternalId msgId) -> do
|
||||
submitPendingMsg c cData sq
|
||||
let sqs' = filter isActiveSndQ sqs
|
||||
pure $ Right (msgId, if null sqs' then Nothing else Just (cData, sqs', msgId))
|
||||
where
|
||||
@@ -1135,67 +1118,54 @@ enqueueSavedMessageB :: (AgentMonad' m, Foldable t) => AgentClient -> t (ConnDat
|
||||
enqueueSavedMessageB c reqs = do
|
||||
-- saving to the database is in the start to avoid race conditions when delivery is read from queue before it is saved
|
||||
void $ withStoreBatch' c $ \db -> concatMap (storeDeliveries db) reqs
|
||||
forM_ reqs $ \(cData, sqs, msgId) ->
|
||||
forM sqs $ \sq -> do
|
||||
void . runExceptT $ resumeMsgDelivery c cData sq
|
||||
let mId = InternalId msgId
|
||||
queuePendingMsgs c sq [mId]
|
||||
forM_ reqs $ \(cData, sqs, _) ->
|
||||
forM sqs $ submitPendingMsg c cData
|
||||
where
|
||||
storeDeliveries :: DB.Connection -> (ConnData, [SndQueue], AgentMsgId) -> [IO ()]
|
||||
storeDeliveries db (ConnData {connId}, sqs, msgId) = do
|
||||
let mId = InternalId msgId
|
||||
in map (\sq -> createSndMsgDelivery db connId sq mId) sqs
|
||||
|
||||
resumeMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> m ()
|
||||
resumeMsgDelivery c cData@ConnData {connId} sq@SndQueue {server, sndId} = do
|
||||
let qKey = (server, sndId)
|
||||
unlessM (queueDelivering qKey) $
|
||||
async (runSmpQueueMsgDelivery c cData sq)
|
||||
>>= \a -> atomically (TM.insert qKey a $ smpQueueMsgDeliveries c)
|
||||
unlessM msgsQueued $
|
||||
withStore' c (\db -> getPendingMsgs db connId sq)
|
||||
>>= queuePendingMsgs' c False sq
|
||||
resumeMsgDelivery :: forall m. AgentMonad' m => AgentClient -> ConnData -> SndQueue -> m ()
|
||||
resumeMsgDelivery = void .:. getDeliveryWorker
|
||||
|
||||
getDeliveryWorker :: AgentMonad' m => AgentClient -> ConnData -> SndQueue -> m (Worker, TMVar ())
|
||||
getDeliveryWorker c cData sq = do
|
||||
atomically (getWorker >>= maybe createWorker resumeWorker) >>= \wl -> runWorker wl $> wl
|
||||
where
|
||||
queueDelivering qKey = atomically $ TM.member qKey (smpQueueMsgDeliveries c)
|
||||
msgsQueued = atomically $ isJust <$> TM.lookupInsert (server, sndId) True (pendingMsgsQueued c)
|
||||
qAddr = qAddress sq
|
||||
getWorker = TM.lookup qAddr $ smpDeliveryWorkers c
|
||||
resumeWorker wl = hasWorkToDo (fst wl) $> wl
|
||||
deleteWorker wId = mapM_ $ \(w, _) -> when (wId == workerId w) $ TM.delete qAddr $ smpDeliveryWorkers c
|
||||
createWorker = do
|
||||
retryLock <- newEmptyTMVar
|
||||
wl <- (,retryLock) <$> newWorker c
|
||||
TM.insert qAddr wl $ smpDeliveryWorkers c
|
||||
pure wl
|
||||
runWorker (w@Worker {workerId = wId, doWork}, retryLock) =
|
||||
runWorkerAsync w . void . runExceptT $
|
||||
runSmpQueueMsgDelivery c cData sq doWork retryLock
|
||||
`agentFinally` atomically (getWorker >>= deleteWorker wId)
|
||||
|
||||
queuePendingMsgs :: AgentMonad' m => AgentClient -> SndQueue -> [InternalId] -> m ()
|
||||
queuePendingMsgs c = queuePendingMsgs' c True
|
||||
submitPendingMsg :: AgentMonad' m => AgentClient -> ConnData -> SndQueue -> m ()
|
||||
submitPendingMsg c cData sq = do
|
||||
atomically $ modifyTVar' (msgDeliveryOp c) $ \s -> s {opsInProgress = opsInProgress s + 1}
|
||||
resumeMsgDelivery c cData sq
|
||||
|
||||
queuePendingMsgs' :: AgentMonad' m => AgentClient -> Bool -> SndQueue -> [InternalId] -> m ()
|
||||
queuePendingMsgs' c setDeliveryOps sq msgIds = atomically $ do
|
||||
when setDeliveryOps $ modifyTVar' (msgDeliveryOp c) $ \s -> s {opsInProgress = opsInProgress s + length msgIds}
|
||||
-- s <- readTVar (msgDeliveryOp c)
|
||||
-- unsafeIOToSTM $ putStrLn $ "msgDeliveryOp: " <> show (opsInProgress s)
|
||||
(mq, _) <- getPendingMsgQ c sq
|
||||
mapM_ (writeTQueue mq . (,setDeliveryOps)) msgIds
|
||||
|
||||
getPendingMsgQ :: AgentClient -> SndQueue -> STM (TQueue (InternalId, Bool), TMVar ())
|
||||
getPendingMsgQ c SndQueue {server, sndId} = do
|
||||
let qKey = (server, sndId)
|
||||
maybe (newMsgQueue qKey) pure =<< TM.lookup qKey (smpQueueMsgQueues c)
|
||||
where
|
||||
newMsgQueue qKey = do
|
||||
q <- (,) <$> newTQueue <*> newEmptyTMVar
|
||||
TM.insert qKey q $ smpQueueMsgQueues c
|
||||
pure q
|
||||
|
||||
runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> m ()
|
||||
runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, duplexHandshake} sq = do
|
||||
(mq, qLock) <- atomically $ getPendingMsgQ c sq
|
||||
runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> TMVar () -> TMVar () -> m ()
|
||||
runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, duplexHandshake} sq doWork qLock = do
|
||||
ri <- asks $ messageRetryInterval . config
|
||||
forever $ do
|
||||
atomically $ endAgentOperation c AOSndNetwork
|
||||
waitForWork doWork
|
||||
atomically $ throwWhenInactive c
|
||||
atomically $ throwWhenNoDelivery c sq
|
||||
(msgId, deliveryOp) <- atomically $ readTQueue mq
|
||||
atomically $ beginAgentOperation c AOSndNetwork
|
||||
when deliveryOp $ atomically $ endAgentOperation c AOMsgDelivery -- this operation begins in queuePendingMsgs
|
||||
let mId = unId msgId
|
||||
tryAgentError (withStore c $ \db -> getPendingMsgData db connId msgId) >>= \case
|
||||
Left e -> notify $ MERR mId e
|
||||
Right (rq_, PendingMsgData {msgType, msgBody, msgFlags, msgRetryState, internalTs}) -> do
|
||||
let ri' = maybe id updateRetryInterval2 msgRetryState ri
|
||||
withWork c doWork (\db -> getPendingQueueMsg db connId sq) $
|
||||
\(rq_, PendingMsgData {msgId, msgType, msgBody, msgFlags, msgRetryState, internalTs}) -> do
|
||||
atomically $ endAgentOperation c AOMsgDelivery -- this operation begins in submitPendingMsg
|
||||
let mId = unId msgId
|
||||
ri' = maybe id updateRetryInterval2 msgRetryState ri
|
||||
withRetryLock2 ri' qLock $ \riState loop -> do
|
||||
resp <- tryError $ case msgType of
|
||||
AM_CONN_INFO -> sendConfirmation c sq msgBody
|
||||
@@ -1301,7 +1271,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
|
||||
Just (sq', sq'' : sqs') -> do
|
||||
checkSQSwchStatus sq' SSSendingQTEST
|
||||
-- remove the delivery from the map to stop the thread when the delivery loop is complete
|
||||
atomically $ TM.delete (qAddress sq') $ smpQueueMsgQueues c
|
||||
atomically $ TM.delete (qAddress sq') $ smpDeliveryWorkers c
|
||||
withStore' c $ \db -> do
|
||||
when primary $ setSndQueuePrimary db connId sq
|
||||
deletePendingMsgs db connId sq'
|
||||
@@ -1387,19 +1357,19 @@ switchConnection' c connId =
|
||||
_ -> throwError $ CMD PROHIBITED
|
||||
|
||||
switchDuplexConnection :: AgentMonad m => AgentClient -> Connection 'CDuplex -> RcvQueue -> m ConnectionStats
|
||||
switchDuplexConnection c (DuplexConnection cData@ConnData {connId, userId} rqs sqs) rq@RcvQueue {server, dbQueueId, sndId} = do
|
||||
switchDuplexConnection c (DuplexConnection cData@ConnData {connId, userId} rqs sqs) rq@RcvQueue {server, dbQueueId = DBQueueId dbQueueId, sndId} = do
|
||||
checkRQSwchStatus rq RSSwitchStarted
|
||||
clientVRange <- asks $ smpClientVRange . config
|
||||
-- try to get the server that is different from all queues, or at least from the primary rcv queue
|
||||
srvAuth@(ProtoServerWithAuth srv _) <- getNextServer c userId $ map qServer (L.toList rqs) <> map qServer (L.toList sqs)
|
||||
srv' <- if srv == server then getNextServer c userId [server] else pure srvAuth
|
||||
(q, qUri) <- newRcvQueue c userId connId srv' clientVRange SMSubscribe
|
||||
let rq' = (q :: RcvQueue) {primary = True, dbReplaceQueueId = Just dbQueueId}
|
||||
void . withStore c $ \db -> addConnRcvQueue db connId rq'
|
||||
addSubscription c rq'
|
||||
let rq' = (q :: NewRcvQueue) {primary = True, dbReplaceQueueId = Just dbQueueId}
|
||||
rq'' <- withStore c $ \db -> addConnRcvQueue db connId rq'
|
||||
addSubscription c rq''
|
||||
void . enqueueMessages c cData sqs SMP.noMsgFlags $ QADD [(qUri, Just (server, sndId))]
|
||||
rq1 <- withStore' c $ \db -> setRcvSwitchStatus db rq $ Just RSSendingQADD
|
||||
let rqs' = updatedQs rq1 rqs <> [rq']
|
||||
let rqs' = updatedQs rq1 rqs <> [rq'']
|
||||
pure . connectionStats $ DuplexConnection cData rqs' sqs
|
||||
|
||||
abortConnectionSwitch' :: AgentMonad m => AgentClient -> ConnId -> m ConnectionStats
|
||||
@@ -1435,7 +1405,7 @@ synchronizeRatchet' c connId force = withConnLock c connId "synchronizeRatchet"
|
||||
AgentConfig {e2eEncryptVRange} <- asks config
|
||||
g <- asks random
|
||||
(pk1, pk2, e2eParams@(CR.E2ERatchetParams _ k1 k2)) <- atomically . CR.generateE2EParams g $ maxVersion e2eEncryptVRange
|
||||
void $ enqueueRatchetKeyMsgs c cData sqs e2eParams
|
||||
enqueueRatchetKeyMsgs c cData sqs e2eParams
|
||||
withStore' c $ \db -> do
|
||||
setConnRatchetSync db connId RSStarted
|
||||
setRatchetX3dhKeys db connId pk1 pk2 k1 k2
|
||||
@@ -1804,11 +1774,8 @@ sendNtfConnCommands c cmd = do
|
||||
setNtfServers' :: AgentMonad' m => AgentClient -> [NtfServer] -> m ()
|
||||
setNtfServers' c = atomically . writeTVar (ntfServers c)
|
||||
|
||||
-- stopPending = True is used in iOS NSE
|
||||
foregroundAgent' :: AgentMonad' m => AgentClient -> Bool -> m ()
|
||||
foregroundAgent' c stopPending = do
|
||||
-- to prevent race conditions with message deliveries and async commands in the app
|
||||
when stopPending $ stopPendingActions c
|
||||
foregroundAgent' :: AgentMonad' m => AgentClient -> m ()
|
||||
foregroundAgent' c = do
|
||||
atomically $ writeTVar (agentState c) ASForeground
|
||||
mapM_ activate $ reverse agentOperations
|
||||
where
|
||||
@@ -2222,9 +2189,9 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
|
||||
case findQ addr sqs of
|
||||
Just sq -> do
|
||||
logServer "<--" c srv rId "MSG <QCONT>"
|
||||
atomically $ do
|
||||
(_, qLock) <- getPendingMsgQ c sq
|
||||
void $ tryPutTMVar qLock ()
|
||||
atomically $
|
||||
TM.lookup (qAddress sq) (smpDeliveryWorkers c)
|
||||
>>= mapM_ (\(_, retryLock) -> tryPutTMVar retryLock ())
|
||||
Nothing -> qError "QCONT: queue address not found"
|
||||
|
||||
messagesRcvd :: NonEmpty AMessageReceipt -> MsgMeta -> Connection 'CDuplex -> m ()
|
||||
@@ -2259,16 +2226,15 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
|
||||
Just qInfo@(Compatible sqInfo@SMPQueueInfo {queueAddress}) ->
|
||||
case (findQ (qAddress sqInfo) sqs, findQ addr sqs) of
|
||||
(Just _, _) -> qError "QADD: queue address is already used in connection"
|
||||
(_, Just sq@SndQueue {dbQueueId}) -> do
|
||||
(_, Just sq@SndQueue {dbQueueId = DBQueueId dbQueueId}) -> do
|
||||
let (delSqs, keepSqs) = L.partition ((Just dbQueueId ==) . dbReplaceQId) sqs
|
||||
case L.nonEmpty keepSqs of
|
||||
Just sqs' -> do
|
||||
-- move inside case?
|
||||
withStore' c $ \db -> mapM_ (deleteConnSndQueue db connId) delSqs
|
||||
sq_@SndQueue {sndPublicKey, e2ePubKey} <- newSndQueue userId connId qInfo
|
||||
let sq'' = (sq_ :: SndQueue) {primary = True, dbQueueId, dbReplaceQueueId = Just dbQueueId}
|
||||
dbId <- withStore c $ \db -> addConnSndQueue db connId sq''
|
||||
let sq2 = (sq'' :: SndQueue) {dbQueueId = dbId}
|
||||
let sq'' = (sq_ :: NewSndQueue) {primary = True, dbReplaceQueueId = Just dbQueueId}
|
||||
sq2 <- withStore c $ \db -> addConnSndQueue db connId sq''
|
||||
case (sndPublicKey, e2ePubKey) of
|
||||
(Just sndPubKey, Just dhPublicKey) -> do
|
||||
logServer "<--" c srv rId $ "MSG <QADD> " <> logSecret (senderId queueAddress)
|
||||
@@ -2386,7 +2352,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
|
||||
sendReplyKey = do
|
||||
g <- asks random
|
||||
(pk1, pk2, e2eParams@(CR.E2ERatchetParams _ k1 k2)) <- atomically . CR.generateE2EParams g $ version e2eOtherPartyParams
|
||||
void $ enqueueRatchetKeyMsgs c cData' sqs e2eParams
|
||||
enqueueRatchetKeyMsgs c cData' sqs e2eParams
|
||||
pure (pk1, pk2, k1, k2)
|
||||
notifyRatchetSyncError = do
|
||||
let cData'' = cData' {ratchetSyncState = RSRequired} :: ConnData
|
||||
@@ -2444,14 +2410,13 @@ connectReplyQueues c cData@ConnData {userId, connId} ownConnInfo (qInfo :| _) =
|
||||
Nothing -> throwError $ AGENT A_VERSION
|
||||
Just qInfo' -> do
|
||||
sq <- newSndQueue userId connId qInfo'
|
||||
dbQueueId <- withStore c $ \db -> upgradeRcvConnToDuplex db connId sq
|
||||
enqueueConfirmation c cData sq {dbQueueId} ownConnInfo Nothing
|
||||
sq' <- withStore c $ \db -> upgradeRcvConnToDuplex db connId sq
|
||||
enqueueConfirmation c cData sq' ownConnInfo Nothing
|
||||
|
||||
confirmQueueAsync :: forall m. AgentMonad m => Compatible Version -> AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> SubscriptionMode -> m ()
|
||||
confirmQueueAsync v c cData sq srv connInfo e2eEncryption_ subMode = do
|
||||
resumeMsgDelivery c cData sq
|
||||
msgId <- storeConfirmation c cData sq e2eEncryption_ =<< mkAgentConfirmation v c cData sq srv connInfo subMode
|
||||
queuePendingMsgs c sq [msgId]
|
||||
storeConfirmation c cData sq e2eEncryption_ =<< mkAgentConfirmation v c cData sq srv connInfo subMode
|
||||
submitPendingMsg c cData sq
|
||||
|
||||
confirmQueue :: forall m. AgentMonad m => Compatible Version -> AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> SubscriptionMode -> m ()
|
||||
confirmQueue v@(Compatible agentVersion) c cData@ConnData {connId} sq srv connInfo e2eEncryption_ subMode = do
|
||||
@@ -2474,11 +2439,10 @@ mkAgentConfirmation (Compatible agentVersion) c cData sq srv connInfo subMode
|
||||
|
||||
enqueueConfirmation :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
|
||||
enqueueConfirmation c cData sq connInfo e2eEncryption_ = do
|
||||
resumeMsgDelivery c cData sq
|
||||
msgId <- storeConfirmation c cData sq e2eEncryption_ $ AgentConnInfo connInfo
|
||||
queuePendingMsgs c sq [msgId]
|
||||
storeConfirmation c cData sq e2eEncryption_ $ AgentConnInfo connInfo
|
||||
submitPendingMsg c cData sq
|
||||
|
||||
storeConfirmation :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> Maybe (CR.E2ERatchetParams 'C.X448) -> AgentMessage -> m InternalId
|
||||
storeConfirmation :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> Maybe (CR.E2ERatchetParams 'C.X448) -> AgentMessage -> m ()
|
||||
storeConfirmation c ConnData {connId, connAgentVersion} sq e2eEncryption_ agentMsg = withStore c $ \db -> runExceptT $ do
|
||||
internalTs <- liftIO getCurrentTime
|
||||
(internalId, internalSndId, prevMsgHash) <- liftIO $ updateSndIds db connId
|
||||
@@ -2490,20 +2454,17 @@ storeConfirmation c ConnData {connId, connAgentVersion} sq e2eEncryption_ agentM
|
||||
msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash}
|
||||
liftIO $ createSndMsg db connId msgData
|
||||
liftIO $ createSndMsgDelivery db connId sq internalId
|
||||
pure internalId
|
||||
|
||||
enqueueRatchetKeyMsgs :: forall m. AgentMonad m => AgentClient -> ConnData -> NonEmpty SndQueue -> CR.E2ERatchetParams 'C.X448 -> m AgentMsgId
|
||||
enqueueRatchetKeyMsgs :: forall m. AgentMonad m => AgentClient -> ConnData -> NonEmpty SndQueue -> CR.E2ERatchetParams 'C.X448 -> m ()
|
||||
enqueueRatchetKeyMsgs c cData (sq :| sqs) e2eEncryption = do
|
||||
msgId <- enqueueRatchetKey c cData sq e2eEncryption
|
||||
mapM_ (enqueueSavedMessage c cData msgId) $ filter isActiveSndQ sqs
|
||||
pure msgId
|
||||
|
||||
enqueueRatchetKey :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> CR.E2ERatchetParams 'C.X448 -> m AgentMsgId
|
||||
enqueueRatchetKey c cData@ConnData {connId} sq e2eEncryption = do
|
||||
resumeMsgDelivery c cData sq
|
||||
aVRange <- asks $ smpAgentVRange . config
|
||||
msgId <- storeRatchetKey $ maxVersion aVRange
|
||||
queuePendingMsgs c sq [msgId]
|
||||
submitPendingMsg c cData sq
|
||||
pure $ unId msgId
|
||||
where
|
||||
storeRatchetKey :: Version -> m InternalId
|
||||
@@ -2541,7 +2502,7 @@ agentRatchetDecrypt' g db connId rc encAgentMsg = do
|
||||
liftIO $ updateRatchet db connId rc' skippedDiff
|
||||
liftEither $ first (SEAgentError . cryptoError) agentMsgBody_
|
||||
|
||||
newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => UserId -> ConnId -> Compatible SMPQueueInfo -> m SndQueue
|
||||
newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => UserId -> ConnId -> Compatible SMPQueueInfo -> m NewSndQueue
|
||||
newSndQueue userId connId (Compatible (SMPQueueInfo smpClientVersion SMPQueueAddress {smpServer, senderId, dhPublicKey = rcvE2ePubDhKey})) = do
|
||||
C.SignAlg a <- asks $ cmdSignAlg . config
|
||||
g <- asks random
|
||||
@@ -2558,7 +2519,7 @@ newSndQueue userId connId (Compatible (SMPQueueInfo smpClientVersion SMPQueueAdd
|
||||
e2eDhSecret = C.dh' rcvE2ePubDhKey e2ePrivKey,
|
||||
e2ePubKey = Just e2ePubKey,
|
||||
status = New,
|
||||
dbQueueId = 0,
|
||||
dbQueueId = DBNewQueue,
|
||||
primary = True,
|
||||
dbReplaceQueueId = Nothing,
|
||||
sndSwchStatus = Nothing,
|
||||
|
||||
@@ -29,7 +29,6 @@ module Simplex.Messaging.Agent.Client
|
||||
closeAgentClient,
|
||||
closeProtocolServerClients,
|
||||
closeXFTPServerClient,
|
||||
stopPendingActions,
|
||||
runSMPServerTest,
|
||||
runXFTPServerTest,
|
||||
getXFTPWorkPath,
|
||||
@@ -78,6 +77,7 @@ module Simplex.Messaging.Agent.Client
|
||||
agentClientStore,
|
||||
agentDRG,
|
||||
getAgentSubscriptions,
|
||||
Worker (..),
|
||||
SubscriptionsInfo (..),
|
||||
SubInfo (..),
|
||||
AgentOperation (..),
|
||||
@@ -85,7 +85,11 @@ module Simplex.Messaging.Agent.Client
|
||||
AgentState (..),
|
||||
AgentLocks (..),
|
||||
AgentStatsKey (..),
|
||||
newWorker,
|
||||
runWorkerAsync,
|
||||
waitForWork,
|
||||
hasWorkToDo,
|
||||
hasWorkToDo',
|
||||
withWork,
|
||||
agentOperations,
|
||||
agentOperationBracket,
|
||||
@@ -138,7 +142,7 @@ import qualified Data.List.NonEmpty as L
|
||||
import Data.Maybe (isNothing)
|
||||
import Data.Map.Strict (Map)
|
||||
import qualified Data.Map.Strict as M
|
||||
import Data.Maybe (isJust, listToMaybe)
|
||||
import Data.Maybe (listToMaybe)
|
||||
import Data.Set (Set)
|
||||
import qualified Data.Set as S
|
||||
import Data.Text (Text)
|
||||
@@ -207,7 +211,9 @@ import Simplex.Messaging.Version
|
||||
import System.Random (randomR)
|
||||
import System.Timeout (timeout)
|
||||
import UnliftIO (mapConcurrently)
|
||||
import UnliftIO.Async (async)
|
||||
import UnliftIO.Directory (getTemporaryDirectory)
|
||||
import UnliftIO.Exception (bracket)
|
||||
import qualified UnliftIO.Exception as E
|
||||
import UnliftIO.STM
|
||||
|
||||
@@ -241,12 +247,10 @@ data AgentClient = AgentClient
|
||||
activeSubs :: TRcvQueues,
|
||||
pendingSubs :: TRcvQueues,
|
||||
removedSubs :: TMap (UserId, SMPServer, SMP.RecipientId) SMPClientError,
|
||||
pendingMsgsQueued :: TMap SndQAddr Bool,
|
||||
smpQueueMsgQueues :: TMap SndQAddr (TQueue (InternalId, Bool), TMVar ()),
|
||||
smpQueueMsgDeliveries :: TMap SndQAddr (Async ()),
|
||||
workerSeq :: TVar Int,
|
||||
smpDeliveryWorkers :: TMap SndQAddr (Worker, TMVar ()),
|
||||
asyncCmdWorkers :: TMap (Maybe SMPServer) Worker,
|
||||
connCmdsQueued :: TMap ConnId Bool,
|
||||
asyncCmdQueues :: TMap (Maybe SMPServer) (TQueue AsyncCmdId),
|
||||
asyncCmdProcesses :: TMap (Maybe SMPServer) (Async ()),
|
||||
ntfNetworkOp :: TVar AgentOpState,
|
||||
rcvNetworkOp :: TVar AgentOpState,
|
||||
msgDeliveryOp :: TVar AgentOpState,
|
||||
@@ -269,6 +273,24 @@ data AgentClient = AgentClient
|
||||
agentEnv :: Env
|
||||
}
|
||||
|
||||
data Worker = Worker {workerId :: Int, doWork :: TMVar (), action :: TMVar (Maybe (Async ()))}
|
||||
|
||||
newWorker :: AgentClient -> STM Worker
|
||||
newWorker c = do
|
||||
workerId <- stateTVar (workerSeq c) $ \next -> (next, next + 1)
|
||||
doWork <- newTMVar ()
|
||||
action <- newTMVar Nothing
|
||||
pure Worker {workerId, doWork, action}
|
||||
|
||||
runWorkerAsync :: AgentMonad' m => Worker -> m () -> m ()
|
||||
runWorkerAsync Worker {action} worker =
|
||||
bracket
|
||||
(atomically $ takeTMVar action) -- get current action, locking to avoid race conditions
|
||||
(atomically . tryPutTMVar action) -- if it was running (or if start crashes), put it back and unlock (don't lock if it was just started)
|
||||
(\a -> when (isNothing a) start) -- start worker if it's not running
|
||||
where
|
||||
start = atomically . putTMVar action . Just =<< async worker
|
||||
|
||||
data AgentOperation = AONtfNetwork | AORcvNetwork | AOMsgDelivery | AOSndNetwork | AODatabase
|
||||
deriving (Eq, Show)
|
||||
|
||||
@@ -323,12 +345,10 @@ newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
|
||||
activeSubs <- RQ.empty
|
||||
pendingSubs <- RQ.empty
|
||||
removedSubs <- TM.empty
|
||||
pendingMsgsQueued <- TM.empty
|
||||
smpQueueMsgQueues <- TM.empty
|
||||
smpQueueMsgDeliveries <- TM.empty
|
||||
workerSeq <- newTVar 0
|
||||
smpDeliveryWorkers <- TM.empty
|
||||
asyncCmdWorkers <- TM.empty
|
||||
connCmdsQueued <- TM.empty
|
||||
asyncCmdQueues <- TM.empty
|
||||
asyncCmdProcesses <- TM.empty
|
||||
ntfNetworkOp <- newTVar $ AgentOpState False 0
|
||||
rcvNetworkOp <- newTVar $ AgentOpState False 0
|
||||
msgDeliveryOp <- newTVar $ AgentOpState False 0
|
||||
@@ -361,12 +381,10 @@ newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
|
||||
activeSubs,
|
||||
pendingSubs,
|
||||
removedSubs,
|
||||
pendingMsgsQueued,
|
||||
smpQueueMsgQueues,
|
||||
smpQueueMsgDeliveries,
|
||||
workerSeq,
|
||||
smpDeliveryWorkers,
|
||||
asyncCmdWorkers,
|
||||
connCmdsQueued,
|
||||
asyncCmdQueues,
|
||||
asyncCmdProcesses,
|
||||
ntfNetworkOp,
|
||||
rcvNetworkOp,
|
||||
msgDeliveryOp,
|
||||
@@ -608,28 +626,23 @@ closeAgentClient c = liftIO $ do
|
||||
closeProtocolServerClients c xftpClients
|
||||
cancelActions . actions $ reconnections c
|
||||
cancelActions . actions $ asyncClients c
|
||||
stopPendingActions c
|
||||
clearWorkers smpDeliveryWorkers >>= mapM_ (cancelWorker . fst)
|
||||
clearWorkers asyncCmdWorkers >>= mapM_ cancelWorker
|
||||
clear connCmdsQueued
|
||||
atomically . RQ.clear $ activeSubs c
|
||||
atomically . RQ.clear $ pendingSubs c
|
||||
clear subscrConns
|
||||
clear getMsgLocks
|
||||
where
|
||||
clearWorkers :: Ord k => (AgentClient -> TMap k a) -> IO (Map k a)
|
||||
clearWorkers workers = atomically $ swapTVar (workers c) mempty
|
||||
clear :: Monoid m => (AgentClient -> TVar m) -> IO ()
|
||||
clear sel = atomically $ writeTVar (sel c) mempty
|
||||
|
||||
stopPendingActions :: MonadIO m => AgentClient -> m ()
|
||||
stopPendingActions c = liftIO $ do
|
||||
-- messages
|
||||
cancelActions $ smpQueueMsgDeliveries c
|
||||
clear pendingMsgsQueued
|
||||
clear smpQueueMsgQueues
|
||||
-- commands
|
||||
cancelActions $ asyncCmdProcesses c
|
||||
clear connCmdsQueued
|
||||
clear asyncCmdQueues
|
||||
where
|
||||
clear :: Monoid m => (AgentClient -> TVar m) -> IO ()
|
||||
clear sel = atomically $ writeTVar (sel c) mempty
|
||||
cancelWorker :: Worker -> IO ()
|
||||
cancelWorker Worker {doWork, action} = do
|
||||
noWorkToDo doWork
|
||||
atomically (tryTakeTMVar action) >>= mapM_ (mapM_ uninterruptibleCancel)
|
||||
|
||||
waitUntilActive :: AgentClient -> STM ()
|
||||
waitUntilActive c = unlessM (readTVar $ active c) retry
|
||||
@@ -637,13 +650,11 @@ waitUntilActive c = unlessM (readTVar $ active c) retry
|
||||
throwWhenInactive :: AgentClient -> STM ()
|
||||
throwWhenInactive c = unlessM (readTVar $ active c) $ throwSTM ThreadKilled
|
||||
|
||||
-- this function is used to remove workers once delivery is complete, not when it is removed from the map
|
||||
throwWhenNoDelivery :: AgentClient -> SndQueue -> STM ()
|
||||
throwWhenNoDelivery c SndQueue {server, sndId} =
|
||||
unlessM (isJust <$> TM.lookup k (smpQueueMsgQueues c)) $ do
|
||||
TM.delete k $ smpQueueMsgDeliveries c
|
||||
throwWhenNoDelivery c sq =
|
||||
unlessM (TM.member (qAddress sq) $ smpDeliveryWorkers c) $
|
||||
throwSTM ThreadKilled
|
||||
where
|
||||
k = (server, sndId)
|
||||
|
||||
closeProtocolServerClients :: ProtocolServerClient err msg => AgentClient -> (AgentClient -> TMap (TransportSession msg) (ClientVar msg)) -> IO ()
|
||||
closeProtocolServerClients c clientsSel =
|
||||
@@ -860,7 +871,7 @@ mkSMPTSession q = mkTSession (qUserId q) (qServer q) (qConnId q)
|
||||
getSessionMode :: AgentMonad' m => AgentClient -> m TransportSessionMode
|
||||
getSessionMode = fmap sessionMode . readTVarIO . useNetworkConfig
|
||||
|
||||
newRcvQueue :: AgentMonad m => AgentClient -> UserId -> ConnId -> SMPServerWithAuth -> VersionRange -> SubscriptionMode -> m (RcvQueue, SMPQueueUri)
|
||||
newRcvQueue :: AgentMonad m => AgentClient -> UserId -> ConnId -> SMPServerWithAuth -> VersionRange -> SubscriptionMode -> m (NewRcvQueue, SMPQueueUri)
|
||||
newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange subMode = do
|
||||
C.SignAlg a <- asks (cmdSignAlg . config)
|
||||
g <- asks random
|
||||
@@ -884,7 +895,7 @@ newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange subMode = do
|
||||
e2eDhSecret = Nothing,
|
||||
sndId,
|
||||
status = New,
|
||||
dbQueueId = 0,
|
||||
dbQueueId = DBNewQueue,
|
||||
primary = True,
|
||||
dbReplaceQueueId = Nothing,
|
||||
rcvSwchStatus = Nothing,
|
||||
@@ -1227,18 +1238,25 @@ cryptoError = \case
|
||||
where
|
||||
c = AGENT . A_CRYPTO
|
||||
|
||||
waitForWork :: AgentMonad m => TMVar () -> m ()
|
||||
waitForWork :: AgentMonad' m => TMVar () -> m ()
|
||||
waitForWork = void . atomically . readTMVar
|
||||
|
||||
withWork :: AgentMonad m => AgentClient -> TMVar () -> (DB.Connection -> IO (Maybe a)) -> (a -> m ()) -> m ()
|
||||
withWork c doWork getWork action = do
|
||||
r <- withStore' c $ \db -> do
|
||||
r' <- getWork db
|
||||
when (isNothing r') noWorkToDo
|
||||
liftIO $ when (isNothing r') $ noWorkToDo doWork
|
||||
pure r'
|
||||
mapM_ action r
|
||||
where
|
||||
noWorkToDo = void . atomically $ tryTakeTMVar doWork
|
||||
|
||||
noWorkToDo :: TMVar () -> IO ()
|
||||
noWorkToDo = void . atomically . tryTakeTMVar
|
||||
|
||||
hasWorkToDo :: Worker -> STM ()
|
||||
hasWorkToDo = hasWorkToDo' . doWork
|
||||
|
||||
hasWorkToDo' :: TMVar () -> STM ()
|
||||
hasWorkToDo' = void . (`tryPutTMVar` ())
|
||||
|
||||
endAgentOperation :: AgentClient -> AgentOperation -> STM ()
|
||||
endAgentOperation c op = endOperation c op $ case op of
|
||||
|
||||
@@ -149,8 +149,7 @@ processNtfSub c (connId, cmd) = do
|
||||
doWork <- newTMVarIO ()
|
||||
worker <- async $ runWorker c srv doWork `agentFinally` atomically (TM.delete srv ws)
|
||||
atomically $ TM.insert srv (doWork, worker) ws
|
||||
Just (doWork, _) ->
|
||||
void . atomically $ tryPutTMVar doWork ()
|
||||
Just (doWork, _) -> atomically $ hasWorkToDo' doWork
|
||||
|
||||
withNtfServer :: AgentMonad' m => AgentClient -> (NtfServer -> m ()) -> m ()
|
||||
withNtfServer c action = getNtfServer c >>= mapM_ action
|
||||
@@ -288,7 +287,7 @@ rescheduleAction doWork ts actionTs
|
||||
void . atomically $ tryTakeTMVar doWork
|
||||
void . forkIO $ do
|
||||
liftIO $ threadDelay' $ diffToMicroseconds $ diffUTCTime actionTs ts
|
||||
void . atomically $ tryPutTMVar doWork ()
|
||||
atomically $ hasWorkToDo' doWork
|
||||
pure True
|
||||
|
||||
retryOnError :: AgentMonad' m => AgentClient -> Text -> m () -> (AgentErrorType -> m ()) -> AgentErrorType -> m ()
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
{-# LANGUAGE DeriveAnyClass #-}
|
||||
{-# LANGUAGE DuplicateRecordFields #-}
|
||||
{-# LANGUAGE FlexibleContexts #-}
|
||||
{-# LANGUAGE FlexibleInstances #-}
|
||||
{-# LANGUAGE GADTs #-}
|
||||
{-# LANGUAGE KindSignatures #-}
|
||||
{-# LANGUAGE LambdaCase #-}
|
||||
@@ -49,8 +50,26 @@ import Simplex.Messaging.Version
|
||||
|
||||
-- * Queue types
|
||||
|
||||
data QueueStored = QSStored | QSNew
|
||||
|
||||
data SQueueStored (q :: QueueStored) where
|
||||
SQSStored :: SQueueStored 'QSStored
|
||||
SQSNew :: SQueueStored 'QSNew
|
||||
|
||||
data DBQueueId (q :: QueueStored) where
|
||||
DBQueueId :: Int64 -> DBQueueId 'QSStored
|
||||
DBNewQueue :: DBQueueId 'QSNew
|
||||
|
||||
deriving instance Eq (DBQueueId q)
|
||||
|
||||
deriving instance Show (DBQueueId q)
|
||||
|
||||
type RcvQueue = StoredRcvQueue 'QSStored
|
||||
|
||||
type NewRcvQueue = StoredRcvQueue 'QSNew
|
||||
|
||||
-- | A receive queue. SMP queue through which the agent receives messages from a sender.
|
||||
data RcvQueue = RcvQueue
|
||||
data StoredRcvQueue (q :: QueueStored) = RcvQueue
|
||||
{ userId :: UserId,
|
||||
connId :: ConnId,
|
||||
server :: SMPServer,
|
||||
@@ -69,7 +88,7 @@ data RcvQueue = RcvQueue
|
||||
-- | queue status
|
||||
status :: QueueStatus,
|
||||
-- | database queue ID (within connection)
|
||||
dbQueueId :: Int64,
|
||||
dbQueueId :: DBQueueId q,
|
||||
-- | True for a primary or a next primary queue of the connection (next if dbReplaceQueueId is set)
|
||||
primary :: Bool,
|
||||
-- | database queue ID to replace, Nothing if this queue is not replacing another, `Just Nothing` is used for replacing old queues
|
||||
@@ -110,8 +129,12 @@ data ClientNtfCreds = ClientNtfCreds
|
||||
}
|
||||
deriving (Eq, Show)
|
||||
|
||||
type SndQueue = StoredSndQueue 'QSStored
|
||||
|
||||
type NewSndQueue = StoredSndQueue 'QSNew
|
||||
|
||||
-- | A send queue. SMP queue through which the agent sends messages to a recipient.
|
||||
data SndQueue = SndQueue
|
||||
data StoredSndQueue (q :: QueueStored) = SndQueue
|
||||
{ userId :: UserId,
|
||||
connId :: ConnId,
|
||||
server :: SMPServer,
|
||||
@@ -127,7 +150,7 @@ data SndQueue = SndQueue
|
||||
-- | queue status
|
||||
status :: QueueStatus,
|
||||
-- | database queue ID (within connection)
|
||||
dbQueueId :: Int64,
|
||||
dbQueueId :: DBQueueId q,
|
||||
-- | True for a primary or a next primary queue of the connection (next if dbReplaceQueueId is set)
|
||||
primary :: Bool,
|
||||
-- | ID of the queue this one is replacing
|
||||
@@ -194,7 +217,7 @@ instance SMPQueueRec RcvQueue where
|
||||
{-# INLINE qUserId #-}
|
||||
qConnId RcvQueue {connId} = connId
|
||||
{-# INLINE qConnId #-}
|
||||
dbQId RcvQueue {dbQueueId} = dbQueueId
|
||||
dbQId RcvQueue {dbQueueId = DBQueueId qId} = qId
|
||||
{-# INLINE dbQId #-}
|
||||
dbReplaceQId RcvQueue {dbReplaceQueueId} = dbReplaceQueueId
|
||||
{-# INLINE dbReplaceQId #-}
|
||||
@@ -204,7 +227,7 @@ instance SMPQueueRec SndQueue where
|
||||
{-# INLINE qUserId #-}
|
||||
qConnId SndQueue {connId} = connId
|
||||
{-# INLINE qConnId #-}
|
||||
dbQId SndQueue {dbQueueId} = dbQueueId
|
||||
dbQId SndQueue {dbQueueId = DBQueueId qId} = qId
|
||||
{-# INLINE dbQId #-}
|
||||
dbReplaceQId SndQueue {dbReplaceQueueId} = dbReplaceQueueId
|
||||
{-# INLINE dbReplaceQId #-}
|
||||
@@ -316,7 +339,8 @@ ratchetSyncSendProhibited ConnData {ratchetSyncState} =
|
||||
ratchetSyncState `elem` ([RSRequired, RSStarted, RSAgreed] :: [RatchetSyncState])
|
||||
|
||||
data PendingCommand = PendingCommand
|
||||
{ corrId :: ACorrId,
|
||||
{ cmdId :: AsyncCmdId,
|
||||
corrId :: ACorrId,
|
||||
userId :: UserId,
|
||||
connId :: ConnId,
|
||||
command :: AgentCommand
|
||||
|
||||
@@ -50,7 +50,7 @@ module Simplex.Messaging.Agent.Store.SQLite
|
||||
createNewConn,
|
||||
updateNewConnRcv,
|
||||
updateNewConnSnd,
|
||||
createRcvConn,
|
||||
createRcvConn, -- no longer used
|
||||
createSndConn,
|
||||
getConn,
|
||||
getDeletedConn,
|
||||
@@ -107,9 +107,8 @@ module Simplex.Messaging.Agent.Store.SQLite
|
||||
createSndMsgDelivery,
|
||||
getSndMsgViaRcpt,
|
||||
updateSndMsgRcpt,
|
||||
getPendingMsgData,
|
||||
getPendingQueueMsg,
|
||||
updatePendingMsgRIState,
|
||||
getPendingMsgs,
|
||||
deletePendingMsgs,
|
||||
setMsgUserAck,
|
||||
getRcvMsg,
|
||||
@@ -133,8 +132,8 @@ module Simplex.Messaging.Agent.Store.SQLite
|
||||
updateRatchet,
|
||||
-- Async commands
|
||||
createCommand,
|
||||
getPendingCommands,
|
||||
getPendingCommand,
|
||||
getPendingCommandServers,
|
||||
getPendingServerCommand,
|
||||
deleteCommand,
|
||||
-- Notification device token persistence
|
||||
createNtfToken,
|
||||
@@ -272,7 +271,7 @@ import Simplex.Messaging.Parsers (blobFieldParser, defaultJSON, dropPrefix, from
|
||||
import Simplex.Messaging.Protocol
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import Simplex.Messaging.Transport.Client (TransportHost)
|
||||
import Simplex.Messaging.Util (bshow, eitherToMaybe, groupOn, ifM, safeDecodeUtf8, ($>>=), (<$$>))
|
||||
import Simplex.Messaging.Util (bshow, eitherToMaybe, ifM, safeDecodeUtf8, ($>>=), (<$$>))
|
||||
import Simplex.Messaging.Version
|
||||
import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
|
||||
import System.Exit (exitFailure)
|
||||
@@ -530,54 +529,56 @@ deleteUsersWithoutConns db = do
|
||||
createConn_ ::
|
||||
TVar ChaChaDRG ->
|
||||
ConnData ->
|
||||
(ByteString -> IO ()) ->
|
||||
IO (Either StoreError ByteString)
|
||||
(ConnId -> IO a) ->
|
||||
IO (Either StoreError (ConnId, a))
|
||||
createConn_ gVar cData create = checkConstraint SEConnDuplicate $ case cData of
|
||||
ConnData {connId = ""} -> createWithRandomId gVar create
|
||||
ConnData {connId} -> create connId $> Right connId
|
||||
ConnData {connId = ""} -> createWithRandomId' gVar create
|
||||
ConnData {connId} -> Right . (connId,) <$> create connId
|
||||
|
||||
createNewConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> SConnectionMode c -> IO (Either StoreError ConnId)
|
||||
createNewConn db gVar cData@ConnData {userId, connAgentVersion, enableNtfs, duplexHandshake} cMode =
|
||||
createConn_ gVar cData $ \connId -> do
|
||||
DB.execute db "INSERT INTO connections (user_id, conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?,?,?,?,?,?)" (userId, connId, cMode, connAgentVersion, enableNtfs, duplexHandshake)
|
||||
createNewConn db gVar cData@ConnData {userId, connAgentVersion, enableNtfs, duplexHandshake} cMode = do
|
||||
fst <$$> createConn_ gVar cData create
|
||||
where
|
||||
create connId =
|
||||
DB.execute db "INSERT INTO connections (user_id, conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?,?,?,?,?,?)" (userId, connId, cMode, connAgentVersion, enableNtfs, duplexHandshake)
|
||||
|
||||
updateNewConnRcv :: DB.Connection -> ConnId -> RcvQueue -> IO (Either StoreError Int64)
|
||||
updateNewConnRcv :: DB.Connection -> ConnId -> NewRcvQueue -> IO (Either StoreError RcvQueue)
|
||||
updateNewConnRcv db connId rq =
|
||||
getConn db connId $>>= \case
|
||||
(SomeConn _ NewConnection {}) -> updateConn
|
||||
(SomeConn _ RcvConnection {}) -> updateConn -- to allow retries
|
||||
(SomeConn c _) -> pure . Left . SEBadConnType $ connType c
|
||||
where
|
||||
updateConn :: IO (Either StoreError Int64)
|
||||
updateConn :: IO (Either StoreError RcvQueue)
|
||||
updateConn = Right <$> addConnRcvQueue_ db connId rq
|
||||
|
||||
updateNewConnSnd :: DB.Connection -> ConnId -> SndQueue -> IO (Either StoreError Int64)
|
||||
updateNewConnSnd :: DB.Connection -> ConnId -> NewSndQueue -> IO (Either StoreError SndQueue)
|
||||
updateNewConnSnd db connId sq =
|
||||
getConn db connId $>>= \case
|
||||
(SomeConn _ NewConnection {}) -> updateConn
|
||||
(SomeConn _ SndConnection {}) -> updateConn -- to allow retries
|
||||
(SomeConn c _) -> pure . Left . SEBadConnType $ connType c
|
||||
where
|
||||
updateConn :: IO (Either StoreError Int64)
|
||||
updateConn :: IO (Either StoreError SndQueue)
|
||||
updateConn = Right <$> addConnSndQueue_ db connId sq
|
||||
|
||||
createRcvConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> IO (Either StoreError ConnId)
|
||||
createRcvConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> NewRcvQueue -> SConnectionMode c -> IO (Either StoreError (ConnId, RcvQueue))
|
||||
createRcvConn db gVar cData@ConnData {userId, connAgentVersion, enableNtfs, duplexHandshake} q@RcvQueue {server} cMode =
|
||||
createConn_ gVar cData $ \connId -> do
|
||||
serverKeyHash_ <- createServer_ db server
|
||||
DB.execute db "INSERT INTO connections (user_id, conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?,?,?,?,?,?)" (userId, connId, cMode, connAgentVersion, enableNtfs, duplexHandshake)
|
||||
void $ insertRcvQueue_ db connId q serverKeyHash_
|
||||
insertRcvQueue_ db connId q serverKeyHash_
|
||||
|
||||
createSndConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> SndQueue -> IO (Either StoreError ConnId)
|
||||
createSndConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> NewSndQueue -> IO (Either StoreError (ConnId, SndQueue))
|
||||
createSndConn db gVar cData@ConnData {userId, connAgentVersion, enableNtfs, duplexHandshake} q@SndQueue {server} =
|
||||
-- check confirmed snd queue doesn't already exist, to prevent it being deleted by REPLACE in insertSndQueue_
|
||||
ifM (liftIO $ checkConfirmedSndQueueExists_ db q) (pure $ Left SESndQueueExists) $
|
||||
createConn_ gVar cData $ \connId -> do
|
||||
serverKeyHash_ <- createServer_ db server
|
||||
DB.execute db "INSERT INTO connections (user_id, conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?,?,?,?,?,?)" (userId, connId, SCMInvitation, connAgentVersion, enableNtfs, duplexHandshake)
|
||||
void $ insertSndQueue_ db connId q serverKeyHash_
|
||||
insertSndQueue_ db connId q serverKeyHash_
|
||||
|
||||
checkConfirmedSndQueueExists_ :: DB.Connection -> SndQueue -> IO Bool
|
||||
checkConfirmedSndQueueExists_ :: DB.Connection -> NewSndQueue -> IO Bool
|
||||
checkConfirmedSndQueueExists_ db SndQueue {server, sndId} = do
|
||||
fromMaybe False
|
||||
<$> maybeFirstRow
|
||||
@@ -602,39 +603,39 @@ deleteConn db connId =
|
||||
"DELETE FROM connections WHERE conn_id = :conn_id;"
|
||||
[":conn_id" := connId]
|
||||
|
||||
upgradeRcvConnToDuplex :: DB.Connection -> ConnId -> SndQueue -> IO (Either StoreError Int64)
|
||||
upgradeRcvConnToDuplex :: DB.Connection -> ConnId -> NewSndQueue -> IO (Either StoreError SndQueue)
|
||||
upgradeRcvConnToDuplex db connId sq =
|
||||
getConn db connId $>>= \case
|
||||
(SomeConn _ RcvConnection {}) -> Right <$> addConnSndQueue_ db connId sq
|
||||
(SomeConn c _) -> pure . Left . SEBadConnType $ connType c
|
||||
|
||||
upgradeSndConnToDuplex :: DB.Connection -> ConnId -> RcvQueue -> IO (Either StoreError Int64)
|
||||
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
|
||||
_ -> pure $ Left SEConnNotFound
|
||||
|
||||
addConnRcvQueue :: DB.Connection -> ConnId -> RcvQueue -> IO (Either StoreError Int64)
|
||||
addConnRcvQueue :: DB.Connection -> ConnId -> NewRcvQueue -> IO (Either StoreError RcvQueue)
|
||||
addConnRcvQueue db connId rq =
|
||||
getConn db connId >>= \case
|
||||
Right (SomeConn _ DuplexConnection {}) -> Right <$> addConnRcvQueue_ db connId rq
|
||||
Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
|
||||
_ -> pure $ Left SEConnNotFound
|
||||
|
||||
addConnRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO Int64
|
||||
addConnRcvQueue_ :: DB.Connection -> ConnId -> NewRcvQueue -> IO RcvQueue
|
||||
addConnRcvQueue_ db connId rq@RcvQueue {server} = do
|
||||
serverKeyHash_ <- createServer_ db server
|
||||
insertRcvQueue_ db connId rq serverKeyHash_
|
||||
|
||||
addConnSndQueue :: DB.Connection -> ConnId -> SndQueue -> IO (Either StoreError Int64)
|
||||
addConnSndQueue :: DB.Connection -> ConnId -> NewSndQueue -> IO (Either StoreError SndQueue)
|
||||
addConnSndQueue db connId sq =
|
||||
getConn db connId >>= \case
|
||||
Right (SomeConn _ DuplexConnection {}) -> Right <$> addConnSndQueue_ db connId sq
|
||||
Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
|
||||
_ -> pure $ Left SEConnNotFound
|
||||
|
||||
addConnSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO Int64
|
||||
addConnSndQueue_ :: DB.Connection -> ConnId -> NewSndQueue -> IO SndQueue
|
||||
addConnSndQueue_ db connId sq@SndQueue {server} = do
|
||||
serverKeyHash_ <- createServer_ db server
|
||||
insertSndQueue_ db connId sq serverKeyHash_
|
||||
@@ -974,23 +975,26 @@ updateSndMsgRcpt db connId sndMsgId MsgReceipt {agentMsgId, msgRcptStatus} =
|
||||
"UPDATE snd_messages SET rcpt_internal_id = ?, rcpt_status = ? WHERE conn_id = ? AND internal_snd_id = ?"
|
||||
(agentMsgId, msgRcptStatus, connId, sndMsgId)
|
||||
|
||||
getPendingMsgData :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (Maybe RcvQueue, PendingMsgData))
|
||||
getPendingMsgData db connId msgId = do
|
||||
getPendingQueueMsg :: DB.Connection -> ConnId -> SndQueue -> IO (Maybe (Maybe RcvQueue, PendingMsgData))
|
||||
getPendingQueueMsg db connId SndQueue {dbQueueId} = do
|
||||
rq_ <- L.head <$$> getRcvQueuesByConnId_ db connId
|
||||
(rq_,) <$$> firstRow pendingMsgData SEMsgNotFound getMsgData_
|
||||
(rq_,) <$$> maybeFirstRow pendingMsgData getMsgData_
|
||||
where
|
||||
getMsgData_ =
|
||||
DB.query
|
||||
db
|
||||
[sql|
|
||||
SELECT m.msg_type, m.msg_flags, m.msg_body, m.internal_ts, s.retry_int_slow, s.retry_int_fast
|
||||
FROM messages m
|
||||
JOIN snd_messages s ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id
|
||||
WHERE m.conn_id = ? AND m.internal_id = ?
|
||||
SELECT m.internal_id, m.msg_type, m.msg_flags, m.msg_body, m.internal_ts, s.retry_int_slow, s.retry_int_fast
|
||||
FROM snd_message_deliveries d
|
||||
JOIN messages m ON m.conn_id = d.conn_id AND m.internal_id = d.internal_id
|
||||
JOIN snd_messages s ON s.conn_id = d.conn_id AND s.internal_id = d.internal_id
|
||||
WHERE d.conn_id = ? AND d.snd_queue_id = ?
|
||||
ORDER BY d.internal_id ASC
|
||||
LIMIT 1
|
||||
|]
|
||||
(connId, msgId)
|
||||
pendingMsgData :: (AgentMessageType, Maybe MsgFlags, MsgBody, InternalTs, Maybe Int64, Maybe Int64) -> PendingMsgData
|
||||
pendingMsgData (msgType, msgFlags_, msgBody, internalTs, riSlow_, riFast_) =
|
||||
(connId, dbQueueId)
|
||||
pendingMsgData :: (InternalId, AgentMessageType, Maybe MsgFlags, MsgBody, InternalTs, Maybe Int64, Maybe Int64) -> PendingMsgData
|
||||
pendingMsgData (msgId, msgType, msgFlags_, msgBody, internalTs, riSlow_, riFast_) =
|
||||
let msgFlags = fromMaybe SMP.noMsgFlags msgFlags_
|
||||
msgRetryState = RI2State <$> riSlow_ <*> riFast_
|
||||
in PendingMsgData {msgId, msgType, msgFlags, msgBody, msgRetryState, internalTs}
|
||||
@@ -999,11 +1003,6 @@ updatePendingMsgRIState :: DB.Connection -> ConnId -> InternalId -> RI2State ->
|
||||
updatePendingMsgRIState db connId msgId RI2State {slowInterval, fastInterval} =
|
||||
DB.execute db "UPDATE snd_messages SET retry_int_slow = ?, retry_int_fast = ? WHERE conn_id = ? AND internal_id = ?" (slowInterval, fastInterval, connId, msgId)
|
||||
|
||||
getPendingMsgs :: DB.Connection -> ConnId -> SndQueue -> IO [InternalId]
|
||||
getPendingMsgs db connId SndQueue {dbQueueId} =
|
||||
map fromOnly
|
||||
<$> DB.query db "SELECT internal_id FROM snd_message_deliveries WHERE conn_id = ? AND snd_queue_id = ? ORDER BY internal_id ASC" (connId, dbQueueId)
|
||||
|
||||
deletePendingMsgs :: DB.Connection -> ConnId -> SndQueue -> IO ()
|
||||
deletePendingMsgs db connId SndQueue {dbQueueId} =
|
||||
DB.execute db "DELETE FROM snd_message_deliveries WHERE conn_id = ? AND snd_queue_id = ?" (connId, dbQueueId)
|
||||
@@ -1202,7 +1201,7 @@ updateRatchet db connId rc skipped = do
|
||||
forM_ (M.assocs mks) $ \(msgN, mk) ->
|
||||
DB.execute db "INSERT INTO skipped_messages (conn_id, header_key, msg_n, msg_key) VALUES (?, ?, ?, ?)" (connId, hk, msgN, mk)
|
||||
|
||||
createCommand :: DB.Connection -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> IO (Either StoreError AsyncCmdId)
|
||||
createCommand :: DB.Connection -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> IO (Either StoreError ())
|
||||
createCommand db corrId connId srv_ cmd = runExceptT $ do
|
||||
(host_, port_, serverKeyHash_) <- serverFields
|
||||
liftIO $ do
|
||||
@@ -1210,7 +1209,6 @@ createCommand db corrId connId srv_ cmd = runExceptT $ do
|
||||
db
|
||||
"INSERT INTO commands (host, port, corr_id, conn_id, command_tag, command, server_key_hash) VALUES (?,?,?,?,?,?,?)"
|
||||
(host_, port_, corrId, connId, agentCommandTag cmd, cmd, serverKeyHash_)
|
||||
insertedRowId db
|
||||
where
|
||||
serverFields :: ExceptT StoreError IO (Maybe (NonEmpty TransportHost), Maybe ServiceName, Maybe C.KeyHash)
|
||||
serverFields = case srv_ of
|
||||
@@ -1221,38 +1219,49 @@ createCommand db corrId connId srv_ cmd = runExceptT $ do
|
||||
insertedRowId :: DB.Connection -> IO Int64
|
||||
insertedRowId db = fromOnly . head <$> DB.query_ db "SELECT last_insert_rowid()"
|
||||
|
||||
getPendingCommands :: DB.Connection -> ConnId -> IO [(Maybe SMPServer, [AsyncCmdId])]
|
||||
getPendingCommands db connId = do
|
||||
-- `groupOn` is used instead of `groupAllOn` to avoid extra sorting by `server + cmdId`, as the query already sorts by them.
|
||||
getPendingCommandServers :: DB.Connection -> ConnId -> IO [Maybe SMPServer]
|
||||
getPendingCommandServers db connId = do
|
||||
-- TODO review whether this can break if, e.g., the server has another key hash.
|
||||
map (\ids -> (fst $ head ids, map snd ids)) . groupOn fst . map srvCmdId
|
||||
map smpServer
|
||||
<$> DB.query
|
||||
db
|
||||
[sql|
|
||||
SELECT c.host, c.port, COALESCE(c.server_key_hash, s.key_hash), c.command_id
|
||||
SELECT DISTINCT c.host, c.port, COALESCE(c.server_key_hash, s.key_hash)
|
||||
FROM commands c
|
||||
LEFT JOIN servers s ON s.host = c.host AND s.port = c.port
|
||||
WHERE conn_id = ?
|
||||
ORDER BY c.host, c.port, c.command_id ASC
|
||||
|]
|
||||
(Only connId)
|
||||
where
|
||||
srvCmdId (host, port, keyHash, cmdId) = (SMPServer <$> host <*> port <*> keyHash, cmdId)
|
||||
smpServer (host, port, keyHash) = SMPServer <$> host <*> port <*> keyHash
|
||||
|
||||
getPendingCommand :: DB.Connection -> AsyncCmdId -> IO (Either StoreError PendingCommand)
|
||||
getPendingCommand db msgId = do
|
||||
firstRow pendingCommand SECmdNotFound $
|
||||
getPendingServerCommand :: DB.Connection -> Maybe SMPServer -> IO (Maybe PendingCommand)
|
||||
getPendingServerCommand db srv_ = maybeFirstRow pendingCommand $ case srv_ of
|
||||
Nothing ->
|
||||
DB.query_
|
||||
db
|
||||
[sql|
|
||||
SELECT c.command_id, c.corr_id, cs.user_id, c.conn_id, c.command
|
||||
FROM commands c
|
||||
JOIN connections cs USING (conn_id)
|
||||
WHERE c.host IS NULL AND c.port IS NULL
|
||||
ORDER BY c.created_at ASC, c.command_id ASC
|
||||
LIMIT 1
|
||||
|]
|
||||
Just (SMPServer host port _) ->
|
||||
DB.query
|
||||
db
|
||||
[sql|
|
||||
SELECT c.corr_id, cs.user_id, c.conn_id, c.command
|
||||
SELECT c.command_id, c.corr_id, cs.user_id, c.conn_id, c.command
|
||||
FROM commands c
|
||||
JOIN connections cs USING (conn_id)
|
||||
WHERE c.command_id = ?
|
||||
WHERE c.host = ? AND c.port = ?
|
||||
ORDER BY c.created_at ASC, c.command_id ASC
|
||||
LIMIT 1
|
||||
|]
|
||||
(Only msgId)
|
||||
(host, port)
|
||||
where
|
||||
pendingCommand (corrId, userId, connId, command) = PendingCommand {corrId, userId, connId, command}
|
||||
pendingCommand (cmdId, corrId, userId, connId, command) = PendingCommand {cmdId, corrId, userId, connId, command}
|
||||
|
||||
deleteCommand :: DB.Connection -> AsyncCmdId -> IO ()
|
||||
deleteCommand db cmdId =
|
||||
@@ -1571,6 +1580,10 @@ instance ToField QueueStatus where toField = toField . serializeQueueStatus
|
||||
|
||||
instance FromField QueueStatus where fromField = fromTextField_ queueStatusT
|
||||
|
||||
instance ToField (DBQueueId 'QSStored) where toField (DBQueueId qId) = toField qId
|
||||
|
||||
instance FromField (DBQueueId 'QSStored) where fromField x = DBQueueId <$> fromField x
|
||||
|
||||
instance ToField InternalRcvId where toField (InternalRcvId x) = toField x
|
||||
|
||||
instance FromField InternalRcvId where fromField x = InternalRcvId <$> fromField x
|
||||
@@ -1711,8 +1724,8 @@ upsertNtfServer_ db ProtocolServer {host, port, keyHash} = do
|
||||
|
||||
-- * createRcvConn helpers
|
||||
|
||||
insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> Maybe C.KeyHash -> IO Int64
|
||||
insertRcvQueue_ db connId' RcvQueue {..} serverKeyHash_ = do
|
||||
insertRcvQueue_ :: DB.Connection -> ConnId -> NewRcvQueue -> Maybe C.KeyHash -> IO RcvQueue
|
||||
insertRcvQueue_ db connId' rq@RcvQueue {..} serverKeyHash_ = do
|
||||
qId <- newQueueId_ <$> DB.query db "SELECT rcv_queue_id FROM rcv_queues WHERE conn_id = ? ORDER BY rcv_queue_id DESC LIMIT 1" (Only connId')
|
||||
DB.execute
|
||||
db
|
||||
@@ -1721,12 +1734,12 @@ insertRcvQueue_ db connId' RcvQueue {..} serverKeyHash_ = do
|
||||
(host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status, rcv_queue_id, rcv_primary, replace_rcv_queue_id, smp_client_version, server_key_hash) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?);
|
||||
|]
|
||||
((host server, port server, rcvId, connId', rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret) :. (sndId, status, qId, primary, dbReplaceQueueId, smpClientVersion, serverKeyHash_))
|
||||
pure qId
|
||||
pure (rq :: NewRcvQueue) {connId = connId', dbQueueId = qId}
|
||||
|
||||
-- * createSndConn helpers
|
||||
|
||||
insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> Maybe C.KeyHash -> IO Int64
|
||||
insertSndQueue_ db connId' SndQueue {..} serverKeyHash_ = do
|
||||
insertSndQueue_ :: DB.Connection -> ConnId -> NewSndQueue -> Maybe C.KeyHash -> IO SndQueue
|
||||
insertSndQueue_ db connId' sq@SndQueue {..} serverKeyHash_ = do
|
||||
qId <- newQueueId_ <$> DB.query db "SELECT snd_queue_id FROM snd_queues WHERE conn_id = ? ORDER BY snd_queue_id DESC LIMIT 1" (Only connId')
|
||||
DB.execute
|
||||
db
|
||||
@@ -1735,11 +1748,11 @@ insertSndQueue_ db connId' SndQueue {..} serverKeyHash_ = do
|
||||
(host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status, snd_queue_id, snd_primary, replace_snd_queue_id, smp_client_version, server_key_hash) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?);
|
||||
|]
|
||||
((host server, port server, sndId, connId', sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret) :. (status, qId, primary, dbReplaceQueueId, smpClientVersion, serverKeyHash_))
|
||||
pure qId
|
||||
pure (sq :: NewSndQueue) {connId = connId', dbQueueId = qId}
|
||||
|
||||
newQueueId_ :: [Only Int64] -> Int64
|
||||
newQueueId_ [] = 1
|
||||
newQueueId_ (Only maxId : _) = maxId + 1
|
||||
newQueueId_ :: [Only Int64] -> DBQueueId 'QSStored
|
||||
newQueueId_ [] = DBQueueId 1
|
||||
newQueueId_ (Only maxId : _) = DBQueueId (maxId + 1)
|
||||
|
||||
-- * getConn helpers
|
||||
|
||||
@@ -1853,7 +1866,7 @@ rcvQueueQuery =
|
||||
|
||||
toRcvQueue ::
|
||||
(UserId, C.KeyHash, ConnId, NonEmpty TransportHost, ServiceName, SMP.RecipientId, SMP.RcvPrivateSignKey, SMP.RcvDhSecret, C.PrivateKeyX25519, Maybe C.DhSecretX25519, SMP.SenderId, QueueStatus)
|
||||
:. (Int64, Bool, Maybe Int64, Maybe RcvSwitchStatus, Maybe Version, Int)
|
||||
:. (DBQueueId 'QSStored, Bool, Maybe Int64, Maybe RcvSwitchStatus, Maybe Version, Int)
|
||||
:. (Maybe SMP.NtfPublicVerifyKey, Maybe SMP.NtfPrivateSignKey, Maybe SMP.NotifierId, Maybe RcvNtfDhSecret) ->
|
||||
RcvQueue
|
||||
toRcvQueue ((userId, keyHash, connId, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status) :. (dbQueueId, primary, dbReplaceQueueId, rcvSwchStatus, smpClientVersion_, deleteErrors) :. (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_)) =
|
||||
@@ -1894,7 +1907,7 @@ sndQueueQuery =
|
||||
toSndQueue ::
|
||||
(UserId, C.KeyHash, ConnId, NonEmpty TransportHost, ServiceName, SenderId)
|
||||
:. (Maybe C.APublicVerifyKey, SndPrivateSignKey, Maybe C.PublicKeyX25519, C.DhSecretX25519, QueueStatus)
|
||||
:. (Int64, Bool, Maybe Int64, Maybe SndSwitchStatus, Version) ->
|
||||
:. (DBQueueId 'QSStored, Bool, Maybe Int64, Maybe SndSwitchStatus, Version) ->
|
||||
SndQueue
|
||||
toSndQueue
|
||||
( (userId, keyHash, connId, host, port, sndId)
|
||||
@@ -2093,14 +2106,17 @@ updateHashSnd_ dbConn connId SndMsgData {..} =
|
||||
|
||||
-- create record with a random ID
|
||||
createWithRandomId :: TVar ChaChaDRG -> (ByteString -> IO ()) -> IO (Either StoreError ByteString)
|
||||
createWithRandomId gVar create = tryCreate 3
|
||||
createWithRandomId gVar create = fst <$$> createWithRandomId' gVar create
|
||||
|
||||
createWithRandomId' :: forall a. TVar ChaChaDRG -> (ByteString -> IO a) -> IO (Either StoreError (ByteString, a))
|
||||
createWithRandomId' gVar create = tryCreate 3
|
||||
where
|
||||
tryCreate :: Int -> IO (Either StoreError ByteString)
|
||||
tryCreate :: Int -> IO (Either StoreError (ByteString, a))
|
||||
tryCreate 0 = pure $ Left SEUniqueID
|
||||
tryCreate n = do
|
||||
id' <- randomId gVar 12
|
||||
E.try (create id') >>= \case
|
||||
Right _ -> pure $ Right id'
|
||||
Right r -> pure $ Right (id', r)
|
||||
Left e
|
||||
| SQL.sqlError e == SQL.ErrorConstraint -> tryCreate (n - 1)
|
||||
| otherwise -> pure . Left . SEInternal $ bshow e
|
||||
|
||||
@@ -65,6 +65,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230720_delete_expired_
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230722_indexes
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230814_indexes
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230829_crypto_files
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231222_command_created_at
|
||||
import Simplex.Messaging.Encoding.String
|
||||
import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON)
|
||||
import Simplex.Messaging.Transport.Client (TransportHost)
|
||||
@@ -98,7 +99,8 @@ schemaMigrations =
|
||||
("m20230720_delete_expired_messages", m20230720_delete_expired_messages, Just down_m20230720_delete_expired_messages),
|
||||
("m20230722_indexes", m20230722_indexes, Just down_m20230722_indexes),
|
||||
("m20230814_indexes", m20230814_indexes, Just down_m20230814_indexes),
|
||||
("m20230829_crypto_files", m20230829_crypto_files, Just down_m20230829_crypto_files)
|
||||
("m20230829_crypto_files", m20230829_crypto_files, Just down_m20230829_crypto_files),
|
||||
("m20231222_command_created_at", m20231222_command_created_at, Just down_m20231222_command_created_at)
|
||||
]
|
||||
|
||||
-- | The list of migrations in ascending order by date
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
{-# LANGUAGE QuasiQuotes #-}
|
||||
|
||||
module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231222_command_created_at where
|
||||
|
||||
import Database.SQLite.Simple (Query)
|
||||
import Database.SQLite.Simple.QQ (sql)
|
||||
|
||||
m20231222_command_created_at :: Query
|
||||
m20231222_command_created_at =
|
||||
[sql|
|
||||
ALTER TABLE commands ADD COLUMN created_at TEXT NOT NULL DEFAULT(datetime('now'));
|
||||
CREATE INDEX idx_commands_server_commands ON commands(host, port, created_at, command_id);
|
||||
|]
|
||||
|
||||
down_m20231222_command_created_at :: Query
|
||||
down_m20231222_command_created_at =
|
||||
[sql|
|
||||
DROP INDEX idx_commands_server_commands;
|
||||
ALTER TABLE commands DROP COLUMN created_at;
|
||||
|]
|
||||
@@ -229,6 +229,7 @@ CREATE TABLE commands(
|
||||
command BLOB NOT NULL,
|
||||
agent_version INTEGER NOT NULL DEFAULT 1,
|
||||
server_key_hash BLOB,
|
||||
created_at TEXT NOT NULL DEFAULT(datetime('now')),
|
||||
FOREIGN KEY(host, port) REFERENCES servers
|
||||
ON DELETE RESTRICT ON UPDATE CASCADE
|
||||
);
|
||||
@@ -479,3 +480,9 @@ CREATE INDEX idx_encrypted_rcv_message_hashes_created_at ON encrypted_rcv_messag
|
||||
created_at
|
||||
);
|
||||
CREATE INDEX idx_messages_internal_ts ON messages(internal_ts);
|
||||
CREATE INDEX idx_commands_server_commands ON commands(
|
||||
host,
|
||||
port,
|
||||
created_at,
|
||||
command_id
|
||||
);
|
||||
@@ -18,7 +18,7 @@ import qualified Data.Map.Strict as M
|
||||
import Data.Set (Set)
|
||||
import qualified Data.Set as S
|
||||
import Simplex.Messaging.Agent.Protocol (ConnId, UserId)
|
||||
import Simplex.Messaging.Agent.Store (RcvQueue (..))
|
||||
import Simplex.Messaging.Agent.Store (RcvQueue, StoredRcvQueue (..))
|
||||
import Simplex.Messaging.Protocol (RecipientId, SMPServer)
|
||||
import Simplex.Messaging.TMap (TMap)
|
||||
import qualified Simplex.Messaging.TMap as TM
|
||||
|
||||
@@ -115,31 +115,41 @@ pattern Rcvd :: AgentMsgId -> ACommand 'Agent e
|
||||
pattern Rcvd agentMsgId <- RCVD MsgMeta {integrity = MsgOk} [MsgReceipt {agentMsgId, msgRcptStatus = MROk}]
|
||||
|
||||
smpCfgVPrev :: ProtocolClientConfig
|
||||
smpCfgVPrev = (smpCfg agentCfg) {serverVRange = serverVRangePrev}
|
||||
where
|
||||
serverVRangePrev = prevRange $ serverVRange $ smpCfg agentCfg
|
||||
smpCfgVPrev = (smpCfg agentCfg) {serverVRange = prevRange $ serverVRange $ smpCfg agentCfg}
|
||||
|
||||
smpCfgV1 :: ProtocolClientConfig
|
||||
smpCfgV1 = (smpCfg agentCfg) {serverVRange = v1Range}
|
||||
|
||||
agentCfgVPrev :: AgentConfig
|
||||
agentCfgVPrev =
|
||||
agentCfg
|
||||
{ smpAgentVRange = smpAgentVRangePrev,
|
||||
smpClientVRange = smpClientVRangePrev,
|
||||
e2eEncryptVRange = e2eEncryptVRangePrev,
|
||||
{ smpAgentVRange = prevRange $ smpAgentVRange agentCfg,
|
||||
smpClientVRange = prevRange $ smpClientVRange agentCfg,
|
||||
e2eEncryptVRange = prevRange $ e2eEncryptVRange agentCfg,
|
||||
smpCfg = smpCfgVPrev
|
||||
}
|
||||
where
|
||||
smpAgentVRangePrev = prevRange $ smpAgentVRange agentCfg
|
||||
smpClientVRangePrev = prevRange $ smpClientVRange agentCfg
|
||||
e2eEncryptVRangePrev = prevRange $ e2eEncryptVRange agentCfg
|
||||
|
||||
agentCfgV1 :: AgentConfig
|
||||
agentCfgV1 =
|
||||
agentCfg
|
||||
{ smpAgentVRange = v1Range,
|
||||
smpClientVRange = v1Range,
|
||||
e2eEncryptVRange = v1Range,
|
||||
smpCfg = smpCfgV1
|
||||
}
|
||||
|
||||
agentCfgRatchetVPrev :: AgentConfig
|
||||
agentCfgRatchetVPrev = agentCfg {e2eEncryptVRange = e2eEncryptVRangePrev}
|
||||
where
|
||||
e2eEncryptVRangePrev = prevRange $ e2eEncryptVRange agentCfg
|
||||
agentCfgRatchetVPrev = agentCfg {e2eEncryptVRange = prevRange $ e2eEncryptVRange agentCfg}
|
||||
|
||||
agentCfgRatchetV1 :: AgentConfig
|
||||
agentCfgRatchetV1 = agentCfg {e2eEncryptVRange = v1Range}
|
||||
|
||||
prevRange :: VersionRange -> VersionRange
|
||||
prevRange vr = vr {maxVersion = maxVersion vr - 1}
|
||||
|
||||
v1Range :: VersionRange
|
||||
v1Range = mkVersionRange 1 1
|
||||
|
||||
runRight_ :: (Eq e, Show e, HasCallStack) => ExceptT e IO () -> Expectation
|
||||
runRight_ action = runExceptT action `shouldReturn` Right ()
|
||||
|
||||
@@ -338,6 +348,9 @@ testMatrix2 t runTest = do
|
||||
it "prev" $ withSmpServer t $ runTestCfg2 agentCfgVPrev agentCfgVPrev 3 runTest
|
||||
it "prev to current" $ withSmpServer t $ runTestCfg2 agentCfgVPrev agentCfg 3 runTest
|
||||
it "current to prev" $ withSmpServer t $ runTestCfg2 agentCfg agentCfgVPrev 3 runTest
|
||||
it "v1" $ withSmpServer t $ runTestCfg2 agentCfgV1 agentCfgV1 4 runTest
|
||||
it "v1 to current" $ withSmpServer t $ runTestCfg2 agentCfgV1 agentCfg 4 runTest
|
||||
it "current to v1" $ withSmpServer t $ runTestCfg2 agentCfg agentCfgV1 4 runTest
|
||||
|
||||
testRatchetMatrix2 :: ATransport -> (AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
|
||||
testRatchetMatrix2 t runTest = do
|
||||
@@ -345,6 +358,9 @@ testRatchetMatrix2 t runTest = do
|
||||
it "ratchet prev" $ withSmpServer t $ runTestCfg2 agentCfgRatchetVPrev agentCfgRatchetVPrev 3 runTest
|
||||
it "ratchets prev to current" $ withSmpServer t $ runTestCfg2 agentCfgRatchetVPrev agentCfg 3 runTest
|
||||
it "ratchets current to prev" $ withSmpServer t $ runTestCfg2 agentCfg agentCfgRatchetVPrev 3 runTest
|
||||
it "ratchet v1" $ withSmpServer t $ runTestCfg2 agentCfgRatchetV1 agentCfgRatchetV1 3 runTest
|
||||
it "ratchets v1 to current" $ withSmpServer t $ runTestCfg2 agentCfgRatchetV1 agentCfg 3 runTest
|
||||
it "ratchets current to v1" $ withSmpServer t $ runTestCfg2 agentCfg agentCfgRatchetV1 3 runTest
|
||||
|
||||
testServerMatrix2 :: ATransport -> (InitialAgentServers -> IO ()) -> Spec
|
||||
testServerMatrix2 t runTest = do
|
||||
@@ -521,9 +537,6 @@ testAsyncServerOffline t = withAgentClients2 $ \alice bob -> do
|
||||
testAsyncHelloTimeout :: HasCallStack => IO ()
|
||||
testAsyncHelloTimeout = do
|
||||
-- this test would only work if any of the agent is v1, there is no HELLO timeout in v2
|
||||
let vr11 = mkVersionRange 1 1
|
||||
smpCfgV1 = (smpCfg agentCfg) {serverVRange = vr11}
|
||||
agentCfgV1 = agentCfg {smpAgentVRange = vr11, smpClientVRange = vr11, e2eEncryptVRange = vr11, smpCfg = smpCfgV1}
|
||||
withAgentClientsCfg2 agentCfgV1 agentCfg {helloTimeout = 1} $ \alice bob -> runRight_ $ do
|
||||
(_, cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
|
||||
disconnectAgentClient alice
|
||||
@@ -956,7 +969,7 @@ testRatchetSyncSuspendForeground t = do
|
||||
|
||||
suspendAgent bob2 0
|
||||
threadDelay 100000
|
||||
foregroundAgent bob2 False
|
||||
foregroundAgent bob2
|
||||
|
||||
withSmpServerStoreMsgLogOn t testPort $ \_ -> do
|
||||
runRight_ $ do
|
||||
@@ -1119,7 +1132,7 @@ testSuspendingAgent =
|
||||
5 <- sendMessage a bId SMP.noMsgFlags "hello 2"
|
||||
get a ##> ("", bId, SENT 5)
|
||||
Nothing <- 100000 `timeout` get b
|
||||
foregroundAgent b False
|
||||
foregroundAgent b
|
||||
get b =##> \case ("", c, Msg "hello 2") -> c == aId; _ -> False
|
||||
|
||||
testSuspendingAgentCompleteSending :: ATransport -> IO ()
|
||||
|
||||
@@ -118,16 +118,16 @@ testConcurrentWrites :: SpecWith (SQLiteStore, SQLiteStore)
|
||||
testConcurrentWrites =
|
||||
it "should complete multiple concurrent write transactions w/t sqlite busy errors" $ \(s1, s2) -> do
|
||||
g <- C.newRandom
|
||||
_ <- withTransaction s1 $ \db ->
|
||||
Right (_, rq) <- withTransaction s1 $ \db ->
|
||||
createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
let ConnData {connId} = cData1
|
||||
concurrently_ (runTest s1 connId) (runTest s2 connId)
|
||||
concurrently_ (runTest s1 connId rq) (runTest s2 connId rq)
|
||||
where
|
||||
runTest :: SQLiteStore -> ConnId -> IO ()
|
||||
runTest st connId = replicateM_ 100 . withTransaction st $ \db -> do
|
||||
runTest :: SQLiteStore -> ConnId -> RcvQueue -> IO ()
|
||||
runTest st connId rq = replicateM_ 100 . withTransaction st $ \db -> do
|
||||
(internalId, internalRcvId, _, _) <- updateRcvIds db connId
|
||||
let rcvMsgData = mkRcvMsgData internalId internalRcvId 0 "0" "hash_dummy"
|
||||
createRcvMsg db connId rcvQueue1 rcvMsgData
|
||||
createRcvMsg db connId rq rcvMsgData
|
||||
|
||||
testCompiledThreadsafe :: SpecWith SQLiteStore
|
||||
testCompiledThreadsafe =
|
||||
@@ -163,7 +163,7 @@ testPrivDhKey = "MC4CAQAwBQYDK2VuBCIEINCzbVFaCiYHoYncxNY8tSIfn0pXcIAhLBfFc0m+gOp
|
||||
testDhSecret :: C.DhSecretX25519
|
||||
testDhSecret = "01234567890123456789012345678901"
|
||||
|
||||
rcvQueue1 :: RcvQueue
|
||||
rcvQueue1 :: NewRcvQueue
|
||||
rcvQueue1 =
|
||||
RcvQueue
|
||||
{ userId = 1,
|
||||
@@ -176,7 +176,7 @@ rcvQueue1 =
|
||||
e2eDhSecret = Nothing,
|
||||
sndId = "2345",
|
||||
status = New,
|
||||
dbQueueId = 1,
|
||||
dbQueueId = DBNewQueue,
|
||||
primary = True,
|
||||
dbReplaceQueueId = Nothing,
|
||||
rcvSwchStatus = Nothing,
|
||||
@@ -185,7 +185,7 @@ rcvQueue1 =
|
||||
deleteErrors = 0
|
||||
}
|
||||
|
||||
sndQueue1 :: SndQueue
|
||||
sndQueue1 :: NewSndQueue
|
||||
sndQueue1 =
|
||||
SndQueue
|
||||
{ userId = 1,
|
||||
@@ -197,7 +197,7 @@ sndQueue1 =
|
||||
e2ePubKey = Nothing,
|
||||
e2eDhSecret = testDhSecret,
|
||||
status = New,
|
||||
dbQueueId = 1,
|
||||
dbQueueId = DBNewQueue,
|
||||
primary = True,
|
||||
dbReplaceQueueId = Nothing,
|
||||
sndSwchStatus = Nothing,
|
||||
@@ -208,28 +208,27 @@ testCreateRcvConn :: SpecWith SQLiteStore
|
||||
testCreateRcvConn =
|
||||
it "should create RcvConnection and add SndQueue" . withStoreTransaction $ \db -> do
|
||||
g <- C.newRandom
|
||||
createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
`shouldReturn` Right "conn1"
|
||||
Right (connId, rq@RcvQueue {dbQueueId}) <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
connId `shouldBe` "conn1"
|
||||
dbQueueId `shouldBe` DBQueueId 1
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rcvQueue1))
|
||||
upgradeRcvConnToDuplex db "conn1" sndQueue1
|
||||
`shouldReturn` Right 1
|
||||
`shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rq))
|
||||
Right sq@SndQueue {dbQueueId = dbQueueId'} <- upgradeRcvConnToDuplex db "conn1" sndQueue1
|
||||
dbQueueId' `shouldBe` DBQueueId 1
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rcvQueue1] [sndQueue1]))
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rq] [sq]))
|
||||
|
||||
testCreateRcvConnRandomId :: SpecWith SQLiteStore
|
||||
testCreateRcvConnRandomId =
|
||||
it "should create RcvConnection and add SndQueue with random ID" . withStoreTransaction $ \db -> do
|
||||
g <- C.newRandom
|
||||
Right connId <- createRcvConn db g cData1 {connId = ""} rcvQueue1 SCMInvitation
|
||||
let rq' = (rcvQueue1 :: RcvQueue) {connId}
|
||||
sq' = (sndQueue1 :: SndQueue) {connId}
|
||||
Right (connId, rq) <- createRcvConn db g cData1 {connId = ""} rcvQueue1 SCMInvitation
|
||||
getConn db connId
|
||||
`shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 {connId} rq'))
|
||||
upgradeRcvConnToDuplex db connId sndQueue1
|
||||
`shouldReturn` Right 1
|
||||
`shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 {connId} rq))
|
||||
Right sq@SndQueue {dbQueueId = dbQueueId'} <- upgradeRcvConnToDuplex db connId sndQueue1
|
||||
dbQueueId' `shouldBe` DBQueueId 1
|
||||
getConn db connId
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} [rq'] [sq']))
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} [rq] [sq]))
|
||||
|
||||
testCreateRcvConnDuplicate :: SpecWith SQLiteStore
|
||||
testCreateRcvConnDuplicate =
|
||||
@@ -243,28 +242,27 @@ testCreateSndConn :: SpecWith SQLiteStore
|
||||
testCreateSndConn =
|
||||
it "should create SndConnection and add RcvQueue" . withStoreTransaction $ \db -> do
|
||||
g <- C.newRandom
|
||||
createSndConn db g cData1 sndQueue1
|
||||
`shouldReturn` Right "conn1"
|
||||
Right (connId, sq@SndQueue {dbQueueId}) <- createSndConn db g cData1 sndQueue1
|
||||
connId `shouldBe` "conn1"
|
||||
dbQueueId `shouldBe` DBQueueId 1
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sndQueue1))
|
||||
upgradeSndConnToDuplex db "conn1" rcvQueue1
|
||||
`shouldReturn` Right 1
|
||||
`shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sq))
|
||||
Right rq@RcvQueue {dbQueueId = dbQueueId'} <- upgradeSndConnToDuplex db "conn1" rcvQueue1
|
||||
dbQueueId' `shouldBe` DBQueueId 1
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rcvQueue1] [sndQueue1]))
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rq] [sq]))
|
||||
|
||||
testCreateSndConnRandomID :: SpecWith SQLiteStore
|
||||
testCreateSndConnRandomID =
|
||||
it "should create SndConnection and add RcvQueue with random ID" . withStoreTransaction $ \db -> do
|
||||
g <- C.newRandom
|
||||
Right connId <- createSndConn db g cData1 {connId = ""} sndQueue1
|
||||
let rq' = (rcvQueue1 :: RcvQueue) {connId}
|
||||
sq' = (sndQueue1 :: SndQueue) {connId}
|
||||
Right (connId, sq) <- createSndConn db g cData1 {connId = ""} sndQueue1
|
||||
getConn db connId
|
||||
`shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 {connId} sq'))
|
||||
upgradeSndConnToDuplex db connId rcvQueue1
|
||||
`shouldReturn` Right 1
|
||||
`shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 {connId} sq))
|
||||
Right (rq@RcvQueue {dbQueueId = dbQueueId'}) <- upgradeSndConnToDuplex db connId rcvQueue1
|
||||
dbQueueId' `shouldBe` DBQueueId 1
|
||||
getConn db connId
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} [rq'] [sq']))
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} [rq] [sq]))
|
||||
|
||||
testCreateSndConnDuplicate :: SpecWith SQLiteStore
|
||||
testCreateSndConnDuplicate =
|
||||
@@ -280,17 +278,17 @@ testGetRcvConn =
|
||||
let smpServer = SMPServer "smp.simplex.im" "5223" testKeyHash
|
||||
let recipientId = "1234"
|
||||
g <- C.newRandom
|
||||
_ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
Right (_, rq) <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
getRcvConn db smpServer recipientId
|
||||
`shouldReturn` Right (rcvQueue1, SomeConn SCRcv (RcvConnection cData1 rcvQueue1))
|
||||
`shouldReturn` Right (rq, SomeConn SCRcv (RcvConnection cData1 rq))
|
||||
|
||||
testDeleteRcvConn :: SpecWith SQLiteStore
|
||||
testDeleteRcvConn =
|
||||
it "should create RcvConnection and delete it" . withStoreTransaction $ \db -> do
|
||||
g <- C.newRandom
|
||||
_ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
Right (_, rq) <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rcvQueue1))
|
||||
`shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rq))
|
||||
deleteConn db "conn1"
|
||||
`shouldReturn` ()
|
||||
getConn db "conn1"
|
||||
@@ -300,9 +298,9 @@ testDeleteSndConn :: SpecWith SQLiteStore
|
||||
testDeleteSndConn =
|
||||
it "should create SndConnection and delete it" . withStoreTransaction $ \db -> do
|
||||
g <- C.newRandom
|
||||
_ <- createSndConn db g cData1 sndQueue1
|
||||
Right (_, sq) <- createSndConn db g cData1 sndQueue1
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sndQueue1))
|
||||
`shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sq))
|
||||
deleteConn db "conn1"
|
||||
`shouldReturn` ()
|
||||
getConn db "conn1"
|
||||
@@ -312,10 +310,10 @@ testDeleteDuplexConn :: SpecWith SQLiteStore
|
||||
testDeleteDuplexConn =
|
||||
it "should create DuplexConnection and delete it" . withStoreTransaction $ \db -> do
|
||||
g <- C.newRandom
|
||||
_ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
_ <- upgradeRcvConnToDuplex db "conn1" sndQueue1
|
||||
Right (_, rq) <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
Right sq <- upgradeRcvConnToDuplex db "conn1" sndQueue1
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rcvQueue1] [sndQueue1]))
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rq] [sq]))
|
||||
deleteConn db "conn1"
|
||||
`shouldReturn` ()
|
||||
getConn db "conn1"
|
||||
@@ -337,7 +335,7 @@ testUpgradeRcvConnToDuplex =
|
||||
e2ePubKey = Nothing,
|
||||
e2eDhSecret = testDhSecret,
|
||||
status = New,
|
||||
dbQueueId = 1,
|
||||
dbQueueId = DBNewQueue,
|
||||
sndSwchStatus = Nothing,
|
||||
primary = True,
|
||||
dbReplaceQueueId = Nothing,
|
||||
@@ -366,7 +364,7 @@ testUpgradeSndConnToDuplex =
|
||||
e2eDhSecret = Nothing,
|
||||
sndId = "4567",
|
||||
status = New,
|
||||
dbQueueId = 1,
|
||||
dbQueueId = DBNewQueue,
|
||||
rcvSwchStatus = Nothing,
|
||||
primary = True,
|
||||
dbReplaceQueueId = Nothing,
|
||||
@@ -384,42 +382,42 @@ testSetRcvQueueStatus :: SpecWith SQLiteStore
|
||||
testSetRcvQueueStatus =
|
||||
it "should update status of RcvQueue" . withStoreTransaction $ \db -> do
|
||||
g <- C.newRandom
|
||||
_ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
Right (_, rq) <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rcvQueue1))
|
||||
setRcvQueueStatus db rcvQueue1 Confirmed
|
||||
`shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rq))
|
||||
setRcvQueueStatus db rq Confirmed
|
||||
`shouldReturn` ()
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rcvQueue1 {status = Confirmed}))
|
||||
`shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rq {status = Confirmed}))
|
||||
|
||||
testSetSndQueueStatus :: SpecWith SQLiteStore
|
||||
testSetSndQueueStatus =
|
||||
it "should update status of SndQueue" . withStoreTransaction $ \db -> do
|
||||
g <- C.newRandom
|
||||
_ <- createSndConn db g cData1 sndQueue1
|
||||
Right (_, sq) <- createSndConn db g cData1 sndQueue1
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sndQueue1))
|
||||
setSndQueueStatus db sndQueue1 Confirmed
|
||||
`shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sq))
|
||||
setSndQueueStatus db sq Confirmed
|
||||
`shouldReturn` ()
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sndQueue1 {status = Confirmed}))
|
||||
`shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sq {status = Confirmed}))
|
||||
|
||||
testSetQueueStatusDuplex :: SpecWith SQLiteStore
|
||||
testSetQueueStatusDuplex =
|
||||
it "should update statuses of RcvQueue and SndQueue in DuplexConnection" . withStoreTransaction $ \db -> do
|
||||
g <- C.newRandom
|
||||
_ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
_ <- upgradeRcvConnToDuplex db "conn1" sndQueue1
|
||||
Right (_, rq) <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
Right sq <- upgradeRcvConnToDuplex db "conn1" sndQueue1
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rcvQueue1] [sndQueue1]))
|
||||
setRcvQueueStatus db rcvQueue1 Secured
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rq] [sq]))
|
||||
setRcvQueueStatus db rq Secured
|
||||
`shouldReturn` ()
|
||||
let rq' = (rcvQueue1 :: RcvQueue) {status = Secured}
|
||||
sq' = (sndQueue1 :: SndQueue) {status = Confirmed}
|
||||
let rq' = (rq :: RcvQueue) {status = Secured}
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rq'] [sndQueue1]))
|
||||
setSndQueueStatus db sndQueue1 Confirmed
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rq'] [sq]))
|
||||
setSndQueueStatus db sq Confirmed
|
||||
`shouldReturn` ()
|
||||
let sq' = (sq :: SndQueue) {status = Confirmed}
|
||||
getConn db "conn1"
|
||||
`shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rq'] [sq']))
|
||||
|
||||
@@ -461,11 +459,11 @@ testCreateRcvMsg =
|
||||
it "should reserve internal ids and create a RcvMsg" $ \st -> do
|
||||
g <- C.newRandom
|
||||
let ConnData {connId} = cData1
|
||||
_ <- withTransaction st $ \db -> do
|
||||
Right (_, rq) <- withTransaction st $ \db -> do
|
||||
createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
withTransaction st $ \db -> do
|
||||
testCreateRcvMsg_ db 0 "" connId rcvQueue1 $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "hash_dummy"
|
||||
testCreateRcvMsg_ db 1 "hash_dummy" connId rcvQueue1 $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "new_hash_dummy"
|
||||
testCreateRcvMsg_ db 0 "" connId rq $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "hash_dummy"
|
||||
testCreateRcvMsg_ db 1 "hash_dummy" connId rq $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "new_hash_dummy"
|
||||
|
||||
mkSndMsgData :: InternalId -> InternalSndId -> MsgHash -> SndMsgData
|
||||
mkSndMsgData internalId internalSndId internalHash =
|
||||
@@ -502,15 +500,15 @@ testCreateRcvAndSndMsgs :: SpecWith SQLiteStore
|
||||
testCreateRcvAndSndMsgs =
|
||||
it "should create multiple RcvMsg and SndMsg, correctly ordering internal Ids and returning previous state" $ \st -> do
|
||||
let ConnData {connId} = cData1
|
||||
_ <- withTransaction st $ \db -> do
|
||||
Right (_, rq) <- withTransaction st $ \db -> do
|
||||
g <- C.newRandom
|
||||
createRcvConn db g cData1 rcvQueue1 SCMInvitation
|
||||
withTransaction st $ \db -> do
|
||||
_ <- upgradeRcvConnToDuplex db "conn1" sndQueue1
|
||||
testCreateRcvMsg_ db 0 "" connId rcvQueue1 $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "rcv_hash_1"
|
||||
testCreateRcvMsg_ db 1 "rcv_hash_1" connId rcvQueue1 $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "rcv_hash_2"
|
||||
testCreateRcvMsg_ db 0 "" connId rq $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "rcv_hash_1"
|
||||
testCreateRcvMsg_ db 1 "rcv_hash_1" connId rq $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "rcv_hash_2"
|
||||
testCreateSndMsg_ db "" connId $ mkSndMsgData (InternalId 3) (InternalSndId 1) "snd_hash_1"
|
||||
testCreateRcvMsg_ db 2 "rcv_hash_2" connId rcvQueue1 $ mkRcvMsgData (InternalId 4) (InternalRcvId 3) 3 "3" "rcv_hash_3"
|
||||
testCreateRcvMsg_ db 2 "rcv_hash_2" connId rq $ mkRcvMsgData (InternalId 4) (InternalRcvId 3) 3 "3" "rcv_hash_3"
|
||||
testCreateSndMsg_ db "snd_hash_1" connId $ mkSndMsgData (InternalId 5) (InternalSndId 2) "snd_hash_2"
|
||||
testCreateSndMsg_ db "snd_hash_2" connId $ mkSndMsgData (InternalId 6) (InternalSndId 3) "snd_hash_3"
|
||||
|
||||
|
||||
Reference in new issue
Block a user