diff --git a/model/task.go b/model/task.go index 2922688..32ce8ef 100644 --- a/model/task.go +++ b/model/task.go @@ -27,16 +27,17 @@ func (TaskPlan) TableName() string { // TaskExecution 任务执行记录 type TaskExecution struct { - ID int64 `gorm:"primaryKey;column:id;type:BIGINT;not null" json:"id"` - TaskID string `gorm:"column:task_id;type:VARCHAR(128)" json:"taskId"` - CommandID string `gorm:"column:command_id;type:VARCHAR(64)" json:"commandId"` - DockID string `gorm:"column:dock_id;type:VARCHAR(64);not null" json:"dockId"` - DroneSN string `gorm:"column:drone_sn;type:VARCHAR(32)" json:"droneSn"` - StartTime *time.Time `gorm:"column:start_time" json:"startTime"` - EndTime *time.Time `gorm:"column:end_time" json:"endTime"` - Status string `gorm:"column:status;type:VARCHAR(16);default:pending" json:"status"` + ID int64 `gorm:"primaryKey;column:id;type:BIGINT;not null" json:"id"` + TaskID string `gorm:"column:task_id;type:VARCHAR(128)" json:"taskId"` + CommandID string `gorm:"column:command_id;type:VARCHAR(64)" json:"commandId"` + DockID string `gorm:"column:dock_id;type:VARCHAR(64);not null" json:"dockId"` + DroneSN string `gorm:"column:drone_sn;type:VARCHAR(32)" json:"droneSn"` + StartTime *time.Time `gorm:"column:start_time" json:"startTime"` + EndTime *time.Time `gorm:"column:end_time" json:"endTime"` + Status string `gorm:"column:status;type:VARCHAR(16);default:pending" json:"status"` + ResultCode string `gorm:"column:result_code;type:VARCHAR(64)" json:"resultCode"` TrajectoryJSON json.RawMessage `gorm:"column:trajectory_json;type:JSON" json:"trajectoryJson"` - CreatedAt time.Time `gorm:"column:created_at" json:"createdAt"` + CreatedAt time.Time `gorm:"column:created_at" json:"createdAt"` } func (TaskExecution) TableName() string { diff --git a/mqtt/mqtt.go b/mqtt/mqtt.go index e6d6db7..27b6d80 100644 --- a/mqtt/mqtt.go +++ b/mqtt/mqtt.go @@ -55,17 +55,35 @@ func Subscribe(topics map[string]byte, handler mqtt.MessageHandler) { func resubscribe(c mqtt.Client) { for topic, qos := range subTopics { - if token := c.Subscribe(topic, qos, subHandler); token.Wait() && token.Error() != nil { + if token := c.Subscribe(topic, qos, inboundHandler); token.Wait() && token.Error() != nil { logger.ERROR("订阅失败 topic="+topic, token.Error()) } } } +func inboundHandler(c mqtt.Client, msg mqtt.Message) { + //logger.INFO("MQTT 收到消息", + // "topic="+msg.Topic(), + // "qos=", msg.Qos(), + // "retained=", msg.Retained(), + // "payload="+string(msg.Payload()), + //) + if subHandler != nil { + subHandler(c, msg) + } +} + // Publish 发布消息到指定 topic func Publish(topic string, qos byte, retained bool, payload []byte) error { if client == nil || !client.IsConnected() { return errors.New("mqtt client not connected") } + logger.INFO("MQTT 发送消息", + "topic="+topic, + "qos=", qos, + "retained=", retained, + "payload="+string(payload), + ) token := client.Publish(topic, qos, retained, payload) token.Wait() return token.Error() diff --git a/service/account_resource_service.go b/service/account_resource_service.go index 5f28f96..2e14668 100644 --- a/service/account_resource_service.go +++ b/service/account_resource_service.go @@ -24,7 +24,7 @@ func (s *AccountService) GetDownloadPage(userID int64, req *vo.UsagePageReq) (*c return nil, common.ErrInternal } var rows []row - if err := db.Scopes(req.Paginate).Order("download_log.id DESC").Find(&rows).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("download_log.created_at DESC, download_log.id DESC").Find(&rows).Error; err != nil { return nil, common.ErrInternal } list := make([]vo.DownloadRecordVO, 0, len(rows)) diff --git a/service/alarm_service.go b/service/alarm_service.go index 42147c1..2fc6950 100644 --- a/service/alarm_service.go +++ b/service/alarm_service.go @@ -44,7 +44,7 @@ func (s *AlarmService) GetPage(userID int64, isAdmin bool, req *vo.AlarmPageReq) return nil, common.ErrInternal } var list []model.Alarm - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("triggered_at DESC, id DESC").Find(&list).Error; err != nil { logger.ERROR("查询告警列表失败", err) return nil, common.ErrInternal } diff --git a/service/billing_service.go b/service/billing_service.go index 3fbe351..1c6d76c 100644 --- a/service/billing_service.go +++ b/service/billing_service.go @@ -173,7 +173,7 @@ func (b *BillingService) GetUsagePage(userID int64, req *vo.UsagePageReq) (*comm return nil, common.ErrInternal } var list []model.TrafficUsageLog - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { return nil, common.ErrInternal } return common.Page(req.Pagination, total, list), nil @@ -190,7 +190,7 @@ func (b *BillingService) GetOrderPage(userID int64, req *vo.OrderPageReq) (*comm return nil, common.ErrInternal } var list []model.TrafficOrder - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { return nil, common.ErrInternal } return common.Page(req.Pagination, total, list), nil @@ -263,7 +263,7 @@ func (b *BillingService) ListSimCards(userID int64, req *vo.SimCardPageReq) (*co return nil, common.ErrInternal } var list []model.SimCard - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("updated_at DESC, id DESC").Find(&list).Error; err != nil { return nil, common.ErrInternal } return common.Page(req.Pagination, total, list), nil @@ -283,7 +283,7 @@ func (b *BillingService) GetSimRechargeLogPage(userID int64, req *vo.SimRecharge return nil, common.ErrInternal } var list []model.SimRechargeLog - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { return nil, common.ErrInternal } return common.Page(req.Pagination, total, list), nil diff --git a/service/command_service.go b/service/command_service.go index 3272880..f372cd2 100644 --- a/service/command_service.go +++ b/service/command_service.go @@ -24,7 +24,11 @@ type CommandService struct{} var DefaultCommandService = &CommandService{} -const maxRetryCount = 2 +const ( + maxRetryCount = 2 + workflowStartTimeout = 2 * time.Minute + workflowProgressTimeout = 5 * time.Minute +) // Dispatch 下发指令:校验在线 → 落库 → MQTT 发布 func (s *CommandService) Dispatch(userID int64, isAdmin bool, id int64, req *vo.CommandReq) (*model.DeviceCommandLog, *common.BusiError) { @@ -133,9 +137,16 @@ func failTaskExecution(cmd model.DeviceCommandLog, reason string) { if reason == "" { reason = "COMMAND_REJECTED" } - _ = common.DB.Model(&model.TaskExecution{}). + result := common.DB.Model(&model.TaskExecution{}). Where("dock_id = ? AND command_id = ? AND status IN ?", cmd.DockID, strconv.FormatInt(cmd.ID, 10), []string{"pending", "running"}). - Updates(map[string]any{"status": "failed", "end_time": time.Now()}) + Updates(map[string]any{"status": "failed", "result_code": reason, "end_time": time.Now()}) + if result.Error != nil { + logger.ERROR("更新任务执行失败状态失败", result.Error) + return + } + if result.RowsAffected > 0 { + DefaultTrajectoryStore.Finalize(cmd.DockID, "") + } } // DispatchToDock 持久化并下发已完成权限与在线校验的设备指令。 @@ -232,6 +243,10 @@ func (s *CommandService) retryTimeout() { if cmd.AckedAt == nil { continue } + if cmd.CommandType == "workflow.start_task" { + s.checkWorkflowTimeout(cmd, now) + continue + } ttl := int64(cmd.TTLMs) if ttl <= 0 { ttl = 30000 @@ -242,6 +257,48 @@ func (s *CommandService) retryTimeout() { } } +func (s *CommandService) checkWorkflowTimeout(cmd model.DeviceCommandLog, now time.Time) { + commandID := strconv.FormatInt(cmd.ID, 10) + var executions []model.TaskExecution + executionQuery := common.DB.Where("dock_id = ? AND command_id = ? AND status IN ?", cmd.DockID, commandID, []string{"pending", "running"}).Find(&executions) + if executionQuery.Error != nil { + logger.ERROR("查询工作流执行记录失败", executionQuery.Error) + return + } + if executionQuery.RowsAffected == 0 { + return + } + + var workflow model.WorkflowState + err := common.DB.Where("dock_id = ? AND command_id = ?", cmd.DockID, commandID).First(&workflow).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + if now.Sub(*cmd.AckedAt) > workflowStartTimeout { + s.timeoutWorkflow(cmd, "WORKFLOW_START_TIMEOUT") + } + return + } + if err != nil { + logger.ERROR("查询工作流状态失败", err) + return + } + if workflow.State == "running" && now.Sub(workflow.UpdatedAt) > workflowProgressTimeout { + s.timeoutWorkflow(cmd, "WORKFLOW_PROGRESS_TIMEOUT") + } +} + +func (s *CommandService) timeoutWorkflow(cmd model.DeviceCommandLog, reason string) { + result := common.DB.Model(&model.DeviceCommandLog{}). + Where("id = ? AND status = ?", cmd.ID, "acked"). + Updates(map[string]any{"status": "terminal", "ack_result_code": reason}) + if result.Error != nil { + logger.ERROR("更新工作流超时指令失败", result.Error) + return + } + if result.RowsAffected > 0 { + failTaskExecution(cmd, reason) + } +} + // commandEnvelope 构造下行指令通用外层(文档 §5.1) func (s *CommandService) commandEnvelope(cmd *model.DeviceCommandLog, requestID string, params map[string]any) *mqtt.Envelope { return mqtt.NewEnvelope(requestID, cmd.DockID, cmd.DroneSN, map[string]any{ @@ -284,7 +341,7 @@ func (s *CommandService) GetPage(userID int64, isAdmin bool, req *vo.CommandPage return nil, common.ErrInternal } var list []model.DeviceCommandLog - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("sent_at DESC, id DESC").Find(&list).Error; err != nil { logger.ERROR("查询指令日志失败", err) return nil, common.ErrInternal } diff --git a/service/dock_service.go b/service/dock_service.go index bfb5b8c..27941e9 100644 --- a/service/dock_service.go +++ b/service/dock_service.go @@ -53,7 +53,7 @@ func (s *DockService) GetPage(userID int64, isAdmin bool, req *vo.DockPageReq) ( return nil, common.ErrInternal } var list []model.Dock - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { logger.ERROR("查询机巢列表失败", err) return nil, common.ErrInternal } diff --git a/service/drone_service.go b/service/drone_service.go index ddccea1..de79869 100644 --- a/service/drone_service.go +++ b/service/drone_service.go @@ -32,7 +32,7 @@ func (s *DroneService) GetPage(userID int64, isAdmin bool, req *vo.DronePageReq) return nil, common.ErrInternal } var list []model.Drone - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { logger.ERROR("查询无人机列表失败", err) return nil, common.ErrInternal } diff --git a/service/execution_service.go b/service/execution_service.go index 0a116bc..bd652a7 100644 --- a/service/execution_service.go +++ b/service/execution_service.go @@ -35,7 +35,7 @@ func (s *ExecutionService) GetPage(userID int64, isAdmin bool, req *vo.Execution return nil, common.ErrInternal } var list []model.TaskExecution - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { logger.ERROR("查询执行记录失败", err) return nil, common.ErrInternal } diff --git a/service/firmware_service.go b/service/firmware_service.go index 4fd05ab..2fbb297 100644 --- a/service/firmware_service.go +++ b/service/firmware_service.go @@ -44,7 +44,7 @@ func (s *FirmwareService) GetPage(req *vo.FirmwarePageReq) (*common.PageResponse return nil, common.ErrInternal } var list []model.Firmware - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { logger.ERROR("查询固件列表失败", err) return nil, common.ErrInternal } @@ -54,7 +54,7 @@ func (s *FirmwareService) GetPage(req *vo.FirmwarePageReq) (*common.PageResponse // GetReleasedList 已发布固件列表(供普通用户选择升级) func (s *FirmwareService) GetReleasedList() ([]model.Firmware, *common.BusiError) { var list []model.Firmware - if err := common.DB.Where("status = ?", "released").Order("id DESC").Find(&list).Error; err != nil { + if err := common.DB.Where("status = ?", "released").Order("updated_at DESC, id DESC").Find(&list).Error; err != nil { logger.ERROR("查询已发布固件失败", err) return nil, common.ErrInternal } diff --git a/service/invoice_service.go b/service/invoice_service.go index 80c9a0c..2eb267d 100644 --- a/service/invoice_service.go +++ b/service/invoice_service.go @@ -18,7 +18,7 @@ var DefaultInvoiceService = &InvoiceService{} func (s *InvoiceService) ListProfiles(userID int64) ([]model.InvoiceProfile, *common.BusiError) { var profiles []model.InvoiceProfile - if err := common.DB.Where("user_id = ?", userID).Order("is_default DESC, id DESC").Find(&profiles).Error; err != nil { + if err := common.DB.Where("user_id = ?", userID).Order("is_default DESC, updated_at DESC, id DESC").Find(&profiles).Error; err != nil { return nil, common.ErrInternal } return profiles, nil @@ -67,7 +67,7 @@ func (s *InvoiceService) DeleteProfile(userID, profileID int64) *common.BusiErro func (s *InvoiceService) ListEligibleOrders(userID int64) ([]model.TrafficOrder, *common.BusiError) { var orders []model.TrafficOrder - if err := common.DB.Where("user_id = ? AND pay_status = ? AND NOT EXISTS (SELECT 1 FROM invoice_request WHERE invoice_request.order_id = traffic_order.id)", userID, "paid").Order("paid_at DESC").Find(&orders).Error; err != nil { + if err := common.DB.Where("user_id = ? AND pay_status = ? AND NOT EXISTS (SELECT 1 FROM invoice_request WHERE invoice_request.order_id = traffic_order.id)", userID, "paid").Order("paid_at DESC, id DESC").Find(&orders).Error; err != nil { return nil, common.ErrInternal } return orders, nil @@ -104,7 +104,7 @@ func (s *InvoiceService) CreateRequest(userID int64, req *vo.InvoiceRequestCreat func (s *InvoiceService) ListRequests(userID int64) ([]model.InvoiceRequest, *common.BusiError) { var requests []model.InvoiceRequest - if err := common.DB.Where("user_id = ?", userID).Order("requested_at DESC").Find(&requests).Error; err != nil { + if err := common.DB.Where("user_id = ?", userID).Order("requested_at DESC, id DESC").Find(&requests).Error; err != nil { return nil, common.ErrInternal } return requests, nil diff --git a/service/live_service.go b/service/live_service.go index 7874900..546ba6c 100644 --- a/service/live_service.go +++ b/service/live_service.go @@ -63,7 +63,7 @@ func (s *LiveService) Join(userID int64, isAdmin bool, dockID string, req *vo.Li if err := tx.Set("gorm:query_option", "FOR UPDATE").Where("dock_id = ?", dockID).First(&model.Dock{}).Error; err != nil { return err } - if err := tx.Where("dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", dockID).Order("created_at DESC").First(&session).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { + if err := tx.Where("dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", dockID).Order("created_at DESC, id DESC").First(&session).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { return err } if session.ID == "" { @@ -230,7 +230,7 @@ func (s *LiveService) Stop(userID int64, isAdmin bool, dockID string) *common.Bu return common.ErrInternal } var active model.LiveSession - if err := common.DB.Where("dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", dockID).Order("id DESC").First(&active).Error; err != nil { + if err := common.DB.Where("dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", dockID).Order("created_at DESC, id DESC").First(&active).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return common.ErrLiveNotFound } @@ -274,7 +274,7 @@ func (s *LiveService) GetPage(userID int64, isAdmin bool, req *vo.LivePageReq) ( return nil, common.ErrInternal } var list []model.LiveSession - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { return nil, common.ErrInternal } return common.Page(req.Pagination, total, list), nil @@ -287,7 +287,7 @@ func (s *LiveService) GetPlayURL(userID int64, isAdmin bool, dockID string) (*vo return nil, busiErr } var session model.LiveSession - if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("dock_id = ? AND phase = 'streaming'", dockID).Order("id DESC").First(&session).Error; err != nil { + if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("dock_id = ? AND phase = 'streaming'", dockID).Order("created_at DESC, id DESC").First(&session).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, common.ErrLiveNotFound } diff --git a/service/operation_log_service.go b/service/operation_log_service.go index 64ec96a..8cba48e 100644 --- a/service/operation_log_service.go +++ b/service/operation_log_service.go @@ -65,7 +65,7 @@ func (s *OperationLogService) GetPage(req *vo.OperationLogPageReq) (*common.Page return nil, common.ErrInternal } var list []model.OperationLog - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { logger.ERROR("查询操作日志失败", err) return nil, common.ErrInternal } diff --git a/service/route_service.go b/service/route_service.go index fdb3f56..5984e61 100644 --- a/service/route_service.go +++ b/service/route_service.go @@ -31,7 +31,7 @@ func (s *RouteService) GetPage(userID int64, isAdmin bool, req *vo.RoutePageReq) return nil, common.ErrInternal } var list []model.Route - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { logger.ERROR("查询航线列表失败", err) return nil, common.ErrInternal } diff --git a/service/system_service.go b/service/system_service.go index de29656..660718b 100644 --- a/service/system_service.go +++ b/service/system_service.go @@ -38,7 +38,7 @@ func (s *SystemService) GetUserPage(req *vo.UserPageReq) (*common.PageResponse[m return nil, common.ErrInternal } var list []model.User - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { logger.ERROR("查询用户列表失败", err) return nil, common.ErrInternal } diff --git a/service/task_service.go b/service/task_service.go index 171fa51..4431e8a 100644 --- a/service/task_service.go +++ b/service/task_service.go @@ -43,7 +43,7 @@ func (s *TaskService) GetPage(userID int64, isAdmin bool, req *vo.TaskPageReq) ( return nil, common.ErrInternal } var tasks []model.TaskPlan - if err := db.Scopes(req.Paginate).Order("created_at DESC").Find(&tasks).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&tasks).Error; err != nil { logger.ERROR("查询任务列表失败", err) return nil, common.ErrInternal } diff --git a/service/user_service.go b/service/user_service.go index 8ee7c35..c317973 100644 --- a/service/user_service.go +++ b/service/user_service.go @@ -144,7 +144,7 @@ func (s *UserService) ChangePassword(userID int64, req *vo.ChangePasswordReq) *c func (s *UserService) ListSessions(userID, currentSessionID int64) ([]vo.SessionVO, *common.BusiError) { var sessions []model.UserLoginSession - if err := common.DB.Where("user_id = ? AND revoked_at IS NULL", userID).Order("last_active_at DESC").Find(&sessions).Error; err != nil { + if err := common.DB.Where("user_id = ? AND revoked_at IS NULL", userID).Order("last_active_at DESC, id DESC").Find(&sessions).Error; err != nil { return nil, common.ErrInternal } result := make([]vo.SessionVO, 0, len(sessions)) diff --git a/service/video_service.go b/service/video_service.go index 9c39ca1..c54da00 100644 --- a/service/video_service.go +++ b/service/video_service.go @@ -186,7 +186,7 @@ func (s *VideoService) GetPage(userID int64, isAdmin bool, req *vo.VideoPageReq) return nil, common.ErrInternal } var list []model.Video - if err := db.Scopes(req.Paginate).Order("id DESC").Find(&list).Error; err != nil { + if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { return nil, common.ErrInternal } return common.Page(req.Pagination, total, list), nil diff --git a/service/workflow_service.go b/service/workflow_service.go index c93080f..5cd3eb1 100644 --- a/service/workflow_service.go +++ b/service/workflow_service.go @@ -82,7 +82,7 @@ func (s *WorkflowService) Upsert(dockID, requestID string, in *WorkflowStateIn) // 回写 task_execution,优先按 command_id 精确关联,兼容旧设备时只回退到最新未终态记录。 if (in.CommandID != "" || in.TaskID != "") && (in.State == "running" || isTerminalWorkflowState(in.State)) { - updates := map[string]any{"status": in.State} + updates := map[string]any{"status": in.State, "result_code": in.ResultCode} if in.State == "running" { updates["start_time"] = now } else { diff --git a/sql/001_schema.sql b/sql/001_schema.sql index 3abfe3f..1df75b3 100644 --- a/sql/001_schema.sql +++ b/sql/001_schema.sql @@ -176,6 +176,7 @@ CREATE TABLE IF NOT EXISTS task_execution ( start_time DATETIME, end_time DATETIME, status VARCHAR(16) DEFAULT 'pending', + result_code VARCHAR(64), trajectory_json JSON, created_at DATETIME ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/sql/010_task_workflow_timeout.sql b/sql/010_task_workflow_timeout.sql new file mode 100644 index 0000000..fb7ed21 --- /dev/null +++ b/sql/010_task_workflow_timeout.sql @@ -0,0 +1,2 @@ +ALTER TABLE task_execution + ADD COLUMN result_code VARCHAR(64) NULL; diff --git a/tool/snowflake.go b/tool/snowflake.go index 8ca4de2..fa1748e 100644 --- a/tool/snowflake.go +++ b/tool/snowflake.go @@ -10,7 +10,7 @@ const ( // 起始时间戳 (2023-01-01 00:00:00 UTC) epoch int64 = 1672531200000 - timestampBits = 28 // 时间戳位数(约17年) + timestampBits = 41 // 时间戳位数(约69年) workerIDBits = 5 // 工作机器ID所占位数 sequenceBits = 12 // 序列号所占位数 @@ -37,7 +37,7 @@ func init() { func NewSnowflake(workerID int64) *Snowflake { if workerID < 0 || workerID > maxWorkerID { - panic(errors.New("worker ID must be between 0 and 1023")) + panic(errors.New("worker ID must be between 0 and 31")) } return &Snowflake{ timestamp: 0, @@ -52,42 +52,38 @@ func (s *Snowflake) NextID() (int64, error) { s.mu.Lock() defer s.mu.Unlock() - now := time.Now().UnixMilli() - now = (now - epoch) & (-1 ^ (-1 << timestampBits)) + maxTimestamp := int64(1< maxTimestamp { + return -1, errors.New("timestamp overflow") } if now < s.lastTime { waitTime := s.lastTime - now if waitTime > 100 { - now = s.lastTime + 1 - if now > (1< (1< maxTimestamp { + return -1, errors.New("timestamp overflow") } } - if s.lastTime == now { + if now == s.lastTime { s.sequence = (s.sequence + 1) & maxSequence if s.sequence == 0 { for now <= s.lastTime { - now = (time.Now().UnixMilli() - epoch) & (-1 ^ (-1 << timestampBits)) - if now <= 0 { - now = s.lastTime + 1 - break + time.Sleep(time.Millisecond) + now = time.Now().UnixMilli() - epoch + if now > maxTimestamp { + return -1, errors.New("timestamp overflow") } } } @@ -96,19 +92,7 @@ func (s *Snowflake) NextID() (int64, error) { } s.lastTime = now - - id := (now << timestampShift) | - (s.workerID << workerIDShift) | - s.sequence - - if id == 0 { - s.sequence = 1 - id = (now << timestampShift) | - (s.workerID << workerIDShift) | - s.sequence - } - - return id, nil + return (now << timestampShift) | (s.workerID << workerIDShift) | s.sequence, nil } // NextID 全局函数,使用默认实例生成ID