Queue syncs in RabbitMQ with retries and a dead-letter queue

README.md

@@ -79,11 +79,27 @@ Both carry `occurredAt`, when the change was made: events about one project can
 Listeners use `@TransactionalEventListener`, so they only see committed changes. `SyncOnProjectChange`
 starts a background sync (as `POST /api/v2/projects/{id}/sync` does) when a project is created or re-keyed
 (pointed at another SDLC project), so its name comes from the SDLC system without the client asking. A sync
-that can't start right then (one is running, rate limited, no free background slot) is logged and skipped.
-The listener runs on the request thread after commit; it only starts the sync, but with Redis unreachable the
-lock attempt adds about a second to the request (see Lock below).
+that can't start right then (one is running, rate limited, RabbitMQ unreachable) is logged and skipped.
+The listener runs on the request thread after commit; it only queues the sync, but with Redis or RabbitMQ
+unreachable the failed attempt adds about a second (Redis) or up to `spring.rabbitmq.connection-timeout` (2s)
+to the request.
 Turn it off with `resourceapi.sync.on-change=false` (tests do, except `SyncOnProjectChangeTest`).
 
+## Sync queue (RabbitMQ)
+`POST /api/v2/projects/{id}/sync` queues the sync in the durable RabbitMQ queue `resourceapi.project-sync`
+(`resourceapi.sync.queue`) and answers `202 Accepted` once the broker confirmed it has the request; a consumer
+(`ProjectSyncConsumer`) refreshes the project's name from its SDLC system. Requests survive restarts, and a burst
+of them waits in the queue instead of being refused. If RabbitMQ is unreachable, the request gets `503` with
+`Retry-After`.
+
+- Each instance runs `spring.rabbitmq.listener.simple.concurrency` (4) consumers on virtual threads, taking one
+  request at a time (`prefetch=1`), so waiting requests go to whichever instance is free.
+- A sync failing for a reason that may pass (SDLC system unreachable, `5xx` or `429`, our own SDLC rate limit) is
+  retried after 1s and 2s (`resourceapi.sync.max-retries`, `retry-delay`, `retry-multiplier`); a sync that still
+  fails, or fails for another reason (e.g. `403`), goes to the dead-letter queue `<queue>.dead`, where it can be
+  inspected and moved back (e.g. with the management UI's shovel). A project deleted in the meantime is skipped.
+- The project's sync lock (see Redis below) is released however the sync ends.
+
 ## Notifications (RabbitMQ)
 Other systems learn about project changes from RabbitMQ (`spring.rabbitmq.*`, default `localhost:5672`):
 `ProjectCreated` and `ProjectUpdated` are sent as JSON to the topic exchange `resourceapi.projects`
@@ -131,9 +147,10 @@ optimization: if Redis is unreachable, cache reads and writes are logged and ski
 500 ms Redis timeouts) and requests are served without it.
 
 **Lock** ([Redisson](https://redisson.org/) `RLock`): at most one `POST /api/v2/projects/{id}/sync` per project
-runs at a time, across all instances; a second one gets `409 Conflict` until the first ends. The lock is
-released when the background sync ends and expires after `resourceapi.sync.lock-lease-time` (60s) in case
-an instance dies mid-sync. If Redis can't be reached, the sync runs without the lock (a duplicate sync only
+is queued or running at a time, across all instances; a second one gets `409 Conflict` until the first ends. The
+lock is taken by the request and released by whichever instance runs the sync when it ends (the request carries
+the lock's owner id), and expires after `resourceapi.sync.lock-lease-time` (60s) in case an instance dies
+mid-sync or the request waits in the queue longer than that. If Redis can't be reached, the sync runs without the lock (a duplicate sync only
 costs an extra SDLC call; the failed lock request takes about a second). Redisson connects lazily and uses the
 same Redis connection settings; its keys are prefixed with `resourceapi.redisson.key-prefix`. Only a
 standalone Redis is supported for the lock: with Sentinel or Cluster the application still starts and syncs
@@ -146,7 +163,7 @@ available again one period later, which is also what `Retry-After` says):
 | Limit | Default | When exceeded |
 |---|---|---|
 | Calls to one SDLC system (`resourceapi.rate-limit.sdlc.*`), protecting its API quota; cache hits don't count | 10 per 1s, waiting up to 1s for a permit | `503` + `Retry-After` (our own limit, not the SDLC system's failure); a background sync logs it |
-| `POST /api/v2/projects/{id}/sync` per project (`resourceapi.rate-limit.sync.*`); requests refused anyway (unknown project, sync already running) don't count, one refused because no background slot is free (`503`) does | 6 per minute | `429` + `Retry-After` |
+| `POST /api/v2/projects/{id}/sync` per project (`resourceapi.rate-limit.sync.*`); requests refused anyway (unknown project, sync already running) don't count, one refused because RabbitMQ is unreachable (`503`) does | 6 per minute | `429` + `Retry-After` |
 | All `/api/**` requests per client IP (`resourceapi.rate-limit.client.*`); behind a reverse proxy the IP comes from `X-Forwarded-For`, trusted only from internal proxy addresses (`server.forward-headers-strategy=native`, see below) | 100 per second | `429` + `Retry-After` |
 
 A changed rate is applied to the existing limiter when an instance first uses it; unused limiters expire from
