[Python] Add async components for WebBridgeController

[VanitasCodes]

Async equivalents of the threading components, explored as part of Issue #335.

AsyncSubscriberPool is an async equivalent of SubscriberPool using zmq.asyncio. AsyncHeartbeatReceiver is an async equivalent of HeartbeatChecker with full lives and stale connection logic. CombinedBridge is a prototype combining CHIRP discovery in a thread with heartbeat tracking, CMDP receiving, and CSCP commands all running in the asyncio event loop.

Tested against PyRandomTransmitter through a full FSM cycle. The bridge detected 4 state transitions, received 7 log messages, and sent 3 commands successfully. The architecture reduces threading from 5 threads plus task queue to 1 thread for CHIRP plus asyncio.

To run, start PyRandomTransmitter in one terminal with python -m constellation.satellites.PyRandomTransmitter -g test -n Sat1 and run the bridge in another with python -m python.constellation.core.async_experimental.bridge.

Related: Issue #335

[Copilot]

Pull request overview

This PR introduces an experimental asyncio-based bridge for the Constellation Python core, providing async equivalents of existing threaded ZMQ subscriber pooling and heartbeat monitoring, plus a prototype “combined bridge” that runs heartbeat/CMDP processing in an event loop while keeping CHIRP discovery in a thread.

Changes:

• Adds AsyncSubscriberPool (async ZMQ SUB socket pool using zmq.asyncio).
• Adds AsyncHeartbeatReceiver (async heartbeat polling with lives/stale-connection tracking).
• Adds CombinedBridge prototype wiring CHIRP discovery (thread) + heartbeat + CMDP receive + CSCP command sending.

Reviewed changes

Copilot reviewed 3 out of 4 changed files in this pull request and generated 13 comments.

File	Description
python/constellation/core/async_experimental/bridge.py	Prototype combined bridge coordinating CHIRP discovery, async heartbeat/CMDP processing, and CSCP command dispatch.
python/constellation/core/async_experimental/async_pools.py	Async subscriber pool abstraction for CMDP-style PUB/SUB sockets.
python/constellation/core/async_experimental/async_heartbeat.py	Async heartbeat receiver that tracks satellite state and liveness via CHP messages.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[VanitasCodes]

@stephanlachnit Ready for your feedback. Copilot flagged some cleanup items (SPDX headers, unused imports, DEPART handling) I'll address those based on your direction on the overall approach.

[simonspa]

Woooow, sorry for the Copilot noise, this is incredibly annoying. I (think I) have switched it off now for the repo.

[stephanlachnit]

I think this is quite promising - it looks like the code changes required to switch to async are not too big. If I get it right, we use asyncio.gather instead of starting thread for the background tasks.

[VanitasCodes]

@stephanlachnit Yes, that's exactly right. CHIRP stays as the one thread since it uses raw UDP multicast rather than ZeroMQ so zmq.asyncio doesn't help there.

One thing I wanted to check is whether you'd prefer we integrate with the existing CHIRPManager for discovery or keep it self contained in the bridge.

[VanitasCodes]

@stephanlachnit AsyncHeartbeatReceiver now delegates socket management to an internal AsyncSubscriberPool, subscribing to "" at construction so every socket added via add_satellite() gets it automatically. Also fixed subscribe(topic, uuid=uuid) mutating _topics for per-socket calls, and replaced asyncio.to_thread for CSCP with a ThreadPoolExecutor(max_workers=1) to keep all REQ socket operations on one thread. Tested through the full cycle, same results as before.

[stephanlachnit]

AsyncHeartbeatReceiver now delegates socket management to an internal AsyncSubscriberPool, subscribing to "" at construction so every socket added via add_satellite() gets it automatically

Nice, this is how I imagined it!

[VanitasCodes]

@stephanlachnit Implemented async CHIRP using asyncio.DatagramProtocol. The multicast receive socket is set up with all the IP_ADD_MEMBERSHIP calls as before, then handed to the event loop via create_datagram_endpoint. datagram_received runs on the event loop so add_satellite and add_socket are called directly without call_soon_threadsafe. Send sockets stay synchronous since UDP sendto writes to the kernel buffer without blocking. _setup_transmitter is now an async method that schedules the blocking CSCP connect in the executor. Tested this one as well through the full cycle and same results as before, now with no dedicated threads at all.

[stephanlachnit]

I've taken a look the async CHIRP code, and I think we need to restructure it a bit - it works well for discovering services at a single point and lays out the async part, but for a general purpose CHIRP class the API needs to be adjusted.

This is what I roughly have in mind:

• The socket should stay in a separate class like the MulticastSocket right now
• The socket should be owned and managed by a CHIRPListener class, which takes care of the message decoding and calling callbacks
• The CHIRPListener class should have three public methods:
  • register_callback(callback_id, callback_func) to register callbacks
  • unregister_callback(callback_id) to remove callbacks
  • send_chirp_message(message) to send a CHIRP message via the socket
