fix: more nondeterminism issues with solana

chain-signatures/node/src/protocol/signature.rs

@@ -1164,6 +1164,12 @@ struct SignTask {
 }
 
 impl SignTask {
+    fn should_restart_phase(&mut self, contract_state: &ProtocolState) -> bool {
+        let was_running = self.governance.is_running;
+        let is_running = self.refresh_governance(contract_state);
+        !was_running && is_running
+    }
+
     async fn run(
         mut self,
         indexed: IndexedSignRequest,
@@ -1184,10 +1190,17 @@ impl SignTask {
         loop {
             tokio::select! {
                 Some(contract_state) = contract_watcher.next_state() => {
-                    is_running = self.refresh_governance(&contract_state);
-                    if is_running {
-                        // we're back into running, reset to SignOrganizer
+                    let should_restart_phase = self.should_restart_phase(&contract_state);
+                    is_running = self.governance.is_running;
+                    if should_restart_phase {
+                        // We only reset after a real pause => resume transition.
                         phase = SignPhase::Organizing(SignOrganizer);
+                    } else if is_running {
+                        tracing::info!(
+                            ?sign_id,
+                            gov = ?self.governance,
+                            "signature task refreshed governance while running"
+                        );
                     } else {
                         tracing::info!(
                             ?sign_id,
@@ -1650,4 +1663,87 @@ mod tests {
         assert_eq!(sign_task.governance.threshold, 1);
         assert_eq!(sign_task.governance.me, Participant::from(0));
     }
+
+    #[test]
+    fn sign_task_only_restarts_after_pause_resume() {
+        let account_id: near_account_id::AccountId = "p-0".parse().unwrap();
+        let mut participants = Participants::default();
+        participants.insert(&Participant::from(0), ParticipantInfo::new(0));
+
+        let governance = GovernanceInfo {
+            me: Participant::from(0),
+            threshold: 1,
+            epoch: 0,
+            public_key: k256::AffinePoint::default(),
+            participants: [Participant::from(0)].into_iter().collect(),
+            is_running: true,
+        };
+
+        let redis_cfg = deadpool_redis::Config::from_url("redis://127.0.0.1/");
+        let pool = redis_cfg.create_pool(Some(Runtime::Tokio1)).unwrap();
+        let presignatures = Presignature::storage(&pool, &account_id);
+        let (_inbox, _outbox, msg_channel) = MessageChannel::new();
+        let (rpc_tx, _rpc_rx) = mpsc::channel(1);
+        let rpc_channel = RpcChannel { tx: rpc_tx };
+        let (contract, _tx) = ContractStateWatcher::with_running(
+            &account_id,
+            k256::AffinePoint::default(),
+            1,
+            participants.clone(),
+        );
+
+        let mut sign_task = SignTask {
+            governance,
+            sign_id: SignId::new([0u8; 32]),
+            presignatures,
+            msg: msg_channel,
+            rpc: rpc_channel,
+            backlog: Backlog::new(),
+            cfg: ProtocolConfig::default(),
+            contract,
+            is_proposer: Arc::new(AtomicBool::new(false)),
+            node_account_id: account_id,
+        };
+
+        let running = ProtocolState::Running(RunningContractState {
+            epoch: 1,
+            public_key: k256::AffinePoint::default(),
+            participants: participants.clone(),
+            candidates: Default::default(),
+            join_votes: Default::default(),
+            leave_votes: Default::default(),
+            threshold: 1,
+        });
+
+        assert!(!sign_task.should_restart_phase(&running));
+        assert!(sign_task.governance.is_running);
+        assert_eq!(sign_task.governance.epoch, 1);
+
+        let resharing = ProtocolState::Resharing(ResharingContractState {
+            old_epoch: 1,
+            old_participants: participants.clone(),
+            new_participants: participants.clone(),
+            threshold: 1,
+            public_key: k256::AffinePoint::default(),
+            finished_votes: Default::default(),
+            cancel_votes: Default::default(),
+        });
+
+        assert!(!sign_task.should_restart_phase(&resharing));
+        assert!(!sign_task.governance.is_running);
+
+        let resumed = ProtocolState::Running(RunningContractState {
+            epoch: 2,
+            public_key: k256::AffinePoint::default(),
+            participants,
+            candidates: Default::default(),
+            join_votes: Default::default(),
+            leave_votes: Default::default(),
+            threshold: 1,
+        });
+
+        assert!(sign_task.should_restart_phase(&resumed));
+        assert!(sign_task.governance.is_running);
+        assert_eq!(sign_task.governance.epoch, 2);
+    }
 }

integration-tests/Cargo.toml

@@ -17,6 +17,7 @@ ciborium.workspace = true
 clap.workspace = true
 deadpool-redis.workspace = true
 futures = "0.3"
+fs2 = "0.4"
 generic-array = { version = "0.14.7", default-features = false }
 hex.workspace = true
 hyper.workspace = true

integration-tests/src/actions/sign.rs

@@ -574,13 +574,10 @@ async fn wait_for_signature_responded_event(
     timeout: Duration,
 ) -> anyhow::Result<SolSignatureResponse> {
     let rpc_client = RpcClient::new(rpc_http_url);
-    let pubsub_client = PubsubClient::new(rpc_ws_url.as_str()).await?;
-
     let filter = RpcTransactionLogsFilter::Mentions(vec![program_id.to_string()]);
     let config = RpcTransactionLogsConfig {
         commitment: Some(CommitmentConfig::confirmed()),
     };
-    let (mut stream, _unsubscribe) = pubsub_client.logs_subscribe(filter, config).await?;
 
     let mut seen = HashSet::new();
     let program_invoke_prefix = format!("Program {} invoke [", program_id);
@@ -589,70 +586,89 @@ async fn wait_for_signature_responded_event(
     tokio::pin!(deadline);
 
     loop {
-        tokio::select! {
-            _ = &mut deadline => {
-                anyhow::bail!("timeout waiting for respond on sol");
-            }
-            maybe = stream.next() => {
-                let Some(response) = maybe else {
-                    anyhow::bail!("sol signature respond log stream closed unexpectedly");
-                };
-
-                if response.value.err.is_some() {
-                    continue;
-                }
+        let pubsub_client = PubsubClient::new(rpc_ws_url.as_str()).await?;
+        let (mut stream, _unsubscribe) = pubsub_client
+            .logs_subscribe(filter.clone(), config.clone())
+            .await?;
 
-                let logs = &response.value.logs;
-                if !logs.iter().any(|log| log.contains(RESPOND_EVENT_HINT)) {
-                    continue;
-                }
-                if !logs.iter().any(|log| log.starts_with(&program_invoke_prefix)) {
-                    continue;
+        let stream_result = loop {
+            tokio::select! {
+                _ = &mut deadline => {
+                    anyhow::bail!("timeout waiting for respond on sol");
                 }
+                maybe = stream.next() => {
+                    let Some(response) = maybe else {
+                        tracing::warn!("sol signature respond log stream closed unexpectedly, reconnecting");
+                        break Ok::<Option<SolSignatureResponse>, anyhow::Error>(None);
+                    };
+
+                    if response.value.err.is_some() {
+                        continue;
+                    }
 
-                let sig_text = &response.value.signature;
-                let Ok(tx_signature) = solana_sdk::signature::Signature::from_str(sig_text) else {
-                    tracing::warn!(tx_signature = sig_text, "invalid solana signature string in respond logs");
-                    continue;
-                };
+                    let logs = &response.value.logs;
+                    if !logs.iter().any(|log| log.contains(RESPOND_EVENT_HINT)) {
+                        continue;
+                    }
+                    if !logs.iter().any(|log| log.starts_with(&program_invoke_prefix)) {
+                        continue;
+                    }
 
-                if !seen.insert(tx_signature) {
-                    continue;
-                }
+                    let sig_text = &response.value.signature;
+                    let Ok(tx_signature) = solana_sdk::signature::Signature::from_str(sig_text) else {
+                        tracing::warn!(tx_signature = sig_text, "invalid solana signature string in respond logs");
+                        continue;
+                    };
 
-                match parse_signature_responded_events(&rpc_client, &tx_signature, &program_id).await {
-                    Ok(events) => {
-                        for event in events {
-                            tracing::info!(
-                                request_id = %hex::encode(event.request_id),
-                                tx_signature = %tx_signature,
-                                "received SignatureRespondedEvent via CPI logs",
-                            );
+                    if !seen.insert(tx_signature) {
+                        continue;
+                    }
 
-                            match parse_sol_signature(&event.signature) {
-                                Ok((signature, recovery_id)) => {
-                                    return Ok(SolSignatureResponse {
-                                        request_id: event.request_id,
-                                        signature,
-                                        recovery_id,
-                                    });
-                                }
-                                Err(err) => {
-                                    tracing::warn!(?err, "failed to parse sol signature from SignatureRespondedEvent");
+                    match parse_signature_responded_events(&rpc_client, &tx_signature, &program_id).await {
+                        Ok(events) => {
+                            for event in events {
+                                tracing::info!(
+                                    request_id = %hex::encode(event.request_id),
+                                    tx_signature = %tx_signature,
+                                    "received SignatureRespondedEvent via CPI logs",
+                                );
+
+                                match parse_sol_signature(&event.signature) {
+                                    Ok((signature, recovery_id)) => {
+                                        return Ok(SolSignatureResponse {
+                                            request_id: event.request_id,
+                                            signature,
+                                            recovery_id,
+                                        });
+                                    }
+                                    Err(err) => {
+                                        tracing::warn!(
+                                            ?err,
+                                            tx_signature = %tx_signature,
+                                            request_id = %hex::encode(event.request_id),
+                                            "failed to parse sol signature from SignatureRespondedEvent",
+                                        );
+                                    }
                                 }
                             }
                         }
-                    }
-                    Err(err) => {
-                        tracing::warn!(
-                            ?err,
-                            tx_signature = %tx_signature,
-                            "failed to parse SignatureRespondedEvent from respond transaction",
-                        );
+                        Err(err) => {
+                            tracing::warn!(
+                                ?err,
+                                tx_signature = %tx_signature,
+                                "failed to parse SignatureRespondedEvent from respond transaction",
+                            );
+                        }
                     }
                 }
             }
+        };
+
+        if let Some(response) = stream_result? {
+            return Ok(response);
         }
+
+        tokio::time::sleep(Duration::from_secs(1)).await;
     }
 }
 
@@ -766,33 +782,38 @@ pub async fn wait_for_respond_bidirectional(
 
     let event_unsub = program
         .on(move |_ctx, event: RespondBidirectionalEvent| {
+            if event.request_id != expected_request_id {
+                return;
+            }
+
+            let Ok(mut sender) = tx.lock() else {
+                tracing::warn!("failed to lock RespondBidirectionalEvent sender");
+                return;
+            };
+
+            let Some(sender) = sender.take() else {
+                return;
+            };
+
             tracing::info!(
                 request_id = %hex::encode(event.request_id),
                 responder = ?event.responder,
                 serialized_output_len = event.serialized_output.len(),
                 "received RespondBidirectionalEvent",
             );
 
-            if event.request_id != expected_request_id {
-                return;
-            }
-
-            let signature_result = parse_sol_signature(&event.signature);
-            if let Ok(mut sender) = tx.lock() {
-                if let Some(sender) = sender.take() {
-                    let outcome = signature_result.map(|(signature, recovery_id)| {
-                        SolRespondBidirectionalOutcome {
-                            request_id: event.request_id,
-                            responder: event.responder.to_string(),
-                            serialized_output: event.serialized_output.clone(),
-                            signature,
-                            recovery_id,
-                        }
-                    });
-                    if sender.send(outcome).is_err() {
-                        tracing::error!("failed to send RespondBidirectionalEvent outcome");
-                    }
+            let outcome = parse_sol_signature(&event.signature).map(|(signature, recovery_id)| {
+                SolRespondBidirectionalOutcome {
+                    request_id: event.request_id,
+                    responder: event.responder.to_string(),
+                    serialized_output: event.serialized_output.clone(),
+                    signature,
+                    recovery_id,
                 }
+            });
+
+            if sender.send(outcome).is_err() {
+                tracing::error!("failed to send RespondBidirectionalEvent outcome");
             }
         })
         .await?;