fix(ingestion): accumulate component progress lines into document.progress_msg (#18356)

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

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) {