From ccf503a79897c3ca0c84bbbf6a8dbea15b297edb Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 1 Nov 2026 05:55:31 +0000 Subject: [PATCH 1/5] Publish project events; sync a project's name after create or re-key ProjectService publishes ProjectCreated and ProjectUpdated (with the changed fields) inside the changing transaction. SyncOnProjectChange listens after commit and starts a background sync when a project is created or pointed at another SDLC project, so its name comes from the SDLC system without a separate sync request (resourceapi.sync.on-change, off in tests). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012MLyPXGHeCc1fvTeB8JnT2 --- README.md | 16 +++ .../resourceapi/events/ProjectCreated.java | 5 + .../resourceapi/events/ProjectUpdated.java | 24 ++++ .../services/SyncOnProjectChange.java | 44 ++++++ .../services/impl/ProjectServiceImpl.java | 29 +++- src/main/resources/application.properties | 2 + .../resourceapi/events/ProjectEventsTest.java | 110 +++++++++++++++ .../events/SyncOnProjectChangeTest.java | 129 ++++++++++++++++++ .../services/impl/ProjectServiceImplTest.java | 2 +- .../resources/config/application.properties | 3 + 10 files changed, 362 insertions(+), 2 deletions(-) create mode 100644 src/main/java/com/company/resourceapi/events/ProjectCreated.java create mode 100644 src/main/java/com/company/resourceapi/events/ProjectUpdated.java create mode 100644 src/main/java/com/company/resourceapi/services/SyncOnProjectChange.java create mode 100644 src/test/java/com/company/resourceapi/events/ProjectEventsTest.java create mode 100644 src/test/java/com/company/resourceapi/events/SyncOnProjectChangeTest.java diff --git a/README.md b/README.md index 3629212..c7c9728 100644 --- a/README.md +++ b/README.md @@ -65,6 +65,22 @@ 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. + +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). +Turn it off with `resourceapi.sync.on-change=false` (tests do, except `SyncOnProjectChangeTest`). + ## 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` diff --git a/src/main/java/com/company/resourceapi/events/ProjectCreated.java b/src/main/java/com/company/resourceapi/events/ProjectCreated.java new file mode 100644 index 0000000..1472a50 --- /dev/null +++ b/src/main/java/com/company/resourceapi/events/ProjectCreated.java @@ -0,0 +1,5 @@ +package com.company.resourceapi.events; + +/** A project was created. Published inside the creating transaction; listeners see it after commit. */ +public record ProjectCreated(long projectId, long sdlcSystemId, String externalId, String name) { +} diff --git a/src/main/java/com/company/resourceapi/events/ProjectUpdated.java b/src/main/java/com/company/resourceapi/events/ProjectUpdated.java new file mode 100644 index 0000000..96fc5f6 --- /dev/null +++ b/src/main/java/com/company/resourceapi/events/ProjectUpdated.java @@ -0,0 +1,24 @@ +package com.company.resourceapi.events; + +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. + */ +public record ProjectUpdated(long projectId, long sdlcSystemId, String externalId, String name, + Set changed) { + + 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); + } +} diff --git a/src/main/java/com/company/resourceapi/services/SyncOnProjectChange.java b/src/main/java/com/company/resourceapi/services/SyncOnProjectChange.java new file mode 100644 index 0000000..f913c4c --- /dev/null +++ b/src/main/java/com/company/resourceapi/services/SyncOnProjectChange.java @@ -0,0 +1,44 @@ +package com.company.resourceapi.services; + +import com.company.resourceapi.events.ProjectCreated; +import com.company.resourceapi.events.ProjectUpdated; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; +import org.springframework.stereotype.Component; +import org.springframework.transaction.event.TransactionalEventListener; + +/** + * Fetches a project's name from its SDLC system when the project is created or pointed at another + * 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. + */ +@Component +@ConditionalOnBooleanProperty(name = "resourceapi.sync.on-change", matchIfMissing = true) +@RequiredArgsConstructor +@Slf4j +public class SyncOnProjectChange { + + private final ProjectSyncService projectSyncService; + + @TransactionalEventListener + void onCreated(ProjectCreated event) { + requestSync(event.projectId()); + } + + @TransactionalEventListener(condition = "#event.rekeyed") + void onRekeyed(ProjectUpdated event) { + requestSync(event.projectId()); + } + + private void requestSync(long projectId) { + try { + projectSyncService.requestSync(projectId); + } catch (Exception e) { + log.info("Sync of project {} after its change not started: {}", projectId, e.getMessage()); + } + } +} diff --git a/src/main/java/com/company/resourceapi/services/impl/ProjectServiceImpl.java b/src/main/java/com/company/resourceapi/services/impl/ProjectServiceImpl.java index 8aef1bf..21fba70 100755 --- a/src/main/java/com/company/resourceapi/services/impl/ProjectServiceImpl.java +++ b/src/main/java/com/company/resourceapi/services/impl/ProjectServiceImpl.java @@ -1,14 +1,20 @@ package com.company.resourceapi.services.impl; import com.company.resourceapi.entities.Project; +import com.company.resourceapi.events.ProjectCreated; +import com.company.resourceapi.events.ProjectUpdated; +import com.company.resourceapi.events.ProjectUpdated.Field; import com.company.resourceapi.exceptions.ConflictException; import com.company.resourceapi.exceptions.NotFoundException; import com.company.resourceapi.repositories.ProjectRepository; import com.company.resourceapi.services.ProjectService; import jakarta.validation.ConstraintViolationException; import jakarta.validation.Validator; +import java.util.EnumSet; +import java.util.Objects; import java.util.Optional; import lombok.RequiredArgsConstructor; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @@ -17,13 +23,18 @@ public class ProjectServiceImpl implements ProjectService { private final ProjectRepository projectRepository; private final Validator validator; + private final ApplicationEventPublisher events; public Project getProject(long id) throws NotFoundException { return getProjectStream(id).orElseThrow(() -> new NotFoundException(Project.class, id)); } + @Transactional public Project createProject(Project project) { - return projectRepository.save(project); + var created = projectRepository.save(project); + events.publishEvent(new ProjectCreated(created.getId(), created.getSdlcSystem().getId(), + created.getExternalId(), created.getName())); + return created; } public Optional getProject(String externalId, long sdlcSystemId) { @@ -40,6 +51,7 @@ public Project updateProject(long id, ProjectUpdate update) throws NotFoundExcep var project = getProject(id); var externalIdBefore = project.getExternalId(); var sdlcSystemBefore = project.getSdlcSystem(); + var nameBefore = project.getName(); update.applyTo(project); var violations = validator.validate(project); @@ -55,6 +67,21 @@ public Project updateProject(long id, ProjectUpdate update) throws NotFoundExcep .formatted(project.getExternalId(), project.getSdlcSystem().getId())); } + var changed = EnumSet.noneOf(Field.class); + if (!Objects.equals(project.getName(), nameBefore)) { + changed.add(Field.NAME); + } + if (!project.getExternalId().equals(externalIdBefore)) { + changed.add(Field.EXTERNAL_ID); + } + if (project.getSdlcSystem().getId() != sdlcSystemBefore.getId()) { + changed.add(Field.SDLC_SYSTEM); + } + if (!changed.isEmpty()) { + events.publishEvent(new ProjectUpdated(id, project.getSdlcSystem().getId(), + project.getExternalId(), project.getName(), changed)); + } + // Managed entity: flushed on commit, only the changed columns (@DynamicUpdate) return project; } diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index ade8268..2ccfddf 100755 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -27,6 +27,8 @@ resourceapi.cache.external-projects-ttl=5m resourceapi.cache.missing-external-projects-ttl=30s # redisson (locks, rate limiters) uses the same Redis; its object names get this prefix 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) resourceapi.sync.lock-lease-time=60s # distributed rate limits (Redisson): calls to one SDLC system (503 + Retry-After when exceeded after diff --git a/src/test/java/com/company/resourceapi/events/ProjectEventsTest.java b/src/test/java/com/company/resourceapi/events/ProjectEventsTest.java new file mode 100644 index 0000000..2e7b897 --- /dev/null +++ b/src/test/java/com/company/resourceapi/events/ProjectEventsTest.java @@ -0,0 +1,110 @@ +package com.company.resourceapi.events; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.company.resourceapi.entities.Project; +import com.company.resourceapi.entities.SdlcSystem; +import com.company.resourceapi.events.ProjectUpdated.Field; +import com.company.resourceapi.exceptions.ConflictException; +import com.company.resourceapi.repositories.SdlcSystemRepository; +import com.company.resourceapi.services.ProjectService; +import java.util.Set; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.event.ApplicationEvents; +import org.springframework.test.context.event.RecordApplicationEvents; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Which events ProjectService publishes. */ +@SpringBootTest +@RecordApplicationEvents +class ProjectEventsTest { + + @Autowired + private ProjectService projectService; + + @Autowired + private SdlcSystemRepository sdlcSystemRepository; + + @Autowired + private ApplicationEvents events; + + private SdlcSystem sdlcSystem; + private SdlcSystem otherSdlcSystem; + + @BeforeEach + void setup() { + sdlcSystem = sdlcSystemRepository.save(newSdlcSystem()); + otherSdlcSystem = sdlcSystemRepository.save(newSdlcSystem()); + } + + @Test + void givenCreatedProjectThenProjectCreatedIsPublished() { + var project = projectService.createProject(newProject("EVENTS-1", "Name")); + + assertThat(events.stream(ProjectCreated.class)).containsExactly( + new ProjectCreated(project.getId(), sdlcSystem.getId(), "EVENTS-1", "Name")); + } + + @Test + void givenRenamedProjectThenProjectUpdatedNamesTheChange() throws Exception { + var id = projectService.createProject(newProject("EVENTS-2", "Name")).getId(); + + projectService.updateProject(id, project -> project.setName("New name")); + + assertThat(events.stream(ProjectUpdated.class)).containsExactly( + new ProjectUpdated(id, sdlcSystem.getId(), "EVENTS-2", "New name", Set.of(Field.NAME))); + } + + @Test + void givenProjectMovedToAnotherSdlcProjectThenItIsRekeyed() throws Exception { + var id = projectService.createProject(newProject("EVENTS-3", "Name")).getId(); + + projectService.updateProject(id, project -> { + project.setExternalId("EVENTS-3B"); + project.setSdlcSystem(otherSdlcSystem); + }); + + var event = events.stream(ProjectUpdated.class).findFirst().orElseThrow(); + assertThat(event.changed()).containsExactlyInAnyOrder(Field.EXTERNAL_ID, Field.SDLC_SYSTEM); + assertThat(event.isRekeyed()).isTrue(); + assertThat(event.sdlcSystemId()).isEqualTo(otherSdlcSystem.getId()); + } + + @Test + void givenUpdateThatChangesNothingThenNothingIsPublished() throws Exception { + var id = projectService.createProject(newProject("EVENTS-4", "Name")).getId(); + + projectService.updateProject(id, project -> project.setName("Name")); + + assertThat(events.stream(ProjectUpdated.class)).isEmpty(); + } + + @Test + void givenRejectedUpdateThenNothingIsPublished() throws Exception { + projectService.createProject(newProject("EVENTS-5", "Name")); + var id = projectService.createProject(newProject("EVENTS-6", "Name")).getId(); + + assertThatThrownBy(() -> projectService.updateProject(id, project -> project.setExternalId("EVENTS-5"))) + .isInstanceOf(ConflictException.class); + + assertThat(events.stream(ProjectUpdated.class)).isEmpty(); + } + + private Project newProject(String externalId, String name) { + var project = new Project(); + project.setExternalId(externalId); + project.setName(name); + project.setSdlcSystem(sdlcSystem); + return project; + } + + private static SdlcSystem newSdlcSystem() { + var system = new SdlcSystem(); + system.setBaseUrl("http://sdlc.example.com"); + return system; + } +} diff --git a/src/test/java/com/company/resourceapi/events/SyncOnProjectChangeTest.java b/src/test/java/com/company/resourceapi/events/SyncOnProjectChangeTest.java new file mode 100644 index 0000000..ee6652b --- /dev/null +++ b/src/test/java/com/company/resourceapi/events/SyncOnProjectChangeTest.java @@ -0,0 +1,129 @@ +package com.company.resourceapi.events; + +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 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.ProjectRepository; +import com.company.resourceapi.repositories.SdlcSystemRepository; +import com.github.tomakehurst.wiremock.client.WireMock; +import com.github.tomakehurst.wiremock.junit5.WireMockExtension; +import java.time.Duration; +import java.time.Instant; +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.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc; +import org.springframework.http.MediaType; +import org.springframework.test.web.servlet.MockMvc; + +/** A project's name is fetched from its SDLC system after it is created or re-keyed. */ +@SpringBootTest(properties = "resourceapi.sync.on-change=true") +@AutoConfigureMockMvc +class SyncOnProjectChangeTest { + + @RegisterExtension + static WireMockExtension sdlcSystemServer = WireMockExtension.newInstance() + .options(wireMockConfig().dynamicPort()) + .build(); + + @Autowired + private MockMvc mockMvc; + + @Autowired + private SdlcSystemRepository sdlcSystemRepository; + + @Autowired + private ProjectRepository projectRepository; + + private SdlcSystem sdlcSystem; + + @BeforeEach + void setup() { + sdlcSystem = new SdlcSystem(); + sdlcSystem.setBaseUrl(sdlcSystemServer.baseUrl()); + sdlcSystem = sdlcSystemRepository.save(sdlcSystem); + stubProject("FIRST", "First from SDLC"); + stubProject("SECOND", "Second from SDLC"); + } + + @Test + void givenProjectCreatedWithoutNameThenNameComesFromSdlcSystem() throws Exception { + var id = create("FIRST"); + + await(() -> "First from SDLC".equals(nameOf(id))); + } + + @Test + void givenProjectPointedAtAnotherSdlcProjectThenItsNameIsFetched() throws Exception { + var id = create("FIRST"); + await(() -> "First from SDLC".equals(nameOf(id))); + + patchProject(id, "{\"externalId\": \"SECOND\"}"); + + await(() -> "Second from SDLC".equals(nameOf(id))); + } + + @Test + void givenOnlyNameChangedThenSdlcSystemIsNotAsked() throws Exception { + var id = create("FIRST"); + await(() -> "First from SDLC".equals(nameOf(id))); + // the sync's own rename doesn't start another sync either + int calls = sdlcCalls(); + + patchProject(id, "{\"name\": \"Renamed here\"}"); + + Thread.sleep(300); + assertThat(sdlcCalls()).isEqualTo(calls); + assertThat(nameOf(id)).isEqualTo("Renamed here"); + } + + private long create(String externalId) throws Exception { + var location = mockMvc.perform(post(ProjectRestController.ENDPOINT) + .contentType(MediaType.APPLICATION_JSON) + .content("{\"externalId\": \"%s\", \"sdlcSystem\": {\"id\": %d}}" + .formatted(externalId, sdlcSystem.getId()))) + .andExpect(status().isCreated()) + .andReturn().getResponse().getHeader("Location"); + return Long.parseLong(location.substring(location.lastIndexOf('/') + 1)); + } + + private void patchProject(long id, String body) throws Exception { + mockMvc.perform(patch(ProjectRestController.ENDPOINT + "/" + id) + .contentType(MediaType.APPLICATION_JSON) + .content(body)) + .andExpect(status().isOk()); + } + + private String nameOf(long id) { + return projectRepository.findById(id).orElseThrow().getName(); + } + + private static int sdlcCalls() { + return sdlcSystemServer.findAll(getRequestedFor(urlPathMatching("/rest/api/2/project/.*"))).size(); + } + + private static void stubProject(String key, String name) { + sdlcSystemServer.stubFor(WireMock.get(urlEqualTo("/rest/api/2/project/" + key)) + .willReturn(okJson("{\"key\": \"%s\", \"name\": \"%s\"}".formatted(key, name)))); + } + + 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/impl/ProjectServiceImplTest.java b/src/test/java/com/company/resourceapi/services/impl/ProjectServiceImplTest.java index 85716a7..0c724c1 100755 --- a/src/test/java/com/company/resourceapi/services/impl/ProjectServiceImplTest.java +++ b/src/test/java/com/company/resourceapi/services/impl/ProjectServiceImplTest.java @@ -36,7 +36,7 @@ public class ProjectServiceImplTest { @BeforeEach void setup() { projectService = new ProjectServiceImpl(projectRepository, - Validation.buildDefaultValidatorFactory().getValidator()); + Validation.buildDefaultValidatorFactory().getValidator(), event -> { }); var sdlcSystem = new SdlcSystem(); sdlcSystem.setId(1); diff --git a/src/test/resources/config/application.properties b/src/test/resources/config/application.properties index 67b2d67..fb0015f 100644 --- a/src/test/resources/config/application.properties +++ b/src/test/resources/config/application.properties @@ -2,3 +2,6 @@ spring.datasource.url=jdbc:h2:mem:test-${random.uuid};MODE=MySQL;DATABASE_TO_LOWER=TRUE # tests send many requests from one address; ClientRateLimitIntegrationTest sets its own limit 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 From 7e80f34739802828fa30016a6ecd0eb8f1b2d94c Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 1 Nov 2026 06:11:25 +0000 Subject: [PATCH 2/5] Send project events to RabbitMQ through an outbox Spring Modulith's JDBC event publication registry stores ProjectCreated and ProjectUpdated in event_publication within the changing transaction. ProjectNotifier sends them after commit to the resourceapi.projects topic exchange (project.created / project.updated, JSON, traceparent) on its own virtual-thread executor, and only counts a send once the broker confirmed it; the row is then deleted. UnsentNotifications resends what is left, one instance at a time (Redis lock). Delivery is at least once. Modulith's RabbitMQ externalization isn't used: in module-listener mode it doesn't wait for publisher confirms, so a refused message was counted as sent (caught by the test that deletes the exchange). Also: spring.task.execution.mode=force keeps Boot's application task executor next to the notification executor; failed fire-and-forget tasks are logged in one line. RabbitMQ in compose.yaml and in tests (Testcontainers, an exchange per test context). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012MLyPXGHeCc1fvTeB8JnT2 --- README.md | 29 ++- compose.yaml | 6 + docs/docker.md | 8 +- pom.xml | 28 +++ .../resourceapi/config/AsyncConfig.java | 35 ++++ .../notifications/NotificationsConfig.java | 59 ++++++ .../NotificationsProperties.java | 24 +++ .../notifications/ProjectNotifier.java | 54 ++++++ .../notifications/UnsentNotifications.java | 69 +++++++ src/main/resources/application.properties | 14 ++ src/main/resources/schema.sql | 16 ++ .../RabbitTestContainerInitializer.java | 45 +++++ .../NotificationsIntegrationTest.java | 170 ++++++++++++++++++ src/test/resources/META-INF/spring.factories | 3 +- 14 files changed, 553 insertions(+), 7 deletions(-) create mode 100644 src/main/java/com/company/resourceapi/notifications/NotificationsConfig.java create mode 100644 src/main/java/com/company/resourceapi/notifications/NotificationsProperties.java create mode 100644 src/main/java/com/company/resourceapi/notifications/ProjectNotifier.java create mode 100644 src/main/java/com/company/resourceapi/notifications/UnsentNotifications.java create mode 100644 src/test/java/com/company/resourceapi/RabbitTestContainerInitializer.java create mode 100644 src/test/java/com/company/resourceapi/notifications/NotificationsIntegrationTest.java diff --git a/README.md b/README.md index c7c9728..4763c8d 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,28 @@ The listener runs on the request thread after commit; it only starts the sync, b lock attempt adds about a second to the request (see Lock below). Turn it off with `resourceapi.sync.on-change=false` (tests do, except `SyncOnProjectChangeTest`). +## 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 (its own virtual-thread executor, so a slow broker + doesn't hold up requests) and waits for the broker's confirmation (publisher confirms). Only then is the row + deleted. +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). + +Delivery is therefore 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` @@ -134,9 +156,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): diff --git a/compose.yaml b/compose.yaml index 942b34f..7042d06 100644 --- a/compose.yaml +++ b/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" diff --git a/docs/docker.md b/docs/docker.md index 0805423..6c2c53a 100644 --- a/docs/docker.md +++ b/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 ``` diff --git a/pom.xml b/pom.xml index 98f9ef3..80c6358 100644 --- a/pom.xml +++ b/pom.xml @@ -18,6 +18,7 @@ 25 3.2.0 4.8.0 + 2.1.2 3.13.2 0.8.16 @@ -65,6 +66,21 @@ redisson ${redisson.version} + + + org.springframework.boot + spring-boot-starter-amqp + + + + org.springframework.modulith + spring-modulith-events-jdbc + + + + org.springframework.modulith + spring-modulith-events-jackson + org.apache.httpcomponents.client5 @@ -125,6 +141,18 @@ + + + + org.springframework.modulith + spring-modulith-bom + ${spring-modulith.version} + pom + import + + + + 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/notifications/NotificationsConfig.java b/src/main/java/com/company/resourceapi/notifications/NotificationsConfig.java new file mode 100644 index 0000000..7161e5d --- /dev/null +++ b/src/main/java/com/company/resourceapi/notifications/NotificationsConfig.java @@ -0,0 +1,59 @@ +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: its limit is for syncs, and a burst of + * project changes shouldn't get notifications rejected. Virtual threads, in the caller's trace; at + * most 16 sends at once, further ones wait for a free slot. + */ + @Bean + SimpleAsyncTaskExecutor notificationExecutor() { + var executor = new SimpleAsyncTaskExecutor("notify-"); + executor.setVirtualThreads(true); + executor.setConcurrencyLimit(16); + 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/resources/application.properties b/src/main/resources/application.properties index 2ccfddf..7ed62ab 100755 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -6,6 +6,8 @@ management.tracing.sampling.probability=1.0 # virtual threads for request handling and the @Async task executor spring.threads.virtual.enabled=true # at most 8 background tasks at once; more are rejected (503) instead of piling up +# 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 @@ -45,6 +47,18 @@ 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 +spring.rabbitmq.template.observation-enabled=true +# 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..1e59a4f --- /dev/null +++ b/src/test/java/com/company/resourceapi/RabbitTestContainerInitializer.java @@ -0,0 +1,45 @@ +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: 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 + * 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 + propertySources.addFirst(new MapPropertySource("rabbitTestContainerExchange", Map.of( + "resourceapi.notifications.exchange", "test-" + UUID.randomUUID()))); + } + + 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..bc33fa1 --- /dev/null +++ b/src/test/java/com/company/resourceapi/notifications/NotificationsIntegrationTest.java @@ -0,0 +1,170 @@ +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.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.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 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); + } + + 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/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 From 8b8113aca7079e40faf36220b186b32659846d84 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 1 Nov 2026 06:20:43 +0000 Subject: [PATCH 3/5] 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 From 86da1631092e87fa120910f71df5cc6e36f9808c Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 1 Nov 2026 08:05:07 +0000 Subject: [PATCH 4/5] Add occurredAt to project events Events about one project can reach a listener or another system out of order (sent in parallel, or retried later), so each carries when the change was made and an event older than the last applied can be discarded. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012MLyPXGHeCc1fvTeB8JnT2 --- README.md | 3 +++ .../resourceapi/events/ProjectCreated.java | 12 ++++++++++-- .../resourceapi/events/ProjectUpdated.java | 6 +++++- .../services/impl/ProjectServiceImpl.java | 5 +++-- .../resourceapi/events/ProjectEventsTest.java | 16 ++++++++++++---- 5 files changed, 33 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index c7c9728..cbae985 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,9 @@ changes a project: - `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 diff --git a/src/main/java/com/company/resourceapi/events/ProjectCreated.java b/src/main/java/com/company/resourceapi/events/ProjectCreated.java index 1472a50..06c2142 100644 --- a/src/main/java/com/company/resourceapi/events/ProjectCreated.java +++ b/src/main/java/com/company/resourceapi/events/ProjectCreated.java @@ -1,5 +1,13 @@ package com.company.resourceapi.events; -/** A project was created. Published inside the creating transaction; listeners see it after commit. */ -public record ProjectCreated(long projectId, long sdlcSystemId, String externalId, String name) { +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) { } diff --git a/src/main/java/com/company/resourceapi/events/ProjectUpdated.java b/src/main/java/com/company/resourceapi/events/ProjectUpdated.java index 96fc5f6..4928aa0 100644 --- a/src/main/java/com/company/resourceapi/events/ProjectUpdated.java +++ b/src/main/java/com/company/resourceapi/events/ProjectUpdated.java @@ -1,13 +1,17 @@ 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 changed) { + Set changed, Instant occurredAt) { public enum Field { NAME, EXTERNAL_ID, SDLC_SYSTEM diff --git a/src/main/java/com/company/resourceapi/services/impl/ProjectServiceImpl.java b/src/main/java/com/company/resourceapi/services/impl/ProjectServiceImpl.java index 21fba70..5492e87 100755 --- a/src/main/java/com/company/resourceapi/services/impl/ProjectServiceImpl.java +++ b/src/main/java/com/company/resourceapi/services/impl/ProjectServiceImpl.java @@ -10,6 +10,7 @@ import com.company.resourceapi.services.ProjectService; import jakarta.validation.ConstraintViolationException; import jakarta.validation.Validator; +import java.time.Instant; import java.util.EnumSet; import java.util.Objects; import java.util.Optional; @@ -33,7 +34,7 @@ public Project getProject(long id) throws NotFoundException { public Project createProject(Project project) { var created = projectRepository.save(project); events.publishEvent(new ProjectCreated(created.getId(), created.getSdlcSystem().getId(), - created.getExternalId(), created.getName())); + created.getExternalId(), created.getName(), Instant.now())); return created; } @@ -79,7 +80,7 @@ public Project updateProject(long id, ProjectUpdate update) throws NotFoundExcep } if (!changed.isEmpty()) { events.publishEvent(new ProjectUpdated(id, project.getSdlcSystem().getId(), - project.getExternalId(), project.getName(), changed)); + project.getExternalId(), project.getName(), changed, Instant.now())); } // Managed entity: flushed on commit, only the changed columns (@DynamicUpdate) diff --git a/src/test/java/com/company/resourceapi/events/ProjectEventsTest.java b/src/test/java/com/company/resourceapi/events/ProjectEventsTest.java index 2e7b897..0693b39 100644 --- a/src/test/java/com/company/resourceapi/events/ProjectEventsTest.java +++ b/src/test/java/com/company/resourceapi/events/ProjectEventsTest.java @@ -8,6 +8,7 @@ import com.company.resourceapi.exceptions.ConflictException; import com.company.resourceapi.repositories.SdlcSystemRepository; import com.company.resourceapi.services.ProjectService; +import java.time.Instant; import java.util.Set; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -43,10 +44,13 @@ void setup() { @Test void givenCreatedProjectThenProjectCreatedIsPublished() { + var before = Instant.now(); var project = projectService.createProject(newProject("EVENTS-1", "Name")); - assertThat(events.stream(ProjectCreated.class)).containsExactly( - new ProjectCreated(project.getId(), sdlcSystem.getId(), "EVENTS-1", "Name")); + var event = events.stream(ProjectCreated.class).findFirst().orElseThrow(); + assertThat(event).usingRecursiveComparison().ignoringFields("occurredAt").isEqualTo( + new ProjectCreated(project.getId(), sdlcSystem.getId(), "EVENTS-1", "Name", null)); + assertThat(event.occurredAt()).isBetween(before, Instant.now()); } @Test @@ -55,8 +59,12 @@ void givenRenamedProjectThenProjectUpdatedNamesTheChange() throws Exception { projectService.updateProject(id, project -> project.setName("New name")); - assertThat(events.stream(ProjectUpdated.class)).containsExactly( - new ProjectUpdated(id, sdlcSystem.getId(), "EVENTS-2", "New name", Set.of(Field.NAME))); + var event = events.stream(ProjectUpdated.class).findFirst().orElseThrow(); + assertThat(event).usingRecursiveComparison().ignoringFields("occurredAt").isEqualTo( + new ProjectUpdated(id, sdlcSystem.getId(), "EVENTS-2", "New name", Set.of(Field.NAME), null)); + // later than the creation's + assertThat(event.occurredAt()).isAfterOrEqualTo(events.stream(ProjectCreated.class).findFirst() + .orElseThrow().occurredAt()); } @Test From 83bbaa181c7a80ddee72fd8afbda54d39fd019b9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 1 Nov 2026 08:10:14 +0000 Subject: [PATCH 5/5] Refuse notification sends when all senders are busy instead of waiting With 16 sends each waiting for an unreachable broker, further sends made the committing request wait for a free slot, coupling API latency to the broker. A refused send leaves its event in the outbox, sent again later. README: consumers should order notifications by occurredAt. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012MLyPXGHeCc1fvTeB8JnT2 --- README.md | 11 ++-- .../notifications/NotificationsConfig.java | 8 ++- .../NotificationsIntegrationTest.java | 60 +++++++++++++++++++ 3 files changed, 72 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 4627c75..a7d2d64 100644 --- a/README.md +++ b/README.md @@ -94,14 +94,17 @@ No notification is lost, even if RabbitMQ is down or the application stops right 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 (its own virtual-thread executor, so a slow broker - doesn't hold up requests) and waits for the broker's confirmation (publisher confirms). Only then is the row - deleted. +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). -Delivery is therefore at least once: a consumer may see the same notification twice (e.g. the broker confirmed +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. diff --git a/src/main/java/com/company/resourceapi/notifications/NotificationsConfig.java b/src/main/java/com/company/resourceapi/notifications/NotificationsConfig.java index 7161e5d..5a7739a 100644 --- a/src/main/java/com/company/resourceapi/notifications/NotificationsConfig.java +++ b/src/main/java/com/company/resourceapi/notifications/NotificationsConfig.java @@ -38,15 +38,17 @@ TopicExchange projectsExchange(NotificationsProperties properties) { } /** - * Sends run here, not on the application task executor: its limit is for syncs, and a burst of - * project changes shouldn't get notifications rejected. Virtual threads, in the caller's trace; at - * most 16 sends at once, further ones wait for a free slot. + * 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; } diff --git a/src/test/java/com/company/resourceapi/notifications/NotificationsIntegrationTest.java b/src/test/java/com/company/resourceapi/notifications/NotificationsIntegrationTest.java index bc33fa1..b97a694 100644 --- a/src/test/java/com/company/resourceapi/notifications/NotificationsIntegrationTest.java +++ b/src/test/java/com/company/resourceapi/notifications/NotificationsIntegrationTest.java @@ -12,6 +12,7 @@ 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; @@ -25,6 +26,7 @@ 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; @@ -57,6 +59,9 @@ class NotificationsIntegrationTest { @Autowired private UnsentNotifications unsentNotifications; + @Autowired + private SimpleAsyncTaskExecutor notificationExecutor; + @Autowired private JdbcTemplate jdbcTemplate; @@ -128,6 +133,61 @@ void givenNotificationThatCouldNotBeSentThenItIsSentLater() throws Exception { 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)