Address review feedback for LMDB state store

- Move class attribute after docstring for proper documentation tooling
- Simplify metadata to only store end_block and end_hash (remove unused start_parent_hash)
- Return False for empty batch_ids in is_processed (align with InMemoryStreamStateStore)
- Remove prev_hash from resume watermark BlockRange (was semantically incorrect)
- Add comment explaining resume watermark start=end=end_block pattern

Author: incrypto32

src/amp/streaming/lmdb_state.py

@@ -18,7 +18,6 @@
 
 
 class LMDBStreamStateStore(StreamStateStore):
-    env: lmdb.Environment
     """
     Generic LMDB-based state store for tracking processed batches.
 
@@ -39,9 +38,11 @@ class LMDBStreamStateStore(StreamStateStore):
 
     Metadata database layout:
     - Key: {connection_name}|{table_name}|{network}
-    - Value: JSON with {end_block, end_hash, start_parent_hash} (max processed block)
+    - Value: JSON with {end_block, end_hash} (max processed block for resume)
     """
 
+    env: lmdb.Environment
+
     def __init__(
         self,
         connection_name: str,
@@ -95,12 +96,11 @@ def _serialize_batch(self, batch: BatchIdentifier) -> bytes:
         }
         return json.dumps(batch_value_dict).encode('utf-8')
 
-    def _serialize_metadata(self, end_block: int, end_hash: str, start_parent_hash: str) -> bytes:
+    def _serialize_metadata(self, end_block: int, end_hash: str) -> bytes:
         """Serialize metadata to JSON bytes."""
         meta_value_dict = {
             'end_block': end_block,
             'end_hash': end_hash,
-            'start_parent_hash': start_parent_hash,
         }
         return json.dumps(meta_value_dict).encode('utf-8')
 
@@ -121,7 +121,7 @@ def is_processed(self, connection_name: str, table_name: str, batch_ids: List[Ba
             True only if ALL batches are already processed
         """
         if not batch_ids:
-            return True
+            return False
 
         with self.env.begin(db=self.batches_db) as txn:
             for batch_id in batch_ids:
@@ -161,7 +161,7 @@ def mark_processed(self, connection_name: str, table_name: str, batch_ids: List[
                         should_update = True
 
                 if should_update:
-                    meta_value = self._serialize_metadata(batch.end_block, batch.end_hash, batch.start_parent_hash)
+                    meta_value = self._serialize_metadata(batch.end_block, batch.end_hash)
                     txn.put(meta_key, meta_value, db=self.metadata_db)
 
         self.logger.debug(f'Marked {len(batch_ids)} batches as processed in {table_name}')
@@ -205,13 +205,15 @@ def get_resume_position(
                     _, _, network = self._parse_key(key)
                     meta_data = self._deserialize_batch(value)
 
+                    # BlockRange here represents a resume watermark (open-ended), not a processed range.
+                    # start=end=end_block means "resume streaming from this block onwards".
+                    # See ReorgAwareStream for how this watermark is used for crash recovery.
                     ranges.append(
                         BlockRange(
                             network=network,
                             start=meta_data['end_block'],
                             end=meta_data['end_block'],
                             hash=meta_data.get('end_hash'),
-                            prev_hash=meta_data.get('start_parent_hash'),
                         )
                     )
 
@@ -298,9 +300,7 @@ def invalidate_from_block(
                 if remaining_batches:
                     remaining_batches.sort(key=lambda b: b['end_block'])
                     max_batch = remaining_batches[-1]
-                    meta_value = self._serialize_metadata(
-                        max_batch['end_block'], max_batch.get('end_hash'), max_batch.get('start_parent_hash')
-                    )
+                    meta_value = self._serialize_metadata(max_batch['end_block'], max_batch.get('end_hash'))
                     txn.put(meta_key, meta_value, db=self.metadata_db)
                 else:
                     txn.delete(meta_key, db=self.metadata_db)

tests/unit/test_lmdb_state.py

@@ -47,7 +47,7 @@ def test_mark_and_check_processed(self, lmdb_store):
 
     def test_is_processed_empty_list(self, lmdb_store):
         result = lmdb_store.is_processed('conn1', 'table1', [])
-        assert result is True
+        assert result is False
 
     def test_multiple_batches_all_must_be_processed(self, lmdb_store):
         batch_id1 = BatchIdentifier('ethereum', 100, 200, '0xabc')
@@ -106,7 +106,6 @@ def test_get_resume_position_single_network(self, lmdb_store):
         assert watermark.ranges[0].network == 'ethereum'
         assert watermark.ranges[0].end == 400
         assert watermark.ranges[0].hash == '0x123'
-        assert watermark.ranges[0].prev_hash == '0xparent3'
 
     def test_get_resume_position_multiple_networks(self, lmdb_store):
         eth_batch = BatchIdentifier('ethereum', 100, 200, '0xabc', '0xeth_parent')
@@ -285,7 +284,6 @@ def test_resume_position_with_many_out_of_order_batches(self, lmdb_store):
         watermark = lmdb_store.get_resume_position('conn1', 'table1')
         assert watermark.ranges[0].end == 800
         assert watermark.ranges[0].hash == '0x9'
-        assert watermark.ranges[0].prev_hash == '0xp9'
 
 
 class TestIntegrationScenarios: