mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-23 13:02:28 +00:00
tests: execute in parallel, agent: fix race condition with files and when switching queue (#1932)
* tests: execute in parallel * fix test * fix postgresql build * fix test * fix test * fix some tests * fix tests * fix tests * fix test * update schema * fix test * update schemas * diff * fix race * fixes * update schemas * fix tests --------- Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com>
This commit is contained in:
EvgenyandEvgeny @ SimpleX Chat authored and GitHub committed
1 parent
1d39da91c5
commit
9d83c0378e
40 files changed
+1157
-970
No files matched your search
@@ -279,6 +279,7 @@ retryOnError name loop done e = do
|
||||
|
||||
rcvWorkerInternalError :: AgentClient -> DBRcvFileId -> RcvFileId -> Maybe RcvFileId -> Maybe FilePath -> AgentErrorType -> AM ()
|
||||
rcvWorkerInternalError c rcvFileId rcvFileEntityId redirectEntityId_ tmpPath err = do
|
||||
liftIO $ throwWhenInactive c
|
||||
lift $ forM_ tmpPath (removePath <=< toFSFilePath)
|
||||
withStore' c $ \db -> updateRcvFileError db rcvFileId (show err)
|
||||
notify c (fromMaybe rcvFileEntityId redirectEntityId_) (RFERR err)
|
||||
@@ -318,9 +319,6 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
|
||||
withStore' c (`updateRcvFileComplete` rcvFileId)
|
||||
Just RcvFileRedirect {redirectFileInfo, redirectDbId} -> do
|
||||
let RedirectFileInfo {size = redirectSize, digest = redirectDigest} = redirectFileInfo
|
||||
lift $ forM_ tmpPath (removePath <=< toFSFilePath)
|
||||
liftIO $ waitUntilForeground c
|
||||
withStore' c (`updateRcvFileComplete` rcvFileId)
|
||||
-- proceed with redirect
|
||||
yaml <- liftError (FILE . FILE_IO . show) (CF.readFile $ CryptoFile fsSavePath cfArgs) `allFinally` (lift $ toFSFilePath fsSavePath >>= removePath)
|
||||
next@FileDescription {chunks = nextChunks} <- case strDecode (LB.toStrict yaml) of
|
||||
@@ -330,9 +328,12 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
|
||||
| dstSize /= redirectSize -> throwE . FILE $ REDIRECT "size mismatch"
|
||||
| dstDigest /= redirectDigest -> throwE . FILE $ REDIRECT "digest mismatch"
|
||||
| otherwise -> pure fd
|
||||
liftIO $ waitUntilForeground c
|
||||
-- register and download chunks from the actual file
|
||||
withStore c $ \db -> updateRcvFileRedirect db redirectDbId next
|
||||
withStore c $ \db -> updateRcvFileComplete db rcvFileId >> updateRcvFileRedirect db redirectDbId next
|
||||
forM_ nextChunks (downloadChunk c)
|
||||
flip catchAllErrors (\e -> logError $ "XFTP rcv worker error: " <> tshow e) $
|
||||
lift $ forM_ tmpPath (removePath <=< toFSFilePath)
|
||||
where
|
||||
getChunkPaths :: [RcvFileChunk] -> AM [FilePath]
|
||||
getChunkPaths [] = pure []
|
||||
@@ -512,6 +513,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
|
||||
|
||||
sndWorkerInternalError :: AgentClient -> DBSndFileId -> SndFileId -> Maybe FilePath -> AgentErrorType -> AM ()
|
||||
sndWorkerInternalError c sndFileId sndFileEntityId prefixPath err = do
|
||||
liftIO $ throwWhenInactive c
|
||||
lift . forM_ prefixPath $ removePath <=< toFSFilePath
|
||||
withStore' c $ \db -> updateSndFileError db sndFileId (show err)
|
||||
notify c sndFileEntityId $ SFERR err
|
||||
@@ -733,6 +735,7 @@ runXFTPDelWorker c srv Worker {doWork} = do
|
||||
|
||||
delWorkerInternalError :: AgentClient -> Int64 -> AgentErrorType -> AM ()
|
||||
delWorkerInternalError c deletedSndChunkReplicaId e = do
|
||||
liftIO $ throwWhenInactive c
|
||||
withStore' c $ \db -> deleteDeletedSndChunkReplica db deletedSndChunkReplicaId
|
||||
notify c "" $ SFERR e
|
||||
|
||||
|
||||
@@ -2360,9 +2360,9 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
|
||||
-- we may add more statistics special to queue rotation later on,
|
||||
-- not accounting secure during rotation for now:
|
||||
-- atomically $ incSMPServerStat c userId server connSecured
|
||||
withStore' c $ \db -> setRcvQueueStatus db rq' Secured
|
||||
rq1' <- withStore' c $ \db -> setRcvQueueStatus db rq' Secured >> setRcvSwitchStatus db rq1 (Just RSSendingQUSE)
|
||||
void . enqueueMessages c cData sqs SMP.noMsgFlags $ QUSE [((server, sndId), True)]
|
||||
rq1' <- withStore' c $ \db -> setRcvSwitchStatus db rq1 (Just RSSendingQUSE) <* deleteCommand db cmdId
|
||||
withStore' c (`deleteCommand` cmdId)
|
||||
let rqs' = updatedQs rq1' rqs
|
||||
conn' = DuplexConnection cData rqs' sqs
|
||||
cStats <- connectionStats c conn'
|
||||
@@ -4118,15 +4118,15 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
|
||||
withStore' c $ \db -> copyPendingSndDeliveries db sq sq2
|
||||
enqueueCommand c "" connId (Just $ qServer sq2) $ AInternalCommand $ ICQSndSecure (snd $ qAddress sq2)
|
||||
pure SSSecuringQueue
|
||||
else do
|
||||
let sqInfo' = (sqInfo :: SMPQueueInfo) {queueAddress = queueAddress {dhPublicKey}}
|
||||
void . enqueueMessages c cData' sqs SMP.noMsgFlags $ QKEY [(sqInfo', C.toPublic sndPrivateKey)]
|
||||
pure SSSendingQKEY
|
||||
else pure SSSendingQKEY
|
||||
sq1 <- withStore' c $ \db -> setSndSwitchStatus db sq $ Just swchStatus
|
||||
let sqs'' = updatedQs sq1 sqs' <> [sq2]
|
||||
conn' = DuplexConnection cData' rqs sqs''
|
||||
cStats <- connectionStats c conn'
|
||||
notify $ SWITCH QDSnd SPStarted cStats
|
||||
when (connAgentVersion < rpcAddressSMPAgentVersion) $ do
|
||||
let sqInfo' = (sqInfo :: SMPQueueInfo) {queueAddress = queueAddress {dhPublicKey}}
|
||||
void . enqueueMessages c cData' sqs SMP.noMsgFlags $ QKEY [(sqInfo', C.toPublic sndPrivateKey)]
|
||||
_ -> qError "QADD: won't delete all snd queues in connection"
|
||||
_ -> qError "QADD: replaced queue address is not found in connection"
|
||||
_ -> throwE $ AGENT A_VERSION
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
|
||||
|
||||
SET statement_timeout = 0;
|
||||
SET lock_timeout = 0;
|
||||
SET idle_in_transaction_session_timeout = 0;
|
||||
@@ -1531,6 +1529,3 @@ ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_queues
|
||||
|
||||
ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_queues
|
||||
ADD CONSTRAINT snd_queues_host_port_fkey FOREIGN KEY (host, port) REFERENCES smp_agent_test_protocol_schema.servers(host, port) ON UPDATE CASCADE ON DELETE RESTRICT;
|
||||
|
||||
|
||||
|
||||
@@ -1,6 +1,3 @@
|
||||
|
||||
|
||||
|
||||
SET statement_timeout = 0;
|
||||
SET lock_timeout = 0;
|
||||
SET idle_in_transaction_session_timeout = 0;
|
||||
@@ -310,7 +307,3 @@ ALTER TABLE ONLY ntf_server.subscriptions
|
||||
|
||||
ALTER TABLE ONLY ntf_server.subscriptions
|
||||
ADD CONSTRAINT subscriptions_token_id_fkey FOREIGN KEY (token_id) REFERENCES ntf_server.tokens(token_id) ON UPDATE RESTRICT ON DELETE CASCADE;
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -719,6 +719,7 @@ importStoreLogToDatabase logPath storeLogFile dbOpts = do
|
||||
ps <- newJournalMsgStore logPath $ PQStoreCfg storeCfg
|
||||
sCnt <- batchInsertServices services' $ postgresQueueStore ps
|
||||
qCnt <- batchInsertQueues @(JournalQueue 'QSMemory) True queues $ postgresQueueStore ps
|
||||
closeMsgStore ps
|
||||
renameFile storeLogFile $ storeLogFile <> ".bak"
|
||||
pure (sCnt, qCnt)
|
||||
|
||||
@@ -745,6 +746,7 @@ exportDatabaseToStoreLog logPath dbOpts storeLogFilePath = do
|
||||
Sum sCnt <- foldServiceRecs (postgresQueueStore ps) $ \sr -> logNewService sl sr $> Sum (1 :: Int)
|
||||
Sum qCnt <- foldQueueRecs True True (postgresQueueStore ps) $ \(rId, qr) -> logCreateQueue sl rId qr $> Sum (1 :: Int)
|
||||
closeStoreLog sl
|
||||
closeMsgStore ps
|
||||
pure (sCnt, qCnt)
|
||||
#endif
|
||||
|
||||
|
||||
@@ -141,4 +141,8 @@ httpGet ResolverEnv {manager, baseUrl, authHdr, timeoutMicro, maxResponseBytes}
|
||||
else do
|
||||
bs <- brReadSome (responseBody res) (maxResponseBytes + 1)
|
||||
pure $ if BL.length bs > fromIntegral maxResponseBytes then Left BodyTooLarge else Right bs
|
||||
pure (either (Left . HttpFailure) id result)
|
||||
pure (either (Left . httpError) id result)
|
||||
where
|
||||
httpError = \case
|
||||
HC.HttpExceptionRequest _ HC.ResponseTimeout -> ResolverTimeout
|
||||
e -> HttpFailure e
|
||||
@@ -1,6 +1,3 @@
|
||||
|
||||
|
||||
|
||||
SET statement_timeout = 0;
|
||||
SET lock_timeout = 0;
|
||||
SET idle_in_transaction_session_timeout = 0;
|
||||
@@ -552,7 +549,3 @@ ALTER TABLE ONLY smp_server.msg_queues
|
||||
|
||||
ALTER TABLE ONLY smp_server.msg_queues
|
||||
ADD CONSTRAINT msg_queues_rcv_service_id_fkey FOREIGN KEY (rcv_service_id) REFERENCES smp_server.services(service_id) ON UPDATE RESTRICT ON DELETE SET NULL;
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user