feat(outbox): attempt cap → DLQ + admin endpoint + 6 unit-тестов

После N (default 1000) неудачных попыток публикации событие маркируется dlq_at и исключается из polling. Без cap poller бесконечно ретраил мёртвые сообщения (orphan topic, ACL deny), забивая логи и lag-метрику. - 0011-outbox-dlq.xml: dlq_at column + перестроенный idx_outbox_unpublished (исключает DLQ) + idx_outbox_dlq для admin-листинга. - OutboxPoller: ordinis.outbox.attempt-max (default 1000), markDlq() при достижении cap, новый counter ordinis_outbox_dlq_total. - OutboxLagGauge: ordinis_outbox_dlq_size gauge — алерт на > 0. - OutboxEventRepository: countPending() (без DLQ), findByDlqAtIsNotNull(Pageable). - OutboxAdminController: GET /admin/outbox/{stats,dlq} для UI. - 6 unit-тестов через mock-maker-subclass (JDK 25 ломает Mockito inline).
2026-05-04 15:43:56 +03:00
parent f7efb23e09
commit e182b30c58
10 changed files with 391 additions and 9 deletions
@@ -57,6 +57,20 @@ ordinis:
    enabled: ${ORDINIS_OUTBOX_ENABLED:true}
    poll-interval-ms: ${ORDINIS_OUTBOX_POLL_INTERVAL_MS:1000}
    batch-size: ${ORDINIS_OUTBOX_BATCH_SIZE:100}
+    # Cap на ретраи: после N попыток событие → DLQ (dlq_at), poller его пропускает.
+    # 1000 при poll=1s ≈ ~17 минут активных ошибок (или быстрее на fail-fast топиках).
+    attempt-max: ${ORDINIS_OUTBOX_ATTEMPT_MAX:1000}
+
+springdoc:
+  api-docs:
+    path: /v3/api-docs
+  swagger-ui:
+    path: /swagger-ui.html
+    operations-sorter: method
+    tags-sorter: alpha
+    display-request-duration: true
+  show-actuator: false
+  packages-to-scan: cloud.nstart.terravault.ordinis.restapi.web

 ---
 # DEV profile: подключается к cluster PG+Kafka через kubectl port-forward.
