core: fix relay request worker retry limit (#6931)

* core: fix relay request worker retry limit

* update plan

* update plan

* update plan

* wip

* schema

* wip

* wip

* schema

* update

* remove comment

* rework

* schema

* update

* schema

* update

* plans

* corrections

* add 1 second

* remove + 1

* add +1 to schedule

* changes

* updated schemas

---------

Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com>
Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>

src/Simplex/Chat/Store/Groups.hs

@@ -1515,8 +1515,8 @@ setGroupInProgressDone db GroupInfo {groupId} = do
     "UPDATE groups SET creating_in_progress = 0, updated_at = ? WHERE group_id = ?"
     (currentTs, groupId)
 
-createRelayRequestGroup :: DB.Connection -> VersionRangeChat -> User -> GroupRelayInvitation -> InvitationId -> VersionRangeChat -> ExceptT StoreError IO (GroupInfo, GroupMember)
-createRelayRequestGroup db vr user@User {userId} GroupRelayInvitation {fromMember, fromMemberProfile, relayMemberId, groupLink} invId reqChatVRange = do
+createRelayRequestGroup :: DB.Connection -> VersionRangeChat -> User -> GroupRelayInvitation -> InvitationId -> VersionRangeChat -> Int64 -> ExceptT StoreError IO (GroupInfo, GroupMember)
+createRelayRequestGroup db vr user@User {userId} GroupRelayInvitation {fromMember, fromMemberProfile, relayMemberId, groupLink} invId reqChatVRange initialDelay = do
   currentTs <- liftIO getCurrentTime
   -- Create group with placeholder profile
   let Profile {displayName = fromMemberLDN} = fromMemberProfile
@@ -1532,7 +1532,7 @@ createRelayRequestGroup db vr user@User {userId} GroupRelayInvitation {fromMembe
         }
   (groupId, _groupLDN) <- createGroup_ db userId placeholderProfile Nothing Nothing True (Just RSInvited) Nothing currentTs
   -- Store relay request data for recovery
-  liftIO $ setRelayRequestData_ groupId
+  liftIO $ setRelayRequestData_ groupId currentTs
   ownerMemberId <- insertOwner_ currentTs groupId
   let relayMember = MemberIdRole relayMemberId GRRelay
   -- TODO [member keys] should relays use member keys?
@@ -1541,7 +1541,7 @@ createRelayRequestGroup db vr user@User {userId} GroupRelayInvitation {fromMembe
   g <- getGroupInfo db vr user groupId
   pure (g, ownerMember)
   where
-    setRelayRequestData_ groupId =
+    setRelayRequestData_ groupId currentTs =
       DB.execute
         db
         [sql|
@@ -1549,10 +1549,12 @@ createRelayRequestGroup db vr user@User {userId} GroupRelayInvitation {fromMembe
           SET relay_request_inv_id = ?,
               relay_request_group_link = ?,
               relay_request_peer_chat_min_version = ?,
-              relay_request_peer_chat_max_version = ?
+              relay_request_peer_chat_max_version = ?,
+              relay_request_delay = ?,
+              relay_request_execute_at = ?
           WHERE group_id = ?
         |]
-        (Binary invId, groupLink, minVersion reqChatVRange, maxVersion reqChatVRange, groupId)
+        (Binary invId, groupLink, minVersion reqChatVRange, maxVersion reqChatVRange, initialDelay, currentTs, groupId)
     insertOwner_ currentTs groupId = do
       let MemberIdRole {memberId, memberRole} = fromMember
           VersionRange minV maxV = reqChatVRange

src/Simplex/Chat/Store/Postgres/Migrations.hs

@@ -28,6 +28,7 @@ import Simplex.Chat.Store.Postgres.Migrations.M20260108_chat_indices
 import Simplex.Chat.Store.Postgres.Migrations.M20260122_has_link
 import Simplex.Chat.Store.Postgres.Migrations.M20260222_chat_relays
 import Simplex.Chat.Store.Postgres.Migrations.M20260403_item_viewed
+import Simplex.Chat.Store.Postgres.Migrations.M20260429_relay_request_retries
 import Simplex.Messaging.Agent.Store.Shared (Migration (..))
 
 schemaMigrations :: [(String, Text, Maybe Text)]
@@ -55,7 +56,8 @@ schemaMigrations =
     ("20260108_chat_indices", m20260108_chat_indices, Just down_m20260108_chat_indices),
     ("20260122_has_link", m20260122_has_link, Just down_m20260122_has_link),
     ("20260222_chat_relays", m20260222_chat_relays, Just down_m20260222_chat_relays),
-    ("20260403_item_viewed", m20260403_item_viewed, Just down_m20260403_item_viewed)
+    ("20260403_item_viewed", m20260403_item_viewed, Just down_m20260403_item_viewed),
+    ("20260429_relay_request_retries", m20260429_relay_request_retries, Just down_m20260429_relay_request_retries)
   ]
 
 -- | The list of migrations in ascending order by date

src/Simplex/Chat/Store/Postgres/Migrations/M20260429_relay_request_retries.hs

@@ -0,0 +1,23 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Chat.Store.Postgres.Migrations.M20260429_relay_request_retries where
+
+import Data.Text (Text)
+import Text.RawString.QQ (r)
+
+m20260429_relay_request_retries :: Text
+m20260429_relay_request_retries =
+  [r|
+ALTER TABLE groups ADD COLUMN relay_request_retries BIGINT NOT NULL DEFAULT 0;
+ALTER TABLE groups ADD COLUMN relay_request_delay BIGINT NOT NULL DEFAULT 0;
+ALTER TABLE groups ADD COLUMN relay_request_execute_at TIMESTAMPTZ NOT NULL DEFAULT (now());
+|]
+
+down_m20260429_relay_request_retries :: Text
+down_m20260429_relay_request_retries =
+  [r|
+ALTER TABLE groups DROP COLUMN relay_request_retries;
+ALTER TABLE groups DROP COLUMN relay_request_delay;
+ALTER TABLE groups DROP COLUMN relay_request_execute_at;
+|]

src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql

@@ -959,7 +959,10 @@ CREATE TABLE test_chat_schema.groups (
     root_priv_key bytea,
     root_pub_key bytea,
     member_priv_key bytea,
-    public_member_count bigint
+    public_member_count bigint,
+    relay_request_retries bigint DEFAULT 0 NOT NULL,
+    relay_request_delay bigint DEFAULT 0 NOT NULL,
+    relay_request_execute_at timestamp with time zone DEFAULT now() NOT NULL
 );
 
 

src/Simplex/Chat/Store/RelayRequests.hs

@@ -9,13 +9,15 @@
 module Simplex.Chat.Store.RelayRequests
   ( hasPendingRelayRequests,
     getNextPendingRelayRequest,
+    updateRelayRequestRetries,
     setRelayRequestErr,
   )
 where
 
+import Data.Int (Int64)
 import Data.Maybe (fromMaybe)
 import Data.Text (Text)
-import Data.Time.Clock (getCurrentTime)
+import Data.Time.Clock (UTCTime, getCurrentTime)
 import Simplex.Chat.Store.Shared
 import Simplex.Chat.Types
 import Simplex.Chat.Types.Shared
@@ -64,7 +66,7 @@ getNextPendingRelayRequest db =
             WHERE relay_own_status = ?
               AND relay_request_failed = 0
               AND relay_request_err_reason IS NULL
-            ORDER BY group_id ASC
+            ORDER BY relay_request_execute_at ASC
             LIMIT 1
           |]
           (Only RSInvited)
@@ -76,18 +78,27 @@ getNextPendingRelayRequest db =
           [sql|
             SELECT
               relay_request_inv_id, relay_request_group_link,
-              relay_request_peer_chat_min_version, relay_request_peer_chat_max_version
+              relay_request_peer_chat_min_version, relay_request_peer_chat_max_version,
+              relay_request_delay, relay_request_retries, created_at, relay_request_execute_at
             FROM groups
             WHERE group_id = ?
           |]
           (Only groupId)
       where
-        toRelayRequestData :: (Maybe InvitationId, Maybe ShortLinkContact, Maybe VersionChat, Maybe VersionChat) -> Either StoreError (GroupId, RelayRequestData)
+        toRelayRequestData :: (Maybe InvitationId, Maybe ShortLinkContact, Maybe VersionChat, Maybe VersionChat, Int64, Int, UTCTime, UTCTime) -> Either StoreError (GroupId, RelayRequestData)
         toRelayRequestData = \case
-          (Just relayInvId, Just reqGroupLink, Just minV, Just maxV) ->
-            Right (groupId, RelayRequestData {relayInvId, reqGroupLink, reqChatVRange = fromMaybe (versionToRange maxV) $ safeVersionRange minV maxV})
+          (Just relayInvId, Just reqGroupLink, Just minV, Just maxV, reqDelay, reqRetries, reqCreatedAt, reqExecuteAt) ->
+            Right (groupId, RelayRequestData {relayInvId, reqGroupLink, reqChatVRange = fromMaybe (versionToRange maxV) $ safeVersionRange minV maxV, reqDelay, reqRetries, reqCreatedAt, reqExecuteAt})
           _ -> Left $ SEInternalError "missing relay request data"
 
+updateRelayRequestRetries :: DB.Connection -> GroupId -> Int64 -> UTCTime -> IO ()
+updateRelayRequestRetries db groupId delay executeAt = do
+  currentTs <- getCurrentTime
+  DB.execute
+    db
+    "UPDATE groups SET relay_request_retries = relay_request_retries + 1, relay_request_delay = ?, relay_request_execute_at = ?, updated_at = ? WHERE group_id = ?"
+    (delay, executeAt, currentTs, groupId)
+
 markRelayRequestFailed :: DB.Connection -> GroupId -> IO ()
 markRelayRequestFailed db groupId = do
   currentTs <- getCurrentTime

src/Simplex/Chat/Store/SQLite/Migrations.hs

@@ -151,6 +151,7 @@ import Simplex.Chat.Store.SQLite.Migrations.M20260108_chat_indices
 import Simplex.Chat.Store.SQLite.Migrations.M20260122_has_link
 import Simplex.Chat.Store.SQLite.Migrations.M20260222_chat_relays
 import Simplex.Chat.Store.SQLite.Migrations.M20260403_item_viewed
+import Simplex.Chat.Store.SQLite.Migrations.M20260429_relay_request_retries
 import Simplex.Messaging.Agent.Store.Shared (Migration (..))
 
 schemaMigrations :: [(String, Query, Maybe Query)]
@@ -301,7 +302,8 @@ schemaMigrations =
     ("20260108_chat_indices", m20260108_chat_indices, Just down_m20260108_chat_indices),
     ("20260122_has_link", m20260122_has_link, Just down_m20260122_has_link),
     ("20260222_chat_relays", m20260222_chat_relays, Just down_m20260222_chat_relays),
-    ("20260403_item_viewed", m20260403_item_viewed, Just down_m20260403_item_viewed)
+    ("20260403_item_viewed", m20260403_item_viewed, Just down_m20260403_item_viewed),
+    ("20260429_relay_request_retries", m20260429_relay_request_retries, Just down_m20260429_relay_request_retries)
   ]
 
 -- | The list of migrations in ascending order by date

