feat(drive-abci): preliminary support for event subscriptions in Platform

packages/dapi-grpc/protos/platform/v0/platform.proto

@@ -6,6 +6,63 @@ package org.dash.platform.dapi.v0;
 
 import "google/protobuf/timestamp.proto";
 
+// Platform events subscription (v0)
+message PlatformSubscriptionRequest {
+  message PlatformSubscriptionRequestV0 {
+    string client_subscription_id = 1;
+    PlatformFilterV0 filter = 2;
+  }
+  oneof version { PlatformSubscriptionRequestV0 v0 = 1; }
+}
+
+message PlatformSubscriptionResponse {
+  message PlatformSubscriptionResponseV0 {
+    string client_subscription_id = 1;
+    PlatformEventV0 event = 2;
+  }
+  oneof version { PlatformSubscriptionResponseV0 v0 = 1; }
+}
+
+// Initial placeholder filter and event to be refined during integration
+// Filter for StateTransitionResult events
+message StateTransitionResultFilter {
+  // When set, only match StateTransitionResult events for this tx hash.
+  optional bytes tx_hash = 1;
+}
+
+message PlatformFilterV0 {
+  oneof kind {
+    bool all = 1;             // subscribe to all platform events
+    bool block_committed = 2; // subscribe to BlockCommitted events only
+    StateTransitionResultFilter state_transition_result =
+        3; // subscribe to StateTransitionResult events (optionally filtered by
+           // tx_hash)
+  }
+}
+
+message PlatformEventV0 {
+  message BlockMetadata {
+    uint64 height = 1 [ jstype = JS_STRING ];
+    uint64 time_ms = 2 [ jstype = JS_STRING ];
+    bytes block_id_hash = 3;
+  }
+
+  message BlockCommitted {
+    BlockMetadata meta = 1;
+    uint32 tx_count = 2;
+  }
+
+  message StateTransitionFinalized {
+    BlockMetadata meta = 1;
+    bytes tx_hash = 2;
+  }
+
+  oneof event {
+    BlockCommitted block_committed = 1;
+    StateTransitionFinalized state_transition_finalized = 2;
+  }
+}
+
 service Platform {
   rpc broadcastStateTransition(BroadcastStateTransitionRequest)
       returns (BroadcastStateTransitionResponse);
@@ -104,6 +161,10 @@ service Platform {
   rpc getGroupActions(GetGroupActionsRequest) returns (GetGroupActionsResponse);
   rpc getGroupActionSigners(GetGroupActionSignersRequest)
       returns (GetGroupActionSignersResponse);
+
+  // Bi-directional stream for multiplexed platform events subscriptions
+  rpc SubscribePlatformEvents(PlatformSubscriptionRequest)
+      returns (stream PlatformSubscriptionResponse);
 }
 
 // Proof message includes cryptographic proofs for validating responses

packages/rs-dapi/doc/DESIGN.md

@@ -264,7 +264,7 @@ Operational notes:
 - Access logging: HTTP/JSON-RPC and gRPC traffic share the same access logging layer when configured, so all protocols emit uniform access entries
 
 - Platform event streaming is handled via a direct upstream proxy:
-  - `subscribePlatformEvents` simply forwards every inbound command stream to a single Drive connection and relays responses back without multiplexing
+  - `subscribePlatformEvents` forwards each inbound subscription request to Drive and relays the resulting event stream without additional multiplexing
 
 #### Key Features
 - **Modular Organization**: Complex methods separated into dedicated modules for maintainability
@@ -283,12 +283,11 @@ Operational notes:
 rs-dapi exposes `subscribePlatformEvents` as a server-streaming endpoint and currently performs a straightforward pass-through to rs-drive-abci.
 
 - Public interface:
-  - Bi-directional gRPC stream: `subscribePlatformEvents(request stream PlatformEventsCommand) -> (response stream PlatformEventsResponse)`.
-  - Commands (`Add`, `Remove`, `Ping`) and responses (`Event`, `Ack`, `Error`) stay in their protobuf `V0` envelopes end-to-end.
+  - Server-streaming gRPC method: `subscribePlatformEvents(PlatformSubscriptionRequest) -> (stream PlatformSubscriptionResponse)`.
+  - Requests carry versioned `V0` envelopes with the client subscription id and filter; responses return the same id alongside the event payload.
 
 - Upstream behavior:
-  - Each client stream obtains its own upstream Drive connection; tokio channels forward commands upstream and pipe responses back downstream without pooling.
-  - The `EventMux` from `rs-dash-event-bus` is retained for future multiplexing work but does not alter traffic today.
+  - Each client stream obtains its own upstream Drive connection; responses are forwarded downstream as they arrive without intermediate pooling or command channels.
 
 - Observability:
   - Standard `tracing` logging wraps the forwarders, and the proxy participates in the existing `/metrics` exporter via shared counters.

packages/rs-dapi/src/main.rs

@@ -111,7 +111,7 @@ impl Cli {
                     "rs-dapi server initializing",
                 );
 
-                let mut server_future = run_server(config, access_logger);
+                let server_future = run_server(config, access_logger);
                 tokio::pin!(server_future);
 
                 let outcome = tokio::select! {
@@ -130,8 +130,8 @@ impl Cli {
                     }
                 };
 
-                if let Some(result) = outcome {
-                    if let Err(e) = result {
+                if let Some(result) = outcome
+                    && let Err(e) = result {
                         error!("Server error: {}", e);
 
                         // Check if this is a connection-related error and set appropriate exit code
@@ -161,7 +161,6 @@ impl Cli {
                             }
                         }
                     }
-                }
                 Ok(())
             }
             Commands::Config => dump_config(&config),

