test(webhook): re-enable WebhookE2ETest через no-op scheduler + manual ticks
Fix CI flakiness: тест больше не полагается на @Scheduled тики каждые 200ms. NoSchedulingConfig подменяет TaskScheduler на no-op stub — @Scheduled методы регистрируются, но никогда не запускаются. Тест вручную дёргает dispatcher.matchTick() / sendTick() — synchronous, deterministic, нет race с background threads. Что изменилось: - @Disabled убран - Inject WebhookDispatcher, manual matchTick + sendTick после setup - Await timeouts: 60s → 5s (manual ticks → нет 200ms scheduler delay) - Pool=4 / send-interval-ms / match-interval-ms убраны (не нужны) - Send-timeout-ms 1000 → 2000 (с запасом, но scheduler не работает) - @AutoConfigureMockMvc сохранён, port=RANDOM_PORT, allowed-hosts тот же Side-fix: @Autowired на Spring constructor WebhookTestPingService (добавлен ранее package-private constructor для unit-тестов сделал выбор конструктора неоднозначным; Spring 6 strict — требует explicit @Autowired когда есть >1 candidate). Локально: 2/2 PASSED 1.85s (было 9.9s — 5× быстрее). CI: 14 e2e всех тестов passed (Smoke, OutboxKafka, Auth, Bitemporal, Webhook).
This commit is contained in:
Zimin A.N.
+103
-64
@@ -6,6 +6,7 @@ import cloud.nstart.terravault.ordinis.domain.webhook.WebhookDeliveryRepository;
|
||||
import cloud.nstart.terravault.ordinis.domain.webhook.WebhookSubscription;
|
||||
import cloud.nstart.terravault.ordinis.domain.webhook.WebhookSubscriptionRepository;
|
||||
import cloud.nstart.terravault.ordinis.outbox.webhook.HmacSigner;
|
||||
import cloud.nstart.terravault.ordinis.outbox.webhook.WebhookDispatcher;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.sun.net.httpserver.HttpServer;
|
||||
@@ -15,7 +16,12 @@ import org.junit.jupiter.api.Test;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.boot.test.context.TestConfiguration;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.scheduling.TaskScheduler;
|
||||
import org.springframework.scheduling.Trigger;
|
||||
import org.springframework.test.context.ActiveProfiles;
|
||||
import org.springframework.test.context.DynamicPropertyRegistry;
|
||||
import org.springframework.test.context.DynamicPropertySource;
|
||||
@@ -25,10 +31,14 @@ import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.Delayed;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.awaitility.Awaitility.await;
|
||||
@@ -42,13 +52,23 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
|
||||
* imitate consumer's webhook receiver.</li>
|
||||
* <li>Создаём WebhookSubscription указывающий на этот endpoint.</li>
|
||||
* <li>POST a record — попадает в outbox_events.</li>
|
||||
* <li>WebhookDispatcher должен matchTick → INSERT delivery →
|
||||
* sendTick → POST на наш HttpServer → mark delivered.</li>
|
||||
* <li>Тест вручную вызывает {@code dispatcher.matchTick()} и
|
||||
* {@code dispatcher.sendTick()} — никакого фонового scheduler'а.</li>
|
||||
* <li>Проверяем: receiver получил body + valid HMAC signature +
|
||||
* Ordinis headers; delivery row mark'нут как delivered.</li>
|
||||
* </ol>
|
||||
*
|
||||
* <p>SSRF guard разрешает localhost через {@code allowed-hosts} config.
|
||||
* <p><b>Почему no-op scheduler:</b> ранее тест полагался на @Scheduled tick'и
|
||||
* с интервалом 200ms. На CI postgres testcontainer с reuse=true тащил leftover
|
||||
* subscriptions с dead URLs из предыдущих runs — scheduler в фоне забивал
|
||||
* thread pool retry'ями ДО того как наш @BeforeEach успевал truncate. Result:
|
||||
* matchTick на наш event не доходил, ConditionTimeout 60s. Три итерации фиксов
|
||||
* (cleanup leftovers, pool=4, send-timeout=1s) лечили симптомы, не причину.
|
||||
*
|
||||
* <p>Финальный fix: {@link NoSchedulingConfig} подменяет {@link TaskScheduler}
|
||||
* на no-op. @Scheduled методы регистрируются, но никогда не запускаются.
|
||||
* Тест вручную дёргает matchTick/sendTick — synchronous, deterministic. Нет
|
||||
* race с background threads, нет dependency на 200ms timer'ы.
|
||||
*/
|
||||
@SpringBootTest(
|
||||
webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
|
||||
@@ -57,16 +77,10 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
|
||||
// dispatcher читает outbox_events напрямую, Kafka не нужна.
|
||||
"ordinis.outbox.enabled=false",
|
||||
"ordinis.webhook.enabled=true",
|
||||
"ordinis.webhook.match-interval-ms=200",
|
||||
"ordinis.webhook.send-interval-ms=200",
|
||||
// Send timeout оставляем агрессивным на случай если manual sendTick
|
||||
// встретит leftover dead URL (хоть и cleanup в @BeforeEach должен убрать).
|
||||
"ordinis.webhook.send-timeout-ms=2000",
|
||||
"ordinis.webhook.allowed-hosts=localhost,127.0.0.1",
|
||||
// CI: postgres testcontainer withReuse(true) тащит leftover
|
||||
// deliveries с dead URLs из предыдущих runs. Default send-timeout
|
||||
// 5000ms × N leftovers + scheduling pool=1 → блокирует matchTick
|
||||
// на минуты, наш event не доходит до receiver. Жмём timeout до
|
||||
// 1s + thread pool 4 чтобы matchTick/sendTick не сериализовались.
|
||||
"ordinis.webhook.send-timeout-ms=1000",
|
||||
"spring.task.scheduling.pool.size=4",
|
||||
// Spring auto-config Kafka не нужно — нет KafkaTemplate потребителей
|
||||
// без OutboxPoller. Иначе при старте Boot попытается подключиться
|
||||
// к localhost:9092 и завесит контекст.
|
||||
@@ -75,18 +89,6 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
|
||||
})
|
||||
@AutoConfigureMockMvc
|
||||
@ActiveProfiles("test")
|
||||
@org.junit.jupiter.api.Disabled(
|
||||
"Flaky на CI runner: ConditionTimeout 60s × 2 теста несмотря на 3 итерации " +
|
||||
"fix'ов (cleanup leftovers @BeforeEach, scheduling pool=4, send-timeout=1s, " +
|
||||
"truncate outbox_events). Локально 2/2 PASS за 9.9s. Корень — race между " +
|
||||
"Spring @Scheduled threads, dispatcher @PostConstruct, testcontainer reuse " +
|
||||
"и наш @BeforeEach truncate. Логика покрыта 26 unit tests (HmacSigner + " +
|
||||
"SsrfGuard + DispatcherMatch + DeliveryLifecycle), e2e добавляет sanity " +
|
||||
"на full pipeline — недостаточный ROI чтобы фиксить дальше. " +
|
||||
"Re-enable: либо мигрировать на dedicated CI runner с warm Docker images, " +
|
||||
"либо переписать без testcontainer reuse (slower CI), либо использовать " +
|
||||
"@Sql(executionPhase = BEFORE_TEST_CLASS) чтобы truncate бежал ДО Spring " +
|
||||
"context инициализации dispatcher'а.")
|
||||
class WebhookE2ETest {
|
||||
|
||||
@DynamicPropertySource
|
||||
@@ -96,11 +98,28 @@ class WebhookE2ETest {
|
||||
E2ESupport.registerPostgresOnlyProperties(registry);
|
||||
}
|
||||
|
||||
/**
|
||||
* Подменяет TaskScheduler на no-op. Spring ScheduledAnnotationBeanPostProcessor
|
||||
* вызовет {@code scheduleWithFixedDelay} для matchTick/sendTick — наш stub
|
||||
* вернёт cancelled future, ничего не запустит.
|
||||
*
|
||||
* <p>marked @Primary чтобы перебить дефолтный Spring Boot bean.
|
||||
*/
|
||||
@TestConfiguration
|
||||
static class NoSchedulingConfig {
|
||||
@Bean
|
||||
@Primary
|
||||
TaskScheduler noopTaskScheduler() {
|
||||
return new NoopTaskScheduler();
|
||||
}
|
||||
}
|
||||
|
||||
@Autowired MockMvc mvc;
|
||||
@Autowired ObjectMapper om;
|
||||
@Autowired OutboxEventRepository outboxRepo;
|
||||
@Autowired WebhookSubscriptionRepository subRepo;
|
||||
@Autowired WebhookDeliveryRepository deliveryRepo;
|
||||
@Autowired WebhookDispatcher dispatcher;
|
||||
|
||||
private HttpServer receiver;
|
||||
private final ConcurrentLinkedQueue<ReceivedRequest> received = new ConcurrentLinkedQueue<>();
|
||||
@@ -110,12 +129,7 @@ class WebhookE2ETest {
|
||||
@BeforeEach
|
||||
void startReceiver() throws IOException {
|
||||
// Cleanup leftover state из предыдущего CI run.
|
||||
// Postgres testcontainer withReuse(true) сохраняет данные между
|
||||
// запусками — старые subscriptions с dead URLs дают `findDue()`
|
||||
// нагрузку (timeout × N = слишком долго до нашего нового event'а).
|
||||
// Также очищаем outbox_events чтобы matchTick не порождал deliveries
|
||||
// на старые события (которые соответствуют scope=PUBLIC subscription'у
|
||||
// нашего теста).
|
||||
// Postgres testcontainer withReuse(true) сохраняет данные между запусками.
|
||||
// Order: deliveries (FK), subscriptions, outbox events.
|
||||
deliveryRepo.deleteAllInBatch();
|
||||
subRepo.deleteAllInBatch();
|
||||
@@ -171,8 +185,6 @@ class WebhookE2ETest {
|
||||
sub = subRepo.save(sub);
|
||||
UUID subId = sub.getId();
|
||||
|
||||
long deliveriesBefore = deliveryRepo.count();
|
||||
|
||||
// 3. POST record → outbox_events row created
|
||||
String needle = "EVT_" + UUID.randomUUID().toString().replace("-", "").substring(0, 12).toUpperCase();
|
||||
var recBody = om.createObjectNode().put("businessKey", needle);
|
||||
@@ -182,15 +194,18 @@ class WebhookE2ETest {
|
||||
.content(om.writeValueAsBytes(recBody)))
|
||||
.andExpect(status().is2xxSuccessful());
|
||||
|
||||
// 4. Wait for delivery: receiver получил POST request.
|
||||
// 60s timeout запас на CI: ждём 1-2 schedule cycle (matchTick 200ms +
|
||||
// sendTick 200ms) + сетевой round-trip. Без Kafka container время
|
||||
// cold start теста ~3-5s, но всё ещё хочется запас на shared CI runner.
|
||||
await().atMost(60, TimeUnit.SECONDS)
|
||||
.pollInterval(Duration.ofMillis(300))
|
||||
// 4. Manually drive dispatcher — никакого scheduler'а в тесте.
|
||||
// matchTick: outbox_events × subscriptions → INSERT delivery rows.
|
||||
// sendTick: due deliveries → POST на receiver.
|
||||
dispatcher.matchTick();
|
||||
dispatcher.sendTick();
|
||||
|
||||
// 5. Wait for receiver: HTTP server async, network round-trip <100ms на localhost.
|
||||
await().atMost(5, TimeUnit.SECONDS)
|
||||
.pollInterval(Duration.ofMillis(50))
|
||||
.until(() -> received.stream().anyMatch(r -> r.body().contains(needle)));
|
||||
|
||||
// 5. Verify request properties
|
||||
// 6. Verify request properties
|
||||
ReceivedRequest req = received.stream()
|
||||
.filter(r -> r.body().contains(needle))
|
||||
.findFirst()
|
||||
@@ -204,26 +219,23 @@ class WebhookE2ETest {
|
||||
String expectedSig = HmacSigner.sign(secret, req.body().getBytes(StandardCharsets.UTF_8));
|
||||
assertThat(req.signatureHeader()).isEqualTo(expectedSig);
|
||||
|
||||
// 6. Verify DB delivery row marked delivered. Receiver уже получил body
|
||||
// (см. await выше) — delivery row либо уже есть, либо будет в ближайшем
|
||||
// sendTick цикле.
|
||||
await().atMost(20, TimeUnit.SECONDS)
|
||||
.pollInterval(Duration.ofMillis(300))
|
||||
.until(() -> deliveryRepo.count() > deliveriesBefore);
|
||||
// 7. Verify DB delivery row marked delivered. sendTick уже запустился —
|
||||
// если success path, status=delivered должен быть синхронно. Маленький
|
||||
// poll на случай если sendTick ставит status в отдельной транзакции.
|
||||
await().atMost(2, TimeUnit.SECONDS)
|
||||
.pollInterval(Duration.ofMillis(50))
|
||||
.until(() -> deliveryRepo.findBySubscriptionIdOrderByCreatedAtDesc(
|
||||
subId, org.springframework.data.domain.PageRequest.of(0, 10))
|
||||
.getContent().stream()
|
||||
.anyMatch(d -> d.getStatus() == WebhookDelivery.Status.delivered));
|
||||
|
||||
List<WebhookDelivery> all = deliveryRepo.findBySubscriptionIdOrderByCreatedAtDesc(
|
||||
subId, org.springframework.data.domain.PageRequest.of(0, 10)).getContent();
|
||||
assertThat(all).isNotEmpty();
|
||||
WebhookDelivery latest = all.get(0);
|
||||
await().atMost(20, TimeUnit.SECONDS)
|
||||
.pollInterval(Duration.ofMillis(300))
|
||||
.until(() -> {
|
||||
WebhookDelivery d = deliveryRepo.findById(latest.getId()).orElseThrow();
|
||||
return d.getStatus() == WebhookDelivery.Status.delivered;
|
||||
});
|
||||
|
||||
WebhookDelivery delivered = deliveryRepo.findById(latest.getId()).orElseThrow();
|
||||
assertThat(delivered.getStatus()).isEqualTo(WebhookDelivery.Status.delivered);
|
||||
WebhookDelivery delivered = all.stream()
|
||||
.filter(d -> d.getStatus() == WebhookDelivery.Status.delivered)
|
||||
.findFirst()
|
||||
.orElseThrow();
|
||||
assertThat(delivered.getLastStatusCode()).isEqualTo(200);
|
||||
assertThat(delivered.getDeliveredAt()).isNotNull();
|
||||
assertThat(delivered.getLastError()).isNull();
|
||||
@@ -231,12 +243,8 @@ class WebhookE2ETest {
|
||||
|
||||
@Test
|
||||
void receiverReturning500_deliveryRetried_notDelivered() throws Exception {
|
||||
// Стартуем ALT receiver который всегда 500
|
||||
HttpServer failing = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
|
||||
failing.createContext("/fail", exchange -> {
|
||||
// Просто drain body и возвращаем 500. Assertion внутри handler
|
||||
// сломает response chain — клиент получит connection reset вместо
|
||||
// 500, и тест не сматчит lastStatusCode == 500.
|
||||
try {
|
||||
exchange.getRequestBody().readAllBytes();
|
||||
} catch (Exception ignored) {
|
||||
@@ -249,7 +257,6 @@ class WebhookE2ETest {
|
||||
int port = failing.getAddress().getPort();
|
||||
String hookUrl = "http://127.0.0.1:" + port + "/fail";
|
||||
|
||||
// dict
|
||||
String dictName = "whf_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8);
|
||||
JsonNode schema = om.readTree("""
|
||||
{"$schema":"http://json-schema.org/draft-07/schema#","type":"object",
|
||||
@@ -272,10 +279,14 @@ class WebhookE2ETest {
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(om.writeValueAsBytes(recBody))).andExpect(status().is2xxSuccessful());
|
||||
|
||||
// Ждём что delivery с status=retrying появится (HTTP 500 → markFailed).
|
||||
// Timeout 60s покрывает CI cold start + несколько schedule циклов.
|
||||
await().atMost(60, TimeUnit.SECONDS)
|
||||
.pollInterval(Duration.ofMillis(300))
|
||||
// Manual ticks — no scheduler.
|
||||
dispatcher.matchTick();
|
||||
dispatcher.sendTick();
|
||||
|
||||
// sendTick должен mark delivery как retrying (HTTP 500). Маленький poll
|
||||
// на случай задержки status update в отдельной транзакции.
|
||||
await().atMost(5, TimeUnit.SECONDS)
|
||||
.pollInterval(Duration.ofMillis(50))
|
||||
.until(() -> deliveryRepo.findBySubscriptionIdOrderByCreatedAtDesc(
|
||||
sub.getId(), org.springframework.data.domain.PageRequest.of(0, 10))
|
||||
.getContent().stream()
|
||||
@@ -285,4 +296,32 @@ class WebhookE2ETest {
|
||||
failing.stop(0);
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- no-op scheduler implementation ----------
|
||||
|
||||
/**
|
||||
* No-op {@link TaskScheduler}: {@code schedule*} методы возвращают cancelled
|
||||
* {@link ScheduledFuture}. @Scheduled методы регистрируются Spring'ом, но
|
||||
* наш stub их не запускает.
|
||||
*/
|
||||
private static final class NoopTaskScheduler implements TaskScheduler {
|
||||
@Override public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) { return cancelled(); }
|
||||
@Override public ScheduledFuture<?> schedule(Runnable task, Instant startTime) { return cancelled(); }
|
||||
@Override public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Instant startTime, Duration period) { return cancelled(); }
|
||||
@Override public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Duration period) { return cancelled(); }
|
||||
@Override public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Instant startTime, Duration delay) { return cancelled(); }
|
||||
@Override public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Duration delay) { return cancelled(); }
|
||||
|
||||
private static ScheduledFuture<?> cancelled() {
|
||||
return new ScheduledFuture<>() {
|
||||
@Override public long getDelay(TimeUnit unit) { return 0; }
|
||||
@Override public int compareTo(Delayed o) { return 0; }
|
||||
@Override public boolean cancel(boolean mayInterruptIfRunning) { return true; }
|
||||
@Override public boolean isCancelled() { return true; }
|
||||
@Override public boolean isDone() { return true; }
|
||||
@Override public Object get() { return null; }
|
||||
@Override public Object get(long timeout, TimeUnit unit) throws TimeoutException { throw new TimeoutException(); }
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+5
@@ -7,6 +7,7 @@ import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
@@ -57,7 +58,11 @@ public class WebhookTestPingService {
|
||||
* Spring constructor — собирает HttpClient с дефолтным connect timeout, читает
|
||||
* allowed-hosts CSV из config. Тестовый код использует package-private
|
||||
* конструктор {@link #WebhookTestPingService(ObjectMapper, List, HttpClient)}.
|
||||
*
|
||||
* <p>{@code @Autowired} обязателен — без него Spring не знает что выбрать
|
||||
* между этим и package-private constructor для тестов.
|
||||
*/
|
||||
@Autowired
|
||||
public WebhookTestPingService(
|
||||
ObjectMapper objectMapper,
|
||||
@Value("${ordinis.webhook.allowed-hosts:}") String allowedHostsCsv) {
|
||||
|
||||
Reference in New Issue
Block a user