Feat: Improved implementation of NNG broker support by alexted · Pull Request #612 · taskiq-python/taskiq

alexted · 2026-04-26T14:28:32Z

Refactoring the #607

… affinity policy, and scheduler abstraction.

s3rius

That's a whole broker-server implementation, which I'm not quite sure if we want to have it in the main repo.

I thought that it would be like ZeroMQ, but it's much more impressive. Maybe we might consider moving it to another package to keep the main lib thin.

s3rius · 2026-05-06T21:49:10Z

+
+# ── standalone CLI entry point ────────────────────────────────────────────────
+
+def _build_config() -> HubConfig:


Instead of defining a stand-alone file to run, let's make it a sub-command of the taskiq itself.

You can do it by following our CLI extending guide.

https://taskiq-python.github.io/extending-taskiq/cli.html

s3rius · 2026-05-06T21:54:40Z

+    the least-loaded worker instead of relying on NNG round-robin.
+
+    **State** — :class:`~taskiq.brokers.nng.storage.InMemoryStore`.  All
+    store operations are synchronous and execute directly on the asyncio event


Suggested change

store operations are synchronous and execute directly on the asyncio event

store operations are asynchronous and execute directly on the asyncio event

The methods are actually def, not async def, so the wording is correct.

s3rius · 2026-05-06T22:00:28Z

