feat: Async HTTP outcalls refunds

rs/consensus/src/consensus/metrics.rs

@@ -188,6 +188,7 @@ pub(crate) struct FinalizerMetrics {
     pub canister_http_timeouts_delivered: IntCounter,
     pub canister_http_divergences_delivered: IntCounter,
     pub canister_http_out_of_cycles_delivered: IntCounter,
+    pub canister_http_async_receipts_delivered: IntCounter,
     pub canister_http_flexible_candid_failures: IntCounter,
     pub canister_http_flexible_errors_delivered: IntCounter,
     pub canister_http_payload_bytes_delivered: Histogram,
@@ -304,6 +305,10 @@ impl FinalizerMetrics {
                 "canister_http_out_of_cycles_delivered",
                 "Total number of canister http messages delivered as out of cycles",
             ),
+            canister_http_async_receipts_delivered: metrics_registry.int_counter(
+                "canister_http_async_receipts_delivered",
+                "Total number of canister http asynchronous receipts delivered",
+            ),
             canister_http_flexible_candid_failures: metrics_registry.int_counter(
                 "canister_http_flexible_candid_failures",
                 "Total number of flexible canister http responses skipped due to candid encoding/decoding failures",
@@ -371,6 +376,8 @@ impl FinalizerMetrics {
             .inc_by(batch_stats.canister_http.divergence_responses as u64);
         self.canister_http_out_of_cycles_delivered
             .inc_by(batch_stats.canister_http.out_of_cycles as u64);
+        self.canister_http_async_receipts_delivered
+            .inc_by(batch_stats.canister_http.async_receipts as u64);
 
         let flexible_ok_candid_failures = batch_stats
             .canister_http

rs/https_outcalls/consensus/benches/payload_validation.rs

@@ -1,14 +1,14 @@
 //! Benchmark for the validation of canister HTTP outcall
 //! payloads.
 
-use std::collections::{BTreeMap, BTreeSet, HashSet};
+use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
 
 use criterion::{BenchmarkId, Criterion, black_box, criterion_group, criterion_main};
 
 use ic_consensus_mocks::{Dependencies, DependenciesBuilder};
 use ic_crypto_temp_crypto::{NodeKeysToGenerate, TempCryptoComponent};
-use ic_https_outcalls_consensus::payload_builder::CanisterHttpPayloadBuilderImpl;
+use ic_https_outcalls_consensus::payload_builder::{CanisterHttpPayloadBuilderImpl, PastPayloads};
 use ic_https_outcalls_pricing::fees::{flexible_initial_spent, non_flexible_initial_spent};
 use ic_interfaces::crypto::BasicSigner;
 use ic_interfaces_registry::RegistryClient;
@@ -148,7 +148,7 @@ fn bench_payload_verification(c: &mut Criterion) {
                     black_box(target.builder.validate_canister_http_payload_impl(
                         black_box(&target.payload),
                         black_box(&target.validation_context),
-                        black_box(HashSet::new()),
+                        black_box(PastPayloads::default()),
                     ))
                     .expect("validation failed");
                 })
@@ -470,6 +470,7 @@ impl<'a> PayloadAssembler<'a> {
             divergence_responses,
             flexible_responses,
             flexible_errors: vec![],
+            async_receipts: vec![],
         };
 
         assert!(

rs/https_outcalls/consensus/src/gossip.rs

@@ -37,8 +37,13 @@ impl<Pool: CanisterHttpPool> BouncerFactory<CanisterHttpResponseId, Pool>
                 &latest_state.get_ref().metadata.subnet_call_context_manager;
             let known_request_ids: BTreeSet<_> = subnet_call_context_manger
                 .canister_http_request_contexts
-                .iter()
-                .map(|item| *item.0)
+                .keys()
+                .chain(
+                    subnet_call_context_manger
+                        .delivered_canister_http_request_contexts
+                        .keys(),
+                )
+                .copied()
                 .collect();
             let next_callback_id = subnet_call_context_manger.next_callback_id();
             (known_request_ids, next_callback_id)
@@ -72,3 +77,145 @@ impl<Pool: CanisterHttpPool> BouncerFactory<CanisterHttpResponseId, Pool>
         std::time::Duration::from_secs(3)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use ic_artifact_pool::canister_http_pool::CanisterHttpPoolImpl;
+    use ic_interfaces_state_manager::Labeled;
+    use ic_logger::replica_logger::no_op_logger;
+    use ic_metrics::MetricsRegistry;
+    use ic_registry_subnet_type::SubnetType;
+    use ic_replicated_state::metadata_state::subnet_call_context_manager::SubnetCallContext;
+    use ic_test_utilities::state_manager::RefMockStateManager;
+    use ic_test_utilities_types::{
+        ids::{node_test_id, subnet_test_id},
+        messages::RequestBuilder,
+    };
+    use ic_types::{
+        Height, NumberOfNodes, RegistryVersion, ReplicaVersion,
+        canister_http::{
+            CanisterHttpMethod, CanisterHttpPaymentReceipt, CanisterHttpRequestContext,
+            CanisterHttpResponseMetadata, CanisterHttpResponseReceipt, PricingVersion,
+            RefundStatus, Replication,
+        },
+        crypto::{BasicSig, BasicSigOf, CryptoHash, CryptoHashOf, Signed},
+        signature::BasicSignature,
+        time::UNIX_EPOCH,
+    };
+    use ic_types_cycles::CanisterCyclesCostSchedule;
+    use std::sync::Arc;
+
+    fn request_context() -> CanisterHttpRequestContext {
+        CanisterHttpRequestContext {
+            request: RequestBuilder::new().build(),
+            url: String::new(),
+            max_response_bytes: None,
+            headers: vec![],
+            body: None,
+            http_method: CanisterHttpMethod::GET,
+            transform: None,
+            time: UNIX_EPOCH,
+            replication: Replication::FullyReplicated,
+            pricing_version: PricingVersion::PayAsYouGo,
+            refund_status: RefundStatus::default(),
+            registry_version: RegistryVersion::from(1),
+            subnet_size: NumberOfNodes::from(13),
+            cost_schedule: CanisterCyclesCostSchedule::Normal,
+        }
+    }
+
+    fn share_id(callback_id: CallbackId) -> CanisterHttpResponseId {
+        Signed {
+            content: CanisterHttpResponseReceipt {
+                metadata: CanisterHttpResponseMetadata {
+                    id: callback_id,
+                    content_hash: CryptoHashOf::new(CryptoHash(vec![])),
+                    content_size: 0,
+                    is_reject: false,
+                    replica_version: ReplicaVersion::default(),
+                },
+                payment_receipt: CanisterHttpPaymentReceipt::default(),
+            },
+            signature: BasicSignature {
+                signature: BasicSigOf::new(BasicSig(vec![])),
+                signer: node_test_id(0),
+            },
+        }
+    }
+
+    /// The `next_callback_id` of the state [`test_bouncer`] builds.
+    const NEXT_CALLBACK_ID: u64 = 3;
+
+    /// A bouncer over a state that has handed out callback ids 0, 1 and 2 — so
+    /// `next_callback_id` is [`NEXT_CALLBACK_ID`] — of which 0 is still awaiting a
+    /// response, 1 has been responded to, and 2 is gone for good.
+    fn test_bouncer() -> Bouncer<CanisterHttpResponseId> {
+        let mut state = ReplicatedState::new(subnet_test_id(0), SubnetType::Application);
+        let contexts = &mut state.metadata.subnet_call_context_manager;
+        // Advance `next_callback_id` to NEXT_CALLBACK_ID
+        for _ in 0..NEXT_CALLBACK_ID {
+            contexts.push_context(SubnetCallContext::CanisterHttpRequest(request_context()));
+        }
+        contexts.canister_http_request_contexts.clear();
+        contexts
+            .canister_http_request_contexts
+            .insert(CallbackId::new(0), request_context());
+        contexts
+            .delivered_canister_http_request_contexts
+            .insert(CallbackId::new(1), request_context());
+
+        let state_manager = Arc::new(RefMockStateManager::default());
+        state_manager
+            .get_mut()
+            .expect_get_latest_state()
+            .return_const(Labeled::new(Height::new(1), Arc::new(state)));
+
+        let gossip = CanisterHttpGossipImpl::new(state_manager);
+        let pool = CanisterHttpPoolImpl::new(MetricsRegistry::new(), no_op_logger());
+        gossip.new_bouncer(&pool)
+    }
+
+    #[test]
+    fn shares_of_delivered_contexts_are_wanted() {
+        let bouncer = test_bouncer();
+
+        assert_eq!(bouncer(&share_id(CallbackId::new(0))), BouncerValue::Wants);
+        assert_eq!(bouncer(&share_id(CallbackId::new(1))), BouncerValue::Wants);
+        // A request that is neither pending nor delivered is settled for good.
+        assert_eq!(
+            bouncer(&share_id(CallbackId::new(2))),
+            BouncerValue::Unwanted
+        );
+    }
+
+    #[test]
+    fn shares_of_upcoming_requests_are_wanted() {
+        let bouncer = test_bouncer();
+
+        // The very next id execution will hand out, ...
+        assert_eq!(
+            bouncer(&share_id(CallbackId::new(NEXT_CALLBACK_ID))),
+            BouncerValue::Wants
+        );
+        // ... and everything up to the far edge of the look-ahead window.
+        assert_eq!(
+            bouncer(&share_id(CallbackId::new(
+                NEXT_CALLBACK_ID + MAX_NUMBER_OF_REQUESTS_AHEAD
+            ))),
+            BouncerValue::Wants
+        );
+    }
+
+    #[test]
+    fn shares_beyond_the_look_ahead_window_are_stashed() {
+        let bouncer = test_bouncer();
+
+        assert_eq!(
+            bouncer(&share_id(CallbackId::new(
+                NEXT_CALLBACK_ID + MAX_NUMBER_OF_REQUESTS_AHEAD + 1
+            ))),
+            BouncerValue::MaybeWantsLater
+        );
+    }
+}