Merge pull request #745 from apache/master

[pull] master from apache:master

Author: pull[bot]

server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java

@@ -189,7 +189,8 @@ private Mono<Task.Result> executeTask(AcknowledgableDelivery delivery) {
             .flatMap(taskId -> Mono.fromCallable(() -> new String(delivery.getBody(), StandardCharsets.UTF_8))
                 .flatMap(bodyValue -> deserialize(bodyValue, taskId))
                 .doOnNext(task -> delivery.ack())
-                .flatMap(task -> executeOnWorker(taskId, task)))
+                .flatMap(task -> executeOnWorker(taskId, task))
+                .doOnSuccess(result -> LOGGER.info("Executed task {} yield {}", taskId, result)))
             .onErrorResume(error -> {
                 Optional<Object> taskId = Optional.ofNullable(delivery.getProperties())
                     .flatMap(props -> Optional.ofNullable(props.getHeaders()))
@@ -211,6 +212,7 @@ private Mono<Task> deserialize(String json, TaskId taskId) {
     }
 
     private Mono<Task.Result> executeOnWorker(TaskId taskId, Task task) {
+        LOGGER.info("Executing task {} ({}) ", taskId, task.getClass());
         return worker.executeTask(new TaskWithId(taskId, task))
             .timeout(rabbitMQConfiguration.getTaskQueueConsumerTimeout())
             .onErrorResume(error -> {
@@ -274,14 +276,22 @@ public void submit(TaskWithId taskWithId) {
                 .build();
 
             OutboundMessage outboundMessage = new OutboundMessage(EXCHANGE_NAME, ROUTING_KEY, basicProperties, payload);
-            sender.send(Mono.just(outboundMessage)).block();
+            sender.send(Mono.just(outboundMessage))
+                .onErrorResume(e -> {
+                    LOGGER.error("Publishing task {} failed", taskWithId.getId(), e);
+                    return Mono.from(worker.fail(taskWithId.getId(), Mono.just(Optional.empty()), "Publishing task failed", e))
+                        .then(Mono.error(e));
+                })
+                .block();
+            LOGGER.info("Submitted task {} ({})", taskWithId.getId(), taskWithId.getTask().getClass());
         } catch (JsonProcessingException e) {
             throw new RuntimeException(e);
         }
     }
 
     @Override
     public void cancel(TaskId taskId) {
+        LOGGER.info("Requesting cancel for task {}", taskId);
         sendCancelRequestsQueue.emitNext(taskId, FAIL_FAST);
     }
 

src/adr/0075-deleted-message-vault.md

@@ -0,0 +1,63 @@
+# 75. Deletes Message Vault
+
+Date: 2025-12-19
+
+## Status
+
+Accepted (lazy consensus) & implemented.
+
+This ADR is written at posteriori in order to capture knowlege of the team following the write up 
+of [0074-dedicated-eventbus-for-message-content-deletion.md](0074-dedicated-eventbus-for-message-content-deletion.md)
+
+## Context
+
+In standard email systems, when a message is deleted—whether by a user, an administrator, or automatically—it is often permanently lost. This creates several 
+problems in environments where email is a critical business record.
+
+As such we want a mechanism that protects against:
+
+ - **Accidental or malicious deletion**: Users can accidentally delete important emails, or malicious actors may intentionally remove messages to hide evidence. Recovery 
+shall be possible.
+ - **Legal and regulatory compliance**: Many organizations must comply with regulations that require retention of business communications and the ability to produce 
+deleted emails during audits or legal discovery.
+ - **Administrator control and traceability**: administrator needs to control and understand destructive actions done on user account (date of deletion and what had been deleted).
+
+However we want a clear **separation of user experience and data retention**. Users expect that “delete” means the message disappears from their mailbox.
+
+However incident recovery, and Recovery Point Objectives are a non objective that shall be addressed through global database level backups.
+
+## Decision
+
+Provide a James mailbox plugin, bundled in Guice application, the **Deleted Message Vault**.
+
+Provide an implementation of this vault atop the object store, which stores data in a time organized fashion (per month) in a dedicated buckets. Minimal 
+metadata are to be kept onto the Cassandra / Postgres  database.
+
+Provide a webadmin endpoints for restoring user data, with a minimal and simple fliter logic to allow restoring specific content. This action is not exposed to the end user.
+
+Provide a webadmin endpoint to access deleted messages if need be.
+
+Provide a webadmin endpoint to delete the vault data that no longer needs to be retained.
+
+Plug this vault onto the mailbox deletion process. We leverage [0029-Cassandra-mailbox-deletion-cleanup.md](0029-Cassandra-mailbox-deletion-cleanup.md) asynchronous deletion listener as well as
+the [0074-dedicated-eventbus-for-message-content-deletion.md](0074-dedicated-eventbus-for-message-content-deletion.md) dedicated event bus for effective content deletion to do so. Please
+note that blob deduplication mentioned in [0049-deduplicated-blobs-gs-with-bloom-filters.md](0049-deduplicated-blobs-gs-with-bloom-filters.md) needs to be active. 
+
+## Consequences
+
+Aforementioned objectives are attained.
+
+An extra copy is needed upon deletes which can be expensive. That is why we needed [0074-dedicated-eventbus-for-message-content-deletion.md](0074-dedicated-eventbus-for-message-content-deletion.md) 
+especially upon large mailbox deletion.
+
+Disk space impact:
+ - Deleted message vault is not counted onto the user quota
+ - Content is not deduplicated onto the deleted message vault
+
+# References 
+
+ - [0029-Cassandra-mailbox-deletion-cleanup.md](0029-Cassandra-mailbox-deletion-cleanup.md)
+ - [0049-deduplicated-blobs-gs-with-bloom-filters.md](0049-deduplicated-blobs-gs-with-bloom-filters.md)
+ - [0074-dedicated-eventbus-for-message-content-deletion.md](0074-dedicated-eventbus-for-message-content-deletion.md)
+ - [Asynchronous deletions with DeletedMessageVault on top of Cassandra](https://issues.apache.org/jira/browse/JAMES-3882)
+