This is an automated email from the ASF dual-hosted git repository. quantranhong1999 pushed a commit to branch 3.9.x in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 7762f6085efd77e5df9f981319bf4bdf6db725fd Author: Benoit TELLIER <[email protected]> AuthorDate: Sun Aug 16 16:33:29 2026 +0700 [FIX] Task cleanup should purge stalled tasks --- .../webadmin/services/TasksCleanupService.java | 47 +++++++++++- .../webadmin/routes/TasksCleanupRoutesTest.java | 86 +++++++++++++++++++++- 2 files changed, 129 insertions(+), 4 deletions(-) diff --git a/server/protocols/webadmin/webadmin-cassandra/src/main/java/org/apache/james/webadmin/services/TasksCleanupService.java b/server/protocols/webadmin/webadmin-cassandra/src/main/java/org/apache/james/webadmin/services/TasksCleanupService.java index 6609083407..677908210c 100644 --- a/server/protocols/webadmin/webadmin-cassandra/src/main/java/org/apache/james/webadmin/services/TasksCleanupService.java +++ b/server/protocols/webadmin/webadmin-cassandra/src/main/java/org/apache/james/webadmin/services/TasksCleanupService.java @@ -20,6 +20,7 @@ package org.apache.james.webadmin.services; import java.time.Instant; +import java.time.ZonedDateTime; import java.util.concurrent.atomic.AtomicLong; import jakarta.inject.Inject; @@ -27,6 +28,7 @@ import jakarta.inject.Inject; import org.apache.commons.lang3.tuple.Pair; import org.apache.james.eventsourcing.eventstore.EventStore; import org.apache.james.task.Task; +import org.apache.james.task.TaskExecutionDetails; import org.apache.james.task.TaskId; import org.apache.james.task.TaskManager; import org.apache.james.task.eventsourcing.TaskAggregateId; @@ -108,8 +110,7 @@ public class TasksCleanupService { private Flux<Pair<TaskId, Task.Result>> removeTask(Instant beforeDate) { return Flux.from(taskExecutionDetailsProjection.listDetailsByBeforeDate(beforeDate)) - .filter(oldTaskDetail -> !(oldTaskDetail.getStatus().equals(TaskManager.Status.WAITING) - || oldTaskDetail.getStatus().equals(TaskManager.Status.IN_PROGRESS))) + .filterWhen(oldTaskDetail -> canBeRemoved(oldTaskDetail, beforeDate)) .flatMap(oldTaskDetail -> Mono.from(eventStore.remove(new TaskAggregateId(oldTaskDetail.getTaskId()))) .then(Mono.from(taskExecutionDetailsProjection.remove(oldTaskDetail))) .then(Mono.just(Pair.of(oldTaskDetail.getTaskId(), Task.Result.COMPLETED))) @@ -119,6 +120,48 @@ public class TasksCleanupService { })); } + private Mono<Boolean> canBeRemoved(TaskExecutionDetails taskDetail, Instant beforeDate) { + if (!isUnfinished(taskDetail)) { + return Mono.just(true); + } + if (isStale(taskDetail, beforeDate)) { + LOGGER.warn("Removing task {} of type {}, left in {} status with no activity since {}: " + + "the node running it died before completing it.", + taskDetail.getTaskId().asString(), taskDetail.getType().asString(), + taskDetail.getStatus().getValue(), lastActivity(taskDetail)); + return Mono.just(true); + } + // Still showing signs of life, yet a task without any event can not be running either: its history + // was lost. As cancelling it has no effect, removing it is the only way to get rid of such an entry. + return Mono.from(eventStore.getEventsOfAggregate(new TaskAggregateId(taskDetail.getTaskId()))) + .map(history -> history.getEventsJava().isEmpty()); + } + + private boolean isUnfinished(TaskExecutionDetails taskDetail) { + return taskDetail.getStatus().equals(TaskManager.Status.WAITING) + || taskDetail.getStatus().equals(TaskManager.Status.IN_PROGRESS); + } + + /** + * A task being executed keeps refreshing its execution details. One left untouched since the cleanup + * horizon lost the node running it, and would otherwise stay in the listing forever: such an entry can + * neither be cancelled nor deleted through any other route. + */ + private boolean isStale(TaskExecutionDetails taskDetail, Instant beforeDate) { + return lastActivity(taskDetail).isBefore(beforeDate); + } + + private Instant lastActivity(TaskExecutionDetails taskDetail) { + Instant lastKnownDate = taskDetail.getStartedDate() + .map(ZonedDateTime::toInstant) + .orElseGet(() -> taskDetail.getSubmittedDate().toInstant()); + + return taskDetail.getAdditionalInformation() + .map(TaskExecutionDetails.AdditionalInformation::timestamp) + .filter(lastKnownDate::isBefore) + .orElse(lastKnownDate); + } + private static void doOnNext(Pair<TaskId, Task.Result> next, Context context) { context.incrementProcessedTaskCount(); if (Task.Result.COMPLETED.equals(next.getValue())) { diff --git a/server/protocols/webadmin/webadmin-cassandra/src/test/java/org/apache/james/webadmin/routes/TasksCleanupRoutesTest.java b/server/protocols/webadmin/webadmin-cassandra/src/test/java/org/apache/james/webadmin/routes/TasksCleanupRoutesTest.java index cf91f44c08..fd9887210a 100644 --- a/server/protocols/webadmin/webadmin-cassandra/src/test/java/org/apache/james/webadmin/routes/TasksCleanupRoutesTest.java +++ b/server/protocols/webadmin/webadmin-cassandra/src/test/java/org/apache/james/webadmin/routes/TasksCleanupRoutesTest.java @@ -319,11 +319,12 @@ public class TasksCleanupRoutesTest { @ParameterizedTest @MethodSource(value = "inProgressStatus") - void tasksCleanupShouldNotRemoveInProgressTask(TaskManager.Status status) { + void tasksCleanupShouldRemoveStaleUnfinishedTask(TaskManager.Status status) { + // Not updated since the cleanup horizon: the node running it is gone. TaskExecutionDetails taskExecutionDetail = new TaskExecutionDetails(TaskId.generateTaskId(), TaskType.of("type"), status, - ZonedDateTime.now(), + ZonedDateTime.now().minus(20, ChronoUnit.DAYS), new Hostname("foo"), Optional::empty, Optional.empty(), @@ -333,6 +334,87 @@ public class TasksCleanupRoutesTest { Optional.empty(), Optional.empty()); + taskExecutionDetailsProjection.update(taskExecutionDetail); + Created event = new Created(new TaskAggregateId(taskExecutionDetail.taskId()), EventId.first(), new MemoryReferenceWithCounterTask((counter) -> Task.Result.COMPLETED), new Hostname("foo")); + Mono.from(eventStore.append(event)).block(); + + String taskId = given() + .queryParam("olderThan", "15day") + .delete() + .jsonPath() + .get("taskId"); + + given() + .basePath(TasksRoutes.BASE) + .when() + .get(taskId + "/await") + .then() + .body("status", is("completed")) + .body("taskId", is(taskId)) + .body("type", is("tasks-cleanup")) + .body("additionalInformation.removedTaskCount", is(1)) + .body("additionalInformation.processedTaskCount", is(1)); + + assertThat(taskExecutionDetailsProjection.list().size()) + .isEqualTo(0); + } + + @Test + void tasksCleanupShouldRemoveUnfinishedTaskWhenEventsAreMissing() { + // Still recently active, yet the event store holds no history for it: the entry can not be + // running, and neither cancelling nor deleting it individually gets rid of it. + TaskExecutionDetails taskExecutionDetail = new TaskExecutionDetails(TaskId.generateTaskId(), + TaskType.of("type"), + TaskManager.Status.IN_PROGRESS, + ZonedDateTime.now().minus(20, ChronoUnit.DAYS), + new Hostname("foo"), + Optional::empty, + Optional.of(ZonedDateTime.now()), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty()); + + taskExecutionDetailsProjection.update(taskExecutionDetail); + + String taskId = given() + .queryParam("olderThan", "15day") + .delete() + .jsonPath() + .get("taskId"); + + given() + .basePath(TasksRoutes.BASE) + .when() + .get(taskId + "/await") + .then() + .body("status", is("completed")) + .body("taskId", is(taskId)) + .body("type", is("tasks-cleanup")) + .body("additionalInformation.removedTaskCount", is(1)) + .body("additionalInformation.processedTaskCount", is(1)); + + assertThat(taskExecutionDetailsProjection.list().size()) + .isEqualTo(0); + } + + @Test + void tasksCleanupShouldNotRemoveInProgressTask() { + // Started recently: the task is genuinely being executed, whatever the age of its submission. + TaskExecutionDetails taskExecutionDetail = new TaskExecutionDetails(TaskId.generateTaskId(), + TaskType.of("type"), + TaskManager.Status.IN_PROGRESS, + ZonedDateTime.now().minus(20, ChronoUnit.DAYS), + new Hostname("foo"), + Optional::empty, + Optional.of(ZonedDateTime.now()), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty()); + taskExecutionDetailsProjection.update(taskExecutionDetail); TaskAggregateId taskAggregateId = new TaskAggregateId(taskExecutionDetail.taskId()); Created event = new Created(taskAggregateId, EventId.first(), new MemoryReferenceWithCounterTask((counter) -> Task.Result.COMPLETED), new Hostname("foo")); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
