xftp: agent error handling (#675)

- chunk download doesn't loop on permanent errors
- decryption errors are considered permanent - local worker doesn't retry
- update replica retries; to do - consider use for this field, or remove it
- rcv file Error status - to prevent repeat reads of chunks for download, files for decryption; also plan to use it for filtering on cleanup
- error string saved in separate field for debugging (not part of status type)
- agent event for rcv file errors

src/Simplex/FileTransfer/Agent.hs

@@ -13,10 +13,11 @@ module Simplex.FileTransfer.Agent
     receiveFile,
     addXFTPWorker,
     -- Sending files
-    sendFile,
+    _sendFile,
   )
 where
 
+import Control.Logger.Simple (logError)
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Reader
@@ -31,7 +32,7 @@ import Simplex.FileTransfer.Types
 import Simplex.FileTransfer.Util (uniqueCombine)
 import Simplex.Messaging.Agent.Client
 import Simplex.Messaging.Agent.Env.SQLite
-import Simplex.Messaging.Agent.Protocol (ACommand (FRCVD), AParty (..), AgentErrorType (INTERNAL))
+import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite
@@ -39,7 +40,7 @@ import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Protocol (XFTPServer)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (whenM)
+import Simplex.Messaging.Util (tshow, whenM)
 import UnliftIO
 import UnliftIO.Directory
 import qualified UnliftIO.Exception as E
@@ -77,31 +78,44 @@ runXFTPWorker c srv doWork = do
     void . atomically $ readTMVar doWork
     agentOperationBracket c AORcvNetwork throwWhenInactive runXftpOperation
   where
+    noWorkToDo = void . atomically $ tryTakeTMVar doWork
     runXftpOperation :: m ()
     runXftpOperation = do
       nextChunk <- withStore' c (`getNextRcvChunkToDownload` srv)
       case nextChunk of
         Nothing -> noWorkToDo
-        Just fc@RcvFileChunk {rcvChunkId, delay} -> do
+        Just RcvFileChunk {rcvFileId, replicas = []} -> workerInternalError c rcvFileId "chunk has no replicas"
+        Just fc@RcvFileChunk {rcvFileId, rcvChunkId, delay, replicas = replica@RcvFileChunkReplica {rcvChunkReplicaId} : _} -> do
           ri <- asks $ reconnectInterval . config
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
           withRetryInterval ri' $ \delay' loop ->
-            downloadFileChunk fc
-              `catchError` \_ -> do
-                withStore' c $ \db -> updateRcvFileChunkDelay db rcvChunkId delay'
-                -- TODO don't loop on permanent errors
-                -- TODO increase replica retries count
-                loop
-    noWorkToDo = void . atomically $ tryTakeTMVar doWork
-    downloadFileChunk :: RcvFileChunk -> m ()
-    downloadFileChunk RcvFileChunk {userId, rcvFileId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath, replicas = replica : _} = do
+            downloadFileChunk fc replica
+              `catchError` retryOnError delay' loop (workerInternalError c rcvFileId . show)
+          where
+            retryOnError :: Int -> m () -> (AgentErrorType -> m ()) -> AgentErrorType -> m ()
+            retryOnError chunkDelay loop done e = do
+              logError $ "XFTP worker error: " <> tshow e
+              if temporaryAgentError e
+                then retryLoop
+                else done e
+              where
+                retryLoop = do
+                  withStore' c $ \db -> do
+                    updateRcvFileChunkDelay db rcvChunkId chunkDelay
+                    increaseRcvChunkReplicaRetries db rcvChunkReplicaId
+                  atomically $ endAgentOperation c AORcvNetwork
+                  atomically $ throwWhenInactive c
+                  atomically $ beginAgentOperation c AORcvNetwork
+                  loop
+    downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> m ()
+    downloadFileChunk RcvFileChunk {userId, rcvFileId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath} replica = do
       chunkPath <- uniqueCombine fileTmpPath $ show chunkNo
       let chunkSpec = XFTPRcvChunkSpec chunkPath (unFileSize chunkSize) (unFileDigest digest)
       agentXFTPDownloadChunk c userId replica chunkSpec
       fileReceived <- withStore c $ \db -> runExceptT $ do
         -- both actions can be done in a single store method
-        fd <- ExceptT $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId rcvFileId chunkPath
-        let fileReceived = allChunksReceived fd
+        f <- ExceptT $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId rcvFileId chunkPath
+        let fileReceived = allChunksReceived f
         when fileReceived $
           liftIO $ updateRcvFileStatus db rcvFileId RFSReceived
         pure fileReceived
@@ -113,7 +127,14 @@ runXFTPWorker c srv doWork = do
         allChunksReceived :: RcvFile -> Bool
         allChunksReceived RcvFile {chunks} =
           all (\RcvFileChunk {replicas} -> any received replicas) chunks
-    downloadFileChunk _ = throwError $ INTERNAL "no replica"
+
+workerInternalError :: AgentMonad m => AgentClient -> RcvFileId -> String -> m ()
+workerInternalError c rcvFileId internalErrStr = do
+  withStore' c $ \db -> updateRcvFileError db rcvFileId internalErrStr
+  notifyInternalError c rcvFileId internalErrStr
+
+notifyInternalError :: (MonadUnliftIO m) => AgentClient -> RcvFileId -> String -> m ()
+notifyInternalError AgentClient {subQ} rcvFileId internalErrStr = atomically $ writeTBQueue subQ ("", "", FRCVERR rcvFileId $ INTERNAL internalErrStr)
 
 runXFTPLocalWorker :: forall m. AgentMonad m => AgentClient -> TMVar () -> m ()
 runXFTPLocalWorker c@AgentClient {subQ} doWork = do
