fix: preserve rule node execution counter when copying messages in delay and deduplication nodes

Delay and deduplication rule nodes were creating brand new TbMsg objects
instead of copying the original, which reset the ruleNodeExecCounter to 0.
This allowed bypassing the maxRuleNodeExecutionsPerMessage limit.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java

@@ -137,7 +137,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
                 defaultCtx.tellFailure(msg.getMsg(), e);
             }
         } else {
-            tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more then " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
+            tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more than " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
         }
     }
 
@@ -160,7 +160,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
                     msg.getCtx().tellFailure(msg.getMsg(), e);
                 }
             } else {
-                tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more then " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
+                tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more than " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
             }
         }
     }

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

@@ -110,11 +110,10 @@ public final class TbMsg implements Serializable {
                 .build();
     }
 
-    public TbMsg copyWithNewCtx() {
+    public TbMsgBuilder copyWithNewCtx() {
         return copy()
                 .ctx(ctx.copy())
-                .callback(TbMsgCallback.EMPTY)
-                .build();
+                .callback(TbMsgCallback.EMPTY);
     }
 
     private TbMsg(String queueName, UUID id, long ts, TbMsgType internalType, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data,

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java

@@ -41,6 +41,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
@@ -177,15 +178,11 @@ public class TbMsgDeduplicationNode implements TbNode {
                         }
                     }
                     if (resultMsg != null) {
-                        String queueName1 = queueName != null ? queueName : resultMsg.getQueueName();
-                        deduplicationResults.add(TbMsg.newMsg()
-                                .queueName(queueName1)
-                                .type(resultMsg.getType())
-                                .originator(resultMsg.getOriginator())
-                                .customerId(resultMsg.getCustomerId())
-                                .copyMetaData(resultMsg.getMetaData())
-                                .data(resultMsg.getData())
-                                .build());
+                        var msgBuilder = resultMsg.copyWithNewCtx().id(UUID.randomUUID());
+                        if (queueName != null) {
+                            msgBuilder.queueName(queueName);
+                        }
+                        deduplicationResults.add(msgBuilder.build());
                     }
                 }
                 packBoundsOpt = findValidPack(msgList, deduplicationTimeoutMs);

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java

