fix: adjust for unit testing against kaolin

Author: Matthias Zimmermann

src/arkiv/contract.py

@@ -188,7 +188,7 @@
         mungers=[default_root_munger],
     ),
     "query": Method(
-        json_rpc_method=RPCEndpoint("golembase_query"),
+        json_rpc_method=RPCEndpoint("arkiv_query"),
         mungers=[default_root_munger],
     ),
 }

src/arkiv/events.py

@@ -1,4 +1,20 @@
-"""Event filtering and monitoring for Arkiv entities."""
+"""Event filtering and monitoring for Arkiv entities.
+
+This module provides event watching functionality for Arkiv smart contract events.
+It supports two polling strategies:
+
+1. Filter-based polling (eth_newFilter + eth_getFilterChanges):
+   - More efficient, uses persistent filters on the node
+   - Works with local nodes and some RPC providers
+   - Automatically tried first
+
+2. Log polling (eth_getLogs):
+   - Maximum compatibility, works with all RPC providers
+   - Used as fallback when filter creation fails (e.g., Kaolin testnet blocks eth_newFilter)
+   - Polls for new logs on each interval
+
+The implementation automatically detects which method to use based on provider capabilities.
+"""
 
 from __future__ import annotations
 
@@ -42,7 +58,8 @@ class EventFilter(EventFilterBase[SyncCallback]):
     """
     Handle for watching entity events using HTTP polling.
 
-    Uses polling-based filter with get_new_entries() for event monitoring.
+    Uses direct log polling via eth_getLogs for maximum compatibility.
+    This approach works with RPC providers that don't support eth_newFilter.
     WebSocket providers are not supported by the sync Arkiv client.
 
     Inherits shared event parsing logic from EventFilterBase.
@@ -73,6 +90,12 @@ def __init__(
         self._thread: threading.Thread | None = None
         self._filter: LogFilter | None = None
 
+        # Track last processed block for log polling
+        self._last_block: int | None = None
+
+        # Try filter-based approach first (fallback to log polling if it fails)
+        self._use_filter = True
+
         if auto_start:
             self.start()
 
@@ -86,13 +109,24 @@ def start(self) -> None:
 
         logger.info(f"Starting event filter for {self.event_type}")
 
-        # Create the Web3 filter using base class helper
-        # Note: base class returns LogFilter | AsyncLogFilter, but for sync we only use LogFilter
-        filter_result = self._create_filter()
-        assert isinstance(filter_result, LogFilter), (
-            "Expected LogFilter for sync client"
-        )
-        self._filter = filter_result
+        # Try to create a Web3 filter first (works with most local nodes)
+        # If it fails (403 Forbidden from Kaolin), we'll use log polling instead
+        try:
+            filter_result = self._create_filter()
+            assert isinstance(filter_result, LogFilter), (
+                "Expected LogFilter for sync client"
+            )
+            self._filter = filter_result
+            self._use_filter = True
+            logger.info(f"Using filter-based polling for {self.event_type}")
+        except Exception as e:
+            logger.warning(f"Filter creation failed ({e}), falling back to log polling")
+            self._use_filter = False
+            # Initialize last_block for log polling
+            if self.from_block == "latest":
+                self._last_block = self.contract.w3.eth.block_number
+            else:
+                self._last_block = int(self.from_block) - 1
 
         # Start polling thread
         self._running = True
@@ -138,14 +172,20 @@ def _poll_loop(self) -> None:
 
         while self._running:
             try:
-                # Get new entries from filter
-                if self._filter:
-                    logs: list[LogReceipt] = self._filter.get_new_entries()
-                    for log in logs:
-                        try:
-                            self._process_log(log)
-                        except Exception as e:
-                            logger.error(f"Error processing event: {e}", exc_info=True)
+                if self._use_filter:
+                    # Filter-based approach (works with local nodes)
+                    if self._filter:
+                        logs: list[LogReceipt] = self._filter.get_new_entries()
+                        for log in logs:
+                            try:
+                                self._process_log(log)
+                            except Exception as e:
+                                logger.error(
+                                    f"Error processing event: {e}", exc_info=True
+                                )
+                else:
+                    # Log polling approach (works with Kaolin and other restricted RPC providers)
+                    self._poll_logs()
 
                 # Sleep before next poll
                 time.sleep(self._poll_interval)
@@ -156,6 +196,55 @@ def _poll_loop(self) -> None:
 
         logger.debug(f"Poll loop ended for {self.event_type}")
 
+    def _poll_logs(self) -> None:
+        """
+        Poll for logs using eth_getLogs.
+
+        This is a fallback method for RPC providers that don't support eth_newFilter.
+        """
+        try:
+            current_block = self.contract.w3.eth.block_number
+
+            # Only query if there are new blocks
+            if self._last_block is not None and current_block <= self._last_block:
+                return
+
+            # Get the contract event
+            event_name = self._get_contract_event_name()
+            contract_event = self.contract.events[event_name]
+
+            # Calculate from_block for this poll
+            from_block = (
+                (self._last_block + 1)
+                if self._last_block is not None
+                else current_block
+            )
+
+            # Get logs for new blocks
+            logs = contract_event.get_logs(
+                from_block=from_block,
+                to_block=current_block,
+            )
+
+            # Process each log
+            for log in logs:
+                try:
+                    self._process_log(log)
+                except Exception as e:
+                    logger.error(f"Error processing event: {e}", exc_info=True)
+
+            # Update last processed block
+            self._last_block = current_block
+
+            if logs:
+                logger.debug(
+                    f"Processed {len(logs)} {self.event_type} events "
+                    f"from blocks {from_block} to {current_block}"
+                )
+
+        except Exception as e:
+            logger.error(f"Error polling logs: {e}", exc_info=True)
+
     def _process_log(self, log: LogReceipt) -> None:
         """
         Process a single log receipt and trigger sync callback.