@@ -73,6 +73,9 @@ public class OutboxEvent {
  @Column(name = "last_error", columnDefinition = "TEXT")
  private String lastError;

+  @Column(name = "dlq_at")
+  private OffsetDateTime dlqAt;
+
  protected OutboxEvent() {}

  public OutboxEvent(
@@ -103,6 +106,15 @@ public class OutboxEvent {
    this.lastError = error;
  }

+  public void markDlq(String error) {
+    this.lastError = error;
+    this.dlqAt = OffsetDateTime.now();
+  }
+
+  public boolean isDlq() {
+    return dlqAt != null;
+  }
+
  public Long getId() { return id; }
  public String getEventType() { return eventType; }
  public String getAggregateType() { return aggregateType; }
@@ -118,6 +130,7 @@ public class OutboxEvent {
  public String getSpanId() { return spanId; }
  public Integer getAttemptCount() { return attemptCount; }
  public String getLastError() { return lastError; }
+  public OffsetDateTime getDlqAt() { return dlqAt; }

  public void setDictionaryName(String v) { this.dictionaryName = v; }
  public void setTraceId(String v) { this.traceId = v; }
@@ -1,5 +1,6 @@
 package cloud.nstart.terravault.ordinis.domain.outbox;

+import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Query;
@@ -9,10 +10,35 @@ import java.util.List;
 public interface OutboxEventRepository extends JpaRepository<OutboxEvent, Long> {

  /**
-   * Unpublished events для polling. Использует partial index {@code idx_outbox_unpublished}.
+   * Unpublished events для polling. Использует partial index
+   * {@code idx_outbox_unpublished} (where published_at IS NULL AND dlq_at IS NULL).
   */
-  @Query("SELECT o FROM OutboxEvent o WHERE o.publishedAt IS NULL ORDER BY o.id ASC")
+  @Query("""
+      SELECT o FROM OutboxEvent o
+      WHERE o.publishedAt IS NULL AND o.dlqAt IS NULL
+      ORDER BY o.id ASC
+      """)
  List<OutboxEvent> findUnpublished(Pageable pageable);

+  /**
+   * Сколько ожидает публикации (без DLQ). Метрика
+   * {@code ordinis_outbox_pending_events}.
+   */
+  @Query("SELECT COUNT(o) FROM OutboxEvent o WHERE o.publishedAt IS NULL AND o.dlqAt IS NULL")
+  long countPending();
+
+  /**
+   * Сколько в DLQ. Метрика {@code ordinis_outbox_dlq_size} — растёт ⇒ алерт.
+   */
+  long countByDlqAtIsNotNull();
+
+  /**
+   * DLQ-листинг для admin UI. Партиальный индекс {@code idx_outbox_dlq} по
+   * {@code dlq_at DESC}.
+   */
+  Page<OutboxEvent> findByDlqAtIsNotNull(Pageable pageable);
+
+  /** @deprecated используйте {@link #countPending()} (не считает DLQ). */
+  @Deprecated
  long countByPublishedAtIsNull();
 }
@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<databaseChangeLog
+  xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog
+                      http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.27.xsd">
+
+  <!--
+    Outbox DLQ: после N неудачных попыток публикации (ordinis.outbox.attempt-max,
+    default 1000) события маркируются dlq_at и исключаются из поллинга. Без cap
+    poller бесконечно ретраит мёртвые сообщения (orphan topic, тупик ACL),
+    забивая лог и Prometheus alert ordinis_outbox_pending_events.
+  -->
+
+  <changeSet id="0011-1-outbox-dlq-column" author="ordinis">
+    <addColumn tableName="outbox_events">
+      <column name="dlq_at" type="TIMESTAMPTZ"/>
+    </addColumn>
+  </changeSet>
+
+  <changeSet id="0011-2-outbox-rebuild-pending-index" author="ordinis">
+    <comment>
+      Pending index должен исключать dlq_at — иначе poller продолжит "видеть"
+      DLQ-записи и ordinis_outbox_pending_events будет показывать их в lag.
+    </comment>
+    <sql>
+      DROP INDEX IF EXISTS idx_outbox_unpublished;
+      CREATE INDEX idx_outbox_unpublished ON outbox_events (created_at)
+      WHERE published_at IS NULL AND dlq_at IS NULL;
+    </sql>
+    <rollback>
+      <sql>
+        DROP INDEX IF EXISTS idx_outbox_unpublished;
+        CREATE INDEX idx_outbox_unpublished ON outbox_events (created_at)
+        WHERE published_at IS NULL;
+      </sql>
+    </rollback>
+  </changeSet>
+
+  <changeSet id="0011-3-outbox-dlq-index" author="ordinis">
+    <comment>Партиальный индекс для быстрого листинга DLQ в admin UI.</comment>
+    <sql>
+      CREATE INDEX idx_outbox_dlq ON outbox_events (dlq_at DESC)
+      WHERE dlq_at IS NOT NULL;
+    </sql>
+    <rollback>
+      <sql>DROP INDEX IF EXISTS idx_outbox_dlq;</sql>
+    </rollback>
+  </changeSet>
+
+</databaseChangeLog>
@@ -15,5 +15,6 @@
  <include file="changes/0008-fix-scope-check-uppercase.xml"/>
  <include file="changes/0009-fix-records-scope-check.xml"/>
  <include file="changes/0010-envers-tables.xml"/>
+  <include file="changes/0011-outbox-dlq.xml"/>

 </databaseChangeLog>
@@ -8,8 +8,13 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Component;

 /**
- * Gauge {@code ordinis_outbox_pending_events} — сколько unpublished. Алертится
- * Prometheus'ом если растёт (poller не успевает или Kafka недоступна).
+ * Gauges:
+ * <ul>
+ *   <li>{@code ordinis_outbox_pending_events} — unpublished без DLQ. Растёт ⇒
+ *       poller не успевает либо Kafka недоступна.</li>
+ *   <li>{@code ordinis_outbox_dlq_size} — события в DLQ. Любое значение &gt; 0 ⇒
+ *       требуется ручной разбор (orphan topic, ACL, broken payload).</li>
+ * </ul>
 */
@Component
@ConditionalOnProperty(name = "ordinis.outbox.enabled", havingValue = "true", matchIfMissing = false)
@@ -26,8 +31,13 @@ public class OutboxLagGauge {
  @PostConstruct
  void register() {
    Gauge.builder("ordinis_outbox_pending_events", repository,
-            r -> (double) r.countByPublishedAtIsNull())
-        .description("Количество outbox events ожидающих публикацию в Kafka")
+            r -> (double) r.countPending())
+        .description("Количество outbox events ожидающих публикацию (без DLQ)")
+        .register(meterRegistry);
+
+    Gauge.builder("ordinis_outbox_dlq_size", repository,
+            r -> (double) r.countByDlqAtIsNotNull())
+        .description("Количество outbox events в DLQ (требуют ручного разбора)")
        .register(meterRegistry);
  }
 }
@@ -25,6 +25,11 @@ import java.util.concurrent.TimeoutException;
 * @Scheduled poller для outbox. Sweep каждые {@code ordinis.outbox.poll-interval-ms},
 * batch size {@code ordinis.outbox.batch-size}. Активируется только если
 * {@code ordinis.outbox.enabled=true} (отключаем в read-api / projection-writer).
+ *
+ * <p>После {@code ordinis.outbox.attempt-max} (default 1000) неудачных попыток
+ * событие маркируется {@code dlq_at} и исключается из дальнейшего polling. Без
+ * cap poller бесконечно ретраит мёртвые сообщения (orphan topic, ACL deny),
+ * забивая лог и lag-метрику.
 */
@Component
@ConditionalOnProperty(name = "ordinis.outbox.enabled", havingValue = "true", matchIfMissing = false)
@@ -37,6 +42,7 @@ public class OutboxPoller {
  private final com.fasterxml.jackson.databind.ObjectMapper objectMapper;
  private final Counter publishedCounter;
  private final Counter failedCounter;
+  private final Counter dlqCounter;
  private final Timer publishTimer;

  @Value("${ordinis.outbox.batch-size:100}")
@@ -45,6 +51,9 @@ public class OutboxPoller {
  @Value("${ordinis.outbox.send-timeout-ms:10000}")
  private long sendTimeoutMs;

+  @Value("${ordinis.outbox.attempt-max:1000}")
+  private int attemptMax;
+
  public OutboxPoller(
      OutboxEventRepository repository,
      KafkaTemplate<String, String> kafkaTemplate,
@@ -59,6 +68,9 @@ public class OutboxPoller {
    this.failedCounter = Counter.builder("ordinis_outbox_publish_errors_total")
        .description("Количество ошибок публикации в Kafka")
        .register(meterRegistry);
+    this.dlqCounter = Counter.builder("ordinis_outbox_dlq_total")
+        .description("Количество событий перенесённых в DLQ (attempt cap)")
+        .register(meterRegistry);
    this.publishTimer = Timer.builder("ordinis_kafka_publish_duration_seconds")
        .description("Время публикации одного события в Kafka")
        .register(meterRegistry);
@@ -103,10 +115,21 @@ public class OutboxPoller {
  }

  private void markFailed(OutboxEvent event, String error) {
-    log.warn("Failed to publish outbox event id={} topic={}: {}",
-        event.getId(), event.getKafkaTopic(), error);
    event.markFailed(error);
-    repository.save(event);
    failedCounter.increment();
+
+    if (event.getAttemptCount() != null && event.getAttemptCount() >= attemptMax) {
+      event.markDlq(error);
+      dlqCounter.increment();
+      log.error(
+          "Outbox event id={} topic={} → DLQ after {} attempts. Last error: {}",
+          event.getId(), event.getKafkaTopic(), event.getAttemptCount(), error);
+    } else {
+      log.warn(
+          "Failed to publish outbox event id={} topic={} attempt={}/{}: {}",
+          event.getId(), event.getKafkaTopic(),
+          event.getAttemptCount(), attemptMax, error);
+    }
+    repository.save(event);
  }
 }
@@ -0,0 +1,170 @@
+package cloud.nstart.terravault.ordinis.outbox;
+
+import cloud.nstart.terravault.ordinis.domain.DataScope;
+import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEvent;
+import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.domain.Pageable;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeoutException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class OutboxPollerTest {
+
+  private OutboxEventRepository repo;
+  private KafkaTemplate<String, String> kafka;
+  private OutboxPoller poller;
+
+  @BeforeEach
+  void setUp() {
+    repo = mock(OutboxEventRepository.class);
+    kafka = mock(KafkaTemplate.class);
+    poller = new OutboxPoller(repo, kafka, new ObjectMapper(), new SimpleMeterRegistry());
+    ReflectionTestUtils.setField(poller, "batchSize", 100);
+    ReflectionTestUtils.setField(poller, "sendTimeoutMs", 1_000L);
+    ReflectionTestUtils.setField(poller, "attemptMax", 3);
+  }
+
+  private OutboxEvent newEvent() {
+    return new OutboxEvent(
+        "DICT_RECORD_CREATED",
+        "DictionaryRecord",
+        "spacecraft:RESURS-P-3",
+        DataScope.PUBLIC,
+        JsonNodeFactory.instance.objectNode().put("foo", "bar"),
+        "ordinis.cuod.events.public",
+        "spacecraft:RESURS-P-3");
+  }
+
+  @Test
+  void empty_batch_does_nothing() {
+    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of());
+
+    poller.poll();
+
+    verify(kafka, never()).send(anyString(), anyString(), anyString());
+    verify(repo, never()).save(any());
+  }
+
+  @Test
+  void successful_publish_marks_published() {
+    OutboxEvent event = newEvent();
+    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(event));
+
+    SendResult<String, String> result = mock(SendResult.class);
+    when(result.getRecordMetadata()).thenReturn(
+        new RecordMetadata(new TopicPartition("ordinis.cuod.events.public", 0), 0L, 0, 0L, 0, 0));
+    when(kafka.send(eq("ordinis.cuod.events.public"), anyString(), anyString()))
+        .thenReturn(CompletableFuture.completedFuture(result));
+
+    poller.poll();
+
+    assertThat(event.getPublishedAt()).isNotNull();
+    assertThat(event.getDlqAt()).isNull();
+    verify(repo).save(event);
+  }
+
+  @Test
+  void failure_below_cap_marks_failed_not_dlq() {
+    OutboxEvent event = newEvent();
+    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(event));
+
+    CompletableFuture<SendResult<String, String>> failed = new CompletableFuture<>();
+    failed.completeExceptionally(new TimeoutException("kafka down"));
+    when(kafka.send(anyString(), anyString(), anyString())).thenReturn(failed);
+
+    poller.poll();
+
+    assertThat(event.getAttemptCount()).isEqualTo(1);
+    assertThat(event.getDlqAt()).isNull();
+    assertThat(event.getLastError()).contains("kafka down");
+    verify(repo).save(event);
+  }
+
+  @Test
+  void failure_at_cap_marks_dlq() {
+    OutboxEvent event = newEvent();
+    // Симулируем что 2 предыдущих попытки уже были.
+    event.markFailed("prev 1");
+    event.markFailed("prev 2");
+    assertThat(event.getAttemptCount()).isEqualTo(2);
+
+    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(event));
+
+    CompletableFuture<SendResult<String, String>> failed = new CompletableFuture<>();
+    failed.completeExceptionally(new TimeoutException("still down"));
+    when(kafka.send(anyString(), anyString(), anyString())).thenReturn(failed);
+
+    poller.poll();
+
+    assertThat(event.getAttemptCount()).isEqualTo(3);
+    assertThat(event.getDlqAt()).isNotNull();
+    assertThat(event.isDlq()).isTrue();
+    assertThat(event.getLastError()).contains("still down");
+    verify(repo).save(event);
+  }
+
+  @Test
+  void failure_above_cap_stays_dlq_not_re_marked() {
+    OutboxEvent event = newEvent();
+    for (int i = 0; i < 3; i++) event.markFailed("err " + i);
+    event.markDlq("first dlq");
+    java.time.OffsetDateTime firstDlqAt = event.getDlqAt();
+
+    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of());
+
+    poller.poll();
+
+    // Poller не должен подобрать DLQ событие — findUnpublished фильтрует dlq_at IS NULL.
+    assertThat(event.getDlqAt()).isEqualTo(firstDlqAt);
+  }
+
+  @Test
+  void mixed_batch_publishes_some_dlqs_others() {
+    OutboxEvent ok = newEvent();
+    OutboxEvent doomed = newEvent();
+    doomed.markFailed("p1");
+    doomed.markFailed("p2");
+
+    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(ok, doomed));
+
+    SendResult<String, String> result = mock(SendResult.class);
+    when(result.getRecordMetadata()).thenReturn(
+        new RecordMetadata(new TopicPartition("ordinis.cuod.events.public", 0), 0L, 0, 0L, 0, 0));
+    CompletableFuture<SendResult<String, String>> okFut = CompletableFuture.completedFuture(result);
+    CompletableFuture<SendResult<String, String>> badFut = new CompletableFuture<>();
+    badFut.completeExceptionally(new TimeoutException("doomed"));
+
+    // Первый event получает success, второй — failure.
+    when(kafka.send(anyString(), anyString(), anyString()))
+        .thenReturn(okFut)
+        .thenReturn(badFut);
+
+    poller.poll();
+
+    assertThat(ok.getPublishedAt()).isNotNull();
+    assertThat(ok.getDlqAt()).isNull();
+    assertThat(doomed.getDlqAt()).isNotNull();
+    verify(repo, times(2)).save(any());
+  }
+}
@@ -0,0 +1 @@
+mock-maker-subclass
@@ -0,0 +1,73 @@
+package cloud.nstart.terravault.ordinis.restapi.web;
+
+import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEvent;
+import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Sort;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Admin endpoints для outbox observability. Только чтение — replay делается
+ * вручную через SQL (UPDATE outbox_events SET dlq_at=NULL, attempt_count=0 WHERE id=...).
+ * Авто-replay сделаем когда появится первый реальный кейс DLQ.
+ */
+@RestController
+@RequestMapping("/api/v1/admin/outbox")
+public class OutboxAdminController {
+
+  private final OutboxEventRepository repository;
+
+  public OutboxAdminController(OutboxEventRepository repository) {
+    this.repository = repository;
+  }
+
+  @GetMapping("/stats")
+  public Map<String, Long> stats() {
+    return Map.of(
+        "pending", repository.countPending(),
+        "dlq", repository.countByDlqAtIsNotNull()
+    );
+  }
+
+  @GetMapping("/dlq")
+  public Page<DlqEntry> dlq(
+      @RequestParam(defaultValue = "0") int page,
+      @RequestParam(defaultValue = "50") int size) {
+    Page<OutboxEvent> events = repository.findByDlqAtIsNotNull(
+        PageRequest.of(page, Math.min(size, 200), Sort.by(Sort.Direction.DESC, "dlqAt")));
+    return events.map(DlqEntry::from);
+  }
+
+  public record DlqEntry(
+      Long id,
+      String eventType,
+      String dictionaryName,
+      String aggregateId,
+      String kafkaTopic,
+      Integer attemptCount,
+      String lastError,
+      OffsetDateTime createdAt,
+      OffsetDateTime dlqAt) {
+
+    static DlqEntry from(OutboxEvent e) {
+      return new DlqEntry(
+          e.getId(),
+          e.getEventType(),
+          e.getDictionaryName(),
+          e.getAggregateId(),
+          e.getKafkaTopic(),
+          e.getAttemptCount(),
+          e.getLastError(),
+          e.getCreatedAt(),
+          e.getDlqAt());
+    }
+  }
+}