feat(syslog): 优化大日志通道并增加异常连续消费兜底

- 128 KiB 及以上日志默认进入大日志 Topic
- InfluxDB 原文及达梦大字段超过 256 KiB 时截断保存
- 截断内容追加 HB_XDR_TRUNCATED 标记
- InfluxDB 写入失败时继续执行规则解析和达梦写入
- 未知或系统性异常优先写入普通/大日志 DLT
- DLT 写入失败时记录特殊告警并提交 offset，避免阻塞后续消息
- DLT 失败等待时间默认限制为 1 秒
- 增加连续性放行、DLT 成功及失败监控指标
- 更新 Docker Compose、v3 升级文档及相关测试

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java

@@ -43,6 +43,7 @@ import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.time.LocalDateTime;
+import java.nio.charset.StandardCharsets;
 
 import com.common.util.Sm4Util;
 @Slf4j
@@ -53,12 +54,19 @@ public class SysLogProcessor {
     @Qualifier("logNormalProcessorExecutor")
     private Executor taskExecutor;
 
+    @Autowired
+    @Qualifier("largeLogProcessorExecutor")
+    private Executor largeTaskExecutor;
+
     @Value("${app.processor.batch-size:100}")
     private int batchSize;
 
     @Value("${app.processor.process-timeout-ms:120000}")
     private long processTimeoutMs;
 
+    @Value("${syslog.large-listener.processing-batch-max-bytes:8388608}")
+    private long largeProcessingBatchMaxBytes;
+
     @Autowired
     private AppProperties appProperties;
 
@@ -132,7 +140,47 @@ public class SysLogProcessor {
             containerFactory = "rawKafkaListenerContainerFactory"
     )
     public void consumeAndParallelProcess(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
-        consumeBatch(records, ack, false, "RAW");
+        consumeBatch(records, ack, false, "RAW", taskExecutor);
+    }
+
+    @KafkaListener(
+            id = "syslog-consumer-rule-raw-large",
+            groupId = "${syslog.large-listener.group-id:${spring.kafka.consumer.group-id}-large}",
+            autoStartup = "${syslog.large-listener.enabled:true}",
+            topics = "${spring.kafka.consumer.large-topic:agent-syslog-topic.large}",
+            containerFactory = "largeRawKafkaListenerContainerFactory"
+    )
+    public void consumeLargeMessages(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
+        consumeLargeInByteChunks(records, ack, false, "RAW_LARGE");
+    }
+
+    @KafkaListener(
+            id = "syslog-consumer-rule-large-retry-replay",
+            groupId = "${syslog.large-reliability.consumer-group-id:${spring.kafka.consumer.group-id}-large-retry}",
+            autoStartup = "${syslog.large-reliability.consumer.enabled:true}",
+            topics = {
+                    "${syslog.large-reliability.retry-1m-topic:${spring.kafka.consumer.large-topic}.RETRY.1M}",
+                    "${syslog.large-reliability.retry-10m-topic:${spring.kafka.consumer.large-topic}.RETRY.10M}",
+                    "${syslog.large-reliability.retry-1h-topic:${spring.kafka.consumer.large-topic}.RETRY.1H}",
+                    "${syslog.large-reliability.replay-topic:${spring.kafka.consumer.large-topic}.REPLAY}"
+            },
+            containerFactory = "largeRawKafkaListenerContainerFactory"
+    )
+    public void consumeLargeRetryAndReplay(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
+        consumeLargeInByteChunks(records, ack, true, "RETRY_REPLAY_LARGE");
+    }
+
+    private void consumeLargeInByteChunks(List<ConsumerRecord<String, String>> records,
+                                          Acknowledgment ack, boolean enforceRetryDue, String lane) {
+        if (records.isEmpty()) {
+            ack.acknowledge();
+            return;
+        }
+        for (List<ConsumerRecord<String, String>> chunk : partitionRecordsByBytes(
+                records, Math.max(1L, largeProcessingBatchMaxBytes))) {
+            processBatch(chunk, enforceRetryDue, lane, largeTaskExecutor);
+        }
+        ack.acknowledge();
     }
 
     @KafkaListener(
@@ -148,7 +196,7 @@ public class SysLogProcessor {
             concurrency = "${syslog.reliability.listener.concurrency:1}"
     )
     public void consumeRetryAndReplay(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
-        consumeBatch(records, ack, true, "RETRY_REPLAY");
+        consumeBatch(records, ack, true, "RETRY_REPLAY", taskExecutor);
     }
 
     /**
@@ -157,11 +205,17 @@ public class SysLogProcessor {
      * 其写库结果不参与 Kafka ACK。
      */
     private void consumeBatch(List<ConsumerRecord<String, String>> records, Acknowledgment ack,
-                              boolean enforceRetryDue, String lane) {
+                              boolean enforceRetryDue, String lane, Executor normalizationExecutor) {
         if (records.isEmpty()) {
             ack.acknowledge();
             return;
         }
+        processBatch(records, enforceRetryDue, lane, normalizationExecutor);
+        ack.acknowledge();
+    }
+
+    private void processBatch(List<ConsumerRecord<String, String>> records,
+                              boolean enforceRetryDue, String lane, Executor normalizationExecutor) {
         long batchStartedAt = System.currentTimeMillis();
         long prepareStartedAt = batchStartedAt;
         BatchMetrics metrics = new BatchMetrics();
@@ -189,10 +243,9 @@ public class SysLogProcessor {
         long influxMs = System.currentTimeMillis() - influxStartedAt;
 
         long normalizeStartedAt = System.currentTimeMillis();
-        normalizePreparedMessages(persistedMessages, metrics);
+        normalizePreparedMessages(persistedMessages, metrics, normalizationExecutor);
         long normalizeMs = System.currentTimeMillis() - normalizeStartedAt;
 
-        ack.acknowledge();
         totalProcessed.addAndGet(records.size());
         log.info("Syslog批次完成: lane={}, topicPartitions={}, records={}, prepared={}, persisted={}, "
                         + "success={}, filtered={}, retry={}, dlt={}, normalizedRows={}, dmSuccessfulRows={}, "
@@ -218,6 +271,32 @@ public class SysLogProcessor {
                 System.currentTimeMillis() - batchStartedAt);
     }
 
+    static List<List<ConsumerRecord<String, String>>> partitionRecordsByBytes(
+            List<ConsumerRecord<String, String>> records, long maxBytes) {
+        if (maxBytes <= 0L) throw new IllegalArgumentException("maxBytes 必须大于 0");
+        List<List<ConsumerRecord<String, String>>> result = new ArrayList<>();
+        List<ConsumerRecord<String, String>> current = new ArrayList<>();
+        long currentBytes = 0L;
+        for (ConsumerRecord<String, String> record : records) {
+            String value = record.value();
+            long bytes = value == null ? 0L : value.getBytes(StandardCharsets.UTF_8).length;
+            if (!current.isEmpty() && bytes > maxBytes - currentBytes) {
+                result.add(current);
+                current = new ArrayList<>();
+                currentBytes = 0L;
+            }
+            current.add(record);
+            currentBytes = Math.addExact(currentBytes, bytes);
+            if (currentBytes >= maxBytes) {
+                result.add(current);
+                current = new ArrayList<>();
+                currentBytes = 0L;
+            }
+        }
+        if (!current.isEmpty()) result.add(current);
+        return result;
+    }
+
     /**
      * 方案二：分批多线程处理
      * 将大批次拆分为小批次进行并行处理
@@ -422,6 +501,16 @@ public class SysLogProcessor {
         if (messages.isEmpty()) {
             return messages;
         }
+        List<PreparedMessage> persisted = new ArrayList<>(messages.size());
+        long configuredBatchBytes = rawStorageProperties.getWriteBatchMaxBytes();
+        long effectiveBatchBytes = configuredBatchBytes > 0L ? configuredBatchBytes : Long.MAX_VALUE;
+        for (List<PreparedMessage> chunk : partitionPreparedByPointBytes(messages, effectiveBatchBytes)) {
+            persisted.addAll(persistRawChunk(chunk, metrics));
+        }
+        return persisted;
+    }
+
+    private List<PreparedMessage> persistRawChunk(List<PreparedMessage> messages, BatchMetrics metrics) {
         List<Point> points = new ArrayList<>(messages.size());
         for (PreparedMessage message : messages) {
             points.add(message.point);
@@ -452,6 +541,31 @@ public class SysLogProcessor {
         }
     }
 
+    static List<List<PreparedMessage>> partitionPreparedByPointBytes(
+            List<PreparedMessage> messages, long maxBytes) {
+        if (maxBytes <= 0L) throw new IllegalArgumentException("InfluxDB 批次字节上限必须大于 0");
+        List<List<PreparedMessage>> result = new ArrayList<>();
+        List<PreparedMessage> current = new ArrayList<>();
+        long currentBytes = 0L;
+        for (PreparedMessage message : messages) {
+            long bytes = message.point.toLineProtocol().getBytes(StandardCharsets.UTF_8).length + 1L;
+            if (!current.isEmpty() && bytes > maxBytes - currentBytes) {
+                result.add(current);
+                current = new ArrayList<>();
+                currentBytes = 0L;
+            }
+            current.add(message);
+            currentBytes = Math.addExact(currentBytes, bytes);
+            if (currentBytes >= maxBytes) {
+                result.add(current);
+                current = new ArrayList<>();
+                currentBytes = 0L;
+            }
+        }
+        if (!current.isEmpty()) result.add(current);
+        return result;
+    }
+
     /**
      * 历史单条兼容入口：原文已可靠持久化后，执行标准化并立即落库。
      * 活动 Kafka 批处理链路使用 prepareNormalization()，避免在计算线程中逐条开启达梦事务。
@@ -484,6 +598,11 @@ public class SysLogProcessor {
      * 批内有界并行标准化。任务异常必须汇总回 listener；超时或系统性失败时整批不 ACK。
      */
     void normalizePreparedMessages(List<PreparedMessage> messages, BatchMetrics metrics) {
+        normalizePreparedMessages(messages, metrics, taskExecutor);
+    }
+
+    private void normalizePreparedMessages(List<PreparedMessage> messages, BatchMetrics metrics,
+                                           Executor normalizationExecutor) {
         if (messages.isEmpty()) {
             return;
         }
@@ -509,7 +628,7 @@ public class SysLogProcessor {
                         return new PreparedNormalization(message, null,
                                 routeFailure(message.record, message.context, failure));
                     }
-                }, taskExecutor));
+                }, normalizationExecutor));
             }
             CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                     .get(processTimeoutMs, TimeUnit.MILLISECONDS);

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java

