Merge branch 'master' into ep/socks-mode

src/Simplex/Messaging/Agent.hs

@@ -4388,8 +4388,9 @@ agentRatchetDecrypt' :: TVar ChaChaDRG -> DB.Connection -> ConnId -> CR.RatchetX
 agentRatchetDecrypt' g db connId rc encAgentMsg = do
   skipped <- liftIO $ getSkippedMsgKeys db connId
   (agentMsgBody_, rc', skippedDiff) <- withExceptT (SEAgentError . cryptoError) $ CR.rcDecrypt g rc skipped encAgentMsg
+  agentMsgBody <- liftEither $ first (SEAgentError . cryptoError) agentMsgBody_
   liftIO $ updateRatchet db connId rc' skippedDiff
-  liftEither $ bimap (SEAgentError . cryptoError) (,CR.rcRcvKEM rc') agentMsgBody_
+  pure (agentMsgBody, CR.rcRcvKEM rc')
 
 newSndQueue :: UserId -> ConnId -> Compatible SMPQueueInfo -> Maybe (C.APrivateAuthKey) -> AM' (NewSndQueue, C.PublicKeyX25519)
 newSndQueue userId connId (Compatible (SMPQueueInfo smpClientVersion SMPQueueAddress {smpServer, senderId, queueMode, dhPublicKey = rcvE2ePubDhKey})) sndKey_ = do

src/Simplex/Messaging/Client.hs

@@ -196,6 +196,7 @@ data PClient v err msg = PClient
     transportHost :: TransportHost,
     tcpConnectTimeout :: NetworkTimeout,
     tcpTimeout :: NetworkTimeout,
+    proxiedRelayVRange :: VersionRange v,
     sendPings :: TVar Bool,
     lastReceived :: TVar UTCTime,
     timeoutErrorCount :: TVar Int,
@@ -241,6 +242,7 @@ smpClientStub g sessionId thVersion thAuth = do
               transportHost = "localhost",
               tcpConnectTimeout,
               tcpTimeout,
+              proxiedRelayVRange = supportedClientSMPRelayVRange,
               sendPings,
               lastReceived,
               timeoutErrorCount,
@@ -482,6 +484,7 @@ data ProtocolClientConfig v = ProtocolClientConfig
     serviceCredentials :: Maybe ServiceCredentials,
     -- | client-server protocol version range
     serverVRange :: VersionRange v,
+    proxiedRelayVRange :: VersionRange v,
     -- | agree shared session secret (used in SMP proxy for additional encryption layer)
     agreeSecret :: Bool,
     -- | Whether connecting client is a proxy server. See comment in ClientHandshake
@@ -500,6 +503,7 @@ defaultClientConfig clientALPN useSNI serverVRange =
       clientALPN,
       serviceCredentials = Nothing,
       serverVRange,
+      proxiedRelayVRange = serverVRange,
       agreeSecret = False,
       proxyServer = False,
       useSNI
@@ -510,6 +514,7 @@ defaultSMPClientConfig :: ProtocolClientConfig SMPVersion
 defaultSMPClientConfig =
   (defaultClientConfig (Just alpnSupportedSMPHandshakes) False supportedClientSMPRelayVRange)
     { defaultTransport = (show defaultSMPPort, transport @TLS),
+      proxiedRelayVRange = supportedClientSMPRelayVRange,
       agreeSecret = True
     }
 {-# INLINE defaultSMPClientConfig #-}
@@ -573,7 +578,7 @@ type SMPTransportSession = TransportSession BrokerMsg
 -- A single queue can be used for multiple 'SMPClient' instances,
 -- as 'SMPServerTransmission' includes server information.
 getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> NetworkRequestMode -> TransportSession msg -> ProtocolClientConfig v -> [HostName] -> Maybe (TBQueue (ServerTransmissionBatch v err msg)) -> UTCTime -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
-getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, clientALPN, serviceCredentials, serverVRange, agreeSecret, proxyServer, useSNI} presetDomains msgQ proxySessTs disconnected = do
+getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, clientALPN, serviceCredentials, serverVRange, proxiedRelayVRange, agreeSecret, proxyServer, useSNI} presetDomains msgQ proxySessTs disconnected = do
   case chooseTransportHost networkConfig (host srv) of
     Right useHost ->
       (getCurrentTime >>= mkProtocolClient useHost >>= runClient useTransport useHost)
@@ -598,6 +603,7 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
             transportHost,
             tcpConnectTimeout,
             tcpTimeout,
+            proxiedRelayVRange,
             sendPings,
             lastReceived,
             timeoutErrorCount,
@@ -1120,10 +1126,10 @@ deleteSMPQueues = okSMPCommands DEL
 -- send PRXY :: SMPServer -> Maybe BasicAuth -> Command Sender
 -- receives PKEY :: SessionId -> X.CertificateChain -> X.SignedExact X.PubKey -> BrokerMsg
 connectSMPProxiedRelay :: SMPClient -> NetworkRequestMode -> SMPServer -> Maybe BasicAuth -> ExceptT SMPClientError IO ProxiedRelay
-connectSMPProxiedRelay c@ProtocolClient {client_ = PClient {tcpConnectTimeout, tcpTimeout}} nm relayServ@ProtocolServer {port = relayPort, keyHash = C.KeyHash kh} proxyAuth =
+connectSMPProxiedRelay c@ProtocolClient {client_ = PClient {tcpConnectTimeout, tcpTimeout, proxiedRelayVRange}} nm relayServ@ProtocolServer {port = relayPort, keyHash = C.KeyHash kh} proxyAuth =
   sendProtocolCommand_ c nm Nothing tOut Nothing NoEntity (Cmd SProxiedClient (PRXY relayServ proxyAuth)) >>= \case
     PKEY sId vr (CertChainPubKey chain key) ->
-      case supportedClientSMPRelayVRange `compatibleVersion` vr of
+      case proxiedRelayVRange `compatibleVersion` vr of
         Nothing -> throwE $ transportErr TEVersion
         Just (Compatible v) -> do
           relayKey <- liftEitherWith (const $ transportErr $ TEHandshake IDENTITY) =<< liftIO (runExceptT $ validateRelay chain key)
@@ -1174,7 +1180,7 @@ instance StrEncoding ProxyClientError where
 -- consider how to process slow responses - is it handled somehow locally or delegated to the caller
 -- this method is used in the client
 -- sends PFWD :: C.PublicKeyX25519 -> EncTransmission -> Command Sender
--- receives PRES :: EncResponse -> BrokerMsg -- proxy to client
+-- receives PRES :: Maybe C.CbNonce -> EncResponse -> BrokerMsg -- proxy to client
 
 -- When client sends message via proxy, there may be one successful scenario and 9 error scenarios
 -- as shown below (WTF stands for unexpected response, ??? for response that failed to parse).
@@ -1233,14 +1239,14 @@ proxySMPCommand c@ProtocolClient {thParams = proxyThParams, client_ = PClient {c
     TBError e _ : _ -> throwE $ PCETransportError e
     TBTransmission s _ : _ -> pure s
     TBTransmissions s _ _ : _ -> pure s
-  et <- liftEitherWith PCECryptoError $ EncTransmission <$> C.cbEncrypt cmdSecret nonce b paddedProxiedTLength
+  et <- liftEitherWith PCECryptoError $ EncTransmission <$> C.cbEncrypt cmdSecret (encTransmissionNonce v nonce) b paddedProxiedTLength
   -- proxy interaction errors are wrapped
   let tOut = Just $ 2 * netTimeoutInt tcpTimeout nm
   tryE (sendProtocolCommand_ c nm (Just nonce) tOut Nothing (EntityId sessionId) (Cmd SProxiedClient (PFWD v cmdPubKey et))) >>= \case
     Right r -> case r of
-      PRES (EncResponse er) -> do
+      PRES nonce_ (EncResponse er) -> do
         -- server interaction errors are thrown directly
-        t' <- liftEitherWith PCECryptoError $ C.cbDecrypt cmdSecret (C.reverseNonce nonce) er
+        t' <- liftEitherWith PCECryptoError $ C.cbDecrypt cmdSecret (fromMaybe (C.reverseNonce nonce) nonce_) er
         case tParse serverThParams t' of
           t'' :| [] -> case tDecodeClient serverThParams t'' of
             (_, _, cmd) -> case cmd of
@@ -1258,10 +1264,10 @@ proxySMPCommand c@ProtocolClient {thParams = proxyThParams, client_ = PClient {c
 
 -- this method is used in the proxy
 -- sends RFWD :: EncFwdTransmission -> Command Sender
--- receives RRES :: EncFwdResponse -> BrokerMsg
+-- receives RRES :: Maybe C.CbNonce -> EncFwdResponse -> BrokerMsg
 -- proxy should send PRES to the client with EncResponse
 -- Always uses background timeout mode
-forwardSMPTransmission :: SMPClient -> CorrId -> VersionSMP -> C.PublicKeyX25519 -> EncTransmission -> ExceptT SMPClientError IO EncResponse
+forwardSMPTransmission :: SMPClient -> CorrId -> VersionSMP -> C.PublicKeyX25519 -> EncTransmission -> ExceptT SMPClientError IO (Maybe C.CbNonce, EncResponse)
 forwardSMPTransmission c@ProtocolClient {thParams, client_ = PClient {clientCorrId = g}} fwdCorrId fwdVersion fwdKey fwdTransmission = do
   -- prepare params
   sessSecret <- case thAuth thParams of
@@ -1273,11 +1279,11 @@ forwardSMPTransmission c@ProtocolClient {thParams, client_ = PClient {clientCorr
       eft = EncFwdTransmission $ C.cbEncryptNoPad sessSecret nonce (smpEncode fwdT)
   -- send
   sendProtocolCommand_ c NRMBackground (Just nonce) Nothing Nothing NoEntity (Cmd SProxyService (RFWD eft)) >>= \case
-    RRES (EncFwdResponse efr) -> do
+    RRES nonce_ (EncFwdResponse efr) -> do
       -- unwrap
       r' <- liftEitherWith PCECryptoError $ C.cbDecryptNoPad sessSecret (C.reverseNonce nonce) efr
       FwdResponse {fwdCorrId = _, fwdResponse} <- liftEitherWith (const $ PCEResponseError BLOCK) $ smpDecode r'
-      pure fwdResponse
+      pure (nonce_, fwdResponse)
     r -> throwE $ unexpectedResponse r
 
 -- get queue information - always sent interactively

src/Simplex/Messaging/Protocol.hs

@@ -168,6 +168,7 @@ module Simplex.Messaging.Protocol
     EncFwdTransmission (..),
     EncResponse (..),
     EncTransmission (..),
+    encTransmissionNonce,
     FwdResponse (..),
     FwdTransmission (..),
     NameRecord (..),
@@ -241,6 +242,7 @@ import Data.Attoparsec.ByteString.Char8 (Parser, (<?>))
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (bimap, first)
 import Data.Bits (xor)
+import qualified Data.ByteArray as BA
 import qualified Data.ByteString as BS
 import qualified Data.ByteString.Base64 as B64
 import Data.ByteString.Char8 (ByteString)
@@ -279,7 +281,7 @@ import Simplex.Messaging.ServiceScheme
 import Simplex.Messaging.SimplexName (LabelHash, SimplexDomain (..), SimplexTLD (..), fullDomainName, labelHash)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (TransportHost, TransportHosts (..))
-import Simplex.Messaging.Util (bshow, eitherToMaybe, safeDecodeUtf8, (<$?>))
+import Simplex.Messaging.Util (bshow, eitherToMaybe, packZipWith, safeDecodeUtf8, (<$?>))
 import Simplex.Messaging.Version
 import Simplex.Messaging.Version.Internal
 
@@ -701,6 +703,11 @@ instance Encoding NewNtfCreds where
 newtype EncTransmission = EncTransmission ByteString
   deriving (Show)
 
+encTransmissionNonce :: VersionSMP -> C.CbNonce -> C.CbNonce
+encTransmissionNonce v nonce@(C.CbNonce s)
+  | v >= fwdNoncesSMPVersion = C.cbNonce $ packZipWith xor (smpEncode v) s <> BS.drop 2 s
+  | otherwise = nonce
+
 data FwdTransmission = FwdTransmission
   { fwdCorrId :: CorrId,
     fwdVersion :: VersionSMP,
@@ -736,8 +743,8 @@ data BrokerMsg where
   NMSG :: C.CbNonce -> EncNMsgMeta -> BrokerMsg
   -- Should include certificate chain
   PKEY :: SessionId -> VersionRangeSMP -> CertChainPubKey -> BrokerMsg -- TLS-signed server key for proxy shared secret and initial sender key
-  RRES :: EncFwdResponse -> BrokerMsg -- relay to proxy
-  PRES :: EncResponse -> BrokerMsg -- proxy to client
+  RRES :: Maybe C.CbNonce -> EncFwdResponse -> BrokerMsg -- relay to proxy
+  PRES :: Maybe C.CbNonce -> EncResponse -> BrokerMsg -- proxy to client
   END :: BrokerMsg
   ENDS :: Int64 -> IdsHash -> BrokerMsg
   DELD :: BrokerMsg
@@ -1323,7 +1330,10 @@ instance ProtocolTypeI p => FromJSON (ProtocolServer p) where
   parseJSON = strParseJSON "ProtocolServer"
 
 newtype BasicAuth = BasicAuth {unBasicAuth :: ByteString}
-  deriving (Eq, Ord, Show)
+  deriving (Ord, Show)
+
+instance Eq BasicAuth where
+  BasicAuth s == BasicAuth s' = BA.constEq s s'
 
 instance IsString BasicAuth where fromString = BasicAuth . B.pack
 
@@ -1981,8 +1991,8 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
     NID nId srvNtfDh -> e (NID_, ' ', nId, srvNtfDh)
     NMSG nmsgNonce encNMsgMeta -> e (NMSG_, ' ', nmsgNonce, encNMsgMeta)
     PKEY sid vr certKey -> e (PKEY_, ' ', sid, vr, certKey)
-    RRES (EncFwdResponse encBlock) -> e (RRES_, ' ', Tail encBlock)
-    PRES (EncResponse encBlock) -> e (PRES_, ' ', Tail encBlock)
+    RRES nonce_ (EncFwdResponse encBlock) -> fwdResp RRES_ nonce_ encBlock
+    PRES nonce_ (EncResponse encBlock) -> fwdResp PRES_ nonce_ encBlock
     END -> e END_
     ENDS n idsHash -> serviceResp ENDS_ n idsHash
     DELD -> e DELD_
@@ -2006,6 +2016,9 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
       serviceResp tag n idsHash
         | v >= rcvServiceSMPVersion = e (tag, ' ', n, idsHash)
         | otherwise = e (tag, ' ', n)
+      fwdResp tag nonce_ encBlock
+        | v >= fwdNoncesSMPVersion = e (tag, ' ', nonce_, Tail encBlock)
+        | otherwise = e (tag, ' ', Tail encBlock)
 
   protocolP v = \case
     MSG_ -> do
@@ -2037,8 +2050,8 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
     NID_ -> NID <$> _smpP <*> smpP
     NMSG_ -> NMSG <$> _smpP <*> smpP
     PKEY_ -> PKEY <$> _smpP <*> smpP <*> smpP
-    RRES_ -> RRES <$> (EncFwdResponse . unTail <$> _smpP)
-    PRES_ -> PRES <$> (EncResponse . unTail <$> _smpP)
+    RRES_ -> fwdRespP RRES EncFwdResponse
+    PRES_ -> fwdRespP PRES EncResponse
     END_ -> pure END
     ENDS_ -> serviceRespP ENDS
     DELD_ -> pure DELD
@@ -2055,6 +2068,9 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
       serviceRespP resp
         | v >= rcvServiceSMPVersion = resp <$> _smpP <*> smpP
         | otherwise = resp <$> _smpP <*> pure mempty
+      fwdRespP :: (Maybe C.CbNonce -> a -> BrokerMsg) -> (ByteString -> a) -> Parser BrokerMsg
+      fwdRespP resp enc = resp <$> (A.space *> nonceP) <*> (enc <$> A.takeByteString)
+      nonceP = if v >= fwdNoncesSMPVersion then smpP else pure Nothing
 
   fromProtocolError = \case
     PECmdSyntax -> CMD SYNTAX
@@ -2071,7 +2087,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
     -- PONG response must not have queue ID
     PONG -> noEntityMsg
     PKEY {} -> noEntityMsg
-    RRES _ -> noEntityMsg
+    RRES {} -> noEntityMsg
     ALLS -> noEntityMsg
     RNAME {} -> noEntityMsg
     -- other broker responses must have queue ID

src/Simplex/Messaging/Server.hs

@@ -1092,6 +1092,7 @@ controlPortAuth h user admin role auth = do
   readTVarIO role >>= \case
     CPRNone -> do
       atomically $ writeTVar role $! newRole
+      when (newRole == CPRNone) $ logWarn "ControlPort: failed auth"
       hPutStrLn h $ currentRole newRole
     r -> hPutStrLn h $ currentRole r <> if r == newRole then "" else ", start new session to change."
   where
@@ -1461,7 +1462,7 @@ client
             inc own pRequests
             forkProxiedCmd $ do
               liftIO (runExceptT (forwardSMPTransmission smp corrId fwdV pubKey encBlock) `E.catches` clientHandlers)  >>= \case
-                Right r -> PRES r <$ inc own pSuccesses
+                Right (nonce_, r) -> PRES nonce_ r <$ inc own pSuccesses
                 Left e -> ERR (smpProxyError e) <$ case e of
                   PCEProtocolError {} -> inc own pSuccesses
                   _ -> inc own pErrorsOther
@@ -2130,7 +2131,7 @@ client
           unless (fwdVersion `isCompatible` thServerVRange thParams') $ throwE $ transportErr TEVersion
           let clientSecret = C.dh' fwdKey serverPrivKey
               clientNonce = C.cbNonce $ bs fwdCorrId
-          b <- liftEitherWith (const CRYPTO) $ C.cbDecrypt clientSecret clientNonce et
+          b <- liftEitherWith (const CRYPTO) $ C.cbDecrypt clientSecret (encTransmissionNonce fwdVersion clientNonce) et
           let clntTHParams = smpTHParamsSetVersion fwdVersion thParams'
           -- only allowing single forwarded transactions
           t' <- case tParse clntTHParams b of
@@ -2143,9 +2144,13 @@ client
                   TBError _ _ : _ -> throwE BLOCK
                   TBTransmission b' _ : _ -> pure b'
                   TBTransmissions b' _ _ : _ -> pure b'
-                r2 <- liftEitherWith (const BLOCK) $ EncResponse <$> C.cbEncrypt clientSecret (C.reverseNonce clientNonce) r' paddedProxiedTLength
+                nonce_ <-
+                  if fwdVersion >= fwdNoncesSMPVersion
+                    then Just <$> (atomically . C.randomCbNonce =<< asks random)
+                    else pure Nothing
+                r2 <- liftEitherWith (const BLOCK) $ EncResponse <$> C.cbEncrypt clientSecret (fromMaybe (C.reverseNonce clientNonce) nonce_) r' paddedProxiedTLength
                 let fr = FwdResponse {fwdCorrId, fwdResponse = r2}
-                pure $ RRES $ EncFwdResponse $ C.cbEncryptNoPad sessSecret (C.reverseNonce proxyNonce) (smpEncode fr)
+                pure $ RRES nonce_ $ EncFwdResponse $ C.cbEncryptNoPad sessSecret (C.reverseNonce proxyNonce) (smpEncode fr)
           -- the inner response, or Nothing if forked (RSLV).
           r_ <- lift (rejectOrVerify clntThAuth t') >>= \case
             -- rejectOrVerify filters allowed commands, no need to repeat it here.

src/Simplex/Messaging/Server/MsgStore/Postgres.hs

@@ -114,7 +114,9 @@ instance MsgStoreClass PostgresMsgStore where
     where
       st = dbStore $ queueStore_ ms
       oldMsg = now - ttl
-      batchSize = 10000 :: Int
+      -- expired messages read per page in expire_old_messages, and the page is one
+      -- transaction: queues in it stay row-locked against SEND and ACK until it commits.
+      batchSize = 100 :: Int
       toMessageStats (expiredMsgsCount, storedMsgsCount, storedQueues) =
         MessageStats {expiredMsgsCount, storedMsgsCount, storedQueues}
 
@@ -363,8 +365,8 @@ deleteAllMessages ms =
       db
       [sql|
         UPDATE msg_queues
-        SET msg_queue_size = 0, msg_can_write = TRUE, msg_queue_expire = FALSE
-        WHERE msg_queue_size != 0 OR msg_can_write = FALSE OR msg_queue_expire = TRUE
+        SET msg_queue_size = 0, msg_can_write = TRUE
+        WHERE msg_queue_size != 0 OR msg_can_write = FALSE
       |]
 
 updateQueueCounts :: PostgresMsgStore -> IO ()
@@ -384,16 +386,15 @@ updateQueueCounts ms =
       db
       [sql|
         UPDATE msg_queues
-        SET msg_queue_size = 0, msg_can_write = TRUE, msg_queue_expire = FALSE
-        WHERE msg_queue_size != 0 OR msg_can_write = FALSE OR msg_queue_expire = TRUE
+        SET msg_queue_size = 0, msg_can_write = TRUE
+        WHERE msg_queue_size != 0 OR msg_can_write = FALSE
       |]
     void $ DB.execute_
       db
       [sql|
         UPDATE msg_queues q
         SET msg_queue_size = s.size,
-            msg_can_write = s.quota_count = 0,
-            msg_queue_expire = s.size > s.quota_count
+            msg_can_write = s.quota_count = 0
         FROM queue_stats s
         WHERE q.recipient_id = s.recipient_id
       |]

src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs

@@ -22,7 +22,8 @@ serverSchemaMigrations =
     ("20250903_store_messages", m20250903_store_messages, Just down_m20250903_store_messages),
     ("20250915_queue_ids_hash", m20250915_queue_ids_hash, Just down_m20250915_queue_ids_hash),
     ("20260916_prometheus_indexes", m20260916_prometheus_indexes, Just down_m20260916_prometheus_indexes),
-    ("20260917_msg_queues_hot", m20260917_msg_queues_hot, Just down_m20260917_msg_queues_hot)
+    ("20260917_msg_queues_hot", m20260917_msg_queues_hot, Just down_m20260917_msg_queues_hot),
+    ("20260918_expire_messages", m20260918_expire_messages, Just down_m20260918_expire_messages)
   ]
 
 -- | The list of migrations in ascending order by date
@@ -734,3 +735,504 @@ ALTER TABLE msg_queues RESET (fillfactor, autovacuum_vacuum_scale_factor, autova
 ALTER TABLE messages RESET (autovacuum_vacuum_scale_factor, autovacuum_analyze_scale_factor, toast.autovacuum_vacuum_scale_factor);
 ALTER TABLE services RESET (fillfactor, autovacuum_vacuum_threshold, autovacuum_vacuum_scale_factor);
     |]
+
+m20260918_expire_messages :: Text
+m20260918_expire_messages =
+  [r|
+CREATE INDEX idx_messages_expire ON messages (msg_ts, recipient_id) WHERE NOT msg_quota;
+
+DROP INDEX idx_messages_recipient_id_msg_ts;
+
+DROP INDEX idx_msg_queues_expire;
+
+DROP PROCEDURE expire_old_messages(bigint, integer);
+
+CREATE PROCEDURE expire_old_messages(IN p_old_ts bigint, IN batch_size integer, OUT r_expired_msgs_count bigint, OUT r_stored_msgs_count bigint, OUT r_stored_queues bigint)
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  rids BYTEA[];
+  rid BYTEA;
+  last_ts BIGINT := -1;
+  last_rid BYTEA := '\x';
+  next_ts BIGINT;
+  next_rid BYTEA;
+  del_count BIGINT;
+  total_deleted BIGINT := 0;
+BEGIN
+  LOOP
+    -- The page is scanned in (msg_ts, recipient_id) order, so its last row is the next
+    -- keyset cursor. Advancing it past every row read, including queues left unexpired
+    -- because delete_expired_msgs skipped a locked row or raised, is what terminates the
+    -- loop; re-reading from the start instead would repeat those queues forever.
+    SELECT array_agg(DISTINCT recipient_id),
+           (array_agg(msg_ts ORDER BY msg_ts DESC, recipient_id DESC))[1],
+           (array_agg(recipient_id ORDER BY msg_ts DESC, recipient_id DESC))[1]
+    INTO rids, next_ts, next_rid
+    FROM (
+      SELECT msg_ts, recipient_id
+      FROM messages
+      WHERE NOT msg_quota
+        AND msg_ts < p_old_ts
+        AND (msg_ts, recipient_id) > (last_ts, last_rid)
+      ORDER BY msg_ts ASC, recipient_id ASC
+      LIMIT batch_size
+    ) m;
+
+    EXIT WHEN rids IS NULL;
+
+    FOREACH rid IN ARRAY rids
+    LOOP
+      BEGIN
+        del_count := delete_expired_msgs(rid, p_old_ts);
+        total_deleted := total_deleted + del_count;
+      EXCEPTION WHEN OTHERS THEN
+        RAISE WARNING 'STORE, expire_old_messages, error expiring queue %: %', encode(rid, 'base64'), SQLERRM;
+        CONTINUE;
+      END;
+    END LOOP;
+    COMMIT;
+
+    last_ts := next_ts;
+    last_rid := next_rid;
+  END LOOP;
+
+  r_expired_msgs_count := total_deleted;
+  r_stored_msgs_count := (SELECT COUNT(1) FROM messages);
+  r_stored_queues := (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL);
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION delete_expired_msgs(p_recipient_id bytea, p_old_ts bigint) RETURNS bigint
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  q_size BIGINT;
+  keep_min_id BIGINT;
+  del_count BIGINT;
+BEGIN
+  SELECT msg_queue_size INTO q_size
+  FROM msg_queues
+  WHERE recipient_id = p_recipient_id AND deleted_at IS NULL
+  FOR UPDATE SKIP LOCKED;
+
+  IF NOT FOUND OR q_size = 0 THEN
+    RETURN 0;
+  END IF;
+
+  SELECT MIN(message_id) INTO keep_min_id
+  FROM messages WHERE recipient_id = p_recipient_id AND msg_ts >= p_old_ts AND msg_quota = FALSE;
+
+  IF keep_min_id IS NULL THEN
+    DELETE FROM messages WHERE recipient_id = p_recipient_id AND msg_quota = FALSE;
+  ELSE
+    DELETE FROM messages WHERE recipient_id = p_recipient_id AND message_id < keep_min_id AND msg_quota = FALSE;
+  END IF;
+
+  GET DIAGNOSTICS del_count = ROW_COUNT;
+  IF del_count > 0 THEN
+    UPDATE msg_queues
+    SET msg_can_write = msg_can_write OR msg_queue_size <= del_count,
+        msg_queue_size = GREATEST(msg_queue_size - del_count, 0)
+    WHERE recipient_id = p_recipient_id;
+  END IF;
+  RETURN del_count;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION write_message(p_recipient_id bytea, p_msg_id bytea, p_msg_ts bigint, p_msg_quota boolean, p_msg_ntf_flag boolean, p_msg_body bytea, p_quota integer) RETURNS TABLE(quota_written boolean, was_empty boolean)
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  q_can_write BOOLEAN;
+  q_size BIGINT;
+BEGIN
+  SELECT msg_can_write, msg_queue_size INTO q_can_write, q_size
+  FROM msg_queues
+  WHERE recipient_id = p_recipient_id AND deleted_at IS NULL
+  FOR UPDATE;
+
+  IF q_can_write OR q_size = 0 THEN
+    quota_written := p_msg_quota OR q_size >= p_quota;
+    was_empty := q_size = 0;
+
+    INSERT INTO messages(recipient_id, msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body)
+    VALUES (p_recipient_id, p_msg_id, p_msg_ts, quota_written, p_msg_ntf_flag AND NOT quota_written, CASE WHEN quota_written THEN '' :: BYTEA ELSE p_msg_body END);
+
+    UPDATE msg_queues
+    SET msg_can_write = NOT quota_written,
+        msg_queue_size = msg_queue_size + 1
+    WHERE recipient_id = p_recipient_id;
+
+    RETURN QUERY VALUES (quota_written, was_empty);
+  END IF;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION try_del_msg(p_recipient_id bytea, p_msg_id bytea) RETURNS TABLE(r_msg_id bytea, r_msg_ts bigint, r_msg_quota boolean, r_msg_ntf_flag boolean, r_msg_body bytea)
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  q_size BIGINT;
+  msg RECORD;
+BEGIN
+  SELECT msg_queue_size INTO q_size
+  FROM msg_queues
+  WHERE recipient_id = p_recipient_id AND deleted_at IS NULL
+  FOR UPDATE;
+
+  IF NOT FOUND THEN
+    RETURN;
+  END IF;
+
+  SELECT message_id, msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body
+  INTO msg
+  FROM messages
+  WHERE recipient_id = p_recipient_id
+  ORDER BY message_id ASC LIMIT 1;
+
+  IF NOT FOUND THEN
+    IF q_size != 0 THEN
+      UPDATE msg_queues
+      SET msg_can_write = TRUE,
+          msg_queue_size = 0
+      WHERE recipient_id = p_recipient_id;
+    END IF;
+    RETURN;
+  END IF;
+
+  IF msg.msg_id = p_msg_id THEN
+    DELETE FROM messages WHERE message_id = msg.message_id;
+    IF FOUND THEN
+      UPDATE msg_queues
+      SET msg_can_write = msg_can_write OR msg_queue_size <= 1,
+          msg_queue_size = GREATEST(msg_queue_size - 1, 0)
+      WHERE recipient_id = p_recipient_id;
+      RETURN QUERY VALUES (msg.msg_id, msg.msg_ts, msg.msg_quota, msg.msg_ntf_flag, msg.msg_body);
+    END IF;
+  END IF;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION try_del_peek_msg(p_recipient_id bytea, p_msg_id bytea) RETURNS TABLE(r_msg_id bytea, r_msg_ts bigint, r_msg_quota boolean, r_msg_ntf_flag boolean, r_msg_body bytea)
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  q_size BIGINT;
+  msg RECORD;
+  msg_deleted BOOLEAN;
+BEGIN
+  SELECT msg_queue_size INTO q_size
+  FROM msg_queues
+  WHERE recipient_id = p_recipient_id AND deleted_at IS NULL
+  FOR UPDATE;
+
+  IF NOT FOUND THEN
+    RETURN;
+  END IF;
+
+  SELECT message_id, msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body
+  INTO msg
+  FROM messages
+  WHERE recipient_id = p_recipient_id
+  ORDER BY message_id ASC LIMIT 1;
+
+  IF NOT FOUND THEN
+    IF q_size != 0 THEN
+      UPDATE msg_queues
+      SET msg_can_write = TRUE,
+          msg_queue_size = 0
+      WHERE recipient_id = p_recipient_id;
+    END IF;
+    RETURN;
+  END IF;
+
+  IF msg.msg_id = p_msg_id THEN
+    DELETE FROM messages WHERE message_id = msg.message_id;
+
+    msg_deleted := FOUND;
+    IF msg_deleted THEN
+      RETURN QUERY VALUES (msg.msg_id, msg.msg_ts, msg.msg_quota, msg.msg_ntf_flag, msg.msg_body);
+    END IF;
+
+    SELECT msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body
+    INTO msg
+    FROM messages
+    WHERE recipient_id = p_recipient_id
+    ORDER BY message_id ASC LIMIT 1;
+
+    IF FOUND THEN
+      RETURN QUERY VALUES (msg.msg_id, msg.msg_ts, msg.msg_quota, msg.msg_ntf_flag, msg.msg_body);
+      IF msg_deleted THEN
+        UPDATE msg_queues
+        SET msg_can_write = msg_can_write OR msg_queue_size <= 1,
+            msg_queue_size = GREATEST(msg_queue_size - 1, 0)
+        WHERE recipient_id = p_recipient_id;
+      END IF;
+    ELSIF msg_deleted OR q_size != 0 THEN
+      UPDATE msg_queues
+      SET msg_can_write = TRUE,
+          msg_queue_size = 0
+      WHERE recipient_id = p_recipient_id;
+    END IF;
+  ELSE
+    RETURN QUERY VALUES (msg.msg_id, msg.msg_ts, msg.msg_quota, msg.msg_ntf_flag, msg.msg_body);
+  END IF;
+END;
+$$;
+
+ALTER TABLE msg_queues DROP COLUMN msg_queue_expire;
+    |]
+
+down_m20260918_expire_messages :: Text
+down_m20260918_expire_messages =
+  [r|
+ALTER TABLE msg_queues ADD COLUMN msg_queue_expire boolean NOT NULL DEFAULT FALSE;
+
+-- ADD COLUMN already defaulted every row to FALSE, so the backfill only needs the rows
+-- that become TRUE, which avoids rewriting the whole table. The flag is only read to pick
+-- queues for delete_expired_msgs, and that returns 0 without deleting when msg_queue_size
+-- is 0, so excluding empty queues here cannot change what expires.
+UPDATE msg_queues q
+SET msg_queue_expire = TRUE
+WHERE msg_queue_size > 0
+  AND EXISTS (SELECT 1 FROM messages m WHERE m.recipient_id = q.recipient_id AND NOT m.msg_quota);
+
+CREATE INDEX idx_msg_queues_expire ON msg_queues (recipient_id) WHERE deleted_at IS NULL AND msg_queue_expire;
+
+CREATE INDEX idx_messages_recipient_id_msg_ts ON messages (recipient_id, msg_ts);
+
+DROP INDEX idx_messages_expire;
+
+CREATE OR REPLACE FUNCTION delete_expired_msgs(p_recipient_id bytea, p_old_ts bigint) RETURNS bigint
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  q_size BIGINT;
+  keep_min_id BIGINT;
+  del_count BIGINT;
+BEGIN
+  SELECT msg_queue_size INTO q_size
+  FROM msg_queues
+  WHERE recipient_id = p_recipient_id AND deleted_at IS NULL
+  FOR UPDATE SKIP LOCKED;
+
+  IF NOT FOUND OR q_size = 0 THEN
+    RETURN 0;
+  END IF;
+
+  SELECT MIN(message_id) INTO keep_min_id
+  FROM messages WHERE recipient_id = p_recipient_id AND msg_ts >= p_old_ts AND msg_quota = FALSE;
+
+  IF keep_min_id IS NULL THEN
+    DELETE FROM messages WHERE recipient_id = p_recipient_id AND msg_quota = FALSE;
+  ELSE
+    DELETE FROM messages WHERE recipient_id = p_recipient_id AND message_id < keep_min_id AND msg_quota = FALSE;
+  END IF;
+
+  GET DIAGNOSTICS del_count = ROW_COUNT;
+  IF del_count > 0 THEN
+    UPDATE msg_queues
+    SET msg_can_write = msg_can_write OR msg_queue_size <= del_count,
+        msg_queue_expire = msg_queue_size > del_count AND keep_min_id IS NOT NULL,
+        msg_queue_size = GREATEST(msg_queue_size - del_count, 0)
+    WHERE recipient_id = p_recipient_id;
+  END IF;
+  RETURN del_count;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION write_message(p_recipient_id bytea, p_msg_id bytea, p_msg_ts bigint, p_msg_quota boolean, p_msg_ntf_flag boolean, p_msg_body bytea, p_quota integer) RETURNS TABLE(quota_written boolean, was_empty boolean)
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  q_can_write BOOLEAN;
+  q_size BIGINT;
+BEGIN
+  SELECT msg_can_write, msg_queue_size INTO q_can_write, q_size
+  FROM msg_queues
+  WHERE recipient_id = p_recipient_id AND deleted_at IS NULL
+  FOR UPDATE;
+
+  IF q_can_write OR q_size = 0 THEN
+    quota_written := p_msg_quota OR q_size >= p_quota;
+    was_empty := q_size = 0;
+
+    INSERT INTO messages(recipient_id, msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body)
+    VALUES (p_recipient_id, p_msg_id, p_msg_ts, quota_written, p_msg_ntf_flag AND NOT quota_written, CASE WHEN quota_written THEN '' :: BYTEA ELSE p_msg_body END);
+
+    UPDATE msg_queues
+    SET msg_can_write = NOT quota_written,
+        msg_queue_expire = TRUE,
+        msg_queue_size = msg_queue_size + 1
+    WHERE recipient_id = p_recipient_id;
+
+    RETURN QUERY VALUES (quota_written, was_empty);
+  END IF;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION try_del_msg(p_recipient_id bytea, p_msg_id bytea) RETURNS TABLE(r_msg_id bytea, r_msg_ts bigint, r_msg_quota boolean, r_msg_ntf_flag boolean, r_msg_body bytea)
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  q_size BIGINT;
+  msg RECORD;
+BEGIN
+  SELECT msg_queue_size INTO q_size
+  FROM msg_queues
+  WHERE recipient_id = p_recipient_id AND deleted_at IS NULL
+  FOR UPDATE;
+
+  IF NOT FOUND THEN
+    RETURN;
+  END IF;
+
+  SELECT message_id, msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body
+  INTO msg
+  FROM messages
+  WHERE recipient_id = p_recipient_id
+  ORDER BY message_id ASC LIMIT 1;
+
+  IF NOT FOUND THEN
+    IF q_size != 0 THEN
+      UPDATE msg_queues
+      SET msg_can_write = TRUE,
+          msg_queue_expire = FALSE,
+          msg_queue_size = 0
+      WHERE recipient_id = p_recipient_id;
+    END IF;
+    RETURN;
+  END IF;
+
+  IF msg.msg_id = p_msg_id THEN
+    DELETE FROM messages WHERE message_id = msg.message_id;
+    IF FOUND THEN
+      UPDATE msg_queues
+      SET msg_can_write = msg_can_write OR msg_queue_size <= 1,
+          msg_queue_expire = msg_queue_size > 1,
+          msg_queue_size = GREATEST(msg_queue_size - 1, 0)
+      WHERE recipient_id = p_recipient_id;
+      RETURN QUERY VALUES (msg.msg_id, msg.msg_ts, msg.msg_quota, msg.msg_ntf_flag, msg.msg_body);
+    END IF;
+  END IF;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION try_del_peek_msg(p_recipient_id bytea, p_msg_id bytea) RETURNS TABLE(r_msg_id bytea, r_msg_ts bigint, r_msg_quota boolean, r_msg_ntf_flag boolean, r_msg_body bytea)
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  q_size BIGINT;
+  msg RECORD;
+  msg_deleted BOOLEAN;
+BEGIN
+  SELECT msg_queue_size INTO q_size
+  FROM msg_queues
+  WHERE recipient_id = p_recipient_id AND deleted_at IS NULL
+  FOR UPDATE;
+
+  IF NOT FOUND THEN
+    RETURN;
+  END IF;
+
+  SELECT message_id, msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body
+  INTO msg
+  FROM messages
+  WHERE recipient_id = p_recipient_id
+  ORDER BY message_id ASC LIMIT 1;
+
+  IF NOT FOUND THEN
+    IF q_size != 0 THEN
+      UPDATE msg_queues
+      SET msg_can_write = TRUE,
+          msg_queue_expire = FALSE,
+          msg_queue_size = 0
+      WHERE recipient_id = p_recipient_id;
+    END IF;
+    RETURN;
+  END IF;
+
+  IF msg.msg_id = p_msg_id THEN
+    DELETE FROM messages WHERE message_id = msg.message_id;
+
+    msg_deleted := FOUND;
+    IF msg_deleted THEN
+      RETURN QUERY VALUES (msg.msg_id, msg.msg_ts, msg.msg_quota, msg.msg_ntf_flag, msg.msg_body);
+    END IF;
+
+    SELECT msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body
+    INTO msg
+    FROM messages
+    WHERE recipient_id = p_recipient_id
+    ORDER BY message_id ASC LIMIT 1;
+
+    IF FOUND THEN
+      RETURN QUERY VALUES (msg.msg_id, msg.msg_ts, msg.msg_quota, msg.msg_ntf_flag, msg.msg_body);
+      IF msg_deleted THEN
+        UPDATE msg_queues
+        SET msg_can_write = msg_can_write OR msg_queue_size <= 1,
+            msg_queue_expire = msg_queue_size > 1,
+            msg_queue_size = GREATEST(msg_queue_size - 1, 0)
+        WHERE recipient_id = p_recipient_id;
+      END IF;
+    ELSIF msg_deleted OR q_size != 0 THEN
+      UPDATE msg_queues
+      SET msg_can_write = TRUE,
+          msg_queue_expire = FALSE,
+          msg_queue_size = 0
+      WHERE recipient_id = p_recipient_id;
+    END IF;
+  ELSE
+    RETURN QUERY VALUES (msg.msg_id, msg.msg_ts, msg.msg_quota, msg.msg_ntf_flag, msg.msg_body);
+  END IF;
+END;
+$$;
+
+DROP PROCEDURE expire_old_messages(bigint, integer);
+
+CREATE PROCEDURE expire_old_messages(IN p_old_ts bigint, IN batch_size integer, OUT r_expired_msgs_count bigint, OUT r_stored_msgs_count bigint, OUT r_stored_queues bigint)
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  rids BYTEA[];
+  rid BYTEA;
+  last_rid BYTEA := '\x';
+  del_count BIGINT;
+  total_deleted BIGINT := 0;
+BEGIN
+  LOOP
+    SELECT array_agg(recipient_id)
+    INTO rids
+    FROM (
+      SELECT recipient_id
+      FROM msg_queues
+      WHERE deleted_at IS NULL
+        AND msg_queue_expire = TRUE
+        AND recipient_id > last_rid
+      ORDER BY recipient_id ASC
+      LIMIT batch_size
+    ) qs;
+
+    EXIT WHEN rids IS NULL OR cardinality(rids) = 0;
+
+    FOREACH rid IN ARRAY rids
+    LOOP
+      BEGIN
+        del_count := delete_expired_msgs(rid, p_old_ts);
+        total_deleted := total_deleted + del_count;
+      EXCEPTION WHEN OTHERS THEN
+        RAISE WARNING 'STORE, expire_old_messages, error expiring queue %: %', encode(rid, 'base64'), SQLERRM;
+        CONTINUE;
+      END;
+      COMMIT;
+    END LOOP;
+    last_rid := rids[cardinality(rids)];
+  END LOOP;
+
+  r_expired_msgs_count := total_deleted;
+  r_stored_msgs_count := (SELECT COUNT(1) FROM messages);
+  r_stored_queues := (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL);
+END;
+$$;
+    |]

src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql

@@ -47,7 +47,6 @@ BEGIN
   IF del_count > 0 THEN
     UPDATE msg_queues
     SET msg_can_write = msg_can_write OR msg_queue_size <= del_count,
-        msg_queue_expire = msg_queue_size > del_count AND keep_min_id IS NOT NULL,
         msg_queue_size = GREATEST(msg_queue_size - del_count, 0)
     WHERE recipient_id = p_recipient_id;
   END IF;
@@ -63,24 +62,33 @@ CREATE PROCEDURE smp_server.expire_old_messages(IN p_old_ts bigint, IN batch_siz
 DECLARE
   rids BYTEA[];
   rid BYTEA;
+  last_ts BIGINT := -1;
   last_rid BYTEA := '\x';
+  next_ts BIGINT;
+  next_rid BYTEA;
   del_count BIGINT;
   total_deleted BIGINT := 0;
 BEGIN
   LOOP
-    SELECT array_agg(recipient_id)
-    INTO rids
+    -- The page is scanned in (msg_ts, recipient_id) order, so its last row is the next
+    -- keyset cursor. Advancing it past every row read, including queues left unexpired
+    -- because delete_expired_msgs skipped a locked row or raised, is what terminates the
+    -- loop; re-reading from the start instead would repeat those queues forever.
+    SELECT array_agg(DISTINCT recipient_id),
+           (array_agg(msg_ts ORDER BY msg_ts DESC, recipient_id DESC))[1],
+           (array_agg(recipient_id ORDER BY msg_ts DESC, recipient_id DESC))[1]
+    INTO rids, next_ts, next_rid
     FROM (
-      SELECT recipient_id
-      FROM msg_queues
-      WHERE deleted_at IS NULL
-        AND msg_queue_expire = TRUE
-        AND recipient_id > last_rid
-      ORDER BY recipient_id ASC
+      SELECT msg_ts, recipient_id
+      FROM messages
+      WHERE NOT msg_quota
+        AND msg_ts < p_old_ts
+        AND (msg_ts, recipient_id) > (last_ts, last_rid)
+      ORDER BY msg_ts ASC, recipient_id ASC
       LIMIT batch_size
-    ) qs;
+    ) m;
 
-    EXIT WHEN rids IS NULL OR cardinality(rids) = 0;
+    EXIT WHEN rids IS NULL;
 
     FOREACH rid IN ARRAY rids
     LOOP
@@ -91,9 +99,11 @@ BEGIN
         RAISE WARNING 'STORE, expire_old_messages, error expiring queue %: %', encode(rid, 'base64'), SQLERRM;
         CONTINUE;
       END;
-      COMMIT;
     END LOOP;
-    last_rid := rids[cardinality(rids)];
+    COMMIT;
+
+    last_ts := next_ts;
+    last_rid := next_rid;
   END LOOP;
 
   r_expired_msgs_count := total_deleted;
@@ -195,7 +205,6 @@ BEGIN
     IF q_size != 0 THEN
       UPDATE msg_queues
       SET msg_can_write = TRUE,
-          msg_queue_expire = FALSE,
           msg_queue_size = 0
       WHERE recipient_id = p_recipient_id;
     END IF;
@@ -207,7 +216,6 @@ BEGIN
     IF FOUND THEN
       UPDATE msg_queues
       SET msg_can_write = msg_can_write OR msg_queue_size <= 1,
-          msg_queue_expire = msg_queue_size > 1,
           msg_queue_size = GREATEST(msg_queue_size - 1, 0)
       WHERE recipient_id = p_recipient_id;
       RETURN QUERY VALUES (msg.msg_id, msg.msg_ts, msg.msg_quota, msg.msg_ntf_flag, msg.msg_body);
@@ -245,7 +253,6 @@ BEGIN
     IF q_size != 0 THEN
       UPDATE msg_queues
       SET msg_can_write = TRUE,
-          msg_queue_expire = FALSE,
           msg_queue_size = 0
       WHERE recipient_id = p_recipient_id;
     END IF;
@@ -271,14 +278,12 @@ BEGIN
       IF msg_deleted THEN
         UPDATE msg_queues
         SET msg_can_write = msg_can_write OR msg_queue_size <= 1,
-            msg_queue_expire = msg_queue_size > 1,
             msg_queue_size = GREATEST(msg_queue_size - 1, 0)
         WHERE recipient_id = p_recipient_id;
       END IF;
     ELSIF msg_deleted OR q_size != 0 THEN
       UPDATE msg_queues
       SET msg_can_write = TRUE,
-          msg_queue_expire = FALSE,
           msg_queue_size = 0
       WHERE recipient_id = p_recipient_id;
     END IF;
@@ -348,7 +353,6 @@ BEGIN
 
     UPDATE msg_queues
     SET msg_can_write = NOT quota_written,
-        msg_queue_expire = TRUE,
         msg_queue_size = msg_queue_size + 1
     WHERE recipient_id = p_recipient_id;
 
@@ -440,7 +444,6 @@ CREATE TABLE smp_server.msg_queues (
     rcv_service_id bytea,
     ntf_service_id bytea,
     msg_can_write boolean DEFAULT true NOT NULL,
-    msg_queue_expire boolean DEFAULT false NOT NULL,
     msg_queue_size bigint DEFAULT 0 NOT NULL
 )
 WITH (fillfactor='80', autovacuum_vacuum_scale_factor='0.02', autovacuum_analyze_scale_factor='0.01', autovacuum_vacuum_cost_limit='1000');
@@ -485,6 +488,10 @@ ALTER TABLE ONLY smp_server.services
 
 
 
+CREATE INDEX idx_messages_expire ON smp_server.messages USING btree (msg_ts, recipient_id) WHERE (NOT msg_quota);
+
+
+
 CREATE INDEX idx_messages_recipient_id_message_id ON smp_server.messages USING btree (recipient_id, message_id);
 
 
@@ -493,14 +500,6 @@ CREATE INDEX idx_messages_recipient_id_msg_quota ON smp_server.messages USING bt
 
 
 
-CREATE INDEX idx_messages_recipient_id_msg_ts ON smp_server.messages USING btree (recipient_id, msg_ts);
-
-
-
-CREATE INDEX idx_msg_queues_expire ON smp_server.msg_queues USING btree (recipient_id) WHERE ((deleted_at IS NULL) AND msg_queue_expire);
-
-
-
 CREATE UNIQUE INDEX idx_msg_queues_link_id ON smp_server.msg_queues USING btree (link_id);
 
 

src/Simplex/Messaging/Transport.hs

@@ -54,6 +54,7 @@ module Simplex.Messaging.Transport
     namesSMPVersion,
     serverInfoSMPVersion,
     nameAvailSMPVersion,
+    fwdNoncesSMPVersion,
     simplexMQVersion,
     smpBlockSize,
     TransportConfig (..),
@@ -177,6 +178,7 @@ smpBlockSize = 16384
 -- 20 - public namespaces resolver, RSLV command (6/20/2026)
 -- 21 - server public information in handshake (7/5/2026)
 -- 22 - RNAME answers name availability as well as the record (7/25/2026)
+-- 23 - version in forwarded command nonce, random nonce in forwarded responses (10/2/2026)
 
 data SMPVersion
 
@@ -218,6 +220,9 @@ serverInfoSMPVersion = VersionSMP 21
 nameAvailSMPVersion :: VersionSMP
 nameAvailSMPVersion = VersionSMP 22
 
+fwdNoncesSMPVersion :: VersionSMP
+fwdNoncesSMPVersion = VersionSMP 23
+
 minClientSMPRelayVersion :: VersionSMP
 minClientSMPRelayVersion = VersionSMP 14
 
@@ -225,20 +230,20 @@ minServerSMPRelayVersion :: VersionSMP
 minServerSMPRelayVersion = VersionSMP 14
 
 currentClientSMPRelayVersion :: VersionSMP
-currentClientSMPRelayVersion = VersionSMP 22
+currentClientSMPRelayVersion = VersionSMP 23
 
 currentServerSMPRelayVersion :: VersionSMP
-currentServerSMPRelayVersion = VersionSMP 22
+currentServerSMPRelayVersion = VersionSMP 23
 
 -- Max SMP protocol version to be used in e2e encrypted connection between
 -- client and server, as defined by SMP proxy. Normally set below the current
 -- version to prevent client version fingerprinting by the destination relays
--- when clients upgrade at different times. Pinned to the current version (22)
--- for this release because a proxied RSLV only carries availability from
--- nameAvailSMPVersion (22), so the one-version anti-fingerprinting buffer does
--- not apply yet; it reappears once the current version advances past 22.
+-- when clients upgrade at different times. Pinned to the current version (23)
+-- for this release because forwarded commands use the nonces from
+-- fwdNoncesSMPVersion (23), so the one-version anti-fingerprinting buffer does
+-- not apply yet; it reappears once the current version advances past 23.
 proxiedSMPRelayVersion :: VersionSMP
-proxiedSMPRelayVersion = VersionSMP 22
+proxiedSMPRelayVersion = VersionSMP 23
 
 -- minimal supported protocol version is 14
 supportedClientSMPRelayVRange :: VersionRangeSMP