chore(agent): impl parent-child comms w/ retry for SenderAccountTask

Author: suchapalaver

crates/tap-agent/src/agent/sender_account.rs

@@ -117,8 +117,7 @@ type RavMap = HashMap<Address, u128>;
 type Balance = U256;
 
 /// Information for Ravs that are abstracted away from the SignedRav itself
-#[derive(Debug, Default, PartialEq, Eq)]
-#[cfg_attr(any(test, feature = "test"), derive(Clone))]
+#[derive(Debug, Default, PartialEq, Eq, Clone)]
 pub struct RavInformation {
     /// Allocation Id of a Rav
     pub allocation_id: Address,
@@ -159,7 +158,7 @@ impl From<&tap_graph::v2::SignedRav> for RavInformation {
 /// It has different logic depending on the variant
 #[derive(Debug)]
 #[cfg_attr(any(test, feature = "test"), derive(educe::Educe))]
-#[cfg_attr(any(test, feature = "test"), educe(PartialEq, Eq, Clone))]
+#[cfg_attr(any(test, feature = "test"), educe(PartialEq, Eq))]
 pub enum ReceiptFees {
     /// Adds the receipt value to the fee tracker
     ///
@@ -175,10 +174,7 @@ pub enum ReceiptFees {
     /// If not, signalize the fee_tracker to apply proper backoff
     RavRequestResponse(
         UnaggregatedReceipts,
-        #[cfg_attr(
-            any(test, feature = "test"),
-            educe(PartialEq(ignore), Clone(method(clone_rav_result)))
-        )]
+        #[cfg_attr(any(test, feature = "test"), educe(PartialEq(ignore)))]
         anyhow::Result<Option<RavInformation>>,
     ),
     /// Ignores all logic and simply retry Allow/Deny and Rav Request logic
@@ -189,20 +185,29 @@ pub enum ReceiptFees {
     Retry,
 }
 
-#[cfg(any(test, feature = "test"))]
-fn clone_rav_result(
-    res: &anyhow::Result<Option<RavInformation>>,
-) -> anyhow::Result<Option<RavInformation>> {
-    match res {
-        Ok(val) => Ok(val.clone()),
-        Err(_) => Err(anyhow::anyhow!("Some error")),
+impl Clone for ReceiptFees {
+    fn clone(&self) -> Self {
+        match self {
+            ReceiptFees::NewReceipt(value, timestamp) => {
+                ReceiptFees::NewReceipt(*value, *timestamp)
+            }
+            ReceiptFees::UpdateValue(receipts) => ReceiptFees::UpdateValue(*receipts),
+            ReceiptFees::RavRequestResponse(receipts, result) => {
+                // For the Result<Option<RavInformation>>, we need to handle it carefully
+                // since anyhow::Error doesn't implement Clone
+                let cloned_result = match result {
+                    Ok(val) => Ok(val.clone()),
+                    Err(_) => Err(anyhow::anyhow!("Cloned error from original anyhow result")),
+                };
+                ReceiptFees::RavRequestResponse(*receipts, cloned_result)
+            }
+            ReceiptFees::Retry => ReceiptFees::Retry,
+        }
     }
 }
 
 /// Enum containing all types of messages that a [SenderAccount] can receive
 #[derive(Debug)]
-#[cfg_attr(any(test, feature = "test"), derive(educe::Educe))]
-#[cfg_attr(any(test, feature = "test"), educe(PartialEq, Eq, Clone))]
 pub enum SenderAccountMessage {
     /// Updates the sender balance and
     UpdateBalanceAndLastRavs(Balance, RavMap),
@@ -224,24 +229,104 @@ pub enum SenderAccountMessage {
     UpdateRav(RavInformation),
     #[cfg(test)]
     /// Returns the sender fee tracker, used for tests
-    GetSenderFeeTracker(
-        #[educe(PartialEq(ignore), Clone(method(crate::test::actors::clone_rpc_reply)))]
-        ractor::RpcReplyPort<SenderFeeTracker>,
-    ),
+    GetSenderFeeTracker(ractor::RpcReplyPort<SenderFeeTracker>),
     #[cfg(test)]
     /// Returns the Deny status, used for tests
-    GetDeny(
-        #[educe(PartialEq(ignore), Clone(method(crate::test::actors::clone_rpc_reply)))]
-        ractor::RpcReplyPort<bool>,
-    ),
+    GetDeny(ractor::RpcReplyPort<bool>),
     #[cfg(test)]
     /// Returns if the scheduler is enabled, used for tests
-    IsSchedulerEnabled(
-        #[educe(PartialEq(ignore), Clone(method(crate::test::actors::clone_rpc_reply)))]
-        ractor::RpcReplyPort<bool>,
-    ),
+    IsSchedulerEnabled(ractor::RpcReplyPort<bool>),
+}
+
+// Manual Clone implementation to handle RpcReplyPort fields properly
+impl Clone for SenderAccountMessage {
+    fn clone(&self) -> Self {
+        match self {
+            SenderAccountMessage::UpdateBalanceAndLastRavs(balance, rav_map) => {
+                SenderAccountMessage::UpdateBalanceAndLastRavs(*balance, rav_map.clone())
+            }
+            SenderAccountMessage::UpdateAllocationIds(ids) => {
+                SenderAccountMessage::UpdateAllocationIds(ids.clone())
+            }
+            SenderAccountMessage::NewAllocationId(id) => SenderAccountMessage::NewAllocationId(*id),
+            SenderAccountMessage::UpdateReceiptFees(id, fees) => {
+                SenderAccountMessage::UpdateReceiptFees(*id, fees.clone())
+            }
+            SenderAccountMessage::UpdateInvalidReceiptFees(id, fees) => {
+                SenderAccountMessage::UpdateInvalidReceiptFees(*id, *fees)
+            }
+            SenderAccountMessage::UpdateRav(rav) => SenderAccountMessage::UpdateRav(rav.clone()),
+            #[cfg(test)]
+            SenderAccountMessage::GetSenderFeeTracker(_reply_port) => {
+                // For tests, create a dummy reply port - this is needed for message forwarding but
+                // the actual reply port can't be meaningfully cloned
+                let (tx, _rx) = tokio::sync::oneshot::channel();
+                SenderAccountMessage::GetSenderFeeTracker(ractor::RpcReplyPort::from(tx))
+            }
+            #[cfg(test)]
+            SenderAccountMessage::GetDeny(_reply_port) => {
+                // For tests, create a dummy reply port
+                let (tx, _rx) = tokio::sync::oneshot::channel();
+                SenderAccountMessage::GetDeny(ractor::RpcReplyPort::from(tx))
+            }
+            #[cfg(test)]
+            SenderAccountMessage::IsSchedulerEnabled(_reply_port) => {
+                // For tests, create a dummy reply port
+                let (tx, _rx) = tokio::sync::oneshot::channel();
+                SenderAccountMessage::IsSchedulerEnabled(ractor::RpcReplyPort::from(tx))
+            }
+        }
+    }
 }
 
+#[cfg(any(test, feature = "test"))]
+impl PartialEq for SenderAccountMessage {
+    fn eq(&self, other: &Self) -> bool {
+        match (self, other) {
+            (
+                SenderAccountMessage::UpdateBalanceAndLastRavs(balance1, rav_map1),
+                SenderAccountMessage::UpdateBalanceAndLastRavs(balance2, rav_map2),
+            ) => balance1 == balance2 && rav_map1 == rav_map2,
+            (
+                SenderAccountMessage::UpdateAllocationIds(ids1),
+                SenderAccountMessage::UpdateAllocationIds(ids2),
+            ) => ids1 == ids2,
+            (
+                SenderAccountMessage::NewAllocationId(id1),
+                SenderAccountMessage::NewAllocationId(id2),
+            ) => id1 == id2,
+            (
+                SenderAccountMessage::UpdateReceiptFees(id1, fees1),
+                SenderAccountMessage::UpdateReceiptFees(id2, fees2),
+            ) => id1 == id2 && fees1 == fees2,
+            (
+                SenderAccountMessage::UpdateInvalidReceiptFees(id1, fees1),
+                SenderAccountMessage::UpdateInvalidReceiptFees(id2, fees2),
+            ) => id1 == id2 && fees1 == fees2,
+            (SenderAccountMessage::UpdateRav(rav1), SenderAccountMessage::UpdateRav(rav2)) => {
+                rav1 == rav2
+            }
+            #[cfg(test)]
+            // For RpcReplyPort messages, we ignore the reply port in comparison (like educe did)
+            (
+                SenderAccountMessage::GetSenderFeeTracker(_),
+                SenderAccountMessage::GetSenderFeeTracker(_),
+            ) => true,
+            #[cfg(test)]
+            (SenderAccountMessage::GetDeny(_), SenderAccountMessage::GetDeny(_)) => true,
+            #[cfg(test)]
+            (
+                SenderAccountMessage::IsSchedulerEnabled(_),
+                SenderAccountMessage::IsSchedulerEnabled(_),
+            ) => true,
+            _ => false,
+        }
+    }
+}
+
+#[cfg(any(test, feature = "test"))]
+impl Eq for SenderAccountMessage {}
+
 /// A SenderAccount manages the receipts accounting between the indexer and the sender across
 /// multiple allocations.
 ///