fix(taiko-client-rs): retry aggregation errors instead of dropping proofs

aggregate_proofs_by_type cleared the buffer and returned on any
non-Pending error, permanently dropping already-generated proofs whose
Proposed events were already marked handled (no scanner replay). Go
retries all aggregation errors with constant backoff and only clears on
shutdown. Mirror that: retry non-terminal errors, keeping the buffer.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: davidtaikocha

packages/taiko-client-rs/crates/prover/src/submitter/submitter.rs

@@ -288,13 +288,18 @@ impl Pipeline {
                     let _ = self.channels.batch_proofs_tx.send(batch).await;
                     return Ok(());
                 }
+                // Retry every non-success outcome with constant backoff, keeping
+                // the buffered proofs (Go `AggregateProofsByType` wraps the
+                // aggregation in `backoff.Retry`; the in-progress vs. error
+                // distinction is logging-only). Dropping the buffer here would
+                // lose already-generated proofs whose `Proposed` events are
+                // already marked handled, so they would never be re-proven.
                 Err(ProverError::Raiko(RaikoError::Pending(_))) => {
                     tokio::time::sleep(self.cfg.proof_polling_interval).await;
                 }
                 Err(err) => {
-                    let ids: Vec<u64> = items.iter().map(ProofResponse::proposal_id).collect();
-                    buffer.clear_items(&ids);
-                    return Err(err);
+                    tracing::warn!(%err, ?proof_type, "aggregate proofs failed, retrying");
+                    tokio::time::sleep(self.cfg.proof_polling_interval).await;
                 }
             }
         }
@@ -572,9 +577,27 @@ impl ProofSubmitter {
             return Ok(());
         }
 
-        if let Err(err) = self.tx_manager.send(candidate).await {
-            ProverMetrics::submission_errors().inc();
-            return Err(ProverError::from(err));
+        let receipt = match self.tx_manager.send(candidate).await {
+            Ok(receipt) => receipt,
+            Err(err) => {
+                ProverMetrics::submission_errors().inc();
+                return Err(ProverError::from(err));
+            }
+        };
+
+        // base-tx-manager returns `Ok(receipt)` even for a transaction that reached
+        // confirmation depth but reverted, so the status must be checked explicitly.
+        // A revert means the proofs did not land; surface it so the caller resends
+        // the requests rather than dropping them, matching Go `prover.go:302-314`
+        // (reverted/unretryable -> `ClearProofBuffers(batchProof, true)`).
+        if !receipt.status() {
+            ProverMetrics::submission_reverted().inc();
+            tracing::error!(
+                tx_hash = %receipt.transaction_hash,
+                ?batch.batch_ids,
+                "prove transaction reverted on-chain; resending proof requests"
+            );
+            return Err(ProverError::SubmissionReverted);
         }
 
         ProverMetrics::proofs_sent().inc_by(batch.batch_ids.len() as u64);
@@ -927,20 +950,25 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn aggregate_clears_buffer_on_terminal_error() {
+    async fn aggregate_retries_error_without_dropping_buffer() {
         let producer = Arc::new(MockProducer::default());
+        // First aggregation attempt fails; the retry (empty queue -> Ok) succeeds.
         producer
             .batch
             .lock()
             .await
             .push_back(Err(ProverError::Other(anyhow::anyhow!("raiko down"))));
-        let h = harness(producer, Duration::from_secs(3_600), 2);
+        let mut h = harness(producer, Duration::from_secs(3_600), 2);
         let buffer = h.pipeline.buffers().get(&ProofType::Sgx).unwrap();
         buffer.write(response(5, ProofType::Sgx)).unwrap();
+        buffer.write(response(6, ProofType::Sgx)).unwrap();
 
-        let err = h.pipeline.aggregate_proofs_by_type(ProofType::Sgx).await.unwrap_err();
-        assert!(err.to_string().contains("raiko down"));
-        assert_eq!(buffer.len(), 0, "terminal error clears the buffer");
+        // The transient error is retried, not surfaced, and the proofs are kept.
+        h.pipeline.aggregate_proofs_by_type(ProofType::Sgx).await.unwrap();
+
+        let batch = h.batch_rx.try_recv().unwrap();
+        assert_eq!(batch.batch_ids, vec![5, 6], "buffered proofs aggregated on retry, not dropped");
+        assert_eq!(buffer.len(), 2, "buffer retained until post-submission clear");
     }
 
     #[tokio::test]