Add last_valid_hash to reorg messages

- Pass through block hash from server in reorg detection
- Include last_valid_hash in Kafka reorg messages
- Update integration tests to verify new field

Author: incrypto32

src/amp/loaders/implementations/kafka_loader.py

@@ -151,6 +151,7 @@ def _handle_reorg(self, invalidation_ranges: List[BlockRange], table_name: str,
                     'network': invalidation_range.network,
                     'start_block': invalidation_range.start,
                     'end_block': invalidation_range.end,
+                    'last_valid_hash': invalidation_range.hash,
                 }
 
                 self._producer.send(

src/amp/streaming/reorg.py

@@ -107,6 +107,7 @@ def _detect_reorg(self, current_ranges: List[BlockRange]) -> List[BlockRange]:
                         network=current_range.network,
                         start=current_range.start,
                         end=max(current_range.end, prev_range.end),
+                        hash=prev_range.hash,
                     )
                     invalidation_ranges.append(invalidation)
 

tests/integration/test_kafka_loader.py

@@ -112,8 +112,8 @@ def test_handle_reorg(self, kafka_test_config):
         topic_name = 'test_reorg_topic'
 
         invalidation_ranges = [
-            BlockRange(network='ethereum', start=100, end=200),
-            BlockRange(network='polygon', start=500, end=600),
+            BlockRange(network='ethereum', start=100, end=200, hash='0xabc123'),
+            BlockRange(network='polygon', start=500, end=600, hash='0xdef456'),
         ]
 
         with loader:
@@ -138,13 +138,15 @@ def test_handle_reorg(self, kafka_test_config):
         assert msg1.value['network'] == 'ethereum'
         assert msg1.value['start_block'] == 100
         assert msg1.value['end_block'] == 200
+        assert msg1.value['last_valid_hash'] == '0xabc123'
 
         msg2 = messages[1]
         assert msg2.key == b'reorg:polygon'
         assert msg2.value['_type'] == 'reorg'
         assert msg2.value['network'] == 'polygon'
         assert msg2.value['start_block'] == 500
         assert msg2.value['end_block'] == 600
+        assert msg2.value['last_valid_hash'] == '0xdef456'
 
     def test_streaming_with_reorg(self, kafka_test_config):
         loader = KafkaLoader(kafka_test_config)
@@ -165,7 +167,7 @@ def test_streaming_with_reorg(self, kafka_test_config):
         )
 
         reorg_response = ResponseBatch.reorg_batch(
-            invalidation_ranges=[BlockRange(network='ethereum', start=110, end=200)]
+            invalidation_ranges=[BlockRange(network='ethereum', start=110, end=200, hash='0xdef456')]
         )
 
         response3 = ResponseBatch.data_batch(
@@ -211,6 +213,7 @@ def test_streaming_with_reorg(self, kafka_test_config):
         assert reorg_messages[0].value['network'] == 'ethereum'
         assert reorg_messages[0].value['start_block'] == 110
         assert reorg_messages[0].value['end_block'] == 200
+        assert reorg_messages[0].value['last_valid_hash'] == '0xdef456'
 
         data_ids = [msg.value['id'] for msg in data_messages]
         assert data_ids == [1, 2, 3, 4, 5, 6]