SONAR-16645 - Push Events Scheduled Purge

server/sonar-db-dao/src/main/java/org/sonar/db/pushevent/PushEventDao.java

@@ -19,11 +19,14 @@
  */
 package org.sonar.db.pushevent;
 
+import java.util.Set;
 import org.sonar.api.utils.System2;
 import org.sonar.core.util.UuidFactory;
 import org.sonar.db.Dao;
 import org.sonar.db.DbSession;
 
+import static org.sonar.db.DatabaseUtils.executeLargeUpdates;
+
 public class PushEventDao implements Dao {
 
   private final UuidFactory uuidFactory;
@@ -39,7 +42,10 @@ public class PushEventDao implements Dao {
       event.setUuid(uuidFactory.create());
     }
 
-    event.setCreatedAt(system2.now());
+    if (event.getCreatedAt() == null) {
+      event.setCreatedAt(system2.now());
+    }
+
     mapper(dbSession).insert(event);
     return event;
   }
@@ -48,6 +54,15 @@ public class PushEventDao implements Dao {
     return mapper(dbSession).selectByUuid(uuid);
   }
 
+
+  public Set<String> selectUuidsOfExpiredEvents(DbSession dbSession, long timestamp) {
+    return mapper(dbSession).selectUuidsOfExpiredEvents(timestamp);
+  }
+
+  public void deleteByUuids(DbSession dbSession, Set<String> pushEventUuids) {
+    executeLargeUpdates(pushEventUuids, mapper(dbSession)::deleteByUuids);
+  }
+
   private static PushEventMapper mapper(DbSession session) {
     return session.getMapper(PushEventMapper.class);
   }

server/sonar-db-dao/src/main/java/org/sonar/db/pushevent/PushEventDto.java

@@ -19,11 +19,13 @@
  */
 package org.sonar.db.pushevent;
 
