ntf: read rcv queue separately from ntf subscription, fix deletion race condition (#423)

src/Simplex/Messaging/Agent.hs

@@ -622,11 +622,10 @@ deleteConnection' c connId =
     delete :: RcvQueue -> m ()
     delete rq = do
       deleteQueue c rq
-      ns <- asks ntfSupervisor
-      atomically $ do
-        removeSubscription c connId
-        sendNtfSubCommand ns (connId, NSCDelete)
+      atomically $ removeSubscription c connId
       withStore' c (`deleteConn` connId)
+      ns <- asks ntfSupervisor
+      atomically $ sendNtfSubCommand ns (connId, NSCDelete)
 
 -- | Change servers to be used for creating new queues, in Reader monad
 setSMPServers' :: AgentMonad m => AgentClient -> NonEmpty SMPServer -> m ()

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -124,18 +124,18 @@ runNtfWorker c srv doWork = forever $ do
       ts <- liftIO getCurrentTime
       case nextSub_ of
         Nothing -> noWorkToDo
-        Just (ntfSub@NtfSubscription {connId, smpServer, ntfSubId}, ntfSubAction, RcvQueue {clientNtfCreds}) ->
+        Just (ntfSub@NtfSubscription {connId, smpServer, ntfSubId}, ntfSubAction) ->
           unlessM (rescheduleAction doWork ts ntfSub) $
             case ntfSubAction of
-              NSACreate -> case clientNtfCreds of
-                Just ClientNtfCreds {ntfPrivateKey, notifierId}
-                  | ntfTknStatus == NTActive -> do
+              NSACreate -> do
+                rq_ <- withStore' c (`getRcvQueue` connId)
+                case (rq_, ntfTknStatus) of
+                  (Right RcvQueue {clientNtfCreds = Just ClientNtfCreds {ntfPrivateKey, notifierId}}, NTActive) -> do
                     nSubId <- agentNtfCreateSubscription c tknId tkn (SMPQueueNtf smpServer notifierId) ntfPrivateKey
                     let actionTs = addUTCTime 30 ts
                     withStore' c $ \db ->
                       updateNtfSubscription db connId ntfSub {ntfSubId = Just nSubId, ntfSubStatus = NASCreated NSNew, ntfSubActionTs = actionTs} (NtfSubAction NSACheck)
-                  | otherwise -> ntfInternalError c connId "NSACreate - token not active"
-                _ -> ntfInternalError c connId "NSACreate - no notifier key or ID"
+                  _ -> ntfInternalError c connId "NSACreate - no notifier queue credentials or token not active"
               NSACheck -> case ntfSubId of
                 Just nSubId ->
                   agentNtfCheckSubscription c nSubId tkn >>= \case
@@ -177,24 +177,27 @@ runNtfSMPWorker c srv doWork = forever $ do
       ts <- liftIO getCurrentTime
       case nextSub_ of
         Nothing -> noWorkToDo
-        Just (ntfSub@NtfSubscription {connId, ntfServer}, ntfSubAction, rq) ->
+        Just (ntfSub@NtfSubscription {connId, ntfServer}, ntfSubAction) ->
           unlessM (rescheduleAction doWork ts ntfSub) $
             case ntfSubAction of
-              NSASmpKey
-                | ntfTknStatus == NTActive -> do
-                  C.SignAlg a <- asks (cmdSignAlg . config)
-                  (ntfPublicKey, ntfPrivateKey) <- liftIO $ C.generateSignatureKeyPair a
-                  (rcvNtfPubDhKey, rcvNtfPrivDhKey) <- liftIO C.generateKeyPair'
-                  (notifierId, rcvNtfSrvPubDhKey) <- enableQueueNotifications c rq ntfPublicKey rcvNtfPubDhKey
-                  let rcvNtfDhSecret = C.dh' rcvNtfSrvPubDhKey rcvNtfPrivDhKey
-                  withStore' c $ \st -> do
-                    setRcvQueueNtfCreds st connId ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret}
-                    updateNtfSubscription st connId ntfSub {ntfQueueId = Just notifierId, ntfSubStatus = NASKey, ntfSubActionTs = ts} (NtfSubAction NSACreate)
-                  ns <- asks ntfSupervisor
-                  atomically $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer)
-                | otherwise -> ntfInternalError c connId "NSASmpKey - token not active"
+              NSASmpKey -> do
+                rq_ <- withStore' c (`getRcvQueue` connId)
+                case (rq_, ntfTknStatus) of
+                  (Right rq, NTActive) -> do
+                    C.SignAlg a <- asks (cmdSignAlg . config)
+                    (ntfPublicKey, ntfPrivateKey) <- liftIO $ C.generateSignatureKeyPair a
+                    (rcvNtfPubDhKey, rcvNtfPrivDhKey) <- liftIO C.generateKeyPair'
+                    (notifierId, rcvNtfSrvPubDhKey) <- enableQueueNotifications c rq ntfPublicKey rcvNtfPubDhKey
+                    let rcvNtfDhSecret = C.dh' rcvNtfSrvPubDhKey rcvNtfPrivDhKey
+                    withStore' c $ \st -> do
+                      setRcvQueueNtfCreds st connId ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret}
+                      updateNtfSubscription st connId ntfSub {ntfQueueId = Just notifierId, ntfSubStatus = NASKey, ntfSubActionTs = ts} (NtfSubAction NSACreate)
+                    ns <- asks ntfSupervisor
+                    atomically $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer)
+                  _ -> ntfInternalError c connId "NSASmpKey - no rcv queue or token not active"
               NSASmpDelete -> do
