Merge pull request #15111 from volodymyr-babak/edge-store-sync-as-attribute

Added syncInProgress as edge attribute

application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java

@@ -78,6 +78,7 @@ import org.thingsboard.server.service.edge.rpc.processor.telemetry.TelemetryEdge
 import org.thingsboard.server.service.edge.rpc.processor.user.UserProcessor;
 import org.thingsboard.server.service.edge.rpc.sync.EdgeRequestsService;
 import org.thingsboard.server.service.executors.GrpcCallbackExecutorService;
+import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
 
 import java.util.EnumMap;
 import java.util.List;
@@ -104,6 +105,9 @@ public class EdgeContextComponent {
     }
 
     // services
+    @Autowired
+    private TelemetrySubscriptionService tsSubService;
+
     @Autowired
     private AdminSettingsService adminSettingsService;
 

application/src/main/java/org/thingsboard/server/service/edge/rpc/AttributeSaveCallback.java

@@ -0,0 +1,43 @@
+/**
+ * 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.server.service.edge.rpc;
+
+import com.google.common.util.concurrent.FutureCallback;
+import jakarta.annotation.Nullable;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.common.data.id.EdgeId;
+import org.thingsboard.server.common.data.id.TenantId;
+
+@Slf4j
+@AllArgsConstructor
+public class AttributeSaveCallback implements FutureCallback<Void> {
+
+    private final TenantId tenantId;
+    private final EdgeId edgeId;
+    private final String key;
+    private final Object value;
+
+    @Override
+    public void onSuccess(@Nullable Void result) {
+        log.trace("[{}][{}] Successfully updated attribute [{}] with value [{}]", tenantId, edgeId, key, value);
+    }
+
+    @Override
+    public void onFailure(Throwable t) {
+        log.warn("[{}][{}] Failed to update attribute [{}] with value [{}]", tenantId, edgeId, key, value, t);
+    }
+}

application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java

@@ -135,9 +135,6 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
     @Lazy
     private EdgeContextComponent ctx;
 
-    @Autowired
-    private TelemetrySubscriptionService tsSubService;
-
     @Autowired
     private TbClusterService clusterService;
 
@@ -553,14 +550,14 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
     private void save(TenantId tenantId, EdgeId edgeId, String key, long value) {
         log.debug("[{}][{}] Updating long edge telemetry [{}] [{}]", tenantId, edgeId, key, value);
         if (persistToTelemetry) {
-            tsSubService.saveTimeseries(TimeseriesSaveRequest.builder()
+            ctx.getTsSubService().saveTimeseries(TimeseriesSaveRequest.builder()
                     .tenantId(tenantId)
                     .entityId(edgeId)
                     .entry(new LongDataEntry(key, value))
                     .callback(new AttributeSaveCallback(tenantId, edgeId, key, value))
                     .build());
         } else {
-            tsSubService.saveAttributes(AttributesSaveRequest.builder()
+            ctx.getTsSubService().saveAttributes(AttributesSaveRequest.builder()
                     .tenantId(tenantId)
                     .entityId(edgeId)
                     .scope(AttributeScope.SERVER_SCOPE)
@@ -573,14 +570,14 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
     private void save(TenantId tenantId, EdgeId edgeId, String key, boolean value) {
         log.debug("[{}][{}] Updating boolean edge telemetry [{}] [{}]", tenantId, edgeId, key, value);
         if (persistToTelemetry) {
-            tsSubService.saveTimeseries(TimeseriesSaveRequest.builder()
+            ctx.getTsSubService().saveTimeseries(TimeseriesSaveRequest.builder()
                     .tenantId(tenantId)
                     .entityId(edgeId)
                     .entry(new BooleanDataEntry(key, value))
                     .callback(new AttributeSaveCallback(tenantId, edgeId, key, value))
                     .build());
         } else {
-            tsSubService.saveAttributes(AttributesSaveRequest.builder()
+            ctx.getTsSubService().saveAttributes(AttributesSaveRequest.builder()
                     .tenantId(tenantId)
                     .entityId(edgeId)
                     .scope(AttributeScope.SERVER_SCOPE)
@@ -590,32 +587,6 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
         }
     }
 
-    private static class AttributeSaveCallback implements FutureCallback<Void> {
-
-        private final TenantId tenantId;
-        private final EdgeId edgeId;
-        private final String key;
-        private final Object value;
-
-        AttributeSaveCallback(TenantId tenantId, EdgeId edgeId, String key, Object value) {
-            this.tenantId = tenantId;
-            this.edgeId = edgeId;
-            this.key = key;
-            this.value = value;
-        }
-
-        @Override
-        public void onSuccess(@Nullable Void result) {
-            log.trace("[{}][{}] Successfully updated attribute [{}] with value [{}]", tenantId, edgeId, key, value);
-        }
-
-        @Override
-        public void onFailure(Throwable t) {
-            log.warn("[{}][{}] Failed to update attribute [{}] with value [{}]", tenantId, edgeId, key, value, t);
-        }
-
-    }
-
     private void pushRuleEngineMessage(TenantId tenantId, Edge edge, long ts, TbMsgType msgType) {
         try {
             EdgeId edgeId = edge.getId();

application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java

@@ -25,6 +25,7 @@ import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.springframework.data.util.Pair;
+import org.thingsboard.rule.engine.api.AttributesSaveRequest;
 import org.thingsboard.server.common.data.AttributeScope;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.EdgeUtils;
@@ -37,6 +38,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BooleanDataEntry;
 import org.thingsboard.server.common.data.kv.LongDataEntry;
 import org.thingsboard.server.common.data.kv.StringDataEntry;
 import org.thingsboard.server.common.data.limit.LimitedApi;
@@ -195,7 +197,7 @@ public abstract class EdgeGrpcSession implements Closeable {
                             }
                             startSyncProcess(fullSync);
                         } else {
-                            syncInProgress = false;
+                            updateSyncInProgress(false);
                         }
                     }
                     if (requestMsg.getMsgType().equals(RequestMsgType.UPLINK_RPC_MESSAGE)) {
@@ -241,6 +243,11 @@ public abstract class EdgeGrpcSession implements Closeable {
         log.debug("[{}] onConfigurationUpdate [{}]", sessionId, edge);
         this.tenantId = edge.getTenantId();
         this.edge = edge;
+        if (!this.edge.getCustomerId().equals(edge.getCustomerId())) {
+            // do not send edge configuration message on customer update
+            // message send by separate flow from assign_to or unassing_from customer
+            return;
+        }
         EdgeUpdateMsg edgeConfig = EdgeUpdateMsg.newBuilder()
                 .setConfiguration(EdgeMsgConstructorUtils.constructEdgeConfiguration(edge)).build();
         ResponseMsg edgeConfigMsg = ResponseMsg.newBuilder()
@@ -252,7 +259,7 @@ public abstract class EdgeGrpcSession implements Closeable {
     public void startSyncProcess(boolean fullSync) {
         if (!syncInProgress) {
             log.info("[{}][{}][{}] Staring edge sync process", tenantId, edge.getId(), sessionId);
-            syncInProgress = true;
+            updateSyncInProgress(true);
             interruptGeneralProcessingOnSync();
             doSync(new EdgeSyncCursor(ctx, edge, fullSync));
         } else {
@@ -398,6 +405,18 @@ public abstract class EdgeGrpcSession implements Closeable {
         ctx.getAttributesService().save(tenantId, edge.getId(), AttributeScope.SERVER_SCOPE, attributeKvEntry);
     }
 
+    private void updateSyncInProgress(Boolean value) {
+        this.syncInProgress = value;
+
+        ctx.getTsSubService().saveAttributes(AttributesSaveRequest.builder()
+                .tenantId(tenantId)
+                .entityId(edge.getId())
+                .scope(AttributeScope.SERVER_SCOPE)
+                .entry(new BooleanDataEntry(DataConstants.EDGE_SYNC_IN_PROGRESS_ATTR_KEY, value))
+                .callback(new AttributeSaveCallback(tenantId, edge.getId(), DataConstants.EDGE_SYNC_IN_PROGRESS_ATTR_KEY, value))
+                .build());
+    }
+
     private void interruptGeneralProcessingOnSync() {
         log.debug("[{}][{}][{}] Sync process started. General processing interrupted!", tenantId, edge.getId(), sessionId);
         stopCurrentSendDownlinkMsgsTask(true);
@@ -766,7 +785,7 @@ public abstract class EdgeGrpcSession implements Closeable {
     }
 
     private void markSyncCompletedSendEdgeEventUpdate() {
-        syncInProgress = false;
+        updateSyncInProgress(false);
         ctx.getClusterService().onEdgeEventUpdate(new EdgeEventUpdateMsg(edge.getTenantId(), edge.getId()));
     }
 

application/src/test/java/org/thingsboard/server/edge/EdgeTest.java

@@ -15,10 +15,12 @@
  */
 package org.thingsboard.server.edge;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import org.junit.Assert;
 import org.junit.Test;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.Customer;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.edge.Edge;
 import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.EntityId;
