优化

event-bridge-service/src/test/java/com/baiyue/eventbridge/service/EventBridgeReliabilitySqlContractTest.java

@@ -0,0 +1,39 @@
+package com.baiyue.eventbridge.service;
+
+import com.baiyue.eventbridge.mapper.BridgeInboxMapper;
+import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.annotations.Update;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class EventBridgeReliabilitySqlContractTest {
+    @Test
+    void schedulingAndLeaseSqlCoverRetryDeadAndExpiredLeaseTakeover() throws Exception {
+        String candidates = sql(BridgeInboxMapper.class.getMethod("dispatchCandidates", int.class), Select.class);
+        String claim = sql(BridgeInboxMapper.class.getMethod(
+            "claim", String.class, String.class, long.class), Update.class);
+        String failed = sql(BridgeInboxMapper.class.getMethod(
+            "markFailed", String.class, String.class, int.class, long.class, String.class), Update.class);
+        String processed = sql(BridgeInboxMapper.class.getMethod(
+            "markProcessed", String.class, String.class), Update.class);
+
+        assertThat(candidates).contains("status IN ('RECEIVED','RETRY')", "next_retry_at <= CURRENT_TIMESTAMP(3)",
+            "status = 'PROCESSING'", "lease_until < CURRENT_TIMESTAMP(3)", "ORDER BY received_at");
+        assertThat(claim).contains("status='PROCESSING'", "lease_owner=#{owner}", "lease_until=DATE_ADD",
+            "status='PROCESSING' AND lease_until < CURRENT_TIMESTAMP(3)");
+        assertThat(failed).contains("retry_count=retry_count+1", "status=CASE", "THEN 'DEAD' ELSE 'RETRY'",
+            "next_retry_at=CASE", "THEN NULL", "last_error=#{error}", "lease_owner=#{owner}");
+        assertThat(processed).contains("status='PROCESSED'", "next_retry_at=NULL", "last_error=NULL",
+            "status='PROCESSING'", "lease_owner=#{owner}");
+    }
+
+    private <A extends java.lang.annotation.Annotation> String sql(Method method, Class<A> type) {
+        A annotation = method.getAnnotation(type);
+        String[] statements = annotation instanceof Select select ? select.value() : ((Update) annotation).value();
+        return String.join(" ", Arrays.asList(statements)).replaceAll("\\s+", " ");
+    }
+}

event-bridge-service/src/test/java/com/baiyue/eventbridge/service/OverseasEventDispatcherTest.java

@@ -12,6 +12,7 @@ import org.springframework.test.util.ReflectionTestUtils;
 
 import java.net.URI;
 import java.time.Duration;
+import java.util.List;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.*;
@@ -103,6 +104,59 @@ class OverseasEventDispatcherTest {
         assertThat(updatedAttempt().getStatus()).isEqualTo("DELIVERED");
     }
 
+    @Test
+    void downstream503RetriesThenDeliversWithDiagnosticAttemptsInBothDirections() throws Exception {
+        for (String target : List.of("OVERSEAS", "CN")) {
+            reset(inboxMapper, attemptMapper, eventClient, firstlegEventClient);
+            event.setSourceRegion("CN".equals(target) ? "OVERSEAS" : "CN");
+            event.setTargetRegion(target);
+            event.setRetryCount(0);
+            when(inboxMapper.claim(eq("event-1"), anyString(), anyLong())).thenReturn(1);
+            when(inboxMapper.selectById("event-1")).thenReturn(event);
+            when(inboxMapper.nextAttemptNo("event-1")).thenReturn(1, 2);
+            doAnswer(invocation -> {
+                event.setRetryCount(1);
+                event.setStatus("RETRY");
+                event.setLastError(invocation.getArgument(4));
+                event.setNextRetryAt(java.time.LocalDateTime.now().plusSeconds(
+                    ((Number) invocation.getArgument(3)).longValue()));
+                return 1;
+            }).when(inboxMapper).markFailed(eq("event-1"), anyString(), eq(3), anyLong(), anyString());
+            when(inboxMapper.markProcessed(eq("event-1"), anyString())).thenAnswer(invocation -> {
+                event.setStatus("PROCESSED");
+                event.setLastError(null);
+                event.setNextRetryAt(null);
+                return 1;
+            });
+            if ("CN".equals(target)) {
+                when(firstlegEventClient.deliver(event))
+                    .thenThrow(new IllegalStateException("HTTP 503 code=S0503 requestId=req-cn traceId=trace-cn"))
+                    .thenReturn("200");
+            } else {
+                when(eventClient.deliver(event))
+                    .thenThrow(new IllegalStateException("HTTP 503 code=S0503 requestId=req-overseas traceId=trace-overseas"))
+                    .thenReturn("200");
+            }
+
+            dispatcher.dispatchOne("event-1");
+            assertThat(event.getStatus()).isEqualTo("RETRY");
+            assertThat(event.getNextRetryAt()).isNotNull();
+            assertThat(event.getLastError()).contains("HTTP 503", "requestId=", "traceId=");
+            dispatcher.dispatchOne("event-1");
+
+            ArgumentCaptor<BridgeDeliveryAttempt> attempts = ArgumentCaptor.forClass(BridgeDeliveryAttempt.class);
+            verify(attemptMapper, times(2)).updateById(attempts.capture());
+            assertThat(attempts.getAllValues()).extracting(BridgeDeliveryAttempt::getAttemptNo)
+                .containsExactly(1, 2);
+            assertThat(attempts.getAllValues()).extracting(BridgeDeliveryAttempt::getStatus)
+                .containsExactly("RETRY", "DELIVERED");
+            assertThat(attempts.getAllValues().get(0).getErrorMessage())
+                .contains("HTTP 503", "requestId=", "traceId=");
+            assertThat(event.getStatus()).isEqualTo("PROCESSED");
+            assertThat(event.getLastError()).isNull();
+        }
+    }
+
     private BridgeDeliveryAttempt updatedAttempt() {
         ArgumentCaptor<BridgeDeliveryAttempt> captor = ArgumentCaptor.forClass(BridgeDeliveryAttempt.class);
         verify(attemptMapper).updateById(captor.capture());