From 44fb21eb6a1a3210ecf52f810a8c2002d4255b80 Mon Sep 17 00:00:00 2001 From: break Date: Thu, 2 Jul 2026 15:13:22 +0900 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=AF=BC=E5=85=A5=E9=87=8D?= =?UTF-8?q?=E8=AF=95=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/task/device_import.go | 13 ++++++++++++- internal/task/iot_card_import.go | 13 ++++++++++++- 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/internal/task/device_import.go b/internal/task/device_import.go index 0ac6591..0448e52 100644 --- a/internal/task/device_import.go +++ b/internal/task/device_import.go @@ -85,7 +85,18 @@ func (h *DeviceImportHandler) HandleDeviceImport(ctx context.Context, task *asyn return asynq.SkipRetry } - if importTask.Status != model.ImportTaskStatusPending { + switch importTask.Status { + case model.ImportTaskStatusPending: + // 正常首次处理 + case model.ImportTaskStatusProcessing: + // 上次 worker 中途中断（重启/崩溃），重置计数重新处理 + // 已入库的设备会被 ExistsByDeviceNoBatch 识别为 skip，不会重复写入 + h.logger.Warn("检测到导入任务上次处理中途中断，重置进度重新处理", + zap.Uint("task_id", payload.TaskID), + ) + h.importTaskStore.UpdateProgress(ctx, importTask.ID, 0, 0, 0) + default: + // 已完成或失败，不重复处理 h.logger.Info("导入任务已处理，跳过", zap.Uint("task_id", payload.TaskID), zap.Int("status", importTask.Status), diff --git a/internal/task/iot_card_import.go b/internal/task/iot_card_import.go index 84fd9f1..82f891d 100644 --- a/internal/task/iot_card_import.go +++ b/internal/task/iot_card_import.go @@ -92,7 +92,18 @@ func (h *IotCardImportHandler) HandleIotCardImport(ctx context.Context, task *as return asynq.SkipRetry } - if importTask.Status != model.ImportTaskStatusPending { + switch importTask.Status { + case model.ImportTaskStatusPending: + // 正常首次处理 + case model.ImportTaskStatusProcessing: + // 上次 worker 中途中断（重启/崩溃），重置计数重新处理 + // 已入库的卡会被 ExistsByICCIDBatch 识别为 skip，不会重复写入 + h.logger.Warn("检测到导入任务上次处理中途中断，重置进度重新处理", + zap.Uint("task_id", payload.TaskID), + ) + h.importTaskStore.UpdateProgress(ctx, importTask.ID, 0, 0, 0) + default: + // 已完成或失败，不重复处理 h.logger.Info("导入任务已处理，跳过", zap.Uint("task_id", payload.TaskID), zap.Int("status", importTask.Status),