Add Message Syncronisation

[alistair-english]

Background

Many ROS nodes need to process messages from multiple topics that correspond to the same point in time. The standard ROS solution is message_filters, which pairs messages by timestamp using either exact or approximate matching.

Today, jig generates independent subscribers, each with their own callback. There is no way to declare that two or more subscribers should deliver their messages together. Users must either mark subscribers as manually_created and wire up message_filters themselves (defeating the purpose of jig), or implement ad-hoc buffering in their callbacks.

The immediate driver is node_dual_gnss_odometry, which must pair NavSatFix messages from two receivers before computing position and heading. But time-synchronised multi-topic processing is a common pattern across sensor fusion, control, and perception nodes.

Design

Sync groups are declared inline in the subscribers list alongside regular subscribers. A regular subscriber has a topic and type; a sync group has a name, policy, and a list of topics. Both live in the same list and map to the same sn.subscribers.* namespace in generated code.

Interface YAML

subscribers:
  # regular subscriber
  - topic: odom
    type: nav_msgs/msg/Odometry
    qos:
      reliability: RELIABLE

  # sync group — delivers paired messages via a single callback
  - name: dual_fix
    policy: approximate       # "approximate" or "exact"
    queue_size: 10
    max_interval: 0.05        # seconds; only valid when policy is "approximate"
    topics:
      - topic: fix_left
        type: sensor_msgs/msg/NavSatFix
        qos:
          history: 10
          reliability: BEST_EFFORT
      - topic: fix_right
        type: sensor_msgs/msg/NavSatFix
        qos:
          history: 10
          reliability: BEST_EFFORT

Regular subscribers and sync groups are distinguished structurally: a regular subscriber has topic + type, a sync group has name + policy + topics. The JSON schema uses oneOf with required fields to discriminate automatically.

Semantics

• A sync group appears in the Subscribers struct alongside regular subscribers. There is no separate sync_groups namespace.
• Each sync group has a single set_callback() that delivers all matched messages together.
• Topics within a sync group define their own topic, type, and qos — they are self-contained.
• A sync group must contain at least 2 and at most 9 topics (the message_filters template arity limit).
• for_each_param subscribers cannot be used within a sync group.
• Regular subscribers not part of any sync group behave exactly as they do today.

Schema Changes

Update the subscribers array items to use oneOf with two variants:

subscribers:
  type: array
  items:
    oneOf:
      - $ref: "#/$defs/subscriber"
      - $ref: "#/$defs/syncGroup"

Add syncGroup to $defs:

syncGroup:
  type: object
  description: "Synchronise multiple topics by timestamp"
  required:
    - name
    - policy
    - queue_size
    - topics
  additionalProperties: false
  properties:
    name:
      type: string
      description: "Field name for this sync group in generated code"
      pattern: '^[a-zA-Z_][a-zA-Z0-9_]*$'
    policy:
      enum: [approximate, exact]
      description: "Synchronisation policy"
    queue_size:
      type: integer
      minimum: 1
      description: "Size of the message queue per topic"
    max_interval:
      type: number
      minimum: 0
      exclusiveMinimum: true
      description: "Maximum time difference (seconds) between messages. Only valid with approximate policy."
    topics:
      type: array
      description: "Topics to synchronise"
      items:
        $ref: "#/$defs/syncGroupTopic"
      minItems: 2
      maxItems: 9

syncGroupTopic:
  type: object
  description: "A topic within a sync group"
  required:
    - topic
    - type
  additionalProperties: false
  properties:
    topic:
      type: string
      description: "Topic name"
    type:
      type: string
      description: "ROS message type"
    qos:
      $ref: "#/$defs/qos"

Validation Rules (Code Generator)

The JSON schema handles structural validation (required fields, types, min/max items). The code generator validates semantic rules that JSON schema cannot express:

1. max_interval is required when policy: approximate and forbidden when policy: exact.
2. Sync group name must not collide with any regular subscriber's field name.
3. Topics within a sync group cannot use for_each_param.

Generated C++ Code

Given the YAML above, the generated interface header produces:

Additional includes:

#include <message_filters/subscriber.h>
#include <message_filters/synchronizer.h>
#include <message_filters/sync_policies/approximate_time.h>
// or: #include <message_filters/sync_policies/exact_time.h>

Sync group member in the Subscribers struct:

