feat(platform)!: subscribe to committed state transitions matching document and address filters

.github/grpc-queries-cache.json

@@ -191,6 +191,9 @@
     },
     "getShieldedNotesCount": {
       "status": "implemented"
+    },
+    "subscribeToStateTransitions": {
+      "status": "implemented"
     }
   }
 }

packages/dapi-grpc/Cargo.toml

@@ -53,13 +53,14 @@ server = [
   "tonic/router",
 ]
 
-serde = ["dep:serde", "dep:serde_bytes", "tenderdash-proto/serde"]
+serde = ["dep:serde", "dep:serde_bytes", "tenderdash-proto/serde", "bytes/serde"]
 mocks = ["serde", "dep:serde_json"]
 
 [dependencies]
 tenderdash-proto = { git = "https://github.com/dashpay/rs-tenderdash-abci", tag = "v1.8.0", default-features = false }
 
 prost = { version = "0.14" }
+bytes = "1"
 futures-core = "0.3.30"
 serde = { version = "1.0.219", optional = true, features = ["derive"] }
 serde_bytes = { version = "0.11.12", optional = true }

packages/dapi-grpc/build.rs

@@ -111,7 +111,7 @@ fn configure_platform(mut platform: MappingConfig) -> MappingConfig {
     // Derive features for versioned messages
     //
     // "GetConsensusParamsRequest" is excluded as this message does not support proofs
-    const VERSIONED_REQUESTS: [&str; 69] = [
+    const VERSIONED_REQUESTS: [&str; 70] = [
         "GetContractModerationStatusRequest",
         "GetContractModerationEntriesRequest",
         "GetContractDocumentRemovalsRequest",
@@ -181,6 +181,7 @@ fn configure_platform(mut platform: MappingConfig) -> MappingConfig {
         "GetShieldedPoolStateRequest",
         "GetShieldedNotesCountRequest",
         "GetShieldedNullifiersRequest",
+        "SubscribeToStateTransitionsRequest",
     ];
 
     const PROOF_ONLY_VERSIONED_REQUESTS: [&str; 1] = ["GetAddressesTrunkStateRequest"];
@@ -401,6 +402,9 @@ fn configure_platform(mut platform: MappingConfig) -> MappingConfig {
         .field_attribute("ResponseMetadata.height", SERDE_WITH_STRING)
         .field_attribute("ResponseMetadata.time_ms", SERDE_WITH_STRING)
         .field_attribute("start_at_ms", SERDE_WITH_STRING)
+        .field_attribute("StateTransitionMatch.block_height", SERDE_WITH_STRING)
+        .field_attribute("StateTransitionMatch.block_time_ms", SERDE_WITH_STRING)
+        .field_attribute("Checkpoint.block_height", SERDE_WITH_STRING)
         .field_attribute("public_key_hash", SERDE_WITH_BYTES)
         .field_attribute("public_key_hashes", SERDE_WITH_BASE64)
         .field_attribute("nullifiers", SERDE_WITH_BASE64)
@@ -415,7 +419,9 @@ fn configure_platform(mut platform: MappingConfig) -> MappingConfig {
         .field_attribute("Proof.grovedb_proof", SERDE_WITH_BYTES)
         .field_attribute("Proof.quorum_hash", SERDE_WITH_BYTES)
         .field_attribute("Proof.signature", SERDE_WITH_BYTES)
-        .field_attribute("Proof.block_id_hash", SERDE_WITH_BYTES);
+        .field_attribute("Proof.block_id_hash", SERDE_WITH_BYTES)
+        // A shared buffer, so DAPI sends one block's transaction to every subscriber uncopied.
+        .bytes(".org.dash.platform.dapi.v0.SubscribeToStateTransitionsResponse.SubscribeToStateTransitionsResponseV0.StateTransitionMatch.state_transition");
 
     #[allow(clippy::let_and_return)]
     platform
@@ -579,6 +585,12 @@ impl MappingConfig {
         self
     }
 
+    #[allow(unused)]
+    fn bytes(mut self, path: &str) -> Self {
+        self.builder = self.builder.bytes(path);
+        self
+    }
+
     #[allow(unused)]
     fn message_attribute(mut self, path: &str, attribute: &str) -> Self {
         self.builder = self.builder.message_attribute(path, attribute);

packages/dapi-grpc/clients/platform/v0/java/org/dash/platform/dapi/v0/PlatformGrpc.java

@@ -2278,6 +2278,37 @@ org.dash.platform.dapi.v0.PlatformOuterClass.GetShieldedNullifiersResponse> getG
     return getGetShieldedNullifiersMethod;
   }
 
+  private static volatile io.grpc.MethodDescriptor<org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsRequest,
+      org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsResponse> getSubscribeToStateTransitionsMethod;
+
+  @io.grpc.stub.annotations.RpcMethod(
+      fullMethodName = SERVICE_NAME + '/' + "subscribeToStateTransitions",
+      requestType = org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsRequest.class,
+      responseType = org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsResponse.class,
+      methodType = io.grpc.MethodDescriptor.MethodType.SERVER_STREAMING)
+  public static io.grpc.MethodDescriptor<org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsRequest,
+      org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsResponse> getSubscribeToStateTransitionsMethod() {
+    io.grpc.MethodDescriptor<org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsRequest, org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsResponse> getSubscribeToStateTransitionsMethod;
+    if ((getSubscribeToStateTransitionsMethod = PlatformGrpc.getSubscribeToStateTransitionsMethod) == null) {
+      synchronized (PlatformGrpc.class) {
+        if ((getSubscribeToStateTransitionsMethod = PlatformGrpc.getSubscribeToStateTransitionsMethod) == null) {
+          PlatformGrpc.getSubscribeToStateTransitionsMethod = getSubscribeToStateTransitionsMethod =
+              io.grpc.MethodDescriptor.<org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsRequest, org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsResponse>newBuilder()
+              .setType(io.grpc.MethodDescriptor.MethodType.SERVER_STREAMING)
+              .setFullMethodName(generateFullMethodName(SERVICE_NAME, "subscribeToStateTransitions"))
+              .setSampledToLocalTracing(true)
+              .setRequestMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
+                  org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsRequest.getDefaultInstance()))
+              .setResponseMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
+                  org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsResponse.getDefaultInstance()))
+              .setSchemaDescriptor(new PlatformMethodDescriptorSupplier("subscribeToStateTransitions"))
+              .build();
+        }
+      }
+    }
+    return getSubscribeToStateTransitionsMethod;
+  }
+
   /**
    * Creates a new async stub that supports all call types for the service
    */
@@ -2858,6 +2889,19 @@ public void getShieldedNullifiers(org.dash.platform.dapi.v0.PlatformOuterClass.G
       io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall(getGetShieldedNullifiersMethod(), responseObserver);
     }
 
+    /**
+     * <pre>
+     * Streams every committed, successfully executed state transition that matches
+     * one of the request's filters, block by block, from a chosen height onward.
+     * Served by DAPI from the Tenderdash block store; see
+     * `SubscribeToStateTransitionsResponse` for the stream contract.
+     * </pre>
+     */
+    public void subscribeToStateTransitions(org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsRequest request,
+        io.grpc.stub.StreamObserver<org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsResponse> responseObserver) {
+      io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall(getSubscribeToStateTransitionsMethod(), responseObserver);
+    }
+
     @java.lang.Override public final io.grpc.ServerServiceDefinition bindService() {
       return io.grpc.ServerServiceDefinition.builder(getServiceDescriptor())
           .addMethod(
@@ -3371,6 +3415,13 @@ public void getShieldedNullifiers(org.dash.platform.dapi.v0.PlatformOuterClass.G
                 org.dash.platform.dapi.v0.PlatformOuterClass.GetShieldedNullifiersRequest,
                 org.dash.platform.dapi.v0.PlatformOuterClass.GetShieldedNullifiersResponse>(
                   this, METHODID_GET_SHIELDED_NULLIFIERS)))
+          .addMethod(
+            getSubscribeToStateTransitionsMethod(),
+            io.grpc.stub.ServerCalls.asyncServerStreamingCall(
+              new MethodHandlers<
+                org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsRequest,
+                org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsResponse>(
+                  this, METHODID_SUBSCRIBE_TO_STATE_TRANSITIONS)))
           .build();
     }
   }
