with retry

client/internal/dns/systemd_linux.go

@@ -31,7 +31,6 @@ const (
 	systemdDbusSetDefaultRouteMethodSuffix = systemdDbusLinkInterface + ".SetDefaultRoute"
 	systemdDbusSetDomainsMethodSuffix      = systemdDbusLinkInterface + ".SetDomains"
 	systemdDbusSetDNSSECMethodSuffix       = systemdDbusLinkInterface + ".SetDNSSEC"
-	systemdDbusSetDNSOverTLSMethodSuffix   = systemdDbusLinkInterface + ".SetDNSOverTLS"
 	systemdDbusResolvConfModeForeign       = "foreign"
 
 	dbusErrorUnknownObject = "org.freedesktop.DBus.Error.UnknownObject"
@@ -103,11 +102,6 @@ func (s *systemdDbusConfigurator) applyDNSConfig(config HostDNSConfig, stateMana
 		log.Warnf("failed to set DNSSEC to 'no': %v", err)
 	}
 
-	// We don't support DNSOverTLS. On some machines this is default on so we explicitly set it to off
-	if err := s.callLinkMethod(systemdDbusSetDNSOverTLSMethodSuffix, dnsSecDisabled); err != nil {
-		log.Warnf("failed to set DNSOverTLS to 'no': %v", err)
-	}
-
 	var (
 		searchDomains []string
 		matchDomains  []string

client/wasm/internal/rdp/cert_validation.go

@@ -73,8 +73,8 @@ func (p *RDCleanPathProxy) validateCertificateWithJS(conn *proxyConnection, cert
 	}
 }
 
-func (p *RDCleanPathProxy) getTLSConfigWithValidation(conn *proxyConnection, requiresCredSSP bool) *tls.Config {
-	config := &tls.Config{
+func (p *RDCleanPathProxy) getTLSConfigWithValidation(conn *proxyConnection) *tls.Config {
+	return &tls.Config{
 		InsecureSkipVerify: true, // We'll validate manually after handshake
 		VerifyConnection: func(cs tls.ConnectionState) error {
 			var certChain [][]byte
@@ -93,15 +93,4 @@ func (p *RDCleanPathProxy) getTLSConfigWithValidation(conn *proxyConnection, req
 			return nil
 		},
 	}
-
-	// CredSSP (NLA) requires TLS 1.2 - it's incompatible with TLS 1.3
-	if requiresCredSSP {
-		config.MinVersion = tls.VersionTLS12
-		config.MaxVersion = tls.VersionTLS12
-	} else {
-		config.MinVersion = tls.VersionTLS12
-		config.MaxVersion = tls.VersionTLS13
-	}
-
-	return config
 }

client/wasm/internal/rdp/rdcleanpath.go

@@ -6,13 +6,11 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/asn1"
-	"errors"
 	"fmt"
 	"io"
 	"net"
 	"sync"
 	"syscall/js"
-	"time"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -21,34 +19,18 @@ const (
 	RDCleanPathVersion     = 3390
 	RDCleanPathProxyHost   = "rdcleanpath.proxy.local"
 	RDCleanPathProxyScheme = "ws"
-
-	rdpDialTimeout = 15 * time.Second
-
-	GeneralErrorCode = 1
-	WSAETimedOut     = 10060
-	WSAEConnRefused  = 10061
-	WSAEConnAborted  = 10053
-	WSAEConnReset    = 10054
-	WSAEGenericError = 10050
 )
 
 type RDCleanPathPDU struct {
-	Version           int64          `asn1:"tag:0,explicit"`
-	Error             RDCleanPathErr `asn1:"tag:1,explicit,optional"`
-	Destination       string         `asn1:"utf8,tag:2,explicit,optional"`
-	ProxyAuth         string         `asn1:"utf8,tag:3,explicit,optional"`
-	ServerAuth        string         `asn1:"utf8,tag:4,explicit,optional"`
-	PreconnectionBlob string         `asn1:"utf8,tag:5,explicit,optional"`
-	X224ConnectionPDU []byte         `asn1:"tag:6,explicit,optional"`
-	ServerCertChain   [][]byte       `asn1:"tag:7,explicit,optional"`
-	ServerAddr        string         `asn1:"utf8,tag:9,explicit,optional"`
-}
-
-type RDCleanPathErr struct {
-	ErrorCode      int16 `asn1:"tag:0,explicit"`
-	HTTPStatusCode int16 `asn1:"tag:1,explicit,optional"`
-	WSALastError   int16 `asn1:"tag:2,explicit,optional"`
-	TLSAlertCode   int8  `asn1:"tag:3,explicit,optional"`
+	Version           int64    `asn1:"tag:0,explicit"`
+	Error             []byte   `asn1:"tag:1,explicit,optional"`
+	Destination       string   `asn1:"utf8,tag:2,explicit,optional"`
+	ProxyAuth         string   `asn1:"utf8,tag:3,explicit,optional"`
+	ServerAuth        string   `asn1:"utf8,tag:4,explicit,optional"`
+	PreconnectionBlob string   `asn1:"utf8,tag:5,explicit,optional"`
+	X224ConnectionPDU []byte   `asn1:"tag:6,explicit,optional"`
+	ServerCertChain   [][]byte `asn1:"tag:7,explicit,optional"`
+	ServerAddr        string   `asn1:"utf8,tag:9,explicit,optional"`
 }
 
 type RDCleanPathProxy struct {
@@ -228,13 +210,9 @@ func (p *RDCleanPathProxy) handleDirectRDP(conn *proxyConnection, firstPacket []
 	destination := conn.destination
 	log.Infof("Direct RDP mode: Connecting to %s via NetBird", destination)
 
-	ctx, cancel := context.WithTimeout(conn.ctx, rdpDialTimeout)
-	defer cancel()
-
-	rdpConn, err := p.nbClient.Dial(ctx, "tcp", destination)
+	rdpConn, err := p.nbClient.Dial(conn.ctx, "tcp", destination)
 	if err != nil {
 		log.Errorf("Failed to connect to %s: %v", destination, err)
-		p.sendRDCleanPathError(conn, newWSAError(err))
 		return
 	}
 	conn.rdpConn = rdpConn
@@ -242,7 +220,6 @@ func (p *RDCleanPathProxy) handleDirectRDP(conn *proxyConnection, firstPacket []
 	_, err = rdpConn.Write(firstPacket)
 	if err != nil {
 		log.Errorf("Failed to write first packet: %v", err)
-		p.sendRDCleanPathError(conn, newWSAError(err))
 		return
 	}
 
@@ -250,7 +227,6 @@ func (p *RDCleanPathProxy) handleDirectRDP(conn *proxyConnection, firstPacket []
 	n, err := rdpConn.Read(response)
 	if err != nil {
 		log.Errorf("Failed to read X.224 response: %v", err)
-		p.sendRDCleanPathError(conn, newWSAError(err))
 		return
 	}
 
@@ -293,52 +269,3 @@ func (p *RDCleanPathProxy) sendToWebSocket(conn *proxyConnection, data []byte) {
 		conn.wsHandlers.Call("send", uint8Array.Get("buffer"))
 	}
 }
-
-func (p *RDCleanPathProxy) sendRDCleanPathError(conn *proxyConnection, pdu RDCleanPathPDU) {
-	data, err := asn1.Marshal(pdu)
-	if err != nil {
-		log.Errorf("Failed to marshal error PDU: %v", err)
-		return
-	}
-	p.sendToWebSocket(conn, data)
-}
-
-func errorToWSACode(err error) int16 {
-	if err == nil {
-		return WSAEGenericError
-	}
-	var netErr *net.OpError
-	if errors.As(err, &netErr) && netErr.Timeout() {
-		return WSAETimedOut
-	}
-	if errors.Is(err, context.DeadlineExceeded) {
-		return WSAETimedOut
-	}
-	if errors.Is(err, context.Canceled) {
-		return WSAEConnAborted
-	}
-	if errors.Is(err, io.EOF) {
-		return WSAEConnReset
-	}
-	return WSAEGenericError
-}
-
-func newWSAError(err error) RDCleanPathPDU {
-	return RDCleanPathPDU{
-		Version: RDCleanPathVersion,
-		Error: RDCleanPathErr{
-			ErrorCode:    GeneralErrorCode,
-			WSALastError: errorToWSACode(err),
-		},
-	}
-}
-
-func newHTTPError(statusCode int16) RDCleanPathPDU {
-	return RDCleanPathPDU{
-		Version: RDCleanPathVersion,
-		Error: RDCleanPathErr{
-			ErrorCode:      GeneralErrorCode,
-			HTTPStatusCode: statusCode,
-		},
-	}
-}

client/wasm/internal/rdp/rdcleanpath_handlers.go

@@ -3,7 +3,6 @@
 package rdp
 
 import (
-	"context"
 	"crypto/tls"
 	"encoding/asn1"
 	"io"
@@ -12,17 +11,11 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-const (
-	// MS-RDPBCGR: confusingly named, actually means PROTOCOL_HYBRID (CredSSP)
-	protocolSSL      = 0x00000001
-	protocolHybridEx = 0x00000008
-)
-
 func (p *RDCleanPathProxy) processRDCleanPathPDU(conn *proxyConnection, pdu RDCleanPathPDU) {
 	log.Infof("Processing RDCleanPath PDU: Version=%d, Destination=%s", pdu.Version, pdu.Destination)
 
 	if pdu.Version != RDCleanPathVersion {
-		p.sendRDCleanPathError(conn, newHTTPError(400))
+		p.sendRDCleanPathError(conn, "Unsupported version")
 		return
 	}
 
@@ -31,13 +24,10 @@ func (p *RDCleanPathProxy) processRDCleanPathPDU(conn *proxyConnection, pdu RDCl
 		destination = pdu.Destination
 	}
 
-	ctx, cancel := context.WithTimeout(conn.ctx, rdpDialTimeout)
-	defer cancel()
-
-	rdpConn, err := p.nbClient.Dial(ctx, "tcp", destination)
+	rdpConn, err := p.nbClient.Dial(conn.ctx, "tcp", destination)
 	if err != nil {
 		log.Errorf("Failed to connect to %s: %v", destination, err)
-		p.sendRDCleanPathError(conn, newWSAError(err))
+		p.sendRDCleanPathError(conn, "Connection failed")
 		p.cleanupConnection(conn)
 		return
 	}
@@ -50,34 +40,6 @@ func (p *RDCleanPathProxy) processRDCleanPathPDU(conn *proxyConnection, pdu RDCl
 	p.setupTLSConnection(conn, pdu)
 }
 
-// detectCredSSPFromX224 checks if the X.224 response indicates NLA/CredSSP is required.
-// Per MS-RDPBCGR spec: byte 11 = TYPE_RDP_NEG_RSP (0x02), bytes 15-18 = selectedProtocol flags.
-// Returns (requiresTLS12, selectedProtocol, detectionSuccessful).
-func (p *RDCleanPathProxy) detectCredSSPFromX224(x224Response []byte) (bool, uint32, bool) {
-	const minResponseLength = 19
-
-	if len(x224Response) < minResponseLength {
-		return false, 0, false
-	}
-
-	// Per X.224 specification:
-	//   x224Response[0] == 0x03: Length of X.224 header (3 bytes)
-	//   x224Response[5] == 0xD0: X.224 Data TPDU code
-	if x224Response[0] != 0x03 || x224Response[5] != 0xD0 {
-		return false, 0, false
-	}
-
-	if x224Response[11] == 0x02 {
-		flags := uint32(x224Response[15]) | uint32(x224Response[16])<<8 |
-			uint32(x224Response[17])<<16 | uint32(x224Response[18])<<24
-
-		hasNLA := (flags & (protocolSSL | protocolHybridEx)) != 0
-		return hasNLA, flags, true
-	}
-
-	return false, 0, false
-}
-
 func (p *RDCleanPathProxy) setupTLSConnection(conn *proxyConnection, pdu RDCleanPathPDU) {
 	var x224Response []byte
 	if len(pdu.X224ConnectionPDU) > 0 {
@@ -85,7 +47,7 @@ func (p *RDCleanPathProxy) setupTLSConnection(conn *proxyConnection, pdu RDClean
 		_, err := conn.rdpConn.Write(pdu.X224ConnectionPDU)
 		if err != nil {
 			log.Errorf("Failed to write X.224 PDU: %v", err)
-			p.sendRDCleanPathError(conn, newWSAError(err))
+			p.sendRDCleanPathError(conn, "Failed to forward X.224")
 			return
 		}
 
@@ -93,32 +55,21 @@ func (p *RDCleanPathProxy) setupTLSConnection(conn *proxyConnection, pdu RDClean
 		n, err := conn.rdpConn.Read(response)
 		if err != nil {
 			log.Errorf("Failed to read X.224 response: %v", err)
-			p.sendRDCleanPathError(conn, newWSAError(err))
+			p.sendRDCleanPathError(conn, "Failed to read X.224 response")
 			return
 		}
 		x224Response = response[:n]
 		log.Debugf("Received X.224 Connection Confirm (%d bytes)", n)
 	}
 
-	requiresCredSSP, selectedProtocol, detected := p.detectCredSSPFromX224(x224Response)
-	if detected {
-		if requiresCredSSP {
-			log.Warnf("Detected NLA/CredSSP (selectedProtocol: 0x%08X), forcing TLS 1.2 for compatibility", selectedProtocol)
-		} else {
-			log.Warnf("No NLA/CredSSP detected (selectedProtocol: 0x%08X), allowing up to TLS 1.3", selectedProtocol)
-		}
-	} else {
-		log.Warnf("Could not detect RDP security protocol, allowing up to TLS 1.3")
-	}
-
-	tlsConfig := p.getTLSConfigWithValidation(conn, requiresCredSSP)
+	tlsConfig := p.getTLSConfigWithValidation(conn)
 
 	tlsConn := tls.Client(conn.rdpConn, tlsConfig)
 	conn.tlsConn = tlsConn
 
 	if err := tlsConn.Handshake(); err != nil {
 		log.Errorf("TLS handshake failed: %v", err)
-		p.sendRDCleanPathError(conn, newWSAError(err))
+		p.sendRDCleanPathError(conn, "TLS handshake failed")
 		return
 	}
 
@@ -155,6 +106,47 @@ func (p *RDCleanPathProxy) setupTLSConnection(conn *proxyConnection, pdu RDClean
 	p.cleanupConnection(conn)
 }
 
+func (p *RDCleanPathProxy) setupPlainConnection(conn *proxyConnection, pdu RDCleanPathPDU) {
+	if len(pdu.X224ConnectionPDU) > 0 {
+		log.Debugf("Forwarding X.224 Connection Request (%d bytes)", len(pdu.X224ConnectionPDU))
+		_, err := conn.rdpConn.Write(pdu.X224ConnectionPDU)
+		if err != nil {
+			log.Errorf("Failed to write X.224 PDU: %v", err)
+			p.sendRDCleanPathError(conn, "Failed to forward X.224")
+			return
+		}
+
+		response := make([]byte, 1024)
+		n, err := conn.rdpConn.Read(response)
+		if err != nil {
+			log.Errorf("Failed to read X.224 response: %v", err)
+			p.sendRDCleanPathError(conn, "Failed to read X.224 response")
+			return
+		}
+
+		responsePDU := RDCleanPathPDU{
+			Version:           RDCleanPathVersion,
+			X224ConnectionPDU: response[:n],
+			ServerAddr:        conn.destination,
+		}
+
+		p.sendRDCleanPathPDU(conn, responsePDU)
+	} else {
+		responsePDU := RDCleanPathPDU{
+			Version:    RDCleanPathVersion,
+			ServerAddr: conn.destination,
+		}
+		p.sendRDCleanPathPDU(conn, responsePDU)
+	}
+
+	go p.forwardConnToWS(conn, conn.rdpConn, "TCP")
+	go p.forwardWSToConn(conn, conn.rdpConn, "TCP")
+
+	<-conn.ctx.Done()
+	log.Debug("TCP connection context done, cleaning up")
+	p.cleanupConnection(conn)
+}
+
 func (p *RDCleanPathProxy) sendRDCleanPathPDU(conn *proxyConnection, pdu RDCleanPathPDU) {
 	data, err := asn1.Marshal(pdu)
 	if err != nil {
@@ -166,6 +158,21 @@ func (p *RDCleanPathProxy) sendRDCleanPathPDU(conn *proxyConnection, pdu RDClean
 	p.sendToWebSocket(conn, data)
 }
 
+func (p *RDCleanPathProxy) sendRDCleanPathError(conn *proxyConnection, errorMsg string) {
+	pdu := RDCleanPathPDU{
+		Version: RDCleanPathVersion,
+		Error:   []byte(errorMsg),
+	}
+
+	data, err := asn1.Marshal(pdu)
+	if err != nil {
+		log.Errorf("Failed to marshal error PDU: %v", err)
+		return
+	}
+
+	p.sendToWebSocket(conn, data)
+}
+
 func (p *RDCleanPathProxy) readWebSocketMessage(conn *proxyConnection) ([]byte, error) {
 	msgChan := make(chan []byte)
 	errChan := make(chan error)

go.mod

@@ -62,7 +62,7 @@ require (
 	github.com/miekg/dns v1.1.59
 	github.com/mitchellh/hashstructure/v2 v2.0.2
 	github.com/nadoo/ipset v0.5.0
-	github.com/netbirdio/management-integrations/integrations v0.0.0-20251010134843-7af36217ac1f
+	github.com/netbirdio/management-integrations/integrations v0.0.0-20250906095204-f87a07690ba0
 	github.com/netbirdio/signal-dispatcher/dispatcher v0.0.0-20250805121659-6b4ac470ca45
 	github.com/okta/okta-sdk-golang/v2 v2.18.0
 	github.com/oschwald/maxminddb-golang v1.12.0

go.sum

@@ -503,8 +503,8 @@ github.com/netbirdio/go-netroute v0.0.0-20240611143515-f59b0e1d3944 h1:TDtJKmM6S
 github.com/netbirdio/go-netroute v0.0.0-20240611143515-f59b0e1d3944/go.mod h1:sHA6TRxjQ6RLbnI+3R4DZo2Eseg/iKiPRfNmcuNySVQ=
 github.com/netbirdio/ice/v4 v4.0.0-20250908184934-6202be846b51 h1:Ov4qdafATOgGMB1wbSuh+0aAHcwz9hdvB6VZjh1mVMI=
 github.com/netbirdio/ice/v4 v4.0.0-20250908184934-6202be846b51/go.mod h1:ZSIbPdBn5hePO8CpF1PekH2SfpTxg1PDhEwtbqZS7R8=
-github.com/netbirdio/management-integrations/integrations v0.0.0-20251010134843-7af36217ac1f h1:XIpRDlpPz3zFUkpwaqDRHjwpQRsf2ZKHggoex1MTafs=
-github.com/netbirdio/management-integrations/integrations v0.0.0-20251010134843-7af36217ac1f/go.mod h1:v0nUbbHbuQnqR7yKIYnKzsLBCswLtp2JctmKYmGgVhc=
+github.com/netbirdio/management-integrations/integrations v0.0.0-20250906095204-f87a07690ba0 h1:9BUqQHPVOGr0edk8EifUBUfTr2Ob0ypAPxtasUApBxQ=
+github.com/netbirdio/management-integrations/integrations v0.0.0-20250906095204-f87a07690ba0/go.mod h1:v0nUbbHbuQnqR7yKIYnKzsLBCswLtp2JctmKYmGgVhc=
 github.com/netbirdio/service v0.0.0-20240911161631-f62744f42502 h1:3tHlFmhTdX9axERMVN63dqyFqnvuD+EMJHzM7mNGON8=
 github.com/netbirdio/service v0.0.0-20240911161631-f62744f42502/go.mod h1:CIMRFEJVL+0DS1a3Nx06NaMn4Dz63Ng6O7dl0qH0zVM=
 github.com/netbirdio/signal-dispatcher/dispatcher v0.0.0-20250805121659-6b4ac470ca45 h1:ujgviVYmx243Ksy7NdSwrdGPSRNE3pb8kEDSpH0QuAQ=

infrastructure_files/docker-compose.yml.tmpl.traefik

@@ -49,7 +49,6 @@ services:
     - traefik.http.routers.netbird-wsproxy-signal.service=netbird-wsproxy-signal
     - traefik.http.services.netbird-wsproxy-signal.loadbalancer.server.port=80
     - traefik.http.routers.netbird-signal.rule=Host(`$NETBIRD_DOMAIN`) && PathPrefix(`/signalexchange.SignalExchange/`)
-    - traefik.http.routers.netbird-signal.service=netbird-signal
     - traefik.http.services.netbird-signal.loadbalancer.server.port=10000
     - traefik.http.services.netbird-signal.loadbalancer.server.scheme=h2c
 

management/server/integrated_validator.go

@@ -136,7 +136,7 @@ func (a MockIntegratedValidator) GetValidatedPeers(_ context.Context, accountID
 	return validatedPeers, nil
 }
 
-func (MockIntegratedValidator) PreparePeer(_ context.Context, accountID string, peer *nbpeer.Peer, peersGroup []string, extraSettings *types.ExtraSettings, temporary bool) *nbpeer.Peer {
+func (MockIntegratedValidator) PreparePeer(_ context.Context, accountID string, peer *nbpeer.Peer, peersGroup []string, extraSettings *types.ExtraSettings) *nbpeer.Peer {
 	return peer
 }
 

management/server/integrations/integrated_validator/interface.go

@@ -3,16 +3,16 @@ package integrated_validator
 import (
 	"context"
 
+	"github.com/netbirdio/netbird/shared/management/proto"
 	nbpeer "github.com/netbirdio/netbird/management/server/peer"
 	"github.com/netbirdio/netbird/management/server/types"
-	"github.com/netbirdio/netbird/shared/management/proto"
 )
 
 // IntegratedValidator interface exists to avoid the circle dependencies
 type IntegratedValidator interface {
 	ValidateExtraSettings(ctx context.Context, newExtraSettings *types.ExtraSettings, oldExtraSettings *types.ExtraSettings, peers map[string]*nbpeer.Peer, userID string, accountID string) error
 	ValidatePeer(ctx context.Context, update *nbpeer.Peer, peer *nbpeer.Peer, userID string, accountID string, dnsDomain string, peersGroup []string, extraSettings *types.ExtraSettings) (*nbpeer.Peer, bool, error)
-	PreparePeer(ctx context.Context, accountID string, peer *nbpeer.Peer, peersGroup []string, extraSettings *types.ExtraSettings, temporary bool) *nbpeer.Peer
+	PreparePeer(ctx context.Context, accountID string, peer *nbpeer.Peer, peersGroup []string, extraSettings *types.ExtraSettings) *nbpeer.Peer
 	IsNotValidPeer(ctx context.Context, accountID string, peer *nbpeer.Peer, peersGroup []string, extraSettings *types.ExtraSettings) (bool, bool, error)
 	GetValidatedPeers(ctx context.Context, accountID string, groups []*types.Group, peers []*nbpeer.Peer, extraSettings *types.ExtraSettings) (map[string]struct{}, error)
 	PeerDeleted(ctx context.Context, accountID, peerID string, extraSettings *types.ExtraSettings) error

management/server/peer.go

@@ -578,7 +578,7 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, accountID, setupKe
 		}
 	}
 
-	newPeer = am.integratedPeerValidator.PreparePeer(ctx, accountID, newPeer, groupsToAdd, settings.Extra, temporary)
+	newPeer = am.integratedPeerValidator.PreparePeer(ctx, accountID, newPeer, groupsToAdd, settings.Extra)
 
 	network, err := am.Store.GetAccountNetwork(ctx, store.LockingStrengthNone, accountID)
 	if err != nil {

management/server/store/cache/dual_key_cache.go

@@ -1,129 +0,0 @@
-package cache
-
-import (
-	"context"
-	"sync"
-)
-
-// DualKeyCache provides a caching mechanism where each entry has two keys:
-// - Primary key (e.g., objectID): used for accessing and invalidating specific entries
-// - Secondary key (e.g., accountID): used for bulk invalidation of all entries with the same secondary key
-type DualKeyCache[K1 comparable, K2 comparable, V any] struct {
-	mu             sync.RWMutex
-	primaryIndex   map[K1]V               // Primary key -> Value
-	secondaryIndex map[K2]map[K1]struct{} // Secondary key -> Set of primary keys
-	reverseLookup  map[K1]K2              // Primary key -> Secondary key
-}
-
-// NewDualKeyCache creates a new dual-key cache
-func NewDualKeyCache[K1 comparable, K2 comparable, V any]() *DualKeyCache[K1, K2, V] {
-	return &DualKeyCache[K1, K2, V]{
-		primaryIndex:   make(map[K1]V),
-		secondaryIndex: make(map[K2]map[K1]struct{}),
-		reverseLookup:  make(map[K1]K2),
-	}
-}
-
-// Get retrieves a value from the cache using the primary key
-func (c *DualKeyCache[K1, K2, V]) Get(ctx context.Context, primaryKey K1) (V, bool) {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	value, ok := c.primaryIndex[primaryKey]
-	return value, ok
-}
-
-// Set stores a value in the cache with both primary and secondary keys
-func (c *DualKeyCache[K1, K2, V]) Set(ctx context.Context, primaryKey K1, secondaryKey K2, value V) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	if oldSecondaryKey, exists := c.reverseLookup[primaryKey]; exists {
-		if primaryKeys, ok := c.secondaryIndex[oldSecondaryKey]; ok {
-			delete(primaryKeys, primaryKey)
-			if len(primaryKeys) == 0 {
-				delete(c.secondaryIndex, oldSecondaryKey)
-			}
-		}
-	}
-
-	c.primaryIndex[primaryKey] = value
-	c.reverseLookup[primaryKey] = secondaryKey
-
-	if _, exists := c.secondaryIndex[secondaryKey]; !exists {
-		c.secondaryIndex[secondaryKey] = make(map[K1]struct{})
-	}
-	c.secondaryIndex[secondaryKey][primaryKey] = struct{}{}
-}
-
-// InvalidateByPrimaryKey removes an entry using the primary key
-func (c *DualKeyCache[K1, K2, V]) InvalidateByPrimaryKey(ctx context.Context, primaryKey K1) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	if secondaryKey, exists := c.reverseLookup[primaryKey]; exists {
-		if primaryKeys, ok := c.secondaryIndex[secondaryKey]; ok {
-			delete(primaryKeys, primaryKey)
-			if len(primaryKeys) == 0 {
-				delete(c.secondaryIndex, secondaryKey)
-			}
-		}
-		delete(c.reverseLookup, primaryKey)
-	}
-
-	delete(c.primaryIndex, primaryKey)
-}
-
-// InvalidateBySecondaryKey removes all entries with the given secondary key
-func (c *DualKeyCache[K1, K2, V]) InvalidateBySecondaryKey(ctx context.Context, secondaryKey K2) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	primaryKeys, exists := c.secondaryIndex[secondaryKey]
-	if !exists {
-		return
-	}
-
-	for primaryKey := range primaryKeys {
-		delete(c.primaryIndex, primaryKey)
-		delete(c.reverseLookup, primaryKey)
-	}
-
-	delete(c.secondaryIndex, secondaryKey)
-}
-
-// InvalidateAll removes all entries from the cache
-func (c *DualKeyCache[K1, K2, V]) InvalidateAll(ctx context.Context) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.primaryIndex = make(map[K1]V)
-	c.secondaryIndex = make(map[K2]map[K1]struct{})
-	c.reverseLookup = make(map[K1]K2)
-}
-
-// Size returns the number of entries in the cache
-func (c *DualKeyCache[K1, K2, V]) Size() int {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	return len(c.primaryIndex)
-}
-
-// GetOrSet retrieves a value from the cache, or sets it using the provided function if not found
-// The loadFunc should return both the value and the secondary key (extracted from the value)
-func (c *DualKeyCache[K1, K2, V]) GetOrSet(ctx context.Context, primaryKey K1, loadFunc func() (V, K2, error)) (V, error) {
-	if value, ok := c.Get(ctx, primaryKey); ok {
-		return value, nil
-	}
-
-	value, secondaryKey, err := loadFunc()
-	if err != nil {
-		var zero V
-		return zero, err
-	}
-
-	c.Set(ctx, primaryKey, secondaryKey, value)
-
-	return value, nil
-}

management/server/store/cache/single_key_cache.go

@@ -1,77 +0,0 @@
-package cache
-
-import (
-	"context"
-	"sync"
-)
-
-// SingleKeyCache provides a simple caching mechanism with a single key
-type SingleKeyCache[K comparable, V any] struct {
-	mu    sync.RWMutex
-	cache map[K]V // Key -> Value
-}
-
-// NewSingleKeyCache creates a new single-key cache
-func NewSingleKeyCache[K comparable, V any]() *SingleKeyCache[K, V] {
-	return &SingleKeyCache[K, V]{
-		cache: make(map[K]V),
-	}
-}
-
-// Get retrieves a value from the cache using the key
-func (c *SingleKeyCache[K, V]) Get(ctx context.Context, key K) (V, bool) {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	value, ok := c.cache[key]
-	return value, ok
-}
-
-// Set stores a value in the cache with the given key
-func (c *SingleKeyCache[K, V]) Set(ctx context.Context, key K, value V) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.cache[key] = value
-}
-
-// Invalidate removes an entry using the key
-func (c *SingleKeyCache[K, V]) Invalidate(ctx context.Context, key K) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	delete(c.cache, key)
-}
-
-// InvalidateAll removes all entries from the cache
-func (c *SingleKeyCache[K, V]) InvalidateAll(ctx context.Context) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.cache = make(map[K]V)
-}
-
-// Size returns the number of entries in the cache
-func (c *SingleKeyCache[K, V]) Size() int {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	return len(c.cache)
-}
-
-// GetOrSet retrieves a value from the cache, or sets it using the provided function if not found
-func (c *SingleKeyCache[K, V]) GetOrSet(ctx context.Context, key K, loadFunc func() (V, error)) (V, error) {
-	if value, ok := c.Get(ctx, key); ok {
-		return value, nil
-	}
-
-	value, err := loadFunc()
-	if err != nil {
-		var zero V
-		return zero, err
-	}
-
-	c.Set(ctx, key, value)
-
-	return value, nil
-}

management/server/store/cache/triple_key_cache.go

@@ -1,242 +0,0 @@
-package cache
-
-import (
-	"context"
-	"sync"
-)
-
-// TripleKeyCache provides a caching mechanism where each entry has three keys:
-// - Primary key (K1): used for accessing and invalidating specific entries
-// - Secondary key (K2): used for bulk invalidation of all entries with the same secondary key
-// - Tertiary key (K3): used for bulk invalidation of all entries with the same tertiary key
-type TripleKeyCache[K1 comparable, K2 comparable, K3 comparable, V any] struct {
-	mu             sync.RWMutex
-	primaryIndex   map[K1]V               // Primary key -> Value
-	secondaryIndex map[K2]map[K1]struct{} // Secondary key -> Set of primary keys
-	tertiaryIndex  map[K3]map[K1]struct{} // Tertiary key -> Set of primary keys
-	reverseLookup  map[K1]keyPair[K2, K3] // Primary key -> Secondary and Tertiary keys
-}
-
-type keyPair[K2 comparable, K3 comparable] struct {
-	secondary K2
-	tertiary  K3
-}
-
-// NewTripleKeyCache creates a new triple-key cache
-func NewTripleKeyCache[K1 comparable, K2 comparable, K3 comparable, V any]() *TripleKeyCache[K1, K2, K3, V] {
-	return &TripleKeyCache[K1, K2, K3, V]{
-		primaryIndex:   make(map[K1]V),
-		secondaryIndex: make(map[K2]map[K1]struct{}),
-		tertiaryIndex:  make(map[K3]map[K1]struct{}),
-		reverseLookup:  make(map[K1]keyPair[K2, K3]),
-	}
-}
-
-// Get retrieves a value from the cache using the primary key
-func (c *TripleKeyCache[K1, K2, K3, V]) Get(ctx context.Context, primaryKey K1) (V, bool) {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	value, ok := c.primaryIndex[primaryKey]
-	return value, ok
-}
-
-// Set stores a value in the cache with primary, secondary, and tertiary keys
-func (c *TripleKeyCache[K1, K2, K3, V]) Set(ctx context.Context, primaryKey K1, secondaryKey K2, tertiaryKey K3, value V) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	if oldKeys, exists := c.reverseLookup[primaryKey]; exists {
-		if primaryKeys, ok := c.secondaryIndex[oldKeys.secondary]; ok {
-			delete(primaryKeys, primaryKey)
-			if len(primaryKeys) == 0 {
-				delete(c.secondaryIndex, oldKeys.secondary)
-			}
-		}
-		if primaryKeys, ok := c.tertiaryIndex[oldKeys.tertiary]; ok {
-			delete(primaryKeys, primaryKey)
-			if len(primaryKeys) == 0 {
-				delete(c.tertiaryIndex, oldKeys.tertiary)
-			}
-		}
-	}
-
-	c.primaryIndex[primaryKey] = value
-	c.reverseLookup[primaryKey] = keyPair[K2, K3]{
-		secondary: secondaryKey,
-		tertiary:  tertiaryKey,
-	}
-
-	if _, exists := c.secondaryIndex[secondaryKey]; !exists {
-		c.secondaryIndex[secondaryKey] = make(map[K1]struct{})
-	}
-	c.secondaryIndex[secondaryKey][primaryKey] = struct{}{}
-
-	if _, exists := c.tertiaryIndex[tertiaryKey]; !exists {
-		c.tertiaryIndex[tertiaryKey] = make(map[K1]struct{})
-	}
-	c.tertiaryIndex[tertiaryKey][primaryKey] = struct{}{}
-}
-
-// InvalidateByPrimaryKey removes an entry using the primary key
-func (c *TripleKeyCache[K1, K2, K3, V]) InvalidateByPrimaryKey(ctx context.Context, primaryKey K1) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	if keys, exists := c.reverseLookup[primaryKey]; exists {
-		if primaryKeys, ok := c.secondaryIndex[keys.secondary]; ok {
-			delete(primaryKeys, primaryKey)
-			if len(primaryKeys) == 0 {
-				delete(c.secondaryIndex, keys.secondary)
-			}
-		}
-		if primaryKeys, ok := c.tertiaryIndex[keys.tertiary]; ok {
-			delete(primaryKeys, primaryKey)
-			if len(primaryKeys) == 0 {
-				delete(c.tertiaryIndex, keys.tertiary)
-			}
-		}
-		delete(c.reverseLookup, primaryKey)
-	}
-
-	delete(c.primaryIndex, primaryKey)
-}
-
-// InvalidateBySecondaryKey removes all entries with the given secondary key
-func (c *TripleKeyCache[K1, K2, K3, V]) InvalidateBySecondaryKey(ctx context.Context, secondaryKey K2) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	primaryKeys, exists := c.secondaryIndex[secondaryKey]
-	if !exists {
-		return
-	}
-
-	for primaryKey := range primaryKeys {
-		if keys, ok := c.reverseLookup[primaryKey]; ok {
-			if tertiaryPrimaryKeys, exists := c.tertiaryIndex[keys.tertiary]; exists {
-				delete(tertiaryPrimaryKeys, primaryKey)
-				if len(tertiaryPrimaryKeys) == 0 {
-					delete(c.tertiaryIndex, keys.tertiary)
-				}
-			}
-		}
-		delete(c.primaryIndex, primaryKey)
-		delete(c.reverseLookup, primaryKey)
-	}
-
-	delete(c.secondaryIndex, secondaryKey)
-}
-
-// InvalidateByTertiaryKey removes all entries with the given tertiary key
-func (c *TripleKeyCache[K1, K2, K3, V]) InvalidateByTertiaryKey(ctx context.Context, tertiaryKey K3) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	primaryKeys, exists := c.tertiaryIndex[tertiaryKey]
-	if !exists {
-		return
-	}
-
-	for primaryKey := range primaryKeys {
-		if keys, ok := c.reverseLookup[primaryKey]; ok {
-			if secondaryPrimaryKeys, exists := c.secondaryIndex[keys.secondary]; exists {
-				delete(secondaryPrimaryKeys, primaryKey)
-				if len(secondaryPrimaryKeys) == 0 {
-					delete(c.secondaryIndex, keys.secondary)
-				}
-			}
-		}
-		delete(c.primaryIndex, primaryKey)
-		delete(c.reverseLookup, primaryKey)
-	}
-
-	delete(c.tertiaryIndex, tertiaryKey)
-}
-
-// InvalidateAll removes all entries from the cache
-func (c *TripleKeyCache[K1, K2, K3, V]) InvalidateAll(ctx context.Context) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.primaryIndex = make(map[K1]V)
-	c.secondaryIndex = make(map[K2]map[K1]struct{})
-	c.tertiaryIndex = make(map[K3]map[K1]struct{})
-	c.reverseLookup = make(map[K1]keyPair[K2, K3])
-}
-
-// Size returns the number of entries in the cache
-func (c *TripleKeyCache[K1, K2, K3, V]) Size() int {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	return len(c.primaryIndex)
-}
-
-// GetOrSet retrieves a value from the cache, or sets it using the provided function if not found
-// The loadFunc should return the value, secondary key, and tertiary key (extracted from the value)
-func (c *TripleKeyCache[K1, K2, K3, V]) GetOrSet(ctx context.Context, primaryKey K1, loadFunc func() (V, K2, K3, error)) (V, error) {
-	if value, ok := c.Get(ctx, primaryKey); ok {
-		return value, nil
-	}
-
-	value, secondaryKey, tertiaryKey, err := loadFunc()
-	if err != nil {
-		var zero V
-		return zero, err
-	}
-
-	c.Set(ctx, primaryKey, secondaryKey, tertiaryKey, value)
-
-	return value, nil
-}
-
-// GetOrSetBySecondaryKey retrieves a value from the cache using the secondary key, or sets it using the provided function if not found
-// The loadFunc should return the value, primary key, secondary key, and tertiary key
-func (c *TripleKeyCache[K1, K2, K3, V]) GetOrSetBySecondaryKey(ctx context.Context, secondaryKey K2, loadFunc func() (V, K1, K3, error)) (V, error) {
-	c.mu.RLock()
-	if primaryKeys, exists := c.secondaryIndex[secondaryKey]; exists && len(primaryKeys) > 0 {
-		for primaryKey := range primaryKeys {
-			if value, ok := c.primaryIndex[primaryKey]; ok {
-				c.mu.RUnlock()
-				return value, nil
-			}
-		}
-	}
-	c.mu.RUnlock()
-
-	value, primaryKey, tertiaryKey, err := loadFunc()
-	if err != nil {
-		var zero V
-		return zero, err
-	}
-
-	c.Set(ctx, primaryKey, secondaryKey, tertiaryKey, value)
-
-	return value, nil
-}
-
-// GetOrSetByTertiaryKey retrieves a value from the cache using the tertiary key, or sets it using the provided function if not found
-// The loadFunc should return the value, primary key, secondary key, and tertiary key
-func (c *TripleKeyCache[K1, K2, K3, V]) GetOrSetByTertiaryKey(ctx context.Context, tertiaryKey K3, loadFunc func() (V, K1, K2, error)) (V, error) {
-	c.mu.RLock()
-	if primaryKeys, exists := c.tertiaryIndex[tertiaryKey]; exists && len(primaryKeys) > 0 {
-		for primaryKey := range primaryKeys {
-			if value, ok := c.primaryIndex[primaryKey]; ok {
-				c.mu.RUnlock()
-				return value, nil
-			}
-		}
-	}
-	c.mu.RUnlock()
-
-	value, primaryKey, secondaryKey, err := loadFunc()
-	if err != nil {
-		var zero V
-		return zero, err
-	}
-
-	c.Set(ctx, primaryKey, secondaryKey, tertiaryKey, value)
-
-	return value, nil
-}

management/server/telemetry/store_metrics.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"time"
 
-	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 )
 
@@ -15,8 +14,6 @@ type StoreMetrics struct {
 	persistenceDurationMicro           metric.Int64Histogram
 	persistenceDurationMs              metric.Int64Histogram
 	transactionDurationMs              metric.Int64Histogram
-	queryDurationMs                    metric.Int64Histogram
-	queryCounter                       metric.Int64Counter
 	ctx                                context.Context
 }
 
@@ -62,29 +59,12 @@ func NewStoreMetrics(ctx context.Context, meter metric.Meter) (*StoreMetrics, er
 		return nil, err
 	}
 
-	queryDurationMs, err := meter.Int64Histogram("management.store.query.duration.ms",
-		metric.WithUnit("milliseconds"),
-		metric.WithDescription("Duration of database query operations with operation type and table name"),
-	)
-	if err != nil {
-		return nil, err
-	}
-
-	queryCounter, err := meter.Int64Counter("management.store.query.count",
-		metric.WithDescription("Count of database query operations with operation type, table name, and status"),
-	)
-	if err != nil {
-		return nil, err
-	}
-
 	return &StoreMetrics{
 		globalLockAcquisitionDurationMicro: globalLockAcquisitionDurationMicro,
 		globalLockAcquisitionDurationMs:    globalLockAcquisitionDurationMs,
 		persistenceDurationMicro:           persistenceDurationMicro,
 		persistenceDurationMs:              persistenceDurationMs,
 		transactionDurationMs:              transactionDurationMs,
-		queryDurationMs:                    queryDurationMs,
-		queryCounter:                       queryCounter,
 		ctx:                                ctx,
 	}, nil
 }
@@ -105,13 +85,3 @@ func (metrics *StoreMetrics) CountPersistenceDuration(duration time.Duration) {
 func (metrics *StoreMetrics) CountTransactionDuration(duration time.Duration) {
 	metrics.transactionDurationMs.Record(metrics.ctx, duration.Milliseconds())
 }
-
-// CountStoreOperation records a store operation with its method name, status, and duration
-func (metrics *StoreMetrics) CountStoreOperation(method string, duration time.Duration) {
-	attrs := []attribute.KeyValue{
-		attribute.String("method", method),
-	}
-
-	metrics.queryDurationMs.Record(metrics.ctx, duration.Milliseconds(), metric.WithAttributes(attrs...))
-	metrics.queryCounter.Add(metrics.ctx, 1, metric.WithAttributes(attrs...))
-}

signal/loadtest/README.md

@@ -0,0 +1,381 @@
+# Signal Server Load Test
+
+Load testing tool for the NetBird signal server.
+
+## Features
+
+- **Rate-based peer pair creation**: Spawn peer pairs at configurable rates (e.g., 10, 20 pairs/sec)
+- **Two exchange modes**:
+  - **Single message**: Each pair exchanges one message for validation
+  - **Continuous exchange**: Pairs continuously exchange messages for a specified duration (e.g., 30 seconds, 10 minutes)
+- **TLS/HTTPS support**: Connect to TLS-enabled signal servers with optional certificate verification
+- **Automatic reconnection**: Optional automatic reconnection with exponential backoff on connection loss
+- **Configurable message interval**: Control message send rate in continuous mode
+- **Message exchange validation**: Validates encrypted body size > 0
+- **Comprehensive metrics**: Tracks throughput, success/failure rates, latency statistics, and reconnection counts
+- **Local server testing**: Tests include embedded signal server for easy development
+- **Worker pool pattern**: Efficient concurrent execution
+- **Graceful shutdown**: Context-based cancellation
+
+## Usage
+
+### Standalone Binary
+
+Build and run the load test as a standalone binary:
+
+```bash
+# Build the binary
+cd signal/loadtest/cmd/signal-loadtest
+go build -o signal-loadtest
+
+# Single message exchange
+./signal-loadtest \
+  -server http://localhost:10000 \
+  -pairs-per-sec 10 \
+  -total-pairs 100 \
+  -message-size 100
+
+# Continuous exchange for 30 seconds
+./signal-loadtest \
+  -server http://localhost:10000 \
+  -pairs-per-sec 10 \
+  -total-pairs 20 \
+  -message-size 200 \
+  -exchange-duration 30s \
+  -message-interval 200ms
+
+# Long-running test (10 minutes)
+./signal-loadtest \
+  -server http://localhost:10000 \
+  -pairs-per-sec 20 \
+  -total-pairs 50 \
+  -message-size 500 \
+  -exchange-duration 10m \
+  -message-interval 100ms \
+  -test-duration 15m \
+  -log-level debug
+
+# TLS server with valid certificate
+./signal-loadtest \
+  -server https://signal.example.com:443 \
+  -pairs-per-sec 10 \
+  -total-pairs 50 \
+  -message-size 100
+
+# TLS server with self-signed certificate
+./signal-loadtest \
+  -server https://localhost:443 \
+  -pairs-per-sec 5 \
+  -total-pairs 10 \
+  -insecure-skip-verify \
+  -log-level debug
+
+# High load test with custom worker pool
+./signal-loadtest \
+  -server http://localhost:10000 \
+  -pairs-per-sec 100 \
+  -total-pairs 1000 \
+  -worker-pool-size 500 \
+  -channel-buffer-size 1000 \
+  -exchange-duration 60s \
+  -log-level info
+
+# Progress reporting - report every 5000 messages
+./signal-loadtest \
+  -server http://localhost:10000 \
+  -pairs-per-sec 50 \
+  -total-pairs 100 \
+  -exchange-duration 5m \
+  -report-interval 5000 \
+  -log-level info
+
+# With automatic reconnection
+./signal-loadtest \
+  -server http://localhost:10000 \
+  -pairs-per-sec 10 \
+  -total-pairs 50 \
+  -exchange-duration 5m \
+  -enable-reconnect \
+  -initial-retry-delay 100ms \
+  -max-reconnect-delay 30s \
+  -log-level debug
+
+# Show help
+./signal-loadtest -h
+```
+
+**Graceful Shutdown:**
+
+The load test supports graceful shutdown via Ctrl+C (SIGINT/SIGTERM):
+- Press Ctrl+C to interrupt the test at any time
+- All active clients will be closed gracefully
+- A final aggregated report will be printed showing metrics up to the point of interruption
+- Shutdown timeout: 5 seconds (after which the process will force exit)
+
+**Available Flags:**
+- `-server`: Signal server URL (http:// or https://) (default: `http://localhost:10000`)
+- `-pairs-per-sec`: Peer pairs created per second (default: 10)
+- `-total-pairs`: Total number of peer pairs (default: 100)
+- `-message-size`: Message size in bytes (default: 100)
+- `-test-duration`: Maximum test duration, 0 = unlimited (default: 0)
+- `-exchange-duration`: Continuous exchange duration per pair, 0 = single message (default: 0)
+- `-message-interval`: Interval between messages in continuous mode (default: 100ms)
+- `-worker-pool-size`: Number of concurrent workers, 0 = auto (pairs-per-sec × 2) (default: 0)
+- `-channel-buffer-size`: Work queue buffer size, 0 = auto (pairs-per-sec × 4) (default: 0)
+- `-report-interval`: Report progress every N messages, 0 = no periodic reports (default: 10000)
+- `-enable-reconnect`: Enable automatic reconnection on connection loss (default: false)
+- `-initial-retry-delay`: Initial delay before first reconnection attempt (default: 100ms)
+- `-max-reconnect-delay`: Maximum delay between reconnection attempts (default: 30s)
+- `-insecure-skip-verify`: Skip TLS certificate verification for self-signed certificates (default: false)
+- `-log-level`: Log level: trace, debug, info, warn, error (default: info)
+
+### Running Tests
+
+```bash
+# Run all tests (includes load tests)
+go test -v -timeout 2m
+
+# Run specific single-message load tests
+go test -v -run TestLoadTest_10PairsPerSecond -timeout 40s
+go test -v -run TestLoadTest_20PairsPerSecond -timeout 40s
+go test -v -run TestLoadTest_SmallBurst -timeout 30s
+
+# Run continuous exchange tests
+go test -v -run TestLoadTest_ContinuousExchange_ShortBurst -timeout 30s
+go test -v -run TestLoadTest_ContinuousExchange_30Seconds -timeout 2m
+go test -v -run TestLoadTest_ContinuousExchange_10Minutes -timeout 15m
+
+# Skip long-running tests in quick runs
+go test -short
+```
+
+### Programmatic Usage
+
+#### Single Message Exchange
+```go
+package main
+
+import (
+    "github.com/netbirdio/netbird/signal/loadtest"
+    "time"
+)
+
+func main() {
+    config := loadtest.LoadTestConfig{
+        ServerURL:      "http://localhost:10000",
+        PairsPerSecond: 10,
+        TotalPairs:     100,
+        MessageSize:    100,
+        TestDuration:   30 * time.Second,
+    }
+
+    lt := loadtest.NewLoadTest(config)
+    if err := lt.Run(); err != nil {
+        panic(err)
+    }
+
+    metrics := lt.GetMetrics()
+    metrics.PrintReport()
+}
+```
+
+#### Continuous Message Exchange
+```go
+package main
+
+import (
+    "github.com/netbirdio/netbird/signal/loadtest"
+    "time"
+)
+
+func main() {
+    config := loadtest.LoadTestConfig{
+        ServerURL:        "http://localhost:10000",
+        PairsPerSecond:   10,
+        TotalPairs:       20,
+        MessageSize:      200,
+        ExchangeDuration: 10 * time.Minute,  // Each pair exchanges messages for 10 minutes
+        MessageInterval:  200 * time.Millisecond,  // Send message every 200ms
+        TestDuration:     15 * time.Minute,  // Overall test timeout
+    }
+
+    lt := loadtest.NewLoadTest(config)
+    if err := lt.Run(); err != nil {
+        panic(err)
+    }
+
+    metrics := lt.GetMetrics()
+    metrics.PrintReport()
+}
+```
+
+## Configuration Options
+
+- **ServerURL**: Signal server URL (e.g., `http://localhost:10000` or `https://signal.example.com:443`)
+- **PairsPerSecond**: Rate at which peer pairs are created (e.g., 10, 20)
+- **TotalPairs**: Total number of peer pairs to create
+- **MessageSize**: Size of test message payload in bytes
+- **TestDuration**: Maximum test duration (optional, 0 = no limit)
+- **ExchangeDuration**: Duration for continuous message exchange per pair (0 = single message)
+- **MessageInterval**: Interval between messages in continuous mode (default: 100ms)
+- **WorkerPoolSize**: Number of concurrent worker goroutines (0 = auto: pairs-per-sec × 2)
+- **ChannelBufferSize**: Work queue buffer size (0 = auto: pairs-per-sec × 4)
+- **ReportInterval**: Report progress every N messages (0 = no periodic reports, default: 10000)
+- **EnableReconnect**: Enable automatic reconnection on connection loss (default: false)
+- **InitialRetryDelay**: Initial delay before first reconnection attempt (default: 100ms)
+- **MaxReconnectDelay**: Maximum delay between reconnection attempts (default: 30s)
+- **InsecureSkipVerify**: Skip TLS certificate verification (for self-signed certificates)
+- **RampUpDuration**: Gradual ramp-up period (not yet implemented)
+
+### Reconnection Handling
+
+The load test supports automatic reconnection on connection loss:
+
+- **Disabled by default**: Connections will fail on any network interruption
+- **When enabled**: Clients automatically reconnect with exponential backoff
+- **Exponential backoff**: Starts at `InitialRetryDelay`, doubles on each failure, caps at `MaxReconnectDelay`
+- **Transparent reconnection**: Message exchange continues after successful reconnection
+- **Metrics tracking**: Total reconnection count is reported
+
+**Use cases:**
+- Testing resilience to network interruptions
+- Validating server restart behavior
+- Simulating flaky network conditions
+- Long-running stability tests
+
+**Example with reconnection:**
+```go
+config := loadtest.LoadTestConfig{
+    ServerURL:         "http://localhost:10000",
+    PairsPerSecond:    10,
+    TotalPairs:        20,
+    ExchangeDuration:  10 * time.Minute,
+    EnableReconnect:   true,
+    InitialRetryDelay: 100 * time.Millisecond,
+    MaxReconnectDelay: 30 * time.Second,
+}
+```
+
+### Performance Tuning
+
+When running high-load tests, you may need to adjust the worker pool and buffer sizes:
+
+- **Default sizing**: Auto-configured based on `PairsPerSecond`
+  - Worker pool: `PairsPerSecond × 2`
+  - Channel buffer: `PairsPerSecond × 4`
+- **For continuous exchange**: Increase worker pool size (e.g., `PairsPerSecond × 5`)
+- **For high pair rates** (>50/sec): Increase both worker pool and buffer proportionally
+- **Signs you need more workers**: Log warnings about "Worker pool saturated"
+
+Example for 100 pairs/sec with continuous exchange:
+```go
+config := LoadTestConfig{
+    PairsPerSecond:    100,
+    WorkerPoolSize:    500,  // 5x pairs/sec
+    ChannelBufferSize: 1000, // 10x pairs/sec
+}
+```
+
+## Metrics
+
+The load test collects and reports:
+
+- **Total Pairs Sent**: Number of peer pairs attempted
+- **Successful Exchanges**: Completed message exchanges
+- **Failed Exchanges**: Failed message exchanges
+- **Total Messages Exchanged**: Count of successfully exchanged messages
+- **Total Errors**: Cumulative error count
+- **Total Reconnections**: Number of automatic reconnections (if enabled)
+- **Throughput**: Pairs per second (actual)
+- **Latency Statistics**: Min, Max, Avg message exchange latency
+
+## Graceful Shutdown Example
+
+You can interrupt a long-running test at any time with Ctrl+C:
+
+```
+./signal-loadtest -server http://localhost:10000 -pairs-per-sec 10 -total-pairs 100 -exchange-duration 10m
+
+# Press Ctrl+C after some time...
+^C
+WARN[0045]
+Received interrupt signal, shutting down gracefully...
+
+=== Load Test Report ===
+Test Duration: 45.234s
+Total Pairs Sent: 75
+Successful Exchanges: 75
+Failed Exchanges: 0
+Total Messages Exchanged: 22500
+Total Errors: 0
+Throughput: 1.66 pairs/sec
+...
+========================
+```
+
+## Test Results
+
+Example output from a 20 pairs/sec test:
+
+```
+=== Load Test Report ===
+Test Duration: 5.055249917s
+Total Pairs Sent: 100
+Successful Exchanges: 100
+Failed Exchanges: 0
+Total Messages Exchanged: 100
+Total Errors: 0
+Throughput: 19.78 pairs/sec
+
+Latency Statistics:
+  Min: 170.375µs
+  Max: 5.176916ms
+  Avg: 441.566µs
+========================
+```
+
+## Architecture
+
+### Client (`client.go`)
+- Manages gRPC connection to signal server
+- Establishes bidirectional stream for receiving messages
+- Sends messages via `Send` RPC method
+- Handles message reception asynchronously
+
+### Load Test Engine (`rate_loadtest.go`)
+- Worker pool pattern for concurrent peer pairs
+- Rate-limited pair creation using ticker
+- Atomic counters for thread-safe metrics collection
+- Graceful shutdown on context cancellation
+
+### Test Suite
+- `loadtest_test.go`: Single pair validation test
+- `rate_loadtest_test.go`: Multiple rate-based load tests and benchmarks
+
+## Implementation Details
+
+### Message Flow
+1. Create sender and receiver clients with unique IDs
+2. Both clients connect to signal server via bidirectional stream
+3. Sender sends encrypted message using `Send` RPC
+4. Signal server forwards message to receiver's stream
+5. Receiver reads message from stream
+6. Validate encrypted body size > 0
+7. Record latency and success metrics
+
+### Concurrency
+- Worker pool size = `PairsPerSecond`
+- Each worker handles multiple peer pairs sequentially
+- Atomic operations for metrics to avoid lock contention
+- Channel-based work distribution
+
+## Future Enhancements
+
+- [x] TLS/HTTPS support for production servers
+- [x] Automatic reconnection with exponential backoff
+- [ ] Ramp-up period implementation
+- [ ] Percentile latency metrics (p50, p95, p99)
+- [ ] Connection reuse for multiple messages per pair
+- [ ] Support for custom message payloads
+- [ ] CSV/JSON metrics export
+- [ ] Real-time metrics dashboard

signal/loadtest/client.go

@@ -0,0 +1,301 @@
+package loadtest
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
+
+	"github.com/netbirdio/netbird/shared/signal/proto"
+)
+
+// Client represents a signal client for load testing
+type Client struct {
+	id         string
+	serverURL  string
+	config     *ClientConfig
+	conn       *grpc.ClientConn
+	client     proto.SignalExchangeClient
+	stream     proto.SignalExchange_ConnectStreamClient
+	ctx        context.Context
+	cancel     context.CancelFunc
+	msgChannel chan *proto.EncryptedMessage
+
+	mu              sync.RWMutex
+	reconnectCount  int64
+	connected       bool
+	receiverStarted bool
+}
+
+// ClientConfig holds optional configuration for the client
+type ClientConfig struct {
+	InsecureSkipVerify bool
+	EnableReconnect    bool
+	MaxReconnectDelay  time.Duration
+	InitialRetryDelay  time.Duration
+}
+
+// NewClient creates a new signal client for load testing
+func NewClient(serverURL, peerID string) (*Client, error) {
+	return NewClientWithConfig(serverURL, peerID, nil)
+}
+
+// NewClientWithConfig creates a new signal client with custom TLS configuration
+func NewClientWithConfig(serverURL, peerID string, config *ClientConfig) (*Client, error) {
+	if config == nil {
+		config = &ClientConfig{}
+	}
+
+	// Set default reconnect delays if not specified
+	if config.EnableReconnect {
+		if config.InitialRetryDelay == 0 {
+			config.InitialRetryDelay = 100 * time.Millisecond
+		}
+		if config.MaxReconnectDelay == 0 {
+			config.MaxReconnectDelay = 30 * time.Second
+		}
+	}
+
+	addr, opts, err := parseServerURL(serverURL, config.InsecureSkipVerify)
+	if err != nil {
+		return nil, fmt.Errorf("parse server URL: %w", err)
+	}
+
+	conn, err := grpc.Dial(addr, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("dial server: %w", err)
+	}
+
+	client := proto.NewSignalExchangeClient(conn)
+	ctx, cancel := context.WithCancel(context.Background())
+
+	return &Client{
+		id:         peerID,
+		serverURL:  serverURL,
+		config:     config,
+		conn:       conn,
+		client:     client,
+		ctx:        ctx,
+		cancel:     cancel,
+		msgChannel: make(chan *proto.EncryptedMessage, 10),
+		connected:  false,
+	}, nil
+}
+
+// Connect establishes a stream connection to the signal server
+func (c *Client) Connect() error {
+	md := metadata.New(map[string]string{proto.HeaderId: c.id})
+	ctx := metadata.NewOutgoingContext(c.ctx, md)
+
+	stream, err := c.client.ConnectStream(ctx)
+	if err != nil {
+		return fmt.Errorf("connect stream: %w", err)
+	}
+
+	if _, err := stream.Header(); err != nil {
+		return fmt.Errorf("receive header: %w", err)
+	}
+
+	c.mu.Lock()
+	c.stream = stream
+	c.connected = true
+	if !c.receiverStarted {
+		c.receiverStarted = true
+		c.mu.Unlock()
+		go c.receiveMessages()
+	} else {
+		c.mu.Unlock()
+	}
+
+	return nil
+}
+
+// reconnectStream reconnects the stream without starting a new receiver goroutine
+func (c *Client) reconnectStream() error {
+	if !c.config.EnableReconnect {
+		return fmt.Errorf("reconnect disabled")
+	}
+
+	delay := c.config.InitialRetryDelay
+	attempt := 0
+
+	for {
+		select {
+		case <-c.ctx.Done():
+			return c.ctx.Err()
+		case <-time.After(delay):
+			attempt++
+			log.Debugf("Client %s reconnect attempt %d (delay: %v)", c.id, attempt, delay)
+
+			md := metadata.New(map[string]string{proto.HeaderId: c.id})
+			ctx := metadata.NewOutgoingContext(c.ctx, md)
+
+			stream, err := c.client.ConnectStream(ctx)
+			if err != nil {
+				log.Debugf("Client %s reconnect attempt %d failed: %v", c.id, attempt, err)
+				delay *= 2
+				if delay > c.config.MaxReconnectDelay {
+					delay = c.config.MaxReconnectDelay
+				}
+				continue
+			}
+
+			if _, err := stream.Header(); err != nil {
+				log.Debugf("Client %s reconnect header failed: %v", c.id, err)
+				delay *= 2
+				if delay > c.config.MaxReconnectDelay {
+					delay = c.config.MaxReconnectDelay
+				}
+				continue
+			}
+
+			c.mu.Lock()
+			c.stream = stream
+			c.connected = true
+			c.reconnectCount++
+			c.mu.Unlock()
+
+			log.Debugf("Client %s reconnected successfully (attempt %d, total reconnects: %d)",
+				c.id, attempt, c.reconnectCount)
+			return nil
+		}
+	}
+}
+
+// SendMessage sends an encrypted message to a remote peer using the Send RPC
+func (c *Client) SendMessage(remotePeerID string, body []byte) error {
+	msg := &proto.EncryptedMessage{
+		Key:       c.id,
+		RemoteKey: remotePeerID,
+		Body:      body,
+	}
+
+	ctx, cancel := context.WithTimeout(c.ctx, 10*time.Second)
+	defer cancel()
+
+	_, err := c.client.Send(ctx, msg)
+	if err != nil {
+		return fmt.Errorf("send message: %w", err)
+	}
+
+	return nil
+}
+
+// ReceiveMessage waits for and returns the next message
+func (c *Client) ReceiveMessage() (*proto.EncryptedMessage, error) {
+	select {
+	case msg := <-c.msgChannel:
+		return msg, nil
+	case <-c.ctx.Done():
+		return nil, c.ctx.Err()
+	}
+}
+
+// Close closes the client connection
+func (c *Client) Close() error {
+	c.cancel()
+	if c.conn != nil {
+		return c.conn.Close()
+	}
+	return nil
+}
+
+func (c *Client) receiveMessages() {
+	for {
+		c.mu.RLock()
+		stream := c.stream
+		c.mu.RUnlock()
+
+		if stream == nil {
+			return
+		}
+
+		msg, err := stream.Recv()
+		if err != nil {
+			// Check if context is cancelled before attempting reconnection
+			select {
+			case <-c.ctx.Done():
+				return
+			default:
+			}
+
+			c.mu.Lock()
+			c.connected = false
+			c.mu.Unlock()
+
+			log.Debugf("Client %s receive error: %v", c.id, err)
+
+			// Attempt reconnection if enabled
+			if c.config.EnableReconnect {
+				if reconnectErr := c.reconnectStream(); reconnectErr != nil {
+					log.Debugf("Client %s reconnection failed: %v", c.id, reconnectErr)
+					return
+				}
+				// Successfully reconnected, continue receiving
+				continue
+			}
+
+			// Reconnect disabled, exit
+			return
+		}
+
+		select {
+		case c.msgChannel <- msg:
+		case <-c.ctx.Done():
+			return
+		}
+	}
+}
+
+// IsConnected returns whether the client is currently connected
+func (c *Client) IsConnected() bool {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.connected
+}
+
+// GetReconnectCount returns the number of reconnections
+func (c *Client) GetReconnectCount() int64 {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.reconnectCount
+}
+
+func parseServerURL(serverURL string, insecureSkipVerify bool) (string, []grpc.DialOption, error) {
+	serverURL = strings.TrimSpace(serverURL)
+	if serverURL == "" {
+		return "", nil, fmt.Errorf("server URL is empty")
+	}
+
+	var addr string
+	var opts []grpc.DialOption
+
+	if strings.HasPrefix(serverURL, "https://") {
+		addr = strings.TrimPrefix(serverURL, "https://")
+		tlsConfig := &tls.Config{
+			MinVersion:         tls.VersionTLS12,
+			InsecureSkipVerify: insecureSkipVerify,
+		}
+		opts = append(opts, grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)))
+	} else if strings.HasPrefix(serverURL, "http://") {
+		addr = strings.TrimPrefix(serverURL, "http://")
+		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	} else {
+		addr = serverURL
+		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	}
+
+	if !strings.Contains(addr, ":") {
+		return "", nil, fmt.Errorf("server URL must include port")
+	}
+
+	return addr, opts, nil
+}

signal/loadtest/cmd/signal-loadtest/integration_test.go

@@ -0,0 +1,128 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"os/exec"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel"
+	"google.golang.org/grpc"
+
+	"github.com/netbirdio/netbird/shared/signal/proto"
+	"github.com/netbirdio/netbird/signal/server"
+)
+
+func TestCLI_SingleMessage(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grpcServer, serverAddr := startTestSignalServer(t, ctx)
+	defer grpcServer.Stop()
+
+	cmd := exec.Command("go", "run", "main.go",
+		"-server", serverAddr,
+		"-pairs-per-sec", "3",
+		"-total-pairs", "5",
+		"-message-size", "50",
+		"-log-level", "warn")
+
+	output, err := cmd.CombinedOutput()
+	require.NoError(t, err, "CLI should execute successfully")
+
+	outputStr := string(output)
+	require.Contains(t, outputStr, "Load Test Report")
+	require.Contains(t, outputStr, "Total Pairs Sent: 5")
+	require.Contains(t, outputStr, "Successful Exchanges: 5")
+	t.Logf("Output:\n%s", outputStr)
+}
+
+func TestCLI_ContinuousExchange(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping continuous exchange CLI test in short mode")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grpcServer, serverAddr := startTestSignalServer(t, ctx)
+	defer grpcServer.Stop()
+
+	cmd := exec.Command("go", "run", "main.go",
+		"-server", serverAddr,
+		"-pairs-per-sec", "2",
+		"-total-pairs", "3",
+		"-message-size", "100",
+		"-exchange-duration", "3s",
+		"-message-interval", "100ms",
+		"-log-level", "warn")
+
+	output, err := cmd.CombinedOutput()
+	require.NoError(t, err, "CLI should execute successfully")
+
+	outputStr := string(output)
+	require.Contains(t, outputStr, "Load Test Report")
+	require.Contains(t, outputStr, "Total Pairs Sent: 3")
+	require.Contains(t, outputStr, "Successful Exchanges: 3")
+	t.Logf("Output:\n%s", outputStr)
+}
+
+func TestCLI_InvalidConfig(t *testing.T) {
+	tests := []struct {
+		name string
+		args []string
+	}{
+		{
+			name: "negative pairs",
+			args: []string{"-pairs-per-sec", "-1"},
+		},
+		{
+			name: "zero total pairs",
+			args: []string{"-total-pairs", "0"},
+		},
+		{
+			name: "negative message size",
+			args: []string{"-message-size", "-100"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			args := append([]string{"run", "main.go"}, tt.args...)
+			cmd := exec.Command("go", args...)
+			output, err := cmd.CombinedOutput()
+			require.Error(t, err, "Should fail with invalid config")
+			require.Contains(t, string(output), "Configuration error")
+		})
+	}
+}
+
+func startTestSignalServer(t *testing.T, ctx context.Context) (*grpc.Server, string) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	grpcServer := grpc.NewServer()
+
+	signalServer, err := server.NewServer(ctx, otel.Meter("cli-test"))
+	require.NoError(t, err)
+
+	proto.RegisterSignalExchangeServer(grpcServer, signalServer)
+
+	go func() {
+		if err := grpcServer.Serve(listener); err != nil {
+			t.Logf("Server stopped: %v", err)
+		}
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+
+	return grpcServer, fmt.Sprintf("http://%s", listener.Addr().String())
+}
+
+func TestMain(m *testing.M) {
+	os.Exit(m.Run())
+}

signal/loadtest/cmd/signal-loadtest/main.go

@@ -0,0 +1,165 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/signal/loadtest"
+)
+
+var (
+	serverURL          string
+	pairsPerSecond     int
+	totalPairs         int
+	messageSize        int
+	testDuration       time.Duration
+	exchangeDuration   time.Duration
+	messageInterval    time.Duration
+	insecureSkipVerify bool
+	workerPoolSize     int
+	channelBufferSize  int
+	reportInterval     int
+	logLevel           string
+	enableReconnect    bool
+	maxReconnectDelay  time.Duration
+	initialRetryDelay  time.Duration
+)
+
+func init() {
+	flag.StringVar(&serverURL, "server", "http://localhost:10000", "Signal server URL (http:// or https://)")
+	flag.IntVar(&pairsPerSecond, "pairs-per-sec", 10, "Number of peer pairs to create per second")
+	flag.IntVar(&totalPairs, "total-pairs", 100, "Total number of peer pairs to create")
+	flag.IntVar(&messageSize, "message-size", 100, "Size of test message in bytes")
+	flag.DurationVar(&testDuration, "test-duration", 0, "Maximum test duration (0 = unlimited)")
+	flag.DurationVar(&exchangeDuration, "exchange-duration", 0, "Duration for continuous message exchange per pair (0 = single message)")
+	flag.DurationVar(&messageInterval, "message-interval", 100*time.Millisecond, "Interval between messages in continuous mode")
+	flag.BoolVar(&insecureSkipVerify, "insecure-skip-verify", false, "Skip TLS certificate verification (use with self-signed certificates)")
+	flag.IntVar(&workerPoolSize, "worker-pool-size", 0, "Number of worker goroutines (0 = auto: pairs-per-sec * 2)")
+	flag.IntVar(&channelBufferSize, "channel-buffer-size", 0, "Channel buffer size (0 = auto: pairs-per-sec * 4)")
+	flag.IntVar(&reportInterval, "report-interval", 10000, "Report progress every N messages (0 = no periodic reports)")
+	flag.StringVar(&logLevel, "log-level", "info", "Log level (trace, debug, info, warn, error)")
+	flag.BoolVar(&enableReconnect, "enable-reconnect", true, "Enable automatic reconnection on connection loss")
+	flag.DurationVar(&maxReconnectDelay, "max-reconnect-delay", 30*time.Second, "Maximum delay between reconnection attempts")
+	flag.DurationVar(&initialRetryDelay, "initial-retry-delay", 100*time.Millisecond, "Initial delay before first reconnection attempt")
+}
+
+func main() {
+	flag.Parse()
+
+	level, err := log.ParseLevel(logLevel)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Invalid log level: %v\n", err)
+		os.Exit(1)
+	}
+	log.SetLevel(level)
+
+	config := loadtest.LoadTestConfig{
+		ServerURL:          serverURL,
+		PairsPerSecond:     pairsPerSecond,
+		TotalPairs:         totalPairs,
+		MessageSize:        messageSize,
+		TestDuration:       testDuration,
+		ExchangeDuration:   exchangeDuration,
+		MessageInterval:    messageInterval,
+		InsecureSkipVerify: insecureSkipVerify,
+		WorkerPoolSize:     workerPoolSize,
+		ChannelBufferSize:  channelBufferSize,
+		ReportInterval:     reportInterval,
+		EnableReconnect:    enableReconnect,
+		MaxReconnectDelay:  maxReconnectDelay,
+		InitialRetryDelay:  initialRetryDelay,
+	}
+
+	if err := validateConfig(config); err != nil {
+		fmt.Fprintf(os.Stderr, "Configuration error: %v\n", err)
+		flag.Usage()
+		os.Exit(1)
+	}
+
+	log.Infof("Signal Load Test Configuration:")
+	log.Infof("  Server URL: %s", config.ServerURL)
+	log.Infof("  Pairs per second: %d", config.PairsPerSecond)
+	log.Infof("  Total pairs: %d", config.TotalPairs)
+	log.Infof("  Message size: %d bytes", config.MessageSize)
+	if config.InsecureSkipVerify {
+		log.Warnf("  TLS certificate verification: DISABLED (insecure)")
+	}
+	if config.TestDuration > 0 {
+		log.Infof("  Test duration: %v", config.TestDuration)
+	}
+	if config.ExchangeDuration > 0 {
+		log.Infof("  Exchange duration: %v", config.ExchangeDuration)
+		log.Infof("  Message interval: %v", config.MessageInterval)
+	} else {
+		log.Infof("  Mode: Single message exchange")
+	}
+	if config.EnableReconnect {
+		log.Infof("  Reconnection: ENABLED")
+		log.Infof("  Initial retry delay: %v", config.InitialRetryDelay)
+		log.Infof("  Max reconnect delay: %v", config.MaxReconnectDelay)
+	}
+	fmt.Println()
+
+	// Set up signal handler for graceful shutdown
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
+
+	lt := loadtest.NewLoadTestWithContext(ctx, config)
+
+	// Run load test in a goroutine
+	done := make(chan error, 1)
+	go func() {
+		done <- lt.Run()
+	}()
+
+	// Wait for completion or signal
+	select {
+	case <-sigChan:
+		log.Warnf("\nReceived interrupt signal, shutting down gracefully...")
+		cancel()
+		// Wait a bit for graceful shutdown
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			log.Warnf("Shutdown timeout, forcing exit")
+		}
+	case err := <-done:
+		if err != nil && err != context.Canceled {
+			log.Errorf("Load test failed: %v", err)
+			os.Exit(1)
+		}
+	}
+
+	metrics := lt.GetMetrics()
+	fmt.Println() // Add blank line before report
+	metrics.PrintReport()
+}
+
+func validateConfig(config loadtest.LoadTestConfig) error {
+	if config.ServerURL == "" {
+		return fmt.Errorf("server URL is required")
+	}
+	if config.PairsPerSecond <= 0 {
+		return fmt.Errorf("pairs-per-sec must be greater than 0")
+	}
+	if config.TotalPairs <= 0 {
+		return fmt.Errorf("total-pairs must be greater than 0")
+	}
+	if config.MessageSize <= 0 {
+		return fmt.Errorf("message-size must be greater than 0")
+	}
+	if config.MessageInterval <= 0 {
+		return fmt.Errorf("message-interval must be greater than 0")
+	}
+	return nil
+}

