Merge branch 'master' into names

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 |
 

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:

scripts/resolver/service/snrc-resolve.py

@@ -42,6 +42,7 @@ Environment:
   SNRC_REGISTRAR_<TLD>   BaseRegistrar (ERC-721) for the TLD; expiry and status
                          (default: mainnet for .testing, empty for .simplex)
   SNRC_CONTROLLER_<TLD>  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:

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)

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

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

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

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 =

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,

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

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)