Add LP registration idempotency and retry logic
Make LP registration resilient to network failures that could waste credentials. When registration succeeds on the gateway but the response is lost (e.g., network drop), clients can retry with the same WG key and get the cached result instead of spending another credential. Gateway-side: - Add check_existing_registration() helper that looks up WG peer and returns cached GatewayData if already registered - Add idempotency check in process_registration() dVPN branch - Only return cached response if bandwidth > 0 (ensures registration was actually completed, not just peer created) - Track idempotent registrations with lp_registration_dvpn_idempotent metric Client-side: - Add register_with_retry() to LpRegistrationClient that acquires credential once and retries handshake+registration on failure - Add handshake_and_register_with_retry() to NestedLpSession for exit gateway registration via forwarding - Add exponential backoff with jitter between retry attempts - Verify outer session validity before nested session retry Both retry methods clear state machine before retry to ensure fresh handshake, and reuse the same credential across all attempts.
This commit is contained in:
durch
committed by
Jędrzej Stuczyński
Jędrzej Stuczyński
parent
379ec1d61e
commit
7c3feaede3
+6
-1
@@ -69,4 +69,9 @@ nym-api/redocly/formatted-openapi.json
|
||||
CLAUDE.md
|
||||
docs
|
||||
.claude
|
||||
.superego
|
||||
.superego
|
||||
|
||||
# Superego (machine-specific paths)
|
||||
.superego/
|
||||
.claude/hooks/superego/
|
||||
.claude/settings.json
|
||||
|
||||
@@ -127,6 +127,69 @@ async fn credential_verification(
|
||||
Ok(allocated)
|
||||
}
|
||||
|
||||
/// Check if WG peer already registered, return cached response if so.
|
||||
///
|
||||
/// This enables idempotent registration: if a client retries registration
|
||||
/// with the same WG public key (e.g., after network failure), we return
|
||||
/// the existing registration data instead of re-processing. This prevents
|
||||
/// wasting credentials on network issues.
|
||||
async fn check_existing_registration(
|
||||
wg_key_str: &str,
|
||||
state: &LpHandlerState,
|
||||
) -> Option<LpRegistrationResponse> {
|
||||
// Need WG data to build GatewayData
|
||||
let wg_data = state.wireguard_data.as_ref()?;
|
||||
|
||||
// Look up existing peer
|
||||
let peer = state.storage.get_wireguard_peer(wg_key_str).await.ok()??;
|
||||
|
||||
// Convert to defguard Peer to access allowed_ips
|
||||
let defguard_peer: Peer = peer.clone().try_into().ok()?;
|
||||
|
||||
// Extract IPv4 and IPv6 from allowed_ips
|
||||
let mut ipv4 = None;
|
||||
let mut ipv6 = None;
|
||||
for ip_mask in &defguard_peer.allowed_ips {
|
||||
match ip_mask.ip {
|
||||
std::net::IpAddr::V4(v4) => ipv4 = Some(v4),
|
||||
std::net::IpAddr::V6(v6) => ipv6 = Some(v6),
|
||||
}
|
||||
}
|
||||
|
||||
let (private_ipv4, private_ipv6) = match (ipv4, ipv6) {
|
||||
(Some(v4), Some(v6)) => (v4, v6),
|
||||
_ => return None, // Incomplete data, treat as new registration
|
||||
};
|
||||
|
||||
// Get current bandwidth
|
||||
let bandwidth = state
|
||||
.ecash_verifier
|
||||
.storage()
|
||||
.get_available_bandwidth(peer.client_id)
|
||||
.await
|
||||
.ok()?
|
||||
.map(|b| b.available)
|
||||
.unwrap_or(0);
|
||||
|
||||
// Only return cached response if bandwidth was actually allocated.
|
||||
// If bandwidth is 0, registration was incomplete (peer exists but
|
||||
// credential verification failed or never completed). Let the caller
|
||||
// proceed with normal registration flow which will handle cleanup.
|
||||
if bandwidth == 0 {
|
||||
return None;
|
||||
}
|
||||
|
||||
Some(LpRegistrationResponse::success(
|
||||
bandwidth,
|
||||
GatewayData {
|
||||
public_key: *wg_data.keypair().public_key(),
|
||||
endpoint: wg_data.config().bind_address,
|
||||
private_ipv4,
|
||||
private_ipv6,
|
||||
},
|
||||
))
|
||||
}
|
||||
|
||||
/// Process an LP registration request
|
||||
pub async fn process_registration(
|
||||
request: LpRegistrationRequest,
|
||||
@@ -150,6 +213,20 @@ pub async fn process_registration(
|
||||
RegistrationMode::Dvpn => {
|
||||
// Track dVPN registration attempts
|
||||
inc!("lp_registration_dvpn_attempts");
|
||||
|
||||
// Check for idempotent re-registration (same WG key already registered)
|
||||
// This allows clients to retry registration after network failures
|
||||
// without wasting credentials
|
||||
let wg_key_str = request.wg_public_key.to_string();
|
||||
if let Some(existing_response) = check_existing_registration(&wg_key_str, state).await {
|
||||
info!(
|
||||
"LP dVPN re-registration for existing peer {} (idempotent)",
|
||||
wg_key_str
|
||||
);
|
||||
inc!("lp_registration_dvpn_idempotent");
|
||||
return existing_response;
|
||||
}
|
||||
|
||||
// Register as WireGuard peer first to get client_id
|
||||
let (gateway_data, client_id) = match register_wg_peer(
|
||||
request.wg_public_key.inner().as_ref(),
|
||||
|
||||
@@ -873,6 +873,115 @@ impl LpRegistrationClient {
|
||||
Ok(gateway_data)
|
||||
}
|
||||
|
||||
/// Register with automatic retry on network failure.
|
||||
///
|
||||
/// This method:
|
||||
/// 1. Acquires credential ONCE
|
||||
/// 2. Performs handshake if not already connected
|
||||
/// 3. Attempts registration
|
||||
/// 4. On network failure, re-establishes connection and retries with same credential
|
||||
/// 5. Gateway idempotency ensures no double-spend even if credential was processed
|
||||
///
|
||||
/// Use this method for resilient registration on unreliable networks (e.g., train
|
||||
/// through tunnel). The gateway's idempotent registration check ensures that if
|
||||
/// a registration succeeds but the response is lost, retrying with the same WG key
|
||||
/// will return the cached result instead of spending a new credential.
|
||||
///
|
||||
/// # Arguments
|
||||
/// * `wg_keypair` - Client's WireGuard x25519 keypair (same key used for all retries)
|
||||
/// * `gateway_identity` - Gateway's ed25519 identity for credential verification
|
||||
/// * `bandwidth_controller` - Provider for bandwidth credentials
|
||||
/// * `ticket_type` - Type of bandwidth ticket to use
|
||||
/// * `max_retries` - Maximum number of retry attempts after initial failure
|
||||
///
|
||||
/// # Returns
|
||||
/// * `Ok(GatewayData)` - Gateway configuration data on successful registration
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns an error if all retry attempts fail.
|
||||
///
|
||||
/// # Note
|
||||
/// Unlike `register()`, this method handles the full flow including handshake.
|
||||
/// Do NOT call `perform_handshake()` before this method.
|
||||
pub async fn register_with_retry(
|
||||
&mut self,
|
||||
wg_keypair: &x25519::KeyPair,
|
||||
gateway_identity: &ed25519::PublicKey,
|
||||
bandwidth_controller: &dyn BandwidthTicketProvider,
|
||||
ticket_type: TicketType,
|
||||
max_retries: u32,
|
||||
) -> Result<GatewayData> {
|
||||
tracing::debug!(
|
||||
"Starting resilient registration (max_retries={})",
|
||||
max_retries
|
||||
);
|
||||
|
||||
// Acquire credential ONCE before any attempts
|
||||
let credential = bandwidth_controller
|
||||
.get_ecash_ticket(ticket_type, *gateway_identity, DEFAULT_TICKETS_TO_SPEND)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
LpClientError::SendRegistrationRequest(format!(
|
||||
"Failed to acquire bandwidth credential: {}",
|
||||
e
|
||||
))
|
||||
})?
|
||||
.data;
|
||||
|
||||
let mut last_error = None;
|
||||
for attempt in 0..=max_retries {
|
||||
if attempt > 0 {
|
||||
// Exponential backoff with jitter: 100ms, 200ms, 400ms, 800ms, 1600ms (capped)
|
||||
let base_delay_ms = 100u64 * (1 << attempt.min(4));
|
||||
let jitter_ms = rand::random::<u64>() % (base_delay_ms / 4 + 1);
|
||||
let delay = std::time::Duration::from_millis(base_delay_ms + jitter_ms);
|
||||
tracing::info!(
|
||||
"Retrying registration (attempt {}) after {:?}",
|
||||
attempt + 1,
|
||||
delay
|
||||
);
|
||||
tokio::time::sleep(delay).await;
|
||||
}
|
||||
|
||||
// Ensure fresh connection and handshake for each attempt
|
||||
// (On retry, the old connection/session may be dead)
|
||||
if self.stream.is_none() || attempt > 0 {
|
||||
// Clear any stale state before re-handshaking
|
||||
self.close();
|
||||
self.state_machine = None;
|
||||
|
||||
if let Err(e) = self.perform_handshake().await {
|
||||
tracing::warn!("Handshake failed on attempt {}: {}", attempt + 1, e);
|
||||
last_error = Some(e);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
match self
|
||||
.register_with_credential(wg_keypair, credential.clone(), ticket_type)
|
||||
.await
|
||||
{
|
||||
Ok(data) => {
|
||||
if attempt > 0 {
|
||||
tracing::info!(
|
||||
"Registration succeeded on retry attempt {}",
|
||||
attempt + 1
|
||||
);
|
||||
}
|
||||
return Ok(data);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("Registration attempt {} failed: {}", attempt + 1, e);
|
||||
last_error = Some(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Err(last_error.unwrap_or_else(|| {
|
||||
LpClientError::Transport("Registration failed after all retries".to_string())
|
||||
}))
|
||||
}
|
||||
|
||||
/// Sends a ForwardPacket message to the entry gateway for forwarding to the exit gateway.
|
||||
///
|
||||
/// This method constructs a ForwardPacket containing the target gateway's identity,
|
||||
|
||||
@@ -607,6 +607,118 @@ impl NestedLpSession {
|
||||
Ok(gateway_data)
|
||||
}
|
||||
|
||||
/// Performs handshake and registration with the exit gateway via forwarding,
|
||||
/// with automatic retry on network failure.
|
||||
///
|
||||
/// This method:
|
||||
/// 1. Acquires credential ONCE
|
||||
/// 2. Performs handshake and registration with exit gateway
|
||||
/// 3. On network failure, clears state and retries with same credential
|
||||
/// 4. Gateway idempotency ensures no double-spend even if credential was processed
|
||||
///
|
||||
/// Use this method for resilient exit registration on unreliable networks (e.g., train
|
||||
/// through tunnel). The gateway's idempotent registration check ensures that if
|
||||
/// a registration succeeds but the response is lost, retrying with the same WG key
|
||||
/// will return the cached result instead of spending a new credential.
|
||||
///
|
||||
/// # Arguments
|
||||
/// * `outer_client` - Connected LP client with established outer session to entry gateway
|
||||
/// * `wg_keypair` - Client's WireGuard x25519 keypair (same key used for all retries)
|
||||
/// * `gateway_identity` - Exit gateway's Ed25519 identity (for credential verification)
|
||||
/// * `bandwidth_controller` - Provider for bandwidth credentials
|
||||
/// * `ticket_type` - Type of bandwidth ticket to use
|
||||
/// * `client_ip` - Client IP address for registration metadata
|
||||
/// * `max_retries` - Maximum number of retry attempts after initial failure
|
||||
///
|
||||
/// # Returns
|
||||
/// * `Ok(GatewayData)` - Exit gateway configuration data on successful registration
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns an error if all retry attempts fail.
|
||||
pub async fn handshake_and_register_with_retry(
|
||||
&mut self,
|
||||
outer_client: &mut LpRegistrationClient,
|
||||
wg_keypair: &x25519::KeyPair,
|
||||
gateway_identity: &ed25519::PublicKey,
|
||||
bandwidth_controller: &dyn BandwidthTicketProvider,
|
||||
ticket_type: TicketType,
|
||||
client_ip: IpAddr,
|
||||
max_retries: u32,
|
||||
) -> Result<GatewayData> {
|
||||
tracing::debug!(
|
||||
"Starting resilient exit registration (max_retries={})",
|
||||
max_retries
|
||||
);
|
||||
|
||||
// Acquire credential ONCE before any attempts
|
||||
let credential = bandwidth_controller
|
||||
.get_ecash_ticket(
|
||||
ticket_type,
|
||||
*gateway_identity,
|
||||
nym_bandwidth_controller::DEFAULT_TICKETS_TO_SPEND,
|
||||
)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
LpClientError::Transport(format!("Failed to acquire bandwidth credential: {}", e))
|
||||
})?
|
||||
.data;
|
||||
|
||||
let mut last_error = None;
|
||||
for attempt in 0..=max_retries {
|
||||
if attempt > 0 {
|
||||
// Verify outer session is still usable before retry
|
||||
if !outer_client.is_handshake_complete() {
|
||||
return Err(LpClientError::Transport(
|
||||
"Outer session lost during retry - caller must re-establish entry gateway connection".to_string()
|
||||
));
|
||||
}
|
||||
|
||||
// Exponential backoff with jitter: 100ms, 200ms, 400ms, 800ms, 1600ms (capped)
|
||||
let base_delay_ms = 100u64 * (1 << attempt.min(4));
|
||||
let jitter_ms = rand::random::<u64>() % (base_delay_ms / 4 + 1);
|
||||
let delay = std::time::Duration::from_millis(base_delay_ms + jitter_ms);
|
||||
tracing::info!(
|
||||
"Retrying exit registration (attempt {}) after {:?}",
|
||||
attempt + 1,
|
||||
delay
|
||||
);
|
||||
tokio::time::sleep(delay).await;
|
||||
|
||||
// Clear state machine before retry - handshake needs fresh start
|
||||
self.state_machine = None;
|
||||
}
|
||||
|
||||
match self
|
||||
.handshake_and_register_with_credential(
|
||||
outer_client,
|
||||
wg_keypair,
|
||||
credential.clone(),
|
||||
ticket_type,
|
||||
client_ip,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(data) => {
|
||||
if attempt > 0 {
|
||||
tracing::info!(
|
||||
"Exit registration succeeded on retry attempt {}",
|
||||
attempt + 1
|
||||
);
|
||||
}
|
||||
return Ok(data);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("Exit registration attempt {} failed: {}", attempt + 1, e);
|
||||
last_error = Some(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Err(last_error.unwrap_or_else(|| {
|
||||
LpClientError::Transport("Exit registration failed after all retries".to_string())
|
||||
}))
|
||||
}
|
||||
|
||||
/// Sends a packet via forwarding through the entry gateway and returns the parsed response.
|
||||
///
|
||||
/// This helper consolidates the send/receive pattern used throughout the handshake:
|
||||
|
||||
Reference in New Issue
Block a user