+import javax.annotation.CheckForNull;
+
 public class PushEventDto {
   private String uuid;
   private String projectUuid;
   private byte[] payload;
-  private long createdAt;
+  private Long createdAt;
 
   public PushEventDto() {
     // nothing to do
@@ -56,7 +58,8 @@ public class PushEventDto {
     return this;
   }
 
-  public long getCreatedAt() {
+  @CheckForNull
+  public Long getCreatedAt() {
     return createdAt;
   }
 

server/sonar-db-dao/src/main/java/org/sonar/db/pushevent/PushEventMapper.java

@@ -19,7 +19,10 @@
  */
 package org.sonar.db.pushevent;
 
+import java.util.List;
+import java.util.Set;
 import javax.annotation.CheckForNull;
+import org.apache.ibatis.annotations.Param;
 
 public interface PushEventMapper {
 
@@ -28,4 +31,8 @@ public interface PushEventMapper {
   @CheckForNull
   PushEventDto selectByUuid(String uuid);
 
+  Set<String> selectUuidsOfExpiredEvents(@Param("timestamp") long timestamp);
+
+  void deleteByUuids(@Param("pushEventUuids") List<String> pushEventUuids);
+
 }

server/sonar-db-dao/src/main/resources/org/sonar/db/pushevent/PushEventMapper.xml

@@ -24,7 +24,7 @@
     )
   </insert>
 
-  <select id="selectByUuid" parameterType="String" resultType="PushEvent">
+   <select id="selectByUuid" parameterType="String" resultType="PushEvent">
     SELECT
     <include refid="pushEventColumns"/>
     FROM push_events pe
@@ -32,4 +32,17 @@
     pe.uuid=#{uuid,jdbcType=VARCHAR}
   </select>
 
+  <select id="selectUuidsOfExpiredEvents" parameterType="long" resultType="string">
+    SELECT
+    pe.uuid
+    FROM push_events pe
+    where
+    pe.created_at &lt;= #{timestamp,jdbcType=BIGINT}
+  </select>
+
+  <delete id="deleteByUuids" parameterType="String">
+    delete from push_events
+    where uuid in <foreach collection="pushEventUuids" open="(" close=")" item="uuid" separator=",">#{uuid,jdbcType=VARCHAR}</foreach>
+  </delete>
+
 </mapper>

server/sonar-db-dao/src/test/java/org/sonar/db/pushevent/PushEventDaoTest.java

@@ -19,6 +19,7 @@
  */
 package org.sonar.db.pushevent;
 
+import java.util.Set;
 import org.junit.Rule;
 import org.junit.Test;
 import org.sonar.api.impl.utils.TestSystem2;
@@ -67,4 +68,50 @@ public class PushEventDaoTest {
 
   }
 
+  @Test
+  public void select_expired_events() {
+    PushEventDto eventDtoFirst = new PushEventDto()
+      .setProjectUuid("project-uuid")
+      .setCreatedAt(1000L)
+      .setPayload("some-event".getBytes(UTF_8));
+
+    PushEventDto eventDtoSecond = new PushEventDto()
+      .setProjectUuid("project-uuid")
+      .setCreatedAt(1000L)
+      .setPayload("some-event".getBytes(UTF_8));
+
+    PushEventDto eventDtoThird = new PushEventDto()
+      .setProjectUuid("project-uuid")
+      .setCreatedAt(2000L)
+      .setPayload("some-event".getBytes(UTF_8));
+
+    underTest.insert(session, eventDtoFirst);
+    underTest.insert(session, eventDtoSecond);
+    underTest.insert(session, eventDtoThird);
+
+    assertThat(underTest.selectUuidsOfExpiredEvents(session, 2000L)).hasSize(3);
+    assertThat(underTest.selectUuidsOfExpiredEvents(session, 1500L)).hasSize(2);
+    assertThat(underTest.selectUuidsOfExpiredEvents(session, 150L)).isEmpty();
+  }
+
+  @Test
+  public void delete_events_in_batches() {
+    PushEventDto eventDtoFirst = new PushEventDto()
+      .setProjectUuid("project-uuid")
+      .setCreatedAt(1000L)
+      .setPayload("some-event".getBytes(UTF_8));
+
+    PushEventDto eventDtoSecond = new PushEventDto()
+      .setProjectUuid("project-uuid")
+      .setCreatedAt(1000L)
+      .setPayload("some-event".getBytes(UTF_8));
+
+    PushEventDto event1 = underTest.insert(session, eventDtoFirst);
+    PushEventDto event2 = underTest.insert(session, eventDtoSecond);
+
+    assertThat(underTest.selectUuidsOfExpiredEvents(db.getSession(), 2000L)).hasSize(2);
+    underTest.deleteByUuids(db.getSession(), Set.of(event1.getUuid(), event2.getUuid()));
+    assertThat(underTest.selectUuidsOfExpiredEvents(db.getSession(), 2000L)).isEmpty();
+  }
+
 }

server/sonar-webserver-pushapi/src/main/java/org/sonar/server/pushapi/ServerPushWsModule.java

@@ -20,6 +20,9 @@
 package org.sonar.server.pushapi;
 
 import org.sonar.core.platform.Module;
+import org.sonar.server.pushapi.scheduler.purge.PushEventsPurgeExecutorServiceImpl;
+import org.sonar.server.pushapi.scheduler.purge.PushEventsPurgeInitializer;
+import org.sonar.server.pushapi.scheduler.purge.PushEventsPurgeSchedulerImpl;
 import org.sonar.server.pushapi.sonarlint.SonarLintClientPermissionsValidator;
 import org.sonar.server.pushapi.sonarlint.SonarLintClientsRegistry;
 import org.sonar.server.pushapi.sonarlint.SonarLintPushAction;
@@ -32,6 +35,11 @@ public class ServerPushWsModule extends Module {
       ServerPushWs.class,
       SonarLintClientPermissionsValidator.class,
       SonarLintClientsRegistry.class,
-      SonarLintPushAction.class);
+      SonarLintPushAction.class,
+
+      // Push Events Purge
+      PushEventsPurgeSchedulerImpl.class,
+      PushEventsPurgeExecutorServiceImpl.class,
+      PushEventsPurgeInitializer.class);
   }
 }

