mirror of
https://github.com/simplex-chat/simplex-chat.git
synced 2026-11-04 10:47:32 +00:00
core, tests: fix flaky tests; use simplexmq with one thread per agent worker (#7812)
* tests: fix group tests * fix test * fix * fix delivery worker fix * watch ci * debug * fix delivery worker not forwarding to moderators during host acceptance * add script, update simplexmq * debug lines * more debug * rerun * debug * fix, debug * simplexmq * simplexmq * fix * debug: log member-not-found call sites, restore hang diagnostics, use simplexmq fork * tests: do not prompt for migration confirmation in strict tables test * ci: run tests with one capability; simplexmq: end inactive sessions without closing under readers * ci: run tests on all capabilities * remote: close the HTTP2 client before its TLS connection * cabal: use warp that kills timed out connection threads without closing their sockets * remove diagnostics: member-not-found log, thread dump, CI watchdog; update simplexmq * badge-service: prevent interleaving of concurrent log lines * tests: wait for badge group broadcasts before closing members * tests: log contact connected events received twice * tests: log member connected events received twice * core: use simplexmq that logs connections emitting CON twice * core: keep the team contact above notes in a new user's chat list * ci: skip macOS and Windows builds while hunting a Linux-only flake * core: use simplexmq that logs the history of connections emitting CON twice * tests: remove the logging of connected events received twice * ci: run macOS and Windows builds again * core: use simplexmq that runs one thread per agent worker * core: use the merged simplex-chat/wai warp and the updated simplexmq * core: use the merged simplexmq * core: update simplexmq * core, badge service: comment single writes of terminal segments and log lines
This commit is contained in:
1 parent
ea679e6e95
commit
a5cf718b3c
14 files changed
+135
-81
No files matched your search
@@ -476,8 +476,8 @@ processChatCommand cxt nm = \case
|
||||
user <- withFastStore $ \db -> do
|
||||
user <- createUserRecordAt db (AgentUserId auId) (isTrue userChatRelay) service p True ts
|
||||
mapM_ (setUserServers db user ts) uss
|
||||
createPresetContactCards db user `catchAllErrors` \_ -> pure ()
|
||||
createNoteFolder db user
|
||||
createPresetContactCards db user `catchAllErrors` \_ -> pure ()
|
||||
pure user
|
||||
atomically . writeTVar u $ Just user
|
||||
pure $ CRActiveUser user
|
||||
|
||||
@@ -25,7 +25,7 @@ import Control.Monad.Reader
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import qualified Data.ByteString.Lazy.Char8 as LB
|
||||
import Data.Either (lefts, partitionEithers, rights)
|
||||
import Data.Either (isLeft, lefts, partitionEithers, rights)
|
||||
import Data.Foldable (foldr', foldrM)
|
||||
import Data.Functor (($>))
|
||||
import Data.Int (Int64)
|
||||
@@ -1130,7 +1130,7 @@ processAgentMessageConn cxt user@User {userId} entity gks_ corrId agentConnId ag
|
||||
XInfoProbeOk probe -> Nothing <$ xInfoProbeOk (COMGroupMember m'') probe
|
||||
BFileChunk sharedMsgId chunk -> Nothing <$ bFileChunkGroup (GIK gInfo' gks) m'' sharedMsgId chunk msgMeta
|
||||
_ -> Nothing <$ messageError ("unsupported message: " <> tshow event)
|
||||
forM deliveryTaskContext_ $ \taskContext -> do
|
||||
forM (if memberStatus m'' == GSMemPendingApproval then Nothing else deliveryTaskContext_) $ \taskContext -> do
|
||||
let contentChanged :: CM ()
|
||||
contentChanged = atomically $ channelContentChanged cc groupId
|
||||
case event of
|
||||
@@ -4193,11 +4193,16 @@ resumeDeliveryTaskWork :: DeliveryWorkerKey -> CM' ()
|
||||
resumeDeliveryTaskWork = void . getDeliveryTaskWorker False
|
||||
|
||||
getDeliveryTaskWorker :: Bool -> DeliveryWorkerKey -> CM' Worker
|
||||
getDeliveryTaskWorker hasWork deliveryKey = do
|
||||
getDeliveryTaskWorker hasWork deliveryKey@(groupId, _) = do
|
||||
ws <- asks deliveryTaskWorkers
|
||||
a <- asks smpAgent
|
||||
getAgentWorker "delivery_task" hasWork a deliveryKey ws $
|
||||
runDeliveryTaskWorker a deliveryKey
|
||||
getAgentWorker "delivery_task" hasWork a deliveryKey ws $ \w ->
|
||||
runDeliveryTaskWorker a deliveryKey w `catchAllErrors` \e -> do
|
||||
whenM (groupDeleted groupId) $ atomically $ TM.delete deliveryKey ws
|
||||
throwError e
|
||||
|
||||
groupDeleted :: GroupId -> CM Bool
|
||||
groupDeleted groupId = isLeft <$> withStore' (\db -> runExceptT $ getUserByGroupId db groupId)
|
||||
|
||||
runDeliveryTaskWorker :: AgentClient -> DeliveryWorkerKey -> Worker -> CM ()
|
||||
runDeliveryTaskWorker a deliveryKey Worker {doWork} = do
|
||||
@@ -4220,7 +4225,7 @@ runDeliveryTaskWorker a deliveryKey Worker {doWork} = do
|
||||
processDeliveryTask task
|
||||
`catchAllErrors` \e -> do
|
||||
withStore' $ \db -> setDeliveryTaskErrStatus db (deliveryTaskId task) (tshow e)
|
||||
eToView e
|
||||
unlessM (groupDeleted groupId) $ eToView e
|
||||
where
|
||||
processDeliveryTask :: MessageDeliveryTask -> CM ()
|
||||
processDeliveryTask task@MessageDeliveryTask {jobScope} =
|
||||
@@ -4261,11 +4266,13 @@ resumeDeliveryJobWork :: DeliveryWorkerKey -> CM' ()
|
||||
resumeDeliveryJobWork = void . getDeliveryJobWorker False
|
||||
|
||||
getDeliveryJobWorker :: Bool -> DeliveryWorkerKey -> CM' Worker
|
||||
getDeliveryJobWorker hasWork deliveryKey = do
|
||||
getDeliveryJobWorker hasWork deliveryKey@(groupId, _) = do
|
||||
ws <- asks deliveryJobWorkers
|
||||
a <- asks smpAgent
|
||||
getAgentWorker "delivery_job" hasWork a deliveryKey ws $
|
||||
runDeliveryJobWorker a deliveryKey
|
||||
getAgentWorker "delivery_job" hasWork a deliveryKey ws $ \w ->
|
||||
runDeliveryJobWorker a deliveryKey w `catchAllErrors` \e -> do
|
||||
whenM (groupDeleted groupId) $ atomically $ TM.delete deliveryKey ws
|
||||
throwError e
|
||||
|
||||
-- TODO [relays] dissemination here is unsigned (relay-asserted profile).
|
||||
-- Future: members sign an XMember on channel join, relay stores it per
|
||||
@@ -4308,7 +4315,7 @@ runDeliveryJobWorker a deliveryKey Worker {doWork} = do
|
||||
processDeliveryJob job
|
||||
`catchAllErrors` \e -> do
|
||||
withStore' $ \db -> setDeliveryJobErrStatus db (deliveryJobId job) (tshow e)
|
||||
eToView e
|
||||
unlessM (groupDeleted groupId) $ eToView e
|
||||
where
|
||||
processDeliveryJob :: MessageDeliveryJob -> CM ()
|
||||
processDeliveryJob job =
|
||||
|
||||
@@ -297,9 +297,9 @@ cancelRemoteHost handlingError = \case
|
||||
closeConnection tls
|
||||
RHSessionConnected {rchClient, tls, rhClient = RemoteHostClient {httpClient}, pollAction} -> do
|
||||
uninterruptibleCancel pollAction
|
||||
unless handlingError $ closeHTTP2Client httpClient `catchAny` (logError . tshow)
|
||||
cancelHostClient rchClient `catchAny` (logError . tshow)
|
||||
closeConnection tls `catchAny` (logError . tshow)
|
||||
unless handlingError $ closeHTTP2Client httpClient `catchAny` (logError . tshow)
|
||||
where
|
||||
cancelPendingSession RHPendingSession {rchClient, rhsWaitSession} = do
|
||||
unless handlingError $ uninterruptibleCancel rhsWaitSession `catchAny` (logError . tshow)
|
||||
|
||||
@@ -388,8 +388,9 @@ printMessage ChatTerminal {termSize = Size {height, width}, nextMessageRow} msg
|
||||
-- because of accounting for control sequences in length
|
||||
putStyled :: MonadTerminal m => StyledString -> m ()
|
||||
putStyled (s1 :<>: s2) = putStyled s1 >> putStyled s2
|
||||
putStyled (Styled [] s) = putString s
|
||||
putStyled (Styled sgr s) = setSGR sgr >> putString s >> resetAttributes
|
||||
-- putText, as putString sends a terminal command per character
|
||||
putStyled (Styled [] s) = putText (T.pack s)
|
||||
putStyled (Styled sgr s) = setSGR sgr >> putText (T.pack s) >> resetAttributes
|
||||
|
||||
setSGR :: MonadTerminal m => [SGR] -> m ()
|
||||
setSGR = mapM_ $ \case
|
||||
|
||||
Reference in new issue
Block a user