@@ -3993,6 +4044,20 @@ public void getShieldedNullifiers(org.dash.platform.dapi.v0.PlatformOuterClass.G
       io.grpc.stub.ClientCalls.asyncUnaryCall(
           getChannel().newCall(getGetShieldedNullifiersMethod(), getCallOptions()), request, responseObserver);
     }
+
+    /**
+     * <pre>
+     * Streams every committed, successfully executed state transition that matches
+     * one of the request's filters, block by block, from a chosen height onward.
+     * Served by DAPI from the Tenderdash block store; see
+     * `SubscribeToStateTransitionsResponse` for the stream contract.
+     * </pre>
+     */
+    public void subscribeToStateTransitions(org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsRequest request,
+        io.grpc.stub.StreamObserver<org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsResponse> responseObserver) {
+      io.grpc.stub.ClientCalls.asyncServerStreamingCall(
+          getChannel().newCall(getSubscribeToStateTransitionsMethod(), getCallOptions()), request, responseObserver);
+    }
   }
 
   /**
@@ -4540,6 +4605,20 @@ public org.dash.platform.dapi.v0.PlatformOuterClass.GetShieldedNullifiersRespons
       return io.grpc.stub.ClientCalls.blockingUnaryCall(
           getChannel(), getGetShieldedNullifiersMethod(), getCallOptions(), request);
     }
+
+    /**
+     * <pre>
+     * Streams every committed, successfully executed state transition that matches
+     * one of the request's filters, block by block, from a chosen height onward.
+     * Served by DAPI from the Tenderdash block store; see
+     * `SubscribeToStateTransitionsResponse` for the stream contract.
+     * </pre>
+     */
+    public java.util.Iterator<org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsResponse> subscribeToStateTransitions(
+        org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsRequest request) {
+      return io.grpc.stub.ClientCalls.blockingServerStreamingCall(
+          getChannel(), getSubscribeToStateTransitionsMethod(), getCallOptions(), request);
+    }
   }
 
   /**
@@ -5235,6 +5314,7 @@ public com.google.common.util.concurrent.ListenableFuture<org.dash.platform.dapi
   private static final int METHODID_GET_SHIELDED_POOL_STATE = 70;
   private static final int METHODID_GET_SHIELDED_NOTES_COUNT = 71;
   private static final int METHODID_GET_SHIELDED_NULLIFIERS = 72;
+  private static final int METHODID_SUBSCRIBE_TO_STATE_TRANSITIONS = 73;
 
   private static final class MethodHandlers<Req, Resp> implements
       io.grpc.stub.ServerCalls.UnaryMethod<Req, Resp>,
@@ -5545,6 +5625,10 @@ public void invoke(Req request, io.grpc.stub.StreamObserver<Resp> responseObserv
           serviceImpl.getShieldedNullifiers((org.dash.platform.dapi.v0.PlatformOuterClass.GetShieldedNullifiersRequest) request,
               (io.grpc.stub.StreamObserver<org.dash.platform.dapi.v0.PlatformOuterClass.GetShieldedNullifiersResponse>) responseObserver);
           break;
+        case METHODID_SUBSCRIBE_TO_STATE_TRANSITIONS:
+          serviceImpl.subscribeToStateTransitions((org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsRequest) request,
+              (io.grpc.stub.StreamObserver<org.dash.platform.dapi.v0.PlatformOuterClass.SubscribeToStateTransitionsResponse>) responseObserver);
+          break;
         default:
           throw new AssertionError();
       }
@@ -5679,6 +5763,7 @@ public static io.grpc.ServiceDescriptor getServiceDescriptor() {
               .addMethod(getGetShieldedPoolStateMethod())
               .addMethod(getGetShieldedNotesCountMethod())
               .addMethod(getGetShieldedNullifiersMethod())
+              .addMethod(getSubscribeToStateTransitionsMethod())
               .build();
         }
       }