store: wire the names env, phases, turns, router

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}

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))

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)

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

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

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

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)

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 ()

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 ->

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