wait for tasks to complete when shutting down

commafeed-server/src/main/java/com/commafeed/CommaFeedApplication.java

@@ -10,7 +10,9 @@ import com.commafeed.security.password.PasswordConstraintValidator;
 import io.quarkus.runtime.ShutdownEvent;
 import io.quarkus.runtime.StartupEvent;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 
+@Slf4j
 @Singleton
 @RequiredArgsConstructor
 public class CommaFeedApplication {
@@ -27,6 +29,8 @@ public class CommaFeedApplication {
 	}
 
 	public void stop(@Observes ShutdownEvent ev) {
+		log.info("shutting down...");
+
 		feedRefreshEngine.stop();
 		taskScheduler.stop();
 	}

commafeed-server/src/main/java/com/commafeed/CommaFeedConfiguration.java

@@ -92,6 +92,12 @@ public interface CommaFeedConfiguration {
 	@ConfigDocSection
 	Websocket websocket();
 
+	/**
+	 * Duration to wait for the feed refresh engine and the task scheduler to stop when the application is shutting down.
+	 */
+	@WithDefault("2s")
+	Duration shutdownTimeout();
+
 	interface HttpClient {
 		/**
 		 * User-Agent string that will be used by the http client, leave empty for the default one.

commafeed-server/src/main/java/com/commafeed/backend/feed/FeedRefreshEngine.java

@@ -24,6 +24,7 @@ import com.commafeed.backend.model.AbstractModel;
 import com.commafeed.backend.model.Feed;
 import com.commafeed.backend.model.FeedEntry;
 import com.commafeed.backend.model.FeedSubscription;
+import com.google.common.util.concurrent.MoreExecutors;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -197,12 +198,12 @@ public class FeedRefreshEngine {
 	}
 
 	public void stop() {
-		this.feedProcessingLoopExecutor.shutdownNow();
+		MoreExecutors.shutdownAndAwaitTermination(this.feedProcessingLoopExecutor, config.shutdownTimeout());
-		this.refillLoopExecutor.shutdownNow();
+		MoreExecutors.shutdownAndAwaitTermination(this.refillLoopExecutor, config.shutdownTimeout());
-		this.refillExecutor.shutdownNow();
+		MoreExecutors.shutdownAndAwaitTermination(this.refillExecutor, config.shutdownTimeout());
-		this.workerExecutor.shutdownNow();
+		MoreExecutors.shutdownAndAwaitTermination(this.workerExecutor, config.shutdownTimeout());
-		this.databaseUpdaterExecutor.shutdownNow();
+		MoreExecutors.shutdownAndAwaitTermination(this.databaseUpdaterExecutor, config.shutdownTimeout());
-		this.notifierExecutor.shutdownNow();
+		MoreExecutors.shutdownAndAwaitTermination(this.notifierExecutor, config.shutdownTimeout());
 	}
 
 	/**

commafeed-server/src/main/java/com/commafeed/backend/service/db/DatabaseCleaningService.java

@@ -54,10 +54,20 @@ public class DatabaseCleaningService {
 		int deleted;
 		long entriesTotal = 0;
 		do {
+			if (Thread.currentThread().isInterrupted()) {
+				log.info("interrupted, stopping cleanup of feeds without subscriptions");
+				return;
+			}
+
 			List<Feed> feeds = unitOfWork.call(() -> feedDAO.findWithoutSubscriptions(1));
 			for (Feed feed : feeds) {
 				long entriesDeleted;
 				do {
+					if (Thread.currentThread().isInterrupted()) {
+						log.info("interrupted, stopping cleanup of feeds without subscriptions");
+						return;
+					}
+
 					entriesDeleted = unitOfWork.call(() -> feedEntryDAO.delete(feed.getId(), batchSize));
 					entriesDeletedMeter.mark(entriesDeleted);
 					entriesTotal += entriesDeleted;
@@ -76,6 +86,11 @@ public class DatabaseCleaningService {
 		long total = 0;
 		long deleted;
 		do {
+			if (Thread.currentThread().isInterrupted()) {
+				log.info("interrupted, stopping cleanup of contents without entries");
+				return;
+			}
+
 			deleted = unitOfWork.call(() -> feedEntryContentDAO.deleteWithoutEntries(batchSize));
 			total += deleted;
 			log.debug("removed {} contents without entries", total);
@@ -87,6 +102,11 @@ public class DatabaseCleaningService {
 		log.info("cleaning entries exceeding feed capacity");
 		long total = 0;
 		while (true) {
+			if (Thread.currentThread().isInterrupted()) {
+				log.info("interrupted, stopping cleanup of entries exceeding feed capacity");
+				return;
+			}
+
 			List<FeedCapacity> feeds = unitOfWork
 					.call(() -> feedEntryDAO.findFeedsExceedingCapacity(maxFeedCapacity, batchSize, keepStarredEntries));
 			if (feeds.isEmpty()) {
@@ -97,6 +117,11 @@ public class DatabaseCleaningService {
 				long remaining = feed.capacity() - maxFeedCapacity;
 				int deleted;
 				do {
+					if (Thread.currentThread().isInterrupted()) {
+						log.info("interrupted, stopping cleanup of entries exceeding feed capacity");
+						return;
+					}
+
 					final long rem = remaining;
 					deleted = unitOfWork.call(() -> feedEntryDAO.deleteOldEntries(feed.id(), Math.min(batchSize, rem), keepStarredEntries));
 					entriesDeletedMeter.mark(deleted);
@@ -114,6 +139,11 @@ public class DatabaseCleaningService {
 		long total = 0;
 		long deleted;
 		do {
+			if (Thread.currentThread().isInterrupted()) {
+				log.info("interrupted, stopping cleanup of old entries");
+				return;
+			}
+
 			deleted = unitOfWork.call(() -> feedEntryDAO.deleteEntriesOlderThan(olderThan, batchSize, keepStarredEntries));
 			entriesDeletedMeter.mark(deleted);
 			total += deleted;
@@ -127,6 +157,11 @@ public class DatabaseCleaningService {
 		long total = 0;
 		long deleted;
 		do {
+			if (Thread.currentThread().isInterrupted()) {
+				log.info("interrupted, stopping cleanup of old read statuses");
+				return;
+			}
+
 			deleted = unitOfWork.call(() -> feedEntryStatusDAO.deleteOldStatuses(olderThan, batchSize));
 			total += deleted;
 			log.debug("removed {} old read statuses", total);
@@ -139,6 +174,11 @@ public class DatabaseCleaningService {
 		long total = 0;
 		long marked;
 		do {
+			if (Thread.currentThread().isInterrupted()) {
+				log.info("interrupted, stopping marking entries as read");
+				return;
+			}
+
 			marked = unitOfWork.call(() -> feedEntryStatusDAO.autoMarkAsRead(batchSize));
 			total += marked;
 			log.debug("marked {} entries as read", total);

commafeed-server/src/main/java/com/commafeed/backend/task/TaskScheduler.java

@@ -6,16 +6,23 @@ import java.util.concurrent.ScheduledExecutorService;
 
 import jakarta.inject.Singleton;
 
-import io.quarkus.arc.All;
+import com.commafeed.CommaFeedConfiguration;
+import com.google.common.util.concurrent.MoreExecutors;
 
+import io.quarkus.arc.All;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
 @Singleton
 public class TaskScheduler {
 
 	private final List<ScheduledTask> tasks;
+	private final CommaFeedConfiguration config;
 	private final ScheduledExecutorService executor;
 
-	public TaskScheduler(@All List<ScheduledTask> tasks) {
+	public TaskScheduler(@All List<ScheduledTask> tasks, CommaFeedConfiguration config) {
 		this.tasks = tasks;
+		this.config = config;
 		this.executor = Executors.newScheduledThreadPool(tasks.size());
 	}
 
@@ -24,6 +31,6 @@ public class TaskScheduler {
 	}
 
 	public void stop() {
-		executor.shutdownNow();
+		MoreExecutors.shutdownAndAwaitTermination(executor, config.shutdownTimeout());
 	}
 }