Merge branch 'master' into ep/pow

src/Simplex/FileTransfer/Agent.hs

@@ -17,11 +17,16 @@ module Simplex.FileTransfer.Agent
     toFSFilePath,
     -- Receiving files
     xftpReceiveFile',
+    xftpPrepareReceiveFile',
+    xftpStartReceiveFile',
     xftpDeleteRcvFile',
     xftpDeleteRcvFiles',
     -- Sending files
     xftpSendFile',
     xftpSendDescription',
+    xftpPrepareSendFile',
+    xftpPrepareSendDescription',
+    xftpStartSendFile',
     deleteSndFileInternal,
     deleteSndFilesInternal,
     deleteSndFileRemote,
@@ -127,7 +132,12 @@ closeXFTPAgent a = do
     stopWorkers workers = atomically (swapTVar workers M.empty) >>= mapM_ (liftIO . cancelWorker)
 
 xftpReceiveFile' :: AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe CryptoFileArgs -> Bool -> AM RcvFileId
-xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks, redirect}) cfArgs approvedRelays = do
+xftpReceiveFile' c userId vfd cfArgs approvedRelays = do
+  fId <- xftpPrepareReceiveFile' c userId vfd cfArgs approvedRelays
+  fId <$ xftpStartReceiveFile' c fId
+
+xftpPrepareReceiveFile' :: AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe CryptoFileArgs -> Bool -> AM RcvFileId
+xftpPrepareReceiveFile' c userId (ValidFileDescription fd@FileDescription {redirect}) cfArgs approvedRelays = do
   g <- asks random
   prefixPath <- lift $ getPrefixPath "rcv.xftp"
   createDirectory prefixPath
@@ -137,7 +147,7 @@ xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks, redi
   lift $ createDirectory =<< toFSFilePath relTmpPath
   lift $ createEmptyFile =<< toFSFilePath relSavePath
   let saveFile = CryptoFile relSavePath cfArgs
-  fId <- case redirect of
+  case redirect of
     Nothing -> withStore c $ \db -> createRcvFile db g userId fd relPrefixPath relTmpPath saveFile approvedRelays
     Just _ -> do
       -- prepare description paths
@@ -149,8 +159,11 @@ xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks, redi
       let saveFileRedirect = CryptoFile relSavePathRedirect $ Just cfArgsRedirect
       -- create download tasks
       withStore c $ \db -> createRcvFileRedirect db g userId fd relPrefixPath relTmpPathRedirect saveFileRedirect relTmpPath saveFile approvedRelays
-  forM_ chunks (downloadChunk c)
-  pure fId
+
+xftpStartReceiveFile' :: AgentClient -> RcvFileId -> AM ()
+xftpStartReceiveFile' c rcvFileEntityId = do
+  srvs <- withStore c (`startPreparedRcvFile` rcvFileEntityId)
+  lift $ forM_ srvs $ void . getXFTPRcvWorker True c . Just
 
 downloadChunk :: AgentClient -> FileChunk -> AM ()
 downloadChunk c FileChunk {replicas = (FileChunkReplica {server} : _)} = do
