fix:threatip sync bug

firewall/ipset_windows.go

@@ -3,9 +3,12 @@
 package firewall
 
 import (
+	"SamWaf/common/zlog"
+	"context"
 	"fmt"
 	"os/exec"
 	"strings"
+	"time"
 )
 
 // 说明：Windows 无 ipset 概念，但 netsh 单条规则的 remoteip= 支持逗号分隔多个 IP/CIDR。
@@ -18,8 +21,15 @@ const (
 	setRulePrefix = "SamWAF_Set_"
 	// maxRemoteIPChars 单条规则 remoteip= 值的字符上限(留足命令行余量)
 	maxRemoteIPChars = 3000
-	// maxShardScan DestroyIPSet 时扫描删除分片规则的序号上限(兜底防死循环)
-	maxShardScan = 200000
+	// maxShardScan DestroyIPSet 时扫描删除分片规则的序号上限(兜底防死循环)。
+	// 单分片 maxRemoteIPChars=3000 字符 ≈ 200 条 IP，4096 个分片 ≈ 80 万条，远超实际订阅规模。
+	maxShardScan = 4096
+	// destroyMissStreak 连续多少个序号"未命中"就认为分片已删完
+	destroyMissStreak = 3
+	// destroyBudget DestroyIPSet 整体时间预算，超了就放弃剩余扫描(防止拖住调用方持有的锁)
+	destroyBudget = 30 * time.Second
+	// netshCmdTimeout 单条 netsh 命令的超时，防个别调用挂死
+	netshCmdTimeout = 5 * time.Second
 )
 
 // SupportsIPSet 报告当前环境是否可用批量封禁(Windows 下等价于系统防火墙可用，带 30s 缓存)
@@ -68,21 +78,66 @@ func (fw *FireWallEngine) FlushIPSet(setName string) error {
 	return fw.DestroyIPSet(setName)
 }
 