src/arkiv/events_async.py

@@ -1,4 +1,20 @@
-"""Async event filtering and monitoring for Arkiv entities."""
+"""Async event filtering and monitoring for Arkiv entities.
+
+This module provides async event watching functionality for Arkiv smart contract events.
+It supports two polling strategies:
+
+1. Filter-based polling (eth_newFilter + eth_getFilterChanges):
+   - More efficient, uses persistent filters on the node
+   - Works with local nodes and some RPC providers
+   - Automatically tried first
+
+2. Log polling (eth_getLogs):
+   - Maximum compatibility, works with all RPC providers
+   - Used as fallback when filter creation fails (e.g., Kaolin testnet blocks eth_newFilter)
+   - Polls for new logs on each interval
+
+The implementation automatically detects which method to use based on provider capabilities.
+"""
 
 from __future__ import annotations
 
@@ -41,7 +57,8 @@ class AsyncEventFilter(EventFilterBase[AsyncCallback]):
     """
     Handle for watching entity events using async HTTP polling.
 
-    Uses async polling-based filter with get_new_entries() for event monitoring.
+    Uses direct log polling via eth_getLogs for maximum compatibility.
+    This approach works with RPC providers that don't support eth_newFilter.
     WebSocket providers are not supported yet (future enhancement).
 
     Inherits shared event parsing logic from EventFilterBase.
@@ -74,6 +91,12 @@ def __init__(
         self._task: asyncio.Task[None] | None = None
         self._filter: LogFilter | None = None
 
+        # Track last processed block for log polling
+        self._last_block: int | None = None
+
+        # Try filter-based approach first (fallback to log polling if it fails)
+        self._use_filter = True
+
     async def _create_filter(self) -> Any:  # type: ignore[override]
         """
         Create a Web3 contract event filter for async HTTP polling.
@@ -102,8 +125,20 @@ async def start(self) -> None:
 
         logger.info(f"Starting async event filter for {self.event_type}")
 
-        # Create the Web3 filter using async helper
-        self._filter = await self._create_filter()
+        # Try to create a Web3 filter first (works with most local nodes)
+        # If it fails (403 Forbidden from Kaolin), we'll use log polling instead
+        try:
+            self._filter = await self._create_filter()
+            self._use_filter = True
+            logger.info(f"Using filter-based polling for {self.event_type}")
+        except Exception as e:
+            logger.warning(f"Filter creation failed ({e}), falling back to log polling")
+            self._use_filter = False
+            # Initialize last_block for log polling
+            if self.from_block == "latest":
+                self._last_block = await self.contract.w3.eth.block_number  # type: ignore[misc]
+            else:
+                self._last_block = int(self.from_block) - 1
 
         # Start async polling task
         self._running = True
@@ -152,14 +187,20 @@ async def _poll_loop(self) -> None:
 
         while self._running:
             try:
