Struct SubscriptionRelay 

pub struct SubscriptionRelay<CD: ContextData, Q: RemoteQuerySubscriber> { /* private fields */ }

Manages predicate registration on remote peer reactor subscriptions.

The SubscriptionRelay provides a resilient, event-driven approach to managing which predicates are registered with remote durable peers. It automatically handles:

• Registering predicates on peer reactor subscriptions when peers connect
• Re-registering predicates when peers disconnect and reconnect
• Retrying failed predicate registration attempts
• Clean teardown when predicates are removed
• Storing ContextData for each predicate to enable proper authorization

This design separates predicate management concerns from the main Node implementation, making it easier to test and reason about predicate lifecycle management.

Public API (for Node integration)

• subscribe_predicate() - Call when local subscriptions are created (parallel to reactor.subscribe)
• unsubscribe_predicate() - Call when local subscriptions are removed (parallel to reactor.unsubscribe)
• notify_peer_connected() - Call when durable peers connect (triggers automatic predicate registration)
• notify_peer_disconnected() - Call when durable peers disconnect (orphans predicate registrations)
• get_status() - Query current state of a predicate registration

Internal/Testing API

• setup_remote_subscriptions() - Internal method for triggering predicate registration with specific peers (called automatically by notify_peer_connected, but exposed for testing)

The relay will automatically handle predicate registration/teardown asynchronously.

Implementations

impl<CD: ContextData, Q: RemoteQuerySubscriber> SubscriptionRelay<CD, Q>

pub fn new() -> Self

pub fn set_node(&self, node: Arc<dyn TNode<CD>>) -> Result<(), ()>

Inject the node (typically a WeakNode for production)

This should be called once during initialization. Returns an error if the node has already been set.

pub fn subscribe_query( &self, query_id: QueryId, collection_id: CollectionId, selection: Selection, context_data: CD, version: u32, livequery: Q, )

Notify the relay that a new predicate needs to be registered on remote peer subscriptions

This should be called whenever a local subscription is established. The relay will track this predicate and automatically attempt to register it with available durable peers.

pub fn update_query( &self, query_id: QueryId, selection: Selection, version: u32, ) -> Result<(), Error>

pub fn unsubscribe_predicate(&self, query_id: QueryId)

Notify the relay that a predicate should be removed from remote peer subscriptions

This will clean up all tracking state and send unsubscribe requests to any remote peers that have this predicate registered.

pub fn notify_peer_disconnected(&self, peer_id: EntityId)

Handle peer disconnection - mark all predicates for that peer as needing re-registration

This should be called when a durable peer disconnects. All predicates registered with that peer will be marked as pending and will be automatically re-registered when the peer reconnects or another suitable peer becomes available.

pub fn notify_peer_connected(&self, peer_id: EntityId)

Handle peer connection - trigger predicate registration on the new peer subscription

This should be called when a new durable peer connects. The relay will automatically attempt to register any pending predicates on the newly connected peer’s subscription.

pub fn get_status(&self, query_id: QueryId) -> Option<Status>

Get the current state of a predicate registration

pub fn get_contexts_for_peer(&self, peer_id: &EntityId) -> HashSet<CD>

Get all unique contexts for predicates established or requested with a specific peer TODO: update the data structure to do this via a direct lookup rather than having to scan the entire map

Trait Implementations

impl<CD: Clone + ContextData, Q: Clone + RemoteQuerySubscriber> Clone for SubscriptionRelay<CD, Q>

fn clone(&self) -> SubscriptionRelay<CD, Q>

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<CD: ContextData, Q: RemoteQuerySubscriber> Default for SubscriptionRelay<CD, Q>

fn default() -> Self

Returns the “default value” for a type.