Async topic validator crashes gossipsub service

[hayotensor]

Summary

I just recently received this issue that shuts down the pubsub service when using an async topic validator:

When using a sync topic validator with the same logic, this issue doesn't occur

Expected behavior

MessageCache to successfully pop the topic history

Actual behavior

The key is attempting to be popped but it apparently doesn't exist

Relevant log output

2026-01-03 16:16:27,482 [ERROR] [async_service.Manager] <Manager[GossipSub] flags=SRcfe>: task heartbeat[daemon=True] exited with error: (b'\x00\x00\x00\x00iY\x86\xff', b'\x00$\x08\x01\x12 /X\x99\xbbp\xa2\xaa9\x93\xc2\xd6f\xa2\x99\x1e[vH"rac\x8e\x1e!\x8b\xff\x91\x0c)\xe7x')
Traceback (most recent call last):
  File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/base.py", line 334, in _run_and_manage_task
    await task.run()
  File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/trio_service.py", line 118, in run
    await self._async_fn(*self._async_fn_args)
  File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/pubsub/gossipsub.py", line 565, in heartbeat
    self.mcache.shift()
  File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/pubsub/mcache.py", line 102, in shift
    self.msgs.pop(entry.mid)
KeyError: (b'\x00\x00\x00\x00iY\x86\xff', b'\x00$\x08\x01\x12 /X\x99\xbbp\xa2\xaa9\x93\xc2\xd6f\xa2\x99\x1e[vH"rac\x8e\x1e!\x8b\xff\x91\x0c)\xe7x')
2026-01-03 16:16:27,496 [ERROR] [__main__] Fatal error: Exceptions from Trio nursery (1 sub-exception)
  + Exception Group Traceback (most recent call last):
  |   File "/home/bob/py-libp2p-subnet-gossip/subnet/cli/run_node.py", line 408, in main
  |     trio.run(server.run)
  |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/trio/_core/_run.py", line 2549, in run
  |     raise runner.main_task_outcome.error
  |   File "/home/bob/py-libp2p-subnet-gossip/subnet/server/server.py", line 134, in run
  |     async with host.run(listen_addrs=listen_addrs), trio.open_nursery() as nursery:
  |   File "/usr/lib/python3.10/contextlib.py", line 217, in __aexit__
  |     await self.gen.athrow(typ, value, traceback)
  |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/host/basic_host.py", line 357, in _run
  |     async with background_trio_service(network):
  |   File "/usr/lib/python3.10/contextlib.py", line 217, in __aexit__
  |     await self.gen.athrow(typ, value, traceback)
  |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/trio_service.py", line 456, in background_trio_service
  |     async with trio.open_nursery() as nursery:
  |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/trio/_core/_run.py", line 1125, in __aexit__
  |     raise combined_error_from_nursery
  | exceptiongroup.ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
  +-+---------------- 1 ----------------
    | Exception Group Traceback (most recent call last):
    |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/trio_service.py", line 461, in background_trio_service
    |     yield manager
    |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/host/basic_host.py", line 374, in _run
    |     yield
    |   File "/home/bob/py-libp2p-subnet-gossip/subnet/server/server.py", line 134, in run
    |     async with host.run(listen_addrs=listen_addrs), trio.open_nursery() as nursery:
    |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/trio/_core/_run.py", line 1125, in __aexit__
    |     raise combined_error_from_nursery
    | exceptiongroup.ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
    +-+---------------- 1 ----------------
      | Exception Group Traceback (most recent call last):
      |   File "/home/bob/py-libp2p-subnet-gossip/subnet/server/server.py", line 161, in run
      |     async with background_trio_service(dht):
      |   File "/usr/lib/python3.10/contextlib.py", line 217, in __aexit__
      |     await self.gen.athrow(typ, value, traceback)
      |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/trio_service.py", line 456, in background_trio_service
      |     async with trio.open_nursery() as nursery:
      |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/trio/_core/_run.py", line 1125, in __aexit__
      |     raise combined_error_from_nursery
      | exceptiongroup.ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
      +-+---------------- 1 ----------------
        | Exception Group Traceback (most recent call last):
        |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/trio_service.py", line 461, in background_trio_service
        |     yield manager
        |   File "/home/bob/py-libp2p-subnet-gossip/subnet/server/server.py", line 176, in run
        |     async with background_trio_service(pubsub):
        |   File "/usr/lib/python3.10/contextlib.py", line 217, in __aexit__
        |     await self.gen.athrow(typ, value, traceback)
        |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/trio_service.py", line 456, in background_trio_service
        |     async with trio.open_nursery() as nursery:
        |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/trio/_core/_run.py", line 1125, in __aexit__
        |     raise combined_error_from_nursery
        | exceptiongroup.ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
        +-+---------------- 1 ----------------
          | Exception Group Traceback (most recent call last):
          |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/trio_service.py", line 461, in background_trio_service
          |     yield manager
          |   File "/home/bob/py-libp2p-subnet-gossip/subnet/server/server.py", line 177, in run
          |     async with background_trio_service(gossipsub):
          |   File "/usr/lib/python3.10/contextlib.py", line 217, in __aexit__
          |     await self.gen.athrow(typ, value, traceback)
          |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/trio_service.py", line 456, in background_trio_service
          |     async with trio.open_nursery() as nursery:
          |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/trio/_core/_run.py", line 1125, in __aexit__
          |     raise combined_error_from_nursery
          | exceptiongroup.ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
          +-+---------------- 1 ----------------
            | Exception Group Traceback (most recent call last):
            |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/trio_service.py", line 248, in run
            |     raise ExceptionGroup(
            | exceptiongroup.ExceptionGroup: Encountered multiple Exceptions:  (1 sub-exception)
            +-+---------------- 1 ----------------
              | Traceback (most recent call last):
              |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/base.py", line 334, in _run_and_manage_task
              |     await task.run()
              |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/tools/async_service/trio_service.py", line 118, in run
              |     await self._async_fn(*self._async_fn_args)
              |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/pubsub/gossipsub.py", line 565, in heartbeat
              |     self.mcache.shift()
              |   File "/home/bob/py-libp2p-subnet-gossip/.venv/lib/python3.10/site-packages/libp2p/pubsub/mcache.py", line 102, in shift
              |     self.msgs.pop(entry.mid)
              | KeyError: (b'\x00\x00\x00\x00iY\x86\xff', b'\x00$\x08\x01\x12 /X\x99\xbbp\xa2\xaa9\x93\xc2\xd6f\xa2\x99\x1e[vH"rac\x8e\x1e!\x8b\xff\x91\x0c)\xe7x')

Possible Solution

Unsure

Environment

Python version:
3.10.12 (main, Nov  4 2025, 08:48:33) [GCC 11.4.0]

Operating System: Linux-6.6.87.2-microsoft-standard-WSL2-x86_64-with-glibc2.35

python -m pip freeze result:
aioquic==1.3.0
alabaster==1.0.0
annotated-types==0.7.0
anyio==1.4.0
asttokens==3.0.1
async-exit-stack==1.0.1
async-generator==1.10
attrs==25.4.0
babel==2.17.0
backports.tarfile==1.2.0
base58==2.1.1
blake3==1.0.8
bracex==2.6
build==1.3.0
bump-my-version==1.2.6
cachetools==6.2.4
certifi==2025.11.12
cffi==2.0.0
cfgv==3.5.0
chardet==5.2.0
charset-normalizer==3.4.4
click==8.3.1
coincurve==21.0.0
colorama==0.4.6
cryptography==46.0.3
cytoolz==1.1.0
decorator==5.2.1
distlib==0.4.0
dnspython==2.8.0
docutils==0.21.2
ecdsa==0.19.1
eth-hash==0.7.1
eth-keys==0.7.0
eth-typing==5.2.1
eth-utils==5.3.1
exceptiongroup==1.3.1
execnet==2.1.2
executing==2.2.1
factory-boy==2.12.0
Faker==40.1.0
fastecdsa==2.3.2
filelock==3.20.2
grpcio==1.76.0
h11==0.16.0
httpcore==1.0.9
httpx==0.28.1
id==1.5.0
identify==2.6.15
idna==3.11
ifaddr==0.2.0
imagesize==1.4.1
importlib_metadata==8.7.1
iniconfig==2.3.0
ipython==8.37.0
jaraco.classes==3.4.0
jaraco.context==6.0.2
jaraco.functools==4.4.0
jedi==0.19.2
jeepney==0.9.0
Jinja2==3.1.6
keyring==25.7.0
libp2p @ git+https://github.com/libp2p/py-libp2p.git@c63724ff360bb8ff02d468249c32a93c587a8f9e
librt==0.7.7
lru-dict==1.4.1
markdown-it-py==4.0.0
MarkupSafe==3.0.3
matplotlib-inline==0.2.1
mdurl==0.1.2
miniupnpc==2.3.3
mmh3==5.2.0
more-itertools==10.8.0
morphys==1.0
multiaddr==0.0.11
mypy==1.19.1
mypy-protobuf==4.0.0
mypy_extensions==1.1.0
netaddr==1.3.0
nh3==0.3.2
nodeenv==1.10.0
noiseprotocol==0.3.1
outcome==1.3.0.post0
p2pclient==0.2.0
packaging==25.0
parso==0.8.5
pathspec==0.12.1
pexpect==4.9.0
platformdirs==4.5.1
pluggy==1.6.0
pre_commit==4.5.1
prompt_toolkit==3.0.52
protobuf==6.33.2
psutil==7.2.1
ptyprocess==0.7.0
pure_eval==0.2.3
py-bip39-bindings==0.3.0
py-cid==0.4.0
py-ed25519-zebra-bindings==1.3.0
py-libp2p-subnet @ file:///home/bob/py-libp2p-subnet-gossip
py-multibase==2.0.0
py-multicodec==1.0.0
py-multihash==3.0.0
py-sr25519-bindings==0.2.3
pyasn1==0.6.1
pyasn1_modules==0.4.2
pycparser==2.23
pycryptodome==3.23.0
pydantic==2.12.5
pydantic-settings==2.12.0
pydantic_core==2.41.5
Pygments==2.19.2
pylsqpack==0.3.23
pymultihash==0.8.2
PyNaCl==1.6.2
pyOpenSSL==25.3.0
pyproject-api==1.10.0
pyproject_hooks==1.2.0
pyrefly==0.17.1
pytest==9.0.2
pytest-rerunfailures==16.1
pytest-timeout==2.4.0
pytest-trio==0.8.0
pytest-xdist==3.8.0
python-baseconv==1.2.2
python-dotenv==1.2.1
PyYAML==6.0.3
questionary==2.1.1
readme_renderer==44.0
requests==2.32.5
requests-toolbelt==1.0.0
rfc3986==2.0.0
rich==14.2.0
rich-click==1.9.5
rocksdict==0.3.29
rpcudp==5.0.1
ruff==0.14.10
scalecodec===v2.0.0a2-ht
SecretStorage==3.5.0
service-identity==24.2.0
six==1.17.0
sniffio==1.3.1
snowballstemmer==3.0.1
sortedcontainers==2.4.0
Sphinx==8.1.3
sphinx-rtd-theme==3.0.2
sphinxcontrib-applehelp==2.0.0
sphinxcontrib-devhelp==2.0.0
sphinxcontrib-htmlhelp==2.1.0
sphinxcontrib-jquery==4.1
sphinxcontrib-jsmath==1.0.1
sphinxcontrib-qthelp==2.0.0
sphinxcontrib-serializinghtml==2.0.0
stack-data==0.6.3
substrate-interface===v1.7.11-ht
tenacity==9.1.2
tomli==2.3.0
tomlkit==0.13.3
toolz==1.1.0
towncrier==24.8.0
tox==4.33.0
traitlets==5.14.3
trio==0.32.0
trio-typing==0.10.0
trio-websocket==0.12.2
twine==6.2.0
types-protobuf==6.32.1.20251210
typing-inspection==0.4.2
typing_extensions==4.15.0
tzdata==2025.3
u-msgpack-python==2.8.0
urllib3==2.6.2
varint==1.0.2
virtualenv==20.35.4
wcmatch==10.1
wcwidth==0.2.14
websocket-client==1.9.0
wsproto==1.3.2
xxhash==3.6.0
zeroconf==0.147.3
zipp==3.23.0

Would you like to work on fixing this bug ?

Maybe

[IronJam11]

I can work on this issue if there is no one else working on this right now.

[hayotensor]

Yes please!

I never got to looking into this or made an easy way to replicate it. Let me know if you need anything

[Winter-Soren]

Hi @hayotensor,

Thank you for reporting this issue! I've analyzed and identified the root cause of the KeyError that occurs when using async topic validators.

The Problem

The issue is a race condition between the GossipSub heartbeat process and message cache operations when async validators are used. Here's what happens:

1. Message Processing Flow: When a message arrives, it goes through validation in pubsub.py:validate_msg(). For async validators, this involves spawning concurrent tasks with timeout handling.

2. Cache Operations: After successful validation, gossipsub.py:publish() calls mcache.put() which adds the message to both:

   • self.msgs dict (for fast lookup)
   • self.history list (for time-based expiration)

3. Heartbeat Process: The GossipSub heartbeat runs independently every heartbeat_interval and calls mcache.shift() to expire old messages.

4. The Race Condition: mcache.shift() assumes that every message in the history exists in the msgs dict, but in rare edge cases (particularly with async validators and their timeout handling), this assumption can be violated.

Hi @IronJam11,
If ya working on this issue, feel free to ask any questions about Gossipsub Module.

[hayotensor]

Great job @Winter-Soren!

Thank you for your work on this

[IronJam11]

Hey, currently reading on gossipsub and pubsub, will relay my progress here with my updates.

[IronJam11]

Do you have any ideas on how i can reproduce this issue?

[IronJam11]

Thank you for the previous insights given by @Winter-Soren, in the previous 2 days I read about gossipsub and pubsub, and i thought of the following fix to this problem. I completely agree that the given problem is due to the race condition. However, I might be wrong somewhere.

he race (with current code):

• Task A and Task B both get the same message
• Both pass the seen-message check before either marks it seen
• Both validate concurrently (async validators can take 0-5 seconds)
• Task A calls mcache.put(msg) → adds to msgs dict and history list
• Task B calls mcache.put(msg) → overwrites msgs dict (count=1), appends to history (count=2)
• Result: 1 msgs entry, 2 history entries

Current mcache.put() has no duplicate check:

def put(self, msg):
    mid = (msg.seqno, msg.from_id)
    self.msgs[mid] = msg              # dict: replaces if exists
    self.history[0].append(entry)     # list: always appends - creates duplicates!

Fix

Add duplicate check to libp2p/pubsub/mcache.py:

def put(self, msg):
    mid = (msg.seqno, msg.from_id)
    topics = tuple(msg.topicIDs)
    
    if mid in self.msgs:  # NEW - check before adding
        return            # NEW - prevents duplicate history entries
    
    self.msgs[mid] = msg
    self.history[0].append(CacheEntry(mid, topics))

def shift(self):
    if len(self.history) > 0:
        last_entries = self.history[len(self.history) - 1]
        for entry in last_entries:
            self.msgs.pop(entry.mid, None)  # CHANGED - defensive, was .pop(entry.mid)
        
        self.history.pop()
        self.history.insert(0, [])

With this fix, Task B can't add a duplicate:

Before fix (current bug):

• Task A: put() → adds to msgs dict + history list
• Task B: put() → overwrites msgs dict, appends to history (duplicate!)
• Result: 1 msgs entry, 2 history entries (wrong)

After fix:

• Task A: put() → check passes, adds to msgs dict + history list
• Task B: put() → check fails (mid already in msgs), returns early
• Result: 1 msgs entry, 1 history entry ✓

Task A runs put() atomically (no await = no interruption). Task B runs after Task A finishes, sees the message already cached, and exits before appending.

py-libp2p uses Trio (cooperative async), which only switches tasks at await points. Since put() is synchronous, it's atomic.

No breaking changes, backward compatible with existing tests.

[IronJam11]

With the help of @mystical-prog, I could see that we put the message "yet to be validated" after the checks (if the message forwarder and message originator are blacklisted) and just before the validation tests.
It should avoid the race condtion in my opinion.

Changes:

        """
        Push a pubsub message to others.

        :param msg_forwarder: the peer who forward us the message.
        :param msg: the message we are going to push out.
        """
        logger.debug("attempting to publish message %s", msg)

        # Check if the message forwarder (source) is in the blacklist. If yes, reject.
        if self.is_peer_blacklisted(msg_forwarder):
            logger.debug(
                "Rejecting message from blacklisted source peer %s", msg_forwarder
            )
            return

        # Check if the message originator (from) is in the blacklist. If yes, reject.
        msg_from_peer = ID(msg.from_id)
        if self.is_peer_blacklisted(msg_from_peer):
            logger.debug(
                "Rejecting message from blacklisted originator peer %s", msg_from_peer
            )
            return

        # If the message is processed before, return(i.e., don't further process the message)  # noqa: E501
        if self._is_msg_seen(msg):
            return
        

>      self._mark_msg_seen(msg) # PASS IT TO THE SEEN CACHE BEFOREHAND 

        try:
            scorer = getattr(self.router, "scorer", None)
            if scorer is not None:
                if not scorer.allow_publish(msg_forwarder, list(msg.topicIDs)):
                    logger.debug(
                        "Rejecting message from %s by publish score gate", msg_forwarder
                    )
                    return
        except Exception:
            # Router may not support scoring; ignore gracefully
            pass

        # Check if signing is required and if so validate the signature
        if self.strict_signing:
            # Validate the signature of the message
            if not signature_validator(msg):
                logger.debug("Signature validation failed for msg: %s", msg)
                return

        # Validate the message with registered topic validators.
        # If the validation failed, return(i.e., don't further process the message).
        try:
            await self.validate_msg(msg_forwarder, msg)
        except ValidationError:
            # Scoring: count invalid messages
            try:
                scorer = getattr(self.router, "scorer", None)
                if scorer is not None:
                    for topic in msg.topicIDs:
                        scorer.on_invalid_message(msg_forwarder, topic)
            except Exception:
                pass
            logger.debug(
                "Topic validation failed: sender %s sent data %s under topic IDs: %s %s:%s",  # noqa: E501
                msg_forwarder,
                msg.data.hex(),
                msg.topicIDs,
                base58.b58encode(msg.from_id).decode(),
                msg.seqno.hex(),
            )
            return


        # Scoring: first delivery for this sender per topic
        try:
            scorer = getattr(self.router, "scorer", None)
            if scorer is not None:
                for topic in msg.topicIDs:
                    scorer.on_first_delivery(msg_forwarder, topic)
        except Exception:
            pass

        # reject messages claiming to be from ourselves but not locally published
        self_id = self.host.get_id()
        if (
            base58.b58encode(msg.from_id).decode() == self_id
            and msg_forwarder != self_id
        ):
            logger.debug(
                "dropping message claiming to be from self but forwarded from %s",
                msg_forwarder,
            )
            return

        self.notify_subscriptions(msg)
        await self.router.publish(msg_forwarder, msg)``` 

[seetadev]

@IronJam11 : Thanks for digging into this, Aaryan — appreciate the thorough analysis and the time you’ve put into understanding both gossipsub and pubsub internals. This is solid progress, and you’re clearly circling the right root cause around async validation + message lifecycle ordering 👍

Your earlier breakdown around duplicate mcache.put() calls and the later refinement around when we mark messages as seen are both thoughtful approaches. I especially like that you’re reasoning explicitly about Trio’s cooperative scheduling and atomic sections — that’s exactly the level of rigor we want here.

I’d strongly encourage you to collaborate closely with @Winter-Soren (Soham) on this. He has already spent time identifying the race window, and pairing up should help converge faster on a fix that’s both correct and minimally invasive. If possible, please sync up with them on Discord and walk through the two approaches together (mcache-side dedup vs. earlier seen-marking) and agree on the cleanest, most idiomatic solution for py-libp2p.

If you want, feel free to propose a concrete reproduction harness or test case together as well — that’ll make review and long-term confidence much easier.

Really great momentum here from both of you. Looking forward to a clean resolution on this one 🚀

[IronJam11]

Hey, @Winter-Soren I would love to know your views on this one.