fix: gate mqtt retries until startup cleanup

water-admin/src/test/java/org/dromara/web/config/MqttCommandAckConfigUnitTest.java

@@ -14,8 +14,8 @@ import static org.assertj.core.api.Assertions.assertThat;
 class MqttCommandAckConfigUnitTest {
 
     @Test
-    void commandAckMaxRetryCountIsThree() throws Exception {
+    void commandAckRetrySettingsMatchRuntimeRequirements() throws Exception {
-        Integer maxRetryCount = null;
+        Map<?, ?> commandAckConfig = null;
         Yaml yaml = new Yaml();
         try (InputStream inputStream = new ClassPathResource("application.yml").getInputStream()) {
             for (Object document : yaml.loadAll(inputStream)) {
@@ -27,17 +27,16 @@ class MqttCommandAckConfigUnitTest {
                     continue;
                 }
                 Object commandAck = mqttConfig.get("command-ack");
-                if (!(commandAck instanceof Map<?, ?> commandAckConfig)) {
+                if (commandAck instanceof Map<?, ?> config) {
-                    continue;
+                    commandAckConfig = config;
-                }
-                Object value = commandAckConfig.get("max-retry-count");
-                if (value instanceof Number number) {
-                    maxRetryCount = number.intValue();
                 }
             }
         }
 
-        assertThat(maxRetryCount).isEqualTo(3);
+        assertThat(commandAckConfig).isNotNull();
+        assertThat(commandAckConfig.get("retry-interval-ms")).isEqualTo(10000);
+        assertThat(commandAckConfig.get("scan-interval-ms")).isEqualTo(5000);
+        assertThat(commandAckConfig.get("max-retry-count")).isEqualTo(3);
     }
 
     @Test

water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java

@@ -37,6 +37,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
     private MqttClientManager mqttClientManager;
     private final MqttProperties mqttProperties;
     private final AppDeviceMapper appDeviceMapper;
+    private volatile boolean startupCleanupReady;
     private static final String DEVICE_STATUS_LOCK_PREFIX = "lock:mqtt:device:status:";
 
     static boolean ackMatchesPendingCommand(String topicDeviceNo, DeviceCommand pendingCommand) {
@@ -133,14 +134,33 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
     }
 
     public int clearPendingCommands() {
-        Set<String> commandIds = pendingIds().readAll();
+        RSet<String> pendingIds = pendingIds();
+        Set<String> commandIds = pendingIds.readAll();
+        pendingIds.removeAll(commandIds);
+        startupCleanupReady = true;
+
+        RuntimeException cleanupFailure = null;
         for (String commandId : commandIds) {
-            RedisUtils.deleteObject(pendingKey(commandId));
+            try {
+                RedisUtils.deleteObject(pendingKey(commandId));
+            } catch (RuntimeException e) {
+                if (cleanupFailure == null) {
+                    cleanupFailure = new RuntimeException("Failed to delete MQTT pending command object: " + commandId, e);
+                } else {
+                    cleanupFailure.addSuppressed(e);
+                }
+            }
+        }
+        if (cleanupFailure != null) {
+            throw cleanupFailure;
         }
-        pendingIds().clear();
         return commandIds.size();
     }
 
+    boolean isStartupCleanupReady() {
+        return startupCleanupReady;
+    }
+
     public void removePending(String commandId) {
         if (StringUtils.isBlank(commandId)) {
             return;

water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandRetryTask.java

@@ -16,7 +16,7 @@ public class MqttCommandRetryTask {
 
     @Scheduled(fixedDelayString = "${mqtt.command-ack.scan-interval-ms:5000}")
     public void retryExpiredCommands() {
-        if (mqttProperties.getCommandAck().isEnabled()) {
+        if (ackService.isStartupCleanupReady() && mqttProperties.getCommandAck().isEnabled()) {
          //   log.info("[设备命令重新下发] 定时器调用成功 ");
             ackService.retryExpiredCommands();
         }

water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java

@@ -23,13 +23,16 @@ import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.test.util.ReflectionTestUtils;
 
 import java.time.Duration;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.*;
@@ -99,35 +102,48 @@ class MqttCommandAckServiceTest {
     }
 
     @Test
-    void clearPendingCommandsDeletesPendingObjectsAndClearsIds() {
+    void clearPendingCommandsRemovesSnapshotBeforeDeletingPendingObjects() {
         MqttProperties properties = new MqttProperties();
         AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class);
         RedissonClient redissonClient = mock(RedissonClient.class);
         RSet<String> pendingIds = mock(RSet.class);
+        Set<String> snapshot = new LinkedHashSet<>(List.of("cmd-1", "cmd-2"));
+        AtomicBoolean snapshotRemoved = new AtomicBoolean();
         when(redissonClient.<String>getSet("mqtt:command:pending:ids")).thenReturn(pendingIds);
-        when(pendingIds.readAll()).thenReturn(Set.of("cmd-1", "cmd-2"));
+        when(pendingIds.readAll()).thenReturn(snapshot);
+        doAnswer(invocation -> {
+            snapshotRemoved.set(true);
+            return true;
+        }).when(pendingIds).removeAll(snapshot);
 
         MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
         try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
             redis.when(RedisUtils::getClient).thenReturn(redissonClient);
+            redis.when(() -> RedisUtils.deleteObject(any(String.class))).thenAnswer(invocation -> {
+                assertThat(snapshotRemoved).isTrue();
+                return true;
+            });
 
             int cleared = service.clearPendingCommands();
 
             assertThat(cleared).isEqualTo(2);
+            assertThat(service.isStartupCleanupReady()).isTrue();
             redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-1"));
             redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-2"));
-            verify(pendingIds).clear();
+            verify(pendingIds).removeAll(snapshot);
+            verify(pendingIds, never()).clear();
         }
     }
 
     @Test
-    void clearPendingCommandsReturnsZeroForEmptySet() {
+    void clearPendingCommandsOpensRetryGateForEmptySet() {
         MqttProperties properties = new MqttProperties();
         AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class);
         RedissonClient redissonClient = mock(RedissonClient.class);
         RSet<String> pendingIds = mock(RSet.class);
+        Set<String> snapshot = Set.of();
         when(redissonClient.<String>getSet("mqtt:command:pending:ids")).thenReturn(pendingIds);
-        when(pendingIds.readAll()).thenReturn(Set.of());
+        when(pendingIds.readAll()).thenReturn(snapshot);
 
         MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
         try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
@@ -136,7 +152,64 @@ class MqttCommandAckServiceTest {
             int cleared = service.clearPendingCommands();
 
             assertThat(cleared).isZero();
-            verify(pendingIds).clear();
+            assertThat(service.isStartupCleanupReady()).isTrue();
+            verify(pendingIds).removeAll(snapshot);
+            verify(pendingIds, never()).clear();
+        }
+    }
+
+    @Test
+    void clearPendingCommandsKeepsRetryGateClosedWhenSnapshotRemovalFails() {
+        MqttProperties properties = new MqttProperties();
+        AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class);
+        RedissonClient redissonClient = mock(RedissonClient.class);
+        RSet<String> pendingIds = mock(RSet.class);
+        Set<String> snapshot = Set.of("cmd-1");
+        when(redissonClient.<String>getSet("mqtt:command:pending:ids")).thenReturn(pendingIds);
+        when(pendingIds.readAll()).thenReturn(snapshot);
+        doThrow(new IllegalStateException("redis unavailable")).when(pendingIds).removeAll(snapshot);
+
+        MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
+        try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
+            redis.when(RedisUtils::getClient).thenReturn(redissonClient);
+
+            assertThat(service.isStartupCleanupReady()).isFalse();
+            assertThatThrownBy(service::clearPendingCommands)
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessage("redis unavailable");
+
+            assertThat(service.isStartupCleanupReady()).isFalse();
+            redis.verify(() -> RedisUtils.deleteObject(any(String.class)), never());
+            verify(pendingIds, never()).clear();
+        }
+    }
+
+    @Test
+    void clearPendingCommandsContinuesDeletingObjectsAndReportsFailures() {
+        MqttProperties properties = new MqttProperties();
+        AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class);
+        RedissonClient redissonClient = mock(RedissonClient.class);
+        RSet<String> pendingIds = mock(RSet.class);
+        Set<String> snapshot = new LinkedHashSet<>(List.of("cmd-1", "cmd-2"));
+        when(redissonClient.<String>getSet("mqtt:command:pending:ids")).thenReturn(pendingIds);
+        when(pendingIds.readAll()).thenReturn(snapshot);
+
+        MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
+        try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
+            redis.when(RedisUtils::getClient).thenReturn(redissonClient);
+            redis.when(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-1"))
+                .thenThrow(new IllegalStateException("delete failed"));
+
+            assertThatThrownBy(service::clearPendingCommands)
+                .isInstanceOf(RuntimeException.class)
+                .hasMessageContaining("cmd-1")
+                .hasCauseInstanceOf(IllegalStateException.class);
+
+            verify(pendingIds).removeAll(snapshot);
+            verify(pendingIds, never()).clear();
+            redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-1"));
+            redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-2"));
+            assertThat(service.isStartupCleanupReady()).isTrue();
         }
     }
 

water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandRetryTaskTest.java

@@ -0,0 +1,36 @@
+package org.dromara.mqtt;
+
+import org.dromara.mqtt.config.properties.MqttProperties;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@Tag("dev")
+class MqttCommandRetryTaskTest {
+
+    @Test
+    void retryExpiredCommandsSkipsRetryBeforeStartupCleanupIsReady() {
+        MqttCommandAckService ackService = mock(MqttCommandAckService.class);
+        when(ackService.isStartupCleanupReady()).thenReturn(false);
+        MqttCommandRetryTask task = new MqttCommandRetryTask(ackService, new MqttProperties());
+
+        task.retryExpiredCommands();
+
+        verify(ackService, never()).retryExpiredCommands();
+    }
+
+    @Test
+    void retryExpiredCommandsRetriesAfterStartupCleanupIsReady() {
+        MqttCommandAckService ackService = mock(MqttCommandAckService.class);
+        when(ackService.isStartupCleanupReady()).thenReturn(true);
+        MqttCommandRetryTask task = new MqttCommandRetryTask(ackService, new MqttProperties());
+
+        task.retryExpiredCommands();
+
+        verify(ackService).retryExpiredCommands();
+    }
+}