[BUG] Missing messages to syndic when zmq_filtering is enabled.

[fanaticize]

Description
Missing message to syndic when zmq_filtering is enabled.

Setup

Salt Version:
          Salt: 3006.2

Python Version:
        Python: 3.10.12 (main, Aug  3 2023, 21:47:10) [GCC 11.2.0]

Dependency Versions:
          cffi: 1.14.6
      cherrypy: unknown
      dateutil: 2.8.1
     docker-py: Not Installed
         gitdb: Not Installed
     gitpython: Not Installed
        Jinja2: 3.1.2
       libgit2: Not Installed
  looseversion: 1.0.2
      M2Crypto: Not Installed
          Mako: Not Installed
       msgpack: 1.0.2
  msgpack-pure: Not Installed
  mysql-python: Not Installed
     packaging: 22.0
     pycparser: 2.21
      pycrypto: Not Installed
  pycryptodome: 3.9.8
        pygit2: Not Installed
  python-gnupg: 0.4.8
        PyYAML: 6.0.1
         PyZMQ: 23.2.0
        relenv: 0.13.3
         smmap: Not Installed
       timelib: 0.2.4
       Tornado: 4.5.3
           ZMQ: 4.3.4

System Versions:
          dist: rhel 8.8 Ootpa
        locale: utf-8
       machine: x86_64
       release: 4.18.0-477.15.1.el8_8.x86_64
        system: Linux
       version: Red Hat Enterprise Linux 8.8 Ootpa

• on-prem machine
• VM (Virtualbox, KVM, etc. please specify)
• VM running on a cloud service, please be explicit and add details
• container (Kubernetes, Docker, containerd, etc. please specify)
• or a combination, please be explicit
• jails if it is FreeBSD
• classic packaging
• onedir packaging
• used bootstrap to install

Steps to Reproduce the behavior

1. Run master, syndic, and minion. Order master, syndic, and minion are different machines.

# order master config
order_masters: True
id: master1
zmq_filtering: True
log_level_logfile: trace

# syndic master config
syndic_master: {master1_ip}
id: master2
zmq_filtering: True
log_level: trace
log_level_logfile: trace

# syndic minion config
id: master2
master: {master1_ip}
zmq_filtering: True

# minion config
master: {master2_ip}
id: minion1
zmq_filtering: True
log_level_logfile: trace

2. Run the script below on the order master.

import salt.client
from multiprocessing import Pool


def run(n):
    client = salt.client.LocalClient('/etc/salt/master')
    client.cmd_async(
        tgt=['minion1'],
        fun='test.echo',
        kwarg={
            'text': str(n)
        },
        timeout = None,
        tgt_type='list',
        ret="",
    )

if __name__ == '__main__':
    with Pool(10) as p:
        p.map(run, range(10))

3. Check the log in minion. I checked that some jobs were not delivered.

[root@minion1 salt]# tail -f minion | grep 'Return data'
2023-08-21 12:34:10,036 [salt.minion      :32  ][TRACE   ][24344] Return data: {'success': True, 'return': '6', 'retcode': 0, 'jid': '20230821033409990304', 'fun': 'test.echo', 'fun_args': [{'text': '6'}]}
2023-08-21 12:34:10,065 [salt.minion      :32  ][TRACE   ][24345] Return data: {'success': True, 'return': '5', 'retcode': 0, 'jid': '20230821033410014176', 'fun': 'test.echo', 'fun_args': [{'text': '5'}]}
2023-08-21 12:34:10,151 [salt.minion      :32  ][TRACE   ][24352] Return data: {'success': True, 'return': '4', 'retcode': 0, 'jid': '20230821033410044037', 'fun': 'test.echo', 'fun_args': [{'text': '4'}]}
2023-08-21 12:34:10,151 [salt.minion      :32  ][TRACE   ][24348] Return data: {'success': True, 'return': '3', 'retcode': 0, 'jid': '20230821033410025261', 'fun': 'test.echo', 'fun_args': [{'text': '3'}]}
2023-08-21 12:34:10,160 [salt.minion      :32  ][TRACE   ][24353] Return data: {'success': True, 'return': '0', 'retcode': 0, 'jid': '20230821033410014211', 'fun': 'test.echo', 'fun_args': [{'text': '0'}]}
2023-08-21 12:34:10,161 [salt.minion      :32  ][TRACE   ][24349] Return data: {'success': True, 'return': '7', 'retcode': 0, 'jid': '20230821033410018362', 'fun': 'test.echo', 'fun_args': [{'text': '7'}]}
2023-08-21 12:34:10,176 [salt.minion      :32  ][TRACE   ][24354] Return data: {'success': True, 'return': '2', 'retcode': 0, 'jid': '20230821033410032961', 'fun': 'test.echo', 'fun_args': [{'text': '2'}]}
2023-08-21 12:34:10,187 [salt.minion      :32  ][TRACE   ][24355] Return data: {'success': True, 'return': '8', 'retcode': 0, 'jid': '20230821033410047772', 'fun': 'test.echo', 'fun_args': [{'text': '8'}]}