@@ -176,10 +193,9 @@ Every request gets a trace id ([Micrometer Tracing](https://micrometer.io/docs/t
 - outgoing `RestClient` calls (e.g. to SDLC systems) send `traceparent`, so the trace continues there
 
 - background work keeps the trace: `@Async` tasks run on virtual threads (`spring.threads.virtual.enabled`) and
-  `ContextPropagatingTaskDecorator` copies the caller's trace context to them. Example:
-  `POST /api/v2/projects/{id}/sync` answers `202 Accepted` and refreshes the project's name from its SDLC system
-  in the background; its log lines and its call to the SDLC system carry the request's trace id. At most
-  `spring.task.execution.simple.concurrency-limit` (8) syncs run at once; further requests get `503` with `Retry-After`
+  `ContextPropagatingTaskDecorator` copies the caller's trace context to them; RabbitMQ messages carry it in a
+  `traceparent` header (Spring AMQP observation) and their listeners continue it. Example: a queued sync's log
+  lines and its call to the SDLC system carry the trace id of the `POST …/sync` request
 
 No spans are exported; add a reporter (e.g. Zipkin or OTLP) to send them to a tracing backend, and lower
 `management.tracing.sampling.probability` (set to `1.0` here) to what that backend can take.

src/main/java/com/company/resourceapi/config/ProjectSyncQueueConfig.java

@@ -0,0 +1,78 @@
+package com.company.resourceapi.config;
+
+import com.company.resourceapi.services.ProjectSyncRequest;
+import java.time.Duration;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueBuilder;
+import org.springframework.amqp.rabbit.config.ContainerCustomizer;
+import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.aot.hint.annotation.RegisterReflectionForBinding;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.context.properties.bind.DefaultValue;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Durable queue of sync requests: a request survives restarts and waits for a free consumer instead of
+ * being refused. A sync that still fails after its retries goes to the dead-letter queue
+ * ({@code <queue>.dead}), where it can be inspected and moved back.
+ */
+@Configuration
+@EnableConfigurationProperties(ProjectSyncQueueConfig.SyncQueueProperties.class)
+// requests are sent and read as JSON
+@RegisterReflectionForBinding(ProjectSyncRequest.class)
+public class ProjectSyncQueueConfig {
+
+    /** Declared on the broker when the application first connects. */
+    @Bean
+    Queue projectSyncQueue(SyncQueueProperties properties) {
+        return QueueBuilder.durable(properties.queue())
+                // through the default exchange, straight to the dead-letter queue
+                .deadLetterExchange("")
+                .deadLetterRoutingKey(properties.deadLetterQueue())
+                .build();
+    }
+
+    @Bean
+    Queue projectSyncDeadLetterQueue(SyncQueueProperties properties) {
+        return QueueBuilder.durable(properties.deadLetterQueue()).build();
+    }
+
+    /**
+     * A message the listener deliberately dead-letters (it logged why) isn't logged again with a stack
+     * trace; any other listener failure still is.
+     */
+    @Bean
+    ContainerCustomizer<SimpleMessageListenerContainer> deadLetteringLogging() {
+        return container -> container.setErrorHandler(new ConditionalRejectingErrorHandler() {
+            @Override
+            protected void log(Throwable failure) {
+                if (causeChainContainsARADRE(failure)) {
+                    logger.debug("Message dead-lettered: " + failure.getMessage());
+                } else {
+                    super.log(failure);
+                }
+            }
+        });
+    }
+
+    /**
+     * @param queue name of the sync request queue
+     * @param maxRetries retries of a sync that failed for a reason that may pass (SDLC system
+     *     unreachable or failing, our own rate limit)
+     * @param retryDelay wait before the first retry, multiplied by {@code retryMultiplier} for each further one
+     */
+    @ConfigurationProperties("resourceapi.sync")
+    public record SyncQueueProperties(
+            @DefaultValue("resourceapi.project-sync") String queue,
+            @DefaultValue("2") long maxRetries,
+            @DefaultValue("1s") Duration retryDelay,
+            @DefaultValue("2") double retryMultiplier) {
+
+        public String deadLetterQueue() {
+            return queue + ".dead";
+        }
+    }
+}

src/main/java/com/company/resourceapi/services/ProjectSyncConsumer.java

@@ -0,0 +1,79 @@
+package com.company.resourceapi.services;
+
+import com.company.resourceapi.config.ProjectSyncQueueConfig.SyncQueueProperties;
+import com.company.resourceapi.exceptions.NotFoundException;
+import com.company.resourceapi.exceptions.RateLimitExceededException;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.AmqpRejectAndDontRequeueException;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.core.retry.RetryException;
+import org.springframework.core.retry.RetryPolicy;
+import org.springframework.core.retry.RetryTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestClientException;
+import org.springframework.web.client.RestClientResponseException;
+
+/**
+ * Runs queued syncs ({@code spring.rabbitmq.listener.simple.*}: virtual threads, in the trace of the
+ * request that queued them). A sync failing for a reason that may pass is retried with a growing
+ * delay; one that still fails goes to the dead-letter queue. The project's sync lock is released
+ * however the sync ends.
+ */
+@Component
+@Slf4j
+public class ProjectSyncConsumer {
+
+    private final ProjectSyncWorker projectSyncWorker;
+    private final ProjectSyncLock projectSyncLock;
+    private final RetryTemplate retryTemplate;
+
+    public ProjectSyncConsumer(ProjectSyncWorker projectSyncWorker, ProjectSyncLock projectSyncLock,
+            SyncQueueProperties properties) {
+        this.projectSyncWorker = projectSyncWorker;
+        this.projectSyncLock = projectSyncLock;
+        this.retryTemplate = new RetryTemplate(RetryPolicy.builder()
+                .maxRetries(properties.maxRetries())
+                .delay(properties.retryDelay())
+                .multiplier(properties.retryMultiplier())
+                .predicate(ProjectSyncConsumer::mayPass)
+                .build());
+    }
+
+    @RabbitListener(queues = "${resourceapi.sync.queue:resourceapi.project-sync}")
+    public void onSyncRequested(ProjectSyncRequest request) {
+        long projectId = request.projectId();
+        try {
+            retryTemplate.execute(() -> {
+                projectSyncWorker.sync(projectId);
+                return null;
+            });
+        } catch (RetryException e) {
+            var failure = e.getLastException();
+            if (failure instanceof NotFoundException) {
+                log.info("Project {} was deleted before its sync ran", projectId);
+                return;
+            }
+            log.warn("Syncing project {} from its SDLC system failed: {}", projectId, describe(failure));
+            // not requeued (it would fail again right away): kept in the dead-letter queue. Without the
+            // failure as cause: logged above, and an error response's message includes the upstream body
+            throw new AmqpRejectAndDontRequeueException("Sync of project " + projectId + " dead-lettered");
+        } finally {
+            projectSyncLock.lease(projectId, request.lockOwnerId()).release();
+        }
+    }
+
+    /** The SDLC system failing or unreachable, or our own rate limit for it: worth trying again. */
+    private static boolean mayPass(Throwable failure) {
+        if (failure instanceof RestClientResponseException response) {
+            return response.getStatusCode().is5xxServerError() || response.getStatusCode().value() == 429;
+        }
+        return failure instanceof RestClientException || failure instanceof RateLimitExceededException;
+    }
+
+    /** The status only: an error response's message would include the upstream body. */
+    private static String describe(Throwable failure) {
+        return failure instanceof RestClientResponseException response
+                ? "HTTP " + response.getStatusCode().value()
+                : failure.getClass().getSimpleName() + ": " + failure.getMessage();
+    }
+}

src/main/java/com/company/resourceapi/services/ProjectSyncLock.java

@@ -55,11 +55,29 @@ public Optional<Lease> tryAcquire(long projectId) {
         }
     }
 
+    /**
+     * The lease taken by {@link #tryAcquire}, possibly in another instance: a sync's lock is taken
+     * where it is requested and released where it runs.
+     *
+     * @param ownerId the lease's {@link Lease#ownerIdOrNull()}
+     */
+    public Lease lease(long projectId, Long ownerId) {
+        if (redisson == null || ownerId == null) {
+            return Lease.NONE;
+        }
+        return new Lease(redisson.getLock("project-sync:" + projectId), ownerId);
+    }
+
     /** A lock that may be held; releasing one that isn't (anymore) only logs. */
     public record Lease(RLock lock, long ownerId) {
 
         static final Lease NONE = new Lease(null, 0);
 
+        /** @return the owner id to hand to whoever releases the lock, null for a sync without the lock */
+        public Long ownerIdOrNull() {
+            return lock == null ? null : ownerId;
+        }
+
         public void release() {
             if (lock == null) {
                 return;

src/main/java/com/company/resourceapi/services/ProjectSyncRequest.java

@@ -0,0 +1,10 @@
+package com.company.resourceapi.services;
+
+/**
+ * A queued sync of a project.
+ *
+ * @param lockOwnerId owner of the project's sync lock, taken by the request and released by the
+ *     consumer when the sync ends; null if the sync runs without the lock
+ */
+public record ProjectSyncRequest(long projectId, Long lockOwnerId) {
+}