fix: add timeout to WaitingForResponses state (30s no progress = fail)

The JoinerState tracks last_progress but never enforced a timeout.
If a connect operation received no acceptances, it would wait indefinitely.

Now we check for timeout at the start of process_message:
- If gateway is set (joiner) and in WaitingForResponses state
- And last_progress exceeds JOINER_PROGRESS_TIMEOUT (30s)
- Return OpError::Timeout to fail the operation

This prevents indefinitely stalled connect operations from blocking
new connection attempts.

Fixes the WaitingForResponses timeout issue from #2173.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: sanity

crates/core/src/operations/connect.rs

@@ -28,6 +28,9 @@ use freenet_stdlib::client_api::HostResponse;
 
 const FORWARD_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(20);
 const RECENCY_COOLDOWN: Duration = Duration::from_secs(30);
+/// Timeout for joiner waiting for responses. If no progress (acceptances or
+/// observed address updates) is made within this duration, the operation fails.
+const JOINER_PROGRESS_TIMEOUT: Duration = Duration::from_secs(30);
 
 /// Top-level message envelope used by the new connect handshake.
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -512,6 +515,11 @@ impl JoinerState {
         self.observed_address = Some(address);
         self.last_progress = now;
     }
+
+    /// Returns true if no progress has been made within the timeout period.
+    pub(crate) fn has_timed_out(&self, now: Instant) -> bool {
+        now.duration_since(self.last_progress) >= JOINER_PROGRESS_TIMEOUT
+    }
 }
 
 /// Placeholder operation wrapper so we can exercise the logic in isolation in
@@ -806,6 +814,21 @@ impl Operation for ConnectOp {
         Box<dyn std::future::Future<Output = Result<OperationResult, OpError>> + Send + 'a>,
     > {
         Box::pin(async move {
+            // Check for joiner timeout before processing any message
+            if self.gateway.is_some() {
+                if let Some(ConnectState::WaitingForResponses(ref state)) = self.state {
+                    if state.has_timed_out(Instant::now()) {
+                        tracing::warn!(
+                            tx = %self.id,
+                            last_progress_secs = state.last_progress.elapsed().as_secs(),
+                            accepted_count = state.accepted.len(),
+                            "connect: joiner timed out waiting for responses"
+                        );
+                        return Err(OpError::Timeout);
+                    }
+                }
+            }
+
             match msg {
                 ConnectMsg::Request { payload, .. } => {
                     let env = RelayEnv::new(op_manager);

crates/core/src/operations/mod.rs

@@ -324,6 +324,8 @@ pub(crate) enum OpError {
     OpNotPresent(Transaction),
     #[error("max number of retries for tx {0} of op type `{1}` reached")]
     MaxRetriesExceeded(Transaction, TransactionType),
+    #[error("operation timed out waiting for progress")]
+    Timeout,
     #[error("op not available")]
     OpNotAvailable(#[from] OpNotAvailable),