Merge branch 'master' into rcv-services

* agent: use primary key index in setRcvServiceAssocs

Previous WHERE rcv_id = ? did not match the (host, port, rcv_id)
primary key prefix and fell back to a table scan via
idx_rcv_queues_client_notice_id. With ~390k rows per queue, each
update in a 1350-row batch scanned the whole table, yielding ~290s
per batch and a multi-hour rcv-services migration.

* agent: pass SMPServer explicitly to setRcvServiceAssocs

Avoid extracting host/port from the first queue inside setRcvServiceAssocs.
The caller already has SMPServer in scope (from tSess) and the call chain
is short, so threading it through is simpler than inspecting the list.
Removes the empty-list guard from setRcvServiceAssocs (it remains in
processRcvServiceAssocs).

simplexmq.cabal

@@ -1,7 +1,7 @@
 cabal-version: 1.12
 
 name:           simplexmq
-version:        6.5.1.0
+version:        6.5.2.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/Messaging/Agent.hs

@@ -3106,7 +3106,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
   unless (null connIds) $ do
     notify' "" $ UP srv connIds
     atomically $ incSMPServerStat' c userId srv connSubscribed $ length connIds
-  readTVarIO serviceRQs >>= processRcvServiceAssocs c
+  readTVarIO serviceRQs >>= processRcvServiceAssocs c srv
   where
     withRcvConn :: SMP.RecipientId -> (forall c. RcvQueue -> Connection c -> AM ()) -> AM' ()
     withRcvConn rId a = do

src/Simplex/Messaging/Agent/Client.hs

@@ -223,7 +223,7 @@ import Data.Set (Set)
 import qualified Data.Set as S
 import Data.Text (Text)
 import Data.Text.Encoding
-import Data.Time (UTCTime, addUTCTime, defaultTimeLocale, diffUTCTime, formatTime, getCurrentTime)
+import Data.Time (UTCTime, addUTCTime, defaultTimeLocale, formatTime, getCurrentTime)
 import Data.Time.Clock.System (getSystemTime)
 import Data.Word (Word16)
 import qualified Data.X509.Validation as XV
@@ -1677,18 +1677,13 @@ subscribeSessQueues_ c withEvents qs = sendClientBatch_ "SUB" False subscribe_ c
     subscribe_ :: SMPClient -> NonEmpty RcvQueueSub -> IO (BatchResponses RcvQueueSub SMPClientError (Maybe ServiceId), Bool)
     subscribe_ smp qs' = do
       let (userId, srv, _) = tSess
-          n = length qs'
-      atomically $ incSMPServerStat' c userId srv connSubAttempts n
-      tStart <- getCurrentTime
-      tNet0 <- getCurrentTime
+      atomically $ incSMPServerStat' c userId srv connSubAttempts $ length qs'
       rs <- sendBatch (\smp' _ -> subscribeSMPQueues smp') smp NRMBackground qs'
-      tNet1 <- getCurrentTime
-      let (okN, permErrN, tempErrN) = countSubResults rs
       cs_ <-
         if withEvents
           then Just . S.fromList . map qConnId . M.elems <$> atomically (SS.getActiveSubs tSess $ currentSubs c)
           else pure Nothing
-      (active, svcN, tDb0, tDb1) <- E.uninterruptibleMask_ $ do
+      active <- E.uninterruptibleMask_ $ do
         (active, (serviceQs, notices)) <- atomically $ do
           r@(_, (_, notices)) <- ifM
             (activeClientSession c tSess sessId)