server/sonar-webserver-pushapi/src/main/java/org/sonar/server/pushapi/scheduler/purge/PushEventsPurgeExecutorService.java

@@ -0,0 +1,27 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2022 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.server.pushapi.scheduler.purge;
+
+import java.util.concurrent.ScheduledExecutorService;
+import org.sonar.api.server.ServerSide;
+
+@ServerSide
+public interface PushEventsPurgeExecutorService extends ScheduledExecutorService {
+}

server/sonar-webserver-pushapi/src/main/java/org/sonar/server/pushapi/scheduler/purge/PushEventsPurgeExecutorServiceImpl.java

@@ -0,0 +1,38 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2022 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.server.pushapi.scheduler.purge;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import org.sonar.server.util.AbstractStoppableScheduledExecutorServiceImpl;
+
+public class PushEventsPurgeExecutorServiceImpl
+  extends AbstractStoppableScheduledExecutorServiceImpl<ScheduledExecutorService>
+  implements PushEventsPurgeExecutorService {
+
+  public PushEventsPurgeExecutorServiceImpl() {
+    super(Executors.newSingleThreadScheduledExecutor(r -> {
+      Thread thread = Executors.defaultThreadFactory().newThread(r);
+      thread.setDaemon(true);
+      thread.setName(String.format("PushEvents-Purge-%d", System.nanoTime()));
+      return thread;
+    }));
+  }
+}

server/sonar-webserver-pushapi/src/main/java/org/sonar/server/pushapi/scheduler/purge/PushEventsPurgeInitializer.java

@@ -0,0 +1,41 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2022 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.server.pushapi.scheduler.purge;
+
+import javax.annotation.Nullable;
+import org.sonar.api.platform.Server;
+import org.sonar.api.platform.ServerStartHandler;
+
+public class PushEventsPurgeInitializer implements ServerStartHandler {
+
+  private final PushEventsPurgeScheduler pushEventsPurgeScheduler;
+
+  public PushEventsPurgeInitializer(@Nullable PushEventsPurgeScheduler pushEventsPurgeScheduler) {
+    this.pushEventsPurgeScheduler = pushEventsPurgeScheduler;
+  }
+
+  @Override
+  public void onServerStart(Server server) {
+    if (pushEventsPurgeScheduler != null) {
+      pushEventsPurgeScheduler.startScheduling();
+    }
+  }
+
+}

server/sonar-webserver-pushapi/src/main/java/org/sonar/server/pushapi/scheduler/purge/PushEventsPurgeScheduler.java

@@ -0,0 +1,27 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2022 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.server.pushapi.scheduler.purge;
+
+import org.sonar.api.server.ServerSide;
+
+@ServerSide
+public interface PushEventsPurgeScheduler {
+  void startScheduling();
+}

server/sonar-webserver-pushapi/src/main/java/org/sonar/server/pushapi/scheduler/purge/PushEventsPurgeSchedulerImpl.java

