core: credit store receipts from a standard work queue

This commit is contained in:
spaced4ndy
2026-10-06 17:10:05 +04:00
parent 0d7efd0bc3
commit eee013dfcb
10 changed files with 210 additions and 228 deletions
+6 -1
View File
@@ -294,6 +294,11 @@ data BadgeWorker = BadgeWorker
badgeWork :: TMVar ()
}
data StoreReceiptWorker = StoreReceiptWorker
{ receiptWorker :: Worker,
parkedReceipts :: TVar (Set Int64)
}
data ChatController = ChatController
{ currentUser :: TVar (Maybe User),
randomPresetServers :: NonEmpty PresetOperator,
@@ -328,7 +333,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 (SessionVar BadgeWorker),
storeReceiptWorkers :: TMap UserId StoreReceiptWorker,
badgeSeq :: TVar Int,
relayGroupLinkChecksAsync :: TVar (Maybe (Async ())),
webPreviewState :: Maybe WebPreviewState,
+84 -66
View File
@@ -105,9 +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.Env.SQLite (ServerCfg (..), ServerRoles (..), allRoles)
import Simplex.Messaging.Agent.Client (getAgentWorker', hasWorkToDo, 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 (..), nextRetryDelay, withRetryInterval)
import Simplex.Messaging.Agent.RetryInterval (RetryInterval (..), withRetryInterval)
import Simplex.Messaging.Agent.Store.Entity
import Simplex.Messaging.Agent.Store.Interface (execSQL)
import Simplex.Messaging.Agent.Store.Shared (upMigration)
@@ -263,7 +264,7 @@ startChatController mainApp enableSndFiles serviceRequests = do
startRelayRequestWorker_
startCleanupManager
mapM_ startBadgeWork users
mapM_ startStoreReceiptWork users
startStoreReceiptWorkers users
void $ forkIO $ mapM_ startExpireCIs users
startRelayChecks users
startWebPreview users
@@ -282,6 +283,10 @@ startChatController mainApp enableSndFiles serviceRequests = do
runExceptT startRelayRequestWorker >>= \case
Left e -> liftIO $ putStrLn $ "Error starting relay request worker: " <> show e
Right _ -> pure ()
startStoreReceiptWorkers users =
runExceptT (mapM_ startStoreReceiptWork users) >>= \case
Left e -> liftIO $ putStrLn $ "Error starting store receipt workers: " <> show e
Right _ -> pure ()
startCleanupManager = do
cleanupAsync <- asks cleanupManagerAsync
readTVarIO cleanupAsync >>= \case
@@ -374,7 +379,7 @@ stopChatController ChatController {smpAgent, agentAsync = s, sndFiles, rcvFiles,
clearMap deliveryJobWorkers >>= mapM_ cancelWorker
clearMap relayRequestWorkers >>= mapM_ cancelWorker
stopBadgeWorkers badgeWorkers
stopBadgeWorkers storeReceiptWorkers
clearMap storeReceiptWorkers >>= mapM_ (cancelWorker . receiptWorker)
closeFiles sndFiles
closeFiles rcvFiles
atomically $ do
@@ -607,7 +612,7 @@ processChatCommand cxt nm = \case
void . forkIO $ startFilesToReceive users
setAllExpireCIFlags True
mapM_ startBadgeWork users
mapM_ startStoreReceiptWork users
mapM_ startStoreReceiptWork users
ok_
APISuspendChat t -> do
chatWriteVar chatActivated False
@@ -5269,9 +5274,9 @@ handOverStoreReceipt presentingUser echoedInvoiceId payment = do
(StoreReceipt {ownerId, status}, newlyHeld) <-
withStore' (\db -> holdStoreReceipt db g presentingUser echoedInvoiceId txRef payment now)
>>= maybe (throwChatError $ CEInternalError "store receipt was not recorded") pure
owner <- withStore $ \db -> getUser db ownerId
-- before the answer, which can throw: no later hand-over of this receipt signals, and a spare signal costs an empty pass
when newlyHeld $ lift $ startStoreReceiptWork owner
when newlyHeld $ lift $ resumeStoreReceiptWork ownerId
owner <- withStore $ \db -> getUser db ownerId
case status of
RSHeld -> badgeStateResponse owner
RSCredited {badgePurchaseId} -> do
@@ -5348,23 +5353,20 @@ badgeNow = asks (badgeCurrentTime . config) >>= liftIO
-- | A signal carries nothing: each pass derives its work from stored state, so a signal lost or
-- duplicated changes no outcome.
startBadgeWork :: User -> CM' ()
startBadgeWork user = whenM (isJust <$> asks (badgeServiceAddress . config)) $ void $ getBadgeWorker badgeWorkers runBadgeWorker user
startStoreReceiptWork :: User -> CM' ()
startStoreReceiptWork user = whenM (isJust <$> asks (badgeServiceAddress . config)) $ void $ getBadgeWorker storeReceiptWorkers runStoreReceiptWorker user
startBadgeWork user = whenM (isJust <$> asks (badgeServiceAddress . config)) $ void $ getBadgeWorker user
-- | Exactly one caller starts the thread and the rest wait for it: the lookup and the create cannot
-- be one transaction, because starting a thread is not STM.
getBadgeWorker :: (ChatController -> TM.TMap UserId (SessionVar BadgeWorker)) -> (UserId -> TMVar () -> CM ()) -> User -> CM' BadgeWorker
getBadgeWorker workers runWorker User {userId} = do
ws <- asks workers
getBadgeWorker :: User -> CM' BadgeWorker
getBadgeWorker User {userId} = do
ws <- asks badgeWorkers
seq' <- asks badgeSeq
now <- liftIO getCurrentTime
withGetSessVar' seq' userId ws now startWorker signalWorker
where
startWorker v = do
badgeWork <- newEmptyTMVarIO
badgeWorkerAsync <- async $ void $ runExceptT $ runWorker userId badgeWork
badgeWorkerAsync <- async $ void $ runExceptT $ runBadgeWorker userId badgeWork
let w = BadgeWorker {badgeWorkerAsync, badgeWork}
w <$ atomically (putTMVar (sessionVar v) w)
signalWorker v = do
@@ -5434,46 +5436,74 @@ waitBadgeWake badgeWork now = \case
badgeMaxWake :: NominalDiffTime
badgeMaxWake = 100 * 365 * nominalDay
-- | The only sender of held store receipts; each row's next_attempt_at is when it is tried again. The
-- wait is floored, so a row still due because its outcome could not be written does not spin the worker.
runStoreReceiptWorker :: UserId -> TMVar () -> CM ()
runStoreReceiptWorker userId work = forever $ do
lift waitChatStartedAndActivated
nextAt_ <- creditStoreReceipts userId `catchAllErrors` \e -> eToView e >> Just <$> badgeNow
now <- badgeNow
ri <- asks $ badgeRetryInterval . config
liftIO $ waitBadgeWake work now $ max (retryFloor ri `addUTCTime` now) <$> nextAt_
startStoreReceiptWork :: User -> CM ()
startStoreReceiptWork User {userId} = do
due <- withStore' $ \db -> getNextDueStoreReceipt db userId S.empty
when (either (const False) isJust due) $ lift $ resumeStoreReceiptWork userId
-- | The due rows are read once, so a pass tries each at most once.
creditStoreReceipts :: UserId -> CM (Maybe UTCTime)
creditStoreReceipts userId = do
now <- badgeNow
due <- withStore' $ \db -> getDueStoreReceipts db userId now
forM_ due $ \r -> creditStoreReceipt userId r `catchAllErrors` eToView
withStore' (`getNextStoreReceiptAttempt` userId)
resumeStoreReceiptWork :: UserId -> CM' ()
resumeStoreReceiptWork userId = do
StoreReceiptWorker {receiptWorker, parkedReceipts} <- getStoreReceiptWorker userId
atomically $ writeTVar parkedReceipts S.empty >> hasWorkToDo receiptWorker
data StoreReceiptOutcome = SROCredited (Maybe User) | SRORefused | SRODeferred
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
runStoreReceiptWorker :: AgentClient -> UserId -> StoreReceiptWorker -> CM ()
runStoreReceiptWorker a userId StoreReceiptWorker {receiptWorker = Worker {doWork}, parkedReceipts} =
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
where
getNextReceipt = do
parked <- readTVarIO parkedReceipts
withStore' $ \db -> getNextDueStoreReceipt db userId parked
data StoreReceiptOutcome = SROCredited (Maybe User) | SRORefused | SRORetry (Maybe Word32) | SROFailed
-- | Under the owner's badge lock, which keeps a profile's requests to the service serial.
creditStoreReceipt :: UserId -> HeldStoreReceipt -> CM ()
creditStoreReceipt userId HeldStoreReceipt {receiptId, stash = stash@BadgeStash {masterKey}, payment, retryDelay} = 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 -> SRODeferred <$ defer (serviceFailure code retryAfter) retryAfter
Left e -> SRODeferred <$ defer (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
SRODeferred -> pure ()
creditStoreReceipt :: AgentClient -> UserId -> HeldStoreReceipt -> CM Bool
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
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
where
attempt = do
user <- withStore $ \db -> getUser db userId
@@ -5482,21 +5512,9 @@ creditStoreReceipt userId HeldStoreReceipt {receiptId, stash = stash@BadgeStash
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}
defer failure retryAfter = do
ri <- asks $ badgeRetryInterval . config
now <- badgeNow
let (delay, retryDelay') = storeReceiptRetry ri failure retryAfter retryDelay
withStore' $ \db -> deferStoreReceipt db receiptId (delay `addUTCTime` now) retryDelay' failure
-- | The service's retryAfter when given; otherwise a failure that can clear grows its own delay, and any
-- other waits a day. Unlike a renewal, an answered internal grows too: a buyer is watching the purchase.
storeReceiptRetry :: RetryInterval -> BadgeIssueFailure -> Maybe Word32 -> Maybe Int64 -> (NominalDiffTime, Maybe Int64)
storeReceiptRetry ri@RetryInterval {initialInterval} failure retryAfter retryDelay
| isJust retryAfter = (badgeRetryAfter ri retryAfter, retryDelay)
| badgeFailureTransient failure = (fromIntegral delay / 1000000, Just delay)
| otherwise = (badgeStalledInterval, retryDelay)
where
delay = maybe initialInterval (\d -> nextRetryDelay 0 d ri) retryDelay
failed delay failure retryAfter = do
withStore' $ \db -> recordStoreReceiptFailure db receiptId delay failure
pure $ if badgeFailureTransient failure then SRORetry retryAfter else SROFailed
-- | 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.
+24 -35
View File
@@ -26,9 +26,8 @@ module Simplex.Chat.Store.Badges
holdStoreReceipt,
createBadgeStoreReceipt,
getOpenStorePurchases,
getDueStoreReceipts,
getNextStoreReceiptAttempt,
deferStoreReceipt,
getNextDueStoreReceipt,
recordStoreReceiptFailure,
refuseStoreReceipt,
deleteUnfundedStoreReceipts,
deleteBadgeCodeRedemption,
@@ -50,7 +49,10 @@ 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
@@ -59,7 +61,7 @@ import Simplex.Chat.Badges.Service (StatementCreditType (..), StatementDebitType
import Simplex.Chat.Badges.Types (BadgeAlertKind, BadgeIssueError (..), BadgeIssueFailure, BadgePurchaseStatus (..), OpenStorePurchase (..))
import Simplex.Chat.PaymentService (ServicePayment)
import Simplex.Chat.PaymentService.Types (StoreTransactionRef (..))
import Simplex.Chat.Store.Shared (insertedRowId)
import Simplex.Chat.Store.Shared (StoreError, insertedRowId)
import Simplex.Chat.Types
import Simplex.Messaging.Agent.Protocol (UserId)
import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..))
@@ -139,8 +141,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 = ?, next_attempt_at = ? WHERE invoice_id = ? AND transaction_ref IS NULL"
(provider, transactionRef, paymentJSON, now, invoiceId)
"UPDATE badge_store_receipts SET provider = ?, transaction_ref = ?, payment = ? WHERE invoice_id = ? AND transaction_ref IS NULL"
(provider, transactionRef, paymentJSON, invoiceId)
getStoreReceipt db txRef >>= \case
Just r -> pure $ Just (r, True)
Nothing -> do
@@ -160,11 +162,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, next_attempt_at, purchase_key, purchase_priv_key, master_key, created_at)
VALUES (?,?,?,?,?,?,?,?,?,?)
(user_id, invoice_id, provider, transaction_ref, payment, purchase_key, purchase_priv_key, master_key, created_at)
VALUES (?,?,?,?,?,?,?,?,?)
ON CONFLICT DO NOTHING
|]
((userId, invoiceId', provider, transactionRef, paymentJSON, now) :. (purchaseKey, purchasePrivKey, Binary mk, now))
((userId, invoiceId', provider, transactionRef, paymentJSON) :. (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)
@@ -219,48 +221,35 @@ data HeldStoreReceipt = HeldStoreReceipt
retryDelay :: Maybe Int64
}
-- | Due only from next_attempt_at, so a pass a signal starts cannot send a deferred receipt early.
getDueStoreReceipts :: DB.Connection -> UserId -> UTCTime -> IO [HeldStoreReceipt]
getDueStoreReceipts db userId now =
map toHeld
getNextDueStoreReceipt :: DB.Connection -> UserId -> Set Int64 -> IO (Either StoreError (Maybe HeldStoreReceipt))
getNextDueStoreReceipt db userId parked =
Right . find notParked . map 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
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 r.next_attempt_at <= ? AND u.shown_badge_id IS NULL
ORDER BY r.next_attempt_at
WHERE r.user_id = ? AND r.payment IS NOT NULL AND u.shown_badge_id IS NULL
ORDER BY r.badge_store_receipt_id
|]
(userId, now)
(Only userId)
where
notParked HeldStoreReceipt {receiptId} = receiptId `S.notMember` parked
toHeld (stashRow@(receiptId, _, _, _) :. (payment, retryDelay)) =
HeldStoreReceipt {receiptId, stash = toBadgeStash BSRStoreReceipt stashRow, payment, retryDelay}
getNextStoreReceiptAttempt :: DB.Connection -> UserId -> IO (Maybe UTCTime)
getNextStoreReceiptAttempt db userId =
fmap join . maybeFirstRow fromOnly $
DB.query
db
[sql|
SELECT MIN(r.next_attempt_at)
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
|]
(Only userId)
deferStoreReceipt :: DB.Connection -> Int64 -> UTCTime -> Maybe Int64 -> BadgeIssueFailure -> IO ()
deferStoreReceipt db receiptId nextAttemptAt retryDelay failure =
recordStoreReceiptFailure :: DB.Connection -> Int64 -> Int64 -> BadgeIssueFailure -> IO ()
recordStoreReceiptFailure db receiptId retryDelay failure =
DB.execute
db
"UPDATE badge_store_receipts SET next_attempt_at = ?, retry_delay = ?, credit_error = ? WHERE badge_store_receipt_id = ? AND payment IS NOT NULL"
(nextAttemptAt, retryDelay, failure, receiptId)
"UPDATE badge_store_receipts SET retry_delay = ?, credit_error = ? WHERE badge_store_receipt_id = ? AND payment IS NOT NULL"
(retryDelay, 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, credit_error = ? WHERE badge_store_receipt_id = ?" (refusal, receiptId)
DB.execute db "UPDATE badge_store_receipts SET payment = NULL, retry_delay = NULL, 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 ()
@@ -296,7 +285,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 WHERE badge_store_receipt_id = ?" (Only storeReceiptId)
forM_ storeReceiptId_ $ \storeReceiptId -> DB.execute db "UPDATE badge_store_receipts SET payment = NULL, retry_delay = NULL WHERE badge_store_receipt_id = ?" (Only storeReceiptId)
getStashBadgePurchase db stash >>= \case
Just purchaseId -> pure (purchaseId, False)
Nothing -> do
@@ -24,7 +24,6 @@ CREATE TABLE badge_store_receipts(
master_key BYTEA NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
payment TEXT,
next_attempt_at TIMESTAMPTZ,
retry_delay BIGINT,
credit_error TEXT,
UNIQUE(provider, transaction_ref)
@@ -307,7 +307,6 @@ 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,
credit_error text
);
@@ -23,7 +23,6 @@ CREATE TABLE badge_store_receipts(
master_key BLOB NOT NULL,
created_at TEXT NOT NULL,
payment TEXT,
next_attempt_at TEXT,
retry_delay INTEGER,
credit_error TEXT,
UNIQUE(provider, transaction_ref)
@@ -993,7 +993,6 @@ CREATE TABLE badge_store_receipts(
master_key BLOB NOT NULL,
created_at TEXT NOT NULL,
payment TEXT,
next_attempt_at TEXT,
retry_delay INTEGER,
credit_error TEXT,
UNIQUE(provider, transaction_ref)