test(tap_integration_tests): ensure rav timestamp cutoff across workspaces in the library (#149)

* test(tap_integration_tests): Added test to check that tap_manager is correctly filtering receipts on timestamp

Signed-off-by: Severiano Sisneros <[email protected]>

* test(tap_integration_tests): Added a test to check that tap_aggregator is correctly rejecting/accepting receipts/rav requests based off timestamp.

Signed-off-by: Severiano Sisneros <[email protected]>

* test(tap_integration_tests): Added a check that RAV aggregate value matches expected value.

Signed-off-by: Severiano Sisneros <[email protected]>

---------

Signed-off-by: Severiano Sisneros <[email protected]>
Co-authored-by: Alexis Asseman <[email protected]>

Author: severiano-sisneros

tap_integration_tests/tests/indexer_mock/mod.rs

@@ -140,7 +140,14 @@ impl<
 
             // Create the aggregate_receipts request params
             let time_stamp_buffer = 0;
-            match request_rav(&self.manager, time_stamp_buffer, &self.aggregator_client).await {
+            match request_rav(
+                &self.manager,
+                time_stamp_buffer,
+                &self.aggregator_client,
+                self.threshold as usize,
+            )
+            .await
+            {
                 Ok(_) => Ok(()),
                 Err(e) => Err(to_rpc_error(e.into(), "Failed to request rav")),
             }
@@ -208,15 +215,11 @@ async fn request_rav<
     manager: &Arc<Mutex<Manager<CA, RCA, RSA, RAVSA>>>, // Mutex-protected manager object for thread safety
     time_stamp_buffer: u64, // Buffer for timestamping, see tap_core for details
     aggregator_client: &(HttpClient, String), // HttpClient for making requests to the tap_aggregator server
+    threshold: usize,
 ) -> Result<()> {
     // Create the aggregate_receipts request params
     let rav_request = manager.lock().await.create_rav_request(time_stamp_buffer)?;
 
-    match rav_request.invalid_receipts.is_empty() {
-        true => Ok(()),
-        false => Err(Error::msg("Invalid receipts found")),
-    }?;
-
     // To-do: Need to add previous RAV, when tap_manager supports replacing receipts
     let params = rpc_params!(
         &aggregator_client.1,
@@ -234,6 +237,14 @@ async fn request_rav<
         let _result =
             manager_guard.verify_and_store_rav(rav_request.expected_rav, remote_rav_result.data)?;
     }
+
+    // For these tests, we expect every receipt to be valid, i.e. there should be no invalid receipts, nor any missing receipts (less than the expected threshold).
+    // If there is throw an error.
+    match rav_request.invalid_receipts.is_empty() && (rav_request.valid_receipts.len() == threshold)
+    {
+        true => Ok(()),
+        false => Err(Error::msg("Invalid receipts found")),
+    }?;
     Ok(())
 }
 

tap_integration_tests/tests/showcase.rs

@@ -18,11 +18,13 @@ use ethers::{
     signers::{coins_bip39::English, LocalWallet, MnemonicBuilder, Signer},
     types::{Address, H160},
 };
-use jsonrpsee::{core::client::ClientT, http_client::HttpClientBuilder, server::ServerHandle};
+use jsonrpsee::{
+    core::client::ClientT, http_client::HttpClientBuilder, rpc_params, server::ServerHandle,
+};
 use rand::{rngs::StdRng, Rng, SeedableRng};
 use rstest::*;
 
-use tap_aggregator::server as agg_server;
+use tap_aggregator::{jsonrpsee_helpers, server as agg_server};
 use tap_core::{
     adapters::{
         collateral_adapter_mock::CollateralAdapterMock,
@@ -31,8 +33,8 @@ use tap_core::{
         receipt_storage_adapter_mock::ReceiptStorageAdapterMock,
     },
     eip_712_signed_message::EIP712SignedMessage,
-    tap_receipt::ReceiptCheck,
-    tap_receipt::{Receipt, ReceivedReceipt},
+    tap_manager::SignedRAV,
+    tap_receipt::{Receipt, ReceiptCheck, ReceivedReceipt},
 };
 
 use crate::indexer_mock;
@@ -80,7 +82,7 @@ fn receipt_threshold_1(num_queries: u64, num_batches: u64) -> u64 {
 // The number of receipts collected before Indexer 2 sends a RAV request
 #[fixture]
 fn receipt_threshold_2(num_queries: u64, num_batches: u64) -> u64 {
-    num_queries * num_batches
+    num_queries * num_batches / 2
 }
 
 // The private key (LocalWallet) and public key (Address) of a Gateway
@@ -279,6 +281,72 @@ async fn requests_2(
     Ok(requests)
 }
 
+#[fixture]
+async fn repeated_timestamp_request(
+    keys_gateway: (LocalWallet, Address),
+    query_price: Vec<u128>,
+    allocation_ids: Vec<H160>,
+    num_batches: u64,
+    receipt_threshold_1: u64,
+) -> Result<Vec<(EIP712SignedMessage<Receipt>, u64)>> {
+    let (gateway_key, _) = keys_gateway;
+
+    // Create signed receipts
+    let mut requests =
+        generate_requests(query_price, num_batches, &gateway_key, allocation_ids[0]).await?;
+
+    // Create a new receipt with the timestamp equal to the latest receipt in the first RAV request batch
+    let repeat_timestamp = requests[receipt_threshold_1 as usize - 1]
+        .0
+        .message
+        .timestamp_ns;
+    let target_receipt = &requests[receipt_threshold_1 as usize].0.message;
+    let repeat_receipt = Receipt {
+        allocation_id: target_receipt.allocation_id,
+        timestamp_ns: repeat_timestamp,
+        nonce: target_receipt.nonce,
+        value: target_receipt.value,
+    };
+
+    // Sign the new receipt and insert it in the second batch
+    requests[receipt_threshold_1 as usize].0 =
+        EIP712SignedMessage::new(repeat_receipt, &gateway_key).await?;
+    Ok(requests)
+}
+
+#[fixture]
+async fn repeated_timestamp_incremented_by_one_request(
+    keys_gateway: (LocalWallet, Address),
+    query_price: Vec<u128>,
+    allocation_ids: Vec<H160>,
+    num_batches: u64,
+    receipt_threshold_1: u64,
+) -> Result<Vec<(EIP712SignedMessage<Receipt>, u64)>> {
+    let (gateway_key, _) = keys_gateway;
+    // Create your Receipt here
+    let mut requests =
+        generate_requests(query_price, num_batches, &gateway_key, allocation_ids[0]).await?;
+
+    // Create a new receipt with the timestamp equal to the latest receipt timestamp+1 in the first RAV request batch
+    let repeat_timestamp = requests[receipt_threshold_1 as usize - 1]
+        .0
+        .message
+        .timestamp_ns
+        + 1;
+    let target_receipt = &requests[receipt_threshold_1 as usize].0.message;
+    let repeat_receipt = Receipt {
+        allocation_id: target_receipt.allocation_id,
+        timestamp_ns: repeat_timestamp,
+        nonce: target_receipt.nonce,
+        value: target_receipt.value,
+    };
+
+    // Sign the new receipt and insert it in the second batch
+    requests[receipt_threshold_1 as usize].0 =
+        EIP712SignedMessage::new(repeat_receipt, &gateway_key).await?;
+    Ok(requests)
+}
+
 #[fixture]
 async fn wrong_requests(
     wrong_keys_gateway: (LocalWallet, Address),
@@ -634,6 +702,175 @@ async fn test_manager_wrong_requestor_keys(
     Ok(())
 }
 
+#[rstest]
+#[tokio::test]
+async fn test_tap_manager_rav_timestamp_cuttoff(
+    #[future] two_indexers_test_servers: Result<
+        (
+            ServerHandle,
+            SocketAddr,
+            ServerHandle,
+            SocketAddr,
+            ServerHandle,
+            SocketAddr,
+        ),
+        Error,
+    >,
+    #[future] repeated_timestamp_request: Result<Vec<(EIP712SignedMessage<Receipt>, u64)>>,
+    #[future] repeated_timestamp_incremented_by_one_request: Result<
+        Vec<(EIP712SignedMessage<Receipt>, u64)>,
+    >,
+    receipt_threshold_1: u64,
+) -> Result<(), Box<dyn std::error::Error>> {
+    // This test checks that tap_core is correctly filtering receipts by timestamp.
+    let (
+        server_handle_1,
+        socket_addr_1,
+        _server_handle_2,
+        socket_addr_2,
+        _gateway_handle,
+        _gateway_addr,
+    ) = two_indexers_test_servers.await?;
+
+    let indexer_1_address = "http://".to_string() + &socket_addr_1.to_string();
+    let indexer_2_address = "http://".to_string() + &socket_addr_2.to_string();
+    let client_1 = HttpClientBuilder::default().build(indexer_1_address)?;
+    let client_2 = HttpClientBuilder::default().build(indexer_2_address)?;
+    let requests = repeated_timestamp_request.await?;
+
+    let mut counter = 1;
+    for (receipt_1, id) in requests {
+        let result = client_1.request("request", (id, receipt_1)).await;
+
+        // The first receipt in the second batch has the same timestamp as the last receipt in the first batch.
+        // TAP manager should ignore this receipt when creating the second RAV request.
+        // The indexer_mock will throw an error if the number of receipts in RAV request is less than the expected number.
+        // An error is expected when requesting the second RAV.
+        if counter == 2 * receipt_threshold_1 {
+            match result {
+                Ok(()) => panic!("Should have failed RAV request"),
+                Err(_) => {}
+            }
+        } else {
+            match result {
+                Ok(()) => {}
+                Err(e) => panic!("Error making receipt request: {:?}", e),
+            }
+        }
+        counter += 1;
+    }
+
+    server_handle_1.stop()?;
+
+    // Here the timestamp first receipt in the second batch is equal to timestamp + 1 of the last receipt in the first batch.
+    // No errors are expected.
+    let requests = repeated_timestamp_incremented_by_one_request.await?;
+    for (receipt_1, id) in requests {
+        let result = client_2.request("request", (id, receipt_1)).await;
+        match result {
+            Ok(()) => {}
+            Err(e) => panic!("Error making receipt request: {:?}", e),
+        }
+    }
+    Ok(())
+}
+
+#[rstest]
+#[tokio::test]
+async fn test_tap_aggregator_rav_timestamp_cuttoff(
+    keys_gateway: (LocalWallet, Address),
+    http_request_size_limit: u32,
+    http_response_size_limit: u32,
+    http_max_concurrent_connections: u32,
+    #[future] repeated_timestamp_request: Result<Vec<(EIP712SignedMessage<Receipt>, u64)>>,
+    #[future] repeated_timestamp_incremented_by_one_request: Result<
+        Vec<(EIP712SignedMessage<Receipt>, u64)>,
+    >,
+    receipt_threshold_1: u64,
+) -> Result<(), Box<dyn std::error::Error>> {
+    // This test checks that tap_aggregator is correctly rejecting receipts with invalid timestamps
+    let (gateway_handle, gateway_addr) = start_gateway_aggregator(
+        keys_gateway,
+        http_request_size_limit,
+        http_response_size_limit,
+        http_max_concurrent_connections,
+    )
+    .await?;
+    let client =
+        HttpClientBuilder::default().build(format!("http://{}", gateway_addr.to_string()))?;
+
+    // This is the first part of the test, two batches of receipts are sent to the aggregator.
+    // The second batch has one receipt with the same timestamp as the latest receipt in the first batch.
+    // The first RAV will have the same timestamp as one receipt in the second batch.
+    // tap_aggregator should reject the second RAV request due to the repeated timestamp.
+    let requests = repeated_timestamp_request.await?;
+    let first_batch = &requests[0..receipt_threshold_1 as usize];
+    let second_batch = &requests[receipt_threshold_1 as usize..2 * receipt_threshold_1 as usize];
+
+    let receipts = first_batch
+        .iter()
+        .map(|(r, _)| r.clone())
+        .collect::<Vec<_>>();
+    let params = rpc_params!(&aggregate_server_api_version(), &receipts, None::<()>);
+    let first_rav_response: jsonrpsee_helpers::JsonRpcResponse<SignedRAV> =
+        client.request("aggregate_receipts", params).await?;
+
+    let receipts = second_batch
+        .iter()
+        .map(|(r, _)| r.clone())
+        .collect::<Vec<_>>();
+    let params = rpc_params!(
+        &aggregate_server_api_version(),
+        &receipts,
+        first_rav_response.data
+    );
+    let second_rav_response: Result<
+        jsonrpsee_helpers::JsonRpcResponse<SignedRAV>,
+        jsonrpsee::core::Error,
+    > = client.request("aggregate_receipts", params).await;
+    match second_rav_response {
+        Ok(_) => panic!("Should have failed RAV request"),
+        Err(_) => {}
+    }
+
+    // This is the second part of the test, two batches of receipts are sent to the aggregator.
+    // The second batch has one receipt with the timestamp = timestamp+1 of the latest receipt in the first batch.
+    // tap_aggregator should accept the second RAV request.
+    let requests = repeated_timestamp_incremented_by_one_request.await?;
+    let first_batch = &requests[0..receipt_threshold_1 as usize];
+    let second_batch = &requests[receipt_threshold_1 as usize..2 * receipt_threshold_1 as usize];
+
+    let receipts = first_batch
+        .iter()
+        .map(|(r, _)| r.clone())
+        .collect::<Vec<_>>();
+    let params = rpc_params!(&aggregate_server_api_version(), &receipts, None::<()>);
+    let first_rav_response: jsonrpsee_helpers::JsonRpcResponse<SignedRAV> =
+        client.request("aggregate_receipts", params).await?;
+
+    let receipts = second_batch
+        .iter()
+        .map(|(r, _)| r.clone())
+        .collect::<Vec<_>>();
+    let params = rpc_params!(
+        &aggregate_server_api_version(),
+        &receipts,
+        first_rav_response.data
+    );
+    let second_rav_response: jsonrpsee_helpers::JsonRpcResponse<SignedRAV> =
+        client.request("aggregate_receipts", params).await?;
+
+    // Compute the expected aggregate value and check that it matches the latest RAV.
+    let mut expected_value = 0;
+    for (receipt, _) in first_batch.iter().chain(second_batch.iter()) {
+        expected_value += receipt.message.value;
+    }
+    assert!(expected_value == second_rav_response.data.message.value_aggregate);
+
+    gateway_handle.stop()?;
+    Ok(())
+}
+
 async fn generate_requests(
     query_price: Vec<u128>,
     num_batches: u64,