From c7b533a8f47da18e93874e3e03aff0575d91b6de Mon Sep 17 00:00:00 2001 From: wb Date: Fri, 28 Aug 2026 17:24:01 +0800 Subject: [PATCH] fix(event): improve event delivery reliability Realtime event loading dropped new events when downstream processing reached the soft queue limit, while the native publisher applied its configured send high-water mark after socket creation. Pause block event loading at 500 queued events and retain pending events for later delivery. Normalize the native queue settings and apply the sendQueueLength before creating the publisher so explicit values take effect while non-positive values preserve the default behavior. --- .../nativequeue/NativeMessageQueue.java | 17 +++++----- .../core/services/event/BlockEventLoad.java | 2 +- .../services/event/RealtimeEventService.java | 10 +++--- .../logsfilter/NativeMessageQueueTest.java | 32 ++++++++++++++++--- .../tron/core/event/BlockEventLoadTest.java | 27 ++++++++++++++++ .../core/event/RealtimeEventServiceTest.java | 29 +++++++++++++++++ 6 files changed, 98 insertions(+), 19 deletions(-) diff --git a/framework/src/main/java/org/tron/common/logsfilter/nativequeue/NativeMessageQueue.java b/framework/src/main/java/org/tron/common/logsfilter/nativequeue/NativeMessageQueue.java index 7d97e6f4ba9..7337f3df175 100644 --- a/framework/src/main/java/org/tron/common/logsfilter/nativequeue/NativeMessageQueue.java +++ b/framework/src/main/java/org/tron/common/logsfilter/nativequeue/NativeMessageQueue.java @@ -27,22 +27,21 @@ public static NativeMessageQueue getInstance() { } public boolean start(int bindPort, int sendQueueLength) { - context = new ZContext(); - publisher = context.createSocket(SocketType.PUB); - - if (Objects.isNull(publisher)) { - return false; - } - - if (bindPort == 0 || bindPort < 0) { + if (bindPort <= 0) { bindPort = DEFAULT_BIND_PORT; } - if (sendQueueLength < 0) { + if (sendQueueLength <= 0) { sendQueueLength = DEFAULT_QUEUE_LENGTH; } + context = new ZContext(); context.setSndHWM(sendQueueLength); + publisher = context.createSocket(SocketType.PUB); + + if (Objects.isNull(publisher)) { + return false; + } String bindAddress = String.format("tcp://*:%d", bindPort); return publisher.bind(bindAddress); diff --git a/framework/src/main/java/org/tron/core/services/event/BlockEventLoad.java b/framework/src/main/java/org/tron/core/services/event/BlockEventLoad.java index 7efc10a8ed5..65c2adf573d 100644 --- a/framework/src/main/java/org/tron/core/services/event/BlockEventLoad.java +++ b/framework/src/main/java/org/tron/core/services/event/BlockEventLoad.java @@ -37,7 +37,7 @@ public class BlockEventLoad { public void init() { executor.scheduleWithFixedDelay(() -> { try { - if (!instance.isBusy()) { + if (!instance.isBusy() && !realtimeEventService.isBusy()) { load(); } } catch (Exception e) { diff --git a/framework/src/main/java/org/tron/core/services/event/RealtimeEventService.java b/framework/src/main/java/org/tron/core/services/event/RealtimeEventService.java index cef16cd81c1..d2ec6a83c36 100644 --- a/framework/src/main/java/org/tron/core/services/event/RealtimeEventService.java +++ b/framework/src/main/java/org/tron/core/services/event/RealtimeEventService.java @@ -29,7 +29,7 @@ public class RealtimeEventService { private static BlockingQueue queue = new LinkedBlockingQueue<>(); - private int maxEventSize = 10000; + private static final int BUSY_EVENT_SIZE = 500; private final ScheduledExecutorService executor = ExecutorServiceManager .newSingleThreadScheduledExecutor("realtime-event"); @@ -56,13 +56,13 @@ public void close() { } public void add(Event event) { - if (queue.size() >= maxEventSize) { - logger.warn("Add event failed, blockId {}.", event.getBlockEvent().getBlockId().getString()); - return; - } queue.offer(event); } + public boolean isBusy() { + return queue.size() >= BUSY_EVENT_SIZE; + } + public synchronized void work() { while (queue.size() > 0) { Event event = queue.poll(); diff --git a/framework/src/test/java/org/tron/common/logsfilter/NativeMessageQueueTest.java b/framework/src/test/java/org/tron/common/logsfilter/NativeMessageQueueTest.java index 5219654977b..1e7aeefbd09 100644 --- a/framework/src/test/java/org/tron/common/logsfilter/NativeMessageQueueTest.java +++ b/framework/src/test/java/org/tron/common/logsfilter/NativeMessageQueueTest.java @@ -1,5 +1,6 @@ package org.tron.common.logsfilter; +import java.lang.reflect.Field; import java.util.concurrent.ExecutorService; import org.junit.After; import org.junit.Assert; @@ -21,10 +22,20 @@ public class NativeMessageQueueTest { @After public void tearDown() { + NativeMessageQueue.getInstance().stop(); ExecutorServiceManager.shutdownAndAwaitTermination(subscriberExecutor, zmqSubscriber); subscriberExecutor = null; } + @Test + public void configuredSendQueueLengthIsAppliedToPublisherSocket() throws Exception { + int sendQueueLength = 2000; + + Assert.assertTrue(NativeMessageQueue.getInstance().start(bindPort, sendQueueLength)); + + Assert.assertEquals(sendQueueLength, getPublisher().getSndHWM()); + } + @Test public void invalidBindPort() { boolean bRet = NativeMessageQueue.getInstance().start(-1111, 0); @@ -33,10 +44,17 @@ public void invalidBindPort() { } @Test - public void invalidSendLength() { - boolean bRet = NativeMessageQueue.getInstance().start(0, -2222); - Assert.assertEquals(true, bRet); - NativeMessageQueue.getInstance().stop(); + public void negativeSendQueueLengthUsesDefaultSndHWM() throws Exception { + Assert.assertTrue(NativeMessageQueue.getInstance().start(bindPort, -1)); + + Assert.assertEquals(1000, getPublisher().getSndHWM()); + } + + @Test + public void zeroSendQueueLengthUsesDefaultSndHWM() throws Exception { + Assert.assertTrue(NativeMessageQueue.getInstance().start(bindPort, 0)); + + Assert.assertEquals(1000, getPublisher().getSndHWM()); } @Test @@ -84,4 +102,10 @@ public void startSubscribeThread() { } }); } + + private ZMQ.Socket getPublisher() throws ReflectiveOperationException { + Field publisherField = NativeMessageQueue.class.getDeclaredField("publisher"); + publisherField.setAccessible(true); + return (ZMQ.Socket) publisherField.get(NativeMessageQueue.getInstance()); + } } diff --git a/framework/src/test/java/org/tron/core/event/BlockEventLoadTest.java b/framework/src/test/java/org/tron/core/event/BlockEventLoadTest.java index 991133fee78..01f52ed2424 100644 --- a/framework/src/test/java/org/tron/core/event/BlockEventLoadTest.java +++ b/framework/src/test/java/org/tron/core/event/BlockEventLoadTest.java @@ -5,9 +5,11 @@ import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ScheduledExecutorService; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; +import org.tron.common.logsfilter.EventPluginLoader; import org.tron.common.utils.ReflectUtils; import org.tron.core.ChainBaseManager; import org.tron.core.capsule.BlockCapsule; @@ -23,6 +25,31 @@ public class BlockEventLoadTest { BlockEventLoad blockEventLoad = new BlockEventLoad(); + @Test(timeout = 2_000) + public void shouldNotLoadWhenRealtimeEventServiceIsBusy() throws Exception { + EventPluginLoader eventPluginLoader = mock(EventPluginLoader.class); + RealtimeEventService realtimeEventService = mock(RealtimeEventService.class); + Manager manager = mock(Manager.class); + ReflectUtils.setFieldValue(blockEventLoad, "instance", eventPluginLoader); + ReflectUtils.setFieldValue(blockEventLoad, "realtimeEventService", realtimeEventService); + ReflectUtils.setFieldValue(blockEventLoad, "manager", manager); + Mockito.when(eventPluginLoader.isBusy()).thenReturn(false); + Mockito.when(realtimeEventService.isBusy()).thenReturn(true); + + Field executorField = BlockEventLoad.class.getDeclaredField("executor"); + executorField.setAccessible(true); + ScheduledExecutorService executor = (ScheduledExecutorService) executorField + .get(blockEventLoad); + try { + blockEventLoad.init(); + + Mockito.verify(realtimeEventService, Mockito.timeout(1_000).atLeastOnce()).isBusy(); + Mockito.verifyNoInteractions(manager); + } finally { + executor.shutdownNow(); + } + } + @Test public void test() throws Exception { Method method = blockEventLoad.getClass().getDeclaredMethod("load"); diff --git a/framework/src/test/java/org/tron/core/event/RealtimeEventServiceTest.java b/framework/src/test/java/org/tron/core/event/RealtimeEventServiceTest.java index f58f725195c..4d6cb71bf85 100644 --- a/framework/src/test/java/org/tron/core/event/RealtimeEventServiceTest.java +++ b/framework/src/test/java/org/tron/core/event/RealtimeEventServiceTest.java @@ -3,8 +3,10 @@ import static org.mockito.Mockito.mock; import com.google.protobuf.ByteString; +import java.lang.reflect.Field; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.BlockingQueue; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; @@ -26,6 +28,33 @@ public class RealtimeEventServiceTest { RealtimeEventService realtimeEventService = new RealtimeEventService(); + @Test + public void shouldBecomeBusyAt500EventsAndRetainLaterEvents() throws Exception { + Field queueField = RealtimeEventService.class.getDeclaredField("queue"); + queueField.setAccessible(true); + BlockingQueue queue = (BlockingQueue) queueField.get(null); + queue.clear(); + + try { + Event event = mock(Event.class); + for (int i = 0; i < 499; i++) { + realtimeEventService.add(event); + } + + Assert.assertFalse(realtimeEventService.isBusy()); + + realtimeEventService.add(event); + Assert.assertTrue(realtimeEventService.isBusy()); + + realtimeEventService.add(event); + Assert.assertEquals(501, queue.size()); + } finally { + queue.clear(); + } + + Assert.assertFalse(realtimeEventService.isBusy()); + } + @Test public void test() throws Exception { BlockEvent be1 = new BlockEvent();