[F36] Flushed event-cache queries ignore finality-only filters

massa-event-cache/src/event_cache.rs

@@ -468,18 +468,31 @@ impl EventCache {
         &self,
         filter: &EventFilter,
     ) -> (Vec<u64>, Vec<SCOutputEvent>) {
+        // All events stored in this cache are final: a query for non-final
+        // events can never match anything here
+        if filter.is_final == Some(false) {
+            return (vec![], vec![]);
+        }
+
         // Step 1
         // Build a (sorted) map with key: (counter value, indent), value: filter
         // Will be used to iterate from the lower count index to the highest count index
         // e.g. if index for emitter address is 10 (index count), and origin operation id is 20
         //      iter over emitter address index then origin operation id index
 
-        let filter_items = from_event_filter(filter);
+        let mut filter_items = from_event_filter(filter);
 
         if filter_items.is_empty() {
-            // Note: will return too many event - user should restrict the filter
-            warn!("Filter item only on is final field, please add more filter parameters");
-            return (vec![], vec![]);
+            if filter.is_final == Some(true) {
+                // Explicit request for all finalized events with no other
+                // criteria: scan the whole event column (bounded below by
+                // max_events_per_query) instead of matching nothing
+                filter_items.push((KeyIndent::Event, FilterItem::SlotStart(Slot::new(0, 0))));
+            } else {
+                // Note: will return too many event - user should restrict the filter
+                warn!("No filter parameter provided, please add filter parameters");
+                return (vec![], vec![]);
+            }
         }
 
         let it = filter_items.iter().map(|(key_indent, filter_item)| {

massa-grpc/src/stream/new_slot_transfers.rs

@@ -43,16 +43,30 @@ pub(crate) async fn new_slot_transfers(
     tokio::spawn({
         let execution_controller = grpc.execution_controller.clone();
         async move {
-            let mut finality = FinalityLevel::Unspecified;
+            // Issue #5066: standardize startup with `new_slot_execution_outputs` by waiting
+            // for the client's first request before emitting anything, so the default filter
+            // is applied only once the client has had a chance to configure it and this stream
+            // never emits data under an implicit default.
+            let mut finality = match in_stream.next().await {
+                Some(Ok(first)) => first.finality_level(),
+                // No usable initial request (error or client disconnected): stream nothing.
+                _ => {
+                    error!("empty request");
+                    return;
+                }
+            };
             loop {
                 select! {
                     // Receive a new slot execution traces from the subscriber
                     event = subscriber.recv() => {
                         match event {
                             Ok((massa_slot_execution_trace, is_final)) => {
+                                // A `FinalityLevel::Unspecified` finality level (the default,
+                                // e.g. when the client sends no explicit level) accepts BOTH
+                                // candidate and final transfers, matching the default filter of
+                                // `FilterNewSlotExec` used by `new_slot_execution_outputs`.
                                 if (finality == FinalityLevel::Final && !is_final) ||
-                                    (finality == FinalityLevel::Candidate && is_final) ||
-                                    (finality == FinalityLevel::Unspecified && !is_final) {
+                                    (finality == FinalityLevel::Candidate && is_final) {
                                     continue;
                                 }
                                 let mut ret_transfers = Vec::new();

massa-grpc/src/tests/mock.rs

@@ -38,7 +38,18 @@ pub(crate) fn grpc_public_service(addr: &SocketAddr) -> MassaPublicGrpc {
     let shared_storage: massa_storage::Storage = massa_storage::Storage::create_root();
     let selector_ctrl = Box::new(MockSelectorController::new());
     let pool_ctrl = Box::new(MockPoolController::new());
-    let execution_ctrl = Box::new(MockExecutionController::new());
+    #[allow(unused_mut)]
+    let mut execution_ctrl = MockExecutionController::new();
+    // The new_slot_transfers stream (execution-trace feature) clones the controller and queries
+    // transfers for each processed slot; stub both so streaming tests don't panic on an un-mocked
+    // call.
+    #[cfg(feature = "execution-trace")]
+    execution_ctrl.expect_clone_box().returning(|| {
+        let mut cloned = Box::new(MockExecutionController::new());
+        cloned.expect_get_transfers_for_slot().returning(|_| None);
+        cloned
+    });
+    let execution_ctrl = Box::new(execution_ctrl);
     let protocol_ctrl = Box::new(MockProtocolController::new());
 
     let endorsement_sender = tokio::sync::broadcast::channel(2000).0;

massa-grpc/src/tests/stream.rs

@@ -3,6 +3,8 @@
 use crate::tests::mock::grpc_public_service;
 use core::panic;
 use massa_consensus_exports::MockConsensusController;
+#[cfg(feature = "execution-trace")]
+use massa_execution_exports::SlotAbiCallStack;
 use massa_execution_exports::{ExecutionOutput, MockExecutionController, SlotExecutionOutput};
 use massa_models::{
     address::Address,
@@ -15,6 +17,8 @@ use massa_models::{
     types::{SetOrKeep, SetUpdateOrDelete},
 };
 use massa_pool_exports::MockPoolController;
+#[cfg(feature = "execution-trace")]
+use massa_proto_rs::massa::api::v1::NewSlotTransfersRequest;
 use massa_proto_rs::massa::{
     api::v1::{
         public_service_client::PublicServiceClient, NewBlocksRequest, NewFilledBlocksRequest,
@@ -1827,3 +1831,84 @@ async fn send_blocks() {
 
     stop_handle.stop();
 }
+
+// Regression test for issue #5066: new_slot_transfers must match new_slot_execution_outputs
+// startup behavior — it waits for the client's first request before emitting anything, and its
+// default finality level (Unspecified) accepts BOTH candidate and final transfers.
+#[cfg(feature = "execution-trace")]
+#[tokio::test]
+async fn new_slot_transfers_default_finality() {
+    let addr: SocketAddr = "[::]:4033".parse().unwrap();
+    let mut public_server = grpc_public_service(&addr);
+    let config = public_server.grpc_config.clone();
+
+    let (trace_tx, _trace_rx) = tokio::sync::broadcast::channel(10);
+    public_server
+        .execution_channels
+        .slot_execution_traces_sender = trace_tx.clone();
+
+    let stop_handle = public_server.serve(&config).await.unwrap();
+
+    // An empty call stack still yields a (transfer-less) response once it passes the finality
+    // filter, which is all we need to observe startup and default-finality behavior.
+    let empty_trace = |slot| SlotAbiCallStack {
+        slot,
+        asc_call_stacks: vec![],
+        deferred_call_stacks: Default::default(),
+        operation_call_stacks: Default::default(),
+    };
+
+    let mut public_client = PublicServiceClient::connect(format!(
+        "grpc://localhost:{}",
+        addr.to_string().split(':').last().unwrap()
+    ))
+    .await
+    .unwrap();
+
+    let (tx_request, rx) = tokio::sync::mpsc::channel(10);
+    let request_stream = tokio_stream::wrappers::ReceiverStream::new(rx);
+    let mut resp_stream = public_client
+        .new_slot_transfers(request_stream)
+        .await
+        .unwrap()
+        .into_inner();
+
+    // Emit a FINAL trace before the client has sent any request. The stream must NOT emit it yet
+    // (it waits for the first request). Before the fix it would be sent immediately.
+    trace_tx.send((empty_trace(Slot::new(1, 0)), true)).unwrap();
+    assert!(
+        tokio::time::timeout(Duration::from_millis(300), resp_stream.next())
+            .await
+            .is_err(),
+        "stream emitted data before the client's first request"
+    );
+
+    // Send an empty request => finality level Unspecified (the default).
+    tx_request
+        .send(NewSlotTransfersRequest { finality_level: 0 })
+        .await
+        .unwrap();
+
+    // The buffered final trace is now delivered (Unspecified accepts final).
+    let final_resp = tokio::time::timeout(Duration::from_secs(5), resp_stream.next())
+        .await
+        .expect("no final transfer response under default finality")
+        .unwrap()
+        .unwrap();
+    assert_eq!(final_resp.slot, Some(Slot::new(1, 0).into()));
+
+    // A CANDIDATE (non-final) trace must ALSO be delivered under the default. Before the fix
+    // Unspecified meant final-only and this would time out.
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    trace_tx
+        .send((empty_trace(Slot::new(1, 1)), false))
+        .unwrap();
+    let cand_resp = tokio::time::timeout(Duration::from_secs(5), resp_stream.next())
+        .await
+        .expect("candidate transfer dropped under default finality (issue #5066)")
+        .unwrap()
+        .unwrap();
+    assert_eq!(cand_resp.slot, Some(Slot::new(1, 1).into()));
+
+    stop_handle.stop();
+}