tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge 7c199d5345 into be3207ab65

pull/15658/merge
Oleksandra Matviienko 2 days ago
committed by GitHub
parent
be3207ab65 7c199d5345
commit
b748e13ce8
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
7 changed files with 257 additions and 42 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
  2. 17
      application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
  3. 8
      application/src/main/resources/thingsboard.yml
  4. 215
      application/src/test/java/org/thingsboard/server/actors/rule/DefaultTbContextTest.java
  5. 4
      common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
  6. 9
      common/message/src/main/java/org/thingsboard/server/common/msg/TbMsgProcessingCtx.java
  7. 42
      common/message/src/test/java/org/thingsboard/server/common/msg/TbMsgProcessingCtxTest.java

4
application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
View File

@ -587,6 +587,10 @@ public class ActorSystemContext {
@Getter
private long ruleChainErrorPersistFrequency;
@Value("${actors.rule.chain.input_loop_max_visits:1}")
@Getter
private int ruleChainInputLoopMaxVisits;
@Value("${actors.rule.node.error_persist_frequency:3000}")
@Getter
private long ruleNodeErrorPersistFrequency;

17
application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
View File

@ -186,15 +186,14 @@ public class DefaultTbContext implements TbContext {
}
RuleChainId selfRuleChainId = nodeCtx.getSelf().getRuleChainId();
RuleNodeId selfId = nodeCtx.getSelf().getId();
if (msg.isAlreadyInStack(selfRuleChainId, selfId)) {
log.warn("[{}] Detected rule chain processing loop for rule node [{}] in rule chain [{}]. " +
"The message will be failed to prevent infinite loop. " +
"Please check the rule chain configuration for circular references.",
nodeCtx.getTenantId(), selfId, selfRuleChainId);
tellFailure(msg, new RuntimeException(
"Detected rule chain processing loop for rule node [" + selfId + "] " +
"in rule chain [" + selfRuleChainId + "]. " +
"Please check the rule chain configuration for circular references."));
int maxVisits = Math.max(1, mainCtx.getRuleChainInputLoopMaxVisits());
if (msg.countOccurrences(selfRuleChainId, selfId) >= maxVisits) {
String reason = "Rule chain input node visit limit " + maxVisits + " reached for rule node ["
+ selfId + "] in rule chain [" + selfRuleChainId + "]. "
+ "If the loop is intentional, raise actors.rule.chain.input_loop_max_visits "
+ "(env TB_RULE_CHAIN_INPUT_LOOP_MAX_VISITS).";
log.warn("[{}] {}", nodeCtx.getTenantId(), reason);
tellFailure(msg, new RuntimeException(reason));
return;
}
TbMsg tbMsg = msg.copy()

8
application/src/main/resources/thingsboard.yml
View File

@ -549,6 +549,14 @@ actors:
chain:
# Errors for particular actors are persisted once per specified amount of milliseconds
error_persist_frequency: "${ACTORS_RULE_CHAIN_ERROR_FREQUENCY:3000}"
# Maximum number of times a single rule chain input node may fire for one message.
# Default 1 = the node fires at most once per message — any revisit (direct A->A or
# indirect A->B->...->A) is failed immediately to prevent infinite loops.
# Raise to N > 1 to allow legacy designs that intentionally loop through the same input
# node up to N times (e.g. paginated external API calls). Values below 1 are clamped to 1.
# Direct self-references (a rule chain input node configured to forward to its own rule
# chain) are always rejected by TbRuleChainInputNode regardless of this setting.
input_loop_max_visits: "${TB_RULE_CHAIN_INPUT_LOOP_MAX_VISITS:1}"
debug_mode_rate_limits_per_tenant:
# Enable/Disable the rate limit of persisted debug events for all rule nodes per tenant
enabled: "${ACTORS_RULE_CHAIN_DEBUG_MODE_RATE_LIMITS_PER_TENANT_ENABLED:true}"

215
application/src/test/java/org/thingsboard/server/actors/rule/DefaultTbContextTest.java
View File

@ -136,10 +136,11 @@ class DefaultTbContextTest {
}
@Test
public void givenMsgWithCurrentNodeAlreadyInStack_whenInput_thenTellFailureToPreventLoop() {
// GIVEN - simulate the second iteration: current node is already in the return stack,
// which happens when forwardMsgToDefaultRuleChain=true and the device's default rule chain
// is the same as the rule chain containing this node
public void givenDefaultCapAndCurrentNodeAlreadyInStack_whenInput_thenTellFailure() {
// GIVEN - default cap = 1: any revisit of the same (chain, node) within one message fails.
// Simulates the second visit of an input node whose first push is still in the stack.
given(mainCtxMock.getRuleChainInputLoopMaxVisits()).willReturn(1);
var callbackMock = mock(TbMsgCallback.class);
given(callbackMock.isMsgValid()).willReturn(true);
@ -157,7 +158,6 @@ class DefaultTbContextTest {
.data(TbMsg.EMPTY_STRING)
.callback(callbackMock)
.build();
// Push current node into the stack - simulates a previous iteration that already forwarded to this rule chain
msg.pushToStack(RULE_CHAIN_ID, RULE_NODE_ID);
var targetRuleChainId = new RuleChainId(UUID.randomUUID());
@ -165,16 +165,215 @@ class DefaultTbContextTest {
// WHEN
defaultTbContext.input(msg, targetRuleChainId);
// THEN - loop detected: tellFailure is called and no message is enqueued
// THEN - failure with the unified visit-limit message that names the property
ArgumentCaptor<TbActorMsg> tellCaptor = ArgumentCaptor.forClass(TbActorMsg.class);
then(chainActorMock).should().tell(tellCaptor.capture());
TbActorMsg capturedTellMsg = tellCaptor.getValue();
assertThat(capturedTellMsg).isInstanceOf(RuleNodeToRuleChainTellNextMsg.class);
RuleNodeToRuleChainTellNextMsg failureMsg = (RuleNodeToRuleChainTellNextMsg) capturedTellMsg;
assertThat(failureMsg.getRelationTypes()).containsOnly(TbNodeConnectionType.FAILURE);
assertThat(failureMsg.getFailureMessage()).containsIgnoringCase("loop");
assertThat(failureMsg.getFailureMessage()).contains("visit limit 1 reached");
assertThat(failureMsg.getFailureMessage()).contains("TB_RULE_CHAIN_INPUT_LOOP_MAX_VISITS");
assertThat(failureMsg.getFailureMessage()).contains("actors.rule.chain.input_loop_max_visits");
then(mainCtxMock).should().getRuleChainInputLoopMaxVisits();
then(mainCtxMock).shouldHaveNoMoreInteractions();
}
@Test
public void givenVisitsBelowCap_whenInput_thenEnqueue() {
// GIVEN - admin raised cap to 5; this is the first visit of the input node for the message
given(mainCtxMock.getRuleChainInputLoopMaxVisits()).willReturn(5);
var tpi = resolve(null);
given(mainCtxMock.resolve(eq(ServiceType.TB_RULE_ENGINE), nullable(String.class), eq(TENANT_ID), eq(TENANT_ID))).willReturn(tpi);
var clusterService = mock(TbClusterService.class);
given(mainCtxMock.getClusterService()).willReturn(clusterService);
var callbackMock = mock(TbMsgCallback.class);
given(callbackMock.isMsgValid()).willReturn(true);
var ruleNode = new RuleNode(RULE_NODE_ID);
ruleNode.setRuleChainId(RULE_CHAIN_ID);
ruleNode.setDebugSettings(DebugSettings.off());
given(nodeCtxMock.getTenantId()).willReturn(TENANT_ID);
given(nodeCtxMock.getSelf()).willReturn(ruleNode);
var msg = TbMsg.newMsg()
.type(TbMsgType.POST_TELEMETRY_REQUEST)
.originator(TENANT_ID)
.copyMetaData(TbMsgMetaData.EMPTY)
.data(TbMsg.EMPTY_STRING)
.callback(callbackMock)
.build();
var targetRuleChainId = new RuleChainId(UUID.randomUUID());
then(mainCtxMock).shouldHaveNoInteractions(); // no message was enqueued
// WHEN
defaultTbContext.input(msg, targetRuleChainId);
// THEN
then(clusterService).should().pushMsgToRuleEngine(eq(tpi), eq(msg.getId()), any(), any());
then(chainActorMock).shouldHaveNoInteractions();
}
@Test
public void givenVisitsAtCap_whenInput_thenTellFailure() {
// GIVEN - cap=2, this input node has already been pushed twice for this message
given(mainCtxMock.getRuleChainInputLoopMaxVisits()).willReturn(2);
var callbackMock = mock(TbMsgCallback.class);
given(callbackMock.isMsgValid()).willReturn(true);
var ruleNode = new RuleNode(RULE_NODE_ID);
ruleNode.setRuleChainId(RULE_CHAIN_ID);
ruleNode.setDebugSettings(DebugSettings.off());
given(nodeCtxMock.getSelf()).willReturn(ruleNode);
given(nodeCtxMock.getTenantId()).willReturn(TENANT_ID);
given(nodeCtxMock.getChainActor()).willReturn(chainActorMock);
var msg = TbMsg.newMsg()
.type(TbMsgType.POST_TELEMETRY_REQUEST)
.originator(TENANT_ID)
.copyMetaData(TbMsgMetaData.EMPTY)
.data(TbMsg.EMPTY_STRING)
.callback(callbackMock)
.build();
// two prior visits of THIS node + an unrelated entry that must NOT contribute to the count
msg.pushToStack(RULE_CHAIN_ID, RULE_NODE_ID);
msg.pushToStack(new RuleChainId(UUID.randomUUID()), new RuleNodeId(UUID.randomUUID()));
msg.pushToStack(RULE_CHAIN_ID, RULE_NODE_ID);
var targetRuleChainId = new RuleChainId(UUID.randomUUID());
// WHEN
defaultTbContext.input(msg, targetRuleChainId);
// THEN - cap failure: tellFailure with message naming the limit, ENV name and yaml key
ArgumentCaptor<TbActorMsg> tellCaptor = ArgumentCaptor.forClass(TbActorMsg.class);
then(chainActorMock).should().tell(tellCaptor.capture());
TbActorMsg capturedTellMsg = tellCaptor.getValue();
assertThat(capturedTellMsg).isInstanceOf(RuleNodeToRuleChainTellNextMsg.class);
RuleNodeToRuleChainTellNextMsg failureMsg = (RuleNodeToRuleChainTellNextMsg) capturedTellMsg;
assertThat(failureMsg.getRelationTypes()).containsOnly(TbNodeConnectionType.FAILURE);
assertThat(failureMsg.getFailureMessage()).contains("visit limit 2 reached");
assertThat(failureMsg.getFailureMessage()).contains("TB_RULE_CHAIN_INPUT_LOOP_MAX_VISITS");
assertThat(failureMsg.getFailureMessage()).contains("actors.rule.chain.input_loop_max_visits");
then(mainCtxMock).should().getRuleChainInputLoopMaxVisits();
then(mainCtxMock).shouldHaveNoMoreInteractions();
}
@Test
public void givenCurrentNodeRepeatedBelowCap_whenInput_thenEnqueue() {
// GIVEN - customer's pagination case: same RuleChainInputNode visited three times already.
// With cap > occurrences it must proceed.
given(mainCtxMock.getRuleChainInputLoopMaxVisits()).willReturn(10);
var tpi = resolve(null);
given(mainCtxMock.resolve(eq(ServiceType.TB_RULE_ENGINE), nullable(String.class), eq(TENANT_ID), eq(TENANT_ID))).willReturn(tpi);
var clusterService = mock(TbClusterService.class);
given(mainCtxMock.getClusterService()).willReturn(clusterService);
var callbackMock = mock(TbMsgCallback.class);
given(callbackMock.isMsgValid()).willReturn(true);
var ruleNode = new RuleNode(RULE_NODE_ID);
ruleNode.setRuleChainId(RULE_CHAIN_ID);
ruleNode.setDebugSettings(DebugSettings.off());
given(nodeCtxMock.getTenantId()).willReturn(TENANT_ID);
given(nodeCtxMock.getSelf()).willReturn(ruleNode);
var msg = TbMsg.newMsg()
.type(TbMsgType.POST_TELEMETRY_REQUEST)
.originator(TENANT_ID)
.copyMetaData(TbMsgMetaData.EMPTY)
.data(TbMsg.EMPTY_STRING)
.callback(callbackMock)
.build();
// three prior iterations through the same input node — count 3 < cap 10
msg.pushToStack(RULE_CHAIN_ID, RULE_NODE_ID);
msg.pushToStack(RULE_CHAIN_ID, RULE_NODE_ID);
msg.pushToStack(RULE_CHAIN_ID, RULE_NODE_ID);
var targetRuleChainId = new RuleChainId(UUID.randomUUID());
// WHEN
defaultTbContext.input(msg, targetRuleChainId);
// THEN
then(clusterService).should().pushMsgToRuleEngine(eq(tpi), eq(msg.getId()), any(), any());
then(chainActorMock).shouldHaveNoInteractions();
}
@Test
public void givenNegativeCapAndCurrentNodeAlreadyInStack_whenInput_thenTellFailure() {
// GIVEN - misconfiguration: admin set a non-positive value. Per yaml contract, values < 1
// are clamped to 1 — strict behavior is preserved instead of failing at startup.
given(mainCtxMock.getRuleChainInputLoopMaxVisits()).willReturn(-3);
var callbackMock = mock(TbMsgCallback.class);
given(callbackMock.isMsgValid()).willReturn(true);
var ruleNode = new RuleNode(RULE_NODE_ID);
ruleNode.setRuleChainId(RULE_CHAIN_ID);
ruleNode.setDebugSettings(DebugSettings.off());
given(nodeCtxMock.getSelf()).willReturn(ruleNode);
given(nodeCtxMock.getTenantId()).willReturn(TENANT_ID);
given(nodeCtxMock.getChainActor()).willReturn(chainActorMock);
var msg = TbMsg.newMsg()
.type(TbMsgType.POST_TELEMETRY_REQUEST)
.originator(TENANT_ID)
.copyMetaData(TbMsgMetaData.EMPTY)
.data(TbMsg.EMPTY_STRING)
.callback(callbackMock)
.build();
msg.pushToStack(RULE_CHAIN_ID, RULE_NODE_ID);
var targetRuleChainId = new RuleChainId(UUID.randomUUID());
// WHEN
defaultTbContext.input(msg, targetRuleChainId);
// THEN - clamped cap = 1, count = 1, fails with limit 1 in the message
ArgumentCaptor<TbActorMsg> tellCaptor = ArgumentCaptor.forClass(TbActorMsg.class);
then(chainActorMock).should().tell(tellCaptor.capture());
TbActorMsg capturedTellMsg = tellCaptor.getValue();
RuleNodeToRuleChainTellNextMsg failureMsg = (RuleNodeToRuleChainTellNextMsg) capturedTellMsg;
assertThat(failureMsg.getFailureMessage()).contains("visit limit 1 reached");
}
@Test
public void givenDefaultCapAndFreshMessage_whenInput_thenEnqueue() {
// GIVEN - default cap = 1, fresh message (count = 0): first visit must pass.
given(mainCtxMock.getRuleChainInputLoopMaxVisits()).willReturn(1);
var tpi = resolve(null);
given(mainCtxMock.resolve(eq(ServiceType.TB_RULE_ENGINE), nullable(String.class), eq(TENANT_ID), eq(TENANT_ID))).willReturn(tpi);
var clusterService = mock(TbClusterService.class);
given(mainCtxMock.getClusterService()).willReturn(clusterService);
var callbackMock = mock(TbMsgCallback.class);
given(callbackMock.isMsgValid()).willReturn(true);
var ruleNode = new RuleNode(RULE_NODE_ID);
ruleNode.setRuleChainId(RULE_CHAIN_ID);
ruleNode.setDebugSettings(DebugSettings.off());
given(nodeCtxMock.getTenantId()).willReturn(TENANT_ID);
given(nodeCtxMock.getSelf()).willReturn(ruleNode);
var msg = TbMsg.newMsg()
.type(TbMsgType.POST_TELEMETRY_REQUEST)
.originator(TENANT_ID)
.copyMetaData(TbMsgMetaData.EMPTY)
.data(TbMsg.EMPTY_STRING)
.callback(callbackMock)
.build();
var targetRuleChainId = new RuleChainId(UUID.randomUUID());
// WHEN
defaultTbContext.input(msg, targetRuleChainId);
// THEN
then(clusterService).should().pushMsgToRuleEngine(eq(tpi), eq(msg.getId()), any(), any());
then(chainActorMock).shouldHaveNoInteractions();
}
@MethodSource

4
common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
View File

@ -253,8 +253,8 @@ public final class TbMsg implements Serializable {
ctx.push(ruleChainId, ruleNodeId);
}
public boolean isAlreadyInStack(RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
return ctx.isAlreadyInStack(ruleChainId, ruleNodeId);
public int countOccurrences(RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
return ctx.countOccurrences(ruleChainId, ruleNodeId);
}
public TbMsgProcessingStackItem popFormStack() {

9
common/message/src/main/java/org/thingsboard/server/common/msg/TbMsgProcessingCtx.java
View File

@ -63,16 +63,17 @@ public final class TbMsgProcessingCtx implements Serializable {
stack.add(new TbMsgProcessingStackItem(ruleChainId, ruleNodeId));
}
public boolean isAlreadyInStack(RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
public int countOccurrences(RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
if (stack == null || stack.isEmpty()) {
return false;
return 0;
}
int count = 0;
for (TbMsgProcessingStackItem item : stack) {
if (ruleChainId.equals(item.getRuleChainId()) && ruleNodeId.equals(item.getRuleNodeId())) {
return true;
count++;
}
}
return false;
return count;
}
public TbMsgProcessingStackItem pop() {

42
common/message/src/test/java/org/thingsboard/server/common/msg/TbMsgProcessingCtxTest.java
View File

@ -29,61 +29,65 @@ class TbMsgProcessingCtxTest {
private final RuleNodeId RULE_NODE_ID = new RuleNodeId(UUID.fromString("1ca5e2ef-1309-41d9-bafa-709e9df0e2a6"));
@Test
void givenEmptyStack_whenIsAlreadyInStack_thenReturnFalse() {
void givenEmptyStack_whenCountOccurrences_thenReturnZero() {
TbMsgProcessingCtx ctx = new TbMsgProcessingCtx();
assertThat(ctx.isAlreadyInStack(RULE_CHAIN_ID, RULE_NODE_ID)).isFalse();
assertThat(ctx.countOccurrences(RULE_CHAIN_ID, RULE_NODE_ID)).isZero();
}
@Test
void givenStackWithDifferentEntry_whenIsAlreadyInStack_thenReturnFalse() {
void givenStackWithoutMatch_whenCountOccurrences_thenReturnZero() {
TbMsgProcessingCtx ctx = new TbMsgProcessingCtx();
ctx.push(new RuleChainId(UUID.randomUUID()), new RuleNodeId(UUID.randomUUID()));
ctx.push(new RuleChainId(UUID.randomUUID()), new RuleNodeId(UUID.randomUUID()));
assertThat(ctx.isAlreadyInStack(RULE_CHAIN_ID, RULE_NODE_ID)).isFalse();
assertThat(ctx.countOccurrences(RULE_CHAIN_ID, RULE_NODE_ID)).isZero();
}
@Test
void givenStackWithMatchingEntry_whenIsAlreadyInStack_thenReturnTrue() {
void givenStackWithSingleMatch_whenCountOccurrences_thenReturnOne() {
TbMsgProcessingCtx ctx = new TbMsgProcessingCtx();
ctx.push(new RuleChainId(UUID.randomUUID()), new RuleNodeId(UUID.randomUUID()));
ctx.push(RULE_CHAIN_ID, RULE_NODE_ID);
ctx.push(new RuleChainId(UUID.randomUUID()), new RuleNodeId(UUID.randomUUID()));
assertThat(ctx.isAlreadyInStack(RULE_CHAIN_ID, RULE_NODE_ID)).isTrue();
assertThat(ctx.countOccurrences(RULE_CHAIN_ID, RULE_NODE_ID)).isEqualTo(1);
}
@Test
void givenStackWithMatchingEntryAmongOthers_whenIsAlreadyInStack_thenReturnTrue() {
void givenStackWithThreeMatches_whenCountOccurrences_thenReturnThree() {
TbMsgProcessingCtx ctx = new TbMsgProcessingCtx();
ctx.push(new RuleChainId(UUID.randomUUID()), new RuleNodeId(UUID.randomUUID()));
ctx.push(RULE_CHAIN_ID, RULE_NODE_ID);
ctx.push(new RuleChainId(UUID.randomUUID()), new RuleNodeId(UUID.randomUUID()));
ctx.push(RULE_CHAIN_ID, RULE_NODE_ID);
ctx.push(RULE_CHAIN_ID, RULE_NODE_ID);
assertThat(ctx.isAlreadyInStack(RULE_CHAIN_ID, RULE_NODE_ID)).isTrue();
assertThat(ctx.countOccurrences(RULE_CHAIN_ID, RULE_NODE_ID)).isEqualTo(3);
}
@Test
void givenStackWithSameChainButDifferentNode_whenIsAlreadyInStack_thenReturnFalse() {
void givenStackWithMatchThenPopped_whenCountOccurrences_thenReturnZero() {
TbMsgProcessingCtx ctx = new TbMsgProcessingCtx();
ctx.push(RULE_CHAIN_ID, new RuleNodeId(UUID.randomUUID()));
ctx.push(RULE_CHAIN_ID, RULE_NODE_ID);
ctx.pop();
assertThat(ctx.isAlreadyInStack(RULE_CHAIN_ID, RULE_NODE_ID)).isFalse();
assertThat(ctx.countOccurrences(RULE_CHAIN_ID, RULE_NODE_ID)).isZero();
}
@Test
void givenStackWithSameNodeButDifferentChain_whenIsAlreadyInStack_thenReturnFalse() {
void givenStackWithSameChainButDifferentNode_whenCountOccurrences_thenReturnZero() {
TbMsgProcessingCtx ctx = new TbMsgProcessingCtx();
ctx.push(new RuleChainId(UUID.randomUUID()), RULE_NODE_ID);
ctx.push(RULE_CHAIN_ID, new RuleNodeId(UUID.randomUUID()));
assertThat(ctx.isAlreadyInStack(RULE_CHAIN_ID, RULE_NODE_ID)).isFalse();
assertThat(ctx.countOccurrences(RULE_CHAIN_ID, RULE_NODE_ID)).isZero();
}
@Test
void givenStackWithEntryThenPopped_whenIsAlreadyInStack_thenReturnFalse() {
void givenStackWithSameNodeButDifferentChain_whenCountOccurrences_thenReturnZero() {
TbMsgProcessingCtx ctx = new TbMsgProcessingCtx();
ctx.push(RULE_CHAIN_ID, RULE_NODE_ID);
ctx.pop();
ctx.push(new RuleChainId(UUID.randomUUID()), RULE_NODE_ID);
assertThat(ctx.isAlreadyInStack(RULE_CHAIN_ID, RULE_NODE_ID)).isFalse();
assertThat(ctx.countOccurrences(RULE_CHAIN_ID, RULE_NODE_ID)).isZero();
}
}

Write Preview
Loading…
Cancel
Save