Improve event delivery and ZeroMQ queue configuration

Start with RealtimeEventService.add() and work(), then trace how BlockEventLoad produces events and how queue pressure could pause and resume loading. Inspect NativeMessageQueue.start() in useNativeQueue mode, focusing on when SndHWM is configured relative to socket creation. Done means pending events apply backpressure without drops and sendQueueLength controls the publisher queue.

Summary

The event service has two issues affecting reliability and configurability:

• RealtimeEventService uses an unbounded LinkedBlockingQueue with an application-level 10,000-entry soft limit. Once that threshold is reached, new events are dropped instead of applying backpressure.
• NativeMessageQueue sets the send high-water mark (SndHWM) only after creating the socket, causing the configured send queue length to be ineffective.

We recommend introducing explicit backpressure between RealtimeEventService and BlockEventLoad, and correcting the timing of the SndHWM configuration.

Root Cause

1. RealtimeEventService drops events when the queue is full without applying backpressure

The queue is an unbounded LinkedBlockingQueue, and a single-threaded scheduler calls work() once per second to consume events in batches. The producer-side add() logic is:

public void add(Event event) {
  if (queue.size() >= maxEventSize) {          // maxEventSize = 10000
    logger.warn("Add event failed, blockId {}.", ...);
    return;                                     // Drop the event directly
  }
  queue.offer(event);
}

When downstream consumption (work(), which runs once per second) cannot keep up with the event production rate and the queue reaches 10,000 entries, subsequent events are dropped directly, with only a warning being logged.

This is a drop-when-full strategy rather than a backpressure mechanism that applies pressure to the producer. Events generated while the queue is full are therefore silently lost.

2. NativeMessageQueue sets SndHWM too late, causing the configuration to be ineffective

In start(), the PUB socket is created first, and context.setSndHWM(sendQueueLength) is called afterward.
In ZeroMQ (JeroMQ), ZContext.setSndHWM only affects sockets created after the context-level setting is applied. The high-water mark is applied to the socket when it is created, so changing the context value afterward does not affect an already-created publisher socket.

As a result, the configured sendQueueLength is silently ignored, and the publisher continues to use the default high-water mark of 1000.

Reproduction

1. RealtimeEventService

Slow down downstream event consumption so that the event production rate remains higher than the consumption rate of once per second.

Once the queue reaches 10,000 entries, observe that subsequent events are dropped, with only an Add event failed warning being logged. The dropped events cannot be recovered.

2. NativeMessageQueue

In useNativeQueue mode, configure an explicit sendQueueLength. After startup, inspect the actual send high-water mark of the publisher.

The actual value remains at the default of 1000 instead of the configured value.

Impact

• RealtimeEventService: When downstream consumption slows down, events beyond the 10,000-entry queue limit are silently dropped, causing real-time event loss and potentially resulting in subscribers missing events. The current drop-when-full + log strategy does not propagate backpressure to upstream producers.
• NativeMessageQueue: The configured send queue length is ineffective, preventing operators from controlling the publisher's send queue through configuration. When downstream consumption slows down and pending messages exceed the default high-water mark, excess messages may be dropped directly by the PUB socket, again resulting in silent event loss. The affected scope is limited to useNativeQueue mode.
• Both issues affect the reliability and configurability of event delivery. They do not affect consensus or asset security.

Suggested Fix

• RealtimeEventService: Introduce an explicit backpressure mechanism, such as an isBusy() method. When the number of pending events reaches the configured threshold, BlockEventLoad should pause loading new events and resume once the queue size falls below the recovery threshold. This prevents events from being dropped when the queue reaches a fixed limit.

• NativeMessageQueue: Move the setSndHWM configuration to before the socket is created, or directly call setSndHWM(sendQueueLength) on the socket itself, so that the configured value takes effect.