Skip to content

Ah/fix minor issue fixes to prod #12

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ahmedhamedaly merged 5 commits into from
Jan 6, 2026
Merged

Ah/fix minor issue fixes to prod #12

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0e5b6c8
feat: added mock tests for the new watcher mode
ahmedhamedaly Jan 5, 2026
4a7dd0a
fix: revert to older version with contract builders instead of local …
ahmedhamedaly Jan 5, 2026
044fb3a
fix: remove unused implementation of the token transfers watcher func…
ahmedhamedaly Jan 5, 2026
e165829
fix: replace sets with dict to maintain order when removing most recent
ahmedhamedaly Jan 5, 2026
4e2cec3
fix: abstracted max blocks per scan to config
ahmedhamedaly Jan 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions Dockerfile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,10 @@ RUN uv sync --frozen
# Copy source code
COPY rofl_oracle/ ./rofl_oracle/

# Copy pre-compiled ABIs directly
COPY abis/ ./abis/
# Copy ABIs from contracts build stage
RUN mkdir -p abis
COPY --from=contracts-builder /contracts/artifacts/contracts/hashi/adapters/Oasis/ROFLAdapter.sol/ROFLAdapter.json ./abis/
COPY --from=contracts-builder /contracts/artifacts/contracts/BlockHeaderRequester.sol/BlockHeaderRequester.json ./abis/

# Run the oracle using module execution
ENTRYPOINT ["uv", "run", "python", "-m", "rofl_oracle"]
11 changes: 11 additions & 0 deletions rofl_oracle/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,11 +202,13 @@ class TokenWatcherModeConfig:
:cvar token_addresses: List of ERC20 token contract addresses to monitor (env: TOKEN_ADDRESSES, comma-separated)
:cvar recipient_addresses: List of recipient addresses to watch for (env: RECIPIENT_ADDRESSES, comma-separated)
:cvar scan_interval: Seconds between scanning for token transfers (env: SCAN_INTERVAL)
:cvar max_blocks_per_scan: Maximum number of blocks to scan per iteration (env: MAX_BLOCKS_PER_SCAN, default: 10)
"""

token_addresses: list[str]
recipient_addresses: list[str]
scan_interval: int
max_blocks_per_scan: int = 10

def __post_init__(self) -> None:
"""Validate token watcher configuration."""
Expand All @@ -219,6 +221,15 @@ def __post_init__(self) -> None:
f"Scan interval too long (max 300s), got {self.scan_interval}"
)

if self.max_blocks_per_scan <= 0:
raise ValueError(
f"Max blocks per scan must be positive, got {self.max_blocks_per_scan}"
)
if self.max_blocks_per_scan > 10000:
raise ValueError(
f"Max blocks per scan too high (max 10000), got {self.max_blocks_per_scan}"
)

if not self.token_addresses or len(self.token_addresses) == 0:
raise ValueError(
"Token watcher mode requires at least one token address"
Expand Down
115 changes: 14 additions & 101 deletions rofl_oracle/header_oracle.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,7 +219,7 @@ async def _initialize(self, config: OracleConfig) -> None:
logger.info(f" Recipient: {addr}")
self.event_listener = None
self.last_scanned_block: int | None = None
self.processed_tx_hashes: set[str] = set()
self.processed_tx_hashes: dict[str, None] = {}
self.max_tx_cache_size = 10000
elif config.oracle_mode == OracleMode.WATCHER:
assert isinstance(config.mode_config, WatcherModeConfig)
Expand Down Expand Up @@ -709,7 +709,10 @@ async def watch_token_transfers(self) -> None:
if start_block > latest_block_number:
return

end_block = min(start_block + 100, latest_block_number)
end_block = min(
start_block + self.config.mode_config.max_blocks_per_scan,
latest_block_number,
)
num_blocks = end_block - start_block + 1
logger.info(
f"Scanning {num_blocks} blocks ({start_block}-{end_block})"
Expand All @@ -718,7 +721,9 @@ async def watch_token_transfers(self) -> None:
blocks_with_transfers = set()

for token_addr in self.config.mode_config.token_addresses:
for recipient_addr in self.config.mode_config.recipient_addresses:
for (
recipient_addr
) in self.config.mode_config.recipient_addresses:
recipient_topic = "0x" + recipient_addr[2:].lower().zfill(
64
)
Expand Down Expand Up @@ -751,16 +756,17 @@ async def watch_token_transfers(self) -> None:
if tx_hash_str in self.processed_tx_hashes:
continue

self.processed_tx_hashes.add(tx_hash_str)
self.processed_tx_hashes[tx_hash_str] = None

if (
len(self.processed_tx_hashes)
> self.max_tx_cache_size
):
to_remove = list(self.processed_tx_hashes)[
: self.max_tx_cache_size // 2
]
self.processed_tx_hashes -= set(to_remove)
keys_to_remove = list(
self.processed_tx_hashes.keys()
)[: self.max_tx_cache_size // 2]
for key in keys_to_remove:
del self.processed_tx_hashes[key]

blocks_with_transfers.add(block_num)
logger.info(
Expand Down Expand Up @@ -809,99 +815,6 @@ async def watch_token_transfers(self) -> None:
except Exception as e:
logger.error(f"Error in token watcher: {e}", exc_info=True)

async def _check_block_for_token_transfers(self, block_number: int) -> bool:
try:
block = self.source_w3.eth.get_block(block_number)
if not block:
return False

tx_hashes = block.get("transactions", [])
if not tx_hashes:
return False

for tx_hash in tx_hashes:
try:
tx = self.source_w3.eth.get_transaction(tx_hash)
if not tx:
continue

tx_to = tx.get("to", "")
if not tx_to or tx_to.lower() not in self.token_addresses:
continue

receipt = self.source_w3.eth.get_transaction_receipt(
tx_hash
)
if not receipt:
continue

for log in receipt.get("logs", []):
if (
log.get("address", "").lower()
not in self.token_addresses
):
continue

topics = log.get("topics", [])
if len(topics) < 3:
continue

event_sig = (
topics[0].hex()
if isinstance(topics[0], bytes)
else topics[0]
)
if (
event_sig.lower()
!= TRANSFER_EVENT_SIGNATURE.lower()
):
continue

to_address_topic = topics[2]
to_address_hex = (
to_address_topic.hex()
if isinstance(to_address_topic, bytes)
else to_address_topic
)
if to_address_hex.startswith("0x"):
to_address_hex = to_address_hex[2:]
to_address = "0x" + to_address_hex[-40:]

if to_address.lower() in self.recipient_addresses:
tx_hash_str = (
tx_hash.hex()
if isinstance(tx_hash, bytes)
else tx_hash
)
if tx_hash_str in self.processed_tx_hashes:
continue

self.processed_tx_hashes.add(tx_hash_str)

if (
len(self.processed_tx_hashes)
> self.max_tx_cache_size
):
to_remove = list(self.processed_tx_hashes)[
: self.max_tx_cache_size // 2
]
self.processed_tx_hashes -= set(to_remove)

token_addr = log.get("address", "unknown")
logger.info(
f"Transfer detected: token={token_addr}, to={to_address}, tx={tx_hash_str}"
)
return True

except Exception:
continue

return False

except Exception as e:
logger.warning(f"Skipping block {block_number}: {e}")
return False

async def shutdown(self) -> None:
"""Gracefully shutdown the oracle."""
logger.info("Shutting down HeaderOracle...")
Expand Down
Loading