tests: execute in parallel, agent: fix race condition with files and when switching queue

.github/workflows/build.yml

@@ -372,6 +372,14 @@ jobs:
             dist-postgres
           key: ${{ steps.keys.outputs.dist }}-${{ env.opt }}-${{ env.branch }}-${{ github.sha }}
 
+      - name: Increase PostgreSQL connection limit
+        if: matrix.should_run == true && matrix.arch == 'x86-64'
+        shell: bash
+        run: |
+          docker exec ${{ job.services.postgres.id }} psql -U postgres -c "ALTER SYSTEM SET max_connections = 500"
+          docker restart ${{ job.services.postgres.id }}
+          timeout 60 sh -c 'until docker exec ${{ job.services.postgres.id }} pg_isready -U postgres; do sleep 1; done'
+
       - name: Test
         if: matrix.should_run == true && matrix.arch == 'x86-64'
         timeout-minutes: 120

contributing/PROJECT.md

@@ -97,6 +97,14 @@ cabal test --test-option=--match="/SMP client agent/functional API/"
 
 Tests require PostgreSQL running on `localhost:5432` when using `-fserver_postgres` or `-fclient_postgres`.
 
+Tests run in parallel, one job per processor. With 4 jobs, the tests open up to 220 PostgreSQL connections with `-fserver_postgres` and up to 420 with both flags, while the default limit is 100. Raise the limit and restart PostgreSQL:
+
+```bash
+psql -U postgres -c "ALTER SYSTEM SET max_connections = 500"
+```
+
+With more than 4 processors, also limit the jobs: `--test-option=--jobs=4`.
+
 Test files are in `tests/` with structure:
 - `Test.hs`: Main runner
 - `AgentTests/`: Agent protocol and connection tests

plans/2026-10-07-parallel-tests.md

@@ -0,0 +1,80 @@
+# Parallel test execution
+
+Every test gets its own ports, directory and Postgres schemas, so hspec runs
+the tree concurrently in one process with `parallel` and `--jobs`.
+
+## Decisions
+
+- `Util` defines the test environment:
+
+  ```haskell
+  data TestEnv = TestEnv
+    { testNo :: Int,
+      portBase :: Int
+    }
+
+  type HasTestEnv = (?testEnv :: TestEnv)
+  ```
+
+- `Util.it`, `fit` and `xit` take `HasTestEnv => a`. The test wrapper
+  creates the environment for each attempt, so a CI retry gets fresh ports,
+  directory and schemas.
+- Environment creation:
+  - `testNo` from a counter, never reused in a run.
+  - `portBase` from a pool `[10000, 10010 .. 19990]`, returned after the test.
+  - directory `tests/tmp/<testNo>` with `xftp-sender-files` and
+    `xftp-recipient-files`, removed after the test.
+- Fixture constants become `HasTestEnv =>` values:
+  - ports: offsets from `portBase`
+
+    | offset | fixture         |
+    |--------|-----------------|
+    | 1      | `testPort`      |
+    | 2      | `testPort2`     |
+    | 3      | `ntfTestPort`   |
+    | 4      | `ntfTestPort2`  |
+    | 5      | `apnsTestPort`  |
+    | 6      | `xftpTestPort`  |
+    | 7      | `xftpTestPort2` |
+
+  - server addresses: built from the ports.
+  - files: in the test directory.
+  - Postgres schemas: `test<testNo>_<name>`.
+- Functions that use fixtures gain `HasTestEnv` in their signatures.
+- `withXFTPServerCfg` creates the server files directory.
+- Removed hooks:
+  - `tests/tmp` creation and removal per test.
+  - per-test database recreation and schema drops.
+  - XFTP `testBracket` and server files directory hooks.
+  - `around (withMsgStore cfg)`: each test opens its store.
+- `tests/tmp` is cleared before the run and removed after it.
+- Postgres databases are created before the run and dropped after it.
+- Schema dumps and the Ntf CLI test keep the production schema names and
+  recreate the database per test.
+- Sequential, after all parallel tests:
+  - SMP proxy tests that change the number of capabilities.
+  - XFTP CLI tests: `withArgs` and stdout capture are process-wide.
+  - server CLI tests: the same, and fixed ports.
+  - tests with wall-clock assertions:
+    - retry intervals
+    - AUTH error timing
+    - agent user network info
+    - agent SMP queue info
+    - agent client notices
+  - schema dumps.
+- The SQLite agent schema dump resets `tests/tmp` after each item.
+- `createRandomFile_` in the XFTP agent tests writes random bytes directly.
+- Test suite `-with-rtsopts=-N1` becomes `-with-rtsopts=-N`.
+- `it "..." . f $ x` becomes `it "..." $ f $ x` where `f` or `x` uses the
+  environment.
+
+## Mechanics
+
+- `tests/Util.hs`: environment, wrapper, `it`, `fit`, `xit`,
+  `eventuallyRemove`, sender and recipient directories.
+- `tests/Test.hs`: `parallel` tree, sequential tail, `tests/tmp` and database
+  brackets around `hspec`.
+- `tests/SMPClient.hs`, `tests/SMPAgentClient.hs`, `tests/NtfClient.hs`,
+  `tests/XFTPClient.hs`, `tests/Fixtures.hs`: fixtures from the environment.
+- Test modules: signatures, literal paths and ports, removed hooks.
+- `simplexmq.cabal`: `-N`.