signal/loadtest/cmd/signal-loadtest/test.sh

@@ -0,0 +1,40 @@
+#!/bin/bash
+set -e
+
+echo "Building signal-loadtest binary..."
+go build -o signal-loadtest
+
+echo ""
+echo "=== Test 1: Single message exchange (5 pairs) ==="
+./signal-loadtest \
+  -server http://localhost:10000 \
+  -pairs-per-sec 5 \
+  -total-pairs 5 \
+  -message-size 50 \
+  -log-level info
+
+echo ""
+echo "=== Test 2: Continuous exchange (3 pairs, 5 seconds) ==="
+./signal-loadtest \
+  -server http://localhost:10000 \
+  -pairs-per-sec 3 \
+  -total-pairs 3 \
+  -message-size 100 \
+  -exchange-duration 5s \
+  -message-interval 200ms \
+  -log-level info
+
+echo ""
+echo "=== Test 3: Progress reporting (10 pairs, 10s, report every 100 messages) ==="
+./signal-loadtest \
+  -server http://localhost:10000 \
+  -pairs-per-sec 10 \
+  -total-pairs 10 \
+  -message-size 100 \
+  -exchange-duration 10s \
+  -message-interval 100ms \
+  -report-interval 100 \
+  -log-level info
+
+echo ""
+echo "All tests completed!"