@@ -62,17 +62,9 @@ public class TbMsgDelayNode implements TbNode {
         if (msg.isTypeOf(TbMsgType.DELAY_TIMEOUT_SELF_MSG)) {
             TbMsg pendingMsg = pendingMsgs.remove(UUID.fromString(msg.getData()));
             if (pendingMsg != null) {
-                ctx.enqueueForTellNext(
-                        TbMsg.newMsg()
-                                .queueName(pendingMsg.getQueueName())
-                                .type(pendingMsg.getType())
-                                .originator(pendingMsg.getOriginator())
-                                .customerId(pendingMsg.getCustomerId())
-                                .copyMetaData(pendingMsg.getMetaData())
-                                .data(pendingMsg.getData())
-                                .build(),
-                        TbNodeConnectionType.SUCCESS
-                );
+                ctx.enqueueForTellNext(pendingMsg.copyWithNewCtx()
+                        .id(UUID.randomUUID())
+                        .build(), TbNodeConnectionType.SUCCESS);
             }
         } else {
             if (pendingMsgs.size() < config.getMaxPendingMsgs()) {

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/external/TbAbstractExternalNode.java

@@ -30,7 +30,7 @@ public abstract class TbAbstractExternalNode implements TbNode {
 
     protected void tellSuccess(TbContext ctx, TbMsg tbMsg) {
         if (forceAck) {
-            ctx.enqueueForTellNext(tbMsg.copyWithNewCtx(), TbNodeConnectionType.SUCCESS);
+            ctx.enqueueForTellNext(tbMsg.copyWithNewCtx().build(), TbNodeConnectionType.SUCCESS);
         } else {
             ctx.tellSuccess(tbMsg);
         }
@@ -39,9 +39,9 @@ public abstract class TbAbstractExternalNode implements TbNode {
     protected void tellFailure(TbContext ctx, TbMsg tbMsg, Throwable t) {
         if (forceAck) {
             if (t == null) {
-                ctx.enqueueForTellNext(tbMsg.copyWithNewCtx(), TbNodeConnectionType.FAILURE);
+                ctx.enqueueForTellNext(tbMsg.copyWithNewCtx().build(), TbNodeConnectionType.FAILURE);
             } else {
-                ctx.enqueueForTellFailure(tbMsg.copyWithNewCtx(), t);
+                ctx.enqueueForTellFailure(tbMsg.copyWithNewCtx().build(), t);
             }
         } else {
             if (t == null) {
@@ -55,7 +55,7 @@ public abstract class TbAbstractExternalNode implements TbNode {
     protected TbMsg ackIfNeeded(TbContext ctx, TbMsg msg) {
         if (forceAck) {
             ctx.ack(msg);
-            return msg.copyWithNewCtx();
+            return msg.copyWithNewCtx().build();
         } else {
             return msg;
         }

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeTest.java

@@ -0,0 +1,106 @@
+/**
+ * Copyright © 2016-2026 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.delay;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.id.RuleNodeId;
+import org.thingsboard.server.common.data.msg.TbMsgType;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.TbMsgProcessingCtx;
+import org.thingsboard.server.common.msg.queue.TbMsgCallback;
+
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.then;
+import static org.mockito.Mockito.lenient;
+
+@ExtendWith(MockitoExtension.class)
+class TbMsgDelayNodeTest {
+
+    final DeviceId deviceId = new DeviceId(UUID.fromString("5770153d-6ca2-4447-8a54-5d8a4538e052"));
+    final RuleNodeId ruleNodeId = new RuleNodeId(UUID.fromString("ee682a85-7f5a-4182-91bc-46e555138fe2"));
+
+    TbMsgDelayNode node;
+
+    @Mock
+    TbContext ctxMock;
+
+    @BeforeEach
+    void setUp() throws TbNodeException {
+        node = new TbMsgDelayNode();
+        var config = new TbMsgDelayNodeConfiguration().defaultConfiguration();
+        node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
+
+        lenient().when(ctxMock.getSelfId()).thenReturn(ruleNodeId);
+    }
+
+    @Test
+    void shouldPreserveRuleNodeCounterAndResetCallbackWhenEnqueuingDelayedMsg() {
+        // GIVEN
+        int ruleNodeExecCounter = 5;
+        var originalMsg = TbMsg.newMsg()
+                .id(UUID.randomUUID())
+                .type(TbMsgType.POST_TELEMETRY_REQUEST)
+                .originator(deviceId)
+                .metaData(TbMsgMetaData.EMPTY)
+                .data("{\"temperature\":42}")
+                .ctx(new TbMsgProcessingCtx(ruleNodeExecCounter))
+                .build();
+
+        String originalMsgId = originalMsg.getId().toString();
+        var tickMsg = TbMsg.newMsg()
+                .type(TbMsgType.DELAY_TIMEOUT_SELF_MSG)
+                .originator(ruleNodeId)
+                .metaData(TbMsgMetaData.EMPTY)
+                .data(originalMsgId)
+                .build();
+        given(ctxMock.newMsg(null, TbMsgType.DELAY_TIMEOUT_SELF_MSG, ruleNodeId, null, TbMsgMetaData.EMPTY, originalMsgId)).willReturn(tickMsg);
+
+        node.onMsg(ctxMock, originalMsg);
+
+        // WHEN
+        node.onMsg(ctxMock, tickMsg);
+
+        // THEN
+        var msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        then(ctxMock).should().enqueueForTellNext(msgCaptor.capture(), eq(TbNodeConnectionType.SUCCESS));
+
+        var enqueuedMsg = msgCaptor.getValue();
+        assertThat(enqueuedMsg).usingRecursiveComparison()
+                .ignoringFields("id", "ts", "callback")
+                .isEqualTo(originalMsg);
+
+        assertThat(enqueuedMsg.getId()).isNotNull().isNotEqualTo(originalMsg.getId());
+        assertThat(enqueuedMsg.getAndIncrementRuleNodeCounter()).isEqualTo(ruleNodeExecCounter);
+        assertThat(enqueuedMsg.getCallback()).isSameAs(TbMsgCallback.EMPTY);
+    }
+
+}

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbMsgDeduplicationNodeTest.java

@@ -47,6 +47,8 @@ import org.thingsboard.server.common.data.msg.TbMsgType;
 import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.TbMsgProcessingCtx;
+import org.thingsboard.server.common.msg.queue.TbMsgCallback;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -79,6 +81,8 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
 
     private TbContext ctx;
 
+    private static final int RULE_NODE_EXEC_COUNTER = 5;
+
     private final ScheduledExecutorService executorService = ThingsBoardExecutors.newSingleThreadScheduledExecutor("de-duplication-node-test");
     private final int deduplicationInterval = 1;
 
@@ -204,6 +208,8 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
         Assertions.assertEquals(firstMsg.getData(), actualMsg.getData());
         Assertions.assertEquals(firstMsg.getMetaData(), actualMsg.getMetaData());
         Assertions.assertEquals(firstMsg.getType(), actualMsg.getType());
+        Assertions.assertEquals(RULE_NODE_EXEC_COUNTER, actualMsg.getAndIncrementRuleNodeCounter());
+        Assertions.assertSame(TbMsgCallback.EMPTY, actualMsg.getCallback());
 
         if (queueName == null) {
             Assertions.assertEquals(firstMsg.getQueueName(), actualMsg.getQueueName());
@@ -257,6 +263,8 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
         Assertions.assertEquals(msgWithLatestTs.getData(), actualMsg.getData());
         Assertions.assertEquals(msgWithLatestTs.getMetaData(), actualMsg.getMetaData());
         Assertions.assertEquals(msgWithLatestTs.getType(), actualMsg.getType());
+        Assertions.assertEquals(RULE_NODE_EXEC_COUNTER, actualMsg.getAndIncrementRuleNodeCounter());
+        Assertions.assertSame(TbMsgCallback.EMPTY, actualMsg.getCallback());
     }
 
     @Test
@@ -402,6 +410,8 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
         Assertions.assertEquals(msgWithLatestTsInFirstPack.getData(), actualMsg.getData());
         Assertions.assertEquals(msgWithLatestTsInFirstPack.getMetaData(), actualMsg.getMetaData());
         Assertions.assertEquals(msgWithLatestTsInFirstPack.getType(), actualMsg.getType());
+        Assertions.assertEquals(RULE_NODE_EXEC_COUNTER, actualMsg.getAndIncrementRuleNodeCounter());
+        Assertions.assertSame(TbMsgCallback.EMPTY, actualMsg.getCallback());
 
         // verify that newMsg is called but content of messages is the same as in the last msg for the second pack.
         actualMsg = resultMsgs.get(1);
@@ -411,6 +421,8 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
         Assertions.assertEquals(msgWithLatestTsInSecondPack.getData(), actualMsg.getData());
         Assertions.assertEquals(msgWithLatestTsInSecondPack.getMetaData(), actualMsg.getMetaData());
         Assertions.assertEquals(msgWithLatestTsInSecondPack.getType(), actualMsg.getType());
+        Assertions.assertEquals(RULE_NODE_EXEC_COUNTER, actualMsg.getAndIncrementRuleNodeCounter());
+        Assertions.assertSame(TbMsgCallback.EMPTY, actualMsg.getCallback());
     }
 
     @Test
@@ -539,6 +551,7 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
                 .originator(deviceId)
                 .copyMetaData(metaData)
                 .data(JacksonUtil.toString(dataNode))
+                .ctx(new TbMsgProcessingCtx(RULE_NODE_EXEC_COUNTER))
                 .build();
     }