feat(bugcollect): add ingestion backpressure

xuqm-bugcollect-service/docs/BugCollect-API-v1.md

@@ -143,6 +143,26 @@ POST /bugcollect/v1/issues/batch
 Content-Type: application/json
 ```
 
+### 摄取过载
+
+`issues/batch` 与 `events/batch` 单批最多 30 条。平台触发容量保护时返回：
+
+```http
+HTTP/1.1 429 Too Many Requests
+Retry-After: 27
+Content-Type: application/json
+```
+
+```json
+{
+  "code": 429,
+  "message": "BUGCOLLECT_RATE_LIMITED",
+  "data": null
+}
+```
+
+SDK 应保留本地队列，在 `Retry-After` 指定时间后重试；不得把该响应解释为永久停用。
+
 #### 请求体
 
 ```json

xuqm-bugcollect-service/docs/BugCollect-Backpressure.md

@@ -0,0 +1,35 @@
+# BugCollect 摄取防雪崩设计
+
+## 目标
+
+线上同一故障可能在短时间内由大量设备同时触发。摄取链路必须保护数据库和服务线程，
+同时不能让 BugCollect 故障阻断宿主 App。
+
+## 权威链路
+
+1. SDK 本地按指纹采样和限频，批量上报最多 30 条。
+2. 服务端按 `appKey` 统计固定窗口内的事件数，默认每分钟 6000 条。
+3. 多实例部署使用 Redis Lua 原子计数；Redis 不可用时退回进程内窗口，继续保护当前实例。
+4. 超限返回 HTTP 429、`message=BUGCOLLECT_RATE_LIMITED` 和标准
+   `Retry-After` 秒数。
+5. RN/Android SDK 保留原队列，按 `Retry-After` 冷却，最长一小时；冷却期间不重复请求。
+6. `BUGCOLLECT_DISABLED` 仍表示租户关闭服务，SDK 清队列并停用。它与临时限流不能混用。
+
+## 配置
+
+| 环境变量 | 默认值 | 说明 |
+|---|---:|---|
+| `BUGCOLLECT_MAX_EVENTS_PER_MINUTE` | `6000` | 单个 appKey 每个窗口允许的事件数，最小 30 |
+| `BUGCOLLECT_RATE_WINDOW_SECONDS` | `60` | 固定窗口长度，最小 10 秒 |
+
+限流属于平台容量保护，不是租户发布配置。租户平台不向开发者暴露“上传网络建议”或
+让租户关闭服务端保护。
+
+## 验收
+
+- 单批 31 条必须在 Bean Validation 阶段拒绝。
+- Redis 可用时，多实例共享同一 appKey 计数。
+- Redis 不可用时，单实例超过本地额度仍返回 429。
+- 429 必须携带正整数 `Retry-After`。
+- SDK 收到 429 后不得丢队列、立即重试或向宿主业务线程抛未处理异常。
+- 服务关闭返回 403 `BUGCOLLECT_DISABLED`，行为不受限流改动影响。

xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/controller/GlobalExceptionHandler.java

@@ -3,6 +3,7 @@ package com.xuqm.bugcollect.controller;
 import com.xuqm.common.exception.BusinessException;
 import com.xuqm.common.model.ApiResponse;
 import com.xuqm.bugcollect.service.BugCollectDisabledException;
+import com.xuqm.bugcollect.service.IngestionRateLimitedException;
 import jakarta.servlet.http.HttpServletRequest;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -30,6 +31,13 @@ public class GlobalExceptionHandler {
                 .body(ApiResponse.error(403, BugCollectDisabledException.ERROR_CODE));
     }
 