simplexmq.cabal

@@ -602,7 +602,7 @@ test-suite simplexmq-test
   default-extensions:
       StrictData
   -- add -fhpc to ghc-options to run tests with coverage
-  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts "-with-rtsopts=-A64M -N1"
+  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts "-with-rtsopts=-A64M -N"
   build-depends:
       base
     , aeson

src/Simplex/FileTransfer/Agent.hs

@@ -279,6 +279,7 @@ retryOnError name loop done e = do
 
 rcvWorkerInternalError :: AgentClient -> DBRcvFileId -> RcvFileId -> Maybe RcvFileId -> Maybe FilePath -> AgentErrorType -> AM ()
 rcvWorkerInternalError c rcvFileId rcvFileEntityId redirectEntityId_ tmpPath err = do
+  liftIO $ throwWhenInactive c
   lift $ forM_ tmpPath (removePath <=< toFSFilePath)
   withStore' c $ \db -> updateRcvFileError db rcvFileId (show err)
   notify c (fromMaybe rcvFileEntityId redirectEntityId_) (RFERR err)
@@ -318,9 +319,6 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
           withStore' c (`updateRcvFileComplete` rcvFileId)
         Just RcvFileRedirect {redirectFileInfo, redirectDbId} -> do
           let RedirectFileInfo {size = redirectSize, digest = redirectDigest} = redirectFileInfo
-          lift $ forM_ tmpPath (removePath <=< toFSFilePath)
-          liftIO $ waitUntilForeground c
-          withStore' c (`updateRcvFileComplete` rcvFileId)
           -- proceed with redirect
           yaml <- liftError (FILE . FILE_IO . show) (CF.readFile $ CryptoFile fsSavePath cfArgs) `allFinally` (lift $ toFSFilePath fsSavePath >>= removePath)
           next@FileDescription {chunks = nextChunks} <- case strDecode (LB.toStrict yaml) of
@@ -330,9 +328,12 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
               | dstSize /= redirectSize -> throwE . FILE $ REDIRECT "size mismatch"
               | dstDigest /= redirectDigest -> throwE . FILE $ REDIRECT "digest mismatch"
               | otherwise -> pure fd
+          liftIO $ waitUntilForeground c
           -- register and download chunks from the actual file
-          withStore c $ \db -> updateRcvFileRedirect db redirectDbId next
+          withStore c $ \db -> updateRcvFileComplete db rcvFileId >> updateRcvFileRedirect db redirectDbId next
           forM_ nextChunks (downloadChunk c)
+          flip catchAllErrors (\e -> logError $ "XFTP rcv worker error: " <> tshow e) $
+            lift $ forM_ tmpPath (removePath <=< toFSFilePath)
       where
         getChunkPaths :: [RcvFileChunk] -> AM [FilePath]
         getChunkPaths [] = pure []
@@ -512,6 +513,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
 
 sndWorkerInternalError :: AgentClient -> DBSndFileId -> SndFileId -> Maybe FilePath -> AgentErrorType -> AM ()
 sndWorkerInternalError c sndFileId sndFileEntityId prefixPath err = do