Expected behavior
All jobs must be delivered normally.

Screenshots
If applicable, add screenshots to help explain your problem.

Versions Report

salt --versions-report

Salt Version:
          Salt: 3006.2

Python Version:
        Python: 3.10.12 (main, Aug  3 2023, 21:47:10) [GCC 11.2.0]

Dependency Versions:
          cffi: 1.14.6
      cherrypy: unknown
      dateutil: 2.8.1
     docker-py: Not Installed
         gitdb: Not Installed
     gitpython: Not Installed
        Jinja2: 3.1.2
       libgit2: Not Installed
  looseversion: 1.0.2
      M2Crypto: Not Installed
          Mako: Not Installed
       msgpack: 1.0.2
  msgpack-pure: Not Installed
  mysql-python: Not Installed
     packaging: 22.0
     pycparser: 2.21
      pycrypto: Not Installed
  pycryptodome: 3.9.8
        pygit2: Not Installed
  python-gnupg: 0.4.8
        PyYAML: 6.0.1
         PyZMQ: 23.2.0
        relenv: 0.13.3
         smmap: Not Installed
       timelib: 0.2.4
       Tornado: 4.5.3
           ZMQ: 4.3.4

System Versions:
          dist: rhel 8.8 Ootpa
        locale: utf-8
       machine: x86_64
       release: 4.18.0-477.15.1.el8_8.x86_64
        system: Linux
       version: Red Hat Enterprise Linux 8.8 Ootpa

Additional context

salt/salt/transport/zeromq.py

Lines 787 to 811 in 8f750fa

	if self.opts["zmq_filtering"]:
	if topic_list:
	for topic in topic_list:
	log.trace("Sending filtered data over publisher %s", self.pub_uri)
	# zmq filters are substring match, hash the topic
	# to avoid collisions
	htopic = salt.utils.stringutils.to_bytes(
	hashlib.sha1(salt.utils.stringutils.to_bytes(topic)).hexdigest()
	)
	yield self.dpub_sock.send(htopic, flags=zmq.SNDMORE)
	yield self.dpub_sock.send(payload)
	log.trace("Filtered data has been sent")
	# Syndic broadcast
	if self.opts.get("order_masters"):
	log.trace("Sending filtered data to syndic")
	yield self.dpub_sock.send(b"syndic", flags=zmq.SNDMORE)
	yield self.dpub_sock.send(payload)
	log.trace("Filtered data has been sent to syndic")
	# otherwise its a broadcast
	else:
	# TODO: constants file for "broadcast"
	log.trace("Sending broadcasted data over publisher %s", self.pub_uri)
	yield self.dpub_sock.send(b"broadcast", flags=zmq.SNDMORE)
	yield self.dpub_sock.send(payload)
	log.trace("Broadcasted data has been sent")

In the above code, the problem seems to occur when multiple jobs are executed at the same time. I found that if I change to use send_multipart as shown below, the message is not missing.

yield self.dpub_sock.send_multipart([htopic, payload])

[welcome]

Hi there! Welcome to the Salt Community! Thank you for making your first contribution. We have a lengthy process for issues and PRs. Someone from the Core Team will follow up as soon as possible. In the meantime, here’s some information that may help as you continue your Salt journey.
Please be sure to review our Code of Conduct. Also, check out some of our community resources including:

• Community Wiki
• Salt’s Contributor Guide
• Join our Community Slack
• IRC on LiberaChat
• Salt Project YouTube channel
• Salt Project Twitch channel

There are lots of ways to get involved in our community. Every month, there are around a dozen opportunities to meet with other contributors and the Salt Core team and collaborate in real time. The best way to keep track is by subscribing to the Salt Community Events Calendar.
If you have additional questions, email us at saltproject@vmware.com. We’re glad you’ve joined our community and look forward to doing awesome things with you!

[dwoz]

Looks like this was fixed by #65681. Closing the issue.