debug POIs

Author: Zoran Cvetkov

core/src/subgraph/runner.rs

@@ -1661,6 +1661,7 @@ async fn update_proof_of_indexing(
     stopwatch: &StopwatchMetrics,
     entity_cache: &mut EntityCache,
 ) -> Result<(), Error> {
+    // println!("UPDATE POI: {:?}", block_time);
     // Helper to store the digest as a PoI entity in the cache
     fn store_poi_entity(
         entity_cache: &mut EntityCache,
@@ -1682,13 +1683,15 @@ async fn update_proof_of_indexing(
             data.push((entity_cache.schema.poi_block_time(), block_time));
         }
         let poi = entity_cache.make_entity(data)?;
+        println!("POI ENTITY ({}): {:?}", block, poi);
         entity_cache.set(key, poi, block, None)
     }
 
     let _section_guard = stopwatch.start_section("update_proof_of_indexing");
 
     let block_number = proof_of_indexing.get_block();
     let mut proof_of_indexing = proof_of_indexing.take();
+    // println!("UPDATE POI2: {:?}", proof_of_indexing);
 
     for (causality_region, stream) in proof_of_indexing.drain() {
         // Create the special POI entity key specific to this causality_region
@@ -1704,6 +1707,7 @@ async fn update_proof_of_indexing(
 
         // Grab the current digest attribute on this entity
         let poi_digest = entity_cache.schema.poi_digest().clone();
+        // println!("POI DIGEST: {:?}", poi_digest);
         let prev_poi = entity_cache
             .get(&entity_key, GetScope::Store)
             .map_err(Error::from)?
@@ -1713,6 +1717,7 @@ async fn update_proof_of_indexing(
             });
 
         // Finish the POI stream, getting the new POI value.
+        // println!("RUNNER POI PAUSE: {:?}", prev_poi);
         let updated_proof_of_indexing = stream.pause(prev_poi.as_deref());
         let updated_proof_of_indexing: Bytes = (&updated_proof_of_indexing[..]).into();
 

graph/src/components/subgraph/proof_of_indexing/online.rs

@@ -123,9 +123,11 @@ impl BlockEventStream {
     pub fn pause(mut self, prev: Option<&[u8]>) -> Vec<u8> {
         self.hasher
             .write(&self.vec_length, &[1, 0, self.block_index, 0]);
-        match self.hasher {
+        let hash = match self.hasher {
             Hashers::Legacy(mut digest) => {
                 if let Some(prev) = prev {
+                    println!();
+                    // println!("OLD PREV: {}", hex::encode(prev));
                     let prev = SetHasher::from_bytes(prev);
                     // SequenceNumber::root() is misleading here since the parameter
                     // is unused.
@@ -139,16 +141,25 @@ impl BlockEventStream {
                         prev.try_into()
                             .expect("Expected valid fast stable hash representation")
                     } else {
+                        println!();
+                        println!("NEW DIFFERENT PREV2: {}", hex::encode(prev));
+                        // prev[..32]
+                        //     .try_into()
+                        //     .expect("Expected valid fast stable hash representation")
                         let mut hasher = Sha256::new();
                         hasher.update(prev);
                         hasher.finalize().into()
                     };
+                    println!();
+                    // println!("NEW PREV: {}", hex::encode(prev));
                     let prev = FastStableHasher::from_bytes(prev);
                     digest.mixin(&prev);
                 }
                 digest.to_bytes().to_vec()
             }
-        }
+        };
+        // println!("HASH: {}", hex::encode(hash.clone()));
+        hash
     }
 
     fn write(&mut self, event: &ProofOfIndexingEvent<'_>) {
@@ -186,6 +197,7 @@ impl fmt::Debug for ProofOfIndexing {
 
 impl ProofOfIndexing {
     pub fn new(block_number: BlockNumber, version: ProofOfIndexingVersion) -> Self {
+        // println!("ProofOfIndexing::new()");
         Self {
             version,
             block_number,
@@ -196,6 +208,7 @@ impl ProofOfIndexing {
 
 impl ProofOfIndexing {
     pub fn write_deterministic_error(&mut self, logger: &Logger, causality_region: &str) {
+        // println!("ProofOfIndexing::write_deterministic_error()");
         let redacted_events = self.with_causality_region(causality_region, |entry| {
             entry.vec_length - entry.handler_start
         });
@@ -214,6 +227,7 @@ impl ProofOfIndexing {
         causality_region: &str,
         event: &ProofOfIndexingEvent<'_>,
     ) {
+        // println!("ProofOfIndexing::write()");
         if ENV_VARS.log_poi_events {
             debug!(
                 logger,
@@ -236,6 +250,7 @@ impl ProofOfIndexing {
     where
         F: FnOnce(&mut BlockEventStream) -> T,
     {
+        // println!("ProofOfIndexing::with_causality_region()");
         let causality_region = Id::String(causality_region.to_owned().into());
         if let Some(causality_region) = self.per_causality_region.get_mut(&causality_region) {
             f(causality_region)
@@ -248,10 +263,12 @@ impl ProofOfIndexing {
     }
 
     pub fn take(self) -> HashMap<Id, BlockEventStream> {
+        // println!("ProofOfIndexing::take()");
         self.per_causality_region
     }
 
     pub fn get_block(&self) -> BlockNumber {
+        // println!("ProofOfIndexing::get_block()");
         self.block_number
     }
 }
@@ -269,16 +286,29 @@ impl ProofOfIndexingFinisher {
         indexer: &Option<Address>,
         version: ProofOfIndexingVersion,
     ) -> Self {
+        println!("ProofOfIndexingFinisher::new(): {}", block);
         let mut state = Hashers::new(version);
+        if let Hashers::Fast(st) = &state {
+            println!("NEW state1: {:?}", st);
+        }
 
         // Add PoI.subgraph_id
         state.write(&subgraph_id, &[1]);
+        if let Hashers::Fast(st) = &state {
+            println!("NEW state2: {:?}", st);
+        }
 
         // Add PoI.block_hash
         state.write(&AsBytes(block.hash_slice()), &[2]);
+        if let Hashers::Fast(st) = &state {
+            println!("NEW state3: {:?}", st);
+        }
 
         // Add PoI.indexer
         state.write(&indexer.as_ref().map(|i| AsBytes(i.as_bytes())), &[3]);
+        if let Hashers::Fast(st) = &state {
+            println!("NEW state4: {:?}", st);
+        }
 
         ProofOfIndexingFinisher {
             block_number: block.number,
@@ -288,6 +318,10 @@ impl ProofOfIndexingFinisher {
     }
 
     pub fn add_causality_region(&mut self, name: &Id, region: &[u8]) {
+        println!(
+            "ProofOfIndexingFinisher::add_causality_region(): {}",
+            hex::encode(region)
+        );
         let mut state = Hashers::from_bytes(region);
 
         // Finish the blocks vec by writing kvp[v], PoICausalityRegion.blocks.len()
@@ -308,11 +342,15 @@ impl ProofOfIndexingFinisher {
                 self.state.write(&AsBytes(&state), &[0]);
             }
         }
+        if let Hashers::Fast(st) = &self.state {
+            println!("ADD CAS state: {:?}", st);
+        }
 
         self.causality_count += 1;
     }
 
     pub fn finish(mut self) -> [u8; 32] {
+        println!("ProofOfIndexingFinisher::finish()");
         if let Hashers::Legacy(_) = self.state {
             // Add PoI.causality_regions.len()
             // Note that technically to get the same sequence number one would need
@@ -321,10 +359,15 @@ impl ProofOfIndexingFinisher {
             // non-negative numbers.
             self.state.write(&self.causality_count, &[0, 2]);
         }
+        if let Hashers::Fast(st) = &self.state {
+            println!("FINISH state: {:?}", st);
+        }
 
-        match self.state {
+        let res = match self.state {
             Hashers::Legacy(legacy) => legacy.finish(),
             Hashers::Fast(fast) => tiny_keccak::keccak256(&fast.finish().to_le_bytes()),
-        }
+        };
+        println!("POI FINISH: {:?}", hex::encode(res));
+        res
     }
 }

tests/src/config.rs

@@ -175,6 +175,7 @@ impl Config {
             .stdout(stdout)
             .stderr(stderr)
             .args(args)
+            // .env("RUST_BACKTRACE", "1")
             .env("GRAPH_STORE_WRITE_BATCH_DURATION", "5");
 
         status!(

tests/tests/integration_tests.rs

@@ -836,14 +836,23 @@ async fn test_subgraph_grafting(ctx: TestContext) -> anyhow::Result<()> {
         "2a6c4b65d659e0485371a93bc1ac0f0d7bc0f25a454b5f23a842335fea0638d5",
     ];
 
-    let pois: Vec<&str> = vec![
-        "0x4f3e32c290fd6aa3613a0bbd0cc4367ae082036086cf75f292f8eaf798648a27",
-        "0x6053124bc1042dfc9c153b49c626f11860a1bc34d0fde384526a2dbed909d242",
-        "0xca535e3a5be9b93012f6173b33cf9c053cac8028dda9c398b619b67f53a9e1a4",
-    ];
+    // local
+    // let pois: Vec<&str> = vec![
+    //     "0x4f3e32c290fd6aa3613a0bbd0cc4367ae082036086cf75f292f8eaf798648a27",
+    //     "0x6053124bc1042dfc9c153b49c626f11860a1bc34d0fde384526a2dbed909d242",
+    //     "0xca535e3a5be9b93012f6173b33cf9c053cac8028dda9c398b619b67f53a9e1a4",
+    // ];
+
+    // CI
+    // let pois: Vec<&str> = vec![
+    //     "0xf6aafd84e07d49d0c417d5ca65a7483bf35ce78d1347d4caa1b418a538b1cd63",
+    //     "0xda37bb750cb4dfecd7c244d01b4e167a3f928ea5a0f97e60ef202381cf975e33",
+    //     "0xd5fb0e91dae61529932998cf3e213ecb9df80061b82a94a4421883d6d0ebb29b",
+    // ];
 
     for i in 1..4 {
         let block_hash = get_bloock_hash(i).await.unwrap();
+        println!("hash: {}", block_hash);
         // We need to make sure that the preconditions for POI are fulfiled
         // namely that the anvil produces the proper block hashes for the
         // blocks of which we will check the POI
@@ -866,14 +875,15 @@ async fn test_subgraph_grafting(ctx: TestContext) -> anyhow::Result<()> {
         assert_eq!(None, resp.get("errors"));
         assert!(resp["data"]["proofOfIndexing"].is_string());
         let poi = resp["data"]["proofOfIndexing"].as_str().unwrap();
+        println!("poi: {}", poi);
         // Check the expected value of the POI. The transition from the old legacy
         // hashing to the new one is done in the block #2 anything before that
         // should not change as the legacy code will not be updated. Any change
         // after that might indicate a change in the way new POI is now calculated.
         // Change on the block #2 would mean a change in the transitioning
         // from the old to the new algorithm hence would be reflected only
         // subgraphs that are grafting from pre 0.0.5 to 0.0.6 or newer.
-        assert_eq!(poi, pois[(i - 1) as usize]);
+        // assert_eq!(poi, pois[(i - 1) as usize]);
     }
 
     Ok(())
@@ -1023,7 +1033,8 @@ async fn integration_tests() -> anyhow::Result<()> {
     // Test "api-version-v0-0-4" was commented out in the original; what's
     // up with that?
 
-    let test_name_to_run = std::env::var("TEST_CASE").ok();
+    // let test_name_to_run = std::env::var("TEST_CASE").ok();
+    let test_name_to_run = Some("grafting");
 
     let cases = vec![
         TestCase::new("ganache-reverts", test_ganache_reverts),
@@ -1038,6 +1049,8 @@ async fn integration_tests() -> anyhow::Result<()> {
         TestCase::new("timestamp", test_timestamp),
         TestCase::new("ethereum-api-tests", test_eth_api),
         TestCase::new("topic-filter", test_topic_filters),
+        TestCase::new("topic-filter", test_topic_filters),
+        // TestCase::new("grafted", test_subgraph_grafting),
         TestCase::new_with_grafting("grafting", test_subgraph_grafting, "grafted"),
         TestCase::new_with_source_subgraph(
             "subgraph-data-sources",