+    @ExceptionHandler(IngestionRateLimitedException.class)
+    public ResponseEntity<ApiResponse<Void>> handle(IngestionRateLimitedException ex) {
+        return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS)
+                .header("Retry-After", Long.toString(ex.retryAfterSeconds()))
+                .body(ApiResponse.error(429, IngestionRateLimitedException.ERROR_CODE));
+    }
+
     @ExceptionHandler(BusinessException.class)
     public ResponseEntity<ApiResponse<Void>> handle(BusinessException ex, HttpServletRequest request) {
         if (ex.getCode() >= 500) {

xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/controller/LogController.java

@@ -7,6 +7,7 @@ import com.xuqm.bugcollect.service.LogService;
 import com.xuqm.bugcollect.service.SourcemapService;
 import com.xuqm.bugcollect.service.WebhookService;
 import com.xuqm.bugcollect.service.BugCollectAvailabilityService;
+import com.xuqm.bugcollect.service.IngestionRateLimiter;
 import jakarta.validation.Valid;
 import org.springframework.data.domain.Page;
 import org.springframework.security.access.AccessDeniedException;
@@ -25,20 +26,28 @@ public class LogController {
     private final SourcemapService sourcemapService;
     private final WebhookService webhookService;
     private final BugCollectAvailabilityService availabilityService;
+    private final IngestionRateLimiter ingestionRateLimiter;
 
     public LogController(LogService logService, SourcemapService sourcemapService,
                          WebhookService webhookService,
-                         BugCollectAvailabilityService availabilityService) {
+                         BugCollectAvailabilityService availabilityService,
+                         IngestionRateLimiter ingestionRateLimiter) {
         this.logService = logService;
         this.sourcemapService = sourcemapService;
         this.webhookService = webhookService;
         this.availabilityService = availabilityService;
+        this.ingestionRateLimiter = ingestionRateLimiter;
     }
 
     // ── Ingestion ──────────────────────────────────────────────────────────────
 
     @PostMapping("/issues/batch")
     public ApiResponse<Void> ingestIssues(@Valid @RequestBody IssueBatchRequest request) {
+        request.events().stream()
+                .collect(java.util.stream.Collectors.groupingBy(
+                        IssueBatchRequest.IssueEventItem::appKey,
+                        java.util.stream.Collectors.counting()))
+                .forEach((appKey, count) -> ingestionRateLimiter.acquire(appKey, count.intValue()));
         request.events().stream()
                 .map(item -> item.appKey() + "\n" + item.platform())
                 .distinct()
@@ -52,6 +61,11 @@ public class LogController {
 
     @PostMapping("/events/batch")
     public ApiResponse<Void> ingestEvents(@Valid @RequestBody EventBatchRequest request) {
+        request.events().stream()
+                .collect(java.util.stream.Collectors.groupingBy(
+                        EventBatchRequest.EventItem::appKey,
+                        java.util.stream.Collectors.counting()))
+                .forEach((appKey, count) -> ingestionRateLimiter.acquire(appKey, count.intValue()));
         request.events().stream()
                 .map(item -> item.appKey() + "\n" + item.platform())
                 .distinct()

xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/dto/EventBatchRequest.java

@@ -3,12 +3,13 @@ package com.xuqm.bugcollect.dto;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.Size;
 import java.util.List;
 
 public record EventBatchRequest(
         @JsonProperty("sentAt") String sentAt,
         @JsonProperty("sdk") IssueBatchRequest.SdkInfo sdk,
-        @JsonProperty("events") @NotEmpty List<EventItem> events
+        @JsonProperty("events") @NotEmpty @Size(max = 30) List<EventItem> events
 ) {
     public record EventItem(
             @JsonProperty("eventId") String eventId,

xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/dto/IssueBatchRequest.java

@@ -3,13 +3,14 @@ package com.xuqm.bugcollect.dto;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.Size;
 import java.util.List;
 import java.util.Map;
 
 public record IssueBatchRequest(
         @JsonProperty("sentAt") String sentAt,
         @JsonProperty("sdk") SdkInfo sdk,
-        @JsonProperty("events") @NotEmpty List<IssueEventItem> events
+        @JsonProperty("events") @NotEmpty @Size(max = 30) List<IssueEventItem> events
 ) {
     public record SdkInfo(
             String name,

xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/service/IngestionRateLimitedException.java

@@ -0,0 +1,23 @@
+package com.xuqm.bugcollect.service;
+
+/**
+ * BugCollect 摄取保护触发时返回给 SDK 的稳定信号。
+ *
+ * <p>客户端收到该信号后应保留本地队列，并至少等待 {@link #retryAfterSeconds()}
+ * 后再尝试，不能把限流当作永久停用。</p>
+ */
+public final class IngestionRateLimitedException extends RuntimeException {
+
+    public static final String ERROR_CODE = "BUGCOLLECT_RATE_LIMITED";
+
+    private final long retryAfterSeconds;
+
+    public IngestionRateLimitedException(long retryAfterSeconds) {
+        super(ERROR_CODE);
+        this.retryAfterSeconds = Math.max(1, retryAfterSeconds);
+    }
+
+    public long retryAfterSeconds() {
+        return retryAfterSeconds;
+    }
+}

xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/service/IngestionRateLimiter.java

@@ -0,0 +1,102 @@
+package com.xuqm.bugcollect.service;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.script.DefaultRedisScript;
+import org.springframework.stereotype.Service;
+
+import java.time.Clock;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 按 appKey 统计摄取事件数，防止线上同一错误集中爆发时压垮数据库。
+ *
+ * <p>Redis Lua 保证多实例计数原子；Redis 暂不可用时退回进程内固定窗口，
+ * 继续提供单实例保护，不把基础设施异常传播给 App。</p>
+ */
+@Service
+public class IngestionRateLimiter {
+
+    private static final Logger log = LoggerFactory.getLogger(IngestionRateLimiter.class);
+    private static final DefaultRedisScript<Long> LIMIT_SCRIPT = new DefaultRedisScript<>("""
+            local current = redis.call('INCRBY', KEYS[1], ARGV[1])
+            if current == tonumber(ARGV[1]) then
+              redis.call('PEXPIRE', KEYS[1], ARGV[3])
+            end
+            if current > tonumber(ARGV[2]) then
+              return redis.call('PTTL', KEYS[1])
+            end
+            return 0
+            """, Long.class);
+
+    private final StringRedisTemplate redis;
+    private final long maxEventsPerWindow;
+    private final long windowMillis;
+    private final Clock clock;
+    private final Map<String, LocalWindow> localWindows = new ConcurrentHashMap<>();
+
+    public IngestionRateLimiter(
+            StringRedisTemplate redis,
+            @Value("${bugcollect-service.ingestion.max-events-per-minute:6000}") long maxEventsPerMinute,
+            @Value("${bugcollect-service.ingestion.window-seconds:60}") long windowSeconds) {
+        this(redis, maxEventsPerMinute, windowSeconds, Clock.systemUTC());
+    }
+
+    IngestionRateLimiter(
+            StringRedisTemplate redis,
+            long maxEventsPerMinute,
+            long windowSeconds,
+            Clock clock) {
+        this.redis = redis;
+        this.maxEventsPerWindow = Math.max(30, maxEventsPerMinute);
+        this.windowMillis = Math.max(10, windowSeconds) * 1_000L;
+        this.clock = clock;
+    }
+
+    public void acquire(String appKey, int eventCount) {
+        if (appKey == null || appKey.isBlank() || eventCount <= 0) return;
+        String normalized = appKey.trim();
+        try {
+            Long retryAfterMillis = redis.execute(
+                    LIMIT_SCRIPT,
+                    List.of("xuqm:bugcollect:ingestion:" + normalized),
+                    Integer.toString(eventCount),
+                    Long.toString(maxEventsPerWindow),
+                    Long.toString(windowMillis));
+            if (retryAfterMillis != null && retryAfterMillis > 0) {
+                throw new IngestionRateLimitedException(toSeconds(retryAfterMillis));
+            }
+            return;
+        } catch (IngestionRateLimitedException ex) {
+            throw ex;
+        } catch (RuntimeException ex) {
+            log.warn("BugCollect Redis 限流不可用，退回进程内保护: appKey={} cause={}",
+                    normalized, ex.getClass().getSimpleName());
+        }
+        acquireLocally(normalized, eventCount);
+    }
+
+    private void acquireLocally(String appKey, int eventCount) {
+        long now = clock.millis();
+        LocalWindow current = localWindows.compute(appKey, (key, previous) -> {
+            if (previous == null || now >= previous.endsAtMillis()) {
+                return new LocalWindow(eventCount, now + windowMillis);
+            }
+            return new LocalWindow(previous.count() + eventCount, previous.endsAtMillis());
+        });
+        if (current.count() > maxEventsPerWindow) {
+            throw new IngestionRateLimitedException(toSeconds(current.endsAtMillis() - now));
+        }
+    }
+
+    private static long toSeconds(long millis) {
+        return Math.max(1, (millis + 999L) / 1_000L);
+    }
+
+    private record LocalWindow(long count, long endsAtMillis) {
+    }
+}

xuqm-bugcollect-service/src/main/resources/application.yml

@@ -5,9 +5,9 @@ spring:
   application:
     name: xuqm-bugcollect-service
   datasource:
-    url: jdbc:mysql://39.107.53.187:3306/xuqm_bugcollect?useUnicode=true&characterEncoding=utf8&useSSL=true&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true
-    username: xuqm
-    password: Xuqm@2026
+    url: ${BUGCOLLECT_DB_URL:jdbc:mysql://127.0.0.1:3306/xuqm_bugcollect?useUnicode=true&characterEncoding=utf8&useSSL=true&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true}
+    username: ${BUGCOLLECT_DB_USERNAME:xuqm}
+    password: ${BUGCOLLECT_DB_PASSWORD:}
     driver-class-name: com.mysql.cj.jdbc.Driver
     hikari:
       minimum-idle: 5
@@ -25,10 +25,10 @@ spring:
         format_sql: true
   data:
     redis:
-      host: redisdev.xuqinmin.com
-      port: 6379
-      password: xuqinmin1022
-      database: 2
+      host: ${BUGCOLLECT_REDIS_HOST:127.0.0.1}
+      port: ${BUGCOLLECT_REDIS_PORT:6379}
+      password: ${BUGCOLLECT_REDIS_PASSWORD:}
+      database: ${BUGCOLLECT_REDIS_DATABASE:2}
       timeout: 10s
       lettuce:
         pool:
@@ -58,6 +58,9 @@ sdk:
 bugcollect-service:
   tenant-sdk-config-url: ${TENANT_SDK_CONFIG_URL:http://127.0.0.1:8081/api/sdk/config}
   availability-cache-seconds: ${BUGCOLLECT_AVAILABILITY_CACHE_SECONDS:30}
+  ingestion:
+    max-events-per-minute: ${BUGCOLLECT_MAX_EVENTS_PER_MINUTE:6000}
+    window-seconds: ${BUGCOLLECT_RATE_WINDOW_SECONDS:60}
   sourcemap:
     storage-dir: ${LOG_SOURCEMAP_DIR:/data/bugcollect-service/sourcemaps}
   webhook:

xuqm-bugcollect-service/src/test/java/com/xuqm/bugcollect/controller/GlobalExceptionHandlerTest.java

@@ -1,6 +1,7 @@
 package com.xuqm.bugcollect.controller;
 
 import com.xuqm.bugcollect.service.BugCollectDisabledException;
+import com.xuqm.bugcollect.service.IngestionRateLimitedException;
 import org.junit.jupiter.api.Test;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -15,4 +16,14 @@ class GlobalExceptionHandlerTest {
         assertThat(response.getBody()).isNotNull();
         assertThat(response.getBody().message()).isEqualTo("BUGCOLLECT_DISABLED");
     }
+
+    @Test
+    void exposesRetryAfterForIngestionBackpressure() {
+        var response = new GlobalExceptionHandler().handle(new IngestionRateLimitedException(27));
+
+        assertThat(response.getStatusCode().value()).isEqualTo(429);
+        assertThat(response.getHeaders().getFirst("Retry-After")).isEqualTo("27");
+        assertThat(response.getBody()).isNotNull();
+        assertThat(response.getBody().message()).isEqualTo("BUGCOLLECT_RATE_LIMITED");
+    }
 }

xuqm-bugcollect-service/src/test/java/com/xuqm/bugcollect/service/IngestionRateLimiterTest.java

@@ -0,0 +1,46 @@
+package com.xuqm.bugcollect.service;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneOffset;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class IngestionRateLimiterTest {
+
+    @Test
+    void usesDistributedRetryWindowReturnedByRedis() {
+        StringRedisTemplate redis = mock(StringRedisTemplate.class);
+        when(redis.execute(any(), any(), any(Object[].class)))
+                .thenReturn(0L, 27_000L);
+        IngestionRateLimiter limiter = new IngestionRateLimiter(
+                redis, 6_000, 60, Clock.systemUTC());
+
+        limiter.acquire("app-key", 30);
+        assertThatThrownBy(() -> limiter.acquire("app-key", 30))
+                .isInstanceOf(IngestionRateLimitedException.class)
+                .satisfies(error -> org.assertj.core.api.Assertions.assertThat(
+                        ((IngestionRateLimitedException) error).retryAfterSeconds()).isEqualTo(27));
+    }
+
+    @Test
+    void fallsBackToLocalWindowWhenRedisIsUnavailable() {
+        StringRedisTemplate redis = mock(StringRedisTemplate.class);
+        when(redis.execute(any(), any(), any(Object[].class)))
+                .thenThrow(new IllegalStateException("redis unavailable"));
+        Clock clock = Clock.fixed(Instant.ofEpochMilli(10_000), ZoneOffset.UTC);
+        IngestionRateLimiter limiter = new IngestionRateLimiter(redis, 30, 60, clock);
+
+        limiter.acquire("app-key", 20);
+        assertThatThrownBy(() -> limiter.acquire("app-key", 11))
+                .isInstanceOf(IngestionRateLimitedException.class)
+                .satisfies(error -> org.assertj.core.api.Assertions.assertThat(
+                        ((IngestionRateLimitedException) error).retryAfterSeconds()).isEqualTo(60));
+    }
+}