@@ -1696,16 +1691,12 @@ subscribeSessQueues_ c withEvents qs = sendClientBatch_ "SUB" False subscribe_ c
             ((False, ([], [])) <$ incSMPServerStat' c userId srv connSubIgnored (length rs))
           unless (null notices) $ takeTMVar $ clientNoticesLock c
           pure r
-        tDb0' <- getCurrentTime
         unless (null serviceQs) $ void $
-          processRcvServiceAssocs c serviceQs `runReaderT` agentEnv c
-        tDb1' <- getCurrentTime
+          processRcvServiceAssocs c srv serviceQs `runReaderT` agentEnv c
         unless (null notices) $ void $
           (processClientNotices c tSess notices `runReaderT` agentEnv c)
             `E.finally` atomically (putTMVar (clientNoticesLock c) ())
-        pure (active, length serviceQs, tDb0', tDb1')
-      tEnd <- getCurrentTime
-      logSubBatchTiming c srv n okN permErrN tempErrN svcN active tStart tNet0 tNet1 tDb0 tDb1 tEnd
+        pure active
       forM_ cs_ $ \cs -> do
         let (errs, okConns) = partitionEithers $ map (\(RcvQueueSub {connId}, r) -> bimap (connId,) (const connId) r) $ L.toList rs
             conns = filter (`S.notMember` cs) okConns
@@ -1723,10 +1714,10 @@ subscribeSessQueues_ c withEvents qs = sendClientBatch_ "SUB" False subscribe_ c
         tSess = transportSession' smp
         sessId = sessionId $ thParams smp
 
-processRcvServiceAssocs :: SMPQueue q => AgentClient -> [q] -> AM' ()
-processRcvServiceAssocs _ [] = pure ()
-processRcvServiceAssocs c serviceQs =
-  withStore' c (`setRcvServiceAssocs` serviceQs) `catchAllErrors'` \e -> do
+processRcvServiceAssocs :: SMPQueue q => AgentClient -> SMPServer -> [q] -> AM' ()
+processRcvServiceAssocs _ _ [] = pure ()
+processRcvServiceAssocs c srv serviceQs =
+  withStore' c (\db -> setRcvServiceAssocs db srv serviceQs) `catchAllErrors'` \e -> do
     logError $ "processRcvServiceAssocs error: " <> tshow e
     notifySub' c "" $ ERR e
 
@@ -1909,51 +1900,6 @@ logServer' :: MonadIO m => ByteString -> AgentClient -> ProtocolServer s -> Byte
 logServer' dir AgentClient {clientId} srv qStr cmdStr =
   logInfo . decodeUtf8 $ B.unwords ["A", "(" <> bshow clientId <> ")", dir, showServer srv, ":", logSecret' qStr, cmdStr]
 
-countSubResults :: NonEmpty (q, Either SMPClientError r) -> (Int, Int, Int)
-countSubResults = foldl' f (0, 0, 0) . L.toList
-  where
-    f (ok, perm, temp) (_, Right _) = (ok + 1, perm, temp)
-    f (ok, perm, temp) (_, Left e)
-      | temporaryClientError e = (ok, perm, temp + 1)
-      | otherwise = (ok, perm + 1, temp)
-
-logSubBatchTiming ::
-  AgentClient ->
-  SMPServer ->
-  Int -> -- batch size
-  Int -> -- ok
-  Int -> -- permanent errors
-  Int -> -- temporary errors
-  Int -> -- service assoc rows
-  Bool -> -- active session (false = replaced/ignored)
-  UTCTime -> -- tStart
-  UTCTime -> -- tNet0 (before sendBatch)
-  UTCTime -> -- tNet1 (after sendBatch)
-  UTCTime -> -- tDb0 (before processRcvServiceAssocs)
-  UTCTime -> -- tDb1 (after processRcvServiceAssocs)
-  UTCTime -> -- tEnd
-  IO ()
-logSubBatchTiming AgentClient {clientId} srv n okN permErrN tempErrN svcN active tStart tNet0 tNet1 tDb0 tDb1 tEnd =
-  logInfo . decodeUtf8 $ B.unwords
-    [ "A",
-      "(" <> bshow clientId <> ")",
-      "SUB-TIMING",
-      showServer srv,
-      "n=" <> bshow n,
-      "ok=" <> bshow okN,
-      "perm=" <> bshow permErrN,
-      "temp=" <> bshow tempErrN,
-      "svc=" <> bshow svcN,
-      "replaced=" <> (if active then "false" else "true"),
-      "net_ms=" <> bshow (ms tNet0 tNet1),
-      "db_ms=" <> bshow (ms tDb0 tDb1),
-      "other_ms=" <> bshow (ms tStart tEnd - ms tNet0 tNet1 - ms tDb0 tDb1),
-      "total_ms=" <> bshow (ms tStart tEnd)
-    ]
-  where
-    ms :: UTCTime -> UTCTime -> Int
-    ms t0 t1 = round (realToFrac (diffUTCTime t1 t0) * 1000 :: Double)
-
 showServer :: ProtocolServer s -> ByteString
 showServer ProtocolServer {host, port} =
   strEncode host <> B.pack (if null port then "" else ':' : port)

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

@@ -2399,12 +2399,18 @@ unassocUserServerRcvQueueSubs' db userId srv@(SMPServer h p kh) = do
 unsetQueuesToSubscribe :: DB.Connection -> IO ()
 unsetQueuesToSubscribe db = DB.execute_ db "UPDATE rcv_queues SET to_subscribe = 0 WHERE to_subscribe = 1"
 
-setRcvServiceAssocs :: SMPQueue q => DB.Connection -> [q] -> IO ()
-setRcvServiceAssocs db rqs = do
+setRcvServiceAssocs :: SMPQueue q => DB.Connection -> SMPServer -> [q] -> IO ()
+setRcvServiceAssocs db ProtocolServer {host, port} rqs =
 #if defined(dbPostgres)
-  DB.execute db "UPDATE rcv_queues SET rcv_service_assoc = 1 WHERE rcv_id IN ?" $ Only $ In (map queueId rqs)
+  DB.execute
+    db
+    "UPDATE rcv_queues SET rcv_service_assoc = 1 WHERE host = ? AND port = ? AND rcv_id IN ?"
+    (host, port, In (map queueId rqs))
 #else
-  DB.executeMany db "UPDATE rcv_queues SET rcv_service_assoc = 1 WHERE rcv_id = ?" $ map (Only . queueId) rqs
+  DB.executeMany
+    db
+    "UPDATE rcv_queues SET rcv_service_assoc = 1 WHERE host = ? AND port = ? AND rcv_id = ?"
+    (map (\q -> (host, port, queueId q)) rqs)
 #endif
 
 removeRcvServiceAssocs :: DB.Connection -> UserId -> SMPServer -> IO ()

src/Simplex/Messaging/Notifications/Server/Push/APNS.hs

@@ -346,7 +346,7 @@ apnsPushProviderClient c@APNSPushClient {nonceDrg, apnsCfg} tkn@NtfTknRec {token
   nonce <- atomically $ C.randomCbNonce nonceDrg
   apnsNtf <- liftEither $ first PPCryptoError $ apnsNotification tkn nonce (paddedNtfLength apnsCfg) pn
   req <- liftIO $ apnsRequest c tknStr apnsNtf
-  HTTP2Response {response, respBody = HTTP2Body {bodyHead}} <- liftHTTPS2 $ sendRequest http2 req Nothing
+  HTTP2Response {response, respBody = HTTP2Body {bodyHead}} <- liftHTTPS2 $ sendRequestDirect http2 req Nothing
   let status = H.responseStatus response
       reason' = maybe "" reason $ J.decodeStrict' bodyHead
   if status == Just N.ok200