test_hot_reload.rs

Overview

This file implements a test harness for evaluating the hot reload capabilities of a networked node system using the MsQuicTransport transport layer. It defines a HotReloadNode struct that wraps a network node instance with facilities to receive, store, and inspect messages asynchronously. The main functionality centers around starting nodes, receiving messages via an asynchronous Tokio task, and verifying message propagation behavior in a multi-node setup. The file includes a Tokio integration test test_hot_reload that initializes multiple nodes, sends broadcast messages, and asserts correct message delivery and reception.

Entities and Functionality

HotReloadNode

A wrapper struct representing a network node instance configured for hot reloading tests. It manages internal node state, message sending channels, and a thread dedicated to receiving incoming messages.

Fields:

• _node: Node<MsQuicTransport>
  The underlying network node instance parameterized over the MsQuicTransport transport layer.

• _direct_tx: NetDirectSender<String, Message>
  Channel sender for direct messages to peers (unused in this file but retained for completeness).

• broadcast_tx: NetBroadcastSender<Message>
  Channel sender for broadcasting messages to all connected peers.

• _peers_rx: tokio::sync::watch::Receiver<HashMap<String, Vec<PeerData>>>
  A Tokio watch receiver that tracks the current mapping of peers in the network.

• received_messages: Arc<std::sync::Mutex<Vec<Message>>>
  Thread-safe, shared storage of received messages for inspection and assertions.

• _receiver_task: std::thread::JoinHandle<()>
  Handle to the spawned thread running the receiver loop that processes incoming messages.

Methods:

async fn start(transport: MsQuicTransport, config: NodeConfig) -> Self

Starts a HotReloadNode instance asynchronously.

• Parameters:

  • transport: The transport layer object used for network communication (MsQuicTransport).

  • config: Configuration object describing node parameters such as addresses and node ID.

• Returns:
  A new instance of HotReloadNode.

• Behavior:

  • Calls Node::start to initialize the underlying node and obtain communication channels.

  • Creates a shared Arc<Mutex<Vec<Message>>> to store received messages.

  • Spawns a dedicated thread running the run_receiver method to continuously listen for incoming messages.

  • Returns a HotReloadNode struct encapsulating all initialized components.

• Usage Example:

  let node = HotReloadNode::start(transport, config).await;
  

fn received_messages(&self) -> Vec<Message>

Returns a clone of the vector containing all messages received by this node.

• Returns:
  A vector of Message objects previously received.

• Usage Example:

  let messages = node.received_messages();
  

fn run_receiver(receiver: String, message_rx: InstrumentedReceiver<IncomingMessage>, received_messages: Arc<std::sync::Mutex<Vec<Message>>>)

Continuously receives and processes incoming messages for a node, running in a separate thread.

• Parameters:

  • receiver: The node identifier as a string, used for tracing output.

  • message_rx: An instrumented MPSC receiver channel for incoming messages.

  • received_messages: Shared, thread-safe vector to store received Message objects.

• Behavior:

  • Loops while messages can be received from message_rx.

  • Calls finish(&None) on received messages to deserialize and extract the payload.

  • Logs trace-level information about received messages.

  • Pushes successfully extracted messages into the shared received messages vector.

  • Logs when the receiver task terminates.

• Note:
  This method is intended to run in a dedicated thread and interacts with the Tokio runtime asynchronously via the MPSC channel.

#[tokio::test] async fn test_hot_reload()

An integration test that verifies hot reload message passing functionality among multiple nodes.

• Setup:

  • Initializes logging.

  • Creates an instance of MsQuicTransport.

  • Configures and starts two HotReloadNode instances with distinct node IDs and gossip addresses.

  • Waits 2 seconds to allow nodes to initialize and establish connections.

• Test Execution:

  • Node 0 broadcasts a Message::Data payload.

  • Waits 1 second to allow messages to propagate.

  • Asserts that Node 0 itself does not receive the broadcasted message.

  • Asserts that all other nodes receive exactly one message.

• Cleanup:

  • Drops the nodes vector to shut down nodes.

  • Waits 1 second before test completion.

• Notes:

  • The test is marked with #[ignore] indicating it is not run by default.

  • There is commented-out code suggesting an intent to test dynamic reconfiguration (hot reload) by modifying the node's bind address, but it is not executed.

Implementation Details and Algorithms

• Message Receiving Loop:
  The run_receiver method uses a blocking loop on the MPSC receiver to process incoming messages. The use of InstrumentedReceiver suggests telemetry or diagnostics are integrated for message reception.

• Thread Synchronization:
  Received messages are stored in a thread-safe Arc<Mutex<Vec<Message>>> to allow concurrent access from the receiver thread and test assertions.

• Node Initialization:
  Nodes are started asynchronously with the Node::start method which returns both the node instance and communication channels for direct and broadcast messaging, as well as peer state monitoring.

• Broadcast Message Testing:
  The test verifies that broadcast messages sent by one node are received by other nodes but not echoed back to the sender, consistent with typical broadcast semantics.

Interaction with Other System Components

• Transport Layer:
  The file relies on the MsQuicTransport from the transport_layer::msquic module for network communication, allowing QUIC-based messaging.

• Node and Messaging Modules:
  It utilizes the Node abstraction from crate::tests::Node, and associated messaging and configuration types (NodeConfig, Message) to simulate node behavior.

• Channels:
  Uses NetBroadcastSender and NetDirectSender from crate::channel for sending messages to peers.

• Telemetry:
  Incoming messages are received through InstrumentedReceiver from telemetry_utils::mpsc, indicating integration with telemetry systems.

• Testing Utilities:
  Uses utilities from crate::tests such as init_logs, node_addr, and gossip_addr for test setup.

Visual Diagram: Structure and Workflow of HotReloadNode

classDiagram
class HotReloadNode {
-_node: Node<MsQuicTransport>
-_direct_tx: NetDirectSender
+broadcast_tx: NetBroadcastSender
-_peers_rx: watch::Receiver<HashMap>
-received_messages: Arc<Mutex<Vec<Message>>>
-_receiver_task: JoinHandle
+start()
+received_messages()
-run_receiver()
}
HotReloadNode --> Node
HotReloadNode --> NetDirectSender
HotReloadNode --> NetBroadcastSender
HotReloadNode --> "Arc<Mutex<Vec<Message>>>"
HotReloadNode --> "Thread (Receiver Task)"

This class diagram shows the main components of the HotReloadNode struct, highlighting its fields and methods, and the relation to the underlying Node, message channels, thread-safe message storage, and the receiver task thread.

For further exploration of asynchronous programming patterns, concurrency primitives, and transport abstractions referenced here, see topics on Asynchronous Rust, Concurrency and Synchronization, and Network Transport Layers. The interaction between message channels and node lifecycle is related to Pub/Sub Messaging and Node Configuration.