template <typename SessionType> struct NodeDualGnssOdometrySubscribers {
    // regular subscribers
    jig::Subscriber<nav_msgs::msg::Odometry, SessionType> odom;

    // sync group: dual_fix
    struct DualFix {
        using Policy = message_filters::sync_policies::ApproximateTime<
            sensor_msgs::msg::NavSatFix,
            sensor_msgs::msg::NavSatFix>;

        message_filters::Subscriber<sensor_msgs::msg::NavSatFix> fix_left;
        message_filters::Subscriber<sensor_msgs::msg::NavSatFix> fix_right;
        std::shared_ptr<message_filters::Synchronizer<Policy>> sync;

        using Callback = std::function<void(
            std::shared_ptr<SessionType>,
            sensor_msgs::msg::NavSatFix::ConstSharedPtr,
            sensor_msgs::msg::NavSatFix::ConstSharedPtr)>;
        Callback callback;

        void set_callback(Callback cb) { callback = std::move(cb); }
    } dual_fix;
};

Session struct:

template <typename DerivedSessionType> struct NodeDualGnssOdometrySession : jig::Session {
    using jig::Session::Session;
    NodeDualGnssOdometryPublishers<DerivedSessionType> publishers;
    NodeDualGnssOdometrySubscribers<DerivedSessionType> subscribers;  // includes sync groups
    // ... services, actions, etc.
    std::shared_ptr<ParamListener> param_listener;
    Params params;
};

Initialisation in create_session():

// init sync group: dual_fix
sn->subscribers.dual_fix.fix_left.subscribe(&sn->node, "fix_left",
    rclcpp::QoS(10).best_effort().get_rmw_qos_profile());
sn->subscribers.dual_fix.fix_right.subscribe(&sn->node, "fix_right",
    rclcpp::QoS(10).best_effort().get_rmw_qos_profile());

sn->subscribers.dual_fix.sync = std::make_shared<
    message_filters::Synchronizer<decltype(sn->subscribers.dual_fix)::Policy>>(
    typename decltype(sn->subscribers.dual_fix)::Policy(10),
    sn->subscribers.dual_fix.fix_left,
    sn->subscribers.dual_fix.fix_right);

sn->subscribers.dual_fix.sync->setMaxIntervalDuration(rclcpp::Duration::from_seconds(0.05));

auto weak_sn = std::weak_ptr<SessionType>(sn);
sn->subscribers.dual_fix.sync->registerCallback(
    [weak_sn](
        sensor_msgs::msg::NavSatFix::ConstSharedPtr msg_0,
        sensor_msgs::msg::NavSatFix::ConstSharedPtr msg_1) {
        auto sn = weak_sn.lock();
        if (!sn) return;
        if (sn->node.get_current_state().id() !=
            lifecycle_msgs::msg::State::PRIMARY_STATE_ACTIVE) return;
        if (sn->subscribers.dual_fix.callback) {
            sn->subscribers.dual_fix.callback(sn, msg_0, msg_1);
        }
    });

Key details:

• message_filters::Subscriber is used instead of jig::Subscriber for synced topics.
• The registered callback includes the same lifecycle state guard that jig::Subscriber uses (only fires when ACTIVE).
• A weak pointer to the session prevents dangling references during shutdown.
• setMaxIntervalDuration is only called for approximate policy.

Generated Python Code

Additional imports:

import message_filters

Sync group in the Subscribers dataclass:

@dataclass
class Subscribers:
    # regular subscribers
    odom: jig.Subscriber = None

    # sync group: dual_fix
    @dataclass
    class DualFix:
        fix_left: message_filters.Subscriber = None
        fix_right: message_filters.Subscriber = None
        sync: message_filters.ApproximateTimeSynchronizer = None
        _callback: Callable | None = None

        def set_callback(self, callback):
            self._callback = callback

    dual_fix: DualFix = field(default_factory=DualFix)

Session dataclass:

@dataclass
class NodeDualGnssOdometrySession(jig.Session):
    publishers: Publishers
    subscribers: Subscribers       # includes sync groups
    # ...
    param_listener: ParamListener
    params: Params

Initialisation in _create_session():

# init sync group: dual_fix
from sensor_msgs.msg import NavSatFix
from rclpy.qos import QoSProfile, ReliabilityPolicy