@@ -26,10 +28,14 @@ import org.thingsboard.server.dao.service.DaoSqlTest;
 import org.thingsboard.server.gen.edge.v1.CustomerUpdateMsg;
 import org.thingsboard.server.gen.edge.v1.EdgeConfiguration;
 import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
+import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope;
 
+import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 @DaoSqlTest
 public class EdgeTest extends AbstractEdgeTest {
 
@@ -76,4 +82,23 @@ public class EdgeTest extends AbstractEdgeTest {
         Assert.assertEquals(savedCustomer.getUuidId().getMostSignificantBits(), customerUpdateMsg.getIdMSB());
         Assert.assertEquals(savedCustomer.getUuidId().getLeastSignificantBits(), customerUpdateMsg.getIdLSB());
     }
+
+    @Test
+    public void testSyncEdge_attributeUpdated() throws Exception {
+        getWsClient().subscribeForAttributes(edge.getId(), TbAttributeSubscriptionScope.SERVER_SCOPE.name(), List.of(DataConstants.EDGE_SYNC_IN_PROGRESS_ATTR_KEY));
+
+        doPost("/api/edge/sync/" + edge.getId());
+
+        // wait for sync to start
+        waitForEdgeSyncInProgressEqualsValue(true);
+
+        // wait for sync to end
+        waitForEdgeSyncInProgressEqualsValue(false);
+    }
+
+    private void waitForEdgeSyncInProgressEqualsValue(Boolean value) {
+        getWsClient().registerWaitForUpdate();
+        JsonNode update = JacksonUtil.toJsonNode(getWsClient().waitForUpdate());
+        assertThat(update.get("data").get(DataConstants.EDGE_SYNC_IN_PROGRESS_ATTR_KEY).get(0).get(1).asBoolean()).isEqualTo(value);
+    }
 }

