SONAR-10512 removing from queue has no effect if task status changed

server/sonar-ce-common/src/main/java/org/sonar/ce/queue/CeQueueImpl.java

@@ -34,6 +34,7 @@ import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
 import org.sonar.api.server.ServerSide;
+import org.sonar.api.utils.log.Loggers;
 import org.sonar.ce.task.CeTask;
 import org.sonar.core.util.UuidFactory;
 import org.sonar.core.util.stream.MoreCollectors;
@@ -41,6 +42,7 @@ import org.sonar.db.DbClient;
 import org.sonar.db.DbSession;
 import org.sonar.db.ce.CeActivityDto;
 import org.sonar.db.ce.CeQueueDto;
+import org.sonar.db.ce.DeleteIf;
 import org.sonar.db.component.ComponentDto;
 import org.sonar.server.organization.DefaultOrganizationProvider;
 import org.sonar.server.property.InternalProperties;
@@ -196,6 +198,24 @@ public class CeQueueImpl implements CeQueue {
     remove(dbSession, q, activityDto);
   }
 
+  protected void remove(DbSession dbSession, CeQueueDto queueDto, CeActivityDto activityDto) {
+    String taskUuid = queueDto.getUuid();
+    CeQueueDto.Status expectedQueueDtoStatus = queueDto.getStatus();
+
+    dbClient.ceActivityDao().insert(dbSession, activityDto);
+    dbClient.ceTaskInputDao().deleteByUuids(dbSession, singleton(taskUuid));
+    int deletedTasks = dbClient.ceQueueDao().deleteByUuid(dbSession, taskUuid, new DeleteIf(expectedQueueDtoStatus));
+
+    if (deletedTasks == 1) {
+      dbSession.commit();
+    } else {
+      Loggers.get(CeQueueImpl.class).debug(
+        "Remove rolled back because task in queue with uuid {} and status {} could not be deleted",
+        taskUuid, expectedQueueDtoStatus);
+      dbSession.rollback();
+    }
+  }
+
   @Override
   public int cancelAll() {
     return cancelAll(false);
@@ -245,13 +265,6 @@ public class CeQueueImpl implements CeQueue {
     }
   }
 
