diff --git a/src/Simplex/FileTransfer/Agent.hs b/src/Simplex/FileTransfer/Agent.hs index 890966888..426539b1d 100644 --- a/src/Simplex/FileTransfer/Agent.hs +++ b/src/Simplex/FileTransfer/Agent.hs @@ -63,6 +63,7 @@ import Simplex.Messaging.Agent.Client import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.RetryInterval +import Simplex.Messaging.Agent.Stats import Simplex.Messaging.Agent.Store.SQLite import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB import qualified Simplex.Messaging.Crypto as C @@ -184,6 +185,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do liftIO $ waitForUserNetwork c + atomically $ incXFTPServerStat c userId srv replDownloadAttempts 1 downloadFileChunk fc replica approvedRelays `catchAgentError` \e -> retryOnError "XFTP rcv worker" (retryLoop loop e delay') (retryDone e) e where @@ -194,7 +196,9 @@ runXFTPRcvWorker c srv Worker {doWork} = do withStore' c $ \db -> updateRcvChunkReplicaDelay db rcvChunkReplicaId replicaDelay atomically $ assertAgentForeground c loop - retryDone = rcvWorkerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) + retryDone e = do + atomically $ incXFTPServerStat c userId srv replDownloadErr 1 + rcvWorkerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) e downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> Bool -> AM () downloadFileChunk RcvFileChunk {userId, rcvFileId, rcvFileEntityId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath} replica approvedRelays = do unlessM ((approvedRelays ||) <$> ipAddressProtected') $ throwE $ FILE NOT_APPROVED @@ -214,6 +218,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do Just RcvFileRedirect {redirectFileInfo = RedirectFileInfo {size = FileSize finalSize}, redirectEntityId} -> (redirectEntityId, finalSize) liftIO . when complete $ updateRcvFileStatus db rcvFileId RFSReceived pure (entityId, complete, RFPROG rcvd total) + atomically $ incXFTPServerStat c userId srv replDownload 1 notify c entityId progress when complete . lift . void $ getXFTPRcvWorker True c Nothing @@ -484,6 +489,7 @@ runXFTPSndWorker c srv Worker {doWork} = do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do liftIO $ waitForUserNetwork c + atomically $ incXFTPServerStat c userId srv replUploadAttempts 1 uploadFileChunk cfg fc replica `catchAgentError` \e -> retryOnError "XFTP snd worker" (retryLoop loop e delay') (retryDone e) e where @@ -494,7 +500,9 @@ runXFTPSndWorker c srv Worker {doWork} = do withStore' c $ \db -> updateSndChunkReplicaDelay db sndChunkReplicaId replicaDelay atomically $ assertAgentForeground c loop - retryDone = sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) + retryDone e = do + atomically $ incXFTPServerStat c userId srv replUploadErr 1 + sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) e uploadFileChunk :: AgentConfig -> SndFileChunk -> SndFileChunkReplica -> AM () uploadFileChunk AgentConfig {xftpMaxRecipientsPerRequest = maxRecipients} sndFileChunk@SndFileChunk {sndFileId, userId, chunkSpec = chunkSpec@XFTPChunkSpec {filePath}, digest = chunkDigest} replica = do replica'@SndFileChunkReplica {sndChunkReplicaId} <- addRecipients sndFileChunk replica @@ -510,6 +518,7 @@ runXFTPSndWorker c srv Worker {doWork} = do let uploaded = uploadedSize chunks total = totalSize chunks complete = all chunkUploaded chunks + atomically $ incXFTPServerStat c userId srv replUpload 1 notify c sndFileEntityId $ SFPROG uploaded total when complete $ do (sndDescr, rcvDescrs) <- sndFileToDescrs sf @@ -651,6 +660,7 @@ runXFTPDelWorker c srv Worker {doWork} = do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do liftIO $ waitForUserNetwork c + atomically $ incXFTPServerStat c userId srv replDeleteAttempts 1 deleteChunkReplica `catchAgentError` \e -> retryOnError "XFTP del worker" (retryLoop loop e delay') (retryDone e) e where @@ -661,10 +671,13 @@ runXFTPDelWorker c srv Worker {doWork} = do withStore' c $ \db -> updateDeletedSndChunkReplicaDelay db deletedSndChunkReplicaId replicaDelay atomically $ assertAgentForeground c loop - retryDone = delWorkerInternalError c deletedSndChunkReplicaId + retryDone e = do + atomically $ incXFTPServerStat c userId srv replDeleteErr 1 + delWorkerInternalError c deletedSndChunkReplicaId e deleteChunkReplica = do agentXFTPDeleteChunk c userId replica withStore' c $ \db -> deleteDeletedSndChunkReplica db deletedSndChunkReplicaId + atomically $ incXFTPServerStat c userId srv replDelete 1 delWorkerInternalError :: AgentClient -> Int64 -> AgentErrorType -> AM () delWorkerInternalError c deletedSndChunkReplicaId e = do diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 8b9532f6d..2ba36e484 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -140,6 +140,7 @@ module Simplex.Messaging.Agent.Client withUserServers, withNextSrv, incSMPServerStat, + incXFTPServerStat, AgentWorkersDetails (..), getAgentWorkersDetails, AgentWorkersSummary (..), @@ -1952,6 +1953,15 @@ incSMPServerStat AgentClient {smpServersStats} userId srv sel n = do modifyTVar' (sel newStats) (+ n) TM.insert (userId, srv) newStats smpServersStats +incXFTPServerStat :: AgentClient -> UserId -> XFTPServer -> (AgentXFTPServerStats -> TVar Int) -> Int -> STM () +incXFTPServerStat AgentClient {xftpServersStats} userId srv sel n = do + TM.lookup (userId, srv) xftpServersStats >>= \case + Just v -> modifyTVar' (sel v) (+ n) + Nothing -> do + newStats <- newAgentXFTPServerStats + modifyTVar' (sel newStats) (+ n) + TM.insert (userId, srv) newStats xftpServersStats + -- - currently used servers - those that have state -- - previously used servers - have stats but no state - they could be used earlier in session, -- or in previous sessions and stats for them were restored