qos_fix_left = QoSProfile(depth=10, reliability=ReliabilityPolicy.BEST_EFFORT)
qos_fix_right = QoSProfile(depth=10, reliability=ReliabilityPolicy.BEST_EFFORT)

sn.subscribers.dual_fix.fix_left = message_filters.Subscriber(
    node, NavSatFix, "fix_left", qos_profile=qos_fix_left)
sn.subscribers.dual_fix.fix_right = message_filters.Subscriber(
    node, NavSatFix, "fix_right", qos_profile=qos_fix_right)

sn.subscribers.dual_fix.sync = message_filters.ApproximateTimeSynchronizer(
    [sn.subscribers.dual_fix.fix_left, sn.subscribers.dual_fix.fix_right],
    queue_size=10,
    slop=0.05)

def _dual_fix_callback(msg_0, msg_1):
    if session.node.current_state != State.PRIMARY_STATE_ACTIVE:
        return
    if sn.subscribers.dual_fix._callback:
        sn.subscribers.dual_fix._callback(sn, msg_0, msg_1)

sn.subscribers.dual_fix.sync.registerCallback(_dual_fix_callback)

Cleanup

Sync group subscribers need to be destroyed during _destroy_entities (Python) and session teardown (C++). The message_filters::Subscriber in C++ is stack-allocated within the sync group struct, so it is destroyed when the session is destroyed. In Python, _destroy_entities should call unregister() on each message_filters.Subscriber.

User Code

CallbackReturn on_configure(std::shared_ptr<Session> sn) {
    sn->subscribers.dual_fix.set_callback(
        [](auto sn, auto left_msg, auto right_msg) {
            // both messages are from the same epoch
            // compute heading, odometry, etc.
        });
    return CallbackReturn::SUCCESS;
}

def on_configure(sn: Session) -> TransitionCallbackReturn:
    sn.subscribers.dual_fix.set_callback(on_dual_fix)
    return TransitionCallbackReturn.SUCCESS

def on_dual_fix(sn, left_msg, right_msg):
    # both messages are from the same epoch
    pass

Dependencies

message_filters should be declared as a dependency of the jig package itself, since synchronisation is a core framework feature. This means users do not need to add message_filters to their own package.xml.

Jinja2 Template Changes

node_interface.hpp.jinja2:

• Add conditional message_filters includes gated on the presence of sync groups.
• Add inner structs for each sync group within the Subscribers struct.
• Add sync group initialisation block in create_session().

node_interface.py.jinja2:

• Add conditional message_filters import.
• Add inner dataclasses for each sync group within the Subscribers dataclass.
• Add sync group initialisation in _create_session().
• Add cleanup in _destroy_entities().

Code Generator Changes

In generate_node_interface.py:

1. Partition the subscribers list into regular subscribers and sync groups (discriminated by the presence of name + policy + topics vs topic + type).
2. Validate semantic rules (max_interval consistency, no for_each_param, no name collisions).
3. Pass regular subscribers to existing template variables.
4. Pass sync group data (name, policy, queue_size, max_interval, topics with types/QoS) as a new template variable.
5. The template generates inner structs/dataclasses within the Subscribers type and corresponding init code.

Alternatives Considered

Separate sync_groups section at top level

The original design placed sync groups as a separate top-level section that referenced subscribers by field name. This created a "claimed subscriber" concept where subscribers were defined in one place but owned by another. Moving sync groups inline within the subscribers list makes ownership explicit and mirrors the code API where both live under sn.subscribers.*.

Separate sn.sync_groups namespace in generated code

We considered giving sync groups their own namespace (sn.sync_groups.dual_fix instead of sn.subscribers.dual_fix). This creates a conceptual split that doesn't match the user's mental model — sync groups are just a way of subscribing to topics. Keeping everything under sn.subscribers gives a single namespace for "things my node listens to".

Buffer-latest-message approach

Instead of message_filters, each subscriber could simply buffer the latest message and recompute on every arrival. This is simpler but loses temporal alignment guarantees. For the GNSS case the receivers run at the same rate so it would likely work, but it is not a general solution and would not belong in jig as a framework feature.

Keep synced subscribers in a separate list with cross-references

Define sync group topics in the main subscribers list, then reference them by name from a sync_groups sub-section. This keeps the subscriber list flat but introduces cross-referencing, validation of references, and the confusing situation where some subscribers in the list don't get their own callbacks. Inline topic definitions within the sync group are more honest about ownership.