From 34981c30a78e8bbd7791131059a9210f9928b62c Mon Sep 17 00:00:00 2001
From: 刘光辉 <347230014@qq.com>
Date: 星期四, 17 九月 2026 09:24:09 +0800
Subject: [PATCH] Merge remote-tracking branch 'origin/master' into master
---
jnpf-biz-common/jnpf-audit-sdk/src/main/java/jnpf/audit/sdk/AuditSpoolReplayer.java | 592 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 592 insertions(+), 0 deletions(-)
diff --git a/jnpf-biz-common/jnpf-audit-sdk/src/main/java/jnpf/audit/sdk/AuditSpoolReplayer.java b/jnpf-biz-common/jnpf-audit-sdk/src/main/java/jnpf/audit/sdk/AuditSpoolReplayer.java
new file mode 100644
index 0000000..46c2406
--- /dev/null
+++ b/jnpf-biz-common/jnpf-audit-sdk/src/main/java/jnpf/audit/sdk/AuditSpoolReplayer.java
@@ -0,0 +1,592 @@
+package jnpf.audit.sdk;
+
+import jnpf.audit.model.AuditEventDTO;
+import jnpf.util.JsonUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * spool 閲嶆斁鍣紙Task 7 Step 3锛涜瘎瀹�#3锛歴pool 蹇呴』鍙仮澶嶆墠绠楀厹搴曪級銆�
+ *
+ * <p>鍛ㄦ湡鎬ф壂鎻� spool 鐩綍鐨� {@code .jsonl} / {@code .retry} 鏂囦欢 鈫� 鍘熷瓙閲嶅懡鍚嶅姞
+ * {@code .replaying.<owner-token>} 鍚庣紑锛堣棰嗭紝token 姣忔鏂扮敓鎴愶紝瑙�"#B+#C"娈碉級鈫� 閫愯鍙嶅簭鍒楀寲璧�
+ * {@link AuditEventPublisher#deliverOnce(AuditEventDTO)}
+ * 锛堟垚鍔熶涪寮冭琛屻�佹姏寮傚父淇濈暀锛夆啋 鍏ㄩ儴鎴愬姛鍒� {@code .replaying}锛涢儴鍒嗗け璐ユ妸鍓╀綑琛屽啓鍏�
+ * {@code <绋冲畾 base 鏂囦欢鍚�>.<UUID>.retry} 鏂版枃浠讹紙base+UUID 淇濊瘉鍚岃疆澶氭枃浠跺け璐ヤ笉鎾炲悕锛岀粓楠屼簲杞懁锛�
+ * base 鍚嶇殑鍙栨硶瑙佷笅鏂�"鍐呭 Important 淇"娈� {@link #stableBaseName(String)}锛夛紝
+ * {@code .retry} 鍐欐垚鍔熷悗鍒犲搴� {@code .replaying}锛堜笉鐣欏鍎匡級锛�<b>缁濅笉鍐欏洖娲诲姩鍘熷悕</b>锛堢粓楠屽洓杞憿锛�
+ * 閲嶅懡鍚嶆湡闂村啓鍏ョ鍙兘宸插湪鍘熷悕鏂板缓鏂囦欢杩藉姞鏂颁簨浠讹紝鍐欏洖鍘熷悕浼氳鐩栦涪鏁版嵁锛夈�傝惤搴撳箓绛夛紙client_event_id锛�
+ * 浣块噸鏀惧ぉ鐒跺畨鍏ㄣ��
+ *
+ * <p>璋冨害鎵嬫锛氳嚜寤哄崟绾跨▼ daemon {@link ScheduledExecutorService}锛堥潪 @Scheduled锛夛紝涓嶄緷璧栧涓�
+ * 鏄惁寮�鍚� @EnableScheduling锛岀敓鍛藉懆鏈熺敱鏈� bean 鑷帶锛堣 task-7-report 鍐崇瓥璇存槑锛夈��
+ * 鐢� {@code audit.spool-replay-enabled}锛堥粯璁� true锛夊紑鍏炽��
+ *
+ * <p><b>鍙屽眰閿佸崗璁紙Codex 鎵规浜屽楠� #1/#2 缁堝眬淇锛�</b>锛氫笌 {@link AuditSpoolWriter} 鍏变韩鍚屼竴
+ * {@link AuditSpoolLock}锛堣繘绋嬪唴 monitor + 璺ㄨ繘绋� {@code spool.lock} FileLock锛夈�傚揩閫熸枃浠舵搷浣滃湪閿佸唴銆�
+ * 鎱㈤�熺綉缁滈噸鏀惧湪閿佸锛屼复鐣屽尯鍒掑垎锛�
+ * <ol>
+ * <li><b>璁ら</b>锛坽@link #replayFile}锛夛細{@code rename鈫�.replaying.<owner-token>} + {@code setLastModified}
+ * 鍒锋柊绉熺害鍦�<b>鍚屼竴涓寸晫鍖�</b>鈥斺��#1 涓� writer 鐨� append 浜掓枼锛堟秷闄�"鍙栬蛋鏂囦欢鐬棿"涓�"姝ｅ湪 append"鐨勭獥鍙�
+ * 绔炴�侊級锛�#2 rename 涓庡埛鏂板師瀛愭垚瀵癸紙娑堥櫎"rename 鎴愬姛鍚庛�佸埛鏂� mtime 鍓嶈瀛ゅ効鍥炴敹璇姠"鐨勭獥鍙ｏ級銆�
+ * 涓寸晫鍖哄唴<b>鍏堥攣鍐呴噸楠�</b>锛�#B锛夊啀 rename銆俽eplayer 璺緞 {@code tryLock} 鎷夸笉鍒板氨璺宠繃鏈疆锛屼笅杞啀鏉ャ��</li>
+ * <li><b>readAllLines</b>锛氶攣澶栤�斺�旇棰嗗悗 {@code .replaying} 宸蹭笉鍦ㄤ换浣曞啓鍏ョ/鎵弿闆嗚绾垮唴锛堝鍎垮洖鏀�
+ * 10 鍒嗛挓闃堝�� + 鍒氬埛鏂扮殑 mtime 淇濇姢锛夛紝璇诲彇瀹夊叏涓斾笉闃诲 writer銆�</li>
+ * <li><b>閫愯 deliverOnce</b>锛氶攣澶栵紙缃戠粶鎱㈡搷浣滐級銆�</li>
+ * <li><b>鏀跺熬</b>锛坽@link #finishReplay}锛夛細鍐� {@code .bad} / 鍐� {@code .retry} / 鍒� {@code .replaying}
+ * 鍦ㄩ攣鍐咃紙鍚勮嚜蹇搷浣滐級銆�</li>
+ * <li><b>瀛ゅ効鍥炴敹 rename</b>锛坽@link #recoverOrphanedReplaying}锛夛細妫�鏌�+rename 鍦ㄩ攣鍐呫��</li>
+ * </ol>
+ * 60 绉掗潤榛樼獥鍙ｏ紙{@link #QUIET_WINDOW_MS}锛�<b>淇濈暀涓虹函浼樺寲</b>锛堝噺灏戞椿璺冩枃浠舵悈鍔級锛岄潪姝ｇ‘鎬т緷璧栤�斺��
+ * 姝ｇ‘鎬у凡鐢遍攣鍗忚淇濊瘉銆�<b>娈嬩綑</b>锛氬崟鏂囦欢閲嶆斁瓒� 10 鍒嗛挓浠嶅彲鑳借鍏跺畠杩涚▼褰撳鍎挎姠璧� 鈫� 鍚庢灉=閲嶅閲嶆斁锛�
+ * 钀藉簱鎸� client_event_id 骞傜瓑淇濊瘉涓嶄涪涓嶉噸鍏ュ簱锛屽彲鎺ュ彈锛泏@code spool.lock} 姘镐笉鍙備笌鎵弿/閲嶆斁/鍒犻櫎
+ * 锛堟壂鎻忚繃婊ゅ凡鎸� {@link AuditSpoolLock#LOCK_FILE_NAME} 鎺掗櫎锛夈��
+ *
+ * <p><b>瑁佸畾淇 D锛堝鍎� .replaying 鎭㈠锛�</b>锛氳繘绋嬭嫢鍦�"鏀瑰悕鎴� .replaying"涔嬪悗銆�"澶勭悊瀹�"涔嬪墠宕╂簝锛�
+ * 浼氱暀涓嬪鍎库�斺�旀壂鎻忛泦鍥哄畾涓� {@code {.jsonl,.retry}}锛屾案杩滀笉浼氳棰嗗畠锛屼簨浠舵案涔呮粸鐣欍�傛瘡杞壂鎻忔椂锛�
+ * 棰濆鎶�<b>淇敼鏃堕棿鏃╀簬锛堝綋鍓嶆椂鍒� - 10 鍒嗛挓锛屽嵆涓や釜璋冨害鍛ㄦ湡锛�</b>鐨� .replaying 鏂囦欢鍘熷瓙鏀瑰悕涓�
+ * {@code <绋冲畾 base 鏂囦欢鍚�>.<UUID>.retry}锛堥攣鍐咃級锛屽洖褰� retry 閫氶亾鐢变笅涓�杞嚜鐒舵秷鍖栵紱涓嶅湪鍥炴敹鍔ㄤ綔
+ * 鍐呯洿鎺ュ鐞嗘枃浠跺唴瀹癸紝閬垮厤涓�"鏈疄渚�/骞跺彂瀹炰緥姝ｅ湪閲嶆斁涓殑鏂囦欢"娣锋穯銆傚勾榫勯槇鍊煎嵆涓洪槻姝㈣鎶細姝ｅ父涓�杞�
+ * 澶勭悊杩滃揩浜庝竴涓皟搴﹀懆鏈燂紝鑻� .replaying 瀛樻椿璺ㄨ秺涓や釜鍛ㄦ湡鎵嶅彲鍒ゅ畾涓哄鍎裤�傝惤搴撴寜 client_event_id
+ * 骞傜瓑锛岄噸澶嶉噸鏀惧畨鍏ㄣ��
+ *
+ * <p><b>Codex 鎵规浜屼笁杞� #B+#C锛堣棰嗗悕 owner-token 鍖� + 閿佸唴閲嶉獙锛�</b>锛氳棰� rename 鐩爣鐢卞浐瀹氬悗缂�
+ * {@code .replaying} 鏀逛负 owner-token 鍖栫殑 {@code .replaying.<ownerUUID>}锛坱oken 姣忔璁ら鏂扮敓鎴愶紝
+ * {@link #REPLAYING_TOKEN_SUFFIX} 璇嗗埆锛夈�傛敹灏�/鍒犻櫎鍙搷浣滄湰 owner token 鐨勮矾寰勨�斺�旈檲鏃� owner锛堣鎶㈠悗鏀跺熬锛�
+ * 涓庢柊 owner 姘镐笉鍚屽悕锛寋@code deleteIfExists} 鐗╃悊涓婁笉鍙兘鍒犳帀鏂� owner 鍒氳棰嗙殑鏂囦欢锛堟秷闄� #C 鍥哄畾鍚嶈法
+ * owner 鍐茬獊锛夈�傚悓鏃惰棰嗭紙{@link #replayFile}锛変笌瀛ゅ効鍥炴敹锛坽@link #recoverOrphanedReplaying}锛夐兘鏀逛负
+ * <b>閿佸唴鍏堥噸楠�</b>锛坋xists + mtime 浠嶈秴闃�/浠嶉潤榛橈級鍐� rename鈥斺�旈攣澶� {@code listFiles} 棰勭瓫鍙綔鍊欓�夋敹闆嗭紝
+ * 娑堥櫎"棰勭瓫鍚庛�侀攣鍐� rename 鍓嶏紝璺緞琚垹闄ゅ苟澶嶇敤涓烘柊鏂囦欢 鈫� 鏃ф鏌ョ粨鏋滆鎼柊鏂囦欢"鐨勭獥鍙ｏ紙#B锛夈��
+ * {@code setLastModified} 澶辫触淇濈暀 warn+缁х画锛歰wner-token 鍞竴鍚嶄笅绉熺害澶辫触鐨勬渶鍧忓悗鏋�=琚鍎垮洖鏀垛啋閲嶅閲嶆斁
+ * 锛堝箓绛夊惛鏀讹級锛屼笉鍐嶆湁涓㈠け璺緞銆�
+ *
+ * <p><b>Codex 鎵规浜屼笁杞� #D锛堜紭闆呴��鍑猴級</b>锛歿@link #destroy()} 鍦� {@code shutdownNow} 鍚�
+ * {@code awaitTermination}({@link #SHUTDOWN_AWAIT_MS}) 绛夊緟閲嶆斁绾跨▼鐪熸閫�鍑猴紙瓒呮椂 warn锛夛紝涓�
+ * {@link AuditSpoolLock} 鐨勭粓姝綅閰嶅悎锛岄伩鍏�"鍋滄満杩斿洖浣嗛噸鏀剧嚎绋嬩粛鍦ㄩ攣鍐呮敹灏�"銆�
+ *
+ * <p><b>鍐呭 Important 淇锛坮etry 鏂囦欢鍚嶆敹鏁涳紝闃� NAME_MAX 鏃犵晫澧為暱锛�</b>锛氶儴鍒嗗け璐ュ啓 {@code .retry}銆�
+ * 瀛ゅ効鍥炴敹鏀瑰悕銆佸潖琛岄殧绂诲啓 {@code .bad}锛屽墠缂�閮藉厛缁� {@link #stableBaseName(String)} 浠庢枃浠跺悕鍓ョ<b>鎵�鏈�</b>
+ * {@code .<UUID>.retry} 灞備笌 {@code .replaying.<UUID>} token 鍚庣紑杩樺師鍑虹ǔ瀹� base锛堝 {@code audit-events-20260723.jsonl}锛夛紝
+ * 鍐嶆嫾<b>鎭板ソ涓�灞�</b>鏂� UUID锛屾枃浠跺悕闀垮害浠庢鏈夌晫锛屼笉闅忓け璐ヨ疆娆″闀裤�俇UID 娈电敤 {@link #RETRY_LAYER_SUFFIX}
+ * 涓ユ牸鍖归厤鏍囧噯 8-4-4-4-12 鍗佸叚杩涘埗鏍煎紡骞堕敋瀹氬瓧绗︿覆鏈熬锛屼笉浼氳鍓ヤ笟鍔℃枃浠跺悕鑷韩鎼哄甫鐨勭偣娈碉紙濡� {@code .jsonl}锛夈��
+ * 锛圕odex 浜旇疆璧凤細{@link #recoverOrphanedWriting()} 鐨勯殧绂绘敼鍚嶄笉鍐嶈蛋鏈満鍒垛�斺�旂洿鎺ュ湪鍘熷悕鍚庤拷鍔�
+ * {@code .stale}锛岃鍏朵笓灞炴锛泏@code .retry.writing} 灞傚墺绂诲垎鏀繚鐣欎綔閫氱敤闃插尽锛屽綋鍓嶆棤璋冪敤鏂逛細瑙﹀彂銆傦級
+ *
+ * <p><b>鍧忚闅旂涓庤劚鏁忥紙Codex 鎵规浜� #3/#4锛�</b>锛氬弽搴忓垪鍖栧け璐ョ殑鍧忚涓嶉潤榛樹涪銆佷笉杩� remaining锛屾敹灏炬椂鎵归噺
+ * 闅旂杩� {@code <绋冲畾 base>.<UUID>.bad}锛堜笉鍙備笌浠讳綍鑷姩閲嶆斁锛岀暀浜哄伐淇琛ュ綍锛岃 {@link #finishReplay}锛夛紱
+ * 鍧忚 ERROR 鏃ュ織鍙鏂囦欢鍚�/琛屽彿/琛岄暱/鍐呭 SHA-256 鍓� 12 浣嶏紙{@link #sha256Prefix12}锛夛紝缁濅笉杈撳嚭琛屽唴瀹规鏂囥��
+ *
+ * <p><b>鍐欏悗鍘熷瓙鍙戝竷瀛ゅ効鍥炴敹锛圕odex 鍥涜疆 Critical#1锛�</b>锛歿@link AuditSpoolWriter} 闄嶇骇璺緞鏀逛负銆屽厛鍐�
+ * 鍞竴鍚� {@code .writing} 涓存椂鏂囦欢銆佸啀鍘熷瓙鏀瑰悕鍙戝竷涓� {@code .retry}銆嶏紙娑堢伃璺ㄨ繘绋嬪崐鍐欑獥鍙ｏ紝瑙佸叾绫�
+ * Javadoc锛夈�傝嫢 writer 鎭板ソ宕╂簝浜庝袱姝ヤ箣闂达紙鎴栨敼鍚嶆湰韬洜鏋佺 IO 寮傚父澶辫触锛夛紝浼氶仐鐣�
+ * {@code <绋冲畾base>.<UUID>.retry.writing} 瀛ゅ効鈥斺�旀壂鎻忛泦鍥哄畾鎺掗櫎 {@code .writing}锛堣
+ * {@link #replayOnce()}锛夛紝涓绘祦绋嬫案杩滀笉浼氳棰嗗畠銆傛瘡杞澶栬皟鐢� {@link #recoverOrphanedWriting()}
+ * 澶勭悊杩欑被瀛ゅ効锛涘叿浣撳姩浣滆涓嬩竴娈碘�斺�擟odex 浜旇疆澶嶉獙鎺ㄧ炕浜嗘湰娈垫渶鍒�"鍥炲綊 .retry 閫氶亾褰撴甯镐簨浠堕噸鏀�"鐨勬柟妗堛��
+ *
+ * <p><b>Codex 浜旇疆淇锛�.writing 闅旂鍖猴紝涓嶈兘浠呭嚟骞撮緞鑷姩鍙戝竷锛�</b>锛氬洓杞増鏈浘鎶婅秴榫�
+ * {@code .retry.writing} 鐩存帴鏀瑰悕鍥炲綊 {@code .retry} 閫氶亾褰撴甯镐簨浠堕噸鏀锯�斺�擟odex 浜旇疆澶嶉獙鎸囧嚭杩欎笉瀹夊叏锛�
+ * degraded writer锛堟湭鎸� FileLock 鐨勯檷绾ц矾寰勶級鑻ユ伆濂藉湪鍏跺敮涓�涓�娆� {@link Files#write} 璋冪敤<b>鍐呴儴</b>
+ * 琚搷浣滅郴缁�/纾佺洏 IO 闃诲瓒呰繃 10 鍒嗛挓锛岃 writer 绾跨▼浠嶇劧瀛樻椿銆佷粛鎸佹湁璇ユ枃浠剁殑鎵撳紑 FD锛屽彧鏄皻鏈繑鍥烇紱
+ * 姝ゆ椂瀹冪殑 {@code .writing} 鏂囦欢鍦ㄥ鍎垮洖鏀剁溂涓凡"瓒呴緞"锛岃嫢鐓ф棫鏀瑰悕鍙戝竷涓� {@code .retry}锛岄噸鏀惧櫒浼氭妸瀹�
+ * 褰撴甯镐簨浠惰棰嗭紙rename鈫抺@code .replaying}锛夈�佽鍒扮┖鏂囦欢锛堝唴瀹硅繕娌＄湡姝ｈ惤鐩橈級銆佸垽瀹�"澶勭悊瀹屾瘯"骞跺垹闄�
+ * {@code .replaying}锛涢殢鍚庡崱浣忕殑 write 璋冪敤鎭㈠锛屽悜鐫�宸茶 unlink 鐨� inode 杩藉姞鈥斺�旀暟鎹潤榛樻秷澶憋紝涓嶇暀
+ * 浠讳綍閿欒鐥曡抗銆傚崐琛屾埅鏂椂鍚岀悊锛氬潖琛岄殧绂绘満鍒跺彧鑳藉惛鏀�"鍙栬蛋鏃跺埢"宸茬粡鍐欏叆鐨勫墠缂�锛屽啓璋冪敤灏氭湭瀹屾垚鐨勫悗鍗婃
+ * 涓�鏍蜂細涓€��
+ *
+ * <p>缁撹锛�<b>瓒呴緞涓嶈冻浠ユ柇瑷� writer 宸叉</b>锛寋@link #recoverOrphanedWriting()} 鍥犳鍙仛<b>闅旂</b>銆�
+ * 缁濅笉鍙戝竷銆佺粷涓嶅垹闄も�斺�旈攣鍐呴噸楠岄�氳繃鍚庢妸 {@code .retry.writing} 鍘熷瓙鏀瑰悕涓� {@code <鍘熷悕>.stale}锛堝悓
+ * 鐩綍锛夛紝{@code log.error} 鍛婅锛堣劚鏁忥細浠呰鏂囦欢鍚�/澶у皬/mtime锛屼笉鍚唴瀹癸級鍚庡嵆姝㈡锛氫笉杩� {@code .retry}
+ * 閫氶亾锛堜笉浼氳閲嶆斁鍣ㄨ棰嗭級銆佷笉杩� {@link #finishReplay} 鏀跺熬锛堜笉浼氳鍒犻櫎锛夈�倇@code .stale} 鎺掗櫎鍦ㄤ竴鍒�
+ * 鎵弿/璁ら涔嬪鈥斺�斾富鎵弿锛坽@link #replayOnce()}锛夊彧璁� {@code .jsonl}/{@code .retry}锛�
+ * {@link #recoverOrphanedReplaying()} 鍙 {@code .replaying.<uuid>}锛寋@link #recoverOrphanedWriting()}
+ * 鑷韩鍙 {@code .retry.writing}锛屼笁鑰呯殑鍚庣紑鍒ゅ畾閮戒笉浼氬尮閰嶄互 {@code .stale} 鏀跺熬鐨勬枃浠跺悕锛屾晠闅旂鍚庣殑
+ * 鏂囦欢鐗╃悊涓婁笉鍙兘琚湰绫讳换浣曢�昏緫鍐嶆瑙︾锛涗篃鏃犻渶绾冲叆 {@link #stableBaseName(String)} 鐨勫墺灞傝寖鍥粹�斺��
+ * {@code .stale} 浠庝笉鍙備笌浠讳綍"鏂板缓 retry/bad 鏂囦欢"鐨勫墠缂�鎺ㄥ銆�
+ *
+ * <p>瀹夊叏鎬ц璇佺殑鏍稿績鏄� POSIX {@code rename(2)} 鐨勮涔夛細<b>rename 涓� unlink</b>鈥斺�斿畠鍙敼鍙樼洰褰曢」鎸囧悜鐨�
+ * 鍚嶅瓧锛岃鍗′綇鐨� write 璋冪敤鎸佹湁鐨勬槸瀵� inode 鐨� FD锛屼笌璺緞鍚嶆棤鍏筹紝鎭㈠鍚庝粛浼氭妸鍐呭瀹屾暣鍐欏叆<b>鍚屼竴涓�
+ * inode</b>锛堢幇鍦ㄦ寕鍦� {@code .stale} 杩欎釜鍚嶅瓧涓嬶級锛屼笉浼氬洜鏀瑰悕鑰屼涪澶辨垨鍐欏亸銆備簨浠跺洜姝や繚鍏ㄥ湪闅旂鍖猴紝绛夊緟
+ * 浜哄伐鏍告煡鍚庢墜宸ヨˉ褰曪紙瀵圭収 {@code clientEventId} 鍘婚噸锛夛紱闅旂鍖烘湰韬案涓嶈浠讳綍鑷姩鍖栨祦绋嬪垹闄わ紝鍥犳
+ * "閲嶆斁鍣ㄦ妸浠嶅湪鍐欑殑鏂囦欢褰撳鍎挎姠璧扳啋绌鸿鈫掑垹闄も啋闈欓粯涓㈠け"杩欐潯璺緞琚�<b>鐗╃悊娑堥櫎</b>锛堜笉鏄潬鏃跺簭绐楀彛鏀剁獎锛�
+ * 鑰屾槸浠ｇ爜涓婃牴鏈笉瀛樺湪鑳藉垹闄� {@code .stale} 鐨勮矾寰勶級銆倃riter 渚ч拡瀵�"鍙戝竷婧愯闅旂"鐨勮嚜鎰堣
+ * {@link AuditSpoolWriter} 绫� Javadoc"Codex 浜旇疆淇锛堣嚜鎰堥噸鍙戯級"娈碘�斺�攎ove 鐩爣婧愯闅旂鍚庝笉浼氳浜嬩欢
+ * 鐪熺殑涓紝鍙細鍦� {@code .stale} 閲岀暀涓�浠借嚦澶氳浜哄伐琛ュ綍鏃堕噸澶嶇殑鍓湰锛堣惤搴撴寜 {@code clientEventId} 骞傜瓑
+ * 鍚告敹锛夈��
+ */
+@Slf4j
+public class AuditSpoolReplayer implements InitializingBean, DisposableBean {
+
+ private static final long DEFAULT_REPLAY_INTERVAL_MS = 300_000L; // spec: fixedDelay 5 鍒嗛挓锛堥粯璁わ紝鍙粡 audit.spool-replay-interval-ms 瑕嗙洊锛�
+ private static final String REPLAYING_SUFFIX = ".replaying";
+ // 瑁佸畾淇 D锛氬鍎� .replaying.<uuid> 骞撮緞闃堝�� = 涓や釜榛樿璋冨害鍛ㄦ湡锛�10 鍒嗛挓锛夛紝闃叉璇姠骞跺彂瀹炰緥姝ｅ湪澶勭悊鐨勬枃浠躲��
+ // 鍒绘剰閿氬畾 DEFAULT_REPLAY_INTERVAL_MS 鑰岄潪瀹炰緥璋冨害鍛ㄦ湡锛氬鍎垮勾榫勯槇鍊兼槸"澶氫箙鎵嶆暍鍒ゅ畾鏂囦欢琚穿婧冮仐寮�"鐨�
+ // 宕╂簝鎭㈠瀹夊叏搴曠嚎锛屼笌"澶氫箙杞涓�娆�"鏄袱涓叧娉ㄧ偣鈥斺�旀妸杞鍛ㄦ湡璋冪煭锛堝娴嬭瘯鐜 15s 鍔犻�熼獙璇侊級涓嶅簲杩炲甫
+ // 缂╁皬杩欐潯瀹夊叏绐楀彛锛屽惁鍒欎竴娆＄暐鎱㈢殑姝ｅ父閲嶆斁灏卞彲鑳借璇垽涓哄鍎垮苟琚苟鍙戝疄渚嬫姠璧帮紙骞傜瓑鍚告敹浣嗗緬澧炴悈鍔級銆�
+ private static final long ORPHAN_REPLAYING_AGE_MS = 2 * DEFAULT_REPLAY_INTERVAL_MS;
+ // Codex 鎵规浜� #1锛氭椿鍔� .jsonl 闈欓粯绐楀彛鈥斺�旇窛浠婁笉瓒� 60 绉掓湰杞烦杩囥�傞攣鍗忚钀藉湴鍚庢绐楀彛闄嶇骇涓虹函浼樺寲
+ // 锛堝噺灏戞椿璺冩枃浠舵悈鍔級锛岄潪姝ｇ‘鎬т緷璧栵紱姝ｇ‘鎬х敱 AuditSpoolLock 鍙屽眰閿佷繚璇�
+ private static final long QUIET_WINDOW_MS = 60_000L;
+ // 鍐呭 Important 淇锛氱簿纭尮閰�"灏鹃儴鎭板ソ涓�灞� <UUID>.retry"锛孶UID 涓ユ牸鎸� 8-4-4-4-12 鍗佸叚杩涘埗鏍煎紡
+ // 閿氬畾瀛楃涓叉湯灏撅紙$锛夛紝閬垮厤璇墺涓氬姟鏂囦欢鍚嶈嚜韬惡甯︾殑鐐规锛堝 audit-events-20260723.jsonl 鐨� .jsonl锛�
+ private static final Pattern RETRY_LAYER_SUFFIX = Pattern.compile(
+ "\\.[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\\.retry$");
+ // Codex 鎵规浜屼笁杞� #B+#C锛氳棰嗗悕 owner-token 鍖栤�斺�攔ename 鐩爣 <src鍘熷悕>.replaying.<ownerUUID>锛�
+ // 姣忔璁ら鏂扮敓鎴� owner token銆傚鍎垮尮閰�/stableBaseName 鍓ョ閮戒互鏈ā寮忚瘑鍒� .replaying.<uuid> 灞傦紝
+ // 閿氬畾瀛楃涓叉湯灏撅紙$锛夛紝UUID 涓ユ牸 8-4-4-4-12 鍗佸叚杩涘埗鏍煎紡
+ private static final Pattern REPLAYING_TOKEN_SUFFIX = Pattern.compile(
+ "\\.replaying\\.[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$");
+ // Codex 鍥涜疆 Critical#1锛欰uditSpoolWriter 闄嶇骇璺緞鍐欏悗鍘熷瓙鍙戝竷鐨勪复鏃跺悕 <绋冲畾base>.<UUID>.retry.writing鈥斺��
+ // 宕╂簝/rename 澶辫触浼氶仐鐣欏鍎匡紝鏈ā寮忎緵鎵弿鎺掗櫎涓� recoverOrphanedWriting() 瀛ゅ効鍥炴敹璇嗗埆锛�
+ // 閿氬畾瀛楃涓叉湯灏撅紙$锛夛紝UUID 涓ユ牸 8-4-4-4-12 鍗佸叚杩涘埗鏍煎紡
+ private static final Pattern RETRY_WRITING_SUFFIX = Pattern.compile(
+ "\\.[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\\.retry\\.writing$");
+ // Codex 浜旇疆锛歳ecoverOrphanedWriting() 闅旂鐩爣鍚庣紑鈥斺�斿彧闅旂涓嶅彂甯冧笉鍒犻櫎銆備换浣曚互姝ょ粨灏剧殑鏂囦欢鍚嶉兘涓嶅啀
+ // 鍖归厤 RETRY_WRITING_SUFFIX/REPLAYING_TOKEN_SUFFIX锛堜簩鑰呭潎瑕佹眰鎭板ソ浠� .retry.writing / .replaying.<uuid>
+ // 鏀跺熬锛夛紝涔熶笉浠� .jsonl/.retry 缁撳熬锛屾晠闅旂鍚庡ぉ鐒朵笉浼氳鏈被浠讳綍鎵弿閫昏緫鍐嶆鍛戒腑锛岃绫� Javadoc"Codex 浜旇疆淇"娈�
+ private static final String STALE_SUFFIX = ".stale";
+ // Codex 鎵规浜屼笁杞� #D锛歞estroy 鏃剁瓑寰呴噸鏀剧嚎绋嬮��鍑虹殑涓婇檺
+ private static final long SHUTDOWN_AWAIT_MS = 5_000L;
+
+ private final AuditEventPublisher publisher;
+ private final File spoolDir;
+ private final boolean enabled;
+ private final AuditSpoolLock spoolLock;
+ // 璋冨害鍛ㄦ湡锛氶粯璁� DEFAULT_REPLAY_INTERVAL_MS锛�5 鍒嗛挓锛夛紝鍙粡 audit.spool-replay-interval-ms 瑕嗙洊銆�
+ // 浠呭奖鍝�"澶氫箙杞涓�娆� spool"锛屼笉褰卞搷瀛ゅ効骞撮緞闃堝�硷紙ORPHAN_REPLAYING_AGE_MS锛屽浐瀹氬畨鍏ㄥ簳绾匡紝瑙佸叾娉ㄩ噴锛夈��
+ // 鐢熶骇榛樿涓嶅姩锛涙祴璇曠幆澧冨彲璋冪煭锛堝 15s锛夊姞閫熼檷绾ч摼鎭㈠閲嶆斁楠岃瘉銆�
+ private final long replayIntervalMs;
+ private ScheduledExecutorService scheduler;
+
+ public AuditSpoolReplayer(AuditEventPublisher publisher, String spoolDir, boolean enabled,
+ AuditSpoolLock spoolLock) {
+ this(publisher, spoolDir, enabled, spoolLock, DEFAULT_REPLAY_INTERVAL_MS);
+ }
+
+ public AuditSpoolReplayer(AuditEventPublisher publisher, String spoolDir, boolean enabled,
+ AuditSpoolLock spoolLock, long replayIntervalMs) {
+ this.publisher = publisher;
+ this.spoolDir = new File(spoolDir);
+ this.enabled = enabled;
+ this.spoolLock = spoolLock;
+ // 鍏滃簳锛氶潪姝ｆ暟锛堥厤缃瑪璇�/0锛夊洖钀介粯璁ゅ懆鏈燂紝缁濅笉璁� scheduleWithFixedDelay 鏀跺埌闈炴硶鍛ㄦ湡
+ this.replayIntervalMs = replayIntervalMs > 0 ? replayIntervalMs : DEFAULT_REPLAY_INTERVAL_MS;
+ }
+
+ @Override
+ public void afterPropertiesSet() {
+ if (!enabled) {
+ log.info("[audit] spool replayer 宸茬鐢� (audit.spool-replay-enabled=false)");
+ return;
+ }
+ scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "audit-spool-replayer");
+ t.setDaemon(true);
+ return t;
+ });
+ // initialDelay 鍙栦笌鍛ㄦ湡涓�鑷达細閬垮厤鍚姩鐬棿涓庝笟鍔′簤 IO锛涢仐鐣� spool 鑷冲寤惰繜涓�涓懆鏈熻閲嶆斁
+ scheduler.scheduleWithFixedDelay(this::replaySafely,
+ replayIntervalMs, replayIntervalMs, TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public void destroy() {
+ if (scheduler == null) {
+ return;
+ }
+ // Codex 鎵规浜屼笁杞� #D锛歴hutdownNow 鍚庣瓑寰呴噸鏀剧嚎绋嬬湡姝ｉ��鍑猴紝閬垮厤"鍋滄満杩斿洖浣嗛噸鏀剧嚎绋嬩粛鍦ㄩ攣鍐�/
+ // 姝ｈ皟 AuditSpoolLock 鏀跺熬"鐨勭獥鍙ｂ�斺�斾笌 AuditSpoolLock.destroy 鐨勭粓姝綅閰嶅悎浼橀泤閫�鍑恒��
+ scheduler.shutdownNow();
+ try {
+ if (!scheduler.awaitTermination(SHUTDOWN_AWAIT_MS, TimeUnit.MILLISECONDS)) {
+ log.warn("[audit] spool 閲嶆斁绾跨▼ {}ms 鍐呮湭閫�鍑猴紙鍙兘鍗″湪 IO/缃戠粶閲嶆斁涓級", SHUTDOWN_AWAIT_MS);
+ }
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt(); // 鎭㈠涓柇鏍囪鍚庤繑鍥�
+ }
+ }
+
+ private void replaySafely() {
+ try {
+ replayOnce();
+ } catch (Throwable t) {
+ // 璋冨害绾跨▼鎶涘紓甯镐細瀵艰嚧 scheduleWithFixedDelay 鍋滄憜锛屽悗缁疆娆′笉鍐嶆墽琛屸�斺�斿繀椤诲悶浣�
+ log.error("[audit] spool 閲嶆斁杞寮傚父", t);
+ }
+ }
+
+ void replayOnce() {
+ if (!spoolDir.isDirectory()) {
+ return;
+ }
+ recoverOrphanedReplaying();
+ // Codex 鍥涜疆 Critical#1锛氬鍎� .retry.writing 鍥炴敹鈥斺�擜uditSpoolWriter 闄嶇骇璺緞鍐欏悗鍘熷瓙鍙戝竷涓柇
+ // 锛堝穿婧�/rename 澶辫触锛夐仐鐣欑殑涓存椂鏂囦欢锛屾壂鎻忛泦鍥哄畾鎺掗櫎 .writing锛屾案杩滀笉浼氳涓嬮潰鐨勪富娴佺▼璁ら
+ recoverOrphanedWriting();
+ // 鎵弿闆嗗彧鍚� .jsonl/.retry锛屾樉寮忔帓闄� spool.lock锛堥攣鏂囦欢姘镐笉鍙備笌鎵弿/閲嶆斁锛変笌 .writing
+ // 锛圕odex 鍥涜疆 Critical#1锛氬啓鍚庡師瀛愬彂甯冪殑涓存椂鏂囦欢锛屽唴瀹瑰彲鑳戒笉瀹屾暣锛岀粷涓嶈兘琚湰鎵弿鐩存帴鍛戒腑锛�
+ // Codex 浜旇疆璧凤紝瓒呴緞 .writing 瀛ゅ効鍙細琚� recoverOrphanedWriting() 闅旂涓� .stale锛屼笉鍐嶅洖褰掓湰鎵弿闆嗭級銆�
+ // .stale锛堥殧绂诲尯锛岃绫� Javadoc"Codex 浜旇疆淇"娈碉級澶╃劧涓嶄互 .jsonl/.retry 缁撳熬锛岃鏈皳璇嶆帓闄わ紝
+ // 鏃犻渶棰濆鏉′欢
+ File[] targets = spoolDir.listFiles((dir, name) ->
+ !name.equals(AuditSpoolLock.LOCK_FILE_NAME)
+ && !name.endsWith(".writing")
+ && (name.endsWith(".jsonl") || name.endsWith(".retry")));
+ if (targets == null) {
+ return;
+ }
+ long now = System.currentTimeMillis();
+ for (File src : targets) {
+ // Codex 鎵规浜� #1锛�60 绉掗潤榛樼獥鍙ｂ�斺�旇窛浠婁笉瓒� 60 绉掔殑 .jsonl 瑙嗕负鍐欏叆绔彲鑳芥椿璺冿紝鏈疆璺宠繃锛�
+ // 涓嬩竴杞啀璁ら锛�<0.5 QPS锛屽欢杩熶竴鍛ㄦ湡鏃犲锛夈�傞攣鍗忚钀藉湴鍚庢璺宠繃浠呬负浼樺寲锛堝噺灏戞椿璺冩枃浠舵悈鍔級锛�
+ // 鍗充究涓嶈烦杩囥�佹纭�т篃鐢� AuditSpoolLock 鐨� append鈫攔ename 浜掓枼淇濊瘉銆�.retry 鐢辨湰閲嶆斁鍣ㄨ嚜寤恒��
+ // 鏃犲閮ㄥ啓鍏ョ锛屼笉鍙楁闄愶紱.bad 鏈氨涓嶅湪鎵弿闆嗭紙涓斾粠涓嶉噸鏀撅級銆�
+ if (src.getName().endsWith(".jsonl")
+ && now - src.lastModified() < QUIET_WINDOW_MS) {
+ continue;
+ }
+ replayFile(src);
+ }
+ }
+
+ /**
+ * 鍐呭 Important 淇锛氫粠鏂囦欢鍚嶅墺绂�<b>鎵�鏈�</b> {@code .<UUID>.retry} 灞備笌 {@code .replaying.<UUID>} token 鍚庣紑锛�
+ * 杩樺師鍑虹ǔ瀹� base name锛堝 {@code audit-events-20260723.jsonl}锛夛紝瑙佺被娉ㄩ噴"鍐呭 Important 淇"娈点��
+ *
+ * <p>鍓ョ椤哄簭锛氬厛鍘绘帀鏈熬 owner-token 鍖栫殑 {@code .replaying.<UUID>} 鍚庣紑锛堣棰嗘椂鏈�鍚庤拷鍔狅紝
+ * 鏄渶澶栧眰锛屾渶澶氫竴灞傦紝鐢� {@link #REPLAYING_TOKEN_SUFFIX} 鍖归厤锛夛紝鍐嶅幓鎺� {@link AuditSpoolWriter}
+ * 闄嶇骇璺緞鍐欏悗鍘熷瓙鍙戝竷鐨勪复鏃跺眰 {@code .<UUID>.retry.writing}锛圕odex 鍥涜疆 Critical#1锛�
+ * {@link #RETRY_WRITING_SUFFIX} 鍖归厤锛屽悓鏍锋渶澶氫竴灞傘�佷笌 replaying token 浜掓枼涓嶅叡瀛橈級锛屾渶鍚庡惊鐜墺绂�
+ * 鏈熬鐨� {@code .<UUID>.retry} 灞傦紝鐩村埌涓嶅啀鍖归厤涓烘锛堢梾鎬佸満鏅笅鍙兘鏈夊灞傦紝鍘嗗彶绱Н鐨勬瘡涓�灞傞兘浼氳鍓ユ帀锛夈��
+ * UUID 娈电敤 {@link #RETRY_LAYER_SUFFIX}/{@link #REPLAYING_TOKEN_SUFFIX}/{@link #RETRY_WRITING_SUFFIX}
+ * 涓ユ牸鍖归厤鏍囧噯 8-4-4-4-12 鍗佸叚杩涘埗鏍煎紡骞堕敋瀹氬瓧绗︿覆鏈熬锛屼笉浼氳鍓ヤ笟鍔℃枃浠跺悕鏈韩鑷甫鐨勭偣娈点��
+ *
+ * @param name 鍘熷鏂囦欢鍚嶏紙鍙兘宸插甫鑻ュ共灞� retry 鍚庣紑 + 涓�灞� replaying token 鎴栦竴灞� writing 涓存椂灞傦紝
+ * 涔熷彲鑳芥槸浠庢湭閲嶆斁杩囩殑鍘熷鏂囦欢鍚嶏級
+ * @return 鍓ョ骞插噣鐨勭ǔ瀹� base name
+ */
+ private String stableBaseName(String name) {
+ String base = name;
+ Matcher replayingMatcher = REPLAYING_TOKEN_SUFFIX.matcher(base);
+ if (replayingMatcher.find()) {
+ base = base.substring(0, replayingMatcher.start());
+ }
+ // Codex 鍥涜疆 Critical#1锛�.retry.writing 鍙細鏄� AuditSpoolWriter 闄嶇骇璺緞杩藉姞鐨勬渶澶栧眰锛堝啓鍚庡師瀛�
+ // 鍙戝竷鐨勪复鏃跺悕锛屼粠涓嶅祵濂楋級锛屼笌 .replaying.<uuid> 浜掓枼鍚屾椂鍑虹幇锛屾晠涔熷彧鍓ユ渶澶氫竴灞�
+ Matcher writingMatcher = RETRY_WRITING_SUFFIX.matcher(base);
+ if (writingMatcher.find()) {
+ base = base.substring(0, writingMatcher.start());
+ }
+ Matcher matcher;
+ while ((matcher = RETRY_LAYER_SUFFIX.matcher(base)).find()) {
+ base = base.substring(0, matcher.start());
+ }
+ return base;
+ }
+
+ /**
+ * 瑁佸畾淇 D锛氬洖鏀跺穿婧冮仐鐣欑殑 .replaying 瀛ゅ効鏂囦欢锛岃绫绘敞閲娿��
+ *
+ * <p>鍙"淇敼鏃堕棿鏃╀簬 now - {@link #ORPHAN_REPLAYING_AGE_MS}"鐨� {@code .replaying.<uuid>} 鐢熸晥鈥斺��
+ * 骞撮緞闃堝�奸槻姝㈣鎶㈠苟鍙戝疄渚嬶紙鎴栨湰瀹炰緥鏈疆绋嶆棭锛夋鍦ㄥ鐞嗕腑鐨勬枃浠讹紱妫�鏌�+rename 鍦�
+ * {@link AuditSpoolLock} 涓寸晫鍖猴紙閿佸崗璁复鐣屽尯 5锛夊畬鎴愶紝浜ゅ洖 {@code .retry} 閫氶亾鐢辨湰杞垨涓嬩竴杞�
+ * {@link #replayFile(File)} 姝ｅ父娑堝寲锛屾湰鏂规硶涓嶇洿鎺ヨ鍙�/澶勭悊鏂囦欢鍐呭銆�
+ *
+ * <p><b>Codex 鎵规浜屼笁杞� #B锛堥攣鍐呴噸楠岋級</b>锛氶攣澶� {@code listFiles} 棰勭瓫鍙綔鍊欓�夋敹闆嗭紱棰勭瓫涓庨攣鍐�
+ * rename 涔嬮棿锛岃璺緞鍙兘宸茶鍒犻櫎骞跺鐢ㄤ负涓�涓�<b>鏂�</b>鏂囦欢锛堝悓鍚嶅阀鍚堬級锛岃嫢娌跨敤棰勭瓫缁撴灉鐩存帴鎼蛋浼氭妸
+ * 鏂版枃浠惰褰撴棫瀛ゅ効銆傛晠閿佸唴 rename 鍓�<b>閲嶆柊纭</b> orphan 浠嶅瓨鍦ㄤ笖 mtime 浠嶈秴闃堝�硷紝鎵嶆墽琛� rename銆�
+ */
+ private void recoverOrphanedReplaying() {
+ File[] orphans = spoolDir.listFiles((dir, name) ->
+ !name.equals(AuditSpoolLock.LOCK_FILE_NAME)
+ && REPLAYING_TOKEN_SUFFIX.matcher(name).find());
+ if (orphans == null || orphans.length == 0) {
+ return;
+ }
+ long now = System.currentTimeMillis();
+ for (final File orphan : orphans) {
+ if (now - orphan.lastModified() < ORPHAN_REPLAYING_AGE_MS) {
+ continue; // 閿佸棰勭瓫锛氭湭瓒呴緞锛屽彲鑳芥琚鐞嗕腑锛屾湰杞笉纰�
+ }
+ String orphanName = orphan.getName();
+ // 鍐呭 Important 淇锛氬厛褰掍竴鍒扮ǔ瀹� base 鍐嶆嫾鎭板ソ涓�灞傛柊 UUID锛岄槻姝㈠鍎垮湪宸茬疮绉灞�
+ // 鍚庣紑鐨勫悕瀛椾笂缁х画杩藉姞銆佹枃浠跺悕鏃犵晫澧為暱
+ final File retry = new File(orphan.getParentFile(),
+ stableBaseName(orphanName) + "." + UUID.randomUUID() + ".retry");
+ final boolean[] moved = {false};
+ try {
+ // 閿佸崗璁复鐣屽尯 5锛氭鏌�+rename 鍦ㄩ攣鍐咃紝涓庡苟鍙戝疄渚�/鏈繘绋嬭棰嗕簰鏂ワ紱tryLock 鎷夸笉鍒板垯鏈疆璺宠繃璇ュ鍎�
+ boolean lockRun = spoolLock.tryLockAndRun(() -> {
+ // Codex#B 閿佸唴閲嶉獙锛氶绛涘悗璺緞鍙兘宸茶鍒犻櫎骞跺鐢ㄤ负鏂版枃浠讹紝閲嶆柊纭浠嶅瓨鍦ㄤ笖浠嶈秴榫勬墠鎼蛋
+ if (!orphan.exists()) {
+ return;
+ }
+ if (System.currentTimeMillis() - orphan.lastModified() < ORPHAN_REPLAYING_AGE_MS) {
+ return;
+ }
+ Files.move(orphan.toPath(), retry.toPath(), StandardCopyOption.ATOMIC_MOVE);
+ moved[0] = true;
+ });
+ if (lockRun && moved[0]) {
+ log.warn("[audit] 鍥炴敹瀛ゅ効 .replaying 鏂囦欢: {} -> {}", orphanName, retry.getName());
+ }
+ // lockRun=false锛堜粬杩涚▼鎸侀攣锛夋垨 moved=false锛堥攣鍐呴噸楠屾湭杩囷級锛氭湰杞烦杩囪瀛ゅ効锛屼笅杞啀鍒�
+ } catch (IOException moveFail) {
+ // 鏀瑰悕澶辫触锛氬彲鑳藉凡琚叾浠栧疄渚�/鏈疆绋嶆棭澶勭悊锛屼笅涓�杞啀鍒や竴娆�
+ log.warn("[audit] 瀛ゅ効 .replaying 鍥炴敹鏀瑰悕澶辫触锛屼笅杞啀璇�: {}", orphanName);
+ }
+ }
+ }
+
+ /**
+ * Codex 浜旇疆淇锛氬洖鏀� {@link AuditSpoolWriter} 闄嶇骇璺緞銆屽啓鍚庡師瀛愬彂甯冦�嶄腑鏂仐鐣欑殑
+ * {@code <绋冲畾base>.<UUID>.retry.writing} 瀛ゅ効鈥斺��<b>鍙殧绂伙紝姘镐笉鍙戝竷涓� .retry锛屾案涓嶅垹闄�</b>銆�
+ *
+ * <p>鍥涜疆鐗堟湰鏇炬妸瓒呴緞瀛ゅ効鐩存帴鏀瑰悕鍥炲綊 {@code .retry} 閫氶亾褰撴甯镐簨浠堕噸鏀撅紝Codex 浜旇疆澶嶉獙鎸囧嚭杩欎笉瀹夊叏锛�
+ * 瓒呴緞锛坢time 鏃╀簬 {@link #ORPHAN_REPLAYING_AGE_MS}锛夊彧鑳借鏄�"杩欎釜鏂囦欢寰堜箙娌℃湁琚洿鏂�"锛屼笉鑳借瘉鏄�
+ * writer 宸茬粡姝讳骸鈥斺�斿畠鍙兘浠嶅崱鍦ㄩ偅鍞竴涓�娆� {@link java.nio.file.Files#write} 璋冪敤鍐呴儴锛堢鐩�/缃戠粶
+ * 瀛樺偍 IO 闃诲锛夛紝FD 浠嶆墦寮�銆佽繘绋嬩粛瀛樻椿銆傝嫢鐓ф棫鍥炲綊 {@code .retry} 閫氶亾锛岄噸鏀惧櫒浼氳棰嗏啋璇诲埌绌烘枃浠�
+ * 锛堟垨宕╂簝璇箟涓嬬殑鍗婅鍓嶇紑锛夆啋鍒ゅ畾澶勭悊瀹屾瘯鈫掑垹闄わ紱鍗′綇鐨� write 鎭㈠鍚庡啓鍚戝凡 unlink 鐨� inode锛屾暟鎹�
+ * 闈欓粯娑堝け锛屼笖涓嶇暀浠讳綍閿欒鏃ュ織銆傚畬鏁磋璇佽绫� Javadoc"Codex 浜旇疆淇锛�.writing 闅旂鍖猴級"娈点��
+ *
+ * <p>淇鍚庡姩浣滐細閿佸唴閲嶉獙锛坋xists + 浠嶈秴榫勶紝闃叉璇姠姝ｅ湪鍐欏叆涓殑鏂囦欢锛夐�氳繃鍚庯紝鎶� {@code .retry.writing}
+ * <b>鍘熷瓙鏀瑰悕涓� {@code <鍘熷悕>.stale}</b>锛堝悓鐩綍锛夊苟 {@code log.error} 鍛婅锛堣劚鏁忥細浠呮枃浠跺悕/澶у皬/mtime锛�
+ * 涓嶅惈鍐呭锛夆�斺�斾笉瑙ｆ瀽鍐呭銆佷笉鍐欏洖浠讳綍鍙鎵弿/璁ら鐨勫悕瀛椼�佷笉鍒犻櫎銆倇@code rename} 涓� unlink锛氬嵆渚胯
+ * writer 闅忓悗鐪熺殑鎭㈠骞舵妸鍐呭鍐欏畬锛屼篃鏄啓杩涘悓涓�涓� inode锛堢幇鍚� {@code .stale}锛夛紝鍐呭涓嶄細鍥犳敼鍚嶈��
+ * 涓㈠け锛屽彧鏄粸鐣欓殧绂诲尯寰呬汉宸ユ牳鏌ヨˉ褰曪紙瀵圭収 {@code clientEventId} 鍘婚噸锛岃惤搴撳ぉ鐒跺箓绛夛級銆倃riter 渚ч拡瀵�
+ * "鍙戝竷婧愯闅旂"鐨勮嚜鎰堣 {@link AuditSpoolWriter} 绫� Javadoc銆�
+ */
+ private void recoverOrphanedWriting() {
+ // RETRY_WRITING_SUFFIX 鍙尮閰嶄互 ".<uuid>.retry.writing" 缁撳熬鐨勬枃浠跺悕鈥斺�旀湰鏂规硶浜у嚭鐨� .stale
+ // 鏂囦欢涓嶅啀浠ユ缁撳熬锛屽ぉ鐒朵笉浼氳涓嬩竴杞壂鎻忛噸鏂板懡涓紙瑙佺被 Javadoc"Codex 浜旇疆淇"娈碉級
+ File[] orphans = spoolDir.listFiles((dir, name) ->
+ !name.equals(AuditSpoolLock.LOCK_FILE_NAME)
+ && RETRY_WRITING_SUFFIX.matcher(name).find());
+ if (orphans == null || orphans.length == 0) {
+ return;
+ }
+ long now = System.currentTimeMillis();
+ for (final File orphan : orphans) {
+ if (now - orphan.lastModified() < ORPHAN_REPLAYING_AGE_MS) {
+ continue; // 閿佸棰勭瓫锛氭湭瓒呴緞锛屽彲鑳芥琚� writer 鍐欏叆涓紙鎴栧崱鍦� Files.write 鍐呴儴锛夛紝鏈疆涓嶇
+ }
+ String orphanName = orphan.getName();
+ // Codex 浜旇疆锛氬彧闅旂锛屽師鍚嶅悗杩藉姞 .stale鈥斺�斾笉鍙戝竷涓� .retry锛堜笉浼氳閲嶆斁鍣ㄨ棰嗭級銆佷笉鍒犻櫎
+ // 锛�.stale 鎺掗櫎鍦ㄤ竴鍒囨壂鎻忎箣澶栵紝瑙佺被 Javadoc"Codex 浜旇疆淇"娈电殑瀹屾暣璁鸿瘉锛�
+ final File stale = new File(orphan.getParentFile(), orphanName + STALE_SUFFIX);
+ final long[] quarantinedSize = {-1L};
+ final long[] quarantinedMtime = {-1L};
+ final boolean[] moved = {false};
+ try {
+ // 閿佸唴閲嶉獙锛堝悓 recoverOrphanedReplaying 鐨� #B 妯″紡锛夛細棰勭瓫鍚庛�侀攣鍐� rename 鍓嶏紝璺緞鍙兘
+ // 宸茶 writer 瀹屾垚鍙戝竷骞跺垹闄ゃ�佹垨琚埆鐨勫疄渚嬪厛涓�姝ラ殧绂伙紝閲嶆柊纭浠嶅瓨鍦ㄤ笖浠嶈秴榫勬墠鍔ㄦ墜
+ boolean lockRun = spoolLock.tryLockAndRun(() -> {
+ if (!orphan.exists()) {
+ return;
+ }
+ if (System.currentTimeMillis() - orphan.lastModified() < ORPHAN_REPLAYING_AGE_MS) {
+ return;
+ }
+ quarantinedSize[0] = orphan.length();
+ quarantinedMtime[0] = orphan.lastModified();
+ Files.move(orphan.toPath(), stale.toPath(), StandardCopyOption.ATOMIC_MOVE);
+ moved[0] = true;
+ });
+ if (lockRun && moved[0]) {
+ // log.error 鑰岄潪 warn锛氳繖鏄渶瑕佷汉宸ユ牳鏌ヨˉ褰曠殑闅旂浜嬩欢锛屼笉鏄彲鑷剤鐨勭灛鏃剁姸鍐�
+ log.error("[audit] 瓒呴緞 .retry.writing 闅旂鑷� .stale锛堜笉鑷姩鍙戝竷/涓嶈嚜鍔ㄥ垹闄わ紝"
+ + "闇�浜哄伐鏍告煡琛ュ綍锛岃 AuditSpoolReplayer 绫� Javadoc"
+ + "\".writing 闅旂鍖哄缃甛"璇存槑娈碉級: file={} size={}bytes mtime={}",
+ stale.getName(), quarantinedSize[0], quarantinedMtime[0]);
+ }
+ // lockRun=false锛堜粬杩涚▼鎸侀攣锛夋垨 moved=false锛堥攣鍐呴噸楠屾湭杩囷級锛氭湰杞烦杩囪瀛ゅ効锛屼笅杞啀鍒�
+ } catch (IOException moveFail) {
+ // 鏀瑰悕澶辫触锛氬彲鑳藉凡琚叾浠栧疄渚�/鏈疆绋嶆棭澶勭悊锛屼笅涓�杞啀鍒や竴娆�
+ log.warn("[audit] 瀛ゅ効 .retry.writing 闅旂鏀瑰悕澶辫触锛屼笅杞啀璇�: {}", orphanName);
+ }
+ }
+ }
+
+ private void replayFile(final File src) {
+ // 閿佸崗璁复鐣屽尯 2锛堣棰嗭級锛欳odex 鎵规浜� #1/#2/#B/#C銆�
+ // #B+#C 璁ら鍚� owner-token 鍖栤�斺�旀瘡娆¤棰嗘柊鐢熸垚 ownerToken锛宺ename 鐩爣 <src鍘熷悕>.replaying.<ownerToken>銆�
+ // 鏀跺熬/鍒犻櫎鍙搷浣滄湰 owner token 鐨勮矾寰勶細闄堟棫 owner锛堣鎶㈠悗鏀跺熬锛変笌鏂� owner 姘镐笉鍚屽悕锛�
+ // deleteIfExists 鐗╃悊涓婁笉鍙兘璇垹浠� owner 鍒氳棰嗙殑鍚屽悕鏂囦欢锛堟秷闄� Codex#C 鍥哄畾鍚嶈法 owner 鍐茬獊锛夈��
+ final File replaying = new File(src.getParentFile(),
+ src.getName() + REPLAYING_SUFFIX + "." + UUID.randomUUID());
+ final boolean isJsonl = src.getName().endsWith(".jsonl");
+ final boolean[] renamed = {false};
+ // rename鈫�.replaying.<owner> 涓� setLastModified 鍒锋柊绉熺害鍦ㄥ悓涓�涓寸晫鍖猴紙杩涚▼鍐� monitor + 璺ㄨ繘绋� FileLock锛夛細
+ // #1锛氫笌 writer 鐨� append 浜掓枼锛屾秷闄�"鍙栬蛋鏂囦欢鐬棿"涓�"姝ｅ湪 append"鐨勭獥鍙ｇ珵鎬侊紱
+ // #2锛歳ename 涓庡埛鏂板師瀛愭垚瀵癸紝娑堥櫎"rename 鎴愬姛鍚庛�佸埛鏂� mtime 鍓嶈瀛ゅ効鍥炴敹璇姠"鐨勭獥鍙ｃ��
+ try {
+ boolean lockRun = spoolLock.tryLockAndRun(() -> {
+ // Codex#B 閿佸唴閲嶉獙锛氶攣澶栭绛涳紙quiet window锛変笌閿佸唴涔嬮棿锛宻rc 鍙兘宸茶浠� owner 鍙栬蛋/鍒犻櫎銆�
+ // 閿佸唴閲嶆柊纭 src 浠嶅瓨鍦紙.jsonl 涓斾粛鍦ㄩ潤榛樼獥鍙ｅ唴璇存槑鍐欏叆绔彲鑳芥椿璺冿紝璁╃粰涓嬭疆锛夋墠璁ら銆�
+ if (!src.exists()) {
+ return;
+ }
+ if (isJsonl && System.currentTimeMillis() - src.lastModified() < QUIET_WINDOW_MS) {
+ return;
+ }
+ Files.move(src.toPath(), replaying.toPath(), StandardCopyOption.ATOMIC_MOVE);
+ // setLastModified 杩斿洖 false 浠呭憡璀︿笉涓柇銆侰odex#C锛歰wner-token 鍞竴鍚嶄笅锛岀绾﹀缓绔嬪け璐ョ殑鏈�鍧�
+ // 鍚庢灉=鏈枃浠惰秴 10 鍒嗛挓鍚庤瀛ゅ効鍥炴敹鈫掗噸澶嶉噸鏀撅紙钀藉簱鎸� client_event_id 骞傜瓑鍚告敹锛夛紝涓嶅啀鏈変涪澶辫矾寰勩��
+ if (!replaying.setLastModified(System.currentTimeMillis())) {
+ log.warn("[audit] 璁ら鍚庡埛鏂� spool 绉熺害 mtime 澶辫触锛堟渶鍧忓悗鏋�=閲嶅閲嶆斁锛屽箓绛夊惛鏀讹級: {}",
+ replaying.getName());
+ }
+ renamed[0] = true;
+ });
+ if (!lockRun) {
+ // FileLock 琚埆鐨勮繘绋嬫寔鏈� 鈫� 鏈疆璺宠繃璇ユ枃浠讹紝涓嬭疆鍐嶈棰�
+ return;
+ }
+ if (!renamed[0]) {
+ // 閿佸唴閲嶉獙鏈繃锛堝凡琚粬 owner 鍙栬蛋/鍒犻櫎锛屾垨 .jsonl 浠嶆椿璺冿級鈫� 鏈疆璺宠繃锛屼笅杞啀鍒�
+ return;
+ }
+ } catch (IOException moveFail) {
+ // 鏀瑰悕澶辫触锛堟琚粬杩涚▼/浠栬疆鍙栬蛋锛夆啋 璺宠繃锛屼笅杞啀璇�
+ log.warn("[audit] spool 鏂囦欢鍗犵敤鏀瑰悕澶辫触锛岃烦杩�: {}", src.getName());
+ return;
+ }
+
+ // 閿佸崗璁复鐣屽尯澶栵紙readAllLines锛夛細璁ら鍚� .replaying 宸茶鏈繘绋嬬嫭鍗狅紝璇诲彇涓嶉樆濉� writer
+ List<String> lines;
+ try {
+ lines = Files.readAllLines(replaying.toPath(), StandardCharsets.UTF_8);
+ } catch (IOException readFail) {
+ // 璇诲け璐ヤ笉鍒� .replaying锛堥伩鍏嶈澶辫触鍗充涪鏁版嵁锛夛紝鐣欏緟浜哄伐/涓嬫杩涚▼澶勭悊
+ log.error("[audit] spool 璇诲彇澶辫触锛屼繚鐣� .replaying: {}", replaying.getName(), readFail);
+ return;
+ }
+
+ // 閿佸崗璁复鐣屽尯澶栵紙閫愯 deliverOnce=缃戠粶鎱㈡搷浣滐級锛氭妸琛屽垎绫讳负 remaining锛堟姇閫掑け璐ュ緟閲嶈瘯锛�
+ // 涓� badLines锛堝弽搴忓垪鍖栧け璐ュ緟闅旂锛夛紝姝ゅ惊鐜笉鍋氫换浣� spool 鏂囦欢鍐欐搷浣�
+ List<String> remaining = new ArrayList<>();
+ List<String> badLines = new ArrayList<>();
+ for (int i = 0; i < lines.size(); i++) {
+ String line = lines.get(i);
+ if (line == null || line.trim().isEmpty()) {
+ continue;
+ }
+ AuditEventDTO event;
+ try {
+ event = JsonUtil.getJsonToBean(line, AuditEventDTO.class);
+ } catch (Exception parseFail) {
+ // Codex 鎵规浜� #3锛氬潖琛岀暀寰呮敹灏炬壒閲忛殧绂诲埌 .bad锛堜笉杩� remaining銆佷笉闈欓粯涓級銆�
+ // Codex 鎵规浜� #4锛氭棩蹇楀彧璁版枃浠跺悕/琛屽彿/琛岄暱/鍐呭 SHA-256 鍓� 12 浣� + 寮傚父绫诲悕锛岀粷涓嶈緭鍑鸿姝ｆ枃
+ // 锛坧arseFail 鍙彇绫诲悕鈥斺�擩ackson 寮傚父姝ｆ枃鍙兘鍐呭祵婧愮墖娈碉紝鏁呬笉鏁翠綋鎵撳嵃锛夈��
+ badLines.add(line);
+ log.error("[audit] spool 鍧忚妫�鍑哄緟闅旂: file={} lineNo={} lineLen={} sha256_12={} cause={}",
+ replaying.getName(), i + 1, line.length(), sha256Prefix12(line),
+ parseFail.getClass().getName());
+ continue;
+ }
+ try {
+ publisher.deliverOnce(event); // 鎴愬姛鈫掍涪寮冭琛岋紱澶辫触鈫掍繚鐣欏緟涓嬭疆
+ } catch (Exception deliverFail) {
+ // 濂借鎶曢�掑け璐ワ紙AuditDeliveryException锛夆啋 淇濈暀寰呬笅杞紝琛屼负涓嶅彉
+ remaining.add(line);
+ }
+ }
+
+ // 閿佸崗璁复鐣屽尯 4锛堟敹灏撅級锛氬啓 .bad / 鍐� .retry / 鍒� .replaying 鍦ㄩ攣鍐咃紙鍚勮嚜蹇搷浣滐級銆�
+ // 鐢� writer 璇箟 lockAndRun锛堝缁堟墽琛岋級锛氭姇閫掑凡瀹屾垚锛屾敹灏惧繀椤昏惤鍦帮紝閬垮厤鐣� .replaying 琚噸澶嶉噸鏀�
+ try {
+ spoolLock.lockAndRun(() -> finishReplay(src, replaying, remaining, badLines));
+ } catch (IOException writeFail) {
+ log.error("[audit] spool 閲嶆斁鏀跺熬鍐欑洏澶辫触锛屼繚鐣� .replaying: {}", replaying.getName(), writeFail);
+ }
+ }
+
+ /**
+ * 閿佸崗璁复鐣屽尯 4锛堟敹灏撅級锛氭壒閲忓啓鍧忚 {@code .bad}銆佸啓鍓╀綑琛� {@code .retry}銆佸垹 {@code .replaying}銆�
+ * 鍏ㄧ▼鐢� {@link AuditSpoolLock#lockAndRun} 淇濇姢锛屼粎鍋氬揩閫熸枃浠舵搷浣溿��
+ *
+ * <p>Codex 鎵规浜� #3锛氬潖琛屾壒閲忛殧绂诲埌 {@code <绋冲畾 base>.<UUID>.bad}锛堜笉閲嶆斁銆佷笉闈欓粯涓級锛涘啓 {@code .bad}
+ * 澶辫触鍒欒鎵瑰潖琛岄��鍥� remaining锛堝畞鍙噸璇曚篃涓嶄涪锛夈�傚墿浣欒锛堝惈鎶曢�掑け璐� + 闅旂澶辫触閫�鍥炵殑鍧忚锛夐潪绌哄垯鍐�
+ * {@code <绋冲畾 base>.<UUID>.retry}锛堢粓楠屽洓杞憿锛氱粷涓嶅啓鍥炴椿鍔ㄥ師鍚嶏級锛屽惁鍒欏垹 {@code .replaying}锛涙湁 remaining
+ * 鏃朵篃鍦� {@code .retry} 鍐欐垚鍔熷悗鍒� {@code .replaying}锛堜笉鐣欏鍎匡級銆�
+ *
+ * @throws IOException 鍐� {@code .retry} 鎴栧垹 {@code .replaying} 澶辫触鈥斺�旇皟鐢ㄦ柟鎹淇濈暀 {@code .replaying} 寰呬笅杞�
+ */
+ private void finishReplay(File src, File replaying, List<String> remaining, List<String> badLines)
+ throws IOException {
+ List<String> retryLines = new ArrayList<>(remaining);
+ if (!badLines.isEmpty()) {
+ try {
+ writeNewSpoolFile(src, ".bad", badLines);
+ } catch (IOException badFail) {
+ // 鍐� .bad 澶辫触锛氬潖琛岄��鍥為噸璇曪紙瀹佸彲閲嶈瘯涔熶笉涓級锛涙棩蹇椾笉鍚鏂囷紝鍙鏂囦欢鍚�/鏉℃暟
+ retryLines.addAll(badLines);
+ log.warn("[audit] spool 鍧忚闅旂鍐欑洏澶辫触锛岄��鍥為噸璇�: file={} count={}",
+ replaying.getName(), badLines.size(), badFail);
+ }
+ }
+ if (retryLines.isEmpty()) {
+ Files.deleteIfExists(replaying.toPath()); // 鍏ㄩ儴鎴愬姛锛氬垹 .replaying
+ } else {
+ // 閮ㄥ垎澶辫触锛氬墿浣欒鍐� <绋冲畾 base>.<UUID>.retry 鏂版枃浠讹紝缁濅笉鍐欏洖娲诲姩鍘熷悕锛堢粓楠屽洓杞憿锛�
+ writeNewSpoolFile(src, ".retry", retryLines);
+ Files.deleteIfExists(replaying.toPath()); // .retry 鍐欐垚鍔熷悗鍒� .replaying锛屼笉鐣欏鍎�
+ }
+ }
+
+ /**
+ * 鎶� {@code lines} 鍐欏叆涓�涓柊寤虹殑 {@code <绋冲畾 base>.<UUID><suffix>} 鏂囦欢锛坽@code .retry} 鎴� {@code .bad}锛夈��
+ * 鍓嶇紑鍏堢粡 {@link #stableBaseName(String)} 褰掍竴锛屽啀鎷兼伆濂戒竴灞傛柊 UUID锛屾枃浠跺悕闀垮害鏈夌晫锛堝唴瀹� Important 淇锛夛紱
+ * {@link StandardOpenOption#CREATE_NEW} 淇濊瘉涓嶈鐩栨棦鏈夋枃浠讹紝UUID 浣挎挒鍚嶆鐜囧彲蹇界暐銆�
+ */
+ private void writeNewSpoolFile(File src, String suffix, List<String> lines) throws IOException {
+ File out = new File(src.getParentFile(),
+ stableBaseName(src.getName()) + "." + UUID.randomUUID() + suffix);
+ StringBuilder sb = new StringBuilder();
+ for (String line : lines) {
+ sb.append(line).append('\n');
+ }
+ Files.write(out.toPath(), sb.toString().getBytes(StandardCharsets.UTF_8),
+ StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);
+ }
+
+ /**
+ * Codex 鎵规浜� #4锛氬潖琛屽唴瀹圭殑 SHA-256 鍗佸叚杩涘埗<b>鍓� 12 浣�</b>锛屼緵鏃ュ織瀹氫綅鍧忚鑰屼笉娉勯湶姝ｆ枃
+ * 锛坰pec "ERROR 鏃ュ織涓嶅惈瀛楁鍊兼鏂�"锛夈�係HA-256 涓� JDK 鏍囬厤绠楁硶锛寋@link NoSuchAlgorithmException}
+ * 鐞嗚涓嶅彲杈撅紱涓囦竴涓嶅彲杈惧垯鍥炶惤鍒板浐瀹氬崰浣嶄覆锛岀粷涓嶅洖閫�鎴愭墦鍗版鏂囥��
+ */
+ private static String sha256Prefix12(String line) {
+ try {
+ MessageDigest md = MessageDigest.getInstance("SHA-256");
+ byte[] digest = md.digest(line.getBytes(StandardCharsets.UTF_8));
+ StringBuilder hex = new StringBuilder(12);
+ for (byte b : digest) {
+ hex.append(Character.forDigit((b >> 4) & 0xF, 16));
+ hex.append(Character.forDigit(b & 0xF, 16));
+ if (hex.length() >= 12) {
+ break;
+ }
+ }
+ return hex.substring(0, 12);
+ } catch (NoSuchAlgorithmException e) {
+ return "sha256-unavailable";
+ }
+ }
+}
--
Gitblit v1.8.0