@@ -0,0 +1,98 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2022 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.server.pushapi.scheduler.purge;
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Set;
+import org.sonar.api.config.Configuration;
+import org.sonar.api.utils.System2;
+import org.sonar.api.utils.log.Logger;
+import org.sonar.api.utils.log.Loggers;
+import org.sonar.db.DbClient;
+import org.sonar.db.DbSession;
+import org.sonar.server.util.GlobalLockManager;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+public class PushEventsPurgeSchedulerImpl implements PushEventsPurgeScheduler {
+  private static final Logger LOG = Loggers.get(PushEventsPurgeSchedulerImpl.class);
+  private static final String LOCK_NAME = "PushEventsPurgeCheck";
+  private static final String INITIAL_DELAY_IN_SECONDS = "sonar.push.events.purge.initial.delay";
+  private static final String ENQUEUE_DELAY_IN_SECONDS = "sonar.push.events.purge.enqueue.delay";
+  private static final int ENQUEUE_LOCK_DELAY_IN_SECONDS = 60;
+
+  private final DbClient dbClient;
+  private final Configuration config;
+  private final GlobalLockManager lockManager;
+  private final PushEventsPurgeExecutorService executorService;
+  private final System2 system;
+
+
+  public PushEventsPurgeSchedulerImpl(DbClient dbClient, Configuration config, GlobalLockManager lockManager,
+    PushEventsPurgeExecutorService executorService, System2 system) {
+    this.dbClient = dbClient;
+    this.executorService = executorService;
+    this.config = config;
+    this.lockManager = lockManager;
+    this.system = system;
+  }
+
+  @Override
+  public void startScheduling() {
+    executorService.scheduleAtFixedRate(this::checks, getInitialDelay(),
+      getEnqueueDelay(), SECONDS);
+  }
+
+  private void checks() {
+    try {
+      // Avoid enqueueing push events purge task multiple times
+      if (!lockManager.tryLock(LOCK_NAME, ENQUEUE_LOCK_DELAY_IN_SECONDS)) {
+        return;
+      }
+      purgeExpiredPushEvents();
+    } catch (Exception e) {
+      LOG.error("Error in Push Events Purge scheduler", e);
+    }
+  }
+
+  private void purgeExpiredPushEvents() {
+    try (DbSession dbSession = dbClient.openSession(false)) {
+      Set<String> uuids = dbClient.pushEventDao().selectUuidsOfExpiredEvents(dbSession, getExpiredTimestamp());
+      LOG.debug(String.format("%s push events to be deleted...", uuids.size()));
+      dbClient.pushEventDao().deleteByUuids(dbSession, uuids);
+      dbSession.commit();
+    }
+  }
+
+  public long getInitialDelay() {
+    return config.getLong(INITIAL_DELAY_IN_SECONDS).orElse(60 * 60L);
+  }
+
+  public long getEnqueueDelay() {
+    return config.getLong(ENQUEUE_DELAY_IN_SECONDS).orElse(60 * 60L);
+  }
+
+  private long getExpiredTimestamp() {
+    return Instant.ofEpochMilli(system.now())
+      .minus(1, ChronoUnit.HOURS)
+      .toEpochMilli();
+  }
+}

server/sonar-webserver-pushapi/src/test/java/org/sonar/server/pushapi/scheduler/purge/PushEventsPurgeExecutorServiceImplTest.java

@@ -0,0 +1,35 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2022 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.server.pushapi.scheduler.purge;
+
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class PushEventsPurgeExecutorServiceImplTest {
+
+  @Test
+  public void constructor_createsProperDeleagateThatIsReadyToAct() {
+    PushEventsPurgeExecutorServiceImpl pushEventsPurgeExecutorService = new PushEventsPurgeExecutorServiceImpl();
+
+    assertThat(pushEventsPurgeExecutorService.isShutdown()).isFalse();
+    assertThat(pushEventsPurgeExecutorService.isTerminated()).isFalse();
+  }
+}

server/sonar-webserver-pushapi/src/test/java/org/sonar/server/pushapi/scheduler/purge/PushEventsPurgeInitializerTest.java

@@ -0,0 +1,46 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2022 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.server.pushapi.scheduler.purge;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class PushEventsPurgeInitializerTest {
+
+  private PushEventsPurgeScheduler pushEventsPurgeScheduler;
+
+  @Before
+  public void before() {
+    pushEventsPurgeScheduler = mock(PushEventsPurgeScheduler.class);
+  }
+
+  @Test
+  public void onServerStart_givenProjectReportScheduler_startScheduling() {
+    PushEventsPurgeInitializer pushEventsPurgeInitializer = new PushEventsPurgeInitializer(pushEventsPurgeScheduler);
+
+    pushEventsPurgeInitializer.onServerStart(null);
+
+    verify(pushEventsPurgeScheduler, times(1)).startScheduling();
+  }
+}

