Project events; sync a project's name after create or re-key

README.md

@@ -65,6 +65,66 @@ injects uses Boot's single request factory, so there is one pool for the applica
 The client is tested against [WireMock](https://wiremock.org/) (`SdlcProjectClientTest`, `ExternalProjectIntegrationTest`,
 and `SdlcProjectClientPoolingTest`, which counts the connections actually opened).
 
+## Events
+`ProjectService` publishes Spring application events (package `events`) inside the transaction that
+changes a project:
+
+- `ProjectCreated`
+- `ProjectUpdated`, with the new state and which fields changed (`NAME`, `EXTERNAL_ID`, `SDLC_SYSTEM`); not
+  published when an update changes nothing. Changes made by a sync from the SDLC system are published too.
+
+Both carry `occurredAt`, when the change was made: events about one project can reach a listener out of order
+(e.g. one retried later), so an event older than the last one applied is stale.
+
+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, RabbitMQ unreachable) is logged and skipped.
+The listener runs on the request thread after commit; it only queues the sync, but with Redis or RabbitMQ
+unreachable the failed attempt adds about a second (Redis) or up to `spring.rabbitmq.connection-timeout` (2s)
+to the request.
+Turn it off with `resourceapi.sync.on-change=false` (tests do, except `SyncOnProjectChangeTest`).
+
+## Sync queue (RabbitMQ)
+`POST /api/v2/projects/{id}/sync` queues the sync in the durable RabbitMQ queue `resourceapi.project-sync`
+(`resourceapi.sync.queue`) and answers `202 Accepted` once the broker confirmed it has the request; a consumer
+(`ProjectSyncConsumer`) refreshes the project's name from its SDLC system. Requests survive restarts, and a burst
+of them waits in the queue instead of being refused. If RabbitMQ is unreachable, the request gets `503` with
+`Retry-After`.
+
+- Each instance runs `spring.rabbitmq.listener.simple.concurrency` (4) consumers on virtual threads, taking one
+  request at a time (`prefetch=1`), so waiting requests go to whichever instance is free.
+- A sync failing for a reason that may pass (SDLC system unreachable, `5xx` or `429`, our own SDLC rate limit) is
+  retried after 1s and 2s (`resourceapi.sync.max-retries`, `retry-delay`, `retry-multiplier`); a sync that still
+  fails, or fails for another reason (e.g. `403`), goes to the dead-letter queue `<queue>.dead`, where it can be
+  inspected and moved back (e.g. with the management UI's shovel). A project deleted in the meantime is skipped.
+- The project's sync lock (see Redis below) is released however the sync ends.
+
+## Notifications (RabbitMQ)
+Other systems learn about project changes from RabbitMQ (`spring.rabbitmq.*`, default `localhost:5672`):
+`ProjectCreated` and `ProjectUpdated` are sent as JSON to the topic exchange `resourceapi.projects`
+(`resourceapi.notifications.exchange`) with routing keys `project.created` and `project.updated`. Messages carry
+the request's trace (`traceparent` header).
+
+No notification is lost, even if RabbitMQ is down or the application stops right after a change (an outbox):
+
+1. Spring Modulith's event publication registry stores each event in the `event_publication` table, in the
+   transaction that changed the project: it exists exactly if the change was committed.
+2. After commit, `ProjectNotifier` sends it in the background and waits for the broker's confirmation (publisher
+   confirms). Only then is the row deleted. Sends run on their own virtual-thread executor, at most 16 at once;
+   when all are busy (e.g. waiting for an unreachable broker) a further send is refused instead of holding up the
+   request (Spring logs the `TaskRejectedException`), and its event stays in the table.
+3. A send that failed or wasn't confirmed leaves the row; `UnsentNotifications` sends such events again
+   (`resourceapi.notifications.resubmit-interval` 1m, for events older than `resubmit-after` 1m). With several
+   instances, one does it at a time (Redis lock).
+
+Notifications about one project can arrive out of order (they are sent in parallel, and a resent one arrives
+late): consumers should use `occurredAt` and ignore one older than the last they applied. Delivery is at least
+once: a consumer may see the same notification twice (e.g. the broker confirmed
+but the application stopped before deleting the row) and should handle that, e.g. by treating them as "the
+project now looks like this". Modulith's own RabbitMQ externalization isn't used: in its default mode it doesn't
+wait for the broker's confirmation, so a message the broker refused would count as sent.
+
 ## API documentation
 The OpenAPI document is served at `/v3/api-docs` and Swagger UI at `/swagger-ui.html`
 ([springdoc-openapi](https://springdoc.org/)), generated from the controllers and their `@Operation`/`@Parameter`
@@ -87,9 +147,10 @@ optimization: if Redis is unreachable, cache reads and writes are logged and ski
 500 ms Redis timeouts) and requests are served without it.
 
 **Lock** ([Redisson](https://redisson.org/) `RLock`): at most one `POST /api/v2/projects/{id}/sync` per project
-runs at a time, across all instances; a second one gets `409 Conflict` until the first ends. The lock is
-released when the background sync ends and expires after `resourceapi.sync.lock-lease-time` (60s) in case
-an instance dies mid-sync. If Redis can't be reached, the sync runs without the lock (a duplicate sync only
+is queued or running at a time, across all instances; a second one gets `409 Conflict` until the first ends. The
+lock is taken by the request and released by whichever instance runs the sync when it ends (the request carries
+the lock's owner id), and expires after `resourceapi.sync.lock-lease-time` (60s) in case an instance dies
+mid-sync or the request waits in the queue longer than that. If Redis can't be reached, the sync runs without the lock (a duplicate sync only
 costs an extra SDLC call; the failed lock request takes about a second). Redisson connects lazily and uses the
 same Redis connection settings; its keys are prefixed with `resourceapi.redisson.key-prefix`. Only a
 standalone Redis is supported for the lock: with Sentinel or Cluster the application still starts and syncs
@@ -102,7 +163,7 @@ available again one period later, which is also what `Retry-After` says):
 | Limit | Default | When exceeded |
 |---|---|---|
 | Calls to one SDLC system (`resourceapi.rate-limit.sdlc.*`), protecting its API quota; cache hits don't count | 10 per 1s, waiting up to 1s for a permit | `503` + `Retry-After` (our own limit, not the SDLC system's failure); a background sync logs it |
-| `POST /api/v2/projects/{id}/sync` per project (`resourceapi.rate-limit.sync.*`); requests refused anyway (unknown project, sync already running) don't count, one refused because no background slot is free (`503`) does | 6 per minute | `429` + `Retry-After` |
+| `POST /api/v2/projects/{id}/sync` per project (`resourceapi.rate-limit.sync.*`); requests refused anyway (unknown project, sync already running) don't count, one refused because RabbitMQ is unreachable (`503`) does | 6 per minute | `429` + `Retry-After` |
 | All `/api/**` requests per client IP (`resourceapi.rate-limit.client.*`); behind a reverse proxy the IP comes from `X-Forwarded-For`, trusted only from internal proxy addresses (`server.forward-headers-strategy=native`, see below) | 100 per second | `429` + `Retry-After` |
 
 A changed rate is applied to the existing limiter when an instance first uses it; unused limiters expire from
@@ -118,9 +179,10 @@ load balancer has another address, set that property to match it; otherwise ever
 the proxy and all clients share one limit. Never match addresses clients can connect from directly, or they
 can choose their own key by sending `X-Forwarded-For`.
 
-**Local development**: `./mvnw spring-boot:run` starts Redis from `compose.yaml` automatically
-(`spring-boot-docker-compose`); with `java -jar`, start it yourself with `docker compose up -d` (it listens on
-`localhost:6379`). Tests start their own Redis with Testcontainers, so they need Docker.
+**Local development**: `./mvnw spring-boot:run` starts Redis and RabbitMQ from `compose.yaml` automatically
+(`spring-boot-docker-compose`); with `java -jar`, start them yourself with `docker compose up -d` (Redis on
+`localhost:6379`, RabbitMQ on `localhost:5672`, its management UI on http://localhost:15672, guest/guest).
+Tests start their own Redis and RabbitMQ with Testcontainers, so they need Docker.
 
 ## Tracing
 Every request gets a trace id ([Micrometer Tracing](https://micrometer.io/docs/tracing) with Brave, W3C trace context):
@@ -131,10 +193,9 @@ Every request gets a trace id ([Micrometer Tracing](https://micrometer.io/docs/t
 - outgoing `RestClient` calls (e.g. to SDLC systems) send `traceparent`, so the trace continues there
 
 - background work keeps the trace: `@Async` tasks run on virtual threads (`spring.threads.virtual.enabled`) and
-  `ContextPropagatingTaskDecorator` copies the caller's trace context to them. Example:
-  `POST /api/v2/projects/{id}/sync` answers `202 Accepted` and refreshes the project's name from its SDLC system
-  in the background; its log lines and its call to the SDLC system carry the request's trace id. At most
-  `spring.task.execution.simple.concurrency-limit` (8) syncs run at once; further requests get `503` with `Retry-After`
+  `ContextPropagatingTaskDecorator` copies the caller's trace context to them; RabbitMQ messages carry it in a
+  `traceparent` header (Spring AMQP observation) and their listeners continue it. Example: a queued sync's log
+  lines and its call to the SDLC system carry the trace id of the `POST …/sync` request
 
 No spans are exported; add a reporter (e.g. Zipkin or OTLP) to send them to a tracing backend, and lower
 `management.tracing.sampling.probability` (set to `1.0` here) to what that backend can take.

compose.yaml

@@ -5,3 +5,9 @@ services:
     image: redis:8-alpine
     ports:
       - "6379:6379"
+  rabbitmq:
+    # management UI on http://localhost:15672 (guest/guest)
+    image: rabbitmq:4-management-alpine
+    ports:
+      - "5672:5672"
+      - "15672:15672"

docs/docker.md

@@ -126,11 +126,13 @@ the build. The executable is built for the architecture of the machine (or `--pl
 ## Configuration
 
 All three images are configured like the application itself, e.g. with environment variables. The
-application uses Redis (see the README); point it at one with `SPRING_DATA_REDIS_HOST`/`SPRING_DATA_REDIS_PORT`
-(without Redis it still works, uncached):
+application uses Redis and RabbitMQ (see the README); point it at them with `SPRING_DATA_REDIS_HOST` and
+`SPRING_RABBITMQ_HOST` (and `_PORT`). Without Redis it still works, uncached; without RabbitMQ notifications
+wait in the database until it is reachable:
 
 ```sh
-docker run --rm -p 8080:8080 -e SPRING_DATA_REDIS_HOST=redis --network my-network resourceapi
+docker run --rm -p 8080:8080 -e SPRING_DATA_REDIS_HOST=redis -e SPRING_RABBITMQ_HOST=rabbitmq \
+    --network my-network resourceapi
 docker run --rm -p 9090:9090 -e SERVER_PORT=9090 resourceapi
 docker run --rm -p 8080:8080 -e SPRING_HTTP_CLIENTS_READ_TIMEOUT=10s resourceapi:aot
 ```

pom.xml

@@ -18,6 +18,7 @@
 		<java.version>25</java.version>
 		<springdoc-openapi.version>3.2.0</springdoc-openapi.version>
 		<redisson.version>4.8.0</redisson.version>
+		<spring-modulith.version>2.1.2</spring-modulith.version>
 		<wiremock.version>3.13.2</wiremock.version>
 		<jacoco.version>0.8.16</jacoco.version>
 		<!-- Overridden by agents such as JaCoCo; keeps @{argLine} resolvable when none is set -->
@@ -65,6 +66,21 @@
 			<artifactId>redisson</artifactId>
 			<version>${redisson.version}</version>
 		</dependency>
+		<dependency>
+			<!-- RabbitMQ: project notifications for other systems -->
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-amqp</artifactId>
+		</dependency>
+		<dependency>
+			<!-- Outbox: events are stored in the database with the change that caused them, then sent on -->
+			<groupId>org.springframework.modulith</groupId>
+			<artifactId>spring-modulith-events-jdbc</artifactId>
+		</dependency>
+		<dependency>
+			<!-- ... stored as JSON -->
+			<groupId>org.springframework.modulith</groupId>
+			<artifactId>spring-modulith-events-jackson</artifactId>
+		</dependency>
 		<dependency>
 			<!-- Pooled HTTP client for RestClient (Boot picks it up when present) -->
 			<groupId>org.apache.httpcomponents.client5</groupId>
@@ -125,6 +141,18 @@
 		</dependency>
 	</dependencies>
 
+	<dependencyManagement>
+		<dependencies>
+			<dependency>
+				<groupId>org.springframework.modulith</groupId>
+				<artifactId>spring-modulith-bom</artifactId>
+				<version>${spring-modulith.version}</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
+		</dependencies>
+	</dependencyManagement>
+
 	<build>
 		<plugins>
 			<plugin>

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,11 +18,40 @@
  */
 @Configuration
 @EnableAsync
+@Slf4j
 public class AsyncConfig {
 
     /** Boot applies a TaskDecorator bean to the application task executor. */
     @Bean
     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).
+     *
+     * <p>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());
+        }
+    }
 }

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

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

src/main/java/com/company/resourceapi/events/ProjectCreated.java

@@ -0,0 +1,13 @@
+package com.company.resourceapi.events;
+
+import java.time.Instant;
+
+/**
+ * A project was created. Published inside the creating transaction; listeners see it after commit.
+ *
+ * @param occurredAt when the change was made; events about one project can reach a listener (or another
+ *     system) out of order, so one older than the last applied is stale
+ */
+public record ProjectCreated(long projectId, long sdlcSystemId, String externalId, String name,
+        Instant occurredAt) {
+}

src/main/java/com/company/resourceapi/events/ProjectUpdated.java

@@ -0,0 +1,28 @@
+package com.company.resourceapi.events;
+
+import java.time.Instant;
+import java.util.Set;
+
+/**
+ * A project was changed, through the API or by a sync from its SDLC system; carries the new state and
+ * which fields changed. Not published when an update changes nothing.
+ *
+ * @param occurredAt when the change was made; events about one project can reach a listener (or another
+ *     system) out of order, so one older than the last applied is stale
+ */
+public record ProjectUpdated(long projectId, long sdlcSystemId, String externalId, String name,
+        Set<Field> changed, Instant occurredAt) {
+
+    public enum Field {
+        NAME, EXTERNAL_ID, SDLC_SYSTEM
+    }
+
+    public ProjectUpdated {
+        changed = Set.copyOf(changed);
+    }
+
+    /** Whether the project now points at another project in an SDLC system. */
+    public boolean isRekeyed() {
+        return changed.contains(Field.EXTERNAL_ID) || changed.contains(Field.SDLC_SYSTEM);
+    }
+}