packages/rs-dapi/src/metrics.rs

@@ -66,12 +66,8 @@ pub enum Metric {
     RequestDuration,
     /// Platform events: active sessions gauge
     PlatformEventsActiveSessions,
-    /// Platform events: commands processed, labels [op]
-    PlatformEventsCommands,
     /// Platform events: forwarded events counter
     PlatformEventsForwardedEvents,
-    /// Platform events: forwarded acks counter
-    PlatformEventsForwardedAcks,
     /// Platform events: forwarded errors counter
     PlatformEventsForwardedErrors,
     /// Platform events: upstream streams started counter
@@ -91,11 +87,9 @@ impl Metric {
             Metric::RequestCount => "rsdapi_requests_total",
             Metric::RequestDuration => "rsdapi_request_duration_seconds",
             Metric::PlatformEventsActiveSessions => "rsdapi_platform_events_active_sessions",
-            Metric::PlatformEventsCommands => "rsdapi_platform_events_commands_total",
             Metric::PlatformEventsForwardedEvents => {
                 "rsdapi_platform_events_forwarded_events_total"
             }
-            Metric::PlatformEventsForwardedAcks => "rsdapi_platform_events_forwarded_acks_total",
             Metric::PlatformEventsForwardedErrors => {
                 "rsdapi_platform_events_forwarded_errors_total"
             }
@@ -120,9 +114,7 @@ impl Metric {
             Metric::PlatformEventsActiveSessions => {
                 "Current number of active Platform events sessions"
             }
-            Metric::PlatformEventsCommands => "Platform events commands processed by operation",
             Metric::PlatformEventsForwardedEvents => "Platform events forwarded to clients",
-            Metric::PlatformEventsForwardedAcks => "Platform acks forwarded to clients",
             Metric::PlatformEventsForwardedErrors => "Platform errors forwarded to clients",
             Metric::PlatformEventsUpstreamStreams => {
                 "Upstream subscribePlatformEvents streams started"
@@ -159,7 +151,6 @@ pub enum Label {
     // TODO: ensure we have a limited set of endpoints, so that cardinality is controlled and we don't overload Prometheus
     Endpoint,
     Status,
-    Op,
 }
 
 impl Label {
@@ -172,7 +163,6 @@ impl Label {
             Label::Protocol => "protocol",
             Label::Endpoint => "endpoint",
             Label::Status => "status",
-            Label::Op => "op",
         }
     }
 }
@@ -251,15 +241,6 @@ pub static REQUEST_DURATION_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
     .expect("create histogram vec")
 });
 
-pub static PLATFORM_EVENTS_COMMANDS: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        Metric::PlatformEventsCommands.name(),
-        Metric::PlatformEventsCommands.help(),
-        &[Label::Op.name()]
-    )
-    .expect("create counter vec")
-});
-
 pub static PLATFORM_EVENTS_FORWARDED_EVENTS: Lazy<IntCounter> = Lazy::new(|| {
     register_int_counter!(
         Metric::PlatformEventsForwardedEvents.name(),
@@ -268,14 +249,6 @@ pub static PLATFORM_EVENTS_FORWARDED_EVENTS: Lazy<IntCounter> = Lazy::new(|| {
     .expect("create counter")
 });
 
-pub static PLATFORM_EVENTS_FORWARDED_ACKS: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        Metric::PlatformEventsForwardedAcks.name(),
-        Metric::PlatformEventsForwardedAcks.help()
-    )
-    .expect("create counter")
-});
-
 pub static PLATFORM_EVENTS_FORWARDED_ERRORS: Lazy<IntCounter> = Lazy::new(|| {
     register_int_counter!(
         Metric::PlatformEventsForwardedErrors.name(),
@@ -533,21 +506,11 @@ pub fn platform_events_active_sessions_dec() {
     PLATFORM_EVENTS_ACTIVE_SESSIONS.dec();
 }
 
-#[inline]
-pub fn platform_events_command(op: &str) {
-    PLATFORM_EVENTS_COMMANDS.with_label_values(&[op]).inc();
-}
-
 #[inline]
 pub fn platform_events_forwarded_event() {
     PLATFORM_EVENTS_FORWARDED_EVENTS.inc();
 }
 
-#[inline]
-pub fn platform_events_forwarded_ack() {
-    PLATFORM_EVENTS_FORWARDED_ACKS.inc();
-}
-
 #[inline]
 pub fn platform_events_forwarded_error() {
     PLATFORM_EVENTS_FORWARDED_ERRORS.inc();

packages/rs-dapi/src/services/platform_service/mod.rs

@@ -4,6 +4,7 @@
 mod broadcast_state_transition;
 mod error_mapping;
 mod get_status;
+mod subscribe_platform_events;
 mod wait_for_state_transition_result;
 
 use dapi_grpc::platform::v0::platform_server::Platform;
@@ -167,6 +168,20 @@ impl PlatformServiceImpl {
 impl Platform for PlatformServiceImpl {
     // Manually implemented methods
 
+    // Streaming: multiplexed platform events
+    type SubscribePlatformEventsStream = tokio_stream::wrappers::ReceiverStream<
+        Result<dapi_grpc::platform::v0::PlatformSubscriptionResponse, dapi_grpc::tonic::Status>,
+    >;
+
+    async fn subscribe_platform_events(
+        &self,
+        request: dapi_grpc::tonic::Request<dapi_grpc::platform::v0::PlatformSubscriptionRequest>,
+    ) -> Result<
+        dapi_grpc::tonic::Response<Self::SubscribePlatformEventsStream>,
+        dapi_grpc::tonic::Status,
+    > {
+        self.subscribe_platform_events_impl(request).await
+    }
     /// Get the status of the whole system
     ///
     /// This method retrieves the current status of Drive, Tenderdash, and other components.

packages/rs-dapi/src/services/platform_service/subscribe_platform_events.rs

@@ -0,0 +1,84 @@
+use crate::{DapiError, metrics};
+use dapi_grpc::platform::v0::{PlatformSubscriptionRequest, PlatformSubscriptionResponse};
+use dapi_grpc::tonic::{Request, Response, Status};
+use futures::StreamExt;
+use std::sync::Arc;
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::ReceiverStream;
+
+use super::PlatformServiceImpl;
+
+const PLATFORM_EVENTS_STREAM_BUFFER: usize = 512;
+
+/// Tracks an active platform events session until all clones drop.
+struct ActiveSessionGuard;
+
+impl ActiveSessionGuard {
+    fn new() -> Arc<Self> {
+        metrics::platform_events_active_sessions_inc();
+        Arc::new(Self)
+    }
+}
+
+impl Drop for ActiveSessionGuard {
+    fn drop(&mut self) {
+        metrics::platform_events_active_sessions_dec();
+    }
+}
+
+impl PlatformServiceImpl {
+    /// Proxy implementation of Platform::subscribePlatformEvents.
+    ///
+    /// Forwards a subscription request upstream to Drive and streams responses back to the caller.
+    pub async fn subscribe_platform_events_impl(
+        &self,
+        request: Request<PlatformSubscriptionRequest>,
+    ) -> Result<Response<ReceiverStream<Result<PlatformSubscriptionResponse, Status>>>, Status>
+    {
+        let active_session = ActiveSessionGuard::new();
+
+        let mut client = self.drive_client.get_client();
+        let uplink_resp = client.subscribe_platform_events(request).await?;
+        metrics::platform_events_upstream_stream_started();
+        let mut uplink_resp_rx = uplink_resp.into_inner();
+
+        // Channel to forward responses back to caller (downlink)
+        let (downlink_resp_tx, downlink_resp_rx) = mpsc::channel::<
+            Result<PlatformSubscriptionResponse, Status>,
+        >(PLATFORM_EVENTS_STREAM_BUFFER);
+
+        // Spawn a task to forward uplink responses -> downlink
+        {
+            let session_handle = active_session;
+            self.workers.spawn(async move {
+                let _session_guard = session_handle;
+                while let Some(msg) = uplink_resp_rx.next().await {
+                    match msg {
+                        Ok(response) => {
+                            metrics::platform_events_forwarded_event();
+                            if downlink_resp_tx.send(Ok(response)).await.is_err() {
+                                tracing::debug!(
+                                    "Platform events downlink response channel closed; stopping forward"
+                                );
+                                break;
+                            }
+                        }
+                        Err(status) => {
+                            metrics::platform_events_forwarded_error();
+                            if downlink_resp_tx.send(Err(status)).await.is_err() {
+                                tracing::debug!(
+                                    "Platform events downlink response channel closed while forwarding error"
+                                );
+                                break;
+                            }
+                        }
+                    }
+                }
+                tracing::debug!("Platform events uplink response stream closed");
+                Err::<(), DapiError>(DapiError::ConnectionClosed)
+            });
+        }
+
+        Ok(Response::new(ReceiverStream::new(downlink_resp_rx)))
+    }
+}