Improve event delivery and ZeroMQ queue configuration

[xxo1shine]

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.

[waynercheung]

@xxo1shine Thanks for the clear report. I verified both points against develop and agree that they are valid bugs.

For RealtimeEventService, one detail is worth clarifying: work() drains the queue until it becomes empty; it is not limited to a fixed-size batch once per second. However, the conclusion still holds—if flush() throughput remains below the producer throughput, the queue reaches the 10,000-entry soft limit and add() drops subsequent events without notifying the caller or retrying them.

This is especially problematic after rollback support was introduced. Losing a normal event causes a gap, while losing an event with removed=true may leave subscribers with events from an orphaned block. In addition, BlockEventLoad may advance BlockEventCache even when the corresponding event was not accepted, so the dropped event will not be loaded again.

I support adding load-side backpressure, but a simple isBusy() check before load() is not sufficient by itself. One load operation can enqueue up to 100 forward blocks plus rollback events, so a batch can cross the threshold after the check.

I suggest the following acceptance criteria:

• Pause only BlockEventLoad, rather than blocking block processing or validation.
• Use high/low watermarks to avoid frequent pause/resume oscillation.
• Admit the complete rollback + forward batch atomically, or reserve enough capacity for the whole batch.
• Do not advance BlockEventCache until the event batch has been accepted.
• Avoid an unbounded blocking put() while holding the manager lock.
• Add queue-depth, pause/resume and dropped-event metrics.
• Cover saturation/recovery, reorg ordering, batch-boundary and shutdown cases in tests.

For NativeMessageQueue, the analysis is correct. In JeroMQ 0.5.3, ZContext.createSocket() copies the context HWM into the new socket. Calling context.setSndHWM() afterward therefore does not update the existing publisher. I would prefer validating the value and calling:

publisher.setSndHWM(sendQueueLength);

after socket creation but before bind(), then asserting publisher.getSndHWM() in a unit test. We should also explicitly define the behavior of sendQueueLength == 0, since ZeroMQ treats zero as an unlimited HWM.

One scope clarification: making SndHWM effective improves configurability, but it does not provide reliable ZeroMQ PUB delivery. PUB sockets remain best-effort and may drop messages for a slow subscriber after that subscriber's HWM is reached. If at-least-once delivery is a requirement, persistence/replay or an acknowledgement-based protocol should be addressed separately; otherwise the best-effort behavior should be documented.

With these acceptance criteria, I support fixing both issues in the next release. The HWM correction is low risk and can be handled independently from the more concurrency-sensitive backpressure change.

[xxo1shine]

@waynercheung, this change will not alter the default value of SndHWM. It only corrects when the setting is applied, ensuring that the user-configured sendQueueLength takes effect on the publisher. The semantics of sendQueueLength == 0 are outside the scope of this change.

The current backpressure approach only pauses subsequent BlockEventLoad operations once the queue reaches the threshold. The queue itself does not reject the current batch when that threshold is reached. Therefore, even if a single load temporarily pushes the queue above the threshold, no rollback or forward events are dropped. Further loading is paused and resumes after the backlog decreases.

In addition, BlockEventLoad runs independently from block processing and validation. Pausing event loading therefore does not block or affect either of those workflows. The current approach is straightforward and has a limited risk surface. Since both issues are well-defined and relatively simple, they can be handled together in the same change.