+  liftIO $ throwWhenInactive c
   lift . forM_ prefixPath $ removePath <=< toFSFilePath
   withStore' c $ \db -> updateSndFileError db sndFileId (show err)
   notify c sndFileEntityId $ SFERR err
@@ -733,6 +735,7 @@ runXFTPDelWorker c srv Worker {doWork} = do
 
 delWorkerInternalError :: AgentClient -> Int64 -> AgentErrorType -> AM ()
 delWorkerInternalError c deletedSndChunkReplicaId e = do
+  liftIO $ throwWhenInactive c
   withStore' c $ \db -> deleteDeletedSndChunkReplica db deletedSndChunkReplicaId
   notify c "" $ SFERR e
 

src/Simplex/Messaging/Agent.hs

@@ -2360,9 +2360,9 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
                     -- we may add more statistics special to queue rotation later on,
                     -- not accounting secure during rotation for now:
                     -- atomically $ incSMPServerStat c userId server connSecured
-                    withStore' c $ \db -> setRcvQueueStatus db rq' Secured
+                    rq1' <- withStore' c $ \db -> setRcvQueueStatus db rq' Secured >> setRcvSwitchStatus db rq1 (Just RSSendingQUSE)
                     void . enqueueMessages c cData sqs SMP.noMsgFlags $ QUSE [((server, sndId), True)]
-                    rq1' <- withStore' c $ \db -> setRcvSwitchStatus db rq1 (Just RSSendingQUSE) <* deleteCommand db cmdId
+                    withStore' c (`deleteCommand` cmdId)
                     let rqs' = updatedQs rq1' rqs
                         conn' = DuplexConnection cData rqs' sqs
                     cStats <- connectionStats c conn'
@@ -4118,15 +4118,15 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                               withStore' c $ \db -> copyPendingSndDeliveries db sq sq2
                               enqueueCommand c "" connId (Just $ qServer sq2) $ AInternalCommand $ ICQSndSecure (snd $ qAddress sq2)
                               pure SSSecuringQueue
-                            else do
-                              let sqInfo' = (sqInfo :: SMPQueueInfo) {queueAddress = queueAddress {dhPublicKey}}
-                              void . enqueueMessages c cData' sqs SMP.noMsgFlags $ QKEY [(sqInfo', C.toPublic sndPrivateKey)]
-                              pure SSSendingQKEY
+                            else pure SSSendingQKEY
                         sq1 <- withStore' c $ \db -> setSndSwitchStatus db sq $ Just swchStatus
                         let sqs'' = updatedQs sq1 sqs' <> [sq2]
                             conn' = DuplexConnection cData' rqs sqs''
                         cStats <- connectionStats c conn'
                         notify $ SWITCH QDSnd SPStarted cStats
+                        when (connAgentVersion < rpcAddressSMPAgentVersion) $ do
+                          let sqInfo' = (sqInfo :: SMPQueueInfo) {queueAddress = queueAddress {dhPublicKey}}
+                          void . enqueueMessages c cData' sqs SMP.noMsgFlags $ QKEY [(sqInfo', C.toPublic sndPrivateKey)]
                       _ -> qError "QADD: won't delete all snd queues in connection"
                   _ -> qError "QADD: replaced queue address is not found in connection"
               _ -> throwE $ AGENT A_VERSION

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql

@@ -1,5 +1,3 @@
-
-
 SET statement_timeout = 0;
 SET lock_timeout = 0;
 SET idle_in_transaction_session_timeout = 0;
@@ -1531,6 +1529,3 @@ ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_queues
 
 ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_queues
     ADD CONSTRAINT snd_queues_host_port_fkey FOREIGN KEY (host, port) REFERENCES smp_agent_test_protocol_schema.servers(host, port) ON UPDATE CASCADE ON DELETE RESTRICT;
-
-
-

src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql

@@ -1,6 +1,3 @@
-
-
-
 SET statement_timeout = 0;
 SET lock_timeout = 0;
 SET idle_in_transaction_session_timeout = 0;
@@ -310,7 +307,3 @@ ALTER TABLE ONLY ntf_server.subscriptions
 
 ALTER TABLE ONLY ntf_server.subscriptions
     ADD CONSTRAINT subscriptions_token_id_fkey FOREIGN KEY (token_id) REFERENCES ntf_server.tokens(token_id) ON UPDATE RESTRICT ON DELETE CASCADE;
