fix: enforce db_file_size via 5-min sharding check and archive cleanup

#838

cmd/samwaf/main.go

@@ -475,6 +475,7 @@ func (m *wafSystenService) run() {
 	globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_RUNTIME_QPS_CLEAN, waftask.TaskLogQpsClean)
 	globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_HOST_QPS_CLEAN, waftask.TaskHostQpsClean)
 	globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_SHARE_DB, waftask.TaskShareDbInfo)
+	globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_SHARE_DB_CHECK, waftask.TaskShareDbInfo)
 	globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_COUNTER, waftask.TaskCounter)
 	globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_DELAY_INFO, waftask.TaskDelayInfo)
 	globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_LOAD_CONFIG, waftask.TaskLoadSettingCron)

enums/task_enum.go

@@ -3,7 +3,8 @@ package enums
 const (
 	TASK_RUNTIME_QPS_CLEAN            = "task_runtime_qps_clean"            //清空运行QPS
 	TASK_HOST_QPS_CLEAN               = "task_host_qps_clean"               //清空主机QPS
-	TASK_SHARE_DB                     = "task_share_db"                     //分库检测
+	TASK_SHARE_DB                     = "task_share_db"                     //分库检测(每天兜底)
+	TASK_SHARE_DB_CHECK               = "task_share_db_check"               //分库高频检测(每5分钟按大小切库)
 	TASK_COUNTER                      = "task_counter"                      //统计
 	TASK_DELAY_INFO                   = "task_delay_info"                   //延迟统计信息
 	TASK_LOAD_CONFIG                  = "task_load_config"                  //获取配置信息

go.sum

@@ -354,6 +354,7 @@ github.com/samwafgo/cap_go_server v1.0.0 h1:1LnITCKtYaw0X3KM4FGT9mh+3RDsyMGHefwX
 github.com/samwafgo/cap_go_server v1.0.0/go.mod h1:PbMuR/mvf5D2vqj6nZDIFuBhYWCQCq4VnKPQstZS2gc=
 github.com/samwafgo/go-wxsqlite3 v1.0.1 h1:Jv3IH1zmp2kyFWFRzz6eZGXSWWCGESPneCAgeYqZAlk=
 github.com/samwafgo/go-wxsqlite3 v1.0.1/go.mod h1:Q72tT7m3Twc1yNqJHmwl3nKR26mgFlguzSnxSAFiLJU=
+github.com/samwafgo/go-wxsqlite3 v1.0.2 h1:ipwEiHq2+jL1qAO01BrPVAXiV6Z+vKBinbKF6rGpIyM=
 github.com/samwafgo/go-wxsqlite3 v1.0.2/go.mod h1:Q72tT7m3Twc1yNqJHmwl3nKR26mgFlguzSnxSAFiLJU=
 github.com/samwafgo/sqlitedriver v1.0.2 h1:QGQrK+r9z9izEpPAW5MSjhK3vg5H45STLVrAiFT0QJU=
 github.com/samwafgo/sqlitedriver v1.0.2/go.mod h1:OtH66Yn4WxfnFA+1VhT2iMYdbKJQgZDExqsDy/1Nz5w=

service/waf_service/waf_sharedb.go

@@ -16,6 +16,11 @@ func (receiver *WafShareDbService) AddApi(shareDb model.ShareDb) error {
 	return nil
 }
 
+// DeleteById 删除一条分库归档记录（按主键 Id）。归档物理文件由调用方负责删除。
+func (receiver *WafShareDbService) DeleteById(id string) error {
+	return global.GWAF_LOCAL_DB.Where("id = ?", id).Delete(&model.ShareDb{}).Error
+}
+
 func (receiver *WafShareDbService) GetListApi(req request.WafShareDbReq) ([]model.ShareDb, int64, error) {
 	var list []model.ShareDb
 	var total int64 = 0

wafdb/migrations_task.go

@@ -428,6 +428,44 @@ func RunTaskInitMigrations(db *gorm.DB) error {
 				return tx.Where("task_method = ?", enums.TASK_STATS_DATA_CLEANUP).Delete(&model.Task{}).Error
 			},
 		},
+		// 迁移3: 新增高频分库检测任务（每5分钟按文件大小切库，兼容老用户）
+		{
+			ID: "202606260004_add_share_db_check_task",
+			Migrate: func(tx *gorm.DB) error {
+				zlog.Info("迁移 202606260004: 新增高频分库检测任务")
+
+				var count int64
+				tx.Model(&model.Task{}).Where("task_method = ?", enums.TASK_SHARE_DB_CHECK).Count(&count)
+				if count > 0 {
+					zlog.Info("高频分库检测任务已存在，跳过", "task_method", enums.TASK_SHARE_DB_CHECK)
+					return nil
+				}
+
+				task := model.Task{
+					BaseOrm: baseorm.BaseOrm{
+						Id:          uuid.GenUUID(),
+						USER_CODE:   global.GWAF_USER_CODE,
+						Tenant_ID:   global.GWAF_TENANT_ID,
+						CREATE_TIME: customtype.JsonTime(time.Now()),
+						UPDATE_TIME: customtype.JsonTime(time.Now()),
+					},
+					TaskName:   "每5分钟按文件大小检测分库",
+					TaskUnit:   enums.TASK_MIN,
+					TaskValue:  5,
+					TaskAt:     "",
+					TaskMethod: enums.TASK_SHARE_DB_CHECK,
+				}
+				if err := tx.Create(&task).Error; err != nil {
+					return fmt.Errorf("创建高频分库检测任务失败: %w", err)
+				}
+				zlog.Info("高频分库检测任务创建成功")
+				return nil
+			},
+			Rollback: func(tx *gorm.DB) error {
+				zlog.Info("回滚 202606260004: 删除高频分库检测任务")
+				return tx.Where("task_method = ?", enums.TASK_SHARE_DB_CHECK).Delete(&model.Task{}).Error
+			},
+		},
 	})
 
 	// 执行迁移