server/sonar-webserver-pushapi/src/test/java/org/sonar/server/pushapi/scheduler/purge/PushEventsPurgeSchedulerImplTest.java

@@ -0,0 +1,139 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2022 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.server.pushapi.scheduler.purge;
+
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.sonar.api.config.Configuration;
+import org.sonar.api.utils.System2;
+import org.sonar.db.DbClient;
+import org.sonar.db.DbSession;
+import org.sonar.db.pushevent.PushEventDao;
+import org.sonar.server.util.AbstractStoppableExecutorService;
+import org.sonar.server.util.GlobalLockManager;
+import org.sonar.server.util.GlobalLockManagerImpl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+public class PushEventsPurgeSchedulerImplTest {
+  private final DbClient dbClient = mock(DbClient.class);
+  private final DbSession dbSession = mock(DbSession.class);
+  private final GlobalLockManager lockManager = mock(GlobalLockManagerImpl.class);
+  private final PushEventDao pushEventDao = mock(PushEventDao.class);
+  private final PushEventsPurgeExecutorServiceImpl executorService = new PushEventsPurgeExecutorServiceImpl();
+  private final Configuration configuration = mock(Configuration.class);
+  private final System2 system2 = mock(System2.class);
+  private final PushEventsPurgeSchedulerImpl underTest = new PushEventsPurgeSchedulerImpl(dbClient, configuration,
+    lockManager, executorService, system2);
+
+  @Before
+  public void prepare() {
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
+  }
+
+  @Test
+  public void doNothingIfLocked() {
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(false);
+
+    underTest.startScheduling();
+
+    executorService.runCommand();
+
+    verifyNoInteractions(dbClient);
+  }
+
+  @Test
+  public void doNothingIfExceptionIsThrown() {
+    when(lockManager.tryLock(any(), anyInt())).thenThrow(new IllegalArgumentException("Oops"));
+
+    underTest.startScheduling();
+
+    executorService.runCommand();
+
+    verifyNoInteractions(dbClient);
+  }
+
+  @Test
+  public void schedulePurgeTaskWhenNotLocked() {
+    when(system2.now()).thenReturn(100000000L);
+    when(dbClient.pushEventDao()).thenReturn(pushEventDao);
+    when(dbClient.openSession(false)).thenReturn(dbSession);
+    when(dbClient.pushEventDao().selectUuidsOfExpiredEvents(any(), anyLong())).thenReturn(Set.of("1", "2"));
+
+    underTest.startScheduling();
+
+    executorService.runCommand();
+
+    ArgumentCaptor<Set> uuidsCaptor = ArgumentCaptor.forClass(Set.class);
+
+    verify(pushEventDao).deleteByUuids(any(), uuidsCaptor.capture());
+    Set<String> uuids = uuidsCaptor.getValue();
+    assertThat(uuids).containsExactlyInAnyOrder("1", "2");
+  }
+
+  private static class PushEventsPurgeExecutorServiceImpl extends AbstractStoppableExecutorService<ScheduledExecutorService>
+    implements PushEventsPurgeExecutorService {
+
+    private Runnable command;
+
+    public PushEventsPurgeExecutorServiceImpl() {
+      super(null);
+    }
+
+    public void runCommand() {
+      command.run();
+    }
+
+    @Override
+    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
+      this.command = command;
+      return null;
+    }
+
+    @Override
+    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+      return null;
+    }
+
+    @Override
+    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
+      return null;
+    }
+
+    @Override
+    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
+      return null;
+    }
+
+  }
+}