-                disableQueueNotifications c rq
+                rq_ <- withStore' c (`getRcvQueue` connId)
+                forM_ rq_ $ \rq -> disableQueueNotifications c rq
                 withStore' c $ \db -> deleteNtfSubscription db connId
     _ -> noWorkToDo
   delay <- asks $ ntfWorkerThrottle . config

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

@@ -793,11 +793,11 @@ deleteNtfSubscription db connId = do
           (Nothing :: Maybe SMP.NotifierId, Nothing :: Maybe NtfSubscriptionId, NASDeleted, False, updatedAt, connId)
       else DB.execute db "DELETE FROM ntf_subscriptions WHERE conn_id = ?" (Only connId)
 
-getNextNtfSubAction :: DB.Connection -> NtfServer -> IO (Maybe (NtfSubscription, NtfSubAction, RcvQueue))
+getNextNtfSubAction :: DB.Connection -> NtfServer -> IO (Maybe (NtfSubscription, NtfSubAction))
 getNextNtfSubAction db ntfServer@(ProtocolServer ntfHost ntfPort _) = do
-  maybeFirstRow ntfSubscription getNtfSubAction_ $>>= \(ntfSub@NtfSubscription {connId}, ntfSubAction) -> do
+  maybeFirstRow ntfSubscription getNtfSubAction_ $>>= \ntfSub@(NtfSubscription {connId}, _) -> do
     DB.execute db "UPDATE ntf_subscriptions SET updated_by_supervisor = ? WHERE conn_id = ?" (False, connId)
-    (ntfSub,ntfSubAction,) <$$> getRcvQueueByConnId_ db connId
+    pure $ Just ntfSub
   where
     getNtfSubAction_ =
       DB.query
@@ -816,11 +816,11 @@ getNextNtfSubAction db ntfServer@(ProtocolServer ntfHost ntfPort _) = do
       let smpServer = SMPServer smpHost smpPort smpKeyHash
        in (NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus, ntfSubActionTs}, ntfSubAction)
 
-getNextNtfSubSMPAction :: DB.Connection -> SMPServer -> IO (Maybe (NtfSubscription, NtfSubSMPAction, RcvQueue))
+getNextNtfSubSMPAction :: DB.Connection -> SMPServer -> IO (Maybe (NtfSubscription, NtfSubSMPAction))
 getNextNtfSubSMPAction db smpServer@(SMPServer smpHost smpPort _) = do
-  maybeFirstRow ntfSubscription getNtfSubAction_ $>>= \(ntfSub@NtfSubscription {connId}, ntfSubAction) -> do
+  maybeFirstRow ntfSubscription getNtfSubAction_ $>>= \ntfSub@(NtfSubscription {connId}, _) -> do
     DB.execute db "UPDATE ntf_subscriptions SET updated_by_supervisor = ? WHERE conn_id = ?" (False, connId)
-    (ntfSub,ntfSubAction,) <$$> getRcvQueueByConnId_ db connId
+    pure $ Just ntfSub
   where
     getNtfSubAction_ =
       DB.query

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220608_v2.hs

@@ -27,7 +27,7 @@ ALTER TABLE rcv_queues ADD COLUMN rcv_ntf_dh_secret BLOB;
 CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues (host, port, ntf_id);
 
 CREATE TABLE ntf_subscriptions (
-  conn_id BLOB NOT NULL, -- ? make nullable for when connection is deleted but we still need to delete subscription
+  conn_id BLOB NOT NULL,
   smp_host TEXT NULL,
   smp_port TEXT NULL,
   smp_ntf_id BLOB,
@@ -42,8 +42,6 @@ CREATE TABLE ntf_subscriptions (
   created_at TEXT NOT NULL DEFAULT (datetime('now')),
   updated_at TEXT NOT NULL DEFAULT (datetime('now')),
   PRIMARY KEY (conn_id),
-  -- FOREIGN KEY (conn_id) REFERENCES connections
-  --   ON DELETE SET NULL ON UPDATE CASCADE,
   FOREIGN KEY (smp_host, smp_port) REFERENCES servers (host, port)
     ON DELETE SET NULL ON UPDATE CASCADE,
   FOREIGN KEY (ntf_host, ntf_port) REFERENCES ntf_servers

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql

@@ -171,7 +171,7 @@ tkn_dh_secret BLOB, -- DH secret for e2e encryption of notifications
 ) WITHOUT ROWID;
 CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues(host, port, ntf_id);
 CREATE TABLE ntf_subscriptions(
-  conn_id BLOB NOT NULL, -- ? make nullable for when connection is deleted but we still need to delete subscription
+  conn_id BLOB NOT NULL,
   smp_host TEXT NULL,
   smp_port TEXT NULL,
   smp_ntf_id BLOB,
@@ -186,8 +186,6 @@ CREATE TABLE ntf_subscriptions(
   created_at TEXT NOT NULL DEFAULT(datetime('now')),
   updated_at TEXT NOT NULL DEFAULT(datetime('now')),
   PRIMARY KEY(conn_id),
-  -- FOREIGN KEY(conn_id) REFERENCES connections
-  -- ON DELETE SET NULL ON UPDATE CASCADE,
   FOREIGN KEY(smp_host, smp_port) REFERENCES servers(host, port)
   ON DELETE SET NULL ON UPDATE CASCADE,
   FOREIGN KEY(ntf_host, ntf_port) REFERENCES ntf_servers