signal/loadtest/loadtest_test.go

@@ -0,0 +1,91 @@
+package loadtest
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel"
+	"google.golang.org/grpc"
+
+	"github.com/netbirdio/netbird/shared/signal/proto"
+	"github.com/netbirdio/netbird/signal/server"
+)
+
+func TestSignalLoadTest_SinglePair(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grpcServer, serverAddr := startTestSignalServer(t, ctx)
+	defer grpcServer.Stop()
+
+	sender, err := NewClient(serverAddr, "sender-peer-id")
+	require.NoError(t, err)
+	defer sender.Close()
+
+	receiver, err := NewClient(serverAddr, "receiver-peer-id")
+	require.NoError(t, err)
+	defer receiver.Close()
+
+	err = sender.Connect()
+	require.NoError(t, err, "Sender should connect successfully")
+
+	err = receiver.Connect()
+	require.NoError(t, err, "Receiver should connect successfully")
+
+	time.Sleep(100 * time.Millisecond)
+
+	testMessage := []byte("test message payload")
+
+	t.Log("Sending message from sender to receiver")
+	err = sender.SendMessage("receiver-peer-id", testMessage)
+	require.NoError(t, err, "Sender should send message successfully")
+
+	t.Log("Waiting for receiver to receive message")
+
+	receiveDone := make(chan struct{})
+	var msg *proto.EncryptedMessage
+	var receiveErr error
+
+	go func() {
+		msg, receiveErr = receiver.ReceiveMessage()
+		close(receiveDone)
+	}()
+
+	select {
+	case <-receiveDone:
+		require.NoError(t, receiveErr, "Receiver should receive message")
+		require.NotNil(t, msg, "Received message should not be nil")
+		require.Greater(t, len(msg.Body), 0, "Encrypted message body size should be greater than 0")
+		require.Equal(t, "sender-peer-id", msg.Key)
+		require.Equal(t, "receiver-peer-id", msg.RemoteKey)
+		t.Log("Message received successfully")
+	case <-time.After(5 * time.Second):
+		t.Fatal("Timeout waiting for message")
+	}
+}
+
+func startTestSignalServer(t *testing.T, ctx context.Context) (*grpc.Server, string) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	grpcServer := grpc.NewServer()
+
+	signalServer, err := server.NewServer(ctx, otel.Meter("test"))
+	require.NoError(t, err)
+
+	proto.RegisterSignalExchangeServer(grpcServer, signalServer)
+
+	go func() {
+		if err := grpcServer.Serve(listener); err != nil {
+			t.Logf("Server stopped: %v", err)
+		}
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+
+	return grpcServer, fmt.Sprintf("http://%s", listener.Addr().String())
+}