-  protected void remove(DbSession dbSession, CeQueueDto queueDto, CeActivityDto activityDto) {
-    dbClient.ceActivityDao().insert(dbSession, activityDto);
-    dbClient.ceQueueDao().deleteByUuid(dbSession, queueDto.getUuid());
-    dbClient.ceTaskInputDao().deleteByUuids(dbSession, singleton(queueDto.getUuid()));
-    dbSession.commit();
-  }
-
   private static class CeQueueDtoToCeTask implements Function<CeQueueDto, CeTask> {
     private final String defaultOrganizationUuid;
     private final Map<String, ComponentDto> componentDtoByUuid;

server/sonar-db-dao/src/main/java/org/sonar/db/ce/CeQueueDao.java

@@ -114,8 +114,12 @@ public class CeQueueDao implements Dao {
     return dto;
   }
 
-  public void deleteByUuid(DbSession session, String uuid) {
-    mapper(session).deleteByUuid(uuid);
+  public int deleteByUuid(DbSession session, String uuid) {
+    return deleteByUuid(session, uuid, null);
+  }
+
+  public int deleteByUuid(DbSession session, String uuid, @Nullable DeleteIf deleteIf) {
+    return mapper(session).deleteByUuid(uuid, deleteIf);
   }
 
   /**

server/sonar-db-dao/src/main/java/org/sonar/db/ce/CeQueueMapper.java

@@ -74,5 +74,5 @@ public interface CeQueueMapper {
     @Param("new") UpdateIf.NewProperties newProperties,
     @Param("old") UpdateIf.OldProperties oldProperties);
 
-  void deleteByUuid(@Param("uuid") String uuid);
+  int deleteByUuid(@Param("uuid") String uuid, @Nullable @Param("deleteIf") DeleteIf deleteIf);
 }

server/sonar-db-dao/src/main/java/org/sonar/db/ce/DeleteIf.java

@@ -0,0 +1,28 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2018 SonarSource SA
+ * mailto:info AT sonarsource DOT com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+package org.sonar.db.ce;
+
+public class DeleteIf {
+  public final CeQueueDto.Status expectedStatus;
+
+  public DeleteIf(CeQueueDto.Status expectedStatus) {
+    this.expectedStatus = expectedStatus;
+  }
+}

server/sonar-db-dao/src/main/resources/org/sonar/db/ce/CeQueueMapper.xml

@@ -255,6 +255,9 @@
       ce_queue
     where
       uuid=#{uuid,jdbcType=VARCHAR}
+      <if test="deleteIf != null">
+        and status = #{deleteIf.expectedStatus,jdbcType=VARCHAR}
+      </if>
   </delete>
 
   <update id="resetTasksWithUnknownWorkerUUIDs">

server/sonar-db-dao/src/test/java/org/sonar/db/ce/CeQueueDaoTest.java

@@ -187,12 +187,45 @@ public class CeQueueDaoTest {
   @Test
   public void test_delete() {
     insertPending(TASK_UUID_1, COMPONENT_UUID_1);
+    insertPending(TASK_UUID_2, COMPONENT_UUID_1);
 
-    underTest.deleteByUuid(db.getSession(), "UNKNOWN");
+    int deletedCount = underTest.deleteByUuid(db.getSession(), "UNKNOWN");
+    assertThat(deletedCount).isEqualTo(0);
     assertThat(underTest.selectByUuid(db.getSession(), TASK_UUID_1)).isPresent();
 
-    underTest.deleteByUuid(db.getSession(), TASK_UUID_1);
-    assertThat(underTest.selectByUuid(db.getSession(), TASK_UUID_1).isPresent()).isFalse();
+    deletedCount = underTest.deleteByUuid(db.getSession(), TASK_UUID_1);
+    assertThat(deletedCount).isEqualTo(1);
+    assertThat(underTest.selectByUuid(db.getSession(), TASK_UUID_1)).isEmpty();
+
+    deletedCount = underTest.deleteByUuid(db.getSession(), TASK_UUID_2, null);
+    assertThat(deletedCount).isEqualTo(1);
+    assertThat(underTest.selectByUuid(db.getSession(), TASK_UUID_2)).isEmpty();
+  }
+
+  @Test
+  public void test_delete_with_expected_status() {
+    insertPending(TASK_UUID_1, COMPONENT_UUID_1);
+    insertInProgress(TASK_UUID_2, COMPONENT_UUID_1);
+
+    int deletedCount = underTest.deleteByUuid(db.getSession(), "UNKNOWN", null);
+    assertThat(deletedCount).isEqualTo(0);
+    assertThat(underTest.selectByUuid(db.getSession(), TASK_UUID_1)).isPresent();
+
+    deletedCount = underTest.deleteByUuid(db.getSession(), TASK_UUID_1, new DeleteIf(IN_PROGRESS));
+    assertThat(deletedCount).isEqualTo(0);
+    assertThat(underTest.selectByUuid(db.getSession(), TASK_UUID_1)).isPresent();
+
+    deletedCount = underTest.deleteByUuid(db.getSession(), TASK_UUID_2, new DeleteIf(PENDING));
+    assertThat(deletedCount).isEqualTo(0);
+    assertThat(underTest.selectByUuid(db.getSession(), TASK_UUID_2)).isPresent();
+
+    deletedCount = underTest.deleteByUuid(db.getSession(), TASK_UUID_1, new DeleteIf(PENDING));
+    assertThat(deletedCount).isEqualTo(1);
+    assertThat(underTest.selectByUuid(db.getSession(), TASK_UUID_1)).isEmpty();
+
+    deletedCount = underTest.deleteByUuid(db.getSession(), TASK_UUID_2, new DeleteIf(IN_PROGRESS));
+    assertThat(deletedCount).isEqualTo(1);
+    assertThat(underTest.selectByUuid(db.getSession(), TASK_UUID_2)).isEmpty();
   }
 
   @Test
@@ -549,6 +582,12 @@ public class CeQueueDaoTest {
     return dto;
   }
 
+  private CeQueueDto insertInProgress(String uuid, String componentUuid) {
+    CeQueueDto ceQueueDto = insertPending(uuid);
+    CeQueueTesting.makeInProgress(db.getSession(), "workerUuid", System2.INSTANCE.now(), ceQueueDto);
+    return underTest.selectByUuid(db.getSession(), uuid).get();
+  }
+
   private static Iterable<Map<String, Object>> upperizeKeys(List<Map<String, Object>> select) {
     return from(select).transform(new Function<Map<String, Object>, Map<String, Object>>() {
       @Nullable