• Then CHIRPManager class (inheriting from CHIRPListener) then takes care of responding to request
• For this, a request_callback property in the CHIRPListener is needed so that the CHIRPManager can handle request
• The CHIRPManager should have the following public methods:
  • request(service_id) to request services
  • register_service(service_id, port) to register a service
  • unregister_service(service_id) to unregister a service
  • emit_offers() to emit offers of registered services
• In the service callbacks should take two arguments: an event type (enum for SERVICE_CONNECTED and SERVICE_DISCONNECTED) and a discovered service (containing the group ID, host ID, service ID, port and IP address).

If you think I missed something let me know!

[VanitasCodes]

@stephanlachnit Restructured async CHIRP per your feedback. AsyncMulticastSocket owns the raw socket setup, AsyncCHIRPListener owns the socket and manages a callback registry (register_callback, unregister_callback, send_chirp_message, request_callback property), AsyncCHIRPManager inherits from it and adds service registry (request, register_service, unregister_service, emit_offers). Service callbacks now take (CHIRPEvent, DiscoveredService).

Also applied the MR !1131 casing fixes to AsyncHeartbeatReceiver. _name_to_uuid now uses case_insensitive_dict, name comparison uses casefold(), and status is tracked from CHP frames.

Results after both changes. 7 state changes (all transitional states now caught, up from 4), 7 logs, 6 metrics, 3 commands, 0 dedicated threads.

Note: These changes are giving significantly better results than before (initially the transitional states weren't getting registered but now they are). I've attached a screenshot of the output.

[VanitasCodes]

If you think I missed something let me know!

One thing I added that wasn't in your spec, an async start(loop) method on AsyncCHIRPListener that calls loop.create_datagram_endpoint. This is the wiring step that hands the receive socket to asyncio. Let me know if you had a different approach in mind for that.

[stephanlachnit]

Overall quite nice, looks very promising! Just needs a few more checks from the original CHIRP code and then I think we can move on.

One thing I added that wasn't in your spec, an async start(loop) method on AsyncCHIRPListener that calls loop.create_datagram_endpoint. This is the wiring step that hands the receive socket to asyncio. Let me know if you had a different approach in mind for that.

Ah right I see the problem, it's not so easy to separate. Maybe we can try the following:

• Move connection_made, datagram_received and start to the AsyncMulticastSocket
• Add a property to the AsyncMulticastSocket for a callback, which is called by datagram_received
• Add a start method to AsyncCHIRPListener which calls AsyncMulticastSocket.start

This should separate the network communication from the CHIRP logic more cleanly

[VanitasCodes]

@stephanlachnit Addressed all the review comments. Nothing seems to be broken, the output is being displayed the same as before.

[stephanlachnit]

Very nice, this looks excellent now! For the bridge this should be sufficient, but I think we have two approaches from here:

• Convert other parts of the framework (should be mainly data transmission and heartbeating) and integrate the async part fully within in the framework, or
• Move towards implementing the web interface and merge the async parts later

Let's discuss this tomorrow.

[VanitasCodes]

@stephanlachnit Added async framework integration classes to async_experimental/: AsyncCHIRPManager, AsyncHeartbeatChecker, AsyncMonitoringListener, AsyncBaseController, AsyncScriptableController.

Each registers its coroutine via _add_com_task() following the same chain pattern as _add_com_thread(). BaseSatelliteFrame gets _com_task_factories, _add_com_task, and _start_com_tasks which runs everything via asyncio.gather.

MRO verified: WebBridgeController(AsyncScriptableController, AsyncMonitoringListener) resolves correctly with one BaseSatelliteFrame at the base.

[VanitasCodes]

@stephanlachnit While setting up WebControl as a separate package, importing constellation.core.async_experimental failed at runtime with ModuleNotFoundError. I investigated and found that meson-python does not auto-discover packages. Since async_experimental was never added there, it was absent from the installed package entirely. So, I added it and made a commit. I'm a bit uncertain about the approach though. Am I on the right track or is there a work-around approach you would like me to take?

I think that the editable install (pip install -e .) is broken for meson-built packages due to _constellationdaq_editable_loader.py creating virtual namespace paths that don't correspond to real directories. Non-editable install (pip install .) is required when working with async_experimental (that's how I made it work atleast).

[stephanlachnit]

I think that the editable install (pip install -e .) is broken for meson-built packages due to _constellationdaq_editable_loader.py creating virtual namespace paths that don't correspond to real directories. Non-editable install (pip install .) is required when working with async_experimental (that's how I made it work atleast).

You need to disable build isolation (this is also noted in https://constellation.pages.desy.de/application_development/intro/install_from_source.html#installing-the-constellation-package):

pip install --no-build-isolation --editable .