From 01d351b54ab59569780de70f2d1f680115b83a2b Mon Sep 17 00:00:00 2001 From: euvre <93761161+euvre@users.noreply.github.com> Date: Fri, 14 Aug 2026 04:22:36 -0700 Subject: [PATCH] Go: align cancel-parse error with Python for un-started documents (#18245) --- internal/service/document/document_ingest.go | 4 + internal/service/document/document_parse.go | 34 +++++- internal/service/document/document_test.go | 113 +++++++++++++++++++ 3 files changed, 146 insertions(+), 5 deletions(-) diff --git a/internal/service/document/document_ingest.go b/internal/service/document/document_ingest.go index fd682a04ce..6ddd5cfe04 100644 --- a/internal/service/document/document_ingest.go +++ b/internal/service/document/document_ingest.go @@ -105,6 +105,10 @@ func (s *DocumentService) Ingest(ctx context.Context, userID string, req *Ingest if run == string(entity.TaskStatusCancel) { if err = s.CancelDocParse(ctx, doc); err != nil { common.Error(fmt.Sprintf("go side, start to process %s, run is cancel", doc.ID), err) + if errors.Is(err, errParseNotRunning) { + // Mirror the Python /documents/ingest endpoint's message. + return common.CodeDataError, errors.New("Cannot cancel a task that is not in RUNNING status") + } return common.CodeDataError, err } if err = s.documentDAO.UpdateByID(ctx, dao.DB, doc.ID, map[string]interface{}{ diff --git a/internal/service/document/document_parse.go b/internal/service/document/document_parse.go index 53959a8809..b385dfbf72 100644 --- a/internal/service/document/document_parse.go +++ b/internal/service/document/document_parse.go @@ -273,7 +273,12 @@ func (s *DocumentService) StopParseDocuments(ctx context.Context, datasetID stri successCount := 0 for _, doc := range docs { if cancelErr := s.CancelDocParse(ctx, doc); cancelErr != nil { - errs = append(errs, cancelErr.Error()) + if errors.Is(cancelErr, errParseNotRunning) { + // Mirror the Python /documents/stop endpoint's message. + errs = append(errs, "Can't stop parsing document that has not started or already completed") + } else { + errs = append(errs, cancelErr.Error()) + } continue } successCount++ @@ -330,19 +335,38 @@ func (s *DocumentService) validateDocsInDataset(ctx context.Context, docIDs []st return docs, nil } +// errParseNotRunning is returned by CancelDocParse when the document is not in +// a cancelable state: its run status is neither RUNNING nor CANCEL and it has +// no in-flight ingestion task. Callers map it to their endpoint-specific +// message (the Python /documents/ingest and /documents/stop messages differ). +var errParseNotRunning = errors.New("parse task is not in running status") + // CancelDocParse stops the ingestion task for the document by calling // RequestStop (STOPPING), then marks the document run status as CANCEL. +// It mirrors the Python cancel precondition: only a document whose run status +// is RUNNING or CANCEL, or one with an in-flight ingestion task +// (CREATED/RUNNING/STOPPING), can be canceled; otherwise errParseNotRunning +// is returned. A missing ingestion task is not an error by itself — cancel is +// then a no-op on the task side, matching Python's cancel_all_task_of. func (s *DocumentService) CancelDocParse(ctx context.Context, doc *entity.Document) error { task, err := s.ingestionTaskDAO.GetByDocumentID(ctx, dao.DB, doc.ID) if err != nil { return fmt.Errorf("failed to get ingestion task for %s: %v", doc.ID, err) } - if task == nil { - return fmt.Errorf("no ingestion task found for document %s", doc.ID) + + docRun := "" + if doc.Run != nil { + docRun = *doc.Run + } + inFlight := task != nil && (task.Status == common.CREATED || task.Status == common.RUNNING || task.Status == common.STOPPING) + if docRun != string(entity.TaskStatusRunning) && docRun != string(entity.TaskStatusCancel) && !inFlight { + return errParseNotRunning } - if _, err = s.ingestionTaskSvc.RequestStop(ctx, task.ID); err != nil { - return fmt.Errorf("failed to stop ingestion task %s: %v", task.ID, err) + if task != nil { + if _, err = s.ingestionTaskSvc.RequestStop(ctx, task.ID); err != nil { + return fmt.Errorf("failed to stop ingestion task %s: %v", task.ID, err) + } } if upErr := s.documentDAO.UpdateByID(ctx, dao.DB, doc.ID, map[string]interface{}{"run": string(entity.TaskStatusCancel)}); upErr != nil { diff --git a/internal/service/document/document_test.go b/internal/service/document/document_test.go index 8df895966b..aede64b5f6 100644 --- a/internal/service/document/document_test.go +++ b/internal/service/document/document_test.go @@ -1331,6 +1331,9 @@ func TestStopParseDocuments_NotRunningOrCancel(t *testing.T) { if !ok || len(errors) == 0 { t.Fatal("expected errors in result") } + if errors[0] != "Can't stop parsing document that has not started or already completed" { + t.Fatalf("unexpected error message: %q", errors[0]) + } } func TestStopParseDocuments_UnfinishedTask(t *testing.T) { @@ -2978,6 +2981,116 @@ func TestIngest_CancelDoesNotDeleteIngestionTask(t *testing.T) { } } +// TestIngest_CancelUnstartedDocument mirrors the Python /documents/ingest +// behavior: canceling a document whose parse never started (run=UNSTART, no +// ingestion task) must be rejected with code 102 and the Python message, not +// an internal "no ingestion task found" error. +func TestIngest_CancelUnstartedDocument(t *testing.T) { + db := setupServiceTestDB(t) + pushServiceDB(t, db) + insertUserTenantForAccessCheck(t, "user-1", "tenant-1") + insertTestKB(t, "kb-1", "tenant-1", 0, 0, 0) + insertTestDocWithRun(t, "doc-1", "kb-1", string(entity.TaskStatusUnstart), 10, 5) + + svc := testDocumentService(t) + ctx := t.Context() + code, err := svc.Ingest(ctx, "user-1", &IngestDocumentRequest{ + DocIDs: []string{"doc-1"}, + Run: string(entity.TaskStatusCancel), + }) + if err == nil { + t.Fatal("expected error when canceling an un-started document") + } + if code != common.CodeDataError { + t.Fatalf("expected code %v, got %v", common.CodeDataError, code) + } + if err.Error() != "Cannot cancel a task that is not in RUNNING status" { + t.Fatalf("unexpected error message: %q", err.Error()) + } +} + +// TestIngest_CancelCompletedDocument: canceling an already finished parse +// (run=DONE, ingestion task COMPLETED) is rejected with the same Python +// message as the un-started case. +func TestIngest_CancelCompletedDocument(t *testing.T) { + db := setupServiceTestDB(t) + pushServiceDB(t, db) + insertUserTenantForAccessCheck(t, "user-1", "tenant-1") + insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5) + insertTestDocWithRun(t, "doc-1", "kb-1", string(entity.TaskStatusDone), 10, 5) + insertTestIngestionTaskWithStatus(t, "task-1", "user-1", "doc-1", "kb-1", common.COMPLETED) + + svc := testDocumentService(t) + ctx := t.Context() + code, err := svc.Ingest(ctx, "user-1", &IngestDocumentRequest{ + DocIDs: []string{"doc-1"}, + Run: string(entity.TaskStatusCancel), + }) + if err == nil { + t.Fatal("expected error when canceling a completed document") + } + if code != common.CodeDataError { + t.Fatalf("expected code %v, got %v", common.CodeDataError, code) + } + if err.Error() != "Cannot cancel a task that is not in RUNNING status" { + t.Fatalf("unexpected error message: %q", err.Error()) + } +} + +// TestIngest_CancelUnstartedWithInFlightTask: run=UNSTART but an ingestion +// task was just enqueued (CREATED) — cancel is allowed, matching the Python +// has_unfinished_task branch. +func TestIngest_CancelUnstartedWithInFlightTask(t *testing.T) { + db := setupServiceTestDB(t) + pushServiceDB(t, db) + insertUserTenantForAccessCheck(t, "user-1", "tenant-1") + insertTestKB(t, "kb-1", "tenant-1", 0, 0, 0) + insertTestDocWithRun(t, "doc-1", "kb-1", string(entity.TaskStatusUnstart), 10, 5) + insertTestIngestionTask(t, "task-1", "user-1", "doc-1", "kb-1") + + svc := testDocumentService(t) + ctx := t.Context() + code, err := svc.Ingest(ctx, "user-1", &IngestDocumentRequest{ + DocIDs: []string{"doc-1"}, + Run: string(entity.TaskStatusCancel), + }) + if err != nil { + t.Fatalf("Ingest(cancel) with in-flight task: %v", err) + } + if code != common.CodeSuccess { + t.Fatalf("expected code %v, got %v", common.CodeSuccess, code) + } + + doc, _ := dao.NewDocumentDAO().GetByID(ctx, db, "doc-1") + if doc == nil || doc.Run == nil || *doc.Run != string(entity.TaskStatusCancel) { + t.Fatalf("expected doc run=CANCEL, got %v", doc.Run) + } +} + +// TestIngest_CancelAgainWithoutTask: re-canceling a document already in +// CANCEL state is accepted even when its ingestion task is gone — Python +// treats run=CANCEL as cancelable and cancel_all_task_of is a no-op. +func TestIngest_CancelAgainWithoutTask(t *testing.T) { + db := setupServiceTestDB(t) + pushServiceDB(t, db) + insertUserTenantForAccessCheck(t, "user-1", "tenant-1") + insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5) + insertTestDocWithRun(t, "doc-1", "kb-1", string(entity.TaskStatusCancel), 10, 5) + + svc := testDocumentService(t) + ctx := t.Context() + code, err := svc.Ingest(ctx, "user-1", &IngestDocumentRequest{ + DocIDs: []string{"doc-1"}, + Run: string(entity.TaskStatusCancel), + }) + if err != nil { + t.Fatalf("Ingest(re-cancel) without task: %v", err) + } + if code != common.CodeSuccess { + t.Fatalf("expected code %v, got %v", common.CodeSuccess, code) + } +} + func TestUpdateRunProgressMirrorsFields(t *testing.T) { db := setupServiceTestDB(t) pushServiceDB(t, db)