From 3c03c6da32748149c9a1739cedb00e4f539b516b Mon Sep 17 00:00:00 2001 From: perfch <154758349+perfch@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:08:15 +0300 Subject: [PATCH] feat(event): opt-in dual-sink for native queue and event plugin Allow the native ZMQ queue and the event plugin to run at the same time, with activation made explicit rather than inferred from a path so that existing native-queue deployments keep their current behavior after an upgrade. - Plugin activation is resolved explicitly in Args via EventPluginConfig.resolveUseEventPlugin(): with the native queue on it only joins when the operator opts in (runPluginWithNativeQueue = true), so a stale "path" in an upgraded config can never silently load a plugin. - Add pluginLoadFailurePolicy ("fail" | "ignore") to define behavior when the plugin cannot load; default "fail" preserves data integrity. - Isolate plugin trigger posting in try/catch so a plugin fault cannot tear down the node event thread; serialize each trigger once and share it with both sinks so the default path keeps its current cost. - Apply plugin back-pressure (isBusy) whenever the plugin is active, including dual-sink mode, in EventPluginLoader and HistoryEventService. - Tests: upgrade regression, dual-sink routing, plugin-failure isolation, load-failure policy, and updated back-pressure semantics. --- .../common/logsfilter/EventPluginConfig.java | 30 ++++ .../tron/core/config/args/EventConfig.java | 9 + common/src/main/resources/reference.conf | 2 + .../common/logsfilter/EventPluginLoader.java | 170 +++++++++++++----- .../java/org/tron/core/config/args/Args.java | 22 ++- .../services/event/HistoryEventService.java | 8 +- framework/src/main/resources/config.conf | 14 +- .../core/event/EventPluginLoaderTest.java | 152 ++++++++++++++-- 8 files changed, 340 insertions(+), 67 deletions(-) diff --git a/common/src/main/java/org/tron/common/logsfilter/EventPluginConfig.java b/common/src/main/java/org/tron/common/logsfilter/EventPluginConfig.java index e2aa0950d72..a9966ad339f 100644 --- a/common/src/main/java/org/tron/common/logsfilter/EventPluginConfig.java +++ b/common/src/main/java/org/tron/common/logsfilter/EventPluginConfig.java @@ -39,6 +39,18 @@ public class EventPluginConfig { @Setter private boolean useNativeQueue; + // Whether the event plugin should be activated. Resolved in Args from the plugin path + // and the runPluginWithNativeQueue opt-in, so the loader never has to infer activation + // from a possibly-stale path on its own. + @Getter + @Setter + private boolean useEventPlugin; + + // See EventConfig#pluginLoadFailurePolicy ("fail" | "ignore"). + @Getter + @Setter + private String pluginLoadFailurePolicy = "fail"; + @Getter @Setter private int bindPort; @@ -52,6 +64,24 @@ public class EventPluginConfig { @Setter private List triggerConfigList; + /** + * Decide whether the event plugin should be an active sink. + * + *

Backward-compatible by design: + *

+ */ + public static boolean resolveUseEventPlugin(boolean hasPluginPath, boolean useNativeQueue, + boolean runPluginWithNativeQueue) { + return hasPluginPath && (!useNativeQueue || runPluginWithNativeQueue); + } + public EventPluginConfig() { pluginPath = ""; serverAddress = ""; diff --git a/common/src/main/java/org/tron/core/config/args/EventConfig.java b/common/src/main/java/org/tron/core/config/args/EventConfig.java index f4378682cc3..b1d21d47c3a 100644 --- a/common/src/main/java/org/tron/core/config/args/EventConfig.java +++ b/common/src/main/java/org/tron/core/config/args/EventConfig.java @@ -25,6 +25,15 @@ public class EventConfig { private String server = ""; private String dbconfig = ""; private boolean contractParse = true; + // When the native queue is enabled, the event plugin is only activated in addition + // to it when this flag is explicitly set to true. This keeps existing native-queue + // deployments (which may still carry a stale "path") on native-only behavior after + // an upgrade. Ignored when useNativeQueue = false (plugin is the only sink then). + private boolean runPluginWithNativeQueue = false; + // What to do when the event plugin fails to load: + // "fail" - abort node startup (default, avoids silently losing plugin data) + // "ignore" - log the error and keep running with the native queue only + private String pluginLoadFailurePolicy = "fail"; // "native" is a Java reserved word; config key cannot match field name directly. // @Setter(NONE) prevents ConfigBeanFactory from requiring a "nativeQueue" key. @Setter(lombok.AccessLevel.NONE) diff --git a/common/src/main/resources/reference.conf b/common/src/main/resources/reference.conf index 25fc4832e55..9fdae6c7dc6 100644 --- a/common/src/main/resources/reference.conf +++ b/common/src/main/resources/reference.conf @@ -888,6 +888,8 @@ event.subscribe = { # dbname|username|password|2 (if collection exists, indexes must be created manually). dbconfig = "" contractParse = true # Whether to parse contract event data. + runPluginWithNativeQueue = false # Run the event plugin alongside the native queue. + pluginLoadFailurePolicy = "fail" # Plugin load failure: "fail" (abort) or "ignore". # Event trigger topics. topics = [ diff --git a/framework/src/main/java/org/tron/common/logsfilter/EventPluginLoader.java b/framework/src/main/java/org/tron/common/logsfilter/EventPluginLoader.java index c0b7afd6779..9356d825ffd 100644 --- a/framework/src/main/java/org/tron/common/logsfilter/EventPluginLoader.java +++ b/framework/src/main/java/org/tron/common/logsfilter/EventPluginLoader.java @@ -9,6 +9,7 @@ import java.util.List; import java.util.Objects; import java.util.Set; +import java.util.function.Consumer; import java.util.stream.Collectors; import lombok.Getter; import lombok.extern.slf4j.Slf4j; @@ -91,6 +92,17 @@ public class EventPluginLoader { @Getter private boolean useNativeQueue = false; + // Whether the event plugin is an active sink. May be turned off at runtime if the + // plugin fails to load under the "ignore" failure policy. + @Getter + private boolean useEventPlugin = false; + + private static final String FAIL_POLICY = "fail"; + + private static final String IGNORE_POLICY = "ignore"; + + private String pluginLoadFailurePolicy = FAIL_POLICY; + public static EventPluginLoader getInstance() { if (Objects.isNull(instance)) { synchronized (EventPluginLoader.class) { @@ -247,12 +259,54 @@ public boolean start(EventPluginConfig config) { this.triggerConfigList = config.getTriggerConfigList(); useNativeQueue = config.isUseNativeQueue(); + useEventPlugin = config.isUseEventPlugin(); + // Re-resolved every start() so no policy state leaks across runs; any unrecognized + // value defaults to the safe "fail" policy rather than silently becoming "ignore". + pluginLoadFailurePolicy = normalizeFailurePolicy(config.getPluginLoadFailurePolicy()); + + if (!useNativeQueue && !useEventPlugin) { + logger.error("No event sink is configured: enable the native queue " + + "or set a plugin path."); + return false; + } - if (config.isUseNativeQueue()) { - return launchNativeQueue(config); + // Start the plugin first so that, under the "fail" policy, a plugin problem is + // surfaced before the native queue starts accepting subscribers. + if (useEventPlugin && !launchEventPlugin(config)) { + if (!IGNORE_POLICY.equals(pluginLoadFailurePolicy)) { + return false; + } + // "ignore" policy: fall back to native-queue-only operation. + logger.warn("Event plugin failed to load; continuing without it " + + "(pluginLoadFailurePolicy = ignore)."); + useEventPlugin = false; + if (!useNativeQueue) { + return false; + } + } + + if (useNativeQueue && !launchNativeQueue(config)) { + // Native queue failed after the plugin already started; stop the plugin so a + // failed startup does not leak its threads/resources. + if (useEventPlugin) { + stopPlugin(); + } + return false; } - return launchEventPlugin(config); + return true; + } + + private String normalizeFailurePolicy(String policy) { + String p = Objects.isNull(policy) ? "" : policy.trim(); + if (IGNORE_POLICY.equalsIgnoreCase(p)) { + return IGNORE_POLICY; + } + if (!p.isEmpty() && !FAIL_POLICY.equalsIgnoreCase(p)) { + logger.warn("Unknown pluginLoadFailurePolicy '{}', falling back to '{}'.", + p, FAIL_POLICY); + } + return FAIL_POLICY; } private void setPluginConfig() { @@ -284,7 +338,7 @@ private void setSingleTriggerConfig(TriggerConfig triggerConfig) { blockLogTriggerSolidified = false; } - if (!useNativeQueue) { + if (useEventPlugin) { setPluginTopic(Trigger.BLOCK_TRIGGER, triggerConfig.getTopic()); } @@ -304,7 +358,7 @@ private void setSingleTriggerConfig(TriggerConfig triggerConfig) { transactionLogTriggerSolidified = false; } - if (!useNativeQueue) { + if (useEventPlugin) { setPluginTopic(Trigger.TRANSACTION_TRIGGER, triggerConfig.getTopic()); } @@ -316,7 +370,7 @@ private void setSingleTriggerConfig(TriggerConfig triggerConfig) { contractEventTriggerEnable = false; } - if (!useNativeQueue) { + if (useEventPlugin) { setPluginTopic(Trigger.CONTRACTEVENT_TRIGGER, triggerConfig.getTopic()); } @@ -332,7 +386,7 @@ private void setSingleTriggerConfig(TriggerConfig triggerConfig) { contractLogTriggerRedundancy = false; } - if (!useNativeQueue) { + if (useEventPlugin) { setPluginTopic(Trigger.CONTRACTLOG_TRIGGER, triggerConfig.getTopic()); } } else if (EventPluginConfig.SOLIDITY_TRIGGER_NAME @@ -342,7 +396,7 @@ private void setSingleTriggerConfig(TriggerConfig triggerConfig) { } else { solidityTriggerEnable = false; } - if (!useNativeQueue) { + if (useEventPlugin) { setPluginTopic(Trigger.SOLIDITY_TRIGGER, triggerConfig.getTopic()); } } else if (EventPluginConfig.SOLIDITY_EVENT_NAME @@ -353,7 +407,7 @@ private void setSingleTriggerConfig(TriggerConfig triggerConfig) { solidityEventTriggerEnable = false; } - if (!useNativeQueue) { + if (useEventPlugin) { setPluginTopic(Trigger.SOLIDITY_EVENT_TRIGGER, triggerConfig.getTopic()); } } else if (EventPluginConfig.SOLIDITY_LOG_NAME @@ -367,19 +421,38 @@ private void setSingleTriggerConfig(TriggerConfig triggerConfig) { solidityLogTriggerEnable = false; solidityLogTriggerRedundancy = false; } - if (!useNativeQueue) { + if (useEventPlugin) { setPluginTopic(Trigger.SOLIDITY_LOG_TRIGGER, triggerConfig.getTopic()); } } } + // Route a trigger to the plugin listeners, keeping a plugin fault from tearing down + // the node's event thread. Serialization is done once by the caller and shared with + // the native queue, so dual-sink mode adds no extra serialization on the hot path. + private void postToPlugin(Consumer handler) { + if (Objects.isNull(eventListeners)) { + return; + } + // Isolate per listener so one misbehaving plugin can neither block delivery to the + // others nor tear down the node's event thread. Errors (e.g. OutOfMemoryError) are + // intentionally not caught and are allowed to propagate. + for (IPluginEventListener listener : eventListeners) { + try { + handler.accept(listener); + } catch (Exception e) { + logger.error("event plugin listener failed to handle trigger", e); + } + } + } + public void postSolidityTrigger(SolidityTrigger trigger) { + String json = toJsonString(trigger); if (useNativeQueue) { - NativeMessageQueue.getInstance() - .publishTrigger(toJsonString(trigger), trigger.getTriggerName()); - } else { - eventListeners.forEach(listener -> - listener.handleSolidityTrigger(toJsonString(trigger))); + NativeMessageQueue.getInstance().publishTrigger(json, trigger.getTriggerName()); + } + if (useEventPlugin) { + postToPlugin(listener -> listener.handleSolidityTrigger(json)); } } @@ -440,6 +513,9 @@ public synchronized boolean isContractLogTriggerRedundancy() { } private void setPluginTopic(int eventType, String topic) { + if (Objects.isNull(eventListeners)) { + return; + } eventListeners.forEach(listener -> listener.setTopic(eventType, topic)); } @@ -514,66 +590,70 @@ public void stopPlugin() { } public void postBlockTrigger(BlockLogTrigger trigger) { + String json = toJsonString(trigger); if (useNativeQueue) { - NativeMessageQueue.getInstance() - .publishTrigger(toJsonString(trigger), trigger.getTriggerName()); - } else { - eventListeners.forEach(listener -> - listener.handleBlockEvent(toJsonString(trigger))); + NativeMessageQueue.getInstance().publishTrigger(json, trigger.getTriggerName()); + } + if (useEventPlugin) { + postToPlugin(listener -> listener.handleBlockEvent(json)); } } public void postSolidityLogTrigger(ContractLogTrigger trigger) { + String json = toJsonString(trigger); if (useNativeQueue) { - NativeMessageQueue.getInstance() - .publishTrigger(toJsonString(trigger), trigger.getTriggerName()); - } else { - eventListeners.forEach(listener -> - listener.handleSolidityLogTrigger(toJsonString(trigger))); + NativeMessageQueue.getInstance().publishTrigger(json, trigger.getTriggerName()); + } + if (useEventPlugin) { + postToPlugin(listener -> listener.handleSolidityLogTrigger(json)); } } public void postSolidityEventTrigger(ContractEventTrigger trigger) { + String json = toJsonString(trigger); if (useNativeQueue) { - NativeMessageQueue.getInstance() - .publishTrigger(toJsonString(trigger), trigger.getTriggerName()); - } else { - eventListeners.forEach(listener -> - listener.handleSolidityEventTrigger(toJsonString(trigger))); + NativeMessageQueue.getInstance().publishTrigger(json, trigger.getTriggerName()); + } + if (useEventPlugin) { + postToPlugin(listener -> listener.handleSolidityEventTrigger(json)); } } public void postTransactionTrigger(TransactionLogTrigger trigger) { + String json = toJsonString(trigger); if (useNativeQueue) { - NativeMessageQueue.getInstance() - .publishTrigger(toJsonString(trigger), trigger.getTriggerName()); - } else { - eventListeners.forEach(listener -> listener.handleTransactionTrigger(toJsonString(trigger))); + NativeMessageQueue.getInstance().publishTrigger(json, trigger.getTriggerName()); + } + if (useEventPlugin) { + postToPlugin(listener -> listener.handleTransactionTrigger(json)); } } public void postContractLogTrigger(ContractLogTrigger trigger) { + String json = toJsonString(trigger); if (useNativeQueue) { - NativeMessageQueue.getInstance() - .publishTrigger(toJsonString(trigger), trigger.getTriggerName()); - } else { - eventListeners.forEach(listener -> - listener.handleContractLogTrigger(toJsonString(trigger))); + NativeMessageQueue.getInstance().publishTrigger(json, trigger.getTriggerName()); + } + if (useEventPlugin) { + postToPlugin(listener -> listener.handleContractLogTrigger(json)); } } public void postContractEventTrigger(ContractEventTrigger trigger) { + String json = toJsonString(trigger); if (useNativeQueue) { - NativeMessageQueue.getInstance() - .publishTrigger(toJsonString(trigger), trigger.getTriggerName()); - } else { - eventListeners.forEach(listener -> - listener.handleContractEventTrigger(toJsonString(trigger))); + NativeMessageQueue.getInstance().publishTrigger(json, trigger.getTriggerName()); + } + if (useEventPlugin) { + postToPlugin(listener -> listener.handleContractEventTrigger(json)); } } public boolean isBusy() { - if (useNativeQueue) { + // Back-pressure guards the plugin's async pending queue, so it applies whenever the + // plugin is an active sink — including dual-sink mode. The native queue has no such + // queue of its own and never reports busy. + if (!useEventPlugin) { return false; } int queueSize = 0; diff --git a/framework/src/main/java/org/tron/core/config/args/Args.java b/framework/src/main/java/org/tron/core/config/args/Args.java index 0bca242606e..2d4ba6de278 100644 --- a/framework/src/main/java/org/tron/core/config/args/Args.java +++ b/framework/src/main/java/org/tron/core/config/args/Args.java @@ -370,10 +370,13 @@ private static void applyEventConfig(EventConfig ec) { epc.setBindPort(nq.getBindport()); epc.setSendQueueLength(nq.getSendqueuelength()); - if (!nq.isUseNativeQueue()) { - if (StringUtils.isNotEmpty(ec.getPath())) { - epc.setPluginPath(ec.getPath().trim()); - } + // Plugin connection settings are copied only when a plugin path is configured, so the + // plugin can run either on its own (native queue off) or alongside the native queue. + // Without a path the plugin cannot be activated, so server/dbconfig are left unset to + // avoid leaving misleading connection settings on EventPluginConfig. + boolean hasPluginPath = StringUtils.isNotEmpty(ec.getPath()); + if (hasPluginPath) { + epc.setPluginPath(ec.getPath().trim()); if (StringUtils.isNotEmpty(ec.getServer())) { epc.setServerAddress(ec.getServer().trim()); } @@ -382,6 +385,17 @@ private static void applyEventConfig(EventConfig ec) { } } + // Resolve plugin activation explicitly (never inferred later from the raw path): + // - native queue OFF: the plugin is the only sink, so activate it whenever a path + // is set (unchanged legacy behavior). + // - native queue ON : the plugin runs alongside the queue ONLY when the operator + // opts in via runPluginWithNativeQueue. A leftover "path" alone never activates + // it, so upgrading a native-queue node cannot suddenly start loading a plugin. + boolean useEventPlugin = EventPluginConfig.resolveUseEventPlugin( + hasPluginPath, nq.isUseNativeQueue(), ec.isRunPluginWithNativeQueue()); + epc.setUseEventPlugin(useEventPlugin); + epc.setPluginLoadFailurePolicy(ec.getPluginLoadFailurePolicy()); + // topics List triggerConfigs = new ArrayList<>(); for (EventConfig.TopicConfig tc : ec.getTopics()) { diff --git a/framework/src/main/java/org/tron/core/services/event/HistoryEventService.java b/framework/src/main/java/org/tron/core/services/event/HistoryEventService.java index 2eccb9fa2a9..f5685133f44 100644 --- a/framework/src/main/java/org/tron/core/services/event/HistoryEventService.java +++ b/framework/src/main/java/org/tron/core/services/event/HistoryEventService.java @@ -67,11 +67,13 @@ private void syncEvent() { if (thread.isInterrupted() || isClosed) { throw new InterruptedException(); } - if (instance.isUseNativeQueue()) { - Thread.sleep(20); - } else if (instance.isBusy()) { + // Respect the plugin's back-pressure first so it holds even in dual-sink mode + // (native queue + plugin), then apply the native queue's own pacing throttle. + if (instance.isBusy()) { Thread.sleep(100); continue; + } else if (instance.isUseNativeQueue()) { + Thread.sleep(20); } BlockEvent blockEvent = blockEventGet.getBlockEvent(tmp); realtimeEventService.flush(blockEvent, false); diff --git a/framework/src/main/resources/config.conf b/framework/src/main/resources/config.conf index 1176dd46311..bd0e529a610 100644 --- a/framework/src/main/resources/config.conf +++ b/framework/src/main/resources/config.conf @@ -441,10 +441,22 @@ committee = { event.subscribe = { enable = false // enable event subscribe, replaces deprecated CLI flag --es native = { - useNativeQueue = true // if true, use native message queue, else use event plugin. + useNativeQueue = true // if true, use the native message queue. By default this is + // exclusive: the event plugin is not loaded while it is on. + // To run both the native queue and the plugin at the same time, + // set runPluginWithNativeQueue = true below. bindport = 5555 // bind port sendqueuelength = 1000 //max length of send queue } + // Run the event plugin alongside the native queue (dual-sink mode). Only takes effect + // when useNativeQueue = true AND a plugin "path" is configured. Defaults to false so + // existing native-queue nodes keep their current behavior after an upgrade even if a + // stale "path" is still present in the config. + runPluginWithNativeQueue = false + // Behavior when the event plugin fails to load: + // "fail" - abort node startup (default; avoids silently dropping plugin data) + // "ignore" - log the error and keep running with the native queue only + pluginLoadFailurePolicy = "fail" version = 0 # Specify the starting block number to sync historical events. Only applicable when version = 1. # After performing a full event sync, set this value to 0 or a negative number. diff --git a/framework/src/test/java/org/tron/core/event/EventPluginLoaderTest.java b/framework/src/test/java/org/tron/core/event/EventPluginLoaderTest.java index 658d42f38d9..0a9e4cc4180 100644 --- a/framework/src/test/java/org/tron/core/event/EventPluginLoaderTest.java +++ b/framework/src/test/java/org/tron/core/event/EventPluginLoaderTest.java @@ -4,40 +4,164 @@ import java.util.ArrayList; import java.util.List; +import org.junit.After; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; +import org.tron.common.logsfilter.EventPluginConfig; import org.tron.common.logsfilter.EventPluginLoader; import org.tron.common.logsfilter.IPluginEventListener; +import org.tron.common.logsfilter.trigger.BlockLogTrigger; import org.tron.common.utils.ReflectUtils; public class EventPluginLoaderTest { + // EventPluginLoader is a process-wide singleton and these tests mutate its private + // fields via reflection. Reset them after each test so state cannot leak into other + // test classes that also use EventPluginLoader.getInstance(). + @After + public void resetLoader() { + EventPluginLoader loader = EventPluginLoader.getInstance(); + ReflectUtils.setFieldValue(loader, "useNativeQueue", false); + ReflectUtils.setFieldValue(loader, "useEventPlugin", false); + ReflectUtils.setFieldValue(loader, "eventListeners", null); + ReflectUtils.setFieldValue(loader, "pluginLoadFailurePolicy", "fail"); + } + + private static IPluginEventListener mockListener(EventPluginLoader loader) { + IPluginEventListener listener = mock(IPluginEventListener.class); + List list = new ArrayList<>(); + list.add(listener); + ReflectUtils.setFieldValue(loader, "eventListeners", list); + return listener; + } + @Test public void testIsBusy() { EventPluginLoader eventPluginLoader = EventPluginLoader.getInstance(); - ReflectUtils.setFieldValue(eventPluginLoader, "useNativeQueue", true); - boolean flag = eventPluginLoader.isBusy(); - Assert.assertFalse(flag); - ReflectUtils.setFieldValue(eventPluginLoader, "useNativeQueue", false); + // Back-pressure is keyed on the plugin being active, not on the native queue. + // Native-queue-only (plugin off) never reports busy. + ReflectUtils.setFieldValue(eventPluginLoader, "useNativeQueue", true); + ReflectUtils.setFieldValue(eventPluginLoader, "useEventPlugin", false); + Assert.assertFalse(eventPluginLoader.isBusy()); - IPluginEventListener p1 = mock(IPluginEventListener.class); - List list = new ArrayList<>(); - list.add(p1); - ReflectUtils.setFieldValue(eventPluginLoader, "eventListeners", list); + // Plugin active (whether standalone or dual-sink) does apply back-pressure. + ReflectUtils.setFieldValue(eventPluginLoader, "useEventPlugin", true); + IPluginEventListener p1 = mockListener(eventPluginLoader); Mockito.when(p1.getPendingSize()).thenReturn(100); - flag = eventPluginLoader.isBusy(); - Assert.assertFalse(flag); + Assert.assertFalse(eventPluginLoader.isBusy()); Mockito.when(p1.getPendingSize()).thenReturn(60000); - flag = eventPluginLoader.isBusy(); - Assert.assertTrue(flag); + Assert.assertTrue(eventPluginLoader.isBusy()); + + // Dual-sink mode: native queue on AND plugin on -> back-pressure still honored. + ReflectUtils.setFieldValue(eventPluginLoader, "useNativeQueue", true); + Assert.assertTrue(eventPluginLoader.isBusy()); Mockito.when(p1.getPendingSize()).thenThrow(new AbstractMethodError()); - flag = eventPluginLoader.isBusy(); - Assert.assertFalse(flag); + Assert.assertFalse(eventPluginLoader.isBusy()); + } + + /** + * Regression for the upgrade scenario raised in review: a native-queue node that still + * carries a stale plugin "path" must NOT start loading the plugin after an upgrade. + * The plugin only joins when the operator explicitly opts in. + */ + @Test + public void testResolveUseEventPlugin() { + // native queue OFF -> plugin is the only sink, active whenever a path is present. + Assert.assertTrue(EventPluginConfig.resolveUseEventPlugin(true, false, false)); + Assert.assertFalse(EventPluginConfig.resolveUseEventPlugin(false, false, false)); + + // native queue ON + stale path + no opt-in -> plugin stays OFF (the key regression). + Assert.assertFalse(EventPluginConfig.resolveUseEventPlugin(true, true, false)); + + // native queue ON + path + explicit opt-in -> dual-sink mode. + Assert.assertTrue(EventPluginConfig.resolveUseEventPlugin(true, true, true)); + + // opt-in without a path is meaningless -> plugin stays OFF. + Assert.assertFalse(EventPluginConfig.resolveUseEventPlugin(false, true, true)); + } + + @Test + public void testStartRequiresASink() { + EventPluginLoader loader = EventPluginLoader.getInstance(); + EventPluginConfig config = new EventPluginConfig(); + config.setUseNativeQueue(false); + config.setUseEventPlugin(false); + Assert.assertFalse(loader.start(config)); + } + + @Test + public void testPostRoutesToPluginOnlyWhenActive() { + EventPluginLoader loader = EventPluginLoader.getInstance(); + IPluginEventListener listener = mockListener(loader); + + // Plugin inactive -> listener is never called. + ReflectUtils.setFieldValue(loader, "useEventPlugin", false); + ReflectUtils.setFieldValue(loader, "useNativeQueue", true); + loader.postBlockTrigger(new BlockLogTrigger()); + Mockito.verify(listener, Mockito.never()).handleBlockEvent(Mockito.anyString()); + + // Dual-sink: plugin active alongside native queue -> listener receives the trigger. + ReflectUtils.setFieldValue(loader, "useEventPlugin", true); + loader.postBlockTrigger(new BlockLogTrigger()); + Mockito.verify(listener, Mockito.times(1)).handleBlockEvent(Mockito.anyString()); + } + + @Test + public void testPluginFailureIsIsolated() { + EventPluginLoader loader = EventPluginLoader.getInstance(); + IPluginEventListener bad = mock(IPluginEventListener.class); + IPluginEventListener good = mock(IPluginEventListener.class); + List listeners = new ArrayList<>(); + listeners.add(bad); + listeners.add(good); + ReflectUtils.setFieldValue(loader, "eventListeners", listeners); + ReflectUtils.setFieldValue(loader, "useEventPlugin", true); + ReflectUtils.setFieldValue(loader, "useNativeQueue", false); + + Mockito.doThrow(new RuntimeException("boom")) + .when(bad).handleBlockEvent(Mockito.anyString()); + + // A misbehaving plugin must not propagate out of the node's event thread, and it + // must not prevent delivery to the other listeners. + loader.postBlockTrigger(new BlockLogTrigger()); + Mockito.verify(good, Mockito.times(1)).handleBlockEvent(Mockito.anyString()); + } + + /** + * With the "ignore" failure policy, a plugin that cannot load must not abort startup; + * the node falls back to native-queue-only and clears the plugin flag. + */ + @Test + public void testLoadFailurePolicyIgnoreDisablesPlugin() { + EventPluginLoader loader = EventPluginLoader.getInstance(); + ReflectUtils.setFieldValue(loader, "eventListeners", null); + + EventPluginConfig config = new EventPluginConfig(); + config.setUseNativeQueue(false); + config.setUseEventPlugin(true); + config.setPluginPath("/non/existent/plugin/path.zip"); + config.setPluginLoadFailurePolicy("ignore"); + // No native queue left as a fallback -> start still fails (nothing to run). + Assert.assertFalse(loader.start(config)); + Assert.assertFalse(loader.isUseEventPlugin()); + + // "fail" policy on a bad path -> startup aborts. + config.setPluginLoadFailurePolicy("fail"); + config.setUseEventPlugin(true); + Assert.assertFalse(loader.start(config)); + + // An unrecognized policy value must not silently behave as "ignore": it is + // normalized to the safe "fail" default, and re-resolved on every start(). + config.setPluginLoadFailurePolicy("typo"); + config.setUseEventPlugin(true); + Assert.assertFalse(loader.start(config)); + Assert.assertEquals("fail", + ReflectUtils.getFieldValue(loader, "pluginLoadFailurePolicy")); } }