-                # Get new entries from filter
-                if self._filter:
-                    logs: list[LogReceipt] = await self._filter.get_new_entries()  # type: ignore[misc]
-                    for log in logs:
-                        try:
-                            await self._process_log(log)
-                        except Exception as e:
-                            logger.error(f"Error processing event: {e}", exc_info=True)
+                if self._use_filter:
+                    # Filter-based approach (works with local nodes)
+                    if self._filter:
+                        logs: list[LogReceipt] = await self._filter.get_new_entries()  # type: ignore[misc]
+                        for log in logs:
+                            try:
+                                await self._process_log(log)
+                            except Exception as e:
+                                logger.error(
+                                    f"Error processing event: {e}", exc_info=True
+                                )
+                else:
+                    # Log polling approach (works with Kaolin and other restricted RPC providers)
+                    await self._poll_logs()
 
                 # Async sleep before next poll
                 await asyncio.sleep(self._poll_interval)
@@ -173,6 +214,55 @@ async def _poll_loop(self) -> None:
 
         logger.debug(f"Async poll loop ended for {self.event_type}")
 
+    async def _poll_logs(self) -> None:
+        """
+        Poll for logs using eth_getLogs (async version).
+
+        This is a fallback method for RPC providers that don't support eth_newFilter.
+        """
+        try:
+            current_block = await self.contract.w3.eth.block_number  # type: ignore[misc]
+
+            # Only query if there are new blocks
+            if self._last_block is not None and current_block <= self._last_block:
+                return
+
+            # Get the contract event
+            event_name = self._get_contract_event_name()
+            contract_event = self.contract.events[event_name]
+
+            # Calculate from_block for this poll
+            from_block = (
+                (self._last_block + 1)
+                if self._last_block is not None
+                else current_block
+            )
+
+            # Get logs for new blocks
+            logs = await contract_event.get_logs(
+                from_block=from_block,
+                to_block=current_block,
+            )
+
+            # Process each log
+            for log in logs:
+                try:
+                    await self._process_log(log)
+                except Exception as e:
+                    logger.error(f"Error processing event: {e}", exc_info=True)
+
+            # Update last processed block
+            self._last_block = current_block
+
+            if logs:
+                logger.debug(
+                    f"Processed {len(logs)} {self.event_type} events "
+                    f"from blocks {from_block} to {current_block}"
+                )
+
+        except Exception as e:
+            logger.error(f"Error polling logs: {e}", exc_info=True)
+
     async def _process_log(self, log: LogReceipt) -> None:
         """
         Process a single event and trigger async callback.

src/arkiv/events_base.py

@@ -65,7 +65,7 @@ def __init__(
         self.callback: CallbackT = callback
         self.from_block: str | int = from_block
         self._running: bool = False
-        self._poll_interval: float = 2.0  # seconds
+        self._poll_interval: float = 1.0  # seconds
 
     @property
     def is_running(self) -> bool:

src/arkiv/node.py

@@ -397,7 +397,7 @@ def _get_command(self) -> str:
         return (
             "--dev "
             "--http "
-            "--http.api 'eth,web3,net,debug,golembase' "
+            "--http.api 'eth,web3,net,debug,golembase,arkiv' "
             f"--http.port {self._http_port} "
             "--http.addr '0.0.0.0' "
             "--http.corsdomain '*' "

tests/test_arkiv_fixture.py

@@ -2,6 +2,8 @@
 
 import logging
 
+import pytest
+
 from arkiv import Arkiv
 from arkiv.account import NamedAccount
 from arkiv.module import ArkivModule
@@ -83,6 +85,7 @@ def test_arkiv_accounts_are_funded(
     logger.info("Arkiv fixture accounts are funded (balances > 0)")
 
 
+@pytest.mark.skip("Skipping ETH transfer not (yet) supported on db chains.")
 def test_arkiv_transfer_eth_account(
     arkiv_client_http: Arkiv, account_1: NamedAccount, account_2: NamedAccount
 ) -> None:
@@ -106,6 +109,7 @@ def test_arkiv_transfer_eth_account(
     logger.info("Arkiv ETH transfer between accounts succeeded (to: NamedAccount)")
 
 
+@pytest.mark.skip("Skipping ETH transfer not (yet) supported on db chains.")
 def test_arkiv_transfer_eth_address(
     arkiv_client_http: Arkiv, account_1: NamedAccount, account_2: NamedAccount
 ) -> None: