# 数据库 SSOT 单线程事件循环重构 — 实现计划 > **For agentic workers:** REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 移除所有 Redis 运行时依赖(RedisBus、RedisStateManager、RedisToolLocker、WriteThroughHelper、ZPopLoop),新增基于 channel 的 ProductionEventLoop 作为唯一生产状态写路径,DB 用 entgo 原子条件更新。 **Architecture:** 新增 `internal/eventloop/` 包(EventLoopMessage、ProductionEventLoop、DBState、hardware worker)。HTTP handler/SignalWatcher/恢复流程只投递消息到 event loop;硬件 worker 执行长耗时动作后回投结果。现有 scheduler 三层 Generator/Filter/Policy 保留并适配 DB 快照输入。 **Tech Stack:** Go 1.23.4,go-zero v1.8.3,ent ORM v0.14.5,PostgreSQL,无 Redis。 --- ## 文件结构规划 ### 新建文件 | 文件 | 职责 | |------|------| | `internal/eventloop/types.go` | EventLoopMessage、CorrelationID、ReplyChan | | `internal/eventloop/loop.go` | ProductionEventLoop:channel 消费、消息分发、调度触发 | | `internal/eventloop/dbstate.go` | DBState:ent 原子条件更新封装 | | `internal/eventloop/dbstate_test.go` | DBState 集成测试 | | `internal/eventloop/worker.go` | HardwareWorker 接口 + robot worker 适配 | | `internal/eventloop/loop_test.go` | EventLoop 单元测试 | | `internal/eventloop/recovery.go` | 启动恢复:扫描 DB、校验一致性、创建 manual_action | | `internal/eventloop/recovery_test.go` | 恢复逻辑测试 | ### 修改文件 | 文件 | 变更 | |------|------| | `internal/config/config.go` | 删除 EventBus/SSOT 配置结构体,Redis 保留但仅非核心缓存 | | `etc/hougai-api.yaml` | 删除 EventBus/SSOT 配置项 | | `internal/svc/service_context.go` | 重写:移除 go-redis 客户端、RedisBus、StateManager、WriteThroughHelper、ZPopLoop,新建 EventLoop+Worker 并注入 | | `internal/processor/job_processor.go` | 大幅删减:移除 ReadyQueue、ssotWriter、toolLocker、scheduleMu、ZPop 路径,API 方法改为向 event loop 投递 command,保留 jobRuntime map 和相关查询 | | `internal/processor/job_runtime.go` | 简化:移除 onStatusSync/ssotVersion/terminalCounted/notifyReady/exchangePairer 回调字段,移除 SetOnCompleteFunc/SetOnErrorFunc/syncSSOTStatus/NotifyReady 方法,保留步骤推进和动作构建 | | `internal/processor/dispatcher.go` | 简化/废弃:不再需要双通道调度器和 replenishing 标志 | | `internal/processor/ready_queue.go` | 删除 | | `internal/processor/signal_watcher.go` | 回调改为向 event loop 投递 MachineDone/InspectionResult | | `internal/processor/replenisher.go` | 简化:不维护 allocator/mutex,只生成补料动作 | | `internal/processor/temp_slot_allocator.go` | 删除 | | `internal/processor/recovery_coordinator.go` | 简化:改为向 event loop 投递 ConfirmRecovery | | `internal/processor/scheduler_adapter.go` | 适配:SystemState 从 DB/loop 快照构建 | | `internal/scheduler/scheduler.go` | 保留 | | `internal/scheduler/constraint_tool_lock.go` | 删除或简化为 no-op | | `internal/scheduler/zpop.go` | 删除 | | `internal/state/redis_manager.go` | 删除 | | `internal/state/state_write_through.go` | 删除 | | `internal/state/tool_lock.go` | 删除 RedisToolLocker,保留 MemoryToolLocker | | `internal/state/keys.go` | 删除 | | `internal/state/manager.go` | 简化:保留 StateManager 接口供测试 mock,删除 ClaimJob/ApplyTaskState 等 SSOT 方法 | | `internal/eventbus/bus.go` | 删除 RedisBus 实现,保留 Bus 接口 + LocalBus | | `internal/eventbus/events.go` | 保留事件类型常量,删除 EntityVersion 字段 | | `internal/recovery/replayer.go` | 删除(不再重放到 Redis) | | `internal/verify/state_verifier.go` | 删除(不再对比 Redis 状态) | ### 删除文件 `internal/processor/ready_queue.go`、`internal/processor/temp_slot_allocator.go`、`internal/processor/ssot_integration_test.go`、`internal/state/redis_manager.go`、`internal/state/state_write_through.go`、`internal/state/keys.go`、`internal/scheduler/zpop.go`、`internal/recovery/replayer.go`、`internal/recovery/replayer_test.go`、`internal/verify/state_verifier.go`、`internal/verify/state_verifier_test.go` --- ## Chunk 1: 配置清理 + EventLoop 类型定义 ### Task 1.1: 删除配置结构体中的 EventBus/SSOT **Files:** - Modify: `internal/config/config.go` - Modify: `etc/hougai-api.yaml` - [ ] **Step 1: 从 config.go 删除 EventBusConf 和 SSOTConf** ```go // 删除这两段 type EventBusConf struct { Enabled bool `json:",default=false"` StreamName string `json:",default=hougai:events"` GroupName string `json:",default=hougai-workers"` } type SSOTConf struct { Enabled bool `json:",default=false"` KeyPrefix string `json:",default=hougai:"` } // 删除 Config 中的字段 // EventBus EventBusConf // SSOT SSOTConf ``` - [ ] **Step 2: 从 hougai-api.yaml 删除 EventBus 和 SSOT 配置块** ```yaml # 删除以下行 EventBus: Enabled: true StreamName: "back_cover:events" GroupName: "back_cover-workers" SSOT: Enabled: true KeyPrefix: "back_cover:" ``` - [ ] **Step 3: 编译验证** ```bash rtk go build ./... ``` ### Task 1.2: 创建 EventLoopMessage 类型 **Files:** - Create: `internal/eventloop/types.go` - [ ] **Step 1: 写 types.go** ```go package eventloop import "time" // MessageType 消息类型 type MessageType string const ( // Command CmdStartOrder MessageType = "START_ORDER" CmdPauseOrder MessageType = "PAUSE_ORDER" CmdResumeOrder MessageType = "RESUME_ORDER" CmdCancelOrder MessageType = "CANCEL_ORDER" CmdSuspendJob MessageType = "SUSPEND_JOB" CmdResumeJob MessageType = "RESUME_JOB" CmdReworkJob MessageType = "REWORK_JOB" CmdConfirmRecovery MessageType = "CONFIRM_RECOVERY" CmdResolveManualAction MessageType = "RESOLVE_MANUAL_ACTION" // ExternalEvent EvtMachineDone MessageType = "MACHINE_DONE" EvtInspectionResult MessageType = "INSPECTION_RESULT" EvtPalletArrived MessageType = "PALLET_ARRIVED" EvtStepTimeout MessageType = "STEP_TIMEOUT" EvtRefillRequested MessageType = "REFILL_REQUESTED" EvtScheduleTick MessageType = "SCHEDULE_TICK" // WorkerResult ResRobotActionSucceeded MessageType = "ROBOT_ACTION_SUCCEEDED" ResRobotActionFailed MessageType = "ROBOT_ACTION_FAILED" ResToolActionSucceeded MessageType = "TOOL_ACTION_SUCCEEDED" ResToolActionFailed MessageType = "TOOL_ACTION_FAILED" ResMachineStartSucceeded MessageType = "MACHINE_START_SUCCEEDED" ResMachineStartFailed MessageType = "MACHINE_START_FAILED" ) // EventLoopMessage 事件循环消息 type EventLoopMessage struct { ID string Type MessageType Payload map[string]any CorrelationID string // worker result 关联的 task.id Reply chan MessageResult CreatedAt time.Time } // MessageResult 消息处理结果(用于同步 reply) type MessageResult struct { Success bool Error string Data map[string]any } ``` - [ ] **Step 2: 编译验证** ```bash rtk go build ./internal/eventloop/... ``` ### Task 1.3: 创建 HardwareWorker 接口 **Files:** - Create: `internal/eventloop/worker.go` - [ ] **Step 1: 写 worker.go(接口定义)** ```go package eventloop import "context" // RobotAction 机器人动作请求 type RobotAction struct { TaskID string Kind string // load/unload/exchange/scan/mark/replenish JobID int MachineID int SlotNo int TargetSlotNo int Params map[string]any } // HardwareWorker 硬件执行器接口 type HardwareWorker interface { Execute(ctx context.Context, action RobotAction) error } // WorkerResultCallback worker 完成后的回调 type WorkerResultCallback func(msg EventLoopMessage) ``` - [ ] **Step 2: 编译验证** ```bash rtk go build ./internal/eventloop/... ``` --- ## Chunk 2: DBState 层(ent 原子条件更新) ### Task 2.1: 创建 DBState 领域状态访问层 **Files:** - Create: `internal/eventloop/dbstate.go` - Create: `internal/eventloop/dbstate_test.go` - [ ] **Step 1: 写 dbstate.go 接口和方法签名** ```go package eventloop import ( "context" "fmt" "strconv" "hougai/constants" "hougai/ent" "hougai/ent/job" "hougai/ent/equipment" "hougai/ent/equipmentslot" "hougai/ent/task" "hougai/ent/workorder" ) // DBState ent 原子条件更新封装,仅由 event loop 使用 type DBState struct { client *ent.Client } func NewDBState(client *ent.Client) *DBState { return &DBState{client: client} } // ClaimJobForAction 原子抢占工件:从 ON_BUFFER 转为 IN_HANDLING // 防御性 WHERE status = expectedStatus,单 event loop 保证不会并发冲突 func (d *DBState) ClaimJobForAction(ctx context.Context, jobID int, expectedStatus constants.JobStatus) error { n, err := d.client.Job.Update(). Where(job.IDEQ(jobID), job.StatusEQ(expectedStatus)). SetStatus(constants.JobStatus_InHandling). AddVersion(1). Save(ctx) if err != nil { return fmt.Errorf("claim job %d: %w", jobID, err) } if n == 0 { return fmt.Errorf("claim job %d: status not %s", jobID, expectedStatus) } return nil } ``` - [ ] **Step 2: 添加更多领域方法** ```go // MoveJobToEquipment 工件上设备:更新 job.position* + equipment_slot.status/currentJobId func (d *DBState) MoveJobToEquipment(ctx context.Context, jobID, equipmentID, slotNo int) error { tx, err := d.client.Tx(ctx) if err != nil { return fmt.Errorf("move job %d to equipment: begin tx: %w", jobID, err) } defer tx.Rollback() posRef := fmt.Sprintf("%d:%d", equipmentID, slotNo) _, err = tx.Job.UpdateOneID(jobID). SetPositionType(constants.PositionType_OnEquipment). SetPositionRefId(posRef). AddVersion(1). Save(ctx) if err != nil { return fmt.Errorf("move job %d: update job: %w", jobID, err) } _, err = tx.EquipmentSlot.Update(). Where( equipmentslot.EquipmentIdEQ(equipmentID), equipmentslot.SlotNoEQ(slotNo), ). SetStatus(constants.SlotStatus_Occupied). SetCurrentJobId(jobID). Save(ctx) if err != nil { return fmt.Errorf("move job %d: update slot: %w", jobID, err) } return tx.Commit() } // MoveJobToBuffer 工件回暂存台 func (d *DBState) MoveJobToBuffer(ctx context.Context, jobID, slotNo int) error { _, err := d.client.Job.UpdateOneID(jobID). SetPositionType(constants.PositionType_OnBuffer). SetPositionRefId(fmt.Sprintf("%d", slotNo)). SetTempSlotNo(slotNo). AddVersion(1). Save(ctx) return err } // CompleteStep 完成当前步骤,推进到下一步 func (d *DBState) CompleteStep(ctx context.Context, jobID, currentStepIndex, nextStepIndex int, contextPatch map[string]any) error { update := d.client.Job.UpdateOneID(jobID). SetCurrentStepIndex(nextStepIndex). AddVersion(1) if contextPatch != nil { update = update.SetContext(contextPatch) } _, err := update.Save(ctx) return err } // FinishJob 工件终态 func (d *DBState) FinishJob(ctx context.Context, jobID, workOrderID int, terminalStatus constants.JobStatus) error { tx, err := d.client.Tx(ctx) if err != nil { return err } defer tx.Rollback() // 更新 job 为终态,条件为当前非终态 _, err = tx.Job.UpdateOneID(jobID). Where(job.StatusNotIn( constants.JobStatus_Completed, constants.JobStatus_Scrapped, )). SetStatus(terminalStatus). ClearTempSlotNo(). // 释放暂存台槽位 AddVersion(1). Save(ctx) if err != nil { return fmt.Errorf("finish job %d: %w", jobID, err) } // 条件递增工单计数器 if terminalStatus == constants.JobStatus_Completed { _, _ = tx.WorkOrder.UpdateOneID(workOrderID).AddFinishedNum(1).Save(ctx) } else if terminalStatus == constants.JobStatus_Scrapped { _, _ = tx.WorkOrder.UpdateOneID(workOrderID).AddFailNum(1).Save(ctx) } return tx.Commit() } // SetEquipmentSlot 更新设备槽位状态 func (d *DBState) SetEquipmentSlot(ctx context.Context, equipmentID, slotNo int, expectedStatus, nextStatus constants.SlotStatus, jobID int) error { n, err := d.client.EquipmentSlot.Update(). Where( equipmentslot.EquipmentIdEQ(equipmentID), equipmentslot.SlotNoEQ(slotNo), equipmentslot.StatusEQ(expectedStatus), ). SetStatus(nextStatus). SetCurrentJobId(jobID). Save(ctx) if err != nil { return fmt.Errorf("set slot %d-%d: %w", equipmentID, slotNo, err) } if n == 0 { return fmt.Errorf("set slot %d-%d: expected %s", equipmentID, slotNo, expectedStatus) } return nil } // GetJobState 读取单个 job 状态 func (d *DBState) GetJobState(ctx context.Context, jobID int) (*ent.Job, error) { return d.client.Job.Get(ctx, jobID) } // GetActiveJobs 获取所有非终态 job func (d *DBState) GetActiveJobs(ctx context.Context) ([]*ent.Job, error) { return d.client.Job.Query(). Where(job.StatusNotIn( constants.JobStatus_Completed, constants.JobStatus_Scrapped, constants.JobStatus_Cancelled, )). All(ctx) } // GetTempSlotOccupancy 扫描非终态 job 的 temp_slot_no 得到 1..8 占用位图 func (d *DBState) GetTempSlotOccupancy(ctx context.Context) (uint8, error) { jobs, err := d.GetActiveJobs(ctx) if err != nil { return 0, err } var bitmap uint8 for _, j := range jobs { if j.TempSlotNo != nil { slot := *j.TempSlotNo if slot >= 1 && slot <= 8 { bitmap |= 1 << (slot - 1) } } } return bitmap, nil } // parsePositionRef 解析 "equipmentID:slotNo" 格式 func parsePositionRef(ref string) (int, int, error) { for i := 0; i < len(ref); i++ { if ref[i] == ':' { a, err1 := strconv.Atoi(ref[:i]) b, err2 := strconv.Atoi(ref[i+1:]) if err1 != nil || err2 != nil { return 0, 0, fmt.Errorf("invalid position ref %q", ref) } return a, b, nil } } return 0, 0, fmt.Errorf("invalid position ref format %q", ref) } // RaiseAlarm 创建报警 func (d *DBState) RaiseAlarm(ctx context.Context, alarmCode, message string, level constants.AlarmLevel, equipmentID, jobID int, source string) error { _, err := d.client.Alarm.Create(). SetAlarmCode(alarmCode). SetAlarmMessage(message). SetLevel(level). SetNillableEquipmentId(&equipmentID). SetNillableJobId(&jobID). SetSource(source). Save(ctx) return err } // CreateManualAction 创建人工恢复动作 func (d *DBState) CreateManualAction(ctx context.Context, actionType constants.ManualActionType, jobID int, description string) error { _, err := d.client.ManualAction.Create(). SetActionType(actionType). SetJobId(jobID). SetContext(map[string]any{"description": description}). SetStatus(constants.ManualActionStatus_Pending). Save(ctx) return err } ``` - [ ] **Step 3: 编译验证** ```bash rtk go build ./internal/eventloop/... ``` --- ## Chunk 3: ProductionEventLoop 核心 ### Task 3.1: 创建 event loop 主循环 **Files:** - Create: `internal/eventloop/loop.go` - [ ] **Step 1: 写 loop.go** ```go package eventloop import ( "context" "fmt" "log/slog" "time" "hougai/ent" "hougai/internal/scheduler" "hougai/internal/station" "hougai/internal/robot" ) // ProductionEventLoop 产线事件循环:唯一生产状态写路径 type ProductionEventLoop struct { db *DBState sched *scheduler.Scheduler registry *station.StationRegistry robotCtrl *robot.Controller entClient *ent.Client // 消息队列 msgCh chan EventLoopMessage // 硬件 worker worker HardwareWorker // 运行时快照 jobRuntimes map[int]*RuntimeSnapshot // SSE/事件日志回调 onStateChanged func(jobID int, status string, positionType string, positionRefID string) onAlarmRaised func(alarmCode, message string, jobID int) // 停止 stopCh chan struct{} } // RuntimeSnapshot 工件运行时快照(轻量,只供调度用) type RuntimeSnapshot struct { JobID int WorkOrderID int Status string PositionType string PositionRefID string StepIndex int TempSlotNo int } func NewProductionEventLoop( entClient *ent.Client, sched *scheduler.Scheduler, registry *station.StationRegistry, robotCtrl *robot.Controller, worker HardwareWorker, ) *ProductionEventLoop { return &ProductionEventLoop{ db: NewDBState(entClient), sched: sched, registry: registry, robotCtrl: robotCtrl, entClient: entClient, worker: worker, msgCh: make(chan EventLoopMessage, 256), jobRuntimes: make(map[int]*RuntimeSnapshot), stopCh: make(chan struct{}), } } // Send 向事件循环投递消息(非阻塞,channel 满时阻塞) func (l *ProductionEventLoop) Send(msg EventLoopMessage) { l.msgCh <- msg } // SendSync 投递消息并等待结果(同步 API 使用) func (l *ProductionEventLoop) SendSync(msg EventLoopMessage, timeout time.Duration) (*MessageResult, error) { replyCh := make(chan MessageResult, 1) msg.Reply = replyCh l.msgCh <- msg select { case result := <-replyCh: return &result, nil case <-time.After(timeout): return nil, fmt.Errorf("event loop timeout") } } // Run 启动事件循环(阻塞,在 goroutine 中运行) func (l *ProductionEventLoop) Run(ctx context.Context) { slog.Info("event loop: started") ticker := time.NewTicker(500 * time.Millisecond) defer ticker.Stop() for { select { case msg := <-l.msgCh: l.handleMessage(ctx, msg) case <-ticker.C: l.trySchedule(ctx) case <-l.stopCh: slog.Info("event loop: stopped") return case <-ctx.Done(): slog.Info("event loop: context done") return } } } // Stop 停止事件循环 func (l *ProductionEventLoop) Stop() { close(l.stopCh) } ``` - [ ] **Step 2: 实现消息分发** ```go func (l *ProductionEventLoop) handleMessage(ctx context.Context, msg EventLoopMessage) { slog.Debug("event loop: handling message", "type", msg.Type, "id", msg.ID) switch msg.Type { // Command case CmdStartOrder: l.handleStartOrder(ctx, msg) case CmdPauseOrder: l.handlePauseOrder(ctx, msg) case CmdResumeOrder: l.handleResumeOrder(ctx, msg) case CmdCancelOrder: l.handleCancelOrder(ctx, msg) case CmdSuspendJob: l.handleSuspendJob(ctx, msg) case CmdResumeJob: l.handleResumeJob(ctx, msg) case CmdReworkJob: l.handleReworkJob(ctx, msg) case CmdConfirmRecovery: l.handleConfirmRecovery(ctx, msg) case CmdResolveManualAction: l.handleResolveManualAction(ctx, msg) // ExternalEvent case EvtMachineDone: l.handleMachineDone(ctx, msg) case EvtInspectionResult: l.handleInspectionResult(ctx, msg) case EvtPalletArrived: l.handlePalletArrived(ctx, msg) case EvtStepTimeout: l.handleStepTimeout(ctx, msg) case EvtRefillRequested: l.handleRefillRequested(ctx, msg) // WorkerResult case ResRobotActionSucceeded, ResRobotActionFailed, ResToolActionSucceeded, ResToolActionFailed, ResMachineStartSucceeded, ResMachineStartFailed: l.handleWorkerResult(ctx, msg) } } // reply 发送同步响应 func (l *ProductionEventLoop) reply(msg EventLoopMessage, success bool, errMsg string, data map[string]any) { if msg.Reply != nil { msg.Reply <- MessageResult{Success: success, Error: errMsg, Data: data} } } ``` - [ ] **Step 3: 编译验证** ```bash rtk go build ./internal/eventloop/... ``` ### Task 3.2: 实现调度触发方法 trySchedule **Files:** - Modify: `internal/eventloop/loop.go` (追加) - [ ] **Step 1: 实现 trySchedule** ```go func (l *ProductionEventLoop) trySchedule(ctx context.Context) { if l.sched == nil { return } // 从 DB 加载活跃 job 快照 activeJobs, err := l.db.GetActiveJobs(ctx) if err != nil { slog.Error("event loop: get active jobs failed", "error", err) return } if len(activeJobs) == 0 { return } // 更新内存快照 for _, j := range activeJobs { ts := j.TempSlotNo tsn := 0 if ts != nil { tsn = *ts } l.jobRuntimes[j.ID] = &RuntimeSnapshot{ JobID: j.ID, WorkOrderID: j.WorkOrderId, Status: string(j.Status), PositionType: string(j.PositionType), PositionRefID: j.PositionRefId, StepIndex: j.CurrentStepIndex, TempSlotNo: tsn, } } // 构建 JobView + SystemState 并调用 scheduler // 这里适配现有 scheduler_adapter.go 的逻辑 candidates := l.runScheduler(ctx, activeJobs) for _, ct := range candidates { l.submitCandidate(ctx, ct) } } ``` - [ ] **Step 2: 编译验证** ```bash rtk go build ./internal/eventloop/... ``` ### Task 3.3: 实现核心 handler 方法(MachineDone / InspectionResult / WorkerResult) **Files:** - Modify: `internal/eventloop/loop.go`(追加) - [ ] **Step 1: 实现 handleMachineDone** ```go func (l *ProductionEventLoop) handleMachineDone(ctx context.Context, msg EventLoopMessage) { machineID := intFromPayload(msg.Payload, "machineId") if machineID == 0 { return } slotNo := intFromPayload(msg.Payload, "slot") jobIDFromPayload := intFromPayload(msg.Payload, "jobId") // 查询设备是否批量完成 equip, err := l.entClient.Equipment.Get(ctx, machineID) if err != nil { slog.Error("event loop: machine done, equipment not found", "machineId", machineID, "error", err) return } if equip.Batch { // 批量完成:所有 OCCUPIED 槽位 → DONE slots, _ := l.entClient.EquipmentSlot.Query(). Where(equipmentslot.EquipmentIdEQ(machineID), equipmentslot.StatusEQ(constants.SlotStatus_Occupied)). All(ctx) for _, s := range slots { l.db.SetEquipmentSlot(ctx, machineID, s.SlotNo, constants.SlotStatus_Occupied, constants.SlotStatus_Done, s.CurrentJobId) l.updateJobToWaitingUnload(ctx, s.CurrentJobId) } } else { // 非批量:确定目标槽位 targetSlot := slotNo if targetSlot < 1 { // 多槽位非批量设备 FIFO slot, _ := l.entClient.EquipmentSlot.Query(). Where(equipmentslot.EquipmentIdEQ(machineID), equipmentslot.StatusEQ(constants.SlotStatus_Occupied)). Order(ent.Asc(equipmentslot.FieldSlotNo)). First(ctx) if slot != nil { targetSlot = slot.SlotNo } } // 交叉校验 jobID slot, _ := l.entClient.EquipmentSlot.Query(). Where(equipmentslot.EquipmentIdEQ(machineID), equipmentslot.SlotNoEQ(targetSlot)). First(ctx) if slot == nil { return } if jobIDFromPayload > 0 && slot.CurrentJobId != jobIDFromPayload { l.db.RaiseAlarm(ctx, "SLOT_JOB_MISMATCH", fmt.Sprintf("machine %d slot %d: expected job %d, got %d", machineID, targetSlot, slot.CurrentJobId, jobIDFromPayload), constants.AlarmLevel_ERROR, machineID, slot.CurrentJobId, "event-loop") return } l.db.SetEquipmentSlot(ctx, machineID, targetSlot, constants.SlotStatus_Occupied, constants.SlotStatus_Done, slot.CurrentJobId) l.updateJobToWaitingUnload(ctx, slot.CurrentJobId) } l.trySchedule(ctx) } ``` - [ ] **Step 2: 实现 handleInspectionResult** ```go func (l *ProductionEventLoop) handleInspectionResult(ctx context.Context, msg EventLoopMessage) { machineID := intFromPayload(msg.Payload, "machineId") jobID := intFromPayload(msg.Payload, "jobId") pass, _ := msg.Payload["pass"].(bool) job, err := l.entClient.Job.Get(ctx, jobID) if err != nil { return } // 写入检测结果到 context ctx_ := job.Context if ctx_ == nil { ctx_ = make(map[string]any) } ctx_["inspectionPass"] = pass _, _ = l.entClient.Job.UpdateOneID(jobID).SetContext(ctx_).Save(ctx) if pass { // 检测通过:推进步骤 l.advanceJobStep(ctx, jobID) } else { // 检测 NG:创建报警 l.db.RaiseAlarm(ctx, "INSPECTION_FAIL", fmt.Sprintf("job %d inspection failed", jobID), constants.AlarmLevel_WARN, machineID, jobID, "event-loop") l.db.FinishJob(ctx, jobID, job.WorkOrderId, constants.JobStatus_Scrapped) } l.trySchedule(ctx) } ``` - [ ] **Step 3: 实现 handleWorkerResult(含 stale result 防护)** ```go func (l *ProductionEventLoop) handleWorkerResult(ctx context.Context, msg EventLoopMessage) { taskID := msg.CorrelationID if taskID == "" { slog.Warn("event loop: worker result without correlation ID") return } // 查询 task 是否仍在 RUNNING task, err := l.entClient.Task.Get(ctx, taskID) if err != nil || task.Status != constants.TaskStatus_Running { slog.Info("event loop: stale worker result ignored", "taskId", taskID, "status", task.Status) return } jobID := intFromPayload(msg.Payload, "jobId") success := msg.Type == ResRobotActionSucceeded || msg.Type == ResToolActionSucceeded || msg.Type == ResMachineStartSucceeded if success { // 标记 task 完成 l.entClient.Task.UpdateOneID(taskID). SetStatus(constants.TaskStatus_Completed). SetCompletedAt(time.Now()). Save(ctx) // 推进 job 步骤 l.advanceJobStep(ctx, jobID) } else { // 标记 task 失败 l.entClient.Task.UpdateOneID(taskID). SetStatus(constants.TaskStatus_Failed). SetCompletedAt(time.Now()). Save(ctx) // job 挂起 l.entClient.Job.UpdateOneID(jobID). SetStatus(constants.JobStatus_Suspended). SetSuspendedReason("action_failed"). Save(ctx) l.db.RaiseAlarm(ctx, "ACTION_FAILED", fmt.Sprintf("job %d task %s failed", jobID, taskID), constants.AlarmLevel_ERROR, 0, jobID, "event-loop") } l.trySchedule(ctx) } ``` - [ ] **Step 4: 实现 advanceJobStep 和 updateJobToWaitingUnload 辅助方法** ```go func (l *ProductionEventLoop) advanceJobStep(ctx context.Context, jobID int) { job, err := l.entClient.Job.Get(ctx, jobID) if err != nil { return } l.db.CompleteStep(ctx, jobID, job.CurrentStepIndex, job.CurrentStepIndex+1, nil) } func (l *ProductionEventLoop) updateJobToWaitingUnload(ctx context.Context, jobID int) { _, _ = l.entClient.Job.UpdateOneID(jobID). SetStatus(constants.JobStatus_WaitingUnload). AddVersion(1). Save(ctx) } ``` - [ ] **Step 5: 添加 intFromPayload 辅助函数** ```go func intFromPayload(p map[string]any, key string) int { switch v := p[key].(type) { case int: return v case float64: return int(v) } return 0 } ``` - [ ] **Step 6: 编译验证** ```bash rtk go build ./internal/eventloop/... ``` --- ## Chunk 4: ServiceContext 重构 ### Task 4.1: 重写 service_context.go **Files:** - Modify: `internal/svc/service_context.go` - Modify: `internal/config/config.go`(删除冗余引用) - [ ] **Step 1: 删除 go-redis 依赖和 Redis/SSOT 初始化代码** ```go // 删除 // import goredis "github.com/redis/go-redis/v9" // 删除 go-redis 客户端初始化 // var goRedisClient *goredis.Client // if c.EventBus.Enabled || c.SSOT.Enabled { ... } // 删除 RedisBus 初始化 // bus = eventbus.NewRedisBus(...) // 删除 RedisStateManager 初始化 // stateMgr = state.NewRedisStateManager(goRedisClient) // 删除事件重放 // replayer := recovery.NewReplayer(entClient, stateMgr) // 删除 SSOT 写穿 // writer := state.NewWriteThroughHelper(stateMgr, true) // 删除 ZPopLoop // go scheduler.ZPopLoop(...) // 删除状态校验器 // verifier := verify.NewStateVerifier(stateMgr, ...) // 删除 RedisToolLocker // toolLocker := state.NewRedisToolLocker(goRedisClient) ``` - [ ] **Step 2: 新增 EventLoop + HardwareWorker 初始化** ```go // 事件总线:始终使用 LocalBus(单进程模式) bus := eventbus.NewLocalBus() // 硬件 worker:封装 robot controller worker := processor.NewRobotWorker(robotCtrl, scanner, marker) // 事件循环引擎 var sched *scheduler.Scheduler sched = scheduler.DefaultScheduler() sched.SetToolLockChecker(nil) // no tool lock needed in single-worker mode eventLoop := eventloop.NewProductionEventLoop( entClient, sched, registry, robotCtrl, worker, ) eventLoop.SetOnStateChanged(func(jobID int, status, positionType, positionRefID string) { bus.Publish(context.Background(), eventbus.Event{ ID: fmt.Sprintf("el-%d-%d", jobID, time.Now().UnixNano()), Type: eventbus.EventJobStatusChanged, Payload: map[string]interface{}{ "jobId": jobID, "status": status, "positionType": positionType, "positionRefId": positionRefID, }, }) }) eventLoop.SetOnAlarmRaised(func(alarmCode, message string, jobID int) { bus.Publish(context.Background(), eventbus.Event{ Type: eventbus.EventAlarmRaised, Payload: map[string]interface{}{ "alarmCode": alarmCode, "message": message, "jobId": jobID, }, }) }) // SignalWatcher 回调改为投递到 event loop signalWatcher.OnMachineDone = func(machineID int) { eventLoop.Send(eventloop.EventLoopMessage{ ID: fmt.Sprintf("sw-%d-%d", machineID, time.Now().UnixNano()), Type: eventloop.EvtMachineDone, Payload: map[string]any{"machineId": machineID}, }) } signalWatcher.OnInspectionResult = func(machineID int, pass bool) { eventLoop.Send(eventloop.EventLoopMessage{ ID: fmt.Sprintf("sw-insp-%d-%d", machineID, time.Now().UnixNano()), Type: eventloop.EvtInspectionResult, Payload: map[string]any{"machineId": machineID, "pass": pass}, }) } // 启动事件循环(在 bus 之后) go eventLoop.Run(context.Background()) ``` - [ ] **Step 3: ServiceContext 字段更新** ```go type ServiceContext struct { Config config.Config EntClient *ent.Client SSEHandler *sse.Handler PLCManager *plc.Manager RobotManager *robot.Manager StationRegistry *station.StationRegistry RecipeLoader *processor.RecipeLoader EventLoop *eventloop.ProductionEventLoop // 替代 OrderProcessor EventBus eventbus.Bus // 仅 LocalBus EventLogWriter *eventlog.EventLogWriter SSEBridge *sse.EventBridge } ``` - [ ] **Step 4: 添加 SSE/EventLog 回调注册任务** 在 `NewProductionEventLoop` 中添加额外回调设置方法,并在 service_context 中串联: ```go // loop.go 追加 func (l *ProductionEventLoop) SetOnStateChanged(fn func(jobID int, status, positionType, positionRefID string)) { l.onStateChanged = fn } func (l *ProductionEventLoop) SetOnAlarmRaised(fn func(alarmCode, message string, jobID int)) { l.onAlarmRaised = fn } ``` 在 loop 的 `handleWorkerResult`、`advanceJobStep`、`updateJobToWaitingUnload` 等方法末尾调用: ```go if l.onStateChanged != nil { l.onStateChanged(jobID, string(job.Status), string(job.PositionType), job.PositionRefId) } ``` 在 service_context 中,EventLoop 初始化后注册: ```go eventLoop.SetOnStateChanged(func(jobID int, status, positionType, positionRefID string) { bus.Publish(context.Background(), eventbus.NewJobStatusChangedEvent( jobID, 0, status, positionType, positionRefID, 0, )) }) eventLoop.SetOnAlarmRaised(func(alarmCode, message string, jobID int) { bus.Publish(context.Background(), eventbus.Event{ ID: fmt.Sprintf("alarm-%d-%d", jobID, time.Now().UnixNano()), Type: eventbus.EventAlarmRaised, Payload: map[string]interface{}{ "alarmCode": alarmCode, "message": message, "jobId": jobID, }, }) }) ``` 同时保留 `bus.SubscribeAll` 订阅所有事件写入 `event_log`: ```go bus.SubscribeAll(func(ctx context.Context, evt eventbus.Event) error { eventLogWriter.Write(evt) return nil }) ``` - [ ] **Step 5: 编译验证** ```bash rtk go build ./... ``` --- ## Chunk 5: Processor 精简 + Worker 适配 ### Task 5.1: 创建 RobotWorker 适配器 **Files:** - Create: `internal/processor/robot_worker.go` - [ ] **Step 1: 实现 HardwareWorker 接口** ```go package processor import ( "context" "log/slog" "hougai/internal/eventloop" "hougai/internal/robot" "hougai/internal/station" ) // RobotWorker 适配 robot.Controller + handheld tools 到 eventloop.HardwareWorker type RobotWorker struct { robotCtrl *robot.Controller scanner station.HandheldTool marker station.HandheldTool } func NewRobotWorker(robotCtrl *robot.Controller, scanner, marker station.HandheldTool) *RobotWorker { return &RobotWorker{robotCtrl: robotCtrl, scanner: scanner, marker: marker} } func (w *RobotWorker) Execute(ctx context.Context, action eventloop.RobotAction) error { slog.Info("robot worker: executing", "taskId", action.TaskID, "kind", action.Kind, "jobId", action.JobID) switch action.Kind { case "load": return w.robotCtrl.LoadToMachine(ctx, action.MachineID, action.SlotNo) case "unload": return w.robotCtrl.UnloadFromMachine(ctx, action.MachineID, action.SlotNo, action.TargetSlotNo) case "exchange": return w.robotCtrl.ExchangeAtMachine(ctx, action.MachineID, action.SlotNo, action.TargetSlotNo) case "replenish": return w.robotCtrl.ReplenishFromDock(ctx, action.TargetSlotNo) case "scan": return w.scanner.Execute(ctx, action.Params) case "mark": return w.marker.Execute(ctx, action.Params) default: slog.Warn("robot worker: unknown action kind", "kind", action.Kind) return nil } } ``` - [ ] **Step 2: 编译验证** ```bash rtk go build ./internal/processor/... ``` ### Task 5.2: 精简 job_processor.go **Files:** - Modify: `internal/processor/job_processor.go` - [ ] **Step 1: 删除 ssotWriter、toolLocker、readyQueue、scheduleMu 字段** ```go // 删除以下字段 // ssotWriter *state.WriteThroughHelper // toolLocker state.ToolLocker // readyQueue *ReadyQueue // scheduleMu sync.Mutex ``` - [ ] **Step 2: 删除 SetSSOTWriter、SetToolLocker、SetStateManager 方法** - [ ] **Step 3: 删除 startSubsystems 方法(由 event loop 替代)** - [ ] **Step 4: 删除 notifyReady、ScheduleAndSubmit、scheduleReadyJobs、scheduleAllActiveJobs、SyncSSOTStatus、ScheduleSingleJob 方法** - [ ] **Step 5: 保留 StartOrder、PauseOrder 等方法签名为公开方法,内部改为向 event loop 投递消息** ```go func (jp *JobProcessor) PauseOrder(eventLoop *eventloop.ProductionEventLoop, orderID int) error { result, err := eventLoop.SendSync(eventloop.EventLoopMessage{ ID: fmt.Sprintf("cmd-pause-order-%d-%d", orderID, time.Now().UnixNano()), Type: eventloop.CmdPauseOrder, Payload: map[string]any{"orderId": orderID}, }, 5*time.Second) if err != nil { return fmt.Errorf("pause order %d: %w", orderID, err) } if !result.Success { return fmt.Errorf("pause order %d: %s", orderID, result.Error) } return nil } ``` - [ ] **Step 6: 编译验证** ```bash rtk go build ./... ``` ### Task 5.3: 精简 job_runtime.go **Files:** - Modify: `internal/processor/job_runtime.go` - [ ] **Step 1: 删除 onStatusSync、ssotVersion、terminalCounted、notifyReady 字段** - [ ] **Step 2: 删除 syncSSOTStatus 调用(syncStatusToDB 中移除 onStatusSync 分支)** - [ ] **Step 3: 保留 buildTask、buildLoadAction、buildUnloadAction 等动作构建方法** - [ ] **Step 4: 编译验证** ```bash rtk go build ./internal/processor/... ``` --- ## Chunk 6: 恢复 & 启动流程 ### Task 6.1: 创建恢复模块 **Files:** - Create: `internal/eventloop/recovery.go` - [ ] **Step 1: 写恢复逻辑** ```go package eventloop import ( "context" "log/slog" "hougai/constants" "hougai/ent/workorder" ) // RecoverOnStartup 启动时执行保守人工确认恢复 // 1. 将所有 IN_PROGRESS 的 work_order 改为 PAUSED // 2. 将所有非终态 job 改为 SUSPENDED // 3. 校验 consistency(temp_slot_no 重复/越界、equipment_slot 与 job 位置不匹配) // 4. 不一致项创建 manual_action func (d *DBState) RecoverOnStartup(ctx context.Context) (recoveredJobs int, manualActions int, err error) { // 将所有 IN_PROGRESS 工单改为 PAUSED paused, err := d.client.WorkOrder.Update(). Where(workorder.StatusEQ(constants.WorkOrderStatus_InProgress)). SetStatus(constants.WorkOrderStatus_Paused). Save(ctx) if err != nil { return 0, 0, err } slog.Info("recovery: paused work orders", "count", paused) // 校验 temp_slot_no 一致性 jobs, err := d.GetActiveJobs(ctx) if err != nil { return 0, 0, err } slotUsers := make(map[int][]int) // tempSlotNo -> []jobID for _, j := range jobs { if j.TempSlotNo != nil { slot := *j.TempSlotNo if slot < 1 || slot > 8 { // 越界槽位 d.CreateManualAction(ctx, constants.ManualActionType_RecoveryRequired, j.ID, "temp_slot_no out of range") manualActions++ continue } slotUsers[slot] = append(slotUsers[slot], j.ID) } } for slot, users := range slotUsers { if len(users) > 1 { slog.Warn("recovery: duplicate temp slot", "slot", slot, "jobs", users) for _, jID := range users { d.CreateManualAction(ctx, constants.ManualActionType_RecoveryRequired, jID, "duplicate temp_slot_no") manualActions++ } } } // 校验 equipment_slot 与 job 位置一致性 for _, j := range jobs { if j.PositionType == constants.PositionType_OnEquipment { equipID, slotNo, slotErr := parsePositionRef(j.PositionRefId) if slotErr != nil { continue } slot, slotErr := d.client.EquipmentSlot.Query(). Where(equipmentslot.EquipmentIdEQ(equipID), equipmentslot.SlotNoEQ(slotNo)). First(ctx) if slotErr != nil || slot.CurrentJobId != j.ID { d.CreateManualAction(ctx, constants.ManualActionType_RecoveryRequired, j.ID, "equipment_slot inconsistent with job.position") manualActions++ } } } return len(jobs), manualActions, nil } ``` - [ ] **Step 2: 编译验证** ```bash rtk go build ./internal/eventloop/... ``` ### Task 6.2: 在 service_context.go 中接入恢复流程 - [ ] **Step 1: 在 eventLoop.Run() 之前添加恢复调用** ```go // 启动恢复 recovered, manualActions, err := eventloop.NewDBState(entClient).RecoverOnStartup(context.Background()) if err != nil { slog.Error("svc: startup recovery failed", "error", err) } else { slog.Info("svc: startup recovery complete", "recoveredJobs", recovered, "manualActions", manualActions) } // 启动事件循环 go eventLoop.Run(context.Background()) ``` --- ## Chunk 7: 清理死代码 ### Task 7.1: 删除 Redis 依赖文件 - [ ] **Step 1: 删除 state 包中 Redis 相关文件** ```bash rm internal/state/redis_manager.go rm internal/state/state_write_through.go rm internal/state/keys.go ``` - [ ] **Step 2: 删除 scheduler 中 Redis 相关文件** ```bash rm internal/scheduler/zpop.go ``` - [ ] **Step 3: 删除 recovery 中 event_log 重放文件** ```bash rm internal/recovery/replayer.go rm internal/recovery/replayer_test.go ``` - [ ] **Step 4: 删除 verify 中 Redis 校验文件** ```bash rm internal/verify/state_verifier.go rm internal/verify/state_verifier_test.go ``` - [ ] **Step 5: 删除 processor 中已废弃文件** ```bash rm internal/processor/ready_queue.go rm internal/processor/ready_queue_test.go rm internal/processor/temp_slot_allocator.go rm internal/processor/temp_slot_allocator_test.go rm internal/processor/ssot_integration_test.go ``` - [ ] **Step 6: 编译验证** ```bash rtk go build ./... ``` --- ## Chunk 8: EventBus 简化 ### Task 8.1: 删除 RedisBus,保留 LocalBus + Bus 接口 **Files:** - Modify: `internal/eventbus/bus.go` - Delete: `internal/eventbus/local.go` 中的 redis import(如果有) - [ ] **Step 1: 从 bus.go 删除 RedisBus 实现,仅保留 Bus 接口和 Event 结构体** - [ ] **Step 2: 确保 LocalBus 完整可用** - [ ] **Step 3: 编译验证** ```bash rtk go build ./internal/eventbus/... ``` --- ## Chunk 9: Handler/Logic 适配 ### Task 9.1: 更新 handler 以使用 EventLoop **Files:** - Modify: `internal/handler/` 下的 handler 文件(通过 api 生成) 由于 handler 是自动生成的,修改应集中在 logic 层: - [ ] **Step 1: 在 logic 文件中将 OrderProcessor 调用替换为 EventLoop.Send/SendSync** ```go // 旧: l.svcCtx.OrderProcessor.StartOrder(ctx, orderID) // 新: l.svcCtx.EventLoop.Send(eventloop.EventLoopMessage{...}) ``` - [ ] **Step 2: 编译验证** ```bash rtk go build ./... ``` --- ## Chunk 10: 测试 ### Task 10.1: DBState 集成测试 **Files:** - Create: `internal/eventloop/dbstate_test.go` - [ ] **Step 1: 写 ClaimJobForAction 测试** ```go func TestDBState_ClaimJobForAction(t *testing.T) { // 使用 ent 内存 SQLite 或测试 PG // 创建 job (ON_BUFFER) → ClaimJobForAction → 验证 IN_HANDLING // 验证重复 Claim 失败 } ``` - [ ] **Step 2: 写 MoveJobToEquipment/FinishJob 事务测试** - [ ] **Step 3: 写 GetTempSlotOccupancy 测试** - [ ] **Step 4: 运行测试** ```bash rtk go test ./internal/eventloop/... -v ``` ### Task 10.2: EventLoop 单元测试 **Files:** - Create: `internal/eventloop/loop_test.go` - [ ] **Step 1: 写消息处理测试(使用 mock worker)** ```go func TestEventLoop_StartOrder(t *testing.T) { // 投递 CmdStartOrder → 验证 job 状态变更 → 验证 reply } ``` - [ ] **Step 2: 写 worker result 关联测试** - [ ] **Step 3: 写 stale result 忽略测试** - [ ] **Step 4: 运行测试** ```bash rtk go test ./internal/eventloop/... -v ``` ### Task 10.3: 恢复测试 **Files:** - Create: `internal/eventloop/recovery_test.go` - [ ] **Step 1: 写 duplicate temp slot 测试** - [ ] **Step 2: 写 temp slot 越界测试** - [ ] **Step 3: 运行测试** ```bash rtk go test ./internal/eventloop/... -v ``` ### Task 10.4: 运行全量测试 - [ ] **Step 1: 修复编译失败** ```bash rtk go build ./... ``` - [ ] **Step 2: 运行测试** ```bash rtk go test ./... 2>&1 | head -100 ``` - [ ] **Step 3: 修复测试失败** --- ## 验收检查 - [ ] `go build ./...` 通过 - [ ] `go vet ./...` 通过 - [ ] 核心流程不依赖 Redis(搜索 `go-redis` import 仅出现在非核心文件) - [ ] 配置不再需要 `EventBus.Enabled`、`SSOT.Enabled` - [ ] ProductionEventLoop 是唯一状态写路径 - [ ] Mock E2E(`go run cmd/mockrun/main.go` 或等价)通过完整工单流程