common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java

@@ -127,6 +127,7 @@ public class DataConstants {
     public static final String EDGE_MSG_SOURCE = "edge";
     public static final String MSG_SOURCE_KEY = "source";
     public static final String EDGE_VERSION_ATTR_KEY = "edgeVersion";
+    public static final String EDGE_SYNC_IN_PROGRESS_ATTR_KEY = "syncInProgress";
 
     public static final String LAST_CONNECTED_GATEWAY = "lastConnectedGateway";
 

dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java

@@ -66,7 +66,6 @@ import org.thingsboard.server.dao.entity.EntityCountService;
 import org.thingsboard.server.dao.eventsourcing.ActionEntityEvent;
 import org.thingsboard.server.dao.eventsourcing.DeleteEntityEvent;
 import org.thingsboard.server.dao.eventsourcing.SaveEntityEvent;
-import org.thingsboard.server.exception.DataValidationException;
 import org.thingsboard.server.dao.relation.RelationService;
 import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.dao.service.DataValidator;
@@ -75,6 +74,7 @@ import org.thingsboard.server.dao.service.Validator;
 import org.thingsboard.server.dao.sql.JpaExecutorService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.user.UserService;
+import org.thingsboard.server.exception.DataValidationException;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -239,9 +239,10 @@ public class EdgeServiceImpl extends AbstractCachedEntityService<EdgeCacheKey, E
             return edge;
         }
         edge.setCustomerId(customerId);
+        Edge result = saveEdge(edge);
         eventPublisher.publishEvent(ActionEntityEvent.builder().tenantId(tenantId).entityId(edgeId)
                 .body(JacksonUtil.toString(customerId)).actionType(ActionType.ASSIGNED_TO_CUSTOMER).build());
-        return saveEdge(edge);
+        return result;
     }
 
     @Override