Run Model

Important: The event listener is implemented with robustness is mind, checking for weird responses from the RCP and trying to re-fetch suspicious blocks. In some situation the listener MAY intentionally call exit(1) or exit(2) to indicate a strong issue. For liveliness, the container should schedule a re-start (e.g. Docker: restart: unless-stopped) (and logs and provider configuration should be checked).

Usage

• Extend this class and provide the mapping of event to event-handler (implement {@link #topicHandlers()}). Handlers SHOULD return a boolean where false will trigger a retry and true removes the event as handled. IMPORTANT: If the listener is configured to not skip failed handling with later retry (i.e., preserve ordering), a handler returning consistently false will block the listener.
• Base-class provides two methods to listen to the logs: polling or streaming (websocket).
  • Polling: See Web3JListener#listenToLogsPolling(String, DefaultBlockParameter, BigInteger, Consumer, Consumer, String...)
  • Streaming: See Web3JListener#listenToLogsStreaming(String, Supplier, Consumer, Consumer, String...)

Features

• Finality buffering with re-validation: polling and streaming wait finalityBlocks number of blocks until a log will be handled. Polling will not poll non-final blocks. Streaming will re-validate a block once it became final (and re-fetched, if it had changed (re-orgs, etc.)).
• Reconciliation: When running "streaming" an optional reconciliation can be run (which is equivalent to polling) which ensures that no log message was skipped. Reconciliation runs periodically (like polling). In streaming mode, when finality processing encounters finalized pending blocks beyond the reconciliation watermark (reconciledToBlock), it schedules a coalesced/debounced reconciliation run to advance the deterministic backfill. This avoids reconciling on every streamed log while still preventing cursor advancement past missing logs. By this, streaming has low latency AND is as secure as polling.
• Cost considerations: Depending on the provider we have
  • low volume: streaming with coarse reconciliation is usually cheaper than frequent polling.
  • high volume: frequent polling may be cheaper than streaming.
• Cursor persistence: A cursor (blockNumber:logIndex) is maintained and persisted indicating the last successfully handled log entry.
• Queue-only processing: handler logic is executed only on the finality thread (single threaded) (to avoid race conditions in tx)
• Gap-repair hardening (for streaming with reconciliation): When streaming is used AND reconciliationIntervalMs > 0, we gate processing by a reconciliation watermark `reconciledToBlock`. The finality thread will not process blocks beyond the highest block range that has been deterministically swept by eth_getLogs. This prevents the cursor from jumping ahead of missing logs such that later reconciliation would be skipped by `#isAlreadyProcessed(Log)`.
• (Optional) Order preservation: in general the logs are processed in their natural order (blockNumber:logIndex). Handler logic is executed only on the finality thread (single threaded) (to avoid race conditions in tx)
• Preflight: provides a helper method to perform a preflight check (use it in your handler to check if a transaction would fail, if not, perform the true transaction).
• Receipt logging: a separate thread will look for the tx confirmation and log the state and cost (gasUsed / status / effectiveGasPrice).
• Sanity check: This base class assumes the contract exposes isOracle(address) or oracle() returning an address for the startup sanity check. Override {@link #sanityCheckOrExit()} to customize/skip.