signal/loadtest/rate_loadtest.go

@@ -0,0 +1,461 @@
+package loadtest
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// LoadTestConfig configuration for the load test
+type LoadTestConfig struct {
+	IDPrefix           string
+	ServerURL          string
+	PairsPerSecond     int
+	TotalPairs         int
+	MessageSize        int
+	TestDuration       time.Duration
+	ExchangeDuration   time.Duration
+	MessageInterval    time.Duration
+	RampUpDuration     time.Duration
+	InsecureSkipVerify bool
+	WorkerPoolSize     int
+	ChannelBufferSize  int
+	ReportInterval     int // Report progress every N messages (0 = no periodic reports)
+	EnableReconnect    bool
+	MaxReconnectDelay  time.Duration
+	InitialRetryDelay  time.Duration
+}
+
+// LoadTestMetrics metrics collected during the load test
+type LoadTestMetrics struct {
+	TotalPairsSent         atomic.Int64
+	TotalMessagesExchanged atomic.Int64
+	TotalErrors            atomic.Int64
+	SuccessfulExchanges    atomic.Int64
+	FailedExchanges        atomic.Int64
+	ActivePairs            atomic.Int64
+	TotalReconnections     atomic.Int64
+
+	mu        sync.Mutex
+	latencies []time.Duration
+	startTime time.Time
+	endTime   time.Time
+}
+
+// PeerPair represents a sender-receiver pair
+type PeerPair struct {
+	sender   *Client
+	receiver *Client
+	pairID   int
+}
+
+// LoadTest manages the load test execution
+type LoadTest struct {
+	config         LoadTestConfig
+	metrics        *LoadTestMetrics
+	ctx            context.Context
+	cancel         context.CancelFunc
+	reporterCtx    context.Context
+	reporterCancel context.CancelFunc
+}
+
+// NewLoadTest creates a new load test instance
+func NewLoadTest(config LoadTestConfig) *LoadTest {
+	ctx, cancel := context.WithCancel(context.Background())
+	return newLoadTestWithContext(ctx, cancel, config)
+}
+
+// NewLoadTestWithContext creates a new load test instance with a custom context
+func NewLoadTestWithContext(ctx context.Context, config LoadTestConfig) *LoadTest {
+	ctx, cancel := context.WithCancel(ctx)
+	return newLoadTestWithContext(ctx, cancel, config)
+}
+
+func newLoadTestWithContext(ctx context.Context, cancel context.CancelFunc, config LoadTestConfig) *LoadTest {
+	reporterCtx, reporterCancel := context.WithCancel(ctx)
+	config.IDPrefix = fmt.Sprintf("%d-", time.Now().UnixNano())
+	return &LoadTest{
+		config:         config,
+		metrics:        &LoadTestMetrics{},
+		ctx:            ctx,
+		cancel:         cancel,
+		reporterCtx:    reporterCtx,
+		reporterCancel: reporterCancel,
+	}
+}
+
+// Run executes the load test
+func (lt *LoadTest) Run() error {
+	lt.metrics.startTime = time.Now()
+	defer func() {
+		lt.metrics.endTime = time.Now()
+	}()
+
+	exchangeInfo := "single message"
+	if lt.config.ExchangeDuration > 0 {
+		exchangeInfo = fmt.Sprintf("continuous for %v", lt.config.ExchangeDuration)
+	}
+
+	workerPoolSize := lt.config.WorkerPoolSize
+	if workerPoolSize == 0 {
+		workerPoolSize = lt.config.PairsPerSecond * 2
+	}
+
+	channelBufferSize := lt.config.ChannelBufferSize
+	if channelBufferSize == 0 {
+		channelBufferSize = lt.config.PairsPerSecond * 4
+	}
+
+	log.Infof("Starting load test: %d pairs/sec, %d total pairs, message size: %d bytes, exchange: %s",
+		lt.config.PairsPerSecond, lt.config.TotalPairs, lt.config.MessageSize, exchangeInfo)
+	log.Infof("Worker pool size: %d, channel buffer: %d", workerPoolSize, channelBufferSize)
+
+	var wg sync.WaitGroup
+	var reporterWg sync.WaitGroup
+	pairChan := make(chan int, channelBufferSize)
+
+	// Start progress reporter if configured
+	if lt.config.ReportInterval > 0 {
+		reporterWg.Add(1)
+		go lt.progressReporter(&reporterWg, lt.config.ReportInterval)
+	}
+
+	for i := 0; i < workerPoolSize; i++ {
+		wg.Add(1)
+		go lt.pairWorker(&wg, pairChan)
+	}
+
+	testCtx := lt.ctx
+	if lt.config.TestDuration > 0 {
+		var testCancel context.CancelFunc
+		testCtx, testCancel = context.WithTimeout(lt.ctx, lt.config.TestDuration)
+		defer testCancel()
+	}
+
+	ticker := time.NewTicker(time.Second / time.Duration(lt.config.PairsPerSecond))
+	defer ticker.Stop()
+
+	pairsCreated := 0
+	for pairsCreated < lt.config.TotalPairs {
+		select {
+		case <-testCtx.Done():
+			log.Infof("Test duration reached or context cancelled")
+			close(pairChan)
+			wg.Wait()
+			return testCtx.Err()
+		case <-ticker.C:
+			select {
+			case pairChan <- pairsCreated:
+				pairsCreated++
+			default:
+				log.Warnf("Worker pool saturated, skipping pair creation")
+			}
+		}
+	}
+
+	log.Infof("All %d pairs queued, waiting for completion...", pairsCreated)
+	close(pairChan)
+	wg.Wait()
+
+	// Cancel progress reporter context after all work is done and wait for it
+	lt.reporterCancel()
+	reporterWg.Wait()
+
+	return nil
+}
+
+func (lt *LoadTest) pairWorker(wg *sync.WaitGroup, pairChan <-chan int) {
+	defer wg.Done()
+
+	for pairID := range pairChan {
+		lt.metrics.ActivePairs.Add(1)
+		if err := lt.executePairExchange(pairID); err != nil {
+			lt.metrics.TotalErrors.Add(1)
+			lt.metrics.FailedExchanges.Add(1)
+			log.Debugf("Pair %d exchange failed: %v", pairID, err)
+		} else {
+			lt.metrics.SuccessfulExchanges.Add(1)
+		}
+		lt.metrics.ActivePairs.Add(-1)
+		lt.metrics.TotalPairsSent.Add(1)
+	}
+}
+
+func (lt *LoadTest) executePairExchange(pairID int) error {
+	senderID := fmt.Sprintf("%ssender-%d", lt.config.IDPrefix, pairID)
+	receiverID := fmt.Sprintf("%sreceiver-%d", lt.config.IDPrefix, pairID)
+
+	clientConfig := &ClientConfig{
+		InsecureSkipVerify: lt.config.InsecureSkipVerify,
+		EnableReconnect:    lt.config.EnableReconnect,
+		MaxReconnectDelay:  lt.config.MaxReconnectDelay,
+		InitialRetryDelay:  lt.config.InitialRetryDelay,
+	}
+
+	sender, err := NewClientWithConfig(lt.config.ServerURL, senderID, clientConfig)
+	if err != nil {
+		return fmt.Errorf("create sender: %w", err)
+	}
+	defer func() {
+		sender.Close()
+		lt.metrics.TotalReconnections.Add(sender.GetReconnectCount())
+	}()
+
+	receiver, err := NewClientWithConfig(lt.config.ServerURL, receiverID, clientConfig)
+	if err != nil {
+		return fmt.Errorf("create receiver: %w", err)
+	}
+	defer func() {
+		receiver.Close()
+		lt.metrics.TotalReconnections.Add(receiver.GetReconnectCount())
+	}()
+
+	if err := sender.Connect(); err != nil {
+		return fmt.Errorf("sender connect: %w", err)
+	}
+
+	if err := receiver.Connect(); err != nil {
+		return fmt.Errorf("receiver connect: %w", err)
+	}
+
+	time.Sleep(50 * time.Millisecond)
+
+	testMessage := make([]byte, lt.config.MessageSize)
+	for i := range testMessage {
+		testMessage[i] = byte(i % 256)
+	}
+
+	if lt.config.ExchangeDuration > 0 {
+		return lt.continuousExchange(pairID, sender, receiver, receiverID, testMessage)
+	}
+
+	return lt.singleExchange(sender, receiver, receiverID, testMessage)
+}
+
+func (lt *LoadTest) singleExchange(sender, receiver *Client, receiverID string, testMessage []byte) error {
+	startTime := time.Now()
+
+	if err := sender.SendMessage(receiverID, testMessage); err != nil {
+		return fmt.Errorf("send message: %w", err)
+	}
+
+	receiveDone := make(chan error, 1)
+	go func() {
+		msg, err := receiver.ReceiveMessage()
+		if err != nil {
+			receiveDone <- err
+			return
+		}
+		if len(msg.Body) == 0 {
+			receiveDone <- fmt.Errorf("empty message body")
+			return
+		}
+		receiveDone <- nil
+	}()
+
+	select {
+	case err := <-receiveDone:
+		if err != nil {
+			return fmt.Errorf("receive message: %w", err)
+		}
+		latency := time.Since(startTime)
+		lt.recordLatency(latency)
+		lt.metrics.TotalMessagesExchanged.Add(1)
+		return nil
+	case <-time.After(5 * time.Second):
+		return fmt.Errorf("timeout waiting for message")
+	case <-lt.ctx.Done():
+		return lt.ctx.Err()
+	}
+}
+
+func (lt *LoadTest) continuousExchange(pairID int, sender, receiver *Client, receiverID string, testMessage []byte) error {
+	exchangeCtx, cancel := context.WithTimeout(lt.ctx, lt.config.ExchangeDuration)
+	defer cancel()
+
+	messageInterval := lt.config.MessageInterval
+	if messageInterval == 0 {
+		messageInterval = 100 * time.Millisecond
+	}
+
+	errChan := make(chan error, 1)
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := lt.receiverLoop(exchangeCtx, receiver, pairID); err != nil && err != context.DeadlineExceeded && err != context.Canceled {
+			select {
+			case errChan <- err:
+			default:
+			}
+		}
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := lt.senderLoop(exchangeCtx, sender, receiverID, testMessage, messageInterval); err != nil && err != context.DeadlineExceeded && err != context.Canceled {
+			select {
+			case errChan <- err:
+			default:
+			}
+		}
+	}()
+
+	wg.Wait()
+
+	select {
+	case err := <-errChan:
+		return err
+	default:
+		return nil
+	}
+}
+
+func (lt *LoadTest) senderLoop(ctx context.Context, sender *Client, receiverID string, message []byte, interval time.Duration) error {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+			startTime := time.Now()
+			if err := sender.SendMessage(receiverID, message); err != nil {
+				lt.metrics.TotalErrors.Add(1)
+				log.Debugf("Send error: %v", err)
+				continue
+			}
+			lt.recordLatency(time.Since(startTime))
+		}
+	}
+}
+
+func (lt *LoadTest) receiverLoop(ctx context.Context, receiver *Client, pairID int) error {
+	for {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+
+		select {
+		case msg, ok := <-receiver.msgChannel:
+			if !ok {
+				return nil
+			}
+			if len(msg.Body) > 0 {
+				lt.metrics.TotalMessagesExchanged.Add(1)
+			}
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(100 * time.Millisecond):
+			continue
+		}
+	}
+}
+
+func (lt *LoadTest) recordLatency(latency time.Duration) {
+	lt.metrics.mu.Lock()
+	defer lt.metrics.mu.Unlock()
+	lt.metrics.latencies = append(lt.metrics.latencies, latency)
+}
+
+// progressReporter prints periodic progress reports
+func (lt *LoadTest) progressReporter(wg *sync.WaitGroup, interval int) {
+	defer wg.Done()
+
+	lastReported := int64(0)
+	ticker := time.NewTicker(1 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-lt.reporterCtx.Done():
+			return
+		case <-ticker.C:
+			currentMessages := lt.metrics.TotalMessagesExchanged.Load()
+			if currentMessages-lastReported >= int64(interval) {
+				elapsed := time.Since(lt.metrics.startTime)
+				activePairs := lt.metrics.ActivePairs.Load()
+				errors := lt.metrics.TotalErrors.Load()
+				reconnections := lt.metrics.TotalReconnections.Load()
+
+				var msgRate float64
+				if elapsed.Seconds() > 0 {
+					msgRate = float64(currentMessages) / elapsed.Seconds()
+				}
+
+				log.Infof("Progress: %d messages exchanged, %d active pairs, %d errors, %d reconnections, %.2f msg/sec, elapsed: %v",
+					currentMessages, activePairs, errors, reconnections, msgRate, elapsed.Round(time.Second))
+
+				lastReported = (currentMessages / int64(interval)) * int64(interval)
+			}
+		}
+	}
+}
+
+// Stop stops the load test
+func (lt *LoadTest) Stop() {
+	lt.cancel()
+	lt.reporterCancel()
+}
+
+// GetMetrics returns the collected metrics
+func (lt *LoadTest) GetMetrics() *LoadTestMetrics {
+	return lt.metrics
+}
+
+// PrintReport prints a summary report of the test results
+func (m *LoadTestMetrics) PrintReport() {
+	duration := m.endTime.Sub(m.startTime)
+
+	fmt.Println("\n=== Load Test Report ===")
+	fmt.Printf("Test Duration: %v\n", duration)
+	fmt.Printf("Total Pairs Sent: %d\n", m.TotalPairsSent.Load())
+	fmt.Printf("Successful Exchanges: %d\n", m.SuccessfulExchanges.Load())
+	fmt.Printf("Failed Exchanges: %d\n", m.FailedExchanges.Load())
+	fmt.Printf("Total Messages Exchanged: %d\n", m.TotalMessagesExchanged.Load())
+	fmt.Printf("Total Errors: %d\n", m.TotalErrors.Load())
+
+	reconnections := m.TotalReconnections.Load()
+	if reconnections > 0 {
+		fmt.Printf("Total Reconnections: %d\n", reconnections)
+	}
+
+	if duration.Seconds() > 0 {
+		throughput := float64(m.SuccessfulExchanges.Load()) / duration.Seconds()
+		fmt.Printf("Throughput: %.2f pairs/sec\n", throughput)
+	}
+
+	m.mu.Lock()
+	latencies := m.latencies
+	m.mu.Unlock()
+
+	if len(latencies) > 0 {
+		var total time.Duration
+		minLatency := latencies[0]
+		maxLatency := latencies[0]
+
+		for _, lat := range latencies {
+			total += lat
+			if lat < minLatency {
+				minLatency = lat
+			}
+			if lat > maxLatency {
+				maxLatency = lat
+			}
+		}
+
+		avg := total / time.Duration(len(latencies))
+		fmt.Printf("\nLatency Statistics:\n")
+		fmt.Printf("  Min: %v\n", minLatency)
+		fmt.Printf("  Max: %v\n", maxLatency)
+		fmt.Printf("  Avg: %v\n", avg)
+	}
+	fmt.Println("========================")
+}

