cluster adapter

ClusterAdapter Implementation Summary

Overview

This document summarizes the implementation of the ClusterAdapter for the Pusher Protocol Server Rust Implementation. The ClusterAdapter provides multi-instance deployment support using UDP-based node discovery, enabling horizontal scaling across multiple server instances.

Requirements Addressed

• Requirement 4.2: Support a cluster adapter for multi-instance deployments using UDP discovery
• Requirement 4.5: Broadcast messages across all instances when using horizontal adapters
• Requirement 4.6: Synchronize channel membership across instances
• Requirement 17.1: Discover other nodes using UDP broadcast
• Requirement 17.2: Support master election in cluster mode
• Requirement 17.3: Handle node promotion and demotion events
• Requirement 17.4: Track active nodes in the cluster
• Requirement 17.5: Support configurable discovery intervals and timeouts

Implementation Details

Core Components

1. ClusterAdapter (src/adapters/cluster.rs)

The main adapter that extends LocalAdapter with cluster-wide capabilities:

• Local Adapter Integration: Wraps a LocalAdapter for managing local socket connections
• Discovery Manager: Manages UDP-based node discovery and master election
• Broadcast Channel: Tokio broadcast channel for distributing cluster messages
• Configuration: Uses ClusterAdapterConfig for port, multicast address, and timeout settings

2. Discovery System

Implements UDP multicast-based node discovery:

• Hello Messages: Sent when a node joins the cluster
• Heartbeat Messages: Sent periodically (every 5 seconds) to maintain presence
• Goodbye Messages: Sent when a node leaves the cluster gracefully
• Node Timeout Detection: Removes nodes that haven't sent heartbeats for 15 seconds
• Master Election: Deterministic election based on node weight and ID

3. Cluster Messages

Four types of cluster messages for synchronization:

• Broadcast: Distribute messages to channel subscribers across all nodes
• AddMember: Synchronize presence channel member additions
• RemoveMember: Synchronize presence channel member removals
• TerminateUser: Terminate user connections across all nodes

Master Election Algorithm

The master election uses a simple but effective algorithm:

1. Each node calculates a weight (based on uptime)
2. Nodes are sorted by weight (descending), then by node ID (ascending)
3. The node with the highest weight becomes master
4. In case of equal weights, the lexicographically smallest ID wins
5. Election is triggered when:
   • A new node joins
   • A node leaves
   • The current master times out

Message Broadcasting

When a message is sent to a channel:

1. The message is delivered to local subscribers via LocalAdapter
2. A ClusterMessage::Broadcast is created
3. The message is sent to the local broadcast channel
4. The message is sent via UDP to all known nodes
5. Each node receives the message and delivers it to their local subscribers

Configuration

The ClusterAdapter is configured via ClusterAdapterConfig:

pub struct ClusterAdapterConfig {
    pub port: u16,                    // UDP port for discovery (default: 11002)
    pub multicast_address: String,    // Multicast address (default: "239.1.1.1")
    pub request_timeout_ms: u64,      // Request timeout (default: 5000)
}

Integration with AppState

The AppState initialization was updated to support cluster mode:

1. Parse adapter driver from options (adapter_driver field)
2. Create appropriate adapter based on driver type
3. Initialize adapter asynchronously via init_async() method
4. For cluster mode, also initialize ClusterRateLimiter

Graceful Degradation

The implementation includes graceful degradation for environments where multicast is not available:

• Multicast join failures are logged as warnings but don't prevent initialization
• Discovery message send failures are logged at debug level
• The adapter continues to function as a LocalAdapter if multicast is unavailable

Testing

Comprehensive tests were added in tests/cluster_adapter_test.rs:

1. Creation Test: Verifies ClusterAdapter can be created and initialized
2. Socket Management Test: Tests adding and removing sockets
3. Channel Management Test: Tests channel subscription and unsubscription
4. Presence Management Test: Tests presence member tracking
5. Master Status Test: Verifies master election works correctly
6. Send Message Test: Tests message broadcasting to subscribers
7. Disconnect Test: Tests graceful shutdown and cleanup

All tests pass successfully.

Usage Example

Configuration

To use the ClusterAdapter, set the adapter_driver option to "cluster":

./soketi-rs --adapter-driver cluster

Programmatic Usage

use soketi_rs::adapters::cluster::ClusterAdapter;
use soketi_rs::config::ClusterAdapterConfig;

let config = ClusterAdapterConfig {
    port: 11002,
    multicast_address: "239.1.1.1".to_string(),
    request_timeout_ms: 5000,
};

let adapter = ClusterAdapter::new(config).await?;
adapter.init().await?;

// Use adapter for socket management, message broadcasting, etc.

Files Modified

1. src/adapters/cluster.rs (NEW): Main ClusterAdapter implementation
2. src/adapters/mod.rs: Added cluster module export
3. src/state.rs: Updated to support async adapter initialization
4. src/main.rs: Updated to call async initialization
5. src/ws_handler.rs: Updated close_all_local_sockets to support ClusterAdapter
6. tests/cluster_adapter_test.rs (NEW): Comprehensive test suite

Future Enhancements

Potential improvements for future iterations:

1. Cluster-wide Queries: Implement methods to query socket counts and channel membership across all nodes
2. Request-Response Pattern: Add request-response mechanism for querying remote nodes
3. Encryption: Add encryption for cluster messages
4. Authentication: Add authentication for node discovery
5. TCP Fallback: Add TCP-based communication as fallback when UDP multicast is unavailable
6. Metrics: Add cluster-specific metrics (node count, message distribution, etc.)
7. Health Checks: Add health check endpoints for cluster status

Conclusion

The ClusterAdapter implementation successfully provides multi-instance deployment support with UDP-based node discovery. It extends the LocalAdapter with cluster-wide message broadcasting and synchronization capabilities, enabling horizontal scaling of the Pusher protocol server.

The implementation follows Rust best practices, includes comprehensive error handling, and provides graceful degradation for environments where multicast is not available. All tests pass successfully, demonstrating the correctness of the implementation.