diff --git a/src/main/java/com/company/resourceapi/config/AsyncConfig.java b/src/main/java/com/company/resourceapi/config/AsyncConfig.java
index f5e3512..7b4a786 100644
--- a/src/main/java/com/company/resourceapi/config/AsyncConfig.java
+++ b/src/main/java/com/company/resourceapi/config/AsyncConfig.java
@@ -1,8 +1,14 @@
package com.company.resourceapi.config;
+import java.util.concurrent.Executor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
+import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.support.ContextPropagatingTaskDecorator;
+import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
/**
@@ -12,6 +18,7 @@
*/
@Configuration
@EnableAsync
+@Slf4j
public class AsyncConfig {
/** Boot applies a TaskDecorator bean to the application task executor. */
@@ -19,4 +26,32 @@ public class AsyncConfig {
ContextPropagatingTaskDecorator contextPropagatingTaskDecorator() {
return new ContextPropagatingTaskDecorator();
}
+
+ /**
+ * {@code @Async} without an executor name runs on Boot's application task executor, also now
+ * that other executors exist (spring.task.execution.mode=force keeps Boot creating it).
+ *
+ * A failed fire-and-forget task (e.g. a notification that RabbitMQ didn't accept, which is
+ * sent again later) is logged in one line instead of with a stack trace for every occurrence.
+ */
+ @Configuration
+ static class AsyncErrorLogging implements AsyncConfigurer {
+
+ private final BeanFactory beanFactory;
+
+ AsyncErrorLogging(BeanFactory beanFactory) {
+ this.beanFactory = beanFactory;
+ }
+
+ @Override
+ public Executor getAsyncExecutor() {
+ return beanFactory.getBean(TaskExecutionAutoConfiguration.APPLICATION_TASK_EXECUTOR_BEAN_NAME, Executor.class);
+ }
+
+ @Override
+ public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
+ return (failure, method, params) -> log.warn("Background task {}.{} failed: {}",
+ method.getDeclaringClass().getSimpleName(), method.getName(), failure.toString());
+ }
+ }
}
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/notifications/NotificationsConfig.java b/src/main/java/com/company/resourceapi/notifications/NotificationsConfig.java
new file mode 100644
index 0000000..5a7739a
--- /dev/null
+++ b/src/main/java/com/company/resourceapi/notifications/NotificationsConfig.java
@@ -0,0 +1,61 @@
+package com.company.resourceapi.notifications;
+
+import com.company.resourceapi.events.ProjectCreated;
+import com.company.resourceapi.events.ProjectUpdated;
+import org.springframework.amqp.core.ExchangeBuilder;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.support.converter.JacksonJsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.aot.hint.annotation.RegisterReflectionForBinding;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.core.task.support.ContextPropagatingTaskDecorator;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+/**
+ * Sends project events to other systems through RabbitMQ, without losing them (an outbox).
+ *
+ * Spring Modulith's event publication registry stores each event in the event_publication table,
+ * in the transaction that changed the project, so it exists exactly if the change was committed.
+ * After commit, {@link ProjectNotifier} sends it to the {@code resourceapi.notifications.exchange}
+ * topic exchange and it is removed once the broker confirmed it. An event that couldn't be sent
+ * (RabbitMQ down or refusing it, application stopped) stays in the table and is sent again by
+ * {@link UnsentNotifications}: delivery is at least once, so consumers should tolerate duplicates.
+ */
+@Configuration
+@EnableScheduling
+@EnableConfigurationProperties(NotificationsProperties.class)
+// events are stored and sent as JSON
+@RegisterReflectionForBinding({ProjectCreated.class, ProjectUpdated.class})
+public class NotificationsConfig {
+
+ /** Declared on the broker when the application first connects. */
+ @Bean
+ TopicExchange projectsExchange(NotificationsProperties properties) {
+ return ExchangeBuilder.topicExchange(properties.exchange()).durable(true).build();
+ }
+
+ /**
+ * Sends run here, not on the application task executor, whose limit is for other background work.
+ * Virtual threads, in the caller's trace. At most 16 sends at once; when all are busy (e.g. each
+ * waiting for an unreachable broker) a further send is refused rather than making the request wait
+ * for a free slot: its event stays in the outbox and is sent again later.
+ */
+ @Bean
+ SimpleAsyncTaskExecutor notificationExecutor() {
+ var executor = new SimpleAsyncTaskExecutor("notify-");
+ executor.setVirtualThreads(true);
+ executor.setConcurrencyLimit(16);
+ executor.setRejectTasksWhenLimitReached(true);
+ executor.setTaskDecorator(new ContextPropagatingTaskDecorator());
+ return executor;
+ }
+
+ /** Messages as JSON (Boot applies it to the RabbitTemplate). */
+ @Bean
+ MessageConverter jsonMessageConverter() {
+ return new JacksonJsonMessageConverter();
+ }
+}
diff --git a/src/main/java/com/company/resourceapi/notifications/NotificationsProperties.java b/src/main/java/com/company/resourceapi/notifications/NotificationsProperties.java
new file mode 100644
index 0000000..fc70d52
--- /dev/null
+++ b/src/main/java/com/company/resourceapi/notifications/NotificationsProperties.java
@@ -0,0 +1,24 @@
+package com.company.resourceapi.notifications;
+
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+import java.time.Duration;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.bind.DefaultValue;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Project notifications for other systems.
+ *
+ * @param exchange RabbitMQ topic exchange the notifications are sent to
+ * @param resubmitInterval how often unsent notifications are looked for
+ * @param resubmitAfter how old an unsent notification must be before it is sent again; longer than a
+ * normal send takes, so one still on its way isn't sent twice
+ */
+@Validated
+@ConfigurationProperties("resourceapi.notifications")
+public record NotificationsProperties(
+ @DefaultValue("resourceapi.projects") @NotBlank String exchange,
+ @DefaultValue("1m") @NotNull Duration resubmitInterval,
+ @DefaultValue("1m") @NotNull Duration resubmitAfter) {
+}
diff --git a/src/main/java/com/company/resourceapi/notifications/ProjectNotifier.java b/src/main/java/com/company/resourceapi/notifications/ProjectNotifier.java
new file mode 100644
index 0000000..874ee07
--- /dev/null
+++ b/src/main/java/com/company/resourceapi/notifications/ProjectNotifier.java
@@ -0,0 +1,54 @@
+package com.company.resourceapi.notifications;
+
+import com.company.resourceapi.events.ProjectCreated;
+import com.company.resourceapi.events.ProjectUpdated;
+import java.time.Duration;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+/**
+ * Sends committed project events to RabbitMQ. A send counts only once the broker confirmed it: if it
+ * fails or isn't confirmed, the listener fails and the event stays in the outbox to be sent again.
+ *
+ *
Runs in the background, so a slow or unreachable broker doesn't hold up the request.
+ */
+@Component
+public class ProjectNotifier {
+
+ public static final String PROJECT_CREATED = "project.created";
+ public static final String PROJECT_UPDATED = "project.updated";
+
+ private static final Duration CONFIRM_TIMEOUT = Duration.ofSeconds(5);
+
+ private final RabbitTemplate rabbitTemplate;
+ private final String exchange;
+
+ public ProjectNotifier(RabbitTemplate rabbitTemplate, NotificationsProperties properties) {
+ this.rabbitTemplate = rabbitTemplate;
+ this.exchange = properties.exchange();
+ }
+
+ @Async("notificationExecutor")
+ @TransactionalEventListener
+ public void on(ProjectCreated event) {
+ send(PROJECT_CREATED, event);
+ }
+
+ @Async("notificationExecutor")
+ @TransactionalEventListener
+ public void on(ProjectUpdated event) {
+ send(PROJECT_UPDATED, event);
+ }
+
+ private void send(String routingKey, Object event) {
+ // a failure leaves the event in the outbox (and is logged by AsyncConfig's handler)
+ rabbitTemplate.invoke(operations -> {
+ operations.convertAndSend(exchange, routingKey, event);
+ // throws if the broker refused the message (nack, closed channel) or didn't answer in time
+ operations.waitForConfirmsOrDie(CONFIRM_TIMEOUT.toMillis());
+ return null;
+ });
+ }
+}
diff --git a/src/main/java/com/company/resourceapi/notifications/UnsentNotifications.java b/src/main/java/com/company/resourceapi/notifications/UnsentNotifications.java
new file mode 100644
index 0000000..ec4243c
--- /dev/null
+++ b/src/main/java/com/company/resourceapi/notifications/UnsentNotifications.java
@@ -0,0 +1,69 @@
+package com.company.resourceapi.notifications;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.modulith.events.IncompleteEventPublications;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * Sends again the events that weren't sent (RabbitMQ was down, the application stopped before
+ * sending, ...). One instance does it at a time, chosen with a Redis lock; without Redis every
+ * instance does it, which can only add duplicates, never lose an event.
+ *
+ *
The registry tracks every {@code @TransactionalEventListener}, so this also re-runs other
+ * listeners that didn't finish, e.g. a sync after create that the application stopped before
+ * starting.
+ */
+@Component
+@Slf4j
+public class UnsentNotifications {
+
+ private static final String LOCK_NAME = "notifications-resubmit";
+
+ private final IncompleteEventPublications incompletePublications;
+ private final RedissonClient redisson;
+ private final Duration resubmitAfter;
+
+ public UnsentNotifications(IncompleteEventPublications incompletePublications,
+ ObjectProvider redisson, NotificationsProperties properties) {
+ this.incompletePublications = incompletePublications;
+ this.redisson = redisson.getIfAvailable();
+ this.resubmitAfter = properties.resubmitAfter();
+ }
+
+ @Scheduled(fixedDelayString = "${resourceapi.notifications.resubmit-interval:1m}",
+ initialDelayString = "${resourceapi.notifications.resubmit-interval:1m}")
+ public void resubmit() {
+ var lock = redisson == null ? null : redisson.getLock(LOCK_NAME);
+ boolean locked = false;
+ if (lock != null) {
+ try {
+ // held at most as long as one round may take; another instance runs the next round
+ locked = lock.tryLock(0, resubmitAfter.toMillis(), TimeUnit.MILLISECONDS);
+ if (!locked) {
+ return;
+ }
+ } catch (Exception e) {
+ log.warn("Resubmission lock unavailable, resubmitting anyway: {}", e.getMessage());
+ }
+ }
+ try {
+ incompletePublications.resubmitIncompletePublicationsOlderThan(resubmitAfter);
+ } catch (RuntimeException e) {
+ log.warn("Resubmitting unsent events failed: {}", e.getMessage());
+ } finally {
+ if (locked) {
+ try {
+ lock.unlock();
+ } catch (RuntimeException e) {
+ // expires with its lease
+ log.debug("Releasing the resubmission lock failed: {}", e.getMessage());
+ }
+ }
+ }
+ }
+}
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 2ccfddf..feb782c 100755
--- a/src/main/resources/application.properties
+++ b/src/main/resources/application.properties
@@ -3,9 +3,11 @@ 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-
spring.task.execution.simple.concurrency-limit=8
spring.task.execution.simple.reject-tasks-when-limit-reached=true
@@ -29,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)
@@ -45,6 +48,29 @@ resourceapi.rate-limit.client.permits=100
resourceapi.rate-limit.client.period=1s
resourceapi.rate-limit.client.max-wait=0s
server.forward-headers-strategy=native
+# rabbitmq (project notifications); spring.rabbitmq.host/port default to localhost:5672. A send waits
+# 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, 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
+resourceapi.notifications.exchange=resourceapi.projects
+resourceapi.notifications.resubmit-interval=1m
+resourceapi.notifications.resubmit-after=1m
# datasource
spring.datasource.url=jdbc:h2:mem:test;MODE=MySQL;DATABASE_TO_LOWER=TRUE
spring.sql.init.platform=h2
diff --git a/src/main/resources/schema.sql b/src/main/resources/schema.sql
index 83bddb9..f967116 100755
--- a/src/main/resources/schema.sql
+++ b/src/main/resources/schema.sql
@@ -18,3 +18,19 @@ CREATE TABLE `project` (
UNIQUE KEY `system_project` (`sdlc_system_id`,`external_id`),
CONSTRAINT `project_sdlc_system_fk` FOREIGN KEY (`sdlc_system_id`) REFERENCES `sdlc_system` (`id`)
);
+
+-- Outbox (Spring Modulith event publication registry): an event is stored here in the transaction that
+-- raised it, once per listener, and removed once that listener has handled it (e.g. sent it to RabbitMQ)
+CREATE TABLE `event_publication` (
+ `id` uuid NOT NULL,
+ `listener_id` varchar(512) NOT NULL,
+ `event_type` varchar(512) NOT NULL,
+ `serialized_event` varchar(4000) NOT NULL,
+ `publication_date` timestamp(9) with time zone NOT NULL,
+ `completion_date` timestamp(9) with time zone DEFAULT NULL,
+ `status` varchar(20),
+ `completion_attempts` int,
+ `last_resubmission_date` timestamp(9) with time zone DEFAULT NULL,
+ PRIMARY KEY (`id`)
+);
+CREATE INDEX `event_publication_by_completion_date_idx` ON `event_publication` (`completion_date`);
diff --git a/src/test/java/com/company/resourceapi/RabbitTestContainerInitializer.java b/src/test/java/com/company/resourceapi/RabbitTestContainerInitializer.java
new file mode 100644
index 0000000..3a83cc2
--- /dev/null
+++ b/src/test/java/com/company/resourceapi/RabbitTestContainerInitializer.java
@@ -0,0 +1,47 @@
+package com.company.resourceapi;
+
+import java.util.Map;
+import java.util.UUID;
+import org.springframework.context.ApplicationContextInitializer;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.env.MapPropertySource;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+
+/**
+ * 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 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 {
+
+ private static final GenericContainer RABBITMQ =
+ new GenericContainer<>(DockerImageName.parse("rabbitmq:4-alpine"))
+ .withExposedPorts(5672)
+ .waitingFor(Wait.forLogMessage(".*Server startup complete.*", 1));
+
+ @Override
+ public void initialize(ConfigurableApplicationContext context) {
+ startOnce();
+ var propertySources = context.getEnvironment().getPropertySources();
+ // defaults: a test's own spring.rabbitmq.* properties win
+ propertySources.addLast(new MapPropertySource("rabbitTestContainer", Map.of(
+ "spring.rabbitmq.host", RABBITMQ.getHost(),
+ "spring.rabbitmq.port", RABBITMQ.getMappedPort(5672))));
+ // overrides application.properties
+ 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() {
+ if (!RABBITMQ.isRunning()) {
+ RABBITMQ.start();
+ }
+ }
+}
diff --git a/src/test/java/com/company/resourceapi/notifications/NotificationsIntegrationTest.java b/src/test/java/com/company/resourceapi/notifications/NotificationsIntegrationTest.java
new file mode 100644
index 0000000..b97a694
--- /dev/null
+++ b/src/test/java/com/company/resourceapi/notifications/NotificationsIntegrationTest.java
@@ -0,0 +1,230 @@
+package com.company.resourceapi.notifications;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.patch;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+import com.company.resourceapi.controllers.ProjectRestController;
+import com.company.resourceapi.entities.SdlcSystem;
+import com.company.resourceapi.repositories.SdlcSystemRepository;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.function.BooleanSupplier;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.amqp.core.AmqpAdmin;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+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.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.http.MediaType;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.web.servlet.MockMvc;
+
+/** Project events reach RabbitMQ through the outbox, also when the first send fails. */
+@SpringBootTest(properties = {
+ // resubmitted by the test, not on a schedule
+ "resourceapi.notifications.resubmit-interval=1h",
+ "resourceapi.notifications.resubmit-after=1ms",
+ // a listener whose condition doesn't match must not leave an event behind in the outbox
+ "resourceapi.sync.on-change=true"})
+@AutoConfigureMockMvc
+class NotificationsIntegrationTest {
+
+ @Autowired
+ private MockMvc mockMvc;
+
+ @Autowired
+ private SdlcSystemRepository sdlcSystemRepository;
+
+ @Autowired
+ private RabbitTemplate rabbitTemplate;
+
+ @Autowired
+ private AmqpAdmin amqpAdmin;
+
+ @Autowired
+ private TopicExchange projectsExchange;
+
+ @Autowired
+ private UnsentNotifications unsentNotifications;
+
+ @Autowired
+ private SimpleAsyncTaskExecutor notificationExecutor;
+
+ @Autowired
+ private JdbcTemplate jdbcTemplate;
+
+ private SdlcSystem sdlcSystem;
+ private Queue queue;
+
+ @BeforeEach
+ void setup() {
+ sdlcSystem = new SdlcSystem();
+ // nothing listens there: the sync started after a create fails fast, which doesn't matter here
+ sdlcSystem.setBaseUrl("http://localhost:1");
+ sdlcSystem = sdlcSystemRepository.save(sdlcSystem);
+ queue = bindQueue();
+ }
+
+ @AfterEach
+ void cleanup() {
+ if (queue != null) {
+ amqpAdmin.deleteQueue(queue.getName());
+ }
+ }
+
+ @Test
+ void givenCreatedProjectThenNotificationIsSentAndOutboxEmptied() throws Exception {
+ var id = create("NOTIFY-1");
+
+ var message = receive();
+ assertThat(message.getMessageProperties().getReceivedRoutingKey()).isEqualTo("project.created");
+ assertThat(body(message)).contains("\"projectId\":" + id).contains("\"externalId\":\"NOTIFY-1\"");
+ // sent in the trace of the request that created the project
+ assertThat(message.getMessageProperties().getHeaders()).containsKey("traceparent");
+ await(() -> outboxSize() == 0);
+ }
+
+ @Test
+ void givenUpdatedProjectThenNotificationNamesTheChange() throws Exception {
+ var id = create("NOTIFY-2");
+ receive();
+
+ mockMvc.perform(patch(ProjectRestController.ENDPOINT + "/" + id)
+ .contentType(MediaType.APPLICATION_JSON)
+ .content("{\"name\": \"Renamed\"}"))
+ .andExpect(status().isOk());
+
+ var message = receive();
+ assertThat(message.getMessageProperties().getReceivedRoutingKey()).isEqualTo("project.updated");
+ assertThat(body(message)).contains("\"projectId\":" + id).contains("\"changed\":[\"NAME\"]");
+ await(() -> outboxSize() == 0);
+ }
+
+ @Test
+ void givenNotificationThatCouldNotBeSentThenItIsSentLater() throws Exception {
+ // the broker refuses the message: here because the exchange is gone, as for any unreachable broker
+ amqpAdmin.deleteExchange(projectsExchange.getName());
+
+ create("NOTIFY-3");
+
+ await(() -> outboxSize() == 1);
+ assertThat(rabbitTemplate.receive(queue.getName(), 300)).isNull();
+
+ // the broker is back
+ amqpAdmin.declareExchange(projectsExchange);
+ amqpAdmin.declareBinding(BindingBuilder.bind(queue).to(projectsExchange).with("project.#"));
+ unsentNotifications.resubmit();
+
+ var message = receive();
+ assertThat(message.getMessageProperties().getReceivedRoutingKey()).isEqualTo("project.created");
+ assertThat(body(message)).contains("\"externalId\":\"NOTIFY-3\"");
+ await(() -> outboxSize() == 0);
+ }
+
+ @Test
+ void givenAllSendersBusyThenRequestDoesNotWaitAndNotificationIsSentLater() throws Exception {
+ // connections (database, Redis, RabbitMQ) already open, so the timing below is the request's own
+ create("NOTIFY-4-WARMUP");
+ receive();
+ await(() -> outboxSize() == 0);
+
+ // every send slot taken, e.g. by sends waiting for an unreachable broker
+ var release = new CountDownLatch(1);
+ var busy = new CountDownLatch(16);
+ for (int i = 0; i < 16; i++) {
+ notificationExecutor.execute(() -> {
+ busy.countDown();
+ awaitQuietly(release);
+ });
+ }
+ busy.await();
+ // in case the request does wait for a slot: freed after 3s, so the test fails instead of hanging
+ Thread.ofVirtual().start(() -> {
+ try {
+ Thread.sleep(3_000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ release.countDown();
+ });
+
+ long started = System.nanoTime();
+ create("NOTIFY-4");
+
+ try {
+ // the request isn't held up by the full executor; the event waits in the outbox
+ assertThat(Duration.ofNanos(System.nanoTime() - started)).isLessThan(Duration.ofSeconds(1));
+ await(() -> outboxSize() == 1);
+ assertThat(rabbitTemplate.receive(queue.getName(), 300)).isNull();
+ } finally {
+ release.countDown();
+ }
+
+ // sent again once a slot is free (the blocked tasks end right after the release)
+ await(() -> {
+ unsentNotifications.resubmit();
+ return outboxSize() == 0;
+ });
+ assertThat(body(receive())).contains("\"externalId\":\"NOTIFY-4\"");
+ }
+
+ private static void awaitQuietly(CountDownLatch latch) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private long create(String externalId) throws Exception {
+ var location = mockMvc.perform(post(ProjectRestController.ENDPOINT)
+ .contentType(MediaType.APPLICATION_JSON)
+ .content("{\"externalId\": \"%s\", \"name\": \"Name\", \"sdlcSystem\": {\"id\": %d}}"
+ .formatted(externalId, sdlcSystem.getId())))
+ .andExpect(status().isCreated())
+ .andReturn().getResponse().getHeader("Location");
+ return Long.parseLong(location.substring(location.lastIndexOf('/') + 1));
+ }
+
+ private Queue bindQueue() {
+ // durable and not auto-delete (each receive() is a short-lived consumer); deleted after the test
+ var testQueue = new Queue("test-" + UUID.randomUUID(), true, false, false);
+ amqpAdmin.declareQueue(testQueue);
+ amqpAdmin.declareBinding(BindingBuilder.bind(testQueue).to(projectsExchange).with("project.#"));
+ return testQueue;
+ }
+
+ private Message receive() {
+ var message = rabbitTemplate.receive(queue.getName(), 5_000);
+ assertThat(message).as("notification within 5s").isNotNull();
+ return message;
+ }
+
+ private static String body(Message message) {
+ return new String(message.getBody(), StandardCharsets.UTF_8);
+ }
+
+ private int outboxSize() {
+ return jdbcTemplate.queryForObject("select count(*) from event_publication", Integer.class);
+ }
+
+ private static void await(BooleanSupplier condition) throws InterruptedException {
+ var deadline = Instant.now().plus(Duration.ofSeconds(5));
+ while (!condition.getAsBoolean()) {
+ assertThat(Instant.now()).as("condition not met within 5s").isBefore(deadline);
+ Thread.sleep(50);
+ }
+ }
+}
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/META-INF/spring.factories b/src/test/resources/META-INF/spring.factories
index 5256c02..28e0f8b 100644
--- a/src/test/resources/META-INF/spring.factories
+++ b/src/test/resources/META-INF/spring.factories
@@ -1,2 +1,3 @@
org.springframework.context.ApplicationContextInitializer=\
-com.company.resourceapi.RedisTestContainerInitializer
+com.company.resourceapi.RedisTestContainerInitializer,\
+com.company.resourceapi.RabbitTestContainerInitializer
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