signal/loadtest/rate_loadtest_test.go

@@ -0,0 +1,305 @@
+package loadtest
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel"
+	"google.golang.org/grpc"
+
+	"github.com/netbirdio/netbird/shared/signal/proto"
+	"github.com/netbirdio/netbird/signal/server"
+)
+
+func TestLoadTest_10PairsPerSecond(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping load test in short mode")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grpcServer, serverAddr := startTestSignalServerForLoad(t, ctx)
+	defer grpcServer.Stop()
+
+	config := LoadTestConfig{
+		ServerURL:      serverAddr,
+		PairsPerSecond: 10,
+		TotalPairs:     50,
+		MessageSize:    100,
+		TestDuration:   30 * time.Second,
+	}
+
+	loadTest := NewLoadTest(config)
+	err := loadTest.Run()
+	require.NoError(t, err)
+
+	metrics := loadTest.GetMetrics()
+	metrics.PrintReport()
+
+	require.Equal(t, int64(50), metrics.TotalPairsSent.Load(), "Should send all 50 pairs")
+	require.Greater(t, metrics.SuccessfulExchanges.Load(), int64(0), "Should have successful exchanges")
+	require.Equal(t, metrics.TotalMessagesExchanged.Load(), metrics.SuccessfulExchanges.Load(), "Messages exchanged should match successful exchanges")
+}
+
+func TestLoadTest_20PairsPerSecond(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping load test in short mode")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grpcServer, serverAddr := startTestSignalServerForLoad(t, ctx)
+	defer grpcServer.Stop()
+
+	config := LoadTestConfig{
+		ServerURL:      serverAddr,
+		PairsPerSecond: 20,
+		TotalPairs:     100,
+		MessageSize:    500,
+		TestDuration:   30 * time.Second,
+	}
+
+	loadTest := NewLoadTest(config)
+	err := loadTest.Run()
+	require.NoError(t, err)
+
+	metrics := loadTest.GetMetrics()
+	metrics.PrintReport()
+
+	require.Equal(t, int64(100), metrics.TotalPairsSent.Load(), "Should send all 100 pairs")
+	require.Greater(t, metrics.SuccessfulExchanges.Load(), int64(0), "Should have successful exchanges")
+}
+
+func TestLoadTest_SmallBurst(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grpcServer, serverAddr := startTestSignalServerForLoad(t, ctx)
+	defer grpcServer.Stop()
+
+	config := LoadTestConfig{
+		ServerURL:      serverAddr,
+		PairsPerSecond: 5,
+		TotalPairs:     10,
+		MessageSize:    50,
+		TestDuration:   10 * time.Second,
+	}
+
+	loadTest := NewLoadTest(config)
+	err := loadTest.Run()
+	require.NoError(t, err)
+
+	metrics := loadTest.GetMetrics()
+	metrics.PrintReport()
+
+	require.Equal(t, int64(10), metrics.TotalPairsSent.Load())
+	require.Greater(t, metrics.SuccessfulExchanges.Load(), int64(5), "At least 50% success rate")
+	require.Less(t, metrics.FailedExchanges.Load(), int64(5), "Less than 50% failure rate")
+}
+
+func TestLoadTest_ContinuousExchange_30Seconds(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping continuous exchange test in short mode")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grpcServer, serverAddr := startTestSignalServerForLoad(t, ctx)
+	defer grpcServer.Stop()
+
+	config := LoadTestConfig{
+		ServerURL:        serverAddr,
+		PairsPerSecond:   5,
+		TotalPairs:       10,
+		MessageSize:      100,
+		ExchangeDuration: 30 * time.Second,
+		MessageInterval:  100 * time.Millisecond,
+		TestDuration:     2 * time.Minute,
+	}
+
+	loadTest := NewLoadTest(config)
+	err := loadTest.Run()
+	require.NoError(t, err)
+
+	metrics := loadTest.GetMetrics()
+	metrics.PrintReport()
+
+	require.Equal(t, int64(10), metrics.TotalPairsSent.Load())
+	require.Greater(t, metrics.TotalMessagesExchanged.Load(), int64(2000), "Should exchange many messages over 30 seconds")
+}
+
+func TestLoadTest_ContinuousExchange_10Minutes(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long continuous exchange test in short mode")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grpcServer, serverAddr := startTestSignalServerForLoad(t, ctx)
+	defer grpcServer.Stop()
+
+	config := LoadTestConfig{
+		ServerURL:        serverAddr,
+		PairsPerSecond:   10,
+		TotalPairs:       20,
+		MessageSize:      200,
+		ExchangeDuration: 10 * time.Minute,
+		MessageInterval:  200 * time.Millisecond,
+		TestDuration:     15 * time.Minute,
+	}
+
+	loadTest := NewLoadTest(config)
+	err := loadTest.Run()
+	require.NoError(t, err)
+
+	metrics := loadTest.GetMetrics()
+	metrics.PrintReport()
+
+	require.Equal(t, int64(20), metrics.TotalPairsSent.Load())
+	require.Greater(t, metrics.TotalMessagesExchanged.Load(), int64(50000), "Should exchange many messages over 10 minutes")
+}
+
+func TestLoadTest_ContinuousExchange_ShortBurst(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grpcServer, serverAddr := startTestSignalServerForLoad(t, ctx)
+	defer grpcServer.Stop()
+
+	config := LoadTestConfig{
+		ServerURL:        serverAddr,
+		PairsPerSecond:   3,
+		TotalPairs:       5,
+		MessageSize:      50,
+		ExchangeDuration: 3 * time.Second,
+		MessageInterval:  100 * time.Millisecond,
+		TestDuration:     10 * time.Second,
+		ReportInterval:   50, // Report every 50 messages for testing
+	}
+
+	loadTest := NewLoadTest(config)
+	err := loadTest.Run()
+	require.NoError(t, err)
+
+	metrics := loadTest.GetMetrics()
+	metrics.PrintReport()
+
+	require.Equal(t, int64(5), metrics.TotalPairsSent.Load())
+	require.Greater(t, metrics.TotalMessagesExchanged.Load(), int64(100), "Should exchange multiple messages in 3 seconds")
+	require.Equal(t, int64(5), metrics.SuccessfulExchanges.Load(), "All pairs should complete successfully")
+}
+
+func TestLoadTest_ReconnectionConfig(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grpcServer, serverAddr := startTestSignalServerForLoad(t, ctx)
+	defer grpcServer.Stop()
+
+	config := LoadTestConfig{
+		ServerURL:         serverAddr,
+		PairsPerSecond:    3,
+		TotalPairs:        5,
+		MessageSize:       50,
+		ExchangeDuration:  2 * time.Second,
+		MessageInterval:   200 * time.Millisecond,
+		TestDuration:      5 * time.Second,
+		EnableReconnect:   true,
+		InitialRetryDelay: 100 * time.Millisecond,
+		MaxReconnectDelay: 2 * time.Second,
+	}
+
+	loadTest := NewLoadTest(config)
+	err := loadTest.Run()
+	require.NoError(t, err)
+
+	metrics := loadTest.GetMetrics()
+	metrics.PrintReport()
+
+	// Test should complete successfully with reconnection enabled
+	require.Equal(t, int64(5), metrics.TotalPairsSent.Load())
+	require.Greater(t, metrics.TotalMessagesExchanged.Load(), int64(0), "Should have exchanged messages")
+	require.Equal(t, int64(5), metrics.SuccessfulExchanges.Load(), "All pairs should complete successfully")
+
+	// Reconnections counter should exist (even if zero for this stable test)
+	reconnections := metrics.TotalReconnections.Load()
+	require.GreaterOrEqual(t, reconnections, int64(0), "Reconnections metric should be tracked")
+}
+
+func BenchmarkLoadTest_Throughput(b *testing.B) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grpcServer, serverAddr := startBenchSignalServer(b, ctx)
+	defer grpcServer.Stop()
+
+	b.Run("5pairs-per-sec", func(b *testing.B) {
+		config := LoadTestConfig{
+			ServerURL:      serverAddr,
+			PairsPerSecond: 5,
+			TotalPairs:     b.N,
+			MessageSize:    100,
+		}
+
+		loadTest := NewLoadTest(config)
+		b.ResetTimer()
+		_ = loadTest.Run()
+		b.StopTimer()
+
+		metrics := loadTest.GetMetrics()
+		b.ReportMetric(float64(metrics.SuccessfulExchanges.Load()), "successful")
+		b.ReportMetric(float64(metrics.FailedExchanges.Load()), "failed")
+	})
+}
+
+func startTestSignalServerForLoad(t *testing.T, ctx context.Context) (*grpc.Server, string) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	grpcServer := grpc.NewServer()
+
+	signalServer, err := server.NewServer(ctx, otel.Meter("test"))
+	require.NoError(t, err)
+
+	proto.RegisterSignalExchangeServer(grpcServer, signalServer)
+
+	go func() {
+		if err := grpcServer.Serve(listener); err != nil {
+			t.Logf("Server stopped: %v", err)
+		}
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+
+	return grpcServer, fmt.Sprintf("http://%s", listener.Addr().String())
+}
+
+func startBenchSignalServer(b *testing.B, ctx context.Context) (*grpc.Server, string) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(b, err)
+
+	grpcServer := grpc.NewServer()
+
+	signalServer, err := server.NewServer(ctx, otel.Meter("bench"))
+	require.NoError(b, err)
+
+	proto.RegisterSignalExchangeServer(grpcServer, signalServer)
+
+	go func() {
+		if err := grpcServer.Serve(listener); err != nil {
+			b.Logf("Server stopped: %v", err)
+		}
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+
+	return grpcServer, fmt.Sprintf("http://%s", listener.Addr().String())
+}