From 57950a86db3ec15a7983336fa8eff824c80bd792 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Wed, 26 Aug 2026 13:30:12 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E9=9B=86=E7=BE=A4=E6=97=A5=E5=BF=97?= =?UTF-8?q?=E9=87=8D=E5=A4=8D=E6=9D=A1=E7=9B=AE=20=E2=80=94=20=E6=9C=AC?= =?UTF-8?q?=E5=9C=B0=20Append=20=E5=90=8C=E6=AD=A5=E6=8E=A8=E8=BF=9B=20Syn?= =?UTF-8?q?ced=20=E6=B0=B4=E4=BD=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因: Append 只推进 Seq 不推进 Synced。本地追加 seq=N 后 Synced 落后， 下一轮收到对端全量附带(含自己的 seq=N)时 ApplyDelta 视为未见过而再次追加 → 审计日志出现同 seq 重复行（实测 .30 出现 seq=12/13/14 各两条）。 修复: Append 将 Synced 一并推进到新条目 seq——本地条目天然已同步， 对端回传的同一 seq 被 watermark 幂等跳过。 测试: TestLocalAppendNotDuplicatedByPeerFullLog 实测: 三台各触发 stop+start 制造多条事件 → 三台均 26 条 / 唯一seq 26 / 重复 0 / 水位一致 (seq=26) --- internal/cluster/ring_log.go | 9 +++++++++ internal/cluster/ring_log_test.go | 19 +++++++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/internal/cluster/ring_log.go b/internal/cluster/ring_log.go index ed03e4a..dd6e04c 100644 --- a/internal/cluster/ring_log.go +++ b/internal/cluster/ring_log.go @@ -50,6 +50,12 @@ func NewClusterLog() *ClusterLog { } // Append adds an entry with the next sequence number (caller supplies kind/data). +// The entry is treated as already-synced (Synced advanced with Seq): a locally +// appended entry exists on this node and will be broadcast; the next full-log +// attachment from a peer must NOT re-apply it as a duplicate. Without this, +// a node that appends seq=N while its Synced lags behind would later receive +// its own seq=N in a peer's full attachment and ApplyDelta would re-add it +// (duplicate rows in the audit log). func (l *ClusterLog) Append(node, kind string, data any) (LogEntry, error) { l.mu.Lock() defer l.mu.Unlock() @@ -61,6 +67,9 @@ func (l *ClusterLog) Append(node, kind string, data any) (LogEntry, error) { } e := LogEntry{Seq: l.Seq, Node: node, Kind: kind, Data: raw, At: time.Now().Unix()} l.Log = append(l.Log, e) + if e.Seq > l.Synced { + l.Synced = e.Seq + } return e, nil } diff --git a/internal/cluster/ring_log_test.go b/internal/cluster/ring_log_test.go index 6f57dfa..0a4173d 100644 --- a/internal/cluster/ring_log_test.go +++ b/internal/cluster/ring_log_test.go @@ -134,3 +134,22 @@ func TestApplyDeltaIdempotentOnFullResend(t *testing.T) { t.Fatalf("wm=%d len=%d, want 3/3", wm, len(dst.Snapshot())) } } + +// TestLocalAppendNotDuplicatedByPeerFullLog: a locally appended entry (seq=N) +// must NOT be re-applied when a peer's full-log attachment carries the same +// seq — this was the source of duplicate rows in the audit log. +func TestLocalAppendNotDuplicatedByPeerFullLog(t *testing.T) { + var mine ClusterLog + if _, err := mine.Append("n1", LogForwardAdd, map[string]string{"local": "web"}); err != nil { + t.Fatal(err) + } + // Simulate the pre-fix bug window: peer attaches [1..N] including our N. + peerHas := mine.Snapshot() + before := len(mine.Snapshot()) + if _, err := mine.ApplyDelta(peerHas); err != nil { + t.Fatalf("apply own entry back: %v", err) + } + if got := len(mine.Snapshot()); got != before { + t.Fatalf("log grew from %d to %d after re-applying own entry", before, got) + } +}