@@ -353,6 +366,11 @@ notify c entId cmd = atomically $ writeTBQueue (subQ c) ("", entId, AEvt (sAEnti
 
 xftpSendFile' :: AgentClient -> UserId -> CryptoFile -> Int -> Maybe Word32 -> AM SndFileId
 xftpSendFile' c userId file numRecipients storageHours = do
+  fId <- xftpPrepareSendFile' c userId file numRecipients storageHours
+  fId <$ xftpStartSendFile' c fId
+
+xftpPrepareSendFile' :: AgentClient -> UserId -> CryptoFile -> Int -> Maybe Word32 -> AM SndFileId
+xftpPrepareSendFile' c userId file numRecipients storageHours = do
   g <- asks random
   prefixPath <- lift $ getPrefixPath "snd.xftp"
   createDirectory prefixPath
@@ -360,12 +378,15 @@ xftpSendFile' c userId file numRecipients storageHours = do
   key <- atomically $ C.randomSbKey g
   nonce <- atomically $ C.randomCbNonce g
   -- saving absolute filePath will not allow to restore file encryption after app update, but it's a short window
-  fId <- withStore c $ \db -> createSndFile db g userId file numRecipients relPrefixPath key nonce Nothing storageHours
-  lift . void $ getXFTPSndWorker True c Nothing
-  pure fId
+  withStore c $ \db -> createSndFile db g userId file numRecipients relPrefixPath key nonce Nothing storageHours
 
 xftpSendDescription' :: AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Int -> AM SndFileId
-xftpSendDescription' c userId (ValidFileDescription fdDirect@FileDescription {size, digest}) numRecipients = do
+xftpSendDescription' c userId vfd numRecipients = do
+  fId <- xftpPrepareSendDescription' c userId vfd numRecipients
+  fId <$ xftpStartSendFile' c fId
+
+xftpPrepareSendDescription' :: AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Int -> AM SndFileId
+xftpPrepareSendDescription' c userId (ValidFileDescription fdDirect@FileDescription {size, digest}) numRecipients = do
   g <- asks random
   prefixPath <- lift $ getPrefixPath "snd.xftp"
   createDirectory prefixPath
@@ -376,9 +397,12 @@ xftpSendDescription' c userId (ValidFileDescription fdDirect@FileDescription {si
   liftError (FILE . FILE_IO . show) $ CF.writeFile file (LB.fromStrict $ strEncode fdDirect)
   key <- atomically $ C.randomSbKey g
   nonce <- atomically $ C.randomCbNonce g
-  fId <- withStore c $ \db -> createSndFile db g userId file numRecipients relPrefixPath key nonce (Just RedirectFileInfo {size, digest}) Nothing
+  withStore c $ \db -> createSndFile db g userId file numRecipients relPrefixPath key nonce (Just RedirectFileInfo {size, digest}) Nothing
+
+xftpStartSendFile' :: AgentClient -> SndFileId -> AM ()
+xftpStartSendFile' c sndFileEntityId = do
+  withStore c (`startPreparedSndFile` sndFileEntityId)
   lift . void $ getXFTPSndWorker True c Nothing
-  pure fId
 
 resumeXFTPSndWork :: AgentClient -> Maybe XFTPServer -> AM' ()
 resumeXFTPSndWork = void .: getXFTPSndWorker False

src/Simplex/FileTransfer/Types.hs

@@ -93,7 +93,8 @@ data RcvFile = RcvFile
   deriving (Show)
 
 data RcvFileStatus
-  = RFSReceiving
+  = RFSPrepared
+  | RFSReceiving
   | RFSReceived
   | RFSDecrypting
   | RFSComplete
@@ -106,6 +107,7 @@ instance ToField RcvFileStatus where toField = toField . textEncode
 
 instance TextEncoding RcvFileStatus where
   textDecode = \case
+    "prepared" -> Just RFSPrepared
     "receiving" -> Just RFSReceiving
     "received" -> Just RFSReceived
     "decrypting" -> Just RFSDecrypting
@@ -113,6 +115,7 @@ instance TextEncoding RcvFileStatus where
     "error" -> Just RFSError
     _ -> Nothing
   textEncode = \case
+    RFSPrepared -> "prepared"
     RFSReceiving -> "receiving"
     RFSReceived -> "received"
     RFSDecrypting -> "decrypting"
@@ -177,7 +180,8 @@ sndFileEncPath :: FilePath -> FilePath
 sndFileEncPath prefixPath = prefixPath </> "xftp.encrypted"
 
 data SndFileStatus
-  = SFSNew -- db record created
+  = SFSPrepared
+  | SFSNew -- db record created
   | SFSEncrypting -- encryption started
   | SFSEncrypted -- encryption complete
   | SFSUploading -- all chunk replicas are created on servers
@@ -191,6 +195,7 @@ instance ToField SndFileStatus where toField = toField . textEncode
 
 instance TextEncoding SndFileStatus where
   textDecode = \case
+    "prepared" -> Just SFSPrepared
     "new" -> Just SFSNew
     "encrypting" -> Just SFSEncrypting
     "encrypted" -> Just SFSEncrypted
@@ -199,6 +204,7 @@ instance TextEncoding SndFileStatus where
     "error" -> Just SFSError
     _ -> Nothing
   textEncode = \case
+    SFSPrepared -> "prepared"
     SFSNew -> "new"
     SFSEncrypting -> "encrypting"
     SFSEncrypted -> "encrypted"

src/Simplex/Messaging/Agent.hs

@@ -127,10 +127,15 @@ module Simplex.Messaging.Agent
     xftpStartWorkers,
     xftpStartSndWorkers,
     xftpReceiveFile,
+    xftpPrepareReceiveFile,
+    xftpStartReceiveFile,
     xftpDeleteRcvFile,
     xftpDeleteRcvFiles,
     xftpSendFile,
     xftpSendDescription,
+    xftpPrepareSendFile,
+    xftpPrepareSendDescription,
+    xftpStartSendFile,
     xftpDeleteSndFileInternal,
     xftpDeleteSndFilesInternal,
     xftpDeleteSndFileRemote,
@@ -190,7 +195,7 @@ import Data.Time.Clock
 import Data.Time.Clock.System (systemToUTCTime)
 import Data.Traversable (mapAccumL)
 import Data.Word (Word16, Word32)
-import Simplex.FileTransfer.Agent (closeXFTPAgent, deleteSndFileInternal, deleteSndFileRemote, deleteSndFilesInternal, deleteSndFilesRemote, startXFTPSndWorkers, startXFTPWorkers, toFSFilePath, xftpDeleteRcvFile', xftpDeleteRcvFiles', xftpReceiveFile', xftpSendDescription', xftpSendFile')
+import Simplex.FileTransfer.Agent (closeXFTPAgent, deleteSndFileInternal, deleteSndFileRemote, deleteSndFilesInternal, deleteSndFilesRemote, startXFTPSndWorkers, startXFTPWorkers, toFSFilePath, xftpDeleteRcvFile', xftpDeleteRcvFiles', xftpPrepareReceiveFile', xftpPrepareSendDescription', xftpPrepareSendFile', xftpReceiveFile', xftpSendDescription', xftpSendFile', xftpStartReceiveFile', xftpStartSendFile')
 import Simplex.FileTransfer.Description (ValidFileDescription)
 import Simplex.FileTransfer.Protocol (FileParty (..))
 import Simplex.FileTransfer.Types (RcvFileId, SndFileId)
@@ -769,6 +774,14 @@ xftpReceiveFile :: AgentClient -> UserId -> ValidFileDescription 'FRecipient ->
 xftpReceiveFile c = withAgentEnv c .:: xftpReceiveFile' c
 {-# INLINE xftpReceiveFile #-}
 
+xftpPrepareReceiveFile :: AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe CryptoFileArgs -> Bool -> AE RcvFileId
+xftpPrepareReceiveFile c = withAgentEnv c .:: xftpPrepareReceiveFile' c
+{-# INLINE xftpPrepareReceiveFile #-}
+
+xftpStartReceiveFile :: AgentClient -> RcvFileId -> AE ()
+xftpStartReceiveFile c = withAgentEnv c . xftpStartReceiveFile' c
+{-# INLINE xftpStartReceiveFile #-}
+
 -- | Delete XFTP rcv file (deletes work files from file system and db records)
 xftpDeleteRcvFile :: AgentClient -> RcvFileId -> IO ()
 xftpDeleteRcvFile c = withAgentEnv' c . xftpDeleteRcvFile' c
@@ -789,6 +802,18 @@ xftpSendDescription :: AgentClient -> UserId -> ValidFileDescription 'FRecipient
 xftpSendDescription c = withAgentEnv c .:. xftpSendDescription' c
 {-# INLINE xftpSendDescription #-}
 
+xftpPrepareSendFile :: AgentClient -> UserId -> CryptoFile -> Int -> Maybe Word32 -> AE SndFileId
+xftpPrepareSendFile c = withAgentEnv c .:: xftpPrepareSendFile' c
+{-# INLINE xftpPrepareSendFile #-}
+
+xftpPrepareSendDescription :: AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Int -> AE SndFileId
+xftpPrepareSendDescription c = withAgentEnv c .:. xftpPrepareSendDescription' c
+{-# INLINE xftpPrepareSendDescription #-}
+
+xftpStartSendFile :: AgentClient -> SndFileId -> AE ()
+xftpStartSendFile c = withAgentEnv c . xftpStartSendFile' c
+{-# INLINE xftpStartSendFile #-}
+
 -- | Delete XFTP snd file internally (deletes work files from file system and db records)
 xftpDeleteSndFileInternal :: AgentClient -> SndFileId -> IO ()
 xftpDeleteSndFileInternal c = withAgentEnv' c . deleteSndFileInternal c
@@ -2143,20 +2168,21 @@ data CommandCompletion = CCMoved | CCCompleted
 
 runCommandProcessing :: AgentClient -> ConnId -> Maybe SMPServer -> Worker -> AM ()
 runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
-  ri <- asks $ messageRetryInterval . config -- different retry interval?
+  AgentConfig {messageRetryInterval = RetryInterval2 {riFast}, commandQuotaRetryInterval} <- asks config
+  let ri = RetryInterval2 {riSlow = commandQuotaRetryInterval, riFast}
   forever $ do
     atomically $ endAgentOperation c AOSndNetwork
     lift $ waitForWork doWork
     liftIO $ throwWhenInactive c
     atomically $ beginAgentOperation c AOSndNetwork
-    withWork c doWork (\db -> getPendingServerCommand db connId server_) $ runProcessCmd (riFast ri)
+    withWork c doWork (\db -> getPendingServerCommand db connId server_) $ runProcessCmd ri
   where
     runProcessCmd ri cmd = do
       pending <- newTVarIO []
       processCmd ri cmd pending
       mapM_ (atomically . writeTBQueue subQ) . reverse =<< readTVarIO pending
-    processCmd :: RetryInterval -> PendingCommand -> TVar [ATransmission] -> AM ()
-    processCmd ri PendingCommand {cmdId, corrId, userId, command} pendingCmds = case command of
+    processCmd :: RetryInterval2 -> PendingCommand -> TVar [ATransmission] -> AM ()
+    processCmd ri PendingCommand {cmdId, corrId, userId, command, createdAt} pendingCmds = case command of
       AClientCommand cmd -> case cmd of
         NEW enableNtfs (ACM cMode) pqEnc subMode useDR -> noServer $ do
           triedHosts <- newTVarIO S.empty
@@ -2351,12 +2377,18 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
             SomeConn _ conn@DuplexConnection {} -> a conn
             _ -> internalErr "command requires duplex connection"
         tryCommand action = tryMoveableCommand (action $> CCCompleted)
-        tryMoveableCommand action = withRetryInterval ri $ \_ loop -> do
+        tryMoveableCommand action = withRetryInterval2 ri $ \_ loop -> do
           liftIO $ waitWhileSuspended c
           liftIO $ waitForUserNetwork c
           tryAllErrors action >>= \case
+            Left e@(SMP _ SMP.QUOTA) -> do
+              AgentConfig {quotaExceededTimeout} <- asks config
+              expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime
+              if createdAt < expireTs
+                then cmdError e
+                else retrySndOp c $ loop RISlow
             Left e
-              | temporaryOrHostError e -> retrySndOp c loop
+              | temporaryOrHostError e -> retrySndOp c $ loop RIFast
               | otherwise -> cmdError e
             Right CCCompleted -> withStore' c (`deleteCommand` cmdId)
             Right CCMoved -> pure () -- command processing moved to another command queue

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -162,6 +162,7 @@ data AgentConfig = AgentConfig
     connDeleteDeliveryTimeout :: NominalDiffTime,
     helloTimeout :: NominalDiffTime,
     quotaExceededTimeout :: NominalDiffTime,
+    commandQuotaRetryInterval :: RetryInterval,
     persistErrorInterval :: NominalDiffTime,
     initialCleanupDelay :: Int64,
     cleanupInterval :: Int64,
@@ -241,6 +242,7 @@ defaultAgentConfig =
       connDeleteDeliveryTimeout = 2 * nominalDay,
       helloTimeout = 2 * nominalDay,
       quotaExceededTimeout = 7 * nominalDay,
+      commandQuotaRetryInterval = RetryInterval {initialInterval = 30_000000, increaseAfter = 0, maxInterval = 3600_000000},
       persistErrorInterval = 3, -- seconds
       initialCleanupDelay = 30 * 1000000, -- 30 seconds
       cleanupInterval = 5 * 60 * 1000000, -- 5 minutes

src/Simplex/Messaging/Agent/RetryInterval.hs

@@ -10,6 +10,7 @@ module Simplex.Messaging.Agent.RetryInterval
     withRetryInterval,
     withRetryIntervalCount,
     withRetryForeground,
+    withRetryInterval2,
     withRetryLock2,
     updateRetryInterval2,
     nextRetryDelay,
@@ -86,9 +87,24 @@ withRetryForeground ri isForeground isOnline action = callAction 0 $ initialInte
                 | otherwise = (elapsed + delay, nextRetryDelay elapsed' delay ri)
           callAction elapsed' delay'
 
+withRetryInterval2 :: forall m. MonadIO m => RetryInterval2 -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
+withRetryInterval2 = withRetryWait2 $ liftIO . threadDelay'
+
 -- This function allows action to toggle between slow and fast retry intervals.
 withRetryLock2 :: forall m. MonadIO m => RetryInterval2 -> TMVar () -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
-withRetryLock2 RetryInterval2 {riSlow, riFast} lock action =
+withRetryLock2 ri lock = withRetryWait2 wait ri
+  where
+    wait delay = do
+      waiting <- newTVarIO True
+      _ <- liftIO . forkIO $ do
+        threadDelay' delay
+        atomically $ whenM (readTVar waiting) $ void $ tryPutTMVar lock ()
+      atomically $ do
+        takeTMVar lock
+        writeTVar waiting False
+
+withRetryWait2 :: forall m. Monad m => (Int64 -> m ()) -> RetryInterval2 -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
+withRetryWait2 wait RetryInterval2 {riSlow, riFast} action =
   callAction (0, initialInterval riSlow) (0, initialInterval riFast)
   where
     callAction :: (Int64, Int64) -> (Int64, Int64) -> m ()
@@ -102,14 +118,6 @@ withRetryLock2 RetryInterval2 {riSlow, riFast} lock action =
           let elapsed' = elapsed + delay
               delay' = nextRetryDelay elapsed' delay ri
           call (elapsed', delay')
-        wait delay = do
-          waiting <- newTVarIO True
-          _ <- liftIO . forkIO $ do
-            threadDelay' delay
-            atomically $ whenM (readTVar waiting) $ void $ tryPutTMVar lock ()
-          atomically $ do
-            takeTMVar lock
-            writeTVar waiting False
 
 nextRetryDelay :: Int64 -> Int64 -> RetryInterval -> Int64
 nextRetryDelay elapsed delay RetryInterval {increaseAfter, maxInterval} =

src/Simplex/Messaging/Agent/Store.hs

@@ -489,7 +489,8 @@ data PendingCommand = PendingCommand
     corrId :: ACorrId,
     userId :: UserId,
     connId :: ConnId,
-    command :: AgentCommand
+    command :: AgentCommand,
+    createdAt :: UTCTime
   }
 
 data AgentCmdType = ACClient | ACInternal

src/Simplex/Messaging/Agent/Store/AgentStore.hs

@@ -215,6 +215,7 @@ module Simplex.Messaging.Agent.Store.AgentStore
     -- Rcv files
     createRcvFile,
     createRcvFileRedirect,
+    startPreparedRcvFile,
     lockRcvFileForUpdate,
     getRcvFile,
     getRcvFileByEntityId,
@@ -236,6 +237,7 @@ module Simplex.Messaging.Agent.Store.AgentStore
     getRcvFilesExpired,
     -- Snd files
     createSndFile,
+    startPreparedSndFile,
     lockSndFileForUpdate,
     getSndFile,
     getSndFileByEntityId,
@@ -1716,7 +1718,7 @@ getPendingServerCommand db connId srv_ = getWorkItem "command" getCmdId getComma
         DB.query
           db
           [sql|
-            SELECT c.corr_id, cs.user_id, c.command
+            SELECT c.corr_id, cs.user_id, c.command, c.created_at
             FROM commands c
             JOIN connections cs USING (conn_id)
             WHERE c.command_id = ?
@@ -1724,7 +1726,7 @@ getPendingServerCommand db connId srv_ = getWorkItem "command" getCmdId getComma
           (Only cmdId)
       where
         err = SEInternal $ "command  " <> bshow cmdId <> " returned []"
-        pendingCommand (corrId, userId, command) = PendingCommand {cmdId, corrId, userId, connId, command}
+        pendingCommand (corrId, userId, command, createdAt) = PendingCommand {cmdId, corrId, userId, connId, command, createdAt}
     markCommandFailed cmdId = DB.execute db "UPDATE commands SET failed = 1 WHERE command_id = ?" (Only cmdId)
 
 updateCommandServer :: DB.Connection -> AsyncCmdId -> SMPServer -> IO (Either StoreError ())
@@ -3163,6 +3165,29 @@ createRcvFileRedirect db gVar userId redirectFd@FileDescription {chunks = redire
           chunks = []
         }
 
+startPreparedRcvFile :: DB.Connection -> RcvFileId -> IO (Either StoreError [XFTPServer])
+startPreparedRcvFile db rcvFileEntityId = runExceptT $ do
+  rcvFileId <- ExceptT $ getRcvFileIdByEntityId_ db rcvFileEntityId
+  liftIO $ do
+    updatedAt <- getCurrentTime
+    DB.execute
+      db
+      "UPDATE rcv_files SET status = ?, updated_at = ? WHERE (rcv_file_id = ? OR redirect_id = ?) AND status = ?"
+      (RFSReceiving, updatedAt, rcvFileId, rcvFileId, RFSPrepared)
+    map toXFTPServer
+      <$> DB.query
+        db
+        [sql|
+          SELECT DISTINCT
+            s.xftp_host, s.xftp_port, s.xftp_key_hash
+          FROM rcv_file_chunk_replicas r
+          JOIN xftp_servers s ON s.xftp_server_id = r.xftp_server_id
+          JOIN rcv_file_chunks c ON c.rcv_file_chunk_id = r.rcv_file_chunk_id
+          JOIN rcv_files f ON f.rcv_file_id = c.rcv_file_id
+          WHERE (f.rcv_file_id = ? OR f.redirect_id = ?) AND r.replica_number = 1
+        |]
+        (rcvFileId, rcvFileId)
+
 insertRcvFile :: DB.Connection -> TVar ChaChaDRG -> UserId -> FileDescription 'FRecipient -> FilePath -> FilePath -> CryptoFile -> Maybe DBRcvFileId -> Maybe RcvFileId -> Bool -> IO (Either StoreError (RcvFileId, DBRcvFileId))
 insertRcvFile db gVar userId FileDescription {size, digest, key, nonce, chunkSize, redirect} prefixPath tmpPath (CryptoFile savePath cfArgs) redirectId_ redirectEntityId_ approvedRelays = runExceptT $ do
   let (redirectDigest_, redirectSize_) = case redirect of
@@ -3173,7 +3198,7 @@ insertRcvFile db gVar userId FileDescription {size, digest, key, nonce, chunkSiz
       DB.execute
         db
         "INSERT INTO rcv_files (rcv_file_entity_id, user_id, size, digest, key, nonce, chunk_size, prefix_path, tmp_path, save_path, save_file_key, save_file_nonce, status, redirect_id, redirect_entity_id, redirect_digest, redirect_size, approved_relays) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
-        ((Binary rcvFileEntityId, userId, size, digest, key, nonce, chunkSize, prefixPath, tmpPath) :. (savePath, fileKey <$> cfArgs, fileNonce <$> cfArgs, RFSReceiving, redirectId_, Binary <$> redirectEntityId_, redirectDigest_, redirectSize_, BI approvedRelays))
+        ((Binary rcvFileEntityId, userId, size, digest, key, nonce, chunkSize, prefixPath, tmpPath) :. (savePath, fileKey <$> cfArgs, fileNonce <$> cfArgs, RFSPrepared, redirectId_, Binary <$> redirectEntityId_, redirectDigest_, redirectSize_, BI approvedRelays))
   rcvFileId <- liftIO $ insertedRowId db
   pure (rcvFileEntityId, rcvFileId)
 
@@ -3478,13 +3503,23 @@ createSndFile db gVar userId (CryptoFile path cfArgs) numRecipients prefixPath k
     DB.execute
       db
       "INSERT INTO snd_files (snd_file_entity_id, user_id, path, src_file_key, src_file_nonce, num_recipients, prefix_path, key, nonce, status, redirect_size, redirect_digest, storage_time) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)"
-      ((Binary sndFileEntityId, userId, path, fileKey <$> cfArgs, fileNonce <$> cfArgs, numRecipients) :. (prefixPath, key, nonce, SFSNew, redirectSize_, redirectDigest_, storageHours))
+      ((Binary sndFileEntityId, userId, path, fileKey <$> cfArgs, fileNonce <$> cfArgs, numRecipients) :. (prefixPath, key, nonce, SFSPrepared, redirectSize_, redirectDigest_, storageHours))
   where
     (redirectSize_, redirectDigest_) =
       case redirect_ of
         Nothing -> (Nothing, Nothing)
         Just RedirectFileInfo {size, digest} -> (Just size, Just digest)
 
+startPreparedSndFile :: DB.Connection -> SndFileId -> IO (Either StoreError ())
+startPreparedSndFile db sndFileEntityId = runExceptT $ do
+  sndFileId <- ExceptT $ getSndFileIdByEntityId_ db sndFileEntityId
+  liftIO $ do
+    updatedAt <- getCurrentTime
+    DB.execute
+      db
+      "UPDATE snd_files SET status = ?, updated_at = ? WHERE snd_file_id = ? AND status = ?"
+      (SFSNew, updatedAt, sndFileId, SFSPrepared)
+
 getSndFileByEntityId :: DB.Connection -> SndFileId -> IO (Either StoreError SndFile)
 getSndFileByEntityId db sndFileEntityId = runExceptT $ do
   sndFileId <- ExceptT $ getSndFileIdByEntityId_ db sndFileEntityId

tests/AgentTests/FunctionalAPITests.hs

@@ -444,6 +444,9 @@ functionalAPITests ps = do
       it "should expire multiple messages" $ testExpireManyMessages ps
       it "should expire one message if quota is exceeded" $ testExpireMessageQuota ps
       it "should expire multiple messages if quota is exceeded" $ testExpireManyMessagesQuota ps
+    describe "joining a full contact address" $ do
+      it "should retry joining until the address has space" $ testJoinFullContactAsync ps
+      it "should fail joining after quota exceeded timeout" $ testJoinFullContactAsyncExpire ps
     it "should drop message after too many receive attempts" $ testDropMsgAfterRcvAttempts ps
 #if !defined(dbPostgres)
     -- TODO [postgres] restore from outdated db backup (we use copyFile/renameFile for sqlite)
@@ -2495,6 +2498,49 @@ testExpireManyMessagesQuota (t, msType) = withSmpServerConfigOn t cfg' testPort
   where
     cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1, maxJournalMsgCount = 2}
 
+testJoinFullContactAsync :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testJoinFullContactAsync (t, msType) = withSmpServerConfigOn t cfg' testPort $ \_ -> do
+  (contactId, qInfo) <- fillContactAddress
+  withAgent 2 agentCfg {commandQuotaRetryInterval = fastRetryInterval} initAgentServers testDB2 $ \bob -> do
+    aliceId <- runRight $ do
+      (aliceId, _) <- A.prepareConnectionToJoin bob 1 True qInfo PQSupportOn
+      A.joinConnectionAsync bob "2" False aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
+      pure aliceId
+    threadDelay 500000
+    noMessages bob "joining should be retried while the address is full"
+    withAgent 1 agentCfg initAgentServers testDB $ \alice -> runRight_ $ do
+      subscribeConnection alice contactId
+      ("", _, A.REQ _ _ _ "carol's connInfo" _ _) <- get alice
+      get bob =##> \case ("2", c, JOINED False) -> c == aliceId; _ -> False
+      ("", _, A.REQ _ _ _ "bob's connInfo" _ _) <- get alice
+      pure ()
+  where
+    cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1, maxJournalMsgCount = 2}
+
+testJoinFullContactAsyncExpire :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testJoinFullContactAsyncExpire (t, msType) = withSmpServerConfigOn t cfg' testPort $ \_ -> do
+  (_, qInfo) <- fillContactAddress
+  withAgent 2 agentCfg {quotaExceededTimeout = 2, commandQuotaRetryInterval = fastRetryInterval} initAgentServers testDB2 $ \bob -> do
+    aliceId <- runRight $ do
+      (aliceId, _) <- A.prepareConnectionToJoin bob 1 True qInfo PQSupportOn
+      A.joinConnectionAsync bob "2" False aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
+      pure aliceId
+    threadDelay 500000
+    noMessages bob "joining should be retried until quota exceeded timeout"
+    get bob =##> \case ("2", c, ERR (SMP _ QUOTA)) -> c == aliceId; _ -> False
+  where
+    cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1, maxJournalMsgCount = 2}
+
+fillContactAddress :: HasCallStack => IO (ConnId, ConnectionRequestUri 'CMContact)
+fillContactAddress = do
+  (contactId, qInfo) <- withAgent 1 agentCfg initAgentServers testDB $ \alice -> runRight $ do
+    (contactId, CCLink qInfo Nothing) <- A.createConnection alice NRMInteractive 1 True True SCMContact Nothing Nothing IKPQOn False SMSubscribe
+    pure (contactId, qInfo)
+  withAgent 3 agentCfg initAgentServers testDB3 $ \carol -> runRight_ $ do
+    (aliceId, _) <- A.prepareConnectionToJoin carol 1 True qInfo PQSupportOn
+    void $ A.joinConnection carol NRMInteractive 1 aliceId True qInfo "carol's connInfo" PQSupportOn SMSubscribe
+  pure (contactId, qInfo)
+
 testDropMsgAfterRcvAttempts :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testDropMsgAfterRcvAttempts ps =
   withSmpServerStoreLogOn ps testPort $ \_ -> do
@@ -4639,8 +4685,7 @@ testServerMultipleIdentities =
         testE2ERatchetParams12
 
 testWaitForUserNetwork :: IO ()
-testWaitForUserNetwork = do
-  a <- getSMPAgentClient' 1 aCfg initAgentServers testDB
+testWaitForUserNetwork = withAgent 1 aCfg initAgentServers testDB $ \a -> do
   noNetworkDelay a
   setUserNetworkInfo a $ UserNetworkInfo UNNone False
   networkDelay a 100000
@@ -4657,8 +4702,7 @@ testWaitForUserNetwork = do
     aCfg = agentCfg {userNetworkInterval = 100000, userOfflineDelay = 0}
 
 testDoNotResetOnlineToOffline :: IO ()
-testDoNotResetOnlineToOffline = do
-  a <- getSMPAgentClient' 1 aCfg initAgentServers testDB
+testDoNotResetOnlineToOffline = withAgent 1 aCfg initAgentServers testDB $ \a -> do
   noNetworkDelay a
   setUserNetworkInfo a $ UserNetworkInfo UNWifi False
   networkDelay a 100000
@@ -4679,8 +4723,7 @@ testDoNotResetOnlineToOffline = do
     aCfg = agentCfg {userNetworkInterval = 100000, userOfflineDelay = 0.1}
 
 testResumeMultipleThreads :: IO ()
-testResumeMultipleThreads = do
-  a <- getSMPAgentClient' 1 aCfg initAgentServers testDB
+testResumeMultipleThreads = withAgent 1 aCfg initAgentServers testDB $ \a -> do
   noNetworkDelay a
   setUserNetworkInfo a $ UserNetworkInfo UNNone False
   vs <-

tests/AgentTests/SQLiteTests.hs

@@ -743,9 +743,11 @@ testGetNextRcvChunkToDownload st = do
   withTransaction st $ \db -> do
     Right Nothing <- getNextRcvChunkToDownload db xftpServer1 86400
 
-    Right _ <- createRcvFile db g 1 rcvFileDescr1 "filepath" "filepath" (CryptoFile "filepath" Nothing) True
+    Right fId1 <- createRcvFile db g 1 rcvFileDescr1 "filepath" "filepath" (CryptoFile "filepath" Nothing) True
+    Right _ <- startPreparedRcvFile db fId1
     DB.execute_ db "UPDATE rcv_file_chunk_replicas SET replica_key = cast('bad' as blob) WHERE rcv_file_chunk_replica_id = 1"
     Right fId2 <- createRcvFile db g 1 rcvFileDescr1 "filepath" "filepath" (CryptoFile "filepath" Nothing) True
+    Right _ <- startPreparedRcvFile db fId2
 
     Left e <- getNextRcvChunkToDownload db xftpServer1 86400
     show e `shouldContain` "ConversionFailed"
@@ -783,6 +785,7 @@ testGetNextSndFileToPrepare st = do
     -- Right _ <- createSndFile db g 1 (CryptoFile "filepath" Nothing) 1 "filepath" testFileSbKey testFileCbNonce Nothing
     -- DB.execute_ db "UPDATE snd_files SET status = 'new', num_recipients = 'bad' WHERE snd_file_id = 1"
     Right fId2 <- createSndFile db g 1 (CryptoFile "filepath" Nothing) 1 "filepath" testFileSbKey testFileCbNonce Nothing Nothing
+    Right _ <- startPreparedSndFile db fId2
     DB.execute_ db "UPDATE snd_files SET status = 'new' WHERE snd_file_id = 2"
 
     -- Left e <- getNextSndFileToPrepare db 86400

tests/AgentTests/ServerChoice.hs

@@ -71,9 +71,7 @@ initServers =
     }
 
 testChooseDifferentOperator :: IO ()
-testChooseDifferentOperator = do
-  c <- getSMPAgentClient' 1 agentCfg initServers testDB
-  runRight_ $ do
+testChooseDifferentOperator = withAgent 1 agentCfg initServers testDB $ \c -> runRight_ $ do
     -- chooses the only operator with storage role
     srv1 <- withAgentEnv c $ getNextServer c 1 storageSrvs []
     liftIO $ srv1 == testOp1Srv1 || srv1 == testOp1Srv2 `shouldBe` True

tests/CoreTests/RetryIntervalTests.hs

@@ -14,8 +14,11 @@ import Util
 retryIntervalTests :: Spec
 retryIntervalTests = do
   describe "Retry interval with 2 modes and lock" $ do
-    testRetryIntervalSameMode
-    testRetryIntervalSwitchMode
+    testRetryIntervalSameMode withRetryNewLock2
+    testRetryIntervalSwitchMode withRetryNewLock2
+  describe "Retry interval with 2 modes" $ do
+    testRetryIntervalSameMode withRetryInterval2
+    testRetryIntervalSwitchMode withRetryInterval2
   describe "Foreground retry interval" $ do
     testRetryForeground
     testRetryToBackground
@@ -41,14 +44,18 @@ testFastRI =
       maxInterval = 40000
     }
 
-testRetryIntervalSameMode :: Spec
-testRetryIntervalSameMode =
+type WithRetry2 = RetryInterval2 -> (RI2State -> (RetryIntervalMode -> IO ()) -> IO ()) -> IO ()
+
+withRetryNewLock2 :: WithRetry2
+withRetryNewLock2 ri action = newEmptyTMVarIO >>= \lock -> withRetryLock2 ri lock action
+
+testRetryIntervalSameMode :: WithRetry2 -> Spec
+testRetryIntervalSameMode withRetry =
   it "should increase elapased time and interval when the mode stays the same" $ do
-    lock <- newEmptyTMVarIO
     intervals <- newTVarIO []
     reportedIntervals <- newTVarIO []
     ts <- newTVarIO =<< getCurrentTime
-    withRetryLock2 testRI lock $ \(RI2State slow fast) loop -> do
+    withRetry testRI $ \(RI2State slow fast) loop -> do
       ints <- addInterval intervals ts
       atomically $ modifyTVar' reportedIntervals ((slow, fast) :)
       when (length ints < 9) $ loop RIFast
@@ -65,14 +72,13 @@ testRetryIntervalSameMode =
                        (20000, 40000)
                      ]
 
-testRetryIntervalSwitchMode :: Spec
-testRetryIntervalSwitchMode =
+testRetryIntervalSwitchMode :: WithRetry2 -> Spec
+testRetryIntervalSwitchMode withRetry =
   it "should increase elapased time and interval when the mode switches" $ do
-    lock <- newEmptyTMVarIO
     intervals <- newTVarIO []
     reportedIntervals <- newTVarIO []
     ts <- newTVarIO =<< getCurrentTime
-    withRetryLock2 testRI lock $ \(RI2State slow fast) loop -> do
+    withRetry testRI $ \(RI2State slow fast) loop -> do
       ints <- addInterval intervals ts
       atomically $ modifyTVar' reportedIntervals ((slow, fast) :)
       when (length ints < 11) $ loop $ if length ints <= 5 then RIFast else RISlow

tests/XFTPAgent.hs

@@ -34,7 +34,7 @@ import Simplex.FileTransfer.Server.Env (AFStoreType, XFTPServerConfig (..), defa
 import Simplex.FileTransfer.Server.Store (STMFileStore)
 import Simplex.FileTransfer.Transport (XFTPErrorType (AUTH))
 import Simplex.FileTransfer.Types (RcvFileId, SndFileId)
-import Simplex.Messaging.Agent (AgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpDeleteSndFileRemote, xftpReceiveFile, xftpSendDescription, xftpStartWorkers)
+import Simplex.Messaging.Agent (AgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpDeleteSndFileRemote, xftpPrepareReceiveFile, xftpPrepareSendFile, xftpReceiveFile, xftpSendDescription, xftpStartReceiveFile, xftpStartSendFile, xftpStartWorkers)
 import qualified Simplex.Messaging.Agent as XA
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..))
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, xftpCfg)
@@ -84,6 +84,8 @@ xftpAgentTests =
       it "should send and receive with encrypted local files" testXFTPAgentSendReceiveEncrypted
       it "should send and receive large file with a redirect" testXFTPAgentSendReceiveRedirect
       it "should send and receive small file without a redirect" testXFTPAgentSendReceiveNoRedirect
+      it "should receive prepared file only after it is started" testXFTPAgentPrepareReceive
+      it "should send prepared file only after it is started" testXFTPAgentPrepareSend
       it "should extend storage time with an entitlement proof and report the granted expiry" $ \_ -> testXFTPAgentEntitlement
       describe "sending and receiving with version negotiation" $ beforeWith (const (pure ())) testXFTPAgentSendReceiveMatrix
       it "should resume receiving file after restart" $ \_ -> testXFTPAgentReceiveRestore
@@ -278,6 +280,56 @@ testXFTPAgentSendReceiveNoRedirect = withXFTPServer $ do
       inBytes <- B.readFile filePathIn
       B.readFile out `shouldReturn` inBytes
 
+testXFTPAgentPrepareReceive :: HasCallStack => AFStoreType -> IO ()
+testXFTPAgentPrepareReceive = withXFTPServer $ do
+  filePath <- createRandomFile
+  (_, _, rfd1, rfd2) <- withAgent 1 agentCfg initAgentServers testDB $ \sndr -> runRight $ testSend sndr filePath
+  rfId2 <- withAgent 2 agentCfg initAgentServers testDB2 $ \rcp -> runRight $ do
+    xftpStartWorkers rcp (Just recipientFiles)
+    rfId1 <- xftpReceiveFile rcp 1 rfd1 Nothing True
+    rfId2 <- xftpPrepareReceiveFile rcp 1 rfd2 Nothing True
+    rfProgress rcp $ mb 18
+    ("", rfId1', RFDONE _) <- rfGet rcp
+    liftIO $ do
+      rfId1' `shouldBe` rfId1
+      timeout 300000 (rfGet rcp) `shouldReturn` Nothing
+    pure rfId2
+  withAgent 3 agentCfg initAgentServers testDB2 $ \rcp' -> runRight_ $ do
+    xftpStartWorkers rcp' (Just recipientFiles)
+    liftIO $ timeout 300000 (rfGet rcp') `shouldReturn` Nothing
+    xftpStartReceiveFile rcp' rfId2
+    rfProgress rcp' $ mb 18
+    ("", rfId2', RFDONE path) <- rfGet rcp'
+    liftIO $ do
+      rfId2' `shouldBe` rfId2
+      file <- B.readFile filePath
+      B.readFile path `shouldReturn` file
+
+testXFTPAgentPrepareSend :: HasCallStack => AFStoreType -> IO ()
+testXFTPAgentPrepareSend = withXFTPServer $ do
+  filePath1 <- createRandomFile' "testfile1"
+  filePath2 <- createRandomFile' "testfile2"
+  sfId2 <- withAgent 1 agentCfg initAgentServers testDB $ \sndr -> runRight $ do
+    xftpStartWorkers sndr (Just senderFiles)
+    sfId1 <- xftpSendFile sndr 1 (CF.plain filePath1) 1
+    sfId2 <- xftpPrepareSendFile sndr 1 (CF.plain filePath2) 1 Nothing
+    sfProgress sndr $ mb 18
+    ("", sfId1', SFDONE _ _) <- sfGet sndr
+    liftIO $ do
+      sfId1' `shouldBe` sfId1
+      timeout 300000 (sfGet sndr) `shouldReturn` Nothing
+    pure sfId2
+  rfd <- withAgent 2 agentCfg initAgentServers testDB $ \sndr' -> runRight $ do
+    xftpStartWorkers sndr' (Just senderFiles)
+    liftIO $ timeout 300000 (sfGet sndr') `shouldReturn` Nothing
+    xftpStartSendFile sndr' sfId2
+    sfProgress sndr' $ mb 18
+    ("", sfId2', SFDONE _ [rfd]) <- sfGet sndr'
+    liftIO $ sfId2' `shouldBe` sfId2
+    pure rfd
+  withAgent 3 agentCfg initAgentServers testDB2 $ \rcp ->
+    runRight_ . void $ testReceive rcp rfd filePath2
+
 testXFTPAgentSendReceiveMatrix :: Spec
 testXFTPAgentSendReceiveMatrix = do
   describe "old server" $ do