Parallelizing eth_newFilter event matching

[317787106]

Background

The JSON-RPC event subscription API eth_newFilter is a stateful service designed for efficiently querying newly generated transaction events. This mechanism works by matching user-defined filter criteria against the Bloom filters of block events, and returning the matched results through the eth_getFilterChanges API.

Currently, the event-matching process is executed in a single-threaded, sequential manner. Under high concurrency with a large number of active subscriptions, the matching latency increases significantly, causing the event processing throughput to fall behind the block synchronization speed. As a result, users experience noticeable delays when retrieving the latest events.

To address this issue, we propose optimizing the existing sequential matching mechanism into a parallel matching model to improve overall throughput and processing efficiency. In addition, an upper limit will be enforced on the number of concurrently created subscriptions. When this limit is exceeded, the system will return a clear and explicit error message to prevent resource exhaustion and further improve user experience.

Rationale

Why should this feature exist?

To improve user experience by preventing delayed delivery of subscribed events.

What are the use-cases?

The combination of eth_newFilter and eth_getFilterChanges can be used to build event-driven WebSocket services for real-time event subscriptions.

Specification

After a FullNode finishes processing a block, all transaction events generated within that block are written into a FIFO queue named filterCapsuleQueue. A background thread, filterProcessLoop, continuously consumes events from this queue and processes them block by block. For each block, the handleLogsFilter method is invoked to match the events against the subscription conditions created via eth_newFilter. Matched events are temporarily stored in a cache and can later be retrieved by users through eth_getFilterChanges.

If a subscription is not accessed within 300 seconds, it is considered expired, and its associated cached data will be automatically cleaned up.

In the current implementation, if there are m active subscriptions, each of the n events in a block must be matched against all subscriptions, resulting in a time complexity of O(m × n). When the number of subscriptions reaches approximately 200,000, the matching time for a single block can take up to 20 seconds, which significantly exceeds the block production interval (3 seconds). This leads to severe event processing lag: existing subscribers experience substantial delays, while newly created subscriptions starting from the latest block may fail to receive any events for an extended period of time.

To resolve this issue, the sequential matching model between block events and subscriptions is optimized by partitioning subscriptions into groups and matching them in parallel against the event list of a block. At the same time, the ordering of events within each individual subscription is strictly preserved.

Since event matching is a CPU-intensive operation, the level of parallelism is intentionally limited to 2–4 threads to avoid negatively impacting block synchronization and other core processing workflows.

Furthermore, to prevent uncontrolled matching latency under extreme subscription volumes, a configuration parameter node.jsonrpc.maxLogFilterNum is introduced to cap the maximum number of log filters that can be created. When the number of active subscriptions exceeds this threshold, new subscription requests will be rejected with a clear error message: “exceed max log filters“.

Service operators can adjust the default maximum subscription limit based on their hardware resources and system load.

Test Specification

Stress testing can be used to evaluate block-level event matching latency and CPU utilization. Test results will be provided in a follow-up update.

Scope of Impact

This change only affects the JSON-RPC eth_newFilter interface.

Implementation

Do you have ideas regarding the implementation of this feature?

Yes.

Are you willing to implement this feature?

Yes.

[lxcmyf]

Here are some questions:

1. Is the subscription group static or dynamic load balancing?

2. How to ensure the event sequence of a single filter?

[317787106]

@lxcmyf For the second question, Each newly generated event is sequentially matched against all filters. If a match is found, the event is cached in the buffer corresponding to that filter. Since events are generated in a strictly ordered manner, and the grouping is applied to filters rather than to events, the events stored in each filter’s cache are guaranteed to remain ordered.

[waynercheung]

@317787106 Hi, it is good idea, and I have a question.

Since event matching is a CPU-intensive operation, the level of parallelism is intentionally limited to 2–4 threads to avoid negatively impacting block synchronization and other core processing workflows.

How was the 2-4 thread range determined? With 200,000 subscriptions and 20-second matching times, would 2-4 threads provide sufficient improvement to bring matching time below the 3-second block interval? Has there been any performance modeling or benchmarking to validate that this thread count can solve the stated problem?

[Sunny6889]

Since the number of transactions/events in a single block is relatively small (typically hundreds), while the number of subscriptions can be huge, just wonder can java-tron iterating by events instead of subscriptions?

An inverted index (e.g., mapping Address/Topic -> List<FilterID>) could allow us to directly locate the relevant filters for each event. This would reduce the matching complexity from O(M_filters * N_events) to O(N_events * k_matched_filters), which is algorithmically much more efficient than just parallelizing the brute-force scan.

Current Approach (Iterating Filters):

// M = 200,000 (filters), N = 200 (events)
// Complexity: M * N = 40,000,000 checks
for (Filter filter : allFilters) {
    for (Event event : blockEvents) {
        if (filter.matches(event)) {
            addResult(filter, event);
        }
    }
}

Proposed Approach (Iterating Events with Index):

// Complexity: N * k (where k is avg matched filters per event, e.g., 50)
// Total: 200 * 50 = 10,000 checks (vs 40M)
for (Event event : blockEvents) {
    // Direct lookup: O(1) or O(log)
    List<Filter> candidates = index.get(event.address); 
    
    if (candidates != null) {
        for (Filter filter : candidates) {
            if (filter.matches(event)) {
                addResult(filter, event);
            }
        }
    }
}

[317787106]

@lxcmyf For the first question, we use java.util.Collection.parallelStream(), where the data is dynamically partitioned and scheduled based on load balancing.

[317787106]

@waynercheung We may rely on a ForkJoinPool, without explicitly specifying the parallelism level. The default configuration is intended to maximize CPU utilization while ensuring block processing is not delayed. It needs more test.

[317787106]

@Sunny6889 Good suggestion. An inverted index is difficult to maintain due to the short lifetime of filters and the additional memory cost, but it is still worth exploring.

[warku123]

Hi @317787106 ,
I’ve reviewed the current implementation and the proposed parallelization. As it stands, handleLogsFilter still calls LogMatch.matchBlock once per active filter for the same block, so the block’s info like txInfoList is fully scanned many times.

Parallelism will help reduce wall‑clock latency, but the repeated work per block remains. As a possible follow‑up, maybe we could find a way to pre‑parsing each block’s logs once, store it and sharing that across filters, or even moving from a “per filter scan of the whole block” model to a “per log dispatch to candidate filters” model to further reduce CPU usage.