@@ -53,6 +53,16 @@ public class SyslogFailureRouter {
     private String retry1hTopic;
     @Value("${syslog.reliability.dlt-topic:${spring.kafka.consumer.topic}.DLT}")
     private String dltTopic;
+    @Value("${spring.kafka.consumer.large-topic:${spring.kafka.consumer.topic}.large}")
+    private String largeOriginalTopic;
+    @Value("${syslog.large-reliability.retry-1m-topic:${spring.kafka.consumer.large-topic}.RETRY.1M}")
+    private String largeRetry1mTopic;
+    @Value("${syslog.large-reliability.retry-10m-topic:${spring.kafka.consumer.large-topic}.RETRY.10M}")
+    private String largeRetry10mTopic;
+    @Value("${syslog.large-reliability.retry-1h-topic:${spring.kafka.consumer.large-topic}.RETRY.1H}")
+    private String largeRetry1hTopic;
+    @Value("${syslog.large-reliability.dlt-topic:${spring.kafka.consumer.large-topic}.DLT}")
+    private String largeDltTopic;
     @Value("${syslog.reliability.publish-timeout-ms:30000}")
     private long publishTimeoutMs;
     @Value("${syslog.reliability.consumer-version:unknown}")
@@ -129,7 +139,11 @@ public class SyslogFailureRouter {
         int nextAttempt = Math.max(0, context.getAttemptCount()) + 1;
         boolean permanent = category == FailureCategory.PERMANENT_DATA;
         boolean exhausted = nextAttempt > 3;
-        String targetTopic = (permanent || exhausted) ? dltTopic : retryTopic(nextAttempt);
+        boolean largeLane = largeOriginalTopic != null
+                && largeOriginalTopic.equals(context.getOriginalTopic());
+        String targetTopic = (permanent || exhausted)
+                ? (largeLane ? largeDltTopic : dltTopic)
+                : retryTopic(nextAttempt, largeLane);
         Instant nextRetryAt = (permanent || exhausted) ? null : Instant.now().plusSeconds(delaySeconds(nextAttempt));
 
         FailureEnvelope source = readEnvelope(record.value());
@@ -256,10 +270,10 @@ public class SyslogFailureRouter {
         return record.timestamp() > 0 ? Instant.ofEpochMilli(record.timestamp()) : Instant.now();
     }
 
-    private String retryTopic(int attempt) {
-        if (attempt == 1) return retry1mTopic;
-        if (attempt == 2) return retry10mTopic;
-        return retry1hTopic;
+    private String retryTopic(int attempt, boolean largeLane) {
+        if (attempt == 1) return largeLane ? largeRetry1mTopic : retry1mTopic;
+        if (attempt == 2) return largeLane ? largeRetry10mTopic : retry10mTopic;
+        return largeLane ? largeRetry1hTopic : retry1hTopic;
     }
 
     private long delaySeconds(int attempt) {

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DirectStatsProjectionWriter.java

@@ -23,9 +23,10 @@ import java.util.SortedMap;
 import java.util.TreeMap;
 
 /**
- * 把当前事实事务的内存聚合结果直接写入小型统计表。
+ * 把当前事实事务的内存聚合结果写入同步投影或可靠异步投影入口。
  *
- * <p>调用方必须已经开启事实事务；本类不会新开事务，也不会吞掉任何数据库异常。</p>
+ * <p>DIRECT 默认只追加一条批次 Outbox；SHADOW/显式回退仍同步更新统计表。调用方必须
+ * 已经开启事实事务，本类不会吞掉任何数据库异常。</p>
  */
 @Slf4j
 @Service
@@ -63,6 +64,7 @@ public class DirectStatsProjectionWriter {
     private final StatsWriteModeProvider modeProvider;
     private final DirectStatsWriteMetrics metrics;
     private final StatsMaterializationPublisher materializationPublisher;
+    private final StatsProjectionOutboxWriter projectionOutboxWriter;
 
     @Value("${stats.incremental.read-enabled:false}")
     private boolean readEnabled;
@@ -91,11 +93,13 @@ public class DirectStatsProjectionWriter {
     public DirectStatsProjectionWriter(JdbcTemplate jdbc,
                                        StatsWriteModeProvider modeProvider,
                                        DirectStatsWriteMetrics metrics,
-                                       StatsMaterializationPublisher materializationPublisher) {
+                                       StatsMaterializationPublisher materializationPublisher,
+                                       StatsProjectionOutboxWriter projectionOutboxWriter) {
         this.jdbc = jdbc;
         this.modeProvider = modeProvider;
         this.metrics = metrics;
         this.materializationPublisher = materializationPublisher;
+        this.projectionOutboxWriter = projectionOutboxWriter;
     }
 
     public WriteResult writeStandardizedRows(List<Map<String, Object>> rows) {
@@ -160,6 +164,8 @@ public class DirectStatsProjectionWriter {
         status.put("today_stats_date_enabled", todayStatsDateEnabled);
         status.put("business_materialization_mode", asynchronousMaterialization()
                 ? "KAFKA_DIRTY_KEY" : "SYNCHRONOUS");
+        status.put("daily_projection_write_mode", asynchronousProjection()
+                ? "DURABLE_OUTBOX" : "SYNCHRONOUS_MERGE");
         return status;
     }
 
@@ -194,6 +200,24 @@ public class DirectStatsProjectionWriter {
         }
         requireTransaction();
 
+        if (asynchronousProjection()) {
+            try {
+                long outboxStarted = System.nanoTime();
+                projectionOutboxWriter.append(StatsProjectionOutboxEvent.fromAccumulator(accumulator));
+                long outboxMs = elapsedMs(outboxStarted);
+                int deviceGroups = effectiveDeviceEntries(accumulator).size();
+                int ruleGroups = effectiveRuleEntries(accumulator).size();
+                metrics.recordSuccess(accumulator.getInputRows(), deviceGroups, ruleGroups,
+                        aggregateMs, outboxMs, 0L);
+                return WriteResult.direct(accumulator.getInputRows(), deviceGroups, ruleGroups,
+                        aggregateMs, outboxMs, 0L);
+            } catch (RuntimeException failure) {
+                metrics.recordFailure();
+                throw new DirectStatsWriteException(
+                        "统计投影 Outbox 写入失败，事实事务必须整体回滚", failure);
+            }
+        }
+
         List<Map.Entry<DirectStatsAccumulator.DailyKey, DirectStatsAccumulator.DeviceDelta>> devices =
                 effectiveDeviceEntries(accumulator);
         List<Map.Entry<DirectStatsAccumulator.DailyKey, DirectStatsAccumulator.RuleDelta>> rules =
@@ -244,6 +268,11 @@ public class DirectStatsProjectionWriter {
                     aggregateMs, 0L, 0L);
         }
         requireTransaction();
+        // Additive positive/negative changes commute in the serialized projection consumer. This
+        // keeps non-normal UPDATE statistics outside the fact transaction as well.
+        if (asynchronousProjection()) {
+            return write(accumulator, aggregateMs);
+        }
         List<DirectStatsAccumulator.DailyKey> keys = new ArrayList<>(accumulator.getDevices().keySet());
         try {
             long mergeStarted = System.nanoTime();
@@ -408,6 +437,11 @@ public class DirectStatsProjectionWriter {
                 && materializationPublisher.isEnabled();
     }
 
+    private boolean asynchronousProjection() {
+        return modeProvider.getMode() == StatsWriteModeProvider.Mode.DIRECT
+                && projectionOutboxWriter != null && projectionOutboxWriter.isEnabled();
+    }
+
     private List<StatsMaterializationCommand> materializationCommands(
             List<Map.Entry<DirectStatsAccumulator.DailyKey, DirectStatsAccumulator.DeviceDelta>> devices,
             List<Map.Entry<DirectStatsAccumulator.DailyKey, DirectStatsAccumulator.RuleDelta>> rules) {

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmLargeTextLimiter.java

@@ -0,0 +1,89 @@
+package com.common.service;
+
+import com.common.entity.SyslogNonNormalMessage;
+import com.common.util.Utf8TextTruncator;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.LongAdder;
+
+/** 达梦持久化边界的大字段保护；0 表示兼容模式下不限制。 */
+@Component
+public class DmLargeTextLimiter {
+    private static final long MIN_ENABLED_LIMIT = 64L;
+
+    @Value("${dm.normal.large-field-max-utf8-bytes:0}")
+    private long normalMaxBytes;
+
+    @Value("${dm.normal.large-fields:http_req_body,http_resp_body,http_req_header,http_resp_header,detail,description,cmdline,shell_cmdline,parent_cmdline,mail_message}")
+    private String normalFieldsValue;
+
+    @Value("${dm.non-normal.large-field-max-utf8-bytes:0}")
+    private long nonNormalMaxBytes;
+
+    private volatile Set<String> normalFields = Collections.emptySet();
+    private final LongAdder normalFieldsTruncated = new LongAdder();
+    private final LongAdder nonNormalFieldsTruncated = new LongAdder();
+
+    @PostConstruct
+    public void validate() {
+        validateLimit(normalMaxBytes, "dm.normal.large-field-max-utf8-bytes");
+        validateLimit(nonNormalMaxBytes, "dm.non-normal.large-field-max-utf8-bytes");
+        Set<String> parsed = new LinkedHashSet<>();
+        if (normalFieldsValue != null) {
+            Arrays.stream(normalFieldsValue.split(","))
+                    .map(String::trim).filter(value -> !value.isEmpty()).forEach(parsed::add);
+        }
+        normalFields = Collections.unmodifiableSet(parsed);
+    }
+
+    private static void validateLimit(long value, String name) {
+        if (value < 0L || (value > 0L && value < MIN_ENABLED_LIMIT)) {
+            throw new IllegalArgumentException(name + " 必须为 0 或至少 " + MIN_ENABLED_LIMIT);
+        }
+    }
+
+    public void truncateNormalRow(Map<String, Object> row) {
+        if (row == null || normalMaxBytes == 0L) return;
+        for (String field : normalFields) {
+            Object value = row.get(field);
+            if (!(value instanceof String)) continue;
+            Utf8TextTruncator.Result result = Utf8TextTruncator.truncate((String) value, normalMaxBytes);
+            if (result.isTruncated()) {
+                row.put(field, result.getValue());
+                normalFieldsTruncated.increment();
+            }
+        }
+    }
+
+    public void truncateNonNormal(SyslogNonNormalMessage message) {
+        if (message == null || nonNormalMaxBytes == 0L) return;
+        message.setSyslogMessage(truncateNonNormalValue(message.getSyslogMessage()));
+        message.setHeaderMessage(truncateNonNormalValue(message.getHeaderMessage()));
+        message.setReasonDetail(truncateNonNormalValue(message.getReasonDetail()));
+    }
+
+    private String truncateNonNormalValue(String value) {
+        Utf8TextTruncator.Result result = Utf8TextTruncator.truncate(value, nonNormalMaxBytes);
+        if (result.isTruncated()) nonNormalFieldsTruncated.increment();
+        return result.getValue();
+    }
+
+    public long getNormalFieldsTruncated() { return normalFieldsTruncated.sum(); }
+    public long getNonNormalFieldsTruncated() { return nonNormalFieldsTruncated.sum(); }
+
+    /** 测试和非 Spring 兼容入口。 */
+    static DmLargeTextLimiter unlimited() {
+        DmLargeTextLimiter limiter = new DmLargeTextLimiter();
+        limiter.normalMaxBytes = 0L;
+        limiter.nonNormalMaxBytes = 0L;
+        limiter.validate();
+        return limiter;
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java

@@ -174,8 +174,8 @@ public class DmStandardizedBatchExecutor {
                 lateAlarmBucketWriter.record(rows);
             }
 
-            // DIRECT/SHADOW 模式在当前事务内按设备、规则聚合写统计；LEGACY 模式会直接返回。
-            // 任一统计 SQL 失败都会回滚本事务的事实行，禁止产生无法定位的统计缺口。
+            // DIRECT 异步投影模式只在当前事实事务追加一条聚合 Outbox；独立 Kafka consumer
+            // 再更新日统计。SHADOW/回退模式仍保留原同步 MERGE。
             timing.statsResult = directStatsProjectionWriter.writeStandardizedRows(rows);
             return null;
         });
@@ -183,7 +183,7 @@ public class DmStandardizedBatchExecutor {
         transactionMetrics.recordSuccess(rows.size(), timing.factFlushMs, timing.commitMs, transactionMs);
         if (transactionLogEnabled && timing.statsResult != null && timing.statsResult.isDirect()) {
             log.info("DIRECT 达梦事务完成: factRows={}, deviceGroups={}, ruleGroups={}, aggregateMs={}, " +
-                            "factFlushMs={}, statsMergeMs={}, materializeMs={}, commitMs={}, transactionMs={}, retry={}",
+                            "factFlushMs={}, statsProjectionStageMs={}, materializeMs={}, commitMs={}, transactionMs={}, retry={}",
                     rows.size(), timing.statsResult.getDeviceGroups(), timing.statsResult.getRuleGroups(),
                     timing.statsResult.getAggregateMs(), timing.factFlushMs,
                     timing.statsResult.getMergeMs(), timing.statsResult.getMaterializeMs(),

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchWriter.java

@@ -60,6 +60,7 @@ public class DmStandardizedBatchWriter {
     private final SyslogNormalDataService normalDataService;
     private final DmStandardizedBatchExecutor batchExecutor;
     private final Executor writerExecutor;
+    private final DmLargeTextLimiter largeTextLimiter;
 
     // 未配置时直接启用本次改造的批写路径；SINGLE 只作为现场故障回退开关保留。
     @Value("${dm.standardized-write.mode:BATCH}")
@@ -89,12 +90,22 @@ public class DmStandardizedBatchWriter {
     private final AtomicInteger consecutiveSystemFailures = new AtomicInteger();
     private volatile long circuitOpenUntilMs;
 
+    @org.springframework.beans.factory.annotation.Autowired
     public DmStandardizedBatchWriter(SyslogNormalDataService normalDataService,
                                      DmStandardizedBatchExecutor batchExecutor,
-                                     @Qualifier("dmStandardizedWriterExecutor") Executor writerExecutor) {
+                                     @Qualifier("dmStandardizedWriterExecutor") Executor writerExecutor,
+                                     DmLargeTextLimiter largeTextLimiter) {
         this.normalDataService = normalDataService;
         this.batchExecutor = batchExecutor;
         this.writerExecutor = writerExecutor;
+        this.largeTextLimiter = largeTextLimiter;
+    }
+
+    /** 保留已有单元测试和离线工具入口，默认不改变历史存储内容。 */
+    public DmStandardizedBatchWriter(SyslogNormalDataService normalDataService,
+                                     DmStandardizedBatchExecutor batchExecutor,
+                                     Executor writerExecutor) {
+        this(normalDataService, batchExecutor, writerExecutor, DmLargeTextLimiter.unlimited());
     }
 
     /**
@@ -148,6 +159,7 @@ public class DmStandardizedBatchWriter {
         List<WriteRequest<T>> prepared = new ArrayList<>(requests.size());
         for (WriteRequest<T> request : requests) {
             try {
+                largeTextLimiter.truncateNormalRow(request.getData());
                 normalDataService.prepareForInsert(request.getData());
                 prepared.add(request);
             } catch (RuntimeException invalidRow) {

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/IngressStatsProjectionWriter.java

@@ -29,6 +29,7 @@ public class IngressStatsProjectionWriter {
     private final PlatformTransactionManager transactionManager;
     private final StatsWriteModeProvider modeProvider;
     private final StatsMaterializationPublisher materializationPublisher;
+    private final StatsProjectionOutboxWriter projectionOutboxWriter;
 
     @Value("${stats.ingress-metadata.enabled:false}")
     private boolean enabled;
@@ -50,11 +51,13 @@ public class IngressStatsProjectionWriter {
     public IngressStatsProjectionWriter(JdbcTemplate jdbc,
                                         PlatformTransactionManager transactionManager,
                                         StatsWriteModeProvider modeProvider,
-                                        StatsMaterializationPublisher materializationPublisher) {
+                                        StatsMaterializationPublisher materializationPublisher,
+                                        StatsProjectionOutboxWriter projectionOutboxWriter) {
         this.jdbc = jdbc;
         this.transactionManager = transactionManager;
         this.modeProvider = modeProvider;
         this.materializationPublisher = materializationPublisher;
+        this.projectionOutboxWriter = projectionOutboxWriter;
     }
 
     public int writeBatch(List<IngressStatsEvent> events) {
@@ -92,6 +95,8 @@ public class IngressStatsProjectionWriter {
                 ? "device_daily_stats+" + shadowDeviceTable : "device_daily_stats");
         status.put("business_materialization_mode", asynchronousMaterialization()
                 ? "KAFKA_DIRTY_KEY" : "SYNCHRONOUS");
+        status.put("daily_projection_write_mode", asynchronousProjection()
+                ? "DURABLE_OUTBOX" : "SYNCHRONOUS_MERGE");
         status.put("successful_batches", successfulBatches.get());
         status.put("received_events", receivedEvents.get());
         status.put("updated_groups", updatedGroups.get());
@@ -109,6 +114,10 @@ public class IngressStatsProjectionWriter {
         TransactionTemplate transaction = new TransactionTemplate(transactionManager);
         transaction.setTimeout(Math.max(1, transactionTimeoutSeconds));
         Integer updated = transaction.execute(status -> {
+            if (asynchronousProjection()) {
+                projectionOutboxWriter.append(StatsProjectionOutboxEvent.fromIngress(aggregation));
+                return aggregation.devices.size() + aggregation.collects.size();
+            }
             int groups = 0;
             boolean asyncMaterialization = asynchronousMaterialization();
             List<StatsMaterializationCommand> commands = asyncMaterialization
@@ -237,6 +246,11 @@ public class IngressStatsProjectionWriter {
                 && materializationPublisher.isEnabled();
     }
 
+    private boolean asynchronousProjection() {
+        return modeProvider.getMode() == StatsWriteModeProvider.Mode.DIRECT
+                && projectionOutboxWriter != null && projectionOutboxWriter.isEnabled();
+    }
+
     private static String safeTable(String table) {
         if (table == null || !table.matches("[A-Za-z][A-Za-z0-9_]*")) {
             throw new IllegalStateException("非法入口设备统计表名: " + table);

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsProjectionDeltaConsumer.java

@@ -0,0 +1,386 @@
+package com.common.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.sql.Date;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** Applies durable projection deltas outside the fact transaction. */
+@Slf4j
+@Component
+public class StatsProjectionDeltaConsumer {
+    private static final String DEVICE_MERGE =
+            "MERGE INTO %s t USING (SELECT ? stat_date,? device_id,? parse_delta," +
+            "? non_log_delta,? receive_time FROM DUAL) s " +
+            "ON (t.stat_date=s.stat_date AND t.device_id=s.device_id) " +
+            "WHEN MATCHED THEN UPDATE SET t.parse_count=t.parse_count+s.parse_delta," +
+            "t.non_log_count=t.non_log_count+s.non_log_delta," +
+            "t.last_receive_time=CASE WHEN s.receive_time IS NULL THEN t.last_receive_time " +
+            "WHEN t.last_receive_time IS NULL OR t.last_receive_time<s.receive_time " +
+            "THEN s.receive_time ELSE t.last_receive_time END," +
+            "t.projection_seq=t.projection_seq+1,t.updated_at=SYSDATE " +
+            "WHEN NOT MATCHED THEN INSERT (stat_date,device_id,parse_count,non_log_count," +
+            "last_receive_time,projection_seq,updated_at) VALUES " +
+            "(s.stat_date,s.device_id,s.parse_delta,s.non_log_delta,s.receive_time,1,SYSDATE)";
+
+    private static final String RULE_MERGE =
+            "MERGE INTO %s t USING (SELECT ? stat_date,? normalize_rule_id,? max_log_time FROM DUAL) s " +
+            "ON (t.stat_date=s.stat_date AND t.normalize_rule_id=s.normalize_rule_id) " +
+            "WHEN MATCHED THEN UPDATE SET t.max_log_time=CASE " +
+            "WHEN t.max_log_time IS NULL OR t.max_log_time<s.max_log_time THEN s.max_log_time " +
+            "ELSE t.max_log_time END,t.projection_seq=t.projection_seq+1,t.updated_at=SYSDATE " +
+            "WHEN NOT MATCHED THEN INSERT (stat_date,normalize_rule_id,max_log_time,projection_seq,updated_at) " +
+            "VALUES (s.stat_date,s.normalize_rule_id,s.max_log_time,1,SYSDATE)";
+
+    private static final String COLLECT_MERGE =
+            "MERGE INTO device_collect_stats t USING (SELECT ? device_collect_id," +
+            "? first_success_time,? last_success_time FROM DUAL) s " +
+            "ON (t.device_collect_id=s.device_collect_id) " +
+            "WHEN MATCHED THEN UPDATE SET t.first_success_time=CASE " +
+            "WHEN s.first_success_time IS NULL THEN t.first_success_time " +
+            "WHEN t.first_success_time IS NULL OR t.first_success_time>s.first_success_time " +
+            "THEN s.first_success_time ELSE t.first_success_time END," +
+            "t.last_success_time=CASE WHEN s.last_success_time IS NULL THEN t.last_success_time " +
+            "WHEN t.last_success_time IS NULL OR t.last_success_time<s.last_success_time " +
+            "THEN s.last_success_time ELSE t.last_success_time END," +
+            "t.projection_seq=t.projection_seq+1,t.updated_at=SYSDATE " +
+            "WHEN NOT MATCHED THEN INSERT (device_collect_id,first_success_time,last_success_time," +
+            "last_fail_time,projection_seq,updated_at) VALUES " +
+            "(s.device_collect_id,s.first_success_time,s.last_success_time,NULL,1,SYSDATE)";
+
+    private final JdbcTemplate jdbc;
+    private final PlatformTransactionManager transactionManager;
+    private final ObjectMapper objectMapper;
+    private final StatsMaterializationPublisher materializationPublisher;
+
+    @Value("${stats.projection-async.enabled:true}") private boolean enabled;
+    @Value("${stats.projection-async.applied-table:stats_projection_applied_event}") private String appliedTable;
+    @Value("${stats.projection-async.outbox-table:stats_projection_outbox}") private String outboxTable;
+    @Value("${stats.direct.device-table:device_daily_stats}") private String deviceTable;
+    @Value("${stats.direct.rule-table:normalize_rule_daily_stats}") private String ruleTable;
+    @Value("${stats.incremental.transaction-timeout-seconds:45}") private int transactionTimeoutSeconds;
+    @Value("${stats.projection-async.max-concurrency-retries:3}") private int maxConcurrencyRetries;
+    @Value("${stats.projection-async.concurrency-retry-backoff-ms:50}") private long retryBackoffMs;
+
+    private final AtomicLong consumed = new AtomicLong();
+    private final AtomicLong appliedEvents = new AtomicLong();
+    private final AtomicLong duplicates = new AtomicLong();
+    private final AtomicLong deviceGroups = new AtomicLong();
+    private final AtomicLong ruleGroups = new AtomicLong();
+    private final AtomicLong collectGroups = new AtomicLong();
+    private final AtomicLong successfulBatches = new AtomicLong();
+    private final AtomicLong failures = new AtomicLong();
+    private final AtomicLong retries = new AtomicLong();
+    private volatile LocalDateTime lastSuccessTime;
+    private volatile LocalDateTime lastFailureTime;
+
+    public StatsProjectionDeltaConsumer(JdbcTemplate jdbc,
+                                        PlatformTransactionManager transactionManager,
+                                        ObjectMapper objectMapper,
+                                        StatsMaterializationPublisher materializationPublisher) {
+        this.jdbc = jdbc;
+        this.transactionManager = transactionManager;
+        this.objectMapper = objectMapper.copy();
+        this.objectMapper.deactivateDefaultTyping();
+        this.materializationPublisher = materializationPublisher;
+    }
+
+    @KafkaListener(
+            id = "syslog-stats-projection",
+            groupId = "${stats.projection-async.group-id:${spring.kafka.consumer.group-id}-stats-projection}",
+            topics = "${stats.projection-async.topic:${spring.kafka.consumer.topic}.STATS.PROJECTION}",
+            autoStartup = "${stats.projection-async.enabled:true}",
+            containerFactory = "statsProjectionKafkaListenerContainerFactory"
+    )
+    public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
+        if (records == null || records.isEmpty()) {
+            acknowledgment.acknowledge();
+            return;
+        }
+        consumed.addAndGet(records.size());
+        List<StatsProjectionOutboxEvent> events = parse(records);
+        long started = System.nanoTime();
+        try {
+            ApplyResult result = applyWithRetry(events);
+            acknowledgment.acknowledge();
+            appliedEvents.addAndGet(result.appliedEvents);
+            duplicates.addAndGet(result.duplicates);
+            deviceGroups.addAndGet(result.deviceGroups);
+            ruleGroups.addAndGet(result.ruleGroups);
+            collectGroups.addAndGet(result.collectGroups);
+            successfulBatches.incrementAndGet();
+            lastSuccessTime = LocalDateTime.now();
+            log.info("统计投影批次完成: received={}, appliedEvents={}, duplicates={}, " +
+                            "deviceGroups={}, ruleGroups={}, collectGroups={}, durationMs={}",
+                    records.size(), result.appliedEvents, result.duplicates,
+                    result.deviceGroups, result.ruleGroups, result.collectGroups,
+                    (System.nanoTime() - started) / 1_000_000L);
+        } catch (RuntimeException failure) {
+            failures.incrementAndGet();
+            lastFailureTime = LocalDateTime.now();
+            throw failure;
+        }
+    }
+
+    private List<StatsProjectionOutboxEvent> parse(List<ConsumerRecord<String, String>> records) {
+        Map<String, StatsProjectionOutboxEvent> unique = new LinkedHashMap<>();
+        for (ConsumerRecord<String, String> record : records) {
+            try {
+                if (!StatsProjectionOutboxEvent.KAFKA_KEY.equals(record.key())) {
+                    throw new IllegalArgumentException("Unexpected stats projection Kafka key: " + record.key());
+                }
+                StatsProjectionOutboxEvent event = objectMapper.readValue(
+                        record.value(), StatsProjectionOutboxEvent.class);
+                event.validate();
+                unique.put(event.getEventId(), event);
+            } catch (Exception invalid) {
+                throw new IllegalStateException("Invalid internal stats projection event at " +
+                        record.topic() + "-" + record.partition() + "@" + record.offset(), invalid);
+            }
+        }
+        return new ArrayList<>(unique.values());
+    }
+
+    ApplyResult applyWithRetry(List<StatsProjectionOutboxEvent> events) {
+        int attempt = 0;
+        while (true) {
+            try {
+                return applyOnce(events);
+            } catch (RuntimeException failure) {
+                if (!DirectStatsProjectionWriter.isConcurrentFailure(failure)
+                        || attempt >= Math.max(0, maxConcurrencyRetries)) {
+                    throw failure;
+                }
+                attempt++;
+                retries.incrementAndGet();
+                sleepBeforeRetry(attempt);
+            }
+        }
+    }
+
+    private ApplyResult applyOnce(List<StatsProjectionOutboxEvent> events) {
+        if (!enabled) throw new IllegalStateException("Stats projection consumer is disabled");
+        TransactionTemplate transaction = new TransactionTemplate(transactionManager);
+        transaction.setTimeout(Math.max(1, transactionTimeoutSeconds));
+        ApplyResult result = transaction.execute(status -> {
+            Set<String> existing = existingEventIds(events);
+            List<StatsProjectionOutboxEvent> fresh = new ArrayList<>();
+            for (StatsProjectionOutboxEvent event : events) {
+                if (!existing.contains(event.getEventId())) fresh.add(event);
+            }
+            ProjectionAggregation aggregation = aggregate(fresh);
+            if (!aggregation.devices.isEmpty()) {
+                jdbc.batchUpdate(String.format(DEVICE_MERGE,
+                        StatsProjectionOutboxWriter.safeTable(deviceTable)), aggregation.deviceParams());
+            }
+            if (!aggregation.rules.isEmpty()) {
+                jdbc.batchUpdate(String.format(RULE_MERGE,
+                        StatsProjectionOutboxWriter.safeTable(ruleTable)), aggregation.ruleParams());
+            }
+            if (!aggregation.collects.isEmpty()) {
+                jdbc.batchUpdate(COLLECT_MERGE, aggregation.collectParams());
+            }
+            if (!fresh.isEmpty()) {
+                List<Object[]> markers = new ArrayList<>(fresh.size());
+                for (StatsProjectionOutboxEvent event : fresh) {
+                    markers.add(new Object[]{event.getEventId(), event.getSchema()});
+                }
+                jdbc.batchUpdate("INSERT INTO " + StatsProjectionOutboxWriter.safeTable(appliedTable) +
+                        " (event_id,event_schema,applied_at) VALUES (?,?,SYSDATE)", markers);
+            }
+            if (!events.isEmpty()) {
+                List<Object[]> acknowledged = new ArrayList<>(events.size());
+                for (StatsProjectionOutboxEvent event : events) {
+                    acknowledged.add(new Object[]{event.getEventId()});
+                }
+                jdbc.batchUpdate("UPDATE " + StatsProjectionOutboxWriter.safeTable(outboxTable) +
+                        " SET status='APPLIED',applied_at=SYSDATE,next_attempt_at=NULL," +
+                        "updated_at=SYSDATE WHERE event_id=?", acknowledged);
+            }
+            Collection<StatsMaterializationCommand> dirty = aggregation.materializationCommands();
+            if (!dirty.isEmpty()) materializationPublisher.enqueueAfterCommit(dirty);
+            return new ApplyResult(fresh.size(), events.size() - fresh.size(),
+                    aggregation.devices.size(), aggregation.rules.size(), aggregation.collects.size());
+        });
+        return result == null ? new ApplyResult(0, events.size(), 0, 0, 0) : result;
+    }
+
+    private Set<String> existingEventIds(List<StatsProjectionOutboxEvent> events) {
+        if (events.isEmpty()) return Collections.emptySet();
+        Set<String> result = new HashSet<>();
+        int chunkSize = 900;
+        for (int start = 0; start < events.size(); start += chunkSize) {
+            int end = Math.min(events.size(), start + chunkSize);
+            StringBuilder placeholders = new StringBuilder();
+            Object[] ids = new Object[end - start];
+            for (int i = start; i < end; i++) {
+                if (placeholders.length() > 0) placeholders.append(',');
+                placeholders.append('?');
+                ids[i - start] = events.get(i).getEventId();
+            }
+            result.addAll(jdbc.queryForList("SELECT event_id FROM " +
+                    StatsProjectionOutboxWriter.safeTable(appliedTable) + " WHERE event_id IN (" +
+                    placeholders + ")", String.class, ids));
+        }
+        return result;
+    }
+
+    static ProjectionAggregation aggregate(List<StatsProjectionOutboxEvent> events) {
+        ProjectionAggregation result = new ProjectionAggregation();
+        for (StatsProjectionOutboxEvent event : events) {
+            event.validate();
+            for (StatsProjectionOutboxEvent.DeviceDelta delta : event.getDevices()) {
+                DeviceKey key = new DeviceKey(delta.date(), delta.getDeviceId());
+                DeviceAggregate value = result.devices.computeIfAbsent(key,
+                        ignored -> new DeviceAggregate());
+                value.parseDelta += delta.getParseDelta();
+                value.nonLogDelta += delta.getNonLogDelta();
+                LocalDateTime receiveTime = delta.receiveTime();
+                if (receiveTime != null && (value.lastReceiveTime == null
+                        || receiveTime.isAfter(value.lastReceiveTime))) {
+                    value.lastReceiveTime = receiveTime;
+                }
+            }
+            for (StatsProjectionOutboxEvent.RuleDelta delta : event.getRules()) {
+                RuleKey key = new RuleKey(delta.date(), delta.getNormalizeRuleId());
+                LocalDateTime value = delta.time();
+                LocalDateTime current = result.rules.get(key);
+                if (current == null || value.isAfter(current)) result.rules.put(key, value);
+            }
+            for (StatsProjectionOutboxEvent.CollectDelta delta : event.getCollects()) {
+                CollectAggregate value = result.collects.computeIfAbsent(delta.getDeviceCollectId(),
+                        ignored -> new CollectAggregate());
+                value.include(delta.firstTime());
+                value.include(delta.lastTime());
+            }
+        }
+        return result;
+    }
+
+    private void sleepBeforeRetry(int attempt) {
+        try {
+            Thread.sleep(Math.max(0L, retryBackoffMs) * attempt);
+        } catch (InterruptedException interrupted) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException("Stats projection retry interrupted", interrupted);
+        }
+    }
+
+    public Map<String, Object> status() {
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("enabled", enabled);
+        result.put("consumed_total", consumed.get());
+        result.put("applied_events_total", appliedEvents.get());
+        result.put("duplicate_events_total", duplicates.get());
+        result.put("device_groups_total", deviceGroups.get());
+        result.put("rule_groups_total", ruleGroups.get());
+        result.put("collect_groups_total", collectGroups.get());
+        result.put("successful_batches", successfulBatches.get());
+        result.put("failures_total", failures.get());
+        result.put("concurrency_retries_total", retries.get());
+        result.put("last_success_time", lastSuccessTime);
+        result.put("last_failure_time", lastFailureTime);
+        return result;
+    }
+
+    static final class ProjectionAggregation {
+        final Map<DeviceKey, DeviceAggregate> devices = new TreeMap<>();
+        final Map<RuleKey, LocalDateTime> rules = new TreeMap<>();
+        final Map<Long, CollectAggregate> collects = new TreeMap<>();
+
+        List<Object[]> deviceParams() {
+            List<Object[]> result = new ArrayList<>(devices.size());
+            for (Map.Entry<DeviceKey, DeviceAggregate> entry : devices.entrySet()) {
+                DeviceAggregate value = entry.getValue();
+                result.add(new Object[]{Date.valueOf(entry.getKey().date), entry.getKey().id,
+                        value.parseDelta, value.nonLogDelta, value.lastReceiveTime});
+            }
+            return result;
+        }
+
+        List<Object[]> ruleParams() {
+            List<Object[]> result = new ArrayList<>(rules.size());
+            for (Map.Entry<RuleKey, LocalDateTime> entry : rules.entrySet()) {
+                result.add(new Object[]{Date.valueOf(entry.getKey().date), entry.getKey().id,
+                        entry.getValue()});
+            }
+            return result;
+        }
+
+        List<Object[]> collectParams() {
+            List<Object[]> result = new ArrayList<>(collects.size());
+            for (Map.Entry<Long, CollectAggregate> entry : collects.entrySet()) {
+                result.add(new Object[]{entry.getKey(), entry.getValue().min, entry.getValue().max});
+            }
+            return result;
+        }
+
+        Collection<StatsMaterializationCommand> materializationCommands() {
+            Set<StatsMaterializationCommand> result = new LinkedHashSet<>();
+            for (DeviceKey key : devices.keySet()) result.add(StatsMaterializationCommand.device(key.date, key.id));
+            for (RuleKey key : rules.keySet()) result.add(StatsMaterializationCommand.rule(key.date, key.id));
+            for (Long id : collects.keySet()) result.add(StatsMaterializationCommand.collect(id));
+            return result;
+        }
+    }
+
+    static class DimensionKey implements Comparable<DimensionKey> {
+        final LocalDate date;
+        final long id;
+        DimensionKey(LocalDate date, long id) { this.date = date; this.id = id; }
+        @Override public int compareTo(DimensionKey other) {
+            int dateOrder = date.compareTo(other.date);
+            return dateOrder != 0 ? dateOrder : Long.compare(id, other.id);
+        }
+        @Override public boolean equals(Object other) {
+            if (this == other) return true;
+            if (!(other instanceof DimensionKey)) return false;
+            DimensionKey key = (DimensionKey) other;
+            return id == key.id && date.equals(key.date);
+        }
+        @Override public int hashCode() { return 31 * date.hashCode() + Long.hashCode(id); }
+    }
+    static final class DeviceKey extends DimensionKey { DeviceKey(LocalDate date, long id) { super(date, id); } }
+    static final class RuleKey extends DimensionKey { RuleKey(LocalDate date, long id) { super(date, id); } }
+    static final class DeviceAggregate { long parseDelta; long nonLogDelta; LocalDateTime lastReceiveTime; }
+    static final class CollectAggregate {
+        LocalDateTime min; LocalDateTime max;
+        void include(LocalDateTime time) {
+            if (time == null) return;
+            if (min == null || time.isBefore(min)) min = time;
+            if (max == null || time.isAfter(max)) max = time;
+        }
+    }
+
+    static final class ApplyResult {
+        final int appliedEvents; final int duplicates;
+        final int deviceGroups; final int ruleGroups; final int collectGroups;
+        ApplyResult(int appliedEvents, int duplicates, int deviceGroups, int ruleGroups, int collectGroups) {
+            this.appliedEvents = appliedEvents; this.duplicates = duplicates;
+            this.deviceGroups = deviceGroups; this.ruleGroups = ruleGroups; this.collectGroups = collectGroups;
+        }
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsProjectionOutboxEvent.java

@@ -0,0 +1,221 @@
+package com.common.service;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * A durable, batch-aggregated statistics delta written beside the fact rows.
+ *
+ * <p>The event is additive/max-min only, so events from different application nodes may be
+ * published in a different order without changing the final projection. Kafka duplicates are
+ * eliminated by {@code eventId} in the projection consumer transaction.</p>
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class StatsProjectionOutboxEvent {
+    public static final String SCHEMA = "SYSLOG_STATS_PROJECTION_V1";
+    public static final String KAFKA_KEY = "SYSLOG_STATS_PROJECTION";
+
+    private String schema = SCHEMA;
+    private String eventId = UUID.randomUUID().toString();
+    private List<DeviceDelta> devices = new ArrayList<>();
+    private List<RuleDelta> rules = new ArrayList<>();
+    private List<CollectDelta> collects = new ArrayList<>();
+
+    public static StatsProjectionOutboxEvent fromAccumulator(DirectStatsAccumulator accumulator) {
+        StatsProjectionOutboxEvent event = new StatsProjectionOutboxEvent();
+        for (Map.Entry<DirectStatsAccumulator.DailyKey, DirectStatsAccumulator.DeviceDelta> entry
+                : accumulator.getDevices().entrySet()) {
+            DirectStatsAccumulator.DeviceDelta value = entry.getValue();
+            if (value.getParseDelta() != 0L || value.getNonLogDelta() != 0L) {
+                event.devices.add(DeviceDelta.counts(entry.getKey().getDate(),
+                        entry.getKey().getDimensionId(), value.getParseDelta(), value.getNonLogDelta()));
+            }
+        }
+        for (Map.Entry<DirectStatsAccumulator.DailyKey, DirectStatsAccumulator.RuleDelta> entry
+                : accumulator.getRules().entrySet()) {
+            if (entry.getValue().getMaxLogTime() != null) {
+                event.rules.add(RuleDelta.max(entry.getKey().getDate(),
+                        entry.getKey().getDimensionId(), entry.getValue().getMaxLogTime()));
+            }
+        }
+        return event;
+    }
+
+    static StatsProjectionOutboxEvent fromIngress(IngressStatsProjectionWriter.Aggregation aggregation) {
+        StatsProjectionOutboxEvent event = new StatsProjectionOutboxEvent();
+        for (Map.Entry<IngressStatsProjectionWriter.DeviceKey, LocalDateTime> entry
+                : aggregation.devices.entrySet()) {
+            event.devices.add(DeviceDelta.received(entry.getKey().date,
+                    entry.getKey().deviceId, entry.getValue()));
+        }
+        for (Map.Entry<Long, IngressStatsProjectionWriter.CollectTimes> entry
+                : aggregation.collects.entrySet()) {
+            event.collects.add(CollectDelta.range(entry.getKey(),
+                    entry.getValue().min, entry.getValue().max));
+        }
+        return event;
+    }
+
+    public void validate() {
+        if (!SCHEMA.equals(schema)) {
+            throw new IllegalArgumentException("Unsupported stats projection schema: " + schema);
+        }
+        if (eventId == null || eventId.trim().isEmpty()) {
+            throw new IllegalArgumentException("Stats projection event is missing eventId");
+        }
+        if (devices == null || rules == null || collects == null) {
+            throw new IllegalArgumentException("Stats projection event contains null collections");
+        }
+        if (devices.isEmpty() && rules.isEmpty() && collects.isEmpty()) {
+            throw new IllegalArgumentException("Stats projection event has no changes");
+        }
+        for (DeviceDelta delta : devices) delta.validate();
+        for (RuleDelta delta : rules) delta.validate();
+        for (CollectDelta delta : collects) delta.validate();
+    }
+
+    public boolean isEmpty() {
+        return devices.isEmpty() && rules.isEmpty() && collects.isEmpty();
+    }
+
+    public String getSchema() { return schema; }
+    public void setSchema(String schema) { this.schema = schema; }
+    public String getEventId() { return eventId; }
+    public void setEventId(String eventId) { this.eventId = eventId; }
+    public List<DeviceDelta> getDevices() { return devices; }
+    public void setDevices(List<DeviceDelta> devices) { this.devices = devices; }
+    public List<RuleDelta> getRules() { return rules; }
+    public void setRules(List<RuleDelta> rules) { this.rules = rules; }
+    public List<CollectDelta> getCollects() { return collects; }
+    public void setCollects(List<CollectDelta> collects) { this.collects = collects; }
+
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    public static class DeviceDelta {
+        private String statDate;
+        private long deviceId;
+        private long parseDelta;
+        private long nonLogDelta;
+        private String lastReceiveTime;
+
+        public DeviceDelta() {}
+
+        static DeviceDelta counts(LocalDate date, long deviceId, long parseDelta, long nonLogDelta) {
+            DeviceDelta result = new DeviceDelta();
+            result.statDate = date.toString();
+            result.deviceId = deviceId;
+            result.parseDelta = parseDelta;
+            result.nonLogDelta = nonLogDelta;
+            return result;
+        }
+
+        static DeviceDelta received(LocalDate date, long deviceId, LocalDateTime receiveTime) {
+            DeviceDelta result = counts(date, deviceId, 0L, 0L);
+            result.lastReceiveTime = receiveTime.toString();
+            return result;
+        }
+
+        void validate() {
+            date();
+            if (deviceId <= 0L) throw new IllegalArgumentException("Invalid device projection id");
+            if (parseDelta == 0L && nonLogDelta == 0L && receiveTime() == null) {
+                throw new IllegalArgumentException("Device projection delta is empty");
+            }
+        }
+
+        public LocalDate date() { return LocalDate.parse(statDate); }
+        public LocalDateTime receiveTime() {
+            return lastReceiveTime == null || lastReceiveTime.trim().isEmpty()
+                    ? null : LocalDateTime.parse(lastReceiveTime);
+        }
+        public String getStatDate() { return statDate; }
+        public void setStatDate(String statDate) { this.statDate = statDate; }
+        public long getDeviceId() { return deviceId; }
+        public void setDeviceId(long deviceId) { this.deviceId = deviceId; }
+        public long getParseDelta() { return parseDelta; }
+        public void setParseDelta(long parseDelta) { this.parseDelta = parseDelta; }
+        public long getNonLogDelta() { return nonLogDelta; }
+        public void setNonLogDelta(long nonLogDelta) { this.nonLogDelta = nonLogDelta; }
+        public String getLastReceiveTime() { return lastReceiveTime; }
+        public void setLastReceiveTime(String lastReceiveTime) { this.lastReceiveTime = lastReceiveTime; }
+    }
+
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    public static class RuleDelta {
+        private String statDate;
+        private long normalizeRuleId;
+        private String maxLogTime;
+
+        public RuleDelta() {}
+
+        static RuleDelta max(LocalDate date, long ruleId, LocalDateTime maxLogTime) {
+            RuleDelta result = new RuleDelta();
+            result.statDate = date.toString();
+            result.normalizeRuleId = ruleId;
+            result.maxLogTime = maxLogTime.toString();
+            return result;
+        }
+
+        void validate() {
+            date();
+            if (normalizeRuleId <= 0L || time() == null) {
+                throw new IllegalArgumentException("Invalid rule projection delta");
+            }
+        }
+
+        public LocalDate date() { return LocalDate.parse(statDate); }
+        public LocalDateTime time() {
+            return maxLogTime == null || maxLogTime.trim().isEmpty()
+                    ? null : LocalDateTime.parse(maxLogTime);
+        }
+        public String getStatDate() { return statDate; }
+        public void setStatDate(String statDate) { this.statDate = statDate; }
+        public long getNormalizeRuleId() { return normalizeRuleId; }
+        public void setNormalizeRuleId(long normalizeRuleId) { this.normalizeRuleId = normalizeRuleId; }
+        public String getMaxLogTime() { return maxLogTime; }
+        public void setMaxLogTime(String maxLogTime) { this.maxLogTime = maxLogTime; }
+    }
+
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    public static class CollectDelta {
+        private long deviceCollectId;
+        private String firstSuccessTime;
+        private String lastSuccessTime;
+
+        public CollectDelta() {}
+
+        static CollectDelta range(long collectId, LocalDateTime first, LocalDateTime last) {
+            CollectDelta result = new CollectDelta();
+            result.deviceCollectId = collectId;
+            result.firstSuccessTime = first == null ? null : first.toString();
+            result.lastSuccessTime = last == null ? null : last.toString();
+            return result;
+        }
+
+        void validate() {
+            if (deviceCollectId <= 0L || (firstTime() == null && lastTime() == null)) {
+                throw new IllegalArgumentException("Invalid collector projection delta");
+            }
+        }
+
+        public LocalDateTime firstTime() {
+            return firstSuccessTime == null || firstSuccessTime.trim().isEmpty()
+                    ? null : LocalDateTime.parse(firstSuccessTime);
+        }
+        public LocalDateTime lastTime() {
+            return lastSuccessTime == null || lastSuccessTime.trim().isEmpty()
+                    ? null : LocalDateTime.parse(lastSuccessTime);
+        }
+        public long getDeviceCollectId() { return deviceCollectId; }
+        public void setDeviceCollectId(long deviceCollectId) { this.deviceCollectId = deviceCollectId; }
+        public String getFirstSuccessTime() { return firstSuccessTime; }
+        public void setFirstSuccessTime(String firstSuccessTime) { this.firstSuccessTime = firstSuccessTime; }
+        public String getLastSuccessTime() { return lastSuccessTime; }
+        public void setLastSuccessTime(String lastSuccessTime) { this.lastSuccessTime = lastSuccessTime; }
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsProjectionOutboxRelay.java

@@ -0,0 +1,151 @@
+package com.common.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.util.concurrent.ListenableFuture;
+
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Relays durable outbox events to the dedicated projection topic.
+ *
+ * <p>Multiple application nodes may relay the same row during a race. The consumer's applied-event
+ * table makes that harmless, so the relay never needs a distributed lease that could stall the
+ * fact path.</p>
+ */
+@Slf4j
+@Component
+public class StatsProjectionOutboxRelay {
+    private final JdbcTemplate jdbc;
+    private final KafkaTemplate<String, String> kafkaTemplate;
+    private final AtomicBoolean running = new AtomicBoolean();
+    private final AtomicLong published = new AtomicLong();
+    private final AtomicLong failures = new AtomicLong();
+    private final AtomicLong cleaned = new AtomicLong();
+    private volatile long lastSuccessAt;
+    private volatile long lastFailureAt;
+
+    @Value("${stats.projection-async.enabled:true}") private boolean enabled;
+    @Value("${stats.projection-async.outbox-table:stats_projection_outbox}") private String outboxTable;
+    @Value("${stats.projection-async.applied-table:stats_projection_applied_event}") private String appliedTable;
+    @Value("${stats.projection-async.topic:${spring.kafka.consumer.topic}.STATS.PROJECTION}") private String topic;
+    @Value("${stats.projection-async.relay-batch-size:100}") private int batchSize;
+    @Value("${stats.projection-async.publish-timeout-ms:10000}") private long publishTimeoutMs;
+    @Value("${stats.projection-async.retry-backoff-ms:1000}") private long retryBackoffMs;
+    @Value("${stats.projection-async.redelivery-interval-ms:60000}") private long redeliveryIntervalMs;
+    @Value("${stats.projection-async.retention-days:14}") private int retentionDays;
+
+    public StatsProjectionOutboxRelay(JdbcTemplate jdbc, KafkaTemplate<String, String> kafkaTemplate) {
+        this.jdbc = jdbc;
+        this.kafkaTemplate = kafkaTemplate;
+    }
+
+    @Scheduled(fixedDelayString = "${stats.projection-async.relay-interval-ms:200}")
+    public void relay() {
+        if (!enabled || !running.compareAndSet(false, true)) return;
+        try {
+            for (OutboxRow row : loadDueRows()) {
+                publish(row);
+            }
+        } catch (RuntimeException failure) {
+            failures.incrementAndGet();
+            lastFailureAt = System.currentTimeMillis();
+            log.error("统计投影 Outbox 扫描失败，下轮自动重试", failure);
+        } finally {
+            running.set(false);
+        }
+    }
+
+    private List<OutboxRow> loadDueRows() {
+        String table = StatsProjectionOutboxWriter.safeTable(outboxTable);
+        String sql = "SELECT event_id,payload FROM (SELECT event_id,payload FROM " + table +
+                " WHERE status IN ('PENDING','FAILED','PUBLISHED') AND (next_attempt_at IS NULL " +
+                "OR next_attempt_at<=SYSDATE) ORDER BY created_at,event_id) WHERE ROWNUM<=?";
+        return jdbc.query(sql, ps -> ps.setInt(1, Math.max(1, batchSize)),
+                (rs, rowNum) -> new OutboxRow(rs.getString(1), rs.getString(2)));
+    }
+
+    private void publish(OutboxRow row) {
+        try {
+            ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(
+                    topic, StatsProjectionOutboxEvent.KAFKA_KEY, row.payload);
+            future.get(Math.max(1L, publishTimeoutMs), TimeUnit.MILLISECONDS);
+            jdbc.update("UPDATE " + StatsProjectionOutboxWriter.safeTable(outboxTable) +
+                            " SET status='PUBLISHED',published_at=SYSDATE,next_attempt_at=?,last_error=NULL," +
+                            "updated_at=SYSDATE WHERE event_id=? AND status<>'APPLIED'",
+                    new Timestamp(System.currentTimeMillis() + Math.max(1000L, redeliveryIntervalMs)),
+                    row.eventId);
+            published.incrementAndGet();
+            lastSuccessAt = System.currentTimeMillis();
+        } catch (InterruptedException interrupted) {
+            Thread.currentThread().interrupt();
+            markFailure(row, interrupted);
+        } catch (Exception failure) {
+            markFailure(row, failure);
+        }
+    }
+
+    private void markFailure(OutboxRow row, Exception failure) {
+        failures.incrementAndGet();
+        lastFailureAt = System.currentTimeMillis();
+        String message = String.valueOf(failure.getMessage());
+        if (message.length() > 1800) message = message.substring(0, 1800);
+        try {
+            jdbc.update("UPDATE " + StatsProjectionOutboxWriter.safeTable(outboxTable) +
+                            " SET status='FAILED',attempt_count=attempt_count+1,next_attempt_at=?," +
+                            "last_error=?,updated_at=SYSDATE WHERE event_id=? AND status<>'APPLIED'",
+                    new Timestamp(System.currentTimeMillis() + Math.max(0L, retryBackoffMs)),
+                    message, row.eventId);
+        } catch (RuntimeException markFailure) {
+            failure.addSuppressed(markFailure);
+        }
+        log.error("统计投影 Outbox 发布失败，将自动重试: eventId={}", row.eventId, failure);
+    }
+
+    @Scheduled(initialDelayString = "${stats.projection-async.cleanup-initial-delay-ms:3600000}",
+            fixedDelayString = "${stats.projection-async.cleanup-interval-ms:3600000}")
+    public void cleanup() {
+        if (!enabled) return;
+        long keepMillis = TimeUnit.DAYS.toMillis(Math.max(1, retentionDays));
+        Timestamp cutoff = new Timestamp(System.currentTimeMillis() - keepMillis);
+        int removed = jdbc.update("DELETE FROM " + StatsProjectionOutboxWriter.safeTable(outboxTable) +
+                " WHERE status='APPLIED' AND applied_at<?", cutoff);
+        removed += jdbc.update("DELETE FROM " + StatsProjectionOutboxWriter.safeTable(appliedTable) +
+                " WHERE applied_at<?", cutoff);
+        cleaned.addAndGet(removed);
+    }
+
+    public Map<String, Object> status() {
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("enabled", enabled);
+        result.put("topic", topic);
+        result.put("running", running.get());
+        result.put("published_total", published.get());
+        result.put("failures_total", failures.get());
+        result.put("cleaned_rows_total", cleaned.get());
+        result.put("last_success_epoch_ms", lastSuccessAt == 0L ? null : lastSuccessAt);
+        result.put("last_failure_epoch_ms", lastFailureAt == 0L ? null : lastFailureAt);
+        return result;
+    }
+
+    private static final class OutboxRow {
+        private final String eventId;
+        private final String payload;
+        private OutboxRow(String eventId, String payload) {
+            this.eventId = eventId;
+            this.payload = payload;
+        }
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsProjectionOutboxWriter.java

@@ -0,0 +1,70 @@
+package com.common.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** Writes one aggregated outbox row in the same transaction as the corresponding fact batch. */
+@Component
+public class StatsProjectionOutboxWriter {
+    private final JdbcTemplate jdbc;
+    private final ObjectMapper objectMapper;
+    private final AtomicLong events = new AtomicLong();
+    private final AtomicLong failures = new AtomicLong();
+
+    @Value("${stats.projection-async.enabled:true}") private boolean enabled;
+    @Value("${stats.projection-async.outbox-table:stats_projection_outbox}") private String outboxTable;
+
+    public StatsProjectionOutboxWriter(JdbcTemplate jdbc, ObjectMapper objectMapper) {
+        this.jdbc = jdbc;
+        this.objectMapper = objectMapper.copy();
+        this.objectMapper.deactivateDefaultTyping();
+    }
+
+    public boolean isEnabled() { return enabled; }
+
+    public int append(StatsProjectionOutboxEvent event) {
+        if (!enabled || event == null || event.isEmpty()) return 0;
+        if (!TransactionSynchronizationManager.isActualTransactionActive()) {
+            throw new IllegalStateException("Stats projection outbox must be written inside a transaction");
+        }
+        event.validate();
+        try {
+            int inserted = jdbc.update("INSERT INTO " + safeTable(outboxTable) +
+                            " (event_id,event_schema,payload,status,attempt_count,next_attempt_at," +
+                            "created_at,updated_at) VALUES (?,?,?,'PENDING',0,SYSDATE,SYSDATE,SYSDATE)",
+                    event.getEventId(), event.getSchema(), objectMapper.writeValueAsString(event));
+            events.incrementAndGet();
+            return inserted;
+        } catch (JsonProcessingException serializationFailure) {
+            failures.incrementAndGet();
+            throw new IllegalStateException("Failed to serialize stats projection outbox event", serializationFailure);
+        } catch (RuntimeException databaseFailure) {
+            failures.incrementAndGet();
+            throw databaseFailure;
+        }
+    }
+
+    public Map<String, Object> status() {
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("enabled", enabled);
+        result.put("outbox_table", outboxTable);
+        result.put("events_written_total", events.get());
+        result.put("write_failures_total", failures.get());
+        return result;
+    }
+
+    static String safeTable(String table) {
+        if (table == null || !table.matches("[A-Za-z][A-Za-z0-9_]*")) {
+            throw new IllegalStateException("Invalid stats projection table: " + table);
+        }
+        return table;
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/SyslogNonNormalMessageServiceImpl.java

@@ -9,6 +9,7 @@ import com.common.mapper.SyslogNonNormalMessageMapper;
 import com.common.service.SyslogNonNormalMessageService;
 import com.common.service.StatsChangeCaptureService;
 import com.common.service.DirectStatsProjectionWriter;
+import com.common.service.DmLargeTextLimiter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -40,6 +41,9 @@ public class SyslogNonNormalMessageServiceImpl
     @Autowired
     private DirectStatsProjectionWriter directStatsProjectionWriter;
 
+    @Autowired(required = false)
+    private DmLargeTextLimiter largeTextLimiter;
+
     @Value("${dm.non-normal-write.batch-size:500}")
     private int nonNormalBatchSize;
 
@@ -47,6 +51,7 @@ public class SyslogNonNormalMessageServiceImpl
     @Transactional(rollbackFor = Exception.class)
     public boolean saveMessage(SyslogNonNormalMessage message) {
         try {
+            truncateLargeFields(message);
             // 设置ID（如果未提供）
             if (message.getId() == null || message.getId().isEmpty()) {
                 message.setId(generateMessageId());
@@ -101,6 +106,7 @@ public class SyslogNonNormalMessageServiceImpl
         try {
             // 为每条记录设置默认值
             for (SyslogNonNormalMessage message : messages) {
+                truncateLargeFields(message);
                 if (message.getId() == null || message.getId().isEmpty()) {
                     message.setId(generateMessageId());
                 }
@@ -159,10 +165,15 @@ public class SyslogNonNormalMessageServiceImpl
         }
     }
 
+    private void truncateLargeFields(SyslogNonNormalMessage message) {
+        if (largeTextLimiter != null) largeTextLimiter.truncateNonNormal(message);
+    }
+
     @Override
     @Transactional(rollbackFor = Exception.class)
     public boolean updateMessage(SyslogNonNormalMessage message) {
         try {
+            truncateLargeFields(message);
             SyslogNonNormalMessage oldValue = getMessageById(message.getId());
             // 设置更新时间
             message.setUpdateTime(LocalDateTime.now());

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/Utf8TextTruncator.java

@@ -0,0 +1,88 @@
+package com.common.util;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+
+/** UTF-8 字节预算截断器；标记属于最终存储值的一部分，且永不切断 Unicode 字符。 */
+public final class Utf8TextTruncator {
+    private static final String MARKER_FORMAT = "...[HB_XDR_TRUNCATED original_utf8_bytes=%d]";
+
+    private Utf8TextTruncator() { }
+
+    public static Result truncate(String text, long maxUtf8Bytes) {
+        if (text == null) return new Result(null, false, 0L, 0L, null);
+        if (maxUtf8Bytes < 0L) throw new IllegalArgumentException("maxUtf8Bytes 不能小于 0");
+
+        byte[] original = text.getBytes(StandardCharsets.UTF_8);
+        long originalBytes = original.length;
+        if (maxUtf8Bytes == 0L || originalBytes <= maxUtf8Bytes) {
+            return new Result(text, false, originalBytes, originalBytes, null);
+        }
+
+        String marker = String.format(MARKER_FORMAT, originalBytes);
+        int markerBytes = marker.getBytes(StandardCharsets.UTF_8).length;
+        if (maxUtf8Bytes < markerBytes) {
+            throw new IllegalArgumentException("UTF-8 截断上限小于截断标记长度: limit="
+                    + maxUtf8Bytes + ", markerBytes=" + markerBytes);
+        }
+        long prefixBudget = maxUtf8Bytes - markerBytes;
+        int end = prefixEnd(text, prefixBudget);
+        String stored = text.substring(0, end) + marker;
+        return new Result(stored, true, originalBytes,
+                stored.getBytes(StandardCharsets.UTF_8).length, sha256(original));
+    }
+
+    private static int prefixEnd(String text, long byteBudget) {
+        long used = 0L;
+        int index = 0;
+        while (index < text.length()) {
+            int codePoint = text.codePointAt(index);
+            int width = utf8Width(codePoint);
+            if (used + width > byteBudget) break;
+            used += width;
+            index += Character.charCount(codePoint);
+        }
+        return index;
+    }
+
+    private static int utf8Width(int codePoint) {
+        if (codePoint <= 0x7f) return 1;
+        if (codePoint <= 0x7ff) return 2;
+        if (codePoint <= 0xffff) return 3;
+        return 4;
+    }
+
+    private static String sha256(byte[] bytes) {
+        try {
+            byte[] digest = MessageDigest.getInstance("SHA-256").digest(bytes);
+            StringBuilder hex = new StringBuilder(digest.length * 2);
+            for (byte item : digest) hex.append(String.format("%02x", item & 0xff));
+            return hex.toString();
+        } catch (Exception impossible) {
+            throw new IllegalStateException("JVM 不支持 SHA-256", impossible);
+        }
+    }
+
+    public static final class Result {
+        private final String value;
+        private final boolean truncated;
+        private final long originalUtf8Bytes;
+        private final long storedUtf8Bytes;
+        private final String originalSha256;
+
+        Result(String value, boolean truncated, long originalUtf8Bytes,
+               long storedUtf8Bytes, String originalSha256) {
+            this.value = value;
+            this.truncated = truncated;
+            this.originalUtf8Bytes = originalUtf8Bytes;
+            this.storedUtf8Bytes = storedUtf8Bytes;
+            this.originalSha256 = originalSha256;
+        }
+
+        public String getValue() { return value; }
+        public boolean isTruncated() { return truncated; }
+        public long getOriginalUtf8Bytes() { return originalUtf8Bytes; }
+        public long getStoredUtf8Bytes() { return storedUtf8Bytes; }
+        public String getOriginalSha256() { return originalSha256; }
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/KafkaReliabilityConfig.java

@@ -1,15 +1,20 @@
 package com.config;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.listener.DefaultErrorHandler;
 import org.springframework.util.backoff.FixedBackOff;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /** Kafka 消费失败边界：业务未确认的消息绝不能被框架默认恢复器静默跳过。 */
 @Configuration
 public class KafkaReliabilityConfig {
@@ -50,6 +55,32 @@ public class KafkaReliabilityConfig {
         return factory;
     }
 
+    /** 大日志使用独立 ConsumerFactory，限制单次 poll 条数并显式放开单条拉取上限。 */
+    @Bean("largeRawKafkaListenerContainerFactory")
+    public ConcurrentKafkaListenerContainerFactory<Object, Object> largeRawKafkaListenerContainerFactory(
+            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
+            ConsumerFactory<Object, Object> consumerFactory,
+            DefaultErrorHandler syslogDefaultErrorHandler,
+            @Value("${syslog.large-listener.concurrency:1}") int concurrency,
+            @Value("${syslog.large-listener.max-poll-records:5}") int maxPollRecords,
+            @Value("${syslog.large-listener.max-partition-fetch-bytes:33554432}") int maxPartitionFetchBytes,
+            @Value("${syslog.large-listener.fetch-max-bytes:67108864}") int fetchMaxBytes) {
+        if (concurrency < 1 || maxPollRecords < 1 || maxPartitionFetchBytes < 1
+                || fetchMaxBytes < maxPartitionFetchBytes) {
+            throw new IllegalArgumentException("大日志 Kafka 并发、poll 和 fetch 参数非法");
+        }
+        Map<String, Object> properties = new HashMap<>(consumerFactory.getConfigurationProperties());
+        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
+        properties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, maxPartitionFetchBytes);
+        properties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, fetchMaxBytes);
+        DefaultKafkaConsumerFactory<Object, Object> largeConsumerFactory =
+                new DefaultKafkaConsumerFactory<>(properties);
+        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
+                configuredFactory(configurer, largeConsumerFactory, syslogDefaultErrorHandler);
+        factory.setConcurrency(concurrency);
+        return factory;
+    }
+
     /** 统计脏键与原始日志隔离；同一维度 key 由 Kafka 分区保证只被一个组成员处理。 */
     @Bean("statsMaterializationKafkaListenerContainerFactory")
     public ConcurrentKafkaListenerContainerFactory<Object, Object> statsMaterializationKafkaListenerContainerFactory(
@@ -66,6 +97,21 @@ public class KafkaReliabilityConfig {
         return factory;
     }
 
+    /**
+     * Projection deltas intentionally use one consumer per instance. All events use one Kafka key,
+     * so only one group member owns the active partition and daily projection rows cannot deadlock.
+     */
+    @Bean("statsProjectionKafkaListenerContainerFactory")
+    public ConcurrentKafkaListenerContainerFactory<Object, Object> statsProjectionKafkaListenerContainerFactory(
+            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
+            ConsumerFactory<Object, Object> consumerFactory,
+            DefaultErrorHandler syslogDefaultErrorHandler) {
+        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
+                configuredFactory(configurer, consumerFactory, syslogDefaultErrorHandler);
+        factory.setConcurrency(1);
+        return factory;
+    }
+
     private ConcurrentKafkaListenerContainerFactory<Object, Object> configuredFactory(
             ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
             ConsumerFactory<Object, Object> consumerFactory,

haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/ThreadPoolConfig.java

@@ -25,6 +25,15 @@ public class ThreadPoolConfig {
     @Value("${app.processor.thread-pool.keep-alive-seconds:60}")
     private int keepAliveSeconds;
 
+    @Value("${syslog.large-processor.thread-pool.core-pool-size:1}")
+    private int largeCorePoolSize;
+
+    @Value("${syslog.large-processor.thread-pool.max-pool-size:2}")
+    private int largeMaxPoolSize;
+
+    @Value("${syslog.large-processor.thread-pool.queue-capacity:10}")
+    private int largeQueueCapacity;
+
     @Value("${dm.standardized-write.writer-concurrency:2}")
     private int dmWriterConcurrency;
 
@@ -54,6 +63,25 @@ public class ThreadPoolConfig {
         return executor;
     }
 
+    /** 大日志解析/规则计算独立线程池，避免占满普通日志的计算线程与任务队列。 */
+    @Bean("largeLogProcessorExecutor")
+    public Executor largeLogProcessorExecutor() {
+        if (largeCorePoolSize < 1 || largeMaxPoolSize < largeCorePoolSize || largeQueueCapacity < 1) {
+            throw new IllegalArgumentException("大日志处理线程池参数非法");
+        }
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(largeCorePoolSize);
+        executor.setMaxPoolSize(largeMaxPoolSize);
+        executor.setQueueCapacity(largeQueueCapacity);
+        executor.setKeepAliveSeconds(keepAliveSeconds);
+        executor.setThreadNamePrefix("large-log-processor-");
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
+        executor.setWaitForTasksToCompleteOnShutdown(true);
+        executor.setAwaitTerminationSeconds(60);
+        executor.initialize();
+        return executor;
+    }
+
     /**
      * 达梦批写使用独立的有界线程池，不能复用规则计算线程池。
      *

haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/DirectStatsController.java

@@ -7,6 +7,9 @@ import com.common.service.DmStandardizedBatchExecutor;
 import com.common.service.StatsMaterializationConsumer;
 import com.common.service.StatsMaterializationPublisher;
 import com.common.service.StatsMaterializationReconciler;
+import com.common.service.StatsProjectionOutboxWriter;
+import com.common.service.StatsProjectionOutboxRelay;
+import com.common.service.StatsProjectionDeltaConsumer;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -25,6 +28,9 @@ public class DirectStatsController {
     private final StatsMaterializationPublisher materializationPublisher;
     private final StatsMaterializationConsumer materializationConsumer;
     private final StatsMaterializationReconciler materializationReconciler;
+    private final StatsProjectionOutboxWriter projectionOutboxWriter;
+    private final StatsProjectionOutboxRelay projectionOutboxRelay;
+    private final StatsProjectionDeltaConsumer projectionDeltaConsumer;
 
     public DirectStatsController(DirectStatsProjectionWriter writer,
                                  IngressStatsProjectionWriter ingressWriter,
@@ -32,7 +38,10 @@ public class DirectStatsController {
                                  DmStandardizedBatchExecutor batchExecutor,
                                  StatsMaterializationPublisher materializationPublisher,
                                  StatsMaterializationConsumer materializationConsumer,
-                                 StatsMaterializationReconciler materializationReconciler) {
+                                 StatsMaterializationReconciler materializationReconciler,
+                                 StatsProjectionOutboxWriter projectionOutboxWriter,
+                                 StatsProjectionOutboxRelay projectionOutboxRelay,
+                                 StatsProjectionDeltaConsumer projectionDeltaConsumer) {
         this.writer = writer;
         this.ingressWriter = ingressWriter;
         this.dayRolloverService = dayRolloverService;
@@ -40,6 +49,9 @@ public class DirectStatsController {
         this.materializationPublisher = materializationPublisher;
         this.materializationConsumer = materializationConsumer;
         this.materializationReconciler = materializationReconciler;
+        this.projectionOutboxWriter = projectionOutboxWriter;
+        this.projectionOutboxRelay = projectionOutboxRelay;
+        this.projectionDeltaConsumer = projectionDeltaConsumer;
     }
 
     @GetMapping("/status")
@@ -51,6 +63,9 @@ public class DirectStatsController {
         status.put("materialization_publisher", materializationPublisher.status());
         status.put("materialization_consumer", materializationConsumer.status());
         status.put("materialization_reconciler", materializationReconciler.status());
+        status.put("projection_outbox_writer", projectionOutboxWriter.status());
+        status.put("projection_outbox_relay", projectionOutboxRelay.status());
+        status.put("projection_delta_consumer", projectionDeltaConsumer.status());
         return ResponseEntity.ok(status);
     }
 }

haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/StorageLimitMetricsController.java

@@ -0,0 +1,26 @@
+package com.controllers;
+
+import com.common.service.DmLargeTextLimiter;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/** 大字段截断观测，指标只包含计数，不暴露设备原文。 */
+@RestController
+@RequestMapping("/api/syslog/storage-limits")
+public class StorageLimitMetricsController {
+    private final DmLargeTextLimiter limiter;
+
+    public StorageLimitMetricsController(DmLargeTextLimiter limiter) { this.limiter = limiter; }
+
+    @GetMapping("/metrics")
+    public Map<String, Long> metrics() {
+        Map<String, Long> values = new LinkedHashMap<>();
+        values.put("dm_normal_fields_truncated_total", limiter.getNormalFieldsTruncated());
+        values.put("dm_non_normal_fields_truncated_total", limiter.getNonNormalFieldsTruncated());
+        return values;
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/InfluxRawPointFactory.java

@@ -1,15 +1,27 @@
 package com.influx.raw;
 
+import com.common.util.Utf8TextTruncator;
 import com.influxdb.client.domain.WritePrecision;
 import com.influxdb.client.write.Point;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 /** 原始 Point 的唯一构造器；V2 的 uuid 只作为 field。 */
 @Component
 public class InfluxRawPointFactory {
     private final RawStorageProperties properties;
+    private final RawInfluxMetrics metrics;
 
-    public InfluxRawPointFactory(RawStorageProperties properties) { this.properties = properties; }
+    @Autowired
+    public InfluxRawPointFactory(RawStorageProperties properties, RawInfluxMetrics metrics) {
+        this.properties = properties;
+        this.metrics = metrics;
+    }
+
+    /** 保留已有单元测试及非 Spring 工具入口。 */
+    public InfluxRawPointFactory(RawStorageProperties properties) {
+        this(properties, new RawInfluxMetrics());
+    }
 
     public RawPoint create(String eventId, String message, String receiveTime,
                            String deviceId, String deviceCollectId, String originalTopic,
@@ -17,21 +29,34 @@ public class InfluxRawPointFactory {
         require(eventId, "event_id"); require(receiveTime, "receive_time");
         require(deviceId, "device_id"); require(deviceCollectId, "device_collect_id");
         require(originalTopic, "originalTopic");
+        Utf8TextTruncator.Result storedMessage = Utf8TextTruncator.truncate(
+                message, properties.getMessageMaxUtf8Bytes());
+        if (storedMessage.isTruncated()) {
+            metrics.messageTruncated(storedMessage.getOriginalUtf8Bytes(), storedMessage.getStoredUtf8Bytes());
+        }
         String measurement = "syslog_security_" + parsed.getMeasurementDate();
         if (properties.getWriteSchema() == RawStorageProperties.WriteSchema.V1) {
             Point point = Point.measurement(measurement)
                     .addTag("device_id", deviceId).addTag("device_collect_id", deviceCollectId)
                     .addTag("uuid", eventId).addTag("topic", originalTopic)
-                    .addField("message", message).addField("receive_time", receiveTime)
+                    .addField("message", storedMessage.getValue()).addField("receive_time", receiveTime)
+                    .addField("message_truncated", storedMessage.isTruncated())
+                    .addField("message_original_bytes", storedMessage.getOriginalUtf8Bytes())
+                    .addField("message_stored_bytes", storedMessage.getStoredUtf8Bytes())
                     .addField("uuid", eventId).time(parsed.getEpochMillis(), WritePrecision.MS);
+            if (storedMessage.isTruncated()) point.addField("message_original_sha256", storedMessage.getOriginalSha256());
             return new RawPoint(point, properties.getV1Bucket(), false);
         }
         if (pointTimeNs == null) throw new IllegalArgumentException("V2 Point 缺少 point_time_ns");
         Point point = Point.measurement(measurement)
                 .addTag("device_id", deviceId).addTag("device_collect_id", deviceCollectId)
                 .addTag("topic", originalTopic).addTag("event_bucket", InfluxEventBucket.bucketOf(eventId))
-                .addField("uuid", eventId).addField("message", message).addField("receive_time", receiveTime)
+                .addField("uuid", eventId).addField("message", storedMessage.getValue()).addField("receive_time", receiveTime)
+                .addField("message_truncated", storedMessage.isTruncated())
+                .addField("message_original_bytes", storedMessage.getOriginalUtf8Bytes())
+                .addField("message_stored_bytes", storedMessage.getStoredUtf8Bytes())
                 .time(pointTimeNs, WritePrecision.NS);
+        if (storedMessage.isTruncated()) point.addField("message_original_sha256", storedMessage.getOriginalSha256());
         return new RawPoint(point, properties.getV2Bucket(), true);
     }
 

haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawInfluxMetrics.java

@@ -16,6 +16,9 @@ public class RawInfluxMetrics {
     private final LongAdder v2FailedBatches = new LongAdder();
     private final LongAdder dualReadDuplicate = new LongAdder();
     private final LongAdder dualReadConflict = new LongAdder();
+    private final LongAdder messagesTruncated = new LongAdder();
+    private final LongAdder originalBytesTruncated = new LongAdder();
+    private final LongAdder storedBytesAfterTruncation = new LongAdder();
 
     public void legacyMissing() { legacyMissing.increment(); }
     public void legacyAssigned() { legacyAssigned.increment(); }
@@ -24,6 +27,11 @@ public class RawInfluxMetrics {
     public void v2FailedBatch() { v2FailedBatches.increment(); }
     public void dualReadDuplicate() { dualReadDuplicate.increment(); }
     public void dualReadConflict() { dualReadConflict.increment(); }
+    public void messageTruncated(long originalBytes, long storedBytes) {
+        messagesTruncated.increment();
+        originalBytesTruncated.add(originalBytes);
+        storedBytesAfterTruncation.add(storedBytes);
+    }
 
     public Map<String, Object> snapshot() {
         Map<String, Object> values = new LinkedHashMap<>();
@@ -36,6 +44,9 @@ public class RawInfluxMetrics {
         values.put("raw_write_v2_failed_batches_total", v2FailedBatches.sum());
         values.put("influx_dual_read_duplicate_total", dualReadDuplicate.sum());
         values.put("influx_dual_read_conflict_total", dualReadConflict.sum());
+        values.put("raw_message_truncated_total", messagesTruncated.sum());
+        values.put("raw_message_truncated_original_bytes_total", originalBytesTruncated.sum());
+        values.put("raw_message_truncated_stored_bytes_total", storedBytesAfterTruncation.sum());
         return values;
     }
 }

haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawStorageProperties.java

@@ -19,6 +19,10 @@ public class RawStorageProperties {
     @Value("${influxdb.raw.legacy-point-time-fallback-enabled:true}") private boolean legacyFallbackEnabled;
     @Value("${influxdb.raw.query-max-rows:10000}") private int queryMaxRows;
     @Value("${influxdb.raw.query-max-response-bytes:10485760}") private long queryMaxResponseBytes;
+    /** 原始 message 单字段的 UTF-8 存储上限；0 表示不限制。 */
+    @Value("${influxdb.raw.message-max-utf8-bytes:0}") private long messageMaxUtf8Bytes;
+    /** 一次同步写请求的估算 Line Protocol 上限；按字节切批，不能只按条数。 */
+    @Value("${influxdb.raw.write-batch-max-bytes:8388608}") private long writeBatchMaxBytes;
 
     private WriteSchema writeSchema;
     private ReadMode readMode;
@@ -35,6 +39,12 @@ public class RawStorageProperties {
         if (queryMaxResponseBytes <= 0L) {
             throw new IllegalArgumentException("influxdb.raw.query-max-response-bytes 必须大于 0");
         }
+        if (messageMaxUtf8Bytes < 0L || (messageMaxUtf8Bytes > 0L && messageMaxUtf8Bytes < 64L)) {
+            throw new IllegalArgumentException("influxdb.raw.message-max-utf8-bytes 必须为 0 或至少 64");
+        }
+        if (writeBatchMaxBytes <= 0L) {
+            throw new IllegalArgumentException("influxdb.raw.write-batch-max-bytes 必须大于 0");
+        }
     }
 
     private static String normalize(String value) {
@@ -49,6 +59,8 @@ public class RawStorageProperties {
     public boolean isLegacyFallbackEnabled() { return legacyFallbackEnabled; }
     public int getQueryMaxRows() { return queryMaxRows; }
     public long getQueryMaxResponseBytes() { return queryMaxResponseBytes; }
+    public long getMessageMaxUtf8Bytes() { return messageMaxUtf8Bytes; }
+    public long getWriteBatchMaxBytes() { return writeBatchMaxBytes; }
 
     public enum WriteSchema { V1, V2 }
     public enum ReadMode { V1_ONLY, V1_V2, V2_ONLY }

haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties

@@ -34,6 +34,10 @@ influxdb.raw.uuid-bucket-count=${INFLUXDB_RAW_UUID_BUCKET_COUNT:4096}
 influxdb.raw.legacy-point-time-fallback-enabled=${INFLUXDB_RAW_LEGACY_POINT_TIME_FALLBACK_ENABLED:true}
 influxdb.raw.query-max-rows=${INFLUXDB_RAW_QUERY_MAX_ROWS:10000}
 influxdb.raw.query-max-response-bytes=${INFLUXDB_RAW_QUERY_MAX_RESPONSE_BYTES:10485760}
+# 0 表示不限制；启用后最终 message（含截断标记）不超过该 UTF-8 字节数。
+influxdb.raw.message-max-utf8-bytes=${INFLUXDB_RAW_MESSAGE_MAX_UTF8_BYTES:1048576}
+# Influx 同步写按 Line Protocol 估算字节切批，避免少量大日志形成超大 HTTP 请求。
+influxdb.raw.write-batch-max-bytes=${INFLUXDB_RAW_WRITE_BATCH_MAX_BYTES:8388608}
 # InfluxDB 2.7 连接超时配置
 influxdb.connection.timeout=30s
 influxdb.connection.read-timeout=30s
@@ -70,16 +74,31 @@ spring.kafka.consumer.auto-offset-reset=latest
 spring.kafka.consumer.enable-auto-commit=false
 spring.kafka.consumer.auto-commit-interval=1000
 spring.kafka.consumer.topic=test-topic
+spring.kafka.consumer.large-topic=${KAFKA_CONSUMER_LARGE_TOPIC:${spring.kafka.consumer.topic}.large}
 # consumer-rule 是当前活跃的 Kafka 原始日志标准化节点，默认启动监听。
 syslog.raw-consumer.enabled=true
 # Raw Topic 当前有 3 个分区，使用独立工厂显式创建 3 个 consumer；可通过环境变量灰度调整。
 syslog.raw-listener.concurrency=${SYSLOG_RAW_LISTENER_CONCURRENCY:3}
+# 大日志使用独立 Topic、消费组和 ConsumerFactory；建议生产环境独立 JVM 部署。
+syslog.large-listener.enabled=${SYSLOG_LARGE_LISTENER_ENABLED:true}
+syslog.large-listener.group-id=${SYSLOG_LARGE_LISTENER_GROUP_ID:${spring.kafka.consumer.group-id}-large}
+syslog.large-listener.concurrency=${SYSLOG_LARGE_LISTENER_CONCURRENCY:1}
+syslog.large-listener.max-poll-records=${SYSLOG_LARGE_LISTENER_MAX_POLL_RECORDS:5}
+syslog.large-listener.max-partition-fetch-bytes=${SYSLOG_LARGE_LISTENER_MAX_PARTITION_FETCH_BYTES:33554432}
+syslog.large-listener.fetch-max-bytes=${SYSLOG_LARGE_LISTENER_FETCH_MAX_BYTES:67108864}
+syslog.large-listener.processing-batch-max-bytes=${SYSLOG_LARGE_PROCESSING_BATCH_MAX_BYTES:8388608}
+# 大日志规则计算不复用普通日志线程池，过载时由 Kafka offset 背压。
+syslog.large-processor.thread-pool.core-pool-size=${SYSLOG_LARGE_PROCESSOR_CORE_POOL_SIZE:1}
+syslog.large-processor.thread-pool.max-pool-size=${SYSLOG_LARGE_PROCESSOR_MAX_POOL_SIZE:2}
+syslog.large-processor.thread-pool.queue-capacity=${SYSLOG_LARGE_PROCESSOR_QUEUE_CAPACITY:10}
 # P0：Retry/DLT 发布与原始消费使用同一 Kafka 集群；Topic 名称全部可由环境变量覆盖。
 spring.kafka.bootstrap-servers=${spring.kafka.consumer.bootstrap-servers}
 spring.kafka.producer.acks=all
 spring.kafka.producer.properties.enable.idempotence=true
 spring.kafka.producer.properties.delivery.timeout.ms=120000
 spring.kafka.producer.properties.request.timeout.ms=30000
+# Retry/DLT 信封也可能携带大日志密文，不能保留 Kafka 默认 1MiB。
+spring.kafka.producer.properties.max.request.size=${KAFKA_RELIABILITY_MAX_REQUEST_SIZE:33554432}
 syslog.reliability.retry-1m-topic=${spring.kafka.consumer.topic}.RETRY.1M
 syslog.reliability.retry-10m-topic=${spring.kafka.consumer.topic}.RETRY.10M
 syslog.reliability.retry-1h-topic=${spring.kafka.consumer.topic}.RETRY.1H
@@ -91,6 +110,14 @@ syslog.reliability.consumer-version=${APP_VERSION:1.0.0}
 syslog.reliability.consumer-group-id=${spring.kafka.consumer.group-id}-retry
 syslog.reliability.consumer.enabled=true
 syslog.reliability.listener.concurrency=1
+# 大日志 Retry/DLT 不回流普通重试通道，避免再次形成队头阻塞。
+syslog.large-reliability.retry-1m-topic=${spring.kafka.consumer.large-topic}.RETRY.1M
+syslog.large-reliability.retry-10m-topic=${spring.kafka.consumer.large-topic}.RETRY.10M
+syslog.large-reliability.retry-1h-topic=${spring.kafka.consumer.large-topic}.RETRY.1H
+syslog.large-reliability.dlt-topic=${spring.kafka.consumer.large-topic}.DLT
+syslog.large-reliability.replay-topic=${spring.kafka.consumer.large-topic}.REPLAY
+syslog.large-reliability.consumer-group-id=${spring.kafka.consumer.group-id}-large-retry
+syslog.large-reliability.consumer.enabled=true
 
 
 spring.kafka.consumer.max-poll-records=500
@@ -152,6 +179,10 @@ dm.standardized-write.mode=BATCH
 dm.standardized-write.batch-size=${DM_STANDARDIZED_WRITE_BATCH_SIZE:500}
 dm.standardized-write.writer-concurrency=2
 dm.standardized-write.writer-queue-capacity=100
+# 标准化大字段和非标原文均在达梦持久化边界截断；0 表示不限制。
+dm.normal.large-field-max-utf8-bytes=${DM_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES:65536}
+dm.normal.large-fields=${DM_NORMAL_LARGE_FIELDS:http_req_body,http_resp_body,http_req_header,http_resp_header,detail,description,cmdline,shell_cmdline,parent_cmdline,mail_message}
+dm.non-normal.large-field-max-utf8-bytes=${DM_NON_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES:262144}
 dm.standardized-write.transaction-timeout-seconds=45
 dm.standardized-write.kafka-batch-deadline-ms=90000
 dm.standardized-write.max-split-depth=10
@@ -181,7 +212,7 @@ spring.datasource.hikari.validation-timeout=5000
 spring.datasource.hikari.leak-detection-threshold=30000
 spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer-rule
 spring.datasource.hikari.auto-commit=false
-# 统计写入采用事实同事务直写，不再为每条结果写 change_log + pending。
+# DIRECT 不再写逐事实 change_log；默认只在事实事务追加一条批次 Outbox，日统计异步投影。
 # 首次部署 DIRECT 前必须先执行 08/09/10/11 脚本并按设计文档排空旧 pending。
 stats.write-mode=DIRECT
 stats.incremental.capture-enabled=false
@@ -197,11 +228,11 @@ stats.direct.control-table=stats_direct_control
 stats.direct.today-stats-date-enabled=true
 stats.direct.max-concurrency-retries=${STATS_DIRECT_MAX_CONCURRENCY_RETRIES:3}
 stats.direct.concurrency-retry-backoff-ms=${STATS_DIRECT_CONCURRENCY_RETRY_BACKOFF_MS:50}
-stats.direct.transaction-log-enabled=false
+stats.direct.transaction-log-enabled=${STATS_DIRECT_TRANSACTION_LOG_ENABLED:false}
 stats.direct.day-rollover-check-cron=5 * * * * ?
 stats.business-zone=Asia/Shanghai
 stats.fact-only.confirm-rebuild-required=false
-# 事实事务只维护小型精确统计表；提交后按维度脏键异步刷新业务展示字段。
+# 日统计投影完成后按维度脏键异步刷新业务展示字段。
 # 三个实例必须配置相同 topic/group-id，Topic 分区数建议与原始 Topic 相同。
 stats.materialization.enabled=${STATS_MATERIALIZATION_ENABLED:true}
 stats.materialization.topic=${STATS_MATERIALIZATION_TOPIC:${spring.kafka.consumer.topic}.STATS.MATERIALIZE}
@@ -216,6 +247,25 @@ stats.materialization.concurrency-retry-backoff-ms=${STATS_MATERIALIZATION_CONCU
 stats.materialization.reconcile-initial-delay-ms=${STATS_MATERIALIZATION_RECONCILE_INITIAL_DELAY_MS:30000}
 stats.materialization.reconcile-interval-ms=${STATS_MATERIALIZATION_RECONCILE_INTERVAL_MS:60000}
 
+# Fact transactions append one aggregated durable event instead of locking the daily projection rows.
+# A dedicated single-lane Kafka consumer applies device/rule/collector projection deltas and then
+# emits the existing idempotent business-materialization dirty keys.
+stats.projection-async.enabled=${STATS_PROJECTION_ASYNC_ENABLED:true}
+stats.projection-async.outbox-table=${STATS_PROJECTION_OUTBOX_TABLE:stats_projection_outbox}
+stats.projection-async.applied-table=${STATS_PROJECTION_APPLIED_TABLE:stats_projection_applied_event}
+stats.projection-async.topic=${STATS_PROJECTION_TOPIC:${spring.kafka.consumer.topic}.STATS.PROJECTION}
+stats.projection-async.group-id=${STATS_PROJECTION_GROUP_ID:${spring.kafka.consumer.group-id}-stats-projection}
+stats.projection-async.relay-interval-ms=${STATS_PROJECTION_RELAY_INTERVAL_MS:200}
+stats.projection-async.relay-batch-size=${STATS_PROJECTION_RELAY_BATCH_SIZE:100}
+stats.projection-async.publish-timeout-ms=${STATS_PROJECTION_PUBLISH_TIMEOUT_MS:10000}
+stats.projection-async.retry-backoff-ms=${STATS_PROJECTION_RETRY_BACKOFF_MS:1000}
+stats.projection-async.redelivery-interval-ms=${STATS_PROJECTION_REDELIVERY_INTERVAL_MS:60000}
+stats.projection-async.max-concurrency-retries=${STATS_PROJECTION_MAX_CONCURRENCY_RETRIES:3}
+stats.projection-async.concurrency-retry-backoff-ms=${STATS_PROJECTION_CONCURRENCY_RETRY_BACKOFF_MS:50}
+stats.projection-async.retention-days=${STATS_PROJECTION_RETENTION_DAYS:14}
+stats.projection-async.cleanup-initial-delay-ms=${STATS_PROJECTION_CLEANUP_INITIAL_DELAY_MS:3600000}
+stats.projection-async.cleanup-interval-ms=${STATS_PROJECTION_CLEANUP_INTERVAL_MS:3600000}
+
 # 标准化规则编译与批次缓存灰度开关
 normalization.jetcache.enabled=true
 normalization.rule-source-cache.enabled=true

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SyslogFailureRouterReliabilityTest.java

@@ -42,6 +42,11 @@ class SyslogFailureRouterReliabilityTest {
         ReflectionTestUtils.setField(router, "retry10mTopic", "raw.RETRY.10M");
         ReflectionTestUtils.setField(router, "retry1hTopic", "raw.RETRY.1H");
         ReflectionTestUtils.setField(router, "dltTopic", "raw.DLT");
+        ReflectionTestUtils.setField(router, "largeOriginalTopic", "raw.large");
+        ReflectionTestUtils.setField(router, "largeRetry1mTopic", "raw.large.RETRY.1M");
+        ReflectionTestUtils.setField(router, "largeRetry10mTopic", "raw.large.RETRY.10M");
+        ReflectionTestUtils.setField(router, "largeRetry1hTopic", "raw.large.RETRY.1H");
+        ReflectionTestUtils.setField(router, "largeDltTopic", "raw.large.DLT");
         ReflectionTestUtils.setField(router, "publishTimeoutMs", 1000L);
         ReflectionTestUtils.setField(router, "consumerVersion", "test");
         record = new ConsumerRecord<>("raw", 2, 42L, "device-1", "ciphertext");
@@ -86,6 +91,22 @@ class SyslogFailureRouterReliabilityTest {
         assertFalse(body.getValue().contains("schema_version"));
     }
 
+    @Test
+    void largeMessageFailureNeverFallsBackIntoNormalRetryTopic() {
+        ConsumerRecord<String, String> largeRecord = new ConsumerRecord<>(
+                "raw.large", 1, 7L, "event", "large-ciphertext");
+        SettableListenableFuture<SendResult<String, String>> acknowledged = new SettableListenableFuture<>();
+        acknowledged.set(null);
+        when(template.send(eq("raw.large.RETRY.1M"), anyString(), anyString()))
+                .thenReturn(acknowledged);
+
+        SyslogFailureRouter.TerminalState state = router.route(
+                largeRecord, router.resolve(largeRecord), new SocketException("reset"));
+
+        assertEquals(SyslogFailureRouter.TerminalState.RETRY_PUBLISHED, state);
+        verify(template).send(eq("raw.large.RETRY.1M"), anyString(), anyString());
+    }
+
     @Test
     void publishFailureNeverBecomesTerminal() {
         SettableListenableFuture<SendResult<String, String>> failed = new SettableListenableFuture<>();

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SyslogKafkaTopologyTest.java

@@ -4,12 +4,17 @@ import org.junit.jupiter.api.Test;
 import org.springframework.kafka.annotation.KafkaListener;
 
 import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 import java.util.List;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import com.common.service.StatsMaterializationConsumer;
+import com.common.service.StatsProjectionDeltaConsumer;
 
 /** 防止多节点 consumer-rule 的监听器 ID 或显式消费组在后续合并中回退。 */
 class SyslogKafkaTopologyTest {
@@ -50,6 +55,56 @@ class SyslogKafkaTopologyTest {
         }, listener.topics());
     }
 
+    @Test
+    void largeRawListenerUsesDedicatedTopicGroupAndFactory() throws Exception {
+        Method method = SysLogProcessor.class.getMethod(
+                "consumeLargeMessages", List.class,
+                org.springframework.kafka.support.Acknowledgment.class);
+        KafkaListener listener = method.getAnnotation(KafkaListener.class);
+
+        assertNotNull(listener);
+        assertEquals("syslog-consumer-rule-raw-large", listener.id());
+        assertEquals("${syslog.large-listener.group-id:${spring.kafka.consumer.group-id}-large}",
+                listener.groupId());
+        assertEquals("largeRawKafkaListenerContainerFactory", listener.containerFactory());
+        assertArrayEquals(new String[]{"${spring.kafka.consumer.large-topic:agent-syslog-topic.large}"},
+                listener.topics());
+    }
+
+    @Test
+    void largeRetryReplayListenerIsAlsoIsolated() throws Exception {
+        Method method = SysLogProcessor.class.getMethod(
+                "consumeLargeRetryAndReplay", List.class,
+                org.springframework.kafka.support.Acknowledgment.class);
+        KafkaListener listener = method.getAnnotation(KafkaListener.class);
+
+        assertNotNull(listener);
+        assertEquals("syslog-consumer-rule-large-retry-replay", listener.id());
+        assertEquals("largeRawKafkaListenerContainerFactory", listener.containerFactory());
+        assertArrayEquals(new String[]{
+                "${syslog.large-reliability.retry-1m-topic:${spring.kafka.consumer.large-topic}.RETRY.1M}",
+                "${syslog.large-reliability.retry-10m-topic:${spring.kafka.consumer.large-topic}.RETRY.10M}",
+                "${syslog.large-reliability.retry-1h-topic:${spring.kafka.consumer.large-topic}.RETRY.1H}",
+                "${syslog.large-reliability.replay-topic:${spring.kafka.consumer.large-topic}.REPLAY}"
+        }, listener.topics());
+    }
+
+    @Test
+    void largePollIsPartitionedByUtf8BytesInsteadOfOnlyRecordCount() {
+        List<ConsumerRecord<String, String>> records = Arrays.asList(
+                new ConsumerRecord<>("large", 0, 0L, "k1", "中文"),
+                new ConsumerRecord<>("large", 0, 1L, "k2", "12"),
+                new ConsumerRecord<>("large", 0, 2L, "k3", "🙂"));
+
+        List<List<ConsumerRecord<String, String>>> chunks =
+                SysLogProcessor.partitionRecordsByBytes(records, 7L);
+
+        assertEquals(2, chunks.size());
+        assertEquals(1, chunks.get(0).size());
+        assertEquals(2, chunks.get(1).size());
+        assertEquals(6, chunks.get(0).get(0).value().getBytes(StandardCharsets.UTF_8).length);
+    }
+
     @Test
     void statsMaterializerUsesDedicatedTopicAndGroup() throws Exception {
         Method method = StatsMaterializationConsumer.class.getMethod(
@@ -65,4 +120,20 @@ class SyslogKafkaTopologyTest {
                 "${stats.materialization.topic:${spring.kafka.consumer.topic}.STATS.MATERIALIZE}"
         }, listener.topics());
     }
+
+    @Test
+    void statsProjectionUsesSeparateSingleLaneTopicAndGroup() throws Exception {
+        Method method = StatsProjectionDeltaConsumer.class.getMethod(
+                "consume", List.class, org.springframework.kafka.support.Acknowledgment.class);
+        KafkaListener listener = method.getAnnotation(KafkaListener.class);
+
+        assertNotNull(listener);
+        assertEquals("syslog-stats-projection", listener.id());
+        assertEquals("${stats.projection-async.group-id:${spring.kafka.consumer.group-id}-stats-projection}",
+                listener.groupId());
+        assertEquals("statsProjectionKafkaListenerContainerFactory", listener.containerFactory());
+        assertArrayEquals(new String[]{
+                "${stats.projection-async.topic:${spring.kafka.consumer.topic}.STATS.PROJECTION}"
+        }, listener.topics());
+    }
 }

haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DirectStatsProjectionWriterTest.java

@@ -119,12 +119,14 @@ class DirectStatsProjectionWriterTest {
     }
 
     @Test
-    void kafkaMaterializationRemovesBusinessTableUpdatesFromFactTransaction() {
+    void durableOutboxRemovesDailyAndBusinessProjectionUpdatesFromFactTransaction() {
         JdbcTemplate jdbc = mock(JdbcTemplate.class);
         StatsMaterializationPublisher publisher = mock(StatsMaterializationPublisher.class);
         when(publisher.isEnabled()).thenReturn(true);
+        StatsProjectionOutboxWriter outbox = mock(StatsProjectionOutboxWriter.class);
+        when(outbox.isEnabled()).thenReturn(true);
         DirectStatsProjectionWriter writer = new DirectStatsProjectionWriter(
-                jdbc, new StatsWriteModeProvider("DIRECT"), new DirectStatsWriteMetrics(), publisher);
+                jdbc, new StatsWriteModeProvider("DIRECT"), new DirectStatsWriteMetrics(), publisher, outbox);
         configure(writer, true);
         TransactionSynchronizationManager.setActualTransactionActive(true);
 
@@ -135,20 +137,19 @@ class DirectStatsProjectionWriterTest {
         row.put("normalize_rule_id", 11L);
         writer.writeStandardizedRow(row);
 
-        ArgumentCaptor<String> sql = ArgumentCaptor.forClass(String.class);
-        verify(jdbc, times(2)).batchUpdate(sql.capture(), anyList());
-        assertTrue(sql.getAllValues().get(0).contains("MERGE INTO device_daily_stats"));
-        assertTrue(sql.getAllValues().get(1).contains("MERGE INTO normalize_rule_daily_stats"));
-        verify(publisher).enqueueAfterCommit(anyCollection());
-        assertTrue("KAFKA_DIRTY_KEY".equals(writer.status().get("business_materialization_mode")));
+        verify(outbox).append(org.mockito.ArgumentMatchers.any(StatsProjectionOutboxEvent.class));
+        verifyNoInteractions(jdbc);
+        assertTrue("DURABLE_OUTBOX".equals(writer.status().get("daily_projection_write_mode")));
     }
 
     private DirectStatsProjectionWriter writer(JdbcTemplate jdbc, String mode, boolean readEnabled) {
         StatsWriteModeProvider modeProvider = new StatsWriteModeProvider(mode);
         StatsMaterializationPublisher publisher = mock(StatsMaterializationPublisher.class);
         when(publisher.isEnabled()).thenReturn(false);
+        StatsProjectionOutboxWriter outbox = mock(StatsProjectionOutboxWriter.class);
+        when(outbox.isEnabled()).thenReturn(false);
         DirectStatsProjectionWriter writer = new DirectStatsProjectionWriter(
-                jdbc, modeProvider, new DirectStatsWriteMetrics(), publisher);
+                jdbc, modeProvider, new DirectStatsWriteMetrics(), publisher, outbox);
         configure(writer, readEnabled);
         return writer;
     }

haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmLargeTextLimiterTest.java

@@ -0,0 +1,62 @@
+package com.common.service;
+
+import com.common.entity.SyslogNonNormalMessage;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class DmLargeTextLimiterTest {
+
+    @Test
+    void normalConfiguredFieldsAreTruncatedButOtherFieldsRemainUntouched() {
+        DmLargeTextLimiter limiter = limiter(96L, "http_req_body,description", 0L);
+        Map<String, Object> row = new HashMap<>();
+        row.put("http_req_body", repeat("中文🙂", 100));
+        row.put("other", repeat("x", 200));
+
+        limiter.truncateNormalRow(row);
+
+        String body = (String) row.get("http_req_body");
+        assertTrue(body.contains("HB_XDR_TRUNCATED"));
+        assertTrue(body.getBytes(StandardCharsets.UTF_8).length <= 96);
+        assertEquals(repeat("x", 200), row.get("other"));
+        assertEquals(1L, limiter.getNormalFieldsTruncated());
+    }
+
+    @Test
+    void allNonNormalLargeColumnsUseTheSameVisibleMarker() {
+        DmLargeTextLimiter limiter = limiter(0L, "", 96L);
+        SyslogNonNormalMessage message = new SyslogNonNormalMessage();
+        message.setSyslogMessage(repeat("原文", 100));
+        message.setHeaderMessage(repeat("头", 100));
+        message.setReasonDetail(repeat("原因", 100));
+
+        limiter.truncateNonNormal(message);
+
+        assertTrue(message.getSyslogMessage().contains("HB_XDR_TRUNCATED"));
+        assertTrue(message.getHeaderMessage().contains("HB_XDR_TRUNCATED"));
+        assertTrue(message.getReasonDetail().contains("HB_XDR_TRUNCATED"));
+        assertEquals(3L, limiter.getNonNormalFieldsTruncated());
+    }
+
+    private DmLargeTextLimiter limiter(long normal, String fields, long nonNormal) {
+        DmLargeTextLimiter limiter = new DmLargeTextLimiter();
+        ReflectionTestUtils.setField(limiter, "normalMaxBytes", normal);
+        ReflectionTestUtils.setField(limiter, "normalFieldsValue", fields);
+        ReflectionTestUtils.setField(limiter, "nonNormalMaxBytes", nonNormal);
+        limiter.validate();
+        return limiter;
+    }
+
+    private static String repeat(String value, int count) {
+        StringBuilder result = new StringBuilder(value.length() * count);
+        for (int i = 0; i < count; i++) result.append(value);
+        return result.toString();
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/IngressStatsProjectionWriterTest.java

@@ -2,6 +2,11 @@ package com.common.service;
 
 import com.common.entity.IngressStatsEvent;
 import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.SimpleTransactionStatus;
+import org.springframework.jdbc.core.JdbcTemplate;
 
 import java.time.LocalDateTime;
 import java.util.Arrays;
@@ -9,6 +14,11 @@ import java.util.ArrayList;
 import java.util.List;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
 
 class IngressStatsProjectionWriterTest {
     @Test
@@ -41,4 +51,28 @@ class IngressStatsProjectionWriterTest {
         assertEquals(Arrays.asList(8L, 2L, 9L), devices);
         assertEquals(Arrays.asList(3L, 4L, 7L), new ArrayList<>(result.collects.keySet()));
     }
+
+    @Test
+    void directAsyncModeWritesOnlyOneOutboxEvent() {
+        JdbcTemplate jdbc = mock(JdbcTemplate.class);
+        PlatformTransactionManager manager = mock(PlatformTransactionManager.class);
+        when(manager.getTransaction(any(TransactionDefinition.class)))
+                .thenReturn(new SimpleTransactionStatus());
+        StatsProjectionOutboxWriter outbox = mock(StatsProjectionOutboxWriter.class);
+        when(outbox.isEnabled()).thenReturn(true);
+        IngressStatsProjectionWriter writer = new IngressStatsProjectionWriter(
+                jdbc, manager, new StatsWriteModeProvider("DIRECT"),
+                mock(StatsMaterializationPublisher.class), outbox);
+        ReflectionTestUtils.setField(writer, "enabled", true);
+        ReflectionTestUtils.setField(writer, "transactionTimeoutSeconds", 45);
+
+        int groups = writer.writeBatch(Arrays.asList(
+                new IngressStatsEvent("a", 7L, 3L, LocalDateTime.of(2026, 9, 12, 8, 0)),
+                new IngressStatsEvent("b", 7L, 3L, LocalDateTime.of(2026, 9, 12, 8, 1))));
+
+        assertEquals(2, groups);
+        verify(outbox).append(org.mockito.ArgumentMatchers.any(StatsProjectionOutboxEvent.class));
+        verifyNoInteractions(jdbc);
+        verify(manager).commit(any());
+    }
 }

haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsProjectionAsyncTest.java

@@ -0,0 +1,151 @@
+package com.common.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.SimpleTransactionStatus;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class StatsProjectionAsyncTest {
+    @AfterEach
+    void clearTransactionState() {
+        TransactionSynchronizationManager.clear();
+    }
+
+    @Test
+    void outboxEventAggregatesOneFactBatchByDimension() {
+        LocalDateTime time = LocalDateTime.of(2026, 9, 12, 8, 0);
+        List<Map<String, Object>> rows = new ArrayList<>();
+        for (long ruleId : Arrays.asList(11L, 12L, 11L)) {
+            Map<String, Object> row = new HashMap<>();
+            row.put("created_at", time);
+            row.put("log_time", time.plusMinutes(ruleId));
+            row.put("device_id", 7L);
+            row.put("normalize_rule_id", ruleId);
+            rows.add(row);
+        }
+
+        StatsProjectionOutboxEvent event = StatsProjectionOutboxEvent.fromAccumulator(
+                new DirectStatsAccumulator().addStandardizedRows(rows));
+
+        event.validate();
+        assertEquals(1, event.getDevices().size());
+        assertEquals(3L, event.getDevices().get(0).getParseDelta());
+        assertEquals(2, event.getRules().size());
+    }
+
+    @Test
+    void outboxWriterRequiresAndUsesFactTransaction() {
+        JdbcTemplate jdbc = mock(JdbcTemplate.class);
+        StatsProjectionOutboxWriter writer = new StatsProjectionOutboxWriter(jdbc, new ObjectMapper());
+        ReflectionTestUtils.setField(writer, "enabled", true);
+        ReflectionTestUtils.setField(writer, "outboxTable", "stats_projection_outbox");
+        StatsProjectionOutboxEvent event = oneDeviceEvent("event-1");
+
+        assertThrows(IllegalStateException.class, () -> writer.append(event));
+
+        TransactionSynchronizationManager.setActualTransactionActive(true);
+        writer.append(event);
+        verify(jdbc).update(org.mockito.ArgumentMatchers.contains("INSERT INTO stats_projection_outbox"),
+                org.mockito.ArgumentMatchers.eq("event-1"),
+                org.mockito.ArgumentMatchers.eq(StatsProjectionOutboxEvent.SCHEMA),
+                org.mockito.ArgumentMatchers.anyString());
+    }
+
+    @Test
+    void projectionConsumerAppliesFreshEventAndRecordsIdempotencyMarker() {
+        JdbcTemplate jdbc = mock(JdbcTemplate.class);
+        when(jdbc.queryForList(org.mockito.ArgumentMatchers.anyString(),
+                org.mockito.ArgumentMatchers.eq(String.class), org.mockito.ArgumentMatchers.<Object>any()))
+                .thenReturn(Collections.emptyList());
+        PlatformTransactionManager manager = mock(PlatformTransactionManager.class);
+        when(manager.getTransaction(any(TransactionDefinition.class))).thenReturn(new SimpleTransactionStatus());
+        StatsMaterializationPublisher publisher = mock(StatsMaterializationPublisher.class);
+        StatsProjectionDeltaConsumer consumer = new StatsProjectionDeltaConsumer(
+                jdbc, manager, new ObjectMapper(), publisher);
+        configure(consumer);
+
+        StatsProjectionDeltaConsumer.ApplyResult result = consumer.applyWithRetry(
+                Collections.singletonList(oneDeviceEvent("event-1")));
+
+        assertEquals(1, result.appliedEvents);
+        assertEquals(1, result.deviceGroups);
+        ArgumentCaptor<String> sql = ArgumentCaptor.forClass(String.class);
+        verify(jdbc, times(3)).batchUpdate(sql.capture(), anyList());
+        assertTrue(sql.getAllValues().get(0).contains("MERGE INTO device_daily_stats"));
+        assertTrue(sql.getAllValues().get(1).contains("stats_projection_applied_event"));
+        assertTrue(sql.getAllValues().get(2).contains("SET status='APPLIED'"));
+        verify(publisher).enqueueAfterCommit(org.mockito.ArgumentMatchers.anyCollection());
+        verify(manager).commit(any());
+    }
+
+    @Test
+    void projectionConsumerDoesNotApplyDuplicateDeltaTwice() {
+        JdbcTemplate jdbc = mock(JdbcTemplate.class);
+        when(jdbc.queryForList(org.mockito.ArgumentMatchers.anyString(),
+                org.mockito.ArgumentMatchers.eq(String.class), org.mockito.ArgumentMatchers.<Object>any()))
+                .thenReturn(Collections.singletonList("event-1"));
+        PlatformTransactionManager manager = mock(PlatformTransactionManager.class);
+        when(manager.getTransaction(any(TransactionDefinition.class))).thenReturn(new SimpleTransactionStatus());
+        StatsMaterializationPublisher publisher = mock(StatsMaterializationPublisher.class);
+        StatsProjectionDeltaConsumer consumer = new StatsProjectionDeltaConsumer(
+                jdbc, manager, new ObjectMapper(), publisher);
+        configure(consumer);
+
+        StatsProjectionDeltaConsumer.ApplyResult result = consumer.applyWithRetry(
+                Collections.singletonList(oneDeviceEvent("event-1")));
+
+        assertEquals(0, result.appliedEvents);
+        assertEquals(1, result.duplicates);
+        ArgumentCaptor<String> sql = ArgumentCaptor.forClass(String.class);
+        verify(jdbc).batchUpdate(sql.capture(), anyList());
+        assertTrue(sql.getValue().contains("SET status='APPLIED'"));
+    }
+
+    private StatsProjectionOutboxEvent oneDeviceEvent(String eventId) {
+        DirectStatsAccumulator accumulator = new DirectStatsAccumulator();
+        Map<String, Object> row = new HashMap<>();
+        row.put("created_at", LocalDate.of(2026, 9, 12));
+        row.put("log_time", LocalDateTime.of(2026, 9, 12, 8, 0));
+        row.put("device_id", 7L);
+        StatsProjectionOutboxEvent event = StatsProjectionOutboxEvent.fromAccumulator(
+                accumulator.addStandardizedRow(row));
+        event.setEventId(eventId);
+        return event;
+    }
+
+    private void configure(StatsProjectionDeltaConsumer consumer) {
+        ReflectionTestUtils.setField(consumer, "enabled", true);
+        ReflectionTestUtils.setField(consumer, "appliedTable", "stats_projection_applied_event");
+        ReflectionTestUtils.setField(consumer, "outboxTable", "stats_projection_outbox");
+        ReflectionTestUtils.setField(consumer, "deviceTable", "device_daily_stats");
+        ReflectionTestUtils.setField(consumer, "ruleTable", "normalize_rule_daily_stats");
+        ReflectionTestUtils.setField(consumer, "transactionTimeoutSeconds", 45);
+        ReflectionTestUtils.setField(consumer, "maxConcurrencyRetries", 3);
+        ReflectionTestUtils.setField(consumer, "retryBackoffMs", 0L);
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/util/Utf8TextTruncatorTest.java

@@ -0,0 +1,43 @@
+package com.common.util;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class Utf8TextTruncatorTest {
+    @Test
+    void truncatesWithinBudgetWithoutSplittingChineseOrEmoji() {
+        String source = repeat("中文🙂", 100);
+
+        Utf8TextTruncator.Result result = Utf8TextTruncator.truncate(source, 96L);
+
+        assertTrue(result.isTruncated());
+        assertTrue(result.getValue().endsWith("]"));
+        assertTrue(result.getValue().contains("HB_XDR_TRUNCATED"));
+        assertTrue(result.getValue().getBytes(StandardCharsets.UTF_8).length <= 96);
+        assertEquals(source.getBytes(StandardCharsets.UTF_8).length, result.getOriginalUtf8Bytes());
+        assertNotNull(result.getOriginalSha256());
+        assertFalse(result.getValue().contains("�"));
+    }
+
+    @Test
+    void zeroMeansUnlimitedAndSmallEnabledLimitFailsClearly() {
+        Utf8TextTruncator.Result result = Utf8TextTruncator.truncate("中文🙂", 0L);
+        assertFalse(result.isTruncated());
+        assertEquals("中文🙂", result.getValue());
+        assertThrows(IllegalArgumentException.class,
+                () -> Utf8TextTruncator.truncate(repeat("x", 100), 8L));
+    }
+
+    private static String repeat(String value, int count) {
+        StringBuilder result = new StringBuilder(value.length() * count);
+        for (int i = 0; i < count; i++) result.append(value);
+        return result.toString();
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/test/java/com/influx/raw/InfluxRawV2ContractTest.java

@@ -67,6 +67,25 @@ class InfluxRawV2ContractTest {
         String line=raw.getPoint().toLineProtocol(); assertFalse(line.substring(0,line.indexOf(' ')).contains("uuid=")); assertTrue(line.contains("event_bucket=2085")); assertTrue(line.contains("uuid=\"20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5\"")); assertEquals(WritePrecision.NS,raw.getPoint().getPrecision()); assertEquals("yelangbucket_v2",raw.getBucket());
     }
 
+    @Test void oversizedRawMessageIsMarkedAndCarriesStructuredTruncationMetadata() {
+        RawStorageProperties properties=mock(RawStorageProperties.class);
+        when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2);
+        when(properties.getV2Bucket()).thenReturn("yelangbucket_v2");
+        when(properties.getMessageMaxUtf8Bytes()).thenReturn(128L);
+        RawInfluxMetrics metrics=new RawInfluxMetrics();
+        InfluxRawPointFactory factory=new InfluxRawPointFactory(properties,metrics);
+
+        String line=factory.create("20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5",
+                repeat("中文🙂",100),"20260814000000000","126","1","raw.large",
+                ReceiveTimeParser.parse("20260814000000000"),1_786_636_800_000_000_123L)
+                .getPoint().toLineProtocol();
+
+        assertTrue(line.contains("HB_XDR_TRUNCATED"));
+        assertTrue(line.contains("message_truncated=true"));
+        assertTrue(line.contains("message_original_sha256="));
+        assertEquals(1L,metrics.snapshot().get("raw_message_truncated_total"));
+    }
+
     @Test void onlyMissingPointTimeFallsBack() {
         RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.isLegacyFallbackEnabled()).thenReturn(true); RawInfluxMetrics metrics=new RawInfluxMetrics(); PointTimeResolver resolver=new PointTimeResolver(properties,metrics); long millis=ReceiveTimeParser.parse("20260814000000000").getEpochMillis();
         assertNotNull(resolver.resolve(false,null,millis,true)); assertThrows(IllegalArgumentException.class,()->resolver.resolve(true,"-1",millis,true)); assertThrows(IllegalArgumentException.class,()->resolver.resolve(true,"not-number",millis,true)); assertThrows(IllegalArgumentException.class,()->resolver.resolve(true,Long.toString(millis*1_000_000L+1_000_000L),millis,true)); assertEquals(1L,metrics.snapshot().get("legacy_missing_point_time_total"));
@@ -108,4 +127,10 @@ class InfluxRawV2ContractTest {
         assertTrue(query.indexOf("|> group()") < query.indexOf("|> sort("));
         assertTrue(query.indexOf("|> sort(") < query.indexOf("|> limit(n: 2)"));
     }
+
+    private static String repeat(String value, int count) {
+        StringBuilder result = new StringBuilder(value.length() * count);
+        for (int i=0;i<count;i++) result.append(value);
+        return result.toString();
+    }
 }