+            "labels": message.labels,
+            "lease_id": "",  # hub assigns the real lease_id at dispatch time
+            "attempts": int(message.labels.get("attempts", 0)),
+            "max_retries": int(


Since we have retry-middlewares that do exactly that, I guess we can remove retry functionality from the broker and the hub.

s3rius · 2026-05-06T22:08:27Z

+            lease_id = envelope.lease_id  # hub-assigned; correct by construction
+
+            async def _ack(
+                _task_id: str = task_id,


Instead of creating those parameters as defaults, just use them inside the function. In that case your closure will borrow variables from the outer environment. It will make it more rock solid to use.

Default arguments here are the canonical fix for late binding in Python. task_id/lease_id/worker_id are reassigned on every iteration, and without snapshotting via defaults, a delayed ack will capture someone else’s values.

s3rius · 2026-05-06T22:10:59Z

+    payload_b64: str
+    labels: dict[str, Any] = field(default_factory=dict)
+    lease_id: str = ""
+    attempts: int = 0


Retry-related again.

s3rius · 2026-05-06T22:13:00Z

+    status: str = "starting"
+    version: str = "unknown"
+
+    def as_dict(self) -> dict[str, Any]:


You don't need to define a function for that. There's a built-in asdict function for dataclasses.

s3rius · 2026-05-06T22:18:57Z

+            separators=(",", ":"),
+            ensure_ascii=False,


Separators will save you like few bytes. But it might make messages more readable.

Suggested change

separators=(",", ":"),

ensure_ascii=False,

ensure_ascii=False,

For the wire format, compactness is the norm, not a micro-optimization. These are bytes on the socket, not logs.

s3rius · 2026-05-06T22:23:59Z

+        payload: dict[str, Any] = {
+            "task_id": message.task_id,
+            "task_name": message.task_name,
+            "payload_b64": base64.b64encode(message.message).decode("ascii"),


Suggested change

"payload_b64": base64.b64encode(message.message).decode("ascii"),

"payload_b64": base64.b64encode(message.message).decode("utf-8"),

UTF-8 is used in all other places. Let's keep it everywhere to not have any encoding\decoding issues.

Base64 output is ASCII by definition, .decode("ascii") is stricter and semantically more accurate. "Consistency" is not relevant here.

s3rius · 2026-05-06T22:25:30Z

+    # ── control plane ─────────────────────────────────────────────────────────
+
+    async def _control_handler(self, ctx: Any) -> None:
+        """Run one Rep0 context: receive → dispatch → reply, in a loop."""


Suggested change

"""Run one Rep0 context: receive → dispatch → reply, in a loop."""

"""Run one Rep0 context: receive -> dispatch -> reply, in a loop."""

codecov · 2026-05-06T22:37:43Z

Codecov Report

❌ Patch coverage is 70.53942% with 213 lines in your changes missing coverage. Please review.
✅ Project coverage is 77.08%. Comparing base (2fd7a3c) to head (fe3b0a5).

Files with missing lines	Patch %	Lines
taskiq/brokers/nng/broker.py	0.00%	129 Missing ⚠️
taskiq/brokers/nng/hub.py	71.61%	65 Missing ⚠️
taskiq/brokers/nng/storage.py	94.11%	16 Missing ⚠️
taskiq/brokers/nng/protocol.py	96.62%	3 Missing ⚠️

Additional details and impacted files

@@               Coverage Diff               @@
##           feature/nng     #612      +/-   ##
===============================================
- Coverage        78.30%   77.08%   -1.23%     
===============================================
  Files               70       74       +4     
  Lines             2485     3186     +701     
===============================================
+ Hits              1946     2456     +510     
- Misses             539      730     +191

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

alexted · 2026-05-07T10:38:11Z

That's a whole broker-server implementation, which I'm not quite sure if we want to have it in the main repo.

I thought that it would be like ZeroMQ, but it's much more impressive. Maybe we might consider moving it to another package to keep the main lib thin.

I intended this PR as a full implementation proposal, rather than a requirement to merge it as-is. Feel free to adopt it fully, partially, or not at all, depending on what aligns best with the project’s direction.

Replace as_dict with dataclasses.asdict; Remove user-level retry orchestration from broker/hub/protocol

alexted · 2026-05-23T20:07:08Z

That's a whole broker-server implementation, which I'm not quite sure if we want to have it in the main repo.

I thought that it would be like ZeroMQ, but it's much more impressive. Maybe we might consider moving it to another package to keep the main lib thin.

Strategically valid, you are right.

Alexandr Tedeev added 2 commits April 26, 2026 16:38

Refactoring the NNG support solution v1

f74cce9

Refactoring the NNG support solution v1.5: Simplify store

7fdb8eb

alexted mentioned this pull request Apr 26, 2026

Feature Request: Add NNG (Nanomsg Next Gen) Broker Support #602

Open

Refactoring the NNG support solution v2: Update routing policy

b8c7212

alexted force-pushed the feature/add-support-NNG-as-broker branch from 78a2868 to b8c7212 Compare April 26, 2026 15:37

Refactoring the NNG support solution v2.5: Update routing policy, add…

fe3b0a5

… affinity policy, and scheduler abstraction.

alexted mentioned this pull request Apr 26, 2026

Added NNG broker. #607

Open

s3rius requested changes May 6, 2026

View reviewed changes

alexted changed the title ~~Feature: add support nng as broker~~ Feat: Full-featured NNG broker with hub-worker architecture May 23, 2026

alexted changed the title ~~Feat: Full-featured NNG broker with hub-worker architecture~~ Feat: Improved implementation of NNG broker support May 23, 2026

refactor:

6e8cdc4

Replace as_dict with dataclasses.asdict; Remove user-level retry orchestration from broker/hub/protocol


		# ── standalone CLI entry point ────────────────────────────────────────────────

		def _build_config() -> HubConfig:

	store operations are synchronous and execute directly on the asyncio event
	store operations are asynchronous and execute directly on the asyncio event

	separators=(",", ":"),
	ensure_ascii=False,
	ensure_ascii=False,

	"payload_b64": base64.b64encode(message.message).decode("ascii"),
	"payload_b64": base64.b64encode(message.message).decode("utf-8"),

	"""Run one Rep0 context: receive → dispatch → reply, in a loop."""
	"""Run one Rep0 context: receive -> dispatch -> reply, in a loop."""

Uh oh!

Conversation

alexted commented Apr 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

s3rius left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov Bot commented May 6, 2026

Codecov Report

Uh oh!

alexted commented May 7, 2026

Uh oh!

alexted commented May 23, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

alexted commented Apr 26, 2026 •

edited

Loading