fix(sdk): send schema-typed operands and move off a failing subscription node

Addresses the non-blocking findings of the review of #5285 at a181760:
- The SDK sends clause operands in the form the schema stores them
  (`canonical_operands`), since the wire carries a few primitive types: a
  `u128` for a float field or `u8`s for a byte array no longer arrive as
  text and `u64`s the node refuses.
- When a stream fails because of its node (missing history, an unreadable
  or undecodable block, Tenderdash unreachable), the SDK bans that node
  before reconnecting, so a sticky active set does not reconnect to it until
  the failure budget runs out; ended lifetimes and slow-client errors do not
  ban.
- A document purchase makes the batch owner a recipient, as a token direct
  purchase already did.
- Price IN candidates are bounded and distinct, like any IN clause's.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: PastaPastaPasta

packages/dash-platform-queries/src/subscriptions/mod.rs

@@ -27,7 +27,7 @@ use dpp::prelude::Identifier;
 use drive::query::{ValueClause, WhereClause};
 
 pub use proto::subscribe_request;
-pub use resolved::{FilterMatch, ResolvedFilters};
+pub use resolved::{canonical_operands, FilterMatch, ResolvedFilters};
 
 /// Most filters one subscription may carry.
 pub const MAX_FILTERS: usize = 16;

packages/dash-platform-queries/src/subscriptions/participants.rs

@@ -201,11 +201,15 @@ fn state_transition_owner(state_transition: &StateTransition) -> Identifier {
     state_transition.owner_id().unwrap_or_default()
 }
 
-/// The identity a batch's inner document transition names as recipient: a transfer's new
-/// owner.
-pub(super) fn document_recipient(transition: &DocumentTransition) -> Option<Identifier> {
+/// The identity a batch's inner document transition makes the document's new owner: a
+/// transfer's recipient, or for a purchase the batch owner, who buys it.
+pub(super) fn document_recipient(
+    transition: &DocumentTransition,
+    batch_owner_id: Identifier,
+) -> Option<Identifier> {
     match transition {
         DocumentTransition::Transfer(transfer) => Some(transfer.recipient_owner_id()),
+        DocumentTransition::Purchase(_) => Some(batch_owner_id),
         _ => None,
     }
 }

packages/dash-platform-queries/src/subscriptions/resolved.rs

@@ -278,6 +278,56 @@ impl ResolvedFilters {
     }
 }
 
