refactor(sender_allocation_task): add receipt validation

Author: suchapalaver

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

@@ -37,7 +37,7 @@ pub struct SenderAllocationTask<T: NetworkVersion> {
     _phantom: PhantomData<T>,
 }
 
-/// Simple state structure for the task
+/// Simple state structure for the task (will be enhanced incrementally)
 struct TaskState {
     /// Sum of all receipt fees for the current allocation
     unaggregated_fees: UnaggregatedReceipts,
@@ -115,27 +115,39 @@ impl<T: NetworkVersion> SenderAllocationTask<T> {
         Ok(())
     }
 
-    /// Handle new receipt - simplified version
+    /// Handle new receipt - with basic validation
     async fn handle_new_receipt(
         state: &mut TaskState,
         notification: NewReceiptNotification,
     ) -> anyhow::Result<()> {
-        // For now, just accept all receipts as valid
-        // In the full implementation, this would include validation
-
-        let value = match notification {
-            NewReceiptNotification::V1(ref n) => n.value,
-            NewReceiptNotification::V2(ref n) => n.value,
+        let (id, value, timestamp_ns) = match notification {
+            NewReceiptNotification::V1(ref n) => (n.id, n.value, n.timestamp_ns),
+            NewReceiptNotification::V2(ref n) => (n.id, n.value, n.timestamp_ns),
         };
 
-        let timestamp_ns = match notification {
-            NewReceiptNotification::V1(ref n) => n.timestamp_ns,
-            NewReceiptNotification::V2(ref n) => n.timestamp_ns,
-        };
+        // Basic receipt ID validation - reject already processed receipts
+        if id <= state.unaggregated_fees.last_id {
+            tracing::warn!(
+                allocation_id = ?state.allocation_id,
+                receipt_id = id,
+                last_processed_id = state.unaggregated_fees.last_id,
+                "Rejecting receipt with ID <= last processed ID"
+            );
+            return Ok(()); // Silently ignore duplicate/old receipts
+        }
 
-        // Update local state
+        // Update local state with new receipt
         state.unaggregated_fees.value += value;
         state.unaggregated_fees.counter += 1;
+        state.unaggregated_fees.last_id = id;
+
+        tracing::debug!(
+            allocation_id = ?state.allocation_id,
+            receipt_id = id,
+            value = value,
+            new_total = state.unaggregated_fees.value,
+            "Processed new receipt"
+        );
 
         // Notify parent
         state
@@ -149,14 +161,28 @@ impl<T: NetworkVersion> SenderAllocationTask<T> {
         Ok(())
     }
 
-    /// Handle RAV request - simplified version
+    /// Handle RAV request - enhanced but still simplified version
     async fn handle_rav_request(state: &mut TaskState) -> anyhow::Result<()> {
-        let _start_time = Instant::now();
+        let start_time = Instant::now();
+
+        // Check if there are any receipts to aggregate
+        if state.unaggregated_fees.value == 0 {
+            tracing::debug!(
+                allocation_id = ?state.allocation_id,
+                "No receipts to aggregate, skipping RAV request"
+            );
+            return Ok(());
+        }
 
-        // For now, simulate a successful RAV request
-        // In the full implementation, this would make actual gRPC calls
+        tracing::info!(
+            allocation_id = ?state.allocation_id,
+            receipt_count = state.unaggregated_fees.counter,
+            total_value = state.unaggregated_fees.value,
+            "Creating RAV for aggregated receipts"
+        );
 
-        // Create a dummy RAV info
+        // TODO: Replace with real TAP manager integration
+        // For now, simulate a successful RAV request
         let rav_info = RavInformation {
             allocation_id: match state.allocation_id {
                 AllocationId::Legacy(id) => id.into_inner(),
@@ -168,15 +194,26 @@ impl<T: NetworkVersion> SenderAllocationTask<T> {
             value_aggregate: state.unaggregated_fees.value,
         };
 
+        // Store the fees we're about to clear for the response
+        let fees_to_clear = state.unaggregated_fees;
+
         // Reset local fees since they're now covered by RAV
         state.unaggregated_fees = UnaggregatedReceipts::default();
 
+        let elapsed = start_time.elapsed();
+        tracing::info!(
+            allocation_id = ?state.allocation_id,
+            rav_value = rav_info.value_aggregate,
+            duration_ms = elapsed.as_millis(),
+            "RAV creation completed successfully"
+        );
+
         // Notify parent of successful RAV
         state
             .sender_account_handle
             .cast(SenderAccountMessage::UpdateReceiptFees(
                 state.allocation_id,
-                ReceiptFees::RavRequestResponse(state.unaggregated_fees, Ok(Some(rav_info))),
+                ReceiptFees::RavRequestResponse(fees_to_clear, Ok(Some(rav_info))),
             ))
             .await?;
 
@@ -238,6 +275,97 @@ mod tests {
                 .unwrap();
 
         // Check it's the right type of message
-        matches!(parent_message, SenderAccountMessage::UpdateReceiptFees(..));
+        assert!(matches!(
+            parent_message,
+            SenderAccountMessage::UpdateReceiptFees(..)
+        ));
+    }
+
+    #[tokio::test]
+    async fn test_receipt_id_validation() {
+        let lifecycle = LifecycleManager::new();
+
+        // Create a dummy parent handle for testing
+        let (parent_tx, mut parent_rx) = mpsc::channel(10);
+        let parent_handle = TaskHandle::new_for_test(
+            parent_tx,
+            Some("test_parent".to_string()),
+            std::sync::Arc::new(lifecycle.clone()),
+        );
+
+        let allocation_id =
+            AllocationId::Legacy(thegraph_core::AllocationId::new([1u8; 20].into()));
+
+        let task_handle = SenderAllocationTask::<Legacy>::spawn_simple(
+            &lifecycle,
+            Some("test_allocation".to_string()),
+            allocation_id,
+            parent_handle,
+        )
+        .await
+        .unwrap();
+
+        // Send first receipt (should be accepted)
+        let notification1 = NewReceiptNotification::V1(
+            super::super::sender_accounts_manager::NewReceiptNotificationV1 {
+                id: 100,
+                allocation_id: thegraph_core::AllocationId::new([1u8; 20].into()).into_inner(),
+                signer_address: thegraph_core::alloy::primitives::Address::ZERO,
+                timestamp_ns: 1000,
+                value: 100,
+            },
+        );
+
+        task_handle
+            .cast(SenderAllocationMessage::NewReceipt(notification1))
+            .await
+            .unwrap();
+
+        // Receive first update
+        let _first_message = parent_rx.recv().await.unwrap();
+
+        // Send second receipt with same ID (should be rejected silently)
+        let notification2 = NewReceiptNotification::V1(
+            super::super::sender_accounts_manager::NewReceiptNotificationV1 {
+                id: 100, // Same ID - should be rejected
+                allocation_id: thegraph_core::AllocationId::new([1u8; 20].into()).into_inner(),
+                signer_address: thegraph_core::alloy::primitives::Address::ZERO,
+                timestamp_ns: 2000,
+                value: 200,
+            },
+        );
+
+        task_handle
+            .cast(SenderAllocationMessage::NewReceipt(notification2))
+            .await
+            .unwrap();
+
+        // Send third receipt with higher ID (should be accepted)
+        let notification3 = NewReceiptNotification::V1(
+            super::super::sender_accounts_manager::NewReceiptNotificationV1 {
+                id: 101, // Higher ID - should be accepted
+                allocation_id: thegraph_core::AllocationId::new([1u8; 20].into()).into_inner(),
+                signer_address: thegraph_core::alloy::primitives::Address::ZERO,
+                timestamp_ns: 3000,
+                value: 300,
+            },
+        );
+
+        task_handle
+            .cast(SenderAllocationMessage::NewReceipt(notification3))
+            .await
+            .unwrap();
+
+        // Should only receive one more message (for the third receipt)
+        let second_message =
+            tokio::time::timeout(std::time::Duration::from_millis(100), parent_rx.recv())
+                .await
+                .unwrap()
+                .unwrap();
+
+        assert!(matches!(
+            second_message,
+            SenderAccountMessage::UpdateReceiptFees(..)
+        ));
     }
 }