Skip to content

feat(event): opt-in dual-sink for native queue and event plugin #6878

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
perfch wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

feat(event): opt-in dual-sink for native queue and event plugin #6878

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 30 additions & 0 deletions common/src/main/java/org/tron/common/logsfilter/EventPluginConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -52,6 +64,24 @@ public class EventPluginConfig {
@Setter
private List<TriggerConfig> triggerConfigList;

/**
* Decide whether the event plugin should be an active sink.
*
* <p>Backward-compatible by design:
* <ul>
* <li>native queue OFF: the plugin is the only sink, so it is active whenever a
* plugin path is configured (unchanged legacy behavior);</li>
* <li>native queue ON: the plugin runs alongside the queue only when the operator
* opts in via {@code runPluginWithNativeQueue}. A leftover path alone never
* activates it, so upgrading a native-queue node cannot suddenly load a plugin.
* </li>
* </ul>
*/
public static boolean resolveUseEventPlugin(boolean hasPluginPath, boolean useNativeQueue,
boolean runPluginWithNativeQueue) {
return hasPluginPath && (!useNativeQueue || runPluginWithNativeQueue);
}

public EventPluginConfig() {
pluginPath = "";
serverAddress = "";
Expand Down
9 changes: 9 additions & 0 deletions common/src/main/java/org/tron/core/config/args/EventConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
2 changes: 2 additions & 0 deletions common/src/main/resources/reference.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand Down
170 changes: 125 additions & 45 deletions framework/src/main/java/org/tron/common/logsfilter/EventPluginLoader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -284,7 +338,7 @@ private void setSingleTriggerConfig(TriggerConfig triggerConfig) {
blockLogTriggerSolidified = false;
}

if (!useNativeQueue) {
if (useEventPlugin) {
setPluginTopic(Trigger.BLOCK_TRIGGER, triggerConfig.getTopic());
}

Expand All @@ -304,7 +358,7 @@ private void setSingleTriggerConfig(TriggerConfig triggerConfig) {
transactionLogTriggerSolidified = false;
}

if (!useNativeQueue) {
if (useEventPlugin) {
setPluginTopic(Trigger.TRANSACTION_TRIGGER, triggerConfig.getTopic());
}

Expand All @@ -316,7 +370,7 @@ private void setSingleTriggerConfig(TriggerConfig triggerConfig) {
contractEventTriggerEnable = false;
}

if (!useNativeQueue) {
if (useEventPlugin) {
setPluginTopic(Trigger.CONTRACTEVENT_TRIGGER, triggerConfig.getTopic());
}

Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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<IPluginEventListener> 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));
}
}

Expand Down Expand Up @@ -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));
}

Expand Down Expand Up @@ -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;
Expand Down
22 changes: 18 additions & 4 deletions framework/src/main/java/org/tron/core/config/args/Args.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}
Expand All @@ -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<TriggerConfig> triggerConfigs = new ArrayList<>();
for (EventConfig.TopicConfig tc : ec.getTopics()) {
Expand Down
Loading
Loading