-// DestroyIPSet 逐序号删除该集合的所有分片规则，直到某序号无匹配为止
+// DestroyIPSet 逐序号删除该集合的所有分片规则，直到连续若干个序号都删不动为止。
+//
+// 停止条件刻意不再依赖 netsh 的本地化提示文案：老实现只认英文
+// "No rules match the specified criteria" 和简中"没有与指定标准相匹配的规则"，
+// 一旦 netsh 输出别的内容(非中英文系统、非管理员运行时的"请求的操作需要提升"、
+// 防火墙服务异常等)，break 就永远不会触发，会 fork 二十万次 netsh 空转数小时，
+// 并把调用方持有的同步锁一起焊死。现在改为：**任何一次"没有删成功"都算未命中**，
+// 连续 destroyMissStreak 次未命中即停，另加整体时间预算兜底。宁可少删，绝不空转。
 func (fw *FireWallEngine) DestroyIPSet(setName string) error {
+	// 系统防火墙不可用时一次 fork 都不做(netsh 缺失 / 防火墙未启用)
+	if !fw.SupportsIPSet() {
+		return nil
+	}
+
+	start := time.Now()
+	deleted, miss := 0, 0
 	for seq := 0; seq < maxShardScan; seq++ {
-		ruleName := setShardRuleName(setName, seq)
-		cmd := exec.Command("netsh", "advfirewall", "firewall", "delete", "rule", "name="+ruleName)
-		_, output := fw.executeCommand(cmd)
-		if strings.Contains(output, "No rules match the specified criteria") ||
-			strings.Contains(output, "没有与指定标准相匹配的规则") {
-			// 连续遇到无匹配即认为分片已删完
+		if time.Since(start) > destroyBudget {
+			zlog.Warn("清理防火墙分片规则超出时间预算，已提前结束",
+				"set", setName, "scanned", seq, "deleted", deleted,
+				"elapsed", time.Since(start).Round(time.Millisecond).String())
+			break
+		}
+		if fw.deleteShardRule(setName, seq) {
+			deleted++
+			miss = 0
+			continue
+		}
+		miss++
+		if miss >= destroyMissStreak {
 			break
 		}
 	}
+	if deleted > 0 {
+		zlog.Debug("清理防火墙分片规则完成", "set", setName, "deleted", deleted,
+			"elapsed", time.Since(start).Round(time.Millisecond).String())
+	}
 	return nil
 }
 
+// deleteShardRule 删除单个分片规则，返回是否确实删掉了。
+// 只有 netsh 明确报告删除成功才算命中；出错、超时、输出看不懂一律当作"没有这个分片"。
+func (fw *FireWallEngine) deleteShardRule(setName string, seq int) bool {
+	ctx, cancel := context.WithTimeout(context.Background(), netshCmdTimeout)
+	defer cancel()
+
+	ruleName := setShardRuleName(setName, seq)
+	cmd := exec.CommandContext(ctx, "netsh", "advfirewall", "firewall", "delete", "rule", "name="+ruleName)
+	err, output := fw.executeCommand(cmd)
+	if err != nil {
+		return false
+	}
+	// 已知的"没有这条规则"文案优先判定：部分环境下 netsh 会在该提示后仍打印"确定。"
+	if strings.Contains(output, "No rules match the specified criteria") ||
+		strings.Contains(output, "没有与指定标准相匹配的规则") {
+		return false
+	}
+	return strings.Contains(output, "已删除") || strings.Contains(output, "Deleted")
+}
+
 // shardIPsByLen 把 IP/CIDR 列表按逗号拼接、累计长度不超过 maxLen 切分为多个分片字符串
 func shardIPsByLen(ips []string, maxLen int) []string {
 	var shards []string

model/threat_ip_channel.go

@@ -21,6 +21,11 @@ type ThreatIPChannel struct {
 	LastCount    int    `json:"last_count"`                  // 上次快照收录条数
 	LastStatus   string `gorm:"size:255" json:"last_status"` // 上次同步结果：ok 或错误摘要
 	Remarks      string `gorm:"size:500" json:"remarks"`     // 备注
+
+	// 以下为运行时内存态字段(gorm:"-" 不落库，无需迁移)，仅供列表接口回显"同步中"，
+	// 让前端能在后台拉取期间给出反馈并自动轮询，而不是点完什么都看不到。
+	Syncing       bool  `gorm:"-" json:"syncing"`         // 该渠道当前是否有同步在进行
+	SyncStartedAt int64 `gorm:"-" json:"sync_started_at"` // 本次同步开始时间戳(秒)，Syncing 为 false 时无意义
 }
 
 // TableName 表名

service/waf_service/waf_threat_ip_service.go

@@ -1,9 +1,11 @@
 package waf_service
 
 import (
+	"SamWaf/common/tasklog"
 	"SamWaf/common/uuid"
 	"SamWaf/common/zlog"
 	"SamWaf/customtype"
+	"SamWaf/enums"
 	"SamWaf/firewall"
 	"SamWaf/global"
 	"SamWaf/innerbean"
@@ -14,8 +16,10 @@ import (
 	"SamWaf/waftask/threatip"
 	"errors"
 	"fmt"
+	"runtime/debug"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -23,6 +27,67 @@ import (
 type WafThreatIPService struct {
 	fw     firewall.FireWallEngine
 	syncMu sync.Mutex // 串行化同步，避免多渠道并发拉取造成内核/内存压力
+
+	// inflight 记录"当前正在处理(含排队等锁)的渠道 Code -> 开始时间戳(秒)"。
+	// 纯内存态(重启即清空)，只用于列表接口回显"同步中"，让前端能轮询到结果。
+	inflight sync.Map
+	// lockHolder 当前持有 syncMu 的操作描述，用于在别人拿不到锁时说清楚"是谁占着"
+	lockHolder atomic.Value
+}
+
+// lockWaitOnDemand 等待 syncMu 的上限。
+// 超过就明确报错返回，而不是像以前那样无限期静默排队(用户会看到"提示已开始"却永远没结果)。
+// 取 30s：既能容纳"连着点两个渠道同步"这种正常排队，又不会让失败无限期悬着。
+const lockWaitOnDemand = 30 * time.Second
+
+// syncTriggerManual / syncTriggerSchedule 同步触发方式，进日志与 last_status 便于区分
+const (
+	syncTriggerManual   = "手动"
+	syncTriggerSchedule = "定时"
+)
+
+// tryLockSync 在 wait 时间内尝试获取同步锁，holder 描述本次操作(用于让后来者知道是谁占着)。
+// 用轮询式 TryLock 而非阻塞 Lock：这条链路上任何一次卡死都会把后续所有同步永久堵住，
+// 宁可明确失败并留下日志，也不要无限期静默排队。
+func (r *WafThreatIPService) tryLockSync(wait time.Duration, holder string) bool {
+	deadline := time.Now().Add(wait)
+	for {
+		if r.syncMu.TryLock() {
+			r.lockHolder.Store(holder)
+			return true
+		}
+		if time.Now().After(deadline) {
+			return false
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+}
+
+// unlockSync 释放同步锁并清掉持有者标记
+func (r *WafThreatIPService) unlockSync() {
+	r.lockHolder.Store("")
+	r.syncMu.Unlock()
+}
+
+// busyHint 描述当前占着同步锁的操作，用于错误信息与日志
+func (r *WafThreatIPService) busyHint() string {
+	if h, _ := r.lockHolder.Load().(string); h != "" {
+		return h
+	}
+	return "有其它订阅落地操作正在进行"
+}
+
+// safeGo 起一个带 panic 兜底的后台 goroutine。
+// 以前这里是裸 go func()：一次 panic 会直接崩掉整个 SamWaf 进程，且什么线索都不留。
+func safeGo(name string, fn func()) {
+	go func() {
+		defer func() {
+			if e := recover(); e != nil {
+				zlog.Error(fmt.Sprintf("威胁情报订阅后台任务[%s]异常: %v 调试信息:%s", name, e, debug.Stack()))
+			}
+		}()
+		fn()
+	}()
 }
 
 var WafThreatIPServiceApp = new(WafThreatIPService)
@@ -100,21 +165,28 @@ func (r *WafThreatIPService) ModifyApi(req request.WafThreatIPChannelEditReq) er
 	code := bean.Code
 	enable := req.Enable
 	name := req.Name
-	go func() {
-		r.applyEnableState(code, landTarget, enable)
-		r.notifyOpResult("威胁情报订阅保存生效", name, nil)
-	}()
+	safeGo("保存生效:"+code, func() {
+		err := r.applyEnableState(code, landTarget, enable)
+		r.notifyOpResult("威胁情报订阅保存生效", name, err)
+	})
 	return nil
 }
 
 // applyEnableState 依据启用状态把某渠道落地到位：系统集合的增删 + 重建 WAF 应用层并集。
 // 供编辑保存后异步调用；串行化(syncMu)避免与同步任务并发改动内核/内存态。
-func (r *WafThreatIPService) applyEnableState(code, landTarget string, enable int) {
-	r.syncMu.Lock()
-	defer r.syncMu.Unlock()
+func (r *WafThreatIPService) applyEnableState(code, landTarget string, enable int) error {
+	start := time.Now()
+	if !r.tryLockSync(lockWaitOnDemand, fmt.Sprintf("渠道[%s]保存生效", code)) {
+		msg := "订阅落地生效跳过：" + r.busyHint()
+		zlog.Warn("威胁情报订阅落地生效未能获取同步锁", "channel", code, "detail", r.busyHint())
+		return errors.New(msg)
+	}
+	defer r.unlockSync()
+
+	zlog.Info("威胁情报订阅落地生效开始", "channel", code, "land", landTarget, "enable", enable)
 	if enable == 0 {
 		// 停用：清理系统集合(WAF 并集在下方重建时自动排除该渠道)
-		r.fw.DestroyIPSet(setNameForChannel(code))
+		r.destroySystemSet(code)
 	} else if landTarget == model.ThreatLandSystem || landTarget == model.ThreatLandBoth {
 		// 启用且落地含系统层：把快照重新灌回系统集合(重启/停用会丢，需按需回灌)
 		if r.fw.SupportsIPSet() {
@@ -123,12 +195,29 @@ func (r *WafThreatIPService) applyEnableState(code, landTarget string, enable in
 					zlog.Warn("启用渠道系统层落地失败", "channel", code, "error", err.Error())
 				}
 			}
+		} else {
+			zlog.Warn("当前环境不支持 ipset 批量封禁，跳过系统层落地", "channel", code)
 		}
 	} else {
 		// 启用但落地不含系统层：清理可能残留的系统集合
-		r.fw.DestroyIPSet(setNameForChannel(code))
+		r.destroySystemSet(code)
 	}
 	r.RebuildWAFUnion()
+	zlog.Info("威胁情报订阅落地生效完成", "channel", code, "elapsed", time.Since(start).Round(time.Millisecond).String())
+	return nil
+}
+
+// destroySystemSet 清理某渠道在系统防火墙上的集合。
+// 只在系统防火墙确实可用时才动手：纯 WAF 落地的渠道从没往系统层写过东西，
+// 无条件调用只会白白 fork 一堆 iptables/netsh(Windows 上还可能长时间空转并占着同步锁)。
+func (r *WafThreatIPService) destroySystemSet(code string) {
+	if !r.fw.SupportsIPSet() {
+		zlog.Debug("系统防火墙不可用，跳过系统层集合清理", "channel", code)
+		return
+	}
+	if err := r.fw.DestroyIPSet(setNameForChannel(code)); err != nil {
+		zlog.Warn("清理系统层集合失败", "channel", code, "error", err.Error())
+	}
 }
 
 // notifyOpResult 把某异步操作结果推送到管理端通知中心(WebSocket 实时)。
@@ -164,13 +253,18 @@ func (r *WafThreatIPService) DelApi(id string) error {
 		return err
 	}
 	// 异步落地：撤系统集合 + 重建并集
-	go func() {
-		r.syncMu.Lock()
-		r.fw.DestroyIPSet(setNameForChannel(code))
+	safeGo("删除生效:"+code, func() {
+		if !r.tryLockSync(lockWaitOnDemand, fmt.Sprintf("渠道[%s]删除生效", code)) {
+			zlog.Warn("威胁情报订阅删除生效未能获取同步锁", "channel", code, "detail", r.busyHint())
+			r.notifyOpResult("威胁情报订阅删除生效", name, errors.New("落地生效跳过："+r.busyHint()+"；下次同步或重启后会自动对齐"))
+			return
+		}
+		defer r.unlockSync() // 用 defer：以前是手动 Unlock，中途 panic 会永久泄漏锁，之后所有同步全卡死
+		r.destroySystemSet(code)
 		r.RebuildWAFUnion()
-		r.syncMu.Unlock()
+		zlog.Info("威胁情报订阅删除生效完成", "channel", code)
 		r.notifyOpResult("威胁情报订阅删除生效", name, nil)
-	}()
+	})
 	return nil
 }
 
@@ -189,10 +283,21 @@ func (r *WafThreatIPService) GetListApi(req request.WafThreatIPChannelSearchReq)
 		if c := r.snapshotCount(list[i].Code); c >= 0 {
 			list[i].LastCount = c
 		}
+		// 回填内存态"同步中"，前端据此显示进行中并自动轮询
+		list[i].Syncing, list[i].SyncStartedAt = r.syncingOf(list[i].Code)
 	}
 	return list, total, err
 }
 
+// syncingOf 查某渠道当前是否正在同步，以及本次同步的开始时间戳
+func (r *WafThreatIPService) syncingOf(code string) (bool, int64) {
+	if v, ok := r.inflight.Load(code); ok {
+		startedAt, _ := v.(int64)
+		return true, startedAt
+	}
+	return false, 0
+}
+
 // ---------------- 订阅落地只读浏览(方案三：原页"订阅来源"Tab) ----------------
 
 // LandedChannelSummary 某渠道在指定层的落地汇总
@@ -201,10 +306,12 @@ type LandedChannelSummary struct {
 	Name          string `json:"name"`
 	LandTarget    string `json:"land_target"`
 	Enable        int    `json:"enable"`
-	Count         int    `json:"count"`          // 实际生效(已落地)条数：启用=快照条数，停用=0(已从防火墙/WAF移除)
-	SnapshotCount int    `json:"snapshot_count"` // 快照收录条数(不论启用与否，供停用时提示"再启用可回灌")
-	LastSyncAt    int64  `json:"last_sync_at"`   // 上次同步时间戳(秒)
-	LastStatus    string `json:"last_status"`    // 上次同步结果
+	Count         int    `json:"count"`           // 实际生效(已落地)条数：启用=快照条数，停用=0(已从防火墙/WAF移除)
+	SnapshotCount int    `json:"snapshot_count"`  // 快照收录条数(不论启用与否，供停用时提示"再启用可回灌")
+	LastSyncAt    int64  `json:"last_sync_at"`    // 上次同步时间戳(秒)
+	LastStatus    string `json:"last_status"`     // 上次同步结果
+	Syncing       bool   `json:"syncing"`         // 当前是否有同步在进行(内存态)
+	SyncStartedAt int64  `json:"sync_started_at"` // 本次同步开始时间戳(秒)
 }
 
 // GetLandedSummary 返回落地到指定层的各渠道汇总。
@@ -226,6 +333,7 @@ func (r *WafThreatIPService) GetLandedSummary(land string) []LandedChannelSummar
 		if ch.Enable == 0 {
 			landed = 0
 		}
+		syncing, syncStartedAt := r.syncingOf(ch.Code)
 		out = append(out, LandedChannelSummary{
 			Code:          ch.Code,
 			Name:          ch.Name,
@@ -235,6 +343,8 @@ func (r *WafThreatIPService) GetLandedSummary(land string) []LandedChannelSummar
 			SnapshotCount: snapCnt,
 			LastSyncAt:    ch.LastSyncAt,
 			LastStatus:    ch.LastStatus,
+			Syncing:       syncing,
+			SyncStartedAt: syncStartedAt,
 		})
 	}
 	return out
@@ -308,47 +418,100 @@ func (r *WafThreatIPService) SyncByIdApi(id string) error {
 	if err := global.GWAF_LOCAL_DB.Where("id = ?", id).First(&bean).Error; err != nil {
 		return errors.New("记录不存在")
 	}
-	go func() {
-		err := r.SyncChannel(bean)
+	zlog.Info("威胁情报订阅收到手动同步请求", "channel", bean.Code, "name", bean.Name,
+		"url", bean.URL, "parser", bean.ParserType, "land", bean.LandTarget)
+	safeGo("手动同步:"+bean.Code, func() {
+		err := r.syncChannelWithTrigger(bean, syncTriggerManual)
 		r.notifyOpResult("威胁情报订阅同步", bean.Name, err)
-	}()
+	})
 	return nil
 }
 
 // SyncChannel 拉取→解析→与上次快照比对→(突变保护)→保存快照→落地(系统 ipset + WAF 并集)。
 func (r *WafThreatIPService) SyncChannel(ch model.ThreatIPChannel) error {
-	r.syncMu.Lock()
-	defer r.syncMu.Unlock()
+	return r.syncChannelWithTrigger(ch, syncTriggerSchedule)
+}
 
-	raw, err := threatip.Fetch(ch.URL)
+// syncChannelWithTrigger 同 SyncChannel，额外带上触发方式(手动/定时)用于日志与状态回显。
+//
+// 这条链路以前失败时是完全静默的(四个失败分支都只写 DB、不打日志，失败通知又只走一次性 WebSocket)，
+// 用户点完"立即同步"只能看到"已开始"然后什么都没有。所以这里每个阶段都留了日志与耗时，
+// 并把手动触发也挂进任务日志，方便在"任务日志"页直接看到全过程。
+func (r *WafThreatIPService) syncChannelWithTrigger(ch model.ThreatIPChannel, trigger string) error {
+	// 让本 goroutine 的 zlog 输出同时落到任务日志文件(与定时任务同一个任务名)。
+	// 定时路径下 ExecuteTask 已经设过任务上下文，这里就不能重复设——否则同步完第一个渠道时
+	// 的 defer Clear 会把整个 SyncAllDue 剩余过程的任务日志上下文一起清掉。
+	if _, ok := tasklog.GetCurrentTask(); !ok {
+		tasklog.SetCurrentTask(enums.TASK_THREAT_IP_SYNC)
+		defer tasklog.ClearCurrentTask()
+	}
+
+	start := time.Now()
+
+	// 先标记"同步中"再去抢锁：排队等锁的这段时间前端也该看到进行中，
+	// 否则轮询会以为已经结束而提前收工。无论从哪个分支返回都会清掉。
+	r.inflight.Store(ch.Code, start.Unix())
+	defer r.inflight.Delete(ch.Code)
+
+	if !r.tryLockSync(lockWaitOnDemand, fmt.Sprintf("渠道[%s]%s同步", ch.Code, trigger)) {
+		msg := fmt.Sprintf("已跳过(%s触发)：等待 %s 仍未轮到，%s", trigger, lockWaitOnDemand, r.busyHint())
+		zlog.Warn("威胁情报订阅同步未能获取同步锁", "channel", ch.Code, "trigger", trigger, "detail", r.busyHint())
+		r.markSyncFail(ch.Id, msg)
+		return errors.New(msg)
+	}
+	defer r.unlockSync()
+
+	zlog.Info("威胁情报订阅同步开始", "channel", ch.Code, "trigger", trigger, "url", ch.URL, "parser", ch.ParserType,
+		"waited", time.Since(start).Round(time.Millisecond).String())
+
+	raw, stat, err := threatip.FetchWithStat(ch.URL)
 	if err != nil {
-		r.markSyncStatus(ch.Id, "拉取失败: "+err.Error(), 0)
+		zlog.Error(fmt.Sprintf("威胁情报订阅拉取失败 channel=%s trigger=%s error=%s", ch.Code, trigger, err.Error()))
+		r.markSyncFail(ch.Id, "拉取失败: "+err.Error())
 		return err
 	}
+	zlog.Info("威胁情报订阅拉取完成", "channel", ch.Code, "status", stat.StatusCode,
+		"bytes", stat.Bytes, "elapsed", stat.Elapsed.Round(time.Millisecond).String())
+
+	parseStart := time.Now()
 	parseRes, err := threatip.ParseByType(ch.ParserType, strings.NewReader(string(raw)), ch.Threshold)
 	if err != nil {
-		r.markSyncStatus(ch.Id, "解析失败: "+err.Error(), 0)
+		zlog.Error(fmt.Sprintf("威胁情报订阅解析失败 channel=%s parser=%s error=%s", ch.Code, ch.ParserType, err.Error()))
+		r.markSyncFail(ch.Id, "解析失败: "+err.Error())
 		return err
 	}
 	newIPs := parseRes.IPs
+	zlog.Info("威胁情报订阅解析完成", "channel", ch.Code, "valid", len(newIPs), "dropped", parseRes.Dropped,
+		"elapsed", time.Since(parseStart).Round(time.Millisecond).String())
+
+	// 源返回 200 但一条有效数据都没有(常见于返回了验证页/错误页)，按失败处理，避免把空快照当成功
+	if len(newIPs) == 0 {
+		msg := fmt.Sprintf("解析结果为空(丢弃非法行 %d，响应 %d 字节)，可能订阅源返回了非预期内容或解析格式选错", parseRes.Dropped, stat.Bytes)
+		zlog.Error(fmt.Sprintf("威胁情报订阅解析结果为空 channel=%s detail=%s", ch.Code, msg))
+		r.markSyncFail(ch.Id, msg)
+		return errors.New(msg)
+	}
 
 	// 载入上次快照做突变保护与 sha 比对
 	oldIPs, oldSha := r.loadSnapshot(ch.Code)
 	if len(oldIPs) > shrinkGuardMin && len(newIPs) < len(oldIPs)/shrinkGuardRatio {
 		msg := fmt.Sprintf("疑似源异常：新快照 %d 条 远少于上次 %d 条，已跳过应用(丢弃非法行 %d)", len(newIPs), len(oldIPs), parseRes.Dropped)
 		zlog.Warn("威胁情报订阅突变保护", "channel", ch.Code, "detail", msg)
-		r.markSyncStatus(ch.Id, msg, len(oldIPs))
+		r.markSyncFail(ch.Id, msg)
 		return errors.New(msg)
 	}
 
 	payload, sha, count, err := threatip.EncodeSnapshot(newIPs)
 	if err != nil {
-		r.markSyncStatus(ch.Id, "快照编码失败: "+err.Error(), 0)
+		zlog.Error(fmt.Sprintf("威胁情报订阅快照编码失败 channel=%s error=%s", ch.Code, err.Error()))
+		r.markSyncFail(ch.Id, "快照编码失败: "+err.Error())
 		return err
 	}
 	if sha == oldSha && oldSha != "" {
 		// 无变化：仅更新同步时间
-		r.markSyncStatus(ch.Id, "ok(无变化)", count)
+		zlog.Info("威胁情报订阅内容无变化，跳过落地", "channel", ch.Code, "count", count,
+			"elapsed", time.Since(start).Round(time.Millisecond).String())
+		r.markSyncOK(ch.Id, fmt.Sprintf("ok(无变化，%s触发，耗时%s)", trigger, time.Since(start).Round(time.Millisecond)), count)
 		return nil
 	}
 
@@ -356,29 +519,36 @@ func (r *WafThreatIPService) SyncChannel(ch model.ThreatIPChannel) error {
 
 	// 保存新快照(替换该渠道旧快照)
 	if err := r.saveSnapshot(ch.Code, payload, sha, count); err != nil {
-		r.markSyncStatus(ch.Id, "保存快照失败: "+err.Error(), 0)
+		zlog.Error(fmt.Sprintf("威胁情报订阅保存快照失败 channel=%s error=%s", ch.Code, err.Error()))
+		r.markSyncFail(ch.Id, "保存快照失败: "+err.Error())
 		return err
 	}
 
 	// 落地系统防火墙(该渠道私有集合，全量重建)
 	if ch.LandTarget == model.ThreatLandSystem || ch.LandTarget == model.ThreatLandBoth {
 		if r.fw.SupportsIPSet() {
+			landStart := time.Now()
 			if err := r.fw.RestoreIPSet(setNameForChannel(ch.Code), newIPs); err != nil {
 				zlog.Warn("威胁情报系统层落地失败", "channel", ch.Code, "error", err.Error())
+			} else {
+				zlog.Info("威胁情报系统层落地完成", "channel", ch.Code, "count", count,
+					"elapsed", time.Since(landStart).Round(time.Millisecond).String())
 			}
 		} else {
 			zlog.Warn("当前环境不支持 ipset 批量封禁，跳过系统层落地", "channel", ch.Code)
 		}
 	} else {
-		// 落地目标不含系统层：清理可能残留的系统集合
-		r.fw.DestroyIPSet(setNameForChannel(ch.Code))
+		// 落地目标不含系统层：清理可能残留的系统集合(系统防火墙不可用时内部会直接跳过)
+		r.destroySystemSet(ch.Code)
 	}
 
 	// 落地 WAF 应用层(重建全局并集)
 	r.RebuildWAFUnion()
 
-	r.markSyncStatus(ch.Id, fmt.Sprintf("ok(+%d/-%d，丢弃%d)", len(added), len(removed), parseRes.Dropped), count)
-	zlog.Info("威胁情报订阅同步完成", "channel", ch.Code, "count", count, "added", len(added), "removed", len(removed))
+	elapsed := time.Since(start).Round(time.Millisecond)
+	r.markSyncOK(ch.Id, fmt.Sprintf("ok(+%d/-%d，丢弃%d，%s触发，耗时%s)", len(added), len(removed), parseRes.Dropped, trigger, elapsed), count)
+	zlog.Info("威胁情报订阅同步完成", "channel", ch.Code, "trigger", trigger, "count", count,
+		"added", len(added), "removed", len(removed), "elapsed", elapsed.String())
 	return nil
 }
 
@@ -405,8 +575,10 @@ func (r *WafThreatIPService) RebuildWAFUnion() {
 	for ip := range uniq {
 		items = append(items, ip)
 	}
+	buildStart := time.Now()
 	ipset.SetGlobalThreatMatcher(ipset.BuildMatchSet(items))
-	zlog.Info("威胁情报 WAF 并集已重建", "total", len(items))
+	zlog.Info("威胁情报 WAF 并集已重建", "total", len(items),
+		"elapsed", time.Since(buildStart).Round(time.Millisecond).String())
 }
 
 // RestoreAllOnStartup 进程启动时重放：把各启用渠道的快照重新灌入系统 ipset(内存态，重启会丢)，
@@ -436,15 +608,18 @@ func (r *WafThreatIPService) SyncAllDue() {
 	var channels []model.ThreatIPChannel
 	global.GWAF_LOCAL_DB.Where("enable = 1").Find(&channels)
 	now := time.Now().Unix()
+	due := 0
 	for _, ch := range channels {
 		interval := int64(defaultInt(ch.IntervalHour, 24)) * 3600
 		if ch.LastSyncAt > 0 && now-ch.LastSyncAt < interval {
 			continue // 未到期
 		}
-		if err := r.SyncChannel(ch); err != nil {
+		due++
+		if err := r.syncChannelWithTrigger(ch, syncTriggerSchedule); err != nil {
 			zlog.Warn("威胁情报订阅同步失败", "channel", ch.Code, "error", err.Error())
 		}
 	}
+	zlog.Info("威胁情报订阅定时检查完成", "enabled", len(channels), "due", due)
 }
 
 // ---------------- 内部辅助 ----------------
@@ -484,14 +659,32 @@ func (r *WafThreatIPService) saveSnapshot(code string, payload []byte, sha strin
 	return global.GWAF_LOCAL_DB.Create(snap).Error
 }
 
-// markSyncStatus 更新渠道同步状态字段
-func (r *WafThreatIPService) markSyncStatus(id, status string, count int) {
-	if err := global.GWAF_LOCAL_DB.Model(&model.ThreatIPChannel{}).Where("id = ?", id).Updates(map[string]interface{}{
-		"LastSyncAt":  time.Now().Unix(),
-		"LastCount":   count,
-		"LastStatus":  truncateStatus(status),
-		"UPDATE_TIME": customtype.JsonTime(time.Now()),
-	}).Error; err != nil {
+// markSyncOK 同步成功后回写：刷新同步时间、收录条数与状态
+func (r *WafThreatIPService) markSyncOK(id, status string, count int) {
+	r.updateSyncFields(id, map[string]interface{}{
+		"LastSyncAt": time.Now().Unix(),
+		"LastCount":  count,
+		"LastStatus": truncateStatus(status),
+	})
+}
+
+// markSyncFail 同步失败后回写：只写状态原因。
+//
+// 刻意不动 LastSyncAt、LastCount：
+//   - 以前失败也把 LastSyncAt 刷成当前时间，SyncAllDue 的到期判断就把失败当成了成功，
+//     导致整整一个周期(默认 24h)不再重试——这正是用户说的"后面再也没成功"；
+//     不刷新之后，下一次每小时的定时任务会自然重试。
+//   - 以前失败还会把 LastCount 抹成 0，把上一次的真实条数丢掉。
+func (r *WafThreatIPService) markSyncFail(id, status string) {
+	r.updateSyncFields(id, map[string]interface{}{
+		"LastStatus": truncateStatus(status),
+	})
+}
+
+// updateSyncFields 统一回写同步相关字段(自动补 UPDATE_TIME)
+func (r *WafThreatIPService) updateSyncFields(id string, fields map[string]interface{}) {
+	fields["UPDATE_TIME"] = customtype.JsonTime(time.Now())
+	if err := global.GWAF_LOCAL_DB.Model(&model.ThreatIPChannel{}).Where("id = ?", id).Updates(fields).Error; err != nil {
 		zlog.Warn("威胁情报订阅状态回写失败", "id", id, "error", err.Error())
 	}
 }

waftask/threatip/fetch.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strings"
 	"time"
 )
 
@@ -13,25 +14,66 @@ var fetchClient = &http.Client{Timeout: 2 * time.Minute}
 // maxFetchBytes 单次拉取的响应体上限(防超大/投毒响应耗尽内存)：64MB
 const maxFetchBytes = 64 * 1024 * 1024
 
+// bodyPeekLen 非 200 时截取响应正文做摘要的长度(用于区分"真的挂了"与"返回了HTML错误页/验证页")
+const bodyPeekLen = 200
+
+// FetchStat 一次拉取的可观测信息，供调用方打日志/回写状态。
+type FetchStat struct {
+	StatusCode int           // HTTP 状态码，0 表示连接阶段就失败
+	Bytes      int           // 实际读取到的字节数
+	Elapsed    time.Duration // 整体耗时(含连接、TLS、读完 body)
+}
+
 // Fetch 拉取订阅源原始内容。限制最大读取字节数，返回原始 body 字节。
 // 注意：调用方需确保 url 已登记进对外访问白名单(见 CLAUDE.md)，且订阅默认关闭、由用户显式启用。
 func Fetch(url string) ([]byte, error) {
+	data, _, err := FetchWithStat(url)
+	return data, err
+}
+
+// FetchWithStat 与 Fetch 相同，但额外返回本次拉取的状态码/字节数/耗时。
+// 失败时错误信息里会带上 URL、耗时与(非 200 时的)响应正文摘要——这条链路以前失败是完全静默的，
+// 用户只能看到"没反应"，所以这里刻意把上下文塞足，便于日志与"上次状态"里直接看出原因。
+func FetchWithStat(url string) ([]byte, FetchStat, error) {
+	start := time.Now()
+	stat := FetchStat{}
+
 	req, err := http.NewRequest(http.MethodGet, url, nil)
 	if err != nil {
-		return nil, err
+		stat.Elapsed = time.Since(start)
+		return nil, stat, fmt.Errorf("构造请求失败(url=%s): %v", url, err)
 	}
 	req.Header.Set("User-Agent", "SamWaf-ThreatIP")
+
 	resp, err := fetchClient.Do(req)
 	if err != nil {
-		return nil, err
+		stat.Elapsed = time.Since(start)
+		return nil, stat, fmt.Errorf("请求失败(url=%s, 耗时=%s): %v", url, stat.Elapsed.Round(time.Millisecond), err)
 	}
 	defer resp.Body.Close()
+	stat.StatusCode = resp.StatusCode
+
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("拉取失败，HTTP 状态码: %d", resp.StatusCode)
+		peek, _ := io.ReadAll(io.LimitReader(resp.Body, bodyPeekLen))
+		stat.Elapsed = time.Since(start)
+		return nil, stat, fmt.Errorf("HTTP 状态码 %d(url=%s, 耗时=%s, 正文摘要=%q)",
+			resp.StatusCode, url, stat.Elapsed.Round(time.Millisecond), summarize(peek))
 	}
+
 	data, err := io.ReadAll(io.LimitReader(resp.Body, maxFetchBytes))
+	stat.Bytes = len(data)
+	stat.Elapsed = time.Since(start)
 	if err != nil {
-		return nil, err
+		return nil, stat, fmt.Errorf("读取响应失败(url=%s, 已读=%d字节, 耗时=%s): %v",
+			url, stat.Bytes, stat.Elapsed.Round(time.Millisecond), err)
 	}
-	return data, nil
+	return data, stat, nil
+}
+
+// summarize 把响应片段压成单行摘要(去换行、压空白)，避免污染日志与 last_status
+func summarize(b []byte) string {
+	s := strings.TrimSpace(string(b))
+	s = strings.ReplaceAll(s, "\r", " ")
+	s = strings.ReplaceAll(s, "\n", " ")
+	return strings.Join(strings.Fields(s), " ")
 }