Browse Source

Add direct self-reference loop detection to TbRuleChainInputNode

The stack-based check in DefaultTbContext.input() only catches a loop
on the second Kafka round-trip through the same rule chain. When the
target rule chain equals the node's own rule chain (direct self-loop),
the first iteration is always unnecessary and should be rejected
immediately before any enqueue occurs.

Add an eager check in TbRuleChainInputNode.onMsg(): if the resolved
targetRuleChainId equals ctx.getSelf().getRuleChainId() (the rule chain
that contains this node), tellFailure() is called right away.

Together with the existing stack-based check in DefaultTbContext.input()
this forms a two-layer defence:
- Layer 1 (TbRuleChainInputNode): direct loops A→A, caught immediately
- Layer 2 (DefaultTbContext): indirect cycles A→B→A, caught on second pass

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
pull/15102/head
Sergey Matvienko 8 months ago
parent
commit
6615c45cb5
  1. 7
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNode.java
  2. 70
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNodeTest.java

7
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNode.java

@ -76,6 +76,13 @@ public class TbRuleChainInputNode implements TbNode {
public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException { public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException {
RuleChainId targetRuleChainId = forwardMsgToDefaultRuleChain ? RuleChainId targetRuleChainId = forwardMsgToDefaultRuleChain ?
getOriginatorDefaultRuleChainId(ctx, msg).orElse(ruleChainId) : ruleChainId; getOriginatorDefaultRuleChainId(ctx, msg).orElse(ruleChainId) : ruleChainId;
if (targetRuleChainId.equals(ctx.getSelf().getRuleChainId())) {
ctx.tellFailure(msg, new RuntimeException(
"TbRuleChainInputNode in rule chain [" + targetRuleChainId +
"] is configured to forward messages back to the same rule chain it belongs to. " +
"This would cause an infinite loop. Please check the rule chain configuration."));
return;
}
ctx.input(msg, targetRuleChainId); ctx.input(msg, targetRuleChainId);
} }

70
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNodeTest.java

@ -41,8 +41,10 @@ import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.RuleNodeId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.data.rule.RuleNode;
import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.common.msg.TbMsgMetaData;
@ -52,8 +54,9 @@ import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.any; import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy; import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when; import static org.mockito.Mockito.when;
@ -64,6 +67,8 @@ public class TbRuleChainInputNodeTest extends AbstractRuleNodeUpgradeTest {
private final TenantId TENANT_ID = new TenantId(UUID.fromString("4ba69ea5-6b27-42df-ab66-e7a727a67027")); private final TenantId TENANT_ID = new TenantId(UUID.fromString("4ba69ea5-6b27-42df-ab66-e7a727a67027"));
private final DeviceId DEVICE_ID = new DeviceId(UUID.fromString("97731954-2147-4176-8f1a-d14f1b73e4e6")); private final DeviceId DEVICE_ID = new DeviceId(UUID.fromString("97731954-2147-4176-8f1a-d14f1b73e4e6"));
private final AssetId ASSET_ID = new AssetId(UUID.fromString("841a47bd-4e8e-4ea5-88e6-420da0d70e51")); private final AssetId ASSET_ID = new AssetId(UUID.fromString("841a47bd-4e8e-4ea5-88e6-420da0d70e51"));
// A stable "current" rule chain ID: this node lives here. Must differ from any target used in tests.
private final RuleChainId CURRENT_RULE_CHAIN_ID = new RuleChainId(UUID.fromString("11111111-1111-1111-1111-111111111111"));
private TbRuleChainInputNode node; private TbRuleChainInputNode node;
private TbRuleChainInputNodeConfiguration config; private TbRuleChainInputNodeConfiguration config;
@ -149,6 +154,7 @@ public class TbRuleChainInputNodeTest extends AbstractRuleNodeUpgradeTest {
when(ctxMock.getTenantId()).thenReturn(TENANT_ID); when(ctxMock.getTenantId()).thenReturn(TENANT_ID);
when(ctxMock.getDeviceProfileCache()).thenReturn(deviceProfileCacheMock); when(ctxMock.getDeviceProfileCache()).thenReturn(deviceProfileCacheMock);
when(deviceProfileCacheMock.get(any(TenantId.class), any(DeviceId.class))).thenReturn(deviceProfile); when(deviceProfileCacheMock.get(any(TenantId.class), any(DeviceId.class))).thenReturn(deviceProfile);
when(ctxMock.getSelf()).thenReturn(ruleNodeInCurrentChain());
node.init(ctxMock, nodeConfiguration); node.init(ctxMock, nodeConfiguration);
@ -182,6 +188,7 @@ public class TbRuleChainInputNodeTest extends AbstractRuleNodeUpgradeTest {
when(ctxMock.getTenantId()).thenReturn(TENANT_ID); when(ctxMock.getTenantId()).thenReturn(TENANT_ID);
when(ctxMock.getAssetProfileCache()).thenReturn(assetProfileCacheMock); when(ctxMock.getAssetProfileCache()).thenReturn(assetProfileCacheMock);
when(assetProfileCacheMock.get(any(TenantId.class), any(AssetId.class))).thenReturn(assetProfile); when(assetProfileCacheMock.get(any(TenantId.class), any(AssetId.class))).thenReturn(assetProfile);
when(ctxMock.getSelf()).thenReturn(ruleNodeInCurrentChain());
node.init(ctxMock, nodeConfiguration); node.init(ctxMock, nodeConfiguration);
@ -214,6 +221,7 @@ public class TbRuleChainInputNodeTest extends AbstractRuleNodeUpgradeTest {
when(ctxMock.getTenantId()).thenReturn(TENANT_ID); when(ctxMock.getTenantId()).thenReturn(TENANT_ID);
when(ctxMock.getDeviceProfileCache()).thenReturn(deviceProfileCacheMock); when(ctxMock.getDeviceProfileCache()).thenReturn(deviceProfileCacheMock);
when(deviceProfileCacheMock.get(any(TenantId.class), any(DeviceId.class))).thenReturn(deviceProfile); when(deviceProfileCacheMock.get(any(TenantId.class), any(DeviceId.class))).thenReturn(deviceProfile);
when(ctxMock.getSelf()).thenReturn(ruleNodeInCurrentChain());
node.init(ctxMock, nodeConfiguration); node.init(ctxMock, nodeConfiguration);
@ -242,6 +250,7 @@ public class TbRuleChainInputNodeTest extends AbstractRuleNodeUpgradeTest {
when(ctxMock.getTenantId()).thenReturn(TENANT_ID); when(ctxMock.getTenantId()).thenReturn(TENANT_ID);
when(ctxMock.getAssetProfileCache()).thenReturn(assetProfileCacheMock); when(ctxMock.getAssetProfileCache()).thenReturn(assetProfileCacheMock);
when(assetProfileCacheMock.get(any(TenantId.class), any(AssetId.class))).thenReturn(assetProfile); when(assetProfileCacheMock.get(any(TenantId.class), any(AssetId.class))).thenReturn(assetProfile);
when(ctxMock.getSelf()).thenReturn(ruleNodeInCurrentChain());
node.init(ctxMock, nodeConfiguration); node.init(ctxMock, nodeConfiguration);
@ -266,6 +275,8 @@ public class TbRuleChainInputNodeTest extends AbstractRuleNodeUpgradeTest {
config.setForwardMsgToDefaultRuleChain(false); config.setForwardMsgToDefaultRuleChain(false);
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
when(ctxMock.getSelf()).thenReturn(ruleNodeInCurrentChain());
node.init(ctxMock, nodeConfiguration); node.init(ctxMock, nodeConfiguration);
//WHEN //WHEN
@ -277,6 +288,56 @@ public class TbRuleChainInputNodeTest extends AbstractRuleNodeUpgradeTest {
assertThat(ruleChainArgumentCaptor.getValue()).isEqualTo(ruleChainIdFromConfig); assertThat(ruleChainArgumentCaptor.getValue()).isEqualTo(ruleChainIdFromConfig);
} }
@Test
public void givenForwardMsgToDefaultIsTrueAndDeviceDefaultRuleChainIsSameAsCurrent_whenOnMsg_thenTellFailureToPreventDirectLoop() throws TbNodeException {
// GIVEN - device's default rule chain == the rule chain that contains this node (self-loop)
// This is caught immediately, before any Kafka round-trip
DeviceProfile deviceProfile = new DeviceProfile();
deviceProfile.setDefaultRuleChainId(CURRENT_RULE_CHAIN_ID);
TbMsg msg = getMsg(DEVICE_ID);
String fallbackRuleChainIdStr = "acbc924f-7f95-4a9b-a854-e4822deb74c7";
config.setRuleChainId(fallbackRuleChainIdStr);
config.setForwardMsgToDefaultRuleChain(true);
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
when(ctxMock.getTenantId()).thenReturn(TENANT_ID);
when(ctxMock.getDeviceProfileCache()).thenReturn(deviceProfileCacheMock);
when(deviceProfileCacheMock.get(any(TenantId.class), any(DeviceId.class))).thenReturn(deviceProfile);
when(ctxMock.getSelf()).thenReturn(ruleNodeInCurrentChain());
node.init(ctxMock, nodeConfiguration);
// WHEN
node.onMsg(ctxMock, msg);
// THEN - tellFailure immediately; ctx.input() must never be called
verify(ctxMock).tellFailure(eq(msg), any(RuntimeException.class));
verify(ctxMock, never()).input(any(), any());
}
@Test
public void givenConfiguredRuleChainIsSameAsCurrent_whenOnMsg_thenTellFailureToPreventDirectLoop() throws TbNodeException {
// GIVEN - static config points to the same rule chain that contains this node
config.setRuleChainId(CURRENT_RULE_CHAIN_ID.getId().toString());
config.setForwardMsgToDefaultRuleChain(false);
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
TbMsg msg = getMsg(DEVICE_ID);
when(ctxMock.getSelf()).thenReturn(ruleNodeInCurrentChain());
node.init(ctxMock, nodeConfiguration);
// WHEN
node.onMsg(ctxMock, msg);
// THEN - tellFailure immediately; ctx.input() must never be called
verify(ctxMock).tellFailure(eq(msg), any(RuntimeException.class));
verify(ctxMock, never()).input(any(), any());
}
private static Stream<Arguments> givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() { private static Stream<Arguments> givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() {
return Stream.of( return Stream.of(
//config for version 0 //config for version 0
@ -299,6 +360,13 @@ public class TbRuleChainInputNodeTest extends AbstractRuleNodeUpgradeTest {
return node; return node;
} }
/** Creates a RuleNode that lives in {@link #CURRENT_RULE_CHAIN_ID}. */
private RuleNode ruleNodeInCurrentChain() {
RuleNode ruleNode = new RuleNode(new RuleNodeId(UUID.fromString("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")));
ruleNode.setRuleChainId(CURRENT_RULE_CHAIN_ID);
return ruleNode;
}
private TbMsg getMsg(EntityId entityId) { private TbMsg getMsg(EntityId entityId) {
return TbMsg.newMsg() return TbMsg.newMsg()
.type(TbMsgType.POST_TELEMETRY_REQUEST) .type(TbMsgType.POST_TELEMETRY_REQUEST)

Loading…
Cancel
Save