From c3012173e346b06b5625de5268dc157e74d29fca Mon Sep 17 00:00:00 2001 From: JRoberts <8711996+jr-simplex@users.noreply.github.com> Date: Thu, 30 Jun 2022 20:18:12 +0400 Subject: [PATCH] ntf: recover from failed/in-deletion subscriptions (#446) * ntf: recover from failed/in-deletion subscriptions * check NASNew status on subscription recover * carryOnWithDeletion * logs * SCHK --- .../Messaging/Agent/NtfSubSupervisor.hs | 81 ++++++++++++------- src/Simplex/Messaging/Agent/Store/SQLite.hs | 41 +++++++--- src/Simplex/Messaging/Notifications/Server.hs | 4 +- src/Simplex/Messaging/Notifications/Types.hs | 11 +++ src/Simplex/Messaging/Util.hs | 6 ++ 5 files changed, 103 insertions(+), 40 deletions(-) diff --git a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs index 8e863d8fc..4decd8c93 100644 --- a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs +++ b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs @@ -18,6 +18,7 @@ where import Control.Concurrent.Async (Async, uninterruptibleCancel) import Control.Concurrent.STM (stateTVar) +import Control.Logger.Simple (logInfo) import Control.Monad import Control.Monad.Except import Control.Monad.IO.Unlift (MonadUnliftIO) @@ -40,7 +41,7 @@ import Simplex.Messaging.Notifications.Types import Simplex.Messaging.Protocol import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM -import Simplex.Messaging.Util (unlessM) +import Simplex.Messaging.Util (tshow, unlessM) import System.Random (randomR) import UnliftIO (async) import UnliftIO.Concurrent (forkIO, threadDelay) @@ -56,14 +57,17 @@ runNtfSupervisor c = forever $ do Right _ -> return () processNtfSub :: forall m. AgentMonad m => AgentClient -> (ConnId, NtfSupervisorCommand) -> m () -processNtfSub c (connId, cmd) = do +processNtfSub c@AgentClient {subQ} (connId, cmd) = do + logInfo $ "processNtfSub - connId = " <> tshow connId <> " - cmd = " <> tshow cmd ntfServer_ <- getNtfServer c + ts <- liftIO getCurrentTime case cmd of NSCCreate -> do (sub_, RcvQueue {server = smpServer, clientNtfCreds}) <- withStore c $ \db -> runExceptT $ do sub_ <- liftIO $ getNtfSubscription db connId q <- ExceptT $ getRcvQueue db connId pure (sub_, q) + logInfo $ "processNtfSub, NSCCreate - sub_ = " <> tshow sub_ case (sub_, ntfServer_) of (Nothing, Just ntfServer) -> do currentTime <- liftIO getCurrentTime @@ -79,27 +83,39 @@ processNtfSub c (connId, cmd) = do -- TODO - SMP worker can create Ntf worker on NKEY completion addNtfSMPWorker smpServer addNtfWorker ntfServer - (Just _, Just ntfServer) -> do - -- TODO subscription may have to be updated depending on current state: - -- TODO - e.g., if it was previously marked for deletion action has to be updated - -- TODO - should action depend on subscription status or always be NSASmpKey (NSACreate if notifierId exists) - -- TODO in case worker is currently deleting it? When deleting worker should check for updated_by_supervisor - -- TODO and if it is set perform update instead of delete. If worker was not deleting it yet it should - -- TODO idempotently replay commands. + (Just (sub@NtfSubscription {ntfSubStatus}, action), Just ntfServer) -> do + case action of + -- action was set to NULL after worker internal error + Nothing -> resetSubscription + Just (a, _) -> + -- subscription was marked for deletion + when (isDeleteNtfSubAction a) $ + if ntfSubStatus == NASNew || ntfSubStatus == NASOff || ntfSubStatus == NASDeleted + then resetSubscription + else withStore' c $ \db -> + supervisorUpdateNtfSubscription db sub {ntfServer} (NtfSubNTFAction NSACreate) ts addNtfSMPWorker smpServer addNtfWorker ntfServer - _ -> pure () -- error - notification server not configured + where + resetSubscription :: m () + resetSubscription = withStore' c $ \db -> + supervisorUpdateNtfSubscription db sub {ntfQueueId = Nothing, ntfServer, ntfSubId = Nothing, ntfSubStatus = NASNew} (NtfSubSMPAction NSASmpKey) ts + _ -> pure () -- err "NSCCreate - notification server not configured" NSCDelete -> do - withStore' c $ \db -> markNtfSubscriptionForDeletion db connId (NtfSubNTFAction NSADelete) - case ntfServer_ of - (Just ntfServer) -> addNtfWorker ntfServer - _ -> pure () + sub_ <- withStore' c $ \db -> do + supervisorUpdateNtfSubAction db connId (NtfSubNTFAction NSADelete) ts + getNtfSubscription db connId + logInfo $ "processNtfSub, NSCDelete - sub_ = " <> tshow sub_ + case sub_ of + (Just (NtfSubscription {ntfServer}, _)) -> addNtfWorker ntfServer + _ -> pure () -- err "NSCDelete - no subscription" NSCSmpDelete -> do withStore' c (`getRcvQueue` connId) >>= \case - Right RcvQueue {server = smpServer} -> do - withStore' c $ \db -> markNtfSubscriptionForDeletion db connId (NtfSubSMPAction NSASmpDelete) + Right rq@RcvQueue {server = smpServer} -> do + logInfo $ "processNtfSub, NSCSmpDelete - rq = " <> tshow rq + withStore' c $ \db -> supervisorUpdateNtfSubAction db connId (NtfSubSMPAction NSASmpDelete) ts addNtfSMPWorker smpServer - Left _ -> pure () + _ -> err "NSCSmpDelete - no rcv queue" NSCNtfWorker ntfServer -> addNtfWorker ntfServer NSCNtfSMPWorker smpServer -> @@ -121,18 +137,22 @@ processNtfSub c (connId, cmd) = do atomically $ TM.insert srv (doWork, worker) ws Just (doWork, _) -> void . atomically $ tryPutTMVar doWork () + err :: String -> m () + err internalErrStr = atomically $ writeTBQueue subQ ("", connId, AP.ERR $ AP.INTERNAL internalErrStr) runNtfWorker :: forall m. AgentMonad m => AgentClient -> NtfServer -> TMVar () -> m () runNtfWorker c srv doWork = forever $ do void . atomically $ readTMVar doWork nextSub_ <- withStore' c (`getNextNtfSubNTFAction` srv) + logInfo $ "runNtfWorker, nextSub_ " <> tshow nextSub_ case nextSub_ of Nothing -> noWorkToDo Just a@(NtfSubscription {connId}, _, _) -> do ri <- asks $ reconnectInterval . config withRetryInterval ri $ \loop -> processAction a - `catchError` ( \e -> + `catchError` ( \e -> do + logInfo $ "runNtfWorker, error " <> tshow e case e of BROKER NETWORK -> loop BROKER TIMEOUT -> loop @@ -157,7 +177,7 @@ runNtfWorker c srv doWork = forever $ do -- TODO smaller retry until Active, less frequently (daily?) once Active let actionTs' = addUTCTime 30 ts withStore' c $ \db -> - updateNtfSubscription db connId sub {ntfSubId = Just nSubId, ntfSubStatus = NASCreated NSNew} (NtfSubNTFAction NSACheck) actionTs' + updateNtfSubscription db sub {ntfSubId = Just nSubId, ntfSubStatus = NASCreated NSNew} (NtfSubNTFAction NSACheck) actionTs' _ -> ntfInternalError c connId "NSACreate - no notifier queue credentials" _ -> ntfInternalError c connId "NSACreate - no active token" NSACheck -> @@ -173,12 +193,15 @@ runNtfWorker c srv doWork = forever $ do NSADelete -> case ntfSubId of Just nSubId -> (getNtfToken >>= \tkn -> forM_ tkn $ agentNtfDeleteSubscription c nSubId) - `E.finally` do - withStore' c $ \db -> - updateNtfSubscription db connId sub {ntfSubId = Nothing, ntfSubStatus = NASOff} (NtfSubSMPAction NSASmpDelete) ts - ns <- asks ntfSupervisor - atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCNtfSMPWorker smpServer) - Nothing -> ntfInternalError c connId "NSADelete - no subscription ID" + `E.finally` carryOnWithDeletion + Nothing -> carryOnWithDeletion + where + carryOnWithDeletion :: m () + carryOnWithDeletion = do + withStore' c $ \db -> + updateNtfSubscription db sub {ntfSubId = Nothing, ntfSubStatus = NASOff} (NtfSubSMPAction NSASmpDelete) ts + ns <- asks ntfSupervisor + atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCNtfSMPWorker smpServer) where updateSubNextCheck ts toStatus = do checkInterval <- asks $ ntfSubCheckInterval . config @@ -186,19 +209,21 @@ runNtfWorker c srv doWork = forever $ do updateSub (NASCreated toStatus) (NtfSubNTFAction NSACheck) nextCheckTs updateSub toStatus toAction actionTs' = withStore' c $ \db -> - updateNtfSubscription db connId sub {ntfSubStatus = toStatus} toAction actionTs' + updateNtfSubscription db sub {ntfSubStatus = toStatus} toAction actionTs' runNtfSMPWorker :: forall m. AgentMonad m => AgentClient -> SMPServer -> TMVar () -> m () runNtfSMPWorker c srv doWork = forever $ do void . atomically $ readTMVar doWork nextSub_ <- withStore' c (`getNextNtfSubSMPAction` srv) + logInfo $ "runNtfSMPWorker, nextSub_ " <> tshow nextSub_ case nextSub_ of Nothing -> noWorkToDo Just a@(NtfSubscription {connId}, _, _) -> do ri <- asks $ reconnectInterval . config withRetryInterval ri $ \loop -> processAction a - `catchError` ( \e -> + `catchError` ( \e -> do + logInfo $ "runNtfSMPWorker, error " <> tshow e case e of BROKER NETWORK -> loop BROKER TIMEOUT -> loop @@ -224,7 +249,7 @@ runNtfSMPWorker c srv doWork = forever $ do let rcvNtfDhSecret = C.dh' rcvNtfSrvPubDhKey rcvNtfPrivDhKey withStore' c $ \st -> do setRcvQueueNtfCreds st connId ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret} - updateNtfSubscription st connId sub {ntfQueueId = Just notifierId, ntfSubStatus = NASKey} (NtfSubNTFAction NSACreate) ts + updateNtfSubscription st sub {ntfQueueId = Just notifierId, ntfSubStatus = NASKey} (NtfSubNTFAction NSACreate) ts ns <- asks ntfSupervisor atomically $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer) _ -> ntfInternalError c connId "NSASmpKey - no active token" diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 0dacb98d8..c4b92f2d5 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -75,7 +75,8 @@ module Simplex.Messaging.Agent.Store.SQLite -- Notification subscription persistence getNtfSubscription, createNtfSubscription, - markNtfSubscriptionForDeletion, + supervisorUpdateNtfSubscription, + supervisorUpdateNtfSubAction, updateNtfSubscription, setNullNtfSubscriptionAction, deleteNtfSubscription, @@ -712,14 +713,14 @@ removeNtfToken db NtfToken {deviceToken = DeviceToken provider token, ntfServer |] (provider, token, host, port) -getNtfSubscription :: DB.Connection -> ConnId -> IO (Maybe NtfSubscription) +getNtfSubscription :: DB.Connection -> ConnId -> IO (Maybe (NtfSubscription, Maybe (NtfSubAction, NtfActionTs))) getNtfSubscription db connId = maybeFirstRow ntfSubscription $ DB.query db [sql| SELECT s.host, s.port, s.key_hash, ns.ntf_host, ns.ntf_port, ns.ntf_key_hash, - nsb.smp_ntf_id, nsb.ntf_sub_id, nsb.ntf_sub_status + nsb.smp_ntf_id, nsb.ntf_sub_id, nsb.ntf_sub_status, nsb.ntf_sub_action, nsb.ntf_sub_smp_action, nsb.ntf_sub_action_ts FROM ntf_subscriptions nsb JOIN servers s ON s.host = nsb.smp_host AND s.port = nsb.smp_port JOIN ntf_servers ns USING (ntf_host, ntf_port) @@ -727,10 +728,14 @@ getNtfSubscription db connId = |] (Only connId) where - ntfSubscription (smpHost, smpPort, smpKeyHash, ntfHost, ntfPort, ntfKeyHash, ntfQueueId, ntfSubId, ntfSubStatus) = + ntfSubscription (smpHost, smpPort, smpKeyHash, ntfHost, ntfPort, ntfKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, ntfAction_, smpAction_, actionTs_) = let smpServer = SMPServer smpHost smpPort smpKeyHash ntfServer = ProtocolServer ntfHost ntfPort ntfKeyHash - in NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus} + action = case (ntfAction_, smpAction_, actionTs_) of + (Just ntfAction, Nothing, Just actionTs) -> Just (NtfSubNTFAction ntfAction, actionTs) + (Nothing, Just smpAction, Just actionTs) -> Just (NtfSubSMPAction smpAction, actionTs) + _ -> Nothing + in (NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus}, action) createNtfSubscription :: DB.Connection -> NtfSubscription -> NtfSubAction -> NtfActionTs -> IO () createNtfSubscription db ntfSubscription action actionTs = do @@ -749,8 +754,22 @@ createNtfSubscription db ntfSubscription action actionTs = do where (ntfSubAction, ntfSubSMPAction) = ntfSubAndSMPAction action -markNtfSubscriptionForDeletion :: DB.Connection -> ConnId -> NtfSubAction -> IO () -markNtfSubscriptionForDeletion db connId ntfAction = do +supervisorUpdateNtfSubscription :: DB.Connection -> NtfSubscription -> NtfSubAction -> NtfActionTs -> IO () +supervisorUpdateNtfSubscription db NtfSubscription {connId, ntfQueueId, ntfServer = (ProtocolServer ntfHost ntfPort _), ntfSubId, ntfSubStatus} action actionTs = do + updatedAt <- getCurrentTime + DB.execute + db + [sql| + UPDATE ntf_subscriptions + SET smp_ntf_id = ?, ntf_host = ?, ntf_port = ?, ntf_sub_id = ?, ntf_sub_status = ?, ntf_sub_action = ?, ntf_sub_smp_action = ?, ntf_sub_action_ts = ?, updated_by_supervisor = ?, updated_at = ? + WHERE conn_id = ? + |] + ((ntfQueueId, ntfHost, ntfPort, ntfSubId) :. (ntfSubStatus, ntfSubAction, ntfSubSMPAction, actionTs, True, updatedAt, connId)) + where + (ntfSubAction, ntfSubSMPAction) = ntfSubAndSMPAction action + +supervisorUpdateNtfSubAction :: DB.Connection -> ConnId -> NtfSubAction -> NtfActionTs -> IO () +supervisorUpdateNtfSubAction db connId action actionTs = do updatedAt <- getCurrentTime DB.execute db @@ -759,12 +778,12 @@ markNtfSubscriptionForDeletion db connId ntfAction = do SET ntf_sub_action = ?, ntf_sub_smp_action = ?, ntf_sub_action_ts = ?, updated_by_supervisor = ?, updated_at = ? WHERE conn_id = ? |] - (ntfSubAction, ntfSubSMPAction, updatedAt, True, updatedAt, connId) + (ntfSubAction, ntfSubSMPAction, actionTs, True, updatedAt, connId) where - (ntfSubAction, ntfSubSMPAction) = ntfSubAndSMPAction ntfAction + (ntfSubAction, ntfSubSMPAction) = ntfSubAndSMPAction action -updateNtfSubscription :: DB.Connection -> ConnId -> NtfSubscription -> NtfSubAction -> NtfActionTs -> IO () -updateNtfSubscription db connId NtfSubscription {ntfQueueId, ntfSubId, ntfSubStatus} action actionTs = do +updateNtfSubscription :: DB.Connection -> NtfSubscription -> NtfSubAction -> NtfActionTs -> IO () +updateNtfSubscription db NtfSubscription {connId, ntfQueueId, ntfSubId, ntfSubStatus} action actionTs = do r <- maybeFirstRow fromOnly $ DB.query db "SELECT updated_by_supervisor FROM ntf_subscriptions WHERE conn_id = ?" (Only connId) forM_ r $ \updatedBySupervisor -> do updatedAt <- getCurrentTime diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index f6d3dbf37..bbef83255 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/src/Simplex/Messaging/Notifications/Server.hs @@ -440,7 +440,9 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu if notifierKey == registeredNKey then NRSubId subId else NRErr AUTH - SCHK -> pure $ NRSub status + SCHK -> do + logDebug "SCHK" + pure $ NRSub status SDEL -> do logDebug "SDEL" st <- asks store diff --git a/src/Simplex/Messaging/Notifications/Types.hs b/src/Simplex/Messaging/Notifications/Types.hs index ab6ac13f7..d679d00f2 100644 --- a/src/Simplex/Messaging/Notifications/Types.hs +++ b/src/Simplex/Messaging/Notifications/Types.hs @@ -81,6 +81,17 @@ newNtfToken deviceToken ntfServer (ntfPubKey, ntfPrivKey) ntfDhKeys ntfMode = } data NtfSubAction = NtfSubNTFAction NtfSubNTFAction | NtfSubSMPAction NtfSubSMPAction + deriving (Show) + +isDeleteNtfSubAction :: NtfSubAction -> Bool +isDeleteNtfSubAction = \case + NtfSubNTFAction a -> case a of + NSACreate -> False + NSACheck -> False + NSADelete -> True + NtfSubSMPAction a -> case a of + NSASmpKey -> False + NSASmpDelete -> True type NtfActionTs = UTCTime diff --git a/src/Simplex/Messaging/Util.hs b/src/Simplex/Messaging/Util.hs index 5c0876196..c1e4146d6 100644 --- a/src/Simplex/Messaging/Util.hs +++ b/src/Simplex/Messaging/Util.hs @@ -10,6 +10,8 @@ import Control.Monad.Trans.Except import Data.Bifunctor (first) import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B +import Data.Text (Text) +import qualified Data.Text as T import UnliftIO.Async raceAny_ :: MonadUnliftIO m => [m a] -> m () @@ -32,6 +34,10 @@ bshow :: Show a => a -> ByteString bshow = B.pack . show {-# INLINE bshow #-} +tshow :: Show a => a -> Text +tshow = T.pack . show +{-# INLINE tshow #-} + maybeWord :: (a -> ByteString) -> Maybe a -> ByteString maybeWord f = maybe "" $ B.cons ' ' . f {-# INLINE maybeWord #-}