fix: 集群日志重复条目 — 本地 Append 同步推进 Synced 水位

根因: 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

@@ -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
 }
 

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)
+	}
+}