diff --git a/scripts/resolver/README.md b/scripts/resolver/README.md index 6ba327c6a..995267837 100644 --- a/scripts/resolver/README.md +++ b/scripts/resolver/README.md @@ -46,7 +46,8 @@ docker compose logs --tail=20 reth ``` This is the long pole (~1 day on mainnet). Until reth is synced the resolver -returns `502`. +returns `502`. A TLD whose controller has a name policy then still answers `502` +until `docker compose restart resolver` (see Configuring addresses). ## Verify @@ -357,7 +358,8 @@ from before the enum stores a boolean, whose `true` decodes as 1, which is why ### Configuring addresses -The resolver reads three contracts, each configured per TLD. +The resolver reads three contracts, each configured per TLD, and the name +policy a controller names. The **registry** answers who owns a node, and `/resolve` reads the records from it. The **registrar** (ERC-721) holds `nameExpires` and `GRACE_PERIOD`, which @@ -372,6 +374,19 @@ held-back name reads as not reserved, and no name can be priced. All three default to the mainnet `.testing` deployment. `.simplex` is unset until it is deployed. +Newer contracts, which `.simplex` will run, keep `reservedNames` in a **name +policy** (`SimplexNamePolicy`), with the same `Reason`, and their controller +answers no `reservedNames` call. The resolver reads each controller's +`namePolicy()` once at startup, before its workers start, and reads reservations +from the policy it names. `.testing`'s controller is from before name policies, +so the read reverts. A read that reverts, answers nothing, names no policy or +fails with an RPC error is logged as `no_name_policy`, and reservations are read +from the controller. For a controller that has a name policy, every lookup under +its TLD then answers 502 until a restart: after an RPC error, or when the node +has not yet reached the controller or its `setNamePolicy`. A node that +does not answer stops the resolver, and the container restarts. After the +controller's owner calls `setNamePolicy`, restart the resolver. + The controller default is the **proxy**, not `SimplexControllerImpl`. Storage lives in the proxy, so the implementation address answers nothing. The two deployment files use different names for that proxy: @@ -418,13 +433,15 @@ Each event is one line: UTC time, level, event name, then `key=value` fields. | Event | Level | Meaning | |---|---|---| -| `listening` | info | started: address, workers, RPC endpoint, registries, trusted proxies | +| `listening` | info | started: address, workers, RPC endpoint, registries, name policies, trusted proxies | | `request` | info | one answered request; `/health` only at debug | | `upstream_error` | warn | a read from the node failed; the caller got 502 | +| `no_name_policy` | warn | at startup, a controller's `namePolicy()` named no policy (see Configuring addresses); its reservations are read from the controller | | `multicall_unavailable` | warn | Multicall3 did not answer; rounds are sent as plain batches | | `client_gone` | warn | the caller hung up before the answer, usually an smp-server past its timeout | | `http_error` | warn | a malformed request; idle keep-alive timeouts only at debug | | `request_failed` | error | a request failed unexpectedly, with its traceback | +| `startup_failed` | error | the node did not answer the startup read of the name policies; the container restarts | | `worker_exited`, `worker_failed` | error | a worker stopped; the others stop too, so the container restarts | | `stopping` | info | stopped by a signal | diff --git a/scripts/resolver/docker-compose.yml b/scripts/resolver/docker-compose.yml index 1e716357b..a0978b6b6 100644 --- a/scripts/resolver/docker-compose.yml +++ b/scripts/resolver/docker-compose.yml @@ -141,6 +141,10 @@ services: # starting before reth has finished snapshot replay just yields a few # 502s until the chain is queryable. The upstream reth image doesn't # ship a HEALTHCHECK, so we can't gate on healthy. + # The startup read of the name policies is the exception: until reth + # answers, the resolver exits and restarts, and a TLD whose name policy + # reth has not yet reached answers 502 until the resolver restarts + # (README, "Configuring addresses"). reth: condition: service_started environment: diff --git a/scripts/resolver/service/snrc-resolve.py b/scripts/resolver/service/snrc-resolve.py index 741e512f9..5ab92a4a4 100755 --- a/scripts/resolver/service/snrc-resolve.py +++ b/scripts/resolver/service/snrc-resolve.py @@ -42,6 +42,7 @@ Environment: SNRC_REGISTRAR_ BaseRegistrar (ERC-721) for the TLD; expiry and status (default: mainnet for .testing, empty for .simplex) SNRC_CONTROLLER_ SimplexController (proxy) for the TLD; reservations, + read from the policy its `namePolicy()` names, if any, and through its `prices()` oracle what registering costs (default: mainnet for .testing, empty for .simplex) SNRC_PORT Listen port (default: 8000) @@ -140,6 +141,10 @@ CONTROLLERS = { "simplex": os.environ.get("SNRC_CONTROLLER_SIMPLEX", ""), # not deployed yet } +# The name policy each TLD's controller names, which holds its reservations in +# place of the controller; set at startup from `name_policies()`. +NAME_POLICIES = {} + # `reservedNames` holds a SimplexController.Reason; 0 means not reserved. A # controller from before the enum stores a bool, whose `true` decodes as 1, # which is why 1 reads as "internal". @@ -520,7 +525,27 @@ def reservation_reason(tld: str, token: int) -> int: controller = CONTROLLERS.get(tld) if not controller: return 0 - return decode_uint(eth_call(*reserved_call(controller, token))) + return decode_uint(eth_call(*reserved_call(NAME_POLICIES.get(tld, controller), token))) + + +def name_policies() -> dict: + """The name policy of each configured controller that names one. A controller + that reverts, answers nothing or names no policy is logged, and its reservations + are read from it; an RPC error is taken as a revert. Raises when the node does + not answer.""" + policies = {} + for tld, controller in CONTROLLERS.items(): + if not controller: + continue + try: + policy = decode_address(eth_call(*name_policy_call(controller))) + if policy == ZERO_ADDR: + raise RuntimeError("namePolicy() is unset") + except RuntimeError as e: + log_event(logging.WARNING, "no_name_policy", tld=tld, controller=controller, message=str(e), reservations="controller") + continue + policies[tld] = policy + return policies def pricing_params(tld: str): @@ -786,8 +811,12 @@ def label_call(registrar: str, token: int): return registrar, selector("labelOf(uint256)") + encode_uint(token) -def reserved_call(controller: str, token: int): - return controller, selector("reservedNames(bytes32)") + encode_uint(token) +def reserved_call(contract: str, token: int): + return contract, selector("reservedNames(bytes32)") + encode_uint(token) + + +def name_policy_call(controller: str): + return controller, selector("namePolicy()") def prices_call(contract: str): @@ -830,7 +859,7 @@ def lookup_reads(name: str): if len(labels) == 2 and is_encoded_labelhash(labels[0]): reads.append(eth_call_read(*label_call(registrar, token))) if CONTROLLERS.get(tld): - reads.append(eth_call_read(*reserved_call(CONTROLLERS[tld], token))) + reads.append(eth_call_read(*reserved_call(NAME_POLICIES.get(tld, CONTROLLERS[tld]), token))) registry = REGISTRIES.get(tld) if registry: node = node_of(name) @@ -1464,6 +1493,9 @@ def supervise(workers: int): stop() stop_on(signum, frame) + # each worker opens its own connections: one shared would mix their replies + while not _rpc_pool.empty(): + _rpc_pool.get_nowait().close() # before forking, so a signal during startup cannot orphan the workers signal.signal(signal.SIGTERM, stop_and_exit) signal.signal(signal.SIGINT, stop_and_exit) @@ -1488,6 +1520,11 @@ def supervise(workers: int): def main(): setup_logging() + try: + NAME_POLICIES.update(name_policies()) + except Exception: + log_event(logging.ERROR, "startup_failed", exc_info=True) + sys.exit(1) log_event( logging.INFO, "listening", @@ -1496,6 +1533,7 @@ def main(): workers=WORKERS, rpc=public_endpoint(RPC), registries=",".join(f"{tld}={addr or '-'}" for tld, addr in REGISTRIES.items()), + name_policies=",".join(f"{tld}={NAME_POLICIES.get(tld, '-')}" for tld in CONTROLLERS), trusted_proxies=",".join(map(str, TRUSTED_PROXIES)) or None, ) if WORKERS > 1: diff --git a/scripts/resolver/service/test_snrc_resolve.py b/scripts/resolver/service/test_snrc_resolve.py index 55ea17e43..2cf6a792c 100644 --- a/scripts/resolver/service/test_snrc_resolve.py +++ b/scripts/resolver/service/test_snrc_resolve.py @@ -1141,6 +1141,7 @@ class FakeChain: REGISTRY = "0x58fc46996d975c57883564648bda5206d1a0102b" REGISTRAR = "0xef47eb4384b46c89e4482a677c2cbcbd2a6fd85a" CONTROLLER = "0x281ca41311c2aa808c917c4674639d7567b75714" + NAME_POLICY = "0x4b7e0c1a9d2f3e5b6c8a7d9e0f1b2c3d4e5f6a7b" ORACLE = "0x1e0c9a2b9d1a4c8f7b3e5d6a9c2f4b8e1d7a3c50" OWNER = "0xd83bd7e0e6b8a4c1f2593a7b0c4e8d1a6f9b2c37" RESOLVER = "0x80fa2b1c3d4e5f60718293a4b5c6d7e8f9012345" @@ -1275,7 +1276,7 @@ class FakeNodeTestCase(unittest.TestCase): def setUp(self): self.node = FakeNode() - self._saved = (snrc.RPC, snrc.RPC_URL, snrc.REGISTRIES, snrc.REGISTRARS, snrc.CONTROLLERS) + self._saved = (snrc.RPC, snrc.RPC_URL, snrc.REGISTRIES, snrc.REGISTRARS, snrc.CONTROLLERS, snrc.NAME_POLICIES) snrc.RPC = self.node.url snrc.RPC_URL = urlparse(snrc.RPC) snrc.REGISTRIES = {"testing": FakeChain.REGISTRY, "simplex": ""} @@ -1286,7 +1287,7 @@ class FakeNodeTestCase(unittest.TestCase): def tearDown(self): self._drain_pool() - snrc.RPC, snrc.RPC_URL, snrc.REGISTRIES, snrc.REGISTRARS, snrc.CONTROLLERS = self._saved + snrc.RPC, snrc.RPC_URL, snrc.REGISTRIES, snrc.REGISTRARS, snrc.CONTROLLERS, snrc.NAME_POLICIES = self._saved self.node.stop() def _drain_pool(self): @@ -1398,6 +1399,12 @@ class BatchedReadsTests(FakeNodeTestCase): def test_v1_answers_the_same(self): self.assert_same_answer(snrc.resolve, "acme.testing", 2) + def test_reservations_held_by_a_name_policy_are_read_in_the_same_rounds(self): + snrc.NAME_POLICIES = {"testing": FakeChain.NAME_POLICY} + self.node.chain.answers[snrc.reserved_call(FakeChain.NAME_POLICY, snrc.label_token("free"))] = "0x" + snrc.encode_uint(1) + status, body = self.assert_same_answer(snrc.registration, "free.testing", 3) + self.assertEqual(body["registration"], {"type": "reserved", "reservedReason": "internal"}) + def test_without_multicall_a_round_is_still_one_batch(self): self.node.multicall = False with self.assertLogs("snrc_resolve", "WARNING") as logs: @@ -1436,6 +1443,37 @@ class BatchedReadsTests(FakeNodeTestCase): self.assertEqual(trips, 0) +class NamePoliciesTests(FakeNodeTestCase): + def test_a_controller_naming_a_policy_is_mapped_to_it(self): + self.node.chain.answers[snrc.name_policy_call(FakeChain.CONTROLLER)] = "0x" + snrc.encode_uint(int(FakeChain.NAME_POLICY, 16)) + self.assertEqual(snrc.name_policies(), {"testing": FakeChain.NAME_POLICY}) + + def test_a_controller_from_before_name_policies_keeps_the_reservations_and_is_logged(self): + with self.assertLogs("snrc_resolve", "WARNING") as logs: + self.assertEqual(snrc.name_policies(), {}) + [record] = logs.records + self.assertEqual((record.getMessage(), record.fields["tld"], record.fields["reservations"]), ("no_name_policy", "testing", "controller")) + + def test_an_unconfigured_controller_is_not_read(self): + snrc.CONTROLLERS = {"simplex": ""} + self.assertEqual(self.requests_made(snrc.name_policies), ({}, 0)) + + def test_an_address_without_a_contract_is_read_as_no_policy(self): + self.node.chain.answers[snrc.name_policy_call(FakeChain.CONTROLLER)] = "0x" + with self.assertLogs("snrc_resolve", "WARNING"): + self.assertEqual(snrc.name_policies(), {}) + + def test_a_controller_naming_no_policy_is_left_out_and_logged(self): + self.node.chain.answers[snrc.name_policy_call(FakeChain.CONTROLLER)] = "0x" + snrc.encode_uint(0) + with self.assertLogs("snrc_resolve", "WARNING"): + self.assertEqual(snrc.name_policies(), {}) + + def test_a_node_that_does_not_answer_stops_the_read(self): + self.node.status = 502 + with self.assertRaises(HTTPError): + snrc.name_policies() + + class Aggregate3Tests(unittest.TestCase): def test_calls_are_encoded_as_multicall3_reads_them(self): calls = [(FakeChain.REGISTRY, "0x0178b8bf" + "11" * 32), (FakeChain.RESOLVER, "0x59d1d43c" + "22" * 100)] @@ -1705,6 +1743,17 @@ class KeepAliveTests(unittest.TestCase): self.assertEqual(sock.recv(1), b"") +class StartupTests(unittest.TestCase): + def test_a_node_that_does_not_answer_stops_the_service(self): + node = FakeNode() + self.addCleanup(node.stop) + node.status = 502 + env = dict(os.environ, SNRC_RPC=node.url, SNRC_WORKERS="1") + service = subprocess.run([sys.executable, os.path.join(_HERE, "snrc-resolve.py")], env=env, stderr=subprocess.PIPE, text=True, timeout=10) + self.assertEqual(service.returncode, 1) + self.assertRegex(service.stderr, r"ERROR startup_failed") + + @unittest.skipUnless(sys.platform.startswith("linux"), "reads worker processes from /proc") class WorkerProcessesTests(unittest.TestCase): """Workers share the port, and the service stops as a whole, so the @@ -1715,6 +1764,8 @@ class WorkerProcessesTests(unittest.TestCase): with socket.socket() as s: s.bind(("127.0.0.1", 0)) self.port = s.getsockname()[1] + self.node.chain.answers[snrc.name_policy_call(FakeChain.CONTROLLER)] = "0x" + snrc.encode_uint(int(FakeChain.NAME_POLICY, 16)) + self.node.chain.answers[snrc.reserved_call(FakeChain.NAME_POLICY, snrc.label_token("acme"))] = "0x" + snrc.encode_uint(2) env = dict(os.environ, SNRC_RPC=self.node.url, SNRC_BIND="127.0.0.1", SNRC_PORT=str(self.port), SNRC_WORKERS="2", SNRC_REGISTRY_TESTING=FakeChain.REGISTRY, SNRC_REGISTRAR_TESTING=FakeChain.REGISTRAR, SNRC_CONTROLLER_TESTING=FakeChain.CONTROLLER) self.service = subprocess.Popen([sys.executable, os.path.join(_HERE, "snrc-resolve.py")], env=env, stderr=subprocess.PIPE, text=True) @@ -1767,6 +1818,11 @@ class WorkerProcessesTests(unittest.TestCase): with urlopen(f"http://127.0.0.1:{self.port}/v2/resolve/acme.testing", timeout=5) as res: self.assertEqual(json.loads(res.read())["registration"]["type"], "registered") + def test_workers_read_reservations_from_the_policy_on_their_own_connections(self): + with urlopen(f"http://127.0.0.1:{self.port}/v2/resolve/acme.testing", timeout=5) as res: + self.assertEqual(json.loads(res.read())["registration"]["reservedReason_"], "trademark") + self.assertEqual(self.node.connections, 2) + def test_stopping_the_service_stops_every_worker(self): self.service.send_signal(signal.SIGTERM) self.assertEqual(self.service.wait(timeout=5), 0) diff --git a/simplexmq.cabal b/simplexmq.cabal index ecf9a0c7c..ba1456992 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -1,7 +1,7 @@ cabal-version: 3.0 name: simplexmq -version: 7.1.0.14 +version: 7.1.0.16 synopsis: SimpleXMQ message broker description: This package includes <./docs/Simplex-Messaging-Server.html server>, <./docs/Simplex-Messaging-Client.html client> and diff --git a/src/Simplex/Messaging/Names/Record.hs b/src/Simplex/Messaging/Names/Record.hs index 461bec16e..ed77eca20 100644 --- a/src/Simplex/Messaging/Names/Record.hs +++ b/src/Simplex/Messaging/Names/Record.hs @@ -1,4 +1,6 @@ +{-# LANGUAGE DataKinds #-} {-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE DerivingVia #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} @@ -11,6 +13,9 @@ module Simplex.Messaging.Names.Record NameResponse (..), NameRegistration (..), NamePricing (..), + NameTerms (..), + TermsDoc (..), + ZstdTerms (..), USDCents (..), NameReservedReason (..), ) @@ -19,6 +24,7 @@ where import Data.Aeson (FromJSON (..), ToJSON (..)) import qualified Data.Aeson as J import qualified Data.Aeson.TH as JQ +import Data.ByteString (ByteString) import Data.Int (Int64) import Data.Map.Strict (Map) import Data.Text (Text) @@ -85,7 +91,11 @@ data NameRegistration nameRecord :: NameRecord } | -- | Held by nobody, and registrable now. - NRAvailable {pricing :: NamePricing} + NRAvailable + { pricing :: NamePricing, + -- | absent only from a router before v24 + nameTerms :: Maybe NameTerms + } | -- | Held back by the registry, and not for sale at its price. NRReserved {reservedReason :: NameReservedReason} deriving (Eq, Show) @@ -101,6 +111,27 @@ data NamePricing = NamePricing } deriving (Eq, Show) +-- | The terms a buyer of the name accepts, which the name policy keeps. +data NameTerms + = -- | the general terms + NTGeneral {termsDoc :: TermsDoc} + | -- | a community name's community terms, which include the general ones + NTCommunity {termsDoc :: TermsDoc} + deriving (Eq, Show) + +-- | Terms as the name policy keeps them: a link to one file at one commit, whose keccak256 a buyer accepts. +data TermsDoc = TermsDoc + { -- | ends in #sha256= of the file + termsLink :: Text, + -- | the file in one zstd frame, when it fits the answer + termsZstd :: Maybe ZstdTerms + } + deriving (Eq, Show) + +newtype ZstdTerms = ZstdTerms ByteString + deriving newtype (Eq, Show, StrEncoding) + deriving (ToJSON, FromJSON) via (StrJSON "ZstdTerms" ZstdTerms) + -- | Why the registry holds a name back. data NameReservedReason = -- | held for SimpleX @@ -137,6 +168,10 @@ instance FromJSON NameReservedReason where $(JQ.deriveJSON defaultJSON ''NamePricing) +$(JQ.deriveJSON defaultJSON ''TermsDoc) + +$(JQ.deriveJSON (taggedObjectJSON $ dropPrefix "NT") ''NameTerms) + -- taggedObjectJSON, not sumTypeJSON: this JSON is the RNAME payload and the -- resolver contract, so it must not vary with the swift build flag. $(JQ.deriveJSON (taggedObjectJSON $ dropPrefix "NR") ''NameRegistration) diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 30133885d..27364a950 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -2135,7 +2135,7 @@ client ts <- getSystemSeconds atomically $ whenM (sameClient rc rcv) $ readTVar delivered >>= \case - Just _ -> pure () -- if a message was already delivered, should not deliver more + Just _ -> writeTVar st NoSub -- if a message was already delivered, should not deliver more Nothing -> do -- a separate thread is needed because it blocks when client sndQ is full. deliver sndQ' s ts diff --git a/tests/RSLVTests.hs b/tests/RSLVTests.hs index 0d096bccb..e9b931314 100644 --- a/tests/RSLVTests.hs +++ b/tests/RSLVTests.hs @@ -191,7 +191,7 @@ testRslvAvailable = testSMPClient @TLS $ \h -> do (corrId, _entId, resp) <- sendRslv h "na01" (domain "ghost.simplex") corrId `shouldBe` CorrId "na01" - resp `shouldBe` Right (RNAME (resolved (NRAvailable testPricing))) + resp `shouldBe` Right (RNAME (resolved (NRAvailable testPricing Nothing))) testRslvReserved :: HasTestEnv => IO () testRslvReserved = diff --git a/tests/SMPNamesTests.hs b/tests/SMPNamesTests.hs index 7f28a0b34..452c99428 100644 --- a/tests/SMPNamesTests.hs +++ b/tests/SMPNamesTests.hs @@ -163,7 +163,7 @@ availabilitySpec = do answers heldBackBody . resolved $ NRRegistered {expires = Just (RoundedSystemTime 1813853483), graceUntil = Just (RoundedSystemTime 1821629483), reservedReason_ = Just NRRInternal, nameRecord = testNameRecord} it "an unregistered name answers with the price" $ - answers availableBody (resolved NRAvailable {pricing = testPricing}) + answers availableBody (resolved NRAvailable {pricing = testPricing, nameTerms = Nothing}) it "reserved carries the reason and no price" $ answers reservedBody (resolved $ NRReserved NRRTrademark) -- losing the reservation would offer a name that cannot be registered @@ -175,7 +175,7 @@ availabilitySpec = do (\a -> J.eitherDecodeStrict (LB.toStrict (J.encode a)) `shouldBe` Right a) [ registeredAlice, NRRegistered {expires = Nothing, graceUntil = Nothing, reservedReason_ = Just NRRInternal, nameRecord = testNameRecord}, - NRAvailable {pricing = testPricing}, + NRAvailable {pricing = testPricing, nameTerms = Nothing}, NRReserved NRRInternal, NRReserved NRRTrademark, NRReserved NRRCommunity, diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index 146a1d46f..18b9a972b 100644 --- a/tests/ServerTests.hs +++ b/tests/ServerTests.hs @@ -31,9 +31,9 @@ import qualified Data.ByteString.Char8 as B import Data.Foldable (foldrM) import Data.Hashable (hash) import qualified Data.IntSet as IS -import Data.List.NonEmpty (NonEmpty) +import Data.List.NonEmpty (NonEmpty (..)) import Data.List (isPrefixOf) -import Data.Maybe (catMaybes) +import Data.Maybe (catMaybes, isJust) import Data.String (IsString (..)) import Data.Time.Clock (NominalDiffTime, diffUTCTime, getCurrentTime) import Text.Read (readMaybe) @@ -49,6 +49,7 @@ import Simplex.Messaging.Protocol import Simplex.Messaging.Server.Env.STM (AStoreType (..), ServerConfig (..), ServerStoreCfg (..)) import Simplex.Messaging.Server.Expiration import Simplex.Messaging.Server.MsgStore.Types (SMSType (..), SQSType (..)) +import Simplex.Messaging.Server.QueueStore.QueueInfo (QSub (..), QSubThread (..), QueueInfo (..)) import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (..)) import Simplex.Messaging.Server.StoreLog (StoreLogRecord (..)) import Simplex.Messaging.Transport @@ -87,6 +88,7 @@ serverTests = do describe "GET & SUB commands" testGetSubCommands describe "Exceeding queue quota" testExceedQueueQuota describe "Concurrent sending and delivery" testConcurrentSendDelivery + describe "SUB while delivery waits" testSubPendingDelivery describe "Service message subscriptions" $ do testServiceDeliverSubscribe testServiceUpgradeAndDowngrade @@ -681,6 +683,67 @@ testConcurrentSendDelivery = Resp "4" _ OK <- signSendRecv rh rKey ("4", rId, ACK mId2) pure () +testSubPendingDelivery :: SpecWith (ASrvTransport, AStoreType) +testSubPendingDelivery = + it "should deliver message after SUB while delivery waits for full send queue" $ \(ATransport t, msType) -> + smpTest2 t msType $ \rh@THandle {params} sh -> do + g <- C.newRandom + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let newQueue h = do + Resp "" NoEntity (Ids rId sId srvDh) <- signSendRecv h rKey ("", NoEntity, New rPub dhPub) + pure (rId, sId, decryptMsgV3 $ C.dh' srvDh dhPriv) + (rId, sId, dec) <- newQueue rh + fillers <- replicateM 4 $ (\(fId, fsId, _) -> (fId, fsId)) <$> newQueue rh + (_, probeSId, _) <- newQueue sh + C.CbNonce corrId <- atomically $ C.randomCbNonce g + lnkId <- EntityId <$> atomically (C.randomBytes 24 g) + let linkData = (EncDataBytes "", EncDataBytes $ B.replicate 13000 '-') + qrd = QRContact $ Just (lnkId, (EntityId $ B.take 24 $ C.sha3_384 corrId, linkData)) + Resp _ NoEntity (IDS _) <- signSendRecv sh rKey (corrId, NoEntity, NEW (NewQueueReq rPub dhPub Nothing SMOnlyCreate (Just qrd) Nothing)) + let TransmissionForAuth {tToSend = lget} = encodeTransmissionForAuth params (CorrId "", lnkId, LGET) + TransmissionForAuth {tToSend = probe} = encodeTransmissionForAuth params (CorrId "p", probeSId, _SEND "probe") + void $ tPut rh $ Right (Nothing, lget) :| replicate 1019 (Right (Nothing, lget)) + Right () <- tPut1 rh (Nothing, probe) + Just (Resp "" _ (Msg _ _)) <- 5000000 `timeout` tGet1 sh + forM_ fillers $ \(_, fsId) -> do + Resp "f" _ OK <- sendRecv sh ("", "f", fsId, _SEND "filler") + pure () + Resp "1" _ OK <- sendRecv sh ("", "1", sId, _SEND "hello") + signSend_ rh rKey Nothing ("2", rId, SUB) + threadDelay 500000 + let drain st@(n, _, mId_) = case mId_ of + Just mId | n == length fillers -> pure mId + _ -> 5000000 `timeout` tGetClient rh >>= maybe (error "drain timeout") (foldM received st) >>= drain + received st@(n, subscribed, mId_) = \case + Resp "" _ (LNK _ _) -> pure st + Resp "p" _ OK -> pure st + Resp "2" _ (SOK Nothing) -> pure (n, True, mId_) + Resp "" rId' (Msg mId msg) + | rId' == rId -> do + dec mId msg `shouldBe` Right "hello" + pure (n, subscribed, if subscribed then Just mId else mId_) + | rId' `elem` map fst fillers -> pure (n + 1, subscribed, mId_) + r -> error $ "unexpected event: " <> show r + mId <- drain (0 :: Int, False, Nothing) + Resp "3" _ OK <- signSendRecv rh rKey ("3", rId, ACK mId) + Resp "4" _ OK <- sendRecv sh ("", "4", sId, _SEND "hello 2") + signSend_ rh rKey Nothing ("5", rId, QUE) + void $ + receiveInAnyOrder + rh + [ \case + Resp "" _ (Msg mId2 msg2) -> do + dec mId2 msg2 `shouldBe` Right "hello 2" + pure $ Just () + _ -> pure Nothing, + \case + Resp "5" _ (INFO QueueInfo {qiSub, qiSize}) -> do + (qSubThread <$> qiSub, isJust $ qDelivered =<< qiSub, qiSize) `shouldBe` (Just QNoSub, True, 1) + pure $ Just () + _ -> pure Nothing + ] + testServiceDeliverSubscribe :: SpecWith (ASrvTransport, AStoreType) testServiceDeliverSubscribe = it "should create queue as service and subscribe with SUBS after reconnect" $ \(at@(ATransport t), msType) -> do diff --git a/tests/Util.hs b/tests/Util.hs index dbd824eba..5f12e4dd8 100644 --- a/tests/Util.hs +++ b/tests/Util.hs @@ -119,7 +119,7 @@ eventuallyRemove path retries = case retries of where action = removeDirectoryRecursive path -data TestWrapper a = TestWrapper (HasTestEnv => a) +data TestWrapper a = TestWrapper String (HasTestEnv => a) -- TODO [ntfdb] running wiht LogWarn level shows potential issue "Queue count differs" testLogLevel :: LogLevel @@ -127,7 +127,7 @@ testLogLevel = LogError instance Example a => Example (TestWrapper a) where type Arg (TestWrapper a) = Arg a - evaluateExample (TestWrapper action) params hooks state = do + evaluateExample (TestWrapper label action) params hooks state = do ci <- envCI runTest `E.catches` [E.Handler (onTestFailure ci), E.Handler (onTestException ci)] where @@ -140,13 +140,13 @@ instance Example a => Example (TestWrapper a) where onTestFailure :: Bool -> ResultStatus -> IO Result onTestFailure ci = \case Failure loc_ reason | ci -> do - putStrLn $ "Test failed: location " ++ show loc_ ++ ", reason: " ++ show reason + putStrLn $ "Test failed: " ++ show label ++ ", location " ++ show loc_ ++ ", reason: " ++ show reason retryTest r -> E.throwIO r onTestException :: Bool -> SomeException -> IO Result onTestException False e = E.throwIO e onTestException True e = do - putStrLn $ "Test exception: " ++ show e + putStrLn $ "Test exception: " ++ show label ++ ", " ++ show e retryTest retryTest = do putStrLn "Retrying with more logs..." @@ -157,10 +157,10 @@ envCI :: IO Bool envCI = (Just "true" ==) <$> lookupEnv "CI" it :: (HasCallStack, Example a) => String -> (HasTestEnv => a) -> SpecWith (Arg a) -it label action = Hspec.it label (TestWrapper action) +it label action = Hspec.it label (TestWrapper label action) fit :: (HasCallStack, Example a) => String -> (HasTestEnv => a) -> SpecWith (Arg a) fit label action = focus $ it label action xit :: (HasCallStack, Example a) => String -> (HasTestEnv => a) -> SpecWith (Arg a) -xit label action = Hspec.xit label (TestWrapper action) +xit label action = Hspec.xit label (TestWrapper label action)