src/Simplex/Chat/Store/SQLite/Migrations/M20260429_relay_request_retries.hs

@@ -0,0 +1,22 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Chat.Store.SQLite.Migrations.M20260429_relay_request_retries where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20260429_relay_request_retries :: Query
+m20260429_relay_request_retries =
+  [sql|
+ALTER TABLE groups ADD COLUMN relay_request_retries INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE groups ADD COLUMN relay_request_delay INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE groups ADD COLUMN relay_request_execute_at TEXT NOT NULL DEFAULT(datetime('now'));
+|]
+
+down_m20260429_relay_request_retries :: Query
+down_m20260429_relay_request_retries =
+  [sql|
+ALTER TABLE groups DROP COLUMN relay_request_retries;
+ALTER TABLE groups DROP COLUMN relay_request_delay;
+ALTER TABLE groups DROP COLUMN relay_request_execute_at;
+|]

src/Simplex/Chat/Store/SQLite/Migrations/chat_query_plans.txt

@@ -695,7 +695,8 @@ SEARCH delivery_jobs USING INTEGER PRIMARY KEY (rowid=?)
 Query: 
             SELECT
               relay_request_inv_id, relay_request_group_link,
-              relay_request_peer_chat_min_version, relay_request_peer_chat_max_version
+              relay_request_peer_chat_min_version, relay_request_peer_chat_max_version,
+              relay_request_delay, relay_request_retries, created_at, relay_request_execute_at
             FROM groups
             WHERE group_id = ?
           