-
-
-
-

src/Simplex/Messaging/Server/Main.hs

@@ -719,6 +719,7 @@ importStoreLogToDatabase logPath storeLogFile dbOpts = do
   ps <- newJournalMsgStore logPath $ PQStoreCfg storeCfg
   sCnt <- batchInsertServices services' $ postgresQueueStore ps
   qCnt <- batchInsertQueues @(JournalQueue 'QSMemory) True queues $ postgresQueueStore ps
+  closeMsgStore ps
   renameFile storeLogFile $ storeLogFile <> ".bak"
   pure (sCnt, qCnt)
 
@@ -745,6 +746,7 @@ exportDatabaseToStoreLog logPath dbOpts storeLogFilePath = do
   Sum sCnt <- foldServiceRecs (postgresQueueStore ps) $ \sr -> logNewService sl sr $> Sum (1 :: Int)
   Sum qCnt <- foldQueueRecs True True (postgresQueueStore ps) $ \(rId, qr) -> logCreateQueue sl rId qr $> Sum (1 :: Int)
   closeStoreLog sl
+  closeMsgStore ps
   pure (sCnt, qCnt)
 #endif
 

src/Simplex/Messaging/Server/Names/HttpResolver.hs

@@ -141,4 +141,8 @@ httpGet ResolverEnv {manager, baseUrl, authHdr, timeoutMicro, maxResponseBytes}
       else do
         bs <- brReadSome (responseBody res) (maxResponseBytes + 1)
         pure $ if BL.length bs > fromIntegral maxResponseBytes then Left BodyTooLarge else Right bs
-  pure (either (Left . HttpFailure) id result)
+  pure (either (Left . httpError) id result)
+  where
+    httpError = \case
+      HC.HttpExceptionRequest _ HC.ResponseTimeout -> ResolverTimeout
+      e -> HttpFailure e

src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql

@@ -1,6 +1,3 @@
-
-
-
 SET statement_timeout = 0;
 SET lock_timeout = 0;
 SET idle_in_transaction_session_timeout = 0;
@@ -552,7 +549,3 @@ ALTER TABLE ONLY smp_server.msg_queues
 
 ALTER TABLE ONLY smp_server.msg_queues
     ADD CONSTRAINT msg_queues_rcv_service_id_fkey FOREIGN KEY (rcv_service_id) REFERENCES smp_server.services(service_id) ON UPDATE RESTRICT ON DELETE SET NULL;
-
-
-
-

tests/AgentTests.hs

@@ -19,17 +19,12 @@ import Simplex.Messaging.Server.Env.STM (AStoreType (..))
 import Simplex.Messaging.Transport (ASrvTransport)
 import Test.Hspec hiding (fit, it)
 
-#if defined(dbPostgres)
-import Fixtures
-import Simplex.Messaging.Agent.Store.Postgres.Util (dropAllSchemasExceptSystem)
-#else
+#if !defined(dbPostgres)
 import AgentTests.SQLiteTests (storeTests)
 #endif
 
 #if defined(dbServerPostgres)
 import AgentTests.NotificationTests (notificationTests)
-import SMPClient (postgressBracket)
-import NtfClient (ntfTestServerDBConnectInfo)
 #endif
 
 agentCoreTests :: Spec
@@ -42,16 +37,10 @@ agentCoreTests = do
 
 agentTests :: (ASrvTransport, AStoreType) -> Spec
 agentTests ps = do
-#if defined(dbPostgres)
-  after_ (dropAllSchemasExceptSystem testDBConnectInfo) $ do
-#else
-  do
-#endif
-    describe "Functional API" $ functionalAPITests ps
-    describe "Chosen servers" serverChoiceTests
+  describe "Functional API" $ functionalAPITests ps
+  describe "Chosen servers" serverChoiceTests
 #if defined(dbServerPostgres)
-    around_ (postgressBracket ntfTestServerDBConnectInfo) $
-      describe "Notification tests" $ notificationTests ps
+  describe "Notification tests" $ notificationTests ps
 #endif
 #if !defined(dbPostgres)
   describe "SQLite store" storeTests