+/// `filter` with every clause operand in the form the document type stores its field in, as
+/// it should be sent: the wire carries a few primitive types, so an operand the matcher accepts
+/// in another representation (a `u128` for a float field, `u8`s for a byte array) would arrive
+/// as something else. Filters other than typed document filters are returned unchanged.
+pub fn canonical_operands(
+    filter: &StateTransitionFilter,
+    data_contract: impl Fn(&Identifier) -> Option<Arc<DataContract>>,
+    platform_version: &PlatformVersion,
+) -> Result<StateTransitionFilter, String> {
+    let StateTransitionFilter::Documents(document_filter) = filter else {
+        return Ok(filter.clone());
+    };
+    let Some(document_type_name) = &document_filter.document_type_name else {
+        return Ok(filter.clone());
+    };
+    let contract = data_contract(&document_filter.data_contract_id).ok_or_else(|| {
+        format!(
+            "data contract {} not found",
+            document_filter.data_contract_id
+        )
+    })?;
+    let document_type = contract
+        .document_type_for_name(document_type_name)
+        .map_err(|e| e.to_string())?;
+    let mut canonical = document_filter.clone();
+    for action in &mut canonical.actions {
+        for clause in action
+            .new_document_where
+            .iter_mut()
+            .chain(action.original_document_where.iter_mut())
+        {
+            canonicalize_where_clause(document_type, clause, platform_version)
+                .map_err(|e| e.to_string())?;
+        }
+        if let Some(price) = &mut action.price {
+            price.value = match &price.value {
+                Value::Array(values) => Value::Array(
+                    values
+                        .iter()
+                        .map(|value| value.to_integer::<u64>().map(Value::U64))
+                        .collect::<Result<_, _>>()
+                        .map_err(|e| e.to_string())?,
+                ),
+                value => Value::U64(value.to_integer::<u64>().map_err(|e| e.to_string())?),
+            };
+        }
+    }
+    Ok(StateTransitionFilter::Documents(canonical))
+}
+
 impl ResolvedFilter {
     /// Whether one inner transition of a batch owned by `batch_owner_id` matches.
     fn matches_batched(
@@ -295,7 +345,8 @@ impl ResolvedFilter {
                 BatchedTransitionRef::Document(transition),
             ) => {
                 role.admits(Role::Recipient)
-                    && document_recipient(transition).is_some_and(|id| identity_ids.contains(&id))
+                    && document_recipient(transition, batch_owner_id)
+                        .is_some_and(|id| identity_ids.contains(&id))
             }
             (
                 ResolvedFilter::Identities { identity_ids, role },
@@ -565,6 +616,21 @@ fn action_clauses(
     if kind != DocumentAction::UpdatePrice && action.price.is_some() {
         return Err("a price clause applies only to update price".to_string());
     }
+    // Price IN candidates are bounded and distinct, like any IN clause's.
+    if let Some(price) = action
+        .price
+        .as_ref()
+        .filter(|price| price.operator == WhereOperator::In)
+    {
+        let candidates = WhereClause {
+            field: "$price".to_string(),
+            operator: WhereOperator::In,
+            value: price.value.clone(),
+        };
+        if let Some(error) = candidates.in_values().errors.first() {
+            return Err(error.to_string());
+        }
+    }
 
     let document_type = contract
         .document_type_for_name(document_type_name)

packages/dash-platform-queries/src/subscriptions/tests.rs

@@ -734,3 +734,118 @@ fn should_match_a_shielded_token_unshield_by_token_and_recipient() {
     .unwrap();
     assert!(matches(&recipient, &unshield).is_some());
 }
+
+#[test]
+fn should_match_a_document_buyer_as_recipient() {
+    use dpp::state_transition::batch_transition::batched_transition::document_purchase_transition::v0::DocumentPurchaseTransitionV0;
+    use dpp::state_transition::batch_transition::batched_transition::document_purchase_transition::DocumentPurchaseTransition;
+    let contract = contract();
+    let buyer = id(6);
+    let purchase = BatchedTransition::Document(DocumentTransition::Purchase(
+        DocumentPurchaseTransition::V0(DocumentPurchaseTransitionV0 {
+            base: document_base(&contract, "niceDocument", 10),
+            revision: 2,
+            price: 100,
+        }),
+    ));
+    let filters = resolve(
+        vec![StateTransitionFilter::Identities {
+            identity_ids: vec![buyer],
+            role: Role::Recipient,
+        }],
+        &contract,
+    )
+    .unwrap();
+    assert_eq!(
+        matches(
+            &filters,
+            &batch(buyer, vec![create_document(&contract, 11, "x"), purchase])
+        ),
+        Some(FilterMatch {
+            matched_filters: vec![0],
+            matched_batch_positions: vec![1],
+        })
+    );
+}
+
+#[test]
+fn should_bound_price_in_candidates() {
+    use drive::query::ValueClause;
+    let contract = contract();
+    let price_in = |values: Vec<Value>| {
+        StateTransitionFilter::Documents(
+            DocumentFilter::new(contract.id())
+                .with_document_type("niceDocument")
+                .with_action(
+                    DocumentActionMatch::new(DocumentAction::UpdatePrice).with_price(ValueClause {
+                        operator: WhereOperator::In,
+                        value: Value::Array(values),
+                    }),
+                ),
+        )
+    };
+    let too_many = price_in((0..20_000u64).map(|n| Value::U64(n % 3)).collect());
+    let duplicates = price_in(vec![Value::U64(1), Value::U64(1)]);
+    for filter in [too_many, duplicates] {
+        assert!(resolve(vec![filter.clone()], &contract).is_err());
+        let wire = StateTransitionFilter::from_proto(0, filter.to_proto().unwrap()).unwrap();
+        assert!(resolve(vec![wire], &contract).is_err());
+    }
+    assert!(resolve(
+        vec![price_in(vec![Value::U64(1), Value::U64(2)])],
+        &contract
+    )
+    .is_ok());
+}
+
+#[test]
+fn should_send_operands_as_the_schema_stores_them() {
+    use dpp::data_contract::DataContractFactory;
+    use dpp::platform_value::platform_value;
+    let documents = platform_value!({
+        "reading": {
+            "type": "object",
+            "properties": {
+                "value": { "type": "number", "position": 0 },
+                "blob": { "type": "array", "byteArray": true, "maxItems": 8, "position": 1 }
+            },
+            "additionalProperties": false
+        }
+    });
+    let contract = Arc::new(
+        DataContractFactory::new(PlatformVersion::latest().protocol_version)
+            .expect("factory")
+            .create_with_value_config(id(1), 1, documents, None, None)
+            .expect("the contract parses")
+            .data_contract_owned(),
+    );
+    let clause = |field: &str, value| drive::query::WhereClause {
+        field: field.to_string(),
+        operator: WhereOperator::Equal,
+        value,
+    };
+    let filter = StateTransitionFilter::Documents(
+        DocumentFilter::new(contract.id())
+            .with_document_type("reading")
+            .with_action(
+                DocumentActionMatch::new(DocumentAction::Create)
+                    .with_new_document_where(clause("value", Value::U128(1 << 64)))
+                    .with_new_document_where(clause(
+                        "blob",
+                        Value::Array(vec![Value::U8(1), Value::U8(2)]),
+                    )),
+            ),
+    );
+    resolve(vec![filter.clone()], &contract).expect("resolves natively");
+    // Sent as given, the operands arrive as text and u64s, which the node refuses.
+    let raw = StateTransitionFilter::from_proto(0, filter.to_proto().unwrap()).unwrap();
+    assert!(resolve(vec![raw], &contract).is_err());
+    let canonical = canonical_operands(
+        &filter,
+        |_| Some(contract.clone()),
+        PlatformVersion::latest(),
+    )
+    .expect("canonical operands");
+    let wire = StateTransitionFilter::from_proto(0, canonical.to_proto().unwrap()).unwrap();
+    resolve(vec![wire], &contract).expect("resolves after the wire round trip");
+}

packages/rs-sdk/src/platform/subscriptions.rs

@@ -28,7 +28,7 @@ use dapi_grpc::platform::v0::subscribe_to_state_transitions_response::Version;
 use dapi_grpc::platform::v0::{SubscribeToStateTransitionsRequest, SubscribeToStateTransitionsResponse};
 use dapi_grpc::tonic::{Code, Status, Streaming};
 use dash_platform_queries::subscriptions::{
-    subscribe_request, ResolvedFilters, StateTransitionFilter, MAX_FILTERS,
+    canonical_operands, subscribe_request, ResolvedFilters, StateTransitionFilter, MAX_FILTERS,
 };
 use dpp::dashcore::hashes::{sha256, Hash};
 use dpp::data_contract::DataContract;
@@ -37,7 +37,7 @@ use dpp::serialization::PlatformDeserializableUntrusted;
 use dpp::state_transition::StateTransition;
 use dpp::version::PlatformVersion;
 use rs_dapi_client::transport::{sleep, TransportError};
-use rs_dapi_client::{DapiClientError, DapiRequestExecutor, IntoInner, RequestSettings};
+use rs_dapi_client::{Address, DapiClientError, DapiRequestExecutor, RequestSettings};
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
 use std::time::Duration;
@@ -93,6 +93,8 @@ pub struct StateTransitionSubscription {
     stream: Option<Streaming<SubscribeToStateTransitionsResponse>>,
     /// Messages received on the current stream.
     stream_messages: u32,
+    /// The node serving the current stream.
+    stream_address: Option<Address>,
     /// Where the next stream starts; `None` until the first checkpoint of a live-only start.
     resume_from: Option<u64>,
     consecutive_failures: u32,
@@ -142,6 +144,14 @@ impl Sdk {
                 data_contract_ids: data_contract_ids.into_iter().collect(),
             });
         }
+        // Send operands in the form the schema stores them, so the node reads what was meant.
+        let filters = filters
+            .iter()
+            .map(|filter| {
+                canonical_operands(filter, |id| contracts.get(id).cloned(), self.version())
+            })
+            .collect::<Result<Vec<_>, _>>()
+            .map_err(Error::Config)?;
         let resolved = ResolvedFilters::resolve(
             filters.clone(),
             |id| contracts.get(id).cloned(),
@@ -156,6 +166,7 @@ impl Sdk {
             resolved,
             stream: None,
             stream_messages: 0,
+            stream_address: None,
             resume_from: from_block_height,
             consecutive_failures: 0,
             contracts_to_refresh: BTreeSet::new(),
@@ -201,6 +212,19 @@ impl StateTransitionSubscription {
             };
             if let Some(status) = &failure {
                 tracing::debug!(code = ?status.code(), message = status.message(), "state transition stream broke; resuming");
+                // Resume elsewhere when the node itself is at fault, rather than reconnecting
+                // to it; a stream ended by its lifetime or the client says nothing of the node.
+                if node_at_fault(status) {
+                    if let Some(address) = &self.stream_address {
+                        self.sdk.address_list().ban_with_reason(
+                            address,
+                            Some(format!(
+                                "state transition stream failed: {}",
+                                status.message()
+                            )),
+                        );
+                    }
+                }
             }
             self.stream = None;
             self.consecutive_failures += 1;
@@ -231,14 +255,16 @@ impl StateTransitionSubscription {
         })
     }
 
-    async fn open(&self) -> Result<Streaming<SubscribeToStateTransitionsResponse>, Error> {
+    async fn open(&mut self) -> Result<Streaming<SubscribeToStateTransitionsResponse>, Error> {
         let request: SubscribeToStateTransitionsRequest =
             subscribe_request(&self.filters, self.resume_from)?;
-        self.sdk
-            .execute(request, RequestSettings::default())
-            .await
-            .into_inner()
-            .map_err(Error::from)
+        match self.sdk.execute(request, RequestSettings::default()).await {
+            Ok(response) => {
+                self.stream_address = Some(response.address);
+                Ok(response.inner)
+            }
+            Err(error) => Err(Error::from(error.inner)),
+        }
     }
 
     /// Open a new stream after a break, backing off after consecutive failures.
@@ -368,6 +394,20 @@ impl StateTransitionSubscription {
     }
 }
 
+/// Whether a stream failure says the serving node cannot serve this subscription (it lacks
+/// history, cannot read or decode a block, or Tenderdash is unreachable), so another node
+/// should be tried; not an ended stream lifetime, a cancellation or a slow client.
+fn node_at_fault(status: &Status) -> bool {
+    matches!(
+        status.code(),
+        Code::OutOfRange
+            | Code::FailedPrecondition
+            | Code::Unavailable
+            | Code::Internal
+            | Code::Unknown
+    )
+}
+
 /// Whether a stream that failed with `status` can continue on a new stream.
 fn resumable(status: &Status) -> bool {
     matches!(
@@ -429,6 +469,7 @@ mod tests {
             resolved,
             stream: None,
             stream_messages: 0,
+            stream_address: None,
             resume_from: None,
             consecutive_failures: 0,
             contracts_to_refresh: BTreeSet::new(),
@@ -751,4 +792,25 @@ mod tests {
             .expect("skipped, not failed")
             .is_none());
     }
+
+    #[test]
+    fn should_move_off_a_node_only_when_it_is_at_fault() {
+        for code in [
+            Code::OutOfRange,
+            Code::FailedPrecondition,
+            Code::Unavailable,
+            Code::Internal,
+        ] {
+            assert!(node_at_fault(&Status::new(code, "")), "{code:?}");
+        }
+        // An ended lifetime, a cancellation or a slow client is not the node's fault.
+        for code in [
+            Code::DeadlineExceeded,
+            Code::Cancelled,
+            Code::ResourceExhausted,
+            Code::Aborted,
+        ] {
+            assert!(!node_at_fault(&Status::new(code, "")), "{code:?}");
+        }
+    }
 }