From 884807eeb8b41958267f4a17055f101a9ae5634a Mon Sep 17 00:00:00 2001 From: andreasarf <68648816+andreasarf@users.noreply.github.com> Date: Sun, 1 Nov 2026 05:51:23 +0000 Subject: [PATCH 1/2] chore: update JaCoCo coverage badges --- .github/badges/branches.svg | 2 +- .github/badges/jacoco.svg | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/badges/branches.svg b/.github/badges/branches.svg index 2cfc1e4..b94bd5f 100644 --- a/.github/badges/branches.svg +++ b/.github/badges/branches.svg @@ -1 +1 @@ -branches88.2% \ No newline at end of file +branches84.8% \ No newline at end of file diff --git a/.github/badges/jacoco.svg b/.github/badges/jacoco.svg index c730bee..3e267a3 100644 --- a/.github/badges/jacoco.svg +++ b/.github/badges/jacoco.svg @@ -1 +1 @@ -coverage97.8% \ No newline at end of file +coverage98.2% \ No newline at end of file From 8b8113aca7079e40faf36220b186b32659846d84 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 1 Nov 2026 06:20:43 +0000 Subject: [PATCH 2/2] Queue syncs in RabbitMQ with retries and a dead-letter queue POST /sync now puts a ProjectSyncRequest on the durable queue resourceapi.project-sync (202 once the broker confirmed it, 503 if RabbitMQ is unreachable) instead of running it on the in-memory @Async executor. Requests survive restarts and wait for a free consumer instead of being refused. ProjectSyncConsumer (4 consumers per instance, virtual threads, prefetch 1, trace continued from the request) runs the sync with Spring's RetryTemplate: SDLC failures that may pass (unreachable, 5xx, 429, our own rate limit) are retried after 1s and 2s; a sync that still fails, or fails otherwise, is dead-lettered to .dead and logged in one line, without the upstream body. The per-project lock is taken by the request and released by the consumer via the owner id carried in the message. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012MLyPXGHeCc1fvTeB8JnT2 --- README.md | 38 ++-- .../config/ProjectSyncQueueConfig.java | 78 ++++++++ .../services/ProjectSyncConsumer.java | 79 ++++++++ .../resourceapi/services/ProjectSyncLock.java | 18 ++ .../services/ProjectSyncRequest.java | 10 + .../services/ProjectSyncService.java | 30 ++- .../services/ProjectSyncWorker.java | 79 ++++---- .../services/SyncOnProjectChange.java | 5 +- .../resourceapi/web/ApiExceptionHandler.java | 8 + src/main/resources/application.properties | 20 +- .../RabbitTestContainerInitializer.java | 10 +- .../services/ProjectSyncBackpressureTest.java | 81 -------- .../services/ProjectSyncQueueTest.java | 179 ++++++++++++++++++ .../services/RabbitUnavailableTest.java | 50 +++++ .../resources/config/application.properties | 2 + 15 files changed, 529 insertions(+), 158 deletions(-) create mode 100644 src/main/java/com/company/resourceapi/config/ProjectSyncQueueConfig.java create mode 100644 src/main/java/com/company/resourceapi/services/ProjectSyncConsumer.java create mode 100644 src/main/java/com/company/resourceapi/services/ProjectSyncRequest.java delete mode 100644 src/test/java/com/company/resourceapi/services/ProjectSyncBackpressureTest.java create mode 100644 src/test/java/com/company/resourceapi/services/ProjectSyncQueueTest.java create mode 100644 src/test/java/com/company/resourceapi/services/RabbitUnavailableTest.java diff --git a/README.md b/README.md index 4763c8d..74c48ea 100644 --- a/README.md +++ b/README.md @@ -76,11 +76,27 @@ changes a project: 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 `.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` @@ -125,9 +141,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 @@ -140,7 +157,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 @@ -170,10 +187,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. diff --git a/src/main/java/com/company/resourceapi/config/ProjectSyncQueueConfig.java b/src/main/java/com/company/resourceapi/config/ProjectSyncQueueConfig.java new file mode 100644 index 0000000..d375b2d --- /dev/null +++ b/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 .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 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"; + } + } +} diff --git a/src/main/java/com/company/resourceapi/services/ProjectSyncConsumer.java b/src/main/java/com/company/resourceapi/services/ProjectSyncConsumer.java new file mode 100644 index 0000000..de5884f --- /dev/null +++ b/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(); + } +} diff --git a/src/main/java/com/company/resourceapi/services/ProjectSyncLock.java b/src/main/java/com/company/resourceapi/services/ProjectSyncLock.java index 76c58f2..1ec7bbe 100644 --- a/src/main/java/com/company/resourceapi/services/ProjectSyncLock.java +++ b/src/main/java/com/company/resourceapi/services/ProjectSyncLock.java @@ -55,11 +55,29 @@ public Optional 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; diff --git a/src/main/java/com/company/resourceapi/services/ProjectSyncRequest.java b/src/main/java/com/company/resourceapi/services/ProjectSyncRequest.java new file mode 100644 index 0000000..6a5dfd3 --- /dev/null +++ b/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) { +} diff --git a/src/main/java/com/company/resourceapi/services/ProjectSyncService.java b/src/main/java/com/company/resourceapi/services/ProjectSyncService.java index 06ab4b0..62cccbb 100644 --- a/src/main/java/com/company/resourceapi/services/ProjectSyncService.java +++ b/src/main/java/com/company/resourceapi/services/ProjectSyncService.java @@ -1,31 +1,38 @@ package com.company.resourceapi.services; +import com.company.resourceapi.config.ProjectSyncQueueConfig.SyncQueueProperties; import com.company.resourceapi.config.RateLimitProperties; import com.company.resourceapi.exceptions.NotFoundException; import com.company.resourceapi.exceptions.RateLimitExceededException; import com.company.resourceapi.exceptions.RateLimitExceededException.Scope; import com.company.resourceapi.exceptions.SyncInProgressException; +import java.time.Duration; import lombok.RequiredArgsConstructor; +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.stereotype.Service; -/** Starts background syncs, at most one per project at a time across all instances. */ +/** Queues syncs, at most one per project at a time across all instances. */ @Service @RequiredArgsConstructor public class ProjectSyncService { + private static final Duration CONFIRM_TIMEOUT = Duration.ofSeconds(5); + private final ProjectService projectService; private final ProjectSyncLock projectSyncLock; - private final ProjectSyncWorker projectSyncWorker; private final RateLimiters rateLimiters; private final RateLimitProperties rateLimits; + private final RabbitTemplate rabbitTemplate; + private final SyncQueueProperties syncQueue; /** - * Starts refreshing the project from its SDLC system in the background. + * Queues refreshing the project from its SDLC system; {@link ProjectSyncConsumer} runs it. * * @throws NotFoundException if the project doesn't exist * @throws RateLimitExceededException if the project was synced too often recently - * @throws SyncInProgressException if a sync of the project is still running - * @throws org.springframework.core.task.TaskRejectedException if no background slot is free + * @throws SyncInProgressException if a sync of the project is queued or running + * @throws AmqpException if the request couldn't be queued (RabbitMQ unreachable or refusing it) */ public void requestSync(long projectId) throws NotFoundException { // an unknown project is reported now, not only in the background @@ -43,10 +50,15 @@ public void requestSync(long projectId) throws NotFoundException { } try { - // the background task releases the lock when it ends, however it ends - projectSyncWorker.syncName(projectId).whenComplete((result, failure) -> lease.release()); - } catch (RuntimeException e) { - // not started, e.g. a full task executor + // the consumer releases the lock when the sync ends, however it ends + var request = new ProjectSyncRequest(projectId, lease.ownerIdOrNull()); + rabbitTemplate.invoke(operations -> { + operations.convertAndSend("", syncQueue.queue(), request); + // accepted (202) only once the broker has the request + operations.waitForConfirmsOrDie(CONFIRM_TIMEOUT.toMillis()); + return null; + }); + } catch (AmqpException e) { lease.release(); throw e; } diff --git a/src/main/java/com/company/resourceapi/services/ProjectSyncWorker.java b/src/main/java/com/company/resourceapi/services/ProjectSyncWorker.java index ddab6f5..15557c7 100644 --- a/src/main/java/com/company/resourceapi/services/ProjectSyncWorker.java +++ b/src/main/java/com/company/resourceapi/services/ProjectSyncWorker.java @@ -1,18 +1,14 @@ package com.company.resourceapi.services; -import static java.util.concurrent.CompletableFuture.completedFuture; - -import java.util.concurrent.CompletableFuture; +import com.company.resourceapi.exceptions.NotFoundException; import java.util.concurrent.atomic.AtomicBoolean; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; -import org.springframework.web.client.RestClientResponseException; /** - * Example of background work: refreshes a project's name from its SDLC system. Started through - * {@link ProjectSyncService#requestSync}, which makes sure only one runs per project. + * Refreshes a project's name from its SDLC system. Run by {@link ProjectSyncConsumer} for requests + * queued by {@link ProjectSyncService#requestSync}, which makes sure only one runs per project. */ @Service @RequiredArgsConstructor @@ -24,50 +20,39 @@ public class ProjectSyncWorker { private final ExternalProjectService externalProjectService; /** - * Runs on a virtual thread of the application task executor, in the caller's trace. Failures are - * logged here, since nobody waits for the result of a fire-and-forget call. - * - * @return completes when the sync has finished, exceptionally if it failed + * @throws NotFoundException if the project was deleted since the sync was requested + * @throws org.springframework.web.client.RestClientException if the SDLC system failed or was unreachable + * @throws com.company.resourceapi.exceptions.RateLimitExceededException if our own limit for calls to + * the SDLC system was reached */ - @Async - public CompletableFuture syncName(long projectId) { - try { - var project = projectService.getProject(projectId); - var externalId = project.getExternalId(); - var sdlcSystemId = project.getSdlcSystem().getId(); + public void sync(long projectId) throws NotFoundException { + var project = projectService.getProject(projectId); + var externalId = project.getExternalId(); + var sdlcSystemId = project.getSdlcSystem().getId(); - var externalProject = externalProjectService.refreshProject(project.getSdlcSystem(), externalId); - if (externalProject.isEmpty()) { - log.warn("Project {} ({}) is unknown to its SDLC system", projectId, externalId); - return completedFuture(null); - } - var externalName = externalProject.get().name(); - if (externalName == null || externalName.equals(project.getName())) { - log.info("Project {} name is unchanged in its SDLC system", projectId); - return completedFuture(null); - } + var externalProject = externalProjectService.refreshProject(project.getSdlcSystem(), externalId); + if (externalProject.isEmpty()) { + log.warn("Project {} ({}) is unknown to its SDLC system", projectId, externalId); + return; + } + var externalName = externalProject.get().name(); + if (externalName == null || externalName.equals(project.getName())) { + log.info("Project {} name is unchanged in its SDLC system", projectId); + return; + } - var applied = new AtomicBoolean(); - projectService.updateProject(projectId, stored -> { - // the project may have been re-keyed while its SDLC system was being asked - if (stored.getExternalId().equals(externalId) && stored.getSdlcSystem().getId() == sdlcSystemId) { - stored.setName(externalName); - applied.set(true); - } - }); - if (applied.get()) { - log.info("Synced project {} name from its SDLC system", projectId); - } else { - log.info("Project {} changed while syncing; SDLC system name not applied", projectId); + var applied = new AtomicBoolean(); + projectService.updateProject(projectId, stored -> { + // the project may have been re-keyed while its SDLC system was being asked + if (stored.getExternalId().equals(externalId) && stored.getSdlcSystem().getId() == sdlcSystemId) { + stored.setName(externalName); + applied.set(true); } - return completedFuture(null); - } catch (Exception e) { - // the status only: an error response's message would include the upstream body - var reason = e instanceof RestClientResponseException response - ? "HTTP " + response.getStatusCode().value() - : e.getClass().getSimpleName() + ": " + e.getMessage(); - log.warn("Syncing project {} from its SDLC system failed: {}", projectId, reason); - return CompletableFuture.failedFuture(e); + }); + if (applied.get()) { + log.info("Synced project {} name from its SDLC system", projectId); + } else { + log.info("Project {} changed while syncing; SDLC system name not applied", projectId); } } } diff --git a/src/main/java/com/company/resourceapi/services/SyncOnProjectChange.java b/src/main/java/com/company/resourceapi/services/SyncOnProjectChange.java index f913c4c..24875e0 100644 --- a/src/main/java/com/company/resourceapi/services/SyncOnProjectChange.java +++ b/src/main/java/com/company/resourceapi/services/SyncOnProjectChange.java @@ -13,8 +13,9 @@ * project there, so clients don't have to request a sync themselves. * *

Runs after the change is committed (a rolled-back change starts nothing), on the request - * thread, which only starts the sync: the sync itself runs in the background. A sync that can't start - * now (one is running, rate limited, no free slot) is logged and skipped; it can be requested later. + * thread, which only queues the sync: the sync itself runs in the background. A sync that can't be + * queued now (one is queued or running, rate limited, RabbitMQ unreachable) is logged and skipped; it + * can be requested later. */ @Component @ConditionalOnBooleanProperty(name = "resourceapi.sync.on-change", matchIfMissing = true) diff --git a/src/main/java/com/company/resourceapi/web/ApiExceptionHandler.java b/src/main/java/com/company/resourceapi/web/ApiExceptionHandler.java index cdb0b5d..842b5e1 100644 --- a/src/main/java/com/company/resourceapi/web/ApiExceptionHandler.java +++ b/src/main/java/com/company/resourceapi/web/ApiExceptionHandler.java @@ -10,6 +10,7 @@ import com.company.resourceapi.exceptions.RateLimitExceededException; import com.company.resourceapi.exceptions.SyncInProgressException; import lombok.extern.slf4j.Slf4j; +import org.springframework.amqp.AmqpException; import org.springframework.core.task.TaskRejectedException; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.ExceptionHandler; @@ -50,6 +51,13 @@ ResponseEntity handleRateLimitExceeded(RateLimitExceededException e) { return ResponseEntity.status(status).header(RETRY_AFTER, String.valueOf(e.getRetryAfterSeconds())).build(); } + /** RabbitMQ is unreachable or refused a message the request needed to queue (e.g. a sync). */ + @ExceptionHandler(AmqpException.class) + ResponseEntity handleMessagingFailure(AmqpException e) { + log.warn("Message broker unavailable: {}", e.getMessage()); + return ResponseEntity.status(SERVICE_UNAVAILABLE).header(RETRY_AFTER, "5").build(); + } + /** The background task executor is at its concurrency limit; the client can retry shortly. */ @ExceptionHandler(TaskRejectedException.class) ResponseEntity handleTaskRejected(TaskRejectedException e) { diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 7ed62ab..feb782c 100755 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -3,9 +3,9 @@ spring.application.name=resourceapi # (incoming W3C and B3 headers are both accepted). Nothing is exported; when adding a reporter # (e.g. Zipkin/OTLP), lower the sampling probability to what the tracing backend can take. management.tracing.sampling.probability=1.0 -# virtual threads for request handling and the @Async task executor +# virtual threads for request handling, the @Async task executor and RabbitMQ listeners spring.threads.virtual.enabled=true -# at most 8 background tasks at once; more are rejected (503) instead of piling up +# at most 8 @Async tasks at once; more are rejected instead of piling up (syncs are queued in RabbitMQ) # created even though the application has other executors (notifications use their own) spring.task.execution.mode=force spring.task.execution.thread-name-prefix=async- @@ -31,7 +31,8 @@ resourceapi.cache.missing-external-projects-ttl=30s resourceapi.redisson.key-prefix=resourceapi: # sync a project from its SDLC system after it is created or re-keyed (pointed at another SDLC project) resourceapi.sync.on-change=true -# a project's sync lock expires after this even if never released (e.g. the instance died) +# a project's sync lock is held from the request until its sync ends, and expires after this even if never +# released (e.g. the instance died); a request waiting in the queue longer than this loses its lock resourceapi.sync.lock-lease-time=60s # distributed rate limits (Redisson): calls to one SDLC system (503 + Retry-After when exceeded after # waiting max-wait), and sync requests per project (429 + Retry-After) @@ -51,8 +52,19 @@ server.forward-headers-strategy=native # for the broker's confirmation before the event counts as sent spring.rabbitmq.connection-timeout=2s spring.rabbitmq.publisher-confirm-type=simple -# trace context (traceparent) on sent messages +# trace context (traceparent) on sent messages, continued by the listeners spring.rabbitmq.template.observation-enabled=true +spring.rabbitmq.listener.simple.observation-enabled=true +# sync consumers per instance; each takes one request at a time, so waiting requests go to whichever +# instance is free. A failing listener's message goes to the dead-letter queue, never back to the queue +spring.rabbitmq.listener.simple.concurrency=4 +spring.rabbitmq.listener.simple.prefetch=1 +spring.rabbitmq.listener.simple.default-requeue-rejected=false +# sync requests (ProjectSyncQueueConfig); failed syncs are retried after 1s and 2s, then dead-lettered +resourceapi.sync.queue=resourceapi.project-sync +resourceapi.sync.max-retries=2 +resourceapi.sync.retry-delay=1s +resourceapi.sync.retry-multiplier=2 # outbox (Spring Modulith): events sent to RabbitMQ are deleted from event_publication once confirmed; # unsent ones are sent again after resourceapi.notifications.resubmit-after spring.modulith.events.completion-mode=delete diff --git a/src/test/java/com/company/resourceapi/RabbitTestContainerInitializer.java b/src/test/java/com/company/resourceapi/RabbitTestContainerInitializer.java index 1e59a4f..3a83cc2 100644 --- a/src/test/java/com/company/resourceapi/RabbitTestContainerInitializer.java +++ b/src/test/java/com/company/resourceapi/RabbitTestContainerInitializer.java @@ -13,8 +13,8 @@ * Gives every test application context a real RabbitMQ: one container, started on first use and * shared by all contexts of the test run (registered in META-INF/spring.factories). * - *

Each context gets its own notifications exchange: contexts have their own in-memory database, - * so the same project id can be sent by two contexts. A test can still point at another broker with + *

Each context gets its own notifications exchange and sync queue: contexts have their own + * in-memory database, so the same project id means different projects in two contexts. A test can still point at another broker with * its own {@code spring.rabbitmq.*} properties. */ public class RabbitTestContainerInitializer implements ApplicationContextInitializer { @@ -33,8 +33,10 @@ public void initialize(ConfigurableApplicationContext context) { "spring.rabbitmq.host", RABBITMQ.getHost(), "spring.rabbitmq.port", RABBITMQ.getMappedPort(5672)))); // overrides application.properties - propertySources.addFirst(new MapPropertySource("rabbitTestContainerExchange", Map.of( - "resourceapi.notifications.exchange", "test-" + UUID.randomUUID()))); + var contextId = "test-" + UUID.randomUUID(); + propertySources.addFirst(new MapPropertySource("rabbitTestContainerNames", Map.of( + "resourceapi.notifications.exchange", contextId, + "resourceapi.sync.queue", contextId + "-sync"))); } private static synchronized void startOnce() { diff --git a/src/test/java/com/company/resourceapi/services/ProjectSyncBackpressureTest.java b/src/test/java/com/company/resourceapi/services/ProjectSyncBackpressureTest.java deleted file mode 100644 index 947f5fa..0000000 --- a/src/test/java/com/company/resourceapi/services/ProjectSyncBackpressureTest.java +++ /dev/null @@ -1,81 +0,0 @@ -package com.company.resourceapi.services; - -import static com.github.tomakehurst.wiremock.client.WireMock.get; -import static com.github.tomakehurst.wiremock.client.WireMock.okJson; -import static com.github.tomakehurst.wiremock.client.WireMock.urlPathMatching; -import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig; -import static org.assertj.core.api.Assertions.assertThat; -import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; -import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header; -import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; - -import com.company.resourceapi.controllers.ProjectRestController; -import com.company.resourceapi.entities.Project; -import com.company.resourceapi.entities.SdlcSystem; -import com.company.resourceapi.repositories.ProjectRepository; -import com.company.resourceapi.repositories.SdlcSystemRepository; -import com.github.tomakehurst.wiremock.junit5.WireMockExtension; -import java.time.Duration; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.RegisterExtension; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc; -import org.springframework.test.web.servlet.MockMvc; - -/** Background syncs are bounded: beyond the executor's limit they are rejected, not queued. */ -@SpringBootTest(properties = { - "spring.task.execution.simple.concurrency-limit=1", - // this test retries refused syncs; the per-project sync rate limit isn't what it is about - "resourceapi.rate-limit.sync.permits=1000"}) -@AutoConfigureMockMvc -class ProjectSyncBackpressureTest { - - @RegisterExtension - static WireMockExtension sdlcSystemServer = WireMockExtension.newInstance() - .options(wireMockConfig().dynamicPort()) - .build(); - - @Autowired - private MockMvc mockMvc; - - @Autowired - private SdlcSystemRepository sdlcSystemRepository; - - @Autowired - private ProjectRepository projectRepository; - - @Test - void givenBusyTaskExecutorWhenSyncThenReturnStatusServiceUnavailable() throws Exception { - // given: the only background slot stays busy for 1s - var sdlcSystem = new SdlcSystem(); - sdlcSystem.setBaseUrl(sdlcSystemServer.baseUrl()); - sdlcSystem = sdlcSystemRepository.save(sdlcSystem); - var syncUrl = syncUrl(sdlcSystem, "BUSY"); - var otherSyncUrl = syncUrl(sdlcSystem, "OTHER"); - sdlcSystemServer.stubFor(get(urlPathMatching("/rest/api/2/project/.*")) - .willReturn(okJson("{\"key\": \"BUSY\", \"name\": \"Busy\"}").withFixedDelay(1_000))); - - // when / then: another project's sync (the same project's would be refused by its lock) - mockMvc.perform(post(syncUrl)).andExpect(status().isAccepted()); - mockMvc.perform(post(otherSyncUrl)) - .andExpect(status().isServiceUnavailable()) - .andExpect(header().string("Retry-After", "1")); - - // the rejected sync released its lock: it can be started once the first sync freed the slot - var deadline = System.nanoTime() + Duration.ofSeconds(10).toNanos(); - int status; - while ((status = mockMvc.perform(post(otherSyncUrl)).andReturn().getResponse().getStatus()) == 503) { - assertThat(System.nanoTime()).as("no free slot within 10s").isLessThan(deadline); - Thread.sleep(100); - } - assertThat(status).isEqualTo(202); - } - - private String syncUrl(SdlcSystem sdlcSystem, String externalId) { - var project = new Project(); - project.setExternalId(externalId); - project.setSdlcSystem(sdlcSystem); - return ProjectRestController.ENDPOINT + "/" + projectRepository.save(project).getId() + "/sync"; - } -} diff --git a/src/test/java/com/company/resourceapi/services/ProjectSyncQueueTest.java b/src/test/java/com/company/resourceapi/services/ProjectSyncQueueTest.java new file mode 100644 index 0000000..1be4c54 --- /dev/null +++ b/src/test/java/com/company/resourceapi/services/ProjectSyncQueueTest.java @@ -0,0 +1,179 @@ +package com.company.resourceapi.services; + +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse; +import static com.github.tomakehurst.wiremock.client.WireMock.get; +import static com.github.tomakehurst.wiremock.client.WireMock.getRequestedFor; +import static com.github.tomakehurst.wiremock.client.WireMock.okJson; +import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo; +import static com.github.tomakehurst.wiremock.client.WireMock.urlPathMatching; +import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig; +import static com.github.tomakehurst.wiremock.stubbing.Scenario.STARTED; +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +import com.company.resourceapi.config.ProjectSyncQueueConfig.SyncQueueProperties; +import com.company.resourceapi.controllers.ProjectRestController; +import com.company.resourceapi.entities.Project; +import com.company.resourceapi.entities.SdlcSystem; +import com.company.resourceapi.repositories.ProjectRepository; +import com.company.resourceapi.repositories.SdlcSystemRepository; +import com.github.tomakehurst.wiremock.junit5.WireMockExtension; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.function.BooleanSupplier; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.springframework.amqp.core.AmqpAdmin; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc; +import org.springframework.test.web.servlet.MockMvc; + +/** Syncs are queued in RabbitMQ, retried when the SDLC system fails, and dead-lettered if it keeps failing. */ +@SpringBootTest(properties = { + // one consumer, so requests have to wait for it + "spring.rabbitmq.listener.simple.concurrency=1", + "spring.rabbitmq.listener.simple.max-concurrency=1"}) +@AutoConfigureMockMvc +class ProjectSyncQueueTest { + + @RegisterExtension + static WireMockExtension sdlcSystemServer = WireMockExtension.newInstance() + .options(wireMockConfig().dynamicPort()) + .build(); + + @Autowired + private MockMvc mockMvc; + + @Autowired + private SdlcSystemRepository sdlcSystemRepository; + + @Autowired + private ProjectRepository projectRepository; + + @Autowired + private RabbitTemplate rabbitTemplate; + + @Autowired + private AmqpAdmin amqpAdmin; + + @Autowired + private SyncQueueProperties syncQueue; + + private SdlcSystem sdlcSystem; + + @BeforeEach + void setup() { + sdlcSystem = new SdlcSystem(); + sdlcSystem.setBaseUrl(sdlcSystemServer.baseUrl()); + sdlcSystem = sdlcSystemRepository.save(sdlcSystem); + amqpAdmin.purgeQueue(syncQueue.deadLetterQueue(), false); + } + + @Test + void givenMoreRequestsThanConsumersThenAllAreAcceptedAndSynced() throws Exception { + sdlcSystemServer.stubFor(get(urlPathMatching("/rest/api/2/project/QUEUED-.*")) + .willReturn(okJson("{\"key\": \"Q\", \"name\": \"Synced\"}").withFixedDelay(200))); + var ids = new ArrayList(); + for (int i = 0; i < 5; i++) { + ids.add(project("QUEUED-" + i)); + } + + // nothing refused for lack of a free slot: the requests wait in the queue + for (long id : ids) { + mockMvc.perform(post(syncUrl(id))).andExpect(status().isAccepted()); + } + + await(() -> ids.stream().allMatch(id -> "Synced".equals(nameOf(id))), Duration.ofSeconds(10)); + } + + @Test + void givenSdlcSystemFailingOnceThenSyncSucceedsOnRetry() throws Exception { + var url = "/rest/api/2/project/FLAKY"; + sdlcSystemServer.stubFor(get(urlEqualTo(url)).inScenario("flaky").whenScenarioStateIs(STARTED) + .willReturn(aResponse().withStatus(503)).willSetStateTo("recovered")); + sdlcSystemServer.stubFor(get(urlEqualTo(url)).inScenario("flaky").whenScenarioStateIs("recovered") + .willReturn(okJson("{\"key\": \"FLAKY\", \"name\": \"After retry\"}"))); + long id = project("FLAKY"); + + mockMvc.perform(post(syncUrl(id))).andExpect(status().isAccepted()); + + await(() -> "After retry".equals(nameOf(id)), Duration.ofSeconds(5)); + sdlcSystemServer.verify(2, getRequestedFor(urlEqualTo(url))); + } + + @Test + void givenSdlcSystemKeepsFailingThenRequestIsDeadLetteredAndLockReleased() throws Exception { + var url = "/rest/api/2/project/BROKEN"; + sdlcSystemServer.stubFor(get(urlEqualTo(url)).willReturn(aResponse().withStatus(500))); + long id = project("BROKEN"); + + mockMvc.perform(post(syncUrl(id))).andExpect(status().isAccepted()); + + // the first try and 2 retries, then kept for inspection + assertThat(body(receiveDeadLetter())).contains("\"projectId\":" + id); + sdlcSystemServer.verify(3, getRequestedFor(urlEqualTo(url))); + // the lock was released: a new sync can be requested right away + await(() -> statusOfSync(id) == 202, Duration.ofSeconds(2)); + } + + @Test + void givenSdlcSystemRejectingTheRequestThenItIsNotRetried() throws Exception { + var url = "/rest/api/2/project/FORBIDDEN"; + sdlcSystemServer.stubFor(get(urlEqualTo(url)).willReturn(aResponse().withStatus(403))); + long id = project("FORBIDDEN"); + + mockMvc.perform(post(syncUrl(id))).andExpect(status().isAccepted()); + + assertThat(body(receiveDeadLetter())).contains("\"projectId\":" + id); + sdlcSystemServer.verify(1, getRequestedFor(urlEqualTo(url))); + } + + private Message receiveDeadLetter() { + var deadLetter = rabbitTemplate.receive(syncQueue.deadLetterQueue(), 5_000); + assertThat(deadLetter).as("dead-lettered within 5s").isNotNull(); + return deadLetter; + } + + private static String body(Message message) { + return new String(message.getBody(), StandardCharsets.UTF_8); + } + + private long project(String externalId) { + var project = new Project(); + project.setExternalId(externalId); + project.setName("Old name"); + project.setSdlcSystem(sdlcSystem); + return projectRepository.save(project).getId(); + } + + private int statusOfSync(long id) { + try { + return mockMvc.perform(post(syncUrl(id))).andReturn().getResponse().getStatus(); + } catch (Exception e) { + throw new IllegalStateException(e); + } + } + + private String nameOf(long id) { + return projectRepository.findById(id).orElseThrow().getName(); + } + + private static String syncUrl(long id) { + return ProjectRestController.ENDPOINT + "/" + id + "/sync"; + } + + private static void await(BooleanSupplier condition, Duration timeout) throws InterruptedException { + var deadline = Instant.now().plus(timeout); + while (!condition.getAsBoolean()) { + assertThat(Instant.now()).as("condition not met within " + timeout).isBefore(deadline); + Thread.sleep(50); + } + } +} diff --git a/src/test/java/com/company/resourceapi/services/RabbitUnavailableTest.java b/src/test/java/com/company/resourceapi/services/RabbitUnavailableTest.java new file mode 100644 index 0000000..3bc7ada --- /dev/null +++ b/src/test/java/com/company/resourceapi/services/RabbitUnavailableTest.java @@ -0,0 +1,50 @@ +package com.company.resourceapi.services; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +import com.company.resourceapi.controllers.ProjectRestController; +import java.time.Duration; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc; +import org.springframework.test.web.servlet.MockMvc; + +/** Without RabbitMQ the application still starts and serves requests; only syncs are refused. */ +@SpringBootTest(properties = { + "spring.rabbitmq.port=1", + "spring.rabbitmq.connection-timeout=200ms"}) +@AutoConfigureMockMvc +class RabbitUnavailableTest { + + @Autowired + private MockMvc mockMvc; + + @Autowired + private ProjectSyncLock projectSyncLock; + + @Test + void givenRabbitUnreachableWhenSyncThenServiceUnavailableAndLockReleased() throws Exception { + long started = System.nanoTime(); + + mockMvc.perform(post(ProjectRestController.ENDPOINT + "/1/sync")) + .andExpect(status().isServiceUnavailable()) + .andExpect(header().string("Retry-After", "5")); + + assertThat(Duration.ofNanos(System.nanoTime() - started)).isLessThan(Duration.ofSeconds(3)); + // not left locked: the lock can be taken again + var lease = projectSyncLock.tryAcquire(1); + assertThat(lease).isPresent(); + lease.get().release(); + } + + @Test + void givenRabbitUnreachableThenProjectsCanStillBeRead() throws Exception { + mockMvc.perform(org.springframework.test.web.servlet.request.MockMvcRequestBuilders + .get(ProjectRestController.ENDPOINT + "/1")) + .andExpect(status().isOk()); + } +} diff --git a/src/test/resources/config/application.properties b/src/test/resources/config/application.properties index fb0015f..3e03429 100644 --- a/src/test/resources/config/application.properties +++ b/src/test/resources/config/application.properties @@ -5,3 +5,5 @@ resourceapi.rate-limit.client.permits=100000 # no automatic syncs after creating or re-keying a project: tests would call SDLC systems they don't # expect (SyncOnProjectChangeTest turns it on) resourceapi.sync.on-change=false +# failed syncs are retried quickly +resourceapi.sync.retry-delay=100ms