fix sm (#36)

* fix sm

* fix advisories

Author: lvboudre

Cargo.lock

@@ -16,6 +16,10 @@ The minor version will be incremented upon a breaking change and the patch versi
 
 ### Breaking
 
+## [0.2.2]
+
+- Fixed `FumaroleSM.need_new_blockchain_events` to follow rust's implementation [#35](https://github.com/rpcpool/yellowstone-fumarole/issues/35).
+
 ## [0.2.1]
 
 - Fixed `OrderedSet.popfirst` + missing `await self.aclose()` in `run` method [PR](https://github.com/rpcpool/yellowstone-fumarole/pull/33) [@Yolley](https://github.com/Yolley)

python/yellowstone-fumarole-client/CHANGELOG.md

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "yellowstone-fumarole-client"
-version = "0.2.1"
+version = "0.2.2"
 homepage = "https://github.com/rpcpool/yellowstone-fumarole"
 repository = "https://github.com/rpcpool/yellowstone-fumarole"
 description = "Yellowstone Fumarole Python Client"

python/yellowstone-fumarole-client/pyproject.toml

@@ -133,6 +133,8 @@ def __init__(
         self.commit_interval = commit_interval
         self.gc_interval = gc_interval
         self.max_concurrent_download = max_concurrent_download
+        self.poll_hist_inflight = False
+        self.commit_offset_inflight = False
 
         # holds metadata about the download task
         self.download_tasks = dict()
@@ -167,10 +169,12 @@ def _handle_control_response(self, control_response: ControlResponse):
 
         match response_field:
             case "poll_hist":
+                self.poll_hist_inflight = False
                 poll_hist = control_response.poll_hist
                 LOGGER.debug(f"Received poll history {len(poll_hist.events)} events")
                 self.sm.queue_blockchain_event(poll_hist.events)
             case "commit_offset":
+                self.commit_offset_inflight = False
                 commit_offset = control_response.commit_offset
                 LOGGER.debug(f"Received commit offset: {commit_offset}")
                 self.sm.update_committed_offset(commit_offset.offset)
@@ -181,9 +185,12 @@ def _handle_control_response(self, control_response: ControlResponse):
 
     async def poll_history_if_needed(self):
         """Poll the history if the state machine needs new events."""
+        if self.poll_hist_inflight:
+            return
         if self.sm.need_new_blockchain_events():
             cmd = self._build_poll_history_cmd(self.sm.committable_offset)
             await self.control_plane_tx.put(cmd)
+            self.poll_hist_inflight = True
 
     def commitment_level(self):
         """Gets the commitment level from the subscribe request."""
@@ -241,9 +248,12 @@ async def _force_commit_offset(self):
 
     async def _commit_offset(self):
         self.last_commit = time.time()
+        if self.commit_offset_inflight:
+            return
         if self.sm.last_committed_offset < self.sm.committable_offset:
             LOGGER.debug(f"Committing offset {self.sm.committable_offset}")
             await self._force_commit_offset()
+            self.commit_offset_inflight = True
 
     async def _drain_slot_status(self):
         """Drains the slot status from the state machine and sends updates to the Dragonsmouth outlet."""

python/yellowstone-fumarole-client/yellowstone_fumarole_client/runtime/aio.py

@@ -322,4 +322,6 @@ def processed_offset_queue_len(self) -> int:
 
     def need_new_blockchain_events(self) -> bool:
         """Check if new blockchain events are needed."""
-        return not self.slot_status_update_queue and not self.blocked_slot_status_update
+        MINIMUM_UNPROCESSED_BLOCKCHAIN_EVENT = 10
+        return len(self.unprocessed_blockchain_event) < MINIMUM_UNPROCESSED_BLOCKCHAIN_EVENT \
+            or (not self.slot_status_update_queue and not self.blocked_slot_status_update)