Mesh Routing and Forwarding

do_connect(twin::Twin)

Connect to the endpoint declared with REMBUS_BASE_URL env variable.

REMBUS_BASE_URL default to ws://127.0.0.1:8338

A component is considered anonymous when a different and random UUID is used as component identifier each time the application connect to the broker.

Performs the transport-level connection and, for non pre-shared-key authentication, the identity/attestation handshake (see authenticate and await_challenge) that seeds this node's mesh routing tables with the peer's exports via update_tables.

authenticate(router::Router, twin::Twin)

Client side of the connection handshake: declare this node's identity to the peer by sending an IdentityMsg (carrying the bare component id twin.uid.id and the local listener ports) and, if challenged, reply with the signed attestation (see attestate).

On success, the peer's response carries its view of the mesh (its own local methods/subscriptions plus whatever it already knows through other connected twins, computed by get_topics on the peer side) and is applied locally with update_tables, seeding router.topic_impls/ router.topic_interests for the twin representing this link. This is the mirror operation of attestation, which runs on the peer that accepts the connection.

No-op if twin is anonymous or the transport does not require authentication.

await_challenge(router::Router, twin::Twin)

Client side counterpart of do_connect for brokers configured with connection_mode === authenticated: instead of proactively sending an IdentityMsg (see authenticate), wait for the peer to send a challenge first (registered as a pending future keyed by CONNECTION_ID) and block until it is answered, raising on timeout or authentication failure. Table synchronization still happens the same way once the attestation completes, via update_tables.

attestation(router, twin, msg, authenticate=true)

Server side of the connection handshake, run when an IdentityMsg is received from a connecting component: authenticate the client (unless authenticate is false), bind its identity with setidentity, and reply with get_topics(router, twin) as the response payload — this router's own exports (local methods/subscribers plus whatever is already reachable through other connected twins). The peer applies this payload with update_tables on its side (see authenticate), which is how a newly joined mesh node immediately learns the routing state of the broker it just connected to.

If authentication fails, the websocket is closed instead.

update_tables(router::Router, twin::Twin, exports)

Register the peer's already-known topics into the local mesh routing tables when a connection is (re)established.

exports is the two-element list [exposers, subscribers] returned by the peer's get_topics (via the attestation response, see authenticate): exports[1] are topics the peer implements, exports[2] are topics the peer (or some twin reachable through it) is subscribed to. For each, twin is added to router.topic_impls/ router.topic_interests, so this router can immediately forward RPC requests and Pub/Sub messages towards the peer without waiting for individual expose/subscribe admin commands to propagate.

A no-op if exports is nothing or ismultipath is false.

get_topics(r::Router, target::Twin)

Return [topic_impls(r, target),topic_interests(r, target)], the pair of exports lists sent to target as the reason payload of a successful attestation response (see attestation). The connecting peer applies them with update_tables, completing the handshake half of mesh table synchronization (the other half being the SETUP_CMD admin command, see admin_command).

topic_impls(router::Router, target::Twin)

Compute the list of topics target should be told this router implements (or knows an implementor for): the router's own local Julia methods (router.local_function, excluding subscribers and built-ins) unioned, via _topics, with every topic in router.topic_impls that has an implementor other than target itself.

Used to build the exports sent back in the attestation response — see get_topics and attestation.

topic_interests(router::Router, target::Twin)

Compute the list of topics target should be told this router (or some twin reachable through it) is subscribed to: the router's own local subscriber callbacks (router.local_function filtered by router.local_subscriber) unioned, via _topics, with every topic in router.topic_interests that has a subscriber other than target itself.

Used to build the exports sent back in the attestation response — see get_topics and attestation.

_topics(results, target::Twin, topic_map)

Shared helper for topic_impls and topic_interests: extend the results collection with every key of topic_map (either router.topic_impls or router.topic_interests) that has at least one twin other than target associated with it, then return results as a Vector (CBOR encodes vectors more compactly than sets).

Excluding target ensures a peer is never told about its own topic — relevant when the attestation response is computed for the very twin whose link is being (re)established.

setidentity(router, twin, msg; isauth=false)

Update twin identity parameters.

Re-keys router.id_twin from the (likely ephemeral) previous id to rid(twin) after twin.uid is set from msg.cid, so the mesh's neighbor-lookup table (id_twin, used by admin_broadcast and direct-RPC target resolution) reflects the component's definitive name as soon as it authenticates.

reconnect(twin::Twin, url::RbURL)

Attempt to reconnect to the given url.

The url is the primary url of the twin or one of its failover urls.`

On a successful do_connect, re-synchronizes mesh state with twin_setup (sends the explicit SETUP_CMD admin command with this node's current exports, see admin_command): since a reconnection may land on a broker that lost all prior incremental subscribe/expose state for this link, the full local configuration is replayed in one shot rather than relying on the one-time attestation exchange used for a fresh connection (see authenticate). If this re-sync fails, the connection is torn down again rather than left in an inconsistent state.

If this twin exposes listener ports (it is itself a broker) and the new connection succeeded, any other currently-connected node is disconnected too, forcing them to reconnect to the (possibly new) primary broker rather than staying attached to a stale failover.

reconnect(twin::Twin)

Reconnection loop driver: repeatedly try twin.failovers in order (sleeping reconnect_period between attempts) via reconnect(twin, url) until one succeeds or the twin is closing. Installed as the HR_CONN_DOWN handler in handle_connection, it is what makes a mesh link self-heal (and re-run mesh table synchronization) after a transient network failure.

handle_connection(twin::Twin, failovers)

Setup the twin connection handling and initiate the connection.

• Add failover URLs to the twin and setup the reconnect handler.
• Setup the reconnect handler.
• Connect the twin.

remove_twin(router::Router, twin::Twin)

Remove twin from router.id_twin, dropping it from the set of direct mesh neighbors reachable from router (it will no longer be a candidate for admin_broadcast, direct-RPC target resolution, or any other neighbor lookup). Does not touch topic_impls/topic_interests; that cleanup is cleanup's responsibility.

destroy_twin(twin, router)

Remove the twin from the system.

Detach twin, shut down its Visor process, and remove it from the router with remove_twin, fully evicting it as a mesh neighbor. Called for anonymous (unnamed) twins when their receiver loop ends; named twins are kept around (just detached) to allow reconnection, see end_receiver.

end_receiver(twin::Twin)

Called when a twin's receiver loop terminates (connection closed): keep named twins around (just detached, so reconnection can reuse the same mesh-routing entries once the link comes back up) but fully evict anonymous twins with destroy_twin, since an anonymous component has no stable identity to reconnect against.

command_permitted(router::Router, twin::Twin)

Return true unless router requires authentication (router_isauthenticated) and twin has not authenticated (isauthenticated) and is not connected from the loopback interface (peer_isloopback), in which case the socket is closed and false is returned. Guards every inbound Pub/Sub (pubsub_msg), RPC (rpc_request) and admin (admin_msg) message before any mesh-routing logic (authorization, broadcast, forwarding) is applied.

pubsub_msg(router::Router, msg, relay_rule=always_true)

Handle an inbound PubSubMsg: after the command_permitted/ isauthorized checks, (QoS2 de-duplication aside) deliver it in two complementary ways so it reaches every interested party regardless of how far away they are in the mesh:

• local_subscribers(router, msg.twin, msg): run any Julia callback bound directly on this router for msg.topic (exact or glob match);
• broadcast_msg(router, msg): put! the message onto the inbox of every twin in router.topic_interests[msg.topic] (plus glob subscribers), which includes twins that are themselves neighbor brokers — causing pubsub_msg to run again on the next hop and the message to keep walking the mesh towards the real subscribers. See the Mesh Routing and Forwarding guide.

broadcast_msg is skipped unless relay_rule(msg.twin.uid) holds: a server (see server_task) relays only the messages it publishes itself.

Returns false (without delivering) if the permission/authorization checks fail, true otherwise.

admin_msg(router::Router, msg)

Entry point for inbound AdminReqMsgs received on a twin's connection: after the command_permitted check, dispatch to admin_command and deliver the result back onto twin's own inbox:

• if the command returned EnableReactiveMsg (the REACTIVE_CMD case), start a dedicated start_reactive process to replay data-at-rest and reply STS_SUCCESS directly;
• otherwise the ResMsg produced by admin_command is delivered as-is (it may need to travel back through respond/sendto_origin if the requestor is several hops away).

Returns false without doing anything else if the permission check fails.

rpc_request(router::Router, msg, implementor_rule)

Handle an inbound RpcReqMsg, after the command_permitted/ isauthorized checks:

• Direct RPC (msg.target !== nothing, i.e. the caller named a specific component or group): resolve msg.target against router.id_twin — either an exact id match, or a "<group>@..." pooled-member match (in which case select_twin applies the router's load-balancing policy among the matching members) — and delegate delivery to manage_target. STS_TARGET_NOT_FOUND is replied if no matching twin is connected here.
• Routable RPC (msg.target === nothing): try the router's own local_fn first; otherwise, if implementor_rule(twin.uid) allows it (used to prevent re-forwarding loops/policy restrictions upstream), call find_implementor(router, msg) to pick a connected exposer — possibly itself a neighbor broker, in which case this same function runs again there, hopping the request towards the real implementor across the mesh.

Returns msg unchanged (the various outcomes are delivered as side effects by put!ing responses/forwarded requests onto the relevant twins' inboxes).

manage_target(router, twin, target_twin, msg)

Deliver a directed RPC request/response msg (one that named an explicit msg.target component or group, see rpc_request) to target_twin:

• STS_TARGET_DOWN back to twin if target_twin's socket is closed;
• resolve it locally via local_fn if target_twin === twin (loopback call to a method exposed by the requestor itself);
• otherwise forward msg onto target_twin's inbox, provided the topic is actually known (router.topic_impls or a built-in command) — letting target_twin (possibly itself a neighbor broker) continue resolving or relaying it further along the mesh; STS_METHOD_NOT_FOUND is replied otherwise.

sendto_origin(twin, msg)

Short-circuit a response that targets a locally-pending direct request, without going through the full broadcast_msg/mesh-relay path.

If twin.socket.direct has a future registered under msg.id (i.e. this twin itself issued the matching direct request — such as an admin command or a direct RPC call awaiting synchronous completion), fulfill that future with msg, cancel its timeout timer and return true. Returns false when there is no such pending future, meaning the caller (respond, admin_broadcast) must fall back to delivering msg through the normal inbox/broadcast mechanism so it keeps propagating towards its real destination across the mesh.

Method-specific overloads exist for FutureResponse (always false, nothing to resolve) and WsPing (sends a WebSocket ping frame instead of resolving a future).

isadmin(router, twin, cmd)

Return true if twin is listed in router.admins, i.e. is allowed to execute the administration command cmd.

Logs an @error and returns false otherwise. Used as a guard at the beginning of every privileged admin command handler (private_topic, public_topic, authorize, unauthorize, SHUTDOWN_CMD, BROKER_CONFIG_CMD, ...).

isauthorized(router::Router, twin::Twin, topic::AbstractString)

Return true if topic is public (absent from router.topic_auth) or twin is explicitly listed in router.topic_auth[topic].

Every mesh hop re-evaluates this check independently with its own local topic_auth table, so a private topic stays protected regardless of how many brokers a subscribe/expose/publish/request message has to cross: there is no "trusted upstream" shortcut, each router enforces authorization on its own.

private_topic(router, twin, msg)

Administration command handler for PRIVATE_TOPIC_CMD: declares msg.topic private by creating an (initially empty) entry in router.topic_auth. Requires twin to be an admin (see isadmin).

This is a purely local-router operation: it is not flooded across the mesh by mark_and_broadcast, so a topic must be declared private on each broker where it should be protected.

private_topic(rb, topic::AbstractString)

Set the specified topic to private.

The component must have the admin role to change the privacy level.

public_topic(router, twin, msg)

Administration command handler for PUBLIC_TOPIC_CMD: resets msg.topic back to public by deleting its entry from router.topic_auth. Requires twin to be an admin (see isadmin).

Like private_topic, this is a local-router-only operation.

public_topic(rb, topic::AbstractString)

Set the specified topic to public.

The component must have the admin role to change the privacy level.

authorize(router, twin, msg)

Administration command handler for AUTHORIZE_CMD: grants the component identified by msg.data[CID] access to msg.topic, whether to:

• publish or subscribe to a private topic;
• make RPC requests to, or expose, a remote method.

Requires twin to be an admin and msg.data[CID] to be a non-empty component id. The topic is implicitly declared private (an empty router.topic_auth[msg.topic] entry is created if missing) before the grant is recorded, same as private_topic this is local to the router handling the command.

function authorize(rb, client::AbstractString, topic::AbstractString)

Authorize the client component to use the private topic.

The component must have the admin role for granting topic accessibility.

unauthorize(router, twin, msg)

Administration command handler for UNAUTHORIZE_CMD: revokes the grant previously given with authorize for the component identified by msg.data[CID] on msg.topic. Requires twin to be an admin and msg.data[CID] to be a non-empty component id.

function unauthorize(rb, client::AbstractString, topic::AbstractString)

Revoke authorization to the client component for use of the private topic.

The component must have the admin role for revoking topic accessibility.

shutdown_broker(router)

Administration action for SHUTDOWN_CMD: shut down the broker's Visor supervisor tree, terminating the broker process. Invoked asynchronously by admin_command so the STS_SUCCESS response can still be sent back to the caller before the process goes down.

color_admin(tw::Twin, msg)

Send msg to the neighbor tw, after "coloring" it: append rid(tw) to msg.data["touch"] (creating the list if absent) just before transmitting.

This is the mechanism that marks a link as already crossed by an admin command, so that tw (or whichever router/component re-broadcasts the message further) never bounces it back to a twin whose id is already in touch. See admin_broadcast and the Mesh Routing and Forwarding guide for the full flood/anti-loop scheme.

admin_broadcast(router::Router, twin::Twin, msg::RembusMsg)

Flood msg to every named, open, directly-connected neighbor of router (router.id_twin), except:

• twin itself, the originator of the command;
• any neighbor whose rid is already present in msg.data["touch"] (already reached through another path — see color_admin).

Each twin that is actually sent the message is marked via color_admin before transmission. This is the link-level half of the mesh flood; the router-level half (stopping re-processing once a router has already seen the command) is mark_and_broadcast, which calls this function.

Finally, if twin registered a future for msg.id (a direct, synchronous admin request), it is resolved here with STS_SUCCESS, so the originator gets its response immediately without waiting for the flood to finish propagating through the rest of the mesh.

mark_and_broadcast(router, twin, msg)

Router-level loop guard around admin_broadcast for mesh-wide subscribe/expose/unsubscribe/unexpose propagation.

Stamps msg.data["rmark"] with router.eid:

• if router.eid is already present, this router has already processed msg (the flood looped back to it through a cycle in the mesh graph); the function returns false without re-broadcasting, so the caller must not reapply the corresponding local table update either;
• otherwise router.eid is appended, admin_broadcast is called to flood msg to this router's neighbors, and true is returned so the caller proceeds to update its local topic_interests/topic_impls.

Together with the per-link touch marker used by admin_broadcast, this guarantees a command reaches every router in the mesh exactly once, regardless of topology (tree, ring, or arbitrary graph).

ismultipath(router)

Return true if router should maintain mesh routing tables (topic_interests/topic_impls) at all, i.e. it could have more than one outstanding route: it is a real broker or a pool component, as opposed to a plain single-link component.

Currently always returns true; kept as an explicit gate (and extension point) around every table update in admin_command and update_tables so the optimization for non-routing components can be reintroduced without touching call sites.

admin_command(router::Router, twin, msg::AdminReqMsg)

Dispatch and execute an administration command received from twin, based on msg.data[COMMAND]. This is the single entry point for every mesh-routing state change as well as for broker administration (topic privacy, authorization, configuration, shutdown, logging level).

Mesh-routing-relevant commands:

• SETUP_CMD: bulk-register twin's already-known "subscribers" and "exposers" into router.topic_interests/router.topic_impls (see update_tables for the symmetric client-side counterpart), then reply with EnableReactiveMsg. Used to (re)synchronize a link's full state in one shot, typically on reconnection (see twin_setup / reconnect).
• SUBSCRIBE_CMD / EXPOSE_CMD / UNSUBSCRIBE_CMD / UNEXPOSE_CMD: after an isauthorized check, propagate the change mesh-wide via mark_and_broadcast and, only if that router had not already processed this command, apply the corresponding local update to router.topic_interests/router.topic_impls (and twin.msg_from for subscriptions).
• PRIVATE_TOPICS_CONFIG_CMD, PRIVATE_TOPIC_CMD / PUBLIC_TOPIC_CMD, AUTHORIZE_CMD / UNAUTHORIZE_CMD: manage router.topic_auth (see private_topic, public_topic, authorize, unauthorize).
• REACTIVE_CMD, BROKER_CONFIG_CMD, LOAD_CONFIG_CMD, SAVE_CONFIG_CMD, SHUTDOWN_CMD, ENABLE_DEBUG_CMD, DISABLE_DEBUG_CMD: broker-local administration, gated by isadmin where privileged.

Returns a ResMsg carrying the resulting status and, for query-style commands, the requested data.