@@ -126,14 +147,8 @@ runXFTPLocalWorker c@AgentClient {subQ} doWork = do
       nextFile <- withStore' c getNextRcvFileToDecrypt
       case nextFile of
         Nothing -> noWorkToDo
-        Just fd -> do
-          ri <- asks $ reconnectInterval . config
-          withRetryInterval ri $ \_ loop ->
-            decryptFile fd
-              `catchError` \_ -> do
-                -- TODO don't loop on permanent errors
-                -- TODO fixed number of retries instead of exponential backoff?
-                loop
+        Just f@RcvFile {rcvFileId} ->
+          decryptFile f `catchError` (workerInternalError c rcvFileId . show)
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     decryptFile :: RcvFile -> m ()
     decryptFile RcvFile {rcvFileId, key, nonce, tmpPath, saveDir, chunks} = do
@@ -183,15 +198,16 @@ runXFTPLocalWorker c@AgentClient {subQ} doWork = do
             )
             LB.empty
 
-sendFile :: AgentMonad m => AgentClient -> UserId -> FilePath -> FilePath -> m Int64
-sendFile c userId xftpPath filePath = do
+-- _sendFile :: AgentMonad m => AgentClient -> UserId -> FilePath -> FilePath -> m Int64
+_sendFile :: AgentClient -> UserId -> FilePath -> FilePath -> m Int64
+_sendFile _c _userId _xftpPath _filePath = do
   -- db: create file in status New without chunks
   -- add local snd worker for encryption
   -- return file id to client
   undefined
 
-runXFTPSndLocalWorker :: forall m. AgentMonad m => AgentClient -> TMVar () -> m ()
-runXFTPSndLocalWorker c@AgentClient {subQ} doWork = do
+_runXFTPSndLocalWorker :: forall m. AgentMonad m => AgentClient -> TMVar () -> m ()
+_runXFTPSndLocalWorker _c doWork = do
   forever $ do
     void . atomically $ readTMVar doWork
     runXftpOperation
@@ -202,8 +218,8 @@ runXFTPSndLocalWorker c@AgentClient {subQ} doWork = do
       -- ? (or Encrypted to retry create? - see below)
       -- with fixed retries (?) encryptFile
       undefined
-    encryptFile :: SndFile -> m ()
-    encryptFile sndFile = do
+    _encryptFile :: SndFile -> m ()
+    _encryptFile _sndFile = do
       -- if enc path exists, remove it
       -- if enc path doesn't exist:
       --   - choose enc path
@@ -225,8 +241,8 @@ runXFTPSndLocalWorker c@AgentClient {subQ} doWork = do
       -- ? and add XFTP snd workers for uploading chunks.
       undefined
 
-runXFTPSndWorker :: forall m. AgentMonad m => AgentClient -> XFTPServer -> TMVar () -> m ()
-runXFTPSndWorker c srv doWork = do
+_runXFTPSndWorker :: forall m. AgentMonad m => AgentClient -> XFTPServer -> TMVar () -> m ()
+_runXFTPSndWorker c _srv doWork = do
   forever $ do
     void . atomically $ readTMVar doWork
     agentOperationBracket c AOSndNetwork throwWhenInactive runXftpOperation
@@ -238,8 +254,8 @@ runXFTPSndWorker c srv doWork = do
       --   - with fixed retries, repeat N times:
       --     check if other files are in upload, delay (see xftpSndFiles in XFTPAgent)
       undefined
-    uploadFileChunk :: SndFileChunk -> m ()
-    uploadFileChunk sndFileChunk = do
+    _uploadFileChunk :: SndFileChunk -> m ()
+    _uploadFileChunk _sndFileChunk = do
       -- add file id to xftpSndFiles
       -- XFTP upload chunk
       -- db: update replica status to Uploaded, return SndFile

src/Simplex/FileTransfer/Types.hs

@@ -51,12 +51,12 @@ data RcvFile = RcvFile
   }
   deriving (Eq, Show)
 
--- TODO add error status?
 data RcvFileStatus
   = RFSReceiving
   | RFSReceived
   | RFSDecrypting
   | RFSComplete
+  | RFSError
   deriving (Eq, Show)
 
 instance FromField RcvFileStatus where fromField = fromTextField_ textDecode
@@ -69,12 +69,14 @@ instance TextEncoding RcvFileStatus where
     "received" -> Just RFSReceived
     "decrypting" -> Just RFSDecrypting
     "complete" -> Just RFSComplete
+    "error" -> Just RFSError
     _ -> Nothing
   textEncode = \case
     RFSReceiving -> "receiving"
     RFSReceived -> "received"
     RFSDecrypting -> "decrypting"
     RFSComplete -> "complete"
+    RFSError -> "error"
 
 data RcvFileChunk = RcvFileChunk
   { userId :: Int64,
@@ -96,7 +98,7 @@ data RcvFileChunkReplica = RcvFileChunkReplica
     replicaId :: ChunkReplicaId,
     replicaKey :: C.APrivateSignKey,
     received :: Bool,
-    acknowledged :: Bool,
+    -- acknowledged :: Bool,
     retries :: Int
   }
   deriving (Eq, Show)