From 19a3ab623086e8494217e9dcb7462c565b7e6252 Mon Sep 17 00:00:00 2001 From: Alexander Bondarenko <486682+dpwiz@users.noreply.github.com> Date: Fri, 26 Apr 2024 15:41:53 +0300 Subject: [PATCH] fix DebugDelivery --- src/Simplex/Chat.hs | 163 +++++++++++++-------------------- src/Simplex/Chat/Controller.hs | 31 +++++-- src/Simplex/Chat/View.hs | 2 +- 3 files changed, 84 insertions(+), 112 deletions(-) diff --git a/src/Simplex/Chat.hs b/src/Simplex/Chat.hs index efb20c8ffb..fdf97cb9f6 100644 --- a/src/Simplex/Chat.hs +++ b/src/Simplex/Chat.hs @@ -231,7 +231,7 @@ newChatController inputQ <- newTBQueueIO tbqSize outputQ <- newTBQueueIO tbqSize connNetworkStatuses <- atomically TM.empty - agentConnStatuses <- atomically TM.empty + agentDeliveryStatuses <- atomically TM.empty subscriptionMode <- newTVarIO SMSubscribe chatLock <- newEmptyTMVarIO entityLocks <- atomically TM.empty @@ -268,7 +268,7 @@ newChatController inputQ, outputQ, connNetworkStatuses, - agentConnStatuses, + agentDeliveryStatuses, subscriptionMode, chatLock, entityLocks, @@ -2153,55 +2153,10 @@ processChatCommand' vr = \case chatMigrations <- map upMigration <$> withStore' (Migrations.getCurrent . DB.conn) agentMigrations <- withAgent getAgentMigrations pure $ CRVersionInfo {versionInfo, chatMigrations, agentMigrations} - DebugAcks -> do - acs <- mapM readTVarIO =<< readTVarIO =<< asks agentConnStatuses - fmap (CRDebugAcks . M.fromList) . forM (M.toList acs) $ \(acId@(AgentConnId acId'), agentConnStatus) -> do - user_ <- withStore' (`getUserByAConnId` acId) - - rq' <- withAgent $ \ac -> liftIO . (`runReaderT` agentEnv ac) . runExceptT $ do - AS.RcvQueue {server, rcvId, status, smpClientVersion} <- AC.withStore ac (`ADB.getPrimaryRcvQueue` acId') - pure (server, (rcvId, status, smpClientVersion)) - - (inActive, inPending) <- case (user_, rq') of - (Just User {userId}, Right (srv, (rcvId', _, _))) -> do - let tSess = (userId, srv, Just acId') - withAgent $ \AgentClient {activeSubs, pendingSubs} -> do - active <- atomically (RQ.getSessQueues tSess activeSubs) - pending <- atomically (RQ.getSessQueues tSess pendingSubs) - pure - ( any (\AS.RcvQueue {rcvId} -> rcvId == rcvId') active, - any (\AS.RcvQueue {rcvId} -> rcvId == rcvId') pending - ) - _ -> pure (False, False) - - entity_ <- forM user_ $ \user -> withStore (\db -> getConnectionEntity db vr user acId) - conn_ <- forM entity_ $ \case - RcvDirectMsgConnection {entityConnection} -> pure entityConnection - RcvGroupMsgConnection {entityConnection} -> pure entityConnection - SndFileConnection {entityConnection} -> pure entityConnection - RcvFileConnection {entityConnection} -> pure entityConnection - UserContactConnection {entityConnection} -> pure entityConnection - - let AgentConnStatus {lastCmd, lastCmdTag, lastMsg, ackSent, okRcvd} = agentConnStatus - pure - ( decodeLatin1 $ strEncode acId, - DebugAck - { lastCmd = Just (decodeLatin1 $ strEncode lastCmdTag, lastCmd), - lastMsg, - lastAck = ackSent, - lastOK = okRcvd, - inActive, - inPending, - server = either (const Nothing) (\(SMP.ProtocolServer {host, port}, _) -> Just (decodeLatin1 . strEncode $ L.head host, port)) rq', - hasSMPClient = True, -- TODO - hasSubWorker = False, -- TODO - hasDeliveryWorker = False, -- TODO - connStatus_ = (\Connection {connStatus} -> connStatus) <$> conn_, - -- agent connstatus? - connAuthErrors = (\c@Connection {authErrCounter} -> (authErrCounter, connDisabled c)) <$> conn_, - createdAt = (\Connection {createdAt} -> createdAt) <$> conn_ - } - ) + DebugDelivery -> do + ads <- mapM readTVarIO =<< readTVarIO =<< asks agentDeliveryStatuses + let collect (acId, ds) = if agentDeliveryOk ds then Nothing else Just (decodeLatin1 $ strEncode acId, ds) + pure $ CRDebugDelivery . M.fromList . mapMaybe collect $ M.toList ads DebugLocks -> lift $ do chatLockName <- atomically . tryReadTMVar =<< asks chatLock chatEntityLocks <- getLocks =<< asks entityLocks @@ -2528,8 +2483,6 @@ processChatCommand' vr = \case toView $ CRNewChatItem user (AChatItem SCTDirect SMDSnd (DirectChat ct) ci) forM_ (timed_ >>= timedDeleteAt') $ startProximateTimedItemThread user (ChatRef CTDirect contactId, chatItemId' ci) - drgRandomBytes :: Int -> CM ByteString - drgRandomBytes n = asks random >>= atomically . C.randomBytes n privateGetUser :: UserId -> CM User privateGetUser userId = tryChatError (withStore (`getUser` userId)) >>= \case @@ -3583,17 +3536,12 @@ expireChatItems user@User {userId} ttl sync = do forM_ membersToDelete $ \m -> withStore' $ \db -> deleteGroupMember db user m processAgentMessage :: ACorrId -> ConnId -> ACommand 'Agent 'AEConn -> CM () -processAgentMessage _ connId (DEL_RCVQ srv qId err_) = do - let acId = AgentConnId connId - asks agentConnStatuses >>= atomically . TM.delete acId - toView $ CRAgentRcvQueueDeleted acId srv (AgentQueueId qId) err_ -processAgentMessage _ connId DEL_CONN = do - let acId = AgentConnId connId - asks agentConnStatuses >>= atomically . TM.delete acId - toView $ CRAgentConnDeleted acId +processAgentMessage _ connId (DEL_RCVQ srv qId err_) = toView $ CRAgentRcvQueueDeleted (AgentConnId connId) srv (AgentQueueId qId) err_ +processAgentMessage _ connId DEL_CONN = toView $ CRAgentConnDeleted (AgentConnId connId) processAgentMessage corrId connId msg = do let acId = AgentConnId connId - lift $ trackAgentConn acId msg + acTag = aCommandTag msg + when (acTag == MSG_ || acTag == RCVD_) . lift $ trackNewDelivery acId (acTag == MSG_) lockEntity <- critical (withStore (`getChatLockEntity` acId)) withEntityLock "processAgentMessage" lockEntity $ do vr <- chatVersionRange @@ -3602,24 +3550,15 @@ processAgentMessage corrId connId msg = do Just user -> processAgentMessageConn vr user corrId connId msg `catchChatError` (toView . CRChatError (Just user)) _ -> throwChatError $ CENoConnectionUser acId -trackAgentConn :: AgentConnId -> ACommand 'Agent 'AEConn -> CM' () -trackAgentConn acId msg = do +-- TODO: clean up deliveries +trackNewDelivery :: AgentConnId -> Bool -> CM' () +trackNewDelivery acId isMSG = do now <- liftIO getCurrentTime - asks agentConnStatuses >>= atomically . TM.alterF (updateConn now) acId + asks agentDeliveryStatuses >>= atomically . TM.alterF (updateConn now) acId where - updateConn now = \case - Nothing -> Just <$> newTVar (status now Nothing Nothing Nothing) - Just v -> Just v <$ modifyTVar' v (\AgentConnStatus {lastMsg, ackSent, okRcvd} -> status now lastMsg ackSent okRcvd) - status now lastMsg ackSent okRcvd = AgentConnStatus - { lastCmd = now, - lastCmdTag, - lastMsg = if isMSG then Just now else lastMsg, - ackSent, - okRcvd = if isOK then Just now else okRcvd - } - lastCmdTag = aCommandTag msg - isMSG = lastCmdTag == MSG_ - isOK = lastCmdTag == OK_ + updateConn lastCmd = \case + Nothing -> Just <$> newTVar AgentDeliveryStatus {lastCmd, isMSG, connId = Nothing, eventTag = Nothing,ackSent = Nothing, pendingAcks = M.empty} + Just v -> Just v <$ modifyTVar' v (\AgentDeliveryStatus {pendingAcks} -> AgentDeliveryStatus {lastCmd, isMSG, connId = Nothing, eventTag = Nothing,ackSent = Nothing, pendingAcks = M.filter not pendingAcks}) -- CRITICAL error will be shown to the user as alert with restart button in Android/desktop apps. -- SEDBBusyError will only be thrown on IO exceptions or SQLError during DB queries, @@ -3834,15 +3773,20 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage = _ -> toView $ CRSubscriptionEnd user entity MSGNTF smpMsgInfo -> toView $ CRNtfMessage user entity $ ntfMsgInfo smpMsgInfo _ -> case entity of - RcvDirectMsgConnection conn contact_ -> + RcvDirectMsgConnection conn contact_ -> do + storeDeliveryConn conn processDirectMessage agentMessage entity conn contact_ - RcvGroupMsgConnection conn gInfo m -> + RcvGroupMsgConnection conn gInfo m -> do + storeDeliveryConn conn processGroupMessage agentMessage entity conn gInfo m - RcvFileConnection conn ft -> + RcvFileConnection conn ft -> do + storeDeliveryConn conn processRcvFileConn agentMessage entity conn ft - SndFileConnection conn ft -> + SndFileConnection conn ft -> do + storeDeliveryConn conn processSndFileConn agentMessage entity conn ft - UserContactConnection conn uc -> + UserContactConnection conn uc -> do + storeDeliveryConn conn processUserContactRequest agentMessage entity conn uc where updateConnStatus :: ConnectionEntity -> CM ConnectionEntity @@ -3852,7 +3796,8 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage = withStore' $ \db -> updateConnectionStatus db conn connStatus pure $ updateEntityConnStatus acEntity connStatus Nothing -> pure acEntity - + storeDeliveryConn :: Connection -> CM () + storeDeliveryConn Connection {connId} = lift $ agentDeliveryStatus (AgentConnId agentConnId) $ \ad -> ad {connId = Just connId} agentMsgConnStatus :: ACommand 'Agent e -> Maybe ConnStatus agentMsgConnStatus = \case CONF {} -> Just ConnRequested @@ -3931,6 +3876,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage = (ct', conn') <- updateContactPQRcv user ct conn pqEncryption checkIntegrityCreateItem (CDDirectRcv ct') msgMeta `catchChatError` \_ -> pure () (conn'', msg@RcvMessage {chatMsgEvent = ACME _ event}) <- saveDirectRcvMSG conn' msgMeta msgBody + lift $ agentDeliveryStatus (AgentConnId agentConnId) $ \ad -> ad {eventTag = Just $! tshow (toCMEventTag event)} let ct'' = ct' {activeConn = Just conn''} :: Contact assertDirectAllowed user MDRcv ct'' $ toCMEventTag event case event of @@ -4351,6 +4297,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage = processEvent :: MsgEncodingI e => ChatMessage e -> CM () processEvent chatMsg = do (m', conn', msg@RcvMessage {chatMsgEvent = ACME _ event}) <- saveGroupRcvMsg user groupId m conn msgMeta msgBody chatMsg + lift $ agentDeliveryStatus (AgentConnId agentConnId) $ \ad -> ad {eventTag = Just $! tshow (toCMEventTag event)} case event of XMsgNew mc -> memberCanSend m' $ newGroupContentMessage gInfo m' mc msg brokerTs False XMsgFileDescr sharedMsgId fileDescr -> memberCanSend m' $ groupMessageFileDescription gInfo m' sharedMsgId fileDescr @@ -4689,16 +4636,20 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage = withCompletedCommand :: forall e. AEntityI e => Connection -> ACommand 'Agent e -> (CommandData -> CM ()) -> CM () withCompletedCommand Connection {connId} agentMsg action = do let agentMsgTag = APCT (sAEntity @e) $ aCommandTag agentMsg - cmdData_ <- withStore' $ \db -> getCommandDataByCorrId db user corrId - case cmdData_ of - Just cmdData@CommandData {cmdId, cmdConnId = Just cmdConnId', cmdFunction} - | connId == cmdConnId' && (agentMsgTag == commandExpectedResponse cmdFunction || agentMsgTag == APCT SAEConn ERR_) -> do - withStore' $ \db -> deleteCommand db user cmdId - action cmdData - | otherwise -> err cmdId $ "not matching connection id or unexpected response, corrId = " <> show corrId - Just CommandData {cmdId, cmdConnId = Nothing} -> err cmdId $ "no command connection id, corrId = " <> show corrId - Nothing -> throwChatError . CEAgentCommandError $ "command not found, corrId = " <> show corrId + if agentMsgTag == APCT SAEConn OK_ && corrId /= "" then markDelivery else do + cmdData_ <- withStore' $ \db -> getCommandDataByCorrId db user corrId + case cmdData_ of + Just cmdData@CommandData {cmdId, cmdConnId = Just cmdConnId', cmdFunction} + | connId == cmdConnId' && (agentMsgTag == commandExpectedResponse cmdFunction || agentMsgTag == APCT SAEConn ERR_) -> do + withStore' $ \db -> deleteCommand db user cmdId + action cmdData + | otherwise -> err cmdId $ "not matching connection id or unexpected response, corrId = " <> show corrId + Just CommandData {cmdId, cmdConnId = Nothing} -> err cmdId $ "no command connection id, corrId = " <> show corrId + Nothing -> throwChatError . CEAgentCommandError $ "command not found, corrId = " <> show corrId where + markDelivery = lift $ agentDeliveryStatus (AgentConnId agentConnId) $ \ad@AgentDeliveryStatus {pendingAcks} -> ad {pendingAcks = M.adjust (const True) ackKey pendingAcks} + where + ackKey = decodeLatin1 $ strEncode corrId err cmdId msg = do withStore' $ \db -> updateCommandStatus db user cmdId CSError throwChatError . CEAgentCommandError $ msg @@ -4724,12 +4675,12 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage = where ackMsg :: MsgMeta -> Maybe MsgReceiptInfo -> CM () ackMsg MsgMeta {recipient = (msgId, _)} rcpt = do - withAgent $ \a -> ackMessageAsync a "" cId msgId rcpt - acs <- asks agentConnStatuses - liftIO $ do - let acId = AgentConnId cId - now <- getCurrentTime - atomically $ TM.lookup acId acs >>= mapM_ (\v -> modifyTVar' v $ \cs -> cs {ackSent = Just now}) + ackId <- drgRandomBytes 24 + withAgent $ \a -> ackMessageAsync a ackId cId msgId rcpt + now <- liftIO getCurrentTime + let ackKey = decodeLatin1 $ strEncode ackId + lift . agentDeliveryStatus (AgentConnId agentConnId) $ \ad@AgentDeliveryStatus {pendingAcks} -> + ad {ackSent = Just (now, ackKey), pendingAcks = M.insert ackKey False pendingAcks} sentMsgDeliveryEvent :: Connection -> AgentMsgId -> CM () sentMsgDeliveryEvent Connection {connId} msgId = @@ -6028,6 +5979,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage = processForwardedMsg author chatMsg = do let body = LB.toStrict $ J.encode msg rcvMsg@RcvMessage {chatMsgEvent = ACME _ event} <- saveGroupFwdRcvMsg user groupId m author body chatMsg + lift $ agentDeliveryStatus (AgentConnId agentConnId) $ \ad -> ad {eventTag = Just $! tshow (toCMEventTag event)} case event of XMsgNew mc -> memberCanSend author $ newGroupContentMessage gInfo author mc rcvMsg msgTs True XMsgFileDescr sharedMsgId fileDescr -> memberCanSend author $ groupMessageFileDescription gInfo author sharedMsgId fileDescr @@ -6652,9 +6604,10 @@ sendPendingGroupMessages user GroupMember {groupMemberId, localDisplayName} conn -- TODO [batch send] refactor direct message processing same as groups (e.g. checkIntegrity before processing) saveDirectRcvMSG :: Connection -> MsgMeta -> MsgBody -> CM (Connection, RcvMessage) -saveDirectRcvMSG conn@Connection {connId} agentMsgMeta msgBody = +saveDirectRcvMSG conn@Connection {connId, agentConnId} agentMsgMeta msgBody = case parseChatMessages msgBody of [Right (ACMsg _ ChatMessage {chatVRange, msgId = sharedMsgId_, chatMsgEvent})] -> do + lift $ agentDeliveryStatus agentConnId $ \ad -> ad {eventTag = Just $! tshow (toCMEventTag chatMsgEvent)} conn' <- updatePeerChatVRange conn chatVRange let agentMsgId = fst $ recipient agentMsgMeta newMsg = NewRcvMessage {chatMsgEvent, msgBody} @@ -7357,7 +7310,7 @@ chatCommandP = "/_download " *> (APIDownloadStandaloneFile <$> A.decimal <* A.space <*> strP_ <*> cryptoFileP), ("/quit" <|> "/q" <|> "/exit") $> QuitChat, ("/version" <|> "/v") $> ShowVersion, - "/debug acks" $> DebugAcks, + "/debug delivery" $> DebugDelivery, "/debug locks" $> DebugLocks, "/debug event " *> (DebugEvent <$> jsonP), "/get stats" $> GetAgentStats, @@ -7519,6 +7472,9 @@ timeItToView s action = do toView' $ CRTimedAction s diff pure a +drgRandomBytes :: Int -> CM ByteString +drgRandomBytes n = asks random >>= atomically . C.randomBytes n + mkValidName :: String -> String mkValidName = reverse . dropWhile isSpace . fst3 . foldl' addChar ("", '\NUL', 0 :: Int) where @@ -7563,3 +7519,8 @@ xftpSndFileRedirect user ftId vfd = do dummyFileDescr :: FileDescr dummyFileDescr = FileDescr {fileDescrText = "", fileDescrPartNo = 0, fileDescrComplete = False} + +agentDeliveryStatus :: AgentConnId -> (AgentDeliveryStatus -> AgentDeliveryStatus) -> CM' () +agentDeliveryStatus acId f = do + ads <- asks agentDeliveryStatuses + atomically $ TM.lookup acId ads >>= mapM_ (\v -> modifyTVar' v f) diff --git a/src/Simplex/Chat/Controller.hs b/src/Simplex/Chat/Controller.hs index 35270a836c..381927f64e 100644 --- a/src/Simplex/Chat/Controller.hs +++ b/src/Simplex/Chat/Controller.hs @@ -207,7 +207,7 @@ data ChatController = ChatController inputQ :: TBQueue String, outputQ :: TBQueue (Maybe CorrId, Maybe RemoteHostId, ChatResponse), connNetworkStatuses :: TMap AgentConnId NetworkStatus, - agentConnStatuses :: TMap AgentConnId (TVar AgentConnStatus), + agentDeliveryStatuses :: TMap AgentConnId (TVar AgentDeliveryStatus), subscriptionMode :: TVar SubscriptionMode, chatLock :: Lock, entityLocks :: TMap ChatLockEntity Lock, @@ -234,14 +234,20 @@ data ChatController = ChatController contactMergeEnabled :: TVar Bool } -data AgentConnStatus = AgentConnStatus +data AgentDeliveryStatus = AgentDeliveryStatus { lastCmd :: UTCTime, - lastCmdTag :: ACommandTag 'Agent 'AEConn, - lastMsg :: Maybe UTCTime, -- no message yet / got an MSG to ack - ackSent :: Maybe UTCTime, - okRcvd :: Maybe UTCTime -- ACK delivered, resulting in OK or MSG + isMSG :: Bool, -- False for RCVD + connId :: Maybe Int64, -- chat connection ID + eventTag :: Maybe Text, -- tshow of ACMEventTag (for JSON instances) + ackSent :: Maybe (UTCTime, Text), -- strEncode of random CorrId + pendingAcks :: Map Text Bool } - deriving (Show) + deriving (Show) -- for ChatResponse + +agentDeliveryOk :: AgentDeliveryStatus -> Bool +agentDeliveryOk AgentDeliveryStatus {ackSent, pendingAcks} = case ackSent of + Nothing -> False + Just (_, corrId) -> M.lookup corrId pendingAcks == Just True && and pendingAcks data HelpSection = HSMain | HSFiles | HSGroups | HSContacts | HSMyAddress | HSIncognito | HSMarkdown | HSMessages | HSRemote | HSSettings | HSDatabase deriving (Show) @@ -498,7 +504,8 @@ data ChatCommand | APIStandaloneFileInfo FileDescriptionURI | QuitChat | ShowVersion - | DebugAcks + | DebugDelivery + -- | DebugConnection Int64 | DebugLocks | DebugEvent ChatResponse | GetAgentStats @@ -746,7 +753,7 @@ data ChatResponse | CRContactPQEnabled {user :: User, contact :: Contact, pqEnabled :: PQEncryption} | CRSQLResult {rows :: [Text]} | CRSlowSQLQueries {chatQueries :: [SlowSQLQuery], agentQueries :: [SlowSQLQuery]} - | CRDebugAcks {debugAcks :: Map DebugAckKey DebugAck} + | CRDebugDelivery {debugDelivery :: Map Text AgentDeliveryStatus} | CRDebugLocks {chatLockName :: Maybe String, chatEntityLocks :: Map String String, agentLocks :: AgentLocks} | CRAgentStats {agentStats :: [[String]]} | CRAgentWorkersDetails {agentWorkersDetails :: AgentWorkersDetails} @@ -1495,7 +1502,11 @@ $(JQ.deriveJSON (sumTypeJSON $ dropPrefix "RCSR") ''RemoteCtrlStopReason) $(JQ.deriveJSON (sumTypeJSON $ dropPrefix "RHSR") ''RemoteHostStopReason) -$(JQ.deriveJSON defaultJSON ''DebugAck) +$(JQ.deriveJSON defaultJSON ''AgentDeliveryStatus) + +-- $(JQ.deriveJSON defaultJSON ''DebugDelivery) + +-- $(JQ.deriveJSON defaultJSON ''DebugConnection) $(JQ.deriveJSON (sumTypeJSON $ dropPrefix "CR") ''ChatResponse) diff --git a/src/Simplex/Chat/View.hs b/src/Simplex/Chat/View.hs index 8a8e68de96..df288149e2 100644 --- a/src/Simplex/Chat/View.hs +++ b/src/Simplex/Chat/View.hs @@ -351,7 +351,7 @@ responseToView hu@(currentRH, user_) ChatConfig {logLevel, showReactions, showRe <> (" :: avg: " <> sShow timeAvg <> " ms") <> (" :: " <> plain (T.unwords $ T.lines query)) in ("Chat queries" : map viewQuery chatQueries) <> [""] <> ("Agent queries" : map viewQuery agentQueries) - CRDebugAcks {debugAcks} -> [plain $ LB.unpack (J.encode debugAcks)] + CRDebugDelivery ads -> [plain $ LB.unpack (J.encode ads)] CRDebugLocks {chatLockName, chatEntityLocks, agentLocks} -> [ maybe "no chat lock" (("chat lock: " <>) . plain) chatLockName, plain $ "chat entity locks: " <> LB.unpack (J.encode chatEntityLocks),