@@ -993,11 +994,12 @@ Query:
             WHERE relay_own_status = ?
               AND relay_request_failed = 0
               AND relay_request_err_reason IS NULL
-            ORDER BY group_id ASC
+            ORDER BY relay_request_execute_at ASC
             LIMIT 1
           
 Plan:
 SCAN groups
+USE TEMP B-TREE FOR ORDER BY
 
 Query: 
             SELECT i.chat_item_id
@@ -1775,7 +1777,9 @@ Query:
           SET relay_request_inv_id = ?,
               relay_request_group_link = ?,
               relay_request_peer_chat_min_version = ?,
-              relay_request_peer_chat_max_version = ?
+              relay_request_peer_chat_max_version = ?,
+              relay_request_delay = ?,
+              relay_request_execute_at = ?
           WHERE group_id = ?
         
 Plan:

src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql

@@ -173,7 +173,10 @@ CREATE TABLE groups(
   root_priv_key BLOB,
   root_pub_key BLOB,
   member_priv_key BLOB,
-  public_member_count INTEGER, -- received
+  public_member_count INTEGER,
+  relay_request_retries INTEGER NOT NULL DEFAULT 0,
+  relay_request_delay INTEGER NOT NULL DEFAULT 0,
+  relay_request_execute_at TEXT NOT NULL DEFAULT(datetime('now')), -- received
   FOREIGN KEY(user_id, local_display_name)
   REFERENCES display_names(user_id, local_display_name)
   ON DELETE CASCADE