rfc about drift detection, and SALL to mark end of message delivery

src/Simplex/Messaging/Agent.hs

@@ -200,6 +200,7 @@ import Simplex.Messaging.Protocol
     ErrorType (AUTH),
     MsgBody,
     MsgFlags (..),
+    IdsHash,
     NtfServer,
     ProtoServerWithAuth (..),
     ProtocolServer (..),
@@ -465,7 +466,7 @@ resubscribeConnections :: AgentClient -> [ConnId] -> AE (Map ConnId (Either Agen
 resubscribeConnections c = withAgentEnv c . resubscribeConnections' c
 {-# INLINE resubscribeConnections #-}
 
-subscribeClientServices :: AgentClient -> UserId -> AE (Map SMPServer (Either AgentErrorType Int64))
+subscribeClientServices :: AgentClient -> UserId -> AE (Map SMPServer (Either AgentErrorType (Int64, IdsHash)))
 subscribeClientServices c = withAgentEnv c . subscribeClientServices' c
 {-# INLINE subscribeClientServices #-}
 
@@ -1361,7 +1362,8 @@ resubscribeConnections' c connIds = do
   -- union is left-biased, so results returned by subscribeConnections' take precedence
   (`M.union` r) <$> subscribeConnections' c connIds'
 
-subscribeClientServices' :: AgentClient -> UserId -> AM (Map SMPServer (Either AgentErrorType Int64))
+-- TODO [serts rcv] compare hash with lock
+subscribeClientServices' :: AgentClient -> UserId -> AM (Map SMPServer (Either AgentErrorType (Int64, IdsHash)))
 subscribeClientServices' c userId =
   ifM useService subscribe $ throwError $ CMD PROHIBITED "no user service allowed"
   where

src/Simplex/Messaging/Agent/Client.hs

@@ -254,6 +254,7 @@ import Simplex.Messaging.Protocol
     ErrorType,
     MsgFlags (..),
     MsgId,
+    IdsHash,
     NtfServer,
     NtfServerWithAuth,
     ProtoServer,
@@ -1424,6 +1425,7 @@ newRcvQueue_ c nm userId connId (ProtoServerWithAuth srv auth) vRange cqrd enabl
       (ntfKeys, ntfCreds) <- liftIO $ mkNtfCreds a g smp
       (thParams smp,ntfKeys,) <$> createSMPQueue smp nm nonce_ rKeys dhKey auth subMode (queueReqData cqrd) ntfCreds
   -- TODO [certs rcv] validate that serviceId is the same as in the client session, fail otherwise
+  -- possibly, it should allow returning Nothing - it would indicate incorrect old version
   liftIO . logServer "<--" c srv NoEntity $ B.unwords ["IDS", logSecret rcvId, logSecret sndId]
   shortLink <- mkShortLinkCreds thParams' qik
   let rq =
@@ -1575,7 +1577,7 @@ subscribeQueues c qs = do
         processSubResults = mapM_ $ uncurry $ processSubResult c sessId
         resubscribe = resubscribeSMPSession c tSess `runReaderT` env
 
-subscribeClientService :: AgentClient -> UserId -> SMPServer -> AM Int64
+subscribeClientService :: AgentClient -> UserId -> SMPServer -> AM (Int64, IdsHash)
 subscribeClientService c userId srv =
   withLogClient c NRMBackground (userId, srv, Nothing) B.empty "SUBS" $
     (`subscribeService` SMP.SRecipientService) . connectedClient

src/Simplex/Messaging/Client.hs

@@ -884,12 +884,12 @@ nsubResponse_ = \case
 {-# INLINE nsubResponse_ #-}
 
 -- This command is always sent in background request mode
-subscribeService :: forall p. (PartyI p, ServiceParty p) => SMPClient -> SParty p -> ExceptT SMPClientError IO Int64
+subscribeService :: forall p. (PartyI p, ServiceParty p) => SMPClient -> SParty p -> ExceptT SMPClientError IO (Int64, IdsHash)
 subscribeService c party = case smpClientService c of
   Just THClientService {serviceId, serviceKey} -> do
     liftIO $ enablePings c
     sendSMPCommand c NRMBackground (Just (C.APrivateAuthKey C.SEd25519 serviceKey)) serviceId subCmd >>= \case
-      SOKS n -> pure n
+      SOKS n idsHash -> pure (n, idsHash)
       r -> throwE $ unexpectedResponse r
     where
       subCmd :: Command p

src/Simplex/Messaging/Client/Agent.hs

@@ -479,14 +479,14 @@ smpSubscribeService ca smp srv serviceSub@(serviceId, _) = case smpClientService
             (True <$ processSubscription r)
             (pure False)
       if ok
-        then case r of
-          Right n -> notify ca $ CAServiceSubscribed srv serviceSub n
+        then case r of -- TODO [certs rcv] compare hash
+          Right (n, _idsHash) -> notify ca $ CAServiceSubscribed srv serviceSub n
           Left e
             | smpClientServiceError e -> notifyUnavailable
             | temporaryClientError e -> reconnectClient ca srv
             | otherwise -> notify ca $ CAServiceSubError srv serviceSub e
         else reconnectClient ca srv
-    processSubscription = mapM_ $ \n -> do
+    processSubscription = mapM_ $ \(n, _idsHash) -> do -- TODO [certs rcv] validate hash here?
       setActiveServiceSub ca srv $ Just ((serviceId, n), sessId)
       setPendingServiceSub ca srv Nothing
     serviceAvailable THClientService {serviceRole, serviceId = serviceId'} =

src/Simplex/Messaging/Protocol.hs

@@ -139,6 +139,7 @@ module Simplex.Messaging.Protocol
     RcvMessage (..),
     MsgId,
     MsgBody,
+    IdsHash,
     MaxMessageLen,
     MaxRcvMessageLen,
     EncRcvMsgBody (..),
@@ -692,11 +693,13 @@ data BrokerMsg where
   -- | Service subscription success - confirms when queue was associated with the service
   SOK :: Maybe ServiceId -> BrokerMsg
   -- | The number of queues subscribed with SUBS command
-  SOKS :: Int64 -> BrokerMsg
+  SOKS :: Int64 -> IdsHash -> BrokerMsg
   -- MSG v1/2 has to be supported for encoding/decoding
   -- v1: MSG :: MsgId -> SystemTime -> MsgBody -> BrokerMsg
   -- v2: MsgId -> SystemTime -> MsgFlags -> MsgBody -> BrokerMsg
   MSG :: RcvMessage -> BrokerMsg
+  -- sent once delivering messages to SUBS command is complete
+  SALL :: BrokerMsg
   NID :: NotifierId -> RcvNtfPublicDhKey -> BrokerMsg
   NMSG :: C.CbNonce -> EncNMsgMeta -> BrokerMsg
   -- Should include certificate chain
@@ -933,6 +936,7 @@ data BrokerMsgTag
   | SOK_
   | SOKS_
   | MSG_
+  | SALL_
   | NID_
   | NMSG_
   | PKEY_
@@ -1025,6 +1029,7 @@ instance Encoding BrokerMsgTag where
     SOK_ -> "SOK"
     SOKS_ -> "SOKS"
     MSG_ -> "MSG"
+    SALL_ -> "SALL"
     NID_ -> "NID"
     NMSG_ -> "NMSG"
     PKEY_ -> "PKEY"
@@ -1046,6 +1051,7 @@ instance ProtocolMsgTag BrokerMsgTag where
     "SOK" -> Just SOK_
     "SOKS" -> Just SOKS_
     "MSG" -> Just MSG_
+    "SALL" -> Just SALL_
     "NID" -> Just NID_
     "NMSG" -> Just NMSG_
     "PKEY" -> Just PKEY_
@@ -1448,6 +1454,8 @@ type MsgId = ByteString
 -- | SMP message body.
 type MsgBody = ByteString
 
+type IdsHash = ByteString
+
 data ProtocolErrorType = PECmdSyntax | PECmdUnknown | PESession | PEBlock
 
 -- | Type for protocol errors.
@@ -1807,9 +1815,12 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
     SOK serviceId_
       | v >= serviceCertsSMPVersion -> e (SOK_, ' ', serviceId_)
       | otherwise -> e OK_ -- won't happen, the association with the service requires v >= serviceCertsSMPVersion
-    SOKS n -> e (SOKS_, ' ', n)
+    SOKS n idsHash
+      | v >= rcvServiceSMPVersion -> e (SOKS_, ' ', n, idsHash)
+      | otherwise -> e (SOKS_, ' ', n)
     MSG RcvMessage {msgId, msgBody = EncRcvMsgBody body} ->
       e (MSG_, ' ', msgId, Tail body)
+    SALL -> e SALL_
     NID nId srvNtfDh -> e (NID_, ' ', nId, srvNtfDh)
     NMSG nmsgNonce encNMsgMeta -> e (NMSG_, ' ', nmsgNonce, encNMsgMeta)
     PKEY sid vr certKey -> e (PKEY_, ' ', sid, vr, certKey)
@@ -1836,6 +1847,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
       MSG . RcvMessage msgId <$> bodyP
       where
         bodyP = EncRcvMsgBody . unTail <$> smpP
+    SALL_ -> pure SALL
     IDS_
       | v >= newNtfCredsSMPVersion -> ids smpP smpP smpP smpP
       | v >= serviceCertsSMPVersion -> ids smpP smpP smpP nothing
@@ -1856,7 +1868,9 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
           pure $ IDS QIK {rcvId, sndId, rcvPublicDhKey, queueMode, linkId, serviceId, serverNtfCreds}
     LNK_ -> LNK <$> _smpP <*> smpP
     SOK_ -> SOK <$> _smpP
-    SOKS_ -> SOKS <$> _smpP
+    SOKS_
+      | v >= rcvServiceSMPVersion -> SOKS <$> _smpP <*> smpP
+      | otherwise -> SOKS <$> _smpP <*> pure B.empty
     NID_ -> NID <$> _smpP <*> smpP
     NMSG_ -> NMSG <$> _smpP <*> smpP
     PKEY_ -> PKEY <$> _smpP <*> smpP <*> smpP
@@ -1886,6 +1900,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
     PONG -> noEntityMsg
     PKEY {} -> noEntityMsg
     RRES _ -> noEntityMsg
+    SALL -> noEntityMsg
     -- other broker responses must have queue ID
     _
       | B.null entId -> Left $ CMD NO_ENTITY

src/Simplex/Messaging/Server.hs

@@ -1790,17 +1790,18 @@ client
         subscribeServiceMessages serviceId =
           sharedSubscribeService SRecipientService serviceId subscribers serviceSubscribed serviceSubsCount >>= \case
             Left e -> pure $ ERR e
-            Right (hasSub, count) -> do
+            Right (hasSub, (count, idsHash)) -> do
               unless hasSub $ forkClient clnt "deliverServiceMessages" $ liftIO $ deliverServiceMessages count
-              pure $ SOKS count
+              pure $ SOKS count idsHash
           where
             deliverServiceMessages expectedCnt = do
               (qCnt, _msgCnt, _dupCnt, _errCnt) <- foldRcvServiceMessages ms serviceId deliverQueueMsg (0, 0, 0, 0)
+              atomically $ writeTBQueue msgQ [(NoCorrId, NoEntity, SALL)]
               -- TODO [cert rcv] compare with expected
               logNote $ "Service subscriptions for " <> tshow serviceId <> " (" <> tshow qCnt <> " queues)"
             deliverQueueMsg :: (Int, Int, Int, Int) -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO (Int, Int, Int, Int)
             deliverQueueMsg (!qCnt, !msgCnt, !dupCnt, !errCnt) rId = \case
-              Left e -> pure (qCnt + 1, msgCnt, dupCnt, errCnt + 1) -- TODO deliver subscription error
+              Left e -> pure (qCnt + 1, msgCnt, dupCnt, errCnt + 1) -- TODO [certs rcv] deliver subscription error
               Right qMsg_ -> case qMsg_ of
                 Nothing -> pure (qCnt + 1, msgCnt, dupCnt, errCnt)
                 Just (qr, msg) ->
@@ -1823,15 +1824,15 @@ client
 
         subscribeServiceNotifications :: ServiceId -> M s BrokerMsg
         subscribeServiceNotifications serviceId =
-          either ERR (SOKS . snd) <$> sharedSubscribeService SNotifierService serviceId ntfSubscribers ntfServiceSubscribed ntfServiceSubsCount
+          either ERR (uncurry SOKS . snd) <$> sharedSubscribeService SNotifierService serviceId ntfSubscribers ntfServiceSubscribed ntfServiceSubsCount
 
-        sharedSubscribeService :: (PartyI p, ServiceParty p) => SParty p -> ServiceId -> ServerSubscribers s -> (Client s -> TVar Bool) -> (Client s -> TVar Int64) -> M s (Either ErrorType (Bool, Int64))
+        sharedSubscribeService :: (PartyI p, ServiceParty p) => SParty p -> ServiceId -> ServerSubscribers s -> (Client s -> TVar Bool) -> (Client s -> TVar Int64) -> M s (Either ErrorType (Bool, (Int64, IdsHash)))
         sharedSubscribeService party serviceId srvSubscribers clientServiceSubscribed clientServiceSubs = do
           subscribed <- readTVarIO $ clientServiceSubscribed clnt
           liftIO $ runExceptT $
             (subscribed,)
               <$> if subscribed
-                then readTVarIO (clientServiceSubs clnt)
+                then (,B.empty) <$> readTVarIO (clientServiceSubs clnt) -- TODO [certs rcv] get IDs hash
                 else do
                   count' <- ExceptT $ getServiceQueueCount @(StoreQueue s) (queueStore ms) party serviceId
                   incCount <- atomically $ do
@@ -1839,7 +1840,7 @@ client
                     count <- swapTVar (clientServiceSubs clnt) count'
                     pure $ count' - count
                   atomically $ writeTQueue (subQ srvSubscribers) (CSService serviceId incCount, clientId)
-                  pure count'
+                  pure (count', B.empty) -- TODO [certs rcv] get IDs hash
 
         acknowledgeMsg :: MsgId -> StoreQueue s -> QueueRec -> M s (Transmission BrokerMsg)
         acknowledgeMsg msgId q qr =

src/Simplex/Messaging/Transport.hs

@@ -55,6 +55,7 @@ module Simplex.Messaging.Transport
     shortLinksSMPVersion,
     serviceCertsSMPVersion,
     newNtfCredsSMPVersion,
+    rcvServiceSMPVersion,
     simplexMQVersion,
     smpBlockSize,
     TransportConfig (..),
@@ -168,6 +169,7 @@ smpBlockSize = 16384
 -- 15 - short links, with associated data passed in NEW of LSET command (3/30/2025)
 -- 16 - service certificates (5/31/2025)
 -- 17 - create notification credentials with NEW (7/12/2025)
+-- 18 - service subscriptions to messages (8/25/2025)
 
 data SMPVersion
 
@@ -213,6 +215,9 @@ serviceCertsSMPVersion = VersionSMP 16
 newNtfCredsSMPVersion :: VersionSMP
 newNtfCredsSMPVersion = VersionSMP 17
 
+rcvServiceSMPVersion :: VersionSMP
+rcvServiceSMPVersion = VersionSMP 18
+
 minClientSMPRelayVersion :: VersionSMP
 minClientSMPRelayVersion = VersionSMP 6
 
@@ -220,13 +225,13 @@ minServerSMPRelayVersion :: VersionSMP
 minServerSMPRelayVersion = VersionSMP 6
 
 currentClientSMPRelayVersion :: VersionSMP
-currentClientSMPRelayVersion = VersionSMP 17
+currentClientSMPRelayVersion = VersionSMP 18
 
 legacyServerSMPRelayVersion :: VersionSMP
 legacyServerSMPRelayVersion = VersionSMP 6
 
 currentServerSMPRelayVersion :: VersionSMP
-currentServerSMPRelayVersion = VersionSMP 17
+currentServerSMPRelayVersion = VersionSMP 18
 
 -- Max SMP protocol version to be used in e2e encrypted
 -- connection between client and server, as defined by SMP proxy.
@@ -234,7 +239,7 @@ currentServerSMPRelayVersion = VersionSMP 17
 -- to prevent client version fingerprinting by the
 -- destination relays when clients upgrade at different times.
 proxiedSMPRelayVersion :: VersionSMP
-proxiedSMPRelayVersion = VersionSMP 16
+proxiedSMPRelayVersion = VersionSMP 17
 
 -- minimal supported protocol version is 6
 -- TODO remove code that supports sending commands without batching
@@ -818,7 +823,7 @@ smpClientHandshake c ks_ keyHash@(C.KeyHash kh) vRange proxyServer serviceKeys_
           serviceKeys = case serviceKeys_ of
             Just sks | v >= serviceCertsSMPVersion && certificateSent c -> Just sks
             _ -> Nothing
-          clientService = mkClientService <$> serviceKeys
+          clientService = mkClientService v =<< serviceKeys
           hs = SMPClientHandshake {smpVersion = v, keyHash, authPubKey = fst <$> ks_, proxyServer, clientService}
       sendHandshake th hs
       service <- mapM getClientService serviceKeys
@@ -826,10 +831,12 @@ smpClientHandshake c ks_ keyHash@(C.KeyHash kh) vRange proxyServer serviceKeys_
     Nothing -> throwE TEVersion
   where
     th@THandle {params = THandleParams {sessionId}} = smpTHandle c
-    mkClientService :: (ServiceCredentials, C.KeyPairEd25519) -> SMPClientHandshakeService
-    mkClientService (ServiceCredentials {serviceRole, serviceCreds, serviceSignKey}, (k, _)) =
-      let sk = C.signX509 serviceSignKey $ C.publicToX509 k
-       in SMPClientHandshakeService {serviceRole, serviceCertKey = CertChainPubKey (fst serviceCreds) sk}
+    mkClientService :: VersionSMP -> (ServiceCredentials, C.KeyPairEd25519) -> Maybe SMPClientHandshakeService
+    mkClientService v (ServiceCredentials {serviceRole, serviceCreds, serviceSignKey}, (k, _))
+      | serviceRole == SRMessaging && v < rcvServiceSMPVersion = Nothing
+      | otherwise =
+          let sk = C.signX509 serviceSignKey $ C.publicToX509 k
+           in Just SMPClientHandshakeService {serviceRole, serviceCertKey = CertChainPubKey (fst serviceCreds) sk}
     getClientService :: (ServiceCredentials, C.KeyPairEd25519) -> ExceptT TransportError IO THClientService
     getClientService (ServiceCredentials {serviceRole, serviceCertHash}, (_, pk)) =
       getHandshake th >>= \case