mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-27 22:34:59 +00:00
ntf: batch ntf worker command processing
This commit is contained in:
spaced4ndy
@@ -70,6 +70,7 @@ module Simplex.Messaging.Agent.Client
|
||||
agentNtfDeleteToken,
|
||||
agentNtfEnableCron,
|
||||
agentNtfCreateSubscription,
|
||||
agentNtfCreateSubscriptions,
|
||||
agentNtfCheckSubscription,
|
||||
agentNtfDeleteSubscription,
|
||||
agentXFTPDownloadChunk,
|
||||
@@ -1730,6 +1731,28 @@ agentNtfCreateSubscription :: AgentClient -> NtfTokenId -> NtfToken -> SMPQueueN
|
||||
agentNtfCreateSubscription c tknId NtfToken {ntfServer, ntfPrivKey} smpQueue nKey =
|
||||
withNtfClient c ntfServer tknId "SNEW" $ \ntf -> ntfCreateSubscription ntf ntfPrivKey (NewNtfSub tknId smpQueue nKey)
|
||||
|
||||
type CreateSubNtfReq = (NtfSubscription, ClientNtfCreds)
|
||||
|
||||
agentNtfCreateSubscriptions :: AgentClient -> NtfTokenId -> NtfToken -> [CreateSubNtfReq] -> AM' [(CreateSubNtfReq, Either AgentErrorType NtfSubscriptionId)]
|
||||
agentNtfCreateSubscriptions c tknId NtfToken {ntfServer, ntfPrivKey} subsCreds = do
|
||||
case L.nonEmpty subsCreds of
|
||||
Just subsCreds' -> do
|
||||
let tSess = (0, ntfServer, Nothing)
|
||||
tryAgentError' (getNtfServerClient c tSess) >>= \case
|
||||
Left e -> pure $ L.toList $ L.map (,Left e) subsCreds'
|
||||
Right ntf -> liftIO $ do
|
||||
logServer' "-->" c ntfServer (bshow (length subsCreds') <> " subscriptions") "SNEW"
|
||||
r <- L.zip subsCreds' <$> ntfCreateSubscriptions ntf ntfPrivKey (L.map toNewSub subsCreds')
|
||||
pure $ L.toList $ L.map agentError r
|
||||
where
|
||||
agentError = second . first $ protocolClientError NTF $ clientServer ntf
|
||||
Nothing -> pure []
|
||||
where
|
||||
-- newSubs = L.map toNewSub subsCreds
|
||||
toNewSub :: CreateSubNtfReq -> NewNtfEntity 'Subscription
|
||||
toNewSub (NtfSubscription {smpServer}, ClientNtfCreds {ntfPrivateKey, notifierId}) =
|
||||
NewNtfSub tknId (SMPQueueNtf smpServer notifierId) ntfPrivateKey
|
||||
|
||||
agentNtfCheckSubscription :: AgentClient -> NtfSubscriptionId -> NtfToken -> AM NtfSubStatus
|
||||
agentNtfCheckSubscription c subId NtfToken {ntfServer, ntfPrivKey} =
|
||||
withNtfClient c ntfServer subId "SCHK" $ \ntf -> ntfCheckSubscription ntf ntfPrivKey subId
|
||||
|
||||
@@ -43,7 +43,7 @@ import Simplex.Messaging.Agent.Store
|
||||
import Simplex.Messaging.Agent.Store.SQLite
|
||||
import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Notifications.Protocol (NtfSubStatus (..), NtfTknStatus (..), SMPQueueNtf (..))
|
||||
import Simplex.Messaging.Notifications.Protocol (NtfSubStatus (..), NtfSubscriptionId, NtfTknStatus (..), SMPQueueNtf (..))
|
||||
import Simplex.Messaging.Notifications.Types
|
||||
import Simplex.Messaging.Protocol (NtfServer, sameSrvAddr)
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
@@ -198,18 +198,73 @@ runNtfWorker c srv Worker {doWork} =
|
||||
ExceptT $ agentOperationBracket c AONtfNetwork throwWhenInactive $ runExceptT runNtfOperation
|
||||
where
|
||||
runNtfOperation :: AM ()
|
||||
runNtfOperation =
|
||||
withWork c doWork (`getNextNtfSubNTFAction` srv) $
|
||||
\nextSub@(NtfSubscription {connId}, _, _) -> do
|
||||
runNtfOperation = do
|
||||
ntfBatchSize <- asks $ ntfBatchSize . config
|
||||
withWorkItems c doWork (\db -> getNextNtfSubNTFActions db srv ntfBatchSize) $ \nextSubs -> do
|
||||
logInfo $ "runNtfWorker - length nextSubs = " <> tshow (length nextSubs)
|
||||
let (creates, checks, deletes, rotates) = splitActions nextSubs
|
||||
retrySubActions c creates createSubs
|
||||
-- retrySubActions c checks checkSubs
|
||||
-- retrySubActions c deletes deleteSubs
|
||||
-- retrySubActions c rotates rotateSubs
|
||||
pure ()
|
||||
splitActions :: NonEmpty (NtfSubscription, NtfSubNTFAction, NtfActionTs) -> ([NtfSubscription], [NtfSubscription], [NtfSubscription], [NtfSubscription])
|
||||
splitActions = foldr addAction ([], [], [], [])
|
||||
where
|
||||
addAction action (creates, checks, deletes, rotates) = case action of
|
||||
(sub, NSACreate, _) -> (sub : creates, checks, deletes, rotates)
|
||||
(sub, NSACheck, _) -> (creates, sub : checks, deletes, rotates)
|
||||
(sub, NSADelete, _) -> (creates, checks, sub : deletes, rotates)
|
||||
(sub, NSARotate, _) -> (creates, checks, deletes, sub : rotates)
|
||||
createSubs :: [NtfSubscription] -> AM' [NtfSubscription]
|
||||
createSubs ntfSubs =
|
||||
getNtfToken >>= \case
|
||||
Just tkn@NtfToken {ntfServer, ntfTokenId = Just tknId, ntfTknStatus = NTActive, ntfMode = NMInstant} -> do
|
||||
subsRqs_ <- zip ntfSubs <$> withStoreBatch c (\db -> map (getQueue db) ntfSubs)
|
||||
let (errs1, subsCreds) = splitSubs subsRqs_
|
||||
-- todo incNtfServerStat ntfCreateAttempts
|
||||
rs <- agentNtfCreateSubscriptions c tknId tkn subsCreds
|
||||
let (subsCreds', errs2, successes) = splitResults rs
|
||||
ntfSubs' = map fst subsCreds'
|
||||
errs2' = map (first (ntfSubConnId . fst)) errs2
|
||||
nSubIds = map (first fst) successes
|
||||
-- todo incNtfServerStat ntfCreated
|
||||
ts <- liftIO getCurrentTime
|
||||
let checkTs = addUTCTime 30 ts
|
||||
(errs3, _) <- partitionErrs (ntfSubConnId . fst) nSubIds <$> withStoreBatch' c (\db -> map (updateSubNSACheck db checkTs) nSubIds)
|
||||
workerErrors c $ errs1 <> errs2' <> errs3
|
||||
pure ntfSubs'
|
||||
_ -> do
|
||||
let errs = map (\sub -> (ntfSubConnId sub, INTERNAL "NSACreate - no active token")) ntfSubs
|
||||
workerErrors c errs
|
||||
pure []
|
||||
where
|
||||
getQueue :: DB.Connection -> NtfSubscription -> IO (Either AgentErrorType RcvQueue)
|
||||
getQueue db NtfSubscription {connId} = first storeError <$> getPrimaryRcvQueue db connId
|
||||
splitSubs :: [(NtfSubscription, Either AgentErrorType RcvQueue)] -> ([(ConnId, AgentErrorType)], [(NtfSubscription, ClientNtfCreds)])
|
||||
splitSubs = foldr splitSub ([], [])
|
||||
where
|
||||
splitSub (sub, Right RcvQueue {clientNtfCreds = Just creds}) (errs, ss) = (errs, (sub, creds) : ss)
|
||||
splitSub (sub, Right _) (errs, ss) = ((ntfSubConnId sub, INTERNAL "NSACreate - no notifier queue credentials") : errs, ss)
|
||||
splitSub (sub, Left e) (errs, ss) = ((ntfSubConnId sub, e) : errs, ss)
|
||||
updateSubNSACheck :: DB.Connection -> UTCTime -> (NtfSubscription, NtfSubscriptionId) -> IO ()
|
||||
updateSubNSACheck db checkTs (sub, nSubId) = updateNtfSubscription db sub {ntfSubId = Just nSubId, ntfSubStatus = NASCreated NSNew} (NSANtf NSACheck) checkTs
|
||||
-- -------------------- below - old code --------------------
|
||||
runNtfOperation' :: AM ()
|
||||
runNtfOperation' = do
|
||||
ntfBatchSize <- asks $ ntfBatchSize . config
|
||||
withWorkItems c doWork (\db -> getNextNtfSubNTFActions db srv ntfBatchSize) $ \case
|
||||
[nextSub@(NtfSubscription {connId}, _, _)] -> do
|
||||
logInfo $ "runNtfWorker, nextSub " <> tshow nextSub
|
||||
ri <- asks $ reconnectInterval . config
|
||||
withRetryInterval ri $ \_ loop -> do
|
||||
liftIO $ waitWhileSuspended c
|
||||
liftIO $ waitForUserNetwork c
|
||||
processSub nextSub
|
||||
processSub' nextSub
|
||||
`catchAgentError` retryOnError c "NtfWorker" loop (workerInternalError c connId . show)
|
||||
processSub :: (NtfSubscription, NtfSubNTFAction, NtfActionTs) -> AM ()
|
||||
processSub (sub@NtfSubscription {userId, connId, smpServer, ntfSubId}, action, actionTs) = do
|
||||
_ -> pure ()
|
||||
processSub' :: (NtfSubscription, NtfSubNTFAction, NtfActionTs) -> AM ()
|
||||
processSub' (sub@NtfSubscription {userId, connId, smpServer, ntfSubId}, action, actionTs) = do
|
||||
ts <- liftIO getCurrentTime
|
||||
unlessM (lift $ rescheduleAction doWork ts actionTs) $
|
||||
case action of
|
||||
@@ -286,26 +341,14 @@ runNtfSMPWorker c srv Worker {doWork} = forever $ do
|
||||
withWorkItems c doWork (\db -> getNextNtfSubSMPActions db srv ntfBatchSize) $ \nextSubs -> do
|
||||
logInfo $ "runNtfSMPWorker - length nextSubs = " <> tshow (length nextSubs)
|
||||
let (creates, deletes) = splitActions nextSubs
|
||||
retrySubActions creates createNotifierKeys
|
||||
retrySubActions deletes deleteNotifierKeys
|
||||
retrySubActions c creates createNotifierKeys
|
||||
retrySubActions c deletes deleteNotifierKeys
|
||||
splitActions :: NonEmpty (NtfSubSMPAction, NtfSubscription) -> ([NtfSubscription], [NtfSubscription])
|
||||
splitActions = foldr addAction ([], [])
|
||||
where
|
||||
addAction action (creates, deletes) = case action of
|
||||
(NSASmpKey, sub) -> (sub : creates, deletes)
|
||||
(NSASmpDelete, sub) -> (creates, sub : deletes)
|
||||
retrySubActions :: [NtfSubscription] -> ([NtfSubscription] -> AM' [NtfSubscription]) -> AM ()
|
||||
retrySubActions subs action = do
|
||||
v <- newTVarIO subs
|
||||
ri <- asks $ reconnectInterval . config
|
||||
withRetryInterval ri $ \_ loop -> do
|
||||
liftIO $ waitWhileSuspended c
|
||||
liftIO $ waitForUserNetwork c
|
||||
subs' <- readTVarIO v
|
||||
retrySubs <- lift $ action subs'
|
||||
unless (null retrySubs) $ do
|
||||
atomically $ writeTVar v retrySubs
|
||||
retryNetworkLoop c loop
|
||||
createNotifierKeys :: [NtfSubscription] -> AM' [NtfSubscription]
|
||||
createNotifierKeys ntfSubs =
|
||||
getNtfToken >>= \case
|
||||
@@ -363,15 +406,29 @@ runNtfSMPWorker c srv Worker {doWork} = forever $ do
|
||||
pure (sub, rq)
|
||||
deleteSub :: DB.Connection -> RcvQueue -> IO ()
|
||||
deleteSub db rq = deleteNtfSubscription db (qConnId rq)
|
||||
-- (temporary errs, other errs, successes)
|
||||
splitResults :: [(a, Either AgentErrorType r)] -> ([a], [(a, AgentErrorType)], [(a, r)])
|
||||
splitResults = foldr' addRes ([], [], [])
|
||||
where
|
||||
addRes (a, r_) (as, errs, rs) = case r_ of
|
||||
Right r -> (as, errs, (a, r) : rs)
|
||||
Left e
|
||||
| temporaryOrHostError e -> (a : as, errs, rs)
|
||||
| otherwise -> (as, (a, e) : errs, rs)
|
||||
|
||||
retrySubActions :: AgentClient -> [NtfSubscription] -> ([NtfSubscription] -> AM' [NtfSubscription]) -> AM ()
|
||||
retrySubActions c subs action = do
|
||||
v <- newTVarIO subs
|
||||
ri <- asks $ reconnectInterval . config
|
||||
withRetryInterval ri $ \_ loop -> do
|
||||
liftIO $ waitWhileSuspended c
|
||||
liftIO $ waitForUserNetwork c
|
||||
subs' <- readTVarIO v
|
||||
retrySubs <- lift $ action subs'
|
||||
unless (null retrySubs) $ do
|
||||
atomically $ writeTVar v retrySubs
|
||||
retryNetworkLoop c loop
|
||||
|
||||
-- (temporary errs, other errs, successes)
|
||||
splitResults :: [(a, Either AgentErrorType r)] -> ([a], [(a, AgentErrorType)], [(a, r)])
|
||||
splitResults = foldr' addRes ([], [], [])
|
||||
where
|
||||
addRes (a, r_) (as, errs, rs) = case r_ of
|
||||
Right r -> (as, errs, (a, r) : rs)
|
||||
Left e
|
||||
| temporaryOrHostError e -> (a : as, errs, rs)
|
||||
| otherwise -> (as, (a, e) : errs, rs)
|
||||
|
||||
rescheduleAction :: TMVar () -> UTCTime -> UTCTime -> AM' Bool
|
||||
rescheduleAction doWork ts actionTs
|
||||
|
||||
@@ -160,7 +160,7 @@ module Simplex.Messaging.Agent.Store.SQLite
|
||||
setNullNtfSubscriptionAction,
|
||||
deleteNtfSubscription,
|
||||
deleteNtfSubscription',
|
||||
getNextNtfSubNTFAction,
|
||||
getNextNtfSubNTFActions,
|
||||
markNtfSubActionNtfFailed_, -- exported for tests
|
||||
getNextNtfSubSMPActions,
|
||||
markNtfSubActionSMPFailed_, -- exported for tests
|
||||
@@ -1627,14 +1627,14 @@ deleteNtfSubscription' :: DB.Connection -> ConnId -> IO ()
|
||||
deleteNtfSubscription' db connId = do
|
||||
DB.execute db "DELETE FROM ntf_subscriptions WHERE conn_id = ?" (Only connId)
|
||||
|
||||
getNextNtfSubNTFAction :: DB.Connection -> NtfServer -> IO (Either StoreError (Maybe (NtfSubscription, NtfSubNTFAction, NtfActionTs)))
|
||||
getNextNtfSubNTFAction db ntfServer@(NtfServer ntfHost ntfPort _) =
|
||||
getWorkItem "ntf NTF" getNtfConnId getNtfSubAction (markNtfSubActionNtfFailed_ db)
|
||||
getNextNtfSubNTFActions :: DB.Connection -> NtfServer -> Int -> IO (Either StoreError [Either StoreError (NtfSubscription, NtfSubNTFAction, NtfActionTs)])
|
||||
getNextNtfSubNTFActions db ntfServer@(NtfServer ntfHost ntfPort _) ntfBatchSize =
|
||||
getWorkItems "ntf NTF" getNtfConnIds getNtfSubAction (markNtfSubActionNtfFailed_ db)
|
||||
where
|
||||
getNtfConnId :: IO (Maybe ConnId)
|
||||
getNtfConnId =
|
||||
maybeFirstRow fromOnly $
|
||||
DB.query
|
||||
getNtfConnIds :: IO [ConnId]
|
||||
getNtfConnIds =
|
||||
map fromOnly
|
||||
<$> DB.query
|
||||
db
|
||||
[sql|
|
||||
SELECT conn_id
|
||||
@@ -1642,9 +1642,9 @@ getNextNtfSubNTFAction db ntfServer@(NtfServer ntfHost ntfPort _) =
|
||||
WHERE ntf_host = ? AND ntf_port = ? AND ntf_sub_action IS NOT NULL
|
||||
AND (ntf_failed = 0 OR updated_by_supervisor = 1)
|
||||
ORDER BY ntf_sub_action_ts ASC
|
||||
LIMIT 1
|
||||
LIMIT ?
|
||||
|]
|
||||
(ntfHost, ntfPort)
|
||||
(ntfHost, ntfPort, ntfBatchSize)
|
||||
getNtfSubAction :: ConnId -> IO (Either StoreError (NtfSubscription, NtfSubNTFAction, NtfActionTs))
|
||||
getNtfSubAction connId = do
|
||||
markUpdatedByWorker db connId
|
||||
|
||||
@@ -63,11 +63,13 @@ module Simplex.Messaging.Client
|
||||
forwardSMPTransmission,
|
||||
getSMPQueueInfo,
|
||||
sendProtocolCommand,
|
||||
sendProtocolCommands,
|
||||
|
||||
-- * Supporting types and client configuration
|
||||
ProtocolClientError (..),
|
||||
SMPClientError,
|
||||
ProxyClientError (..),
|
||||
Response (..),
|
||||
unexpectedResponse,
|
||||
ProtocolClientConfig (..),
|
||||
NetworkConfig (..),
|
||||
|
||||
@@ -7,6 +7,8 @@ module Simplex.Messaging.Notifications.Client where
|
||||
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.Trans.Except
|
||||
import Data.List.NonEmpty (NonEmpty (..))
|
||||
import qualified Data.List.NonEmpty as L
|
||||
import Data.Word (Word16)
|
||||
import Simplex.Messaging.Client
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
@@ -51,6 +53,15 @@ ntfCreateSubscription c pKey newSub =
|
||||
NRSubId subId -> pure subId
|
||||
r -> throwE $ unexpectedResponse r
|
||||
|
||||
ntfCreateSubscriptions :: NtfClient -> C.APrivateAuthKey -> NonEmpty (NewNtfEntity 'Subscription) -> IO (NonEmpty (Either NtfClientError NtfSubscriptionId))
|
||||
ntfCreateSubscriptions c pKey newSubs = L.map process <$> sendProtocolCommands c cs
|
||||
where
|
||||
cs = L.map (\newSub -> (Just pKey, NoEntity, NtfCmd SSubscription $ SNEW newSub)) newSubs
|
||||
process (Response _ r) = case r of
|
||||
Right (NRSubId subId) -> Right subId
|
||||
Right r' -> Left $ unexpectedResponse r'
|
||||
Left e -> Left e
|
||||
|
||||
ntfCheckSubscription :: NtfClient -> C.APrivateAuthKey -> NtfSubscriptionId -> ExceptT NtfClientError IO NtfSubStatus
|
||||
ntfCheckSubscription c pKey subId =
|
||||
sendNtfCommand c (Just pKey) subId SCHK >>= \case
|
||||
|
||||
Reference in New Issue
Block a user