fix(query): substrate events emit "callAddress" twice

crates/query/src/query/substrate.rs

@@ -135,7 +135,6 @@ item_field_selection! {
         this.call_address,
         this.name,
         this.phase,
-        this.call_address,
         this.topics,
         [this.args]: Json,
     }}

crates/query/tests/fixtures.rs

@@ -1,12 +1,90 @@
 use std::{
+    collections::HashSet,
+    fmt,
     io::ErrorKind,
     path::{Path, PathBuf}
 };
 
+use serde::de::{self, Deserialize, Deserializer, MapAccess, SeqAccess, Visitor};
 use sqd_query::{Chunk, JsonArrayWriter, Query};
 
-fn execute_query(chunk: &dyn Chunk, query_file: impl AsRef<Path>) -> anyhow::Result<Vec<u8>> {
-    let query = Query::from_json_bytes(&std::fs::read(query_file)?)?;
+/// Deserialization target that accepts any JSON document, but fails if some
+/// object in it repeats a property name.
+///
+/// `serde_json::Value` cannot be used for this: its map keeps a single entry
+/// per key, so a repeated property is silently collapsed and the defect never
+/// reaches an assertion. Every fixture comparison in this file parses the
+/// response, so a duplicate property is invisible to all of them.
+struct UniqueKeys;
+
+impl<'de> Deserialize<'de> for UniqueKeys {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        deserializer.deserialize_any(UniqueKeysVisitor)
+    }
+}
+
+struct UniqueKeysVisitor;
+
+impl<'de> Visitor<'de> for UniqueKeysVisitor {
+    type Value = UniqueKeys;
+
+    fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.write_str("any JSON value")
+    }
+
+    fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Self::Value, A::Error> {
+        let mut seen: HashSet<String> = HashSet::new();
+        while let Some(key) = map.next_key::<String>()? {
+            map.next_value::<UniqueKeys>()?;
+            if !seen.insert(key.clone()) {
+                return Err(de::Error::custom(format!("duplicate property {:?}", key)));
+            }
+        }
+        Ok(UniqueKeys)
+    }
+
+    fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Self::Value, A::Error> {
+        while seq.next_element::<UniqueKeys>()?.is_some() {}
+        Ok(UniqueKeys)
+    }
+
+    fn visit_some<D: Deserializer<'de>>(self, d: D) -> Result<Self::Value, D::Error> {
+        Deserialize::deserialize(d)
+    }
+
+    fn visit_bool<E: de::Error>(self, _: bool) -> Result<Self::Value, E> {
+        Ok(UniqueKeys)
+    }
+    fn visit_i64<E: de::Error>(self, _: i64) -> Result<Self::Value, E> {
+        Ok(UniqueKeys)
+    }
+    fn visit_u64<E: de::Error>(self, _: u64) -> Result<Self::Value, E> {
+        Ok(UniqueKeys)
+    }
+    fn visit_f64<E: de::Error>(self, _: f64) -> Result<Self::Value, E> {
+        Ok(UniqueKeys)
+    }
+    fn visit_str<E: de::Error>(self, _: &str) -> Result<Self::Value, E> {
+        Ok(UniqueKeys)
+    }
+    fn visit_unit<E: de::Error>(self) -> Result<Self::Value, E> {
+        Ok(UniqueKeys)
+    }
+    fn visit_none<E: de::Error>(self) -> Result<Self::Value, E> {
+        Ok(UniqueKeys)
+    }
+}
+
+/// A response must be a valid JSON document in which no object repeats a
+/// property name (RFC 8259 s4). Checked on the raw bytes, before any parse.
+fn assert_unique_keys(bytes: &[u8], query_file: &Path) {
+    if let Err(err) = serde_json::from_slice::<UniqueKeys>(bytes) {
+        panic!("{}: {}", query_file.display(), err)
+    }
+}
+
+fn execute_query_bytes(chunk: &dyn Chunk, query_json: &[u8]) -> anyhow::Result<Vec<u8>> {
+    let query = Query::from_json_bytes(query_json)?;
     let data = Vec::with_capacity(4 * 1024 * 1024);
     let mut writer = JsonArrayWriter::new(data);
     if let Some(mut blocks) = query.compile().execute(chunk)? {
@@ -15,12 +93,19 @@ fn execute_query(chunk: &dyn Chunk, query_file: impl AsRef<Path>) -> anyhow::Res
     Ok(writer.finish()?)
 }
 
+fn execute_query(chunk: &dyn Chunk, query_file: impl AsRef<Path>) -> anyhow::Result<Vec<u8>> {
+    execute_query_bytes(chunk, &std::fs::read(query_file)?)
+}
+
 fn test_fixture(chunk: &dyn Chunk, query_file: PathBuf) {
     let case_dir = query_file.parent().unwrap();
     let result_file = case_dir.join("result.json");
 
     let actual: serde_json::Value = match execute_query(chunk, &query_file) {
-        Ok(bytes) => serde_json::from_slice(&bytes).unwrap(),
+        Ok(bytes) => {
+            assert_unique_keys(&bytes, &query_file);
+            serde_json::from_slice(&bytes).unwrap()
+        }
         Err(err) => serde_json::Value::String(err.to_string())
     };
 
@@ -49,12 +134,12 @@ fn test_fixture(chunk: &dyn Chunk, query_file: PathBuf) {
 
 #[cfg(feature = "parquet")]
 mod parquet {
-    use std::path::PathBuf;
+    use std::path::{Path, PathBuf};
 
     use rstest::rstest;
     use sqd_query::ParquetChunk;
 
-    use crate::test_fixture;
+    use crate::{assert_unique_keys, execute_query_bytes, test_fixture};
 
     #[rstest]
     fn query(#[files("fixtures/*/queries/*/query.json")] query_file: PathBuf) {
@@ -63,6 +148,49 @@ mod parquet {
         let chunk = ParquetChunk::new(chunk_dir.to_str().unwrap());
         test_fixture(&chunk, query_file)
     }
+
+    /// A substrate event must carry `callAddress` exactly once.
+    ///
+    /// The projection used to list the column twice, so the property was
+    /// written twice into every event object. No fixture could catch it: the
+    /// comparison parses both sides, and a JSON parser keeps one entry per
+    /// key, collapsing the duplicate before any assertion runs.
+    ///
+    /// This test therefore asserts on the raw response bytes, and does not
+    /// depend on any fixture's `query.json` continuing to select the field.
+    #[test]
+    fn substrate_event_emits_call_address_once() {
+        let chunk = ParquetChunk::new("fixtures/moonbeam/chunk");
+        let query = br#"{
+            "type": "substrate",
+            "fields": {"event": {"index": true, "callAddress": true, "name": true}},
+            "events": [{}]
+        }"#;
+
+        let bytes = execute_query_bytes(&chunk, query).unwrap();
+
+        // The duplicate lives in the bytes, not in the parsed document.
+        assert_unique_keys(&bytes, Path::new("substrate_event_emits_call_address_once"));
+
+        // Guard against a vacuous pass: the query must actually have returned
+        // events, and each must actually carry the property under test.
+        let blocks: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+        let events: Vec<&serde_json::Value> = blocks
+            .as_array()
+            .unwrap()
+            .iter()
+            .filter_map(|block| block.get("events"))
+            .flat_map(|events| events.as_array().unwrap())
+            .collect();
+
+        assert!(!events.is_empty(), "fixture chunk returned no events");
+        for event in &events {
+            assert!(
+                event.get("callAddress").is_some(),
+                "selected property is missing from {event}"
+            );
+        }
+    }
 }
 
 #[cfg(feature = "storage")]