From a33ee7d4a09a249098a47453b5214a482702f1cb Mon Sep 17 00:00:00 2001 From: Haruko386 Date: Fri, 21 Aug 2026 15:41:40 +0800 Subject: [PATCH] fix: dataset ingestion log has no other type instead of success (#18605) --- .../ingestion/service/ingestion_service.go | 50 ++++++++-- internal/ingestion/task/pipeline_executor.go | 95 +++++++++++++++---- .../ingestion/task/pipeline_executor_test.go | 83 ++++++++++++++++ internal/service/ingestion_task_service.go | 10 +- 4 files changed, 207 insertions(+), 31 deletions(-) diff --git a/internal/ingestion/service/ingestion_service.go b/internal/ingestion/service/ingestion_service.go index 46d882b824..d0a5b24dec 100644 --- a/internal/ingestion/service/ingestion_service.go +++ b/internal/ingestion/service/ingestion_service.go @@ -599,17 +599,29 @@ func (e *Ingestor) runTask(ctx context.Context, task *entity.IngestionTask) bool case <-ctx.Done(): common.Info(fmt.Sprintf("Task %s cancelled", task.ID)) e.markCancelProgress(task) - return e.markStopped(context.Background(), task.ID) + stopped := e.markStopped(context.Background(), task.ID) + if stopped { + e.recordTerminalPipelineLog(context.Background(), task, string(entity.TaskStatusCancel)) + } + return stopped default: } if err := e.ingestionTaskSvc.IncrementRunCount(ctx, task.ID); err != nil { common.Error(fmt.Sprintf("Failed to increment run count for task %s", task.ID), err) - return e.markFailed(ctx, task.ID) + ok := e.markFailed(ctx, task.ID) + if ok { + e.recordTerminalPipelineLog(ctx, task, string(entity.TaskStatusFail)) + } + return ok } if err := e.ingestionTaskSvc.ClearComponentProgress(ctx, task.ID); err != nil { common.Error(fmt.Sprintf("Failed to clear previous component progress for task %s", task.ID), err) - return e.markFailed(ctx, task.ID) + ok := e.markFailed(ctx, task.ID) + if ok { + e.recordTerminalPipelineLog(ctx, task, string(entity.TaskStatusFail)) + } + return ok } // This is a new run (IncrementRunCount succeeded). Any Redis cancel flag @@ -630,15 +642,27 @@ func (e *Ingestor) runTask(ctx context.Context, task *entity.IngestionTask) bool if errors.Is(err, context.Canceled) { common.Info(fmt.Sprintf("Task %s cancelled during pipeline", task.ID)) e.markCancelProgress(task) - return e.markStopped(ctx, task.ID) + stopped := e.markStopped(ctx, task.ID) + if stopped { + e.recordTerminalPipelineLog(ctx, task, string(entity.TaskStatusCancel)) + } + return stopped } if errors.Is(err, context.DeadlineExceeded) { common.Info(fmt.Sprintf("Task %s timed out during pipeline", task.ID)) e.markTimeoutProgress(task) - return e.markFailed(ctx, task.ID) + ok := e.markFailed(ctx, task.ID) + if ok { + e.recordTerminalPipelineLog(ctx, task, string(entity.TaskStatusFail)) + } + return ok } common.Error(fmt.Sprintf("Task %s failed", task.ID), err) - return e.markFailed(ctx, task.ID) + ok := e.markFailed(ctx, task.ID) + if ok { + e.recordTerminalPipelineLog(ctx, task, string(entity.TaskStatusFail)) + } + return ok } if err := e.completeTask(ctx, task.ID); err != nil { @@ -981,6 +1005,20 @@ func (e *Ingestor) defaultRunDocumentTask(ctx context.Context, ingestionTask *en return nil } +func (e *Ingestor) recordTerminalPipelineLog(ctx context.Context, ingestionTask *entity.IngestionTask, status string) { + if ingestionTask == nil || status == "" { + return + } + ctx = context.WithoutCancel(ctx) + if err := taskpkg.RecordPipelineLog(ctx, dao.DB, taskpkg.PipelineLogInput{ + KbID: ingestionTask.DatasetID, + DocumentID: ingestionTask.DocumentID, + Status: status, + }); err != nil { + common.Warn(fmt.Sprintf("record terminal pipeline log for task %s document %s: %v", ingestionTask.ID, ingestionTask.DocumentID, err)) + } +} + // Stop gracefully shuts down the ingestor. It cancels the root context so // idle workers exit immediately and in-flight pipelines abort at their next // ctx.Err() check, then waits for workers to return. The wait is bounded by diff --git a/internal/ingestion/task/pipeline_executor.go b/internal/ingestion/task/pipeline_executor.go index dfab1be12e..496b989018 100644 --- a/internal/ingestion/task/pipeline_executor.go +++ b/internal/ingestion/task/pipeline_executor.go @@ -198,7 +198,7 @@ func (s *PipelineExecutor) Execute(ctx context.Context) (*PipelineResult, error) } if pipelineDSL != "" { - s.recordPipelineLog(ctx, dao.DB, s.taskCtx.Doc.ID, pipelineDSL, "done") + s.recordPipelineLog(context.WithoutCancel(ctx), dao.DB, s.taskCtx.Doc.ID, pipelineDSL, "") } return result, nil @@ -628,22 +628,65 @@ func wikiActiveStates(output map[string]any) ([]kccommon.WikiMapActiveState, err } } -func (s *PipelineExecutor) recordPipelineLog(ctx context.Context, db *gorm.DB, docID, dsl, status string) { +// PipelineLogInput contains the identifiers and optional snapshots needed to +// persist a pipeline operation log without constructing a PipelineExecutor. +type PipelineLogInput struct { + TenantID string + KbID string + DocumentID string + PipelineID string + DSL string + Status string + Document entity.Document +} + +// RecordPipelineLog persists a pipeline operation log without requiring +// executor setup. Callers that already know a terminal state should pass it in +// Status; otherwise the writer falls back to the latest document.run value. +func RecordPipelineLog(ctx context.Context, db *gorm.DB, input PipelineLogInput) error { + return recordPipelineLog(ctx, db, input, dao.NewPipelineOperationLogDAO().Create) +} + +func recordPipelineLog( + ctx context.Context, + db *gorm.DB, + input PipelineLogInput, + createFunc func(ctx context.Context, db *gorm.DB, log *entity.PipelineOperationLog) error, +) error { var dslMap entity.JSONMap - if err := json.Unmarshal([]byte(dsl), &dslMap); err != nil { - dslMap = entity.JSONMap{"raw": dsl} + if strings.TrimSpace(input.DSL) == "" { + dslMap = entity.JSONMap{} + } else if err := json.Unmarshal([]byte(input.DSL), &dslMap); err != nil { + dslMap = entity.JSONMap{"raw": input.DSL} } // The task context contains the document snapshot loaded when the task // started. Reload it here so the operation log reflects the final progress // state written by the progress sink, matching the Python operation-log // creation path. - doc := s.taskCtx.Doc + doc := input.Document + if doc.ID == "" { + doc.ID = input.DocumentID + doc.KbID = input.KbID + } if db != nil { - if persisted, err := dao.NewDocumentDAO().GetByID(ctx, db, docID); err == nil && persisted != nil { + if persisted, err := dao.NewDocumentDAO().GetByID(ctx, db, input.DocumentID); err == nil && persisted != nil { doc = *persisted } else if err != nil { - common.Warn(fmt.Sprintf("failed to reload document %s for pipeline log: %v", docID, err)) + common.Warn(fmt.Sprintf("failed to reload document %s for pipeline log: %v", input.DocumentID, err)) + } + } + if input.KbID == "" { + input.KbID = doc.KbID + } + if input.PipelineID == "" && doc.PipelineID != nil { + input.PipelineID = strings.TrimSpace(*doc.PipelineID) + } + if input.TenantID == "" && db != nil && input.KbID != "" { + if kb, err := dao.NewKnowledgebaseDAO().GetByID(ctx, db, input.KbID); err == nil && kb != nil { + input.TenantID = kb.TenantID + } else if err != nil { + return fmt.Errorf("load knowledgebase %s for pipeline log: %w", input.KbID, err) } } @@ -655,22 +698,22 @@ func (s *PipelineExecutor) recordPipelineLog(ctx context.Context, db *gorm.DB, d pipelineTitle := doc.ParserID pipelineAvatar := doc.Thumbnail var pipelineID *string - if s.taskCtx.PipelineID != "" { - pipelineID = &s.canvasID - if db != nil { - if canvas, err := dao.NewUserCanvasDAO().GetByID(ctx, db, s.canvasID); err == nil && canvas != nil { + if input.PipelineID != "" { + pipelineID = &input.PipelineID + if db != nil && strings.TrimSpace(input.DSL) != "" { + if canvas, err := dao.NewUserCanvasDAO().GetByID(ctx, db, input.PipelineID); err == nil && canvas != nil { if canvas.Title != nil { pipelineTitle = *canvas.Title } pipelineAvatar = canvas.Avatar } else if err != nil && !errors.Is(err, dao.ErrUserCanvasNotFound) { - common.Warn(fmt.Sprintf("failed to reload pipeline %s for operation log: %v", s.canvasID, err)) + common.Warn(fmt.Sprintf("failed to reload pipeline %s for operation log: %v", input.PipelineID, err)) } } } - operationStatus := status - if doc.Run != nil && *doc.Run != "" { + operationStatus := input.Status + if operationStatus == "" && doc.Run != nil && *doc.Run != "" { operationStatus = *doc.Run } statusValue := "1" @@ -687,9 +730,9 @@ func (s *PipelineExecutor) recordPipelineLog(ctx context.Context, db *gorm.DB, d } log := &entity.PipelineOperationLog{ ID: utility.GenerateUUID(), - TenantID: s.Tenant().ID, - KbID: s.KB().ID, - DocumentID: docID, + TenantID: input.TenantID, + KbID: input.KbID, + DocumentID: input.DocumentID, PipelineID: pipelineID, PipelineTitle: &pipelineTitle, TaskType: string(entity.PipelineTaskTypeParse), @@ -707,7 +750,23 @@ func (s *PipelineExecutor) recordPipelineLog(ctx context.Context, db *gorm.DB, d Avatar: pipelineAvatar, Status: &statusValue, } - if err := s.logCreateFunc(ctx, db, log); err != nil { + return createFunc(ctx, db, log) +} + +func (s *PipelineExecutor) recordPipelineLog(ctx context.Context, db *gorm.DB, docID, dsl, status string) { + pipelineID := "" + if s.taskCtx.PipelineID != "" { + pipelineID = s.canvasID + } + if err := recordPipelineLog(ctx, db, PipelineLogInput{ + TenantID: s.Tenant().ID, + KbID: s.KB().ID, + DocumentID: docID, + PipelineID: pipelineID, + DSL: dsl, + Status: status, + Document: s.taskCtx.Doc, + }, s.logCreateFunc); err != nil { common.Warn(fmt.Sprintf("failed to record pipeline log: %v", err)) } } diff --git a/internal/ingestion/task/pipeline_executor_test.go b/internal/ingestion/task/pipeline_executor_test.go index 6c94417b0f..ab50c9a7e6 100644 --- a/internal/ingestion/task/pipeline_executor_test.go +++ b/internal/ingestion/task/pipeline_executor_test.go @@ -2,6 +2,7 @@ package task import ( "context" + "errors" "sync" "testing" "time" @@ -315,6 +316,44 @@ func TestRecordPipelineLog_ValidJSONParsed(t *testing.T) { } } +func TestRecordPipelineLog_SharedWriterTerminalWithoutDSL(t *testing.T) { + cleanup := setupPipelineExecutorTestDB(t) + defer cleanup() + + docName := "terminal.pdf" + run := "1" + if err := RecordPipelineLog(t.Context(), dao.DB, PipelineLogInput{ + TenantID: "tenant-1", + KbID: "kb-1", + DocumentID: "doc-1", + Status: "3", + Document: entity.Document{ + ID: "doc-1", + KbID: "kb-1", + ParserID: "naive", + ParserConfig: entity.JSONMap{}, + SourceType: "local", + Type: "pdf", + Name: &docName, + Suffix: ".pdf", + Run: &run, + }, + }); err != nil { + t.Fatalf("RecordPipelineLog: %v", err) + } + + var log entity.PipelineOperationLog + if err := dao.DB.First(&log, "document_id = ?", "doc-1").Error; err != nil { + t.Fatalf("load pipeline log: %v", err) + } + if log.OperationStatus != "3" { + t.Fatalf("OperationStatus = %q, want explicit terminal status", log.OperationStatus) + } + if len(log.DSL) != 0 { + t.Fatalf("DSL = %v, want empty object for terminal writer without DSL", log.DSL) + } +} + func TestRecordPipelineLog_BuiltinUsesParserIDFallback(t *testing.T) { cleanup := setupPipelineExecutorTestDB(t) defer cleanup() @@ -609,6 +648,50 @@ func TestPipelineExecutor_Run_MainFlowWithStubs(t *testing.T) { } } +func TestPipelineExecutor_Execute_DoesNotLogFailedRun(t *testing.T) { + logged := false + svc := mustNewPipelineExecutor(t, makeTaskCtx(), "flow-1", 0). + WithLoadDSLFunc(func(ctx context.Context, canvasID string) (string, string, error) { + return `{"nodes":[{"id":"n1"}],"edges":[]}`, canvasID, nil + }). + WithRunPipelineFunc(func(ctx context.Context, dsl string) (map[string]any, string, error) { + return nil, dsl, errors.New("pipeline failed") + }). + WithLogCreateFunc(func(ctx context.Context, db *gorm.DB, log *entity.PipelineOperationLog) error { + logged = true + return nil + }) + + if _, err := svc.Execute(context.Background()); err == nil { + t.Fatal("Execute error = nil, want failure") + } + if logged { + t.Fatal("executor must not log failed runs before ingestor writes final document status") + } +} + +func TestPipelineExecutor_Execute_DoesNotLogCanceledRun(t *testing.T) { + logged := false + svc := mustNewPipelineExecutor(t, makeTaskCtx(), "flow-1", 0). + WithLoadDSLFunc(func(ctx context.Context, canvasID string) (string, string, error) { + return `{"nodes":[{"id":"n1"}],"edges":[]}`, canvasID, nil + }). + WithRunPipelineFunc(func(ctx context.Context, dsl string) (map[string]any, string, error) { + return nil, dsl, context.Canceled + }). + WithLogCreateFunc(func(ctx context.Context, db *gorm.DB, log *entity.PipelineOperationLog) error { + logged = true + return nil + }) + + if _, err := svc.Execute(context.Background()); !errors.Is(err, context.Canceled) { + t.Fatalf("Execute error = %v, want context.Canceled", err) + } + if logged { + t.Fatal("executor must not log canceled runs before ingestor writes final document status") + } +} + // TestPipelineExecutor_Execute_PropagatesContext verifies the ctx passed to // Execute is the ctx received by runPipelineFunc - the task context must flow // through to the pipeline run. diff --git a/internal/service/ingestion_task_service.go b/internal/service/ingestion_task_service.go index 55674b9f78..60a8ff321b 100644 --- a/internal/service/ingestion_task_service.go +++ b/internal/service/ingestion_task_service.go @@ -515,9 +515,8 @@ func (s *IngestionTaskService) lastRunCount(ctx context.Context, taskID string) // failure. ListAllForAdmin reads run_count back to render the attempt number. // // A corrupted run_count value in an existing row is skipped (the row is -// ignored). A failure to persist the new row is best-effort (logged) and -// does not return an error — matching the legacy semantics that the run -// proceeds even if the counter write fails. +// ignored). A failure to persist the new row is returned so the caller can +// fail the task before running the pipeline. func (s *IngestionTaskService) IncrementRunCount(ctx context.Context, taskID string) error { prevCount, _ := s.lastRunCount(ctx, taskID) @@ -525,8 +524,5 @@ func (s *IngestionTaskService) IncrementRunCount(ctx context.Context, taskID str TaskID: taskID, Checkpoint: entity.JSONMap{stepKeyRunCount: prevCount + 1}, } - if err := s.ingestionTaskLogDAO.Create(ctx, dao.DB, entry); err != nil { - common.Error(fmt.Sprintf("Failed to persist run_count for task %s", taskID), err) - } - return nil + return s.ingestionTaskLogDAO.Create(ctx, dao.DB, entry) }