diff --git a/apps/simplex-store-service/src/StoreService/Names/Env.hs b/apps/simplex-store-service/src/StoreService/Names/Env.hs new file mode 100644 index 0000000000..325ff4538c --- /dev/null +++ b/apps/simplex-store-service/src/StoreService/Names/Env.hs @@ -0,0 +1,25 @@ +{-# LANGUAGE NamedFieldPuns #-} + +module StoreService.Names.Env (NamesEnv (..), newNamesEnv) where + +import Simplex.Chat.Controller (ChatController) +import Simplex.Messaging.Agent.Protocol (UserId) +import Simplex.Messaging.Agent.Store.Common (DBStore) +import StoreService.Config (NamesConfig) +import StoreService.Names.Phases (Phases, newPhases) +import StoreService.Names.Turns (Turns, newTurns) + +data NamesEnv = NamesEnv + { neChat :: ChatController, + neUserId :: UserId, + neConfig :: NamesConfig, + neStore :: DBStore, + nePhases :: Phases, + neTurns :: Turns + } + +newNamesEnv :: ChatController -> UserId -> NamesConfig -> DBStore -> IO NamesEnv +newNamesEnv neChat neUserId neConfig neStore = do + nePhases <- newPhases + neTurns <- newTurns + pure NamesEnv {neChat, neUserId, neConfig, neStore, nePhases, neTurns} diff --git a/apps/simplex-store-service/src/StoreService/Names/Phases.hs b/apps/simplex-store-service/src/StoreService/Names/Phases.hs new file mode 100644 index 0000000000..ec4a3c7ec6 --- /dev/null +++ b/apps/simplex-store-service/src/StoreService/Names/Phases.hs @@ -0,0 +1,57 @@ +{-# LANGUAGE NamedFieldPuns #-} + +module StoreService.Names.Phases (Phases, RegistrationId (..), newPhases, publish, awaitChange) where + +import Control.Concurrent.STM +import Control.Exception (bracket) +import Data.Int (Int64) +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Data.Text (Text) + +newtype RegistrationId = RegistrationId Text + deriving (Eq, Ord, Show) + +-- | wVersion only grows, so a publish that arrives after a newer one cannot wake anyone. +data Watch = Watch {wVersion :: TVar Int64, wRefs :: TVar Int} + +newtype Phases = Phases (TVar (Map RegistrationId Watch)) + +newPhases :: IO Phases +newPhases = Phases <$> newTVarIO Map.empty + +-- | Call after the transaction that wrote this version commits. +publish :: Phases -> RegistrationId -> Int64 -> STM () +publish (Phases watches) rid version = + readTVar watches >>= mapM_ (\Watch {wVersion} -> modifyTVar' wVersion (max version)) . Map.lookup rid + +subscribe :: Phases -> RegistrationId -> STM Watch +subscribe (Phases watches) rid = + readTVar watches >>= \ws -> case Map.lookup rid ws of + Just watch@Watch {wRefs} -> watch <$ modifyTVar' wRefs (+ 1) + Nothing -> do + watch <- Watch <$> newTVar minBound <*> newTVar 1 + watch <$ writeTVar watches (Map.insert rid watch ws) + +release :: Phases -> RegistrationId -> STM () +release (Phases watches) rid = + readTVar watches >>= \ws -> mapM_ (drop' ws) (Map.lookup rid ws) + where + drop' ws Watch {wRefs} = do + n <- pred <$> readTVar wRefs + if n <= 0 then writeTVar watches (Map.delete rid ws) else writeTVar wRefs n + +-- | True once the version differs from @seen@, False after @micros@ without a change. +-- Subscribes before reading the version, so a publish between the read and the wait is not missed. +awaitChange :: Phases -> RegistrationId -> IO Int64 -> Int64 -> Int -> IO Bool +awaitChange phases rid readVersion seen micros = + bracket (atomically $ subscribe phases rid) (const . atomically $ release phases rid) $ \Watch {wVersion} -> do + current <- readVersion + atomically $ modifyTVar' wVersion (max current) + if current /= seen + then pure True + else do + timer <- registerDelay micros + atomically $ + (readTVar wVersion >>= \v -> if v > seen then pure True else retry) + `orElse` (False <$ (readTVar timer >>= check)) diff --git a/apps/simplex-store-service/src/StoreService/Names/Turns.hs b/apps/simplex-store-service/src/StoreService/Names/Turns.hs new file mode 100644 index 0000000000..0ff9968594 --- /dev/null +++ b/apps/simplex-store-service/src/StoreService/Names/Turns.hs @@ -0,0 +1,46 @@ +{-# LANGUAGE NamedFieldPuns #-} + +module StoreService.Names.Turns (Turns, newTurns, withLabelTurn, withLabelTurns, withCodeTurn) where + +import Control.Concurrent.STM +import Control.Exception (bracket, bracket_) +import Data.ByteString (ByteString) +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Simplex.Messaging.SimplexName (LabelHash (..)) + +-- | Each lock counts the requests holding or waiting for it, and its entry goes when none are left. +type TurnLocks = TVar (Map ByteString (TMVar (), Int)) + +data Turns = Turns {labelTurns :: TurnLocks, codeTurns :: TurnLocks} + +newTurns :: IO Turns +newTurns = Turns <$> newTVarIO Map.empty <*> newTVarIO Map.empty + +withLabelTurn :: Turns -> LabelHash -> IO a -> IO a +withLabelTurn Turns {labelTurns} (LabelHash h) = withTurn labelTurns h + +-- | Taken in hash order, so two requests moving between the same names cannot each hold one and wait for the other. +withLabelTurns :: Turns -> LabelHash -> LabelHash -> IO a -> IO a +withLabelTurns turns a@(LabelHash ha) b@(LabelHash hb) action + | ha == hb = withLabelTurn turns a action + | ha < hb = withLabelTurn turns a $ withLabelTurn turns b action + | otherwise = withLabelTurn turns b $ withLabelTurn turns a action + +-- | Keyed by the code's SHA-256 hash. +withCodeTurn :: Turns -> ByteString -> IO a -> IO a +withCodeTurn Turns {codeTurns} = withTurn codeTurns + +withTurn :: TurnLocks -> ByteString -> IO a -> IO a +withTurn locks key action = + bracket (atomically enter) (const $ atomically leave) $ \lock -> + bracket_ (atomically $ takeTMVar lock) (atomically $ putTMVar lock ()) action + where + enter = do + held <- readTVar locks + case Map.lookup key held of + Just (lock, n) -> lock <$ writeTVar locks (Map.insert key (lock, n + 1) held) + Nothing -> do + lock <- newTMVar () + lock <$ writeTVar locks (Map.insert key (lock, 1) held) + leave = modifyTVar' locks (Map.update (\(lock, n) -> if n <= 1 then Nothing else Just (lock, n - 1)) key) diff --git a/apps/simplex-store-service/src/StoreService/Service.hs b/apps/simplex-store-service/src/StoreService/Service.hs index 334d081422..764f369cd9 100644 --- a/apps/simplex-store-service/src/StoreService/Service.hs +++ b/apps/simplex-store-service/src/StoreService/Service.hs @@ -20,9 +20,10 @@ where import StoreService.Catalog (defaultCatalog) import StoreService.Codes (issueFailedText, issueOneCode, singleUse) -import StoreService.Config (BadgeIssuerKey (..), ServiceConfig (..), readServiceConfig) +import StoreService.Config (BadgeIssuerKey (..), ChainConfig (..), NamesConfig (..), ServiceConfig (..), readServiceConfig) import StoreService.Group (GroupEvent, ensureManagedGroup, groupEvent, hasTracker, refreshTracker, revokeWithTracker, runGroupLane) import StoreService.Group.Command (badgeTypeP, codeP, maxMonths, textTokenP) +import StoreService.Names.Env (NamesEnv, newNamesEnv) import StoreService.Options import StoreService.Poller (newPollerEnv, newReadHints, runPoller) import StoreService.Providers.BTCPay (btcpayProvider) @@ -150,18 +151,27 @@ storeService opts@StoreServiceOpts {serviceConfigFile} cfg env = do simplexChatCore cfg {chatHooks} (mkChatOpts opts) $ \_ cc -> raceAny_ [readEvents cc, startLanes cc] where serviceLanes :: Waiters -> ChatController -> ServiceConfig -> IO [IO ()] - serviceLanes ws ChatController {chatStore} sc = do + serviceLanes ws cc@ChatController {chatStore} sc = do -- Seed before the listener accepts anything, since every checkout is priced from these. seedServiceCatalog chatStore btc <- maybe (pure []) (fmap (: []) . btcpayProvider) (btcpay sc) str <- maybe (pure []) (fmap (: []) . stripeProvider) (stripe sc) let providers = btc <> str hints <- newReadHints - webEnv <- newWebEnv chatStore sc ws hints providers + namesEnv <- forM (names sc) (startNames cc) + webEnv <- newWebEnv chatStore sc ws hints providers namesEnv exportWebapp (listener sc) (stripe sc) pollerEnv <- newPollerEnv chatStore ws hints providers (poll sc) pure [runWebListener webEnv, runPoller pollerEnv] +startNames :: ChatController -> NamesConfig -> IO NamesEnv +startNames cc@ChatController {chatStore} nc@NamesConfig {nTLD, nChain = ChainConfig {cChainId}} = + readTVarIO (currentUser cc) >>= \case + Nothing -> putStrLn "No current user" >> exitFailure + Just User {userId} -> do + logInfo $ "names: on, TLD " <> safeDecodeUtf8 (strEncode nTLD) <> ", chain " <> tshow cChainId + newNamesEnv cc userId nc chatStore + seedServiceCatalog :: DBStore -> IO () seedServiceCatalog st = do now <- truncateToSecond <$> getCurrentTime diff --git a/apps/simplex-store-service/src/StoreService/Web/Names.hs b/apps/simplex-store-service/src/StoreService/Web/Names.hs new file mode 100644 index 0000000000..cc2f969e0c --- /dev/null +++ b/apps/simplex-store-service/src/StoreService/Web/Names.hs @@ -0,0 +1,25 @@ +{-# LANGUAGE NamedFieldPuns #-} + +module StoreService.Web.Names (Route (..), namesRoutes, namesApp) where + +import Data.Text (Text) +import Network.HTTP.Types (Method) +import Network.Wai (Request, ResponseReceived, pathInfo) +import StoreService.Names.Env (NamesEnv) +import StoreService.Web.Http (HttpEnv, Limit, Respond, limited, only) + +data Route = Route + { rMethod :: Method, + rLimit :: Limit, + rHandler :: Request -> Respond -> IO ResponseReceived + } + +namesRoutes :: NamesEnv -> [Text] -> Maybe Route +namesRoutes _ _ = Nothing + +-- | Nothing for a path no route knows. The limit is taken only once the route and the method match. +namesApp :: HttpEnv -> ([Text] -> Maybe Route) -> Request -> Respond -> Maybe (IO ResponseReceived) +namesApp http routes req respond = serve <$> routes (pathInfo req) + where + serve Route {rMethod, rLimit, rHandler} = + only rMethod req respond $ limited http rLimit req respond $ rHandler req respond diff --git a/apps/simplex-store-service/src/StoreService/Web/Server.hs b/apps/simplex-store-service/src/StoreService/Web/Server.hs index 7a6318c1b6..c033953981 100644 --- a/apps/simplex-store-service/src/StoreService/Web/Server.hs +++ b/apps/simplex-store-service/src/StoreService/Web/Server.hs @@ -19,8 +19,10 @@ import StoreService.Config (BTCPayConfig (..), ListenerConfig (..), ServiceConfi import StoreService.Poller (ReadHints, queueReadHint) import StoreService.Providers (OrderDraft (..), Provider (..), ProviderError (..), ProviderInvoice (..), WebhookError (..)) import StoreService.Store.Invoices (CreateError (..), InvoicePayment (..), InvoiceRow (..), NewInvoice (..), cancelOpenInvoice, codeHashExists, createInvoiceRows, cryptoCurrencyText, getInvoice, getInvoiceByProviderRef, invoiceStatusText, newInvoiceId, paymentHolds, readCatalogRows, textToInvoiceStatus, truncateToSecond) +import StoreService.Names.Env (NamesEnv) import StoreService.Waiters (Seen, Waiters, awaitStatus, publish) import StoreService.Web.Http +import StoreService.Web.Names (namesApp, namesRoutes) import Control.Concurrent.STM import qualified Control.Exception as E import StoreService.Log (logError, logInfo, logWarn) @@ -60,14 +62,16 @@ data WebEnv = WebEnv weHints :: ReadHints, weHttp :: HttpEnv, -- | @Nothing@ when no Stripe key is configured, and the pristine @static_dir@ shell is served instead. - weShell :: Maybe LB.ByteString + weShell :: Maybe LB.ByteString, + -- | @Nothing@ when the ini has no @[names]@ section. + weNames :: Maybe NamesEnv } -newWebEnv :: DBStore -> ServiceConfig -> Waiters -> ReadHints -> [Provider] -> IO WebEnv -newWebEnv weStore weConfig weWaiters weHints weProviders = do +newWebEnv :: DBStore -> ServiceConfig -> Waiters -> ReadHints -> [Provider] -> Maybe NamesEnv -> IO WebEnv +newWebEnv weStore weConfig weWaiters weHints weProviders weNames = do weHttp <- newHttpEnv (listener weConfig) weShell <- prepareShell (listener weConfig) (stripe weConfig) - pure WebEnv {weStore, weConfig, weWaiters, weProviders, weHoldMicros = holdMicros, weHints, weHttp, weShell} + pure WebEnv {weStore, weConfig, weWaiters, weProviders, weHoldMicros = holdMicros, weHints, weHttp, weShell, weNames} listenerConfig :: WebEnv -> ListenerConfig listenerConfig WebEnv {weConfig} = listener weConfig @@ -98,9 +102,10 @@ webApp env@WebEnv {weHttp} req respond = case pathInfo req of ["api", "invoice", iid, "cancel"] -> only "POST" req respond $ limited weHttp createLimit req respond $ cancelInvoiceHandler env (InvoiceId iid) respond ["webhooks", "btcpay"] -> only "POST" req respond $ webhookHandler env PPCrypto "POST /webhooks/btcpay" req respond ["webhooks", "stripe"] -> only "POST" req respond $ webhookHandler env PPStripe "POST /webhooks/stripe" req respond - _ -> respond notFound + _ -> fromMaybe (respond notFound) names where serving = lServeWebapp (listenerConfig env) + names = weNames env >>= \n -> namesApp weHttp (namesRoutes n) req respond -- | A @".."@ check would miss separators WAI has already decoded, an absolute path, and a symlink out of the tree. serveStatic :: WebEnv -> [Text] -> Respond -> IO ResponseReceived diff --git a/simplex-chat.cabal b/simplex-chat.cabal index 2812fc3435..267d4f8f51 100644 --- a/simplex-chat.cabal +++ b/simplex-chat.cabal @@ -442,6 +442,11 @@ executable simplex-store-service StoreService.Group StoreService.Group.Command StoreService.Log + StoreService.Names.Chain + StoreService.Names.Env + StoreService.Names.Phases + StoreService.Names.Registration + StoreService.Names.Turns StoreService.Options StoreService.Orders StoreService.Poller @@ -454,6 +459,7 @@ executable simplex-store-service StoreService.Store.Migrate StoreService.Waiters StoreService.Web.Http + StoreService.Web.Names StoreService.Web.Server Paths_simplex_chat ghc-options: -O2 -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-unused-packages -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=missing-methods -Werror=incomplete-uni-patterns -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -Werror=name-shadowing -threaded -rtsopts @@ -739,6 +745,7 @@ test-suite simplex-chat-test Bots.StoreService.GroupIntegrationTests Bots.StoreService.GroupTests Bots.StoreService.NameRegistrationTests + Bots.StoreService.NamesWiringTests Bots.StoreService.StripeTests Bots.StoreService.WaitersTests Bots.StoreService.WebTests @@ -761,7 +768,10 @@ test-suite simplex-chat-test StoreService.Group.Command StoreService.Log StoreService.Names.Chain + StoreService.Names.Env + StoreService.Names.Phases StoreService.Names.Registration + StoreService.Names.Turns StoreService.Options StoreService.Orders StoreService.Poller @@ -774,6 +784,7 @@ test-suite simplex-chat-test StoreService.Store.Migrate StoreService.Waiters StoreService.Web.Http + StoreService.Web.Names StoreService.Web.Server Paths_simplex_chat if flag(client_postgres) diff --git a/tests/Bots/StoreService/NamesWiringTests.hs b/tests/Bots/StoreService/NamesWiringTests.hs new file mode 100644 index 0000000000..87df5a43ef --- /dev/null +++ b/tests/Bots/StoreService/NamesWiringTests.hs @@ -0,0 +1,225 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE OverloadedStrings #-} + +module Bots.StoreService.NamesWiringTests (namesWiringTests) where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (async, concurrently_, replicateConcurrently_, wait) +import qualified Control.Concurrent.Async as Async +import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) +import Control.Concurrent.STM +import qualified Control.Exception as E +import Control.Monad (replicateM_) +import qualified Data.Aeson as J +import qualified Data.Aeson.KeyMap as KM +import qualified Data.ByteString.Char8 as BC +import qualified Data.ByteString.Lazy as LB +import Data.Maybe (fromMaybe, isNothing) +import Network.HTTP.Client (Manager, Response, defaultManagerSettings, httpLbs, newManager, parseRequest, responseBody, responseHeaders, responseStatus) +import qualified Network.HTTP.Client as HTTP +import Network.HTTP.Types (Method, hCacheControl, hContentType, status200) +import Network.HTTP.Types.Status (statusCode) +import qualified Network.Wai.Handler.Warp as Warp +import Simplex.Messaging.SimplexName (labelHash) +import StoreService.Config (ListenerConfig (..)) +import StoreService.Names.Phases +import StoreService.Names.Turns +import StoreService.Web.Http +import StoreService.Web.Names (Route (..), namesApp) +import System.Timeout (timeout) +import Test.Hspec + +namesWiringTests :: Spec +namesWiringTests = describe "StoreService names: wiring" $ do + describe "phases" $ do + it "wakes a waiter on a newer published version, not on an older or the same one" testPublishWakesOnNewer + it "answers at once when the version read is not the one seen" testVersionAlreadyMoved + it "catches a publish that lands between subscribing and reading" testPublishBeforeRead + it "gives up after the hold when nothing is published" testHoldTimesOut + it "is not woken by another registration's publish" testOtherRegistration + describe "router" $ do + it "answers a known path with another method 405 with Allow, spending no token" testWrongMethod + it "leaves an unknown path to the caller" testUnknownPath + describe "turns" $ do + it "runs two holders of one label hash one after the other" testOneLabelSerialized + it "runs holders of different label hashes at once" testDifferentLabelsConcurrent + it "runs two holders of one code hash one after the other" testOneCodeSerialized + it "holds both label hashes in withLabelTurns, and one when they are equal" testLabelTurnsHoldBoth + it "never deadlocks withLabelTurns A->B against B->A" testLabelTurnsNoDeadlock + it "releases a turn whose holder throws" testTurnReleasedOnThrow + +-- | Long enough that only a missed wake-up could make a test wait it out. +longHold :: Int +longHold = 10000000 + +-- | How long a test waits for something that must happen promptly. +prompt :: Int +prompt = 2000000 + +parkDelay :: Int +parkDelay = 100000 + +rid1, rid2 :: RegistrationId +rid1 = RegistrationId "reg-1" +rid2 = RegistrationId "reg-2" + +testPublishWakesOnNewer :: IO () +testPublishWakesOnNewer = do + phases <- newPhases + parked <- newEmptyMVar + waiter <- async $ awaitChange phases rid1 (putMVar parked () >> pure 3) 3 longHold + takeMVar parked + atomically $ publish phases rid1 2 + atomically $ publish phases rid1 3 + threadDelay parkDelay + Async.poll waiter >>= (`shouldSatisfy` isNothing) + atomically $ publish phases rid1 4 + timeout prompt (wait waiter) `shouldReturn` Just True + +testVersionAlreadyMoved :: IO () +testVersionAlreadyMoved = do + phases <- newPhases + timeout prompt (awaitChange phases rid1 (pure 5) 3 longHold) `shouldReturn` Just True + +testPublishBeforeRead :: IO () +testPublishBeforeRead = do + phases <- newPhases + -- the row read here is the old one, though version 4 was committed and published while it ran + let staleRead = atomically (publish phases rid1 4) >> pure 3 + timeout prompt (awaitChange phases rid1 staleRead 3 longHold) `shouldReturn` Just True + +testHoldTimesOut :: IO () +testHoldTimesOut = do + phases <- newPhases + timeout prompt (awaitChange phases rid1 (pure 3) 3 parkDelay) `shouldReturn` Just False + +testOtherRegistration :: IO () +testOtherRegistration = do + phases <- newPhases + parked <- newEmptyMVar + waiter <- async $ awaitChange phases rid1 (putMVar parked () >> pure 1) 1 (3 * parkDelay) + takeMVar parked + atomically $ publish phases rid2 7 + timeout prompt (wait waiter) `shouldReturn` Just False + +type Client = (Manager, String) + +probeLimit :: Limit +probeLimit = Limit {lmName = "probe", lmPerMinute = 1} + +withProbeRouter :: (Client -> IO a) -> IO a +withProbeRouter action = do + http <- newHttpEnv ListenerConfig {lHost = "127.0.0.1", lPort = 0, lStaticDir = "", lServeWebapp = False, lWebappExportDir = Nothing, lTrustForwardedFor = False, lNamesStaticDir = Nothing, lNamesExportDir = Nothing, lNamesHosts = []} + let routes = \case + ["api", "probe", _] -> Just Route {rMethod = "GET", rLimit = probeLimit, rHandler = \_ respond -> respond (jsonResponse status200 (J.object []))} + _ -> Nothing + app req respond = fromMaybe (respond notFound) (namesApp http routes req respond) + mgr <- newManager defaultManagerSettings + Warp.testWithApplication (pure app) $ \port -> action (mgr, "http://127.0.0.1:" <> show port) + +send :: Client -> Method -> String -> IO (Response LB.ByteString) +send (mgr, base) verb target = do + req <- parseRequest (base <> target) + httpLbs req {HTTP.method = verb} mgr + +errorTag :: Response LB.ByteString -> Maybe J.Value +errorTag r = case J.decode (responseBody r) of + Just (J.Object o) -> KM.lookup "error" o + _ -> Nothing + +testWrongMethod :: IO () +testWrongMethod = withProbeRouter $ \client -> do + replicateM_ 2 $ do + r <- send client "POST" "/api/probe/x" + statusCode (responseStatus r) `shouldBe` 405 + lookup "Allow" (responseHeaders r) `shouldBe` Just "GET" + lookup hContentType (responseHeaders r) `shouldBe` Just "application/json" + lookup hCacheControl (responseHeaders r) `shouldBe` Just "no-store" + errorTag r `shouldBe` Just (J.String "method_not_allowed") + allowed <- send client "GET" "/api/probe/x" + statusCode (responseStatus allowed) `shouldBe` 200 + limitedR <- send client "GET" "/api/probe/x" + statusCode (responseStatus limitedR) `shouldBe` 429 + errorTag limitedR `shouldBe` Just (J.String "rate_limited") + lookup "Retry-After" (responseHeaders limitedR) `shouldSatisfy` maybe False (BC.all (`elem` ['0' .. '9'])) + +testUnknownPath :: IO () +testUnknownPath = withProbeRouter $ \client -> do + r <- send client "POST" "/api/probe" + statusCode (responseStatus r) `shouldBe` 404 + errorTag r `shouldBe` Just (J.String "not_found") + +-- | Runs the holders concurrently and returns the most that were inside at once. +mostInside :: Int -> ((IO () -> IO ()) -> IO ()) -> IO Int +mostInside n holder = do + inside <- newTVarIO (0 :: Int) + most <- newTVarIO 0 + let enter = atomically $ do + k <- (+ 1) <$> readTVar inside + writeTVar inside k + modifyTVar' most (max k) + leave = atomically $ modifyTVar' inside (subtract 1) + replicateConcurrently_ n $ holder (\work -> enter >> work >> leave) + readTVarIO most + +testOneLabelSerialized :: IO () +testOneLabelSerialized = do + turns <- newTurns + r <- timeout (5 * prompt) $ mostInside 8 $ \inTurn -> withLabelTurn turns (labelHash "alice1") (inTurn (threadDelay 20000)) + r `shouldBe` Just 1 + +testOneCodeSerialized :: IO () +testOneCodeSerialized = do + turns <- newTurns + r <- timeout (5 * prompt) $ mostInside 8 $ \inTurn -> withCodeTurn turns (BC.replicate 32 'c') (inTurn (threadDelay 20000)) + r `shouldBe` Just 1 + +-- | Each holder waits inside its turn for the other to be inside its own, which only concurrent turns allow. +testDifferentLabelsConcurrent :: IO () +testDifferentLabelsConcurrent = do + turns <- newTurns + aInside <- newEmptyMVar + bInside <- newEmptyMVar + r <- + timeout prompt $ + concurrently_ + (withLabelTurn turns (labelHash "alice1") (putMVar aInside () >> takeMVar bInside)) + (withLabelTurn turns (labelHash "bobby2") (putMVar bInside () >> takeMVar aInside)) + r `shouldBe` Just () + +testLabelTurnsHoldBoth :: IO () +testLabelTurnsHoldBoth = do + turns <- newTurns + let a = labelHash "alice1" + b = labelHash "bobby2" + withLabelTurns turns a b $ do + timeout parkDelay (withLabelTurn turns a (pure ())) `shouldReturn` Nothing + timeout parkDelay (withLabelTurn turns b (pure ())) `shouldReturn` Nothing + timeout prompt (withLabelTurns turns a a (pure ())) `shouldReturn` Just () + +testLabelTurnsNoDeadlock :: IO () +testLabelTurnsNoDeadlock = do + turns <- newTurns + let a = labelHash "alice1" + b = labelHash "bobby2" + rounds = 200 + r <- + timeout (5 * prompt) $ + concurrently_ + (replicateConcurrently_ rounds (withLabelTurns turns a b (threadDelay 50))) + (replicateConcurrently_ rounds (withLabelTurns turns b a (threadDelay 50))) + r `shouldBe` Just () + +data Boom = Boom + deriving (Show) + +instance E.Exception Boom + +testTurnReleasedOnThrow :: IO () +testTurnReleasedOnThrow = do + turns <- newTurns + let a = labelHash "alice1" + E.try (withLabelTurn turns a (E.throwIO Boom)) >>= \case + Left Boom -> pure () + Right () -> expectationFailure "the holder's exception must reach the caller" + timeout prompt (withLabelTurn turns a (pure ())) `shouldReturn` Just () diff --git a/tests/Bots/StoreService/WebTests.hs b/tests/Bots/StoreService/WebTests.hs index 3634923275..54f7a394a4 100644 --- a/tests/Bots/StoreService/WebTests.hs +++ b/tests/Bots/StoreService/WebTests.hs @@ -1040,7 +1040,7 @@ withListener providers rethrow hold st cfg action = do let runner = if rethrow then Warp.testWithApplicationSettings else Warp.withApplicationSettings waiters <- newWaiters hints <- newReadHints - built <- newWebEnv st cfg waiters hints providers + built <- newWebEnv st cfg waiters hints providers Nothing let env = built {weHoldMicros = hold} mgr <- newManager defaultManagerSettings runner (webSettings (listener cfg)) (pure (webApp env)) $ \prt -> diff --git a/tests/Test.hs b/tests/Test.hs index ab4934c616..77b3970e7d 100644 --- a/tests/Test.hs +++ b/tests/Test.hs @@ -10,6 +10,7 @@ import Bots.StoreService.ConfigTests import Bots.StoreService.GroupIntegrationTests import Bots.StoreService.GroupTests import Bots.StoreService.NameRegistrationTests +import Bots.StoreService.NamesWiringTests import Bots.StoreService.StripeTests import Bots.StoreService.WaitersTests import Bots.StoreService.WebTests @@ -87,6 +88,7 @@ main = do badgeStripeTests badgeNameRegistrationTests namesConfigTests + namesWiringTests describe "SimpleX chat markdown" markdownTests describe "JSON Tests" jsonTests describe "Member relations" memberRelationsTests