store: hold race tests at a gated chain read

tests/Bots/StoreService/FakeEthNode.hs

@@ -6,6 +6,7 @@ module Bots.StoreService.FakeEthNode
   ( FakeEthNode (..),
     CallAnswer (..),
     Fault (..),
+    Gate,
     withFakeEthNode,
     fakeChainConfig,
     answerCall,
@@ -13,6 +14,10 @@ module Bots.StoreService.FakeEthNode
     forgetCall,
     setFault,
     clearFault,
+    gateNext,
+    awaitGate,
+    releaseGate,
+    pausedAt,
     setBlock,
     setFees,
     revertDryRun,
@@ -37,6 +42,7 @@ module Bots.StoreService.FakeEthNode
 where
 
 import Control.Concurrent (threadDelay)
+import Control.Concurrent.Async (Async, async, cancel)
 import Control.Concurrent.STM
 import Control.Exception (finally)
 import Data.Aeson ((.:), (.=))
@@ -63,6 +69,7 @@ import Simplex.Messaging.Util (eitherToMaybe)
 import StoreService.Config (ChainConfig (..))
 import StoreService.Names.Abi (hex)
 import StoreService.Names.Chain (TxRequest (..), decodeSignedTx)
+import System.Timeout (timeout)
 
 data FakeEthNode = FakeEthNode
   { fnUrl :: String,
@@ -81,6 +88,10 @@ data Fault
   | Answers J.Value
   | -- | answers as usual after this many microseconds
     Slow Int
+  | -- | the next request waits at the gate until it is released, then answers as usual; later ones pass
+    Gated Gate
+
+data Gate = Gate {gArrived :: TMVar (), gRelease :: TMVar ()}
 
 data FakeState = FakeState
   { fsCalls :: Map (ByteString, ByteString) CallAnswer,
@@ -182,6 +193,30 @@ setFault FakeEthNode {fnState} method f = atomically $ modifyTVar' fnState $ \s
 clearFault :: FakeEthNode -> Text -> IO ()
 clearFault FakeEthNode {fnState} method = atomically $ modifyTVar' fnState $ \s -> s {fsFaults = M.delete method $ fsFaults s}
 
+-- | The next request for the method stops at the gate it returns.
+gateNext :: FakeEthNode -> Text -> IO Gate
+gateNext node method = do
+  gate <- Gate <$> newEmptyTMVarIO <*> newEmptyTMVarIO
+  gate <$ setFault node method (Gated gate)
+
+-- | Returns once a request is held at the gate.
+awaitGate :: Gate -> IO ()
+awaitGate Gate {gArrived} = atomically $ readTMVar gArrived
+
+releaseGate :: Gate -> IO ()
+releaseGate Gate {gRelease} = atomically $ putTMVar gRelease ()
+
+-- | Far longer than a request takes to reach its first chain read.
+gateArrival :: Int
+gateArrival = 5000000
+
+-- | The action, run until its first request for the method is held at the gate.
+pausedAt :: FakeEthNode -> Text -> IO a -> IO (Async a, Gate)
+pausedAt node method action = do
+  gate <- gateNext node method
+  inFlight <- async action
+  timeout gateArrival (awaitGate gate) >>= maybe (cancel inFlight >> fail ("no " <> T.unpack method <> " reached the gate")) (const $ pure (inFlight, gate))
+
 revertFault :: ByteString -> Fault
 revertFault = RpcError 3 "execution reverted" . Just
 
@@ -243,12 +278,22 @@ fakeApp st req respond = do
         Just (RpcError code message revertData) -> reply $ rpcError code message revertData
         Just NotJson -> respond $ responseLBS ok200 [] "<html>Sign in to provider-api-key</html>"
         Just (Answers v) -> reply ["result" .= v]
-        Just (Slow delay) -> threadDelay delay >> (reply =<< atomically (answer st method params))
-        Nothing -> reply =<< atomically (answer st method params)
+        Just (Slow delay) -> threadDelay delay >> answered method params
+        Just (Gated Gate {gArrived, gRelease}) -> do
+          atomically $ putTMVar gArrived ()
+          atomically $ takeTMVar gRelease `orElse` (readTVar st >>= check . fsClosed)
+          answered method params
+        Nothing -> answered method params
     Nothing -> respond $ responseLBS badRequest400 [] "not a JSON-RPC request"
   where
     request = J.withObject "request" $ \o -> (,) <$> o .: "method" <*> o .: "params"
-    fault method = M.lookup method . fsFaults <$> readTVarIO st
+    -- a gate holds one request only
+    fault method = atomically $ do
+      s <- readTVar st
+      case M.lookup method $ fsFaults s of
+        f@(Just Gated {}) -> f <$ writeTVar st s {fsFaults = M.delete method $ fsFaults s}
+        f -> pure f
+    answered method params = reply =<< atomically (answer st method params)
     reply r = respond $ responseLBS ok200 [(hContentType, "application/json")] $ J.encode $ J.object $ ["jsonrpc" .= ("2.0" :: Text), "id" .= (1 :: Int)] <> r
 
 answer :: TVar FakeState -> Text -> [J.Value] -> STM [JT.Pair]

tests/Bots/StoreService/NamesOwnerTests.hs

@@ -16,7 +16,7 @@ import Control.Concurrent (threadDelay)
 import Control.Concurrent.Async (Async, async, cancel, wait)
 import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
 import Control.Concurrent.STM (atomically, check)
-import Control.Monad (forM, forM_, unless, void)
+import Control.Monad (forM, forM_, void)
 import qualified Data.Aeson as J
 import qualified Data.Aeson.KeyMap as KM
 import Data.ByteString (ByteString)
@@ -50,7 +50,7 @@ import StoreService.Names.Orders (OrdersEnv (..), signedRequestSeconds)
 import StoreService.Names.Phases (RegistrationId (..), phasesWaiting)
 import StoreService.Names.Signatures (OrderFields (..), findMessage, parseAnyCaseAddress, personalDigest, retargetMessage, takenCodeMessage)
 import StoreService.Names.Store hiding (findLimit)
-import StoreService.Names.Turns (turnsCount, withLabelTurn)
+import StoreService.Names.Turns (withLabelTurn)
 import StoreService.Store (RevokeResult (..))
 import StoreService.Web.Http (createLimit, findLimit, holdMicros, readLimit)
 import StoreService.Web.Names (Route (..), namesApp)
@@ -319,12 +319,6 @@ codeRow Ctx {cStore} PrepaidCode {pcId} = withTransaction cStore $ \db -> fromJu
 within :: String -> IO a -> IO a
 within what action = timeout 10000000 action >>= maybe (error $ what <> " did not finish") pure
 
--- | Waits until the turns have at least this many label or code hashes held or waited on.
-awaitTurns :: Ctx -> Int -> IO ()
-awaitTurns Ctx {cOrders} n = within "turns" loop
-  where
-    loop = turnsCount (oeTurns cOrders) >>= \count -> unless (count >= n) (threadDelay 1000 >> loop)
-
 -- | A registration poll at the order's version, once the server holds it.
 held :: Ctx -> NameRegistration -> IO (Async (Int, J.Object))
 held ctx@Ctx {cOrders} NameRegistration {nrId, nrVersion} = do
@@ -340,14 +334,6 @@ woken poll = timeout 5000000 (wait poll) >>= maybe (cancel poll >> error "the po
 parkDelay :: Int
 parkDelay = 200000
 
--- | Each eth_call's delay in the lost-race tests, far longer than a request takes to read its registration.
-slowRead :: Int
-slowRead = 3000000
-
--- | How long after the turns are held the lost-race tests change the order: past the request's read, before its write.
-raceDelay :: Int
-raceDelay = 1000000
-
 -- tests
 
 testFind :: IO ()
@@ -679,12 +665,10 @@ testMoveLostRace = withOwner $ \ctx@Ctx {cNode} -> do
   answerFree ctx "alicebakers"
   r <- paidIn ctx aliceOrder failedPhase
   to <- moveTo ctx "alicebakers" 10
-  setFault cNode "eth_call" (Slow slowRead)
-  moving <- async $ move ctx keyA r to
-  awaitTurns ctx 2
-  threadDelay raceDelay
+  -- held at the register check's ownerOf read, inside the turns
+  (moving, gate) <- pausedAt cNode "eth_call" $ move ctx keyA r to
   changed <- rewrite ctx r id
-  clearFault cNode "eth_call"
+  releaseGate gate
   answer <- within "the move" $ wait moving
   view <- getView ctx changed
   answer `shouldBe` refusedWithView "registration_changed" view
@@ -821,12 +805,10 @@ testCodeLostRace :: IO ()
 testCodeLostRace = withOwner $ \ctx@Ctx {cNode, cStore} -> do
   r <- takenTwelve ctx
   let (hashText, hash) = takenCodeHash 1
-  setFault cNode "eth_call" (Slow slowRead)
-  issuing <- async $ takeCode ctx keyA r hashText 8
-  awaitTurns ctx 1
-  threadDelay raceDelay
+  -- held at the length read, inside the turn
+  (issuing, gate) <- pausedAt cNode "eth_call" $ takeCode ctx keyA r hashText 8
   changed <- rewrite ctx r id
-  clearFault cNode "eth_call"
+  releaseGate gate
   answer <- within "the code" $ wait issuing
   view <- getView ctx changed
   answer `shouldBe` refusedWithView "registration_changed" view

tests/Bots/StoreService/NamesRevealTests.hs

@@ -13,10 +13,10 @@ import Bots.StoreService.NamesOwnerTests (answerFree, moveBody, moveTo, postMove
 import Bots.StoreService.NamesPhasesTests (change, held, prompt, registrationPath, seenVersion, woken)
 import Bots.StoreService.NamesStoreTests (ownerA)
 import Bots.StoreService.WebTests (bounded, jsonObject, statusOf, webGet, webRequestBody)
-import Control.Concurrent.Async (Async, async, cancel, wait)
+import Control.Concurrent.Async (Async, async, wait)
 import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
 import Control.Concurrent.STM
-import Control.Monad (forM_, replicateM, when)
+import Control.Monad (forM_, replicateM)
 import Crypto.Number.Serialize (i2ospOf_, os2ip)
 import qualified Crypto.PubKey.ECC.Types as ECC
 import qualified Data.Aeson as J
@@ -32,19 +32,16 @@ import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock.POSIX (posixSecondsToUTCTime, utcTimeToPOSIXSeconds)
 import Data.Word (Word64)
-import Network.HTTP.Client (Manager, RequestBody (..), defaultManagerSettings, httpLbs, method, newManager, parseRequest, requestBody, requestHeaders, responseBody, responseStatus)
+import Network.HTTP.Client (defaultManagerSettings, newManager)
 import Network.HTTP.Types (hContentType)
-import Network.Wai (Application, responseLBS, strictRequestBody)
+import Network.Wai (Application)
 import qualified Network.Wai.Handler.Warp as Warp
 import Simplex.Chat.PaymentService.Types (InvoiceStatus (..))
 import Simplex.Messaging.Agent.Store.Common (withTransaction)
 import Simplex.Messaging.Crypto.Secp256k1 (RecoverableSignature (..), Secp256k1PrivateKey, signRecoverable)
 import Simplex.Messaging.Eth.EIP712 (serializeSignature)
 import Simplex.Messaging.SimplexName (LabelHash (..), SimplexTLD (..), labelHash)
-import StoreService.Config (ChainConfig (..))
 import StoreService.Names.Abi (hex, uint256)
-import StoreService.Names.Chain (newChainClient)
-import StoreService.Names.Contracts (newContracts)
 import StoreService.Names.Links (linkMaxBytes)
 import StoreService.Names.Orders (OrdersEnv (..))
 import StoreService.Names.Phases (RegistrationId (..))
@@ -85,49 +82,26 @@ namesRevealTests = describe "StoreService names: reveal" $ do
 
 -- fixtures
 
--- | The fake node behind a proxy that, once armed, holds the next eth_call until released.
-data Gate = Gate {gArmed :: TVar Bool, gArrived :: TMVar (), gRelease :: TMVar ()}
-
-data Fx = Fx {fCtx :: Ctx, fGate :: Gate, fApp :: Application}
+data Fx = Fx {fCtx :: Ctx, fApp :: Application}
 
 -- | The registration routes as namesRoutes registers them, the reveal's and another name's included.
 withReveals :: (Fx -> IO a) -> IO a
-withReveals action = withOrders $ \ctx@Ctx {cNode, cOrders, cWeb, cRandom} -> do
-  gate <- Gate <$> newTVarIO False <*> newEmptyTMVarIO <*> newEmptyTMVarIO
-  mgr <- newManager defaultManagerSettings
+withReveals action = withOrders $ \ctx@Ctx {cNode, cOrders, cWeb} -> do
   answerCall cNode fakeSponsor noncesCall $ Returns $ uint256 (0 :: Int)
-  Warp.testWithApplication (pure $ gatedNode gate mgr cNode) $ \nodePort -> do
-    ct <- newContracts TLDSimplex =<< newChainClient cRandom (fakeChainConfig cNode) {cRpcUrl = "http://127.0.0.1:" <> show nodePort}
-    let orders = cOrders {oeContracts = ct}
-        web = cWeb {weOrders = Just orders}
-        routes = \case
-          ["api", "registration", rid] -> Just Route {rMethod = "GET", rLimit = readLimit, rHandler = registrationHandler orders holdMicros rid}
-          ["api", "registration", rid, "reveal"] -> Just Route {rMethod = "POST", rLimit = readLimit, rHandler = revealHandler orders rid}
-          ["api", "registration", rid, "name"] -> Just Route {rMethod = "POST", rLimit = createLimit, rHandler = retargetHandler orders rid}
-          _ -> Nothing
-        app req respond = fromMaybe (webApp web req respond) (namesApp (weHttp web) routes req respond)
-    Warp.testWithApplication (pure app) $ \port ->
-      action Fx {fCtx = ctx {cOrders = orders, cWeb = web, cClient = (mgr, "http://127.0.0.1:" <> show port)}, fGate = gate, fApp = app}
-
-gatedNode :: Gate -> Manager -> FakeEthNode -> Application
-gatedNode Gate {gArmed, gArrived, gRelease} mgr FakeEthNode {fnUrl} req respond = do
-  body <- strictRequestBody req
-  paused <- atomically $ readTVar gArmed >>= \armed -> armed <$ when armed (writeTVar gArmed False >> putTMVar gArrived ())
-  when paused $ atomically $ takeTMVar gRelease
-  forward <- parseRequest fnUrl
-  r <- httpLbs forward {method = "POST", requestBody = RequestBodyLBS body, requestHeaders = [(hContentType, "application/json")]} mgr
-  respond $ responseLBS (responseStatus r) [(hContentType, "application/json")] (responseBody r)
+  let web = cWeb {weOrders = Just cOrders}
+      routes = \case
+        ["api", "registration", rid] -> Just Route {rMethod = "GET", rLimit = readLimit, rHandler = registrationHandler cOrders holdMicros rid}
+        ["api", "registration", rid, "reveal"] -> Just Route {rMethod = "POST", rLimit = readLimit, rHandler = revealHandler cOrders rid}
+        ["api", "registration", rid, "name"] -> Just Route {rMethod = "POST", rLimit = createLimit, rHandler = retargetHandler cOrders rid}
+        _ -> Nothing
+      app req respond = fromMaybe (webApp web req respond) (namesApp (weHttp web) routes req respond)
+  mgr <- newManager defaultManagerSettings
+  Warp.testWithApplication (pure app) $ \port ->
+    action Fx {fCtx = ctx {cWeb = web, cClient = (mgr, "http://127.0.0.1:" <> show port)}, fApp = app}
 
 -- | The request, held at its first chain read: a reveal's nonce read, its last check before it writes, or /name's name check, in its turns.
-pausedAtChainRead :: Gate -> IO a -> IO (Async a)
-pausedAtChainRead Gate {gArmed, gArrived} request = do
-  atomically $ writeTVar gArmed True
-  inFlight <- async request
-  timeout prompt (atomically $ takeTMVar gArrived) >>= maybe (cancel inFlight >> fail "the request did not reach a chain read") pure
-  pure inFlight
-
-releaseRead :: Gate -> IO ()
-releaseRead Gate {gRelease} = atomically $ putTMVar gRelease ()
+pausedAtChainRead :: Ctx -> IO a -> IO (Async a, Gate)
+pausedAtChainRead Ctx {cNode} = pausedAt cNode "eth_call"
 
 orderSeconds :: Word64
 orderSeconds = floor $ utcTimeToPOSIXSeconds orderTime
@@ -461,14 +435,14 @@ testNoRefusalRecorded = withReveals $ \Fx {fCtx = ctx} -> do
 
 -- | A writer that takes no turn, as the worker re-committing, moves the version while the reveal reads the nonce.
 testLostRace :: IO ()
-testLostRace = bounded "lost race" $ withReveals $ \Fx {fCtx = ctx, fGate} -> do
+testLostRace = bounded "lost race" $ withReveals $ \Fx {fCtx = ctx} -> do
   let overBound = dynamis {rContactLink = Just $ padded "a" (linkMaxBytes + 1)}
   forM_ [(alice, "accepted"), (overBound, "links refused")] $ \(r, what :: String) -> do
     rid <- committed ctx (orderOf r)
     body <- ownerBody ctx r
-    inFlight <- pausedAtChainRead fGate (reveal ctx rid body)
+    (inFlight, gate) <- pausedAtChainRead ctx (reveal ctx rid body)
     v <- change ctx rid $ \x -> x {nrRevealAfter = (+ 60) <$> nrRevealAfter x, nrRevealBefore = (+ 60) <$> nrRevealBefore x}
-    releaseRead fGate
+    releaseGate gate
     (status, answer) <- wait inFlight
     let view = registrationOf $ fromJust answer
     (what, status, field "error" =<< answer, field "version" view, field "label" view, field "linksRefused" view, field "revealBefore" view)
@@ -476,15 +450,15 @@ testLostRace = bounded "lost race" $ withReveals $ \Fx {fCtx = ctx, fGate} -> do
     (what,) . unwritten <$> stored ctx rid `shouldReturn` (what, (v, Nothing, False))
 
 testRacesName :: IO ()
-testRacesName = bounded "races /name" $ withReveals $ \Fx {fCtx = ctx, fGate} -> do
+testRacesName = bounded "races /name" $ withReveals $ \Fx {fCtx = ctx} -> do
   -- the reveal holds the name's turn while it reads the nonce, so /name waits and then finds it revealed
   rid <- committed ctx (orderOf alice)
   answerFree ctx "alicebakers"
   toAlicebakers <- moveTo ctx "alicebakers" 1
-  inFlight <- pausedAtChainRead fGate (ownerReveal ctx rid alice)
+  (inFlight, gate) <- pausedAtChainRead ctx (ownerReveal ctx rid alice)
   move <- async $ postMove ctx rid =<< moveBody ctx keyA rid toAlicebakers
   queuedOn ctx (labelHash "alicebakery") 2
-  releaseRead fGate
+  releaseGate gate
   wait inFlight `shouldReturn` accepted
   wait move `shouldReturn` badRequest
   NameRegistration {nrLabelHash, nrReveal, nrMoves} <- stored ctx rid
@@ -496,10 +470,10 @@ testRacesName = bounded "races /name" $ withReveals $ \Fx {fCtx = ctx, fGate} ->
   body <- ownerBody ctx overBound
   reveal ctx refusedRid body `shouldReturn` refused 400 "invalid_link"
   toDynamit <- moveTo ctx "dynamit" 2
-  move' <- pausedAtChainRead fGate (postMove ctx refusedRid =<< moveBody ctx keyA refusedRid toDynamit)
+  (move', gate') <- pausedAtChainRead ctx (postMove ctx refusedRid =<< moveBody ctx keyA refusedRid toDynamit)
   late <- async $ reveal ctx refusedRid body
   queuedOn ctx (labelHash "dynamis") 2
-  releaseRead fGate
+  releaseGate gate'
   (200, Just movedView) <- wait move'
   (status, answer) <- wait late
   let view = registrationOf $ fromJust answer