waftask/task_clean_archive.go

@@ -0,0 +1,74 @@
+package waftask
+
+import (
+	"SamWaf/common/zlog"
+	"SamWaf/enums"
+	"SamWaf/global"
+	"SamWaf/service/waf_service"
+	"SamWaf/utils"
+	"SamWaf/wafdb"
+	"SamWaf/wafdb/dialect"
+	"os"
+	"time"
+)
+
+// CleanExpiredArchiveShard 清理超过保留期(GDATA_DELETE_INTERVAL 天)的归档日志分片文件。
+//
+// 背景：高频切库(见 TaskShareDbInfo)后会产生大量 local_log_<ts>.db 归档文件，
+// 而原有的删历史逻辑只对 live 库做 DELETE，从不清理归档文件，会很快堆满磁盘。
+// 真正的 N 天保留改由本函数按分片 EndTime 删除整个过期归档文件实现。
+//
+// 仅 SQLite(文件型驱动)生效；MySQL 的归档是表，由其各自策略处理。
+func CleanExpiredArchiveShard() {
+	innerLogName := "CleanExpiredArchiveShard"
+	if !dialect.Get().IsFileBased() {
+		return
+	}
+
+	cutoff := time.Now().AddDate(0, 0, -int(global.GDATA_DELETE_INTERVAL))
+	currentDir := utils.GetCurrentDir()
+
+	shards, err := waf_service.WafShareDbServiceApp.GetAllShareDbApi()
+	if err != nil {
+		zlog.Error(innerLogName, "获取分库列表失败", err)
+		return
+	}
+
+	removedFiles := 0
+	for _, shard := range shards {
+		// 只处理日志分片；跳过 live 库本身
+		if shard.DbLogicType != "log" {
+			continue
+		}
+		if shard.FileName == "" || shard.FileName == enums.DB_LOG || shard.FileName == "local_log.db" {
+			continue
+		}
+		// 整个分片(截止 EndTime)早于保留期才删除
+		if !time.Time(shard.EndTime).Before(cutoff) {
+			continue
+		}
+
+		// 删除前先关闭可能已按需打开的连接，避免删正在查询的文件
+		wafdb.CloseManualLogDb(shard.FileName)
+
+		dbPath := currentDir + "/data/" + shard.FileName
+		for _, p := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} {
+			if err := os.Remove(p); err != nil && !os.IsNotExist(err) {
+				zlog.Warn(innerLogName, "删除归档文件失败", "file", p, "error", err.Error())
+			}
+		}
+		removedFiles++
+
+		// 删除归档元数据记录
+		if err := waf_service.WafShareDbServiceApp.DeleteById(shard.Id); err != nil {
+			zlog.Warn(innerLogName, "删除归档记录失败", "file", shard.FileName, "error", err.Error())
+		} else {
+			zlog.Info(innerLogName, "已清理过期归档分片", "file", shard.FileName,
+				"end_time", time.Time(shard.EndTime).Format("2006-01-02 15:04:05"))
+		}
+	}
+
+	if removedFiles > 0 {
+		zlog.Info(innerLogName, "归档清理完成", "removed", removedFiles, "cutoff", cutoff.Format("2006-01-02"))
+	}
+}

waftask/task_db_sharding.go

@@ -54,10 +54,15 @@ func TaskShareDbInfo() {
 	if dialect.Get().IsFileBased() {
 		fileInfo, err := os.Stat(dbFilePath)
 		if err == nil {
-			fileSizeMB := fileInfo.Size() / (1024 * 1024) // 转换为MB
+			totalBytes := fileInfo.Size()
+			// 把 -wal 大小一并计入：WAL 模式下数据可能暂存在 wal 文件，主库文件显小会导致漏判
+			if walInfo, werr := os.Stat(dbFilePath + "-wal"); werr == nil {
+				totalBytes += walInfo.Size()
+			}
+			fileSizeMB := totalBytes / (1024 * 1024) // 转换为MB
 			if fileSizeMB > global.GDATA_SHARE_DB_FILE_SIZE {
 				needSharding = true
-				shardingReason = fmt.Sprintf("文件大小(%dMB)超过限制(%dMB)", fileSizeMB, global.GDATA_SHARE_DB_FILE_SIZE)
+				shardingReason = fmt.Sprintf("文件大小(%dMB,含WAL)超过限制(%dMB)", fileSizeMB, global.GDATA_SHARE_DB_FILE_SIZE)
 			}
 		} else {
 			zlog.Error(innerLogName, "获取数据库文件大小失败:", err)
@@ -122,11 +127,17 @@ func TaskShareDbInfo() {
 					zlog.Error(innerLogName, "切换关闭时候错误", err)
 				}
 			}
+			// 等待连接彻底关闭（Count 报错即连接已关闭，可安全重命名文件）。
+			// 加最大重试上限，避免连接异常未报错时无限循环卡住切库（高频切库后该段执行更频繁）。
 			var testTotal int64
-			for {
+			for attempt := 0; attempt < 10; attempt++ {
 				testError := global.GWAF_LOCAL_LOG_DB.Model(&innerbean.WebLog{}).Count(&testTotal).Error
 				if testError != nil {
-					zlog.Error(innerLogName, "检测数据", testError)
+					zlog.Debug(innerLogName, "连接已关闭，可切库", testError)
+					break
+				}
+				if attempt == 9 {
+					zlog.Warn(innerLogName, "等待日志库连接关闭超时，强制继续切库")
 					break
 				}
 				time.Sleep(1 * time.Second)