From ff87acec0b11c3f06b983b2df970cbdde9011046 Mon Sep 17 00:00:00 2001 From: shum Date: Thu, 22 Oct 2026 06:39:44 +0000 Subject: [PATCH] store: hold race tests at a gated chain read --- tests/Bots/StoreService/FakeEthNode.hs | 51 +++++++++++++- tests/Bots/StoreService/NamesOwnerTests.hs | 34 +++------ tests/Bots/StoreService/NamesRevealTests.hs | 78 +++++++-------------- 3 files changed, 82 insertions(+), 81 deletions(-) diff --git a/tests/Bots/StoreService/FakeEthNode.hs b/tests/Bots/StoreService/FakeEthNode.hs index cb3af9b213..91d6adff6d 100644 --- a/tests/Bots/StoreService/FakeEthNode.hs +++ b/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 [] "Sign in to provider-api-key" 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] diff --git a/tests/Bots/StoreService/NamesOwnerTests.hs b/tests/Bots/StoreService/NamesOwnerTests.hs index fd0f9a93f8..4b6470da03 100644 --- a/tests/Bots/StoreService/NamesOwnerTests.hs +++ b/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 diff --git a/tests/Bots/StoreService/NamesRevealTests.hs b/tests/Bots/StoreService/NamesRevealTests.hs index 050d5e54f2..ea78752aa8 100644 --- a/tests/Bots/StoreService/NamesRevealTests.hs +++ b/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