core: schedule store receipt attempts in the database

src/Simplex/Chat.hs

@@ -72,6 +72,7 @@ defaultChatConfig =
       badgeServiceAddress = Just $ either error id $ strDecode "https://smp5.simplex.im/a#ooSNWlEZTO2RPE0Ff5ZoybAs5zEhWLMlQrXesnhaZHM",
       badgeCurrentTime = getCurrentTime,
       badgeRetryInterval = RetryInterval {initialInterval = 30_000000, increaseAfter = 0, maxInterval = 3600_000000},
+      badgeConsecutiveRetries = 3,
       confirmMigrations = MCConsole,
       -- this property should NOT use operator = Nothing
       -- non-operator servers can be passed via options

src/Simplex/Chat/Controller.hs

@@ -153,6 +153,8 @@ data ChatConfig = ChatConfig
     badgeCurrentTime :: IO UTCTime,
     -- how long a badge worker waits before repeating a renewal that failed for a passing reason
     badgeRetryInterval :: RetryInterval,
+    -- attempts a store receipt gets in one turn before its next one waits for the stored schedule
+    badgeConsecutiveRetries :: Int,
     confirmMigrations :: MigrationConfirmation,
     presetServers :: PresetServers,
     shortLinkPresetServers :: NonEmpty SMPServer,
@@ -294,11 +296,6 @@ data BadgeWorker = BadgeWorker
     badgeWork :: TMVar ()
   }
 
