diff --git a/internal/ingestion/service/progress_sink.go b/internal/ingestion/service/progress_sink.go index 9150479942..c48b1f86b6 100644 --- a/internal/ingestion/service/progress_sink.go +++ b/internal/ingestion/service/progress_sink.go @@ -19,6 +19,8 @@ package service import ( "context" "fmt" + "strings" + "sync" "sync/atomic" "ragflow/internal/common" @@ -29,6 +31,11 @@ import ( documentpkg "ragflow/internal/service/document" ) +// progressLogMaxChars bounds the accumulated run log mirrored into +// document.progress_msg. Mirrors Python's TASK_MAX_LOG_LENGTH +// (api/db/services/task_service.py). +const progressLogMaxChars = 3000 + // progressSink implements pipeline.ProgressSink. It is the single writer of // the document / ingestion_task_log / ingestion_task.component_total tables // for a pipeline run: the pipeline reports component lifecycle events here @@ -44,6 +51,12 @@ type progressSink struct { // which eino fires from concurrent parallel-branch goroutines. Atomic because // the two access paths share no other synchronization. total atomic.Int64 + // logMu guards the accumulated run log below: OnComponentProgress fires + // from concurrent parallel-branch goroutines, so both the lazy seed and + // the append must hold the same mutex. + logMu sync.Mutex + seeded bool + log strings.Builder } // docProgressSvc is the subset of *service.DocumentService the sink needs to @@ -51,6 +64,7 @@ type progressSink struct { // tests can inject a stub and assert the mirror call without depending on the // full DocumentService surface. type docProgressSvc interface { + GetDocumentByID(ctx context.Context, docID string) (*documentpkg.DocumentResponse, error) UpdateRunProgress(ctx context.Context, docID string, progress float64, run, progressMsg string) error } @@ -90,11 +104,64 @@ func (s *progressSink) OnComponentProgress(ctx context.Context, ev pipeline.Prog return } progress, run := deriveDocumentProgress(agg, int(total)) - if err = s.docSvc.UpdateRunProgress(ctx, ev.DocumentID, progress, run, ev.Message); err != nil { + if err = s.docSvc.UpdateRunProgress(ctx, ev.DocumentID, progress, run, s.accumulateLog(ctx, ev.DocumentID, ev.Message)); err != nil { common.Error(fmt.Sprintf("progressSink: mirror progress to document %s for task %s failed: %v", ev.DocumentID, ev.TaskID, err), err) } } +// accumulateLog appends one component lifecycle line to the run log and +// returns the full accumulated log for mirroring into document.progress_msg. +// The document-details dialog renders progress_msg verbatim as a multi-line +// log (whitespace-pre-line), matching Python's append-and-trim semantics in +// TaskService.update_progress; a single overwrite would leave the dialog +// showing only the latest line. The buffer is lazily seeded from the +// document row on the first event so a resumed run keeps the lines logged +// before the crash/redelivery instead of restarting empty (StartRunning +// resets the row to "" on a fresh run, which then seeds an empty log). +func (s *progressSink) accumulateLog(ctx context.Context, docID, msg string) string { + s.logMu.Lock() + defer s.logMu.Unlock() + if !s.seeded { + s.seeded = true + doc, err := s.docSvc.GetDocumentByID(ctx, docID) + switch { + case err != nil: + common.Warn(fmt.Sprintf("progressSink: seed run log from document %s: %v", docID, err)) + case doc != nil && doc.ProgressMsg != nil: + s.log.WriteString(*doc.ProgressMsg) + } + } + if msg == "" { + return s.log.String() + } + if s.log.Len() > 0 { + s.log.WriteByte('\n') + } + s.log.WriteString(msg) + log := trimLogHead(s.log.String(), progressLogMaxChars) + if len(log) != s.log.Len() { + s.log.Reset() + s.log.WriteString(log) + } + return log +} + +// trimLogHead drops whole lines from the head of text until the remainder +// fits maxChars, keeping the newest lines. Mirrors Python's +// trim_header_by_lines: text without a fitting newline boundary is returned +// unchanged. +func trimLogHead(text string, maxChars int) string { + if len(text) <= maxChars { + return text + } + for i := 0; i < len(text); i++ { + if text[i] == '\n' && len(text)-i <= maxChars { + return text[i+1:] + } + } + return text +} + // deriveDocumentProgress computes the document-level progress (0..1) and run // label ("0".."4", matching Python's document.run enum) from the aggregated // ingestion_task_log. This logic is owned by the sink (the document-table diff --git a/internal/ingestion/service/progress_sink_test.go b/internal/ingestion/service/progress_sink_test.go index 5f4e49da4b..0c1cff2bc6 100644 --- a/internal/ingestion/service/progress_sink_test.go +++ b/internal/ingestion/service/progress_sink_test.go @@ -18,7 +18,9 @@ package service import ( "context" + "fmt" "runtime" + "strings" "sync" "testing" @@ -144,6 +146,12 @@ type stubDocProgressSvc struct { gotRun string gotMsg string calls int + doc *document.DocumentResponse + docErr error +} + +func (s *stubDocProgressSvc) GetDocumentByID(ctx context.Context, docID string) (*document.DocumentResponse, error) { + return s.doc, s.docErr } func (s *stubDocProgressSvc) UpdateRunProgress(ctx context.Context, docID string, progress float64, run, progressMsg string) error { @@ -215,6 +223,147 @@ func TestProgressSinkPersistsViaService(t *testing.T) { } } +// TestProgressSinkAccumulatesProgressLog pins the core fix: document.progress_msg +// is the accumulated multi-line run log, not just the latest component line. +// The document-details dialog renders progress_msg verbatim (whitespace-pre-line). +func TestProgressSinkAccumulatesProgressLog(t *testing.T) { + db := testutil.SetupTestDB(t) + cleanup := testutil.ReplaceDBForTest(t, db) + defer cleanup() + _, _, docID, taskID := testutil.SeedTestData(t, db, testutil.WithPipelineID("flow-1")) + + ctx := t.Context() + sink := newProgressSink(ctx, servicepkg.NewIngestionTaskService()) + stub := &stubDocProgressSvc{} + sink.docSvc = stub + sink.OnComponentTotal(ctx, taskID, 2) + + sink.OnComponentProgress(ctx, pipeline.ProgressEvent{ + TaskID: taskID, DocumentID: docID, Component: "File", Phase: 1, Message: "File:naive Done", + }) + sink.OnComponentProgress(ctx, pipeline.ProgressEvent{ + TaskID: taskID, DocumentID: docID, Component: "Parser", Phase: 1, Message: "Parser Done", + }) + + want := "File:naive Done\nParser Done" + if stub.gotMsg != want { + t.Fatalf("progress_msg = %q, want multi-line log %q", stub.gotMsg, want) + } +} + +// TestProgressSinkSeedsLogFromDocument verifies the accumulated log keeps the +// lines already stored on the document row, so a resumed run (post-crash +// redelivery) appends to the pre-crash log instead of restarting empty. +func TestProgressSinkSeedsLogFromDocument(t *testing.T) { + db := testutil.SetupTestDB(t) + cleanup := testutil.ReplaceDBForTest(t, db) + defer cleanup() + _, _, docID, taskID := testutil.SeedTestData(t, db, testutil.WithPipelineID("flow-1")) + + ctx := t.Context() + sink := newProgressSink(ctx, servicepkg.NewIngestionTaskService()) + seed := "File:naive Done" + stub := &stubDocProgressSvc{doc: &document.DocumentResponse{ProgressMsg: &seed}} + sink.docSvc = stub + sink.OnComponentTotal(ctx, taskID, 2) + + sink.OnComponentProgress(ctx, pipeline.ProgressEvent{ + TaskID: taskID, DocumentID: docID, Component: "Parser", Phase: 1, Message: "Parser Done", + }) + + want := "File:naive Done\nParser Done" + if stub.gotMsg != want { + t.Fatalf("progress_msg = %q, want seeded multi-line log %q", stub.gotMsg, want) + } +} + +// TestProgressSinkTrimsLogHead verifies the accumulated log is head-trimmed at +// line boundaries once it exceeds progressLogMaxChars, keeping the newest lines +// (Python trim_header_by_lines parity). +func TestProgressSinkTrimsLogHead(t *testing.T) { + db := testutil.SetupTestDB(t) + cleanup := testutil.ReplaceDBForTest(t, db) + defer cleanup() + _, _, docID, taskID := testutil.SeedTestData(t, db, testutil.WithPipelineID("flow-1")) + + ctx := t.Context() + sink := newProgressSink(ctx, servicepkg.NewIngestionTaskService()) + stub := &stubDocProgressSvc{} + sink.docSvc = stub + sink.OnComponentTotal(ctx, taskID, 2) + + head := strings.Repeat("h", 2000) + tail := strings.Repeat("t", 2000) + sink.OnComponentProgress(ctx, pipeline.ProgressEvent{ + TaskID: taskID, DocumentID: docID, Component: "File", Phase: 1, Message: head, + }) + sink.OnComponentProgress(ctx, pipeline.ProgressEvent{ + TaskID: taskID, DocumentID: docID, Component: "Parser", Phase: 1, Message: tail, + }) + + if len(stub.gotMsg) > progressLogMaxChars { + t.Fatalf("progress_msg len = %d, exceeds %d", len(stub.gotMsg), progressLogMaxChars) + } + if stub.gotMsg != tail { + t.Fatalf("progress_msg keeps the newest line: got prefix %q, want %q", stub.gotMsg[:20], tail[:20]) + } +} + +// TestProgressSink_Log_NoDataRace hits the accumulated log buffer directly: +// OnComponentProgress fires from concurrent parallel-branch goroutines, so the +// lazy seed and the append must be mutex-guarded. The direct call avoids the +// test-DB serialization inside OnComponentProgress that would mask the race. +func TestProgressSink_Log_NoDataRace(t *testing.T) { + db := testutil.SetupTestDB(t) + cleanup := testutil.ReplaceDBForTest(t, db) + defer cleanup() + + ctx := t.Context() + sink := newProgressSink(ctx, servicepkg.NewIngestionTaskService()) + sink.docSvc = &stubDocProgressSvc{} + + const n = 30 + var wg sync.WaitGroup + start := make(chan struct{}) + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + sink.accumulateLog(ctx, "doc-1", fmt.Sprintf("line-%d", i)) + }(i) + } + close(start) + wg.Wait() + + got := sink.accumulateLog(ctx, "doc-1", "") + if lines := strings.Split(got, "\n"); len(lines) != n { + t.Fatalf("accumulated log lines = %d, want %d: %q", len(lines), n, got) + } +} + +func TestTrimLogHead(t *testing.T) { + tests := []struct { + name string + text string + max int + want string + }{ + {name: "short text unchanged", text: "a\nb", max: 10, want: "a\nb"}, + {name: "exactly max unchanged", text: "a\nb", max: 3, want: "a\nb"}, + {name: "drops head line", text: "l1\nl2\nl3", max: 6, want: "l2\nl3"}, + {name: "drops lines until remainder fits", text: "l1\nl2\nl3", max: 5, want: "l3"}, + {name: "no fitting newline unchanged", text: "abcdef", max: 3, want: "abcdef"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := trimLogHead(tt.text, tt.max); got != tt.want { + t.Errorf("trimLogHead(%q, %d) = %q, want %q", tt.text, tt.max, got, tt.want) + } + }) + } +} + // TestProgressSinkEmptyDocumentIDSkipsMirror verifies the log row is still // recorded when no owning document is bound, but the document mirror is skipped. func TestProgressSinkEmptyDocumentIDSkipsMirror(t *testing.T) {