-data StoreReceiptWorker = StoreReceiptWorker
-  { receiptWorker :: Worker,
-    parkedReceipts :: TVar (Set Int64)
-  }
-
 data ChatController = ChatController
   { currentUser :: TVar (Maybe User),
     randomPresetServers :: NonEmpty PresetOperator,
@@ -333,7 +330,7 @@ data ChatController = ChatController
     relayRequestWorkers :: TMap Int Worker, -- single global worker with key 1 is used to fit into existing worker management framework
     -- one badge worker per user: badge state is per profile, and one profile must not stall another
     badgeWorkers :: TMap UserId (SessionVar BadgeWorker),
-    storeReceiptWorkers :: TMap UserId StoreReceiptWorker,
+    storeReceiptWorkers :: TMap UserId Worker,
     badgeSeq :: TVar Int,
     relayGroupLinkChecksAsync :: TVar (Maybe (Async ())),
     webPreviewState :: Maybe WebPreviewState,

src/Simplex/Chat/Library/Commands.hs

@@ -105,10 +105,10 @@ import qualified Simplex.Chat.Util as U
 import Simplex.Chat.Web (webPreviewWorker)
 import Simplex.FileTransfer.Description (FileDescriptionURI (..), maxFileSizeHard)
 import Simplex.Messaging.Agent
-import Simplex.Messaging.Agent.Client (getAgentWorker', hasWorkToDo, waitForUserNetwork, waitForWork, waitWhileSuspended, withWork_)
+import Simplex.Messaging.Agent.Client (getAgentWorker, waitForUserNetwork, waitForWork, waitWhileSuspended, withWork_)
 import Simplex.Messaging.Agent.Env.SQLite (ServerCfg (..), ServerRoles (..), Worker (..), allRoles)
 import Simplex.Messaging.Agent.Protocol
-import Simplex.Messaging.Agent.RetryInterval (RetryInterval (..), withRetryInterval)
+import Simplex.Messaging.Agent.RetryInterval (RetryInterval (..), nextRetryDelay, withRetryInterval, withRetryIntervalCount)
 import Simplex.Messaging.Agent.Store.Entity
 import Simplex.Messaging.Agent.Store.Interface (execSQL)
 import Simplex.Messaging.Agent.Store.Shared (upMigration)
@@ -379,7 +379,7 @@ stopChatController ChatController {smpAgent, agentAsync = s, sndFiles, rcvFiles,
   clearMap deliveryJobWorkers >>= mapM_ cancelWorker
   clearMap relayRequestWorkers >>= mapM_ cancelWorker
   stopBadgeWorkers badgeWorkers
-  clearMap storeReceiptWorkers >>= mapM_ (cancelWorker . receiptWorker)
+  clearMap storeReceiptWorkers >>= mapM_ cancelWorker
   closeFiles sndFiles
   closeFiles rcvFiles
   atomically $ do
@@ -5438,72 +5438,72 @@ badgeMaxWake = 100 * 365 * nominalDay
 
 startStoreReceiptWork :: User -> CM ()
 startStoreReceiptWork User {userId} = do
-  due <- withStore' $ \db -> getNextDueStoreReceipt db userId S.empty
-  when (either (const False) isJust due) $ lift $ resumeStoreReceiptWork userId
+  held <- withStore' $ \db -> getNextHeldStoreReceipt db userId
+  when (either (const False) isJust held) $ lift $ resumeStoreReceiptWork userId
 
 resumeStoreReceiptWork :: UserId -> CM' ()
 resumeStoreReceiptWork userId = do
-  StoreReceiptWorker {receiptWorker, parkedReceipts} <- getStoreReceiptWorker userId
-  atomically $ writeTVar parkedReceipts S.empty >> hasWorkToDo receiptWorker
-
-getStoreReceiptWorker :: UserId -> CM' StoreReceiptWorker
-getStoreReceiptWorker userId = do
   ws <- asks storeReceiptWorkers
   a <- asks smpAgent
-  getAgentWorker' receiptWorker newStoreReceiptWorker "store_receipt" False a userId ws $
-    runStoreReceiptWorker a userId
-  where
-    newStoreReceiptWorker w = StoreReceiptWorker w <$> newTVar S.empty
+  void $ getAgentWorker "store_receipt" True a userId ws $ runStoreReceiptWorker a userId
 
-runStoreReceiptWorker :: AgentClient -> UserId -> StoreReceiptWorker -> CM ()
-runStoreReceiptWorker a userId StoreReceiptWorker {receiptWorker = Worker {doWork}, parkedReceipts} =
+runStoreReceiptWorker :: AgentClient -> UserId -> Worker -> CM ()
+runStoreReceiptWorker a userId Worker {doWork} = do
+  scheduled <- newTVarIO S.empty
   forever $ do
     lift $ waitForWork doWork
-    withWork_ a doWork getNextReceipt $ \r@HeldStoreReceipt {receiptId} ->
-      (creditStoreReceipt a userId r `catchAllErrors` \e -> False <$ eToView e) >>= \case
-        True -> pure ()
-        -- unlike the agent's workers, a final failure keeps the row: the purchase is paid for, so the next trigger tries it again
-        False -> atomically $ modifyTVar' parkedReceipts $ S.insert receiptId
+    withWork_ a doWork (getDueReceipt scheduled) $ creditStoreReceipt a userId
   where
-    getNextReceipt = do
-      parked <- readTVarIO parkedReceipts
-      withStore' $ \db -> getNextDueStoreReceipt db userId parked
+    getDueReceipt scheduled =
+      withStore' (`getNextHeldStoreReceipt` userId) >>= \case
+        Right (Just r@HeldStoreReceipt {receiptId, nextAttemptAt}) -> do
+          now <- badgeNow
+          if nextAttemptAt <= now
+            then pure $ Right (Just r)
+            else Right Nothing <$ scheduleReceipt scheduled receiptId nextAttemptAt
+        r -> pure r
+    scheduleReceipt scheduled receiptId at = do
+      new <- atomically $ stateTVar scheduled $ \ids -> (S.notMember receiptId ids, S.insert receiptId ids)
+      when new $ do
+        clock <- asks $ badgeCurrentTime . config
+        RetryInterval {maxInterval} <- asks $ badgeRetryInterval . config
+        void . liftIO . forkIO $ do
+          waitBadgeTime clock maxInterval at
+          atomically $ modifyTVar' scheduled $ S.delete receiptId
+          void $ atomically $ tryPutTMVar doWork ()
 
-data StoreReceiptOutcome = SROCredited (Maybe User) | SRORefused | SRORetry (Maybe Word32) | SROFailed
+waitBadgeTime :: IO UTCTime -> Int64 -> UTCTime -> IO ()
+waitBadgeTime clock step at = do
+  remaining <- diffToMicroseconds . diffUTCTime at <$> clock
+  when (remaining > 0) $ threadDelay' (min remaining step) >> waitBadgeTime clock step at
+
+data StoreReceiptOutcome = SROCredited (Maybe User) | SRORefused | SRORetry | SRODeferred
 
 -- | Under the owner's badge lock, which keeps a profile's requests to the service serial.
-creditStoreReceipt :: AgentClient -> UserId -> HeldStoreReceipt -> CM Bool
+creditStoreReceipt :: AgentClient -> UserId -> HeldStoreReceipt -> CM ()
 creditStoreReceipt a userId HeldStoreReceipt {receiptId, stash = stash@BadgeStash {masterKey}, payment, retryDelay} = do
-  ri <- asks $ badgeRetryInterval . config
-  notBefore <- newTVarIO Nothing
-  withRetryInterval (maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) retryDelay) $ \delay loop -> do
+  ChatConfig {badgeRetryInterval = ri, badgeConsecutiveRetries} <- asks config
+  withRetryIntervalCount (maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) retryDelay) $ \n delay loop -> do
     liftIO $ waitWhileSuspended a
     liftIO $ waitForUserNetwork a
-    now <- badgeNow
-    readTVarIO notBefore >>= \case
-      Just at | now < at -> loop
-      _ -> do
-        outcome <- withEntityLock "badgePurchase" (CLBadgeUser userId) $
-          tryAllErrors attempt >>= \case
-            Right (Right (present_, _)) -> pure $ SROCredited present_
-            Right (Left (code, retryAfter))
-              | storeReceiptRefused code -> SRORefused <$ withStore' (\db -> refuseStoreReceipt db receiptId $ serviceFailure code retryAfter)
-              | otherwise -> failed delay (serviceFailure code retryAfter) retryAfter
-            Left e -> failed delay (badgeIssueFailure e) Nothing
-        -- only a settlement is announced, and the app's sweep in answer signals this worker only for a newly held receipt, so it cannot loop
-        case outcome of
-          SROCredited present_ -> do
-            user <- withStore $ \db -> getUser db userId
-            toView . CEvtBadgeChanged user =<< getUserBadgeState user
-            toView $ CEvtStorePurchaseSettled user
-            -- last, so a failed broadcast cannot lose the settlement; outside the badge lock, as it takes the chat lock
-            mapM_ presentUserBadgeToContacts present_
-            pure True
-          SRORefused -> True <$ (withStore (`getUser` userId) >>= toView . CEvtStorePurchaseSettled)
-          SRORetry retryAfter_ -> do
-            atomically $ writeTVar notBefore $ (`addUTCTime` now) . fromIntegral <$> retryAfter_
-            loop
-          SROFailed -> pure False
+    outcome <- withEntityLock "badgePurchase" (CLBadgeUser userId) $
+      tryAllErrors attempt >>= \case
+        Right (Right (present_, _)) -> pure $ SROCredited present_
+        Right (Left (code, retryAfter))
+          | storeReceiptRefused code -> SRORefused <$ withStore' (\db -> refuseStoreReceipt db receiptId $ serviceFailure code retryAfter)
+          | otherwise -> failed ri delay (serviceFailure code retryAfter) retryAfter
+        Left e -> failed ri delay (badgeIssueFailure e) Nothing
+    -- only a settlement is announced, and the app's sweep in answer signals this worker only for a newly held receipt, so it cannot loop
+    case outcome of
+      SROCredited present_ -> do
+        user <- withStore $ \db -> getUser db userId
+        toView . CEvtBadgeChanged user =<< getUserBadgeState user
+        toView $ CEvtStorePurchaseSettled user
+        -- last, so a failed broadcast cannot lose the settlement; outside the badge lock, as it takes the chat lock
+        mapM_ presentUserBadgeToContacts present_
+      SRORefused -> withStore (`getUser` userId) >>= toView . CEvtStorePurchaseSettled
+      SRORetry | n + 1 < badgeConsecutiveRetries -> loop
+      _ -> pure ()
   where
     attempt = do
       user <- withStore $ \db -> getUser db userId
@@ -5512,9 +5512,15 @@ creditStoreReceipt a userId HeldStoreReceipt {receiptId, stash = stash@BadgeStas
       payment' <- maybe (throwChatError $ CEInternalError "held store payment does not decode") pure $ decodeJSON payment
       requestBadgeStash NRMBackground user sendTarget stash BSCPurchaseBadge {masterKey, payment = payment', upgrade = Nothing}
     serviceFailure code retryAfter = BIFServiceError {code = boundedServiceErrorCode code, retryable = isJust retryAfter}
-    failed delay failure retryAfter = do
-      withStore' $ \db -> recordStoreReceiptFailure db receiptId delay failure
-      pure $ if badgeFailureTransient failure then SRORetry retryAfter else SROFailed
+    failed ri delay failure retryAfter = do
+      now <- badgeNow
+      let (retryIn, outcome)
+            | isJust retryAfter = (badgeRetryAfter ri retryAfter, SRODeferred)
+            -- a buyer is waiting, so unlike a renewal a failure that can clear on its own never waits the day
+            | badgeFailureTransient failure = (fromIntegral delay / 1000000, SRORetry)
+            | otherwise = (badgeStalledInterval, SRODeferred)
+      withStore' $ \db -> recordStoreReceiptFailure db receiptId (nextRetryDelay 0 delay ri) (retryIn `addUTCTime` now) failure
+      pure outcome
 
 -- | Retire what has ended, renew what is due, then report the next wake. Waking early, late or not
 -- at all changes only timing: each run reads stored state and works out what to do.

src/Simplex/Chat/Store/Badges.hs

@@ -26,7 +26,7 @@ module Simplex.Chat.Store.Badges
     holdStoreReceipt,
     createBadgeStoreReceipt,
     getOpenStorePurchases,
-    getNextDueStoreReceipt,
+    getNextHeldStoreReceipt,
     recordStoreReceiptFailure,
     refuseStoreReceipt,
     deleteUnfundedStoreReceipts,
@@ -49,10 +49,7 @@ import qualified Data.Aeson as J
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Int (Int64)
-import Data.List (find)
 import Data.Maybe (isJust, mapMaybe)
-import Data.Set (Set)
-import qualified Data.Set as S
 import Data.Text (Text)
 import Data.Time.Clock (UTCTime)
 import Simplex.Chat.Badges
@@ -141,8 +138,8 @@ holdStoreReceipt db g User {userId} invoiceId_ txRef@StoreTransactionRef {provid
       forM_ invoiceId_ $ \invoiceId ->
         DB.execute
           db
-          "UPDATE badge_store_receipts SET provider = ?, transaction_ref = ?, payment = ? WHERE invoice_id = ? AND transaction_ref IS NULL"
-          (provider, transactionRef, paymentJSON, invoiceId)
+          "UPDATE badge_store_receipts SET provider = ?, transaction_ref = ?, payment = ?, next_attempt_at = ? WHERE invoice_id = ? AND transaction_ref IS NULL"
+          (provider, transactionRef, paymentJSON, now, invoiceId)
       getStoreReceipt db txRef >>= \case
         Just r -> pure $ Just (r, True)
         Nothing -> do
@@ -162,11 +159,11 @@ holdStoreReceipt db g User {userId} invoiceId_ txRef@StoreTransactionRef {provid
         db
         [sql|
           INSERT INTO badge_store_receipts
-            (user_id, invoice_id, provider, transaction_ref, payment, purchase_key, purchase_priv_key, master_key, created_at)
-          VALUES (?,?,?,?,?,?,?,?,?)
+            (user_id, invoice_id, provider, transaction_ref, payment, next_attempt_at, purchase_key, purchase_priv_key, master_key, created_at)
+          VALUES (?,?,?,?,?,?,?,?,?,?)
           ON CONFLICT DO NOTHING
         |]
-        ((userId, invoiceId', provider, transactionRef, paymentJSON) :. (purchaseKey, purchasePrivKey, Binary mk, now))
+        ((userId, invoiceId', provider, transactionRef, paymentJSON, now) :. (purchaseKey, purchasePrivKey, Binary mk, now))
 
 -- | held is a CASE because in Postgres a comparison is boolean, which BoolInt rejects.
 getStoreReceipt :: DB.Connection -> StoreTransactionRef -> IO (Maybe StoreReceipt)
@@ -218,38 +215,46 @@ data HeldStoreReceipt = HeldStoreReceipt
   { receiptId :: Int64,
     stash :: BadgeStash,
     payment :: Text,
+    nextAttemptAt :: UTCTime,
     retryDelay :: Maybe Int64
   }
 
-getNextDueStoreReceipt :: DB.Connection -> UserId -> Set Int64 -> IO (Either StoreError (Maybe HeldStoreReceipt))
-getNextDueStoreReceipt db userId parked =
-  Right . find notParked . map toHeld
-    <$> DB.query
+getNextHeldStoreReceipt :: DB.Connection -> UserId -> IO (Either StoreError (Maybe HeldStoreReceipt))
+getNextHeldStoreReceipt db userId =
+  fmap Right . maybeFirstRow toHeld $
+    DB.query
       db
       [sql|
-        SELECT r.badge_store_receipt_id, r.purchase_key, r.purchase_priv_key, r.master_key, r.payment, r.retry_delay
+        SELECT r.badge_store_receipt_id, r.purchase_key, r.purchase_priv_key, r.master_key, r.payment, r.next_attempt_at, r.retry_delay
         FROM badge_store_receipts r
         JOIN users u ON u.user_id = r.user_id
         WHERE r.user_id = ? AND r.payment IS NOT NULL AND u.shown_badge_id IS NULL
-        ORDER BY r.badge_store_receipt_id
+        ORDER BY r.next_attempt_at, r.retry_count, r.badge_store_receipt_id
+        LIMIT 1
       |]
       (Only userId)
   where
-    notParked HeldStoreReceipt {receiptId} = receiptId `S.notMember` parked
-    toHeld (stashRow@(receiptId, _, _, _) :. (payment, retryDelay)) =
-      HeldStoreReceipt {receiptId, stash = toBadgeStash BSRStoreReceipt stashRow, payment, retryDelay}
+    toHeld (stashRow@(receiptId, _, _, _) :. (payment, nextAttemptAt, retryDelay)) =
+      HeldStoreReceipt {receiptId, stash = toBadgeStash BSRStoreReceipt stashRow, payment, nextAttemptAt, retryDelay}
 
-recordStoreReceiptFailure :: DB.Connection -> Int64 -> Int64 -> BadgeIssueFailure -> IO ()
-recordStoreReceiptFailure db receiptId retryDelay failure =
+recordStoreReceiptFailure :: DB.Connection -> Int64 -> Int64 -> UTCTime -> BadgeIssueFailure -> IO ()
+recordStoreReceiptFailure db receiptId retryDelay nextAttemptAt failure =
   DB.execute
     db
-    "UPDATE badge_store_receipts SET retry_delay = ?, credit_error = ? WHERE badge_store_receipt_id = ? AND payment IS NOT NULL"
-    (retryDelay, failure, receiptId)
+    [sql|
+      UPDATE badge_store_receipts
+      SET retry_delay = ?, next_attempt_at = ?, retry_count = retry_count + 1, credit_error = ?
+      WHERE badge_store_receipt_id = ? AND payment IS NOT NULL
+    |]
+    (retryDelay, nextAttemptAt, failure, receiptId)
 
 -- | Kept rather than deleted, so that a later hand-over of the same transaction is answered from the record.
 refuseStoreReceipt :: DB.Connection -> Int64 -> BadgeIssueFailure -> IO ()
 refuseStoreReceipt db receiptId refusal =
-  DB.execute db "UPDATE badge_store_receipts SET payment = NULL, retry_delay = NULL, credit_error = ? WHERE badge_store_receipt_id = ?" (refusal, receiptId)
+  DB.execute
+    db
+    "UPDATE badge_store_receipts SET payment = NULL, next_attempt_at = NULL, retry_delay = NULL, retry_count = 0, credit_error = ? WHERE badge_store_receipt_id = ?"
+    (refusal, receiptId)
 
 -- | A held record is paid for and a credited one is referenced by its purchase, so neither is deleted.
 deleteUnfundedStoreReceipts :: DB.Connection -> UTCTime -> IO ()
@@ -285,7 +290,7 @@ deleteBadgeCodeRedemption db User {userId} code =
 createStashBadgePurchase :: DB.Connection -> User -> BadgeStash -> BadgeCredential -> UTCTime -> IO (Int64, Bool)
 createStashBadgePurchase db User {userId} stash credential now = do
   -- the credit settles a held receipt in the transaction that stores its purchase, a replay included
-  forM_ storeReceiptId_ $ \storeReceiptId -> DB.execute db "UPDATE badge_store_receipts SET payment = NULL, retry_delay = NULL WHERE badge_store_receipt_id = ?" (Only storeReceiptId)
+  forM_ storeReceiptId_ $ \storeReceiptId -> DB.execute db "UPDATE badge_store_receipts SET payment = NULL, next_attempt_at = NULL, retry_delay = NULL, retry_count = 0 WHERE badge_store_receipt_id = ?" (Only storeReceiptId)
   getStashBadgePurchase db stash >>= \case
     Just purchaseId -> pure (purchaseId, False)
     Nothing -> do

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

@@ -24,7 +24,9 @@ CREATE TABLE badge_store_receipts(
   master_key BYTEA NOT NULL,
   created_at TIMESTAMPTZ NOT NULL,
   payment TEXT,
+  next_attempt_at TIMESTAMPTZ,
   retry_delay BIGINT,
+  retry_count BIGINT NOT NULL DEFAULT 0,
   credit_error TEXT,
   UNIQUE(provider, transaction_ref)
 );

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

@@ -307,7 +307,9 @@ CREATE TABLE test_chat_schema.badge_store_receipts (
     master_key bytea NOT NULL,
     created_at timestamp with time zone NOT NULL,
     payment text,
+    next_attempt_at timestamp with time zone,
     retry_delay bigint,
+    retry_count bigint DEFAULT 0 NOT NULL,
     credit_error text
 );
 

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

@@ -23,7 +23,9 @@ CREATE TABLE badge_store_receipts(
   master_key BLOB NOT NULL,
   created_at TEXT NOT NULL,
   payment TEXT,
+  next_attempt_at TEXT,
   retry_delay INTEGER,
+  retry_count INTEGER NOT NULL DEFAULT 0,
   credit_error TEXT,
   UNIQUE(provider, transaction_ref)
 ) STRICT;

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

@@ -993,7 +993,9 @@ CREATE TABLE badge_store_receipts(
   master_key BLOB NOT NULL,
   created_at TEXT NOT NULL,
   payment TEXT,
+  next_attempt_at TEXT,
   retry_delay INTEGER,
+  retry_count INTEGER NOT NULL DEFAULT 0,
   credit_error TEXT,
   UNIQUE(provider, transaction_ref)
 ) STRICT;